diff --git a/docs/guide/grpc/client.md b/docs/guide/grpc/client.md index 24fdbb24f..9504cf15f 100644 --- a/docs/guide/grpc/client.md +++ b/docs/guide/grpc/client.md @@ -106,9 +106,23 @@ builder.Services.AddWolverineGrpcClient(o => }); ``` -On the server side of a Wolverine→Wolverine hop, the envelope headers are read back in the -`WolverineGrpcServicePropagationInterceptor` already shipped with the adapter, so a call chain -spanning multiple Wolverine services keeps a single correlation identity without any user wiring. +On the server side of a Wolverine→Wolverine hop, `WolverineGrpcServicePropagationInterceptor` +(registered automatically by `AddWolverineGrpc()`) reads the `correlation-id` and `tenant-id` +headers back off the inbound call and applies them to the ambient `IMessageContext` before the +service method runs. Only those two identifiers round-trip this way — `parent-id` and +`conversation-id` live on an `Envelope`, and there is no envelope yet at this point in the +pipeline (`IMessageContext.Envelope` is null outside of message handling). Once the service +method calls `Bus.InvokeAsync(...)`, Wolverine copies the now-set `CorrelationId`/`TenantId` onto +the freshly created envelope automatically, and Marten/Polecat tenant-scoped session frames pick +it up the same way they would for any other transport — no further wiring needed. + +Cross-process parent/trace correlation for gRPC hops is handled separately by OpenTelemetry +`Activity` propagation over HTTP/2 (W3C traceparent) — see [How gRPC Handlers Work](./handlers). +The `parent-id`/`conversation-id` headers still go out on the wire for diagnostic/non-Wolverine +consumers, but nothing on the Wolverine side reads them back in. + +Server-side propagation can be disabled with `WolverineGrpcOptions.PropagateEnvelopeHeaders = +false`, the counterpart to the client-side `PropagateEnvelopeHeaders` option above. ## `RpcException` → typed-exception translation @@ -226,6 +240,8 @@ it lands inside both Wolverine interceptors, which is what you almost always wan | `WolverineGrpcClientPropagationInterceptor` | Stamps envelope headers on each call. | | `WolverineGrpcClientExceptionInterceptor` | Translates `RpcException` to typed .NET exceptions per `MapRpcException` + the default table. | | `WolverineGrpcExceptionMapper.MapToException(rpc)` | Public default mapping table; use from custom interceptors if needed. | +| `WolverineGrpcServicePropagationInterceptor` | Server-side counterpart — reads `correlation-id`/`tenant-id` off inbound calls onto `IMessageContext`. Registered by `AddWolverineGrpc()`. | +| `WolverineGrpcOptions.PropagateEnvelopeHeaders` | Server-side off-switch for the interceptor above. Defaults to `true`. | ## See also diff --git a/src/Samples/OrderChainWithGrpc/InventoryServer/ReserveStockHandler.cs b/src/Samples/OrderChainWithGrpc/InventoryServer/ReserveStockHandler.cs index 4d012e375..ba6752ba0 100644 --- a/src/Samples/OrderChainWithGrpc/InventoryServer/ReserveStockHandler.cs +++ b/src/Samples/OrderChainWithGrpc/InventoryServer/ReserveStockHandler.cs @@ -7,11 +7,14 @@ namespace OrderChainWithGrpc.InventoryServer; /// Wolverine handler for . The gRPC service forwards to the bus /// and the bus invokes this handler — the handler itself has no gRPC coupling. /// -/// The envelope identifiers read from below were *not* -/// stamped by code in this process. They were unpacked from inbound gRPC metadata by -/// WolverineGrpcServicePropagationInterceptor, which is wired up automatically -/// by AddWolverineGrpc(). Echoing them back on the reply is what lets the sample -/// visually prove the chain preserved identity end-to-end. +/// and +/// below were *not* stamped by code in this process. They were unpacked from inbound gRPC +/// metadata by WolverineGrpcServicePropagationInterceptor, which is wired up +/// automatically by AddWolverineGrpc(), before Bus.InvokeAsync ran this +/// handler. Echoing them back on the reply is what lets the sample visually prove those two +/// identifiers preserved identity end-to-end. ParentIdFromUpstream is a different +/// mechanism — it reflects OpenTelemetry +/// propagation over the HTTP/2 traceparent header, not this interceptor. /// /// public static class ReserveStockHandler diff --git a/src/Wolverine.Grpc.Tests/ServerPropagation/DisabledPropagationFixture.cs b/src/Wolverine.Grpc.Tests/ServerPropagation/DisabledPropagationFixture.cs new file mode 100644 index 000000000..2a52ad0eb --- /dev/null +++ b/src/Wolverine.Grpc.Tests/ServerPropagation/DisabledPropagationFixture.cs @@ -0,0 +1,61 @@ +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using ProtoBuf.Grpc.Client; +using ProtoBuf.Grpc.Server; +using Xunit; + +namespace Wolverine.Grpc.Tests.ServerPropagation; + +/// +/// A standalone in-process host (not the shared ) +/// with turned off, so tests can +/// prove the interceptor honors the off-switch without affecting the shared fixture's +/// default-on host used by every other propagation test. +/// +public class DisabledPropagationFixture : IAsyncLifetime +{ + private WebApplication? _app; + public GrpcChannel? Channel { get; private set; } + + public async Task InitializeAsync() + { + var builder = WebApplication.CreateBuilder([]); + builder.WebHost.UseTestServer(); + + builder.Host.UseWolverine(opts => + { + opts.ApplicationAssembly = typeof(DisabledPropagationFixture).Assembly; + opts.Discovery.DisableConventionalDiscovery(); + opts.Discovery.IncludeType(typeof(PropagationEchoHandler)); + }); + + builder.Services.AddCodeFirstGrpc(); + builder.Services.AddWolverineGrpc(o => o.PropagateEnvelopeHeaders = false); + + _app = builder.Build(); + _app.UseRouting(); + _app.MapGrpcService(); + + await _app.StartAsync(); + + var handler = _app.GetTestServer().CreateHandler(); + Channel = GrpcChannel.ForAddress("http://localhost", new GrpcChannelOptions + { + HttpHandler = handler + }); + } + + public async Task DisposeAsync() + { + Channel?.Dispose(); + if (_app != null) + { + await _app.StopAsync(); + await _app.DisposeAsync(); + } + } + + public IPropagationEchoService CreateClient() => Channel!.CreateGrpcService(); +} diff --git a/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoContracts.cs b/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoContracts.cs new file mode 100644 index 000000000..5bc8d75e4 --- /dev/null +++ b/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoContracts.cs @@ -0,0 +1,33 @@ +using ProtoBuf; +using ProtoBuf.Grpc; +using System.ServiceModel; + +namespace Wolverine.Grpc.Tests.ServerPropagation; + +/// +/// Code-first gRPC contract whose server implementation forwards to the Wolverine bus. The +/// invoked handler reads the ambient and echoes back whatever +/// put there from inbound headers — +/// used to prove the server-side half of a Wolverine-to-Wolverine propagation hop actually +/// lands in the handler, not just on the wire. +/// +[ServiceContract] +public interface IPropagationEchoService +{ + Task Echo(PropagationEchoRequest request, CallContext context = default); +} + +[ProtoContract] +public class PropagationEchoRequest +{ +} + +[ProtoContract] +public class PropagationEchoReply +{ + [ProtoMember(1)] + public string? CorrelationId { get; set; } + + [ProtoMember(2)] + public string? TenantId { get; set; } +} diff --git a/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoGrpcService.cs b/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoGrpcService.cs new file mode 100644 index 000000000..9a440219f --- /dev/null +++ b/src/Wolverine.Grpc.Tests/ServerPropagation/PropagationEchoGrpcService.cs @@ -0,0 +1,35 @@ +using ProtoBuf.Grpc; +using Wolverine.Grpc; + +namespace Wolverine.Grpc.Tests.ServerPropagation; + +/// +/// Code-first gRPC service that delegates to the Wolverine bus. The "GrpcService" suffix lets +/// MapWolverineGrpcServices() discover and map this type automatically. +/// +public class PropagationEchoGrpcService : WolverineGrpcServiceBase, IPropagationEchoService +{ + public PropagationEchoGrpcService(IMessageBus bus) : base(bus) + { + } + + public Task Echo(PropagationEchoRequest request, CallContext context = default) + => Bus.InvokeAsync(request, context.CancellationToken); +} + +/// +/// Wolverine handler for . Reads whatever +/// already stamped onto the ambient +/// before this handler ran. +/// +public static class PropagationEchoHandler +{ + public static PropagationEchoReply Handle(PropagationEchoRequest request, IMessageContext context) + { + return new PropagationEchoReply + { + CorrelationId = context.CorrelationId, + TenantId = context.TenantId + }; + } +} diff --git a/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_disabled_tests.cs b/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_disabled_tests.cs new file mode 100644 index 000000000..6e921ee7a --- /dev/null +++ b/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_disabled_tests.cs @@ -0,0 +1,38 @@ +using Grpc.Core; +using Shouldly; +using Xunit; + +namespace Wolverine.Grpc.Tests.ServerPropagation; + +/// +/// Proves = false actually turns the +/// server-side interceptor off, even when a caller sends the envelope headers on the wire. +/// +public class server_propagation_disabled_tests : IClassFixture +{ + private readonly DisabledPropagationFixture _fixture; + + public server_propagation_disabled_tests(DisabledPropagationFixture fixture) + { + _fixture = fixture; + } + + [Fact] + public async Task handler_does_not_see_headers_when_propagation_is_disabled() + { + var headers = new Metadata + { + { EnvelopeConstants.CorrelationIdKey, "corr-should-be-ignored" }, + { EnvelopeConstants.TenantIdKey, "tenant-should-be-ignored" } + }; + + var client = _fixture.CreateClient(); + var reply = await client.Echo(new PropagationEchoRequest(), new CallOptions(headers: headers)); + + // Neither field is null/empty by default — MessageBus always auto-assigns a CorrelationId, + // and an untenanted context reads back Wolverine's internal default-tenant sentinel rather + // than null. The meaningful assertion is that neither header's specific value won. + reply.CorrelationId.ShouldNotBe("corr-should-be-ignored"); + reply.TenantId.ShouldNotBe("tenant-should-be-ignored"); + } +} diff --git a/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_interceptor_tests.cs b/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_interceptor_tests.cs new file mode 100644 index 000000000..bd58d8690 --- /dev/null +++ b/src/Wolverine.Grpc.Tests/ServerPropagation/server_propagation_interceptor_tests.cs @@ -0,0 +1,132 @@ +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.DependencyInjection; +using ProtoBuf.Grpc.Client; +using Shouldly; +using Wolverine.Grpc.Client; +using Wolverine.Grpc.Tests.Client; +using Xunit; + +namespace Wolverine.Grpc.Tests.ServerPropagation; + +/// +/// End-to-end tests for . Unlike +/// (which only asserts what lands on the +/// wire), these assert what the invoked Wolverine handler actually sees on +/// after both the client-side and server-side interceptors have +/// run — this is the full Wolverine-to-Wolverine hop the OrderChainWithGrpc sample and the +/// gRPC client docs describe. +/// +[Collection("grpc-client")] +public class server_propagation_interceptor_tests +{ + private readonly WolverineGrpcClientFixture _fixture; + + public server_propagation_interceptor_tests(WolverineGrpcClientFixture fixture) + { + _fixture = fixture; + } + + private ServiceProvider BuildContainer(TestMessageContext? ctx) + { + var services = new ServiceCollection(); + services.AddLogging(); + + if (ctx != null) + { + services.AddScoped(_ => ctx); + services.AddScoped(sp => sp.GetRequiredService()); + } + + services.AddWolverineGrpcClient(o => + { + o.Address = new Uri("http://localhost"); + }).ConfigureChannel(c => c.HttpHandler = _fixture.ServerHandler); + + return services.BuildServiceProvider(); + } + + [Fact] + public async Task handler_sees_correlation_and_tenant_id_propagated_from_the_caller() + { + var ctx = new TestMessageContext + { + CorrelationId = "corr-hop-1", + TenantId = "tenant-hop-1" + }; + + await using var provider = BuildContainer(ctx); + using var scope = provider.CreateScope(); + var client = scope.ServiceProvider.GetRequiredService(); + + var reply = await client.Echo(new PropagationEchoRequest()); + + // These values only round-trip if the client interceptor stamped the headers AND the + // server interceptor read them back onto the ambient IMessageContext BEFORE Bus.InvokeAsync + // ran the handler — proving the full hop, not just the wire format. + reply.CorrelationId.ShouldBe("corr-hop-1"); + reply.TenantId.ShouldBe("tenant-hop-1"); + } + + [Fact] + public async Task handler_does_not_see_a_specific_tenant_id_when_caller_has_no_message_context_in_scope() + { + await using var provider = BuildContainer(ctx: null); + using var scope = provider.CreateScope(); + var client = scope.ServiceProvider.GetRequiredService(); + + var reply = await client.Echo(new PropagationEchoRequest()); + + // Neither field is asserted as null/empty here: MessageBus always assigns a CorrelationId + // from Activity.Current?.RootId (or a fresh GUID), and an untenanted context reads back + // Wolverine's internal default-tenant sentinel rather than null — neither is something this + // interceptor controls or suppresses. What it does control is whether a *specific* caller + // value shows up uninvited, which is what the round-trip tests above already prove positively. + reply.TenantId.ShouldNotBe("tenant-hop-1"); + reply.TenantId.ShouldNotBe("tenant-only"); + } + + [Fact] + public async Task tenant_id_alone_propagates_without_a_correlation_id_header() + { + // TestMessageContext's parameterless constructor auto-generates a CorrelationId — clear + // it explicitly so this test isolates tenant-id propagation on its own (no correlation-id + // header goes out on the wire). + var ctx = new TestMessageContext + { + CorrelationId = null, + TenantId = "tenant-only" + }; + + await using var provider = BuildContainer(ctx); + using var scope = provider.CreateScope(); + var client = scope.ServiceProvider.GetRequiredService(); + + var reply = await client.Echo(new PropagationEchoRequest()); + + reply.TenantId.ShouldBe("tenant-only"); + } + + [Fact] + public async Task an_arbitrary_non_wolverine_caller_can_also_drive_tenant_id_via_the_raw_header() + { + // No AddWolverineGrpcClient, no IMessageContext anywhere in this process — an ordinary + // grpc-dotnet caller setting the header by hand, same as an external service or gateway + // would. This is the scenario from the original Discord question: does something read + // an inbound tenant-id header at all? It does, independent of who sent it. + using var channel = GrpcChannel.ForAddress("http://localhost", new GrpcChannelOptions + { + HttpHandler = _fixture.ServerHandler + }); + var client = channel.CreateGrpcService(); + + var headers = new Metadata + { + { EnvelopeConstants.TenantIdKey, "external-caller-tenant" } + }; + + var reply = await client.Echo(new PropagationEchoRequest(), new CallOptions(headers: headers)); + + reply.TenantId.ShouldBe("external-caller-tenant"); + } +} diff --git a/src/Wolverine.Grpc.Tests/add_wolverine_grpc_idempotency_tests.cs b/src/Wolverine.Grpc.Tests/add_wolverine_grpc_idempotency_tests.cs index 0c0347536..40b5120c6 100644 --- a/src/Wolverine.Grpc.Tests/add_wolverine_grpc_idempotency_tests.cs +++ b/src/Wolverine.Grpc.Tests/add_wolverine_grpc_idempotency_tests.cs @@ -46,6 +46,63 @@ public void repeat_calls_do_not_stack_the_exception_interceptor() .ShouldBe(1); } + [Fact] + public void single_call_registers_propagation_interceptor_exactly_once() + { + var services = new ServiceCollection(); + services.AddOptions(); + + services.AddWolverineGrpc(); + + var grpcOptions = services.BuildServiceProvider() + .GetRequiredService>().Value; + + grpcOptions.Interceptors.Count(r => r.Type == typeof(WolverineGrpcServicePropagationInterceptor)) + .ShouldBe(1); + } + + [Fact] + public void repeat_calls_do_not_stack_the_propagation_interceptor() + { + var services = new ServiceCollection(); + services.AddOptions(); + + services.AddWolverineGrpc(); + services.AddWolverineGrpc(); + services.AddWolverineGrpc(); + + var grpcOptions = services.BuildServiceProvider() + .GetRequiredService>().Value; + + grpcOptions.Interceptors.Count(r => r.Type == typeof(WolverineGrpcServicePropagationInterceptor)) + .ShouldBe(1); + } + + [Fact] + public void exception_interceptor_runs_outermost_of_the_propagation_interceptor() + { + // Registration order determines wrapping order for grpc-dotnet server interceptors — the + // first-added interceptor is outermost. Exception translation must stay outermost so it + // can also translate anything the propagation interceptor itself throws, symmetric to the + // client-side interceptor order (docs/guide/grpc/client.md "Ordering and composition"). + var services = new ServiceCollection(); + services.AddOptions(); + + services.AddWolverineGrpc(); + + var grpcOptions = services.BuildServiceProvider() + .GetRequiredService>().Value; + + var exceptionIndex = grpcOptions.Interceptors + .ToList() + .FindIndex(r => r.Type == typeof(WolverineGrpcExceptionInterceptor)); + var propagationIndex = grpcOptions.Interceptors + .ToList() + .FindIndex(r => r.Type == typeof(WolverineGrpcServicePropagationInterceptor)); + + exceptionIndex.ShouldBeLessThan(propagationIndex); + } + [Fact] public void repeat_calls_do_not_stack_the_grpc_graph_registration() { diff --git a/src/Wolverine.Grpc/WolverineGrpcExtensions.cs b/src/Wolverine.Grpc/WolverineGrpcExtensions.cs index f18b6699f..143ac9c76 100644 --- a/src/Wolverine.Grpc/WolverineGrpcExtensions.cs +++ b/src/Wolverine.Grpc/WolverineGrpcExtensions.cs @@ -84,9 +84,14 @@ public static IServiceCollection AddWolverineGrpc( new GrpcEndpointDescriptorSource(sp.GetRequiredService())); services.AddSingleton(); + services.AddSingleton(); services.Configure(opts => { + // Exception translation stays outermost so it can also translate anything thrown by + // the propagation interceptor itself, symmetric to the client-side interceptor order + // (see docs/guide/grpc/client.md "Ordering and composition"). opts.Interceptors.Add(); + opts.Interceptors.Add(); }); configure?.Invoke(options); diff --git a/src/Wolverine.Grpc/WolverineGrpcOptions.cs b/src/Wolverine.Grpc/WolverineGrpcOptions.cs index dfc607fd2..39c24c513 100644 --- a/src/Wolverine.Grpc/WolverineGrpcOptions.cs +++ b/src/Wolverine.Grpc/WolverineGrpcOptions.cs @@ -16,6 +16,14 @@ public sealed class WolverineGrpcOptions { internal MiddlewarePolicy Middleware { get; } = new(); + /// + /// Whether reads the + /// correlation-id/tenant-id envelope headers off inbound calls onto the + /// ambient before the service method runs. Defaults to + /// true. The client-side counterpart is WolverineGrpcClientOptions.PropagateEnvelopeHeaders. + /// + public bool PropagateEnvelopeHeaders { get; set; } = true; + /// /// Structural policies applied to all discovered gRPC chains during bootstrapping. /// Analogous to WolverineHttpOptions.Policies — use when you need typed access diff --git a/src/Wolverine.Grpc/WolverineGrpcServicePropagationInterceptor.cs b/src/Wolverine.Grpc/WolverineGrpcServicePropagationInterceptor.cs new file mode 100644 index 000000000..bab481e71 --- /dev/null +++ b/src/Wolverine.Grpc/WolverineGrpcServicePropagationInterceptor.cs @@ -0,0 +1,134 @@ +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.DependencyInjection; + +namespace Wolverine.Grpc; + +/// +/// Server-side gRPC interceptor that reads the correlation-id and tenant-id +/// envelope headers off an inbound call and applies them to the ambient, request-scoped +/// before the service method runs. The client-side counterpart is +/// — this is what actually reads +/// those headers back on the receiving end of a Wolverine-to-Wolverine hop. +/// +/// +/// +/// Only correlation-id and tenant-id are round-tripped this way. +/// and are the +/// only envelope identifiers that can be set on the ambient context before a message +/// has been invoked — parent-id and conversation-id only exist on an +/// , and there is no envelope yet at this point ( is null outside of message handling). Once the service +/// method calls Bus.InvokeAsync(...), Wolverine copies the now-set +/// CorrelationId/TenantId onto the freshly created envelope automatically — no +/// further wiring needed, and Marten/Polecat tenant-scoped session frames pick it up the same +/// way they would for any other transport. +/// +/// +/// Cross-process parent/trace correlation for gRPC hops is handled separately by +/// OpenTelemetry Activity propagation over HTTP/2 (W3C traceparent) — see +/// How gRPC Handlers Work. +/// +/// +/// Both fields are set unconditionally when the corresponding header is present — the same +/// convention every other Wolverine transport uses for a freshly received message (see +/// MessageContext.ReadEnvelope: CorrelationId = originalEnvelope.CorrelationId, +/// no "only if empty" check). This matters for correlation-id specifically: a brand +/// new / already has a non-empty, +/// auto-generated (seeded from +/// 's root id, or a fresh GUID) before this +/// interceptor ever runs — gating on "currently empty" would silently never apply an inbound +/// header. +/// +/// +/// Registered automatically by , +/// and runs inside so an exception it throws +/// is still translated, symmetric to how propagation sits inside exception translation on the +/// client. +/// +/// +public sealed class WolverineGrpcServicePropagationInterceptor : Interceptor +{ + private readonly WolverineGrpcOptions _options; + + public WolverineGrpcServicePropagationInterceptor(WolverineGrpcOptions options) + { + _options = options; + } + + public override Task UnaryServerHandler( + TRequest request, + ServerCallContext context, + UnaryServerMethod continuation) + { + Propagate(context); + return continuation(request, context); + } + + public override Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + Propagate(context); + return continuation(requestStream, context); + } + + public override Task ServerStreamingServerHandler( + TRequest request, + IServerStreamWriter responseStream, + ServerCallContext context, + ServerStreamingServerMethod continuation) + { + Propagate(context); + return continuation(request, responseStream, context); + } + + public override Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + Propagate(context); + return continuation(requestStream, responseStream, context); + } + + private void Propagate(ServerCallContext context) + { + if (!_options.PropagateEnvelopeHeaders) return; + + var services = context.GetHttpContext().RequestServices; + + // IMessageContext is a scoped service — bail cleanly when there is no active Wolverine scope. + var ctx = services.GetService(typeof(IMessageContext)) as IMessageContext; + if (ctx == null) return; + + var correlationId = Find(context.RequestHeaders, EnvelopeConstants.CorrelationIdKey); + if (!string.IsNullOrEmpty(correlationId)) + { + ctx.CorrelationId = correlationId; + } + + var tenantId = Find(context.RequestHeaders, EnvelopeConstants.TenantIdKey); + if (!string.IsNullOrEmpty(tenantId)) + { + ctx.TenantId = tenantId; + } + } + + private static string? Find(Metadata headers, string key) + { + foreach (var entry in headers) + { + if (string.Equals(entry.Key, key, StringComparison.OrdinalIgnoreCase)) + { + return entry.Value; + } + } + + return null; + } +}