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

dev runs working

Browse Source
This commit is contained in:
Eric Allam
2024-08-09 14:22:17 +01:00
committed by Eric Allam
parent 6bf65838af
commit 7130e305fd
7 changed files with 229 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli-v3/src/build/plugins.ts
+39 -24
View File
@@ -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",
};
}
);
}
},
};
}
packages/cli-v3/src/dev/backgroundWorker.ts
+14 -88
View File
@@ -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<string, BackgroundWorker> = 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<TaskRunExecutionResult> {
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`;
}
}
}
packages/cli-v3/src/dev/devOutput.ts
+110 -1
View File
@@ -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`;
}
}
}
packages/cli-v3/src/dev/workerRuntime.ts
+1 -3
View File
@@ -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 }) => {
packages/cli-v3/src/entryPoints/dev.ts
+60 -33
View File
@@ -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();
packages/cli-v3/src/utilities/eventBus.ts
+3 -1
View File
@@ -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<T extends keyof EventBusEvents> = EventBusEvents[T];
packages/core/src/v3/schemas/common.ts
+2
View File
@@ -79,6 +79,7 @@ export type TaskRunStringError = z.infer<typeof TaskRunStringError>;
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",