fix(grpc+consumer): preserve context logger enrichment end-to-end

internal/events/rabbitmq/consumer/handler.go

@@ -40,7 +40,16 @@ func (r *rabbitmqConsumer) handler(msgs <-chan amqp.Delivery, handler events.Han
 			),
 		)
 
-		logger := r.logger.With().Str("topic", topic).Ctx(ctx).Logger()
+		logCtx := r.logger.With().Str("topic", topic).Ctx(ctx)
+		// Enrich with the OTEL trace_id so every log across the message
+		// lifecycle (decode errors, ack/nack failures, retries, and the
+		// application handler itself) carries the same id as the producer
+		// span. ExtractTrace above already restored the trace context from
+		// the message headers.
+		if sc := trace.SpanContextFromContext(ctx); sc.HasTraceID() {
+			logCtx = logCtx.Str("trace_id", sc.TraceID().String())
+		}
+		logger := logCtx.Logger()
 		ctx = logger.WithContext(ctx)
 		ctx = thunderContext.ContextWithMetadata(ctx, metadataFromAmqpTable(msg.Headers))
 		// ensures that the correlation ID is propagated or generated

pkg/grpc/server.go

@@ -26,12 +26,7 @@ type NewServerParams struct {
 func NewServer(params NewServerParams) *BareServer {
 	grpcServer := &BareServer{}
 
-	// We want to add the MetadataPropagator interceptor first and
-	// logger interceptor last.
-	params.Interceptors = append(
-		[]grpc.UnaryServerInterceptor{UnaryServerMetadataPropagator},
-		append(params.Interceptors, grpcLoggerInterceptor(params.Logger))...,
-	)
+	params.Interceptors = composeInterceptors(params.Logger, params.Interceptors)
 
 	// default max message size is 4MB
 	maxReceiveMessageSize := 4 * 1024 * 1024
@@ -81,3 +76,19 @@ func grpcLoggerInterceptor(logger *zerolog.Logger) func(context.Context, interfa
 		return h, err
 	}
 }
+
+// composeInterceptors builds the server's unary interceptor chain:
+// MetadataPropagator first, then the base logger, then the supplied
+// interceptors. The base logger MUST come before supplied interceptors:
+// they enrich the context logger (e.g. audit fields, trace_id), and if
+// the base logger ran after them it would replace the context logger and
+// silently drop that enrichment before the handler runs.
+func composeInterceptors(logger *zerolog.Logger, supplied []grpc.UnaryServerInterceptor) []grpc.UnaryServerInterceptor {
+	return append(
+		[]grpc.UnaryServerInterceptor{
+			UnaryServerMetadataPropagator,
+			grpcLoggerInterceptor(logger),
+		},
+		supplied...,
+	)
+}

pkg/grpc/server_test.go

@@ -0,0 +1,103 @@
+package grpc
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"google.golang.org/grpc"
+)
+
+// chain replicates grpc.ChainUnaryInterceptor semantics: interceptors run in
+// slice order, each wrapping the next, with the handler innermost.
+func chain(interceptors []grpc.UnaryServerInterceptor, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) grpc.UnaryHandler {
+	chained := handler
+	for i := len(interceptors) - 1; i >= 0; i-- {
+		ic := interceptors[i]
+		next := chained
+		chained = func(ctx context.Context, req interface{}) (interface{}, error) {
+			return ic(ctx, req, info, next)
+		}
+	}
+	return chained
+}
+
+// TestComposedChainPreservesSuppliedEnrichment is the regression for the
+// interceptor-ordering bug: the base logger interceptor used to be appended
+// AFTER supplied interceptors, replacing the context logger and silently
+// dropping their enrichment (trace_id, audit fields) before the handler ran.
+//
+// Unlike a hand-assembled chain, this test takes the interceptor slice from
+// composeInterceptors — the same production function NewServer uses — so
+// reverting the ordering there makes this test fail.
+func TestComposedChainPreservesSuppliedEnrichment(t *testing.T) {
+	var buf bytes.Buffer
+	logger := zerolog.New(&buf)
+
+	// Supplied interceptor: enriches the context logger, exactly like
+	// backend-commons logs.UnaryServerInterceptor does for trace_id.
+	enriching := func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (interface{}, error) {
+		l := zerolog.Ctx(ctx).With().
+			Str("trace_id", "trace-abc-123").
+			Str("grpc_method", info.FullMethod).
+			Logger()
+		return handler(l.WithContext(ctx), req)
+	}
+
+	// Compose through the PRODUCTION function used by NewServer.
+	interceptors := composeInterceptors(&logger, []grpc.UnaryServerInterceptor{enriching})
+
+	// Handler logs from its context, like real service handlers do.
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		zerolog.Ctx(ctx).Info().Msg("handled")
+		return "ok", nil
+	}
+
+	info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"}
+	if _, err := chain(interceptors, info, handler)(context.Background(), "req"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	var entry map[string]any
+	if err := json.Unmarshal(buf.Bytes(), &entry); err != nil {
+		t.Fatalf("failed to parse handler log: %v\noutput: %s", err, buf.String())
+	}
+
+	if entry["trace_id"] != "trace-abc-123" {
+		t.Errorf("handler log trace_id = %v, want trace-abc-123 (supplied enrichment was dropped)", entry["trace_id"])
+	}
+	if entry["grpc_method"] != "/test.Service/Do" {
+		t.Errorf("handler log grpc_method = %v, want /test.Service/Do", entry["grpc_method"])
+	}
+}
+
+// TestComposedChainBaseLoggerReachesHandlerWithoutSupplied guards the base
+// behavior: with no supplied interceptors, the handler still gets the base
+// logger in its context.
+func TestComposedChainBaseLoggerReachesHandlerWithoutSupplied(t *testing.T) {
+	var buf bytes.Buffer
+	logger := zerolog.New(&buf)
+
+	interceptors := composeInterceptors(&logger, nil)
+
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		zerolog.Ctx(ctx).Info().Str("marker", "base").Msg("handled")
+		return "ok", nil
+	}
+
+	info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"}
+	if _, err := chain(interceptors, info, handler)(context.Background(), "req"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !bytes.Contains(buf.Bytes(), []byte(`"marker":"base"`)) {
+		t.Errorf("handler did not log through the base logger: %s", buf.String())
+	}
+}