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]