diff --git a/apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx b/apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx index 54e679d5f..91bd08880 100644 --- a/apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx +++ b/apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx @@ -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(); - const isLoading = fetcher.state !== "idle"; + const isLoading = fetcher.state === "loading"; useEffect(() => { fetcher.load(`/resources/taskruns/${runFriendlyId}/replay`); diff --git a/packages/cli-v3/src/entryPoints/deploy-executor.ts b/packages/cli-v3/src/entryPoints/deploy-executor.ts index 131ba5576..551111280 100644 --- a/packages/cli-v3/src/entryPoints/deploy-executor.ts +++ b/packages/cli-v3/src/entryPoints/deploy-executor.ts @@ -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(); diff --git a/packages/cli-v3/src/entryPoints/deploy.ts b/packages/cli-v3/src/entryPoints/deploy.ts index 5745d9542..ccd28ae3c 100644 --- a/packages/cli-v3/src/entryPoints/deploy.ts +++ b/packages/cli-v3/src/entryPoints/deploy.ts @@ -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) => { diff --git a/packages/cli-v3/src/entryPoints/dev-executor.ts b/packages/cli-v3/src/entryPoints/dev-executor.ts index 2c2580c8c..18febd3d4 100644 --- a/packages/cli-v3/src/entryPoints/dev-executor.ts +++ b/packages/cli-v3/src/entryPoints/dev-executor.ts @@ -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(); }, }, }); diff --git a/packages/cli-v3/src/executions/taskRunProcess.ts b/packages/cli-v3/src/executions/taskRunProcess.ts index 2741821ba..039739715 100644 --- a/packages/cli-v3/src/executions/taskRunProcess.ts +++ b/packages/cli-v3/src/executions/taskRunProcess.ts @@ -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 = []; + 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 { @@ -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; + + constructor(private readonly doFlush: () => Promise) { + this._flushPromise = this.doFlush(); + } + + waitForCompletion() { + return this._flushPromise; + } +} diff --git a/packages/cli-v3/src/package.json b/packages/cli-v3/src/package.json deleted file mode 100644 index 3dbc1ca59..000000000 --- a/packages/cli-v3/src/package.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "type": "module" -} diff --git a/packages/cli-v3/src/utilities/tempDirectories.ts b/packages/cli-v3/src/utilities/tempDirectories.ts index ec341bf99..3dc33fff3 100644 --- a/packages/cli-v3/src/utilities/tempDirectories.ts +++ b/packages/cli-v3/src/utilities/tempDirectories.ts @@ -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. diff --git a/packages/core/src/v3/schemas/messages.ts b/packages/core/src/v3/schemas/messages.ts index bba2e99a0..900fa2905 100644 --- a/packages/core/src/v3/schemas/messages.ts +++ b/packages/core/src/v3/schemas/messages.ts @@ -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 = { diff --git a/packages/core/src/v3/zodIpc.ts b/packages/core/src/v3/zodIpc.ts index 40dd32b2f..5965d8cb5 100644 --- a/packages/core/src/v3/zodIpc.ts +++ b/packages/core/src/v3/zodIpc.ts @@ -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 +>; diff --git a/references/v3-catalog/src/trigger/other.ts b/references/v3-catalog/src/trigger/other.ts index 3f24baca2..c68af1d39 100644 --- a/references/v3-catalog/src/trigger/other.ts +++ b/references/v3-catalog/src/trigger/other.ts @@ -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 }) => { diff --git a/references/v3-catalog/src/trigger/retries.ts b/references/v3-catalog/src/trigger/retries.ts index fe3756189..4741c65f8 100644 --- a/references/v3-catalog/src/trigger/retries.ts +++ b/references/v3-catalog/src/trigger/retries.ts @@ -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"; diff --git a/references/v3-catalog/trigger.config.ts b/references/v3-catalog/trigger.config.ts index c34ad2388..007b5856a 100644 --- a/references/v3-catalog/trigger.config.ts +++ b/references/v3-catalog/trigger.config.ts @@ -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, },