The fan-out logic in captureRawProviderStream has several critical issues:

1. Data Loss & Inconsistency: Using select with a default case causes events to be silently dropped if either the pipeline or the client is slow. This results in corrupted streams for the client and incorrect usage tracking for the pipeline. It should block instead to apply backpressure.
2. Race Condition: The current defer ordering closes the channels before RawStreamErr is populated. Consumers calling Err() immediately after the stream ends (when Next() returns false) may receive a nil error instead of the actual provider error.
3. Context Cancellation: The goroutine does not check for ctx.Done(), which can lead to it hanging if a channel send blocks while the request is aborted. Note that while long-lived application-level goroutines may omit cancellation, request-scoped goroutines like this one must handle it.

		go func() {
			defer stream.Close()
			defer close(outbound.state.RawStreamCh)
			defer close(pipelineCh)
			defer func() {
				outbound.state.RawStreamErr = stream.Err()
			}()

			for stream.Next() {
				event := stream.Current()
				select {
				case pipelineCh <- event:
				case <-ctx.Done():
					return
				}

				select {
				case outbound.state.RawStreamCh <- event:
				case <-ctx.Done():
					return
				}
			}
		}()

The Next() method performs a blocking receive on the channel without considering context cancellation. This can lead to goroutine leaks if the producer is stuck and the request context is cancelled. The passThroughChannelStream should ideally hold a reference to the request context and check ctx.Done() in its Next() method.