v2: Better handle recovering from platform communication errors by auto-yielding back to the platform in case of temporary API failures

.changeset/strange-sheep-pull.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+v2: Better handle recovering from platform communication errors by auto-yielding back to the platform in case of temporary API failures

apps/webapp/package.json

@@ -9,6 +9,7 @@
     "build:remix": "remix build",
     "build:server": "esbuild --platform=node --format=cjs ./server.ts --outdir=build",
     "dev": "cross-env PORT=3030 remix dev -c \"node ./build/server.js\"",
+    "dev:worker": "cross-env NODE_PATH=../../node_modules/.pnpm/node_modules node ./build/server.js",
     "format": "prettier --write .",
     "lint": "eslint --cache --cache-location ./node_modules/.cache/eslint .",
     "start": "cross-env NODE_ENV=production node --max-old-space-size=8192 ./build/server.js",
@@ -237,4 +238,4 @@
   "engines": {
     "node": ">=16.0.0"
   }
-}
+}

packages/core/src/v3/apiClient/index.ts

@@ -1,5 +1,6 @@
 import { context, propagation } from "@opentelemetry/api";
-import { ZodFetchOptions, zodfetch } from "../zodfetch";
+import { version } from "../../../package.json";
+import { APIError } from "../apiErrors";
 import {
   BatchTaskRunExecutionResult,
   BatchTriggerTaskRequestBody,
@@ -19,10 +20,7 @@ import {
   UpdateScheduleOptions,
 } from "../schemas";
 import { taskContext } from "../task-context-api";
-import { getEnvVar } from "../utils/getEnv";
+import { ZodFetchOptions, zodfetch } from "../zodfetch";
-import { SafeAsyncLocalStorage } from "../utils/safeAsyncLocalStorage";
-import { APIError } from "../apiErrors";
-import { version } from "../../../package.json";
 
 export type TriggerOptions = {
   spanParentAsLink?: boolean;

packages/trigger-sdk/src/apiClient.ts

@@ -74,6 +74,18 @@ export type RunRecord = {
   event: ApiEventLog;
 };
 
+export class UnknownVersionError extends Error {
+  constructor(version: string) {
+    super(`Unknown version ${version}`);
+  }
+}
+
+const MAX_RETRIES = 8;
+const EXPONENT_FACTOR = 2;
+const MIN_DELAY_IN_MS = 80;
+const MAX_DELAY_IN_MS = 2000;
+const JITTER_IN_MS = 50;
+
 export class ApiClient {
   #apiUrl: string;
   #options: ApiClientOptions;
@@ -129,11 +141,10 @@ export class ApiClient {
   ) {
     const apiKey = await this.#apiKey();
 
-    this.#logger.debug("Running Task", {
+    this.#logger.debug(`[ApiClient] runTask ${task.displayKey}`);
-      task,
-    });
 
     return await zodfetchWithVersions(
+      this.#logger,
       {
         [API_VERSIONS.LAZY_LOADED_CACHED_TASKS]: RunTaskResponseWithCachedTasksBodySchema,
       },
@@ -771,6 +782,7 @@ async function zodfetchWithVersions<
   TUnversionedResponseBodySchema extends z.ZodTypeAny,
   TOptional extends boolean = false,
 >(
+  logger: Logger,
   versionedSchemaMap: TVersionedResponseBodyMap,
   unversionedSchema: TUnversionedResponseBodySchema,
   url: string,
@@ -785,66 +797,132 @@ async function zodfetchWithVersions<
     ? VersionedResponseBody<TVersionedResponseBodyMap, TUnversionedResponseBodySchema> | undefined
     : VersionedResponseBody<TVersionedResponseBodyMap, TUnversionedResponseBodySchema>
 > {
-  const response = await fetch(url, requestInitWithCache(requestInit));
+  try {
+    const fullRequestInit = requestInitWithCache(requestInit);
 
-  if (
+    const response = await fetch(url, fullRequestInit);
-    (!requestInit || requestInit.method === "GET") &&
-    response.status === 404 &&
-    options?.optional
-  ) {
-    // @ts-ignore
-    return;
-  }
 
-  if (response.status >= 400 && response.status < 500) {
+    logger.debug(`[ApiClient] zodfetchWithVersions ${url} (attempt ${retryCount + 1})`, {
-    const body = await response.json();
-
-    throw new Error(body.error);
-  }
-
-  if (response.status >= 500 && retryCount < 6) {
-    // retry with exponential backoff and jitter
-    const delay = exponentialBackoff(retryCount + 1, 2, 50, 1150, 50);
-
-    await new Promise((resolve) => setTimeout(resolve, delay));
-
-    return zodfetchWithVersions(
-      versionedSchemaMap,
-      unversionedSchema,
       url,
-      requestInit,
+      retryCount,
-      options,
+      requestHeaders: fullRequestInit?.headers,
-      retryCount + 1
+      responseHeaders: Object.fromEntries(response.headers.entries()),
-    );
+    });
-  }
 
-  if (response.status !== 200) {
+    if (
-    throw new Error(
+      (!requestInit || requestInit.method === "GET") &&
-      options?.errorMessage ?? `Failed to fetch ${url}, got status code ${response.status}`
+      response.status === 404 &&
-    );
+      options?.optional
-  }
+    ) {
+      // @ts-ignore
+      return;
+    }
 
-  const jsonBody = await response.json();
+    if (response.status >= 400 && response.status < 500) {
+      const rawBody = await safeResponseText(response);
+      const body = safeJsonParse(rawBody);
 
-  const version = response.headers.get("trigger-version");
+      logger.error(`[ApiClient] zodfetchWithVersions failed with ${response.status}`, {
+        url,
+        retryCount,
+        requestHeaders: fullRequestInit?.headers,
+        responseHeaders: Object.fromEntries(response.headers.entries()),
+        status: response.status,
+        rawBody,
+      });
+
+      if (body && body.error) {
+        throw new Error(body.error);
+      } else {
+        throw new Error(rawBody);
+      }
+    }
+
+    if (response.status >= 500 && retryCount < MAX_RETRIES) {
+      // retry with exponential backoff and jitter
+      const delay = exponentialBackoff(retryCount + 1);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return zodfetchWithVersions(
+        logger,
+        versionedSchemaMap,
+        unversionedSchema,
+        url,
+        requestInit,
+        options,
+        retryCount + 1
+      );
+    }
+
+    if (response.status !== 200) {
+      const rawBody = await safeResponseText(response);
+
+      logger.error(`[ApiClient] zodfetchWithVersions failed with ${response.status}`, {
+        url,
+        retryCount,
+        requestHeaders: fullRequestInit?.headers,
+        responseHeaders: Object.fromEntries(response.headers.entries()),
+        status: response.status,
+        rawBody,
+      });
+
+      throw new Error(
+        options?.errorMessage ?? `Failed to fetch ${url}, got status code ${response.status}`
+      );
+    }
+
+    const jsonBody = await response.json();
+
+    const version = response.headers.get("trigger-version");
+
+    if (!version) {
+      return {
+        version: "unversioned",
+        body: unversionedSchema.parse(jsonBody),
+      };
+    }
+
+    const versionedSchema = versionedSchemaMap[version];
+
+    if (!versionedSchema) {
+      throw new UnknownVersionError(version);
+    }
 
-  if (!version) {
     return {
-      version: "unversioned",
+      version,
-      body: unversionedSchema.parse(jsonBody),
+      body: versionedSchema.parse(jsonBody),
     };
+  } catch (error) {
+    if (error instanceof UnknownVersionError) {
+      throw error;
+    }
+
+    logger.error(`[ApiClient] zodfetchWithVersions failed with a connection error`, {
+      url,
+      retryCount,
+      error,
+    });
+
+    if (retryCount < MAX_RETRIES) {
+      // retry with exponential backoff and jitter
+      const delay = exponentialBackoff(retryCount + 1);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return zodfetchWithVersions(
+        logger,
+        versionedSchemaMap,
+        unversionedSchema,
+        url,
+        requestInit,
+        options,
+        retryCount + 1
+      );
+    }
+
+    throw error;
   }
-
-  const versionedSchema = versionedSchemaMap[version];
-
-  if (!versionedSchema) {
-    throw new Error(`Unknown version ${version}`);
-  }
-
-  return {
-    version,
-    body: versionedSchema.parse(jsonBody),
-  };
 }
 
 function requestInitWithCache(requestInit?: RequestInit): RequestInit {
@@ -873,9 +951,9 @@ async function fetchHead(
   };
   const response = await fetch(url, requestInitWithCache(requestInit));
 
-  if (response.status >= 500 && retryCount < 6) {
+  if (response.status >= 500 && retryCount < MAX_RETRIES) {
     // retry with exponential backoff and jitter
-    const delay = exponentialBackoff(retryCount + 1, 2, 50, 1150, 50);
+    const delay = exponentialBackoff(retryCount + 1);
 
     await new Promise((resolve) => setTimeout(resolve, delay));
 
@@ -897,56 +975,80 @@ async function zodfetch<TResponseSchema extends z.ZodTypeAny, TOptional extends
 ): Promise<
   TOptional extends true ? z.infer<TResponseSchema> | undefined : z.infer<TResponseSchema>
 > {
-  const response = await fetch(url, requestInitWithCache(requestInit));
+  try {
+    const response = await fetch(url, requestInitWithCache(requestInit));
 
-  if (
+    if (
-    (!requestInit || requestInit.method === "GET") &&
+      (!requestInit || requestInit.method === "GET") &&
-    response.status === 404 &&
+      response.status === 404 &&
-    options?.optional
+      options?.optional
-  ) {
+    ) {
-    // @ts-ignore
+      // @ts-ignore
-    return;
+      return;
+    }
+
+    if (response.status >= 400 && response.status < 500) {
+      const body = await response.json();
+
+      throw new Error(body.error);
+    }
+
+    if (response.status >= 500 && retryCount < MAX_RETRIES) {
+      // retry with exponential backoff and jitter
+      const delay = exponentialBackoff(retryCount + 1);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return zodfetch(schema, url, requestInit, options, retryCount + 1);
+    }
+
+    if (response.status !== 200) {
+      throw new Error(
+        options?.errorMessage ?? `Failed to fetch ${url}, got status code ${response.status}`
+      );
+    }
+
+    const jsonBody = await response.json();
+
+    return schema.parse(jsonBody);
+  } catch (error) {
+    if (retryCount < MAX_RETRIES) {
+      // retry with exponential backoff and jitter
+      const delay = exponentialBackoff(retryCount + 1);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return zodfetch(schema, url, requestInit, options, retryCount + 1);
+    }
+
+    throw error;
   }
-
-  if (response.status >= 400 && response.status < 500) {
-    const body = await response.json();
-
-    throw new Error(body.error);
-  }
-
-  if (response.status >= 500 && retryCount < 6) {
-    // retry with exponential backoff and jitter
-    const delay = exponentialBackoff(retryCount + 1, 2, 50, 1150, 50);
-
-    await new Promise((resolve) => setTimeout(resolve, delay));
-
-    return zodfetch(schema, url, requestInit, options, retryCount + 1);
-  }
-
-  if (response.status !== 200) {
-    throw new Error(
-      options?.errorMessage ?? `Failed to fetch ${url}, got status code ${response.status}`
-    );
-  }
-
-  const jsonBody = await response.json();
-
-  return schema.parse(jsonBody);
 }
 
-function exponentialBackoff(
+// First retry will have a delay of 80ms, second 160ms, third 320ms, etc.
-  retryCount: number,
+function exponentialBackoff(retryCount: number): number {
-  exponential: number,
-  minDelay: number,
-  maxDelay: number,
-  jitter: number
-): number {
   // Calculate the delay using the exponential backoff formula
-  const delay = Math.min(Math.pow(exponential, retryCount) * minDelay, maxDelay);
+  const delay = Math.min(Math.pow(EXPONENT_FACTOR, retryCount) * MIN_DELAY_IN_MS, MAX_DELAY_IN_MS);
 
   // Calculate the jitter
-  const jitterValue = Math.random() * jitter;
+  const jitterValue = Math.random() * JITTER_IN_MS;
 
   // Return the calculated delay with jitter
   return delay + jitterValue;
 }
+
+function safeJsonParse(rawBody: string) {
+  try {
+    return JSON.parse(rawBody);
+  } catch (error) {
+    return;
+  }
+}
+
+async function safeResponseText(response: Response) {
+  try {
+    return await response.text();
+  } catch (error) {
+    return "";
+  }
+}

packages/trigger-sdk/src/io.ts

@@ -1,6 +1,7 @@
 import {
   API_VERSIONS,
   CachedTask,
+  CompleteTaskBodyV2Input,
   ConnectionAuth,
   CronOptions,
   ErrorWithStackSchema,
@@ -11,6 +12,7 @@ import {
   FetchTimeoutOptions,
   InitialStatusUpdate,
   IntervalOptions,
+  RunTaskBodyInput,
   RunTaskOptions,
   SendEvent,
   SendEventOptions,
@@ -1161,19 +1163,18 @@ export class IO {
 
     const runOptions = { ...(options ?? {}), parseOutput: undefined };
 
-    const response = await this._apiClient.runTask(
+    const response = await this.#doRunTask({
-      this._id,
+      idempotencyKey,
-      {
+      displayKey: typeof cacheKey === "string" ? cacheKey : undefined,
-        idempotencyKey,
+      noop: false,
-        displayKey: typeof cacheKey === "string" ? cacheKey : undefined,
+      ...(runOptions ?? {}),
-        noop: false,
+      parentId,
-        ...(runOptions ?? {}),
+    });
-        parentId,
+
-      },
+    if (!response) {
-      {
+      this.#forceYield("failed_task_run");
-        cachedTasksCursor: this._cachedTasksCursor,
+      throw new Error("Failed to run task"); // this shouldn't actually happen, because forceYield will throw
-      }
+    }
-    );
 
     const task =
       response.version === API_VERSIONS.LAZY_LOADED_CACHED_TASKS
@@ -1267,11 +1268,16 @@ export class IO {
 
         this.#detectAutoYield("before_complete_task", 500, task, output);
 
-        const completedTask = await this._apiClient.completeTask(this._id, task.id, {
+        const completedTask = await this.#doCompleteTask(task.id, {
           output,
           properties: task.outputProperties ?? undefined,
         });
 
+        if (!completedTask) {
+          this.#forceYield("before_complete_task", task, output);
+          throw new Error("Failed to complete task"); // this shouldn't actually happen, because forceYield will throw
+        }
+
         if (completedTask.forceYield) {
           this._logger.debug("Forcing yield after task completed", {
             idempotencyKey,
@@ -1448,6 +1454,24 @@ export class IO {
     this._cachedTasks.set(task.idempotencyKey, task);
   }
 
+  async #doRunTask(task: RunTaskBodyInput) {
+    try {
+      return await this._apiClient.runTask(this._id, task, {
+        cachedTasksCursor: this._cachedTasksCursor,
+      });
+    } catch (error) {
+      return;
+    }
+  }
+
+  async #doCompleteTask(id: string, task: CompleteTaskBodyV2Input) {
+    try {
+      return await this._apiClient.completeTask(this._id, id, task);
+    } catch (error) {
+      return;
+    }
+  }
+
   #detectAutoYield(location: string, threshold: number = 1500, task?: ServerTask, output?: string) {
     const timeRemaining = this.#getRemainingTimeInMillis();
 
@@ -1469,11 +1493,24 @@ export class IO {
     }
   }
 
-  #forceYield(location: string) {
+  #forceYield(location: string, task?: ServerTask, output?: string) {
     const timeRemaining = this.#getRemainingTimeInMillis();
 
     if (timeRemaining) {
-      throw new AutoYieldExecutionError(location, timeRemaining, this.#getTimeElapsed());
+      if (task) {
+        throw new AutoYieldWithCompletedTaskExecutionError(
+          task.id,
+          task.outputProperties ?? [],
+          {
+            location,
+            timeRemaining,
+            timeElapsed: this.#getTimeElapsed(),
+          },
+          output
+        );
+      } else {
+        throw new AutoYieldExecutionError(location, timeRemaining, this.#getTimeElapsed());
+      }
     }
   }
 

packages/trigger-sdk/src/triggerClient.ts

@@ -198,11 +198,15 @@ export class TriggerClient {
   constructor(options: Prettify<TriggerClientOptions>) {
     this.id = options.id;
     this.#options = options;
-    this.#client = new ApiClient(this.#options);
     this.#internalLogger = new Logger("trigger.dev", this.#options.verbose ? "debug" : "log", [
       "output",
       "noopTasksSet",
     ]);
+    this.#client = new ApiClient({
+      logLevel: this.#options.verbose ? "debug" : "log",
+      ...this.#options,
+    });
+
     this.#envStore = new KeyValueStore(this.#client);
   }
 

references/job-catalog/src/stressTest.ts

@@ -23,23 +23,14 @@ client.defineJob({
         `task-${i}`,
         async (task) => {
           return {
-            output: "a".repeat(300 * 1024),
+            output: "a".repeat(30),
           };
         },
         { name: `Task ${i}` }
       );
-    }
 
-    // Now run a single task with 5MB output
+      await new Promise((resolve) => setTimeout(resolve, 2000));
-    await io.runTask(
+    }
-      `task-5mb`,
-      async (task) => {
-        return {
-          output: "a".repeat(5 * 1024 * 1024),
-        };
-      },
-      { name: `Task 5MB` }
-    );
 
     // Now do a wait for 5 seconds
     await io.wait("wait", 5);