From 5baad8a591b7d7cea9ece8de27658d7281356500 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 27 Oct 2026 09:54:04 -0700 Subject: [PATCH 1/3] Support opt-in unary client deadlines Apply optional per-call gRPC deadlines without changing default timeouts, long polls or streams. Preserve caller cancellation without erasing prior transport failures and verify shared-channel recovery over loopback HTTP/2. Fixes #824 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1e2ddb82-9306-4e21-8e6c-1c753d09f684 --- .../DurableTaskSchedulerClientExtensions.cs | 1 + .../DurableTaskSchedulerClientOptions.cs | 11 + .../Grpc/ChannelRecreatingCallInvoker.cs | 18 +- .../DefaultUnaryCallTimeoutInterceptor.cs | 75 ++++++ src/Client/Grpc/GrpcDurableTaskClient.cs | 10 +- .../Grpc/GrpcDurableTaskClientOptions.cs | 14 + src/Client/Grpc/README.md | 40 +++ ...rableTaskSchedulerClientExtensionsTests.cs | 42 ++- ...DurableTaskClientChannelRecreationTests.cs | 46 +++- .../GrpcDurableTaskClientInterceptorsTests.cs | 173 ++++++++++++- .../ClientDeadlineTests.cs | 241 ++++++++++++++++++ 11 files changed, 642 insertions(+), 29 deletions(-) create mode 100644 src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs create mode 100644 test/Grpc.IntegrationTests/ClientDeadlineTests.cs diff --git a/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs b/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs index a88de06ee..bdbdf6f03 100644 --- a/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs +++ b/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.DefaultUnaryCallTimeout = source.DefaultUnaryCallTimeout; // Create a cache key that includes all properties that affect CreateChannel behavior. // This ensures channels are reused for the same configuration diff --git a/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs b/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs index eac29155d..518bdf336 100644 --- a/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs +++ b/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs @@ -69,6 +69,17 @@ public string ResourceId /// public ClientRetryOptions? RetryOptions { get; set; } + /// + /// Gets or sets the default timeout for each unary gRPC call, excluding waits for orchestration start or completion. + /// The default is , which adds no deadline. Must be positive when set. + /// + /// + /// Earlier interceptor deadlines are preserved and streaming calls are not affected. + /// Expiration fails the call with ; it does not retry the operation + /// or guarantee that the server stopped processing it. This option is captured when the client is constructed. + /// + public TimeSpan? DefaultUnaryCallTimeout { get; set; } + /// /// Creates a new instance of from a connection string. /// diff --git a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs index f0928876b..cbe4a662c 100644 --- a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs +++ b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs @@ -22,16 +22,6 @@ namespace Microsoft.DurableTask.Client.Grpc; /// sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable { - /// - /// Methods on which a response is expected behavior - /// (long-poll-style waits) and must NOT be counted toward the recreate threshold. - /// - static readonly HashSet DeadlineExceededAllowedMethods = new(StringComparer.Ordinal) - { - "/TaskHubSidecarService/WaitForInstanceCompletion", - "/TaskHubSidecarService/WaitForInstanceStart", - }; - readonly Func> recreator; readonly int failureThreshold; readonly TimeSpan minRecreateInterval; @@ -237,6 +227,12 @@ void RecordSuccess() void RecordFailure(StatusCode status, string methodFullName) { + if (status == StatusCode.Cancelled) + { + // Local cancellation is not evidence of either a transport failure or a successful round-trip. + 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 +245,7 @@ void RecordFailure(StatusCode status, string methodFullName) bool counts = status switch { StatusCode.Unavailable => true, - StatusCode.DeadlineExceeded => !DeadlineExceededAllowedMethods.Contains(methodFullName), + StatusCode.DeadlineExceeded => !DefaultUnaryCallTimeoutInterceptor.IsLongPollingMethod(methodFullName), _ => false, }; diff --git a/src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs b/src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs new file mode 100644 index 000000000..6f7c9fe33 --- /dev/null +++ b/src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs @@ -0,0 +1,75 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core.Interceptors; + +namespace Microsoft.DurableTask.Client.Grpc; + +/// +/// Applies the configured timeout without replacing an earlier deadline or timing out long polls and streams. +/// +sealed class DefaultUnaryCallTimeoutInterceptor : Interceptor +{ + readonly TimeSpan timeout; + + /// + /// Initializes a new instance of the class. + /// + /// The positive timeout applied to each unary call. + public DefaultUnaryCallTimeoutInterceptor(TimeSpan timeout) + { + if (timeout <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(timeout), timeout, "DefaultUnaryCallTimeout must be positive."); + } + + this.timeout = timeout; + } + + /// + public override TResponse BlockingUnaryCall( + TRequest request, + ClientInterceptorContext context, + BlockingUnaryCallContinuation continuation) + => continuation(request, this.WithDeadline(context)); + + /// + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + => continuation(request, this.WithDeadline(context)); + + /// + /// Identifies calls whose normal operation can wait indefinitely for an orchestration state change. + /// + /// The fully qualified gRPC method name. + /// Whether the call is a long poll. + internal static bool IsLongPollingMethod(string methodFullName) + => methodFullName is "/TaskHubSidecarService/WaitForInstanceCompletion" + or "/TaskHubSidecarService/WaitForInstanceStart"; + + ClientInterceptorContext WithDeadline( + ClientInterceptorContext context) + where TRequest : class + where TResponse : class + { + if (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)); + } +} diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index f240d722d..19e2ea79d 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -656,10 +656,18 @@ await this.largePayloadPurgeClient.SetLargePayloadAutoPurgeAsync( static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { + DefaultUnaryCallTimeoutInterceptor? timeoutInterceptor = options.DefaultUnaryCallTimeout 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; } diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index f684aeac6..bdaf6bd94 100644 --- a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs +++ b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs @@ -15,6 +15,20 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions /// public string? Address { get; set; } + /// + /// Gets or sets the default timeout for each unary gRPC call, excluding + /// and + /// . + /// The default is , which adds no deadline. + /// + /// + /// Must be positive when set. An earlier deadline supplied by an interceptor is preserved. + /// Streaming calls are not affected. Expiration fails the call with ; + /// it does not retry the operation or guarantee that the server stopped processing it. + /// This option is captured when the client is constructed. + /// + public TimeSpan? DefaultUnaryCallTimeout { get; set; } + /// /// Gets or sets the gRPC channel to use. Will supersede when provided. /// diff --git a/src/Client/Grpc/README.md b/src/Client/Grpc/README.md index f7e830a0a..70f73f354 100644 --- a/src/Client/Grpc/README.md +++ b/src/Client/Grpc/README.md @@ -4,4 +4,44 @@ Commonly used types: - `GrpcDurableTaskClient` - `GrpcDurableTaskClientOptions` +## Optional unary call deadlines + +By default, the client adds no gRPC deadline. To bound individual unary calls, configure a positive +`DefaultUnaryCallTimeout` when registering the client: + +```csharp +services.AddDurableTaskClient(builder => builder.UseGrpc(options => +{ + options.Address = "http://localhost:4001"; + options.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30); +})); +``` + +For Azure Durable Task Scheduler, configure the same option on the scheduler options: + +```csharp +services.AddDurableTaskClient(builder => builder.UseDurableTaskScheduler( + connectionString, + options => options.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30))); +``` + +The timeout is captured when the client is constructed and bounds each unary RPC, including any +configured gRPC retries. An earlier deadline set through `GrpcDurableTaskClientOptions.Interceptors` +is preserved. Extremely large timeouts are capped at the latest finite UTC deadline supported by +`DateTime`. Caller cancellation remains independent and is not counted as a transport failure or +treated as proof of a healthy connection. + +`WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls such as +`GetOrchestrationHistoryAsync` are excluded; use their cancellation tokens to bound those operations. +The option does not change worker streams or configure HTTP/2 keepalive. + +A stalled unary call expires with `StatusCode.DeadlineExceeded`, which can contribute to the existing +channel recreation threshold. Scheduler-managed clients retain their cached-channel recovery and +interceptors. Supplying an external `CallInvoker` still opts out of channel recreation. + +**A deadline does not guarantee that the server stopped processing the request.** This applies to +mutations such as scheduling, signaling, and purging as well as reads. This option adds no automatic +retry; determine the outcome before retrying a mutation. Channel recreation benefits subsequent +calls, not the failed call, and does not resolve an underlying server outage. + For more information, see https://github.com/microsoft/durabletask-dotnet \ No newline at end of file diff --git a/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs b/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs index 7f53b4d88..0efe32bf3 100644 --- a/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs +++ b/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.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30))); + services.AddDurableTaskClient("unbounded", builder => builder.UseDurableTaskScheduler(connectionString)); + await using ServiceProvider provider = services.BuildServiceProvider(); + IOptionsFactory factory = + provider.GetRequiredService>(); + IOptionsMonitor schedulerOptions = + provider.GetRequiredService>(); + + // Act + GrpcDurableTaskClientOptions bounded = factory.Create("bounded"); + GrpcDurableTaskClientOptions unbounded = factory.Create("unbounded"); + schedulerOptions.Get("bounded").DefaultUnaryCallTimeout = TimeSpan.FromSeconds(10); + GrpcDurableTaskClientOptions updated = factory.Create("bounded"); + + // Assert + bounded.DefaultUnaryCallTimeout.Should().Be(TimeSpan.FromSeconds(30)); + unbounded.DefaultUnaryCallTimeout.Should().BeNull(); + updated.DefaultUnaryCallTimeout.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"); } -} +} \ No newline at end of file diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs index 6945ead53..85d37a201 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs @@ -152,6 +152,40 @@ public async Task AsyncUnaryCall_SuccessAfterFailure_ResetsConsecutiveFailureCou } } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task UnaryCall_CancelledBetweenTransportFailures_DoesNotResetFailureCount(bool blocking) + { + // Arrange + int callIndex = 0; + CallbackHttpMessageHandler handler = new((request, cancellationToken) => + Task.FromResult(CreateFailureResponse( + Interlocked.Increment(ref callIndex) == 2 ? StatusCode.Cancelled : StatusCode.Unavailable, + "interrupted call"))); + using GrpcChannel channel = CreateChannel("http://cancelled.client.test", handler); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Internal.ChannelRecreateFailureThreshold = 2; + options.Internal.MinRecreateInterval = TimeSpan.Zero; + TaskCompletionSource recreateRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); + options.SetChannelRecreator((existingChannel, ct) => + { + recreateRequested.TrySetResult(); + return Task.FromResult(existingChannel); + }); + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + + // Act + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Cancelled, blocking); + + // Assert + GetConsecutiveFailures(callInvoker).Should().Be(1, "caller cancellation is not a successful transport round-trip"); + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); + await recreateRequested.Task.WaitAsync(TimeSpan.FromSeconds(5)); + } + [Fact] public async Task AsyncUnaryCall_DeadlineExceededOnAllowedLongPollMethod_DoesNotCountTowardThreshold() { @@ -447,10 +481,18 @@ static long InvokeToStopwatchTicks(TimeSpan interval) static async Task AssertRpcFailureAsync(CallInvoker callInvoker) => await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable); - static async Task AssertRpcFailureAsync(CallInvoker callInvoker, Method method, StatusCode expectedStatus) + static async Task AssertRpcFailureAsync( + CallInvoker callInvoker, Method method, StatusCode expectedStatus, bool blocking = false) { Func act = async () => { + if (blocking) + { + callInvoker.BlockingUnaryCall( + method, host: null, new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1)), request: "ping"); + return; + } + using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( method, host: null, @@ -544,4 +586,4 @@ protected override Task SendAsync(HttpRequestMessage reques return this.callback(request, cancellationToken); } } -} +} \ No newline at end of file diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs index 641f7d74b..8501ba080 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs @@ -190,6 +190,123 @@ public async Task GetCallInvoker_MultipleInterceptors_RunsInListOrder() await disposable.DisposeAsync(); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task GetCallInvoker_WithTimeout_BoundsEachUnaryCallAndPreservesCallOptions(bool blocking) + { + // Arrange + StubCallInvoker external = new(); + using CancellationTokenSource cancellation = new(); + Metadata headers = new() { { "test-header", "preserved" } }; + TimeSpan timeout = TimeSpan.FromSeconds(30); + GrpcDurableTaskClientOptions options = new() { CallInvoker = external, DefaultUnaryCallTimeout = timeout }; + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + + // Act + CallProbe.Invoke(callInvoker, blocking: blocking); + DateTime before = DateTime.UtcNow; + CallProbe.Invoke(callInvoker, new CallOptions(headers, deadline: DateTime.UtcNow.AddMinutes(1), + cancellationToken: cancellation.Token), blocking: blocking); + DateTime after = DateTime.UtcNow; + + // Assert + external.LastOptions.Deadline.Should().BeOnOrAfter(before + timeout).And.BeOnOrBefore(after + timeout); + external.LastOptions.Deadline!.Value.Kind.Should().Be(DateTimeKind.Utc); + external.LastOptions.CancellationToken.Should().Be(cancellation.Token); + external.LastOptions.Headers.Should().BeSameAs(headers); + } + + [Fact] + public async Task GetCallInvoker_WithTimeout_PreservesEarlierUserInterceptorDeadline() + { + // Arrange + StubCallInvoker external = new(); + DateTime earlier = DateTime.UtcNow.AddSeconds(1); + GrpcDurableTaskClientOptions options = new() + { + CallInvoker = external, + DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30), + }; + options.Interceptors.Add(new EarlierDeadlineInterceptor(earlier)); + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + + // Act + CallProbe.Invoke(callInvoker); + + // Assert + external.LastOptions.Deadline.Should().Be(earlier); + } + + [Theory] + [InlineData("WaitForInstanceStart")] + [InlineData("WaitForInstanceCompletion")] + [InlineData("StreamInstanceHistory")] + public async Task GetCallInvoker_WithTimeout_DoesNotChangeWaitOrStreamingDeadlines(string methodName) + { + // Arrange + StubCallInvoker external = new(); + GrpcDurableTaskClientOptions options = new() + { + CallInvoker = external, + DefaultUnaryCallTimeout = TimeSpan.FromSeconds(1), + }; + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + DateTime explicitDeadline = DateTime.UtcNow.AddMinutes(1); + + // Act + CallProbe.Invoke(callInvoker, methodName: methodName); + DateTime? defaultDeadline = external.LastOptions.Deadline; + CallProbe.Invoke(callInvoker, new CallOptions(deadline: explicitDeadline), methodName: methodName); + + // Assert + defaultDeadline.Should().BeNull(); + external.LastOptions.Deadline.Should().Be(explicitDeadline); + } + + [Theory] + [InlineData(0L)] + [InlineData(-1L)] + [InlineData(long.MinValue)] + public void Constructor_WithNonPositiveTimeout_Throws(long ticks) + { + // Arrange + GrpcDurableTaskClientOptions options = new() + { + CallInvoker = new StubCallInvoker(), + DefaultUnaryCallTimeout = TimeSpan.FromTicks(ticks), + }; + + // Act + Action construct = () => new GrpcDurableTaskClient("test", options, NullLogger.Instance); + + // Assert + construct.Should().Throw().WithMessage("*DefaultUnaryCallTimeout must be positive*"); + } + + [Fact] + public async Task GetCallInvoker_WithMaximumTimeout_UsesFiniteDeadlineWithoutOverflow() + { + // Arrange + StubCallInvoker external = new(); + GrpcDurableTaskClientOptions options = new() + { + CallInvoker = external, + DefaultUnaryCallTimeout = TimeSpan.MaxValue, + }; + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + + // Act + CallProbe.Invoke(callInvoker); + + // Assert + external.LastOptions.Deadline.Should().Be(new DateTime(DateTime.MaxValue.Ticks - 1, DateTimeKind.Utc)); + } + static (AsyncDisposable Disposable, CallInvoker CallInvoker) InvokeGetCallInvoker( GrpcDurableTaskClientOptions options) => Invoke(GetCallInvokerMethod, options); @@ -215,11 +332,25 @@ 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(); + public static void Invoke( + CallInvoker invoker, CallOptions options = default, bool blocking = false, string methodName = "Probe") + { + Method method = new( + methodName == "StreamInstanceHistory" ? MethodType.ServerStreaming : MethodType.Unary, + "TaskHubSidecarService", methodName, Marshaller, Marshaller); + if (method.Type == MethodType.ServerStreaming) + { + invoker.AsyncServerStreamingCall(method, null, options, new ProbeMessage()).Dispose(); + } + else if (blocking) + { + invoker.BlockingUnaryCall(method, null, options, new ProbeMessage()); + } + else + { + invoker.AsyncUnaryCall(method, null, options, new ProbeMessage()).Dispose(); + } + } } sealed class ProbeMessage @@ -227,6 +358,15 @@ sealed class ProbeMessage public string Value { get; set; } = string.Empty; } + sealed class EarlierDeadlineInterceptor(DateTime deadline) : Interceptor + { + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + => continuation(request, new(context.Method, context.Host, context.Options.WithDeadline(deadline))); + } + sealed class RecordingInterceptor : Interceptor { readonly string tag; @@ -256,6 +396,8 @@ sealed class StubCallInvoker : CallInvoker public int CallCount => Volatile.Read(ref this.callCount); + public CallOptions LastOptions { get; private set; } + public static AsyncUnaryCall EmptyCall() where TResponse : class { @@ -271,12 +413,16 @@ public override AsyncUnaryCall AsyncUnaryCall( Method method, string? host, CallOptions options, TRequest request) { Interlocked.Increment(ref this.callCount); + this.LastOptions = options; return EmptyCall(); } public override TResponse BlockingUnaryCall( Method method, string? host, CallOptions options, TRequest request) - => throw new NotSupportedException(); + { + this.LastOptions = options; + return Activator.CreateInstance(); + } public override AsyncClientStreamingCall AsyncClientStreamingCall( Method method, string? host, CallOptions options) @@ -288,6 +434,17 @@ public override AsyncDuplexStreamingCall AsyncDuplexStreami public override AsyncServerStreamingCall AsyncServerStreamingCall( Method method, string? host, CallOptions options, TRequest request) - => throw new NotSupportedException(); + { + this.LastOptions = options; + return new(new EmptyStreamReader(), Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, () => new Metadata(), () => { }); + } + } + + sealed class EmptyStreamReader : IAsyncStreamReader + { + public T Current => throw new InvalidOperationException("The stream is empty."); + + public Task MoveNext(CancellationToken cancellationToken) => Task.FromResult(false); } -} +} \ No newline at end of file diff --git a/test/Grpc.IntegrationTests/ClientDeadlineTests.cs b/test/Grpc.IntegrationTests/ClientDeadlineTests.cs new file mode 100644 index 000000000..a12c6dbbd --- /dev/null +++ b/test/Grpc.IntegrationTests/ClientDeadlineTests.cs @@ -0,0 +1,241 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Buffers.Binary; +using System.Diagnostics; +using System.Net; +using System.Threading.Channels; +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Client.Grpc; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Moq; +using Xunit.Abstractions; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Grpc.Tests; + +public class ClientDeadlineTests(ITestOutputHelper output) +{ + static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10); + static readonly TimeSpan CallTimeout = TimeSpan.FromMilliseconds(500); + + [Fact] + public async Task StalledReads_WithInterleavedCancellation_RecreateOnceAndRecoverSharedChannelAsync() + { + // Arrange + await using StalledSidecar sidecar = await StalledSidecar.StartAsync(); + ServiceCollection services = new(); + services.AddLogging(); + Mock payloadStore = new(MockBehavior.Strict); + payloadStore.Setup(store => store.IsKnownPayloadToken(It.IsAny())) + .Returns((string value) => value == "test-payload"); + payloadStore.Setup(store => store.DownloadAsync("test-payload", It.IsAny())) + .ReturnsAsync("\"restored\""); + services.AddSingleton(payloadStore.Object); + services.AddDurableTaskClient("deadline", builder => + { + builder.UseDurableTaskScheduler( + $"Endpoint={sidecar.Address};TaskHub=test;Authentication=None", + options => options.DefaultUnaryCallTimeout = CallTimeout); + builder.UseExternalizedPayloads(); + }); + await using ServiceProvider provider = services.BuildServiceProvider(); + IOptionsFactory factory = + provider.GetRequiredService>(); + GrpcDurableTaskClientOptions firstOptions = factory.Create("deadline"); + GrpcDurableTaskClientOptions peerOptions = factory.Create("deadline"); + using GrpcChannel initialChannel = firstOptions.Channel!; + Assert.Same(initialChannel, peerOptions.Channel); + RecreationLogger firstLogger = new(); + RecreationLogger peerLogger = new(); + await using DurableTaskClient first = new GrpcDurableTaskClient("deadline", firstOptions, firstLogger); + await using DurableTaskClient peer = new GrpcDurableTaskClient("deadline", peerOptions, peerLogger); + + // Act + for (int i = 0; i < 4; i++) + { + await this.AssertDeadlineAsync(first, sidecar); + } + + using (CancellationTokenSource cancellation = new()) + { + Task cancelled = first.GetInstanceAsync("cancel", cancellation.Token); + await sidecar.NextCallAsync(); + cancellation.Cancel(); + RpcException error = await Assert.ThrowsAsync(() => cancelled.WaitAsync(Timeout)); + Assert.Equal(StatusCode.Cancelled, error.StatusCode); + } + + await Task.WhenAll(Enumerable.Range(0, 4).Select(_ => this.AssertDeadlineAsync(first, sidecar))); + await firstLogger.Recreated.Task.WaitAsync(Timeout); + GrpcDurableTaskClientOptions refreshedOptions = factory.Create("deadline"); + OrchestrationMetadata? recovered = await first.GetInstanceAsync("recovered", getInputsAndOutputs: true).WaitAsync(Timeout); + ObservedCall recoveredCall = await sidecar.NextCallAsync(); + + for (int i = 0; i < 5; i++) + { + await this.AssertDeadlineAsync(peer, sidecar); + } + + await peerLogger.Recreated.Task.WaitAsync(Timeout); + OrchestrationMetadata? peerRecovered = await peer.GetInstanceAsync("recovered", getInputsAndOutputs: true).WaitAsync(Timeout); + ObservedCall peerCall = await sidecar.NextCallAsync(); + + // Assert + Assert.Equal("healthy", recovered?.Name); + Assert.Equal("healthy", peerRecovered?.Name); + Assert.Equal("restored", recovered!.ReadOutputAs()); + Assert.Equal("restored", peerRecovered!.ReadOutputAs()); + payloadStore.Verify(store => store.DownloadAsync("test-payload", It.IsAny()), Times.Exactly(2)); + Assert.NotSame(initialChannel, refreshedOptions.Channel); + Assert.Same(refreshedOptions.Channel, factory.Create("deadline").Channel); + Assert.Equal(1, firstLogger.RecreateCount); + Assert.Equal(1, peerLogger.RecreateCount); + Assert.NotEqual(sidecar.StalledConnection, recoveredCall.Connection); + Assert.Equal(recoveredCall.Connection, peerCall.Connection); + Assert.All(sidecar.Calls, call => + { + Assert.Equal("test", call.TaskHub); + Assert.False(string.IsNullOrEmpty(call.Deadline)); + }); + output.WriteLine($"Recreations: first={firstLogger.RecreateCount}, peer={peerLogger.RecreateCount}; " + + $"connections: stalled={sidecar.StalledConnection}, recovered={recoveredCall.Connection}, peer={peerCall.Connection}"); + } + + async Task AssertDeadlineAsync(DurableTaskClient client, StalledSidecar sidecar) + { + using CancellationTokenSource guard = new(Timeout); + Stopwatch elapsed = Stopwatch.StartNew(); + Task read = client.GetInstanceAsync("stalled", guard.Token); + ObservedCall call = await sidecar.NextCallAsync(); + RpcException error = await Assert.ThrowsAsync(() => read.WaitAsync(Timeout)); + Assert.Equal(StatusCode.DeadlineExceeded, error.StatusCode); + Assert.Equal(sidecar.StalledConnection, call.Connection); + output.WriteLine($"Stalled GetInstance: {error.StatusCode}, {elapsed.Elapsed.TotalMilliseconds:F0} ms, grpc-timeout={call.Deadline}"); + } + + sealed class RecreationLogger : ILogger + { + int recreateCount; + + public int RecreateCount => Volatile.Read(ref this.recreateCount); + + public TaskCompletionSource Recreated { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + if (eventId.Id == 80) + { + Interlocked.Increment(ref this.recreateCount); + } + else if (eventId.Id == 81) + { + this.Recreated.TrySetResult(); + } + } + } + + sealed record ObservedCall(string Connection, string TaskHub, string Deadline); + + sealed class StalledSidecar : IAsyncDisposable + { + readonly Channel calls = Channel.CreateUnbounded(); + readonly System.Collections.Concurrent.ConcurrentQueue observedCalls = new(); + IHost host = null!; + string? stalledConnection; + + public string Address { get; private set; } = string.Empty; + + public string? StalledConnection => this.stalledConnection; + + public ObservedCall[] Calls => this.observedCalls.ToArray(); + + public static async Task StartAsync() + { + StalledSidecar sidecar = new(); + sidecar.host = Host.CreateDefaultBuilder() + .ConfigureLogging(logging => logging.ClearProviders()) + .ConfigureWebHostDefaults(web => web + .UseKestrel(options => options.Listen(IPAddress.Loopback, 0, endpoint => endpoint.Protocols = HttpProtocols.Http2)) + .Configure(app => app.Run(sidecar.HandleAsync))) + .Build(); + await sidecar.host.StartAsync().WaitAsync(Timeout); + sidecar.Address = Assert.Single(sidecar.host.Services.GetRequiredService() + .Features.Get()!.Addresses); + return sidecar; + } + + public Task NextCallAsync() => this.calls.Reader.ReadAsync().AsTask().WaitAsync(Timeout); + + public async ValueTask DisposeAsync() + { + using IHost host = this.host; + await host.StopAsync().WaitAsync(Timeout); + } + + async Task HandleAsync(HttpContext context) + { + Assert.Equal("/TaskHubSidecarService/GetInstance", context.Request.Path.Value); + await context.Request.Body.CopyToAsync(Stream.Null, context.RequestAborted); + ObservedCall call = new(context.Connection.Id, context.Request.Headers["taskhub"].ToString(), + context.Request.Headers["grpc-timeout"].ToString()); + this.observedCalls.Enqueue(call); + Interlocked.CompareExchange(ref this.stalledConnection, call.Connection, null); + await this.calls.Writer.WriteAsync(call, context.RequestAborted); + if (call.Connection == this.stalledConnection) + { + try + { + await Task.Delay(System.Threading.Timeout.Infinite, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + // This connection accepts requests but sends no gRPC response. + } + + return; + } + + P.GetInstanceResponse response = new() + { + Exists = true, + OrchestrationState = new() + { + Name = "healthy", + InstanceId = "recovered", + Output = "test-payload", + CreatedTimestamp = Timestamp.FromDateTime(DateTime.UnixEpoch), + LastUpdatedTimestamp = Timestamp.FromDateTime(DateTime.UnixEpoch), + }, + }; + byte[] message = response.ToByteArray(); + byte[] frame = new byte[5 + message.Length]; + BinaryPrimitives.WriteInt32BigEndian(frame.AsSpan(1), message.Length); + message.CopyTo(frame, 5); + context.Response.ContentType = "application/grpc"; + context.Response.DeclareTrailer("grpc-status"); + await context.Response.Body.WriteAsync(frame, context.RequestAborted); + context.Response.AppendTrailer("grpc-status", "0"); + } + } +} \ No newline at end of file From 92b4c5be2d6be111ad3b537b0662b12f6c861273 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 27 Oct 2026 11:32:38 -0700 Subject: [PATCH 2/3] Refine client timeout cancellation and validation Rename the unreleased option to GrpcCallTimeout, distinguish cancelled caller tokens from server Cancelled responses, and share long-poll classification independently of timeout interception. Join and dispose failed test-host startup and assert the actual 500 ms RPC budget. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1e2ddb82-9306-4e21-8e6c-1c753d09f684 --- .../DurableTaskSchedulerClientExtensions.cs | 2 +- .../DurableTaskSchedulerClientOptions.cs | 5 +- .../Grpc/ChannelRecreatingCallInvoker.cs | 34 ++++---- ...eptor.cs => GrpcCallTimeoutInterceptor.cs} | 19 ++--- src/Client/Grpc/GrpcDurableTaskClient.cs | 2 +- .../Grpc/GrpcDurableTaskClientOptions.cs | 4 +- src/Client/Grpc/ProtoUtils.cs | 9 ++ src/Client/Grpc/README.md | 15 ++-- ...rableTaskSchedulerClientExtensionsTests.cs | 10 +-- ...DurableTaskClientChannelRecreationTests.cs | 76 +++++++++++++++-- .../GrpcDurableTaskClientInterceptorsTests.cs | 12 +-- .../ClientDeadlineTests.cs | 85 +++++++++++++++++-- 12 files changed, 205 insertions(+), 68 deletions(-) rename src/Client/Grpc/{DefaultUnaryCallTimeoutInterceptor.cs => GrpcCallTimeoutInterceptor.cs} (69%) diff --git a/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs b/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs index bdbdf6f03..bbb3a63db 100644 --- a/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs +++ b/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs @@ -147,7 +147,7 @@ public void Configure(string? name, GrpcDurableTaskClientOptions options) string optionsName = name ?? Options.DefaultName; DurableTaskSchedulerClientOptions source = this.schedulerOptions.Get(optionsName); - options.DefaultUnaryCallTimeout = source.DefaultUnaryCallTimeout; + 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 diff --git a/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs b/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs index 518bdf336..623bdc672 100644 --- a/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs +++ b/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs @@ -76,9 +76,10 @@ public string ResourceId /// /// Earlier interceptor deadlines are preserved and streaming calls are not affected. /// Expiration fails the call with ; it does not retry the operation - /// or guarantee that the server stopped processing it. This option is captured when the client is constructed. + /// 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. /// - public TimeSpan? DefaultUnaryCallTimeout { get; set; } + public TimeSpan? GrpcCallTimeout { get; set; } /// /// Creates a new instance of from a connection string. diff --git a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs index cbe4a662c..34aca5b54 100644 --- a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs +++ b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs @@ -28,11 +28,8 @@ sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable 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 onUnaryCallCompleted; // Cancelled in DisposeAsync so an in-flight RecreateAsync stops promptly and does not leak the @@ -86,7 +83,7 @@ public override TResponse BlockingUnaryCall( } catch (RpcException ex) { - this.RecordFailure(ex.StatusCode, method.FullName); + this.RecordFailure(ex.StatusCode, method.FullName, options.CancellationToken); throw; } } @@ -96,7 +93,7 @@ public override AsyncUnaryCall AsyncUnaryCall( { TransportState current = Volatile.Read(ref this.state); AsyncUnaryCall call = current.Invoker.AsyncUnaryCall(method, host, options, request); - this.ObserveOutcome(call.ResponseAsync, method.FullName); + this.ObserveOutcome(call.ResponseAsync, method.FullName, options.CancellationToken); return call; } @@ -193,15 +190,12 @@ 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); @@ -209,14 +203,16 @@ void ObserveOutcome(Task responseAsync, string methodFullName) 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); } } @@ -225,11 +221,11 @@ 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) + if (status == StatusCode.Cancelled && cancellation.IsCancellationRequested) { - // Local cancellation is not evidence of either a transport failure or a successful round-trip. + // gRPC does not identify cancellation's origin; an actually cancelled caller token is neutral. return; } @@ -245,7 +241,7 @@ void RecordFailure(StatusCode status, string methodFullName) bool counts = status switch { StatusCode.Unavailable => true, - StatusCode.DeadlineExceeded => !DefaultUnaryCallTimeoutInterceptor.IsLongPollingMethod(methodFullName), + StatusCode.DeadlineExceeded => !ProtoUtils.IsLongPollingMethod(methodFullName), _ => false, }; diff --git a/src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs b/src/Client/Grpc/GrpcCallTimeoutInterceptor.cs similarity index 69% rename from src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs rename to src/Client/Grpc/GrpcCallTimeoutInterceptor.cs index 6f7c9fe33..1503af8fe 100644 --- a/src/Client/Grpc/DefaultUnaryCallTimeoutInterceptor.cs +++ b/src/Client/Grpc/GrpcCallTimeoutInterceptor.cs @@ -8,20 +8,20 @@ namespace Microsoft.DurableTask.Client.Grpc; /// /// Applies the configured timeout without replacing an earlier deadline or timing out long polls and streams. /// -sealed class DefaultUnaryCallTimeoutInterceptor : Interceptor +sealed class GrpcCallTimeoutInterceptor : Interceptor { readonly TimeSpan timeout; /// - /// Initializes a new instance of the class. + /// Initializes a new instance of the class. /// /// The positive timeout applied to each unary call. - public DefaultUnaryCallTimeoutInterceptor(TimeSpan timeout) + public GrpcCallTimeoutInterceptor(TimeSpan timeout) { if (timeout <= TimeSpan.Zero) { throw new ArgumentOutOfRangeException( - nameof(timeout), timeout, "DefaultUnaryCallTimeout must be positive."); + nameof(timeout), timeout, "GrpcCallTimeout must be positive."); } this.timeout = timeout; @@ -41,21 +41,12 @@ public override AsyncUnaryCall AsyncUnaryCall( AsyncUnaryCallContinuation continuation) => continuation(request, this.WithDeadline(context)); - /// - /// Identifies calls whose normal operation can wait indefinitely for an orchestration state change. - /// - /// The fully qualified gRPC method name. - /// Whether the call is a long poll. - internal static bool IsLongPollingMethod(string methodFullName) - => methodFullName is "/TaskHubSidecarService/WaitForInstanceCompletion" - or "/TaskHubSidecarService/WaitForInstanceStart"; - ClientInterceptorContext WithDeadline( ClientInterceptorContext context) where TRequest : class where TResponse : class { - if (IsLongPollingMethod(context.Method.FullName)) + if (ProtoUtils.IsLongPollingMethod(context.Method.FullName)) { return context; } diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 19e2ea79d..f18c75653 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -656,7 +656,7 @@ await this.largePayloadPurgeClient.SetLargePayloadAutoPurgeAsync( static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { - DefaultUnaryCallTimeoutInterceptor? timeoutInterceptor = options.DefaultUnaryCallTimeout is TimeSpan timeout + GrpcCallTimeoutInterceptor? timeoutInterceptor = options.GrpcCallTimeout is TimeSpan timeout ? new(timeout) : null; AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker core); diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index bdaf6bd94..6cd1fdb99 100644 --- a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs +++ b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs @@ -25,9 +25,9 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions /// Must be positive when set. An earlier deadline supplied by an interceptor is preserved. /// Streaming calls are not affected. Expiration fails the call with ; /// it does not retry the operation or guarantee that the server stopped processing it. - /// This option is captured when the client is constructed. + /// This option is captured when the client is constructed and bounds each RPC, not the whole client operation. /// - public TimeSpan? DefaultUnaryCallTimeout { get; set; } + public TimeSpan? GrpcCallTimeout { get; set; } /// /// Gets or sets the gRPC channel to use. Will supersede when provided. diff --git a/src/Client/Grpc/ProtoUtils.cs b/src/Client/Grpc/ProtoUtils.cs index 33d9e8cf2..ae970dbcb 100644 --- a/src/Client/Grpc/ProtoUtils.cs +++ b/src/Client/Grpc/ProtoUtils.cs @@ -92,6 +92,15 @@ public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy( return dedupeStatuses; } + /// + /// Identifies RPCs that can wait indefinitely for an orchestration state change. + /// + /// The fully qualified gRPC method name. + /// Whether the method is a long poll. + 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. /// /// Converts to . diff --git a/src/Client/Grpc/README.md b/src/Client/Grpc/README.md index 70f73f354..c4acbabbb 100644 --- a/src/Client/Grpc/README.md +++ b/src/Client/Grpc/README.md @@ -7,13 +7,13 @@ Commonly used types: ## Optional unary call deadlines By default, the client adds no gRPC deadline. To bound individual unary calls, configure a positive -`DefaultUnaryCallTimeout` when registering the client: +`GrpcCallTimeout` when registering the client: ```csharp services.AddDurableTaskClient(builder => builder.UseGrpc(options => { options.Address = "http://localhost:4001"; - options.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30); + options.GrpcCallTimeout = TimeSpan.FromSeconds(30); })); ``` @@ -22,14 +22,17 @@ For Azure Durable Task Scheduler, configure the same option on the scheduler opt ```csharp services.AddDurableTaskClient(builder => builder.UseDurableTaskScheduler( connectionString, - options => options.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30))); + options => options.GrpcCallTimeout = TimeSpan.FromSeconds(30))); ``` -The timeout is captured when the client is constructed and bounds each unary RPC, including any -configured gRPC retries. An earlier deadline set through `GrpcDurableTaskClientOptions.Interceptors` +The timeout is captured when the client is constructed and bounds each unary RPC, not a whole +client operation, including any configured gRPC retries within that RPC. +An earlier deadline set through `GrpcDurableTaskClientOptions.Interceptors` is preserved. Extremely large timeouts are capped at the latest finite UTC deadline supported by `DateTime`. Caller cancellation remains independent and is not counted as a transport failure or -treated as proof of a healthy connection. +treated as proof of a healthy connection. A `Cancelled` response with no cancelled caller token +still resets the failure count like other application responses. gRPC does not identify the source +of a cancellation when server and caller cancellation race. `WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls such as `GetOrchestrationHistoryAsync` are excluded; use their cancellation tokens to bound those operations. diff --git a/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs b/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs index 0efe32bf3..a14272542 100644 --- a/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs +++ b/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs @@ -301,7 +301,7 @@ public async Task UseDurableTaskScheduler_UnaryTimeout_IsNamedAndDoesNotChangeCh ServiceCollection services = new(); string connectionString = $"Endpoint=http://localhost;Authentication=None;TaskHub={ValidTaskHub}"; services.AddDurableTaskClient("bounded", builder => builder.UseDurableTaskScheduler( - connectionString, options => options.DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30))); + connectionString, options => options.GrpcCallTimeout = TimeSpan.FromSeconds(30))); services.AddDurableTaskClient("unbounded", builder => builder.UseDurableTaskScheduler(connectionString)); await using ServiceProvider provider = services.BuildServiceProvider(); IOptionsFactory factory = @@ -312,13 +312,13 @@ public async Task UseDurableTaskScheduler_UnaryTimeout_IsNamedAndDoesNotChangeCh // Act GrpcDurableTaskClientOptions bounded = factory.Create("bounded"); GrpcDurableTaskClientOptions unbounded = factory.Create("unbounded"); - schedulerOptions.Get("bounded").DefaultUnaryCallTimeout = TimeSpan.FromSeconds(10); + schedulerOptions.Get("bounded").GrpcCallTimeout = TimeSpan.FromSeconds(10); GrpcDurableTaskClientOptions updated = factory.Create("bounded"); // Assert - bounded.DefaultUnaryCallTimeout.Should().Be(TimeSpan.FromSeconds(30)); - unbounded.DefaultUnaryCallTimeout.Should().BeNull(); - updated.DefaultUnaryCallTimeout.Should().Be(TimeSpan.FromSeconds(10)); + bounded.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(30)); + unbounded.GrpcCallTimeout.Should().BeNull(); + updated.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(10)); updated.Channel.Should().BeSameAs(bounded.Channel); } diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs index 85d37a201..71fade357 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs @@ -155,7 +155,7 @@ public async Task AsyncUnaryCall_SuccessAfterFailure_ResetsConsecutiveFailureCou [Theory] [InlineData(false)] [InlineData(true)] - public async Task UnaryCall_CancelledBetweenTransportFailures_DoesNotResetFailureCount(bool blocking) + public async Task UnaryCall_ServerCancelledBetweenTransportFailures_ResetsFailureCount(bool blocking) { // Arrange int callIndex = 0; @@ -167,6 +167,50 @@ public async Task UnaryCall_CancelledBetweenTransportFailures_DoesNotResetFailur GrpcDurableTaskClientOptions options = new() { Channel = channel }; options.Internal.ChannelRecreateFailureThreshold = 2; options.Internal.MinRecreateInterval = TimeSpan.Zero; + using CancellationTokenSource cancellation = new(); + int recreatorCalls = 0; + options.SetChannelRecreator((existingChannel, ct) => + { + Interlocked.Increment(ref recreatorCalls); + return Task.FromResult(existingChannel); + }); + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + + // Act + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Cancelled, blocking, cancellation.Token); + + // Assert + cancellation.IsCancellationRequested.Should().BeFalse(); + GetConsecutiveFailures(callInvoker).Should().Be(0, "a server-returned Cancelled is an application response"); + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); + GetConsecutiveFailures(callInvoker).Should().Be(1); + recreatorCalls.Should().Be(0); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task UnaryCall_CallerCancelledBetweenTransportFailures_DoesNotResetFailureCount(bool blocking) + { + // Arrange + int callIndex = 0; + TaskCompletionSource requestStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + CallbackHttpMessageHandler handler = new(async (request, cancellationToken) => + { + if (Interlocked.Increment(ref callIndex) == 2) + { + requestStarted.TrySetResult(); + await Task.Delay(Timeout.Infinite, cancellationToken); + } + + return CreateFailureResponse(StatusCode.Unavailable, "transport failure"); + }); + using GrpcChannel channel = CreateChannel("http://caller-cancelled.client.test", handler); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Internal.ChannelRecreateFailureThreshold = 2; + options.Internal.MinRecreateInterval = TimeSpan.Zero; TaskCompletionSource recreateRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); options.SetChannelRecreator((existingChannel, ct) => { @@ -175,12 +219,29 @@ public async Task UnaryCall_CancelledBetweenTransportFailures_DoesNotResetFailur }); (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); await using AsyncDisposable cleanup = disposable; + using CancellationTokenSource cancellation = new(); // Act await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); - await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Cancelled, blocking); + using AsyncUnaryCall? call = blocking ? null : callInvoker.AsyncUnaryCall( + TestMethod, host: null, new CallOptions(cancellationToken: cancellation.Token), request: "ping"); + Task response = blocking + ? Task.Run(() => callInvoker.BlockingUnaryCall( + TestMethod, host: null, new CallOptions(cancellationToken: cancellation.Token), request: "ping")) + : call!.ResponseAsync; + try + { + await requestStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + } + finally + { + cancellation.Cancel(); + } + + RpcException error = await Assert.ThrowsAsync(() => response.WaitAsync(TimeSpan.FromSeconds(5))); // Assert + error.StatusCode.Should().Be(StatusCode.Cancelled); GetConsecutiveFailures(callInvoker).Should().Be(1, "caller cancellation is not a successful transport round-trip"); await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, blocking); await recreateRequested.Task.WaitAsync(TimeSpan.FromSeconds(5)); @@ -482,21 +543,26 @@ static async Task AssertRpcFailureAsync(CallInvoker callInvoker) => await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable); static async Task AssertRpcFailureAsync( - CallInvoker callInvoker, Method method, StatusCode expectedStatus, bool blocking = false) + CallInvoker callInvoker, + Method method, + StatusCode expectedStatus, + bool blocking = false, + CancellationToken cancellation = default) { Func act = async () => { if (blocking) { callInvoker.BlockingUnaryCall( - method, host: null, new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1)), request: "ping"); + method, host: null, new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1), cancellationToken: cancellation), + request: "ping"); return; } using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( method, host: null, - new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1)), + new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1), cancellationToken: cancellation), request: "ping"); await call.ResponseAsync; diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs index 8501ba080..63bf4d3e0 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientInterceptorsTests.cs @@ -200,7 +200,7 @@ public async Task GetCallInvoker_WithTimeout_BoundsEachUnaryCallAndPreservesCall using CancellationTokenSource cancellation = new(); Metadata headers = new() { { "test-header", "preserved" } }; TimeSpan timeout = TimeSpan.FromSeconds(30); - GrpcDurableTaskClientOptions options = new() { CallInvoker = external, DefaultUnaryCallTimeout = timeout }; + GrpcDurableTaskClientOptions options = new() { CallInvoker = external, GrpcCallTimeout = timeout }; (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); await using AsyncDisposable cleanup = disposable; @@ -227,7 +227,7 @@ public async Task GetCallInvoker_WithTimeout_PreservesEarlierUserInterceptorDead GrpcDurableTaskClientOptions options = new() { CallInvoker = external, - DefaultUnaryCallTimeout = TimeSpan.FromSeconds(30), + GrpcCallTimeout = TimeSpan.FromSeconds(30), }; options.Interceptors.Add(new EarlierDeadlineInterceptor(earlier)); (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); @@ -251,7 +251,7 @@ public async Task GetCallInvoker_WithTimeout_DoesNotChangeWaitOrStreamingDeadlin GrpcDurableTaskClientOptions options = new() { CallInvoker = external, - DefaultUnaryCallTimeout = TimeSpan.FromSeconds(1), + GrpcCallTimeout = TimeSpan.FromSeconds(1), }; (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); await using AsyncDisposable cleanup = disposable; @@ -277,14 +277,14 @@ public void Constructor_WithNonPositiveTimeout_Throws(long ticks) GrpcDurableTaskClientOptions options = new() { CallInvoker = new StubCallInvoker(), - DefaultUnaryCallTimeout = TimeSpan.FromTicks(ticks), + GrpcCallTimeout = TimeSpan.FromTicks(ticks), }; // Act Action construct = () => new GrpcDurableTaskClient("test", options, NullLogger.Instance); // Assert - construct.Should().Throw().WithMessage("*DefaultUnaryCallTimeout must be positive*"); + construct.Should().Throw().WithMessage("*GrpcCallTimeout must be positive*"); } [Fact] @@ -295,7 +295,7 @@ public async Task GetCallInvoker_WithMaximumTimeout_UsesFiniteDeadlineWithoutOve GrpcDurableTaskClientOptions options = new() { CallInvoker = external, - DefaultUnaryCallTimeout = TimeSpan.MaxValue, + GrpcCallTimeout = TimeSpan.MaxValue, }; (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); await using AsyncDisposable cleanup = disposable; diff --git a/test/Grpc.IntegrationTests/ClientDeadlineTests.cs b/test/Grpc.IntegrationTests/ClientDeadlineTests.cs index a12c6dbbd..8d1325daa 100644 --- a/test/Grpc.IntegrationTests/ClientDeadlineTests.cs +++ b/test/Grpc.IntegrationTests/ClientDeadlineTests.cs @@ -3,6 +3,7 @@ using System.Buffers.Binary; using System.Diagnostics; +using System.Globalization; using System.Net; using System.Threading.Channels; using Google.Protobuf; @@ -50,7 +51,7 @@ public async Task StalledReads_WithInterleavedCancellation_RecreateOnceAndRecove { builder.UseDurableTaskScheduler( $"Endpoint={sidecar.Address};TaskHub=test;Authentication=None", - options => options.DefaultUnaryCallTimeout = CallTimeout); + options => options.GrpcCallTimeout = CallTimeout); builder.UseExternalizedPayloads(); }); await using ServiceProvider provider = services.BuildServiceProvider(); @@ -116,6 +117,63 @@ public async Task StalledReads_WithInterleavedCancellation_RecreateOnceAndRecove + $"connections: stalled={sidecar.StalledConnection}, recovered={recoveredCall.Connection}, peer={peerCall.Connection}"); } + [Fact] + public async Task SidecarStartup_WhenItFails_DisposesHostAndPreservesFailure() + { + // Arrange + InvalidOperationException failure = new("startup failure"); + Mock host = new(); + host.Setup(value => value.StartAsync(It.IsAny())).ThrowsAsync(failure); + + // Act + InvalidOperationException actual = await Assert.ThrowsAsync( + () => StalledSidecar.StartAsync(host.Object)); + + // Assert + Assert.Same(failure, actual); + host.Verify(value => value.Dispose(), Times.Once); + } + + [Fact] + public async Task SidecarStartup_WhenCancelled_FinishesStartupAndDisposesHost() + { + // Arrange + using CancellationTokenSource cancellation = new(); + TaskCompletionSource started = new(TaskCreationOptions.RunContinuationsAsynchronously); + bool startupFinished = false; + Mock host = new(); + host.Setup(value => value.StartAsync(It.IsAny())) + .Returns(async (CancellationToken token) => + { + started.TrySetResult(); + try + { + await Task.Delay(System.Threading.Timeout.Infinite, token); + } + finally + { + startupFinished = true; + } + }); + + // Act + Task startup = StalledSidecar.StartAsync(host.Object, cancellation.Token); + try + { + await started.Task.WaitAsync(Timeout); + } + finally + { + cancellation.Cancel(); + } + + await Assert.ThrowsAnyAsync(() => startup.WaitAsync(Timeout)); + + // Assert + Assert.True(startupFinished); + host.Verify(value => value.Dispose(), Times.Once); + } + async Task AssertDeadlineAsync(DurableTaskClient client, StalledSidecar sidecar) { using CancellationTokenSource guard = new(Timeout); @@ -125,6 +183,9 @@ async Task AssertDeadlineAsync(DurableTaskClient client, StalledSidecar sidecar) RpcException error = await Assert.ThrowsAsync(() => read.WaitAsync(Timeout)); Assert.Equal(StatusCode.DeadlineExceeded, error.StatusCode); Assert.Equal(sidecar.StalledConnection, call.Connection); + Assert.EndsWith("m", call.Deadline); + int milliseconds = int.Parse(call.Deadline.AsSpan(0, call.Deadline.Length - 1), CultureInfo.InvariantCulture); + Assert.InRange(milliseconds, 1, (int)CallTimeout.TotalMilliseconds); output.WriteLine($"Stalled GetInstance: {error.StatusCode}, {elapsed.Elapsed.TotalMilliseconds:F0} ms, grpc-timeout={call.Deadline}"); } @@ -170,19 +231,29 @@ sealed class StalledSidecar : IAsyncDisposable public ObservedCall[] Calls => this.observedCalls.ToArray(); - public static async Task StartAsync() + public static async Task StartAsync(IHost? host = null, CancellationToken cancellation = default) { StalledSidecar sidecar = new(); - sidecar.host = Host.CreateDefaultBuilder() + sidecar.host = host ?? Host.CreateDefaultBuilder() .ConfigureLogging(logging => logging.ClearProviders()) .ConfigureWebHostDefaults(web => web .UseKestrel(options => options.Listen(IPAddress.Loopback, 0, endpoint => endpoint.Protocols = HttpProtocols.Http2)) .Configure(app => app.Run(sidecar.HandleAsync))) .Build(); - await sidecar.host.StartAsync().WaitAsync(Timeout); - sidecar.Address = Assert.Single(sidecar.host.Services.GetRequiredService() - .Features.Get()!.Addresses); - return sidecar; + using CancellationTokenSource startup = CancellationTokenSource.CreateLinkedTokenSource(cancellation); + startup.CancelAfter(Timeout); + try + { + await sidecar.host.StartAsync(startup.Token); + sidecar.Address = Assert.Single(sidecar.host.Services.GetRequiredService() + .Features.Get()!.Addresses); + return sidecar; + } + catch + { + sidecar.host.Dispose(); + throw; + } } public Task NextCallAsync() => this.calls.Reader.ReadAsync().AsTask().WaitAsync(Timeout); From 563aa3615260487b5fe19f55e9bce530221141b0 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 27 Oct 2026 11:54:51 -0700 Subject: [PATCH 3/3] Clarify timeout exclusions and shorten client docs Explain why the shared long-poll classifier excludes both default call deadlines and expected deadline failures. Keep one concise client configuration example and the essential timeout contract. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1e2ddb82-9306-4e21-8e6c-1c753d09f684 --- src/Client/Grpc/ProtoUtils.cs | 3 ++- src/Client/Grpc/README.md | 36 ++++++----------------------------- 2 files changed, 8 insertions(+), 31 deletions(-) diff --git a/src/Client/Grpc/ProtoUtils.cs b/src/Client/Grpc/ProtoUtils.cs index ae970dbcb..93633a192 100644 --- a/src/Client/Grpc/ProtoUtils.cs +++ b/src/Client/Grpc/ProtoUtils.cs @@ -93,7 +93,8 @@ public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy( } /// - /// Identifies RPCs that can wait indefinitely for an orchestration state change. + /// Identifies long-poll RPCs excluded from the default call timeout. Their expected + /// DeadlineExceeded responses also do not count as transport failures. /// /// The fully qualified gRPC method name. /// Whether the method is a long poll. diff --git a/src/Client/Grpc/README.md b/src/Client/Grpc/README.md index c4acbabbb..0a0461075 100644 --- a/src/Client/Grpc/README.md +++ b/src/Client/Grpc/README.md @@ -6,8 +6,8 @@ Commonly used types: ## Optional unary call deadlines -By default, the client adds no gRPC deadline. To bound individual unary calls, configure a positive -`GrpcCallTimeout` when registering the client: +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 => @@ -17,34 +17,10 @@ services.AddDurableTaskClient(builder => builder.UseGrpc(options => })); ``` -For Azure Durable Task Scheduler, configure the same option on the scheduler options: +The same property is available on `DurableTaskSchedulerClientOptions` when using `UseDurableTaskScheduler`. +`WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls are excluded. -```csharp -services.AddDurableTaskClient(builder => builder.UseDurableTaskScheduler( - connectionString, - options => options.GrpcCallTimeout = TimeSpan.FromSeconds(30))); -``` - -The timeout is captured when the client is constructed and bounds each unary RPC, not a whole -client operation, including any configured gRPC retries within that RPC. -An earlier deadline set through `GrpcDurableTaskClientOptions.Interceptors` -is preserved. Extremely large timeouts are capped at the latest finite UTC deadline supported by -`DateTime`. Caller cancellation remains independent and is not counted as a transport failure or -treated as proof of a healthy connection. A `Cancelled` response with no cancelled caller token -still resets the failure count like other application responses. gRPC does not identify the source -of a cancellation when server and caller cancellation race. - -`WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls such as -`GetOrchestrationHistoryAsync` are excluded; use their cancellation tokens to bound those operations. -The option does not change worker streams or configure HTTP/2 keepalive. - -A stalled unary call expires with `StatusCode.DeadlineExceeded`, which can contribute to the existing -channel recreation threshold. Scheduler-managed clients retain their cached-channel recovery and -interceptors. Supplying an external `CallInvoker` still opts out of channel recreation. - -**A deadline does not guarantee that the server stopped processing the request.** This applies to -mutations such as scheduling, signaling, and purging as well as reads. This option adds no automatic -retry; determine the outcome before retrying a mutation. Channel recreation benefits subsequent -calls, not the failed call, and does not resolve an underlying server outage. +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 \ No newline at end of file