From 98046c58ae37bdbef113e64fe9470366fdd565d4 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 7 Aug 2024 22:23:29 +0100 Subject: [PATCH] bringing back the background worker stuff --- packages/cli-v3/src/apiClient.ts | 6 +- packages/cli-v3/src/commands/dev.tsx | 1 + packages/cli-v3/src/dev/backgroundWorker.ts | 1013 +++++++++++++++++ packages/cli-v3/src/dev/devSession.ts | 19 +- packages/cli-v3/src/dev/errors.ts | 103 ++ packages/cli-v3/src/dev/workerRuntime.ts | 272 +++++ packages/cli-v3/src/utilities/dotEnv.ts | 21 + packages/core/src/v3/schemas/build.ts | 16 +- packages/core/src/v3/schemas/messages.ts | 21 +- packages/core/src/v3/schemas/schemas.ts | 37 +- packages/core/src/v3/task-catalog/catalog.ts | 6 +- packages/core/src/v3/task-catalog/index.ts | 10 +- .../src/v3/task-catalog/noopTaskCatalog.ts | 6 +- .../v3/task-catalog/standardTaskCatalog.ts | 8 +- packages/core/src/v3/taskContext/index.ts | 4 +- packages/core/src/v3/taskContext/types.ts | 4 +- packages/core/src/v3/types/index.ts | 7 +- packages/core/src/v3/workers/taskExecutor.ts | 7 +- packages/core/src/v3/zodMessageHandler.ts | 5 + 19 files changed, 1489 insertions(+), 77 deletions(-) create mode 100644 packages/cli-v3/src/dev/backgroundWorker.ts create mode 100644 packages/cli-v3/src/dev/errors.ts create mode 100644 packages/cli-v3/src/dev/workerRuntime.ts create mode 100644 packages/cli-v3/src/utilities/dotEnv.ts diff --git a/packages/cli-v3/src/apiClient.ts b/packages/cli-v3/src/apiClient.ts index a042a25d1..5f9eb62b8 100644 --- a/packages/cli-v3/src/apiClient.ts +++ b/packages/cli-v3/src/apiClient.ts @@ -21,11 +21,9 @@ import { import { zodfetch, ApiError } from "@trigger.dev/core/v3/zodfetch"; export class CliApiClient { - private readonly apiURL: string; - constructor( - apiURL: string, - private readonly accessToken?: string + public readonly apiURL: string, + public readonly accessToken?: string ) { this.apiURL = apiURL.replace(/\/$/, ""); } diff --git a/packages/cli-v3/src/commands/dev.tsx b/packages/cli-v3/src/commands/dev.tsx index 2d5b7269b..6150ac12a 100644 --- a/packages/cli-v3/src/commands/dev.tsx +++ b/packages/cli-v3/src/commands/dev.tsx @@ -152,6 +152,7 @@ async function startDev(options: StartDevOptions) { initialMode="local" showInteractiveDevSession={true} client={projectClient.client} + dashboardUrl={options.login.dashboardUrl} /> ); } diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts new file mode 100644 index 000000000..83fc41ecb --- /dev/null +++ b/packages/cli-v3/src/dev/backgroundWorker.ts @@ -0,0 +1,1013 @@ +import { + BuildManifest, + CreateBackgroundWorkerResponse, + ServerBackgroundWorker, + TaskRunBuiltInError, + TaskRunError, + TaskRunErrorCodes, + TaskRunExecution, + TaskRunExecutionPayload, + TaskRunExecutionResult, + TaskRunFailedExecutionResult, + WorkerManifest, + childToWorkerMessages, + correctErrorStackTrace, + formatDurationMilliseconds, + workerToChildMessages, +} from "@trigger.dev/core/v3"; +import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler"; +import { Evt } from "evt"; +import { ChildProcess, fork } from "node:child_process"; +import { + chalkError, + chalkGrey, + chalkLink, + chalkRun, + chalkSuccess, + chalkTask, + chalkWarning, + chalkWorker, + cliLink, + prettyPrintDate, +} from "../utilities/cliOutput.js"; + +import { logger } from "../utilities/logger.js"; +import { + CancelledProcessError, + CleanupProcessError, + SigKillTimeoutProcessError, + TaskMetadataParseError, + UncaughtExceptionError, + UnexpectedExitError, + getFriendlyErrorMessage, +} from "./errors.js"; + +export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"]; +export class BackgroundWorkerCoordinator { + public onTaskCompleted: Evt<{ + backgroundWorkerId: string; + completion: TaskRunExecutionResult; + worker: BackgroundWorker; + execution: TaskRunExecution; + }> = new Evt(); + public onTaskFailedToRun: Evt<{ + backgroundWorkerId: string; + worker: BackgroundWorker; + completion: TaskRunFailedExecutionResult; + }> = new Evt(); + public onWorkerRegistered: Evt<{ + worker: BackgroundWorker; + id: string; + record: CreateBackgroundWorkerResponse; + }> = new Evt(); + + /** + * @deprecated use onWorkerTaskRunHeartbeat instead + */ + public onWorkerTaskHeartbeat: Evt<{ + id: string; + backgroundWorkerId: string; + worker: BackgroundWorker; + }> = new Evt(); + public onWorkerTaskRunHeartbeat: Evt<{ + id: string; + backgroundWorkerId: string; + worker: BackgroundWorker; + }> = new Evt(); + public onWorkerDeprecated: Evt<{ worker: BackgroundWorker; id: string }> = new Evt(); + private _backgroundWorkers: Map = new Map(); + + constructor(private baseURL: string) { + this.onTaskCompleted.attach(async ({ completion }) => { + if (!completion.ok && typeof completion.retry !== "undefined") { + return; + } + + await this.#notifyWorkersOfTaskCompletion(completion); + }); + + this.onTaskFailedToRun.attach(async ({ completion }) => { + await this.#notifyWorkersOfTaskCompletion(completion); + }); + } + + async #notifyWorkersOfTaskCompletion(completion: TaskRunExecutionResult) { + for (const worker of this._backgroundWorkers.values()) { + await worker.taskRunCompletedNotification(completion); + } + } + + get currentWorkers() { + return Array.from(this._backgroundWorkers.entries()).map(([id, worker]) => ({ + id, + worker, + })); + } + + async cancelRun(id: string, taskRunId: string) { + const worker = this._backgroundWorkers.get(id); + + if (!worker) { + logger.error(`Could not find worker ${id}`); + return; + } + + await worker.cancelRun(taskRunId); + } + + async registerWorker(record: CreateBackgroundWorkerResponse, worker: BackgroundWorker) { + for (const [workerId, existingWorker] of this._backgroundWorkers.entries()) { + if (workerId === record.id) { + continue; + } + + existingWorker.deprecate(); + this.onWorkerDeprecated.post({ worker: existingWorker, id: workerId }); + } + + this._backgroundWorkers.set(record.id, worker); + this.onWorkerRegistered.post({ worker, id: record.id, record }); + + worker.onTaskRunHeartbeat.attach((id) => { + this.onWorkerTaskRunHeartbeat.post({ id, backgroundWorkerId: record.id, worker }); + }); + } + + close() { + for (const worker of this._backgroundWorkers.values()) { + worker.close(); + } + + this._backgroundWorkers.clear(); + } + + async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId?: string) { + const worker = this._backgroundWorkers.get(id); + + if (!worker) { + logger.error(`Could not find worker ${id}`); + return; + } + + try { + const completion = await worker.executeTaskRun(payload, this.baseURL); + + this.onTaskCompleted.post({ + completion, + execution: payload.execution, + worker, + backgroundWorkerId: id, + }); + + return completion; + } catch (error) { + this.onTaskFailedToRun.post({ + backgroundWorkerId: id, + worker, + completion: { + ok: false, + id: payload.execution.run.id, + retry: undefined, + error: + error instanceof Error + ? { + type: "BUILT_IN_ERROR", + name: error.name, + message: error.message, + stackTrace: error.stack ?? "", + } + : { + type: "BUILT_IN_ERROR", + name: "UnknownError", + message: String(error), + stackTrace: "", + }, + }, + }); + } + + return; + } +} + +export type BackgroundWorkerOptions = { + env: Record; + cwd: string; +}; + +export class BackgroundWorker { + private _initialized: boolean = false; + private _handler = new ZodMessageHandler({ + schema: childToWorkerMessages, + }); + + public onTaskRunHeartbeat: Evt = new Evt(); + private _onClose: Evt = new Evt(); + + public deprecated: boolean = false; + public manifest: WorkerManifest | undefined; + public serverWorker: ServerBackgroundWorker | undefined; + public stderr: Array = []; + + _taskRunProcesses: Map = new Map(); + private _taskRunProcessesBeingKilled: Set = new Set(); + + private _closed: boolean = false; + + constructor( + public build: BuildManifest, + public params: BackgroundWorkerOptions + ) {} + + deprecate() { + this.deprecated = true; + } + + close() { + if (this._closed) { + return; + } + + this._closed = true; + + this.onTaskRunHeartbeat.detach(); + + // We need to close all the task run processes + for (const taskRunProcess of this._taskRunProcesses.values()) { + taskRunProcess.cleanup(true); + } + + // Delete worker files + this._onClose.post(); + } + + get inProgressRuns(): Array { + return Array.from(this._taskRunProcesses.keys()); + } + + async initialize() { + if (this._initialized) { + throw new Error("Worker already initialized"); + } + + let resolved = false; + + logger.debug("Initializing worker", { build: this.build, params: this.params }); + + this.manifest = await new Promise((resolve, reject) => { + const child = fork(this.build.workerEntryPoint, { + stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"], + cwd: this.params.cwd, + env: this.params.env, + }); + + // Set a timeout to kill the child process if it doesn't respond + const timeout = setTimeout(() => { + if (resolved) { + return; + } + + resolved = true; + child.kill(); + reject(new Error("Worker timed out")); + }, 20_000); + + child.on("message", async (msg: any) => { + const message = this._handler.parseMessage(msg); + + if (!message.success) { + clearTimeout(timeout); + resolved = true; + reject(new Error(`Failed to parse message: ${message.error}`)); + child.kill(); + return; + } + + if (message.data.type === "INDEX_COMPLETE" && !resolved) { + clearTimeout(timeout); + resolved = true; + resolve(message.data.payload.manifest); + child.kill(); + } else if (message.data.type === "UNCAUGHT_EXCEPTION") { + clearTimeout(timeout); + resolved = true; + reject( + new UncaughtExceptionError(message.data.payload.error, message.data.payload.origin) + ); + child.kill(); + } else if (message.data.type === "TASKS_FAILED_TO_PARSE") { + clearTimeout(timeout); + resolved = true; + reject( + new TaskMetadataParseError(message.data.payload.zodIssues, message.data.payload.tasks) + ); + child.kill(); + } + }); + + child.on("exit", (code) => { + if (!resolved) { + clearTimeout(timeout); + resolved = true; + reject(new Error(`Worker exited with code ${code}`)); + } + }); + + child.stdout?.on("data", (data) => { + logger.log(data.toString()); + }); + + child.stderr?.on("data", (data) => { + this.stderr.push(data.toString()); + }); + }); + + this._initialized = true; + } + + // We need to notify all the task run processes that a task run has completed, + // in case they are waiting for it through triggerAndWait + async taskRunCompletedNotification(completion: TaskRunExecutionResult) { + for (const taskRunProcess of this._taskRunProcesses.values()) { + taskRunProcess.taskRunCompletedNotification(completion); + } + } + + #prefixedMessage(payload: TaskRunExecutionPayload, message: string = "") { + return `[${payload.execution.run.id}.${payload.execution.attempt.number}] ${message}`; + } + + async #getFreshTaskRunProcess( + payload: TaskRunExecutionPayload, + messageId?: string + ): Promise { + logger.debug(this.#prefixedMessage(payload, "getFreshTaskRunProcess()")); + + if (!this.serverWorker) { + throw new Error("Worker not registered"); + } + + if (!this.manifest) { + throw new Error("Worker not initialized"); + } + + this._closed = false; + + logger.debug(this.#prefixedMessage(payload, "killing current task run process before attempt")); + + await this.#killCurrentTaskRunProcessBeforeAttempt(payload.execution.run.id); + + logger.debug(this.#prefixedMessage(payload, "creating new task run process")); + + const processOptions: TaskRunProcessOptions = { + payload, + build: this.build, + env: { + ...this.params.env, + ...payload.environment, + }, + serverWorker: this.serverWorker, + workerManifest: this.manifest, + messageId, + }; + + const taskRunProcess = new TaskRunProcess(processOptions); + + taskRunProcess.onExit.attach(({ pid }) => { + logger.debug(this.#prefixedMessage(payload, "onExit()"), { pid }); + + const taskRunProcess = this._taskRunProcesses.get(payload.execution.run.id); + + // Only delete the task run process if the pid matches + if (taskRunProcess?.pid === pid) { + this._taskRunProcesses.delete(payload.execution.run.id); + } + + if (pid) { + this._taskRunProcessesBeingKilled.delete(pid); + } + }); + + taskRunProcess.onIsBeingKilled.attach((pid) => { + if (pid) { + this._taskRunProcessesBeingKilled.add(pid); + } + }); + + taskRunProcess.onTaskRunHeartbeat.attach((id) => { + this.onTaskRunHeartbeat.post(id); + }); + + await taskRunProcess.initialize(); + + this._taskRunProcesses.set(payload.execution.run.id, taskRunProcess); + + return taskRunProcess; + } + + async #killCurrentTaskRunProcessBeforeAttempt(runId: string) { + const taskRunProcess = this._taskRunProcesses.get(runId); + + if (!taskRunProcess) { + logger.debug(`[${runId}] no current task process to kill`); + return; + } + + logger.debug(`[${runId}] killing current task process`, { + pid: taskRunProcess.pid, + }); + + if (taskRunProcess.isBeingKilled) { + if (this._taskRunProcessesBeingKilled.size > 1) { + await this.#tryGracefulExit(taskRunProcess); + } else { + // If there's only one or none being killed, don't do anything so we can create a fresh one in parallel + } + } else { + // It's not being killed, so kill it + if (this._taskRunProcessesBeingKilled.size > 0) { + await this.#tryGracefulExit(taskRunProcess); + } else { + // There's none being killed yet, so we can kill it without waiting. We still set a timeout to kill it forcefully just in case it sticks around. + taskRunProcess.kill("SIGTERM", 5_000).catch(() => {}); + } + } + } + + async #tryGracefulExit( + taskRunProcess: TaskRunProcess, + kill = false, + initialSignal: number | NodeJS.Signals = "SIGTERM" + ) { + try { + const initialExit = taskRunProcess.onExit.waitFor(5_000); + + if (kill) { + taskRunProcess.kill(initialSignal); + } + + await initialExit; + } catch (error) { + logger.error("TaskRunProcess graceful kill timeout exceeded", error); + + this.#tryForcefulExit(taskRunProcess); + } + } + + async #tryForcefulExit(taskRunProcess: TaskRunProcess) { + try { + const forcedKill = taskRunProcess.onExit.waitFor(5_000); + taskRunProcess.kill("SIGKILL"); + await forcedKill; + } catch (error) { + logger.error("TaskRunProcess forced kill timeout exceeded", error); + throw new SigKillTimeoutProcessError(); + } + } + + async cancelRun(taskRunId: string) { + const taskRunProcess = this._taskRunProcesses.get(taskRunId); + + if (!taskRunProcess) { + return; + } + + await taskRunProcess.cancel(); + } + + // We need to fork the process before we can execute any tasks + async executeTaskRun( + payload: TaskRunExecutionPayload, + baseURL: string, + messageId?: string + ): Promise { + if (this._closed) { + throw new Error("Worker is closed"); + } + + if (!this.manifest) { + throw new Error("Worker not initialized"); + } + + if (!this.serverWorker) { + throw new Error("Worker not registered"); + } + + const { execution } = payload; + // ○ Mar 27 09:17:25.653 -> View logs | 20240326.20 | create-avatar | run_slufhjdfiv8ejnrkw9dsj.1 + + const logsUrl = `${baseURL}/runs/${execution.run.id}`; + + const pipe = chalkGrey("|"); + const bullet = chalkGrey("○"); + const link = chalkLink(cliLink("View logs", logsUrl)); + let timestampPrefix = chalkGrey(prettyPrintDate(payload.execution.attempt.startedAt)); + const workerPrefix = chalkWorker(this.serverWorker.version); + const taskPrefix = chalkTask(execution.task.id); + const runId = chalkRun(`${execution.run.id}.${execution.attempt.number}`); + + logger.log( + `${bullet} ${timestampPrefix} ${chalkGrey( + "->" + )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId}` + ); + + const now = performance.now(); + + const completion = await this.#doExecuteTaskRun(payload, messageId); + + const elapsed = performance.now() - now; + + const retryingText = chalkGrey( + !completion.ok && completion.skippedRetrying + ? " (retrying skipped)" + : !completion.ok && completion.retry !== undefined + ? ` (retrying in ${completion.retry.delay}ms)` + : "" + ); + + const resultText = !completion.ok + ? completion.error.type === "INTERNAL_ERROR" && + (completion.error.code === TaskRunErrorCodes.TASK_EXECUTION_ABORTED || + completion.error.code === TaskRunErrorCodes.TASK_RUN_CANCELLED) + ? chalkWarning("Cancelled") + : `${chalkError("Error")}${retryingText}` + : chalkSuccess("Success"); + + const errorText = !completion.ok + ? formatErrorLog(completion.error) + : "retry" in completion + ? `retry in ${completion.retry}ms` + : ""; + + const elapsedText = chalkGrey(`(${formatDurationMilliseconds(elapsed, { style: "short" })})`); + + timestampPrefix = chalkGrey(prettyPrintDate()); + + logger.log( + `${bullet} ${timestampPrefix} ${chalkGrey( + "->" + )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId} ${pipe} ${resultText} ${elapsedText}${errorText}` + ); + + return completion; + } + + async #doExecuteTaskRun( + payload: TaskRunExecutionPayload, + messageId?: string + ): Promise { + try { + const taskRunProcess = await this.#getFreshTaskRunProcess(payload, messageId); + + logger.debug(this.#prefixedMessage(payload, "executing task run"), { + pid: taskRunProcess.pid, + }); + + const result = await taskRunProcess.execute(); + + // Always kill the worker + await taskRunProcess.cleanup(true); + + if (result.ok) { + return result; + } + + const error = result.error; + + if (error.type === "BUILT_IN_ERROR") { + const mappedError = await this.#correctError(error, payload.execution); + + return { + ...result, + error: mappedError, + }; + } + + return result; + } catch (e) { + if (e instanceof CancelledProcessError) { + return { + id: payload.execution.attempt.id, + ok: false, + retry: undefined, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.TASK_RUN_CANCELLED, + }, + }; + } + + if (e instanceof CleanupProcessError) { + return { + id: payload.execution.attempt.id, + ok: false, + retry: undefined, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.TASK_EXECUTION_ABORTED, + }, + }; + } + + if (e instanceof UnexpectedExitError) { + return { + id: payload.execution.attempt.id, + ok: false, + retry: undefined, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE, + message: getFriendlyErrorMessage(e.code, e.signal, e.stderr), + stackTrace: e.stderr, + }, + }; + } + + return { + id: payload.execution.attempt.id, + ok: false, + retry: undefined, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.TASK_EXECUTION_FAILED, + }, + }; + } + } + + async #correctError( + error: TaskRunBuiltInError, + execution: TaskRunExecution + ): Promise { + return { + ...error, + stackTrace: correctErrorStackTrace(error.stackTrace, this.params.cwd), + }; + } +} + +type TaskRunProcessOptions = { + payload: TaskRunExecutionPayload; + build: BuildManifest; + env: Record; + cwd?: string; + // this is the "index" data + workerManifest: WorkerManifest; + // this is the worker on the server data + serverWorker: ServerBackgroundWorker; + messageId?: string; +}; + +class TaskRunProcess { + private _handler = new ZodMessageHandler({ + schema: childToWorkerMessages, + }); + private _sender: ZodMessageSender; + private _child: ChildProcess | undefined; + private _childPid?: number; + private _attemptPromises: Map< + string, + { resolver: (value: TaskRunExecutionResult) => void; rejecter: (err?: any) => void } + > = new Map(); + private _attemptStatuses: Map = new Map(); + private _currentExecution: TaskRunExecution | undefined; + private _isBeingKilled: boolean = false; + private _isBeingCancelled: boolean = false; + private _stderr: Array = []; + /** + * @deprecated use onTaskRunHeartbeat instead + */ + public onTaskHeartbeat: Evt = new Evt(); + public onTaskRunHeartbeat: Evt = new Evt(); + public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> = + new Evt(); + public onIsBeingKilled: Evt = new Evt(); + + constructor(public readonly options: TaskRunProcessOptions) { + this._sender = new ZodMessageSender({ + schema: workerToChildMessages, + sender: async (message) => { + if (this._child?.connected && !this._isBeingKilled && !this._child.killed) { + this._child.send(message); + } + }, + }); + } + + async cancel() { + this._isBeingCancelled = true; + + await this.cleanup(true); + } + + get runId() { + return this.options.payload.execution.run.id; + } + + get isTest() { + return this.options.payload.execution.run.isTest; + } + + get payload() { + return this.options.payload; + } + + async initialize() { + const { env, build, cwd } = this.options; + + const fullEnv = { + ...(this.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}), + ...env, + }; + + logger.debug(`[${this.runId}] initializing task run process`, { + env: fullEnv, + path: build.workerEntryPoint, + cwd, + }); + + this._child = fork(build.workerEntryPoint, { + stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"], + cwd, + env: fullEnv, + execArgv: ["--trace-uncaught", "--no-warnings=ExperimentalWarning"], + }); + + this._childPid = this._child?.pid; + + this._child.on("message", this.#handleMessage.bind(this)); + this._child.on("exit", this.#handleExit.bind(this)); + this._child.stdout?.on("data", this.#handleLog.bind(this)); + this._child.stderr?.on("data", this.#handleStdErr.bind(this)); + } + + async cleanup(kill: boolean = false) { + if (kill && this._isBeingKilled) { + return; + } + + if (kill) { + this._isBeingKilled = true; + this.onIsBeingKilled.post(this._child?.pid); + } + + logger.debug(`[${this.runId}] cleaning up task run process`, { kill, pid: this.pid }); + + await this._sender.send("CLEANUP", { + flush: true, + kill, + }); + + // FIXME: Something broke READY_TO_DISPOSE. We never receive it, so we always have to kill the process after the timeout below. + + if (!kill) { + return; + } + + // Set a timeout to kill the child process if it hasn't been killed within 5 seconds + setTimeout(() => { + if (this._child && !this._child.killed) { + logger.debug(`[${this.runId}] killing task run process after timeout`, { pid: this.pid }); + + this._child.kill(); + } + }, 5000); + } + + async execute(): Promise { + let resolver: (value: TaskRunExecutionResult) => void; + let rejecter: (err?: any) => void; + + const promise = new Promise((resolve, reject) => { + resolver = resolve; + rejecter = reject; + }); + + this._attemptStatuses.set(this.payload.execution.attempt.id, "PENDING"); + + // @ts-expect-error - We know that the resolver and rejecter are defined + this._attemptPromises.set(payload.execution.attempt.id, { resolver, rejecter }); + + const { execution, traceContext } = this.payload; + + this._currentExecution = execution; + + await this._sender.send("EXECUTE_TASK_RUN", { + execution, + traceContext, + metadata: this.options.serverWorker, + }); + + const result = await promise; + + this._currentExecution = undefined; + + return result; + } + + taskRunCompletedNotification(completion: TaskRunExecutionResult) { + if (!completion.ok && typeof completion.retry !== "undefined") { + return; + } + + if (completion.id === this.runId) { + // We don't need to notify the task run process if it's the same as the one we're running + return; + } + + logger.debug(`[${this.runId}] task run completed notification`, { + completion, + }); + + this._sender.send("TASK_RUN_COMPLETED_NOTIFICATION", { + version: "v2", + completion, + }); + } + + async #handleMessage(msg: any) { + const message = this._handler.parseMessage(msg); + + if (!message.success) { + logger.error(`Dropping message: ${message.error}`, { message }); + return; + } + + switch (message.data.type) { + case "TASK_RUN_COMPLETED": { + const { result, execution } = message.data.payload; + + logger.debug(`[${this.runId}] task run completed`, { + result, + }); + + const promiseStatus = this._attemptStatuses.get(execution.attempt.id); + + if (promiseStatus !== "PENDING") { + return; + } + + this._attemptStatuses.set(execution.attempt.id, "RESOLVED"); + + const attemptPromise = this._attemptPromises.get(execution.attempt.id); + + if (!attemptPromise) { + return; + } + + const { resolver } = attemptPromise; + + resolver(result); + + break; + } + case "READY_TO_DISPOSE": { + logger.debug(`[${this.runId}] task run process is ready to dispose`); + + this.#kill(); + + break; + } + case "TASK_HEARTBEAT": { + if (this.options.messageId) { + this.onTaskRunHeartbeat.post(this.options.messageId); + } else { + this.onTaskHeartbeat.post(message.data.payload.id); + } + + break; + } + } + } + + async #handleExit(code: number | null, signal: NodeJS.Signals | null) { + logger.debug(`[${this.runId}] handle task run process exit`, { code, signal, pid: this.pid }); + + // Go through all the attempts currently pending and reject them + for (const [id, status] of this._attemptStatuses.entries()) { + if (status === "PENDING") { + this._attemptStatuses.set(id, "REJECTED"); + + const attemptPromise = this._attemptPromises.get(id); + + if (!attemptPromise) { + continue; + } + + const { rejecter } = attemptPromise; + + if (this._isBeingCancelled) { + rejecter(new CancelledProcessError()); + } else if (this._isBeingKilled) { + rejecter(new CleanupProcessError()); + } else { + rejecter( + new UnexpectedExitError( + code ?? -1, + signal, + this._stderr.length ? this._stderr.join("\n") : undefined + ) + ); + } + } + } + + this.onExit.post({ code, signal, pid: this.pid }); + } + + #handleLog(data: Buffer) { + if (!this._currentExecution) { + logger.log(`${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`); + + return; + } + + const runId = chalkRun( + `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}` + ); + + logger.log( + `${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${data.toString()}` + ); + } + + #handleStdErr(data: Buffer) { + if (this._isBeingKilled) { + return; + } + + if (!this._currentExecution) { + logger.log(`${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`); + + return; + } + + const runId = chalkRun( + `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}` + ); + + const errorLine = data.toString(); + + logger.log( + `${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${errorLine}` + ); + + if (this._stderr.length > 100) { + this._stderr.shift(); + } + this._stderr.push(errorLine); + } + + #kill() { + logger.debug(`[${this.runId}] #kill()`, { pid: this.pid }); + + if (this._child && !this._child.killed) { + this._child?.kill(); + } + } + + async kill(signal?: number | NodeJS.Signals, timeoutInMs?: number) { + logger.debug(`[${this.runId}] killing task run process`, { + signal, + timeoutInMs, + pid: this.pid, + }); + + this._isBeingKilled = true; + + const killTimeout = this.onExit.waitFor(timeoutInMs); + + this.onIsBeingKilled.post(this._child?.pid); + this._child?.kill(signal); + + if (timeoutInMs) { + await killTimeout; + } + } + + get isBeingKilled() { + return this._isBeingKilled || this._child?.killed; + } + + get pid() { + return this._childPid; + } +} + +function formatErrorLog(error: TaskRunError) { + switch (error.type) { + case "INTERNAL_ERROR": { + return ""; + } + case "STRING_ERROR": { + return `\n\n${chalkError("X Error:")} ${error.raw}\n`; + } + case "CUSTOM_ERROR": { + return `\n\n${chalkError("X Error:")} ${error.raw}\n`; + } + case "BUILT_IN_ERROR": { + return `\n\n${error.stackTrace.replace(/^Error: /, chalkError("X Error: "))}\n`; + } + } +} diff --git a/packages/cli-v3/src/dev/devSession.ts b/packages/cli-v3/src/dev/devSession.ts index e5fab6a3c..333cffba0 100644 --- a/packages/cli-v3/src/dev/devSession.ts +++ b/packages/cli-v3/src/dev/devSession.ts @@ -21,9 +21,11 @@ import { type DevCommandOptions } from "../commands/dev.js"; import { logger } from "../utilities/logger.js"; import { EphemeralDirectory, getTmpDir } from "../utilities/tempDirectories.js"; import { copyManifestToDir } from "../build/manifests.js"; +import { startWorkerRuntime } from "./workerRuntime.js"; export type DevSessionOptions = { name: string | undefined; + dashboardUrl: string; initialMode: "local"; showInteractiveDevSession: boolean | undefined; rawConfig: ResolvedConfig; @@ -31,9 +33,23 @@ export type DevSessionOptions = { client: CliApiClient; }; -export async function startDevSession({ rawConfig }: DevSessionOptions) { +export async function startDevSession({ + rawConfig, + name, + rawArgs, + client, + dashboardUrl, +}: DevSessionOptions) { const destination = getTmpDir(rawConfig.workingDir, "build"); + const runtime = await startWorkerRuntime({ + name, + config: rawConfig, + args: rawArgs, + client, + dashboardUrl, + }); + logger.debug("Starting dev session", { destination: destination.path, rawConfig }); const externalsExtension = createExternalsBuildExtension("dev", rawConfig); @@ -124,6 +140,7 @@ export async function startDevSession({ rawConfig }: DevSessionOptions) { destination.remove(); stopBundling?.().catch((error) => {}); + runtime.shutdown().catch((error) => {}); }, }; } diff --git a/packages/cli-v3/src/dev/errors.ts b/packages/cli-v3/src/dev/errors.ts new file mode 100644 index 000000000..265caea88 --- /dev/null +++ b/packages/cli-v3/src/dev/errors.ts @@ -0,0 +1,103 @@ +import { z } from "zod"; + +export class UncaughtExceptionError extends Error { + constructor( + public readonly originalError: { name: string; message: string; stack?: string }, + public readonly origin: "uncaughtException" | "unhandledRejection" + ) { + super(`Uncaught exception: ${originalError.message}`); + + this.name = "UncaughtExceptionError"; + } +} + +export class TaskMetadataParseError extends Error { + constructor( + public readonly zodIssues: z.ZodIssue[], + public readonly tasks: any + ) { + super(`Failed to parse task metadata`); + + this.name = "TaskMetadataParseError"; + } +} + +export class UnexpectedExitError extends Error { + constructor( + public code: number, + public signal: NodeJS.Signals | null, + public stderr: string | undefined + ) { + super(`Unexpected exit with code ${code}`); + + this.name = "UnexpectedExitError"; + } +} + +export class CleanupProcessError extends Error { + constructor() { + super("Cancelled"); + + this.name = "CleanupProcessError"; + } +} + +export class CancelledProcessError extends Error { + constructor() { + super("Cancelled"); + + this.name = "CancelledProcessError"; + } +} + +export class SigKillTimeoutProcessError extends Error { + constructor() { + super("Process kill timeout"); + + this.name = "SigKillTimeoutProcessError"; + } +} + +export class GracefulExitTimeoutError extends Error { + constructor() { + super("Graceful exit timeout"); + + this.name = "GracefulExitTimeoutError"; + } +} + +export function getFriendlyErrorMessage( + code: number, + signal: NodeJS.Signals | null, + stderr: string | undefined, + dockerMode = true +) { + const message = (text: string) => { + if (signal) { + return `[${signal}] ${text}`; + } else { + return text; + } + }; + + if (code === 137) { + if (dockerMode) { + return message( + "Process ran out of memory! Try choosing a machine preset with more memory for this task." + ); + } else { + // Note: containerState reason and message should be checked to clarify the error + return message( + "Process most likely ran out of memory, but we can't be certain. Try choosing a machine preset with more memory for this task." + ); + } + } + + if (stderr?.includes("OOMErrorHandler")) { + return message( + "Process ran out of memory! Try choosing a machine preset with more memory for this task." + ); + } + + return message(`Process exited with code ${code}.`); +} diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts new file mode 100644 index 000000000..46a5d9514 --- /dev/null +++ b/packages/cli-v3/src/dev/workerRuntime.ts @@ -0,0 +1,272 @@ +import { + BuildManifest, + clientWebsocketMessages, + SemanticInternalAttributes, + serverWebsocketMessages, + TaskRunExecutionLazyAttemptPayload, +} from "@trigger.dev/core/v3"; +import { ResolvedConfig } from "@trigger.dev/core/v3/build"; +import { ClientRequestArgs } from "node:http"; +import { WebSocket } from "partysocket"; +import { ClientOptions, WebSocket as wsWebSocket } from "ws"; +import { CliApiClient } from "../apiClient.js"; +import { DevCommandOptions } from "../commands/dev.js"; +import { chalkError } from "../utilities/cliOutput.js"; +import { logger } from "../utilities/logger.js"; +import { BackgroundWorker, BackgroundWorkerCoordinator } from "./backgroundWorker.js"; +import { + MessagePayloadFromSchema, + ZodMessageHandler, + ZodMessageSender, +} from "@trigger.dev/core/v3/zodMessageHandler"; +import { resolveDotEnvVars } from "../utilities/dotEnv.js"; + +export interface WorkerRuntime { + shutdown(): Promise; + initializeWorker(manifest: BuildManifest): Promise; +} + +export type WorkerRuntimeOptions = { + name: string | undefined; + config: ResolvedConfig; + args: DevCommandOptions; + client: CliApiClient; + dashboardUrl: string; +}; + +export async function startWorkerRuntime(options: WorkerRuntimeOptions): Promise { + const runtime = new DevWorkerRuntime(options); + + await runtime.init(); + + return runtime; +} + +class DevWorkerRuntime implements WorkerRuntime { + private websocket: WebSocket; + private backgroundWorkerCoordinator: BackgroundWorkerCoordinator; + private sender: ZodMessageSender; + private websocketMessageHandler: ZodMessageHandler; + + constructor(public readonly options: WorkerRuntimeOptions) { + const websocketUrl = new URL(this.options.client.apiURL); + websocketUrl.protocol = websocketUrl.protocol.replace("http", "ws"); + websocketUrl.pathname = `/ws`; + + this.sender = new ZodMessageSender({ + schema: clientWebsocketMessages, + sender: async (message) => { + this.websocket.send(JSON.stringify(message)); + }, + }); + + this.backgroundWorkerCoordinator = new BackgroundWorkerCoordinator( + `${options.dashboardUrl}/projects/v3/${options.config.project}` + ); + + this.backgroundWorkerCoordinator.onWorkerTaskRunHeartbeat.attach( + async ({ worker, backgroundWorkerId, id }) => { + await this.sender.send("BACKGROUND_WORKER_MESSAGE", { + backgroundWorkerId, + data: { + type: "TASK_RUN_HEARTBEAT", + id, + }, + }); + } + ); + + this.backgroundWorkerCoordinator.onTaskCompleted.attach( + async ({ backgroundWorkerId, completion, execution }) => { + await this.sender.send("BACKGROUND_WORKER_MESSAGE", { + backgroundWorkerId, + data: { + type: "TASK_RUN_COMPLETED", + completion, + execution, + }, + }); + } + ); + + this.backgroundWorkerCoordinator.onTaskFailedToRun.attach( + async ({ backgroundWorkerId, completion }) => { + await this.sender.send("BACKGROUND_WORKER_MESSAGE", { + backgroundWorkerId, + data: { + type: "TASK_RUN_FAILED_TO_RUN", + completion, + }, + }); + } + ); + + this.backgroundWorkerCoordinator.onWorkerRegistered.attach(async ({ id, worker, record }) => { + await this.sender.send("READY_FOR_TASKS", { + backgroundWorkerId: id, + }); + }); + + this.websocketMessageHandler = new ZodMessageHandler({ + schema: serverWebsocketMessages, + messages: { + SERVER_READY: async (payload) => { + await this.#serverReady(payload); + }, + BACKGROUND_WORKER_MESSAGE: async (payload) => { + await this.#backgroundWorkerMessage(payload); + }, + }, + }); + + this.websocket = new WebSocket(websocketUrl.href, [], { + WebSocket: WebsocketFactory(this.options.client.accessToken!), + connectionTimeout: 10000, + maxRetries: 10, + minReconnectionDelay: 1000, + maxReconnectionDelay: 30000, + reconnectionDelayGrowFactor: 1.4, // This leads to the following retry times: 1, 1.4, 1.96, 2.74, 3.84, 5.38, 7.53, 10.54, 14.76, 20.66 + maxEnqueuedMessages: 250, + }); + + this.websocket.addEventListener("open", async (event) => { + logger.debug("WebSocket opened", { event }); + }); + + this.websocket.addEventListener("close", (event) => { + logger.debug("WebSocket closed", { event }); + }); + + this.websocket.addEventListener("error", (event) => { + logger.log(`${chalkError("WebSocketError:")} ${event.error.message}`); + }); + + this.websocket.addEventListener("message", this.#handleWebsocketMessage.bind(this)); + } + + async init(): Promise {} + + async shutdown(): Promise { + this.websocket.close(); + } + + async initializeWorker(manifest: BuildManifest, options?: { cwd?: string }): Promise { + const env = await this.#getEnvVars(); + + const backgroundWorker = new BackgroundWorker(manifest, { + env, + cwd: this.options.config.workingDir, + }); + + await backgroundWorker.initialize(); + } + + async #getEnvVars(): Promise> { + const environmentVariablesResponse = await this.options.client.getEnvironmentVariables( + this.options.config.project + ); + + const processEnv = gatherProcessEnv(); + const dotEnvVars = resolveDotEnvVars(); + + return { + ...processEnv, + ...(environmentVariablesResponse.success ? environmentVariablesResponse.data.variables : {}), + ...dotEnvVars, + TRIGGER_API_URL: this.options.client.apiURL, + TRIGGER_SECRET_KEY: this.options.client.accessToken!, + OTEL_EXPORTER_OTLP_COMPRESSION: "none", + OTEL_RESOURCE_ATTRIBUTES: JSON.stringify({ + [SemanticInternalAttributes.PROJECT_DIR]: this.options.config.workingDir, + }), + }; + } + + async #handleWebsocketMessage(event: MessageEvent) { + try { + const data = JSON.parse( + typeof event.data === "string" ? event.data : new TextDecoder("utf-8").decode(event.data) + ); + + await this.websocketMessageHandler.handleMessage(data); + } catch (error) { + if (error instanceof Error) { + logger.error("Error while handling websocket message", { error: error.message }); + } else { + logger.error( + "Unkown error while handling websocket message, use `-l debug` for additional output" + ); + logger.debug("Error while handling websocket message", { error }); + } + } + } + + async #serverReady( + payload: MessagePayloadFromSchema<"SERVER_READY", typeof serverWebsocketMessages> + ) { + for (const worker of this.backgroundWorkerCoordinator.currentWorkers) { + await this.sender.send("READY_FOR_TASKS", { + backgroundWorkerId: worker.id, + inProgressRuns: worker.worker.inProgressRuns, + }); + } + } + + async #backgroundWorkerMessage( + payload: MessagePayloadFromSchema<"BACKGROUND_WORKER_MESSAGE", typeof serverWebsocketMessages> + ) { + const message = payload.data; + + logger.debug( + `Received message from worker ${payload.backgroundWorkerId}`, + JSON.stringify({ workerMessage: message }) + ); + + switch (message.type) { + case "CANCEL_ATTEMPT": { + // Need to cancel the attempt somehow here + this.backgroundWorkerCoordinator.cancelRun(payload.backgroundWorkerId, message.taskRunId); + break; + } + case "EXECUTE_RUN_LAZY_ATTEMPT": { + await this.#executeTaskRunLazyAttempt(payload.backgroundWorkerId, message.payload); + } + } + } + + async #executeTaskRunLazyAttempt(id: string, payload: TaskRunExecutionLazyAttemptPayload) { + const attemptResponse = await this.options.client.createTaskRunAttempt(payload.runId); + + if (!attemptResponse.success) { + throw new Error(`Failed to create task run attempt: ${attemptResponse.error}`); + } + + const execution = attemptResponse.data; + + const completion = await this.backgroundWorkerCoordinator.executeTaskRun( + id, + { execution, traceContext: payload.traceContext, environment: payload.environment }, + payload.messageId + ); + + return { execution, completion }; + } +} + +function WebsocketFactory(apiKey: string) { + return class extends wsWebSocket { + constructor(address: string | URL, options?: ClientOptions | ClientRequestArgs) { + super(address, { ...(options ?? {}), headers: { Authorization: `Bearer ${apiKey}` } }); + } + }; +} + +function gatherProcessEnv() { + const env = { + ...process.env, + NODE_ENV: "development", + }; + + // Filter out undefined values + return Object.fromEntries(Object.entries(env).filter(([key, value]) => value !== undefined)); +} diff --git a/packages/cli-v3/src/utilities/dotEnv.ts b/packages/cli-v3/src/utilities/dotEnv.ts new file mode 100644 index 000000000..0caa7e0de --- /dev/null +++ b/packages/cli-v3/src/utilities/dotEnv.ts @@ -0,0 +1,21 @@ +import dotenv from "dotenv"; +import { resolve } from "node:path"; + +export function resolveDotEnvVars(cwd?: string) { + const result: { [key: string]: string } = {}; + + dotenv.config({ + processEnv: result, + path: [".env", ".env.local", ".env.development.local"].map((p) => + resolve(cwd ?? process.cwd(), p) + ), + }); + + process.env.TRIGGER_API_URL && (result.TRIGGER_API_URL = process.env.TRIGGER_API_URL); + + // remove TRIGGER_API_URL and TRIGGER_SECRET_KEY, since those should be coming from the worker + delete result.TRIGGER_API_URL; + delete result.TRIGGER_SECRET_KEY; + + return result; +} diff --git a/packages/core/src/v3/schemas/build.ts b/packages/core/src/v3/schemas/build.ts index ab5ec93b8..ee3cd1916 100644 --- a/packages/core/src/v3/schemas/build.ts +++ b/packages/core/src/v3/schemas/build.ts @@ -1,12 +1,6 @@ import { z } from "zod"; import { ConfigManifest } from "./config.js"; - -export const TaskFile = z.object({ - entry: z.string(), - out: z.string(), -}); - -export type TaskFile = z.infer; +import { TaskFile, TaskManifest } from "./schemas.js"; export const BuildExternal = z.object({ name: z.string(), @@ -53,14 +47,6 @@ export const IndexMessage = z.object({ export type IndexMessage = z.infer; -export const TaskManifest = z.object({ - id: z.string(), - exportName: z.string(), - file: TaskFile, -}); - -export type TaskManifest = z.infer; - export const WorkerManifest = z.object({ tasks: TaskManifest.array(), }); diff --git a/packages/core/src/v3/schemas/messages.ts b/packages/core/src/v3/schemas/messages.ts index b26550ad5..470a0d575 100644 --- a/packages/core/src/v3/schemas/messages.ts +++ b/packages/core/src/v3/schemas/messages.ts @@ -9,18 +9,15 @@ import { EnvironmentType, ProdTaskRunExecution, ProdTaskRunExecutionPayload, - TaskMetadataWithFilePath, + TaskManifest, TaskRunExecutionLazyAttemptPayload, TaskRunExecutionPayload, WaitReason, } from "./schemas.js"; import { TaskResource } from "./resources.js"; +import { WorkerManifest } from "./build.js"; export const BackgroundWorkerServerMessages = z.discriminatedUnion("type", [ - z.object({ - type: z.literal("EXECUTE_RUNS"), - payloads: z.array(TaskRunExecutionPayload), - }), z.object({ type: z.literal("CANCEL_ATTEMPT"), taskAttemptId: z.string(), @@ -86,13 +83,13 @@ export const BackgroundWorkerClientMessages = z.discriminatedUnion("type", [ export type BackgroundWorkerClientMessages = z.infer; -export const BackgroundWorkerProperties = z.object({ +export const ServerBackgroundWorker = z.object({ id: z.string(), version: z.string(), contentHash: z.string(), }); -export type BackgroundWorkerProperties = z.infer; +export type ServerBackgroundWorker = z.infer; export const clientWebsocketMessages = { READY_FOR_TASKS: z.object({ @@ -116,7 +113,7 @@ export const workerToChildMessages = { version: z.literal("v1").default("v1"), execution: TaskRunExecution, traceContext: z.record(z.unknown()), - metadata: BackgroundWorkerProperties, + metadata: ServerBackgroundWorker, }), TASK_RUN_COMPLETED_NOTIFICATION: z.discriminatedUnion("version", [ z.object({ @@ -160,9 +157,9 @@ export const childToWorkerMessages = { execution: TaskRunExecution, result: TaskRunExecutionResult, }), - TASKS_READY: z.object({ + INDEX_COMPLETE: z.object({ version: z.literal("v1").default("v1"), - tasks: TaskMetadataWithFilePath.array(), + manifest: WorkerManifest, }), TASKS_FAILED_TO_PARSE: TaskMetadataFailedToParseData, TASK_HEARTBEAT: z.object({ @@ -201,7 +198,7 @@ export const ProdChildToWorkerMessages = { TASKS_READY: { message: z.object({ version: z.literal("v1").default("v1"), - tasks: TaskMetadataWithFilePath.array(), + tasks: TaskManifest.array(), }), }, TASKS_FAILED_TO_PARSE: { @@ -248,7 +245,7 @@ export const ProdWorkerToChildMessages = { version: z.literal("v1").default("v1"), execution: TaskRunExecution, traceContext: z.record(z.unknown()), - metadata: BackgroundWorkerProperties, + metadata: ServerBackgroundWorker, }), }, TASK_RUN_COMPLETED_NOTIFICATION: { diff --git a/packages/core/src/v3/schemas/schemas.ts b/packages/core/src/v3/schemas/schemas.ts index 81d97db0a..251f8929b 100644 --- a/packages/core/src/v3/schemas/schemas.ts +++ b/packages/core/src/v3/schemas/schemas.ts @@ -147,38 +147,41 @@ export const ScheduleMetadata = z.object({ timezone: z.string(), }); -export const TaskMetadata = z.object({ +const taskMetadata = { id: z.string(), - packageVersion: z.string(), queue: QueueOptions.optional(), retry: RetryOptions.optional(), machine: MachineConfig.optional(), triggerSource: z.string().optional(), schedule: ScheduleMetadata.optional(), -}); +}; + +export const TaskMetadata = z.object(taskMetadata); export type TaskMetadata = z.infer; -export const TaskFileMetadata = z.object({ - filePath: z.string(), - exportName: z.string(), +export const TaskFile = z.object({ + entry: z.string(), + out: z.string(), }); +export type TaskFile = z.infer; + +const taskFileMetadata = { + file: TaskFile, + exportName: z.string(), +}; + +export const TaskFileMetadata = z.object(taskFileMetadata); + export type TaskFileMetadata = z.infer; -export const TaskMetadataWithFilePath = z.object({ - id: z.string(), - packageVersion: z.string(), - queue: QueueOptions.optional(), - retry: RetryOptions.optional(), - machine: MachineConfig.optional(), - triggerSource: z.string().optional(), - schedule: ScheduleMetadata.optional(), - filePath: z.string(), - exportName: z.string(), +export const TaskManifest = z.object({ + ...taskMetadata, + ...taskFileMetadata, }); -export type TaskMetadataWithFilePath = z.infer; +export type TaskManifest = z.infer; export const PostStartCauses = z.enum(["index", "create", "restore"]); export type PostStartCauses = z.infer; diff --git a/packages/core/src/v3/task-catalog/catalog.ts b/packages/core/src/v3/task-catalog/catalog.ts index 4aec1aacc..2d9b8ac2a 100644 --- a/packages/core/src/v3/task-catalog/catalog.ts +++ b/packages/core/src/v3/task-catalog/catalog.ts @@ -1,12 +1,12 @@ -import { TaskFileMetadata, TaskMetadataWithFilePath } from "../schemas/index.js"; +import { TaskFileMetadata, TaskManifest } from "../schemas/index.js"; import { TaskMetadataWithFunctions } from "../types/index.js"; export interface TaskCatalog { registerTaskMetadata(task: TaskMetadataWithFunctions): void; updateTaskMetadata(id: string, task: Partial): void; registerTaskFileMetadata(id: string, metadata: TaskFileMetadata): void; - getAllTaskMetadata(): Array; - getTaskMetadata(id: string): TaskMetadataWithFilePath | undefined; + listTaskManifests(): Array; + getTaskManifest(id: string): TaskManifest | undefined; getTask(id: string): TaskMetadataWithFunctions | undefined; taskExists(id: string): boolean; } diff --git a/packages/core/src/v3/task-catalog/index.ts b/packages/core/src/v3/task-catalog/index.ts index 3bb802357..8e3d2c39a 100644 --- a/packages/core/src/v3/task-catalog/index.ts +++ b/packages/core/src/v3/task-catalog/index.ts @@ -1,6 +1,6 @@ const API_NAME = "task-catalog"; -import { TaskFileMetadata, TaskMetadataWithFilePath } from "../schemas/index.js"; +import { TaskFileMetadata, TaskManifest } from "../schemas/index.js"; import { TaskMetadataWithFunctions } from "../types/index.js"; import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js"; import { type TaskCatalog } from "./catalog.js"; @@ -41,12 +41,12 @@ export class TaskCatalogAPI { this.#getCatalog().registerTaskFileMetadata(id, metadata); } - public getAllTaskMetadata(): Array { - return this.#getCatalog().getAllTaskMetadata(); + public listTaskManifests(): Array { + return this.#getCatalog().listTaskManifests(); } - public getTaskMetadata(id: string): TaskMetadataWithFilePath | undefined { - return this.#getCatalog().getTaskMetadata(id); + public getTaskManifest(id: string): TaskManifest | undefined { + return this.#getCatalog().getTaskManifest(id); } public getTask(id: string): TaskMetadataWithFunctions | undefined { diff --git a/packages/core/src/v3/task-catalog/noopTaskCatalog.ts b/packages/core/src/v3/task-catalog/noopTaskCatalog.ts index 77572f9dd..bd9a1616e 100644 --- a/packages/core/src/v3/task-catalog/noopTaskCatalog.ts +++ b/packages/core/src/v3/task-catalog/noopTaskCatalog.ts @@ -1,4 +1,4 @@ -import { TaskFileMetadata, TaskMetadataWithFilePath } from "../schemas/index.js"; +import { TaskFileMetadata, TaskManifest } from "../schemas/index.js"; import { TaskMetadataWithFunctions } from "../types/index.js"; import { TaskCatalog } from "./catalog.js"; @@ -15,11 +15,11 @@ export class NoopTaskCatalog implements TaskCatalog { // noop } - getAllTaskMetadata(): Array { + listTaskManifests(): Array { return []; } - getTaskMetadata(id: string): TaskMetadataWithFilePath | undefined { + getTaskManifest(id: string): TaskManifest | undefined { return undefined; } diff --git a/packages/core/src/v3/task-catalog/standardTaskCatalog.ts b/packages/core/src/v3/task-catalog/standardTaskCatalog.ts index c2eb2b8d8..681186a5e 100644 --- a/packages/core/src/v3/task-catalog/standardTaskCatalog.ts +++ b/packages/core/src/v3/task-catalog/standardTaskCatalog.ts @@ -1,4 +1,4 @@ -import { TaskFileMetadata, TaskMetadata, TaskMetadataWithFilePath } from "../schemas/index.js"; +import { TaskFileMetadata, TaskMetadata, TaskManifest } from "../schemas/index.js"; import { TaskMetadataWithFunctions } from "../types/index.js"; import { TaskCatalog } from "./catalog.js"; @@ -45,8 +45,8 @@ export class StandardTaskCatalog implements TaskCatalog { } // Return all the tasks, without the functions - getAllTaskMetadata(): Array { - const result: Array = []; + listTaskManifests(): Array { + const result: Array = []; for (const [id, metadata] of this._taskMetadata) { const fileMetadata = this._taskFileMetadata.get(id); @@ -64,7 +64,7 @@ export class StandardTaskCatalog implements TaskCatalog { return result; } - getTaskMetadata(id: string): TaskMetadataWithFilePath | undefined { + getTaskManifest(id: string): TaskManifest | undefined { const metadata = this._taskMetadata.get(id); const fileMetadata = this._taskFileMetadata.get(id); diff --git a/packages/core/src/v3/taskContext/index.ts b/packages/core/src/v3/taskContext/index.ts index 27959f764..0c5334b82 100644 --- a/packages/core/src/v3/taskContext/index.ts +++ b/packages/core/src/v3/taskContext/index.ts @@ -1,5 +1,5 @@ import { Attributes } from "@opentelemetry/api"; -import { BackgroundWorkerProperties, TaskRunContext } from "../schemas/index.js"; +import { ServerBackgroundWorker, TaskRunContext } from "../schemas/index.js"; import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js"; import { TaskContext } from "./types.js"; import { SemanticInternalAttributes } from "../semanticInternalAttributes.js"; @@ -27,7 +27,7 @@ export class TaskContextAPI { return this.#getTaskContext()?.ctx; } - get worker(): BackgroundWorkerProperties | undefined { + get worker(): ServerBackgroundWorker | undefined { return this.#getTaskContext()?.worker; } diff --git a/packages/core/src/v3/taskContext/types.ts b/packages/core/src/v3/taskContext/types.ts index 22a2ce6a0..71606d947 100644 --- a/packages/core/src/v3/taskContext/types.ts +++ b/packages/core/src/v3/taskContext/types.ts @@ -1,6 +1,6 @@ -import { BackgroundWorkerProperties, TaskRunContext } from "../schemas/index.js"; +import { ServerBackgroundWorker, TaskRunContext } from "../schemas/index.js"; export type TaskContext = { ctx: TaskRunContext; - worker: BackgroundWorkerProperties; + worker: ServerBackgroundWorker; }; diff --git a/packages/core/src/v3/types/index.ts b/packages/core/src/v3/types/index.ts index 8d67d2889..1d02c1553 100644 --- a/packages/core/src/v3/types/index.ts +++ b/packages/core/src/v3/types/index.ts @@ -1,9 +1,4 @@ -import { - RetryOptions, - TaskMetadata, - TaskMetadataWithFilePath, - TaskRunContext, -} from "../schemas/index.js"; +import { RetryOptions, TaskMetadata, TaskManifest, TaskRunContext } from "../schemas/index.js"; import { Prettify } from "./utils.js"; export * from "./utils.js"; diff --git a/packages/core/src/v3/workers/taskExecutor.ts b/packages/core/src/v3/workers/taskExecutor.ts index 7aefc37e6..fba3760f5 100644 --- a/packages/core/src/v3/workers/taskExecutor.ts +++ b/packages/core/src/v3/workers/taskExecutor.ts @@ -3,7 +3,7 @@ import { ConsoleInterceptor } from "../consoleInterceptor.js"; import { parseError, sanitizeError } from "../errors.js"; import { TracingSDK, recordSpanException } from "../otel/index.js"; import { - BackgroundWorkerProperties, + ServerBackgroundWorker, Config, TaskRunContext, TaskRunErrorCodes, @@ -27,6 +27,7 @@ import { accessoryAttributes } from "../utils/styleAttributes.js"; import { UsageMeasurement } from "../usage/types.js"; import { ApiError, RateLimitError } from "../apiClient/errors.js"; import { TriggerConfig } from "../index.js"; +import { pkg } from "../../pkg.js"; export type TaskExecutorOptions = { tracingSDK: TracingSDK; @@ -59,7 +60,7 @@ export class TaskExecutor { async execute( execution: TaskRunExecution, - worker: BackgroundWorkerProperties, + worker: ServerBackgroundWorker, traceContext: Record, usage: UsageMeasurement ): Promise<{ result: TaskRunExecutionResult }> { @@ -78,7 +79,7 @@ export class TaskExecutor { this._tracingSDK.asyncResourceDetector.resolveWithAttributes({ ...taskContext.attributes, - [SemanticInternalAttributes.SDK_VERSION]: this.task.packageVersion, + [SemanticInternalAttributes.SDK_VERSION]: pkg.version, [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript", }); diff --git a/packages/core/src/v3/zodMessageHandler.ts b/packages/core/src/v3/zodMessageHandler.ts index eda4b1fff..7f0ac0c61 100644 --- a/packages/core/src/v3/zodMessageHandler.ts +++ b/packages/core/src/v3/zodMessageHandler.ts @@ -36,6 +36,11 @@ export type MessageFromSchema< payload: z.input; }; +export type MessagePayloadFromSchema< + K extends keyof TMessageCatalog, + TMessageCatalog extends ZodMessageCatalogSchema, +> = z.output; + export type MessageFromCatalog = { [K in keyof TMessageCatalog]: MessageFromSchema; }[keyof TMessageCatalog];