Add run retry feature (#448)

* Add run retry feature
* Expose the other fields of the retry policy
* Make sure we don't use the retry policy when no option is set, retaining the same behavior as before.

Author: slinkydeveloper

packages/restate-e2e-services/exclusions.yaml

@@ -1,7 +1 @@
-exclusions:
-  "alwaysSuspending":
-    - "dev.restate.sdktesting.tests.RunRetry"
-  "default":
-    - "dev.restate.sdktesting.tests.RunRetry"
-  "singleThreadSinglePartition":
-    - "dev.restate.sdktesting.tests.RunRetry"
+exclusions: {}

packages/restate-e2e-services/src/failing.ts

@@ -9,9 +9,11 @@
 
 import * as restate from "@restatedev/restate-sdk";
 import { REGISTRY } from "./services.js";
+import { TerminalError } from "@restatedev/restate-sdk";
 
 let eventualSuccessCalls = 0;
 let eventualSuccessSideEffectCalls = 0;
+let eventualFailureSideEffectCalls = 0;
 
 const service = restate.object({
   name: "Failing",
@@ -46,24 +48,6 @@ const service = restate.object({
       }
     },
 
-    failingSideEffectWithEventualSuccess: async (
-      context: restate.ObjectContext
-    ) => {
-      const successAttempt = await context.run(() => {
-        eventualSuccessSideEffectCalls += 1;
-        const currentAttempt = eventualSuccessSideEffectCalls;
-
-        if (currentAttempt >= 4) {
-          eventualSuccessSideEffectCalls = 0;
-          return currentAttempt;
-        } else {
-          throw new Error("Failed at attempt: " + currentAttempt);
-        }
-      });
-
-      return successAttempt;
-    },
-
     terminallyFailingSideEffect: async (
       ctx: restate.ObjectContext,
       errorMessage: string
@@ -74,6 +58,59 @@ const service = restate.object({
 
       throw new Error("Should not be reached.");
     },
+
+    sideEffectSucceedsAfterGivenAttempts: async (
+      context: restate.ObjectContext,
+      minimumAttempts: number
+    ) => {
+      return await context.run(
+        "failing-side-effect",
+        () => {
+          eventualSuccessSideEffectCalls += 1;
+          const currentAttempt = eventualSuccessSideEffectCalls;
+
+          if (currentAttempt >= minimumAttempts) {
+            eventualSuccessSideEffectCalls = 0;
+            return currentAttempt;
+          } else {
+            throw new Error("Failed at attempt: " + currentAttempt);
+          }
+        },
+        { retryIntervalFactor: 1, initialRetryIntervalMillis: 10 }
+      );
+    },
+
+    sideEffectFailsAfterGivenAttempts: async (
+      context: restate.ObjectContext,
+      retryPolicyMaxRetryCount: number
+    ) => {
+      try {
+        await context.run(
+          "failing-side-effect",
+          () => {
+            eventualFailureSideEffectCalls += 1;
+            throw new Error(
+              "Failed at attempt: " + eventualFailureSideEffectCalls
+            );
+          },
+          {
+            maxRetryAttempts: retryPolicyMaxRetryCount,
+            retryIntervalFactor: 1,
+            initialRetryIntervalMillis: 10,
+          }
+        );
+      } catch (e) {
+        if (e instanceof TerminalError) {
+          context.console.log(
+            `run failed as expected with ${JSON.stringify(e)}`
+          );
+          return eventualFailureSideEffectCalls;
+        }
+        // This is not a TerminalError!
+        throw e;
+      }
+      throw new TerminalError("Side effect was supposed to fail!");
+    },
   },
 });
 

packages/restate-sdk/src/context.ts

@@ -179,11 +179,48 @@ export type RunAction<T> = (() => Promise<T>) | (() => T);
 
 export type RunOptions<T> = {
   serde?: Serde<T>;
+
+  /**
+   * Max number of retry attempts, before giving up.
+   *
+   * When giving up, `ctx.run` will throw a `TerminalError` wrapping the original error message.
+   */
+  maxRetryAttempts?: number;
+
+  /**
+   * Max duration of retries, before giving up.
+   *
+   * When giving up, `ctx.run` will throw a `TerminalError` wrapping the original error message.
+   */
+  maxRetryDurationMillis?: number;
+
+  /**
+   * Initial interval for the first retry attempt.
+   * Retry interval will grow by a factor specified in `retryIntervalFactor`.
+   *
+   * The default is 50 milliseconds.
+   */
+  initialRetryIntervalMillis?: number;
+
+  /**
+   * Max interval between retries.
+   * Retry interval will grow by a factor specified in `retryIntervalFactor`.
+   *
+   * The default is 10 seconds.
+   */
+  maxRetryIntervalMillis?: number;
+
+  /**
+   * Exponentiation factor to use when computing the next retry delay.
+   *
+   * The default value is `2`, meaning retry interval will double at each attempt.
+   */
+  retryIntervalFactor?: number;
 };
 
 /**
  * Call a handler directly avoiding restate's type safety checks.
- * This is a generic machnisim to invoke handlers directly by only knowing
+ * This is a generic mechanism to invoke handlers directly by only knowing
  * the service and handler name, (or key in the case of objects or workflows)
  */
 export type GenericCall<REQ, RES> = {
@@ -275,6 +312,7 @@ export interface Context extends RestateContext {
    *            throw new TerminalError("Payment failed");
    *        } else if (result.paymentGatewayBusy) {
    *            // restate will retry automatically
+   *            // to bound retries, use RunOptions
    *            throw new Exception("Payment gateway busy");
    *        } else {
    *            // success!

packages/restate-sdk/src/context_impl.ts

@@ -347,13 +347,15 @@ export class ContextImpl implements ObjectContext, WorkflowContext {
 
     // We wrap the rest of the execution in this closure to create a future
     const doRun = async () => {
+      const startTime = Date.now();
       let res: T;
       let err;
       try {
         res = await action();
       } catch (e) {
         err = ensureError(e);
       }
+      const attemptDuration = Date.now() - startTime;
 
       // Record the result/failure, get back the handle for the ack.
       let handle;
@@ -366,8 +368,29 @@ export class ContextImpl implements ObjectContext, WorkflowContext {
               message: err.message,
             });
           } else {
-            // TODO plug retries here!
-            throw err;
+            if (
+              options?.retryIntervalFactor === undefined &&
+              options?.initialRetryIntervalMillis === undefined &&
+              options?.maxRetryAttempts === undefined &&
+              options?.maxRetryDurationMillis === undefined &&
+              options?.maxRetryIntervalMillis === undefined
+            ) {
+              // If no retry option was set, simply throw the error.
+              // This will lead to the invoker applying its retry, without the SDK overriding it.
+              throw err;
+            }
+            handle = this.coreVm.sys_run_exit_failure_transient(
+              err.message,
+              err.cause?.toString(),
+              BigInt(attemptDuration),
+              {
+                factor: options?.retryIntervalFactor || 2.0,
+                initial_interval: options?.initialRetryIntervalMillis || 50,
+                max_attempts: options?.maxRetryAttempts,
+                max_duration: options?.maxRetryDurationMillis,
+                max_interval: options?.maxRetryIntervalMillis || 10 * 1000,
+              }
+            );
           }
         } else {
           // eslint-disable-next-line @typescript-eslint/ban-ts-comment