Preserve gRPC channel recreation when externalized payloads are enabled

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<IList<HistoryEvent>> 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<Interceptor> interceptors, CallInvoker invoker)
+        => interceptors.Count == 0 ? invoker : invoker.Intercept(interceptors.ToArray());
+
+    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

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using Grpc.Core.Interceptors;
+
 namespace Microsoft.DurableTask.Client.Grpc;
 
 /// <summary>
@@ -23,6 +25,31 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions
     /// </summary>
     public CallInvoker? CallInvoker { get; set; }
 
+    /// <summary>
+    /// Gets the gRPC interceptors applied to every <see cref="CallInvoker"/> the client builds from its
+    /// configured transport, including invokers rebuilt after the underlying channel is recreated.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// 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 <see cref="CallInvoker"/> in place of <see cref="Channel"/>: an
+    /// externally-supplied invoker opts the client out of gRPC channel recreation, so a wedged connection
+    /// can never be replaced.
+    /// </para>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// <para>
+    /// This collection is captured when the client is constructed, so it must be populated while options are
+    /// being configured (for example from <c>Configure</c> or <c>PostConfigure</c>). Mutating it afterwards
+    /// has no effect on a client that has already been built, including across channel recreation.
+    /// </para>
+    /// </remarks>
+    public IList<Interceptor> Interceptors { get; } = new List<Interceptor>();
+
     /// <summary>
     /// Gets the internal options. These are not exposed directly, but configurable via
     /// <see cref="Internal.InternalOptionsExtensions"/>.

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<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 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;

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<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 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);
             });

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<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,
+                        ApplyInterceptors(this.interceptors, newChannel.CreateCallInvoker()),
+                        newChannel.Target,
+                        default,
+                        newChannel);
                 }
 
                 // Recreator returned the same instance — nothing to swap.
@@ -154,7 +167,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,
+                    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

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
     /// </summary>
     public CallInvoker? CallInvoker { get; set; }
 
+    /// <summary>
+    /// Gets the gRPC interceptors applied to every <see cref="CallInvoker"/> the worker builds from its
+    /// configured transport, including invokers rebuilt after the underlying channel is recreated.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// 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 <see cref="CallInvoker"/> in place of <see cref="Channel"/>: an
+    /// externally-supplied invoker opts the worker out of gRPC channel recreation, so a wedged connection
+    /// can never be replaced.
+    /// </para>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// <para>
+    /// This collection is captured when the worker is constructed, so it must be populated while options are
+    /// being configured (for example from <c>Configure</c> or <c>PostConfigure</c>). Mutating it afterwards
+    /// has no effect on a worker that has already been built, including across channel recreation.
+    /// </para>
+    /// </remarks>
+    public IList<Interceptor> Interceptors { get; } = new List<Interceptor>();
+
     /// <summary>
     /// Gets the collection of capabilities enabled on this worker.
     /// Capabilities are announced to the backend on connection.