diff --git a/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs b/src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs index a88de06e..bbb3a63d 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.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 eac29155..623bdc67 100644 --- a/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs +++ b/src/Client/AzureManaged/DurableTaskSchedulerClientOptions.cs @@ -69,6 +69,18 @@ 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 + /// and bounds each RPC, not the whole client operation. + /// + 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 f0928876..34aca5b5 100644 --- a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs +++ b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs @@ -22,27 +22,14 @@ 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; 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 @@ -96,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; } } @@ -106,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; } @@ -203,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); @@ -219,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); } } @@ -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, }; diff --git a/src/Client/Grpc/GrpcCallTimeoutInterceptor.cs b/src/Client/Grpc/GrpcCallTimeoutInterceptor.cs new file mode 100644 index 00000000..1503af8f --- /dev/null +++ b/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; + +/// +/// Applies the configured timeout without replacing an earlier deadline or timing out long polls and streams. +/// +sealed class GrpcCallTimeoutInterceptor : Interceptor +{ + readonly TimeSpan timeout; + + /// + /// Initializes a new instance of the class. + /// + /// The positive timeout applied to each unary call. + public GrpcCallTimeoutInterceptor(TimeSpan timeout) + { + if (timeout <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(timeout), timeout, "GrpcCallTimeout 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)); + + ClientInterceptorContext WithDeadline( + ClientInterceptorContext 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)); + } +} diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index f240d722..f18c7565 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) { + 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; } diff --git a/src/Client/Grpc/GrpcDurableTaskClientOptions.cs b/src/Client/Grpc/GrpcDurableTaskClientOptions.cs index f684aeac..6cd1fdb9 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 and bounds each RPC, not the whole client operation. + /// + 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 33d9e8cf..93633a19 100644 --- a/src/Client/Grpc/ProtoUtils.cs +++ b/src/Client/Grpc/ProtoUtils.cs @@ -92,6 +92,16 @@ public static P.OrchestrationIdReusePolicy ConvertDedupeStatusesToReusePolicy( return dedupeStatuses; } + /// + /// 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. + 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 f7e830a0..0a046107 100644 --- a/src/Client/Grpc/README.md +++ b/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 \ No newline at end of file diff --git a/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs b/test/Client/AzureManaged.Tests/DurableTaskSchedulerClientExtensionsTests.cs index 7f53b4d8..a1427254 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.GrpcCallTimeout = 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").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"); } -} +} \ No newline at end of file diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs index 6945ead5..71fade35 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs @@ -152,6 +152,101 @@ public async Task AsyncUnaryCall_SuccessAfterFailure_ResetsConsecutiveFailureCou } } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task UnaryCall_ServerCancelledBetweenTransportFailures_ResetsFailureCount(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; + 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) => + { + recreateRequested.TrySetResult(); + return Task.FromResult(existingChannel); + }); + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + await using AsyncDisposable cleanup = disposable; + using CancellationTokenSource cancellation = new(); + + // Act + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable, 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)); + } + [Fact] public async Task AsyncUnaryCall_DeadlineExceededOnAllowedLongPollMethod_DoesNotCountTowardThreshold() { @@ -447,14 +542,27 @@ 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, + CancellationToken cancellation = default) { Func act = async () => { + if (blocking) + { + callInvoker.BlockingUnaryCall( + 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; @@ -544,4 +652,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 641f7d74..63bf4d3e 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, GrpcCallTimeout = 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, + GrpcCallTimeout = 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, + GrpcCallTimeout = 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(), + GrpcCallTimeout = TimeSpan.FromTicks(ticks), + }; + + // Act + Action construct = () => new GrpcDurableTaskClient("test", options, NullLogger.Instance); + + // Assert + construct.Should().Throw().WithMessage("*GrpcCallTimeout must be positive*"); + } + + [Fact] + public async Task GetCallInvoker_WithMaximumTimeout_UsesFiniteDeadlineWithoutOverflow() + { + // Arrange + StubCallInvoker external = new(); + GrpcDurableTaskClientOptions options = new() + { + CallInvoker = external, + GrpcCallTimeout = 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 00000000..8d1325da --- /dev/null +++ b/test/Grpc.IntegrationTests/ClientDeadlineTests.cs @@ -0,0 +1,312 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Buffers.Binary; +using System.Diagnostics; +using System.Globalization; +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.GrpcCallTimeout = 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}"); + } + + [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); + 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); + 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}"); + } + + 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(IHost? host = null, CancellationToken cancellation = default) + { + StalledSidecar sidecar = new(); + 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(); + 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); + + 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