Add retry to gRPC calls that failed due to transient errors (#714)

Author: sophiatev

src/Worker/Grpc/ReconnectBackoff.cs src/Worker/Grpc/GrpcBackoff.cssrc/Worker/Grpc/ReconnectBackoff.cs renamed to src/Worker/Grpc/GrpcBackoff.cs

@@ -6,9 +6,9 @@
 namespace Microsoft.DurableTask.Worker.Grpc;
 
 /// <summary>
-/// Helpers for computing reconnect backoff delays in the gRPC worker.
+/// Helpers for computing reconnect and retry backoff delays in the gRPC worker.
 /// </summary>
-static class ReconnectBackoff
+static class GrpcBackoff
 {
     /// <summary>
     /// Creates a random source for reconnect jitter using an explicit random seed so multiple workers on
@@ -32,10 +32,11 @@ public static Random CreateRandom()
     /// <param name="baseDelay">The base delay used for the exponential growth.</param>
     /// <param name="cap">The maximum delay before jitter is applied.</param>
     /// <param name="random">The random source used for jitter.</param>
+    /// <param name="fullJitter">If true, applies full jitter. If false, applies a smaller jitter that is biased towards the upper bound.</param>
     /// <returns>The computed jittered delay.</returns>
-    public static TimeSpan Compute(int attempt, TimeSpan baseDelay, TimeSpan cap, Random random)
+    public static TimeSpan Compute(int attempt, TimeSpan baseDelay, TimeSpan cap, Random random, bool fullJitter)
     {
-        if (baseDelay <= TimeSpan.Zero)
+        if (baseDelay <= TimeSpan.Zero || cap <= TimeSpan.Zero)
         {
             return TimeSpan.Zero;
         }
@@ -48,13 +49,13 @@ public static TimeSpan Compute(int attempt, TimeSpan baseDelay, TimeSpan cap, Ra
         // Cap the exponent to avoid overflow in 2^attempt for pathological attempt values.
         int safeAttempt = Math.Min(attempt, 30);
 
-        double capMs = Math.Max(0, cap.TotalMilliseconds);
         double exponentialMs = baseDelay.TotalMilliseconds * Math.Pow(2, safeAttempt);
-        double upperBoundMs = Math.Min(capMs, exponentialMs);
+        double upperBoundMs = Math.Min(cap.TotalMilliseconds, exponentialMs);
+
+        double jitteredMs = fullJitter
+            ? random.NextDouble() * upperBoundMs
+            : upperBoundMs + (random.NextDouble() * (upperBoundMs * .2));
 
-        // Full jitter intentionally allows any value in the retry window. The wide spread keeps many
-        // workers that saw the same outage from reconnecting in lockstep against the backend.
-        double jitteredMs = random.NextDouble() * upperBoundMs;
         return TimeSpan.FromMilliseconds(jitteredMs);
     }
 }

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -135,6 +135,29 @@ internal class InternalOptions
         /// </summary>
         public TimeSpan ReconnectBackoffCap { get; set; } = TimeSpan.FromSeconds(30);
 
+        /// <summary>
+        /// Gets or sets the maximum number of attempts the worker will make when retrying a transient
+        /// gRPC call (such as completing or abandoning a work item). Once this many attempts have failed,
+        /// the most recent exception is rethrown. Defaults to 10.
+        /// </summary>
+        public int TransientRetryMaxAttempts { get; set; } = 10;
+
+        /// <summary>
+        /// Gets or sets the initial delay used when computing exponential backoff between retries of a
+        /// transient gRPC call. The delay doubles after each failed attempt, and the exponential component
+        /// is capped at <see cref="TransientRetryBackoffCap"/> before jitter is applied. In the default
+        /// biased-jitter mode, the final delay may therefore slightly exceed
+        /// <see cref="TransientRetryBackoffCap"/>. Defaults to 200 ms.
+        /// </summary>
+        public TimeSpan TransientRetryBackoffBase { get; set; } = TimeSpan.FromMilliseconds(200);
+
+        /// <summary>
+        /// Gets or sets the cap applied to the exponential backoff component between retries of a transient
+        /// gRPC call before jitter is applied. In the default biased-jitter mode, the final computed delay
+        /// may be slightly greater than this value. Defaults to 15 seconds.
+        /// </summary>
+        public TimeSpan TransientRetryBackoffCap { get; set; } = TimeSpan.FromSeconds(15);
+
         /// <summary>
         /// Gets or sets an optional callback invoked when the worker requests a fresh gRPC channel after
         /// repeated connect failures. The callback receives the previously-used channel and should return

src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs

@@ -99,6 +99,9 @@ static partial class Logs
         public static partial void ReceivedHealthPing(this ILogger logger);
 
         [LoggerMessage(EventId = 76, Level = LogLevel.Information, Message = "Work-item stream ended by the backend (graceful close). Will reconnect.")]
-        public static partial void StreamEndedByPeer(this ILogger logger);
+        public static partial void StreamEndedByPeer(this ILogger logger);
+
+        [LoggerMessage(EventId = 77, Level = LogLevel.Warning, Message = "Transient gRPC error for '{OperationName}'. Attempt {Attempt} of {MaxAttempts}. Retrying in {BackoffMs} ms. StatusCode={StatusCode}")]
+        public static partial void TransientGrpcRetry(this ILogger logger, string operationName, int attempt, int maxAttempts, double backoffMs, int statusCode, Exception exception);
     }
 }