Added retry options to fetch

.changeset/quick-hairs-punch.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added retry options to fetch

apps/docs/functions/fetch.mdx

@@ -62,6 +62,11 @@ This is useful if you want to wrap `fetch` to provide an SDK like experience ins
 
 ## Response
 
+<Info>
+  Non-ok responses will currently halt the progress of a run, so `response.ok`
+  is always `true`
+</Info>
+
 The return value of `fetch` is a similar to a normal fetch response, but we will automatically parse the response body as JSON and provide it as `body`, like so:
 
 ```ts
@@ -154,6 +159,40 @@ Which will show in the trigger.dev app as:
 
 ![title](/images/secure-string.png)
 
+## Retrying
+
+By default, we will retry a failed request up to 10 times if it has one of the following status codes: `408`, `429`, `500`, `502`, `503`, `504`. You can override this behavior by providing a `retry` option:
+
+```ts
+await ctx.fetch("Example Key", "http://httpbin.org/get", {
+  retry: {
+    maxAttempts: 5,
+    statusCodes: [408, 429, 500, 502, 503, 504, 521, 522, 524],
+    minTimeout: 1000,
+    maxTimeout: 10000,
+    factor: 1.2,
+  },
+});
+```
+
+| Property    | Description                              | Default                                  |
+| ----------- | ---------------------------------------- | ---------------------------------------- |
+| enabled     | Enables retrying of failed requests      | true                                     |
+| factor      | The exponential factor of backoff        | 1.8                                      |
+| minTimeout  | The minimum amount of ms between retries | 1000                                     |
+| maxTimeout  | The maximum amount of ms between retries | 60000                                    |
+| statusCodes | The HTTP Status Codes that are retryable | `408`, `429`, `500`, `502`, `503`, `504` |
+
+If you'd like to disable retrying, simple pass in the `retry` option with `enabled` set to `false`:
+
+```ts
+await ctx.fetch("Example Key", "http://httpbin.org/get", {
+  retry: {
+    enabled: false,
+  },
+});
+```
+
 ## Params
 
 <ParamField path="key" type="string" required={true}>

apps/webapp/app/models/workflowRunPresenter.server.ts

@@ -125,7 +125,9 @@ async function parseStep(
       const fetchRequest = FetchRequestSchema.parse(original.input);
       const lastFetchResponse = original.fetchRequest.responses[0];
       const lastResponse = lastFetchResponse
-        ? FetchResponseSchema.parse(lastFetchResponse.output)
+        ? FetchResponseSchema.safeParse(lastFetchResponse.output).success
+          ? FetchResponseSchema.parse(lastFetchResponse.output)
+          : undefined
         : undefined;
 
       return {

apps/webapp/app/services/fetches/performFetchRequest.server.ts

@@ -1,5 +1,5 @@
 import type { FetchRequest } from ".prisma/client";
-import type { SecureString } from "@trigger.dev/common-schemas";
+import type { RetrySchema, SecureString } from "@trigger.dev/common-schemas";
 import { FetchRequestSchema } from "@trigger.dev/common-schemas";
 import type {
   NormalizedResponse,
@@ -9,8 +9,6 @@ import type { z } from "zod";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 
-const RETRYABLE_STATUS_CODES = [408, 429, 500, 502, 503, 504];
-
 type CallResponse =
   | {
       stop: true;
@@ -36,12 +34,28 @@ export class PerformFetchRequest {
       return { stop: true };
     }
 
-    const performedRequest = await this.#performRequest(fetchRequest);
+    const request = FetchRequestSchema.parse(fetchRequest.fetch);
+
+    const retryConfig = {
+      enabled: true,
+      maxAttempts: 10,
+      minTimeout: 1000,
+      maxTimeout: 60000,
+      factor: 1.8,
+      statusCodes: [408, 429, 500, 502, 503, 504],
+      ...(request.retry ?? {}),
+    };
+
+    const performedRequest = await this.#performRequest(request, retryConfig);
 
     if (performedRequest.ok) {
       return this.#completeWithSuccess(fetchRequest, performedRequest.response);
     } else if (performedRequest.isRetryable) {
-      return this.#attemptRetry(fetchRequest, performedRequest.response);
+      return this.#attemptRetry(
+        retryConfig,
+        fetchRequest,
+        performedRequest.response
+      );
     } else {
       return this.#completeWithFailure(fetchRequest, performedRequest.response);
     }
@@ -108,21 +122,11 @@ export class PerformFetchRequest {
   }
 
   async #attemptRetry(
+    retry: z.infer<typeof RetrySchema>,
     fetchRequest: FetchRequest,
     response: NormalizedResponse
   ) {
-    if (fetchRequest.retryCount >= 10) {
-      await this.#prismaClient.fetchRequest.update({
-        where: {
-          id: fetchRequest.id,
-        },
-        data: {
-          retryCount: {
-            increment: 1,
-          },
-        },
-      });
-
+    if (fetchRequest.retryCount >= retry.maxAttempts) {
       return this.#completeWithFailure(fetchRequest, response);
     }
 
@@ -143,7 +147,8 @@ export class PerformFetchRequest {
     return {
       stop: false as const,
       retryInSeconds: this.#calculateRetryInSeconds(
-        updatedFetchRequest.retryCount
+        updatedFetchRequest.retryCount,
+        retry
       ),
     };
   }
@@ -151,15 +156,11 @@ export class PerformFetchRequest {
   // Exponential backoff with a configurable factor and a configurable maximum
   #calculateRetryInSeconds(
     retryCount: number,
-    options: { factor: number; maxTimeout: number; minTimeout: number } = {
-      factor: 1.8,
-      minTimeout: 1000,
-      maxTimeout: 60000,
-    }
+    retryOptions: z.infer<typeof RetrySchema>
   ) {
-    const timeout = options.factor ** retryCount * options.minTimeout;
+    const timeout = retryOptions.factor ** retryCount * retryOptions.minTimeout;
 
-    return Math.min(timeout, options.maxTimeout) / 1000;
+    return Math.min(timeout, retryOptions.maxTimeout) / 1000;
   }
 
   async #createResponse(
@@ -182,19 +183,36 @@ export class PerformFetchRequest {
   }
 
   async #performRequest(
-    fetchRequest: FetchRequest
+    request: z.infer<typeof FetchRequestSchema>,
+    retry: z.infer<typeof RetrySchema>
   ): Promise<PerformedRequestResponse> {
-    const request = FetchRequestSchema.parse(fetchRequest.fetch);
-    const requestInit = createFetchRequestInit(request);
+    try {
+      const requestInit = createFetchRequestInit(request);
 
-    const response = await fetch(request.url, requestInit);
+      const response = await fetch(request.url, requestInit);
 
-    const body = await this.#safeGetJson(response);
+      const body = await this.#safeGetJson(response);
 
-    if (response.ok) {
+      if (response.ok) {
+        return {
+          ok: true,
+          isRetryable: false,
+          response: {
+            output: {
+              status: response.status,
+              headers: headersToRecord(response.headers),
+              body,
+            },
+            context: {},
+          },
+        };
+      }
+
+      // Only retry on retryable status codes
       return {
-        ok: true,
-        isRetryable: false,
+        ok: false,
+        isRetryable:
+          retry.statusCodes.includes(response.status) && retry.enabled,
         response: {
           output: {
             status: response.status,
@@ -204,21 +222,33 @@ export class PerformFetchRequest {
           context: {},
         },
       };
+    } catch (error) {
+      if (error instanceof Error) {
+        return {
+          ok: false,
+          isRetryable: false,
+          response: {
+            output: {
+              name: error.name,
+              message: error.message,
+            },
+            context: {},
+          },
+        };
+      } else {
+        return {
+          ok: false,
+          isRetryable: false,
+          response: {
+            output: {
+              name: "UnknownError",
+              message: "Unknown error",
+            },
+            context: {},
+          },
+        };
+      }
     }
-
-    // Only retry on retryable status codes
-    return {
-      ok: false,
-      isRetryable: RETRYABLE_STATUS_CODES.includes(response.status),
-      response: {
-        output: {
-          status: response.status,
-          headers: headersToRecord(response.headers),
-          body,
-        },
-        context: {},
-      },
-    };
   }
 
   #safeGetJson = async (response: Response) => {

apps/webapp/prisma/migrations/20230124224344_add_retry_to_fetch_request/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "FetchRequest" ADD COLUMN     "retry" JSONB;

apps/webapp/prisma/migrations/20230124224734_remove_retry/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `retry` on the `FetchRequest` table. All the data in the column will be lost.
+
+*/
+-- AlterTable
+ALTER TABLE "FetchRequest" DROP COLUMN "retry";

examples/fetch-playground/src/index.ts

@@ -20,6 +20,18 @@ new Trigger({
         .default("GET"),
       headers: z.record(z.string()).optional(),
       body: z.any().optional(),
+      retry: z
+        .object({
+          enabled: z.boolean().default(true),
+          maxAttempts: z.number().default(3),
+          minTimeout: z.number().default(1000),
+          maxTimeout: z.number().default(60000),
+          factor: z.number().default(1.8),
+          statusCodes: z
+            .array(z.number())
+            .default([408, 429, 500, 502, 503, 504]),
+        })
+        .optional(),
     }),
   }),
   run: async (event, ctx) => {
@@ -33,6 +45,7 @@ new Trigger({
       responseSchema: z.any(),
       headers: event.headers,
       body: event.body ? JSON.stringify(event.body) : undefined,
+      retry: event.retry,
     });
 
     await ctx.logger.info("Received the fetch response", {

packages/common-schemas/src/fetch.ts

@@ -8,6 +8,15 @@ export const SecureStringSchema = z.object({
 
 export type SecureString = z.infer<typeof SecureStringSchema>;
 
+export const RetrySchema = z.object({
+  enabled: z.boolean().default(true),
+  factor: z.number().default(1.8),
+  maxTimeout: z.number().default(60000),
+  minTimeout: z.number().default(1000),
+  maxAttempts: z.number().default(10),
+  statusCodes: z.array(z.number()).default([408, 429, 500, 502, 503, 504]),
+});
+
 export const FetchRequestSchema = z.object({
   url: z.string(),
   headers: z.record(z.union([z.string(), SecureStringSchema])).optional(),
@@ -22,6 +31,7 @@ export const FetchRequestSchema = z.object({
     "TRACE",
   ]),
   body: z.any(),
+  retry: RetrySchema.optional(),
 });
 
 export const FetchOutputSchema = z.object({

packages/internal-bridge/src/schemas/server.ts

@@ -1,6 +1,7 @@
 import {
   CustomEventSchema,
   FetchRequestSchema,
+  RetrySchema,
   TriggerMetadataSchema,
   WaitSchema,
 } from "@trigger.dev/common-schemas";

packages/internal-bridge/src/zodRPC.ts

@@ -117,8 +117,10 @@ export class ZodRPC<
 
   public send<K extends keyof SenderSchema>(
     key: K,
-    data: z.infer<SenderSchema[K]["request"]>
+    data: z.input<SenderSchema[K]["request"]>
   ) {
+    this.#logger.debug("Sending call", { key, data });
+
     const id = generateStableId(this.#connection.id, key as string, data);
 
     const message = packageMessage({ id, methodName: key as string, data });

packages/internal-platform/src/messages/schemas/fetchRequests.ts

@@ -2,6 +2,7 @@ import {
   FetchOutputSchema,
   FetchRequestSchema,
   JsonSchema,
+  RetrySchema,
 } from "@trigger.dev/common-schemas";
 import { z } from "zod";
 import {

packages/trigger-sdk/src/client.ts

@@ -335,6 +335,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
                 method: options.method ?? "GET",
                 headers: options.headers,
                 body: options.body,
+                retry: options.retry,
               },
               timestamp: String(highPrecisionTimestamp()),
             });
@@ -508,7 +509,14 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
                           };
                         }
 
-                        console.error(anyError);
+                        const parsedError = z
+                          .object({ name: z.string(), message: z.string() })
+                          .passthrough()
+                          .safeParse(error);
+
+                        if (parsedError.success) {
+                          return parsedError.data;
+                        }
 
                         return {
                           name: "UnknownError",

packages/trigger-sdk/src/localStorage.ts

@@ -1,11 +1,6 @@
 import { AsyncLocalStorage } from "node:async_hooks";
 import { z } from "zod";
-import {
-  FetchOptions,
-  FetchResponse,
-  TriggerCustomEvent,
-  TriggerFetch,
-} from "./types";
+import { TriggerCustomEvent, TriggerFetch } from "./types";
 
 type PerformRequestOptions<TSchema extends z.ZodTypeAny> = {
   service: string;

packages/trigger-sdk/src/types.ts

@@ -31,6 +31,14 @@ export type FetchOptions<
   body?: z.infer<typeof SerializableJsonSchema>;
   headers?: Record<string, string | SecureString>;
   responseSchema?: TResponseBodySchema;
+  retry?: {
+    enabled?: boolean;
+    factor?: number;
+    maxTimeout?: number;
+    minTimeout?: number;
+    maxAttempts?: number;
+    statusCodes?: number[];
+  };
 };
 
 export type FetchResponse<