Support opt-in unary client deadlines

src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs

@@ -147,6 +147,7 @@ public void Configure(string? name, GrpcDurableTaskClientOptions options)
 
             string optionsName = name ?? Options.DefaultName;
             DurableTaskSchedulerClientOptions source = this.schedulerOptions.Get(optionsName);
+            options.GrpcCallTimeout = source.GrpcCallTimeout;
 
             // Create a cache key that includes all properties that affect CreateChannel behavior.
             // This ensures channels are reused for the same configuration

src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs

@@ -69,6 +69,18 @@ public string ResourceId
     /// </summary>
     public ClientRetryOptions? RetryOptions { get; set; }
 
+    /// <summary>
+    /// Gets or sets the default timeout for each unary gRPC call, excluding waits for orchestration start or completion.
+    /// The default is <see langword="null"/>, which adds no deadline. Must be positive when set.
+    /// </summary>
+    /// <remarks>
+    /// Earlier interceptor deadlines are preserved and streaming calls are not affected.
+    /// Expiration fails the call with <see cref="StatusCode.DeadlineExceeded"/>; it does not retry the operation
+    /// or guarantee that the server stopped processing it. This option is captured when the client is constructed
+    /// and bounds each RPC, not the whole client operation.
+    /// </remarks>
+    public TimeSpan? GrpcCallTimeout { get; set; }
+
     /// <summary>
     /// Creates a new instance of <see cref="DurableTaskSchedulerClientOptions"/> from a connection string.
     /// </summary>

src/Client/Grpc/ChannelRecreatingCallInvoker.cs

@@ -22,27 +22,14 @@ namespace Microsoft.DurableTask.Client.Grpc;
 /// </remarks>
 sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable
 {
-    /// <summary>
-    /// Methods on which a <see cref="StatusCode.DeadlineExceeded"/> response is expected behavior
-    /// (long-poll-style waits) and must NOT be counted toward the recreate threshold.
-    /// </summary>
-    static readonly HashSet<string> DeadlineExceededAllowedMethods = new(StringComparer.Ordinal)
-    {
-        "/TaskHubSidecarService/WaitForInstanceCompletion",
-        "/TaskHubSidecarService/WaitForInstanceStart",
-    };
-
     readonly Func<GrpcChannel, CancellationToken, Task<GrpcChannel>> recreator;
     readonly int failureThreshold;
     readonly TimeSpan minRecreateInterval;
     readonly bool ownsChannel;
     readonly ILogger logger;
 
-    // Cached once per invoker instance (instead of once per RPC) so ObserveOutcome's ContinueWith call
-    // does not allocate a new delegate for every unary call. The method-name diagnostic is threaded
-    // through as the ContinueWith `state` argument (a string, already a reference type) instead of a
-    // boxed (self, methodName) tuple, eliminating that per-call delegate and boxed-tuple allocation
-    // (Task.ContinueWith's own internal continuation/Task bookkeeping is unaffected).
+    // Reuse the continuation delegate. Calls without a caller token also reuse the method-name string
+    // as continuation state; cancellable calls carry the token so local cancellation stays neutral.
     readonly Action<Task, object?> onUnaryCallCompleted;
 
     // Cancelled in DisposeAsync so an in-flight RecreateAsync stops promptly and does not leak the
@@ -96,7 +83,7 @@ public override TResponse BlockingUnaryCall<TRequest, TResponse>(
         }
         catch (RpcException ex)
         {
-            this.RecordFailure(ex.StatusCode, method.FullName);
+            this.RecordFailure(ex.StatusCode, method.FullName, options.CancellationToken);
             throw;
         }
     }
@@ -106,7 +93,7 @@ public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
     {
         TransportState current = Volatile.Read(ref this.state);
         AsyncUnaryCall<TResponse> call = current.Invoker.AsyncUnaryCall(method, host, options, request);
-        this.ObserveOutcome(call.ResponseAsync, method.FullName);
+        this.ObserveOutcome(call.ResponseAsync, method.FullName, options.CancellationToken);
         return call;
     }
 
@@ -203,30 +190,29 @@ static TimeSpan ElapsedSince(long previousTimestamp, long nowTimestamp)
         return TimeSpan.FromSeconds((double)elapsedTicks / Stopwatch.Frequency);
     }
 
-    void ObserveOutcome(Task responseAsync, string methodFullName)
+    void ObserveOutcome(Task responseAsync, string methodFullName, CancellationToken cancellation)
     {
-        // Use ContinueWith with TaskScheduler.Default so we don't capture sync context. Both the
-        // continuation delegate (this.onUnaryCallCompleted, cached once per invoker instance) and the
-        // continuation state (methodFullName, already a reference-typed string) are allocation-free
-        // per call: no boxed tuple and no per-call delegate closure.
+        object continuationState = cancellation.CanBeCanceled ? (methodFullName, cancellation) : methodFullName;
         responseAsync.ContinueWith(
             this.onUnaryCallCompleted,
-            methodFullName,
+            continuationState,
             CancellationToken.None,
             TaskContinuationOptions.ExecuteSynchronously,
             TaskScheduler.Default);
     }
 
     void OnUnaryCallCompleted(Task task, object? continuationState)
     {
-        string methodFullName = (string)continuationState!;
         if (task.Status == TaskStatus.RanToCompletion)
         {
             this.RecordSuccess();
         }
         else if (task.Exception?.InnerException is RpcException rpcEx)
         {
-            this.RecordFailure(rpcEx.StatusCode, methodFullName);
+            (string methodFullName, CancellationToken cancellation) = continuationState is string method
+                ? (method, default)
+                : ((string, CancellationToken))continuationState!;
+            this.RecordFailure(rpcEx.StatusCode, methodFullName, cancellation);
         }
     }
 
@@ -235,8 +221,14 @@ void RecordSuccess()
         Volatile.Write(ref this.consecutiveFailures, 0);
     }
 
-    void RecordFailure(StatusCode status, string methodFullName)
+    void RecordFailure(StatusCode status, string methodFullName, CancellationToken cancellation)
     {
+        if (status == StatusCode.Cancelled && cancellation.IsCancellationRequested)
+        {
+            // gRPC does not identify cancellation's origin; an actually cancelled caller token is neutral.
+            return;
+        }
+
         // Only count statuses that indicate an actual transport problem, not application-level errors:
         //   * Unavailable     — half-open connection, peer reset, or dead routing target.
         //   * DeadlineExceeded — the call exceeded the *client-supplied* deadline. This is a
@@ -249,7 +241,7 @@ void RecordFailure(StatusCode status, string methodFullName)
         bool counts = status switch
         {
             StatusCode.Unavailable => true,
-            StatusCode.DeadlineExceeded => !DeadlineExceededAllowedMethods.Contains(methodFullName),
+            StatusCode.DeadlineExceeded => !ProtoUtils.IsLongPollingMethod(methodFullName),
             _ => false,
         };
 

src/Client/Grpc/GrpcCallTimeoutInterceptor.cs

@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Grpc.Core.Interceptors;
+
+namespace Microsoft.DurableTask.Client.Grpc;
+
+/// <summary>
+/// Applies the configured timeout without replacing an earlier deadline or timing out long polls and streams.
+/// </summary>
+sealed class GrpcCallTimeoutInterceptor : Interceptor
+{
+    readonly TimeSpan timeout;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="GrpcCallTimeoutInterceptor"/> class.
+    /// </summary>
+    /// <param name="timeout">The positive timeout applied to each unary call.</param>
+    public GrpcCallTimeoutInterceptor(TimeSpan timeout)
+    {
+        if (timeout <= TimeSpan.Zero)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(timeout), timeout, "GrpcCallTimeout must be positive.");
+        }
+
+        this.timeout = timeout;
+    }
+
+    /// <inheritdoc/>
+    public override TResponse BlockingUnaryCall<TRequest, TResponse>(
+        TRequest request,
+        ClientInterceptorContext<TRequest, TResponse> context,
+        BlockingUnaryCallContinuation<TRequest, TResponse> continuation)
+        => continuation(request, this.WithDeadline(context));
+
+    /// <inheritdoc/>
+    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
+        TRequest request,
+        ClientInterceptorContext<TRequest, TResponse> context,
+        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
+        => continuation(request, this.WithDeadline(context));
+
+    ClientInterceptorContext<TRequest, TResponse> WithDeadline<TRequest, TResponse>(
+        ClientInterceptorContext<TRequest, TResponse> context)
+        where TRequest : class
+        where TResponse : class
+    {
+        if (ProtoUtils.IsLongPollingMethod(context.Method.FullName))
+        {
+            return context;
+        }
+
+        DateTime now = DateTime.UtcNow;
+
+        // Avoid overflow and DateTime.MaxValue, which gRPC treats as no deadline.
+        long ticks = Math.Min(this.timeout.Ticks, DateTime.MaxValue.Ticks - now.Ticks - 1);
+        DateTime deadline = now.AddTicks(ticks);
+        if (context.Options.Deadline is DateTime existing && existing <= deadline)
+        {
+            return context;
+        }
+
+        return new(context.Method, context.Host, context.Options.WithDeadline(deadline));
+    }
+}

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -656,10 +656,18 @@ await this.largePayloadPurgeClient.SetLargePayloadAutoPurgeAsync(
 
     static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker)
     {
+        GrpcCallTimeoutInterceptor? timeoutInterceptor = options.GrpcCallTimeout is TimeSpan timeout
+            ? new(timeout)
+            : null;
         AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker core);
 
+        if (timeoutInterceptor is not null)
+        {
+            core = core.Intercept(timeoutInterceptor);
+        }
+
         // Intercept outside any ChannelRecreatingCallInvoker so the wrapper's internal channel swaps
-        // stay transparent to the configured interceptors.
+        // stay transparent to the configured interceptors. User interceptors can supply an earlier deadline.
         callInvoker = ApplyInterceptors(options.Interceptors, core);
         return disposable;
     }

src/Client/Grpc/GrpcDurableTaskClientOptions.cs

@@ -15,6 +15,20 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions
     /// </summary>
     public string? Address { get; set; }
 
+    /// <summary>
+    /// Gets or sets the default timeout for each unary gRPC call, excluding
+    /// <see cref="DurableTaskClient.WaitForInstanceStartAsync(string, bool, CancellationToken)"/> and
+    /// <see cref="DurableTaskClient.WaitForInstanceCompletionAsync(string, bool, CancellationToken)"/>.
+    /// The default is <see langword="null"/>, which adds no deadline.
+    /// </summary>
+    /// <remarks>
+    /// Must be positive when set. An earlier deadline supplied by an interceptor is preserved.
+    /// Streaming calls are not affected. Expiration fails the call with <see cref="StatusCode.DeadlineExceeded"/>;
+    /// it does not retry the operation or guarantee that the server stopped processing it.
+    /// This option is captured when the client is constructed and bounds each RPC, not the whole client operation.
+    /// </remarks>
+    public TimeSpan? GrpcCallTimeout { get; set; }
+
     /// <summary>
     /// Gets or sets the gRPC channel to use. Will supersede <see cref="CallInvoker" /> when provided.
     /// </summary>

src/Client/Grpc/ProtoUtils.cs

@@ -92,6 +92,16 @@ public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy(
         return dedupeStatuses;
     }
 
+    /// <summary>
+    /// Identifies long-poll RPCs excluded from the default call timeout. Their expected
+    /// <c>DeadlineExceeded</c> responses also do not count as transport failures.
+    /// </summary>
+    /// <param name="methodFullName">The fully qualified gRPC method name.</param>
+    /// <returns>Whether the method is a long poll.</returns>
+    internal static bool IsLongPollingMethod(string methodFullName)
+        => methodFullName is "/TaskHubSidecarService/WaitForInstanceCompletion"
+            or "/TaskHubSidecarService/WaitForInstanceStart";
+
 #pragma warning disable 0618 // Referencing Obsolete member. This is intention as we are only converting it.
     /// <summary>
     /// Converts <see cref="OrchestrationRuntimeStatus" /> to <see cref="P.OrchestrationStatus" />.

src/Client/Grpc/README.md

@@ -4,4 +4,23 @@ Commonly used types:
 - `GrpcDurableTaskClient`
 - `GrpcDurableTaskClientOptions`
 
+## Optional unary call deadlines
+
+Set `GrpcCallTimeout` to a positive value to bound each unary gRPC call, not the whole operation.
+The default is `null` (no deadline); an earlier explicit deadline is preserved.
+
+```csharp
+services.AddDurableTaskClient(builder => builder.UseGrpc(options =>
+{
+    options.Address = "http://localhost:4001";
+    options.GrpcCallTimeout = TimeSpan.FromSeconds(30);
+}));
+```
+
+The same property is available on `DurableTaskSchedulerClientOptions` when using `UseDurableTaskScheduler`.
+`WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls are excluded.
+
+Expiration fails the call with `StatusCode.DeadlineExceeded`. No retries are added, and a timed-out
+mutation may still have been applied by the server.
+
 For more information, see https://github.com/microsoft/durabletask-dotnet

test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs

@@ -294,6 +294,34 @@ public void UseDurableTaskScheduler_WithConnectionStringAndRetryOptions_ShouldCo
         }
     }
 
+    [Fact]
+    public async Task UseDurableTaskScheduler_UnaryTimeout_IsNamedAndDoesNotChangeChannelCache()
+    {
+        // Arrange
+        ServiceCollection services = new();
+        string connectionString = $"Endpoint=http://localhost;Authentication=None;TaskHub={ValidTaskHub}";
+        services.AddDurableTaskClient("bounded", builder => builder.UseDurableTaskScheduler(
+            connectionString, options => options.GrpcCallTimeout = TimeSpan.FromSeconds(30)));
+        services.AddDurableTaskClient("unbounded", builder => builder.UseDurableTaskScheduler(connectionString));
+        await using ServiceProvider provider = services.BuildServiceProvider();
+        IOptionsFactory<GrpcDurableTaskClientOptions> factory =
+            provider.GetRequiredService<IOptionsFactory<GrpcDurableTaskClientOptions>>();
+        IOptionsMonitor<DurableTaskSchedulerClientOptions> schedulerOptions =
+            provider.GetRequiredService<IOptionsMonitor<DurableTaskSchedulerClientOptions>>();
+
+        // Act
+        GrpcDurableTaskClientOptions bounded = factory.Create("bounded");
+        GrpcDurableTaskClientOptions unbounded = factory.Create("unbounded");
+        schedulerOptions.Get("bounded").GrpcCallTimeout = TimeSpan.FromSeconds(10);
+        GrpcDurableTaskClientOptions updated = factory.Create("bounded");
+
+        // Assert
+        bounded.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(30));
+        unbounded.GrpcCallTimeout.Should().BeNull();
+        updated.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(10));
+        updated.Channel.Should().BeSameAs(bounded.Channel);
+    }
+
     [Fact]
     public async Task UseDurableTaskScheduler_SameConfiguration_ReusesSameChannel()
     {
@@ -424,11 +452,11 @@ public async Task UseDurableTaskScheduler_DifferentResourceId_UsesSeparateChanne
         DefaultAzureCredential credential = new DefaultAzureCredential();
 
         // Act - configure two clients with the same endpoint/taskhub but different ResourceId
-        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.ResourceId = "https://durabletask.io";
         });
-        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.ResourceId = "https://custom.durabletask.io";
         });
@@ -487,11 +515,11 @@ public async Task UseDurableTaskScheduler_DifferentAllowInsecureCredentials_Uses
         DefaultAzureCredential credential = new DefaultAzureCredential();
 
         // Act - configure two clients with the same endpoint/taskhub but different AllowInsecureCredentials
-        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.AllowInsecureCredentials = false;
         });
-        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.AllowInsecureCredentials = true;
         });
@@ -522,14 +550,14 @@ public async Task UseDurableTaskScheduler_DifferentRetryOptions_UsesSeparateChan
         DefaultAzureCredential credential = new DefaultAzureCredential();
 
         // Act - configure two clients with the same endpoint/taskhub but different RetryOptions
-        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.RetryOptions = new DurableTaskSchedulerClientOptions.ClientRetryOptions
             {
                 MaxRetries = 3
             };
         });
-        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options => 
+        mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
         {
             options.RetryOptions = new DurableTaskSchedulerClientOptions.ClientRetryOptions
             {
@@ -548,4 +576,4 @@ public async Task UseDurableTaskScheduler_DifferentRetryOptions_UsesSeparateChan
         options2.Channel.Should().NotBeNull();
         options1.Channel.Should().NotBeSameAs(options2.Channel, "different RetryOptions should use different channels");
     }
-}
+}