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

deploy run executions WIP, extracted TaskRunProcess into 1 place

Browse Source
This commit is contained in:
Eric Allam
2024-08-19 17:08:58 +01:00
committed by Eric Allam
parent cc8eb34e10
commit 6cabc8da5f
17 changed files with 1601 additions and 478 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+1 -1
View File
@@ -41,7 +41,7 @@
"type": "node-terminal",
"request": "launch",
"name": "Debug V3 Deploy CLI",
"command": "pnpm exec triggerdev deploy",
"command": "pnpm exec triggerdev deploy --self-hosted",
"cwd": "${workspaceFolder}/references/v3-catalog",
"sourceMaps": true
},
packages/cli-v3/package.json
+1 -1
View File
@@ -116,7 +116,7 @@
"source-map-support": "0.5.21",
"terminal-link": "^3.0.0",
"tiny-invariant": "^1.2.0",
"tinyexec": "^0.1.4",
"tinyexec": "^0.2.0",
"ws": "^8.18.0",
"xdg-app-paths": "^8.3.0",
"zod": "3.23.8",
packages/cli-v3/src/commands/deploy.ts
+1
View File
@@ -355,6 +355,7 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
noCache: options.noCache,
push: options.push,
registryHost,
registry: options.registry,
deploymentId: deployment.id,
deploymentVersion: deployment.version,
imageTag: deployment.imageTag,
packages/cli-v3/src/deploy/buildImage.ts
+16 -2
View File
@@ -430,7 +430,7 @@ async function generateBunContainerfile(buildManifest: BuildManifest) {
.join("\n");
return `
FROM oven/bun:1 AS base
FROM imbios/bun-node:22-debian AS base
RUN apt-get update && apt-get --fix-broken install -y && apt-get install -y --no-install-recommends busybox ca-certificates dumb-init git openssl && apt-get clean && rm -rf /var/lib/apt/lists/*
FROM base AS install
@@ -484,6 +484,13 @@ FROM base AS final
USER bun
WORKDIR /app
ARG TRIGGER_PROJECT_ID
ARG TRIGGER_DEPLOYMENT_ID
ARG TRIGGER_DEPLOYMENT_VERSION
ARG TRIGGER_CONTENT_HASH
ARG TRIGGER_PROJECT_REF
ARG NODE_EXTRA_CA_CERTS
ENV TRIGGER_PROJECT_ID=\${TRIGGER_PROJECT_ID} \
TRIGGER_DEPLOYMENT_ID=\${TRIGGER_DEPLOYMENT_ID} \
TRIGGER_DEPLOYMENT_VERSION=\${TRIGGER_DEPLOYMENT_VERSION} \
@@ -498,7 +505,7 @@ COPY --from=install --chown=bun:bun /app ./
# Copy the index.json file from the indexer stage
COPY --from=indexer --chown=bun:bun /app/index.json ./
ENTRYPOINT [ "dumb-init", "bun", "run", "${buildManifest.workerEntryPoint}" ]
ENTRYPOINT [ "dumb-init", "node", "${buildManifest.workerEntryPoint}" ]
CMD []
`;
}
@@ -576,6 +583,13 @@ FROM base AS final
USER node
WORKDIR /app
ARG TRIGGER_PROJECT_ID
ARG TRIGGER_DEPLOYMENT_ID
ARG TRIGGER_DEPLOYMENT_VERSION
ARG TRIGGER_CONTENT_HASH
ARG TRIGGER_PROJECT_REF
ARG NODE_EXTRA_CA_CERTS
ENV TRIGGER_PROJECT_ID=\${TRIGGER_PROJECT_ID} \
TRIGGER_DEPLOYMENT_ID=\${TRIGGER_DEPLOYMENT_ID} \
TRIGGER_DEPLOYMENT_VERSION=\${TRIGGER_DEPLOYMENT_VERSION} \
packages/cli-v3/src/dev/backgroundWorker.ts
+13 -369
View File
@@ -9,21 +9,16 @@ import {
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
WorkerManifest,
childToWorkerMessages,
correctErrorStackTrace,
indexerToWorkerMessages,
workerToChildMessages,
} from "@trigger.dev/core/v3";
import {
ZodMessageHandler,
ZodMessageSender,
parseMessageFromCatalog,
} from "@trigger.dev/core/v3/zodMessageHandler";
import { parseMessageFromCatalog } from "@trigger.dev/core/v3/zodMessageHandler";
import { Evt } from "evt";
import { ChildProcess, fork } from "node:child_process";
import { chalkError, chalkGrey, chalkRun, prettyPrintDate } from "../utilities/cliOutput.js";
import { fork } from "node:child_process";
import { execPathForRuntime } from "@trigger.dev/core/v3/build";
import { join } from "node:path";
import { TaskRunProcess, TaskRunProcessOptions } from "../executions/taskRunProcess.js";
import { eventBus } from "../utilities/eventBus.js";
import { writeJSONFile } from "../utilities/fileSystem.js";
import { logger } from "../utilities/logger.js";
@@ -35,8 +30,7 @@ import {
UncaughtExceptionError,
UnexpectedExitError,
getFriendlyErrorMessage,
} from "./errors.js";
import { execPathForRuntime } from "@trigger.dev/core/v3/build";
} from "../executions/errors.js";
export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"];
export class BackgroundWorkerCoordinator {
@@ -212,7 +206,7 @@ export class BackgroundWorker {
public serverWorker: ServerBackgroundWorker | undefined;
_taskRunProcesses: Map<string, TaskRunProcess> = new Map();
private _taskRunProcessesBeingKilled: Set<number> = new Set();
private _taskRunProcessesBeingKilled: Map<number, TaskRunProcess> = new Map();
private _closed: boolean = false;
@@ -379,14 +373,10 @@ export class BackgroundWorker {
const processOptions: TaskRunProcessOptions = {
payload,
build: this.build,
env: {
...this.params.env,
...payload.environment,
TRIGGER_WORKER_MANIFEST_PATH: this.workerManifestPath,
NODE_OPTIONS: this.build.loaderEntryPoint
? `--import=${this.build.loaderEntryPoint} ${process.env.NODE_OPTIONS ?? ""}`
: process.env.NODE_OPTIONS ?? "",
},
serverWorker: this.serverWorker,
workerManifest: this.manifest,
@@ -410,9 +400,9 @@ export class BackgroundWorker {
}
});
taskRunProcess.onIsBeingKilled.attach((pid) => {
if (pid) {
this._taskRunProcessesBeingKilled.add(pid);
taskRunProcess.onIsBeingKilled.attach((taskRunProcess) => {
if (taskRunProcess.pid) {
this._taskRunProcessesBeingKilled.set(taskRunProcess.pid, taskRunProcess);
}
});
@@ -420,6 +410,10 @@ export class BackgroundWorker {
this.onTaskRunHeartbeat.post(id);
});
taskRunProcess.onReadyToDispose.attach(async () => {
await taskRunProcess.kill();
});
await taskRunProcess.initialize();
this._taskRunProcesses.set(payload.execution.run.id, taskRunProcess);
@@ -621,353 +615,3 @@ export class BackgroundWorker {
};
}
}
type TaskRunProcessOptions = {
payload: TaskRunExecutionPayload;
build: BuildManifest;
env: Record<string, string>;
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<typeof workerToChildMessages>;
private _child: ChildProcess | undefined;
private _childPid?: number;
private _attemptPromises: Map<
string,
{ resolver: (value: TaskRunExecutionResult) => void; rejecter: (err?: any) => void }
> = new Map();
private _attemptStatuses: Map<string, "PENDING" | "REJECTED" | "RESOLVED"> = new Map();
private _currentExecution: TaskRunExecution | undefined;
private _isBeingKilled: boolean = false;
private _isBeingCancelled: boolean = false;
private _stderr: Array<string> = [];
/**
* @deprecated use onTaskRunHeartbeat instead
*/
public onTaskHeartbeat: Evt<string> = new Evt();
public onTaskRunHeartbeat: Evt<string> = new Evt();
public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
new Evt();
public onIsBeingKilled: Evt<number | undefined> = new Evt();
private _debuggingPort: number | undefined;
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"],
execPath: execPathForRuntime(build.runtime),
});
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<TaskRunExecutionResult> {
let resolver: (value: TaskRunExecutionResult) => void;
let rejecter: (err?: any) => void;
const promise = new Promise<TaskRunExecutionResult>((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(this.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;
}
}
packages/cli-v3/src/entryPoints/deploy-indexer.ts
+7 -1
View File
@@ -173,7 +173,13 @@ async function indexDeployment({
});
}
const workerManifest: WorkerManifest = { tasks, configPath: buildManifest.configPath };
const workerManifest: WorkerManifest = {
tasks,
configPath: buildManifest.configPath,
runtime: buildManifest.runtime,
workerEntryPoint: buildManifest.workerEntryPoint,
loaderEntryPoint: buildManifest.loaderEntryPoint,
};
console.log("Writing index.json", process.cwd());
packages/cli-v3/src/entryPoints/deploy.ts
+1110 -1
View File
File diff suppressed because it is too large Load Diff
packages/cli-v3/src/entryPoints/dev.ts
+13 -20
View File
@@ -1,7 +1,6 @@
import type { Tracer } from "@opentelemetry/api";
import type { Logger } from "@opentelemetry/api-logs";
import {
childToWorkerMessages,
clock,
type HandleErrorFunction,
logger,
@@ -10,10 +9,11 @@ import {
taskCatalog,
TaskRunErrorCodes,
TaskRunExecution,
WorkerToExecutorMessageCatalog,
TriggerConfig,
TriggerTracer,
WorkerManifest,
workerToChildMessages,
ExecutorToWorkerMessageCatalog,
} from "@trigger.dev/core/v3";
import { DevRuntimeManager } from "@trigger.dev/core/v3/dev";
import {
@@ -29,7 +29,7 @@ import {
TracingSDK,
usage,
} from "@trigger.dev/core/v3/workers";
import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
import { readFile } from "node:fs/promises";
import sourceMapSupport from "source-map-support";
import { VERSION } from "../version.js";
@@ -67,13 +67,6 @@ process.on("uncaughtException", function (error, origin) {
}
});
const sender = new ZodMessageSender({
schema: childToWorkerMessages,
sender: async (message) => {
process.send?.(message);
},
});
taskCatalog.setGlobalTaskCatalog(new StandardTaskCatalog());
const durableClock = new DurableClock();
clock.setGlobalClock(durableClock);
@@ -156,10 +149,12 @@ let _execution: TaskRunExecution | undefined;
let _isRunning = false;
let _tracingSDK: TracingSDK | undefined;
const handler = new ZodMessageHandler({
schema: workerToChildMessages,
messages: {
EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }) => {
const zodIpc = new ZodIpcConnection({
listenSchema: ExecutorToWorkerMessageCatalog,
emitSchema: WorkerToExecutorMessageCatalog,
process,
handlers: {
EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }, sender) => {
if (_isRunning) {
console.error("Worker is already running a task");
@@ -184,6 +179,8 @@ const handler = new ZodMessageHandler({
const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn, workerManifest } =
await bootstrap();
_tracingSDK = tracingSDK;
const taskManifest = workerManifest.tasks.find((t) => t.id === execution.task.id);
if (!taskManifest) {
@@ -300,7 +297,7 @@ const handler = new ZodMessageHandler({
}
}
},
CLEANUP: async ({ flush, kill }) => {
CLEANUP: async ({ flush, kill }, sender) => {
if (kill) {
await _tracingSDK?.flush();
// Now we need to exit the process
@@ -314,10 +311,6 @@ const handler = new ZodMessageHandler({
},
});
process.on("message", async (msg: any) => {
await handler.handleMessage(msg);
});
process.title = "trigger-dev-worker";
async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeconds: number = 30) {
@@ -325,7 +318,7 @@ async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeco
while (true) {
if (_isRunning && _execution) {
try {
await sender.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
} catch (err) {
console.error("Failed to send HEARTBEAT message", err);
}
packages/cli-v3/src/entryPoints/indexer.ts
+3
View File
@@ -122,6 +122,9 @@ await sendMessageInCatalog(
manifest: {
tasks,
configPath: buildManifest.configPath,
runtime: buildManifest.runtime,
workerEntryPoint: buildManifest.workerEntryPoint,
loaderEntryPoint: buildManifest.loaderEntryPoint,
},
},
async (msg) => {
packages/cli-v3/src/dev/errors.ts → packages/cli-v3/src/executions/errors.ts
View File
packages/cli-v3/src/executions/taskRunProcess.ts
+407
View File
@@ -0,0 +1,407 @@
import {
ServerBackgroundWorker,
TaskRunExecution,
TaskRunExecutionPayload,
TaskRunExecutionResult,
WorkerToExecutorMessageCatalog,
ExecutorToWorkerMessageCatalog,
WorkerManifest,
} from "@trigger.dev/core/v3";
import {
type WorkerToExecutorProcessConnection,
ZodIpcConnection,
} from "@trigger.dev/core/v3/zodIpc";
import { Evt } from "evt";
import { ChildProcess, fork } from "node:child_process";
import { chalkError, chalkGrey, chalkRun, prettyPrintDate } from "../utilities/cliOutput.js";
import { execPathForRuntime } from "@trigger.dev/core/v3/build";
import { logger } from "../utilities/logger.js";
import {
CancelledProcessError,
CleanupProcessError,
GracefulExitTimeoutError,
UnexpectedExitError,
} from "./errors.js";
import { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
export type OnWaitForDurationMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
"WAIT_FOR_DURATION"
>;
export type OnWaitForTaskMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
"WAIT_FOR_TASK"
>;
export type OnWaitForBatchMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
"WAIT_FOR_BATCH"
>;
export type TaskRunProcessOptions = {
workerManifest: WorkerManifest;
serverWorker: ServerBackgroundWorker;
env: Record<string, string>;
payload: TaskRunExecutionPayload;
cwd?: string;
messageId?: string;
};
export class TaskRunProcess {
private _ipc?: WorkerToExecutorProcessConnection;
private _child: ChildProcess | undefined;
private _childPid?: number;
private _attemptPromises: Map<
string,
{ resolver: (value: TaskRunExecutionResult) => void; rejecter: (err?: any) => void }
> = new Map();
private _attemptStatuses: Map<string, "PENDING" | "REJECTED" | "RESOLVED"> = new Map();
private _currentExecution: TaskRunExecution | undefined;
private _gracefulExitTimeoutElapsed: boolean = false;
private _isBeingKilled: boolean = false;
private _isBeingCancelled: boolean = false;
private _stderr: Array<string> = [];
/**
* @deprecated use onTaskRunHeartbeat instead
*/
public onTaskHeartbeat: Evt<string> = new Evt();
public onTaskRunHeartbeat: Evt<string> = new Evt();
public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
new Evt();
public onIsBeingKilled: Evt<TaskRunProcess> = new Evt();
public onReadyToDispose: Evt<TaskRunProcess> = new Evt();
public onWaitForDuration: Evt<OnWaitForDurationMessage> = new Evt();
public onWaitForTask: Evt<OnWaitForTaskMessage> = new Evt();
public onWaitForBatch: Evt<OnWaitForBatchMessage> = new Evt();
constructor(public readonly options: TaskRunProcessOptions) {}
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, workerManifest, cwd, messageId } = this.options;
const fullEnv = {
...(this.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}),
...env,
NODE_OPTIONS: workerManifest.loaderEntryPoint
? `--import=${workerManifest.loaderEntryPoint} ${
env.NODE_OPTIONS ?? process.env.NODE_OPTIONS ?? ""
}`
: env.NODE_OPTIONS ?? process.env.NODE_OPTIONS ?? "",
};
logger.debug(`[${this.runId}] initializing task run process`, {
env: fullEnv,
path: workerManifest.workerEntryPoint,
cwd,
});
this._child = fork(workerManifest.workerEntryPoint, {
stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
cwd,
env: fullEnv,
execArgv: ["--trace-uncaught", "--no-warnings=ExperimentalWarning"],
execPath: execPathForRuntime(workerManifest.runtime),
});
this._childPid = this._child?.pid;
this._ipc = new ZodIpcConnection({
listenSchema: WorkerToExecutorMessageCatalog,
emitSchema: ExecutorToWorkerMessageCatalog,
process: this._child,
handlers: {
TASK_RUN_COMPLETED: async (message) => {
const { result, execution } = message;
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);
},
READY_TO_DISPOSE: async (message) => {
logger.debug(`[${this.runId}] task run process is ready to dispose`);
this.onReadyToDispose.post(this);
},
TASK_HEARTBEAT: async (message) => {
if (messageId) {
this.onTaskRunHeartbeat.post(messageId);
} else {
logger.debug(
"No message id for task heartbeat, falling back to (deprecated) attempt heartbeat",
{ id: message.id }
);
this.onTaskHeartbeat.post(message.id);
}
},
WAIT_FOR_TASK: async (message) => {
this.onWaitForTask.post(message);
},
WAIT_FOR_BATCH: async (message) => {
this.onWaitForBatch.post(message);
},
WAIT_FOR_DURATION: async (message) => {
this.onWaitForDuration.post(message);
},
},
});
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 = false, gracefulExitTimeoutElapsed = false) {
logger.debug("cleanup()", { kill, gracefulExitTimeoutElapsed });
if (kill && this._isBeingKilled) {
return;
}
if (kill) {
this._isBeingKilled = true;
this.onIsBeingKilled.post(this);
}
logger.debug("Cleaning up task run process", {
kill,
childPid: this._childPid,
realChildPid: this._child?.pid,
});
try {
await this._ipc?.sendWithAck(
"CLEANUP",
{
flush: true,
kill,
},
30_000
);
} catch (error) {
logger.debug("Error while cleaning up task run process", error);
if (kill) {
this.onReadyToDispose.post(this);
}
}
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<TaskRunExecutionResult> {
let resolver: (value: TaskRunExecutionResult) => void;
let rejecter: (err?: any) => void;
const promise = new Promise<TaskRunExecutionResult>((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(this.payload.execution.attempt.id, { resolver, rejecter });
const { execution, traceContext } = this.payload;
this._currentExecution = execution;
if (this._child?.connected && !this._isBeingKilled && !this._child.killed) {
await this._ipc?.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") {
logger.debug(
"Task run completed with error and wants to retry, won't send task run completed notification"
);
return;
}
if (!this._child?.connected || this._isBeingKilled || this._child.killed) {
logger.debug(
"Child process not connected or being killed, can't send task run completed notification"
);
return;
}
this._ipc?.send("TASK_RUN_COMPLETED_NOTIFICATION", {
version: "v2",
completion,
});
}
async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
logger.debug("handling child exit", { code, signal });
// Go through all the attempts currently pending and reject them
for (const [id, status] of this._attemptStatuses.entries()) {
if (status === "PENDING") {
logger.debug("found pending attempt", { id });
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._gracefulExitTimeoutElapsed) {
// Order matters, this has to be before the graceful exit timeout
rejecter(new GracefulExitTimeoutError());
} 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);
this._child?.kill(signal);
if (timeoutInMs) {
await killTimeout;
}
}
get isBeingKilled() {
return this._isBeingKilled || this._child?.killed;
}
get pid() {
return this._childPid;
}
}
packages/core/src/v3/runtime/prodRuntimeManager.ts
+2 -7
View File
@@ -1,12 +1,10 @@
import { clock } from "../clock-api.js";
import {
BatchTaskRunExecutionResult,
ProdChildToWorkerMessages,
ProdWorkerToChildMessages,
TaskRunContext,
TaskRunExecutionResult,
} from "../schemas/index.js";
import { ZodIpcConnection } from "../zodIpc.js";
import { ExecutorToWorkerProcessConnection } from "../zodIpc.js";
import { RuntimeManager } from "./manager.js";
export type ProdRuntimeManagerOptions = {
@@ -24,10 +22,7 @@ export class ProdRuntimeManager implements RuntimeManager {
_waitForDuration: { resolve: (value: void) => void; reject: (err?: any) => void } | undefined;
constructor(
private ipc: ZodIpcConnection<
typeof ProdWorkerToChildMessages,
typeof ProdChildToWorkerMessages
>,
private ipc: ExecutorToWorkerProcessConnection,
private options: ProdRuntimeManagerOptions = {}
) {}
packages/core/src/v3/schemas/build.ts
+3
View File
@@ -66,6 +66,9 @@ export type IndexMessage = z.infer<typeof IndexMessage>;
export const WorkerManifest = z.object({
configPath: z.string(),
tasks: TaskManifest.array(),
workerEntryPoint: z.string(),
loaderEntryPoint: z.string().optional(),
runtime: BuildRuntime,
});
export type WorkerManifest = z.infer<typeof WorkerManifest>;
packages/core/src/v3/schemas/messages.ts
+4 -71
View File
@@ -1,21 +1,20 @@
import { z } from "zod";
import { WorkerManifest } from "./build.js";
import {
MachinePreset,
TaskRunExecution,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
} from "./common.js";
import { TaskResource } from "./resources.js";
import {
EnvironmentType,
ProdTaskRunExecution,
ProdTaskRunExecutionPayload,
TaskManifest,
TaskRunExecutionLazyAttemptPayload,
TaskRunExecutionPayload,
WaitReason,
} from "./schemas.js";
import { TaskResource } from "./resources.js";
import { BuildManifest, WorkerManifest } from "./build.js";
export const BackgroundWorkerServerMessages = z.discriminatedUnion("type", [
z.object({
@@ -108,32 +107,6 @@ export const clientWebsocketMessages = {
}),
};
export const workerToChildMessages = {
INDEX: z.object({}),
EXECUTE_TASK_RUN: z.object({
version: z.literal("v1").default("v1"),
execution: TaskRunExecution,
traceContext: z.record(z.unknown()),
metadata: ServerBackgroundWorker,
}),
TASK_RUN_COMPLETED_NOTIFICATION: z.discriminatedUnion("version", [
z.object({
version: z.literal("v1"),
completion: TaskRunExecutionResult,
execution: TaskRunExecution,
}),
z.object({
version: z.literal("v2"),
completion: TaskRunExecutionResult,
}),
]),
CLEANUP: z.object({
version: z.literal("v1").default("v1"),
flush: z.boolean().default(false),
kill: z.boolean().default(true),
}),
};
export const UncaughtExceptionMessage = z.object({
version: z.literal("v1").default("v1"),
error: z.object({
@@ -152,37 +125,6 @@ export const TaskMetadataFailedToParseData = z.object({
}),
});
export const childToWorkerMessages = {
TASK_RUN_COMPLETED: z.object({
version: z.literal("v1").default("v1"),
execution: TaskRunExecution,
result: TaskRunExecutionResult,
}),
TASK_HEARTBEAT: z.object({
version: z.literal("v1").default("v1"),
id: z.string(),
}),
TASK_RUN_HEARTBEAT: z.object({
version: z.literal("v1").default("v1"),
id: z.string(),
}),
READY_TO_DISPOSE: z.undefined(),
WAIT_FOR_DURATION: z.object({
version: z.literal("v1").default("v1"),
ms: z.number(),
}),
WAIT_FOR_TASK: z.object({
version: z.literal("v1").default("v1"),
id: z.string(),
}),
WAIT_FOR_BATCH: z.object({
version: z.literal("v1").default("v1"),
id: z.string(),
runs: z.string().array(),
}),
UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
};
export const indexerToWorkerMessages = {
INDEX_COMPLETE: z.object({
version: z.literal("v1").default("v1"),
@@ -192,7 +134,7 @@ export const indexerToWorkerMessages = {
UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
};
export const ProdChildToWorkerMessages = {
export const WorkerToExecutorMessageCatalog = {
TASK_RUN_COMPLETED: {
message: z.object({
version: z.literal("v1").default("v1"),
@@ -200,15 +142,6 @@ export const ProdChildToWorkerMessages = {
result: TaskRunExecutionResult,
}),
},
TASKS_READY: {
message: z.object({
version: z.literal("v1").default("v1"),
tasks: TaskManifest.array(),
}),
},
TASKS_FAILED_TO_PARSE: {
message: TaskMetadataFailedToParseData,
},
TASK_HEARTBEAT: {
message: z.object({
version: z.literal("v1").default("v1"),
@@ -244,7 +177,7 @@ export const ProdChildToWorkerMessages = {
},
};
export const ProdWorkerToChildMessages = {
export const ExecutorToWorkerMessageCatalog = {
EXECUTE_TASK_RUN: {
message: z.object({
version: z.literal("v1").default("v1"),
packages/core/src/v3/zodIpc.ts
+14
View File
@@ -11,6 +11,10 @@ import {
import { z } from "zod";
import { ZodSchemaParsedError } from "./zodMessageHandler.js";
import { inspect } from "node:util";
import {
ExecutorToWorkerMessageCatalog,
WorkerToExecutorMessageCatalog,
} from "./schemas/messages.js";
interface ZodIpcMessageSender<TEmitCatalog extends ZodSocketMessageCatalogSchema> {
send<K extends GetSocketMessagesWithoutCallback<TEmitCatalog>>(
@@ -335,3 +339,13 @@ export class ZodIpcConnection<
});
}
}
export type WorkerToExecutorProcessConnection = ZodIpcConnection<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog
>;
export type ExecutorToWorkerProcessConnection = ZodIpcConnection<
typeof ExecutorToWorkerMessageCatalog,
typeof WorkerToExecutorMessageCatalog
>;
pnpm-lock.yaml
Generated
+4 -4
View File
@@ -985,8 +985,8 @@ importers:
specifier: ^1.2.0
version: 1.3.1
tinyexec:
specifier: ^0.1.4
version: 0.1.4
specifier: ^0.2.0
version: 0.2.0
ws:
specifier: ^8.18.0
version: 8.18.0
@@ -26153,8 +26153,8 @@ packages:
resolution: {integrity: sha512-XPaBkWQJdsf3pLKJV9p4qN/S+fm2Oj8AIPo1BTUhg5oxkvm9+SVEGFdhyOz7tTdUTfvxMiAs4sp6/eZO2Ew+pw==}
dev: false
/tinyexec@0.1.4:
resolution: {integrity: sha512-Ba2ELcNnnWkgqnAJBouhcsDsYitbD9LIAVNSz3746u50f+tlF3wO0uB3uqyz8NHFSTpv23qtT47XGDw8pXW5DA==}
/tinyexec@0.2.0:
resolution: {integrity: sha512-au8dwv4xKSDR+Fw52csDo3wcDztPdne2oM1o/7LFro4h6bdFmvyUAeAfX40pwDtzHgRFqz1XWaUqgKS2G83/ig==}
dev: false
/tinyglobby@0.2.2:
references/bun-catalog/package.json
+2 -1
View File
@@ -4,7 +4,8 @@
"private": true,
"type": "module",
"scripts": {
"dev:trigger": "triggerdev dev"
"dev:trigger": "triggerdev dev",
"deploy": "triggerdev deploy"
},
"dependencies": {
"@trigger.dev/sdk": "workspace:*"