Introduce RetryableError (#569)

slinkydeveloper · web-flow · commit 40d129c77509 · 2025-07-09T10:17:16.000+02:00
* Introduce RetryableError, allowing users to customize the retry after when needed.

This is meant to be supported within `ctx.run` primarily.

* Apply feedback:

* Removed fields maxRetryAttempts and maxRetryDuration from RetryableError
* Make sure uses can handle RetryableError within asTerminalError
diff --git a/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings.d.ts b/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings.d.ts
@@ -100,6 +100,7 @@ export class WasmVM {
   notify_input(buffer: Uint8Array): void;
   notify_input_closed(): void;
   notify_error(error_message: string, stacktrace?: string | null): void;
+  notify_error_with_delay_override(error_message: string, stacktrace?: string | null, delay_override?: bigint | null): void;
   notify_error_for_next_command(error_message: string, stacktrace: string | null | undefined, wasm_command_type: WasmCommandType): void;
   notify_error_for_specific_command(error_message: string, stacktrace: string | null | undefined, wasm_command_type: WasmCommandType, command_index: number, command_name?: string | null): void;
   take_output(): any;
@@ -129,6 +130,7 @@ export class WasmVM {
   propose_run_completion_success(handle: number, buffer: Uint8Array): void;
   propose_run_completion_failure(handle: number, value: WasmFailure): void;
   propose_run_completion_failure_transient(handle: number, error_message: string, error_stacktrace: string | null | undefined, attempt_duration: bigint, config?: WasmExponentialRetryConfig | null): void;
+  propose_run_completion_failure_transient_with_delay_override(handle: number, error_message: string, error_stacktrace: string | null | undefined, attempt_duration: bigint, delay_override?: bigint | null, max_retry_attempts_override?: number | null, max_retry_duration_override?: bigint | null): void;
   sys_cancel_invocation(target_invocation_id: string): void;
   sys_write_output_success(buffer: Uint8Array): void;
   sys_write_output_failure(value: WasmFailure): void;
diff --git a/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.js b/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.js
@@ -591,6 +591,18 @@ export class WasmVM {
         var len1 = WASM_VECTOR_LEN;
         wasm.wasmvm_notify_error(this.__wbg_ptr, ptr0, len0, ptr1, len1);
     }
+    /**
+     * @param {string} error_message
+     * @param {string | null} [stacktrace]
+     * @param {bigint | null} [delay_override]
+     */
+    notify_error_with_delay_override(error_message, stacktrace, delay_override) {
+        const ptr0 = passStringToWasm0(error_message, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        const len0 = WASM_VECTOR_LEN;
+        var ptr1 = isLikeNone(stacktrace) ? 0 : passStringToWasm0(stacktrace, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        var len1 = WASM_VECTOR_LEN;
+        wasm.wasmvm_notify_error_with_delay_override(this.__wbg_ptr, ptr0, len0, ptr1, len1, !isLikeNone(delay_override), isLikeNone(delay_override) ? BigInt(0) : delay_override);
+    }
     /**
      * @param {string} error_message
      * @param {string | null | undefined} stacktrace
@@ -973,6 +985,25 @@ export class WasmVM {
             throw takeFromExternrefTable0(ret[0]);
         }
     }
+    /**
+     * @param {number} handle
+     * @param {string} error_message
+     * @param {string | null | undefined} error_stacktrace
+     * @param {bigint} attempt_duration
+     * @param {bigint | null} [delay_override]
+     * @param {number | null} [max_retry_attempts_override]
+     * @param {bigint | null} [max_retry_duration_override]
+     */
+    propose_run_completion_failure_transient_with_delay_override(handle, error_message, error_stacktrace, attempt_duration, delay_override, max_retry_attempts_override, max_retry_duration_override) {
+        const ptr0 = passStringToWasm0(error_message, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        const len0 = WASM_VECTOR_LEN;
+        var ptr1 = isLikeNone(error_stacktrace) ? 0 : passStringToWasm0(error_stacktrace, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        var len1 = WASM_VECTOR_LEN;
+        const ret = wasm.wasmvm_propose_run_completion_failure_transient_with_delay_override(this.__wbg_ptr, handle, ptr0, len0, ptr1, len1, attempt_duration, !isLikeNone(delay_override), isLikeNone(delay_override) ? BigInt(0) : delay_override, isLikeNone(max_retry_attempts_override) ? 0x100000001 : (max_retry_attempts_override) >>> 0, !isLikeNone(max_retry_duration_override), isLikeNone(max_retry_duration_override) ? BigInt(0) : max_retry_duration_override);
+        if (ret[1]) {
+            throw takeFromExternrefTable0(ret[0]);
+        }
+    }
     /**
      * @param {string} target_invocation_id
      */
diff --git a/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm b/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm
diff --git a/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm.d.ts b/packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm.d.ts
@@ -19,6 +19,7 @@ export const wasmvm_get_response_head: (a: number) => number;
 export const wasmvm_notify_input: (a: number, b: number, c: number) => void;
 export const wasmvm_notify_input_closed: (a: number) => void;
 export const wasmvm_notify_error: (a: number, b: number, c: number, d: number, e: number) => void;
+export const wasmvm_notify_error_with_delay_override: (a: number, b: number, c: number, d: number, e: number, f: number, g: bigint) => void;
 export const wasmvm_notify_error_for_next_command: (a: number, b: number, c: number, d: number, e: number, f: number) => void;
 export const wasmvm_notify_error_for_specific_command: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number) => void;
 export const wasmvm_take_output: (a: number) => any;
@@ -48,6 +49,7 @@ export const wasmvm_sys_run: (a: number, b: number, c: number) => [number, numbe
 export const wasmvm_propose_run_completion_success: (a: number, b: number, c: number, d: number) => [number, number];
 export const wasmvm_propose_run_completion_failure: (a: number, b: number, c: any) => [number, number];
 export const wasmvm_propose_run_completion_failure_transient: (a: number, b: number, c: number, d: number, e: number, f: number, g: bigint, h: number) => [number, number];
+export const wasmvm_propose_run_completion_failure_transient_with_delay_override: (a: number, b: number, c: number, d: number, e: number, f: number, g: bigint, h: number, i: bigint, j: number, k: number, l: bigint) => [number, number];
 export const wasmvm_sys_cancel_invocation: (a: number, b: number, c: number) => [number, number];
 export const wasmvm_sys_write_output_success: (a: number, b: number, c: number) => [number, number];
 export const wasmvm_sys_write_output_failure: (a: number, b: any) => [number, number];
diff --git a/packages/restate-sdk-examples/package.json b/packages/restate-sdk-examples/package.json
@@ -29,7 +29,7 @@
     "verify": "npm run format-check && npm run lint && npm run build",
     "release": "",
     "object": "RESTATE_LOGGING=debug tsx ./src/object.ts",
-    "greeter": "RESTATE_LOGGING=debug tsx ./src/greeter_with_options.ts",
+    "greeter": "RESTATE_LOGGING=debug tsx ./src/greeter.ts",
     "zgreeter": "RESTATE_LOGGING=debug tsx ./src/zod_greeter.ts",
     "workflow": "RESTATE_LOGGING=debug tsx ./src/workflow.ts",
     "workflow_client": "RESTATE_LOGGING=debug tsx ./src/workflow_client.ts",
diff --git a/packages/restate-sdk/src/common_api.ts b/packages/restate-sdk/src/common_api.ts
@@ -108,6 +108,8 @@ export {
   RestateError,
   TerminalError,
   TimeoutError,
+  RetryableError,
+  type RetryableErrorOptions,
 } from "./types/errors.js";
 export type {
   LoggerTransport,
diff --git a/packages/restate-sdk/src/context.ts b/packages/restate-sdk/src/context.ts
@@ -313,11 +313,22 @@ export interface Context extends RestateContext {
    *     re-executed on replay (the latest, if the failure happened in the small windows
    *     described above).
    *
+   * You can customize retry options by either:
+   *
+   * - Providing retry policy options in {@link RunOptions}
+   * - Throwing {@link RetryableError}, providing `retryAfter` option. This can be especially useful when interacting with HTTP requests returning the `Retry-After` header. You can combine the usage of throwing {@link RetryableError} with the `maxRetryAttempts`/`maxRetryDuration` from {@link RunOptions}.
+   *
    * @example
    * ```ts
    * const result = await ctx.run(someExternalAction)
    *```
-
+   *
+   * @example
+   * ```ts
+   * // Add some retry options
+   * const result = await ctx.run("my action", someExternalAction, { maxRetryAttempts: 10 })
+   * ```
+   *
    * @example
    * ```ts
    *    await ctx.run("payment action", async () => {
@@ -328,27 +339,41 @@ export interface Context extends RestateContext {
    *        } else if (result.paymentGatewayBusy) {
    *            // restate will retry automatically
    *            // to bound retries, use RunOptions
-   *            throw new Exception("Payment gateway busy");
+   *            throw new Error("Payment gateway busy");
    *        } else {
    *            // success!
    *        }
    *   });
    *
+   * @example
+   * ```ts
+   *    await ctx.run("payment action", async () => {
+   *        const res = fetch(...);
+   *        if (!res.ok) {
+   *          // Read Retry-After header
+   *          const retryAfterHeader = res.headers['Retry-After']
+   *
+   *          // Use RetryableError to customize in how long to retry
+   *          throw RetryableError.from(cause, { retryAfter: { seconds: retryAfterHeader } })
+   *        }
+   *   }, {
+   *       // Retry at most ten times
+   *       maxRetryAttempts: 10
+   *   });
    * ```
    *
    * @param action The function to run.
    */
   run<T>(action: RunAction<T>): RestatePromise<T>;
 
   /**
-   * Run an operation and store the result in Restate. The operation will thus not
-   * be re-run during a later replay, but take the durable result from Restate.
-   *
-   * @param name the action's name
-   * @param action the action to run.
+   * Same as {@link run}, but providing a name, used for observability purposes.
    */
   run<T>(name: string, action: RunAction<T>): RestatePromise<T>;
 
+  /**
+   * See {@link run}
+   */
   run<T>(
     name: string,
     action: RunAction<T>,
diff --git a/packages/restate-sdk/src/context_impl.ts b/packages/restate-sdk/src/context_impl.ts
@@ -38,6 +38,7 @@ import {
   INTERNAL_ERROR_CODE,
   logError,
   RestateError,
+  RetryableError,
   TerminalError,
   UNKNOWN_ERROR_CODE,
 } from "./types/errors.js";
@@ -436,6 +437,22 @@ export class ContextImpl implements ObjectContext, WorkflowContext {
               code: err.code,
               message: err.message,
             });
+          } else if (err instanceof RetryableError) {
+            const maxRetryDuration =
+              options?.maxRetryDuration ?? options?.maxRetryDurationMillis;
+            this.coreVm.propose_run_completion_failure_transient_with_delay_override(
+              handle,
+              err.message,
+              err.stack,
+              BigInt(attemptDuration),
+              err.retryAfter !== undefined
+                ? BigInt(millisOrDurationToMillis(err.retryAfter))
+                : undefined,
+              options?.maxRetryAttempts,
+              maxRetryDuration !== undefined
+                ? BigInt(millisOrDurationToMillis(maxRetryDuration))
+                : undefined
+            );
           } else {
             this.vmLogger.warn(
               `Error when processing ctx.run '${name}'.\n`,
diff --git a/packages/restate-sdk/src/endpoint/handlers/generic.ts b/packages/restate-sdk/src/endpoint/handlers/generic.ts
@@ -13,6 +13,7 @@ import {
   ensureError,
   logError,
   RestateError,
+  RetryableError,
   TerminalError,
 } from "../../types/errors.js";
 import type { ProtocolMode } from "../../types/discovery.js";
@@ -37,6 +38,7 @@ import {
   LogSource,
   RestateLogLevel,
 } from "../../logging/logger_transport.js";
+import { millisOrDurationToMillis } from "@restatedev/restate-sdk-core";
 
 export interface Headers {
   [name: string]: string | string[] | undefined;
@@ -389,6 +391,14 @@ export class GenericHandler implements RestateHandler {
               message: error.message,
             });
             coreVm.sys_end();
+          } else if (error instanceof RetryableError) {
+            coreVm.notify_error_with_delay_override(
+              error.message,
+              error.stack,
+              error.retryAfter !== undefined
+                ? BigInt(millisOrDurationToMillis(error.retryAfter))
+                : undefined
+            );
           } else {
             coreVm.notify_error(error.message, error.stack);
           }
diff --git a/packages/restate-sdk/src/endpoint/handlers/vm/sdk_shared_core_wasm_bindings.d.ts b/packages/restate-sdk/src/endpoint/handlers/vm/sdk_shared_core_wasm_bindings.d.ts
@@ -111,6 +111,11 @@ export class WasmVM {
   notify_input(buffer: Uint8Array): void;
   notify_input_closed(): void;
   notify_error(error_message: string, stacktrace?: string | null): void;
+  notify_error_with_delay_override(
+    error_message: string,
+    stacktrace?: string | null,
+    delay_override?: bigint | null
+  ): void;
   notify_error_for_next_command(
     error_message: string,
     stacktrace: string | null | undefined,
@@ -171,6 +176,15 @@ export class WasmVM {
     attempt_duration: bigint,
     config?: WasmExponentialRetryConfig | null
   ): void;
+  propose_run_completion_failure_transient_with_delay_override(
+    handle: number,
+    error_message: string,
+    error_stacktrace: string | null | undefined,
+    attempt_duration: bigint,
+    delay_override?: bigint | null,
+    max_retry_attempts_override?: number | null,
+    max_retry_duration_override?: bigint | null
+  ): void;
   sys_cancel_invocation(target_invocation_id: string): void;
   sys_write_output_success(buffer: Uint8Array): void;
   sys_write_output_failure(value: WasmFailure): void;
diff --git a/packages/restate-sdk/src/endpoint/handlers/vm/sdk_shared_core_wasm_bindings.js b/packages/restate-sdk/src/endpoint/handlers/vm/sdk_shared_core_wasm_bindings.js
diff --git a/packages/restate-sdk/src/types/errors.ts b/packages/restate-sdk/src/types/errors.ts
@@ -11,6 +11,8 @@
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 
+import type { Duration } from "@restatedev/restate-sdk-core";
+
 export const INTERNAL_ERROR_CODE = 500;
 export const TIMEOUT_ERROR_CODE = 408;
 export const CANCEL_ERROR_CODE = 409;
@@ -129,3 +131,49 @@ export class CancelledError extends TerminalError {
     super("Cancelled", { errorCode: CANCEL_ERROR_CODE });
   }
 }
+
+export interface RetryableErrorOptions {
+  /**
+   * In how long it should retry.
+   */
+  retryAfter?: Duration | number;
+}
+
+/**
+ * Error that Restate will retry. By using this error type within a `ctx.run` closure,
+ * you can dynamically provide the retry delay specified in {@link RetryableErrorOptions}.
+ *
+ * You can wrap another error using {@link from}.
+ */
+export class RetryableError extends RestateError {
+  public name = "RetryableError";
+
+  readonly retryAfter?: Duration | number;
+
+  constructor(
+    message: string,
+    options?: RetryableErrorOptions & {
+      errorCode?: number;
+      cause?: any;
+    }
+  ) {
+    super(message, {
+      errorCode: options?.errorCode,
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+      cause: options?.cause,
+    });
+    this.retryAfter = options?.retryAfter;
+  }
+
+  /**
+   * Create a `RetryableError` from the given cause.
+   */
+  static from(cause: any, options?: RetryableErrorOptions): RetryableError {
+    const error = ensureError(cause);
+    return new RetryableError(error.message, {
+      errorCode: error["errorCode" as keyof typeof error] as number,
+      cause: cause,
+      ...options,
+    });
+  }
+}
diff --git a/sdk-shared-core-wasm-bindings/Cargo.lock b/sdk-shared-core-wasm-bindings/Cargo.lock
diff --git a/sdk-shared-core-wasm-bindings/Cargo.toml b/sdk-shared-core-wasm-bindings/Cargo.toml
@@ -12,7 +12,7 @@ default = ["console_error_panic_hook"]
 
 [dependencies]
 wasm-bindgen = "0.2.100"
-restate-sdk-shared-core = { version = "=0.4.0", features = ["request_identity"] }
+restate-sdk-shared-core = { path = "../../sdk-shared-core", features = ["request_identity"] }
 console_error_panic_hook = { version = "0.1.7", optional = true }
 serde = { version = "1.0.210", features = ["derive"] }
 serde-wasm-bindgen = "0.6.5"
diff --git a/sdk-shared-core-wasm-bindings/src/lib.rs b/sdk-shared-core-wasm-bindings/src/lib.rs
@@ -474,6 +474,18 @@ impl WasmVM {
         use_log_dispatcher!(self, |vm| CoreVM::notify_error(vm, e, None))
     }
 
+    pub fn notify_error_with_delay_override(&mut self, error_message: String, stacktrace: Option<String>, delay_override: Option<u64>) {
+        let mut e = Error::internal(error_message);
+        if let Some(stacktrace) = stacktrace {
+            e = e.with_stacktrace(stacktrace);
+        }
+        if let Some(delay_override) = delay_override {
+            e = e.with_next_retry_delay_override(Duration::from_millis(delay_override))
+        }
+
+        use_log_dispatcher!(self, |vm| CoreVM::notify_error(vm, e, None))
+    }
+
     pub fn notify_error_for_next_command(
         &mut self,
         error_message: String,
@@ -817,6 +829,38 @@ impl WasmVM {
         .map_err(Into::into)
     }
 
+    pub fn propose_run_completion_failure_transient_with_delay_override(
+        &mut self,
+        handle: WasmNotificationHandle,
+        error_message: String,
+        error_stacktrace: Option<String>,
+        attempt_duration: u64,
+        delay_override: Option<u64>,
+        max_retry_attempts_override: Option<u32>,
+        max_retry_duration_override: Option<u64>
+    ) -> Result<(), WasmFailure> {
+        let retry_policy = if delay_override.is_some() || max_retry_attempts_override.is_some() || max_retry_duration_override.is_some() {
+            RetryPolicy::FixedDelay {
+                interval: delay_override.map(Duration::from_millis),
+                max_attempts: max_retry_attempts_override,
+                max_duration: max_retry_duration_override.map(Duration::from_millis),
+            }
+        } else {
+            RetryPolicy::Infinite
+        };
+        use_log_dispatcher!(self, |vm| CoreVM::propose_run_completion(
+            vm,
+            handle.into(),
+            RunExitResult::RetryableFailure {
+                attempt_duration: Duration::from_millis(attempt_duration),
+                error: Error::internal(error_message)
+                    .with_stacktrace(error_stacktrace.unwrap_or_default()),
+            },
+            retry_policy
+        ))
+            .map_err(Into::into)
+    }
+
     pub fn sys_cancel_invocation(
         &mut self,
         target_invocation_id: String,
