diff --git a/.changeset/strange-sheep-pull.md b/.changeset/strange-sheep-pull.md new file mode 100644 index 000000000..518a7949b --- /dev/null +++ b/.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 diff --git a/packages/trigger-sdk/src/apiClient.ts b/packages/trigger-sdk/src/apiClient.ts index c487691f7..e97fe9bfc 100644 --- a/packages/trigger-sdk/src/apiClient.ts +++ b/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", { - task, - }); + this.#logger.debug(`[ApiClient] runTask ${task.displayKey}`); 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 | undefined : VersionedResponseBody > { - const response = await fetch(url, requestInitWithCache(requestInit)); + try { + const fullRequestInit = requestInitWithCache(requestInit); - if ( - (!requestInit || requestInit.method === "GET") && - response.status === 404 && - options?.optional - ) { - // @ts-ignore - return; - } + const response = await fetch(url, fullRequestInit); - 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 zodfetchWithVersions( - versionedSchemaMap, - unversionedSchema, + logger.debug(`[ApiClient] zodfetchWithVersions ${url} (attempt ${retryCount + 1})`, { url, - requestInit, - options, - retryCount + 1 - ); - } + retryCount, + requestHeaders: fullRequestInit?.headers, + responseHeaders: Object.fromEntries(response.headers.entries()), + }); - if (response.status !== 200) { - throw new Error( - options?.errorMessage ?? `Failed to fetch ${url}, got status code ${response.status}` - ); - } + if ( + (!requestInit || requestInit.method === "GET") && + 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", - body: unversionedSchema.parse(jsonBody), + version, + 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 | undefined : z.infer > { - const response = await fetch(url, requestInitWithCache(requestInit)); + try { + const response = await fetch(url, requestInitWithCache(requestInit)); - if ( - (!requestInit || requestInit.method === "GET") && - response.status === 404 && - options?.optional - ) { - // @ts-ignore - return; + if ( + (!requestInit || requestInit.method === "GET") && + response.status === 404 && + options?.optional + ) { + // @ts-ignore + 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( - retryCount: number, - exponential: number, - minDelay: number, - maxDelay: number, - jitter: number -): number { +// First retry will have a delay of 80ms, second 160ms, third 320ms, etc. +function exponentialBackoff(retryCount: 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 ""; + } +} diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index 47c4d0685..b57c4a7c4 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/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( - this._id, - { - idempotencyKey, - displayKey: typeof cacheKey === "string" ? cacheKey : undefined, - noop: false, - ...(runOptions ?? {}), - parentId, - }, - { - cachedTasksCursor: this._cachedTasksCursor, - } - ); + const response = await this.#doRunTask({ + idempotencyKey, + displayKey: typeof cacheKey === "string" ? cacheKey : undefined, + noop: false, + ...(runOptions ?? {}), + parentId, + }); + + if (!response) { + this.#forceYield("failed_task_run"); + 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()); + } } } diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 2edfdacfc..95598cf3f 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -198,11 +198,15 @@ export class TriggerClient { constructor(options: Prettify) { 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); }