From 0a9d5acf66d479264719b39a4bc3ce712603ae78 Mon Sep 17 00:00:00 2001 From: wangbill Date: Sun, 16 Aug 2026 15:20:15 -0400 Subject: [PATCH 1/4] Preserve gRPC channel recreation when externalized payloads are enabled Enabling UseExternalizedPayloads (AzureBlobPayloads) silently disabled gRPC channel recreation on both the worker and the client. IConfigureOptions runs before IPostConfigureOptions. DTS's ConfigureGrpcChannel sets both options.Channel and SetChannelRecreator(...). The AzureBlobPayloads PostConfigure then moved the channel onto an intercepted CallInvoker and nulled options.Channel (it had to, because "Channel supersedes CallInvoker" per GrpcDurableTaskClientOptions). That killed every recreation path: - Worker path 1 guard requires a non-null channel; latestObservedChannel is seeded from grpcOptions.Channel == null. - Worker path 2 requires Channel and CallInvoker both null; CallInvoker was set. - Client's CallInvoker branch explicitly cannot recreate an external channel. Recreation is on by default (ChannelRecreateFailureThreshold = 5), so on backend scale/upgrade/node replacement the worker wedged on a half-open HTTP/2 connection and never recovered until the process was restarted. Fix: add an internal CallInvokerDecorator hook, mirroring the existing SetChannelRecreator idiom. The extension registers a decorator instead of mutating Channel/CallInvoker, and core applies it at every point a CallInvoker is produced - including after a channel recreate. The decorator is applied outside ChannelRecreatingCallInvoker so internal channel swaps stay transparent to the interceptor. Purely additive: with no decorator set, behavior is unchanged. Also fixes a second defect: UseGrpc("http://localhost:4001") (Address-only) previously threw ArgumentException and was unusable with externalized payloads. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Client/Grpc/GrpcDurableTaskClient.cs | 11 + .../Grpc/GrpcDurableTaskClientOptions.cs | 8 + .../Internal/InternalOptionsExtensions.cs | 44 +++ ...ientBuilderExtensions.AzureBlobPayloads.cs | 23 +- ...rkerBuilderExtensions.AzureBlobPayloads.cs | 23 +- src/Worker/Grpc/GrpcDurableTaskWorker.cs | 22 +- .../Grpc/GrpcDurableTaskWorkerOptions.cs | 8 + .../Internal/InternalOptionsExtensions.cs | 45 +++ ...ableTaskClientCallInvokerDecoratorTests.cs | 149 ++++++++++ .../AzureBlobPayloads.Tests.csproj | 4 + ...alizedPayloadsCallInvokerDecoratorTests.cs | 268 ++++++++++++++++++ ...ableTaskWorkerCallInvokerDecoratorTests.cs | 176 ++++++++++++ ...pcDurableTaskWorkerOptionsInternalTests.cs | 52 ++++ 13 files changed, 797 insertions(+), 36 deletions(-) create mode 100644 test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs create mode 100644 test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs create mode 100644 test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 23350d4c..304c74c1 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -8,6 +8,7 @@ using DurableTask.Core.History; using Google.Protobuf.WellKnownTypes; using Microsoft.DurableTask.Client.Entities; +using Microsoft.DurableTask.Client.Grpc.Internal; using Microsoft.DurableTask.Tracing; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -625,6 +626,16 @@ public override async Task> GetOrchestrationHistoryAsync( } static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) + { + AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker undecorated); + + // Decorate outside any ChannelRecreatingCallInvoker so the wrapper's internal channel swaps + // stay transparent to the decorator (and to any interceptor it installs). + callInvoker = options.ApplyCallInvokerDecorator(undecorated); + return disposable; + } + + static AsyncDisposable GetCallInvokerCore(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { Func>? recreator = options.Internal.ChannelRecreator; int threshold = options.Internal.ChannelRecreateFailureThreshold; diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index 126aad34..a563f3f2 100644 --- a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs +++ b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs @@ -57,5 +57,13 @@ internal class InternalOptions /// old channel so in-flight RPCs from peer clients are not interrupted. /// public Func>? ChannelRecreator { get; set; } + + /// + /// Gets or sets an optional decorator applied to every the client builds + /// from its configured transport. Extensions use this to attach interceptors without taking + /// ownership of , which would otherwise disable + /// recreation. + /// + public Func? CallInvokerDecorator { get; set; } } } diff --git a/src/Client/Grpc/Internal/InternalOptionsExtensions.cs b/src/Client/Grpc/Internal/InternalOptionsExtensions.cs index 800848f3..dec02bb6 100644 --- a/src/Client/Grpc/Internal/InternalOptionsExtensions.cs +++ b/src/Client/Grpc/Internal/InternalOptionsExtensions.cs @@ -30,4 +30,48 @@ public static void SetChannelRecreator( { options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator)); } + + /// + /// Sets a callback that decorates every the client builds from its configured + /// transport. Use this instead of replacing with an + /// intercepted : clearing the channel leaves the + /// client with no way to recreate a wedged connection. + /// + /// The gRPC client options. + /// The decorator callback. + /// + /// This is an internal API that supports the DurableTask infrastructure and not subject to + /// the same compatibility standards as public APIs. It may be changed or removed without notice in + /// any release. You should only use it directly in your code with extreme caution and knowing that + /// doing so can result in application failures when updating to a new DurableTask release. + /// + public static void SetCallInvokerDecorator( + this GrpcDurableTaskClientOptions options, + Func decorator) + { + options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator)); + } + + /// + /// Applies the decorator registered by to , + /// returning it unchanged when no decorator is registered. Callers that build a + /// from these options must route it through this method so registered + /// interceptors are not silently dropped. + /// + /// The gRPC client options. + /// The invoker to decorate. + /// The decorated invoker, or when no decorator is registered. + /// + /// This is an internal API that supports the DurableTask infrastructure and not subject to + /// the same compatibility standards as public APIs. It may be changed or removed without notice in + /// any release. You should only use it directly in your code with extreme caution and knowing that + /// doing so can result in application failures when updating to a new DurableTask release. + /// + public static CallInvoker ApplyCallInvokerDecorator( + this GrpcDurableTaskClientOptions options, + CallInvoker invoker) + { + Func? decorator = options.Internal.CallInvokerDecorator; + return decorator is null ? invoker : decorator(invoker); + } } diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 0817bcea..3b481791 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -4,8 +4,8 @@ using Grpc.Core.Interceptors; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Client.Grpc; +using Microsoft.DurableTask.Client.Grpc.Internal; using Microsoft.DurableTask.Converters; -using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; @@ -37,23 +37,12 @@ static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientB .PostConfigure>((opt, store, monitor) => { LargePayloadStorageOptions opts = monitor.Get(builder.Name); - if (opt.Channel is not null) - { - Grpc.Core.CallInvoker invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); - opt.CallInvoker = invoker; - // Ensure client uses the intercepted invoker path - opt.Channel = null; - } - else if (opt.CallInvoker is not null) - { - opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); - } - else - { - throw new ArgumentException( - "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature"); - } + // Register a decorator rather than moving Channel onto an intercepted CallInvoker. + // Clearing Channel would disable the client's gRPC channel recreation, and requiring a + // pre-built Channel/CallInvoker would rule out the Address-only configuration. + opt.SetCallInvokerDecorator( + invoker => invoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts))); }); return builder; diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index b690d288..ea76c02f 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -2,10 +2,10 @@ // Licensed under the MIT License. using Grpc.Core.Interceptors; -using Grpc.Net.Client; using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -61,23 +61,12 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB .PostConfigure>((opt, store, monitor) => { LargePayloadStorageOptions opts = monitor.Get(builder.Name); - if (opt.Channel is not null) - { - var invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); - opt.CallInvoker = invoker; - // Ensure worker uses the intercepted invoker path - opt.Channel = null; - } - else if (opt.CallInvoker is not null) - { - opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); - } - else - { - throw new ArgumentException( - "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature"); - } + // Register a decorator rather than moving Channel onto an intercepted CallInvoker. + // Clearing Channel would disable the worker's gRPC channel recreation, and requiring a + // pre-built Channel/CallInvoker would rule out the Address-only configuration. + opt.SetCallInvokerDecorator( + invoker => invoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts))); opt.Capabilities.Add(P.WorkerCapability.LargePayloads); }); diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index a200cedb..53b2964b 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using System.Diagnostics; +using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.DurableTask.Worker.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -126,7 +127,12 @@ async Task TryRecreateChannelAsync( // The recreator owns the replacement channel lifetime. Return a default disposable // so the caller disposes the previous worker-owned channel exactly once without // carrying that ownership forward to the recreated state. - return new ChannelRecreateResult(true, newChannel.CreateCallInvoker(), newChannel.Target, default, newChannel); + return new ChannelRecreateResult( + true, + this.grpcOptions.ApplyCallInvokerDecorator(newChannel.CreateCallInvoker()), + newChannel.Target, + default, + newChannel); } // Recreator returned the same instance — nothing to swap. @@ -154,7 +160,12 @@ async Task TryRecreateChannelAsync( // This new channel is worker-owned, so hand back a disposable that will shut it down // (and dispose it on frameworks where GrpcChannel implements IDisposable). AsyncDisposable newDisposable = CreateOwnedChannelDisposable(newChannel); - return new ChannelRecreateResult(true, newChannel.CreateCallInvoker(), newChannel.Target, newDisposable, newChannel); + return new ChannelRecreateResult( + true, + this.grpcOptions.ApplyCallInvokerDecorator(newChannel.CreateCallInvoker()), + newChannel.Target, + newDisposable, + newChannel); } catch (OperationCanceledException) when (cancellation.IsCancellationRequested) { @@ -293,6 +304,13 @@ and not AccessViolationException } AsyncDisposable GetCallInvoker(out CallInvoker callInvoker, out string address) + { + AsyncDisposable disposable = this.GetCallInvokerCore(out CallInvoker undecorated, out address); + callInvoker = this.grpcOptions.ApplyCallInvokerDecorator(undecorated); + return disposable; + } + + AsyncDisposable GetCallInvokerCore(out CallInvoker callInvoker, out string address) { if (this.grpcOptions.Channel is GrpcChannel c) { diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index 59c21a00..fecb674c 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -167,6 +167,14 @@ internal class InternalOptions /// public Func>? ChannelRecreator { get; set; } + /// + /// Gets or sets an optional decorator applied to every the worker builds + /// from its configured transport, including invokers rebuilt after a channel recreate. Extensions + /// use this to attach interceptors without taking ownership of + /// , which would otherwise disable recreation. + /// + public Func? CallInvokerDecorator { get; set; } + /// /// Gets or sets a callback that is invoked when activity work items are received or finished. /// diff --git a/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs b/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs index 81ad09d5..ce922826 100644 --- a/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs +++ b/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs @@ -83,6 +83,51 @@ public static void SetChannelRecreator( options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator)); } + /// + /// Sets a callback that decorates every the worker builds from its configured + /// transport, including invokers rebuilt after a channel recreate. Use this instead of replacing + /// with an intercepted + /// : clearing the channel leaves the worker with + /// no way to recreate a wedged connection. + /// + /// The gRPC worker options. + /// The decorator callback. + /// + /// This is an internal API that supports the DurableTask infrastructure and not subject to + /// the same compatibility standards as public APIs. It may be changed or removed without notice in + /// any release. You should only use it directly in your code with extreme caution and knowing that + /// doing so can result in application failures when updating to a new DurableTask release. + /// + public static void SetCallInvokerDecorator( + this GrpcDurableTaskWorkerOptions options, + Func decorator) + { + options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator)); + } + + /// + /// Applies the decorator registered by to , + /// returning it unchanged when no decorator is registered. Callers that build a + /// from these options must route it through this method so registered + /// interceptors are not silently dropped. + /// + /// The gRPC worker options. + /// The invoker to decorate. + /// The decorated invoker, or when no decorator is registered. + /// + /// This is an internal API that supports the DurableTask infrastructure and not subject to + /// the same compatibility standards as public APIs. It may be changed or removed without notice in + /// any release. You should only use it directly in your code with extreme caution and knowing that + /// doing so can result in application failures when updating to a new DurableTask release. + /// + public static CallInvoker ApplyCallInvokerDecorator( + this GrpcDurableTaskWorkerOptions options, + CallInvoker invoker) + { + Func? decorator = options.Internal.CallInvokerDecorator; + return decorator is null ? invoker : decorator(invoker); + } + /// /// Sets the deadline applied to the initial Hello RPC during worker connect. A wedged /// handshake on a half-open HTTP/2 connection no longer hangs the reconnect loop indefinitely. diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs new file mode 100644 index 00000000..31fb5bc3 --- /dev/null +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs @@ -0,0 +1,149 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using Grpc.Core; +using Microsoft.DurableTask.Client.Grpc.Internal; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Client.Grpc.Tests; + +/// +/// Verifies that a registered CallInvokerDecorator is applied on every transport path the client +/// supports, and that it wraps outside so the +/// wrapper's internal channel swaps stay transparent to the decorator. +/// +public class GrpcDurableTaskClientCallInvokerDecoratorTests +{ + static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskClient) + .GetMethod("GetCallInvoker", BindingFlags.Static | BindingFlags.NonPublic)!; + + [Fact] + public async Task GetCallInvoker_ChannelPath_AppliesDecorator() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5201"); + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.SetCallInvokerDecorator(_ => sentinel); + + try + { + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + callInvoker.Should().BeSameAs(sentinel); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public async Task GetCallInvoker_ExternalCallInvokerPath_AppliesDecorator() + { + // Arrange + CallInvoker external = CreateSentinel(); + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskClientOptions options = new() { CallInvoker = external }; + CallInvoker? observed = null; + options.SetCallInvokerDecorator(invoker => + { + observed = invoker; + return sentinel; + }); + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + callInvoker.Should().BeSameAs(sentinel); + observed.Should().BeSameAs(external); + await disposable.DisposeAsync(); + } + + [Fact] + public async Task GetCallInvoker_AddressPath_AppliesDecorator() + { + // Arrange + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskClientOptions options = new() { Address = "http://localhost:5202" }; + options.SetCallInvokerDecorator(_ => sentinel); + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + callInvoker.Should().BeSameAs(sentinel); + await disposable.DisposeAsync(); + } + + [Fact] + public async Task GetCallInvoker_WithRecreator_AppliesDecoratorOutsideRecreatingInvoker() + { + // Arrange: recreation stays enabled, so the core invoker is a ChannelRecreatingCallInvoker. + // The decorator must receive that wrapper (i.e. wrap outside it), otherwise the wrapper's + // internal channel swaps would replace the decorated invoker and drop the interceptor. + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5203"); + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.SetChannelRecreator((existing, ct) => Task.FromResult(existing)); + CallInvoker? observed = null; + options.SetCallInvokerDecorator(invoker => + { + observed = invoker; + return sentinel; + }); + + try + { + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + observed.Should().BeOfType(); + callInvoker.Should().BeSameAs(sentinel); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public async Task GetCallInvoker_WithoutDecorator_ReturnsUndecoratedInvoker() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5204"); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + + try + { + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + // Assert: with no decorator registered the invoker is exactly what core builds today. + callInvoker.Should().BeOfType(channel.CreateCallInvoker().GetType()); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + static CallInvoker CreateSentinel() => GrpcChannel.ForAddress("http://sentinel.invalid").CreateCallInvoker(); + + static (AsyncDisposable Disposable, CallInvoker CallInvoker) InvokeGetCallInvoker( + GrpcDurableTaskClientOptions options) + { + object?[] args = { options, NullLogger.Instance, null }; + AsyncDisposable disposable = (AsyncDisposable)GetCallInvokerMethod.Invoke(null, args)!; + return (disposable, (CallInvoker)args[2]!); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj index 39298f69..40851465 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj +++ b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -7,6 +7,10 @@ $(AssemblyName) + + + + diff --git a/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs new file mode 100644 index 00000000..a514533b --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs @@ -0,0 +1,268 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Google.Protobuf; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Grpc; +using Microsoft.DurableTask.Client.Grpc.Internal; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +/// +/// Verifies that enabling externalized payloads composes with the gRPC transport options instead of +/// replacing them. Previously the extension moved Channel onto an intercepted CallInvoker +/// and nulled Channel, which silently disabled channel recreation on both the worker and the +/// client, and made the Address-only setup unusable. +/// +public class ExternalizedPayloadsCallInvokerDecoratorTests +{ + static readonly Marshaller RequestMarshaller = Marshallers.Create( + r => r.ToByteArray(), P.CreateInstanceRequest.Parser.ParseFrom); + static readonly Marshaller ResponseMarshaller = Marshallers.Create( + r => r.ToByteArray(), P.CreateInstanceResponse.Parser.ParseFrom); + static readonly Method CreateInstanceMethod = new( + MethodType.Unary, + "TaskHubSidecarService", + "StartInstance", + RequestMarshaller, + ResponseMarshaller); + + [Fact] + public void Worker_WithChannel_PreservesChannelSoRecreationStaysEnabled() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc(channel); + + // Act + builder.UseExternalizedPayloads(); + GrpcDurableTaskWorkerOptions options = GetOptions(services); + + // Assert + options.Channel.Should().BeSameAs(channel); + } + + [Fact] + public void Client_WithChannel_PreservesChannelSoRecreationStaysEnabled() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskClientBuilder builder = new(null, services); + builder.UseGrpc(channel); + + // Act + builder.UseExternalizedPayloads(); + GrpcDurableTaskClientOptions options = GetOptions(services); + + // Assert + options.Channel.Should().BeSameAs(channel); + } + + [Fact] + public void Worker_WithAddressOnly_DoesNotThrow() + { + // Arrange + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc("http://localhost:4001"); + builder.UseExternalizedPayloads(); + + // Act + Func act = () => GetOptions(services); + + // Assert + act.Should().NotThrow().Which.Address.Should().Be("http://localhost:4001"); + } + + [Fact] + public void Client_WithAddressOnly_DoesNotThrow() + { + // Arrange + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskClientBuilder builder = new(null, services); + builder.UseGrpc("http://localhost:4001"); + builder.UseExternalizedPayloads(); + + // Act + Func act = () => GetOptions(services); + + // Assert + act.Should().NotThrow().Which.Address.Should().Be("http://localhost:4001"); + } + + [Fact] + public void Worker_WithExternalCallInvoker_PreservesConfiguredInvoker() + { + // Arrange + CallInvoker external = GrpcChannel.ForAddress("http://localhost:4001").CreateCallInvoker(); + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc(opt => opt.CallInvoker = external); + + // Act + builder.UseExternalizedPayloads(); + GrpcDurableTaskWorkerOptions options = GetOptions(services); + + // Assert: the extension no longer mutates the configured invoker; it decorates on use instead. + options.CallInvoker.Should().BeSameAs(external); + options.ApplyCallInvokerDecorator(external).Should().NotBeSameAs(external); + } + + [Fact] + public void Worker_StillAnnouncesLargePayloadsCapability() + { + // Arrange + ServiceCollection services = new(); + services.AddSingleton(new FakePayloadStore()); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc("http://localhost:4001"); + + // Act + builder.UseExternalizedPayloads(); + GrpcDurableTaskWorkerOptions options = GetOptions(services); + + // Assert + options.Capabilities.Should().Contain(P.WorkerCapability.LargePayloads); + } + + [Fact] + public async Task Worker_RegisteredDecorator_ExternalizesLargePayloads() + { + // Arrange + ServiceCollection services = new(); + RecordingPayloadStore store = new(); + services.AddSingleton(store); + services.Configure(o => o.ThresholdBytes = 1); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc("http://localhost:4001"); + builder.UseExternalizedPayloads(); + GrpcDurableTaskWorkerOptions options = GetOptions(services); + + RecordingCallInvoker inner = new(); + CallInvoker decorated = options.ApplyCallInvokerDecorator(inner); + + // Act + await InvokeCreateInstanceAsync(decorated, new string('x', 1024)); + + // Assert + store.UploadCount.Should().Be(1); + inner.LastRequest!.Input.Should().Be(RecordingPayloadStore.Token); + } + + [Fact] + public async Task Client_RegisteredDecorator_ExternalizesLargePayloads() + { + // Arrange + ServiceCollection services = new(); + RecordingPayloadStore store = new(); + services.AddSingleton(store); + services.Configure(o => o.ThresholdBytes = 1); + DefaultDurableTaskClientBuilder builder = new(null, services); + builder.UseGrpc("http://localhost:4001"); + builder.UseExternalizedPayloads(); + GrpcDurableTaskClientOptions options = GetOptions(services); + + RecordingCallInvoker inner = new(); + CallInvoker decorated = options.ApplyCallInvokerDecorator(inner); + + // Act + await InvokeCreateInstanceAsync(decorated, new string('x', 1024)); + + // Assert + store.UploadCount.Should().Be(1); + inner.LastRequest!.Input.Should().Be(RecordingPayloadStore.Token); + } + + static Task InvokeCreateInstanceAsync(CallInvoker invoker, string input) + { + P.CreateInstanceRequest request = new() { InstanceId = "instance", Name = "orchestration", Input = input }; + return invoker.AsyncUnaryCall(CreateInstanceMethod, null, default, request).ResponseAsync; + } + + static TOptions GetOptions(IServiceCollection services) + where TOptions : class + { + ServiceProvider provider = services.BuildServiceProvider(); + return provider.GetRequiredService>().Get(null); + } + + sealed class FakePayloadStore : PayloadStore + { + public override Task DownloadAsync(string token, CancellationToken cancellationToken) + => Task.FromResult(token); + + public override bool IsKnownPayloadToken(string value) => false; + + public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) + => Task.FromResult(payLoad); + } + + sealed class RecordingPayloadStore : PayloadStore + { + public const string Token = "payload-token"; + + int uploadCount; + + public int UploadCount => Volatile.Read(ref this.uploadCount); + + public override Task DownloadAsync(string token, CancellationToken cancellationToken) + => Task.FromResult(token); + + public override bool IsKnownPayloadToken(string value) => value == Token; + + public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) + { + Interlocked.Increment(ref this.uploadCount); + return Task.FromResult(Token); + } + } + + sealed class RecordingCallInvoker : CallInvoker + { + public P.CreateInstanceRequest? LastRequest { get; private set; } + + public override AsyncUnaryCall AsyncUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + { + this.LastRequest = request as P.CreateInstanceRequest; + return new AsyncUnaryCall( + Task.FromResult(Activator.CreateInstance()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + } + + public override TResponse BlockingUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + + public override AsyncClientStreamingCall AsyncClientStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncDuplexStreamingCall AsyncDuplexStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncServerStreamingCall AsyncServerStreamingCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + } +} diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs new file mode 100644 index 00000000..092fdd94 --- /dev/null +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs @@ -0,0 +1,176 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using Grpc.Core; +using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +/// +/// Verifies that a registered CallInvokerDecorator is applied everywhere the worker produces a +/// — including invokers rebuilt after a channel recreate — so extensions can +/// install interceptors without clearing Channel and disabling channel recreation. +/// +public class GrpcDurableTaskWorkerCallInvokerDecoratorTests +{ + static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskWorker) + .GetMethod("GetCallInvoker", BindingFlags.Instance | BindingFlags.NonPublic)!; + static readonly MethodInfo TryRecreateChannelAsyncMethod = typeof(GrpcDurableTaskWorker) + .GetMethod("TryRecreateChannelAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + + [Fact] + public void GetCallInvoker_WithDecorator_ReturnsDecoratedInvoker() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5101"); + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; + CallInvoker? observed = null; + grpcOptions.SetCallInvokerDecorator(invoker => + { + observed = invoker; + return sentinel; + }); + + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); + + // Assert + callInvoker.Should().BeSameAs(sentinel); + observed.Should().NotBeNull().And.NotBeSameAs(sentinel); + address.Should().Be(channel.Target); + } + finally + { + DisposeChannel(channel); + } + } + + [Fact] + public void GetCallInvoker_WithoutDecorator_ReturnsUndecoratedInvoker() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5102"); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); + + // Assert + // Assert: with no decorator registered the invoker is exactly what the channel produces. + callInvoker.Should().BeOfType(channel.CreateCallInvoker().GetType()); + address.Should().Be(channel.Target); + } + finally + { + DisposeChannel(channel); + } + } + + [Fact] + public async Task TryRecreateChannelAsync_ChannelWithRecreatorAndDecorator_RecreatesAndDecorates() + { + // Arrange: this is the shape the AzureBlobPayloads extension used to break — a DTS-configured + // Channel plus recreator. Path 1 requires Channel to still be set, and the invoker the worker + // builds from the replacement channel must still carry the decorator. + GrpcChannel currentChannel = GrpcChannel.ForAddress("http://localhost:5103"); + GrpcChannel recreatedChannel = GrpcChannel.ForAddress("http://localhost:5104"); + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = currentChannel }; + grpcOptions.SetChannelRecreator((channel, ct) => Task.FromResult(recreatedChannel)); + grpcOptions.SetCallInvokerDecorator(_ => sentinel); + + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + // Assert + GetResultProperty(result, "Recreated").Should().BeTrue(); + GetResultProperty(result, "NewChannel").Should().BeSameAs(recreatedChannel); + GetResultProperty(result, "NewCallInvoker").Should().BeSameAs(sentinel); + } + finally + { + DisposeChannel(currentChannel); + DisposeChannel(recreatedChannel); + } + } + + [Fact] + public async Task TryRecreateChannelAsync_WorkerOwnedChannelWithDecorator_DecoratesRebuiltInvoker() + { + // Arrange: Address-only configuration takes the worker-owned rebuild path. + CallInvoker sentinel = CreateSentinel(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Address = "http://localhost:5105" }; + grpcOptions.SetCallInvokerDecorator(_ => sentinel); + + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + GrpcChannel currentChannel = GrpcChannel.ForAddress(grpcOptions.Address); + + try + { + // Act + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + // Assert + GetResultProperty(result, "Recreated").Should().BeTrue(); + GetResultProperty(result, "NewCallInvoker").Should().BeSameAs(sentinel); + + AsyncDisposable newDisposable = GetResultProperty(result, "NewWorkerOwnedDisposable"); + await newDisposable.DisposeAsync(); + } + finally + { + DisposeChannel(currentChannel); + } + } + + static CallInvoker CreateSentinel() => GrpcChannel.ForAddress("http://sentinel.invalid").CreateCallInvoker(); + + static void InvokeGetCallInvoker(GrpcDurableTaskWorker worker, out CallInvoker callInvoker, out string address) + { + object?[] args = { null, null }; + GetCallInvokerMethod.Invoke(worker, args); + callInvoker = (CallInvoker)args[0]!; + address = (string)args[1]!; + } + + static async Task InvokeTryRecreateChannelAsync(GrpcDurableTaskWorker worker, GrpcChannel currentChannel) + { + object?[] args = { CancellationToken.None, default(AsyncDisposable), currentChannel }; + Task task = (Task)TryRecreateChannelAsyncMethod.Invoke(worker, args)!; + await task; + return task.GetType().GetProperty("Result")!.GetValue(task)!; + } + + static T GetResultProperty(object result, string propertyName) + => (T)result.GetType().GetProperty(propertyName)!.GetValue(result)!; + + static void DisposeChannel(GrpcChannel channel) => channel.Dispose(); + + static GrpcDurableTaskWorker CreateWorker(GrpcDurableTaskWorkerOptions grpcOptions) + { + return new GrpcDurableTaskWorker( + name: "Test", + factory: Mock.Of(), + grpcOptions: new OptionsMonitorStub(grpcOptions), + workerOptions: new OptionsMonitorStub(new DurableTaskWorkerOptions()), + services: Mock.Of(), + loggerFactory: NullLoggerFactory.Instance, + orchestrationFilter: null, + exceptionPropertiesProvider: null, + workItemFiltersMonitor: null); + } +} diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs index 87e70484..51bc66df 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs @@ -1,6 +1,8 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using Grpc.Core; +using Grpc.Net.Client; using Microsoft.DurableTask.Worker.Grpc.Internal; namespace Microsoft.DurableTask.Worker.Grpc.Tests; @@ -26,6 +28,56 @@ public void InternalOptions_HasSafeDefaults() internalOptions.TransientRetryMaxAttempts.Should().Be(10); internalOptions.SilentDisconnectTimeout.Should().Be(TimeSpan.FromSeconds(120)); internalOptions.ChannelRecreator.Should().BeNull(); + internalOptions.CallInvokerDecorator.Should().BeNull(); + } + + [Fact] + public void SetCallInvokerDecorator_NullCallback_Throws() + { + // Arrange + GrpcDurableTaskWorkerOptions options = new(); + + // Act + Action act = () => options.SetCallInvokerDecorator(null!); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void ApplyCallInvokerDecorator_NoDecorator_ReturnsOriginalInvoker() + { + // Arrange + GrpcDurableTaskWorkerOptions options = new(); + CallInvoker invoker = GrpcChannel.ForAddress("http://localhost:9101").CreateCallInvoker(); + + // Act + CallInvoker result = options.ApplyCallInvokerDecorator(invoker); + + // Assert + result.Should().BeSameAs(invoker); + } + + [Fact] + public void ApplyCallInvokerDecorator_WithDecorator_ReturnsDecoratedInvoker() + { + // Arrange + GrpcDurableTaskWorkerOptions options = new(); + CallInvoker invoker = GrpcChannel.ForAddress("http://localhost:9102").CreateCallInvoker(); + CallInvoker decorated = GrpcChannel.ForAddress("http://localhost:9103").CreateCallInvoker(); + CallInvoker? observed = null; + options.SetCallInvokerDecorator(inner => + { + observed = inner; + return decorated; + }); + + // Act + CallInvoker result = options.ApplyCallInvokerDecorator(invoker); + + // Assert + result.Should().BeSameAs(decorated); + observed.Should().BeSameAs(invoker); } [Fact] From 3b1d379cecfe26b0476807c873a6264a6120465f Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 19 Aug 2026 13:14:22 -0400 Subject: [PATCH 2/4] Replace internal CallInvokerDecorator hook with public Interceptors collection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses review feedback that the previous revision exposed `public` extension methods (`SetCallInvokerDecorator` / `ApplyCallInvokerDecorator`) in a `.Internal` namespace whose only protection was an XML remark saying "do not use". `ApplyCallInvokerDecorator` was only ever called from the assembly that declared it, so it never needed to be public at all. The one hook that does need to cross an assembly boundary is now a first-class, supported public extensibility point modeled on `Grpc.Net.ClientFactory`'s `IHttpClientBuilder.AddInterceptor()`: public IList Interceptors { get; } added to both `GrpcDurableTaskClientOptions` and `GrpcDurableTaskWorkerOptions`. Net public API surface is smaller than the previous revision (one property per options class instead of four extension methods), and interceptors compose additively so multiple extensions can coexist — something a single `Func` could not do. The functional fix is unchanged. Interceptors are applied at exactly the same sites the decorator was: - `GrpcDurableTaskClient.GetCallInvoker` — outside any `ChannelRecreatingCallInvoker`, so the wrapper's internal channel swaps stay transparent to interceptors. - `GrpcDurableTaskWorker.GetCallInvoker`. - Both `ChannelRecreateResult` construction sites in `GrpcDurableTaskWorker.TryRecreateChannelAsync` (recreator-owned and worker-owned paths), so interceptors survive every channel recreate. The AzureBlobPayloads extension now calls `opt.Interceptors.Add(...)` instead of mutating `Channel`/`CallInvoker`, so enabling `UseExternalizedPayloads` no longer silently disables gRPC channel recreation, and the `Address`-only form `UseGrpc("http://localhost:4001")` no longer throws. Interceptor ordering is list order (first added is outermost); this is documented on both properties and pinned by a test. Tests renamed from `*CallInvokerDecoratorTests` to `*InterceptorsTests` and reworked to assert through the real invoker-building path rather than calling the removed extension method directly. Added coverage for interceptor ordering and for the purely-additive invariant (with an empty `Interceptors` list the produced invoker is reference-identical to the undecorated one) on both worker and client. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Client/Grpc/GrpcDurableTaskClient.cs | 13 +- .../Grpc/GrpcDurableTaskClientOptions.cs | 30 +- .../Internal/InternalOptionsExtensions.cs | 44 --- ...ientBuilderExtensions.AzureBlobPayloads.cs | 7 +- ...rkerBuilderExtensions.AzureBlobPayloads.cs | 7 +- src/Worker/Grpc/GrpcDurableTaskWorker.cs | 13 +- .../Grpc/GrpcDurableTaskWorkerOptions.cs | 29 +- .../Internal/InternalOptionsExtensions.cs | 45 --- ...ableTaskClientCallInvokerDecoratorTests.cs | 149 ------- .../GrpcDurableTaskClientInterceptorsTests.cs | 293 ++++++++++++++ ...> ExternalizedPayloadsInterceptorTests.cs} | 109 ++++-- ...ableTaskWorkerCallInvokerDecoratorTests.cs | 176 --------- .../GrpcDurableTaskWorkerInterceptorsTests.cs | 366 ++++++++++++++++++ ...pcDurableTaskWorkerOptionsInternalTests.cs | 51 --- 14 files changed, 804 insertions(+), 528 deletions(-) delete mode 100644 test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs create mode 100644 test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs rename test/Extensions/AzureBlobPayloads.Tests/{ExternalizedPayloadsCallInvokerDecoratorTests.cs => ExternalizedPayloadsInterceptorTests.cs} (71%) delete mode 100644 test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs create mode 100644 test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 304c74c1..865a83b3 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -7,8 +7,8 @@ using DurableTask.Core.Exceptions; using DurableTask.Core.History; using Google.Protobuf.WellKnownTypes; +using Grpc.Core.Interceptors; using Microsoft.DurableTask.Client.Entities; -using Microsoft.DurableTask.Client.Grpc.Internal; using Microsoft.DurableTask.Tracing; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -627,14 +627,17 @@ public override async Task> GetOrchestrationHistoryAsync( static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { - AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker undecorated); + AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker core); - // Decorate outside any ChannelRecreatingCallInvoker so the wrapper's internal channel swaps - // stay transparent to the decorator (and to any interceptor it installs). - callInvoker = options.ApplyCallInvokerDecorator(undecorated); + // Intercept outside any ChannelRecreatingCallInvoker so the wrapper's internal channel swaps + // stay transparent to the configured interceptors. + callInvoker = ApplyInterceptors(options.Interceptors, core); return disposable; } + static CallInvoker ApplyInterceptors(IList interceptors, CallInvoker invoker) + => interceptors.Count == 0 ? invoker : invoker.Intercept(interceptors.ToArray()); + static AsyncDisposable GetCallInvokerCore(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { Func>? recreator = options.Internal.ChannelRecreator; diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index a563f3f2..78a03b35 100644 --- a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs +++ b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs @@ -1,6 +1,8 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using Grpc.Core.Interceptors; + namespace Microsoft.DurableTask.Client.Grpc; /// @@ -23,6 +25,26 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions /// public CallInvoker? CallInvoker { get; set; } + /// + /// Gets the gRPC interceptors applied to every the client builds from its + /// configured transport, including invokers rebuilt after the underlying channel is recreated. + /// + /// + /// + /// This is the supported way to attach cross-cutting gRPC behavior — authentication headers, tracing, + /// logging, payload externalization — to the Durable Task gRPC client. Prefer it over supplying a + /// pre-built, already-intercepted in place of : an + /// externally-supplied invoker opts the client out of gRPC channel recreation, so a wedged connection + /// can never be replaced. + /// + /// + /// Interceptors run in list order — the first interceptor added is the outermost, so it observes each + /// outgoing call first and each response last. Registration is purely additive: while this collection + /// is empty, the client uses exactly the invoker its configured transport produces. + /// + /// + public IList Interceptors { get; } = new List(); + /// /// Gets the internal options. These are not exposed directly, but configurable via /// . @@ -57,13 +79,5 @@ internal class InternalOptions /// old channel so in-flight RPCs from peer clients are not interrupted. /// public Func>? ChannelRecreator { get; set; } - - /// - /// Gets or sets an optional decorator applied to every the client builds - /// from its configured transport. Extensions use this to attach interceptors without taking - /// ownership of , which would otherwise disable - /// recreation. - /// - public Func? CallInvokerDecorator { get; set; } } } diff --git a/src/Client/Grpc/Internal/InternalOptionsExtensions.cs b/src/Client/Grpc/Internal/InternalOptionsExtensions.cs index dec02bb6..800848f3 100644 --- a/src/Client/Grpc/Internal/InternalOptionsExtensions.cs +++ b/src/Client/Grpc/Internal/InternalOptionsExtensions.cs @@ -30,48 +30,4 @@ public static void SetChannelRecreator( { options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator)); } - - /// - /// Sets a callback that decorates every the client builds from its configured - /// transport. Use this instead of replacing with an - /// intercepted : clearing the channel leaves the - /// client with no way to recreate a wedged connection. - /// - /// The gRPC client options. - /// The decorator callback. - /// - /// This is an internal API that supports the DurableTask infrastructure and not subject to - /// the same compatibility standards as public APIs. It may be changed or removed without notice in - /// any release. You should only use it directly in your code with extreme caution and knowing that - /// doing so can result in application failures when updating to a new DurableTask release. - /// - public static void SetCallInvokerDecorator( - this GrpcDurableTaskClientOptions options, - Func decorator) - { - options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator)); - } - - /// - /// Applies the decorator registered by to , - /// returning it unchanged when no decorator is registered. Callers that build a - /// from these options must route it through this method so registered - /// interceptors are not silently dropped. - /// - /// The gRPC client options. - /// The invoker to decorate. - /// The decorated invoker, or when no decorator is registered. - /// - /// This is an internal API that supports the DurableTask infrastructure and not subject to - /// the same compatibility standards as public APIs. It may be changed or removed without notice in - /// any release. You should only use it directly in your code with extreme caution and knowing that - /// doing so can result in application failures when updating to a new DurableTask release. - /// - public static CallInvoker ApplyCallInvokerDecorator( - this GrpcDurableTaskClientOptions options, - CallInvoker invoker) - { - Func? decorator = options.Internal.CallInvokerDecorator; - return decorator is null ? invoker : decorator(invoker); - } } diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 3b481791..d8607ec6 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -1,10 +1,8 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using Grpc.Core.Interceptors; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Client.Grpc; -using Microsoft.DurableTask.Client.Grpc.Internal; using Microsoft.DurableTask.Converters; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; @@ -38,11 +36,10 @@ static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientB { LargePayloadStorageOptions opts = monitor.Get(builder.Name); - // Register a decorator rather than moving Channel onto an intercepted CallInvoker. + // Register an interceptor rather than moving Channel onto an intercepted CallInvoker. // Clearing Channel would disable the client's gRPC channel recreation, and requiring a // pre-built Channel/CallInvoker would rule out the Address-only configuration. - opt.SetCallInvokerDecorator( - invoker => invoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts))); + opt.Interceptors.Add(new AzureBlobPayloadsSideCarInterceptor(store, opts)); }); return builder; diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index ea76c02f..e1f8387d 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -1,11 +1,9 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using Grpc.Core.Interceptors; using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; -using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -62,11 +60,10 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB { LargePayloadStorageOptions opts = monitor.Get(builder.Name); - // Register a decorator rather than moving Channel onto an intercepted CallInvoker. + // Register an interceptor rather than moving Channel onto an intercepted CallInvoker. // Clearing Channel would disable the worker's gRPC channel recreation, and requiring a // pre-built Channel/CallInvoker would rule out the Address-only configuration. - opt.SetCallInvokerDecorator( - invoker => invoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts))); + opt.Interceptors.Add(new AzureBlobPayloadsSideCarInterceptor(store, opts)); opt.Capabilities.Add(P.WorkerCapability.LargePayloads); }); diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index 53b2964b..0a4330e2 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -2,7 +2,7 @@ // Licensed under the MIT License. using System.Diagnostics; -using Microsoft.DurableTask.Worker.Grpc.Internal; +using Grpc.Core.Interceptors; using Microsoft.DurableTask.Worker.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -129,7 +129,7 @@ async Task TryRecreateChannelAsync( // carrying that ownership forward to the recreated state. return new ChannelRecreateResult( true, - this.grpcOptions.ApplyCallInvokerDecorator(newChannel.CreateCallInvoker()), + ApplyInterceptors(this.grpcOptions.Interceptors, newChannel.CreateCallInvoker()), newChannel.Target, default, newChannel); @@ -162,7 +162,7 @@ async Task TryRecreateChannelAsync( AsyncDisposable newDisposable = CreateOwnedChannelDisposable(newChannel); return new ChannelRecreateResult( true, - this.grpcOptions.ApplyCallInvokerDecorator(newChannel.CreateCallInvoker()), + ApplyInterceptors(this.grpcOptions.Interceptors, newChannel.CreateCallInvoker()), newChannel.Target, newDisposable, newChannel); @@ -305,8 +305,8 @@ and not AccessViolationException AsyncDisposable GetCallInvoker(out CallInvoker callInvoker, out string address) { - AsyncDisposable disposable = this.GetCallInvokerCore(out CallInvoker undecorated, out address); - callInvoker = this.grpcOptions.ApplyCallInvokerDecorator(undecorated); + AsyncDisposable disposable = this.GetCallInvokerCore(out CallInvoker core, out address); + callInvoker = ApplyInterceptors(this.grpcOptions.Interceptors, core); return disposable; } @@ -332,6 +332,9 @@ AsyncDisposable GetCallInvokerCore(out CallInvoker callInvoker, out string addre return CreateOwnedChannelDisposable(c); } + static CallInvoker ApplyInterceptors(IList interceptors, CallInvoker invoker) + => interceptors.Count == 0 ? invoker : invoker.Intercept(interceptors.ToArray()); + static ILogger CreateLogger(ILoggerFactory loggerFactory, DurableTaskWorkerOptions options) { // Use the new, more specific category name for gRPC worker logs diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index fecb674c..e656ce8b 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using Grpc.Core.Interceptors; using Microsoft.DurableTask.Worker.Grpc.Internal; using P = Microsoft.DurableTask.Protobuf; @@ -38,6 +39,26 @@ public sealed class GrpcDurableTaskWorkerOptions : DurableTaskWorkerOptions /// public CallInvoker? CallInvoker { get; set; } + /// + /// Gets the gRPC interceptors applied to every the worker builds from its + /// configured transport, including invokers rebuilt after the underlying channel is recreated. + /// + /// + /// + /// This is the supported way to attach cross-cutting gRPC behavior — authentication headers, tracing, + /// logging, payload externalization — to the Durable Task gRPC worker. Prefer it over supplying a + /// pre-built, already-intercepted in place of : an + /// externally-supplied invoker opts the worker out of gRPC channel recreation, so a wedged connection + /// can never be replaced. + /// + /// + /// Interceptors run in list order — the first interceptor added is the outermost, so it observes each + /// outgoing call first and each response last. Registration is purely additive: while this collection + /// is empty, the worker uses exactly the invoker its configured transport produces. + /// + /// + public IList Interceptors { get; } = new List(); + /// /// Gets the collection of capabilities enabled on this worker. /// Capabilities are announced to the backend on connection. @@ -167,14 +188,6 @@ internal class InternalOptions /// public Func>? ChannelRecreator { get; set; } - /// - /// Gets or sets an optional decorator applied to every the worker builds - /// from its configured transport, including invokers rebuilt after a channel recreate. Extensions - /// use this to attach interceptors without taking ownership of - /// , which would otherwise disable recreation. - /// - public Func? CallInvokerDecorator { get; set; } - /// /// Gets or sets a callback that is invoked when activity work items are received or finished. /// diff --git a/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs b/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs index ce922826..81ad09d5 100644 --- a/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs +++ b/src/Worker/Grpc/Internal/InternalOptionsExtensions.cs @@ -83,51 +83,6 @@ public static void SetChannelRecreator( options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator)); } - /// - /// Sets a callback that decorates every the worker builds from its configured - /// transport, including invokers rebuilt after a channel recreate. Use this instead of replacing - /// with an intercepted - /// : clearing the channel leaves the worker with - /// no way to recreate a wedged connection. - /// - /// The gRPC worker options. - /// The decorator callback. - /// - /// This is an internal API that supports the DurableTask infrastructure and not subject to - /// the same compatibility standards as public APIs. It may be changed or removed without notice in - /// any release. You should only use it directly in your code with extreme caution and knowing that - /// doing so can result in application failures when updating to a new DurableTask release. - /// - public static void SetCallInvokerDecorator( - this GrpcDurableTaskWorkerOptions options, - Func decorator) - { - options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator)); - } - - /// - /// Applies the decorator registered by to , - /// returning it unchanged when no decorator is registered. Callers that build a - /// from these options must route it through this method so registered - /// interceptors are not silently dropped. - /// - /// The gRPC worker options. - /// The invoker to decorate. - /// The decorated invoker, or when no decorator is registered. - /// - /// This is an internal API that supports the DurableTask infrastructure and not subject to - /// the same compatibility standards as public APIs. It may be changed or removed without notice in - /// any release. You should only use it directly in your code with extreme caution and knowing that - /// doing so can result in application failures when updating to a new DurableTask release. - /// - public static CallInvoker ApplyCallInvokerDecorator( - this GrpcDurableTaskWorkerOptions options, - CallInvoker invoker) - { - Func? decorator = options.Internal.CallInvokerDecorator; - return decorator is null ? invoker : decorator(invoker); - } - /// /// Sets the deadline applied to the initial Hello RPC during worker connect. A wedged /// handshake on a half-open HTTP/2 connection no longer hangs the reconnect loop indefinitely. diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs deleted file mode 100644 index 31fb5bc3..00000000 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientCallInvokerDecoratorTests.cs +++ /dev/null @@ -1,149 +0,0 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -using System.Reflection; -using Grpc.Core; -using Microsoft.DurableTask.Client.Grpc.Internal; -using Microsoft.Extensions.Logging.Abstractions; - -namespace Microsoft.DurableTask.Client.Grpc.Tests; - -/// -/// Verifies that a registered CallInvokerDecorator is applied on every transport path the client -/// supports, and that it wraps outside so the -/// wrapper's internal channel swaps stay transparent to the decorator. -/// -public class GrpcDurableTaskClientCallInvokerDecoratorTests -{ - static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskClient) - .GetMethod("GetCallInvoker", BindingFlags.Static | BindingFlags.NonPublic)!; - - [Fact] - public async Task GetCallInvoker_ChannelPath_AppliesDecorator() - { - // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5201"); - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskClientOptions options = new() { Channel = channel }; - options.SetCallInvokerDecorator(_ => sentinel); - - try - { - // Act - (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); - - // Assert - callInvoker.Should().BeSameAs(sentinel); - await disposable.DisposeAsync(); - } - finally - { - channel.Dispose(); - } - } - - [Fact] - public async Task GetCallInvoker_ExternalCallInvokerPath_AppliesDecorator() - { - // Arrange - CallInvoker external = CreateSentinel(); - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskClientOptions options = new() { CallInvoker = external }; - CallInvoker? observed = null; - options.SetCallInvokerDecorator(invoker => - { - observed = invoker; - return sentinel; - }); - - // Act - (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); - - // Assert - callInvoker.Should().BeSameAs(sentinel); - observed.Should().BeSameAs(external); - await disposable.DisposeAsync(); - } - - [Fact] - public async Task GetCallInvoker_AddressPath_AppliesDecorator() - { - // Arrange - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskClientOptions options = new() { Address = "http://localhost:5202" }; - options.SetCallInvokerDecorator(_ => sentinel); - - // Act - (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); - - // Assert - callInvoker.Should().BeSameAs(sentinel); - await disposable.DisposeAsync(); - } - - [Fact] - public async Task GetCallInvoker_WithRecreator_AppliesDecoratorOutsideRecreatingInvoker() - { - // Arrange: recreation stays enabled, so the core invoker is a ChannelRecreatingCallInvoker. - // The decorator must receive that wrapper (i.e. wrap outside it), otherwise the wrapper's - // internal channel swaps would replace the decorated invoker and drop the interceptor. - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5203"); - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskClientOptions options = new() { Channel = channel }; - options.SetChannelRecreator((existing, ct) => Task.FromResult(existing)); - CallInvoker? observed = null; - options.SetCallInvokerDecorator(invoker => - { - observed = invoker; - return sentinel; - }); - - try - { - // Act - (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); - - // Assert - observed.Should().BeOfType(); - callInvoker.Should().BeSameAs(sentinel); - await disposable.DisposeAsync(); - } - finally - { - channel.Dispose(); - } - } - - [Fact] - public async Task GetCallInvoker_WithoutDecorator_ReturnsUndecoratedInvoker() - { - // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5204"); - GrpcDurableTaskClientOptions options = new() { Channel = channel }; - - try - { - // Act - (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); - - // Assert - // Assert: with no decorator registered the invoker is exactly what core builds today. - callInvoker.Should().BeOfType(channel.CreateCallInvoker().GetType()); - await disposable.DisposeAsync(); - } - finally - { - channel.Dispose(); - } - } - - static CallInvoker CreateSentinel() => GrpcChannel.ForAddress("http://sentinel.invalid").CreateCallInvoker(); - - static (AsyncDisposable Disposable, CallInvoker CallInvoker) InvokeGetCallInvoker( - GrpcDurableTaskClientOptions options) - { - object?[] args = { options, NullLogger.Instance, null }; - AsyncDisposable disposable = (AsyncDisposable)GetCallInvokerMethod.Invoke(null, args)!; - return (disposable, (CallInvoker)args[2]!); - } -} diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs new file mode 100644 index 00000000..641f7d74 --- /dev/null +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs @@ -0,0 +1,293 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.DurableTask.Client.Grpc.Internal; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Client.Grpc.Tests; + +/// +/// Verifies that is honored on every transport +/// path the client supports, that interception wraps outside +/// so the wrapper's internal channel swaps stay transparent, and that registering no interceptors leaves +/// the transport invoker completely untouched. +/// +public class GrpcDurableTaskClientInterceptorsTests +{ + static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskClient) + .GetMethod("GetCallInvoker", BindingFlags.Static | BindingFlags.NonPublic)!; + static readonly MethodInfo GetCallInvokerCoreMethod = typeof(GrpcDurableTaskClient) + .GetMethod("GetCallInvokerCore", BindingFlags.Static | BindingFlags.NonPublic)!; + + [Fact] + public void Interceptors_DefaultsToEmptyList() + { + // Arrange + GrpcDurableTaskClientOptions options = new(); + + // Act + IList interceptors = options.Interceptors; + + // Assert + interceptors.Should().NotBeNull().And.BeEmpty(); + } + + [Fact] + public async Task GetCallInvoker_ChannelPath_AppliesInterceptors() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5201"); + List log = new(); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + + try + { + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public async Task GetCallInvoker_AddressPath_AppliesInterceptors() + { + // Arrange + List log = new(); + GrpcDurableTaskClientOptions options = new() { Address = "http://localhost:5202" }; + options.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + await disposable.DisposeAsync(); + } + + [Fact] + public async Task GetCallInvoker_ExternalCallInvokerPath_AppliesInterceptors() + { + // Arrange + StubCallInvoker external = new(); + List log = new(); + GrpcDurableTaskClientOptions options = new() { CallInvoker = external }; + options.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: true)); + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + external.CallCount.Should().Be(1); + await disposable.DisposeAsync(); + } + + [Fact] + public async Task GetCallInvoker_WithRecreator_AppliesInterceptorsOutsideRecreatingInvoker() + { + // Arrange: recreation stays enabled, so the core invoker is a ChannelRecreatingCallInvoker. + // Interception must wrap outside it, otherwise the wrapper's internal channel swaps would replace + // the intercepted invoker and drop the interceptors. + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5203"); + List log = new(); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.SetChannelRecreator((existing, ct) => Task.FromResult(existing)); + options.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + + try + { + // Act + (AsyncDisposable coreDisposable, CallInvoker coreInvoker) = InvokeGetCallInvokerCore(options); + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + CallProbe.Invoke(callInvoker); + + // Assert + coreInvoker.Should().BeOfType(); + callInvoker.Should().NotBeOfType(); + log.Should().Equal("only"); + + await coreDisposable.DisposeAsync(); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public async Task GetCallInvoker_NoInterceptors_ReturnsTransportInvokerUnchanged() + { + // Arrange: an externally-supplied invoker is handed back verbatim by the core builder, so it is + // the one path where the purely-additive invariant can be asserted by reference. + StubCallInvoker external = new(); + GrpcDurableTaskClientOptions options = new() { CallInvoker = external }; + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + callInvoker.Should().BeSameAs(external); + await disposable.DisposeAsync(); + } + + [Fact] + public async Task GetCallInvoker_NoInterceptors_WithRecreator_ReturnsRecreatingInvokerUnwrapped() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5204"); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.SetChannelRecreator((existing, ct) => Task.FromResult(existing)); + + try + { + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + + // Assert + callInvoker.Should().BeOfType(); + await disposable.DisposeAsync(); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public async Task GetCallInvoker_MultipleInterceptors_RunsInListOrder() + { + // Arrange: the documented contract is that the first interceptor added is the outermost, so it + // observes the outgoing call before every interceptor added after it. + StubCallInvoker external = new(); + List log = new(); + GrpcDurableTaskClientOptions options = new() { CallInvoker = external }; + options.Interceptors.Add(new RecordingInterceptor("first", log, passThrough: true)); + options.Interceptors.Add(new RecordingInterceptor("second", log, passThrough: true)); + + // Act + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("first", "second"); + external.CallCount.Should().Be(1); + await disposable.DisposeAsync(); + } + + static (AsyncDisposable Disposable, CallInvoker CallInvoker) InvokeGetCallInvoker( + GrpcDurableTaskClientOptions options) + => Invoke(GetCallInvokerMethod, options); + + static (AsyncDisposable Disposable, CallInvoker CallInvoker) InvokeGetCallInvokerCore( + GrpcDurableTaskClientOptions options) + => Invoke(GetCallInvokerCoreMethod, options); + + static (AsyncDisposable Disposable, CallInvoker CallInvoker) Invoke( + MethodInfo method, GrpcDurableTaskClientOptions options) + { + object?[] args = { options, NullLogger.Instance, null }; + AsyncDisposable disposable = (AsyncDisposable)method.Invoke(null, args)!; + return (disposable, (CallInvoker)args[2]!); + } + + /// + /// Drives a single unary call through an invoker without touching the network: the innermost + /// participant always short-circuits. + /// + static class CallProbe + { + static readonly Marshaller Marshaller = Marshallers.Create( + m => Encoding.UTF8.GetBytes(m.Value), b => new ProbeMessage { Value = Encoding.UTF8.GetString(b) }); + + static readonly Method Method = new( + MethodType.Unary, "Probe", "Probe", Marshaller, Marshaller); + + public static void Invoke(CallInvoker invoker) + => invoker.AsyncUnaryCall(Method, null, default, new ProbeMessage()).Dispose(); + } + + sealed class ProbeMessage + { + public string Value { get; set; } = string.Empty; + } + + sealed class RecordingInterceptor : Interceptor + { + readonly string tag; + readonly IList log; + readonly bool passThrough; + + public RecordingInterceptor(string tag, IList log, bool passThrough) + { + this.tag = tag; + this.log = log; + this.passThrough = passThrough; + } + + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + { + this.log.Add(this.tag); + return this.passThrough ? continuation(request, context) : StubCallInvoker.EmptyCall(); + } + } + + sealed class StubCallInvoker : CallInvoker + { + int callCount; + + public int CallCount => Volatile.Read(ref this.callCount); + + public static AsyncUnaryCall EmptyCall() + where TResponse : class + { + return new AsyncUnaryCall( + Task.FromResult(Activator.CreateInstance()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + } + + public override AsyncUnaryCall AsyncUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + { + Interlocked.Increment(ref this.callCount); + return EmptyCall(); + } + + public override TResponse BlockingUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + + public override AsyncClientStreamingCall AsyncClientStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncDuplexStreamingCall AsyncDuplexStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncServerStreamingCall AsyncServerStreamingCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs similarity index 71% rename from test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs rename to test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs index a514533b..e81598d0 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsCallInvokerDecoratorTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs @@ -1,16 +1,16 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using System.Reflection; using Google.Protobuf; using Grpc.Core; using Grpc.Net.Client; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Client.Grpc; -using Microsoft.DurableTask.Client.Grpc.Internal; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; -using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -20,9 +20,11 @@ namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; /// Verifies that enabling externalized payloads composes with the gRPC transport options instead of /// replacing them. Previously the extension moved Channel onto an intercepted CallInvoker /// and nulled Channel, which silently disabled channel recreation on both the worker and the -/// client, and made the Address-only setup unusable. +/// client, and made the Address-only setup unusable. It now registers an interceptor on the +/// supported Interceptors collection, which the worker and client apply to every invoker they +/// build. /// -public class ExternalizedPayloadsCallInvokerDecoratorTests +public class ExternalizedPayloadsInterceptorTests { static readonly Marshaller RequestMarshaller = Marshallers.Create( r => r.ToByteArray(), P.CreateInstanceRequest.Parser.ParseFrom); @@ -51,6 +53,7 @@ public void Worker_WithChannel_PreservesChannelSoRecreationStaysEnabled() // Assert options.Channel.Should().BeSameAs(channel); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); } [Fact] @@ -69,6 +72,7 @@ public void Client_WithChannel_PreservesChannelSoRecreationStaysEnabled() // Assert options.Channel.Should().BeSameAs(channel); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); } [Fact] @@ -119,9 +123,9 @@ public void Worker_WithExternalCallInvoker_PreservesConfiguredInvoker() builder.UseExternalizedPayloads(); GrpcDurableTaskWorkerOptions options = GetOptions(services); - // Assert: the extension no longer mutates the configured invoker; it decorates on use instead. + // Assert: the extension no longer mutates the configured invoker; it intercepts on use instead. options.CallInvoker.Should().BeSameAs(external); - options.ApplyCallInvokerDecorator(external).Should().NotBeSameAs(external); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); } [Fact] @@ -142,23 +146,16 @@ public void Worker_StillAnnouncesLargePayloadsCapability() } [Fact] - public async Task Worker_RegisteredDecorator_ExternalizesLargePayloads() + public async Task Worker_RegisteredInterceptor_ExternalizesLargePayloads() { // Arrange - ServiceCollection services = new(); - RecordingPayloadStore store = new(); - services.AddSingleton(store); - services.Configure(o => o.ThresholdBytes = 1); - DefaultDurableTaskWorkerBuilder builder = new(null, services); - builder.UseGrpc("http://localhost:4001"); - builder.UseExternalizedPayloads(); - GrpcDurableTaskWorkerOptions options = GetOptions(services); - RecordingCallInvoker inner = new(); - CallInvoker decorated = options.ApplyCallInvokerDecorator(inner); + RecordingPayloadStore store = new(); + ServiceProvider provider = BuildWorkerProvider(inner, store); - // Act - await InvokeCreateInstanceAsync(decorated, new string('x', 1024)); + // Act: build the invoker the same way the running worker does. + CallInvoker invoker = BuildWorkerCallInvoker(provider); + await InvokeCreateInstanceAsync(invoker, new string('x', 1024)); // Assert store.UploadCount.Should().Be(1); @@ -166,23 +163,22 @@ public async Task Worker_RegisteredDecorator_ExternalizesLargePayloads() } [Fact] - public async Task Client_RegisteredDecorator_ExternalizesLargePayloads() + public async Task Client_RegisteredInterceptor_ExternalizesLargePayloads() { // Arrange - ServiceCollection services = new(); + RecordingCallInvoker inner = new(); RecordingPayloadStore store = new(); + ServiceCollection services = new(); services.AddSingleton(store); services.Configure(o => o.ThresholdBytes = 1); DefaultDurableTaskClientBuilder builder = new(null, services); - builder.UseGrpc("http://localhost:4001"); + builder.UseGrpc(opt => opt.CallInvoker = inner); builder.UseExternalizedPayloads(); GrpcDurableTaskClientOptions options = GetOptions(services); - RecordingCallInvoker inner = new(); - CallInvoker decorated = options.ApplyCallInvokerDecorator(inner); - - // Act - await InvokeCreateInstanceAsync(decorated, new string('x', 1024)); + // Act: build the invoker the same way the running client does. + CallInvoker invoker = BuildClientCallInvoker(options); + await InvokeCreateInstanceAsync(invoker, new string('x', 1024)); // Assert store.UploadCount.Should().Be(1); @@ -195,6 +191,65 @@ public async Task Client_RegisteredDecorator_ExternalizesLargePayloads() return invoker.AsyncUnaryCall(CreateInstanceMethod, null, default, request).ResponseAsync; } + static ServiceProvider BuildWorkerProvider(CallInvoker inner, PayloadStore store) + { + ServiceCollection services = new(); + services.AddSingleton(store); + services.Configure(o => o.ThresholdBytes = 1); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc(opt => opt.CallInvoker = inner); + builder.UseExternalizedPayloads(); + return services.BuildServiceProvider(); + } + + /// + /// Builds a through the worker's own private invoker-building path, so the + /// test cannot accidentally re-implement how interceptors are applied. + /// + /// The configured service provider. + /// The invoker the worker would use. + static CallInvoker BuildWorkerCallInvoker(ServiceProvider provider) + { + Type workerType = typeof(GrpcDurableTaskWorkerOptions).Assembly + .GetType("Microsoft.DurableTask.Worker.Grpc.GrpcDurableTaskWorker", throwOnError: true)!; + + object worker = Activator.CreateInstance( + workerType, + new object?[] + { + string.Empty, + Mock.Of(), + provider.GetRequiredService>(), + provider.GetRequiredService>(), + provider, + NullLoggerFactory.Instance, + null, + null, + null, + })!; + + MethodInfo getCallInvoker = workerType.GetMethod( + "GetCallInvoker", BindingFlags.Instance | BindingFlags.NonPublic)!; + object?[] args = { null, null }; + getCallInvoker.Invoke(worker, args); + return (CallInvoker)args[0]!; + } + + /// + /// Builds a through the client's own private invoker-building path, so the + /// test cannot accidentally re-implement how interceptors are applied. + /// + /// The configured client options. + /// The invoker the client would use. + static CallInvoker BuildClientCallInvoker(GrpcDurableTaskClientOptions options) + { + MethodInfo getCallInvoker = typeof(GrpcDurableTaskClient).GetMethod( + "GetCallInvoker", BindingFlags.Static | BindingFlags.NonPublic)!; + object?[] args = { options, NullLogger.Instance, null }; + getCallInvoker.Invoke(null, args); + return (CallInvoker)args[2]!; + } + static TOptions GetOptions(IServiceCollection services) where TOptions : class { diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs deleted file mode 100644 index 092fdd94..00000000 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerCallInvokerDecoratorTests.cs +++ /dev/null @@ -1,176 +0,0 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -using System.Reflection; -using Grpc.Core; -using Microsoft.DurableTask.Worker.Grpc.Internal; -using Microsoft.Extensions.Logging.Abstractions; - -namespace Microsoft.DurableTask.Worker.Grpc.Tests; - -/// -/// Verifies that a registered CallInvokerDecorator is applied everywhere the worker produces a -/// — including invokers rebuilt after a channel recreate — so extensions can -/// install interceptors without clearing Channel and disabling channel recreation. -/// -public class GrpcDurableTaskWorkerCallInvokerDecoratorTests -{ - static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskWorker) - .GetMethod("GetCallInvoker", BindingFlags.Instance | BindingFlags.NonPublic)!; - static readonly MethodInfo TryRecreateChannelAsyncMethod = typeof(GrpcDurableTaskWorker) - .GetMethod("TryRecreateChannelAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; - - [Fact] - public void GetCallInvoker_WithDecorator_ReturnsDecoratedInvoker() - { - // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5101"); - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; - CallInvoker? observed = null; - grpcOptions.SetCallInvokerDecorator(invoker => - { - observed = invoker; - return sentinel; - }); - - GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); - - try - { - // Act - InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); - - // Assert - callInvoker.Should().BeSameAs(sentinel); - observed.Should().NotBeNull().And.NotBeSameAs(sentinel); - address.Should().Be(channel.Target); - } - finally - { - DisposeChannel(channel); - } - } - - [Fact] - public void GetCallInvoker_WithoutDecorator_ReturnsUndecoratedInvoker() - { - // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5102"); - GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; - GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); - - try - { - // Act - InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); - - // Assert - // Assert: with no decorator registered the invoker is exactly what the channel produces. - callInvoker.Should().BeOfType(channel.CreateCallInvoker().GetType()); - address.Should().Be(channel.Target); - } - finally - { - DisposeChannel(channel); - } - } - - [Fact] - public async Task TryRecreateChannelAsync_ChannelWithRecreatorAndDecorator_RecreatesAndDecorates() - { - // Arrange: this is the shape the AzureBlobPayloads extension used to break — a DTS-configured - // Channel plus recreator. Path 1 requires Channel to still be set, and the invoker the worker - // builds from the replacement channel must still carry the decorator. - GrpcChannel currentChannel = GrpcChannel.ForAddress("http://localhost:5103"); - GrpcChannel recreatedChannel = GrpcChannel.ForAddress("http://localhost:5104"); - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = currentChannel }; - grpcOptions.SetChannelRecreator((channel, ct) => Task.FromResult(recreatedChannel)); - grpcOptions.SetCallInvokerDecorator(_ => sentinel); - - GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); - - try - { - // Act - object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); - - // Assert - GetResultProperty(result, "Recreated").Should().BeTrue(); - GetResultProperty(result, "NewChannel").Should().BeSameAs(recreatedChannel); - GetResultProperty(result, "NewCallInvoker").Should().BeSameAs(sentinel); - } - finally - { - DisposeChannel(currentChannel); - DisposeChannel(recreatedChannel); - } - } - - [Fact] - public async Task TryRecreateChannelAsync_WorkerOwnedChannelWithDecorator_DecoratesRebuiltInvoker() - { - // Arrange: Address-only configuration takes the worker-owned rebuild path. - CallInvoker sentinel = CreateSentinel(); - GrpcDurableTaskWorkerOptions grpcOptions = new() { Address = "http://localhost:5105" }; - grpcOptions.SetCallInvokerDecorator(_ => sentinel); - - GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); - GrpcChannel currentChannel = GrpcChannel.ForAddress(grpcOptions.Address); - - try - { - // Act - object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); - - // Assert - GetResultProperty(result, "Recreated").Should().BeTrue(); - GetResultProperty(result, "NewCallInvoker").Should().BeSameAs(sentinel); - - AsyncDisposable newDisposable = GetResultProperty(result, "NewWorkerOwnedDisposable"); - await newDisposable.DisposeAsync(); - } - finally - { - DisposeChannel(currentChannel); - } - } - - static CallInvoker CreateSentinel() => GrpcChannel.ForAddress("http://sentinel.invalid").CreateCallInvoker(); - - static void InvokeGetCallInvoker(GrpcDurableTaskWorker worker, out CallInvoker callInvoker, out string address) - { - object?[] args = { null, null }; - GetCallInvokerMethod.Invoke(worker, args); - callInvoker = (CallInvoker)args[0]!; - address = (string)args[1]!; - } - - static async Task InvokeTryRecreateChannelAsync(GrpcDurableTaskWorker worker, GrpcChannel currentChannel) - { - object?[] args = { CancellationToken.None, default(AsyncDisposable), currentChannel }; - Task task = (Task)TryRecreateChannelAsyncMethod.Invoke(worker, args)!; - await task; - return task.GetType().GetProperty("Result")!.GetValue(task)!; - } - - static T GetResultProperty(object result, string propertyName) - => (T)result.GetType().GetProperty(propertyName)!.GetValue(result)!; - - static void DisposeChannel(GrpcChannel channel) => channel.Dispose(); - - static GrpcDurableTaskWorker CreateWorker(GrpcDurableTaskWorkerOptions grpcOptions) - { - return new GrpcDurableTaskWorker( - name: "Test", - factory: Mock.Of(), - grpcOptions: new OptionsMonitorStub(grpcOptions), - workerOptions: new OptionsMonitorStub(new DurableTaskWorkerOptions()), - services: Mock.Of(), - loggerFactory: NullLoggerFactory.Instance, - orchestrationFilter: null, - exceptionPropertiesProvider: null, - workItemFiltersMonitor: null); - } -} diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs new file mode 100644 index 00000000..73600663 --- /dev/null +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs @@ -0,0 +1,366 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +/// +/// Verifies that is honored everywhere the worker +/// produces a — including the invokers rebuilt on both channel-recreate paths — +/// so extensions can install interceptors without clearing Channel and disabling channel recreation. +/// +public class GrpcDurableTaskWorkerInterceptorsTests +{ + static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskWorker) + .GetMethod("GetCallInvoker", BindingFlags.Instance | BindingFlags.NonPublic)!; + static readonly MethodInfo TryRecreateChannelAsyncMethod = typeof(GrpcDurableTaskWorker) + .GetMethod("TryRecreateChannelAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + + [Fact] + public void Interceptors_DefaultsToEmptyList() + { + // Arrange + GrpcDurableTaskWorkerOptions options = new(); + + // Act + IList interceptors = options.Interceptors; + + // Assert + interceptors.Should().NotBeNull().And.BeEmpty(); + } + + [Fact] + public void GetCallInvoker_ChannelPath_AppliesInterceptors() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5101"); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + address.Should().Be(channel.Target); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public void GetCallInvoker_AddressPath_AppliesInterceptors() + { + // Arrange + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Address = "http://localhost:5102" }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out _); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + } + + [Fact] + public void GetCallInvoker_ExternalCallInvokerPath_AppliesInterceptors() + { + // Arrange + StubCallInvoker external = new(); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { CallInvoker = external }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: true)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out _); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("only"); + external.CallCount.Should().Be(1); + } + + [Fact] + public void GetCallInvoker_NoInterceptors_ReturnsTransportInvokerUnchanged() + { + // Arrange: an externally-supplied invoker is handed back verbatim by the core builder, so it is + // the one path where the purely-additive invariant can be asserted by reference. + StubCallInvoker external = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { CallInvoker = external }; + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out _); + + // Assert + callInvoker.Should().BeSameAs(external); + } + + [Fact] + public void GetCallInvoker_NoInterceptors_ChannelPath_ReturnsChannelInvokerUnwrapped() + { + // Arrange + GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:5103"); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = channel }; + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out string address); + + // Assert + callInvoker.Should().BeOfType(channel.CreateCallInvoker().GetType()); + address.Should().Be(channel.Target); + } + finally + { + channel.Dispose(); + } + } + + [Fact] + public void GetCallInvoker_MultipleInterceptors_RunsInListOrder() + { + // Arrange: the documented contract is that the first interceptor added is the outermost, so it + // observes the outgoing call before every interceptor added after it. + StubCallInvoker external = new(); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { CallInvoker = external }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("first", log, passThrough: true)); + grpcOptions.Interceptors.Add(new RecordingInterceptor("second", log, passThrough: true)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + // Act + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out _); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("first", "second"); + external.CallCount.Should().Be(1); + } + + [Fact] + public async Task TryRecreateChannelAsync_RecreatorPath_InterceptsRebuiltInvoker() + { + // Arrange: this is the shape the AzureBlobPayloads extension used to break — a DTS-configured + // Channel plus recreator. This path requires Channel to still be set, and the invoker built from + // the replacement channel must still carry the configured interceptors. + GrpcChannel currentChannel = GrpcChannel.ForAddress("http://localhost:5104"); + GrpcChannel recreatedChannel = GrpcChannel.ForAddress("http://localhost:5105"); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = currentChannel }; + grpcOptions.SetChannelRecreator((channel, ct) => Task.FromResult(recreatedChannel)); + grpcOptions.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + // Assert + GetResultProperty(result, "Recreated").Should().BeTrue(); + GetResultProperty(result, "NewChannel").Should().BeSameAs(recreatedChannel); + + CallProbe.Invoke(GetResultProperty(result, "NewCallInvoker")); + log.Should().Equal("only"); + } + finally + { + currentChannel.Dispose(); + recreatedChannel.Dispose(); + } + } + + [Fact] + public async Task TryRecreateChannelAsync_WorkerOwnedPath_InterceptsRebuiltInvoker() + { + // Arrange: Address-only configuration takes the worker-owned rebuild path. + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Address = "http://localhost:5106" }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("only", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + GrpcChannel currentChannel = GrpcChannel.ForAddress(grpcOptions.Address); + + try + { + // Act + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + // Assert + GetResultProperty(result, "Recreated").Should().BeTrue(); + + CallProbe.Invoke(GetResultProperty(result, "NewCallInvoker")); + log.Should().Equal("only"); + + AsyncDisposable newDisposable = GetResultProperty(result, "NewWorkerOwnedDisposable"); + await newDisposable.DisposeAsync(); + } + finally + { + currentChannel.Dispose(); + } + } + + [Fact] + public async Task TryRecreateChannelAsync_NoInterceptors_ReturnsChannelInvokerUnwrapped() + { + // Arrange + GrpcChannel currentChannel = GrpcChannel.ForAddress("http://localhost:5107"); + GrpcChannel recreatedChannel = GrpcChannel.ForAddress("http://localhost:5108"); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = currentChannel }; + grpcOptions.SetChannelRecreator((channel, ct) => Task.FromResult(recreatedChannel)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + // Assert + GetResultProperty(result, "NewCallInvoker") + .Should().BeOfType(recreatedChannel.CreateCallInvoker().GetType()); + } + finally + { + currentChannel.Dispose(); + recreatedChannel.Dispose(); + } + } + + static void InvokeGetCallInvoker(GrpcDurableTaskWorker worker, out CallInvoker callInvoker, out string address) + { + object?[] args = { null, null }; + GetCallInvokerMethod.Invoke(worker, args); + callInvoker = (CallInvoker)args[0]!; + address = (string)args[1]!; + } + + static async Task InvokeTryRecreateChannelAsync(GrpcDurableTaskWorker worker, GrpcChannel currentChannel) + { + object?[] args = { CancellationToken.None, default(AsyncDisposable), currentChannel }; + Task task = (Task)TryRecreateChannelAsyncMethod.Invoke(worker, args)!; + await task; + return task.GetType().GetProperty("Result")!.GetValue(task)!; + } + + static T GetResultProperty(object result, string propertyName) + => (T)result.GetType().GetProperty(propertyName)!.GetValue(result)!; + + static GrpcDurableTaskWorker CreateWorker(GrpcDurableTaskWorkerOptions grpcOptions) + { + return new GrpcDurableTaskWorker( + name: "Test", + factory: Mock.Of(), + grpcOptions: new OptionsMonitorStub(grpcOptions), + workerOptions: new OptionsMonitorStub(new DurableTaskWorkerOptions()), + services: Mock.Of(), + loggerFactory: NullLoggerFactory.Instance, + orchestrationFilter: null, + exceptionPropertiesProvider: null, + workItemFiltersMonitor: null); + } + + /// + /// Drives a single unary call through an invoker without touching the network: the innermost + /// participant always short-circuits. + /// + static class CallProbe + { + static readonly Marshaller Marshaller = Marshallers.Create( + m => Encoding.UTF8.GetBytes(m.Value), b => new ProbeMessage { Value = Encoding.UTF8.GetString(b) }); + + static readonly Method Method = new( + MethodType.Unary, "Probe", "Probe", Marshaller, Marshaller); + + public static void Invoke(CallInvoker invoker) + => invoker.AsyncUnaryCall(Method, null, default, new ProbeMessage()).Dispose(); + } + + sealed class ProbeMessage + { + public string Value { get; set; } = string.Empty; + } + + sealed class RecordingInterceptor : Interceptor + { + readonly string tag; + readonly IList log; + readonly bool passThrough; + + public RecordingInterceptor(string tag, IList log, bool passThrough) + { + this.tag = tag; + this.log = log; + this.passThrough = passThrough; + } + + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + { + this.log.Add(this.tag); + return this.passThrough ? continuation(request, context) : StubCallInvoker.EmptyCall(); + } + } + + sealed class StubCallInvoker : CallInvoker + { + int callCount; + + public int CallCount => Volatile.Read(ref this.callCount); + + public static AsyncUnaryCall EmptyCall() + where TResponse : class + { + return new AsyncUnaryCall( + Task.FromResult(Activator.CreateInstance()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + } + + public override AsyncUnaryCall AsyncUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + { + Interlocked.Increment(ref this.callCount); + return EmptyCall(); + } + + public override TResponse BlockingUnaryCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + + public override AsyncClientStreamingCall AsyncClientStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncDuplexStreamingCall AsyncDuplexStreamingCall( + Method method, string? host, CallOptions options) + => throw new NotSupportedException(); + + public override AsyncServerStreamingCall AsyncServerStreamingCall( + Method method, string? host, CallOptions options, TRequest request) + => throw new NotSupportedException(); + } +} diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs index 51bc66df..a6809387 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using Grpc.Core; using Grpc.Net.Client; using Microsoft.DurableTask.Worker.Grpc.Internal; @@ -28,56 +27,6 @@ public void InternalOptions_HasSafeDefaults() internalOptions.TransientRetryMaxAttempts.Should().Be(10); internalOptions.SilentDisconnectTimeout.Should().Be(TimeSpan.FromSeconds(120)); internalOptions.ChannelRecreator.Should().BeNull(); - internalOptions.CallInvokerDecorator.Should().BeNull(); - } - - [Fact] - public void SetCallInvokerDecorator_NullCallback_Throws() - { - // Arrange - GrpcDurableTaskWorkerOptions options = new(); - - // Act - Action act = () => options.SetCallInvokerDecorator(null!); - - // Assert - act.Should().Throw(); - } - - [Fact] - public void ApplyCallInvokerDecorator_NoDecorator_ReturnsOriginalInvoker() - { - // Arrange - GrpcDurableTaskWorkerOptions options = new(); - CallInvoker invoker = GrpcChannel.ForAddress("http://localhost:9101").CreateCallInvoker(); - - // Act - CallInvoker result = options.ApplyCallInvokerDecorator(invoker); - - // Assert - result.Should().BeSameAs(invoker); - } - - [Fact] - public void ApplyCallInvokerDecorator_WithDecorator_ReturnsDecoratedInvoker() - { - // Arrange - GrpcDurableTaskWorkerOptions options = new(); - CallInvoker invoker = GrpcChannel.ForAddress("http://localhost:9102").CreateCallInvoker(); - CallInvoker decorated = GrpcChannel.ForAddress("http://localhost:9103").CreateCallInvoker(); - CallInvoker? observed = null; - options.SetCallInvokerDecorator(inner => - { - observed = inner; - return decorated; - }); - - // Act - CallInvoker result = options.ApplyCallInvokerDecorator(invoker); - - // Assert - result.Should().BeSameAs(decorated); - observed.Should().BeSameAs(invoker); } [Fact] From 2e5123aac2a0ab1e181cd9c095006a0cdfc2843d Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 19 Aug 2026 13:20:35 -0400 Subject: [PATCH 3/4] Drop redundant using left over from the previous revision Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs index a6809387..87e70484 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using Grpc.Net.Client; using Microsoft.DurableTask.Worker.Grpc.Internal; namespace Microsoft.DurableTask.Worker.Grpc.Tests; From a295ed5f7b92c2fb458010569c212e537decc795 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 20 Aug 2026 16:33:05 -0400 Subject: [PATCH 4/4] Snapshot worker interceptors at construction The worker read `grpcOptions.Interceptors` live at three points: once in `GetCallInvoker` at startup, and again on each of the two channel-recreate paths in `TryRecreateChannelAsync`. `Intercept()` builds a frozen `InterceptingCallInvoker` graph that captures interceptor references rather than holding the list, so an `Interceptors.Add(...)` performed after the worker was constructed had no effect at the time it happened, then silently took effect at the next channel recreate. That is reachable, not theoretical: the worker resolves its options via `IOptionsMonitor.Get(name)`, and options instances are cached per name, so any other holder of the same monitor gets the same instance and can mutate the very list the worker re-reads. Because recreate is triggered by external events (backend replacement, node restart, consecutive failures), the delay between the mutation and its activation is nondeterministic. Capture the collection once into a `readonly Interceptor[]` in the constructor and use that snapshot at all three sites. This gives the worker the same read-once semantics the client already had (the client builds its invoker once in its constructor, and `ChannelRecreatingCallInvoker` swaps channels inside the interceptor wrapper), and it matches the existing "do not re-read `this.grpcOptions.Channel` inside the loop" invariant just below. It also removes a `List` thread-safety hazard, since recreate runs on the worker's background loop and could previously race a concurrent `Add()`. Document the read-once contract in the `` on `Interceptors` for both the worker and client options, and add three tests that pin the scenario: interceptors added after construction never take effect, on the recreator path, the worker-owned rebuild path, and the startup invoker. Also dispose the `GrpcChannel` instances that the externalized-payloads tests were leaking. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5e3106c1-d9ec-4666-a1ec-c8e5867f87f8 --- .../Grpc/GrpcDurableTaskClientOptions.cs | 5 + src/Worker/Grpc/GrpcDurableTaskWorker.cs | 17 +++- .../Grpc/GrpcDurableTaskWorkerOptions.cs | 5 + .../ExternalizedPayloadsInterceptorTests.cs | 7 +- .../GrpcDurableTaskWorkerInterceptorsTests.cs | 91 +++++++++++++++++++ 5 files changed, 117 insertions(+), 8 deletions(-) diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index 78a03b35..f684aeac 100644 --- a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs +++ b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs @@ -42,6 +42,11 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions /// outgoing call first and each response last. Registration is purely additive: while this collection /// is empty, the client uses exactly the invoker its configured transport produces. /// + /// + /// This collection is captured when the client is constructed, so it must be populated while options are + /// being configured (for example from Configure or PostConfigure). Mutating it afterwards + /// has no effect on a client that has already been built, including across channel recreation. + /// /// public IList Interceptors { get; } = new List(); diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index 0a4330e2..1ebd6294 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -17,6 +17,7 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker static readonly TimeSpan DeferredDisposeGracePeriod = TimeSpan.FromSeconds(30); readonly GrpcDurableTaskWorkerOptions grpcOptions; + readonly Interceptor[] interceptors; readonly DurableTaskWorkerOptions workerOptions; readonly IServiceProvider services; readonly ILoggerFactory loggerFactory; @@ -49,6 +50,12 @@ public GrpcDurableTaskWorker( : base(name, factory) { this.grpcOptions = Check.NotNull(grpcOptions).Get(name); + + // Snapshot the interceptors once so the chain is fixed for the worker's lifetime. Options instances + // are cached per name, so the configured collection stays reachable and mutable after construction; + // re-reading it when a channel is recreated would let a late mutation silently take effect at an + // externally-triggered recreate, long after the change was made. + this.interceptors = this.grpcOptions.Interceptors.ToArray(); this.workerOptions = Check.NotNull(workerOptions).Get(name); this.services = Check.NotNull(services); this.loggerFactory = Check.NotNull(loggerFactory); @@ -129,7 +136,7 @@ async Task TryRecreateChannelAsync( // carrying that ownership forward to the recreated state. return new ChannelRecreateResult( true, - ApplyInterceptors(this.grpcOptions.Interceptors, newChannel.CreateCallInvoker()), + ApplyInterceptors(this.interceptors, newChannel.CreateCallInvoker()), newChannel.Target, default, newChannel); @@ -162,7 +169,7 @@ async Task TryRecreateChannelAsync( AsyncDisposable newDisposable = CreateOwnedChannelDisposable(newChannel); return new ChannelRecreateResult( true, - ApplyInterceptors(this.grpcOptions.Interceptors, newChannel.CreateCallInvoker()), + ApplyInterceptors(this.interceptors, newChannel.CreateCallInvoker()), newChannel.Target, newDisposable, newChannel); @@ -306,7 +313,7 @@ and not AccessViolationException AsyncDisposable GetCallInvoker(out CallInvoker callInvoker, out string address) { AsyncDisposable disposable = this.GetCallInvokerCore(out CallInvoker core, out address); - callInvoker = ApplyInterceptors(this.grpcOptions.Interceptors, core); + callInvoker = ApplyInterceptors(this.interceptors, core); return disposable; } @@ -332,8 +339,8 @@ AsyncDisposable GetCallInvokerCore(out CallInvoker callInvoker, out string addre return CreateOwnedChannelDisposable(c); } - static CallInvoker ApplyInterceptors(IList interceptors, CallInvoker invoker) - => interceptors.Count == 0 ? invoker : invoker.Intercept(interceptors.ToArray()); + static CallInvoker ApplyInterceptors(Interceptor[] interceptors, CallInvoker invoker) + => interceptors.Length == 0 ? invoker : invoker.Intercept(interceptors); static ILogger CreateLogger(ILoggerFactory loggerFactory, DurableTaskWorkerOptions options) { diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index e656ce8b..930ee99d 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -56,6 +56,11 @@ public sealed class GrpcDurableTaskWorkerOptions : DurableTaskWorkerOptions /// outgoing call first and each response last. Registration is purely additive: while this collection /// is empty, the worker uses exactly the invoker its configured transport produces. /// + /// + /// This collection is captured when the worker is constructed, so it must be populated while options are + /// being configured (for example from Configure or PostConfigure). Mutating it afterwards + /// has no effect on a worker that has already been built, including across channel recreation. + /// /// public IList Interceptors { get; } = new List(); diff --git a/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs index e81598d0..83eaec39 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs @@ -41,7 +41,7 @@ public class ExternalizedPayloadsInterceptorTests public void Worker_WithChannel_PreservesChannelSoRecreationStaysEnabled() { // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); ServiceCollection services = new(); services.AddSingleton(new FakePayloadStore()); DefaultDurableTaskWorkerBuilder builder = new(null, services); @@ -60,7 +60,7 @@ public void Worker_WithChannel_PreservesChannelSoRecreationStaysEnabled() public void Client_WithChannel_PreservesChannelSoRecreationStaysEnabled() { // Arrange - GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); ServiceCollection services = new(); services.AddSingleton(new FakePayloadStore()); DefaultDurableTaskClientBuilder builder = new(null, services); @@ -113,7 +113,8 @@ public void Client_WithAddressOnly_DoesNotThrow() public void Worker_WithExternalCallInvoker_PreservesConfiguredInvoker() { // Arrange - CallInvoker external = GrpcChannel.ForAddress("http://localhost:4001").CreateCallInvoker(); + using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + CallInvoker external = channel.CreateCallInvoker(); ServiceCollection services = new(); services.AddSingleton(new FakePayloadStore()); DefaultDurableTaskWorkerBuilder builder = new(null, services); diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs index 73600663..83bffd0e 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs @@ -246,6 +246,97 @@ public async Task TryRecreateChannelAsync_NoInterceptors_ReturnsChannelInvokerUn } } + [Fact] + public async Task Interceptors_AddedAfterWorkerConstruction_NeverTakeEffect_EvenAfterChannelRecreate() + { + // Arrange: the interceptor chain is captured when the worker is constructed. Options instances are + // cached per name by IOptionsMonitor, so a caller holding the same monitor can mutate the very list + // the worker was configured from. Without a snapshot, the recreate path re-reads that list and the + // late addition silently activates at the next externally-triggered recreate. + // + // The late interceptor is inserted at index 0 so it becomes the outermost link. That keeps the probe + // network-free in both directions: whichever interceptor is outermost short-circuits the call. An + // Add() to the end is the same defect — it mutates the same list the recreate path re-reads — but it + // lands innermost, where the outer short-circuit would mask it and make the assertion vacuous. + GrpcChannel currentChannel = GrpcChannel.ForAddress("http://localhost:5109"); + GrpcChannel recreatedChannel = GrpcChannel.ForAddress("http://localhost:5110"); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Channel = currentChannel }; + grpcOptions.SetChannelRecreator((channel, ct) => Task.FromResult(recreatedChannel)); + grpcOptions.Interceptors.Add(new RecordingInterceptor("at-construction", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + try + { + // Act: mutate the live options collection after the worker was built, then force a recreate. + grpcOptions.Interceptors.Insert(0, new RecordingInterceptor("added-late", log, passThrough: false)); + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + CallProbe.Invoke(GetResultProperty(result, "NewCallInvoker")); + + // Assert: the rebuilt invoker still carries exactly the chain captured at construction. + GetResultProperty(result, "Recreated").Should().BeTrue(); + log.Should().Equal("at-construction"); + log.Should().NotContain("added-late"); + } + finally + { + currentChannel.Dispose(); + recreatedChannel.Dispose(); + } + } + + [Fact] + public async Task Interceptors_AddedAfterWorkerConstruction_NeverTakeEffect_OnWorkerOwnedRecreate() + { + // Arrange: same contract on the Address-only rebuild path, which re-reads the collection separately. + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { Address = "http://localhost:5111" }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("at-construction", log, passThrough: false)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + GrpcChannel currentChannel = GrpcChannel.ForAddress(grpcOptions.Address); + + try + { + // Act + grpcOptions.Interceptors.Insert(0, new RecordingInterceptor("added-late", log, passThrough: false)); + object result = await InvokeTryRecreateChannelAsync(worker, currentChannel); + + CallProbe.Invoke(GetResultProperty(result, "NewCallInvoker")); + + // Assert + GetResultProperty(result, "Recreated").Should().BeTrue(); + log.Should().Equal("at-construction"); + + AsyncDisposable newDisposable = GetResultProperty(result, "NewWorkerOwnedDisposable"); + await newDisposable.DisposeAsync(); + } + finally + { + currentChannel.Dispose(); + } + } + + [Fact] + public void Interceptors_AddedAfterWorkerConstruction_DoNotAffectStartupInvoker() + { + // Arrange + StubCallInvoker external = new(); + List log = new(); + GrpcDurableTaskWorkerOptions grpcOptions = new() { CallInvoker = external }; + grpcOptions.Interceptors.Add(new RecordingInterceptor("at-construction", log, passThrough: true)); + GrpcDurableTaskWorker worker = CreateWorker(grpcOptions); + + // Act + grpcOptions.Interceptors.Insert(0, new RecordingInterceptor("added-late", log, passThrough: true)); + InvokeGetCallInvoker(worker, out CallInvoker callInvoker, out _); + CallProbe.Invoke(callInvoker); + + // Assert + log.Should().Equal("at-construction"); + external.CallCount.Should().Be(1); + } + static void InvokeGetCallInvoker(GrpcDurableTaskWorker worker, out CallInvoker callInvoker, out string address) { object?[] args = { null, null };