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

support for waits and better flushing & process cleanup

Browse Source
This commit is contained in:
Eric Allam
2024-08-20 16:23:27 +01:00
committed by Eric Allam
parent 36b6f66467
commit 5d6488e197
12 changed files with 487 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx
+3 -3
View File
@@ -1,10 +1,10 @@
import { Form, useFetcher, useNavigation, useSubmit } from "@remix-run/react";
import { Form, useNavigation, useSubmit } from "@remix-run/react";
import { useCallback, useEffect, useRef } from "react";
import { UseDataFunctionReturn, useTypedFetcher } from "remix-typedjson";
import { JSONEditor } from "~/components/code/JSONEditor";
import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
import { Button } from "~/components/primitives/Buttons";
import { DialogContent, DialogDescription, DialogHeader } from "~/components/primitives/Dialog";
import { DialogContent, DialogHeader } from "~/components/primitives/Dialog";
import { Header3 } from "~/components/primitives/Headers";
import { InputGroup } from "~/components/primitives/InputGroup";
import { Label } from "~/components/primitives/Label";
@@ -27,7 +27,7 @@ export function ReplayRunDialog({ runFriendlyId, failedRedirect }: ReplayRunDial
function ReplayContent({ runFriendlyId, failedRedirect }: ReplayRunDialogProps) {
const fetcher = useTypedFetcher<typeof loader>();
const isLoading = fetcher.state !== "idle";
const isLoading = fetcher.state === "loading";
useEffect(() => {
fetcher.load(`/resources/taskruns/${runFriendlyId}/replay`);
packages/cli-v3/src/entryPoints/deploy-executor.ts
+59 -44
View File
@@ -34,6 +34,7 @@ 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";
import { setTimeout, setInterval } from "node:timers/promises";
sourceMapSupport.install({
handleUncaughtExceptions: false,
@@ -45,25 +46,31 @@ process.on("uncaughtException", function (error, origin) {
if (error instanceof Error) {
process.send &&
process.send({
type: "UNCAUGHT_EXCEPTION",
payload: {
error: { name: error.name, message: error.message, stack: error.stack },
origin,
type: "EVENT",
message: {
type: "UNCAUGHT_EXCEPTION",
payload: {
error: { name: error.name, message: error.message, stack: error.stack },
origin,
},
version: "v1",
},
version: "v1",
});
} else {
process.send &&
process.send({
type: "UNCAUGHT_EXCEPTION",
payload: {
error: {
name: "Error",
message: typeof error === "string" ? error : JSON.stringify(error),
type: "EVENT",
message: {
type: "UNCAUGHT_EXCEPTION",
payload: {
error: {
name: "Error",
message: typeof error === "string" ? error : JSON.stringify(error),
},
origin,
},
origin,
version: "v1",
},
version: "v1",
});
}
});
@@ -160,8 +167,8 @@ let _isRunning = false;
let _tracingSDK: TracingSDK | undefined;
const zodIpc = new ZodIpcConnection({
listenSchema: ExecutorToWorkerMessageCatalog,
emitSchema: WorkerToExecutorMessageCatalog,
listenSchema: WorkerToExecutorMessageCatalog,
emitSchema: ExecutorToWorkerMessageCatalog,
process,
handlers: {
EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }, sender) => {
@@ -309,20 +316,42 @@ const zodIpc = new ZodIpcConnection({
WAIT_COMPLETED_NOTIFICATION: async () => {
prodRuntimeManager.resumeAfterDuration();
},
CLEANUP: async ({ flush, kill }, sender) => {
if (kill) {
await _tracingSDK?.flush();
// Now we need to exit the process
await sender.send("READY_TO_DISPOSE", undefined);
} else {
if (flush) {
await _tracingSDK?.flush();
}
}
FLUSH: async ({ timeoutInMs }, sender) => {
await flushAll(timeoutInMs);
},
},
});
async function flushAll(timeoutInMs: number = 10_000) {
const now = performance.now();
await Promise.all([flushUsage(timeoutInMs), flushTracingSDK(timeoutInMs)]);
const duration = performance.now() - now;
console.log(`Flushed all in ${duration}ms`);
}
async function flushUsage(timeoutInMs: number = 10_000) {
const now = performance.now();
await Promise.race([prodUsageManager.flush(), setTimeout(timeoutInMs)]);
const duration = performance.now() - now;
console.log(`Flushed usage in ${duration}ms`);
}
async function flushTracingSDK(timeoutInMs: number = 10_000) {
const now = performance.now();
await Promise.race([_tracingSDK?.flush(), setTimeout(timeoutInMs)]);
const duration = performance.now() - now;
console.log(`Flushed tracingSDK in ${duration}ms`);
}
const prodRuntimeManager = new ProdRuntimeManager(zodIpc, {
waitThresholdInMs: parseInt(process.env.TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS ?? "30000", 10),
});
@@ -331,28 +360,14 @@ runtime.setGlobalRuntimeManager(prodRuntimeManager);
process.title = "trigger-dev-worker";
async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeconds: number = 30) {
async function _doHeartbeat() {
while (true) {
if (_isRunning && _execution) {
try {
await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
} catch (err) {
console.error("Failed to send HEARTBEAT message", err);
}
}
await new Promise((resolve) => setTimeout(resolve, 1000 * intervalInSeconds));
for await (const _ of setInterval(15)) {
if (_isRunning && _execution) {
try {
await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
} catch (err) {
console.error("Failed to send HEARTBEAT message", err);
}
}
// Wait for the initial delay
await new Promise((resolve) => setTimeout(resolve, 1000 * initialDelayInSeconds));
// Wait for 5 seconds before the next execution
return _doHeartbeat();
}
console.log(`[${new Date().toISOString()}] Executor started`);
await asyncHeartbeat();
packages/cli-v3/src/entryPoints/deploy.ts
+308 -153
View File
@@ -26,9 +26,11 @@ import { setTimeout as timeout } from "node:timers/promises";
import { logger as cliLogger } from "../utilities/logger.js";
import {
OnWaitForBatchMessage,
OnWaitForDurationMessage,
OnWaitForTaskMessage,
TaskRunProcess,
} from "../executions/taskRunProcess.js";
import { checkpointSafeTimeout, unboundedTimeout } from "@trigger.dev/core/v3/utils/timers";
const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || getRandomPortNumber());
const COORDINATOR_HOST = process.env.COORDINATOR_HOST || "127.0.0.1";
@@ -102,6 +104,8 @@ class ProdWorker {
typeof CoordinatorToProdWorkerMessages
>;
private _taskRunProcess: TaskRunProcess | undefined;
constructor(
port: number,
private workerManifest: WorkerManifest,
@@ -145,8 +149,11 @@ class ProdWorker {
}
async #exitGracefully(gracefulExitTimeoutElapsed = false, exitCode = 0) {
// TODO: close the worker process
// await this.#backgroundWorker.close(gracefulExitTimeoutElapsed);
if (this._taskRunProcess) {
this._taskRunProcess.onTaskRunHeartbeat.detach();
this._taskRunProcess.onWaitForDuration.detach();
await this._taskRunProcess.cleanup(true);
}
if (!gracefulExitTimeoutElapsed) {
// TODO: Maybe add a sensible timeout instead of a conditional to avoid zombies
@@ -187,148 +194,144 @@ class ProdWorker {
}
// MARK: TASK WAIT
#waitForTaskHandlerFactory(workerId?: string) {
return async (message: OnWaitForTaskMessage, replayIdempotencyKey?: string) => {
logger.log("onWaitForTask", { workerId, message });
async #handleOnWaitForTask(message: OnWaitForTaskMessage, replayIdempotencyKey?: string) {
logger.log("onWaitForTask", { message });
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for task", {
nextResumeAfter: this.nextResumeAfter,
});
return;
}
const waitForTask = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for task with backoff", { retry });
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_TASK", {
version: "v2",
friendlyId: message.friendlyId,
attemptFriendlyId: this.attemptFriendlyId,
});
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for task", {
nextResumeAfter: this.nextResumeAfter,
});
if (!waitForTask.success) {
logger.error("Failed to wait for task with backoff", {
cause: waitForTask.cause,
error: waitForTask.error,
});
return;
}
this.#emitUnrecoverableError(
"WaitForTaskFailed",
`${waitForTask.cause}: ${waitForTask.error}`
);
const waitForTask = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for task with backoff", { retry });
return;
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
const { willCheckpointAndRestore } = waitForTask.result;
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_TASK", {
version: "v2",
friendlyId: message.friendlyId,
attemptFriendlyId: this.attemptFriendlyId,
});
});
await this.#prepareForWait("WAIT_FOR_TASK", willCheckpointAndRestore);
if (!waitForTask.success) {
logger.error("Failed to wait for task with backoff", {
cause: waitForTask.cause,
error: waitForTask.error,
});
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForTaskReplay) {
this.waitForTaskReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForTaskReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.#emitUnrecoverableError(
"WaitForTaskFailed",
`${waitForTask.cause}: ${waitForTask.error}`
);
this.waitForTaskReplay.attempt++;
return;
}
const { willCheckpointAndRestore } = waitForTask.result;
await this.#prepareForWait("WAIT_FOR_TASK", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForTaskReplay) {
this.waitForTaskReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForTaskReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.waitForTaskReplay.attempt++;
}
};
}
}
// MARK: BATCH WAIT
#waitForBatchHandlerFactory(workerId?: string) {
return async (message: OnWaitForBatchMessage, replayIdempotencyKey?: string) => {
logger.log("onWaitForBatch", { workerId, message });
async #handleOnWaitForBatch(message: OnWaitForBatchMessage, replayIdempotencyKey?: string) {
logger.log("onWaitForBatch", { message });
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for batch", {
nextResumeAfter: this.nextResumeAfter,
});
return;
}
const waitForBatch = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for batch with backoff", { retry });
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_BATCH", {
version: "v2",
batchFriendlyId: message.batchFriendlyId,
runFriendlyIds: message.runFriendlyIds,
attemptFriendlyId: this.attemptFriendlyId,
});
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for batch", {
nextResumeAfter: this.nextResumeAfter,
});
if (!waitForBatch.success) {
logger.error("Failed to wait for batch with backoff", {
cause: waitForBatch.cause,
error: waitForBatch.error,
});
return;
}
this.#emitUnrecoverableError(
"WaitForBatchFailed",
`${waitForBatch.cause}: ${waitForBatch.error}`
);
const waitForBatch = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for batch with backoff", { retry });
return;
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
const { willCheckpointAndRestore } = waitForBatch.result;
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_BATCH", {
version: "v2",
batchFriendlyId: message.batchFriendlyId,
runFriendlyIds: message.runFriendlyIds,
attemptFriendlyId: this.attemptFriendlyId,
});
});
await this.#prepareForWait("WAIT_FOR_BATCH", willCheckpointAndRestore);
if (!waitForBatch.success) {
logger.error("Failed to wait for batch with backoff", {
cause: waitForBatch.cause,
error: waitForBatch.error,
});
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForBatchReplay) {
this.waitForBatchReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForBatchReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.#emitUnrecoverableError(
"WaitForBatchFailed",
`${waitForBatch.cause}: ${waitForBatch.error}`
);
this.waitForBatchReplay.attempt++;
return;
}
const { willCheckpointAndRestore } = waitForBatch.result;
await this.#prepareForWait("WAIT_FOR_BATCH", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForBatchReplay) {
this.waitForBatchReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForBatchReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.waitForBatchReplay.attempt++;
}
};
}
}
async #prepareForWait(reason: WaitReason, willCheckpointAndRestore: boolean) {
@@ -380,8 +383,7 @@ class ProdWorker {
if (flush) {
// Flush before checkpointing so we don't flush the same spans again after restore
try {
// TODO: flush the telemetry
// await this.#backgroundWorker.flushTelemetry();
await this._taskRunProcess?.cleanup(false);
} catch (error) {
logger.error(
"Failed to flush telemetry while preparing for checkpoint, will proceed anyway",
@@ -411,8 +413,7 @@ class ProdWorker {
this.durationResumeFallback = undefined;
// TODO: signal to the worker that is can resume after duration
// this.#backgroundWorker.waitCompletedNotification();
this._taskRunProcess?.waitCompletedNotification();
}
async #readyForLazyAttempt() {
@@ -501,6 +502,7 @@ class ProdWorker {
const taskRunCompleted = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Submit attempt completion with backoff", { retry });
// TODO: update this to use version: v2
return await this.#coordinatorSocket.socket
.timeout(20_000)
.emitWithAck("TASK_RUN_COMPLETED", {
@@ -636,8 +638,7 @@ class ProdWorker {
if (!completion) continue;
// TODO: signal to the worker that a task run has completed that it was waiting for
// this.#backgroundWorker.taskRunCompletedNotification(completion);
this._taskRunProcess?.taskRunCompletedNotification(completion);
}
},
RESUME_AFTER_DURATION: async (message) => {
@@ -715,6 +716,11 @@ class ProdWorker {
return;
}
await this.#killCurrentTaskRunProcessBeforeAttempt();
this.attemptFriendlyId = createAttempt.result.executionPayload.execution.attempt.id;
this.attemptNumber = createAttempt.result.executionPayload.execution.attempt.number;
const { execution } = createAttempt.result.executionPayload;
const { environment } = message.lazyPayload;
@@ -723,51 +729,58 @@ class ProdWorker {
...environment,
};
const taskRunProcess = new TaskRunProcess({
this._taskRunProcess = new TaskRunProcess({
workerManifest: this.workerManifest,
env,
serverWorker: execution.worker,
payload: createAttempt.result.executionPayload,
});
this._taskRunProcess.onTaskRunHeartbeat.attach((heartbeatId) => {
logger.log("onTaskRunHeartbeat", {
heartbeatId,
});
this.#coordinatorSocket.socket.volatile.emit("TASK_RUN_HEARTBEAT", {
version: "v1",
runId: heartbeatId,
});
});
this._taskRunProcess.onWaitForDuration.attach(this.#handleOnWaitForDuration.bind(this));
this._taskRunProcess.onWaitForTask.attach(this.#handleOnWaitForTask.bind(this));
this._taskRunProcess.onWaitForBatch.attach(this.#handleOnWaitForBatch.bind(this));
logger.log("initializing task run process", {
workerManifest: this.workerManifest,
attemptId: execution.attempt.id,
runId: execution.run.id,
});
await taskRunProcess.initialize();
try {
await this._taskRunProcess.initialize();
logger.log("executing task run process", {
attemptId: execution.attempt.id,
runId: execution.run.id,
});
logger.log("executing task run process", {
attemptId: execution.attempt.id,
runId: execution.run.id,
});
const completion = await taskRunProcess.execute();
const completion = await this._taskRunProcess.execute();
logger.log("completed", completion);
logger.log("completed", completion);
this.completed.add(execution.attempt.id);
this.completed.add(execution.attempt.id);
await this.#submitAttemptCompletion(execution, completion);
await this._taskRunProcess.startFlushingProcess();
// TODO: execute the task run lazy attempt
// try {
// const { completion, execution } =
// await this.#backgroundWorker.executeTaskRunLazyAttempt(message.lazyPayload);
await this.#submitAttemptCompletion(execution, completion);
} catch (error) {
logger.error("Failed to complete lazy attempt", {
error,
});
// logger.log("completed", completion);
// this.completed.add(execution.attempt.id);
// await this.#submitAttemptCompletion(execution, completion);
// } catch (error) {
// logger.error("Failed to complete lazy attempt", {
// error,
// });
// this.#failRun(message.lazyPayload.runId, error);
// }
this.#failRun(message.lazyPayload.runId, error);
}
},
REQUEST_ATTEMPT_CANCELLATION: async (message) => {
if (!this.executing) {
@@ -777,8 +790,7 @@ class ProdWorker {
logger.log("cancelling attempt", { attemptId: message.attemptId, status: this.#status });
// TODO: cancel the attempt
// await this.#backgroundWorker.cancelAttempt(message.attemptId);
await this._taskRunProcess?.cancel();
},
REQUEST_EXIT: async (message) => {
if (message.version === "v2" && message.delayInMs) {
@@ -885,6 +897,135 @@ class ProdWorker {
return coordinatorConnection;
}
// MARK: Handle onWaitForDuration
async #handleOnWaitForDuration(message: OnWaitForDurationMessage) {
logger.log("onWaitForDuration", {
...message,
drift: Date.now() - message.now,
});
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for duration", {
nextResumeAfter: this.nextResumeAfter,
});
return;
}
noResume: {
const { ms, waitThresholdInMs } = message;
const internalTimeout = unboundedTimeout(ms, "internal" as const);
const checkpointSafeInternalTimeout = checkpointSafeTimeout(ms);
if (ms < waitThresholdInMs) {
await internalTimeout;
break noResume;
}
const waitForDuration = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for duration with backoff", { retry });
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
return await this.#coordinatorSocket.socket
.timeout(20_000)
.emitWithAck("WAIT_FOR_DURATION", {
...message,
attemptFriendlyId: this.attemptFriendlyId,
});
});
if (!waitForDuration.success) {
logger.error("Failed to wait for duration with backoff", {
cause: waitForDuration.cause,
error: waitForDuration.error,
});
this.#emitUnrecoverableError(
"WaitForDurationFailed",
`${waitForDuration.cause}: ${waitForDuration.error}`
);
return;
}
const { willCheckpointAndRestore } = waitForDuration.result;
if (!willCheckpointAndRestore) {
await internalTimeout;
break noResume;
}
await this.#prepareForWait("WAIT_FOR_DURATION", willCheckpointAndRestore);
// CHECKPOINTING AFTER THIS LINE
// internalTimeout acts as a backup and will be accurate if the checkpoint never happens
// checkpointSafeInternalTimeout is accurate even after non-simulated restores
await Promise.race([internalTimeout, checkpointSafeInternalTimeout]);
const idempotencyKey = randomUUID();
this.durationResumeFallback = { idempotencyKey };
try {
await this.restoreNotification.waitFor(5_000);
} catch (error) {
logger.error("Did not receive restore notification in time", {
error,
});
}
try {
// The coordinator should cancel any in-progress checkpoints so we don't end up with race conditions
const { checkpointCanceled } = await this.#coordinatorSocket.socket
.timeout(15_000)
.emitWithAck("CANCEL_CHECKPOINT", {
version: "v2",
reason: "WAIT_FOR_DURATION",
});
logger.log("onCancelCheckpoint coordinator response", { checkpointCanceled });
if (checkpointCanceled) {
// If the checkpoint was canceled, we will never be resumed externally with RESUME_AFTER_DURATION, so it's safe to immediately resume
break noResume;
}
logger.log("Waiting for external duration resume as we may have been restored");
setTimeout(() => {
if (!this.durationResumeFallback) {
logger.error("Already resumed after duration, skipping fallback");
return;
}
if (this.durationResumeFallback.idempotencyKey !== idempotencyKey) {
logger.error("Duration resume idempotency key mismatch, skipping fallback");
return;
}
logger.log("Resuming after duration with fallback");
this.#resumeAfterDuration();
}, 15_000);
} catch (error) {
// Just log this for now, but don't automatically resume. Wait for the external checkpoint-based resume.
logger.debug("Checkpoint cancellation timed out", {
message,
error,
});
}
return;
}
this.#resumeAfterDuration();
}
// MARK: REPLAYS
async #handleReplays() {
const backoff = new ExponentialBackoff().type("FullJitter").maxRetries(3);
@@ -918,7 +1059,7 @@ class ProdWorker {
try {
await backoff.wait(attempt + 1);
await this.#waitForTaskHandlerFactory("replay")(message, idempotencyKey);
await this.#handleOnWaitForTask(message, idempotencyKey);
} catch (error) {
if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
logger.error("wait for task replay retry limit exceeded", { error });
@@ -961,7 +1102,7 @@ class ProdWorker {
try {
await backoff.wait(attempt + 1);
await this.#waitForBatchHandlerFactory("replay")(message, idempotencyKey);
await this.#handleOnWaitForBatch(message, idempotencyKey);
} catch (error) {
if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
logger.error("wait for batch replay retry limit exceeded", { error });
@@ -974,6 +1115,20 @@ class ProdWorker {
}
}
async #killCurrentTaskRunProcessBeforeAttempt() {
console.log("killCurrentTaskRunProcessBeforeAttempt()", {
hasTaskRunProcess: !!this._taskRunProcess,
});
if (!this._taskRunProcess) {
return;
}
const currentTaskRunProcess = this._taskRunProcess;
await currentTaskRunProcess.cleanup();
}
// MARK: HTTP SERVER
#createHttpServer() {
const httpServer = createServer(async (req, res) => {
packages/cli-v3/src/entryPoints/dev-executor.ts
+4 -12
View File
@@ -150,8 +150,8 @@ let _isRunning = false;
let _tracingSDK: TracingSDK | undefined;
const zodIpc = new ZodIpcConnection({
listenSchema: ExecutorToWorkerMessageCatalog,
emitSchema: WorkerToExecutorMessageCatalog,
listenSchema: WorkerToExecutorMessageCatalog,
emitSchema: ExecutorToWorkerMessageCatalog,
process,
handlers: {
EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }, sender) => {
@@ -297,16 +297,8 @@ const zodIpc = new ZodIpcConnection({
}
}
},
CLEANUP: async ({ flush, kill }, sender) => {
if (kill) {
await _tracingSDK?.flush();
// Now we need to exit the process
await sender.send("READY_TO_DISPOSE", undefined);
} else {
if (flush) {
await _tracingSDK?.flush();
}
}
FLUSH: async ({ timeoutInMs }, sender) => {
await _tracingSDK?.flush();
},
},
});
packages/cli-v3/src/executions/taskRunProcess.ts
+59 -60
View File
@@ -1,12 +1,11 @@
import {
ExecutorToWorkerMessageCatalog,
ServerBackgroundWorker,
TaskRunExecution,
TaskRunExecutionPayload,
TaskRunExecutionResult,
WorkerToExecutorMessageCatalog,
ExecutorToWorkerMessageCatalog,
WorkerManifest,
SemanticInternalAttributes,
WorkerToExecutorMessageCatalog,
} from "@trigger.dev/core/v3";
import {
type WorkerToExecutorProcessConnection,
@@ -17,25 +16,26 @@ 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 { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
import { logger } from "../utilities/logger.js";
import {
CancelledProcessError,
CleanupProcessError,
GracefulExitTimeoutError,
SigKillTimeoutProcessError,
UnexpectedExitError,
} from "./errors.js";
import { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
export type OnWaitForDurationMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog,
"WAIT_FOR_DURATION"
>;
export type OnWaitForTaskMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog,
"WAIT_FOR_TASK"
>;
export type OnWaitForBatchMessage = InferSocketMessageSchema<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog,
"WAIT_FOR_BATCH"
>;
@@ -63,6 +63,7 @@ export class TaskRunProcess {
private _isBeingKilled: boolean = false;
private _isBeingCancelled: boolean = false;
private _stderr: Array<string> = [];
private _flushingProcess?: FlushingProcess;
/**
* @deprecated use onTaskRunHeartbeat instead
*/
@@ -82,7 +83,16 @@ export class TaskRunProcess {
async cancel() {
this._isBeingCancelled = true;
await this.cleanup(true);
await this.startFlushingProcess();
await this.kill();
}
async cleanup(kill = true) {
await this.startFlushingProcess();
if (kill) {
await this.kill("SIGKILL");
}
}
get runId() {
@@ -130,8 +140,8 @@ export class TaskRunProcess {
this._childPid = this._child?.pid;
this._ipc = new ZodIpcConnection({
listenSchema: WorkerToExecutorMessageCatalog,
emitSchema: ExecutorToWorkerMessageCatalog,
listenSchema: ExecutorToWorkerMessageCatalog,
emitSchema: WorkerToExecutorMessageCatalog,
process: this._child,
handlers: {
TASK_RUN_COMPLETED: async (message) => {
@@ -188,53 +198,18 @@ export class TaskRunProcess {
this._child.stderr?.on("data", this.#handleStdErr.bind(this));
}
async cleanup(kill = false, gracefulExitTimeoutElapsed = false) {
logger.debug("cleanup()", { kill, gracefulExitTimeoutElapsed });
if (kill && this._isBeingKilled) {
async startFlushingProcess() {
if (this._flushingProcess) {
return;
}
if (kill) {
this._isBeingKilled = true;
this.onIsBeingKilled.post(this);
}
this._flushingProcess = new FlushingProcess(() => this.#flush());
}
logger.debug("Cleaning up task run process", {
kill,
childPid: this._childPid,
realChildPid: this._child?.pid,
});
async #flush(timeoutInMs: number = 5_000) {
logger.debug("flushing task run process", { pid: this.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);
await this._ipc?.sendWithAck("FLUSH", { timeoutInMs }, timeoutInMs + 1_000);
}
async execute(): Promise<TaskRunExecutionResult> {
@@ -300,6 +275,17 @@ export class TaskRunProcess {
});
}
waitCompletedNotification() {
if (!this._child?.connected || this._isBeingKilled || this._child.killed) {
console.error(
"Child process not connected or being killed, can't send wait completed notification"
);
return;
}
this._ipc?.send("WAIT_COMPLETED_NOTIFICATION", {});
}
async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
logger.debug("handling child exit", { code, signal });
@@ -337,6 +323,8 @@ export class TaskRunProcess {
}
}
logger.debug("Task run process exited, posting onExit", { code, signal, pid: this.pid });
this.onExit.post({ code, signal, pid: this.pid });
}
@@ -383,14 +371,6 @@ export class TaskRunProcess {
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,
@@ -403,6 +383,13 @@ export class TaskRunProcess {
const killTimeout = this.onExit.waitFor(timeoutInMs);
this.onIsBeingKilled.post(this);
try {
await this._flushingProcess?.waitForCompletion();
} catch (err) {
logger.error("Error flushing task run process", { err });
}
this._child?.kill(signal);
if (timeoutInMs) {
@@ -422,3 +409,15 @@ export class TaskRunProcess {
function executorArgs(workerManifest: WorkerManifest): string[] {
return [];
}
class FlushingProcess {
private _flushPromise: Promise<void>;
constructor(private readonly doFlush: () => Promise<void>) {
this._flushPromise = this.doFlush();
}
waitForCompletion() {
return this._flushPromise;
}
}
packages/cli-v3/src/package.json
-3
View File
@@ -1,3 +0,0 @@
{
"type": "module"
}
packages/cli-v3/src/utilities/tempDirectories.ts
+2 -2
View File
@@ -12,8 +12,8 @@ export interface EphemeralDirectory {
}
/**
* Gets a temporary directory in the project's `.wrangler` folder with the
* specified prefix. We create temporary directories in `.wrangler` as opposed
* Gets a temporary directory in the project's `.trigger` folder with the
* specified prefix. We create temporary directories in `.trigger` as opposed
* to the OS's temporary directory to avoid issues with different drive letters
* on Windows. For example, when `esbuild` outputs a file to a different drive
* than the input sources, the generated source maps are incorrect.
packages/core/src/v3/schemas/messages.ts
+8 -10
View File
@@ -134,7 +134,7 @@ export const indexerToWorkerMessages = {
UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
};
export const WorkerToExecutorMessageCatalog = {
export const ExecutorToWorkerMessageCatalog = {
TASK_RUN_COMPLETED: {
message: z.object({
version: z.literal("v1").default("v1"),
@@ -177,7 +177,7 @@ export const WorkerToExecutorMessageCatalog = {
},
};
export const ExecutorToWorkerMessageCatalog = {
export const WorkerToExecutorMessageCatalog = {
EXECUTE_TASK_RUN: {
message: z.object({
version: z.literal("v1").default("v1"),
@@ -199,19 +199,17 @@ export const ExecutorToWorkerMessageCatalog = {
}),
]),
},
CLEANUP: {
message: z.object({
version: z.literal("v1").default("v1"),
flush: z.boolean().default(false),
kill: z.boolean().default(true),
}),
callback: z.void(),
},
WAIT_COMPLETED_NOTIFICATION: {
message: z.object({
version: z.literal("v1").default("v1"),
}),
},
FLUSH: {
message: z.object({
timeoutInMs: z.number(),
}),
callback: z.void(),
},
};
export const ProviderToPlatformMessages = {
packages/core/src/v3/zodIpc.ts
+5 -5
View File
@@ -341,11 +341,11 @@ export class ZodIpcConnection<
}
export type WorkerToExecutorProcessConnection = ZodIpcConnection<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog
>;
export type ExecutorToWorkerProcessConnection = ZodIpcConnection<
typeof ExecutorToWorkerMessageCatalog,
typeof WorkerToExecutorMessageCatalog
>;
export type ExecutorToWorkerProcessConnection = ZodIpcConnection<
typeof WorkerToExecutorMessageCatalog,
typeof ExecutorToWorkerMessageCatalog
>;
references/v3-catalog/src/trigger/other.ts
+34
View File
@@ -46,6 +46,40 @@ export const waitAminute = task({
},
});
export const triggerAndWaitDep = task({
id: "trigger-and-wait-dep",
run: async (payload: { seconds?: number }) => {
logger.log("logs before");
await waitAminute.triggerAndWait({ seconds: payload.seconds });
logger.log("logs after");
},
});
export const testingErrors = task({
id: "testing-errors",
run: async ({ numberOfFailures = 10 }: { numberOfFailures?: number }, { ctx }) => {
logger.log("logs before");
if (ctx.attempt.number < numberOfFailures) {
throw new Error(`Attempt ${ctx.attempt.number} failed`);
}
logger.log("logs after");
},
});
export const batchTriggerAndWaitDep = task({
id: "batch-trigger-and-wait-dep",
run: async (payload: { seconds?: number }) => {
logger.log("logs before");
await waitAminute.batchTriggerAndWait([
{ payload: { seconds: payload.seconds } },
{ payload: { seconds: payload.seconds } },
]);
logger.log("logs after");
},
});
export const consecutiveDependencies = task({
id: "consecutive-dependencies",
run: async (payload: { seconds?: number }) => {
references/v3-catalog/src/trigger/retries.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { logger, retry, runs, task, wait } from "@trigger.dev/sdk/v3";
import { cache } from "./utils/cache";
import { interceptor } from "./utils/interceptor";
import { cache } from "./utils/cache.js";
import { interceptor } from "./utils/interceptor.js";
import { join } from "node:path";
import { mkdir, writeFile } from "node:fs/promises";
references/v3-catalog/trigger.config.ts
+3 -3
View File
@@ -39,9 +39,9 @@ export default defineConfig({
retries: {
enabledInDev: true,
default: {
maxAttempts: 4,
minTimeoutInMs: 10000,
maxTimeoutInMs: 10000,
maxAttempts: 10,
minTimeoutInMs: 5_000,
maxTimeoutInMs: 30_000,
factor: 2,
randomize: true,
},