Skip to content

Commit 0a712e8

Browse files
YunchuWangYunchuWang
authored
Support opt-in unary client deadlines (#826)
1 parent 2a409da commit 0a712e8

12 files changed

Lines changed: 773 additions & 46 deletions

‎src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,7 @@ public void Configure(string? name, GrpcDurableTaskClientOptions options)
147147

148148
string optionsName = name ?? Options.DefaultName;
149149
DurableTaskSchedulerClientOptions source = this.schedulerOptions.Get(optionsName);
150+
options.GrpcCallTimeout = source.GrpcCallTimeout;
150151

151152
// Create a cache key that includes all properties that affect CreateChannel behavior.
152153
// This ensures channels are reused for the same configuration

‎src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,18 @@ public string ResourceId
6969
/// </summary>
7070
public ClientRetryOptions? RetryOptions { get; set; }
7171

72+
/// <summary>
73+
/// Gets or sets the default timeout for each unary gRPC call, excluding waits for orchestration start or completion.
74+
/// The default is <see langword="null"/>, which adds no deadline. Must be positive when set.
75+
/// </summary>
76+
/// <remarks>
77+
/// Earlier interceptor deadlines are preserved and streaming calls are not affected.
78+
/// Expiration fails the call with <see cref="StatusCode.DeadlineExceeded"/>; it does not retry the operation
79+
/// or guarantee that the server stopped processing it. This option is captured when the client is constructed
80+
/// and bounds each RPC, not the whole client operation.
81+
/// </remarks>
82+
public TimeSpan? GrpcCallTimeout { get; set; }
83+
7284
/// <summary>
7385
/// Creates a new instance of <see cref="DurableTaskSchedulerClientOptions"/> from a connection string.
7486
/// </summary>

‎src/Client/Grpc/ChannelRecreatingCallInvoker.cs‎

Lines changed: 19 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -22,27 +22,14 @@ namespace Microsoft.DurableTask.Client.Grpc;
2222
/// </remarks>
2323
sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable
2424
{
25-
/// <summary>
26-
/// Methods on which a <see cref="StatusCode.DeadlineExceeded"/> response is expected behavior
27-
/// (long-poll-style waits) and must NOT be counted toward the recreate threshold.
28-
/// </summary>
29-
static readonly HashSet<string> DeadlineExceededAllowedMethods = new(StringComparer.Ordinal)
30-
{
31-
"/TaskHubSidecarService/WaitForInstanceCompletion",
32-
"/TaskHubSidecarService/WaitForInstanceStart",
33-
};
34-
3525
readonly Func<GrpcChannel, CancellationToken, Task<GrpcChannel>> recreator;
3626
readonly int failureThreshold;
3727
readonly TimeSpan minRecreateInterval;
3828
readonly bool ownsChannel;
3929
readonly ILogger logger;
4030

41-
// Cached once per invoker instance (instead of once per RPC) so ObserveOutcome's ContinueWith call
42-
// does not allocate a new delegate for every unary call. The method-name diagnostic is threaded
43-
// through as the ContinueWith `state` argument (a string, already a reference type) instead of a
44-
// boxed (self, methodName) tuple, eliminating that per-call delegate and boxed-tuple allocation
45-
// (Task.ContinueWith's own internal continuation/Task bookkeeping is unaffected).
31+
// Reuse the continuation delegate. Calls without a caller token also reuse the method-name string
32+
// as continuation state; cancellable calls carry the token so local cancellation stays neutral.
4633
readonly Action<Task, object?> onUnaryCallCompleted;
4734

4835
// 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>(
9683
}
9784
catch (RpcException ex)
9885
{
99-
this.RecordFailure(ex.StatusCode, method.FullName);
86+
this.RecordFailure(ex.StatusCode, method.FullName, options.CancellationToken);
10087
throw;
10188
}
10289
}
@@ -106,7 +93,7 @@ public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
10693
{
10794
TransportState current = Volatile.Read(ref this.state);
10895
AsyncUnaryCall<TResponse> call = current.Invoker.AsyncUnaryCall(method, host, options, request);
109-
this.ObserveOutcome(call.ResponseAsync, method.FullName);
96+
this.ObserveOutcome(call.ResponseAsync, method.FullName, options.CancellationToken);
11097
return call;
11198
}
11299

@@ -203,30 +190,29 @@ static TimeSpan ElapsedSince(long previousTimestamp, long nowTimestamp)
203190
return TimeSpan.FromSeconds((double)elapsedTicks / Stopwatch.Frequency);
204191
}
205192

206-
void ObserveOutcome(Task responseAsync, string methodFullName)
193+
void ObserveOutcome(Task responseAsync, string methodFullName, CancellationToken cancellation)
207194
{
208-
// Use ContinueWith with TaskScheduler.Default so we don't capture sync context. Both the
209-
// continuation delegate (this.onUnaryCallCompleted, cached once per invoker instance) and the
210-
// continuation state (methodFullName, already a reference-typed string) are allocation-free
211-
// per call: no boxed tuple and no per-call delegate closure.
195+
object continuationState = cancellation.CanBeCanceled ? (methodFullName, cancellation) : methodFullName;
212196
responseAsync.ContinueWith(
213197
this.onUnaryCallCompleted,
214-
methodFullName,
198+
continuationState,
215199
CancellationToken.None,
216200
TaskContinuationOptions.ExecuteSynchronously,
217201
TaskScheduler.Default);
218202
}
219203

220204
void OnUnaryCallCompleted(Task task, object? continuationState)
221205
{
222-
string methodFullName = (string)continuationState!;
223206
if (task.Status == TaskStatus.RanToCompletion)
224207
{
225208
this.RecordSuccess();
226209
}
227210
else if (task.Exception?.InnerException is RpcException rpcEx)
228211
{
229-
this.RecordFailure(rpcEx.StatusCode, methodFullName);
212+
(string methodFullName, CancellationToken cancellation) = continuationState is string method
213+
? (method, default)
214+
: ((string, CancellationToken))continuationState!;
215+
this.RecordFailure(rpcEx.StatusCode, methodFullName, cancellation);
230216
}
231217
}
232218

@@ -235,8 +221,14 @@ void RecordSuccess()
235221
Volatile.Write(ref this.consecutiveFailures, 0);
236222
}
237223

238-
void RecordFailure(StatusCode status, string methodFullName)
224+
void RecordFailure(StatusCode status, string methodFullName, CancellationToken cancellation)
239225
{
226+
if (status == StatusCode.Cancelled && cancellation.IsCancellationRequested)
227+
{
228+
// gRPC does not identify cancellation's origin; an actually cancelled caller token is neutral.
229+
return;
230+
}
231+
240232
// Only count statuses that indicate an actual transport problem, not application-level errors:
241233
// * Unavailable — half-open connection, peer reset, or dead routing target.
242234
// * DeadlineExceeded — the call exceeded the *client-supplied* deadline. This is a
@@ -249,7 +241,7 @@ void RecordFailure(StatusCode status, string methodFullName)
249241
bool counts = status switch
250242
{
251243
StatusCode.Unavailable => true,
252-
StatusCode.DeadlineExceeded => !DeadlineExceededAllowedMethods.Contains(methodFullName),
244+
StatusCode.DeadlineExceeded => !ProtoUtils.IsLongPollingMethod(methodFullName),
253245
_ => false,
254246
};
255247

‎src/Client/Grpc/GrpcCallTimeoutInterceptor.cs‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
// Copyright (c) Microsoft Corporation.
2+
// Licensed under the MIT License.
3+
4+
using Grpc.Core.Interceptors;
5+
6+
namespace Microsoft.DurableTask.Client.Grpc;
7+
8+
/// <summary>
9+
/// Applies the configured timeout without replacing an earlier deadline or timing out long polls and streams.
10+
/// </summary>
11+
sealed class GrpcCallTimeoutInterceptor : Interceptor
12+
{
13+
readonly TimeSpan timeout;
14+
15+
/// <summary>
16+
/// Initializes a new instance of the <see cref="GrpcCallTimeoutInterceptor"/> class.
17+
/// </summary>
18+
/// <param name="timeout">The positive timeout applied to each unary call.</param>
19+
public GrpcCallTimeoutInterceptor(TimeSpan timeout)
20+
{
21+
if (timeout <= TimeSpan.Zero)
22+
{
23+
throw new ArgumentOutOfRangeException(
24+
nameof(timeout), timeout, "GrpcCallTimeout must be positive.");
25+
}
26+
27+
this.timeout = timeout;
28+
}
29+
30+
/// <inheritdoc/>
31+
public override TResponse BlockingUnaryCall<TRequest, TResponse>(
32+
TRequest request,
33+
ClientInterceptorContext<TRequest, TResponse> context,
34+
BlockingUnaryCallContinuation<TRequest, TResponse> continuation)
35+
=> continuation(request, this.WithDeadline(context));
36+
37+
/// <inheritdoc/>
38+
public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
39+
TRequest request,
40+
ClientInterceptorContext<TRequest, TResponse> context,
41+
AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
42+
=> continuation(request, this.WithDeadline(context));
43+
44+
ClientInterceptorContext<TRequest, TResponse> WithDeadline<TRequest, TResponse>(
45+
ClientInterceptorContext<TRequest, TResponse> context)
46+
where TRequest : class
47+
where TResponse : class
48+
{
49+
if (ProtoUtils.IsLongPollingMethod(context.Method.FullName))
50+
{
51+
return context;
52+
}
53+
54+
DateTime now = DateTime.UtcNow;
55+
56+
// Avoid overflow and DateTime.MaxValue, which gRPC treats as no deadline.
57+
long ticks = Math.Min(this.timeout.Ticks, DateTime.MaxValue.Ticks - now.Ticks - 1);
58+
DateTime deadline = now.AddTicks(ticks);
59+
if (context.Options.Deadline is DateTime existing && existing <= deadline)
60+
{
61+
return context;
62+
}
63+
64+
return new(context.Method, context.Host, context.Options.WithDeadline(deadline));
65+
}
66+
}

‎src/Client/Grpc/GrpcDurableTaskClient.cs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -656,10 +656,18 @@ await this.largePayloadPurgeClient.SetLargePayloadAutoPurgeAsync(
656656

657657
static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker)
658658
{
659+
GrpcCallTimeoutInterceptor? timeoutInterceptor = options.GrpcCallTimeout is TimeSpan timeout
660+
? new(timeout)
661+
: null;
659662
AsyncDisposable disposable = GetCallInvokerCore(options, logger, out CallInvoker core);
660663

664+
if (timeoutInterceptor is not null)
665+
{
666+
core = core.Intercept(timeoutInterceptor);
667+
}
668+
661669
// Intercept outside any ChannelRecreatingCallInvoker so the wrapper's internal channel swaps
662-
// stay transparent to the configured interceptors.
670+
// stay transparent to the configured interceptors. User interceptors can supply an earlier deadline.
663671
callInvoker = ApplyInterceptors(options.Interceptors, core);
664672
return disposable;
665673
}

‎src/Client/Grpc/GrpcDurableTaskClientOptions.cs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,20 @@ public sealed class GrpcDurableTaskClientOptions : DurableTaskClientOptions
1515
/// </summary>
1616
public string? Address { get; set; }
1717

18+
/// <summary>
19+
/// Gets or sets the default timeout for each unary gRPC call, excluding
20+
/// <see cref="DurableTaskClient.WaitForInstanceStartAsync(string, bool, CancellationToken)"/> and
21+
/// <see cref="DurableTaskClient.WaitForInstanceCompletionAsync(string, bool, CancellationToken)"/>.
22+
/// The default is <see langword="null"/>, which adds no deadline.
23+
/// </summary>
24+
/// <remarks>
25+
/// Must be positive when set. An earlier deadline supplied by an interceptor is preserved.
26+
/// Streaming calls are not affected. Expiration fails the call with <see cref="StatusCode.DeadlineExceeded"/>;
27+
/// it does not retry the operation or guarantee that the server stopped processing it.
28+
/// This option is captured when the client is constructed and bounds each RPC, not the whole client operation.
29+
/// </remarks>
30+
public TimeSpan? GrpcCallTimeout { get; set; }
31+
1832
/// <summary>
1933
/// Gets or sets the gRPC channel to use. Will supersede <see cref="CallInvoker" /> when provided.
2034
/// </summary>

‎src/Client/Grpc/ProtoUtils.cs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,16 @@ public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy(
9292
return dedupeStatuses;
9393
}
9494

95+
/// <summary>
96+
/// Identifies long-poll RPCs excluded from the default call timeout. Their expected
97+
/// <c>DeadlineExceeded</c> responses also do not count as transport failures.
98+
/// </summary>
99+
/// <param name="methodFullName">The fully qualified gRPC method name.</param>
100+
/// <returns>Whether the method is a long poll.</returns>
101+
internal static bool IsLongPollingMethod(string methodFullName)
102+
=> methodFullName is "/TaskHubSidecarService/WaitForInstanceCompletion"
103+
or "/TaskHubSidecarService/WaitForInstanceStart";
104+
95105
#pragma warning disable 0618 // Referencing Obsolete member. This is intention as we are only converting it.
96106
/// <summary>
97107
/// Converts <see cref="OrchestrationRuntimeStatus" /> to <see cref="P.OrchestrationStatus" />.

‎src/Client/Grpc/README.md‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,4 +4,23 @@ Commonly used types:
44
- `GrpcDurableTaskClient`
55
- `GrpcDurableTaskClientOptions`
66

7+
## Optional unary call deadlines
8+
9+
Set `GrpcCallTimeout` to a positive value to bound each unary gRPC call, not the whole operation.
10+
The default is `null` (no deadline); an earlier explicit deadline is preserved.
11+
12+
```csharp
13+
services.AddDurableTaskClient(builder => builder.UseGrpc(options =>
14+
{
15+
options.Address = "http://localhost:4001";
16+
options.GrpcCallTimeout = TimeSpan.FromSeconds(30);
17+
}));
18+
```
19+
20+
The same property is available on `DurableTaskSchedulerClientOptions` when using `UseDurableTaskScheduler`.
21+
`WaitForInstanceStartAsync`, `WaitForInstanceCompletionAsync`, and streaming calls are excluded.
22+
23+
Expiration fails the call with `StatusCode.DeadlineExceeded`. No retries are added, and a timed-out
24+
mutation may still have been applied by the server.
25+
726
For more information, see https://github.com/microsoft/durabletask-dotnet

‎test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs‎

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -294,6 +294,34 @@ public void UseDurableTaskScheduler_WithConnectionStringAndRetryOptions_ShouldCo
294294
}
295295
}
296296

297+
[Fact]
298+
public async Task UseDurableTaskScheduler_UnaryTimeout_IsNamedAndDoesNotChangeChannelCache()
299+
{
300+
// Arrange
301+
ServiceCollection services = new();
302+
string connectionString = $"Endpoint=http://localhost;Authentication=None;TaskHub={ValidTaskHub}";
303+
services.AddDurableTaskClient("bounded", builder => builder.UseDurableTaskScheduler(
304+
connectionString, options => options.GrpcCallTimeout = TimeSpan.FromSeconds(30)));
305+
services.AddDurableTaskClient("unbounded", builder => builder.UseDurableTaskScheduler(connectionString));
306+
await using ServiceProvider provider = services.BuildServiceProvider();
307+
IOptionsFactory<GrpcDurableTaskClientOptions> factory =
308+
provider.GetRequiredService<IOptionsFactory<GrpcDurableTaskClientOptions>>();
309+
IOptionsMonitor<DurableTaskSchedulerClientOptions> schedulerOptions =
310+
provider.GetRequiredService<IOptionsMonitor<DurableTaskSchedulerClientOptions>>();
311+
312+
// Act
313+
GrpcDurableTaskClientOptions bounded = factory.Create("bounded");
314+
GrpcDurableTaskClientOptions unbounded = factory.Create("unbounded");
315+
schedulerOptions.Get("bounded").GrpcCallTimeout = TimeSpan.FromSeconds(10);
316+
GrpcDurableTaskClientOptions updated = factory.Create("bounded");
317+
318+
// Assert
319+
bounded.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(30));
320+
unbounded.GrpcCallTimeout.Should().BeNull();
321+
updated.GrpcCallTimeout.Should().Be(TimeSpan.FromSeconds(10));
322+
updated.Channel.Should().BeSameAs(bounded.Channel);
323+
}
324+
297325
[Fact]
298326
public async Task UseDurableTaskScheduler_SameConfiguration_ReusesSameChannel()
299327
{
@@ -424,11 +452,11 @@ public async Task UseDurableTaskScheduler_DifferentResourceId_UsesSeparateChanne
424452
DefaultAzureCredential credential = new DefaultAzureCredential();
425453

426454
// Act - configure two clients with the same endpoint/taskhub but different ResourceId
427-
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
455+
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
428456
{
429457
options.ResourceId = "https://durabletask.io";
430458
});
431-
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
459+
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
432460
{
433461
options.ResourceId = "https://custom.durabletask.io";
434462
});
@@ -487,11 +515,11 @@ public async Task UseDurableTaskScheduler_DifferentAllowInsecureCredentials_Uses
487515
DefaultAzureCredential credential = new DefaultAzureCredential();
488516

489517
// Act - configure two clients with the same endpoint/taskhub but different AllowInsecureCredentials
490-
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
518+
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
491519
{
492520
options.AllowInsecureCredentials = false;
493521
});
494-
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
522+
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
495523
{
496524
options.AllowInsecureCredentials = true;
497525
});
@@ -522,14 +550,14 @@ public async Task UseDurableTaskScheduler_DifferentRetryOptions_UsesSeparateChan
522550
DefaultAzureCredential credential = new DefaultAzureCredential();
523551

524552
// Act - configure two clients with the same endpoint/taskhub but different RetryOptions
525-
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
553+
mockBuilder1.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
526554
{
527555
options.RetryOptions = new DurableTaskSchedulerClientOptions.ClientRetryOptions
528556
{
529557
MaxRetries = 3
530558
};
531559
});
532-
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
560+
mockBuilder2.Object.UseDurableTaskScheduler(ValidEndpoint, ValidTaskHub, credential, options =>
533561
{
534562
options.RetryOptions = new DurableTaskSchedulerClientOptions.ClientRetryOptions
535563
{
@@ -548,4 +576,4 @@ public async Task UseDurableTaskScheduler_DifferentRetryOptions_UsesSeparateChan
548576
options2.Channel.Should().NotBeNull();
549577
options1.Channel.Should().NotBeSameAs(options2.Channel, "different RetryOptions should use different channels");
550578
}
551-
}
579+
}

0 commit comments

Comments
 (0)