diff --git a/packages/cli-v3/src/build/plugins.ts b/packages/cli-v3/src/build/plugins.ts index ee08f28b2..365901253 100644 --- a/packages/cli-v3/src/build/plugins.ts +++ b/packages/cli-v3/src/build/plugins.ts @@ -18,7 +18,7 @@ export async function buildPlugins( plugins.push($configPlugin); } - plugins.push(mockServerOnlyPlugin()); + plugins.push(polyshedPlugin()); return plugins; } @@ -42,32 +42,47 @@ export function analyzeMetadataPlugin(): esbuild.Plugin { }; } -export function mockServerOnlyPlugin(): esbuild.Plugin { - return { - name: "trigger-mock-server-only", - setup(build) { - build.onResolve({ filter: /^server-only$/ }, (args) => { - if (args.path !== "server-only") { - return undefined; - } +const polysheds = [ + { + moduleName: "is-core-module", + code: "const { isBuiltin } = require('node:module'); module.exports = isBuiltin;", + }, + { + moduleName: "server-only", + code: "export default true;", + }, +]; - logger.debug(`[trigger-mock-server-only] Bundling ${args.path}`, { - ...args, +export function polyshedPlugin(): esbuild.Plugin { + return { + name: "polyshed", + setup(build) { + for (const polyshed of polysheds) { + build.onResolve({ filter: new RegExp(`^${polyshed.moduleName}$`) }, (args) => { + if (args.path !== polyshed.moduleName) { + return undefined; + } + + return { + path: args.path, + external: false, + namespace: `polyshed-${polyshed.moduleName}`, + }; }); - return { - path: args.path, - external: false, - namespace: "server-only-mock", - }; - }); - - build.onLoad({ filter: /^server-only$/, namespace: "server-only-mock" }, (args) => { - return { - contents: `export default true;`, - loader: "js", - }; - }); + build.onLoad( + { + filter: new RegExp(`^${polyshed.moduleName}$`), + namespace: `polyshed-${polyshed.moduleName}`, + }, + (args) => { + return { + contents: polyshed.code, + loader: "js", + }; + } + ); + } }, }; } diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts index dec9f50f6..014f54f4d 100644 --- a/packages/cli-v3/src/dev/backgroundWorker.ts +++ b/packages/cli-v3/src/dev/backgroundWorker.ts @@ -3,7 +3,6 @@ import { CreateBackgroundWorkerResponse, ServerBackgroundWorker, TaskRunBuiltInError, - TaskRunError, TaskRunErrorCodes, TaskRunExecution, TaskRunExecutionPayload, @@ -12,30 +11,21 @@ import { WorkerManifest, childToWorkerMessages, correctErrorStackTrace, - formatDurationMilliseconds, indexerToWorkerMessages, workerToChildMessages, } from "@trigger.dev/core/v3"; import { - parseMessageFromCatalog, ZodMessageHandler, ZodMessageSender, + parseMessageFromCatalog, } from "@trigger.dev/core/v3/zodMessageHandler"; import { Evt } from "evt"; import { ChildProcess, fork } from "node:child_process"; -import { - chalkError, - chalkGrey, - chalkLink, - chalkRun, - chalkSuccess, - chalkTask, - chalkWarning, - chalkWorker, - cliLink, - prettyPrintDate, -} from "../utilities/cliOutput.js"; +import { chalkError, chalkGrey, chalkRun, prettyPrintDate } from "../utilities/cliOutput.js"; +import { join } from "node:path"; +import { eventBus } from "../utilities/eventBus.js"; +import { writeJSONFile } from "../utilities/fileSystem.js"; import { logger } from "../utilities/logger.js"; import { CancelledProcessError, @@ -46,8 +36,6 @@ import { UnexpectedExitError, getFriendlyErrorMessage, } from "./errors.js"; -import { writeJSONFile } from "../utilities/fileSystem.js"; -import { join } from "node:path"; export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"]; export class BackgroundWorkerCoordinator { @@ -84,7 +72,7 @@ export class BackgroundWorkerCoordinator { public onWorkerDeprecated: Evt<{ worker: BackgroundWorker; id: string }> = new Evt(); private _backgroundWorkers: Map = new Map(); - constructor(private baseURL: string) { + constructor() { this.onTaskCompleted.attach(async ({ completion }) => { if (!completion.ok && typeof completion.retry !== "undefined") { return; @@ -169,7 +157,7 @@ export class BackgroundWorkerCoordinator { } try { - const completion = await worker.executeTaskRun(payload, this.baseURL); + const completion = await worker.executeTaskRun(payload); this.onTaskCompleted.post({ completion, @@ -343,7 +331,7 @@ export class BackgroundWorker { // Write the build manifest to this.build.outputPath/worker.json await writeJSONFile(indexManifestPath, this.manifest, true); - logger.debug("Worker initialized", { index: indexManifestPath }); + logger.debug("Worker initialized", { index: indexManifestPath, path: this.build.outputPath }); } // We need to notify all the task run processes that a task run has completed, @@ -386,6 +374,8 @@ export class BackgroundWorker { env: { ...this.params.env, ...payload.environment, + TRIGGER_BUILD_MANIFEST_PATH: join(this.build.outputPath, "build.json"), + TRIGGER_WORKER_MANIFEST_PATH: join(this.build.outputPath, "index.json"), }, serverWorker: this.serverWorker, workerManifest: this.manifest, @@ -499,7 +489,6 @@ export class BackgroundWorker { // We need to fork the process before we can execute any tasks async executeTaskRun( payload: TaskRunExecutionPayload, - baseURL: string, messageId?: string ): Promise { if (this._closed) { @@ -514,24 +503,7 @@ export class BackgroundWorker { throw new Error("Worker not registered"); } - const { execution } = payload; - // ○ Mar 27 09:17:25.653 -> View logs | 20240326.20 | create-avatar | run_slufhjdfiv8ejnrkw9dsj.1 - - const logsUrl = `${baseURL}/runs/${execution.run.id}`; - - const pipe = chalkGrey("|"); - const bullet = chalkGrey("○"); - const link = chalkLink(cliLink("View logs", logsUrl)); - let timestampPrefix = chalkGrey(prettyPrintDate(payload.execution.attempt.startedAt)); - const workerPrefix = chalkWorker(this.serverWorker.version); - const taskPrefix = chalkTask(execution.task.id); - const runId = chalkRun(`${execution.run.id}.${execution.attempt.number}`); - - logger.log( - `${bullet} ${timestampPrefix} ${chalkGrey( - "->" - )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId}` - ); + eventBus.emit("runStarted", this, payload); const now = performance.now(); @@ -539,37 +511,7 @@ export class BackgroundWorker { const elapsed = performance.now() - now; - const retryingText = chalkGrey( - !completion.ok && completion.skippedRetrying - ? " (retrying skipped)" - : !completion.ok && completion.retry !== undefined - ? ` (retrying in ${completion.retry.delay}ms)` - : "" - ); - - const resultText = !completion.ok - ? completion.error.type === "INTERNAL_ERROR" && - (completion.error.code === TaskRunErrorCodes.TASK_EXECUTION_ABORTED || - completion.error.code === TaskRunErrorCodes.TASK_RUN_CANCELLED) - ? chalkWarning("Cancelled") - : `${chalkError("Error")}${retryingText}` - : chalkSuccess("Success"); - - const errorText = !completion.ok - ? formatErrorLog(completion.error) - : "retry" in completion - ? `retry in ${completion.retry}ms` - : ""; - - const elapsedText = chalkGrey(`(${formatDurationMilliseconds(elapsed, { style: "short" })})`); - - timestampPrefix = chalkGrey(prettyPrintDate()); - - logger.log( - `${bullet} ${timestampPrefix} ${chalkGrey( - "->" - )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId} ${pipe} ${resultText} ${elapsedText}${errorText}` - ); + eventBus.emit("runCompleted", this, payload, completion, elapsed); return completion; } @@ -652,6 +594,7 @@ export class BackgroundWorker { error: { type: "INTERNAL_ERROR", code: TaskRunErrorCodes.TASK_EXECUTION_FAILED, + message: String(e), }, }; } @@ -808,7 +751,7 @@ class TaskRunProcess { this._attemptStatuses.set(this.payload.execution.attempt.id, "PENDING"); // @ts-expect-error - We know that the resolver and rejecter are defined - this._attemptPromises.set(payload.execution.attempt.id, { resolver, rejecter }); + this._attemptPromises.set(this.payload.execution.attempt.id, { resolver, rejecter }); const { execution, traceContext } = this.payload; @@ -1015,20 +958,3 @@ class TaskRunProcess { return this._childPid; } } - -function formatErrorLog(error: TaskRunError) { - switch (error.type) { - case "INTERNAL_ERROR": { - return ""; - } - case "STRING_ERROR": { - return `\n\n${chalkError("X Error:")} ${error.raw}\n`; - } - case "CUSTOM_ERROR": { - return `\n\n${chalkError("X Error:")} ${error.raw}\n`; - } - case "BUILT_IN_ERROR": { - return `\n\n${error.stackTrace.replace(/^Error: /, chalkError("X Error: "))}\n`; - } - } -} diff --git a/packages/cli-v3/src/dev/devOutput.ts b/packages/cli-v3/src/dev/devOutput.ts index ebab3f566..7953afb41 100644 --- a/packages/cli-v3/src/dev/devOutput.ts +++ b/packages/cli-v3/src/dev/devOutput.ts @@ -1,8 +1,21 @@ import { ResolvedConfig } from "@trigger.dev/core/v3/build"; import { DevCommandOptions } from "../commands/dev.js"; import { logger } from "../utilities/logger.js"; -import { chalkGrey, chalkLink, chalkWorker, cliLink } from "../utilities/cliOutput.js"; +import { + chalkError, + chalkGrey, + chalkLink, + chalkRun, + chalkSuccess, + chalkTask, + chalkWarning, + chalkWorker, + cliLink, + prettyPrintDate, +} from "../utilities/cliOutput.js"; import { eventBus, EventBusEventArgs } from "../utilities/eventBus.js"; +import { TaskRunError, TaskRunErrorCodes } from "@trigger.dev/core/v3/schemas"; +import { formatDurationMilliseconds } from "@trigger.dev/core/v3"; export type DevOutputOptions = { name: string | undefined; @@ -14,6 +27,8 @@ export type DevOutputOptions = { export function startDevOutput(options: DevOutputOptions) { const { dashboardUrl, config } = options; + const baseUrl = `${dashboardUrl}/projects/v3/${config.project}`; + const rebuildStarted = (...[target]: EventBusEventArgs<"rebuildStarted">) => { logger.log(chalkGrey("○ Rebuilding background worker…")); }; @@ -47,15 +62,109 @@ export function startDevOutput(options: DevOutputOptions) { ); }; + const runStarted = (...[worker, payload]: EventBusEventArgs<"runStarted">) => { + if (!worker.serverWorker) { + return; + } + + const { execution } = payload; + + // ○ Mar 27 09:17:25.653 -> View logs | 20240326.20 | create-avatar | run_slufhjdfiv8ejnrkw9dsj.1 + const logsUrl = `${baseUrl}/runs/${execution.run.id}`; + const pipe = chalkGrey("|"); + const bullet = chalkGrey("○"); + const link = chalkLink(cliLink("View logs", logsUrl)); + let timestampPrefix = chalkGrey(prettyPrintDate(payload.execution.attempt.startedAt)); + const workerPrefix = chalkWorker(worker.serverWorker.version); + const taskPrefix = chalkTask(execution.task.id); + const runId = chalkRun(`${execution.run.id}.${execution.attempt.number}`); + + logger.log( + `${bullet} ${timestampPrefix} ${chalkGrey( + "->" + )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId}` + ); + }; + + const runCompleted = ( + ...[worker, payload, completion, durationMs]: EventBusEventArgs<"runCompleted"> + ) => { + const { execution } = payload; + + const retryingText = chalkGrey( + !completion.ok && completion.skippedRetrying + ? " (retrying skipped)" + : !completion.ok && completion.retry !== undefined + ? ` (retrying in ${completion.retry.delay}ms)` + : "" + ); + + const resultText = !completion.ok + ? completion.error.type === "INTERNAL_ERROR" && + (completion.error.code === TaskRunErrorCodes.TASK_EXECUTION_ABORTED || + completion.error.code === TaskRunErrorCodes.TASK_RUN_CANCELLED) + ? chalkWarning("Cancelled") + : `${chalkError("Error")}${retryingText}` + : chalkSuccess("Success"); + + const errorText = !completion.ok + ? formatErrorLog(completion.error) + : "retry" in completion + ? `retry in ${completion.retry}ms` + : ""; + + const elapsedText = chalkGrey( + `(${formatDurationMilliseconds(durationMs, { style: "short" })})` + ); + + const timestampPrefix = chalkGrey(prettyPrintDate()); + + const logsUrl = `${baseUrl}/runs/${execution.run.id}`; + const pipe = chalkGrey("|"); + const bullet = chalkGrey("○"); + const link = chalkLink(cliLink("View logs", logsUrl)); + + const workerPrefix = chalkWorker(worker.serverWorker!.version); + const taskPrefix = chalkTask(execution.task.id); + const runId = chalkRun(`${execution.run.id}.${execution.attempt.number}`); + + logger.log( + `${bullet} ${timestampPrefix} ${chalkGrey( + "->" + )} ${link} ${pipe} ${workerPrefix} ${pipe} ${taskPrefix} ${pipe} ${runId} ${pipe} ${resultText} ${elapsedText}${errorText}` + ); + }; + eventBus.on("rebuildStarted", rebuildStarted); eventBus.on("buildStarted", buildStarted); eventBus.on("workerSkipped", workerSkipped); eventBus.on("backgroundWorkerInitialized", backgroundWorkerInitialized); + eventBus.on("runStarted", runStarted); + eventBus.on("runCompleted", runCompleted); return () => { eventBus.off("rebuildStarted", rebuildStarted); eventBus.off("buildStarted", buildStarted); eventBus.off("workerSkipped", workerSkipped); eventBus.off("backgroundWorkerInitialized", backgroundWorkerInitialized); + eventBus.off("runStarted", runStarted); + eventBus.off("runCompleted", runCompleted); }; } + +function formatErrorLog(error: TaskRunError) { + switch (error.type) { + case "INTERNAL_ERROR": { + return ""; + } + case "STRING_ERROR": { + return `\n\n${chalkError("X Error:")} ${error.raw}\n`; + } + case "CUSTOM_ERROR": { + return `\n\n${chalkError("X Error:")} ${error.raw}\n`; + } + case "BUILT_IN_ERROR": { + return `\n\n${error.stackTrace.replace(/^Error: /, chalkError("X Error: "))}\n`; + } + } +} diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts index e86a5fab3..cf853dc30 100644 --- a/packages/cli-v3/src/dev/workerRuntime.ts +++ b/packages/cli-v3/src/dev/workerRuntime.ts @@ -66,9 +66,7 @@ class DevWorkerRuntime implements WorkerRuntime { }, }); - this.backgroundWorkerCoordinator = new BackgroundWorkerCoordinator( - `${options.dashboardUrl}/projects/v3/${options.config.project}` - ); + this.backgroundWorkerCoordinator = new BackgroundWorkerCoordinator(); this.backgroundWorkerCoordinator.onWorkerTaskRunHeartbeat.attach( async ({ worker, backgroundWorkerId, id }) => { diff --git a/packages/cli-v3/src/entryPoints/dev.ts b/packages/cli-v3/src/entryPoints/dev.ts index 2cbce9b9f..37bc6983a 100644 --- a/packages/cli-v3/src/entryPoints/dev.ts +++ b/packages/cli-v3/src/entryPoints/dev.ts @@ -88,7 +88,7 @@ async function bootstrap() { const buildManifest = await loadBuildManifest(); const workerManifest = await loadWorkerManifest(); - const { config, handleError } = await importConfig(process.env.TRIGGER_BUILD_MANIFEST_PATH!); + const { config, handleError } = await importConfig(buildManifest.configPath); const tracingSDK = new TracingSDK({ url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318", @@ -116,6 +116,14 @@ async function bootstrap() { logger.setGlobalTaskLogger(otelTaskLogger); + for (const task of workerManifest.tasks) { + taskCatalog.registerTaskFileMetadata(task.id, { + exportName: task.exportName, + filePath: task.filePath, + entryPoint: task.entryPoint, + }); + } + return { tracer, tracingSDK, @@ -126,31 +134,6 @@ async function bootstrap() { }; } -async function registerTaskFileMetadata(files: Array<{ entry: string; out: string }>) { - for (const file of files) { - console.log("Detecting exported tasks in file", file.out); - - const module = await import(file.out); - - for (const exportName of Object.keys(module)) { - const task = module[exportName]; - - if (!task) { - continue; - } - - if (task[Symbol.for("trigger.dev/task")]) { - if (taskCatalog.taskExists(task.id)) { - taskCatalog.registerTaskFileMetadata(task.id, { - exportName, - filePath: file.entry, - }); - } - } - } - } -} - let _execution: TaskRunExecution | undefined; let _isRunning = false; let _tracingSDK: TracingSDK | undefined; @@ -180,8 +163,58 @@ const handler = new ZodMessageHandler({ return; } + const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn, workerManifest } = + await bootstrap(); + + const taskManifest = workerManifest.tasks.find((t) => t.id === execution.task.id); + + if (!taskManifest) { + console.error(`Could not find task ${execution.task.id}`); + + await sender.send("TASK_RUN_COMPLETED", { + execution, + result: { + ok: false, + id: execution.run.id, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.COULD_NOT_FIND_TASK, + }, + usage: { + durationMs: 0, + }, + }, + }); + + return; + } + + try { + await import(taskManifest.entryPoint); + } catch (err) { + console.error(`Failed to import task ${execution.task.id}`, err); + + await sender.send("TASK_RUN_COMPLETED", { + execution, + result: { + ok: false, + id: execution.run.id, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.COULD_NOT_IMPORT_TASK, + }, + usage: { + durationMs: 0, + }, + }, + }); + + return; + } + process.title = `trigger-dev-worker: ${execution.task.id} ${execution.run.id}`; + // Import the task module const task = taskCatalog.getTask(execution.task.id); if (!task) { @@ -205,9 +238,6 @@ const handler = new ZodMessageHandler({ return; } - const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn, workerManifest } = - await bootstrap(); - const executor = new TaskExecutor(task, { tracer, tracingSDK, @@ -294,7 +324,4 @@ async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeco return _doHeartbeat(); } -// Start the async interval after 30 seconds -asyncHeartbeat().catch((err) => { - console.error("Failed to start asyncHeartbeat", err); -}); +await asyncHeartbeat(); diff --git a/packages/cli-v3/src/utilities/eventBus.ts b/packages/cli-v3/src/utilities/eventBus.ts index fd254b4d8..9fc6869b6 100644 --- a/packages/cli-v3/src/utilities/eventBus.ts +++ b/packages/cli-v3/src/utilities/eventBus.ts @@ -1,4 +1,4 @@ -import { BuildTarget } from "@trigger.dev/core/v3"; +import { BuildTarget, TaskRunExecutionPayload, TaskRunExecutionResult } from "@trigger.dev/core/v3"; import { EventEmitter } from "node:events"; import { BackgroundWorker } from "../dev/backgroundWorker.js"; @@ -7,6 +7,8 @@ export type EventBusEvents = { buildStarted: [BuildTarget]; workerSkipped: []; backgroundWorkerInitialized: [BackgroundWorker]; + runStarted: [BackgroundWorker, TaskRunExecutionPayload]; + runCompleted: [BackgroundWorker, TaskRunExecutionPayload, TaskRunExecutionResult, number]; }; export type EventBusEventArgs = EventBusEvents[T]; diff --git a/packages/core/src/v3/schemas/common.ts b/packages/core/src/v3/schemas/common.ts index f0964cc8a..06ac0eb72 100644 --- a/packages/core/src/v3/schemas/common.ts +++ b/packages/core/src/v3/schemas/common.ts @@ -79,6 +79,7 @@ export type TaskRunStringError = z.infer; export const TaskRunErrorCodes = { COULD_NOT_FIND_EXECUTOR: "COULD_NOT_FIND_EXECUTOR", COULD_NOT_FIND_TASK: "COULD_NOT_FIND_TASK", + COULD_NOT_IMPORT_TASK: "COULD_NOT_IMPORT_TASK", CONFIGURED_INCORRECTLY: "CONFIGURED_INCORRECTLY", TASK_ALREADY_RUNNING: "TASK_ALREADY_RUNNING", TASK_EXECUTION_FAILED: "TASK_EXECUTION_FAILED", @@ -97,6 +98,7 @@ export const TaskRunInternalError = z.object({ code: z.enum([ "COULD_NOT_FIND_EXECUTOR", "COULD_NOT_FIND_TASK", + "COULD_NOT_IMPORT_TASK", "CONFIGURED_INCORRECTLY", "TASK_ALREADY_RUNNING", "TASK_EXECUTION_FAILED",