Fix Trace Stream interceptors when RecvWithContext returns an error (#557)

* Fix Trace Stream interceptors when `RecvWithContext` returns an error

The implementation of the Trace Stream interceptors' `RecvWithContext`
handling did not take into account an error return where the message
(result or payload) would be `nil`. This change changes the way
`RecvWithContext` calls are intercepted so that if the `next` returns an
error, the interceptor will populate the receive context using the
passed in context and not even attempt to gather any OTel trace metadata
from a message (which usually would not be available in the error case,
but may be possible if another Goa interceptor were in the chain).

I don't believe it is worthwhile to support a case where an error is
returned by `next` along with a non-`nil` message, but if we did, we
would need to resort to using reflection due to the current state of Go
generics and `nil` checking:

```go
func traceStreamRecv[Message TraceStreamStreamingRecvMessage](
    ctx context.Context,
    info goa.InterceptorInfo,
    next goa.Endpoint,
    streamingMessage func(any) Message,
) (any, error) {
    msg, err := next(ctx, info.RawPayload())
    propagator := otel.GetTextMapPropagator()
    sm := streamingMessage(msg)
    rc, ok :=
ctx.Value(traceStreamRecvContextKey).(*traceStreamRecvContext)
    if !ok {
        panic(fmt.Errorf("clue interceptors trace stream receive method
called without prior setup (service: %v, method: %v)", info.Service(),
info.Method()))
    }
    if reflect.ValueOf(sm).Elem().IsZero() {
        rc.ctx = ctx
    } else {
        rc.ctx = propagator.Extract(ctx,
propagation.MapCarrier(sm.TraceMetadata()))
    }
    return msg, err
}
```

* Return `nil` instead of `err` since it is unnecessary

Author: douglaswth

interceptors/trace_stream.go

@@ -98,14 +98,18 @@ func traceStreamRecv[Message TraceStreamStreamingRecvMessage](
 	streamingMessage func(any) Message,
 ) (any, error) {
 	msg, err := next(ctx, info.RawPayload())
-	propagator := otel.GetTextMapPropagator()
-	sm := streamingMessage(msg)
 	rc, ok := ctx.Value(traceStreamRecvContextKey).(*traceStreamRecvContext)
 	if !ok {
 		panic(fmt.Errorf("clue interceptors trace stream receive method called without prior setup (service: %v, method: %v)", info.Service(), info.Method()))
 	}
+	if err != nil {
+		rc.ctx = ctx
+		return nil, err
+	}
+	propagator := otel.GetTextMapPropagator()
+	sm := streamingMessage(msg)
 	rc.ctx = propagator.Extract(ctx, propagation.MapCarrier(sm.TraceMetadata()))
-	return msg, err
+	return msg, nil
 }
 
 // traceStreamWrapRecvAndReturnContext is a helper function for wrapped trace stream receive methods

interceptors/trace_stream_client_test.go

@@ -82,9 +82,6 @@ func TestTraceBidirectionalStreamClientInterceptor(t *testing.T) {
 		info.addRawPayload(func() any {
 			return nil
 		})
-		info.addClientStreamingResult(func(res any) *mockTraceStreamingRecvMessage {
-			return newMockTraceStreamingRecvMessage(assert)
-		})
 		info.addService(func() string {
 			return "TestService"
 		})
@@ -157,6 +154,37 @@ func TestTraceBidirectionalStreamClientInterceptor(t *testing.T) {
 		assert.False(payload.hasMore(), "missing expected payload calls")
 	})
 
+	t.Run("receive with error", func(t *testing.T) {
+		var (
+			ctx         = log.Context(context.Background(), log.WithFormat(log.FormatText))
+			info        = newMockTraceStreamInfo(assert.New(t))
+			interceptor = &TraceBidirectionalStreamClientInterceptor[*mockTraceStreamInfo, *mockTraceStreamingSendMessage, *mockTraceStreamingRecvMessage]{}
+			nextCalled  = false
+			next        = func(ctx context.Context, _ any) (any, error) {
+				nextCalled = true
+				return nil, assert.AnError
+			}
+		)
+		info.addCallType(func() goa.InterceptorCallType {
+			return goa.InterceptorStreamingRecv
+		})
+		info.addRawPayload(func() any {
+			return nil
+		})
+
+		ctx = SetupTraceStreamRecvContext(ctx)
+		res, err := interceptor.TraceBidirectionalStream(ctx, info, next)
+		assert.ErrorIs(t, err, assert.AnError)
+		assert.Nil(t, res)
+
+		assert.NotPanics(t, func() {
+			ctx = GetTraceStreamRecvContext(ctx)
+		})
+
+		assert.True(t, nextCalled, "missing expected next call")
+		assert.False(t, info.hasMore(), "missing expected interceptor info calls")
+	})
+
 	t.Run("unary", func(t *testing.T) {
 		var (
 			assert      = assert.New(t)
@@ -205,9 +233,6 @@ func TestTraceServerToClientStreamClientInterceptor(t *testing.T) {
 		info.addRawPayload(func() any {
 			return nil
 		})
-		info.addClientStreamingResult(func(res any) *mockTraceStreamingRecvMessage {
-			return newMockTraceStreamingRecvMessage(assert)
-		})
 		info.addService(func() string {
 			return "TestService"
 		})
@@ -280,6 +305,37 @@ func TestTraceServerToClientStreamClientInterceptor(t *testing.T) {
 		assert.False(payload.hasMore(), "missing expected payload calls")
 	})
 
+	t.Run("receive with error", func(t *testing.T) {
+		var (
+			ctx         = log.Context(context.Background(), log.WithFormat(log.FormatText))
+			info        = newMockTraceStreamInfo(assert.New(t))
+			interceptor = &TraceServerToClientStreamClientInterceptor[*mockTraceStreamInfo, *mockTraceStreamingRecvMessage]{}
+			nextCalled  = false
+			next        = func(ctx context.Context, _ any) (any, error) {
+				nextCalled = true
+				return nil, assert.AnError
+			}
+		)
+		info.addCallType(func() goa.InterceptorCallType {
+			return goa.InterceptorStreamingRecv
+		})
+		info.addRawPayload(func() any {
+			return nil
+		})
+
+		ctx = SetupTraceStreamRecvContext(ctx)
+		res, err := interceptor.TraceServerToClientStream(ctx, info, next)
+		assert.ErrorIs(t, err, assert.AnError)
+		assert.Nil(t, res)
+
+		assert.NotPanics(t, func() {
+			ctx = GetTraceStreamRecvContext(ctx)
+		})
+
+		assert.True(t, nextCalled, "missing expected next call")
+		assert.False(t, info.hasMore(), "missing expected interceptor info calls")
+	})
+
 	t.Run("unary", func(t *testing.T) {
 		var (
 			assert      = assert.New(t)

interceptors/trace_stream_server_test.go

@@ -82,9 +82,6 @@ func TestTraceBidirectionalStreamServerInterceptor(t *testing.T) {
 		info.addRawPayload(func() any {
 			return nil
 		})
-		info.addServerStreamingPayload(func(pay any) *mockTraceStreamingRecvMessage {
-			return newMockTraceStreamingRecvMessage(assert)
-		})
 		info.addService(func() string {
 			return "TestService"
 		})
@@ -157,6 +154,37 @@ func TestTraceBidirectionalStreamServerInterceptor(t *testing.T) {
 		assert.False(payload.hasMore(), "missing expected payload calls")
 	})
 
+	t.Run("receive with error", func(t *testing.T) {
+		var (
+			ctx         = log.Context(context.Background(), log.WithFormat(log.FormatText))
+			info        = newMockTraceStreamInfo(assert.New(t))
+			interceptor = &TraceBidirectionalStreamServerInterceptor[*mockTraceStreamInfo, *mockTraceStreamingRecvMessage, *mockTraceStreamingSendMessage]{}
+			nextCalled  = false
+			next        = func(ctx context.Context, payload any) (any, error) {
+				nextCalled = true
+				return nil, assert.AnError
+			}
+		)
+		info.addCallType(func() goa.InterceptorCallType {
+			return goa.InterceptorStreamingRecv
+		})
+		info.addRawPayload(func() any {
+			return nil
+		})
+
+		ctx = SetupTraceStreamRecvContext(ctx)
+		res, err := interceptor.TraceBidirectionalStream(ctx, info, next)
+		assert.ErrorIs(t, err, assert.AnError)
+		assert.Nil(t, res)
+
+		assert.NotPanics(t, func() {
+			ctx = GetTraceStreamRecvContext(ctx)
+		})
+
+		assert.True(t, nextCalled, "missing expected next call")
+		assert.False(t, info.hasMore(), "missing expected interceptor info calls")
+	})
+
 	t.Run("unary", func(t *testing.T) {
 		var (
 			assert      = assert.New(t)
@@ -285,9 +313,6 @@ func TestTraceClientToServerStreamServerInterceptor(t *testing.T) {
 		info.addRawPayload(func() any {
 			return nil
 		})
-		info.addServerStreamingPayload(func(pay any) *mockTraceStreamingRecvMessage {
-			return newMockTraceStreamingRecvMessage(assert)
-		})
 		info.addService(func() string {
 			return "TestService"
 		})
@@ -360,6 +385,37 @@ func TestTraceClientToServerStreamServerInterceptor(t *testing.T) {
 		assert.False(payload.hasMore(), "missing expected payload calls")
 	})
 
+	t.Run("receive with error", func(t *testing.T) {
+		var (
+			ctx         = log.Context(context.Background(), log.WithFormat(log.FormatText))
+			info        = newMockTraceStreamInfo(assert.New(t))
+			interceptor = &TraceClientToServerStreamServerInterceptor[*mockTraceStreamInfo, *mockTraceStreamingRecvMessage]{}
+			nextCalled  = false
+			next        = func(ctx context.Context, payload any) (any, error) {
+				nextCalled = true
+				return nil, assert.AnError
+			}
+		)
+		info.addCallType(func() goa.InterceptorCallType {
+			return goa.InterceptorStreamingRecv
+		})
+		info.addRawPayload(func() any {
+			return nil
+		})
+
+		ctx = SetupTraceStreamRecvContext(ctx)
+		res, err := interceptor.TraceClientToServerStream(ctx, info, next)
+		assert.ErrorIs(t, err, assert.AnError)
+		assert.Nil(t, res)
+
+		assert.NotPanics(t, func() {
+			ctx = GetTraceStreamRecvContext(ctx)
+		})
+
+		assert.True(t, nextCalled, "missing expected next call")
+		assert.False(t, info.hasMore(), "missing expected interceptor info calls")
+	})
+
 	t.Run("unary", func(t *testing.T) {
 		var (
 			assert      = assert.New(t)