From 42d7a6975f92677a54e79d82606a12c0cf4ec6ba Mon Sep 17 00:00:00 2001 From: ruangervasi-altpay Date: Mon, 13 Jul 2026 22:48:38 -0300 Subject: [PATCH 1/2] fix(grpc+consumer): preserve context logger enrichment end-to-end Two fixes for trace_id propagation into logs: 1. gRPC server: the base logger interceptor ran AFTER supplied interceptors, replacing the context logger and silently dropping any enrichment (trace_id, audit fields) before the handler executed. Reordered to metadata -> base logger -> supplied interceptors -> handler, so supplied interceptors (e.g. backend-commons logs.UnaryServerInterceptor) enrich a logger that actually reaches the handler. Regression test asserts the handler's log line carries the enriched trace_id and grpc_method through the full chain. 2. RabbitMQ consumer: the lifecycle logger (built before the handler runs) now includes the OTEL trace_id extracted from the message headers. This covers decode errors, ack/nack failures, retries and the application handler itself -- not just handler logs enriched by downstream middleware. --- internal/events/rabbitmq/consumer/handler.go | 11 ++- pkg/grpc/server.go | 14 +++- pkg/grpc/server_test.go | 81 ++++++++++++++++++++ 3 files changed, 101 insertions(+), 5 deletions(-) create mode 100644 pkg/grpc/server_test.go diff --git a/internal/events/rabbitmq/consumer/handler.go b/internal/events/rabbitmq/consumer/handler.go index e40b439..4778d54 100644 --- a/internal/events/rabbitmq/consumer/handler.go +++ b/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 diff --git a/pkg/grpc/server.go b/pkg/grpc/server.go index d27fe8b..8be58d9 100644 --- a/pkg/grpc/server.go +++ b/pkg/grpc/server.go @@ -26,11 +26,17 @@ type NewServerParams struct { func NewServer(params NewServerParams) *BareServer { grpcServer := &BareServer{} - // We want to add the MetadataPropagator interceptor first and - // logger interceptor last. + // 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. params.Interceptors = append( - []grpc.UnaryServerInterceptor{UnaryServerMetadataPropagator}, - append(params.Interceptors, grpcLoggerInterceptor(params.Logger))..., + []grpc.UnaryServerInterceptor{ + UnaryServerMetadataPropagator, + grpcLoggerInterceptor(params.Logger), + }, + params.Interceptors..., ) // default max message size is 4MB diff --git a/pkg/grpc/server_test.go b/pkg/grpc/server_test.go new file mode 100644 index 0000000..4a4ef1c --- /dev/null +++ b/pkg/grpc/server_test.go @@ -0,0 +1,81 @@ +package grpc + +import ( + "bytes" + "context" + "encoding/json" + "testing" + + "github.com/rs/zerolog" + "google.golang.org/grpc" +) + +// TestSuppliedInterceptorEnrichmentReachesHandler is the regression for the +// interceptor-ordering bug: the base logger interceptor used to run AFTER +// supplied interceptors, replacing the context logger and silently dropping +// any enrichment (trace_id, audit fields) before the handler executed. +// +// It simulates a supplied interceptor that enriches the context logger (the +// same pattern backend-commons logs.UnaryServerInterceptor uses) and asserts +// the handler's log line actually carries the enriched fields. +func TestSuppliedInterceptorEnrichmentReachesHandler(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 the chain the same way NewServer does. + interceptors := append( + []grpc.UnaryServerInterceptor{ + UnaryServerMetadataPropagator, + grpcLoggerInterceptor(&logger), + }, + 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 + } + + // Manually chain (mirrors grpc.ChainUnaryInterceptor semantics). + chained := handler + for i := len(interceptors) - 1; i >= 0; i-- { + ic := interceptors[i] + next := chained + info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"} + chained = func(ctx context.Context, req interface{}) (interface{}, error) { + return ic(ctx, req, info, next) + } + } + + if _, err := chained(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 (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"]) + } +} From 9ddf2e7612641bc647b442178eaac7300af90743 Mon Sep 17 00:00:00 2001 From: ruangervasi-altpay Date: Sun, 9 Aug 2026 23:01:04 -0300 Subject: [PATCH 2/2] test(grpc): guard interceptor ordering via production composition Review on the consuming side proved the previous test was not a guard: it hardcoded the corrected ordering in the test body and never called NewServer, so reverting the fix left it green. Extract composeInterceptors (used by NewServer) and drive the test through it: reverting the ordering there now fails the test. Also add a base-behavior test (no supplied interceptors). --- pkg/grpc/server.go | 29 ++++++++------- pkg/grpc/server_test.go | 80 ++++++++++++++++++++++++++--------------- 2 files changed, 68 insertions(+), 41 deletions(-) diff --git a/pkg/grpc/server.go b/pkg/grpc/server.go index 8be58d9..71ca068 100644 --- a/pkg/grpc/server.go +++ b/pkg/grpc/server.go @@ -26,18 +26,7 @@ type NewServerParams struct { func NewServer(params NewServerParams) *BareServer { grpcServer := &BareServer{} - // 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. - params.Interceptors = append( - []grpc.UnaryServerInterceptor{ - UnaryServerMetadataPropagator, - grpcLoggerInterceptor(params.Logger), - }, - params.Interceptors..., - ) + params.Interceptors = composeInterceptors(params.Logger, params.Interceptors) // default max message size is 4MB maxReceiveMessageSize := 4 * 1024 * 1024 @@ -87,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..., + ) +} diff --git a/pkg/grpc/server_test.go b/pkg/grpc/server_test.go index 4a4ef1c..2fe1f73 100644 --- a/pkg/grpc/server_test.go +++ b/pkg/grpc/server_test.go @@ -10,15 +10,29 @@ import ( "google.golang.org/grpc" ) -// TestSuppliedInterceptorEnrichmentReachesHandler is the regression for the -// interceptor-ordering bug: the base logger interceptor used to run AFTER -// supplied interceptors, replacing the context logger and silently dropping -// any enrichment (trace_id, audit fields) before the handler executed. +// 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. // -// It simulates a supplied interceptor that enriches the context logger (the -// same pattern backend-commons logs.UnaryServerInterceptor uses) and asserts -// the handler's log line actually carries the enriched fields. -func TestSuppliedInterceptorEnrichmentReachesHandler(t *testing.T) { +// 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) @@ -37,14 +51,8 @@ func TestSuppliedInterceptorEnrichmentReachesHandler(t *testing.T) { return handler(l.WithContext(ctx), req) } - // Compose the chain the same way NewServer does. - interceptors := append( - []grpc.UnaryServerInterceptor{ - UnaryServerMetadataPropagator, - grpcLoggerInterceptor(&logger), - }, - enriching, - ) + // 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) { @@ -52,18 +60,8 @@ func TestSuppliedInterceptorEnrichmentReachesHandler(t *testing.T) { return "ok", nil } - // Manually chain (mirrors grpc.ChainUnaryInterceptor semantics). - chained := handler - for i := len(interceptors) - 1; i >= 0; i-- { - ic := interceptors[i] - next := chained - info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"} - chained = func(ctx context.Context, req interface{}) (interface{}, error) { - return ic(ctx, req, info, next) - } - } - - if _, err := chained(context.Background(), "req"); err != 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) } @@ -73,9 +71,33 @@ func TestSuppliedInterceptorEnrichmentReachesHandler(t *testing.T) { } if entry["trace_id"] != "trace-abc-123" { - t.Errorf("handler log trace_id = %v, want trace-abc-123 (enrichment was dropped)", entry["trace_id"]) + 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()) + } +}