fix: retry worker response delivery on transient gRPC failures

CHANGELOG.md

@@ -16,6 +16,8 @@
 
 ### Fixes
 
+- Retry worker completion and version-rejection responses on transient gRPC failures, reusing
+  the computed response without rerunning user code. Bound SDK sends to ten with shutdown-aware backoff.
 - Cancel pending client wait RPCs on timeout or cancellation without terminating the orchestration.
   Retry server `DEADLINE_EXCEEDED` responses for completion waits with backoff within the original total timeout.
 - Bound each worker sidecar hello attempt to 30 seconds, retry failed connections, and cancel

README.md

@@ -125,6 +125,24 @@ reconnected work-item streams.
 
 You can find more samples in the [examples/azure-managed](./examples/azure-managed) directory.
 
+### Worker response delivery
+
+An activity can return `42` but fail to report that result because of a transient gRPC error.
+Workers retry the same completion response and token instead of running the activity again.
+If a resend is accepted, the backend can advance the workflow using the saved result.
+This also applies to orchestration and entity responses, including version-rejection responses.
+
+The policy follows the .NET worker: up to ten SDK sends for `UNAVAILABLE`, `UNKNOWN`,
+`DEADLINE_EXCEEDED`, or `INTERNAL`, with backoff starting at 200 ms, doubling to a 15-second
+cap before adding 0-20% jitter. Permanent errors and exhausted attempts use the existing
+error logs. Configured gRPC transport retries remain enabled, so ten SDK sends can involve
+more than ten network attempts.
+
+`stop()` cancels retry backoff and in-flight retry RPCs. Already-running work can still send
+its first response during the existing bounded shutdown wait; user code and metadata
+generation are not canceled. Channel retirement and backend lock durations are unchanged.
+Retries do not guarantee connection recovery, acceptance of expired tokens, or exactly-once execution.
+
 ### Reusing orchestration instance IDs
 
 Set the top-level `dedupeStatuses` start option when an instance ID may be reused. The list

packages/durabletask-js/src/utils/backoff.util.ts

@@ -37,10 +37,11 @@ export interface BackoffOptions {
 
   /**
    * Jitter distribution. "symmetric" preserves the default range around the delay;
-   * "full" selects uniformly from zero through the exponential upper bound.
+   * "full" selects uniformly from zero through the exponential upper bound;
+   * "positive" adds jitter above the delay, matching .NET worker delivery retries.
    * @default "symmetric"
    */
-  jitterStrategy?: "symmetric" | "full";
+  jitterStrategy?: "symmetric" | "full" | "positive";
 }
 
 /**
@@ -80,7 +81,7 @@ export class ExponentialBackoff {
   private readonly _multiplier: number;
   private readonly _maxAttempts: number;
   private readonly _jitterFactor: number;
-  private readonly _jitterStrategy: "symmetric" | "full";
+  private readonly _jitterStrategy: "symmetric" | "full" | "positive";
 
   private _currentDelayMs: number;
   private _attemptCount: number;
@@ -135,6 +136,9 @@ export class ExponentialBackoff {
     }
 
     const jitterRange = this._currentDelayMs * this._jitterFactor;
+    if (this._jitterStrategy === "positive") {
+      return Math.floor(this._currentDelayMs + Math.random() * jitterRange);
+    }
     const jitter = Math.random() * jitterRange * 2 - jitterRange;
     return Math.max(0, Math.floor(this._currentDelayMs + jitter));
   }

packages/durabletask-js/src/worker/task-hub-grpc-worker.ts

@@ -905,6 +905,43 @@ export class TaskHubGrpcWorker {
     this._pendingWorkItems.add(handledPromise);
   }
 
+  private async _deliverResponse<TReq, TRes>(
+    method: Parameters<typeof callWithMetadata<TReq, TRes>>[0],
+    request: TReq,
+    retrySignal?: AbortSignal,
+  ): Promise<TRes> {
+    const backoff = new ExponentialBackoff({
+      initialDelayMs: 200,
+      maxDelayMs: 15000,
+      maxAttempts: 9, // Ten SDK sends; configured gRPC transport retries still apply to each send.
+      jitterStrategy: "positive",
+      jitterFactor: 0.2,
+    });
+    for (;;) {
+      try {
+        // Allow the initial response to drain during shutdown; only retries use the captured run signal.
+        return await callWithMetadata(
+          method,
+          request,
+          this._metadataGenerator,
+          backoff.attemptCount === 0 ? undefined : retrySignal,
+        );
+      } catch (error) {
+        const status = error instanceof Error ? this._getGrpcStatus(error) : undefined;
+        if (
+          !backoff.canRetry() ||
+          (status !== grpc.status.UNAVAILABLE &&
+            status !== grpc.status.UNKNOWN &&
+            status !== grpc.status.DEADLINE_EXCEEDED &&
+            status !== grpc.status.INTERNAL)
+        ) {
+          throw error;
+        }
+        await backoff.wait(retrySignal);
+      }
+    }
+  }
+
   /**
    * Executes an orchestrator request and tracks it as a pending work item.
    */
@@ -913,7 +950,8 @@ export class TaskHubGrpcWorker {
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
   ): void {
-    const workPromise = this._executeOrchestratorInternal(req, completionToken, stub);
+    const retrySignal = this._abortController?.signal;
+    const workPromise = this._executeOrchestratorInternal(req, completionToken, stub, retrySignal);
     this._trackPendingWorkItem(workPromise, (error) => {
       WorkerLogs.executionError(this._logger, req.getInstanceid() || "(unknown)", error);
     });
@@ -926,6 +964,7 @@ export class TaskHubGrpcWorker {
     req: pb.OrchestratorRequest,
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
+    retrySignal?: AbortSignal,
   ): Promise<void> {
     const instanceId = req.getInstanceid();
 
@@ -965,7 +1004,7 @@ export class TaskHubGrpcWorker {
         res.setActionsList(actions);
 
         try {
-          await callWithMetadata(stub.completeOrchestratorTask.bind(stub), res, this._metadataGenerator);
+          await this._deliverResponse(stub.completeOrchestratorTask.bind(stub), res, retrySignal);
         } catch (e: unknown) {
           const error = e instanceof Error ? e : new Error(String(e));
           WorkerLogs.completionError(this._logger, instanceId, error);
@@ -983,11 +1022,7 @@ export class TaskHubGrpcWorker {
         try {
           const abandonRequest = new pb.AbandonOrchestrationTaskRequest();
           abandonRequest.setCompletiontoken(completionToken);
-          await callWithMetadata(
-            stub.abandonTaskOrchestratorWorkItem.bind(stub),
-            abandonRequest,
-            this._metadataGenerator,
-          );
+          await this._deliverResponse(stub.abandonTaskOrchestratorWorkItem.bind(stub), abandonRequest, retrySignal);
         } catch (e: unknown) {
           const error = e instanceof Error ? e : new Error(String(e));
           WorkerLogs.completionError(this._logger, instanceId, error);
@@ -1094,7 +1129,7 @@ export class TaskHubGrpcWorker {
     }
 
     try {
-      await callWithMetadata(stub.completeOrchestratorTask.bind(stub), res, this._metadataGenerator);
+      await this._deliverResponse(stub.completeOrchestratorTask.bind(stub), res, retrySignal);
     } catch (e: unknown) {
       const error = e instanceof Error ? e : new Error(String(e));
       WorkerLogs.completionError(this._logger, req.getInstanceid(), error);
@@ -1109,7 +1144,8 @@ export class TaskHubGrpcWorker {
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
   ): void {
-    const workPromise = this._executeActivityInternal(req, completionToken, stub);
+    const retrySignal = this._abortController?.signal;
+    const workPromise = this._executeActivityInternal(req, completionToken, stub, retrySignal);
     this._trackPendingWorkItem(workPromise, (error) => {
       WorkerLogs.workerError(this._logger, error);
     });
@@ -1122,6 +1158,7 @@ export class TaskHubGrpcWorker {
     req: pb.ActivityRequest,
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
+    retrySignal?: AbortSignal,
   ): Promise<void> {
     const instanceId = req.getOrchestrationinstance()?.getInstanceid();
 
@@ -1175,7 +1212,7 @@ export class TaskHubGrpcWorker {
     }
 
     try {
-      await callWithMetadata(stub.completeActivityTask.bind(stub), res, this._metadataGenerator);
+      await this._deliverResponse(stub.completeActivityTask.bind(stub), res, retrySignal);
     } catch (e: unknown) {
       const error = e instanceof Error ? e : new Error(String(e));
       WorkerLogs.activityResponseError(this._logger, req.getName(), req.getTaskid(), instanceId!, error);
@@ -1191,7 +1228,8 @@ export class TaskHubGrpcWorker {
     stub: stubs.TaskHubSidecarServiceClient,
     operationInfos?: pb.OperationInfo[],
   ): void {
-    const workPromise = this._executeEntityInternal(req, completionToken, stub, operationInfos);
+    const retrySignal = this._abortController?.signal;
+    const workPromise = this._executeEntityInternal(req, completionToken, stub, operationInfos, retrySignal);
     this._trackPendingWorkItem(workPromise, (error) => {
       WorkerLogs.workerError(this._logger, error);
     });
@@ -1214,6 +1252,7 @@ export class TaskHubGrpcWorker {
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
     operationInfos?: pb.OperationInfo[],
+    retrySignal?: AbortSignal,
   ): Promise<void> {
     const instanceIdString = req.getInstanceid();
 
@@ -1233,7 +1272,7 @@ export class TaskHubGrpcWorker {
         completionToken,
         `Invalid entity instance id format: '${instanceIdString}'`,
       );
-      await this._sendEntityResult(batchResult, stub);
+      await this._sendEntityResult(batchResult, stub, retrySignal);
       return;
     }
 
@@ -1281,7 +1320,7 @@ export class TaskHubGrpcWorker {
       batchResult.setOperationinfosList(infosToInclude);
     }
 
-    await this._sendEntityResult(batchResult, stub);
+    await this._sendEntityResult(batchResult, stub, retrySignal);
   }
 
   /**
@@ -1292,7 +1331,8 @@ export class TaskHubGrpcWorker {
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
   ): void {
-    const workPromise = this._executeEntityV2Internal(req, completionToken, stub);
+    const retrySignal = this._abortController?.signal;
+    const workPromise = this._executeEntityV2Internal(req, completionToken, stub, retrySignal);
     this._trackPendingWorkItem(workPromise, (error) => {
       WorkerLogs.workerError(this._logger, error);
     });
@@ -1314,6 +1354,7 @@ export class TaskHubGrpcWorker {
     req: pb.EntityRequest,
     completionToken: string,
     stub: stubs.TaskHubSidecarServiceClient,
+    retrySignal?: AbortSignal,
   ): Promise<void> {
     // Convert EntityRequest (V2) to EntityBatchRequest (V1) format
     const batchRequest = new pb.EntityBatchRequest();
@@ -1394,7 +1435,7 @@ export class TaskHubGrpcWorker {
     batchRequest.setOperationsList(operations);
 
     // Delegate to the V1 execution logic with V2 operationInfos
-    await this._executeEntityInternal(batchRequest, completionToken, stub, operationInfos);
+    await this._executeEntityInternal(batchRequest, completionToken, stub, operationInfos, retrySignal);
   }
 
   /**
@@ -1447,9 +1488,10 @@ export class TaskHubGrpcWorker {
   private async _sendEntityResult(
     batchResult: pb.EntityBatchResult,
     stub: stubs.TaskHubSidecarServiceClient,
+    retrySignal?: AbortSignal,
   ): Promise<void> {
     try {
-      await callWithMetadata(stub.completeEntityTask.bind(stub), batchResult, this._metadataGenerator);
+      await this._deliverResponse(stub.completeEntityTask.bind(stub), batchResult, retrySignal);
     } catch (e: any) {
       WorkerLogs.entityResponseDeliveryFailed(this._logger, e);
     }