github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Eric Allam
2024-05-21 14:07:16 +01:00
parent f243eab9c9
commit 3f8b6d8fce
7 changed files with 271 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/strange-sheep-pull.md
+6
View File
@@ -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
+2 -1
View File
@@ -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
+3 -5
View File
@@ -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 { SafeAsyncLocalStorage } from "../utils/safeAsyncLocalStorage";
import { APIError } from "../apiErrors";
import { version } from "../../../package.json";
import { ZodFetchOptions, zodfetch } from "../zodfetch";
export type TriggerOptions = {
spanParentAsLink?: boolean;
packages/trigger-sdk/src/apiClient.ts
+199 -97
View File
@@ -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<TVersionedResponseBodyMap, TUnversionedResponseBodySchema> | undefined
: VersionedResponseBody<TVersionedResponseBodyMap, TUnversionedResponseBodySchema>
> {
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<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 (
(!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 "";
}
}
packages/trigger-sdk/src/io.ts
+53 -16
View File
@@ -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());
}
}
}
packages/trigger-sdk/src/triggerClient.ts
+5 -1
View File
@@ -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
+3 -12
View File
@@ -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 io.runTask(
`task-5mb`,
async (task) => {
return {
output: "a".repeat(5 * 1024 * 1024),
};
},
{ name: `Task 5MB` }
);
await new Promise((resolve) => setTimeout(resolve, 2000));
}
// Now do a wait for 5 seconds
await io.wait("wait", 5);