Reduce shim instance-start polling load with bounded jitter

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,16 @@ namespace Microsoft.DurableTask.Client.OrchestrationServiceClientShim;
 /// <param name="options">The client options.</param>
 class ShimDurableTaskClient(string name, ShimDurableTaskClientOptions options) : DurableTaskClient(name)
 {
+    // Polling parameters for WaitForInstanceStartAsync. PollingInterval matches the historical fixed
+    // 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);
     ShimDurableEntityClient? entities;
 
@@ -270,6 +281,10 @@ public override async Task<OrchestrationMetadata> 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(
@@ -285,7 +300,14 @@ public override async Task<OrchestrationMetadata> WaitForInstanceStartAsync(
                 return metadata;
             }
 
-            await Task.Delay(TimeSpan.FromSeconds(1), cancellation);
+            // 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 this.DelayAsync(delay, cancellation);
         }
     }
 
@@ -344,6 +366,46 @@ await this.TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
         return newInstanceId;
     }
 
+    /// <summary>
+    /// Computes the delay to wait before the next <see cref="WaitForInstanceStartAsync"/> polling attempt.
+    /// </summary>
+    /// <param name="isInitialDelay">
+    /// <see langword="true"/> if this is the first delay computed for a given <see
+    /// cref="WaitForInstanceStartAsync"/> call; <see langword="false"/> for every subsequent delay in
+    /// that call.
+    /// </param>
+    /// <returns>
+    /// When <paramref name="isInitialDelay"/> is <see langword="true"/>, a one-time randomized phase
+    /// offset uniformly distributed in [<see cref="TimeSpan.Zero"/>, <see cref="PollingInterval"/>) that
+    /// desynchronizes concurrent callers. Otherwise, the fixed <see cref="PollingInterval"/> (1 second),
+    /// unjittered, so steady-state polling volume never exceeds the historical rate. In both cases the
+    /// returned delay never exceeds <see cref="PollingInterval"/>, preserving the historical worst-case
+    /// detection latency for <see cref="WaitForInstanceStartAsync"/>.
+    /// </returns>
+    internal static TimeSpan ComputeNextPollingDelay(bool isInitialDelay)
+    {
+        if (isInitialDelay)
+        {
+            return TimeSpan.FromMilliseconds(PollingInterval.TotalMilliseconds * PollingJitter.NextDouble());
+        }
+
+        return PollingInterval;
+    }
+
+    /// <summary>
+    /// Awaits the delay between <see cref="WaitForInstanceStartAsync"/> polling attempts.
+    /// </summary>
+    /// <remarks>
+    /// This is factored out from a direct <see cref="Task.Delay(TimeSpan, CancellationToken)"/> 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.
+    /// </remarks>
+    /// <param name="delay">The delay to await.</param>
+    /// <param name="cancellation">The cancellation token to honor while awaiting the delay.</param>
+    /// <returns>A task that completes after the delay elapses, or is cancelled via <paramref name="cancellation"/>.</returns>
+    internal virtual Task DelayAsync(TimeSpan delay, CancellationToken cancellation) => Task.Delay(delay, cancellation);
+
     [return: NotNullIfNotNull("state")]
     OrchestrationMetadata? ToMetadata(Core.OrchestrationState? state, bool getInputsAndOutputs)
     {
@@ -449,4 +511,37 @@ async Task TerminateTaskOrchestrationWithReusableRunningStatusAndWaitAsync(
             }
         }
     }
+
+    /// <summary>
+    /// A minimal thread-safe random source used to jitter <see cref="WaitForInstanceStartAsync"/> polling
+    /// delays across concurrent callers. <see cref="System.Random"/> 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.
+    /// </summary>
+    static class PollingJitter
+    {
+        static readonly object SyncRoot = new();
+        static readonly Random Shared = CreateSeededRandom();
+
+        /// <summary>
+        /// Returns a thread-safe random double in the range [0.0, 1.0).
+        /// </summary>
+        /// <returns>A random double in the range [0.0, 1.0).</returns>
+        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));
+        }
+    }
 }

src/Worker/Grpc/WorkItemStreamConsumer.cs

@@ -60,13 +60,22 @@ internal static class WorkItemStreamConsumer
     /// reset retry counters that should only count consecutive transport failures.
     /// </param>
     /// <param name="cancellation">Outer worker cancellation token.</param>
+    /// <param name="onSilentDisconnectTimerArmed">
+    /// 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 <see langword="null"/> 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 <paramref name="silentDisconnectTimeout"/> disables
+    /// detection.
+    /// </param>
     /// <returns>The classified outcome plus whether any message was observed.</returns>
     public static async Task<WorkItemStreamResult> ConsumeAsync(
         Func<CancellationToken, IAsyncEnumerable<P.WorkItem>> openStream,
         TimeSpan silentDisconnectTimeout,
         Action<P.WorkItem> 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();
             }
         }
 

test/Client/OrchestrationServiceClientShim.Tests/ShimDurableTaskClientTests.cs

@@ -300,6 +300,161 @@ 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 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 void ComputeNextPollingDelay_InitialDelay_NeverExceedsHistoricalOneSecondCadence()
+    {
+        // 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++)
+        {
+            TimeSpan delay = ShimDurableTaskClient.ComputeNextPollingDelay(isInitialDelay: true);
+            delay.Should().BeGreaterThanOrEqualTo(TimeSpan.Zero);
+            delay.Should().BeLessThan(TimeSpan.FromSeconds(1));
+        }
+    }
+
+    [Fact]
+    public void ComputeNextPollingDelay_SteadyState_ReturnsFixedHistoricalIntervalWithNoJitter()
+    {
+        // 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(isInitialDelay: false);
+            delay.Should().Be(TimeSpan.FromSeconds(1));
+        }
+    }
+
+    [Fact]
+    public async Task WaitForInstanceStart_InstanceNotFound_ThrowsImmediatelyWithoutPolling()
+    {
+        // arrange
+        string instanceId = Guid.NewGuid().ToString();
+        this.orchestrationClient.Setup(m => m.GetOrchestrationStateAsync(instanceId, false))
+            .ReturnsAsync([]);
+
+        // act
+        Func<Task> act = () => this.client.WaitForInstanceStartAsync(instanceId, false, default);
+
+        // assert -- not-found behavior is preserved: no retry/polling delay before throwing.
+        await act.Should().ThrowExactlyAsync<InvalidOperationException>()
+            .WithMessage($"Orchestration with instanceId '{instanceId}' does not exist");
+        this.orchestrationClient.Verify(
+            m => m.GetOrchestrationStateAsync(instanceId, false), Times.Once);
+    }
+
+    [Fact]
+    public async Task WaitForInstanceStart_CancelledDuringPollingDelay_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();
+        DelayObservingShimDurableTaskClient client = new(
+            "test", new ShimDurableTaskClientOptions { Client = this.orchestrationClient.Object });
+
+        // act -- deterministically coordinate cancellation with the polling delay itself (no wall-clock
+        // 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();
+
+        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.
+        // 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<Task> act = () => waitTask;
+        await act.Should().ThrowAsync<OperationCanceledException>();
+        this.orchestrationClient.Verify(
+            m => m.GetOrchestrationStateAsync(instance.InstanceId, false), Times.Once);
+    }
+
+    /// <summary>
+    /// A <see cref="ShimDurableTaskClient"/> test double whose <see cref="DelayAsync"/> override replaces
+    /// the real polling delay with a fully controlled fake: it signals <see cref="DelayEntered"/> 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* <see cref="CancellationToken"/> 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.
+    /// </summary>
+    sealed class DelayObservingShimDurableTaskClient(string name, ShimDurableTaskClientOptions options)
+        : ShimDurableTaskClient(name, options)
+    {
+        readonly TaskCompletionSource delayEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        /// <summary>Gets a task that completes once <see cref="DelayAsync"/> has been called.</summary>
+        public Task DelayEntered => this.delayEntered.Task;
+
+        internal override Task DelayAsync(TimeSpan delay, CancellationToken cancellation)
+        {
+            TaskCompletionSource pending = new(TaskCreationOptions.RunContinuationsAsynchronously);
+            cancellation.Register(() => pending.TrySetCanceled(cancellation));
+            this.delayEntered.TrySetResult();
+            return pending.Task;
+        }
+    }
+
     [Fact]
     public Task ScheduleNewOrchestrationInstance_IdGenerated_NoInput()
         => this.RunScheduleNewOrchestrationInstanceAsync("test", null, null);