triggerdotdev · ericallam · Jan 10, 2025 · Jan 9, 2025 · Jan 10, 2025
diff --git a/.changeset/unlucky-meals-develop.md b/.changeset/unlucky-meals-develop.md
@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+cleanup deprecated background worker files (fixes #1572)
diff --git a/packages/cli-v3/src/commands/dev.ts b/packages/cli-v3/src/commands/dev.ts
@@ -18,6 +18,7 @@ const DevCommandOptions = CommonCommandOptions.extend({
  projectRef: z.string().optional(),
  skipUpdateCheck: z.boolean().default(false),
  envFile: z.string().optional(),
+  keepTmpFiles: z.boolean().default(false),
 });

 export type DevCommandOptions = z.infer<typeof DevCommandOptions>;
@@ -38,6 +39,10 @@ export function configureDevCommand(program: Command) {
      )
      .option("--debug-otel", "Enable OpenTelemetry debugging")
      .option("--skip-update-check", "Skip checking for @trigger.dev package updates")
+      .option(
+        "--keep-tmp-files",
+        "Keep temporary files after the dev session ends, helpful for debugging"
+      )
  ).action(async (options) => {
    wrapCommandAction("dev", DevCommandOptions, options, async (opts) => {
      await devCommand(opts);
@@ -151,6 +156,7 @@ async function startDev(options: StartDevOptions) {
        initialMode: "local",
        dashboardUrl: options.login.dashboardUrl,
        showInteractiveDevSession: true,
+        keepTmpFiles: options.keepTmpFiles,
      });
    }


diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts
@@ -111,6 +111,7 @@ export class BackgroundWorkerCoordinator {
    }

    this._backgroundWorkers.set(worker.serverWorker.id, worker);
+
    this.onWorkerRegistered.post({
      worker,
      id: worker.serverWorker.id,
@@ -126,14 +127,6 @@ export class BackgroundWorkerCoordinator {
    });
  }

-  close() {
-    for (const worker of this._backgroundWorkers.values()) {
-      worker.close();
-    }
-
-    this._backgroundWorkers.clear();
-  }
-
  async executeTaskRun(id: string, payload: TaskRunExecutionPayload, messageId: string) {
    const worker = this._backgroundWorkers.get(id);

@@ -186,11 +179,11 @@ export class BackgroundWorkerCoordinator {
 export type BackgroundWorkerOptions = {
  env: Record<string, string>;
  cwd: string;
+  stop: () => void;
 };

 export class BackgroundWorker {
  public onTaskRunHeartbeat: Evt<string> = new Evt();
-  private _onClose: Evt<void> = new Evt();

  public deprecated: boolean = false;
  public manifest: WorkerManifest | undefined;
@@ -199,33 +192,27 @@ export class BackgroundWorker {
  _taskRunProcesses: Map<string, TaskRunProcess> = new Map();
  private _taskRunProcessesBeingKilled: Map<number, TaskRunProcess> = new Map();

-  private _closed: boolean = false;
-
  constructor(
    public build: BuildManifest,
    public params: BackgroundWorkerOptions
  ) {}

  deprecate() {
-    this.deprecated = true;
-  }
-
-  close() {
-    if (this._closed) {
+    if (this.deprecated) {
      return;
    }

-    this._closed = true;
+    this.deprecated = true;

-    this.onTaskRunHeartbeat.detach();
+    this.#tryStopWorker();
+  }

-    // We need to close all the task run processes
-    for (const taskRunProcess of this._taskRunProcesses.values()) {
-      taskRunProcess.cleanup(true);
-    }
+  #tryStopWorker() {
+    if (this.deprecated && this._taskRunProcesses.size === 0) {
+      logger.debug("Worker deprecated, stopping", { outputPath: this.build.outputPath });

-    // Delete worker files
-    this._onClose.post();
+      this.params.stop();
+    }
  }

  get inProgressRuns(): Array<string> {
@@ -301,8 +288,6 @@ export class BackgroundWorker {
      throw new Error("Worker not initialized");
    }

-    this._closed = false;
-
    logger.debug(this.#prefixedMessage(payload, "killing current task run process before attempt"));

    await this.#killCurrentTaskRunProcessBeforeAttempt(payload.execution.run.id);
@@ -332,6 +317,8 @@ export class BackgroundWorker {
      // Only delete the task run process if the pid matches
      if (taskRunProcess?.pid === pid) {
        this._taskRunProcesses.delete(payload.execution.run.id);
+
+        this.#tryStopWorker();
      }

      if (pid) {
@@ -435,10 +422,6 @@ export class BackgroundWorker {
    payload: TaskRunExecutionPayload,
    messageId: string
  ): Promise<TaskRunExecutionResult> {
-    if (this._closed) {
-      throw new Error("Worker is closed");
-    }
-
    if (!this.manifest) {
      throw new Error("Worker not initialized");
    }

diff --git a/packages/cli-v3/src/dev/devSession.ts b/packages/cli-v3/src/dev/devSession.ts
@@ -23,10 +23,11 @@ import { type DevCommandOptions } from "../commands/dev.js";
 import { eventBus } from "../utilities/eventBus.js";
 import { logger } from "../utilities/logger.js";
 import { resolveFileSources } from "../utilities/sourceFiles.js";
-import { EphemeralDirectory, getTmpDir } from "../utilities/tempDirectories.js";
+import { clearTmpDirs, EphemeralDirectory, getTmpDir } from "../utilities/tempDirectories.js";
 import { VERSION } from "../version.js";
 import { startDevOutput } from "./devOutput.js";
 import { startWorkerRuntime } from "./workerRuntime.js";
+import { existsSync, mkdirSync, rmSync } from "node:fs";

 export type DevSessionOptions = {
  name: string | undefined;
@@ -37,6 +38,7 @@ export type DevSessionOptions = {
  rawArgs: DevCommandOptions;
  client: CliApiClient;
  onErr?: (error: Error) => void;
+  keepTmpFiles: boolean;
 };

 export type DevSessionInstance = {
@@ -49,8 +51,10 @@ export async function startDevSession({
  rawArgs,
  client,
  dashboardUrl,
+  keepTmpFiles,
 }: DevSessionOptions): Promise<DevSessionInstance> {
-  const destination = getTmpDir(rawConfig.workingDir, "build");
+  clearTmpDirs(rawConfig.workingDir);
+  const destination = getTmpDir(rawConfig.workingDir, "build", keepTmpFiles);

  const runtime = await startWorkerRuntime({
    name,
@@ -96,7 +100,7 @@ export async function startDevSession({
    try {
      logger.debug("Updated bundle", { bundle, buildManifest });

-      await runtime.initializeWorker(buildManifest);
+      await runtime.initializeWorker(buildManifest, workerDir?.remove ?? (() => {}));
    } catch (error) {
      if (error instanceof Error) {
        eventBus.emit("backgroundWorkerIndexingError", buildManifest, error);
@@ -124,6 +128,14 @@ export async function startDevSession({
        if (bundled) {
          eventBus.emit("rebuildStarted", "dev");
        }
+
+        const outdir = b.initialOptions.outdir;
+        if (outdir && existsSync(outdir)) {
+          logger.debug("Removing outdir", { outdir });
+
+          rmSync(outdir, { recursive: true, force: true });
+          mkdirSync(outdir, { recursive: true });
+        }
      });
      b.onEnd(async (result: esbuild.BuildResult) => {
        const errors = result.errors;
@@ -141,7 +153,7 @@ export async function startDevSession({
          // First bundle, no need to update bundle
          bundled = true;
        } else {
-          const workerDir = getTmpDir(rawConfig.workingDir, "build");
+          const workerDir = getTmpDir(rawConfig.workingDir, "build", keepTmpFiles);

          await updateBuild(result, workerDir);
        }

diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts
@@ -30,7 +30,7 @@ import { VERSION } from "../version.js";

 export interface WorkerRuntime {
  shutdown(): Promise<void>;
-  initializeWorker(manifest: BuildManifest): Promise<void>;
+  initializeWorker(manifest: BuildManifest, stop: () => void): Promise<void>;
 }

 export type WorkerRuntimeOptions = {
@@ -167,9 +167,10 @@ class DevWorkerRuntime implements WorkerRuntime {
    }
  }

-  async initializeWorker(manifest: BuildManifest, options?: { cwd?: string }): Promise<void> {
+  async initializeWorker(manifest: BuildManifest, stop: () => void): Promise<void> {
    if (this.lastBuild && this.lastBuild.contentHash === manifest.contentHash) {
      eventBus.emit("workerSkipped");
+      stop();
      return;
    }

@@ -178,18 +179,21 @@ class DevWorkerRuntime implements WorkerRuntime {
    const backgroundWorker = new BackgroundWorker(manifest, {
      env,
      cwd: this.options.config.workingDir,
+      stop,
    });

    await backgroundWorker.initialize();

    if (!backgroundWorker.manifest) {
+      stop();
      throw new Error("Could not initialize worker");
    }

    const issues = validateWorkerManifest(backgroundWorker.manifest);

    if (issues.length > 0) {
      issues.forEach((issue) => logger.error(issue));
+      stop();
      return;
    }

@@ -213,6 +217,7 @@ class DevWorkerRuntime implements WorkerRuntime {
    );

    if (!backgroundWorkerRecord.success) {
+      stop();
      throw new Error(backgroundWorkerRecord.error);
    }


diff --git a/packages/cli-v3/src/utilities/tempDirectories.ts b/packages/cli-v3/src/utilities/tempDirectories.ts
@@ -30,8 +30,14 @@ export function getTmpDir(
  const tmpPrefix = path.join(tmpRoot, `${prefix}-`);
  const tmpDir = fs.realpathSync(fs.mkdtempSync(tmpPrefix));

-  let removeDir = keep ? () => {} : () => fs.rmSync(tmpDir, { recursive: true, force: true });
-  let removeExitListener = keep ? () => {} : onExit(removeDir);
+  const removeDir = () => {
+    try {
+      return fs.rmSync(tmpDir, { recursive: true, force: true });
+    } catch (e) {
+      // This sometimes fails on Windows with EBUSY
+    }
+  };
+  const removeExitListener = keep ? () => {} : onExit(removeDir);

  return {
    path: tmpDir,
@@ -41,3 +47,14 @@ export function getTmpDir(
    },
  };
 }
+
+export function clearTmpDirs(projectRoot: string | undefined) {
+  projectRoot ??= process.cwd();
+  const tmpRoot = path.join(projectRoot, ".trigger", "tmp");
+
+  try {
+    fs.rmSync(tmpRoot, { recursive: true, force: true });
+  } catch (e) {
+    // This sometimes fails on Windows with EBUSY
+  }
+}