From a74caede9cf83f09b9385c8eb57b04eef6f09227 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:05:10 -0700 Subject: [PATCH 1/7] Reduce shim instance-start polling load with jitter and backoff (#776) WaitForInstanceStartAsync previously polled GetOrchestrationStateAsync on a fixed 1-second Task.Delay cadence. Under load, many concurrent waiters end up synchronized on the same polling tick, causing bursty spikes of backend requests instead of a smooth request rate. This changes the wait loop to: - Keep the very first status check immediate (no delay), preserving prompt observation of quick-starting orchestrations. - Compute each subsequent delay via a new ComputeNextPollingDelay helper: starts at the historical 1s interval, grows by a 1.5x multiplier per attempt (capped), up to a 5s maximum. - Apply +/-20% jitter (via a lock-guarded, securely-seeded Random in the new PollingJitter helper) so concurrent callers desynchronize instead of polling in lockstep. Cancellation, not-found (throws immediately, no delay), and terminal-state return behavior are unchanged. No public API surface was touched -- WaitForInstanceStartAsync's signature is unchanged and all new members are private implementation details, so this is not a breaking change. Added focused tests covering multi-iteration polling convergence, not-found short-circuiting without polling delay, and cancellation during the backoff delay. Fixes #776 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../ShimDurableTaskClient.cs | 75 ++++++++++++++++- .../ShimDurableTaskClientTests.cs | 83 +++++++++++++++++++ 2 files changed, 156 insertions(+), 2 deletions(-) diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs index 49414f68..8343bb1f 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs @@ -4,6 +4,7 @@ using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Globalization; +using System.Security.Cryptography; using DurableTask.Core; using DurableTask.Core.Exceptions; using DurableTask.Core.History; @@ -27,6 +28,15 @@ namespace Microsoft.DurableTask.Client.OrchestrationServiceClientShim; /// The client options. class ShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) : DurableTaskClient(name) { + // Polling parameters for WaitForInstanceStartAsync. The minimum interval matches the historical + // fixed 1-second polling cadence so quick-starting orchestrations are still observed promptly. The + // interval then grows (bounded by the maximum) for instances that remain pending longer, reducing + // sustained polling load; jitter is applied on top to desynchronize concurrent callers. + const double PollingBackoffMultiplier = 1.5; + const double PollingJitterFactor = 0.2; + static readonly TimeSpan MinPollingInterval = TimeSpan.FromSeconds(1); + static readonly TimeSpan MaxPollingInterval = TimeSpan.FromSeconds(5); + readonly ShimDurableTaskClientOptions options = Check.NotNull(options); ShimDurableEntityClient? entities; @@ -270,7 +280,7 @@ public override async Task WaitForInstanceStartAsync( { Check.NotNullOrEmpty(instanceId); - while (true) + for (int attempt = 0; ; attempt++) { OrchestrationMetadata? metadata = await this.GetInstancesAsync( instanceId, getInputsAndOutputs, cancellation); @@ -285,7 +295,13 @@ public override async Task WaitForInstanceStartAsync( return metadata; } - await Task.Delay(TimeSpan.FromSeconds(1), cancellation); + // Poll with a jittered, gradually-increasing delay. This keeps the first few retries close to + // the historical 1-second cadence -- preserving prompt observation of quick-starting + // orchestrations -- while desynchronizing concurrent waiters (avoiding synchronized polling + // bursts) and reducing steady-state load against the backend for instances that stay + // pending longer. + TimeSpan delay = ComputeNextPollingDelay(attempt); + await Task.Delay(delay, cancellation); } } @@ -344,6 +360,28 @@ await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync( return newInstanceId; } + /// + /// Computes the delay to wait before the next polling attempt. + /// + /// The zero-based index of the poll attempt that just completed. + /// A jittered delay, growing from up to . + static TimeSpan ComputeNextPollingDelay(int attempt) + { + int safeAttempt = Math.Max(attempt, 0); + + // Cap the exponent so this never overflows for pathological attempt counts. + double growth = Math.Pow(PollingBackoffMultiplier, Math.Min(safeAttempt, 8)); + double targetMs = Math.Min( + MinPollingInterval.TotalMilliseconds * growth, + MaxPollingInterval.TotalMilliseconds); + + // Apply +/- 20% jitter so concurrent callers waiting on the same or different instances don't + // converge on synchronized polling bursts against the backend. + double jitterRangeMs = targetMs * PollingJitterFactor; + double jitteredMs = targetMs + (((PollingJitter.NextDouble() * 2.0) - 1.0) * jitterRangeMs); + return TimeSpan.FromMilliseconds(Math.Max(jitteredMs, 0)); + } + [return: NotNullIfNotNull("state")] OrchestrationMetadata? ToMetadata(Core.OrchestrationState? state, bool getInputsAndOutputs) { @@ -449,4 +487,37 @@ async Task TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync( } } } + + /// + /// A minimal thread-safe random source used to jitter polling + /// delays across concurrent callers. is not thread-safe, and its + /// parameterless constructor can produce correlated sequences when many instances are created around + /// the same tick -- which is exactly the kind of synchronized behavior this jitter is meant to avoid. + /// A single, securely-seeded instance guarded by a lock avoids both issues. + /// + static class PollingJitter + { + static readonly object SyncRoot = new(); + static readonly Random Shared = CreateSeededRandom(); + + /// + /// Returns a thread-safe random double in the range [0.0, 1.0). + /// + /// A random double in the range [0.0, 1.0). + public static double NextDouble() + { + lock (SyncRoot) + { + return Shared.NextDouble(); + } + } + + static Random CreateSeededRandom() + { + byte[] seedBytes = new byte[sizeof(int)]; + using RandomNumberGenerator rng = RandomNumberGenerator.Create(); + rng.GetBytes(seedBytes); + return new Random(BitConverter.ToInt32(seedBytes, 0)); + } + } } diff --git a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs index 967d4bde..9d1d2fc4 100644 --- a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs +++ b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs @@ -300,6 +300,89 @@ public async Task WaitForInstanceStart() Validate(metadata, state2, false); } + [Fact] + public async Task WaitForInstanceStart_MultiplePendingPolls_EventuallyReturnsTerminalMetadata() + { + // arrange + DateTimeOffset start = DateTimeOffset.UtcNow; + OrchestrationInstance instance = new() + { + InstanceId = Guid.NewGuid().ToString(), + ExecutionId = Guid.NewGuid().ToString(), + }; + + Core.OrchestrationState pending1 = CreateState("input", start: start); + pending1.OrchestrationInstance = instance; + pending1.OrchestrationStatus = Core.OrchestrationStatus.Pending; + Core.OrchestrationState pending2 = CreateState("input", start: start); + pending2.OrchestrationInstance = instance; + pending2.OrchestrationStatus = Core.OrchestrationStatus.Pending; + Core.OrchestrationState terminal = CreateState("input", start: start); + terminal.OrchestrationInstance = instance; + + this.orchestrationClient.SetupSequence(m => m.GetOrchestrationStateAsync(instance.InstanceId, false)) + .ReturnsAsync([pending1]) + .ReturnsAsync([pending2]) + .ReturnsAsync([terminal]); + + // act + OrchestrationMetadata metadata = await this.client.WaitForInstanceStartAsync( + instance.InstanceId, false, default); + + // assert -- multiple polling iterations (exercising the jittered backoff loop) still converge + // on the terminal state once observed. + this.orchestrationClient.Verify( + m => m.GetOrchestrationStateAsync(instance.InstanceId, false), Times.Exactly(3)); + Validate(metadata, terminal, false); + } + + [Fact] + public async Task WaitForInstanceStart_InstanceNotFound_ThrowsImmediatelyWithoutPolling() + { + // arrange + string instanceId = Guid.NewGuid().ToString(); + this.orchestrationClient.Setup(m => m.GetOrchestrationStateAsync(instanceId, false)) + .ReturnsAsync([]); + + // act + Func act = () => this.client.WaitForInstanceStartAsync(instanceId, false, default); + + // assert -- not-found behavior is preserved: no retry/backoff delay before throwing. + await act.Should().ThrowExactlyAsync() + .WithMessage($"Orchestration with instanceId '{instanceId}' does not exist"); + this.orchestrationClient.Verify( + m => m.GetOrchestrationStateAsync(instanceId, false), Times.Once); + } + + [Fact] + public async Task WaitForInstanceStart_CancelledDuringBackoffDelay_ThrowsTaskCanceledException() + { + // arrange + DateTimeOffset start = DateTimeOffset.UtcNow; + OrchestrationInstance instance = new() + { + InstanceId = Guid.NewGuid().ToString(), + ExecutionId = Guid.NewGuid().ToString(), + }; + + Core.OrchestrationState pending = CreateState("input", start: start); + pending.OrchestrationInstance = instance; + pending.OrchestrationStatus = Core.OrchestrationStatus.Pending; + + this.orchestrationClient.Setup(m => m.GetOrchestrationStateAsync(instance.InstanceId, false)) + .ReturnsAsync([pending]); + + using CancellationTokenSource cts = new(); + + // act -- cancel shortly after the first (immediate) poll so cancellation fires while the loop is + // awaiting the jittered backoff delay rather than before any polling occurs. + cts.CancelAfter(TimeSpan.FromMilliseconds(50)); + Func act = () => this.client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); + + // assert + await act.Should().ThrowAsync(); + } + [Fact] public Task ScheduleNewOrchestrationInstance_IdGenerated_NoInput() => this.RunScheduleNewOrchestrationInstanceAsync("test", null, null); From e2d659f8c702c6731073e5903daae045b40b3293 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:18:59 -0700 Subject: [PATCH 2/7] Fix: cap WaitForInstanceStartAsync polling delay at historical 1s cadence The prior jittered exponential-backoff design (base 1s, cap 5s, +/-20% jitter) could delay detection of a newly-started orchestration by up to ~6s, regressing the historical 1s worst-case detection latency that the original issue explicitly required to be preserved. Replace growth-based backoff with a fixed 1s PollingInterval that jitter can only ever reduce, never exceed. This still desynchronizes concurrent callers (avoiding synchronized polling bursts) while guaranteeing the delay is always in [0.8s, 1.0s] -- so worst-case detection latency never regresses past the historical 1s cadence. ComputeNextPollingDelay is changed from private static to internal static (not a public API surface change) so it can be exercised directly and deterministically by a new unit test enforcing the max-latency bound. Also adds a Stopwatch-based integration test covering repeated pending observations before a terminal transition, asserting the total wait stays well under the historical cadence. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../ShimDurableTaskClient.cs | 52 +++++++-------- .../ShimDurableTaskClientTests.cs | 65 +++++++++++++++++-- 2 files changed, 83 insertions(+), 34 deletions(-) diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs index 8343bb1f..3af7ea4c 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs @@ -28,14 +28,13 @@ namespace Microsoft.DurableTask.Client.OrchestrationServiceClientShim; /// The client options. class ShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) : DurableTaskClient(name) { - // Polling parameters for WaitForInstanceStartAsync. The minimum interval matches the historical - // fixed 1-second polling cadence so quick-starting orchestrations are still observed promptly. The - // interval then grows (bounded by the maximum) for instances that remain pending longer, reducing - // sustained polling load; jitter is applied on top to desynchronize concurrent callers. - const double PollingBackoffMultiplier = 1.5; + // Polling parameters for WaitForInstanceStartAsync. PollingInterval matches the historical fixed + // 1-second polling cadence. Callers rely on prompt (<= 1 second) observation of a status + // transition, so jitter is only ever applied *downward* from this value -- never grown beyond it + // -- to desynchronize concurrent callers (avoiding synchronized polling bursts against the + // backend) without regressing the historical worst-case detection latency. const double PollingJitterFactor = 0.2; - static readonly TimeSpan MinPollingInterval = TimeSpan.FromSeconds(1); - static readonly TimeSpan MaxPollingInterval = TimeSpan.FromSeconds(5); + static readonly TimeSpan PollingInterval = TimeSpan.FromSeconds(1); readonly ShimDurableTaskClientOptions options = Check.NotNull(options); ShimDurableEntityClient? entities; @@ -280,7 +279,7 @@ public override async Task WaitForInstanceStartAsync( { Check.NotNullOrEmpty(instanceId); - for (int attempt = 0; ; attempt++) + while (true) { OrchestrationMetadata? metadata = await this.GetInstancesAsync( instanceId, getInputsAndOutputs, cancellation); @@ -295,12 +294,11 @@ public override async Task WaitForInstanceStartAsync( return metadata; } - // Poll with a jittered, gradually-increasing delay. This keeps the first few retries close to - // the historical 1-second cadence -- preserving prompt observation of quick-starting - // orchestrations -- while desynchronizing concurrent waiters (avoiding synchronized polling - // bursts) and reducing steady-state load against the backend for instances that stay - // pending longer. - TimeSpan delay = ComputeNextPollingDelay(attempt); + // Poll with a delay that is jittered *downward* from the historical 1-second cadence. + // This desynchronizes concurrent waiters (avoiding synchronized polling bursts against the + // backend) while guaranteeing the worst-case detection latency for a status transition + // never exceeds the historical 1-second cadence -- preserving prompt-start observation. + TimeSpan delay = ComputeNextPollingDelay(); await Task.Delay(delay, cancellation); } } @@ -363,23 +361,17 @@ await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync( /// /// Computes the delay to wait before the next polling attempt. /// - /// The zero-based index of the poll attempt that just completed. - /// A jittered delay, growing from up to . - static TimeSpan ComputeNextPollingDelay(int attempt) + /// + /// A delay in the range [ * (1 - ), + /// ]. Jitter only ever reduces the delay, so the returned value never + /// exceeds -- preserving the historical worst-case detection latency + /// for -- while still desynchronizing concurrent callers. + /// + internal static TimeSpan ComputeNextPollingDelay() { - int safeAttempt = Math.Max(attempt, 0); - - // Cap the exponent so this never overflows for pathological attempt counts. - double growth = Math.Pow(PollingBackoffMultiplier, Math.Min(safeAttempt, 8)); - double targetMs = Math.Min( - MinPollingInterval.TotalMilliseconds * growth, - MaxPollingInterval.TotalMilliseconds); - - // Apply +/- 20% jitter so concurrent callers waiting on the same or different instances don't - // converge on synchronized polling bursts against the backend. - double jitterRangeMs = targetMs * PollingJitterFactor; - double jitteredMs = targetMs + (((PollingJitter.NextDouble() * 2.0) - 1.0) * jitterRangeMs); - return TimeSpan.FromMilliseconds(Math.Max(jitteredMs, 0)); + double reduction = PollingJitterFactor * PollingJitter.NextDouble(); + double delayMs = PollingInterval.TotalMilliseconds * (1.0 - reduction); + return TimeSpan.FromMilliseconds(delayMs); } [return: NotNullIfNotNull("state")] diff --git a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs index 9d1d2fc4..344c3c15 100644 --- a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs +++ b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using System.Diagnostics; using DurableTask.Core; using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; @@ -329,13 +330,69 @@ public async Task WaitForInstanceStart_MultiplePendingPolls_EventuallyReturnsTer OrchestrationMetadata metadata = await this.client.WaitForInstanceStartAsync( instance.InstanceId, false, default); - // assert -- multiple polling iterations (exercising the jittered backoff loop) still converge + // assert -- multiple polling iterations (exercising the jittered polling loop) still converge // on the terminal state once observed. this.orchestrationClient.Verify( m => m.GetOrchestrationStateAsync(instance.InstanceId, false), Times.Exactly(3)); Validate(metadata, terminal, false); } + [Fact] + public async Task WaitForInstanceStart_RepeatedPendingObservations_HonorsMaxDetectionLatencyContract() + { + // arrange + DateTimeOffset start = DateTimeOffset.UtcNow; + OrchestrationInstance instance = new() + { + InstanceId = Guid.NewGuid().ToString(), + ExecutionId = Guid.NewGuid().ToString(), + }; + + Core.OrchestrationState pending1 = CreateState("input", start: start); + pending1.OrchestrationInstance = instance; + pending1.OrchestrationStatus = Core.OrchestrationStatus.Pending; + Core.OrchestrationState pending2 = CreateState("input", start: start); + pending2.OrchestrationInstance = instance; + pending2.OrchestrationStatus = Core.OrchestrationStatus.Pending; + Core.OrchestrationState terminal = CreateState("input", start: start); + terminal.OrchestrationInstance = instance; + + this.orchestrationClient.SetupSequence(m => m.GetOrchestrationStateAsync(instance.InstanceId, false)) + .ReturnsAsync([pending1]) + .ReturnsAsync([pending2]) + .ReturnsAsync([terminal]); + + Stopwatch stopwatch = Stopwatch.StartNew(); + + // act + OrchestrationMetadata metadata = await this.client.WaitForInstanceStartAsync( + instance.InstanceId, false, default); + + stopwatch.Stop(); + + // assert -- two "still pending" observations occur before the terminal state is returned, so + // exactly two polling delays are incurred. Each delay is bounded by the historical 1-second + // detection cadence (jitter only ever reduces it), so the worst-case total wait here is ~2 + // seconds. This guards against a regression -- like unconstrained backoff growth -- that would + // push per-poll delays past the historical cadence and violate prompt-start observation. + stopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(2.5)); + Validate(metadata, terminal, false); + } + + [Fact] + public void ComputeNextPollingDelay_NeverExceedsHistoricalOneSecondCadence() + { + // assert -- across many samples, the jittered delay must never exceed the historical 1-second + // polling cadence (jitter only ever reduces the delay), enforcing the max detection-latency + // contract for WaitForInstanceStartAsync, and must always be a non-negative, finite delay. + for (int i = 0; i < 1000; i++) + { + TimeSpan delay = ShimDurableTaskClient.ComputeNextPollingDelay(); + delay.Should().BeGreaterThan(TimeSpan.Zero); + delay.Should().BeLessThanOrEqualTo(TimeSpan.FromSeconds(1)); + } + } + [Fact] public async Task WaitForInstanceStart_InstanceNotFound_ThrowsImmediatelyWithoutPolling() { @@ -347,7 +404,7 @@ public async Task WaitForInstanceStart_InstanceNotFound_ThrowsImmediatelyWithout // act Func act = () => this.client.WaitForInstanceStartAsync(instanceId, false, default); - // assert -- not-found behavior is preserved: no retry/backoff delay before throwing. + // assert -- not-found behavior is preserved: no retry/polling delay before throwing. await act.Should().ThrowExactlyAsync() .WithMessage($"Orchestration with instanceId '{instanceId}' does not exist"); this.orchestrationClient.Verify( @@ -355,7 +412,7 @@ await act.Should().ThrowExactlyAsync() } [Fact] - public async Task WaitForInstanceStart_CancelledDuringBackoffDelay_ThrowsTaskCanceledException() + public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCanceledException() { // arrange DateTimeOffset start = DateTimeOffset.UtcNow; @@ -375,7 +432,7 @@ public async Task WaitForInstanceStart_CancelledDuringBackoffDelay_ThrowsTaskCan using CancellationTokenSource cts = new(); // act -- cancel shortly after the first (immediate) poll so cancellation fires while the loop is - // awaiting the jittered backoff delay rather than before any polling occurs. + // awaiting the jittered polling delay rather than before any polling occurs. cts.CancelAfter(TimeSpan.FromMilliseconds(50)); Func act = () => this.client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); From 500481cc27df0fc4ba28ded89e16e0b5d95d310a Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:53:14 -0700 Subject: [PATCH 3/7] Fix: use one-time initial phase offset instead of per-poll jitter reduction Terra re-review flagged that the previous downward-only jitter design (range [0.8s, 1.0s], mean 0.9s) increases steady-state polling volume by ~11% versus the historical fixed 1s cadence, undermining the load-reduction goal of #776. It also flagged that the Stopwatch-based upper-bound integration test was CI-flaky, since Task.Delay only guarantees a lower bound on elapsed time, not an upper one. Replace the per-iteration downward jitter with a one-time randomized initial phase offset: only the first delay of a given WaitForInstanceStartAsync call is randomized (uniformly in [0, 1s)); every delay after that is the fixed historical 1-second interval, completely unjittered. This still desynchronizes concurrent callers (avoiding synchronized polling bursts) via the one-time phase difference, but no longer inflates steady-state polling volume, since subsequent delays exactly match the historical cadence forever. The worst-case detection latency for any single status transition still never exceeds the historical 1 second. Replace the flaky Stopwatch-based test with two deterministic unit tests against the (internal) delay-computation policy directly: - ComputeNextPollingDelay_InitialDelay_NeverExceedsHistoricalOneSecondCadence asserts 1000 samples of the initial-phase delay are always in [0, 1s). - ComputeNextPollingDelay_SteadyState_ReturnsFixedHistoricalIntervalWithNoJitter asserts every subsequent delay is exactly 1 second, with no jitter and no growth -- proving steady-state polling volume cannot regress in either direction. No public API change; ComputeNextPollingDelay remains an internal-only member of the internal ShimDurableTaskClient class. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../ShimDurableTaskClient.cs | 54 +++++++++----- .../ShimDurableTaskClientTests.cs | 70 ++++++------------- 2 files changed, 57 insertions(+), 67 deletions(-) diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs index 3af7ea4c..2e1f6589 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs @@ -29,11 +29,13 @@ namespace Microsoft.DurableTask.Client.OrchestrationServiceClientShim; class ShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) : DurableTaskClient(name) { // Polling parameters for WaitForInstanceStartAsync. PollingInterval matches the historical fixed - // 1-second polling cadence. Callers rely on prompt (<= 1 second) observation of a status - // transition, so jitter is only ever applied *downward* from this value -- never grown beyond it - // -- to desynchronize concurrent callers (avoiding synchronized polling bursts against the - // backend) without regressing the historical worst-case detection latency. - const double PollingJitterFactor = 0.2; + // 1-second polling cadence and is used, unjittered, for every steady-state delay -- so long-run + // polling volume never exceeds the historical rate. To desynchronize concurrent callers (avoiding + // synchronized polling bursts against the backend) without inflating that steady-state volume, a + // randomized *initial phase offset* -- uniformly distributed in [0, PollingInterval) -- is applied + // exactly once, before the first delay of a given WaitForInstanceStartAsync call. This is a one-time + // cost per call (not repeated per iteration), so it does not change the long-run polling rate, and + // it still never exceeds the historical 1-second worst-case detection latency. static readonly TimeSpan PollingInterval = TimeSpan.FromSeconds(1); readonly ShimDurableTaskClientOptions options = Check.NotNull(options); @@ -279,6 +281,10 @@ public override async Task WaitForInstanceStartAsync( { Check.NotNullOrEmpty(instanceId); + // A one-time randomized phase offset (see ComputeNextPollingDelay) is applied only to the first + // delay of this call so concurrent waiters desynchronize without increasing steady-state + // polling volume beyond the historical fixed 1-second cadence. + bool isInitialDelay = true; while (true) { OrchestrationMetadata? metadata = await this.GetInstancesAsync( @@ -294,11 +300,13 @@ public override async Task WaitForInstanceStartAsync( return metadata; } - // Poll with a delay that is jittered *downward* from the historical 1-second cadence. - // This desynchronizes concurrent waiters (avoiding synchronized polling bursts against the - // backend) while guaranteeing the worst-case detection latency for a status transition - // never exceeds the historical 1-second cadence -- preserving prompt-start observation. - TimeSpan delay = ComputeNextPollingDelay(); + // The first delay is a randomized phase offset (bounded by the historical 1-second + // cadence) that desynchronizes concurrent waiters; every delay after that is the fixed + // historical 1-second interval, unjittered, so steady-state polling volume never exceeds + // the historical rate. Either way, the delay never exceeds 1 second, preserving prompt-start + // observation. + TimeSpan delay = ComputeNextPollingDelay(isInitialDelay); + isInitialDelay = false; await Task.Delay(delay, cancellation); } } @@ -361,17 +369,27 @@ await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync( /// /// Computes the delay to wait before the next polling attempt. /// + /// + /// if this is the first delay computed for a given call; for every subsequent delay in + /// that call. + /// /// - /// A delay in the range [ * (1 - ), - /// ]. Jitter only ever reduces the delay, so the returned value never - /// exceeds -- preserving the historical worst-case detection latency - /// for -- while still desynchronizing concurrent callers. + /// When is , a one-time randomized phase + /// offset uniformly distributed in [, ) that + /// desynchronizes concurrent callers. Otherwise, the fixed (1 second), + /// unjittered, so steady-state polling volume never exceeds the historical rate. In both cases the + /// returned delay never exceeds , preserving the historical worst-case + /// detection latency for . /// - internal static TimeSpan ComputeNextPollingDelay() + internal static TimeSpan ComputeNextPollingDelay(bool isInitialDelay) { - double reduction = PollingJitterFactor * PollingJitter.NextDouble(); - double delayMs = PollingInterval.TotalMilliseconds * (1.0 - reduction); - return TimeSpan.FromMilliseconds(delayMs); + if (isInitialDelay) + { + return TimeSpan.FromMilliseconds(PollingInterval.TotalMilliseconds * PollingJitter.NextDouble()); + } + + return PollingInterval; } [return: NotNullIfNotNull("state")] diff --git a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs index 344c3c15..31be7c5b 100644 --- a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs +++ b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using System.Diagnostics; using DurableTask.Core; using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; @@ -330,66 +329,39 @@ public async Task WaitForInstanceStart_MultiplePendingPolls_EventuallyReturnsTer OrchestrationMetadata metadata = await this.client.WaitForInstanceStartAsync( instance.InstanceId, false, default); - // assert -- multiple polling iterations (exercising the jittered polling loop) still converge - // on the terminal state once observed. + // assert -- multiple polling iterations (exercising the polling loop, including its one-time + // initial phase offset followed by fixed-interval delays) still converge on the terminal state + // once observed. this.orchestrationClient.Verify( m => m.GetOrchestrationStateAsync(instance.InstanceId, false), Times.Exactly(3)); Validate(metadata, terminal, false); } [Fact] - public async Task WaitForInstanceStart_RepeatedPendingObservations_HonorsMaxDetectionLatencyContract() + public void ComputeNextPollingDelay_InitialDelay_NeverExceedsHistoricalOneSecondCadence() { - // arrange - DateTimeOffset start = DateTimeOffset.UtcNow; - OrchestrationInstance instance = new() + // assert -- across many samples, the randomized initial-phase-offset delay must always be + // non-negative and never exceed the historical 1-second polling cadence, enforcing the max + // detection-latency contract for the very first delay of a WaitForInstanceStartAsync call. + for (int i = 0; i < 1000; i++) { - InstanceId = Guid.NewGuid().ToString(), - ExecutionId = Guid.NewGuid().ToString(), - }; - - Core.OrchestrationState pending1 = CreateState("input", start: start); - pending1.OrchestrationInstance = instance; - pending1.OrchestrationStatus = Core.OrchestrationStatus.Pending; - Core.OrchestrationState pending2 = CreateState("input", start: start); - pending2.OrchestrationInstance = instance; - pending2.OrchestrationStatus = Core.OrchestrationStatus.Pending; - Core.OrchestrationState terminal = CreateState("input", start: start); - terminal.OrchestrationInstance = instance; - - this.orchestrationClient.SetupSequence(m => m.GetOrchestrationStateAsync(instance.InstanceId, false)) - .ReturnsAsync([pending1]) - .ReturnsAsync([pending2]) - .ReturnsAsync([terminal]); - - Stopwatch stopwatch = Stopwatch.StartNew(); - - // act - OrchestrationMetadata metadata = await this.client.WaitForInstanceStartAsync( - instance.InstanceId, false, default); - - stopwatch.Stop(); - - // assert -- two "still pending" observations occur before the terminal state is returned, so - // exactly two polling delays are incurred. Each delay is bounded by the historical 1-second - // detection cadence (jitter only ever reduces it), so the worst-case total wait here is ~2 - // seconds. This guards against a regression -- like unconstrained backoff growth -- that would - // push per-poll delays past the historical cadence and violate prompt-start observation. - stopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(2.5)); - Validate(metadata, terminal, false); + TimeSpan delay = ShimDurableTaskClient.ComputeNextPollingDelay(isInitialDelay: true); + delay.Should().BeGreaterThanOrEqualTo(TimeSpan.Zero); + delay.Should().BeLessThan(TimeSpan.FromSeconds(1)); + } } [Fact] - public void ComputeNextPollingDelay_NeverExceedsHistoricalOneSecondCadence() + public void ComputeNextPollingDelay_SteadyState_ReturnsFixedHistoricalIntervalWithNoJitter() { - // assert -- across many samples, the jittered delay must never exceed the historical 1-second - // polling cadence (jitter only ever reduces the delay), enforcing the max detection-latency - // contract for WaitForInstanceStartAsync, and must always be a non-negative, finite delay. - for (int i = 0; i < 1000; i++) + // assert -- every delay after the initial phase offset must be exactly the fixed historical + // 1-second cadence with no jitter and no growth, so steady-state polling volume never exceeds + // (or falls below) the historical rate. This guards against a regression to either unconstrained + // backoff growth or a per-iteration jitter reduction that would increase polling frequency. + for (int i = 0; i < 100; i++) { - TimeSpan delay = ShimDurableTaskClient.ComputeNextPollingDelay(); - delay.Should().BeGreaterThan(TimeSpan.Zero); - delay.Should().BeLessThanOrEqualTo(TimeSpan.FromSeconds(1)); + TimeSpan delay = ShimDurableTaskClient.ComputeNextPollingDelay(isInitialDelay: false); + delay.Should().Be(TimeSpan.FromSeconds(1)); } } @@ -432,7 +404,7 @@ public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCan using CancellationTokenSource cts = new(); // act -- cancel shortly after the first (immediate) poll so cancellation fires while the loop is - // awaiting the jittered polling delay rather than before any polling occurs. + // awaiting the polling delay rather than before any polling occurs. cts.CancelAfter(TimeSpan.FromMilliseconds(50)); Func act = () => this.client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); From 78d527fc5d479897c02efbc7acf9ad73e54c695e Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:03:12 -0700 Subject: [PATCH 4/7] Fix: make cancellation test deterministic via internal delay seam Replace wall-clock cts.CancelAfter(50ms) coordination in WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCanceledException with deterministic coordination: an internal virtual DelayAsync seam on ShimDurableTaskClient lets the test observe (via a TaskCompletionSource) the exact moment the real Task.Delay(delay, cancellation) call for the polling loop has been made, and only then cancels the token. The test also asserts GetOrchestrationStateAsync was called exactly once, proving cancellation ended the wait during the delay itself rather than via a subsequent poll happening to observe an already-cancelled token. No production behavior change: DelayAsync defaults to Task.Delay and is internal (not public API surface), so no breaking-change-check is required. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../ShimDurableTaskClient.cs | 16 ++++++- .../ShimDurableTaskClientTests.cs | 48 ++++++++++++++++--- 2 files changed, 57 insertions(+), 7 deletions(-) diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs index 2e1f6589..642061fa 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs @@ -307,7 +307,7 @@ public override async Task WaitForInstanceStartAsync( // observation. TimeSpan delay = ComputeNextPollingDelay(isInitialDelay); isInitialDelay = false; - await Task.Delay(delay, cancellation); + await this.DelayAsync(delay, cancellation); } } @@ -392,6 +392,20 @@ internal static TimeSpan ComputeNextPollingDelay(bool isInitialDelay) return PollingInterval; } + /// + /// Awaits the delay between polling attempts. + /// + /// + /// This is factored out from a direct call + /// purely as an internal seam: it lets tests deterministically observe (and coordinate around) the + /// moment a polling delay begins -- e.g. to cancel only once the delay is genuinely in progress -- + /// without relying on wall-clock timing assumptions. It does not change production behavior. + /// + /// The delay to await. + /// The cancellation token to honor while awaiting the delay. + /// A task that completes after the delay elapses, or is cancelled via . + internal virtual Task DelayAsync(TimeSpan delay, CancellationToken cancellation) => Task.Delay(delay, cancellation); + [return: NotNullIfNotNull("state")] OrchestrationMetadata? ToMetadata(Core.OrchestrationState? state, bool getInputsAndOutputs) { diff --git a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs index 31be7c5b..24a03071 100644 --- a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs +++ b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs @@ -402,14 +402,50 @@ public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCan .ReturnsAsync([pending]); using CancellationTokenSource cts = new(); + DelayObservingShimDurableTaskClient client = new( + "test", new ShimDurableTaskClientOptions { Client = this.orchestrationClient.Object }); + + // act -- deterministically coordinate cancellation with the polling delay itself (no wall-clock + // guess): wait until the delay seam confirms the real Task.Delay(delay, cancellation) call has + // been made for this loop iteration, THEN cancel. This proves the delay -- not some other code + // path such as a later poll observing an already-cancelled token -- is what ends the wait. + Task waitTask = client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); + await client.DelayEntered.WaitAsync(TimeSpan.FromSeconds(10)); + cts.Cancel(); + + Task completedTask = await Task.WhenAny(waitTask, Task.Delay(TimeSpan.FromSeconds(10))); + + // assert -- the wait must end (via cancellation) promptly, without ever performing a second poll. + // If cancellation were ignored by the delay, the loop would instead complete the full delay and + // poll again (and again, since the mock always returns "pending"), so this also guards against + // that regression by bounding how long the assertion waits before failing. + completedTask.Should().Be(waitTask, "the wait should be cancelled during the polling delay, not time out"); + Func act = () => waitTask; + await act.Should().ThrowAsync(); + this.orchestrationClient.Verify( + m => m.GetOrchestrationStateAsync(instance.InstanceId, false), Times.Once); + } - // act -- cancel shortly after the first (immediate) poll so cancellation fires while the loop is - // awaiting the polling delay rather than before any polling occurs. - cts.CancelAfter(TimeSpan.FromMilliseconds(50)); - Func act = () => this.client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); + /// + /// A test double that signals once the + /// real polling delay () has actually been invoked with + /// the caller's cancellation token, so tests can deterministically coordinate cancellation without + /// relying on wall-clock timing. The underlying delay/cancellation behavior is otherwise unchanged. + /// + sealed class DelayObservingShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) + : ShimDurableTaskClient(name, options) + { + readonly TaskCompletionSource delayEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); - // assert - await act.Should().ThrowAsync(); + /// Gets a task that completes once has been called. + public Task DelayEntered => this.delayEntered.Task; + + internal override Task DelayAsync(TimeSpan delay, CancellationToken cancellation) + { + Task delayTask = base.DelayAsync(delay, cancellation); + this.delayEntered.TrySetResult(); + return delayTask; + } } [Fact] From 3a74e2dc12cd692642ff38c3170807df410d2708 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:10:41 -0700 Subject: [PATCH 5/7] Make cancellation test fully prove token propagation into DelayAsync Terra's fourth review found the previous fix still insufficient: the test double called the real Task.Delay(delay, cancellation) and let it run, so a regression that passed the wrong (or no) token into DelayAsync could still 'pass' the test -- GetInstancesAsync's own cancellation.ThrowIfCancellationRequested() check would throw on the next poll attempt before the mock is invoked again, satisfying the Times.Once assertion even though the delay itself never honored cancellation. Replace the fake delay with one that never completes on its own (no real timer) and completes -- via cancellation -- only when the exact CancellationToken instance passed by the caller is cancelled. This directly proves that cancelling the caller's token is what ends the wait, independent of any other cancellation check in the poll loop. Verified by temporarily reintroducing the regression (passing CancellationToken.None into DelayAsync at the call site) and confirming the test now fails deterministically (times out at the 10s safety net) instead of passing vacuously; reverted after confirming. No production behavior change -- test-only fix. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../ShimDurableTaskClientTests.cs | 31 ++++++++++++------- 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs index 24a03071..ecdf29ae 100644 --- a/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs +++ b/test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs @@ -406,9 +406,11 @@ public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCan "test", new ShimDurableTaskClientOptions { Client = this.orchestrationClient.Object }); // act -- deterministically coordinate cancellation with the polling delay itself (no wall-clock - // guess): wait until the delay seam confirms the real Task.Delay(delay, cancellation) call has - // been made for this loop iteration, THEN cancel. This proves the delay -- not some other code - // path such as a later poll observing an already-cancelled token -- is what ends the wait. + // guess, no real timer): wait until the fake delay seam confirms it has been entered for this + // loop iteration, THEN cancel. The fake delay never completes on its own -- it only completes + // when the *exact* cancellation token passed by production code is cancelled -- so this proves + // that token, and not some other code path (e.g. a later poll's own cancellation check), is what + // ends the wait. Task waitTask = client.WaitForInstanceStartAsync(instance.InstanceId, false, cts.Token); await client.DelayEntered.WaitAsync(TimeSpan.FromSeconds(10)); cts.Cancel(); @@ -416,9 +418,9 @@ public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCan Task completedTask = await Task.WhenAny(waitTask, Task.Delay(TimeSpan.FromSeconds(10))); // assert -- the wait must end (via cancellation) promptly, without ever performing a second poll. - // If cancellation were ignored by the delay, the loop would instead complete the full delay and - // poll again (and again, since the mock always returns "pending"), so this also guards against - // that regression by bounding how long the assertion waits before failing. + // Because the fake delay never completes unless the supplied token is cancelled, a regression that + // passes the wrong (or no) token into the delay would leave the delay -- and thus the wait -- + // pending forever, causing this assertion to time out and fail instead of passing vacuously. completedTask.Should().Be(waitTask, "the wait should be cancelled during the polling delay, not time out"); Func act = () => waitTask; await act.Should().ThrowAsync(); @@ -427,10 +429,14 @@ public async Task WaitForInstanceStart_CancelledDuringPollingDelay_ThrowsTaskCan } /// - /// A test double that signals once the - /// real polling delay () has actually been invoked with - /// the caller's cancellation token, so tests can deterministically coordinate cancellation without - /// relying on wall-clock timing. The underlying delay/cancellation behavior is otherwise unchanged. + /// A test double whose override replaces + /// the real polling delay with a fully controlled fake: it signals as soon + /// as it is called, then returns a task that never completes on its own (no real timer) and completes + /// -- via cancellation -- only when the *exact* supplied by the caller + /// is cancelled. This lets tests deterministically coordinate cancellation with the delay without any + /// wall-clock timing, and proves that cancelling the caller's token is what actually interrupts the + /// pending wait, rather than some unrelated code path (such as a subsequent poll's own cancellation + /// check) coincidentally producing the same observable outcome. /// sealed class DelayObservingShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) : ShimDurableTaskClient(name, options) @@ -442,9 +448,10 @@ sealed class DelayObservingShimDurableTaskClient(string name, ShimDurableTaskCli internal override Task DelayAsync(TimeSpan delay, CancellationToken cancellation) { - Task delayTask = base.DelayAsync(delay, cancellation); + TaskCompletionSource pending = new(TaskCreationOptions.RunContinuationsAsynchronously); + cancellation.Register(() => pending.TrySetCanceled(cancellation)); this.delayEntered.TrySetResult(); - return delayTask; + return pending.Task; } } From c8b66aa6f2f98d664fc75f870a73c9a64efb6fb3 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 18:13:25 -0700 Subject: [PATCH 6/7] Stabilize PerItem_HeartbeatReset_KeepsTimerAlive against CI scheduling flake This test is unrelated to the shim polling work in this PR, but was blocking CI on this branch with a reproducible timeout under scheduling pressure. Root cause: the test slept 150ms (a fixed wall-clock delay) before writing the first channel item, racing against the consumer's 500ms silent-disconnect timer which is armed as soon as ConsumeAsync starts. Under CI scheduling pressure that 150ms sleep could take long enough to let the timer fire before the first item was ever written, causing the test to hang waiting for a signal that would never arrive. Fix (test-only, no production behavior change): write the first item immediately with no a-priori sleep, eliminating that race entirely. To keep a strong regression signal for a missing per-item timer reset, the test now sends several items in sequence with gaps measured from each item's actual processing (via a semaphore signal, not a wall-clock guess): each individual gap (150ms) is comfortably below the 500ms timeout, but their sum (600ms) comfortably exceeds it -- so a regression that only arms the timer once at loop start would fail this test well before the last item is sent. Verified the fix still catches the regression: temporarily removed the per-item ArmSilentDisconnectTimer() call in WorkItemStreamConsumer.cs and confirmed the test failed deterministically; reverted (zero production diff). Ran the test 10x standalone (stable, ~660ms each) and the full Worker.Grpc.Tests suite (137/137 pass). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- .../Grpc.Tests/WorkItemStreamConsumerTests.cs | 52 ++++++++++++------- 1 file changed, 34 insertions(+), 18 deletions(-) diff --git a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs index 2464c0c7..a6bb101b 100644 --- a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs +++ b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs @@ -139,34 +139,50 @@ public async Task OuterCancellation_WithRpcCancelledFromStream_PropagatesExcepti [Fact] public async Task PerItem_HeartbeatReset_KeepsTimerAlive() { - // Feed one item, wait long enough that the original timer would have expired, then complete. - // Synchronize on the first item actually being processed so the second delay is measured from - // the consumer's timer reset instead of from the test thread's write timing. + // Proves the per-item timer reset -- not just a single arm at loop start -- is what keeps the + // stream alive. Several items are sent in sequence: each gap between consecutive items is + // comfortably shorter than the silent-disconnect timeout (so a correct per-item reset never lets + // the timer expire), but the gaps' sum comfortably exceeds the timeout (so a regression that only + // arms the timer once at loop start, and never re-arms it per item, would already have cancelled + // the stream well before the last item is sent). + // + // Every gap is measured starting from the previous item's actual onItem invocation -- signalled + // via a semaphore -- rather than from an a-priori sleep before the very first item. That avoids a + // CI flake where scheduling pressure before the read loop has even started could delay the first + // write past the timeout and spuriously trip a SilentDisconnect that has nothing to do with the + // per-item reset behavior under test. Channel channel = Channel.CreateUnbounded(); TimeSpan timeout = TimeSpan.FromMilliseconds(500); - TaskCompletionSource firstItemProcessed = new(TaskCreationOptions.RunContinuationsAsynchronously); - int itemCount = 0; + TimeSpan perItemGap = TimeSpan.FromMilliseconds(150); + const int itemCount = 5; // 4 gaps * 150ms = 600ms > 500ms timeout: proves reset is required. + + SemaphoreSlim itemProcessed = new(0); Task consumeTask = WorkItemStreamConsumer.ConsumeAsync( openStream: ct => channel.Reader.ReadAllAsync(ct), silentDisconnectTimeout: timeout, - onItem: _ => - { - if (Interlocked.Increment(ref itemCount) == 1) - { - firstItemProcessed.TrySetResult(); - } - }, + onItem: _ => itemProcessed.Release(), onFirstMessage: null, cancellation: CancellationToken.None); - await Task.Delay(TimeSpan.FromMilliseconds(150)); - await channel.Writer.WriteAsync(new P.WorkItem { HealthPing = new P.HealthPing() }); - await firstItemProcessed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + for (int i = 0; i < itemCount; i++) + { + if (i > 0) + { + bool signaled = await itemProcessed.WaitAsync(TimeSpan.FromSeconds(5)); + signaled.Should().BeTrue("item {0} should have been processed (re-arming the timer) within the bounded wait", i); + + await Task.Delay(perItemGap); + } + + await channel.Writer.WriteAsync(new P.WorkItem { HealthPing = new P.HealthPing() }); + } + + // Wait for the final item to be processed before completing the channel, so the last per-item + // reset has actually happened prior to the graceful drain. + bool finalItemSignaled = await itemProcessed.WaitAsync(TimeSpan.FromSeconds(5)); + finalItemSignaled.Should().BeTrue("the final item should have been processed before the stream completes"); - // Without the per-item reset, the original timer would fire before this second item arrives. - await Task.Delay(TimeSpan.FromMilliseconds(400)); - await channel.Writer.WriteAsync(new P.WorkItem { HealthPing = new P.HealthPing() }); channel.Writer.Complete(); WorkItemStreamResult result = await consumeTask; From 8779f982fed8ce25a3d573dedd615c5855058142 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 18:26:26 -0700 Subject: [PATCH 7/7] Make WorkItemStreamConsumer per-item timer reset test fully deterministic The previous fix for the WorkItemStreamConsumerTests CI flake still relied on real per-item delays (150ms each, comfortably under the 500ms timeout) racing against the real silent-disconnect timer. Under CI scheduling pressure a delayed continuation between the "item processed" signal and the next write could still inflate an intended-short gap past the timeout, even though production behavior was correct. Add a test-only observability seam to WorkItemStreamConsumer.ConsumeAsync: an optional onSilentDisconnectTimerArmed callback invoked synchronously every time the silent-disconnect timer is (re-)armed (once before the read loop, once per item). It is null in production (default parameter, purely additive, zero behavior change when unused) and only appended as a trailing optional parameter, so the sole production call site is unaffected. Rewrite PerItem_HeartbeatReset_KeepsTimerAlive to use this seam instead of wall-clock timing: it records the exact interleaving of "armed" and "item" events and asserts the structural invariant directly (one arm before the loop, one re-arm immediately before each item is dispatched) rather than inferring it from elapsed real time. The test now runs in ~20ms with zero timing dependency. Verified red/green: temporarily commented out the per-item ArmSilentDisconnectTimer() call in production, confirmed the test failed deterministically (armed count 1 instead of 6), then reverted (git diff confirms only the intended seam addition remains). Ran the fixed test 15x standalone (all pass, ~20ms each), the full Worker.Grpc.Tests suite (137/137), and the full Client.OrchestrationServiceClientShim.Tests suite (76/76, confirms the core PR content is unaffected). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8aa98c91-6050-44ef-b236-dd1cf43fabac --- src/Worker/Grpc/WorkItemStreamConsumer.cs | 12 ++- .../Grpc.Tests/WorkItemStreamConsumerTests.cs | 82 +++++++++---------- 2 files changed, 51 insertions(+), 43 deletions(-) diff --git a/src/Worker/Grpc/WorkItemStreamConsumer.cs b/src/Worker/Grpc/WorkItemStreamConsumer.cs index cce0b23a..9c39ef7b 100644 --- a/src/Worker/Grpc/WorkItemStreamConsumer.cs +++ b/src/Worker/Grpc/WorkItemStreamConsumer.cs @@ -60,13 +60,22 @@ internal static class WorkItemStreamConsumer /// reset retry counters that should only count consecutive transport failures. /// /// Outer worker cancellation token. + /// + /// Test-only observability hook invoked synchronously every time the silent-disconnect timer is + /// (re-)armed -- once before the read loop starts, and once per item, immediately before that + /// item is dispatched. Always in production; lets tests prove the + /// per-item reset actually happens (and in what order relative to dispatch) without depending on + /// real elapsed time. Never invoked when disables + /// detection. + /// /// The classified outcome plus whether any message was observed. public static async Task ConsumeAsync( Func> openStream, TimeSpan silentDisconnectTimeout, Action onItem, Action? onFirstMessage, - CancellationToken cancellation) + CancellationToken cancellation, + Action? onSilentDisconnectTimerArmed = null) { bool silentDisconnectEnabled = silentDisconnectTimeout > TimeSpan.Zero; @@ -79,6 +88,7 @@ void ArmSilentDisconnectTimer() if (silentDisconnectEnabled) { timeoutSource.CancelAfter(effectiveTimeout); + onSilentDisconnectTimerArmed?.Invoke(); } } diff --git a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs index a6bb101b..ba9104c1 100644 --- a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs +++ b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs @@ -2,7 +2,6 @@ // Licensed under the MIT License. using System.Runtime.CompilerServices; -using System.Threading.Channels; using Grpc.Core; using Microsoft.DurableTask.Worker.Grpc; using P = Microsoft.DurableTask.Protobuf; @@ -140,55 +139,54 @@ public async Task OuterCancellation_WithRpcCancelledFromStream_PropagatesExcepti public async Task PerItem_HeartbeatReset_KeepsTimerAlive() { // Proves the per-item timer reset -- not just a single arm at loop start -- is what keeps the - // stream alive. Several items are sent in sequence: each gap between consecutive items is - // comfortably shorter than the silent-disconnect timeout (so a correct per-item reset never lets - // the timer expire), but the gaps' sum comfortably exceeds the timeout (so a regression that only - // arms the timer once at loop start, and never re-arms it per item, would already have cancelled - // the stream well before the last item is sent). + // stream alive. Earlier versions of this test tried to prove the reset by racing real per-item + // delays (each comfortably under the timeout) against the real silent-disconnect timeout (so + // their sum comfortably exceeded it). That was still flaky under CI scheduling pressure: any + // continuation between the "item processed" signal and the next write could be delayed by the + // thread pool/scheduler, silently inflating an intended-short gap past the timeout even though + // production was correct. // - // Every gap is measured starting from the previous item's actual onItem invocation -- signalled - // via a semaphore -- rather than from an a-priori sleep before the very first item. That avoids a - // CI flake where scheduling pressure before the read loop has even started could delay the first - // write past the timeout and spuriously trip a SilentDisconnect that has nothing to do with the - // per-item reset behavior under test. - Channel channel = Channel.CreateUnbounded(); - TimeSpan timeout = TimeSpan.FromMilliseconds(500); - TimeSpan perItemGap = TimeSpan.FromMilliseconds(150); - const int itemCount = 5; // 4 gaps * 150ms = 600ms > 500ms timeout: proves reset is required. - - SemaphoreSlim itemProcessed = new(0); - - Task consumeTask = WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => channel.Reader.ReadAllAsync(ct), - silentDisconnectTimeout: timeout, - onItem: _ => itemProcessed.Release(), - onFirstMessage: null, - cancellation: CancellationToken.None); - + // This version removes wall-clock timing from the assertion entirely. ConsumeAsync exposes a + // test-only observability hook that fires every time the silent-disconnect timer is (re-)armed: + // once before the read loop starts, and once per item, immediately before that item is + // dispatched to onItem. By recording the exact interleaving of "armed" and "item" events, the + // test proves the structural guarantee directly -- an arm precedes every item, and the total arm + // count is itemCount + 1 -- instead of inferring it from elapsed real time. A regression that + // only arms the timer once at loop start (and never re-arms it per item) fails this assertion + // deterministically, with no dependency on scheduler timing. + const int itemCount = 5; + List events = new(); + int itemIndex = 0; + + P.WorkItem[] items = new P.WorkItem[itemCount]; for (int i = 0; i < itemCount; i++) { - if (i > 0) - { - bool signaled = await itemProcessed.WaitAsync(TimeSpan.FromSeconds(5)); - signaled.Should().BeTrue("item {0} should have been processed (re-arming the timer) within the bounded wait", i); - - await Task.Delay(perItemGap); - } - - await channel.Writer.WriteAsync(new P.WorkItem { HealthPing = new P.HealthPing() }); + items[i] = new P.WorkItem { HealthPing = new P.HealthPing() }; } - // Wait for the final item to be processed before completing the channel, so the last per-item - // reset has actually happened prior to the graceful drain. - bool finalItemSignaled = await itemProcessed.WaitAsync(TimeSpan.FromSeconds(5)); - finalItemSignaled.Should().BeTrue("the final item should have been processed before the stream completes"); - - channel.Writer.Complete(); - - WorkItemStreamResult result = await consumeTask; + WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( + openStream: _ => StreamOf(items), + silentDisconnectTimeout: TimeSpan.FromMilliseconds(500), + onItem: _ => events.Add($"item{itemIndex++}"), + onFirstMessage: null, + cancellation: CancellationToken.None, + onSilentDisconnectTimerArmed: () => events.Add("armed")); result.Outcome.Should().Be(WorkItemStreamOutcome.GracefulDrain); result.FirstMessageObserved.Should().BeTrue(); + + // 1 initial arm (before the loop starts) + 1 re-arm per item. + events.Count(e => e == "armed").Should().Be(itemCount + 1); + + // Every item must be immediately preceded by its own re-arm, and the very first event overall + // is the initial pre-loop arm. + events[0].Should().Be("armed"); + for (int i = 0; i < itemCount; i++) + { + int armedIndex = 1 + (i * 2); + events[armedIndex].Should().Be("armed", "item {0} must be preceded by a timer re-arm", i); + events[armedIndex + 1].Should().Be($"item{i}"); + } } [Fact]