diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts index 4c2804cf8..283129951 100644 --- a/packages/cli-v3/src/dev/backgroundWorker.ts +++ b/packages/cli-v3/src/dev/backgroundWorker.ts @@ -143,7 +143,7 @@ export class BackgroundWorkerCoordinator { this._backgroundWorkers.clear(); } - async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId?: string) { + async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId: string) { const worker = this._backgroundWorkers.get(id); if (!worker) { @@ -152,7 +152,7 @@ export class BackgroundWorkerCoordinator { } try { - const completion = await worker.executeTaskRun(payload); + const completion = await worker.executeTaskRun(payload, messageId); this.onTaskCompleted.post({ completion, @@ -351,7 +351,7 @@ export class BackgroundWorker { async #getFreshTaskRunProcess( payload: TaskRunExecutionPayload, - messageId?: string + messageId: string ): Promise { logger.debug(this.#prefixedMessage(payload, "getFreshTaskRunProcess()")); @@ -494,7 +494,7 @@ export class BackgroundWorker { // We need to fork the process before we can execute any tasks async executeTaskRun( payload: TaskRunExecutionPayload, - messageId?: string + messageId: string ): Promise { if (this._closed) { throw new Error("Worker is closed"); @@ -523,7 +523,7 @@ export class BackgroundWorker { async #doExecuteTaskRun( payload: TaskRunExecutionPayload, - messageId?: string + messageId: string ): Promise { try { const taskRunProcess = await this.#getFreshTaskRunProcess(payload, messageId); diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts index b593a6184..338220e85 100644 --- a/packages/cli-v3/src/dev/workerRuntime.ts +++ b/packages/cli-v3/src/dev/workerRuntime.ts @@ -213,46 +213,6 @@ class DevWorkerRuntime implements WorkerRuntime { eventBus.emit("backgroundWorkerInitialized", backgroundWorker); } - async #fetchTaskFiles( - sources: Record, - tasks: TaskManifest[] - ) { - const tasksGroupedByFile: Record = {}; - - for (const task of tasks) { - if (!tasksGroupedByFile[task.filePath]) { - tasksGroupedByFile[task.filePath] = []; - } - - tasksGroupedByFile[task.filePath]!.push(task); - } - - const taskFiles: Array<{ - taskIds: string[]; - contents: string; - contentHash: string; - filePath: string; - }> = []; - - for (const [filePath, tasks] of Object.entries(tasksGroupedByFile)) { - const source = sources[filePath]; - - if (!source) { - continue; - } - - const taskIds = tasks.map((task) => task.id); - - taskFiles.push({ - ...source, - taskIds, - filePath, - }); - } - - return taskFiles; - } - async #getEnvVars(): Promise> { const environmentVariablesResponse = await this.options.client.getEnvironmentVariables( this.options.config.project diff --git a/packages/cli-v3/src/entryPoints/deploy-executor.ts b/packages/cli-v3/src/entryPoints/deploy-executor.ts index 551111280..988f7cc87 100644 --- a/packages/cli-v3/src/entryPoints/deploy-executor.ts +++ b/packages/cli-v3/src/entryPoints/deploy-executor.ts @@ -360,7 +360,7 @@ runtime.setGlobalRuntimeManager(prodRuntimeManager); process.title = "trigger-dev-worker"; -for await (const _ of setInterval(15)) { +for await (const _ of setInterval(15_000)) { if (_isRunning && _execution) { try { await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id }); diff --git a/packages/cli-v3/src/entryPoints/deploy.ts b/packages/cli-v3/src/entryPoints/deploy.ts index ccd28ae3c..6fb585a75 100644 --- a/packages/cli-v3/src/entryPoints/deploy.ts +++ b/packages/cli-v3/src/entryPoints/deploy.ts @@ -734,6 +734,7 @@ class ProdWorker { env, serverWorker: execution.worker, payload: createAttempt.result.executionPayload, + messageId: message.lazyPayload.messageId, }); this._taskRunProcess.onTaskRunHeartbeat.attach((heartbeatId) => { diff --git a/packages/cli-v3/src/executions/taskRunProcess.ts b/packages/cli-v3/src/executions/taskRunProcess.ts index 039739715..c62d56bb8 100644 --- a/packages/cli-v3/src/executions/taskRunProcess.ts +++ b/packages/cli-v3/src/executions/taskRunProcess.ts @@ -44,9 +44,9 @@ export type TaskRunProcessOptions = { serverWorker: ServerBackgroundWorker; env: Record; payload: TaskRunExecutionPayload; + messageId: string; cwd?: string; - messageId?: string; }; export class TaskRunProcess { @@ -64,10 +64,7 @@ export class TaskRunProcess { private _isBeingCancelled: boolean = false; private _stderr: Array = []; private _flushingProcess?: FlushingProcess; - /** - * @deprecated use onTaskRunHeartbeat instead - */ - public onTaskHeartbeat: Evt = new Evt(); + public onTaskRunHeartbeat: Evt = new Evt(); public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> = new Evt(); @@ -171,15 +168,7 @@ export class TaskRunProcess { 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); - } + this.onTaskRunHeartbeat.post(messageId); }, WAIT_FOR_TASK: async (message) => { this.onWaitForTask.post(message); diff --git a/references/v3-catalog/src/trigger/longRunning.ts b/references/v3-catalog/src/trigger/longRunning.ts index f3181309f..b6b3fc7f5 100644 --- a/references/v3-catalog/src/trigger/longRunning.ts +++ b/references/v3-catalog/src/trigger/longRunning.ts @@ -5,11 +5,11 @@ export const longRunning = task({ run: async (payload: { message: string }, { ctx }) => { logger.info("Long running", { payload }); - await new Promise((resolve) => setTimeout(resolve, 20000)); + await new Promise((resolve) => setTimeout(resolve, 200000)); // 200 seconds await wait.for({ seconds: 10 }); - await new Promise((resolve) => setTimeout(resolve, 20000)); + await new Promise((resolve) => setTimeout(resolve, 200000)); // 200 seconds }, });