diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 23350d4c..865a83b3 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -7,6 +7,7 @@ using DurableTask.Core.Exceptions; using DurableTask.Core.History; using Google.Protobuf.WellKnownTypes; +using Grpc.Core.Interceptors; using Microsoft.DurableTask.Client.Entities; using Microsoft.DurableTask.Tracing; using Microsoft.Extensions.DependencyInjection; @@ -625,6 +626,19 @@ public override async Task> GetOrchestrationHistoryAsync( } static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) + { + AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker core); + + // 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; int threshold = options.Internal.ChannelRecreateFailureThreshold; diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index 126aad34..f684aeac 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,31 @@ 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. + /// + /// + /// 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(); + /// /// Gets the internal options. These are not exposed directly, but configurable via /// . diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 0817bcea..d8607ec6 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -1,11 +1,9 @@ // 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.Converters; -using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; @@ -37,23 +35,11 @@ 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 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.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 b690d288..e1f8387d 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -1,8 +1,6 @@ // Copyright (c) Microsoft Corporation. // 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; @@ -61,23 +59,11 @@ 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 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.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 a200cedb..1ebd6294 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 Grpc.Core.Interceptors; using Microsoft.DurableTask.Worker.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -16,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; @@ -48,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); @@ -126,7 +134,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, + ApplyInterceptors(this.interceptors, newChannel.CreateCallInvoker()), + newChannel.Target, + default, + newChannel); } // Recreator returned the same instance — nothing to swap. @@ -154,7 +167,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, + ApplyInterceptors(this.interceptors, newChannel.CreateCallInvoker()), + newChannel.Target, + newDisposable, + newChannel); } catch (OperationCanceledException) when (cancellation.IsCancellationRequested) { @@ -293,6 +311,13 @@ and not AccessViolationException } AsyncDisposable GetCallInvoker(out CallInvoker callInvoker, out string address) + { + AsyncDisposable disposable = this.GetCallInvokerCore(out CallInvoker core, out address); + callInvoker = ApplyInterceptors(this.interceptors, core); + return disposable; + } + + AsyncDisposable GetCallInvokerCore(out CallInvoker callInvoker, out string address) { if (this.grpcOptions.Channel is GrpcChannel c) { @@ -314,6 +339,9 @@ AsyncDisposable GetCallInvoker(out CallInvoker callInvoker, out string address) return CreateOwnedChannelDisposable(c); } + static CallInvoker ApplyInterceptors(Interceptor[] interceptors, CallInvoker invoker) + => interceptors.Length == 0 ? invoker : invoker.Intercept(interceptors); + 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 59c21a00..930ee99d 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,31 @@ 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. + /// + /// + /// 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(); + /// /// Gets the collection of capabilities enabled on this worker. /// Capabilities are announced to the backend on connection. 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/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/ExternalizedPayloadsInterceptorTests.cs b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs new file mode 100644 index 00000000..83eaec39 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/ExternalizedPayloadsInterceptorTests.cs @@ -0,0 +1,324 @@ +// 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.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +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. It now registers an interceptor on the +/// supported Interceptors collection, which the worker and client apply to every invoker they +/// build. +/// +public class ExternalizedPayloadsInterceptorTests +{ + 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 + using 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); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); + } + + [Fact] + public void Client_WithChannel_PreservesChannelSoRecreationStaysEnabled() + { + // Arrange + using 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); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); + } + + [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 + using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:4001"); + CallInvoker external = channel.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 intercepts on use instead. + options.CallInvoker.Should().BeSameAs(external); + options.Interceptors.Should().ContainSingle().Which.Should().BeOfType(); + } + + [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_RegisteredInterceptor_ExternalizesLargePayloads() + { + // Arrange + RecordingCallInvoker inner = new(); + RecordingPayloadStore store = new(); + ServiceProvider provider = BuildWorkerProvider(inner, store); + + // 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); + inner.LastRequest!.Input.Should().Be(RecordingPayloadStore.Token); + } + + [Fact] + public async Task Client_RegisteredInterceptor_ExternalizesLargePayloads() + { + // Arrange + 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(opt => opt.CallInvoker = inner); + builder.UseExternalizedPayloads(); + GrpcDurableTaskClientOptions options = GetOptions(services); + + // 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); + 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 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 + { + 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/GrpcDurableTaskWorkerInterceptorsTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs new file mode 100644 index 00000000..83bffd0e --- /dev/null +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerInterceptorsTests.cs @@ -0,0 +1,457 @@ +// 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(); + } + } + + [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 }; + 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(); + } +}