Preserve gRPC channel recreation when externalized payloads are enabled

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<IList<HistoryEvent>> 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<GrpcChannel, CancellationToken, Task<GrpcChannel>>? recreator = options.Internal.ChannelRecreator;
         int threshold = options.Internal.ChannelRecreateFailureThreshold;

src/Client/Grpc/GrpcDurableTaskClientOptions.cs

@@ -57,5 +57,13 @@ internal class InternalOptions
         /// old channel so in-flight RPCs from peer clients are not interrupted.
         /// </summary>
         public Func<GrpcChannel, CancellationToken, Task<GrpcChannel>>? ChannelRecreator { get; set; }
+
+        /// <summary>
+        /// Gets or sets an optional decorator applied to every <see cref="CallInvoker"/> the client builds
+        /// from its configured transport. Extensions use this to attach interceptors without taking
+        /// ownership of <see cref="GrpcDurableTaskClientOptions.Channel"/>, which would otherwise disable
+        /// recreation.
+        /// </summary>
+        public Func<CallInvoker, CallInvoker>? CallInvokerDecorator { get; set; }
     }
 }

src/Client/Grpc/Internal/InternalOptionsExtensions.cs

@@ -30,4 +30,48 @@ public static void SetChannelRecreator(
     {
         options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator));
     }
+
+    /// <summary>
+    /// Sets a callback that decorates every <see cref="CallInvoker"/> the client builds from its configured
+    /// transport. Use this instead of replacing <see cref="GrpcDurableTaskClientOptions.Channel"/> with an
+    /// intercepted <see cref="GrpcDurableTaskClientOptions.CallInvoker"/>: clearing the channel leaves the
+    /// client with no way to recreate a wedged connection.
+    /// </summary>
+    /// <param name="options">The gRPC client options.</param>
+    /// <param name="decorator">The decorator callback.</param>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    public static void SetCallInvokerDecorator(
+        this GrpcDurableTaskClientOptions options,
+        Func<CallInvoker, CallInvoker> decorator)
+    {
+        options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator));
+    }
+
+    /// <summary>
+    /// Applies the decorator registered by <see cref="SetCallInvokerDecorator"/> to <paramref name="invoker"/>,
+    /// returning it unchanged when no decorator is registered. Callers that build a
+    /// <see cref="CallInvoker"/> from these options must route it through this method so registered
+    /// interceptors are not silently dropped.
+    /// </summary>
+    /// <param name="options">The gRPC client options.</param>
+    /// <param name="invoker">The invoker to decorate.</param>
+    /// <returns>The decorated invoker, or <paramref name="invoker"/> when no decorator is registered.</returns>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    public static CallInvoker ApplyCallInvokerDecorator(
+        this GrpcDurableTaskClientOptions options,
+        CallInvoker invoker)
+    {
+        Func<CallInvoker, CallInvoker>? decorator = options.Internal.CallInvokerDecorator;
+        return decorator is null ? invoker : decorator(invoker);
+    }
 }

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<PayloadStore, IOptionsMonitor<LargePayloadStorageOptions>>((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;

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<PayloadStore, IOptionsMonitor<LargePayloadStorageOptions>>((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);
             });

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<ChannelRecreateResult> 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<ChannelRecreateResult> 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)
         {

src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs

@@ -167,6 +167,14 @@ internal class InternalOptions
         /// </summary>
         public Func<GrpcChannel, CancellationToken, Task<GrpcChannel>>? ChannelRecreator { get; set; }
 
+        /// <summary>
+        /// Gets or sets an optional decorator applied to every <see cref="CallInvoker"/> the worker builds
+        /// from its configured transport, including invokers rebuilt after a channel recreate. Extensions
+        /// use this to attach interceptors without taking ownership of
+        /// <see cref="GrpcDurableTaskWorkerOptions.Channel"/>, which would otherwise disable recreation.
+        /// </summary>
+        public Func<CallInvoker, CallInvoker>? CallInvokerDecorator { get; set; }
+
         /// <summary>
         /// Gets or sets a callback that is invoked when activity work items are received or finished.
         /// </summary>

src/Worker/Grpc/Internal/InternalOptionsExtensions.cs

@@ -83,6 +83,51 @@ public static void SetChannelRecreator(
         options.Internal.ChannelRecreator = recreator ?? throw new ArgumentNullException(nameof(recreator));
     }
 
+    /// <summary>
+    /// Sets a callback that decorates every <see cref="CallInvoker"/> the worker builds from its configured
+    /// transport, including invokers rebuilt after a channel recreate. Use this instead of replacing
+    /// <see cref="GrpcDurableTaskWorkerOptions.Channel"/> with an intercepted
+    /// <see cref="GrpcDurableTaskWorkerOptions.CallInvoker"/>: clearing the channel leaves the worker with
+    /// no way to recreate a wedged connection.
+    /// </summary>
+    /// <param name="options">The gRPC worker options.</param>
+    /// <param name="decorator">The decorator callback.</param>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    public static void SetCallInvokerDecorator(
+        this GrpcDurableTaskWorkerOptions options,
+        Func<CallInvoker, CallInvoker> decorator)
+    {
+        options.Internal.CallInvokerDecorator = decorator ?? throw new ArgumentNullException(nameof(decorator));
+    }
+
+    /// <summary>
+    /// Applies the decorator registered by <see cref="SetCallInvokerDecorator"/> to <paramref name="invoker"/>,
+    /// returning it unchanged when no decorator is registered. Callers that build a
+    /// <see cref="CallInvoker"/> from these options must route it through this method so registered
+    /// interceptors are not silently dropped.
+    /// </summary>
+    /// <param name="options">The gRPC worker options.</param>
+    /// <param name="invoker">The invoker to decorate.</param>
+    /// <returns>The decorated invoker, or <paramref name="invoker"/> when no decorator is registered.</returns>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    public static CallInvoker ApplyCallInvokerDecorator(
+        this GrpcDurableTaskWorkerOptions options,
+        CallInvoker invoker)
+    {
+        Func<CallInvoker, CallInvoker>? decorator = options.Internal.CallInvokerDecorator;
+        return decorator is null ? invoker : decorator(invoker);
+    }
+
     /// <summary>
     /// Sets the deadline applied to the initial <c>Hello</c> RPC during worker connect. A wedged
     /// handshake on a half-open HTTP/2 connection no longer hangs the reconnect loop indefinitely.