Merge pull request #2183 from triggerdotdev/process-keep-alive

v4: Experimental process keep alive

Author: ericallam

.changeset/polite-badgers-suffer.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+experimental processKeepAlive

apps/supervisor/src/util.ts

@@ -1,5 +1,15 @@
 import { isMacOS, isWindows } from "std-env";
 
+export function normalizeDockerHostUrl(url: string) {
+  const $url = new URL(url);
+
+  if ($url.hostname === "localhost") {
+    $url.hostname = getDockerHostDomain();
+  }
+
+  return $url.toString();
+}
+
 export function getDockerHostDomain() {
   return isMacOS || isWindows ? "host.docker.internal" : "localhost";
 }

apps/supervisor/src/workloadManager/docker.ts

@@ -5,7 +5,7 @@ import {
   type WorkloadManagerOptions,
 } from "./types.js";
 import { env } from "../env.js";
-import { getDockerHostDomain, getRunnerId } from "../util.js";
+import { getDockerHostDomain, getRunnerId, normalizeDockerHostUrl } from "../util.js";
 import Docker from "dockerode";
 import { tryCatch } from "@trigger.dev/core";
 
@@ -78,7 +78,7 @@ export class DockerWorkloadManager implements WorkloadManager {
     ];
 
     if (this.opts.warmStartUrl) {
-      envVars.push(`TRIGGER_WARM_START_URL=${this.opts.warmStartUrl}`);
+      envVars.push(`TRIGGER_WARM_START_URL=${normalizeDockerHostUrl(this.opts.warmStartUrl)}`);
     }
 
     if (this.opts.metadataUrl) {

apps/webapp/app/utils/timelineSpanEvents.ts

@@ -142,9 +142,6 @@ function getFriendlyNameForEvent(event: string, properties?: Record<string, any>
       return "Attempt created";
     }
     case "import": {
-      if (properties && typeof properties.file === "string") {
-        return `Importing ${properties.file}`;
-      }
       return "Importing task file";
     }
     case "lazy_payload": {

apps/webapp/app/v3/otlpExporter.server.ts

@@ -76,7 +76,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
     expect(result.some((event) => event.name === "Dequeued")).toBe(true);
     expect(result.some((event) => event.name === "Launched")).toBe(true);
     expect(result.some((event) => event.name === "Attempt created")).toBe(true);
-    expect(result.some((event) => event.name === "Importing src/trigger/chat.ts")).toBe(true);
+    expect(result.some((event) => event.name === "Importing task file")).toBe(true);
   });
 
   test("should sort events by timestamp", () => {
@@ -86,7 +86,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
     expect(result[0].name).toBe("Dequeued");
     expect(result[1].name).toBe("Attempt created");
     expect(result[2].name).toBe("Launched");
-    expect(result[3].name).toBe("Importing src/trigger/chat.ts");
+    expect(result[3].name).toBe("Importing task file");
   });
 
   test("should calculate offsets correctly from the first event", () => {
@@ -176,7 +176,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
     expect(result.find((e) => e.name === "Attempt created")?.helpText).toBe(
       "An attempt was created for the run"
     );
-    expect(result.find((e) => e.name === "Importing src/trigger/chat.ts")?.helpText).toBe(
+    expect(result.find((e) => e.name === "Importing task file")?.helpText).toBe(
       "A task file was imported"
     );
   });
@@ -187,7 +187,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
     expect(result.find((e) => e.name === "Dequeued")?.duration).toBe(0);
     expect(result.find((e) => e.name === "Launched")?.duration).toBe(127);
     expect(result.find((e) => e.name === "Attempt created")?.duration).toBe(56);
-    expect(result.find((e) => e.name === "Importing src/trigger/chat.ts")?.duration).toBe(67);
+    expect(result.find((e) => e.name === "Importing task file")?.duration).toBe(67);
   });
 
   test("should use fallback name for import event without file property", () => {
@@ -214,7 +214,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
     // Without fork event, import should also be visible for non-admins
     expect(result.length).toBe(2);
     expect(result.some((event) => event.name === "Dequeued")).toBe(true);
-    expect(result.some((event) => event.name === "Importing src/trigger/chat.ts")).toBe(true);
+    expect(result.some((event) => event.name === "Importing task file")).toBe(true);
 
     // create_attempt should still be admin-only
     expect(result.some((event) => event.name === "Attempt created")).toBe(false);

apps/webapp/test/timelineSpanEvents.test.ts

@@ -11,7 +11,7 @@ export default defineConfig({
         singleThread: true,
       },
     },
-    testTimeout: 60_000,
+    testTimeout: 120_000,
     coverage: {
       provider: "v8",
     },

internal-packages/run-engine/vitest.config.ts

@@ -25,6 +25,8 @@ import {
 import pLimit from "p-limit";
 import { resolveLocalEnvVars } from "../utilities/localEnvVars.js";
 import type { Metafile } from "esbuild";
+import { TaskRunProcessPool } from "./taskRunProcessPool.js";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 export type WorkerRuntimeOptions = {
   name: string | undefined;
@@ -67,6 +69,7 @@ class DevSupervisor implements WorkerRuntime {
   private socketConnections = new Set<string>();
 
   private runLimiter?: ReturnType<typeof pLimit>;
+  private taskRunProcessPool?: TaskRunProcessPool;
 
   constructor(public readonly options: WorkerRuntimeOptions) {}
 
@@ -95,6 +98,42 @@ class DevSupervisor implements WorkerRuntime {
 
     this.runLimiter = pLimit(maxConcurrentRuns);
 
+    // Initialize the task run process pool
+    const env = await this.#getEnvVars();
+
+    const enableProcessReuse =
+      typeof this.options.config.experimental_processKeepAlive === "boolean"
+        ? this.options.config.experimental_processKeepAlive
+        : typeof this.options.config.experimental_processKeepAlive === "object"
+        ? this.options.config.experimental_processKeepAlive.enabled
+        : false;
+
+    const maxPoolSize =
+      typeof this.options.config.experimental_processKeepAlive === "object"
+        ? this.options.config.experimental_processKeepAlive.devMaxPoolSize ?? 25
+        : 25;
+
+    const maxExecutionsPerProcess =
+      typeof this.options.config.experimental_processKeepAlive === "object"
+        ? this.options.config.experimental_processKeepAlive.maxExecutionsPerProcess ?? 50
+        : 50;
+
+    if (enableProcessReuse) {
+      logger.debug("[DevSupervisor] Enabling process reuse", {
+        enableProcessReuse,
+        maxPoolSize,
+        maxExecutionsPerProcess,
+      });
+    }
+
+    this.taskRunProcessPool = new TaskRunProcessPool({
+      env,
+      cwd: this.options.config.workingDir,
+      enableProcessReuse,
+      maxPoolSize,
+      maxExecutionsPerProcess,
+    });
+
     this.socket = this.#createSocket();
 
     //start an SSE connection for presence
@@ -111,6 +150,17 @@ class DevSupervisor implements WorkerRuntime {
     } catch (error) {
       logger.debug("[DevSupervisor] shutdown, socket failed to close", { error });
     }
+
+    // Shutdown the task run process pool
+    if (this.taskRunProcessPool) {
+      const [shutdownError] = await tryCatch(this.taskRunProcessPool.shutdown());
+
+      if (shutdownError) {
+        logger.debug("[DevSupervisor] shutdown, task run process pool failed to shutdown", {
+          error: shutdownError,
+        });
+      }
+    }
   }
 
   async initializeWorker(
@@ -293,12 +343,21 @@ class DevSupervisor implements WorkerRuntime {
           continue;
         }
 
+        if (!this.taskRunProcessPool) {
+          logger.debug(`[DevSupervisor] dequeueRuns. No task run process pool`, {
+            run: message.run.friendlyId,
+            worker,
+          });
+          continue;
+        }
+
         //new run
         runController = new DevRunController({
           runFriendlyId: message.run.friendlyId,
           worker: worker,
           httpClient: this.options.client,
           logLevel: this.options.args.logLevel,
+          taskRunProcessPool: this.taskRunProcessPool,
           onFinished: () => {
             logger.debug("[DevSupervisor] Run finished", { runId: message.run.friendlyId });
 
@@ -574,6 +633,10 @@ class DevSupervisor implements WorkerRuntime {
       return;
     }
 
+    if (worker.serverWorker?.version) {
+      this.taskRunProcessPool?.deprecateVersion(worker.serverWorker?.version);
+    }
+
     if (this.#workerHasInProgressRuns(friendlyId)) {
       return;
     }