Latest lockfile

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.changeset/breezy-gorillas-mate.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+better handle task metadata parse errors, and display nicely formatted errors

.changeset/clean-pianos-listen.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+add machine config and secure zod connection

.changeset/cool-glasses-bake.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Configurable log levels in the config file and via env var

.changeset/eight-pumas-float.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Fixed batch otel flushing

.changeset/few-students-share.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fix permissions inside node_modules

.changeset/green-bags-wink.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/react": patch
+---
+
+Fix for shared queryKey between useRunDetails and useRunStatuses

.changeset/khaki-apricots-design.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Retry 429, 500, and connection error API requests to the trigger.dev server

.changeset/late-icons-lie.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Display errors for runs and deployments

.changeset/light-bulldogs-press.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Changed "Worker" to "Version" in the dev command key

.changeset/light-dragons-complain.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Correctly handle self-hosted deploy command errors

.changeset/loud-actors-remember.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Default to retrying enabled in dev when running init

.changeset/many-ligers-pump.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Handle string and non-stringifiable outputs like functions

.changeset/mighty-camels-joke.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Improve error messages during dev/deploy and handle deploy image build issues

.changeset/new-rivers-tell.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Added replayRun function to the SDK

.changeset/odd-poets-own.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Added a Node.js runtime check for the CLI

.changeset/polite-ducks-switch.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fix post start hooks

.changeset/pre.json

@@ -44,8 +44,34 @@
     "@trigger.dev/yalt": "2.3.18"
   },
   "changesets": [
+    "breezy-gorillas-mate",
     "chilled-hornets-move",
+    "clean-pianos-listen",
+    "cool-glasses-bake",
+    "eight-pumas-float",
+    "few-students-share",
+    "green-bags-wink",
+    "khaki-apricots-design",
+    "late-icons-lie",
     "lemon-jobs-repair",
-    "sweet-lizards-press"
+    "light-bulldogs-press",
+    "light-dragons-complain",
+    "loud-actors-remember",
+    "many-ligers-pump",
+    "mighty-camels-joke",
+    "new-rivers-tell",
+    "odd-poets-own",
+    "polite-ducks-switch",
+    "rare-roses-float",
+    "real-planets-stare",
+    "sharp-zebras-serve",
+    "smart-olives-eat",
+    "spicy-lamps-smoke",
+    "strange-ghosts-matter",
+    "stupid-bulldogs-applaud",
+    "sweet-lizards-press",
+    "tall-bees-wave",
+    "tame-guests-know",
+    "tiny-doors-type"
   ]
 }

.changeset/rare-roses-float.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add openssl to prod worker image and allow passing auth token via env var for deploy

.changeset/real-planets-stare.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Fixed incorrect span timings around checkpoints by implementing a precise wall clock that resets after restores

.changeset/sharp-zebras-serve.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Update trigger.dev CLI for new batch otel support

.changeset/smart-olives-eat.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fix CLI logout and add list-profiles command

.changeset/spicy-lamps-smoke.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fixing an issue with bundling @trigger.dev/core/v3 in dev when using pnpm

.changeset/strange-ghosts-matter.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Added DEBUG to the ignored env vars

.changeset/stupid-bulldogs-applaud.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fixed an issue where the trigger.dev package was not being built before publishing to npm

.changeset/tall-bees-wave.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Dynamically import superjson and fix some bundling issues

.changeset/tame-guests-know.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Changed the binary name from trigger.dev to triggerdev to fix a Windows issue

.changeset/tiny-doors-type.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Added cancelRun to the SDK

.env.example

@@ -63,4 +63,13 @@ COORDINATOR_SECRET=coordinator-secret # generate the actual secret with `openssl
 # These are needed for the object store (for handling large payloads/outputs)
 # OBJECT_STORE_BASE_URL="https://{bucket}.{accountId}.r2.cloudflarestorage.com"
 # OBJECT_STORE_ACCESS_KEY_ID=
-# OBJECT_STORE_SECRET_ACCESS_KEY=
+# OBJECT_STORE_SECRET_ACCESS_KEY=
+# RUNTIME_WAIT_THRESHOLD_IN_MS=10000
+
+# These control the server-side internal telemetry
+# INTERNAL_OTEL_TRACE_EXPORTER_URL=<URL to send traces to>
+# INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_NAME=<Header name for the auth token>
+# INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_VALUE=<Auth token value>
+# INTERNAL_OTEL_TRACE_LOGGING_ENABLED=1
+# INTERNAL_OTEL_TRACE_SAMPING_RATE=20 # this means 1/20 traces or 5% of traces will be sampled (sampled = recorded)
+# INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED=0,

.vscode/launch.json

@@ -33,7 +33,7 @@
       "type": "node-terminal",
       "request": "launch",
       "name": "Debug V3 Dev CLI",
-      "command": "pnpm exec trigger.dev dev --log-level debug",
+      "command": "pnpm exec triggerdev dev --log-level debug",
       "cwd": "${workspaceFolder}/references/v3-catalog",
       "sourceMaps": true
     },
@@ -41,7 +41,7 @@
       "type": "node-terminal",
       "request": "launch",
       "name": "Debug V3 Deploy CLI",
-      "command": "pnpm exec trigger.dev deploy",
+      "command": "pnpm exec triggerdev deploy",
       "cwd": "${workspaceFolder}/references/v3-catalog",
       "sourceMaps": true
     },

apps/coordinator/.env.example

@@ -1,3 +1,4 @@
 HTTP_SERVER_PORT=8020
 PLATFORM_ENABLED=true
-PLATFORM_WS_PORT=3030
+PLATFORM_WS_PORT=3030
+SECURE_CONNECTION=false

apps/coordinator/src/index.ts

@@ -18,6 +18,8 @@ collectDefaultMetrics();
 
 const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || 8020);
 const NODE_NAME = process.env.NODE_NAME || "coordinator";
+const DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS = 30_000;
+
 const REGISTRY_HOST = process.env.REGISTRY_HOST || "localhost:5000";
 const CHECKPOINT_PATH = process.env.CHECKPOINT_PATH || "/checkpoints";
 const REGISTRY_TLS_VERIFY = process.env.REGISTRY_TLS_VERIFY === "false" ? "false" : "true";
@@ -26,6 +28,7 @@ const PLATFORM_ENABLED = ["1", "true"].includes(process.env.PLATFORM_ENABLED ??
 const PLATFORM_HOST = process.env.PLATFORM_HOST || "127.0.0.1";
 const PLATFORM_WS_PORT = process.env.PLATFORM_WS_PORT || 3030;
 const PLATFORM_SECRET = process.env.PLATFORM_SECRET || "coordinator-secret";
+const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?? "false");
 
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 
@@ -333,12 +336,19 @@ class TaskCoordinator {
     { resolve: (value: void) => void; reject: (err?: any) => void }
   >();
 
+  #delayThresholdInMs: number;
+
   constructor(
     private port: number,
     private host = "0.0.0.0"
   ) {
     this.#httpServer = this.#createHttpServer();
     this.#checkpointer.initialize();
+    this.#delayThresholdInMs = this.#getDelayThreshold();
+
+    if (process.env.DELAY_THRESHOLD_IN_MS) {
+      this.#delayThresholdInMs = this.#getDelayThreshold();
+    }
 
     const io = new Server(this.#httpServer);
     this.#prodWorkerNamespace = this.#createProdWorkerNamespace(io);
@@ -355,6 +365,27 @@ class TaskCoordinator {
     register.registerMetric(connectedTasksTotal);
   }
 
+  #getDelayThreshold() {
+    if (!process.env.RETRY_DELAY_THRESHOLD_IN_MS) {
+      return DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS;
+    }
+
+    const threshold = parseInt(process.env.RETRY_DELAY_THRESHOLD_IN_MS);
+
+    if (isNaN(threshold)) {
+      logger.log(
+        "RETRY_DELAY_THRESHOLD_IN_MS parses as NaN, must supply integer. Will use default instead.",
+        {
+          RETRY_DELAY_THRESHOLD_IN_MS: process.env.RETRY_DELAY_THRESHOLD_IN_MS,
+          DEFAULT_DELAY_THRESHOLD_IN_MS: DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS,
+        }
+      );
+      return DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS;
+    }
+
+    return threshold;
+  }
+
   #createPlatformSocket() {
     if (!PLATFORM_ENABLED) {
       console.log("INFO: platform connection disabled");
@@ -365,6 +396,7 @@ class TaskCoordinator {
       namespace: "coordinator",
       host: PLATFORM_HOST,
       port: Number(PLATFORM_WS_PORT),
+      secure: SECURE_CONNECTION,
       clientMessages: CoordinatorToPlatformMessages,
       serverMessages: PlatformToCoordinatorMessages,
       authToken: PLATFORM_SECRET,
@@ -627,7 +659,7 @@ class TaskCoordinator {
             return;
           }
 
-          if (completion.retry.delay < 10_000) {
+          if (completion.retry.delay < this.#delayThresholdInMs) {
             completeWithoutCheckpoint(false);
             return;
           }

apps/docker-provider/.env.example

@@ -2,6 +2,7 @@ HTTP_SERVER_PORT=8050
 
 PLATFORM_WS_PORT=3030
 PLATFORM_SECRET=provider-secret
+SECURE_CONNECTION=false
 
 # Use this if you are on macOS
 # COORDINATOR_HOST="host.docker.internal"

apps/docker-provider/src/index.ts

@@ -8,6 +8,7 @@ import {
   TaskOperationsIndexOptions,
 } from "@trigger.dev/core-apps";
 import { setTimeout } from "node:timers/promises";
+import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
 
 const MACHINE_NAME = process.env.MACHINE_NAME || "local";
 const COORDINATOR_PORT = process.env.COORDINATOR_PORT || 8020;
@@ -190,6 +191,9 @@ class DockerTaskOperations implements TaskOperations {
   async delete(opts: { runId: string }) {
     await this.#initialize();
 
+    const containerName = this.#getRunContainerName(opts.runId);
+    await this.#sendPreStop(containerName);
+
     logger.log("noop: delete");
   }
 
@@ -208,6 +212,26 @@ class DockerTaskOperations implements TaskOperations {
   }
 
   async #sendPostStart(containerName: string): Promise<void> {
+    try {
+      const port = await this.#getHttpServerPort(containerName);
+      logger.debug(await this.#runLifecycleCommand(containerName, port, "postStart", "restore"));
+    } catch (error) {
+      logger.error("postStart error", { error });
+      throw new Error("postStart command failed");
+    }
+  }
+
+  async #sendPreStop(containerName: string): Promise<void> {
+    try {
+      const port = await this.#getHttpServerPort(containerName);
+      logger.debug(await this.#runLifecycleCommand(containerName, port, "preStop", "terminate"));
+    } catch (error) {
+      logger.error("preStop error", { error });
+      throw new Error("preStop command failed");
+    }
+  }
+
+  async #getHttpServerPort(containerName: string): Promise<number> {
     // We first get the correct port, which is random during dev as we run with host networking and need to avoid clashes
     // FIXME: Skip this in prod
     const logs = logger.debug(await $`docker logs ${containerName}`);
@@ -219,19 +243,14 @@ class DockerTaskOperations implements TaskOperations {
       throw new Error("failed to extract port from logs");
     }
 
-    try {
-      logger.debug(await this.#runLifecycleCommand(containerName, port, "postStart", "restore"));
-    } catch (error) {
-      logger.error("postStart error", { error });
-      throw new Error("postStart command failed");
-    }
+    return port;
   }
 
-  async #runLifecycleCommand(
+  async #runLifecycleCommand<THookType extends "postStart" | "preStop">(
     containerName: string,
     port: number,
-    type: "postStart" | "preStop",
-    cause: "index" | "create" | "restore",
+    type: THookType,
+    cause: THookType extends "postStart" ? PostStartCauses : PreStopCauses,
     retryCount = 0
   ): Promise<ExecaChildProcess> {
     try {
@@ -244,15 +263,15 @@ class DockerTaskOperations implements TaskOperations {
         `127.0.0.1:${port}/${type}?cause=${cause}`,
       ]);
     } catch (error: any) {
-      if (retryCount < 6) {
-        logger.debug("retriable postStart error", { retryCount, message: error?.message });
+      if (type === "postStart" && retryCount < 6) {
+        logger.debug(`retriable ${type} error`, { retryCount, message: error?.message });
         await setTimeout(exponentialBackoff(retryCount + 1, 2, 50, 1150, 50));
 
         return this.#runLifecycleCommand(containerName, port, type, cause, retryCount + 1);
       }
 
-      logger.error("final postStart error", { message: error?.message });
-      throw new Error(`postStart command failed after ${retryCount - 1} retries`);
+      logger.error(`final ${type} error`, { message: error?.message });
+      throw new Error(`${type} command failed after ${retryCount - 1} retries`);
     }
   }
 }

apps/kubernetes-provider/.env.example

@@ -2,6 +2,7 @@ HTTP_SERVER_PORT=8060
 
 PLATFORM_WS_PORT=3030
 PLATFORM_SECRET=provider-secret
+SECURE_CONNECTION=false
 
 # Use this if you are on macOS
 # COORDINATOR_HOST="host.docker.internal"

apps/kubernetes-provider/package.json

@@ -19,6 +19,7 @@
     "@kubernetes/client-node": "^0.20.0",
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/core-apps": "workspace:*",
+    "p-queue": "^8.0.1",
     "socket.io-client": "^4.7.4"
   },
   "devDependencies": {

apps/kubernetes-provider/src/index.ts

@@ -7,7 +7,9 @@ import {
   TaskOperationsIndexOptions,
   TaskOperationsRestoreOptions,
 } from "@trigger.dev/core-apps";
+import { Machine, PostStartCauses, PreStopCauses, EnvironmentType } from "@trigger.dev/core/v3";
 import { randomUUID } from "crypto";
+import { TaskMonitor } from "./taskMonitor";
 
 const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local";
 const NODE_NAME = process.env.NODE_NAME || "local";
@@ -23,6 +25,10 @@ type Namespace = {
   };
 };
 
+type ComputeResources = {
+  [K in "cpu" | "memory" | "ephemeral-storage"]?: string;
+};
+
 class KubernetesTaskOperations implements TaskOperations {
   #namespace: Namespace;
   #k8sApi: {
@@ -54,16 +60,15 @@ class KubernetesTaskOperations implements TaskOperations {
           template: {
             metadata: {
               labels: {
+                ...this.#getSharedLabels(opts),
                 app: "task-index",
+                "app.kubernetes.io/part-of": "trigger-worker",
+                "app.kubernetes.io/component": "index",
+                deployment: opts.deploymentId,
               },
             },
             spec: {
-              restartPolicy: "Never",
-              imagePullSecrets: [
-                {
-                  name: "registry-trigger",
-                },
-              ],
+              ...this.#defaultPodSpec,
               containers: [
                 {
                   name: this.#getIndexContainerName(opts.shortCode),
@@ -73,17 +78,22 @@ class KubernetesTaskOperations implements TaskOperations {
                       containerPort: 8000,
                     },
                   ],
-                  // resources: {
-                  //   limits: {
-                  //     cpu: "100m",
-                  //     memory: "50Mi",
-                  //   },
-                  // },
-                  env: [
-                    {
-                      name: "DEBUG",
-                      value: "true",
+                  resources: {
+                    limits: {
+                      cpu: "250m",
+                      memory: "0.5G",
+                      "ephemeral-storage": "2Gi",
                     },
+                  },
+                  lifecycle: {
+                    preStop: {
+                      exec: {
+                        command: this.#getLifecycleCommand("preStop", "terminate"),
+                      },
+                    },
+                  },
+                  env: [
+                    ...this.#getSharedEnv(opts.envId),
                     {
                       name: "INDEX_TASKS",
                       value: "true",
@@ -96,42 +106,6 @@ class KubernetesTaskOperations implements TaskOperations {
                       name: "TRIGGER_API_URL",
                       value: opts.apiUrl,
                     },
-                    {
-                      name: "TRIGGER_ENV_ID",
-                      value: opts.envId,
-                    },
-                    {
-                      name: "OTEL_EXPORTER_OTLP_ENDPOINT",
-                      value: OTEL_EXPORTER_OTLP_ENDPOINT,
-                    },
-                    {
-                      name: "HTTP_SERVER_PORT",
-                      value: "8000",
-                    },
-                    {
-                      name: "POD_NAME",
-                      valueFrom: {
-                        fieldRef: {
-                          fieldPath: "metadata.name",
-                        },
-                      },
-                    },
-                    {
-                      name: "COORDINATOR_HOST",
-                      valueFrom: {
-                        fieldRef: {
-                          fieldPath: "status.hostIP",
-                        },
-                      },
-                    },
-                    {
-                      name: "MACHINE_NAME",
-                      valueFrom: {
-                        fieldRef: {
-                          fieldPath: "spec.nodeName",
-                        },
-                      },
-                    },
                   ],
                 },
               ],
@@ -150,16 +124,15 @@ class KubernetesTaskOperations implements TaskOperations {
           name: this.#getRunContainerName(opts.runId),
           namespace: this.#namespace.metadata.name,
           labels: {
+            ...this.#getSharedLabels(opts),
             app: "task-run",
+            "app.kubernetes.io/part-of": "trigger-worker",
+            "app.kubernetes.io/component": "create",
+            run: opts.runId,
           },
         },
         spec: {
-          restartPolicy: "Never",
-          imagePullSecrets: [
-            {
-              name: "registry-trigger",
-            },
-          ],
+          ...this.#defaultPodSpec,
           containers: [
             {
               name: this.#getRunContainerName(opts.runId),
@@ -169,70 +142,28 @@ class KubernetesTaskOperations implements TaskOperations {
                   containerPort: 8000,
                 },
               ],
-              // resources: {
-              //   limits: opts.machine,
-              // },
-              lifecycle: {
-                postStart: {
-                  exec: {
-                    command: this.#getLifecycleCommand("postStart", "create"),
-                  },
+              resources: {
+                requests: {
+                  ...this.#defaultResourceRequests,
                 },
+                limits: {
+                  ...this.#defaultResourceLimits,
+                  ...this.#getResourcesFromMachineConfig(opts.machine),
+                },
+              },
+              lifecycle: {
                 preStop: {
                   exec: {
-                    command: this.#getLifecycleCommand("preStop", "create"),
+                    command: this.#getLifecycleCommand("preStop", "terminate"),
                   },
                 },
               },
               env: [
-                {
-                  name: "DEBUG",
-                  value: "true",
-                },
-                {
-                  name: "HTTP_SERVER_PORT",
-                  value: "8000",
-                },
-                {
-                  name: "TRIGGER_ENV_ID",
-                  value: opts.envId,
-                },
+                ...this.#getSharedEnv(opts.envId),
                 {
                   name: "TRIGGER_RUN_ID",
                   value: opts.runId,
                 },
-                {
-                  name: "TRIGGER_WORKER_VERSION",
-                  value: opts.version,
-                },
-                {
-                  name: "OTEL_EXPORTER_OTLP_ENDPOINT",
-                  value: OTEL_EXPORTER_OTLP_ENDPOINT,
-                },
-                {
-                  name: "POD_NAME",
-                  valueFrom: {
-                    fieldRef: {
-                      fieldPath: "metadata.name",
-                    },
-                  },
-                },
-                {
-                  name: "COORDINATOR_HOST",
-                  valueFrom: {
-                    fieldRef: {
-                      fieldPath: "status.hostIP",
-                    },
-                  },
-                },
-                {
-                  name: "NODE_NAME",
-                  valueFrom: {
-                    fieldRef: {
-                      fieldPath: "spec.nodeName",
-                    },
-                  },
-                },
               ],
               volumeMounts: [
                 {
@@ -261,16 +192,16 @@ class KubernetesTaskOperations implements TaskOperations {
           name: `${this.#getRunContainerName(opts.runId)}-${randomUUID().slice(0, 8)}`,
           namespace: this.#namespace.metadata.name,
           labels: {
+            ...this.#getSharedLabels(opts),
             app: "task-run",
+            "app.kubernetes.io/part-of": "trigger-worker",
+            "app.kubernetes.io/component": "restore",
+            run: opts.runId,
+            checkpoint: opts.checkpointId,
           },
         },
         spec: {
-          restartPolicy: "Never",
-          imagePullSecrets: [
-            {
-              name: "registry-trigger",
-            },
-          ],
+          ...this.#defaultPodSpec,
           initContainers: [
             {
               name: "pull-base-image",
@@ -309,9 +240,15 @@ class KubernetesTaskOperations implements TaskOperations {
                   containerPort: 8000,
                 },
               ],
-              // resources: {
-              //   limits: opts.machine,
-              // },
+              resources: {
+                requests: {
+                  ...this.#defaultResourceRequests,
+                },
+                limits: {
+                  ...this.#defaultResourceLimits,
+                  ...this.#getResourcesFromMachineConfig(opts.machine),
+                },
+              },
               lifecycle: {
                 postStart: {
                   exec: {
@@ -320,7 +257,7 @@ class KubernetesTaskOperations implements TaskOperations {
                 },
                 preStop: {
                   exec: {
-                    command: this.#getLifecycleCommand("preStop", "restore"),
+                    command: this.#getLifecycleCommand("preStop", "terminate"),
                   },
                 },
               },
@@ -355,8 +292,126 @@ class KubernetesTaskOperations implements TaskOperations {
     await this.#getPod(opts.runId, this.#namespace);
   }
 
-  #getLifecycleCommand(type: "postStart" | "preStop", cause: "index" | "create" | "restore") {
-    return ["/bin/sh", "-c", `sleep 1; wget -q -O- 127.0.0.1:8000/${type}?cause=${cause}`];
+  #envTypeToLabelValue(type: EnvironmentType) {
+    switch (type) {
+      case "PRODUCTION":
+        return "prod";
+      case "STAGING":
+        return "stg";
+      case "DEVELOPMENT":
+        return "dev";
+      case "PREVIEW":
+        return "preview";
+    }
+  }
+
+  get #defaultPodSpec(): Omit<k8s.V1PodSpec, "containers"> {
+    return {
+      restartPolicy: "Never",
+      automountServiceAccountToken: false,
+      imagePullSecrets: [
+        {
+          name: "registry-trigger",
+        },
+      ],
+      nodeSelector: {
+        nodetype: "worker",
+      },
+    };
+  }
+
+  get #defaultResourceRequests(): ComputeResources {
+    return {
+      "ephemeral-storage": "2Gi",
+    };
+  }
+
+  get #defaultResourceLimits(): ComputeResources {
+    return {
+      "ephemeral-storage": "10Gi",
+    };
+  }
+
+  #getSharedEnv(envId: string): k8s.V1EnvVar[] {
+    return [
+      {
+        name: "TRIGGER_ENV_ID",
+        value: envId,
+      },
+      {
+        name: "DEBUG",
+        value: process.env.DEBUG ? "1" : "0",
+      },
+      {
+        name: "HTTP_SERVER_PORT",
+        value: "8000",
+      },
+      {
+        name: "OTEL_EXPORTER_OTLP_ENDPOINT",
+        value: OTEL_EXPORTER_OTLP_ENDPOINT,
+      },
+      {
+        name: "POD_NAME",
+        valueFrom: {
+          fieldRef: {
+            fieldPath: "metadata.name",
+          },
+        },
+      },
+      {
+        name: "COORDINATOR_HOST",
+        valueFrom: {
+          fieldRef: {
+            fieldPath: "status.hostIP",
+          },
+        },
+      },
+      {
+        name: "MACHINE_NAME",
+        valueFrom: {
+          fieldRef: {
+            fieldPath: "spec.nodeName",
+          },
+        },
+      },
+    ];
+  }
+
+  #getSharedLabels(
+    opts: TaskOperationsIndexOptions | TaskOperationsCreateOptions | TaskOperationsRestoreOptions
+  ): Record<string, string> {
+    return {
+      env: opts.envId,
+      envtype: this.#envTypeToLabelValue(opts.envType),
+      org: opts.orgId,
+      project: opts.projectId,
+    };
+  }
+
+  #getResourcesFromMachineConfig(config: Machine): ComputeResources {
+    return {
+      cpu: `${config.cpu}`,
+      memory: `${config.memory}G`,
+    };
+  }
+
+  #getLifecycleCommand<THookType extends "postStart" | "preStop">(
+    type: THookType,
+    cause: THookType extends "postStart" ? PostStartCauses : PreStopCauses
+  ) {
+    const retries = 5;
+
+    // This will retry sending the lifecycle hook up to `retries` times
+    // The sleep is required as this may start running before the HTTP server is up
+    const exec = [
+      "/bin/sh",
+      "-c",
+      `for i in $(seq ${retries}); do sleep 1; busybox wget -q -O- 127.0.0.1:8000/${type}?cause=${cause} && break; done`,
+    ];
+
+    logger.log("getLifecycleCommand()", { exec });
+
+    return exec;
   }
 
   #getIndexContainerName(suffix: string) {
@@ -455,3 +510,34 @@ const provider = new ProviderShell({
 });
 
 provider.listen();
+
+const taskMonitor = new TaskMonitor({
+  runtimeEnv: RUNTIME_ENV,
+  onIndexFailure: async (deploymentId, failureInfo) => {
+    logger.log("Indexing failed", { deploymentId, failureInfo });
+
+    try {
+      provider.platformSocket.send("INDEXING_FAILED", {
+        deploymentId,
+        error: {
+          name: `Crashed with exit code ${failureInfo.exitCode}`,
+          message: failureInfo.reason,
+          stack: failureInfo.logs,
+        },
+      });
+    } catch (error) {
+      logger.error(error);
+    }
+  },
+  onRunFailure: async (runId, failureInfo) => {
+    logger.log("Run failed:", { runId, failureInfo });
+
+    try {
+      provider.platformSocket.send("WORKER_CRASHED", { runId, ...failureInfo });
+    } catch (error) {
+      logger.error(error);
+    }
+  },
+});
+
+taskMonitor.start();

apps/kubernetes-provider/src/taskMonitor.ts

@@ -0,0 +1,436 @@
+import * as k8s from "@kubernetes/client-node";
+import { SimpleLogger } from "@trigger.dev/core-apps";
+import { setTimeout } from "timers/promises";
+import PQueue from "p-queue";
+
+type IndexFailureHandler = (
+  deploymentId: string,
+  failureInfo: {
+    exitCode: number;
+    reason: string;
+    logs: string;
+  }
+) => Promise<any>;
+
+type RunFailureHandler = (
+  runId: string,
+  failureInfo: {
+    exitCode: number;
+    reason: string;
+    logs: string;
+  }
+) => Promise<any>;
+
+type TaskMonitorOptions = {
+  runtimeEnv: "local" | "kubernetes";
+  onIndexFailure?: IndexFailureHandler;
+  onRunFailure?: RunFailureHandler;
+  namespace?: string;
+};
+
+export class TaskMonitor {
+  #enabled = false;
+  #logger = new SimpleLogger("[TaskMonitor]");
+  #taskInformer: ReturnType<typeof k8s.makeInformer<k8s.V1Pod>>;
+  #processedPods = new Map<string, number>();
+  #queue = new PQueue({ concurrency: 10 });
+  #k8sClient: {
+    core: k8s.CoreV1Api;
+    kubeConfig: k8s.KubeConfig;
+  };
+
+  private namespace = "default";
+  private fieldSelector = "status.phase=Failed";
+  private labelSelector = "app in (task-index, task-run)";
+
+  constructor(private opts: TaskMonitorOptions) {
+    this.#k8sClient = this.#createK8sClient();
+
+    this.#taskInformer = this.#createTaskInformer();
+    this.#taskInformer.on("connect", this.#onInformerConnected.bind(this));
+    this.#taskInformer.on("error", this.#onInformerError.bind(this));
+    this.#taskInformer.on("update", this.#enqueueOnPodUpdated.bind(this));
+  }
+
+  #createTaskInformer() {
+    const listTasks = () =>
+      this.#k8sClient.core.listNamespacedPod(
+        this.namespace,
+        undefined,
+        undefined,
+        undefined,
+        this.fieldSelector,
+        this.labelSelector
+      );
+
+    // Uses watch with local caching
+    // https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
+    const informer = k8s.makeInformer(
+      this.#k8sClient.kubeConfig,
+      `/api/v1/namespaces/${this.namespace}/pods`,
+      listTasks,
+      this.labelSelector,
+      this.fieldSelector
+    );
+
+    return informer;
+  }
+
+  async #onInformerConnected() {
+    this.#logger.log("Connected");
+  }
+
+  async #onInformerError(error: any) {
+    this.#logger.error("Error:", error);
+
+    // Automatic reconnect
+    await setTimeout(2_000);
+    this.#taskInformer.start();
+  }
+
+  #enqueueOnPodUpdated(pod: k8s.V1Pod) {
+    this.#queue.add(async () => {
+      try {
+        // It would be better to only pass the cache key, but the pod may already be removed from the cache by the time we process it
+        await this.#onPodUpdated(pod);
+      } catch (error) {
+        this.#logger.error("Caught onPodUpdated() error:", error);
+      }
+    });
+  }
+
+  async #onPodUpdated(pod: k8s.V1Pod) {
+    this.#logger.debug(`Updated: ${pod.metadata?.name}`);
+    this.#logger.debug("Updated", JSON.stringify(pod, null, 2));
+
+    // We only care about failures
+    if (pod.status?.phase !== "Failed") {
+      return;
+    }
+
+    const podName = pod.metadata?.name;
+
+    if (!podName) {
+      this.#logger.error("Pod is nameless", { pod });
+      return;
+    }
+
+    const containerStatus = pod.status.containerStatuses?.[0];
+
+    if (!containerStatus?.state) {
+      this.#logger.error("Pod failed, but container status doesn't have state", {
+        status: pod.status,
+      });
+      return;
+    }
+
+    if (this.#processedPods.has(podName)) {
+      this.#logger.debug("Pod update already processed", {
+        podName,
+        timestamp: this.#processedPods.get(podName),
+      });
+      return;
+    }
+
+    this.#processedPods.set(podName, Date.now());
+
+    const podStatus = this.#getPodStatusSummary(pod.status);
+    const containerState = this.#getContainerStateSummary(containerStatus.state);
+    const rawLogs = await this.#getLogTail(podName);
+
+    this.#logger.log(`${podName} failed with:`, {
+      podStatus,
+      containerState,
+      rawLogs,
+    });
+
+    const exitCode = containerState.exitCode ?? -1;
+    const rawReason = podStatus.reason ?? containerState.reason ?? "";
+    const message = podStatus.message ?? containerState.message ?? "";
+
+    let reason = rawReason || "Unknown error";
+    let logs = rawLogs || "";
+
+    switch (rawReason) {
+      case "Error":
+        reason = "Unknown error.";
+        break;
+      case "Evicted":
+        if (message.startsWith("Pod ephemeral local storage usage")) {
+          reason = "Storage limit exceeded.";
+        } else if (message) {
+          reason = `Evicted: ${message}`;
+        } else {
+          reason = "Evicted for unknown reason.";
+        }
+
+        if (logs.startsWith("failed to try resolving symlinks")) {
+          logs = "";
+        }
+        break;
+      case "OOMKilled":
+        reason = "Out of memory! Try increasing the memory on this task.";
+        break;
+      default:
+        break;
+    }
+
+    const failureInfo = {
+      exitCode,
+      reason,
+      logs,
+    };
+
+    const app = pod.metadata?.labels?.app;
+
+    switch (app) {
+      case "task-index":
+        const deploymentId = pod.metadata?.labels?.deployment;
+
+        if (!deploymentId) {
+          this.#logger.error("Index is missing ID", { pod });
+          return;
+        }
+
+        if (this.opts.onIndexFailure) {
+          await this.opts.onIndexFailure(deploymentId, failureInfo);
+        }
+        break;
+      case "task-run":
+        const runId = pod.metadata?.labels?.run;
+
+        if (!runId) {
+          this.#logger.error("Run is missing ID", { pod });
+          return;
+        }
+
+        if (this.opts.onRunFailure) {
+          await this.opts.onRunFailure(runId, failureInfo);
+        }
+        break;
+      default:
+        this.#logger.error("Pod has invalid app label", { pod });
+        return;
+    }
+
+    await this.#deletePod(podName);
+  }
+
+  async #getLogTail(podName: string) {
+    try {
+      const logs = await this.#k8sClient.core.readNamespacedPodLog(
+        podName,
+        this.namespace,
+        undefined,
+        undefined,
+        undefined,
+        1024, // limitBytes
+        undefined,
+        undefined,
+        undefined,
+        20 // tailLines
+      );
+
+      const responseBody = logs.body ?? "";
+
+      if (responseBody.startsWith("unable to retrieve container logs")) {
+        return "";
+      }
+
+      // Type is wrong, body may be undefined
+      return responseBody;
+    } catch (error) {
+      this.#logger.error("Log tail error:", error instanceof Error ? error.message : "unknown");
+      return "";
+    }
+  }
+
+  #getPodStatusSummary(status: k8s.V1PodStatus) {
+    return {
+      reason: status.reason,
+      message: status.message,
+    };
+  }
+
+  #getContainerStateSummary(state: k8s.V1ContainerState) {
+    return {
+      reason: state.terminated?.reason,
+      exitCode: state.terminated?.exitCode,
+      message: state.terminated?.message,
+    };
+  }
+
+  #createK8sClient() {
+    const kubeConfig = new k8s.KubeConfig();
+
+    if (this.opts.runtimeEnv === "local") {
+      kubeConfig.loadFromDefault();
+    } else if (this.opts.runtimeEnv === "kubernetes") {
+      kubeConfig.loadFromCluster();
+    } else {
+      throw new Error(`Unsupported runtime environment: ${this.opts.runtimeEnv}`);
+    }
+
+    return {
+      core: kubeConfig.makeApiClient(k8s.CoreV1Api),
+      kubeConfig: kubeConfig,
+    };
+  }
+
+  #isRecord(candidate: unknown): candidate is Record<string, unknown> {
+    if (typeof candidate !== "object" || candidate === null) {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+  #logK8sError(err: unknown, debugOnly = false) {
+    if (debugOnly) {
+      this.#logger.debug("K8s API Error", err);
+    } else {
+      this.#logger.error("K8s API Error", err);
+    }
+  }
+
+  #handleK8sError(err: unknown) {
+    if (!this.#isRecord(err) || !this.#isRecord(err.body)) {
+      this.#logK8sError(err);
+      return;
+    }
+
+    this.#logK8sError(err, true);
+
+    if (typeof err.body.message === "string") {
+      this.#logK8sError({ message: err.body.message });
+      return;
+    }
+
+    this.#logK8sError({ body: err.body });
+  }
+
+  #printStats(includeMoreDetails = false) {
+    this.#logger.log("Stats:", {
+      cacheSize: this.#taskInformer.list().length,
+      totalProcessed: this.#processedPods.size,
+      ...(includeMoreDetails && {
+        processedPods: this.#processedPods,
+      }),
+    });
+  }
+
+  async #deletePod(name: string) {
+    this.#logger.debug("Deleting pod:", name);
+
+    await this.#k8sClient.core
+      .deleteNamespacedPod(name, this.namespace)
+      .catch(this.#handleK8sError.bind(this));
+  }
+
+  async start() {
+    this.#enabled = true;
+
+    const interval = setInterval(() => {
+      if (!this.#enabled) {
+        clearInterval(interval);
+        return;
+      }
+
+      this.#printStats();
+    }, 300_000);
+
+    await this.#taskInformer.start();
+
+    // this.#launchTests();
+  }
+
+  async stop() {
+    if (!this.#enabled) {
+      return;
+    }
+
+    this.#enabled = false;
+    this.#logger.log("Shutting down..");
+
+    await this.#taskInformer.stop();
+
+    this.#printStats(true);
+  }
+
+  async #launchTests() {
+    const createPod = async (
+      container: k8s.V1Container,
+      name: string,
+      labels?: Record<string, string>
+    ) => {
+      this.#logger.log("Creating pod:", name);
+
+      const pod = {
+        metadata: {
+          name,
+          labels,
+        },
+        spec: {
+          restartPolicy: "Never",
+          automountServiceAccountToken: false,
+          terminationGracePeriodSeconds: 1,
+          containers: [container],
+        },
+      } satisfies k8s.V1Pod;
+
+      await this.#k8sClient.core
+        .createNamespacedPod(this.namespace, pod)
+        .catch(this.#handleK8sError.bind(this));
+    };
+
+    const createOomPod = async (name: string, labels?: Record<string, string>) => {
+      const container = {
+        name,
+        image: "polinux/stress",
+        resources: {
+          limits: {
+            memory: "100Mi",
+          },
+        },
+        command: ["stress"],
+        args: ["--vm", "1", "--vm-bytes", "150M", "--vm-hang", "1"],
+      } satisfies k8s.V1Container;
+
+      await createPod(container, name, labels);
+    };
+
+    const createNonZeroExitPod = async (name: string, labels?: Record<string, string>) => {
+      const container = {
+        name,
+        image: "busybox",
+        command: ["sh"],
+        args: ["-c", "exit 1"],
+      } satisfies k8s.V1Container;
+
+      await createPod(container, name, labels);
+    };
+
+    const createOoDiskPod = async (name: string, labels?: Record<string, string>) => {
+      const container = {
+        name,
+        image: "busybox",
+        command: ["sh"],
+        args: [
+          "-c",
+          "echo creating huge-file..; head -c 1000m /dev/zero > huge-file; ls -lh huge-file; sleep infinity",
+        ],
+        resources: {
+          limits: {
+            "ephemeral-storage": "500Mi",
+          },
+        },
+      } satisfies k8s.V1Container;
+
+      await createPod(container, name, labels);
+    };
+
+    await createNonZeroExitPod("non-zero-exit-task", { app: "task-run", run: "123" });
+    await createOomPod("oom-task", { app: "task-index", deployment: "456" });
+    await createOoDiskPod("ood-task", { app: "task-run", run: "abc" });
+  }
+}

apps/webapp/app/components/navigation/SideMenu.tsx

@@ -57,6 +57,7 @@ import { UserProfilePhoto } from "../UserProfilePhoto";
 import { FreePlanUsage } from "../billing/FreePlanUsage";
 import { Badge } from "../primitives/Badge";
 import { Button } from "../primitives/Buttons";
+import { Callout } from "../primitives/Callout";
 import { ClipboardField } from "../primitives/ClipboardField";
 import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "../primitives/Dialog";
 import { Icon } from "../primitives/Icon";
@@ -187,7 +188,20 @@ export function SideMenu({ user, project, organization, organizations }: SideMen
             />
           </div>
         </div>
+        <div className="m-2">
+          <Callout variant={"info"}>
+            {`This is a ${project.version === "V2" ? "v2" : "v3"} project`}
+          </Callout>
+        </div>
         <div className="flex flex-col gap-1 border-t border-grid-bright p-1">
+          {project.version === "V2" && (
+            <SideMenuItem
+              to="https://trigger.dev/v3-early-access"
+              target="_blank"
+              name="Request access to v3"
+              icon={V3Icon}
+            />
+          )}
           {currentPlan?.subscription?.isPaying === true && (
             <Dialog>
               <DialogTrigger asChild>
@@ -264,7 +278,7 @@ export function SideMenu({ user, project, organization, organizations }: SideMen
             <SideMenuItem
               name="Documentation (v3)"
               icon="docs"
-              to="https://trigger.dev/docs"
+              to="https://trigger.dev/docs/v3"
               data-action="documentation"
               target="_blank"
             />
@@ -556,7 +570,7 @@ function V3ProjectSideMenu({
         data-action="test"
       />
       <SideMenuItem
-        name="API Keys"
+        name="API keys"
         icon={KeyIcon}
         iconColor="text-amber-500"
         to={v3ApiKeysPath(organization, project)}
@@ -586,3 +600,15 @@ function V3ProjectSideMenu({
     </>
   );
 }
+
+function V3Icon() {
+  return (
+    <svg width="16" height="16" viewBox="0 0 16 16" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect width="16" height="16" rx="8" fill="#A8FF53" />
+      <path
+        d="M7.7488 6.17L5.5818 12H3.6678L1.5008 6.17H3.2058L4.6248 10.339L6.0328 6.17H7.7488ZM11.0527 7.292C11.6357 7.303 12.2297 7.006 12.2297 6.28C12.2297 5.774 11.7787 5.433 11.0527 5.433C10.4147 5.433 9.98567 5.741 9.91967 6.214L8.22567 6.126C8.36867 4.861 9.51267 4.014 11.0857 4.014C12.8457 4.014 13.9567 4.806 13.9567 6.049C13.9567 6.951 13.3847 7.534 12.3067 7.776C13.5387 8.04 14.2207 8.777 14.2207 9.855C14.2207 11.274 13.0107 12.176 11.0857 12.176C9.32567 12.176 8.12667 11.197 8.04967 9.712L9.75467 9.646C9.83167 10.405 10.4917 10.757 11.0967 10.757C11.8007 10.757 12.4937 10.394 12.4937 9.591C12.4937 8.81 11.7897 8.425 11.0527 8.447L10.3817 8.458V7.281L11.0527 7.292Z"
+        fill="#15171A"
+      />
+    </svg>
+  );
+}

apps/webapp/app/components/primitives/Dialog.tsx

@@ -43,7 +43,7 @@ const DialogContent = React.forwardRef<
     <DialogPrimitive.Content
       ref={ref}
       className={cn(
-        "fixed z-50 grid w-full gap-4 rounded-b-lg border bg-background-dimmed px-4 pb-6 pt-3.5 shadow-lg animate-in data-[state=open]:fade-in-90 data-[state=open]:slide-in-from-bottom-10 sm:max-w-lg sm:rounded-lg sm:zoom-in-90 data-[state=open]:sm:slide-in-from-bottom-0",
+        "fixed z-50 grid w-full gap-4 rounded-b-lg border bg-background-dimmed px-4 pb-4 pt-3.5 shadow-lg animate-in data-[state=open]:fade-in-90 data-[state=open]:slide-in-from-bottom-10 sm:max-w-lg sm:rounded-lg sm:zoom-in-90 data-[state=open]:sm:slide-in-from-bottom-0",
         className
       )}
       {...props}

apps/webapp/app/components/primitives/PageHeader.tsx

@@ -15,6 +15,7 @@ import { Tabs, TabsProps } from "./Tabs";
 
 type WithChildren = {
   children: React.ReactNode;
+  className?: string;
 };
 
 export function NavBar({ children }: WithChildren) {
@@ -72,8 +73,8 @@ export function PageAccessories({ children }: WithChildren) {
   return <div className="flex items-center gap-3">{children}</div>;
 }
 
-export function PageInfoRow({ children }: WithChildren) {
-  return <div className="flex w-full items-center gap-2">{children}</div>;
+export function PageInfoRow({ children, className }: WithChildren) {
+  return <div className={cn("flex w-full items-center gap-2", className)}>{children}</div>;
 }
 
 export function PageInfoGroup({

apps/webapp/app/components/run/RunOverview.tsx

@@ -124,9 +124,9 @@ export function RunOverview({ run, trigger, showRerun, paths, currentUser }: Run
           {!run.isFinished && <CancelRun runId={run.id} />}
         </PageAccessories>
       </NavBar>
-      <PageBody scrollable={false}>
-        <div className="px-4 pt-4">
-          <PageInfoRow>
+      <PageBody scrollable={false} className="grid grid-rows-[auto_1fr] overflow-hidden">
+        <div className="border-b border-grid-dimmed px-4 py-4">
+          <PageInfoRow className="overflow-hidden">
             <PageInfoGroup>
               <PageInfoProperty
                 icon={<RunStatusIcon status={run.status} className="h-4 w-4" />}
@@ -166,7 +166,7 @@ export function RunOverview({ run, trigger, showRerun, paths, currentUser }: Run
             </PageInfoGroup>
           </PageInfoRow>
         </div>
-        <div className="grid h-full grid-cols-2 gap-2">
+        <div className="grid h-full grid-cols-2 gap-2 overflow-hidden">
           <div className="flex flex-col gap-6 overflow-y-auto py-4 pl-4 pr-2 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
             <div>
               {run.status === "SUCCESS" &&

apps/webapp/app/components/runs/v3/CancelRunDialog.tsx

@@ -1,5 +1,5 @@
 import { StopCircleIcon } from "@heroicons/react/20/solid";
-import { useFetcher } from "@remix-run/react";
+import { Form, useFetcher, useNavigation } from "@remix-run/react";
 import { Button } from "~/components/primitives/Buttons";
 import {
   DialogContent,
@@ -14,29 +14,32 @@ type CancelRunDialogProps = {
 };
 
 export function CancelRunDialog({ runFriendlyId, redirectPath }: CancelRunDialogProps) {
-  const cancelFetcher = useFetcher();
+  const navigation = useNavigation();
+
+  const formAction = `/resources/taskruns/${runFriendlyId}/cancel`;
+  const isLoading = navigation.formAction === formAction;
 
   return (
-    <DialogContent>
+    <DialogContent key="cancel">
       <DialogHeader>Cancel this run?</DialogHeader>
       <DialogDescription>
         Canceling a run will stop execution. If you want to run this later you will have to replay
         the entire run with the original payload.
       </DialogDescription>
       <DialogFooter>
-        <cancelFetcher.Form action={`/resources/taskruns/${runFriendlyId}/cancel`} method="post">
+        <Form action={`/resources/taskruns/${runFriendlyId}/cancel`} method="post">
           <Button
             type="submit"
             name="redirectUrl"
             value={redirectPath}
             variant="danger/small"
-            LeadingIcon={cancelFetcher.state === "idle" ? StopCircleIcon : "spinner-white"}
-            disabled={cancelFetcher.state !== "idle"}
+            LeadingIcon={isLoading ? "spinner-white" : StopCircleIcon}
+            disabled={isLoading}
             shortcut={{ modifiers: ["meta"], key: "enter" }}
           >
-            {cancelFetcher.state === "idle" ? "Cancel run" : "Canceling..."}
+            {isLoading ? "Canceling..." : "Cancel run"}
           </Button>
-        </cancelFetcher.Form>
+        </Form>
       </DialogFooter>
     </DialogContent>
   );

apps/webapp/app/components/runs/v3/ReplayRunDialog.tsx

@@ -0,0 +1,44 @@
+import { ArrowPathIcon } from "@heroicons/react/20/solid";
+import { Form, useFetcher, useNavigation } from "@remix-run/react";
+import { Button } from "~/components/primitives/Buttons";
+import {
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+} from "~/components/primitives/Dialog";
+
+type ReplayRunDialogProps = {
+  runFriendlyId: string;
+  failedRedirect: string;
+};
+
+export function ReplayRunDialog({ runFriendlyId, failedRedirect }: ReplayRunDialogProps) {
+  const navigation = useNavigation();
+
+  const formAction = `/resources/taskruns/${runFriendlyId}/replay`;
+  const isLoading = navigation.formAction === formAction;
+
+  return (
+    <DialogContent key="replay">
+      <DialogHeader>Replay this run?</DialogHeader>
+      <DialogDescription>
+        Replaying a run will create a new run with the same payload and environment as the original.
+      </DialogDescription>
+      <DialogFooter>
+        <Form action={formAction} method="post">
+          <input type="hidden" name="failedRedirect" value={failedRedirect} />
+          <Button
+            type="submit"
+            variant="primary/small"
+            LeadingIcon={isLoading ? "spinner-white" : ArrowPathIcon}
+            disabled={isLoading}
+            shortcut={{ modifiers: ["meta"], key: "enter" }}
+          >
+            {isLoading ? "Replaying..." : "Replay run"}
+          </Button>
+        </Form>
+      </DialogFooter>
+    </DialogContent>
+  );
+}

apps/webapp/app/components/runs/v3/RunFilters.tsx

@@ -35,6 +35,7 @@ export const allTaskRunStatuses = [
   "COMPLETED_WITH_ERRORS",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "CRASHED",
 ] as TaskRunStatusType[];
 
 export const TaskAttemptStatus = z.nativeEnum(TaskRunStatus);

apps/webapp/app/components/runs/v3/SpanEvents.tsx

@@ -66,7 +66,11 @@ function SpanEventError({
 }) {
   return (
     <div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 p-3">
-      <SpanEventHeader title={"Error"} time={spanEvent.time} titleClassName="text-rose-500" />
+      <SpanEventHeader
+        title={exception.type ?? "Error"}
+        time={spanEvent.time}
+        titleClassName="text-rose-500"
+      />
       {exception.message && <Callout variant="error">{exception.message}</Callout>}
       {exception.stacktrace && (
         <CodeBlock

apps/webapp/app/components/runs/v3/TaskRunStatus.tsx

@@ -3,7 +3,7 @@ import {
   BoltSlashIcon,
   BugAntIcon,
   CheckCircleIcon,
-  ClockIcon,
+  FireIcon,
   NoSymbolIcon,
   PauseCircleIcon,
   RectangleStackIcon,
@@ -25,6 +25,7 @@ const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
   INTERRUPTED: "Task has failed because it was interrupted",
   SYSTEM_FAILURE: "Task has failed due to a system failure",
   PAUSED: "Task has been paused by the user",
+  CRASHED: "Task has crashed and won't be retried",
 };
 
 export function descriptionForTaskRunStatus(status: TaskRunStatus): string {
@@ -80,6 +81,8 @@ export function TaskRunStatusIcon({
       return <XCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "SYSTEM_FAILURE":
       return <BugAntIcon className={cn(runStatusClassNameColor(status), className)} />;
+    case "CRASHED":
+      return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
 
     default: {
       const _exhaustiveCheck: never = status;
@@ -109,6 +112,8 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
       return "text-error";
     case "SYSTEM_FAILURE":
       return "text-error";
+    case "CRASHED":
+      return "text-error";
     default: {
       const _exhaustiveCheck: never = status;
       throw new Error(`Non-exhaustive match for value: ${status}`);
@@ -138,6 +143,8 @@ export function runStatusTitle(status: TaskRunStatus): string {
       return "Failed";
     case "SYSTEM_FAILURE":
       return "System failure";
+    case "CRASHED":
+      return "Crashed";
     default: {
       const _exhaustiveCheck: never = status;
       throw new Error(`Non-exhaustive match for value: ${status}`);

apps/webapp/app/components/runs/v3/TaskRunsTable.tsx

@@ -24,10 +24,11 @@ import { formatDuration } from "@trigger.dev/core/v3";
 import { TaskRunStatusCombo } from "./TaskRunStatus";
 import { useEnvironments } from "~/hooks/useEnvironments";
 import { Button, LinkButton } from "~/components/primitives/Buttons";
-import { StopCircleIcon } from "@heroicons/react/20/solid";
+import { ArrowPathIcon, StopCircleIcon } from "@heroicons/react/20/solid";
 import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
 import { CancelRunDialog } from "./CancelRunDialog";
 import { useLocation } from "@remix-run/react";
+import { ReplayRunDialog } from "./ReplayRunDialog";
 
 type RunsTableProps = {
   total: number;
@@ -49,7 +50,6 @@ export function TaskRunsTable({
 }: RunsTableProps) {
   const organization = useOrganization();
   const project = useProject();
-  const location = useLocation();
 
   return (
     <Table>
@@ -110,23 +110,7 @@ export function TaskRunsTable({
                 <TableCell to={path}>
                   {run.createdAt ? <DateTime date={run.createdAt} /> : "–"}
                 </TableCell>
-                {run.isCancellable ? (
-                  <TableCellMenu isSticky>
-                    <Dialog>
-                      <DialogTrigger asChild>
-                        <Button variant="small-menu-item" LeadingIcon={StopCircleIcon}>
-                          Cancel run
-                        </Button>
-                      </DialogTrigger>
-                      <CancelRunDialog
-                        runFriendlyId={run.friendlyId}
-                        redirectPath={`${location.pathname}${location.search}`}
-                      />
-                    </Dialog>
-                  </TableCellMenu>
-                ) : (
-                  <TableCell to={path}>{""}</TableCell>
-                )}
+                <RunActionsCell run={run} path={path} />
               </TableRow>
             );
           })
@@ -144,6 +128,43 @@ export function TaskRunsTable({
   );
 }
 
+function RunActionsCell({ run, path }: { run: RunListItem; path: string }) {
+  const location = useLocation();
+
+  if (!run.isCancellable && !run.isReplayable) return <TableCell to={path}>{""}</TableCell>;
+
+  return (
+    <TableCellMenu isSticky>
+      {run.isCancellable && (
+        <Dialog>
+          <DialogTrigger asChild>
+            <Button variant="small-menu-item" LeadingIcon={StopCircleIcon}>
+              Cancel run
+            </Button>
+          </DialogTrigger>
+          <CancelRunDialog
+            runFriendlyId={run.friendlyId}
+            redirectPath={`${location.pathname}${location.search}`}
+          />
+        </Dialog>
+      )}
+      {run.isReplayable && (
+        <Dialog>
+          <DialogTrigger asChild>
+            <Button variant="small-menu-item" LeadingIcon={ArrowPathIcon}>
+              Replay run
+            </Button>
+          </DialogTrigger>
+          <ReplayRunDialog
+            runFriendlyId={run.friendlyId}
+            failedRedirect={`${location.pathname}${location.search}`}
+          />
+        </Dialog>
+      )}
+    </TableCellMenu>
+  );
+}
+
 function NoRuns({ title }: { title: string }) {
   return (
     <div className="flex items-center justify-center">

apps/webapp/app/entry.server.tsx

@@ -14,9 +14,9 @@ import {
   OperatingSystemContextProvider,
   OperatingSystemPlatform,
 } from "./components/primitives/OperatingSystemProvider";
-import { env } from "./env.server";
 import { getSharedSqsEventConsumer } from "./services/events/sqsEventConsumer";
 import { singleton } from "./utils/singleton";
+import { logger } from "./services/logger.server";
 
 const ABORT_DELAY = 30000;
 
@@ -178,7 +178,7 @@ function logError(error: unknown, request?: Request) {
 
 const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);
 
-export { wss } from "./v3/handleWebsockets.server";
-export { socketIo } from "./v3/handleSocketIo.server";
-export { registryProxy } from "./v3/registryProxy.server";
 export { apiRateLimiter } from "./services/apiRateLimit.server";
+export { socketIo } from "./v3/handleSocketIo.server";
+export { wss } from "./v3/handleWebsockets.server";
+export { registryProxy } from "./v3/registryProxy.server";

apps/webapp/app/env.server.ts

@@ -19,6 +19,7 @@ const EnvironmentSchema = z.object({
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ENV: z.string().default(process.env.NODE_ENV),
+  SERVICE_NAME: z.string().default("trigger.dev webapp"),
   SECRET_STORE: SecretStoreOptionsSchema.default("DATABASE"),
   POSTHOG_PROJECT_KEY: z.string().optional(),
   TELEMETRY_TRIGGER_API_KEY: z.string().optional(),
@@ -63,6 +64,8 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
+  DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
+  DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
 
@@ -85,10 +88,6 @@ const EnvironmentSchema = z.object({
 
   //v3
   V3_ENABLED: z.string().default("false"),
-  OTLP_EXPORTER_TRACES_URL: z.string().optional(),
-  LOG_TELEMETRY: z.string().default("true"),
-  IMAGE_REGISTRY: z.string().default("docker.io"),
-  IMAGE_REPO: z.string().default("task"),
   PROVIDER_SECRET: z.string().default("provider-secret"),
   COORDINATOR_SECRET: z.string().default("coordinator-secret"),
   DEPOT_TOKEN: z.string().optional(),
@@ -97,13 +96,49 @@ const EnvironmentSchema = z.object({
   CONTAINER_REGISTRY_USERNAME: z.string().optional(),
   CONTAINER_REGISTRY_PASSWORD: z.string().optional(),
   DEPLOY_REGISTRY_HOST: z.string().optional(),
-  DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
   OBJECT_STORE_BASE_URL: z.string().optional(),
   OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
   OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
   EVENTS_BATCH_SIZE: z.coerce.number().int().default(100),
   EVENTS_BATCH_INTERVAL: z.coerce.number().int().default(1000),
   EVENTS_DEFAULT_LOG_RETENTION: z.coerce.number().int().default(7),
+  SHARED_QUEUE_CONSUMER_POOL_SIZE: z.coerce.number().int().default(10),
+  SHARED_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(100),
+  SHARED_QUEUE_CONSUMER_NEXT_TICK_INTERVAL_MS: z.coerce.number().int().default(100),
+
+  // Development OTEL environment variables
+  DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
+  // If this is set to 1, then the below variables are used to configure the batch processor for spans and logs
+  DEV_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
+  DEV_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  DEV_OTEL_SPAN_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  DEV_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  DEV_OTEL_SPAN_MAX_QUEUE_SIZE: z.string().default("512"),
+  DEV_OTEL_LOG_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  DEV_OTEL_LOG_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  DEV_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
+
+  PROD_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
+  PROD_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  PROD_OTEL_SPAN_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  PROD_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  PROD_OTEL_SPAN_MAX_QUEUE_SIZE: z.string().default("512"),
+  PROD_OTEL_LOG_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  PROD_OTEL_LOG_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  PROD_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
+
+  RUNTIME_WAIT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000),
+
+  // Internal OTEL environment variables
+  INTERNAL_OTEL_TRACE_EXPORTER_URL: z.string().optional(),
+  INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_NAME: z.string().optional(),
+  INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_VALUE: z.string().optional(),
+  INTERNAL_OTEL_TRACE_LOGGING_ENABLED: z.string().default("1"),
+  // this means 1/20 traces or 5% of traces will be sampled (sampled = recorded)
+  INTERNAL_OTEL_TRACE_SAMPLING_RATE: z.string().default("20"),
+  INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED: z.string().default("0"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/models/organization.server.ts

@@ -11,6 +11,7 @@ import { prisma, PrismaClientOrTransaction } from "~/db.server";
 import { createProject } from "./project.server";
 import { generate } from "random-words";
 import { createApiKeyForEnv, createPkApiKeyForEnv, envSlug } from "./api-key.server";
+import { env } from "~/env.server";
 
 export type { Organization };
 
@@ -62,6 +63,7 @@ export async function createOrganization(
       title,
       slug: uniqueOrgSlug,
       companySize,
+      maximumConcurrencyLimit: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
       members: {
         create: {
           userId: userId,
@@ -103,6 +105,7 @@ export async function createEnvironment(
       pkApiKey,
       shortcode,
       autoEnableInternalSources: type !== "DEVELOPMENT",
+      maximumConcurrencyLimit: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
       organization: {
         connect: {
           id: organization.id,

apps/webapp/app/presenters/v3/DeploymentPresenter.server.ts

@@ -1,8 +1,15 @@
+import {
+  DeploymentErrorData,
+  TaskMetadataFailedToParseData,
+  groupTaskMetadataIssuesByTask,
+} from "@trigger.dev/core/v3";
 import { WorkerDeployment, WorkerDeploymentStatus } from "@trigger.dev/database";
+import { z } from "zod";
 import { PrismaClient, prisma } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
+import { safeJsonParse } from "~/utils/json";
 import { getUsername } from "~/utils/username";
 
 export class DeploymentPresenter {
@@ -51,6 +58,7 @@ export class DeploymentPresenter {
         id: true,
         shortCode: true,
         version: true,
+        errorData: true,
         environment: {
           select: {
             id: true,
@@ -120,7 +128,81 @@ export class DeploymentPresenter {
           userName: getUsername(deployment.environment.orgMember?.user),
         },
         deployedBy: deployment.triggeredBy,
+        errorData: this.#prepareErrorData(deployment.errorData),
       },
     };
   }
+
+  #prepareErrorData(errorData: WorkerDeployment["errorData"]) {
+    if (!errorData) {
+      return;
+    }
+
+    const parsedErrorData = DeploymentErrorData.safeParse(errorData);
+
+    if (!parsedErrorData.success) {
+      return;
+    }
+
+    if (parsedErrorData.data.name === "TaskMetadataParseError") {
+      const errorJson = safeJsonParse(parsedErrorData.data.stack);
+
+      if (errorJson) {
+        const parsedError = TaskMetadataFailedToParseData.safeParse(errorJson);
+
+        if (parsedError.success) {
+          return {
+            name: parsedErrorData.data.name,
+            message: parsedErrorData.data.message,
+            stack: createTaskMetadataFailedErrorStack(parsedError.data),
+          };
+        } else {
+          return {
+            name: parsedErrorData.data.name,
+            message: parsedErrorData.data.message,
+          };
+        }
+      } else {
+        return {
+          name: parsedErrorData.data.name,
+          message: parsedErrorData.data.message,
+        };
+      }
+    }
+
+    return {
+      name: parsedErrorData.data.name,
+      message: parsedErrorData.data.message,
+      stack: parsedErrorData.data.stack,
+    };
+  }
+}
+
+function createTaskMetadataFailedErrorStack(
+  data: z.infer<typeof TaskMetadataFailedToParseData>
+): string {
+  const stack = [];
+
+  const groupedIssues = groupTaskMetadataIssuesByTask(data.tasks, data.zodIssues);
+
+  for (const key in groupedIssues) {
+    const taskWithIssues = groupedIssues[key];
+
+    if (!taskWithIssues) {
+      continue;
+    }
+
+    stack.push("\n");
+    stack.push(`  ❯ ${taskWithIssues.exportName} in ${taskWithIssues.filePath}`);
+
+    for (const issue of taskWithIssues.issues) {
+      if (issue.path) {
+        stack.push(`    x ${issue.path} ${issue.message}`);
+      } else {
+        stack.push(`    x ${issue.message}`);
+      }
+    }
+  }
+
+  return stack.join("\n");
 }

apps/webapp/app/presenters/v3/RunListPresenter.server.ts

@@ -222,6 +222,7 @@ export class RunListPresenter {
           version: run.version,
           taskIdentifier: run.taskIdentifier,
           attempts: Number(run.attempts),
+          isReplayable: true,
           isCancellable: CANCELLABLE_STATUSES.includes(run.status),
           environment: {
             type: environment.type,

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -43,14 +43,14 @@ export class SpanPresenter {
       span.outputType === "application/store"
         ? `/resources/packets/${span.environmentId}/${span.output}`
         : typeof span.output !== "undefined" && span.output !== null
-        ? prettyPrintPacket(span.output, span.outputType ?? undefined)
+        ? await prettyPrintPacket(span.output, span.outputType ?? undefined)
         : undefined;
 
     const payload =
       span.payloadType === "application/store"
         ? `/resources/packets/${span.environmentId}/${span.payload}`
         : typeof span.payload !== "undefined" && span.payload !== null
-        ? prettyPrintPacket(span.payload, span.payloadType ?? undefined)
+        ? await prettyPrintPacket(span.payload, span.payloadType ?? undefined)
         : undefined;
 
     return {

apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts

@@ -69,7 +69,8 @@ export class TestTaskPresenter {
       ON
           tr."taskIdentifier" = bwt.slug
       WHERE
-          bwt."friendlyId" = ${taskFriendlyId}
+          bwt."friendlyId" = ${taskFriendlyId} AND
+          tr."runtimeEnvironmentId" = ${task.runtimeEnvironment.id}
       ORDER BY 
           tr."createdAt" DESC
       LIMIT 5

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers._index/route.tsx

@@ -44,7 +44,7 @@ export default function Integrations() {
 
   return (
     <>
-      <Paragraph variant="small" spacing>
+      <Paragraph variant="small" spacing className="pt-2">
         External Triggers get registered with external APIs, for example a webhook.
       </Paragraph>
       <Table containerClassName="mt-4">

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx

@@ -44,7 +44,7 @@ export default function Integrations() {
 
   return (
     <>
-      <Paragraph variant="small" spacing>
+      <Paragraph variant="small" spacing className="pt-2">
         A Scheduled Trigger runs a Job on a repeated schedule. The schedule can use a CRON
         expression or an interval.
       </Paragraph>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.webhooks/route.tsx

@@ -44,7 +44,7 @@ export default function Integrations() {
 
   return (
     <>
-      <Paragraph variant="small" spacing>
+      <Paragraph variant="small" spacing className="pt-2">
         A Webhook Trigger runs a Job when it receives a matching payload at a registered HTTP
         Endpoint.
       </Paragraph>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers/route.tsx

@@ -32,7 +32,7 @@ export default function Page() {
       </NavBar>
 
       <PageBody scrollable={false}>
-        <div className="grid grid-rows-[auto_1fr] gap-y-4 px-4">
+        <div className="grid max-h-full grid-rows-[auto_1fr] overflow-hidden px-4">
           <PageTabs
             layoutId="triggers"
             tabs={[

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.apikeys/route.tsx

@@ -19,6 +19,7 @@ import {
   TableHeaderCell,
   TableRow,
 } from "~/components/primitives/Table";
+import { TextLink } from "~/components/primitives/TextLink";
 import { useProject } from "~/hooks/useProject";
 import { ApiKeysPresenter } from "~/presenters/v3/ApiKeysPresenter.server";
 import { requireUserId } from "~/services/session.server";
@@ -55,12 +56,12 @@ export default function Page() {
   return (
     <PageContainer>
       <NavBar>
-        <PageTitle title="API Keys" />
+        <PageTitle title="API keys" />
         <PageAccessories>
           <LinkButton
             variant={"minimal/small"}
             LeadingIcon={BookOpenIcon}
-            to={docsPath("/documentation/concepts/environments-endpoints#environments")}
+            to={docsPath("/v3/apikeys")}
           >
             API keys docs
           </LinkButton>
@@ -68,9 +69,10 @@ export default function Page() {
       </NavBar>
       <PageBody>
         <div className={cn("h-full")}>
-          <Header3 spacing>Server API keys</Header3>
+          <Header3 spacing>Secret keys</Header3>
           <Paragraph variant="small" spacing>
-            Server API keys should be used on your server – they give full API access.
+            Secret keys should be used on your server – they give full API access and allow you to{" "}
+            <TextLink to={docsPath("v3/triggering")}>trigger tasks</TextLink> from your backend.
           </Paragraph>
           <Header3 spacing>Public API keys</Header3>
           <Paragraph variant="small" spacing>
@@ -81,7 +83,7 @@ export default function Page() {
               <TableHeader>
                 <TableRow>
                   <TableHeaderCell>Environment</TableHeaderCell>
-                  <TableHeaderCell>Server API key</TableHeaderCell>
+                  <TableHeaderCell>Secret key</TableHeaderCell>
                   <TableHeaderCell>Public API key</TableHeaderCell>
                   <TableHeaderCell>Keys generated</TableHeaderCell>
                   <TableHeaderCell>Latest version</TableHeaderCell>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.deployments.$deploymentParam/route.tsx

@@ -3,6 +3,7 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { UserAvatar } from "~/components/UserProfilePhoto";
+import { CodeBlock } from "~/components/code/CodeBlock";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { Badge } from "~/components/primitives/Badge";
 import { LinkButton } from "~/components/primitives/Buttons";
@@ -156,6 +157,26 @@ export default function Page() {
                 </TableBody>
               </Table>
             </div>
+          ) : deployment.errorData ? (
+            <div className="flex flex-col">
+              {deployment.errorData.stack ? (
+                <CodeBlock
+                  language="markdown"
+                  rowTitle={deployment.errorData.message}
+                  code={deployment.errorData.stack}
+                  maxLines={20}
+                />
+              ) : (
+                <div className="flex flex-col">
+                  <Paragraph
+                    variant="base/bright"
+                    className="w-full border-b border-grid-dimmed py-2.5"
+                  >
+                    {deployment.errorData.message}
+                  </Paragraph>
+                </div>
+              )}
+            </div>
           ) : null}
         </div>
       </div>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.environment-variables.new/route.tsx

@@ -16,6 +16,7 @@ import { FormError } from "~/components/primitives/FormError";
 import { Input } from "~/components/primitives/Input";
 import { InputGroup } from "~/components/primitives/InputGroup";
 import { Label } from "~/components/primitives/Label";
+import { Switch } from "~/components/primitives/Switch";
 import { prisma } from "~/db.server";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
@@ -134,6 +135,8 @@ export default function Page() {
     shouldRevalidate: "onSubmit",
   });
 
+  const [revealAll, setRevealAll] = useState(false);
+
   useEffect(() => {
     setIsOpen(true);
   }, []);
@@ -169,7 +172,15 @@ export default function Page() {
               />
             </InputGroup>
             <InputGroup fullWidth>
-              <Label>Values</Label>
+              <div className="flex items-center justify-between">
+                <Label>Values</Label>
+                <Switch
+                  variant="small"
+                  label="Reveal values"
+                  checked={revealAll}
+                  onCheckedChange={(e) => setRevealAll(e.valueOf())}
+                />
+              </div>
               <div className="grid grid-cols-[auto_1fr] gap-x-2 gap-y-2">
                 {environments.map((environment, index) => {
                   return (
@@ -185,7 +196,11 @@ export default function Page() {
                       >
                         <EnvironmentLabel environment={environment} className="h-5 px-2" />
                       </label>
-                      <Input name={`values[${index}].value`} placeholder="Not set" />
+                      <Input
+                        type={revealAll ? "text" : "password"}
+                        name={`values[${index}].value`}
+                        placeholder="Not set"
+                      />
                     </Fragment>
                   );
                 })}

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.environment-variables/route.tsx

@@ -167,7 +167,7 @@ export default function Page() {
         <PageAccessories>
           <LinkButton
             LeadingIcon={BookOpenIcon}
-            to={docsPath("/documentation/concepts/environments-endpoints#environments")}
+            to={docsPath("v3/deploy-environment-variables")}
             variant="minimal/small"
           >
             Environment variables docs

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -1,4 +1,9 @@
-import { CloudArrowDownIcon, QueueListIcon, StopCircleIcon } from "@heroicons/react/20/solid";
+import {
+  ArrowPathIcon,
+  CloudArrowDownIcon,
+  QueueListIcon,
+  StopCircleIcon,
+} from "@heroicons/react/20/solid";
 import { useParams } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { formatDurationNanoseconds, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
@@ -14,6 +19,7 @@ import { Paragraph } from "~/components/primitives/Paragraph";
 import { Property, PropertyTable } from "~/components/primitives/PropertyTable";
 import { CancelRunDialog } from "~/components/runs/v3/CancelRunDialog";
 import { LiveTimer } from "~/components/runs/v3/LiveTimer";
+import { ReplayRunDialog } from "~/components/runs/v3/ReplayRunDialog";
 import { RunIcon } from "~/components/runs/v3/RunIcon";
 import { SpanEvents } from "~/components/runs/v3/SpanEvents";
 import { SpanTitle } from "~/components/runs/v3/SpanTitle";
@@ -22,7 +28,7 @@ import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptSt
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { redirectWithErrorMessage } from "~/models/message.server";
-import { SpanPresenter } from "~/presenters/v3/SpanPresenter.server";
+import { Span, SpanPresenter } from "~/presenters/v3/SpanPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { cn } from "~/utils/cn";
 import { v3RunPath, v3RunSpanPath, v3SpanParamsSchema, v3TraceSpanPath } from "~/utils/pathBuilder";
@@ -188,24 +194,7 @@ export default function Page() {
             )}
           </div>
           <div className="flex items-center gap-4">
-            {event.isPartial && runParam && (
-              <Dialog>
-                <DialogTrigger asChild>
-                  <Button variant="danger/small" LeadingIcon={StopCircleIcon}>
-                    Cancel run
-                  </Button>
-                </DialogTrigger>
-                <CancelRunDialog
-                  runFriendlyId={event.runId}
-                  redirectPath={v3RunSpanPath(
-                    organization,
-                    project,
-                    { friendlyId: runParam },
-                    { spanId: event.spanId }
-                  )}
-                />
-              </Dialog>
-            )}
+            <RunActionButtons span={event} />
           </div>
         </div>
       ) : null}
@@ -213,6 +202,54 @@ export default function Page() {
   );
 }
 
+function RunActionButtons({ span }: { span: Span }) {
+  const organization = useOrganization();
+  const project = useProject();
+  const { runParam } = useParams();
+
+  if (!runParam) return null;
+
+  if (span.isPartial) {
+    return (
+      <Dialog>
+        <DialogTrigger asChild>
+          <Button variant="danger/small" LeadingIcon={StopCircleIcon}>
+            Cancel run
+          </Button>
+        </DialogTrigger>
+        <CancelRunDialog
+          runFriendlyId={span.runId}
+          redirectPath={v3RunSpanPath(
+            organization,
+            project,
+            { friendlyId: runParam },
+            { spanId: span.spanId }
+          )}
+        />
+      </Dialog>
+    );
+  }
+
+  return (
+    <Dialog>
+      <DialogTrigger asChild>
+        <Button variant="tertiary/small" LeadingIcon={ArrowPathIcon}>
+          Replay run
+        </Button>
+      </DialogTrigger>
+      <ReplayRunDialog
+        runFriendlyId={span.runId}
+        failedRedirect={v3RunSpanPath(
+          organization,
+          project,
+          { friendlyId: runParam },
+          { spanId: span.spanId }
+        )}
+      />
+    </Dialog>
+  );
+}
+
 function PacketDisplay({
   data,
   dataType,
@@ -222,19 +259,25 @@ function PacketDisplay({
   dataType: string;
   title: string;
 }) {
-  if (dataType === "application/store") {
-    return (
-      <div className="flex flex-col">
-        <Paragraph variant="base/bright" className="w-full border-b border-grid-dimmed py-2.5">
-          {title}
-        </Paragraph>
-        <LinkButton LeadingIcon={CloudArrowDownIcon} to={data} variant="tertiary/medium" download>
-          Download
-        </LinkButton>
-      </div>
-    );
-  } else {
-    return <CodeBlock rowTitle={title} code={data} maxLines={20} />;
+  switch (dataType) {
+    case "application/store": {
+      return (
+        <div className="flex flex-col">
+          <Paragraph variant="base/bright" className="w-full border-b border-grid-dimmed py-2.5">
+            {title}
+          </Paragraph>
+          <LinkButton LeadingIcon={CloudArrowDownIcon} to={data} variant="tertiary/medium" download>
+            Download
+          </LinkButton>
+        </div>
+      );
+    }
+    case "text/plain": {
+      return <CodeBlock language="markdown" rowTitle={title} code={data} maxLines={20} />;
+    }
+    default: {
+      return <CodeBlock language="json" rowTitle={title} code={data} maxLines={20} />;
+    }
   }
 }
 

apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts

@@ -0,0 +1,66 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+const ParamsSchema = z.object({
+  environmentId: z.string(),
+});
+
+const RequestBodySchema = z.object({
+  envMaximumConcurrencyLimit: z.number(),
+  orgMaximumConcurrencyLimit: z.number(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  const parsedParams = ParamsSchema.parse(params);
+
+  const rawBody = await request.json();
+  const body = RequestBodySchema.parse(rawBody);
+
+  const environment = await prisma.runtimeEnvironment.update({
+    where: {
+      id: parsedParams.environmentId,
+    },
+    data: {
+      maximumConcurrencyLimit: body.envMaximumConcurrencyLimit,
+      organization: {
+        update: {
+          data: {
+            maximumConcurrencyLimit: body.orgMaximumConcurrencyLimit,
+          },
+        },
+      },
+    },
+    include: {
+      organization: true,
+      project: true,
+    },
+  });
+
+  await marqs?.updateEnvConcurrencyLimits(environment);
+
+  return json({ success: true });
+}

apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts

@@ -0,0 +1,72 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { PrismaErrorSchema, prisma } from "~/db.server";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CancelRunService } from "~/services/runs/cancelRun.server";
+import { ApiRunPresenter } from "~/presenters/ApiRunPresenter.server";
+import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { logger } from "~/services/logger.server";
+
+const ParamsSchema = z.object({
+  /* This is the run friendly ID */
+  runParam: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { runParam } = parsed.data;
+
+  try {
+    const taskRun = await prisma.taskRun.findUnique({
+      where: {
+        friendlyId: runParam,
+      },
+    });
+
+    if (!taskRun) {
+      return json({ error: "Run not found" }, { status: 404 });
+    }
+
+    const service = new ReplayTaskRunService();
+    const newRun = await service.call(taskRun);
+
+    if (!newRun) {
+      return json({ error: "Failed to create new run" }, { status: 400 });
+    }
+
+    return json({
+      id: newRun?.friendlyId,
+    });
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Failed to replay run", {
+        error: {
+          name: error.name,
+          message: error.message,
+          stack: error.stack,
+        },
+        run: runParam,
+      });
+      return json({ error: error.message }, { status: 400 });
+    } else {
+      logger.error("Failed to replay run", { error: JSON.stringify(error), run: runParam });
+      return json({ error: JSON.stringify(error) }, { status: 400 });
+    }
+  }
+}

apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts

@@ -25,7 +25,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const parsed = ParamsSchema.safeParse(params);
 
   if (!parsed.success) {
-    return json({ error: "Invalid or Missing runId" }, { status: 400 });
+    return json({ error: "Invalid or Missing run id" }, { status: 400 });
   }
 
   const { runParam } = parsed.data;

apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts

@@ -0,0 +1,85 @@
+import { parse } from "@conform-to/zod";
+import { ActionFunction, json } from "@remix-run/node";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
+import { logger } from "~/services/logger.server";
+import { v3RunPath } from "~/utils/pathBuilder";
+import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+
+const FormSchema = z.object({
+  failedRedirect: z.string(),
+});
+
+const ParamSchema = z.object({
+  runParam: z.string(),
+});
+
+export const action: ActionFunction = async ({ request, params }) => {
+  const { runParam } = ParamSchema.parse(params);
+
+  const formData = await request.formData();
+  const submission = parse(formData, { schema: FormSchema });
+
+  if (!submission.value) {
+    return json(submission);
+  }
+
+  try {
+    const taskRun = await prisma.taskRun.findUnique({
+      where: {
+        friendlyId: runParam,
+      },
+      include: {
+        project: {
+          include: {
+            organization: true,
+          },
+        },
+      },
+    });
+
+    if (!taskRun) {
+      return redirectWithErrorMessage(submission.value.failedRedirect, request, "Run not found");
+    }
+
+    const replayRunService = new ReplayTaskRunService();
+    const newRun = await replayRunService.call(taskRun);
+
+    if (!newRun) {
+      return redirectWithErrorMessage(
+        submission.value.failedRedirect,
+        request,
+        "Failed to replay run"
+      );
+    }
+
+    const runPath = v3RunPath(
+      {
+        slug: taskRun.project.organization.slug,
+      },
+      { slug: taskRun.project.slug },
+      { friendlyId: newRun.friendlyId }
+    );
+
+    return redirectWithSuccessMessage(runPath, request, `Replaying run`);
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Failed to replay run", {
+        error: {
+          name: error.name,
+          message: error.message,
+          stack: error.stack,
+        },
+      });
+      return redirectWithErrorMessage(submission.value.failedRedirect, request, error.message);
+    } else {
+      logger.error("Failed to replay run", { error });
+      return redirectWithErrorMessage(
+        submission.value.failedRedirect,
+        request,
+        JSON.stringify(error)
+      );
+    }
+  }
+};

apps/webapp/app/services/apiRateLimit.server.ts

@@ -67,6 +67,11 @@ export function authorizationRateLimitMiddleware({
       logger.info(`RateLimiter (${keyPrefix}): request to ${req.path}`);
     }
 
+    // allow OPTIONS requests
+    if (req.method.toUpperCase() === "OPTIONS") {
+      return next();
+    }
+
     //first check if any of the pathMatchers match the request path
     const path = req.path;
     if (
@@ -109,6 +114,7 @@ export function authorizationRateLimitMiddleware({
             status: 401,
             type: "https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/401",
             detail: "No authorization header provided",
+            error: "No authorization header provided",
           },
           null,
           2
@@ -160,6 +166,7 @@ export function authorizationRateLimitMiddleware({
           detail: `Rate limit exceeded ${remaining}/${limit} requests remaining. Retry after ${reset} seconds.`,
           reset: reset,
           limit: limit,
+          error: `Rate limit exceeded ${remaining}/${limit} requests remaining. Retry after ${reset} seconds.`,
         },
         null,
         2
@@ -182,7 +189,16 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
   },
   limiter: Ratelimit.slidingWindow(env.API_RATE_LIMIT_MAX, env.API_RATE_LIMIT_WINDOW as Duration),
   pathMatchers: [/^\/api/],
-  pathWhiteList: ["/api/v1/authorization-code", "/api/v1/token"],
+  // Allow /api/v1/tasks/:id/callback/:secret
+  pathWhiteList: [
+    "/api/v1/authorization-code",
+    "/api/v1/token",
+    /^\/api\/v1\/tasks\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/tasks/$id/callback/$secret
+    /^\/api\/v1\/runs\/[^\/]+\/tasks\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/runs/$runId/tasks/$id/callback/$secret
+    /^\/api\/v1\/http-endpoints\/[^\/]+\/env\/[^\/]+\/[^\/]+$/, // /api/v1/http-endpoints/$httpEndpointId/env/$envType/$shortcode
+    /^\/api\/v1\/sources\/http\/[^\/]+$/, // /api/v1/sources/http/$id
+    /^\/api\/v1\/endpoints\/[^\/]+\/[^\/]+\/index\/[^\/]+$/, // /api/v1/endpoints/$environmentId/$endpointSlug/index/$indexHookIdentifier
+  ],
   log: {
     rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",
     requests: env.API_RATE_LIMIT_REQUEST_LOGS_ENABLED === "1",

apps/webapp/app/utils/json.ts

@@ -1,6 +1,10 @@
 import { z } from "zod";
 
-export function safeJsonParse(json: string): unknown {
+export function safeJsonParse(json?: string): unknown {
+  if (!json) {
+    return;
+  }
+
   try {
     return JSON.parse(json);
   } catch (e) {

apps/webapp/app/utils/throttle.ts

@@ -1,20 +1,20 @@
 //From: https://kettanaito.com/blog/debounce-vs-throttle
 
-/** A very simple throttle. Will execute the function every Xms and discard any other calls during that period. */
+/** A very simple throttle. Will execute the function at the end of each period and discard any other calls during that period. */
 export function throttle(
   func: (...args: any[]) => void,
-  duration: number
+  durationMs: number
 ): (...args: any[]) => void {
-  let shouldWait = false;
+  let isPrimedToFire = false;
 
   return (...args: any[]) => {
-    if (!shouldWait) {
-      func(...args);
-      shouldWait = true;
+    if (!isPrimedToFire) {
+      isPrimedToFire = true;
 
       setTimeout(() => {
-        shouldWait = false;
-      }, duration);
+        func(...args);
+        isPrimedToFire = false;
+      }, durationMs);
     }
   };
 }

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -30,7 +30,7 @@ function parseSecretKey(key: string) {
 const SecretValue = z.object({ secret: z.string() });
 
 export class EnvironmentVariablesRepository implements Repository {
-  constructor(private prismaClient: PrismaClient = prisma) { }
+  constructor(private prismaClient: PrismaClient = prisma) {}
 
   async create(
     projectId: string,
@@ -419,8 +419,49 @@ export class EnvironmentVariablesRepository implements Repository {
         {
           key: "OTEL_EXPORTER_OTLP_ENDPOINT",
           value: env.DEV_OTEL_EXPORTER_OTLP_ENDPOINT ?? env.APP_ORIGIN,
-        }
-      ];
+        },
+      ].concat(
+        env.DEV_OTEL_BATCH_PROCESSING_ENABLED === "1"
+          ? [
+              {
+                key: "OTEL_BATCH_PROCESSING_ENABLED",
+                value: "1",
+              },
+              {
+                key: "OTEL_SPAN_MAX_EXPORT_BATCH_SIZE",
+                value: env.DEV_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE,
+              },
+              {
+                key: "OTEL_SPAN_SCHEDULED_DELAY_MILLIS",
+                value: env.DEV_OTEL_SPAN_SCHEDULED_DELAY_MILLIS,
+              },
+              {
+                key: "OTEL_SPAN_EXPORT_TIMEOUT_MILLIS",
+                value: env.DEV_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS,
+              },
+              {
+                key: "OTEL_SPAN_MAX_QUEUE_SIZE",
+                value: env.DEV_OTEL_SPAN_MAX_QUEUE_SIZE,
+              },
+              {
+                key: "OTEL_LOG_MAX_EXPORT_BATCH_SIZE",
+                value: env.DEV_OTEL_LOG_MAX_EXPORT_BATCH_SIZE,
+              },
+              {
+                key: "OTEL_LOG_SCHEDULED_DELAY_MILLIS",
+                value: env.DEV_OTEL_LOG_SCHEDULED_DELAY_MILLIS,
+              },
+              {
+                key: "OTEL_LOG_EXPORT_TIMEOUT_MILLIS",
+                value: env.DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS,
+              },
+              {
+                key: "OTEL_LOG_MAX_QUEUE_SIZE",
+                value: env.DEV_OTEL_LOG_MAX_QUEUE_SIZE,
+              },
+            ]
+          : []
+      );
     }
 
     return [
@@ -432,6 +473,50 @@ export class EnvironmentVariablesRepository implements Repository {
         key: "TRIGGER_API_URL",
         value: env.APP_ORIGIN,
       },
+      {
+        key: "TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS",
+        value: String(env.RUNTIME_WAIT_THRESHOLD_IN_MS),
+      },
+      ...(env.PROD_OTEL_BATCH_PROCESSING_ENABLED === "1"
+        ? [
+            {
+              key: "OTEL_BATCH_PROCESSING_ENABLED",
+              value: "1",
+            },
+            {
+              key: "OTEL_SPAN_MAX_EXPORT_BATCH_SIZE",
+              value: env.PROD_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE,
+            },
+            {
+              key: "OTEL_SPAN_SCHEDULED_DELAY_MILLIS",
+              value: env.PROD_OTEL_SPAN_SCHEDULED_DELAY_MILLIS,
+            },
+            {
+              key: "OTEL_SPAN_EXPORT_TIMEOUT_MILLIS",
+              value: env.PROD_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS,
+            },
+            {
+              key: "OTEL_SPAN_MAX_QUEUE_SIZE",
+              value: env.PROD_OTEL_SPAN_MAX_QUEUE_SIZE,
+            },
+            {
+              key: "OTEL_LOG_MAX_EXPORT_BATCH_SIZE",
+              value: env.PROD_OTEL_LOG_MAX_EXPORT_BATCH_SIZE,
+            },
+            {
+              key: "OTEL_LOG_SCHEDULED_DELAY_MILLIS",
+              value: env.PROD_OTEL_LOG_SCHEDULED_DELAY_MILLIS,
+            },
+            {
+              key: "OTEL_LOG_EXPORT_TIMEOUT_MILLIS",
+              value: env.PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS,
+            },
+            {
+              key: "OTEL_LOG_MAX_QUEUE_SIZE",
+              value: env.PROD_OTEL_LOG_MAX_QUEUE_SIZE,
+            },
+          ]
+        : []),
     ];
   }
 

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

@@ -3,6 +3,7 @@ import { RandomIdGenerator } from "@opentelemetry/sdk-trace-base";
 import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
 import {
   ExceptionEventProperties,
+  ExceptionSpanEvent,
   PRIMARY_VARIANT,
   SemanticInternalAttributes,
   SpanEvent,
@@ -10,7 +11,7 @@ import {
   SpanMessagingEvent,
   TaskEventStyle,
   correctErrorStackTrace,
-  createPackageAttributesAsJson,
+  createPacketAttributesAsJson,
   flattenAttributes,
   isExceptionSpanEvent,
   omit,
@@ -188,7 +189,7 @@ export class EventRepository {
     const event = events[0];
 
     const output = options?.attributes.output
-      ? createPackageAttributesAsJson(
+      ? await createPacketAttributesAsJson(
           options?.attributes.output,
           options?.attributes.outputType ?? "application/json"
         )
@@ -213,8 +214,9 @@ export class EventRepository {
       style: event.style as Attributes,
       output: output,
       outputType:
-        options?.attributes.outputType === "application/store"
-          ? "application/store"
+        options?.attributes.outputType === "application/store" ||
+        options?.attributes.outputType === "text/plain"
+          ? options?.attributes.outputType
           : "application/json",
       payload: event.payload as Attributes,
       payloadType: event.payloadType,
@@ -254,6 +256,47 @@ export class EventRepository {
     });
   }
 
+  async crashEvent({
+    event,
+    crashedAt,
+    exception,
+  }: {
+    event: TaskEventRecord;
+    crashedAt: Date;
+    exception: ExceptionEventProperties;
+  }) {
+    if (!event.isPartial) {
+      return;
+    }
+
+    await this.insertImmediate({
+      ...omit(event, "id"),
+      isPartial: false,
+      isError: true,
+      isCancelled: false,
+      status: "ERROR",
+      links: event.links ?? [],
+      events: [
+        {
+          name: "exception",
+          time: crashedAt,
+          properties: {
+            exception,
+          },
+        } satisfies ExceptionSpanEvent,
+        ...((event.events as any[]) ?? []),
+      ],
+      duration: calculateDurationFromStart(event.startTime, crashedAt),
+      properties: event.properties as Attributes,
+      metadata: event.metadata as Attributes,
+      style: event.style as Attributes,
+      output: event.output as Attributes,
+      outputType: event.outputType,
+      payload: event.payload as Attributes,
+      payloadType: event.payloadType,
+    });
+  }
+
   async queryEvents(queryOptions: QueryOptions): Promise<TaskEventRecord[]> {
     return await this.db.taskEvent.findMany({
       where: queryOptions,

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

@@ -21,6 +21,7 @@ import { ResumeAttemptService } from "./services/resumeAttempt.server";
 import { DeploymentIndexFailed } from "./services/deploymentIndexFailed.server";
 import { Redis } from "ioredis";
 import { createAdapter } from "@socket.io/redis-adapter";
+import { CrashTaskRunService } from "./services/crashTaskRun.server";
 
 export const socketIo = singleton("socketIo", initalizeIoServer);
 
@@ -135,7 +136,7 @@ function createCoordinatorNamespace(io: Server) {
 
           await service.call(message.deploymentId, message.error);
         } catch (e) {
-          logger.error("Error while indexing failed", { error: e });
+          logger.error("Error while indexing", { error: e });
         }
       },
     },
@@ -151,6 +152,28 @@ function createProviderNamespace(io: Server) {
     authToken: env.PROVIDER_SECRET,
     clientMessages: ProviderToPlatformMessages,
     serverMessages: PlatformToProviderMessages,
+    handlers: {
+      WORKER_CRASHED: async (message) => {
+        try {
+          const service = new CrashTaskRunService();
+
+          await service.call(message.runId, {
+            ...message,
+          });
+        } catch (error) {
+          logger.error("Error while handling crashed worker", { error });
+        }
+      },
+      INDEXING_FAILED: async (message) => {
+        try {
+          const service = new DeploymentIndexFailed();
+
+          await service.call(message.deploymentId, message.error);
+        } catch (e) {
+          logger.error("Error while indexing", { error: e });
+        }
+      },
+    },
   });
 
   return provider.namespace;
@@ -168,6 +191,7 @@ function createSharedQueueConsumerNamespace(io: Server) {
         namespace: sharedQueue.namespace,
         socket,
         logger,
+        poolSize: env.SHARED_QUEUE_CONSUMER_POOL_SIZE,
       });
 
       sharedSocketConnection.onClose.attach((closeEvent) => {

apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -13,15 +13,17 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { CompleteAttemptService } from "../services/completeAttempt.server";
-import { attributesFromAuthenticatedEnv } from "../tracer.server";
+import {
+  SEMINTATTRS_FORCE_RECORDING,
+  attributesFromAuthenticatedEnv,
+  tracer,
+} from "../tracer.server";
 import { DevSubscriber, devPubSub } from "./devPubSub.server";
 import { CancelTaskRunService } from "../services/cancelTaskRun.server";
 
-const tracer = trace.getTracer("devQueueConsumer");
-
 const MessageBody = z.discriminatedUnion("type", [
   z.object({
     type: z.literal("EXECUTE"),
@@ -165,6 +167,11 @@ export class DevQueueConsumer {
 
       logger.debug("Unsubscribed from background worker channel", { id });
     }
+
+    // We need to end the current span
+    if (this._currentSpan) {
+      this._currentSpan.end();
+    }
   }
 
   async #cancelInProgressRunsAndAttempts(reason: string) {
@@ -284,6 +291,10 @@ export class DevQueueConsumer {
         this._currentSpan.setAttribute("tasks.period.failures", this._taskFailures);
         this._currentSpan.setAttribute("tasks.period.successes", this._taskSuccesses);
 
+        logger.debug("Ending DevQueueConsumer.doWork() trace", {
+          isRecording: this._currentSpan.isRecording(),
+        });
+
         this._currentSpan.end();
       }
 
@@ -294,6 +305,7 @@ export class DevQueueConsumer {
           kind: SpanKind.CONSUMER,
           attributes: {
             ...attributesFromAuthenticatedEnv(this.env),
+            [SEMINTATTRS_FORCE_RECORDING]: true,
           },
         },
         ROOT_CONTEXT

apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts

@@ -0,0 +1,119 @@
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { MarQSKeyProducer } from "./types";
+
+const constants = {
+  SHARED_QUEUE: "sharedQueue",
+  CURRENT_CONCURRENCY_PART: "currentConcurrency",
+  CONCURRENCY_LIMIT_PART: "concurrency",
+  ENV_PART: "env",
+  ORG_PART: "org",
+  QUEUE_PART: "queue",
+  CONCURRENCY_KEY_PART: "ck",
+  MESSAGE_PART: "message",
+} as const;
+
+export class MarQSShortKeyProducer implements MarQSKeyProducer {
+  constructor(private _prefix: string) {}
+
+  queueConcurrencyLimitKey(env: AuthenticatedEnvironment, queue: string) {
+    return [this.queueKey(env, queue), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  envConcurrencyLimitKey(env: AuthenticatedEnvironment) {
+    return [this.envKeySection(env.id), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  orgConcurrencyLimitKey(env: AuthenticatedEnvironment) {
+    return [this.orgKeySection(env.organizationId), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string) {
+    return [
+      this.orgKeySection(env.organizationId),
+      this.envKeySection(env.id),
+      this.queueSection(queue),
+    ]
+      .concat(concurrencyKey ? this.concurrencyKeySection(concurrencyKey) : [])
+      .join(":");
+  }
+
+  envSharedQueueKey(env: AuthenticatedEnvironment) {
+    if (env.type === "DEVELOPMENT") {
+      return [
+        this.orgKeySection(env.organizationId),
+        this.envKeySection(env.id),
+        constants.SHARED_QUEUE,
+      ].join(":");
+    }
+
+    return constants.SHARED_QUEUE;
+  }
+
+  concurrencyLimitKeyFromQueue(queue: string) {
+    const concurrencyQueueName = queue.replace(/:ck:.+$/, "");
+
+    return `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  currentConcurrencyKeyFromQueue(queue: string) {
+    return `${queue}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  orgConcurrencyLimitKeyFromQueue(queue: string) {
+    const orgId = this.normalizeQueue(queue).split(":")[1];
+
+    return `${constants.ORG_PART}:${orgId}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  orgCurrentConcurrencyKeyFromQueue(queue: string) {
+    const orgId = this.normalizeQueue(queue).split(":")[1];
+
+    return `${constants.ORG_PART}:${orgId}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  envConcurrencyLimitKeyFromQueue(queue: string) {
+    const envId = this.normalizeQueue(queue).split(":")[3];
+
+    return `${constants.ENV_PART}:${envId}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  envCurrentConcurrencyKeyFromQueue(queue: string) {
+    const envId = this.normalizeQueue(queue).split(":")[3];
+
+    return `${constants.ENV_PART}:${envId}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  messageKey(messageId: string) {
+    return `${constants.MESSAGE_PART}:${messageId}`;
+  }
+
+  private shortId(id: string) {
+    // Return the last 12 characters of the id
+    return id.slice(-12);
+  }
+
+  private envKeySection(envId: string) {
+    return `${constants.ENV_PART}:${this.shortId(envId)}`;
+  }
+
+  private orgKeySection(orgId: string) {
+    return `${constants.ORG_PART}:${this.shortId(orgId)}`;
+  }
+
+  private queueSection(queue: string) {
+    return `${constants.QUEUE_PART}:${queue}`;
+  }
+
+  private concurrencyKeySection(concurrencyKey: string) {
+    return `${constants.CONCURRENCY_KEY_PART}:${concurrencyKey}`;
+  }
+
+  // This removes the leading prefix from the queue name if it exists
+  private normalizeQueue(queue: string) {
+    if (queue.startsWith(this._prefix)) {
+      return queue.slice(this._prefix.length);
+    }
+
+    return queue;
+  }
+}

apps/webapp/app/v3/marqs/priorityStrategy.server.ts

@@ -0,0 +1,150 @@
+import { RedisOptions } from "ioredis";
+import { MarQSQueuePriorityStrategy, PriorityStrategyChoice, QueueWithScores } from "./types";
+import { nanoid } from "nanoid";
+import seedrandom from "seedrandom";
+
+export type DynamicWeightedChoiceStrategyOptions = {
+  initialQueueSelectionSize: number;
+  redis: RedisOptions;
+};
+
+// This implementation of the priority strategy will "react" over time, giving more weight to queues that have been selected less frequently.
+// It will also change the next candidate selection range based on if previous choices only had queues that were at capacity.
+// Some other ideas:
+// - Implement a "cooldown" period for queues that have been selected recently
+// - Implement a "decay" for queues that have been selected recently
+//
+// The "memory" of this strategy is stored in Redis, to coordinate between multiple instances of the webapp (coming soon?)
+export class DynamicWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  constructor(private options: DynamicWeightedChoiceStrategyOptions) {}
+
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice {
+    throw new Error("Method not implemented.");
+  }
+
+  nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }> {
+    throw new Error("Method not implemented.");
+  }
+}
+
+export type SimpleWeightedChoiceStrategyOptions = {
+  queueSelectionCount: number;
+  randomSeed?: string;
+};
+
+export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  private _nextRangesByParentQueue: Map<string, [number, number]> = new Map();
+  private _randomGenerator = seedrandom(this.options.randomSeed);
+
+  constructor(private options: SimpleWeightedChoiceStrategyOptions) {}
+
+  private nextRangeForParentQueue(parentQueue: string) {
+    return this._nextRangesByParentQueue.get(parentQueue) ?? [0, this.options.queueSelectionCount];
+  }
+
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice {
+    const filteredQueues = filterQueuesAtCapacity(queues);
+
+    if (filteredQueues.length === 0) {
+      if (queues.length === this.options.queueSelectionCount) {
+        const nextRangeForParentQueue = this.nextRangeForParentQueue(parentQueue);
+        const nextRange: [number, number] = nextRangeForParentQueue
+          ? [
+              nextRangeForParentQueue[1],
+              nextRangeForParentQueue[1] + this.options.queueSelectionCount,
+            ]
+          : [this.options.queueSelectionCount, this.options.queueSelectionCount * 2];
+        // If all queues are at capacity, and we were passed the max number of queues, then we will slide the window "to the right"
+        this._nextRangesByParentQueue.set(parentQueue, nextRange);
+      } else {
+        this._nextRangesByParentQueue.delete(parentQueue);
+      }
+
+      return { abort: true };
+    }
+
+    this._nextRangesByParentQueue.delete(parentQueue);
+
+    const queueWeights = this.#calculateQueueWeights(filteredQueues);
+
+    return weightedRandomChoice(queueWeights, this._randomGenerator());
+  }
+
+  async nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }> {
+    return { range: this.nextRangeForParentQueue(parentQueue), selectionId: nanoid(24) };
+  }
+
+  // This function calculates the weight of each queue based on the age of the queue and the capacity of the queue, env, and org
+  // First, it normalizes the age, queue capacity, env capacity, and org capacity to a value between 0 and 1 based on the maximum value of each
+  // Then, it calculates the weight of each queue based on the following factors:
+  // - Age is 50% of the weight
+  // - Queue capacity is 30% of the weight
+  // - Env capacity is 10% of the weight
+  // - Org capacity is 10% of the weight
+  #calculateQueueWeights(queues: QueueWithScores[]) {
+    const maximumAge = Math.max(...queues.map((queue) => queue.age));
+    const maximumQueueCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.queue.limit - queue.capacities.queue.current)
+    );
+    const maximumEnvCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.env.limit - queue.capacities.env.current)
+    );
+    const maximumOrgCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.org.limit - queue.capacities.org.current)
+    );
+
+    return queues.map(({ capacities, age, queue }) => {
+      const ageWeight = 0.5 * (age / maximumAge);
+      const queueWeight =
+        0.3 * (1 - (capacities.queue.limit - capacities.queue.current) / maximumQueueCapacity);
+      const envWeight =
+        0.1 * (1 - (capacities.env.limit - capacities.env.current) / maximumEnvCapacity);
+      const orgWeight =
+        0.1 * (1 - (capacities.org.limit - capacities.org.current) / maximumOrgCapacity);
+
+      return {
+        queue,
+        weight: ageWeight + queueWeight + envWeight + orgWeight,
+      };
+    });
+  }
+}
+
+function filterQueuesAtCapacity(queues: QueueWithScores[]) {
+  return queues.filter(
+    (queue) =>
+      queue.capacities.queue.current < queue.capacities.queue.limit &&
+      queue.capacities.env.current < queue.capacities.env.limit &&
+      queue.capacities.org.current < queue.capacities.org.limit
+  );
+}
+
+function weightedRandomChoice(
+  queues: Array<{ queue: string; weight: number }>,
+  randomNumber: number
+) {
+  const totalWeight = queues.reduce((acc, queue) => acc + queue.weight, 0);
+  const randomNum = randomNumber * totalWeight;
+  let weightSum = 0;
+
+  for (const queue of queues) {
+    weightSum += queue.weight;
+    if (randomNum <= weightSum) {
+      return queue.queue;
+    }
+  }
+
+  return queues[queues.length - 1].queue;
+}

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -1,5 +1,6 @@
 import { Context, ROOT_CONTEXT, Span, SpanKind, context, trace } from "@opentelemetry/api";
 import {
+  Machine,
   ProdTaskRunExecution,
   ProdTaskRunExecutionPayload,
   TaskRunError,
@@ -19,30 +20,34 @@ import {
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
-import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
-import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
-import { CancelAttemptService } from "../services/cancelAttempt.server";
-import { socketIo } from "../handleSocketIo.server";
 import { singleton } from "~/utils/singleton";
-import { RestoreCheckpointService } from "../services/restoreCheckpoint.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
+import { generateFriendlyId } from "../friendlyIdentifiers";
+import { socketIo } from "../handleSocketIo.server";
 import { findCurrentWorkerDeployment } from "../models/workerDeployment.server";
+import { CancelAttemptService } from "../services/cancelAttempt.server";
+import { RestoreCheckpointService } from "../services/restoreCheckpoint.server";
+import { tracer } from "../tracer.server";
 
-const tracer = trace.getTracer("sharedQueueConsumer");
+const WithTraceContext = z.object({
+  traceparent: z.string().optional(),
+  tracestate: z.string().optional(),
+});
 
 const MessageBody = z.discriminatedUnion("type", [
-  z.object({
+  WithTraceContext.extend({
     type: z.literal("EXECUTE"),
     taskIdentifier: z.string(),
     checkpointEventId: z.string().optional(),
   }),
-  z.object({
+  WithTraceContext.extend({
     type: z.literal("RESUME"),
     completedAttemptIds: z.string().array(),
     resumableAttemptId: z.string(),
     checkpointEventId: z.string().optional(),
   }),
-  z.object({
+  WithTraceContext.extend({
     type: z.literal("RESUME_AFTER_DURATION"),
     resumableAttemptId: z.string(),
     checkpointEventId: z.string(),
@@ -56,6 +61,7 @@ export type SharedQueueConsumerOptions = {
   traceTimeoutSeconds?: number;
   nextTickInterval?: number;
   interval?: number;
+  parentContext?: Context;
 };
 
 export class SharedQueueConsumer {
@@ -82,6 +88,7 @@ export class SharedQueueConsumer {
       traceTimeoutSeconds: options.traceTimeoutSeconds ?? 60, // 60 seconds
       nextTickInterval: options.nextTickInterval ?? 1000, // 1 second
       interval: options.interval ?? 100, // 100ms
+      parentContext: options.parentContext ?? ROOT_CONTEXT,
     };
   }
 
@@ -138,6 +145,10 @@ export class SharedQueueConsumer {
 
     logger.debug("Stopping shared queue consumer");
     this._enabled = false;
+
+    if (this._currentSpan) {
+      this._currentSpan.end();
+    }
   }
 
   async #cancelInProgressAttempts(reason: string) {
@@ -186,8 +197,17 @@ export class SharedQueueConsumer {
     this.#doWork().finally(() => {});
   }
 
+  #endCurrentSpan() {
+    if (this._currentSpan) {
+      this._currentSpan.setAttribute("tasks.period.failures", this._taskFailures);
+      this._currentSpan.setAttribute("tasks.period.successes", this._taskSuccesses);
+      this._currentSpan.end();
+    }
+  }
+
   async #doWork() {
     if (!this._enabled) {
+      this.#endCurrentSpan();
       return;
     }
 
@@ -198,12 +218,9 @@ export class SharedQueueConsumer {
       this._currentSpanContext === undefined ||
       this._endSpanInNextIteration
     ) {
-      if (this._currentSpan) {
-        this._currentSpan.setAttribute("tasks.period.failures", this._taskFailures);
-        this._currentSpan.setAttribute("tasks.period.successes", this._taskSuccesses);
+      this.#endCurrentSpan();
 
-        this._currentSpan.end();
-      }
+      const parentContext = this._options.parentContext ?? ROOT_CONTEXT;
 
       // Create a new trace
       this._currentSpan = tracer.startSpan(
@@ -211,11 +228,11 @@ export class SharedQueueConsumer {
         {
           kind: SpanKind.CONSUMER,
         },
-        ROOT_CONTEXT
+        parentContext
       );
 
       // Get the span trace context
-      this._currentSpanContext = trace.setSpan(ROOT_CONTEXT, this._currentSpan);
+      this._currentSpanContext = trace.setSpan(parentContext, this._currentSpan);
 
       this._perTraceCountdown = this._options.maximumItemsPerTrace;
       this._lastNewTrace = new Date();
@@ -250,35 +267,12 @@ export class SharedQueueConsumer {
 
     logger.log("dequeueMessageInSharedQueue()", { queueMessage: message });
 
-    const envId = this.#envIdFromQueue(message.queue);
-
-    const environment = await prisma.runtimeEnvironment.findUnique({
-      include: {
-        organization: true,
-        project: true,
-      },
-      where: {
-        id: envId,
-      },
-    });
-
-    if (!environment) {
-      logger.error("Environment not found", {
-        queueMessage: message.data,
-        envId,
-      });
-
-      this.#ackAndDoMoreWork(message.messageId);
-      return;
-    }
-
     const messageBody = MessageBody.safeParse(message.data);
 
     if (!messageBody.success) {
       logger.error("Failed to parse message", {
         queueMessage: message.data,
         error: messageBody.error,
-        env: environment,
       });
 
       this.#ackAndDoMoreWork(message.messageId);
@@ -380,6 +374,7 @@ export class SharedQueueConsumer {
             lockedById: backgroundTask.id,
           },
           include: {
+            runtimeEnvironment: true,
             attempts: {
               take: 1,
               orderBy: { number: "desc" },
@@ -410,7 +405,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
               name: lockedTaskRun.queue,
             },
           },
@@ -436,12 +431,29 @@ export class SharedQueueConsumer {
             backgroundWorkerTaskId: backgroundTask.id,
             status: "PENDING" as const,
             queueId: queue.id,
-            runtimeEnvironmentId: environment.id,
+            runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
+          },
+          include: {
+            backgroundWorkerTask: true,
           },
         });
 
         const isRetry = taskRunAttempt.number > 1;
 
+        const { machineConfig } = taskRunAttempt.backgroundWorkerTask;
+        const machine = Machine.safeParse(machineConfig ?? {});
+
+        if (!machine.success) {
+          logger.error("Failed to parse machine config", {
+            queueMessage: message.data,
+            messageId: message.messageId,
+            attemptId: taskRunAttempt.id,
+            machineConfig,
+          });
+
+          await this.#ackAndDoMoreWork(message.messageId);
+          return;
+        }
         try {
           if (messageBody.data.checkpointEventId) {
             const restoreService = new RestoreCheckpointService();
@@ -470,11 +482,16 @@ export class SharedQueueConsumer {
               backgroundWorkerId: deployment.worker.friendlyId,
               data: {
                 type: "SCHEDULE_ATTEMPT",
-                id: taskRunAttempt.id,
                 image: deployment.imageReference,
-                envId: environment.id,
-                runId: taskRunAttempt.taskRunId,
                 version: deployment.version,
+                machine: machine.data,
+                // identifiers
+                id: taskRunAttempt.id,
+                envId: lockedTaskRun.runtimeEnvironment.id,
+                envType: lockedTaskRun.runtimeEnvironment.type,
+                orgId: lockedTaskRun.runtimeEnvironment.organizationId,
+                projectId: lockedTaskRun.runtimeEnvironment.projectId,
+                runId: taskRunAttempt.taskRunId,
               },
             });
           }
@@ -602,7 +619,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: resumableAttempt.runtimeEnvironmentId,
               name: resumableRun.queue,
             },
           },
@@ -731,10 +748,6 @@ export class SharedQueueConsumer {
     return;
   }
 
-  #envIdFromQueue(queueName: string) {
-    return queueName.split(":")[1];
-  }
-
   #doMoreWork(intervalInMs = this._options.interval) {
     setTimeout(() => this.#doWork(), intervalInMs);
   }

apps/webapp/app/v3/marqs/types.ts

@@ -0,0 +1,79 @@
+import { RedisOptions } from "ioredis";
+import { z } from "zod";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+
+export type QueueCapacity = {
+  current: number;
+  limit: number;
+};
+
+export type QueueCapacities = {
+  queue: QueueCapacity;
+  env: QueueCapacity;
+  org: QueueCapacity;
+};
+
+export type QueueWithScores = {
+  queue: string;
+  capacities: QueueCapacities;
+  age: number;
+};
+
+export interface MarQSKeyProducer {
+  queueConcurrencyLimitKey(env: AuthenticatedEnvironment, queue: string): string;
+  envConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
+  orgConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
+  queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string): string;
+  envSharedQueueKey(env: AuthenticatedEnvironment): string;
+  concurrencyLimitKeyFromQueue(queue: string): string;
+  currentConcurrencyKeyFromQueue(queue: string): string;
+  orgConcurrencyLimitKeyFromQueue(queue: string): string;
+  orgCurrentConcurrencyKeyFromQueue(queue: string): string;
+  envConcurrencyLimitKeyFromQueue(queue: string): string;
+  envCurrentConcurrencyKeyFromQueue(queue: string): string;
+  messageKey(messageId: string): string;
+}
+
+export type PriorityStrategyChoice = string | { abort: true };
+
+export interface MarQSQueuePriorityStrategy {
+  /**
+   * chooseQueue is called to select the next queue to process a message from
+   *
+   * @param queues
+   * @param parentQueue
+   * @param selectionId
+   *
+   * @returns The queue to process the message from, or an object with `abort: true` if no queue is available
+   */
+  chooseQueue(
+    queues: Array<QueueWithScores>,
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice;
+
+  /**
+   * This function is called to get the next candidate selection for the queue
+   * The `range` is used to select the set of queues that will be considered for the next selection (passed to chooseQueue)
+   * The `selectionId` is used to identify the selection and should be passed to chooseQueue
+   *
+   * @param parentQueue The parent queue that holds the candidate queues
+   *
+   * @returns The scores and the selectionId for the next candidate selection
+   */
+  nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }>;
+}
+
+export const MessagePayload = z.object({
+  version: z.literal("1"),
+  data: z.record(z.unknown()),
+  queue: z.string(),
+  messageId: z.string(),
+  timestamp: z.number(),
+  parentQueue: z.string(),
+  concurrencyKey: z.string().optional(),
+});
+
+export type MessagePayload = z.infer<typeof MessagePayload>;

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

@@ -259,6 +259,18 @@ export class RegistryProxy {
         proxyRes.pipe(response, { end: true });
       });
 
+      request.on("close", () => {
+        logger.debug("Client closed the connection");
+        proxyReq.destroy();
+        cleanupTempFile();
+      });
+
+      request.on("abort", () => {
+        logger.debug("Client aborted the connection");
+        proxyReq.destroy(); // Abort the proxied request
+        cleanupTempFile(); // Clean up the temporary file if necessary
+      });
+
       if (tempFilePath) {
         const readStream = createReadStream(tempFilePath);
 
@@ -427,14 +439,22 @@ function initializeProxy() {
   });
 }
 
-async function streamRequestBodyToTempFile(request: IncomingMessage): Promise<string> {
-  const tempDir = await mkdtemp(`${tmpdir()}/`);
-  const tempFilePath = `${tempDir}/requestBody.tmp`;
-  const writeStream = createWriteStream(tempFilePath);
+async function streamRequestBodyToTempFile(request: IncomingMessage): Promise<string | undefined> {
+  try {
+    const tempDir = await mkdtemp(`${tmpdir()}/`);
+    const tempFilePath = `${tempDir}/requestBody.tmp`;
+    const writeStream = createWriteStream(tempFilePath);
 
-  await pipeline(request, writeStream);
+    await pipeline(request, writeStream);
 
-  return tempFilePath;
+    return tempFilePath;
+  } catch (error) {
+    logger.error("Failed to stream request body to temp file", {
+      error: error instanceof Error ? error.message : error,
+    });
+
+    return;
+  }
 }
 
 type DockerImageParts = {

apps/webapp/app/v3/services/cancelAttempt.server.ts

@@ -1,6 +1,6 @@
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

apps/webapp/app/v3/services/cancelTaskRun.server.ts

@@ -1,6 +1,6 @@
-import { TaskRun, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
+import { Prisma, TaskRun, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { devPubSub } from "../marqs/devPubSub.server";
 import { BaseService } from "./baseService.server";
 import { socketIo } from "../handleSocketIo.server";
@@ -23,6 +23,13 @@ const CANCELLABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = [
   "PENDING",
 ];
 
+type ExtendedTaskRunAttempt = Prisma.TaskRunAttemptGetPayload<{
+  include: {
+    runtimeEnvironment: true;
+    backgroundWorker: true;
+  };
+}>;
+
 export type CancelTaskRunServiceOptions = {
   reason?: string;
   cancelAttempts?: boolean;
@@ -87,56 +94,60 @@ export class CancelTaskRunService extends BaseService {
 
     // Cancel any in progress attempts
     if (opts.cancelAttempts) {
-      for (const attempt of cancelledTaskRun.attempts) {
-        if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
-          // Signal the task run attempt to stop
-          await devPubSub.publish(
-            `backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
-            "CANCEL_ATTEMPT",
-            {
-              attemptId: attempt.friendlyId,
-              backgroundWorkerId: attempt.backgroundWorker.friendlyId,
-              taskRunId: cancelledTaskRun.friendlyId,
-            }
-          );
-        } else {
-          switch (attempt.status) {
-            case "EXECUTING": {
-              // We need to send a cancel message to the coordinator
-              socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
-                version: "v1",
-                attemptId: attempt.id,
-                attemptFriendlyId: attempt.friendlyId,
-              });
+      await this.#cancelPotentiallyRunningAttempts(cancelledTaskRun, cancelledTaskRun.attempts);
+    }
+  }
 
-              break;
-            }
-            case "PENDING":
-            case "PAUSED": {
-              logger.debug("Cancelling pending or paused attempt", {
-                attempt,
-              });
+  async #cancelPotentiallyRunningAttempts(run: TaskRun, attempts: ExtendedTaskRunAttempt[]) {
+    for (const attempt of attempts) {
+      if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
+        // Signal the task run attempt to stop
+        await devPubSub.publish(
+          `backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
+          "CANCEL_ATTEMPT",
+          {
+            attemptId: attempt.friendlyId,
+            backgroundWorkerId: attempt.backgroundWorker.friendlyId,
+            taskRunId: run.friendlyId,
+          }
+        );
+      } else {
+        switch (attempt.status) {
+          case "EXECUTING": {
+            // We need to send a cancel message to the coordinator
+            socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
+              version: "v1",
+              attemptId: attempt.id,
+              attemptFriendlyId: attempt.friendlyId,
+            });
 
-              const service = new CancelAttemptService();
+            break;
+          }
+          case "PENDING":
+          case "PAUSED": {
+            logger.debug("Cancelling pending or paused attempt", {
+              attempt,
+            });
 
-              await service.call(
-                attempt.friendlyId,
-                taskRun.id,
-                new Date(),
-                "Task run was cancelled by user"
-              );
+            const service = new CancelAttemptService();
 
-              break;
-            }
-            case "CANCELED":
-            case "COMPLETED":
-            case "FAILED": {
-              // Do nothing
-              break;
-            }
-            default: {
-              assertUnreachable(attempt.status);
-            }
+            await service.call(
+              attempt.friendlyId,
+              run.id,
+              new Date(),
+              "Task run was cancelled by user"
+            );
+
+            break;
+          }
+          case "CANCELED":
+          case "COMPLETED":
+          case "FAILED": {
+            // Do nothing
+            break;
+          }
+          default: {
+            assertUnreachable(attempt.status);
           }
         }
       }

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -12,7 +12,7 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { safeJsonParse } from "~/utils/json";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
 import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
@@ -92,10 +92,7 @@ export class CompleteAttemptService extends BaseService {
       },
     });
 
-    logger.debug("Completed attempt successfully, ACKing message", {
-      serializedOutput: completion.output,
-      outputType: completion.outputType,
-    });
+    logger.debug("Completed attempt successfully, ACKing message");
 
     await marqs?.acknowledgeMessage(taskRunAttempt.taskRunId);
 
@@ -105,7 +102,7 @@ export class CompleteAttemptService extends BaseService {
       attributes: {
         isError: false,
         output:
-          completion.outputType === "application/store"
+          completion.outputType === "application/store" || completion.outputType === "text/plain"
             ? completion.output
             : completion.output
             ? (safeJsonParse(completion.output) as Attributes)

apps/webapp/app/v3/services/crashTaskRun.server.ts

@@ -0,0 +1,157 @@
+import {
+  TaskRun,
+  TaskRunAttempt,
+  TaskRunAttemptStatus,
+  TaskRunStatus,
+} from "@trigger.dev/database";
+import { eventRepository } from "../eventRepository.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { BaseService } from "./baseService.server";
+import { logger } from "~/services/logger.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
+
+export const CRASHABLE_RUN_STATUSES: Array<TaskRunStatus> = [
+  "PENDING",
+  "EXECUTING",
+  "PAUSED",
+  "WAITING_TO_RESUME",
+  "PAUSED",
+  "RETRYING_AFTER_FAILURE",
+];
+
+const CRASHABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = ["EXECUTING", "PAUSED", "PENDING"];
+
+export type CrashTaskRunServiceOptions = {
+  reason?: string;
+  exitCode?: number;
+  logs?: string;
+  crashAttempts?: boolean;
+  crashedAt?: Date;
+};
+
+export class CrashTaskRunService extends BaseService {
+  public async call(runId: string, options?: CrashTaskRunServiceOptions) {
+    const opts = {
+      reason: "Worker crashed",
+      crashAttempts: true,
+      crashedAt: new Date(),
+      ...options,
+    };
+
+    const taskRun = await this._prisma.taskRun.findFirst({
+      where: {
+        id: runId,
+      },
+    });
+
+    if (!taskRun) {
+      logger.error("Task run not found", { runId });
+      return;
+    }
+
+    // Make sure the task run is in a crashable state
+    if (!CRASHABLE_RUN_STATUSES.includes(taskRun.status)) {
+      return;
+    }
+
+    // Remove the task run from the queue if it's there for some reason
+    await marqs?.acknowledgeMessage(taskRun.id);
+
+    // Set the task run status to crashed
+    const crashedTaskRun = await this._prisma.taskRun.update({
+      where: {
+        id: taskRun.id,
+      },
+      data: {
+        status: "CRASHED",
+      },
+      include: {
+        attempts: {
+          where: {
+            status: {
+              in: CRASHABLE_ATTEMPT_STATUSES,
+            },
+          },
+          include: {
+            backgroundWorker: true,
+            runtimeEnvironment: true,
+          },
+        },
+        dependency: true,
+        runtimeEnvironment: {
+          include: {
+            organization: true,
+            project: true,
+          },
+        },
+      },
+    });
+
+    const inProgressEvents = await eventRepository.queryIncompleteEvents({
+      runId: taskRun.friendlyId,
+    });
+
+    logger.debug("Crashing in-progress events", {
+      inProgressEvents: inProgressEvents.map((event) => event.id),
+    });
+
+    await Promise.all(
+      inProgressEvents.map((event) => {
+        return eventRepository.crashEvent({
+          event: event,
+          crashedAt: opts.crashedAt,
+          exception: {
+            type: "Worker crashed",
+            message: opts.reason,
+            stacktrace: opts.logs,
+          },
+        });
+      })
+    );
+
+    if (!opts.crashAttempts) {
+      return;
+    }
+
+    // Cancel any in progress attempts
+    for (const attempt of crashedTaskRun.attempts) {
+      await this.#failAttempt(
+        attempt,
+        crashedTaskRun,
+        new Date(),
+        crashedTaskRun.runtimeEnvironment
+      );
+    }
+  }
+
+  async #failAttempt(
+    attempt: TaskRunAttempt,
+    run: TaskRun,
+    failedAt: Date,
+    environment: AuthenticatedEnvironment
+  ) {
+    return await this.traceWithEnv("failAttempt()", environment, async (span) => {
+      span.setAttribute("taskRunId", run.id);
+      span.setAttribute("attemptId", attempt.id);
+
+      await marqs?.acknowledgeMessage(run.id);
+
+      await this._prisma.taskRunAttempt.update({
+        where: {
+          id: attempt.id,
+        },
+        data: {
+          status: "FAILED",
+          completedAt: failedAt,
+        },
+      });
+
+      if (environment.type === "DEVELOPMENT") {
+        return;
+      }
+
+      await ResumeTaskRunDependenciesService.enqueue(attempt.id, this._prisma);
+    });
+  }
+}

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -4,10 +4,11 @@ import { Prisma, PrismaClientOrTransaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion";
 import { BaseService } from "./baseService.server";
 import { projectPubSub } from "./projectPubSub.server";
+import { env } from "~/env.server";
 
 export class CreateBackgroundWorkerService extends BaseService {
   public async call(
@@ -68,14 +69,39 @@ export class CreateBackgroundWorkerService extends BaseService {
 
       await createBackgroundTasks(body.metadata.tasks, backgroundWorker, environment, this._prisma);
 
-      //send a notification that a new worker has been created
-      await projectPubSub.publish(`project:${project.id}:env:${environment.id}`, "WORKER_CREATED", {
-        environmentId: environment.id,
-        environmentType: environment.type,
-        createdAt: backgroundWorker.createdAt,
-        taskCount: body.metadata.tasks.length,
-        type: "local",
-      });
+      try {
+        //send a notification that a new worker has been created
+        await projectPubSub.publish(
+          `project:${project.id}:env:${environment.id}`,
+          "WORKER_CREATED",
+          {
+            environmentId: environment.id,
+            environmentType: environment.type,
+            createdAt: backgroundWorker.createdAt,
+            taskCount: body.metadata.tasks.length,
+            type: "local",
+          }
+        );
+
+        await marqs?.updateEnvConcurrencyLimits(environment);
+      } catch (err) {
+        logger.error(
+          "Error publishing WORKER_CREATED event or updating global concurrency limits",
+          {
+            error:
+              err instanceof Error
+                ? {
+                  name: err.name,
+                  message: err.message,
+                  stack: err.stack,
+                }
+                : err,
+            project,
+            environment,
+            backgroundWorker,
+          }
+        );
+      }
 
       return backgroundWorker;
     });
@@ -85,7 +111,7 @@ export class CreateBackgroundWorkerService extends BaseService {
 export async function createBackgroundTasks(
   tasks: TaskResource[],
   worker: BackgroundWorker,
-  env: AuthenticatedEnvironment,
+  environment: AuthenticatedEnvironment,
   prisma: PrismaClientOrTransaction
 ) {
   for (const task of tasks) {
@@ -101,10 +127,28 @@ export async function createBackgroundTasks(
           exportName: task.exportName,
           retryConfig: task.retry,
           queueConfig: task.queue,
+          machineConfig: task.machine,
         },
       });
 
-      const queueName = task.queue?.name ?? `task/${task.id}`;
+      let queueName = sanitizeQueueName(task.queue?.name ?? `task/${task.id}`);
+
+      // Check that the queuename is not an empty string
+      if (!queueName) {
+        queueName = sanitizeQueueName(`task/${task.id}`);
+      }
+
+      const concurrencyLimit =
+        typeof task.queue?.concurrencyLimit === "number"
+          ? Math.max(
+            Math.min(
+              task.queue.concurrencyLimit,
+              environment.maximumConcurrencyLimit,
+              environment.organization.maximumConcurrencyLimit
+            ),
+            0
+          )
+          : null;
 
       const taskQueue = await prisma.taskQueue.upsert({
         where: {
@@ -114,13 +158,13 @@ export async function createBackgroundTasks(
           },
         },
         update: {
-          concurrencyLimit: task.queue?.concurrencyLimit,
+          concurrencyLimit,
           rateLimit: task.queue?.rateLimit,
         },
         create: {
           friendlyId: generateFriendlyId("queue"),
           name: queueName,
-          concurrencyLimit: task.queue?.concurrencyLimit,
+          concurrencyLimit,
           runtimeEnvironmentId: worker.runtimeEnvironmentId,
           projectId: worker.projectId,
           rateLimit: task.queue?.rateLimit,
@@ -129,7 +173,11 @@ export async function createBackgroundTasks(
       });
 
       if (taskQueue.concurrencyLimit) {
-        await marqs?.updateQueueConcurrency(env, taskQueue.name, taskQueue.concurrencyLimit);
+        await marqs?.updateQueueConcurrencyLimits(
+          environment,
+          taskQueue.name,
+          taskQueue.concurrencyLimit
+        );
       }
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError) {

apps/webapp/app/v3/services/createCheckpoint.server.ts

@@ -6,7 +6,7 @@ import type {
 } from "@trigger.dev/database";
 import { logger } from "~/services/logger.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server";
 import { BaseService } from "./baseService.server";
 

apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts

@@ -6,6 +6,8 @@ import { BaseService } from "./baseService.server";
 import { createBackgroundTasks } from "./createBackgroundWorker.server";
 import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
 import { projectPubSub } from "./projectPubSub.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { logger } from "~/services/logger.server";
 
 export class CreateDeployedBackgroundWorkerService extends BaseService {
   public async call(
@@ -76,18 +78,23 @@ export class CreateDeployedBackgroundWorkerService extends BaseService {
         },
       });
 
-      //send a notification that a new worker has been created
-      await projectPubSub.publish(
-        `project:${environment.projectId}:env:${environment.id}`,
-        "WORKER_CREATED",
-        {
-          environmentId: environment.id,
-          environmentType: environment.type,
-          createdAt: backgroundWorker.createdAt,
-          taskCount: body.metadata.tasks.length,
-          type: "deployed",
-        }
-      );
+      try {
+        //send a notification that a new worker has been created
+        await projectPubSub.publish(
+          `project:${environment.projectId}:env:${environment.id}`,
+          "WORKER_CREATED",
+          {
+            environmentId: environment.id,
+            environmentType: environment.type,
+            createdAt: backgroundWorker.createdAt,
+            taskCount: body.metadata.tasks.length,
+            type: "deployed",
+          }
+        );
+        await marqs?.updateEnvConcurrencyLimits(environment);
+      } catch (err) {
+        logger.error("Failed to publish WORKER_CREATED event", { err });
+      }
 
       return backgroundWorker;
     });

apps/webapp/app/v3/services/deploymentIndexFailed.server.ts

@@ -2,13 +2,19 @@ import { BaseService } from "./baseService.server";
 
 export class DeploymentIndexFailed extends BaseService {
   public async call(
-    deploymentId: string,
+    maybeFriendlyId: string,
     error: { name: string; message: string; stack?: string }
   ) {
+    const isFriendlyId = maybeFriendlyId.startsWith("deployment_");
+
     const deployment = await this._prisma.workerDeployment.update({
-      where: {
-        friendlyId: deploymentId,
-      },
+      where: isFriendlyId
+        ? {
+            friendlyId: maybeFriendlyId,
+          }
+        : {
+            id: maybeFriendlyId,
+          },
       data: {
         status: "FAILED",
         failedAt: new Date(),

apps/webapp/app/v3/services/indexDeployment.server.ts

@@ -49,9 +49,14 @@ export class IndexDeploymentService extends BaseService {
         version: "v1",
         shortCode: deployment.shortCode,
         imageTag: deployment.imageReference,
-        envId: deployment.environmentId,
         apiKey: deployment.environment.apiKey,
         apiUrl: env.APP_ORIGIN,
+        // identifiers
+        envId: deployment.environmentId,
+        envType: deployment.environment.type,
+        projectId: deployment.projectId,
+        orgId: deployment.environment.organizationId,
+        deploymentId: deployment.id,
       });
 
       logger.debug("Index ACK received", { responses });

apps/webapp/app/v3/services/replayTaskRun.server.ts

@@ -0,0 +1,53 @@
+import { conditionallyImportPacket, parsePacket } from "@trigger.dev/core/v3";
+import { Prisma, TaskRun } from "@trigger.dev/database";
+import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
+import { logger } from "~/services/logger.server";
+import { BaseService } from "./baseService.server";
+import { TriggerTaskService } from "./triggerTask.server";
+
+type ExtendedTaskRunAttempt = Prisma.TaskRunAttemptGetPayload<{
+  include: {
+    runtimeEnvironment: true;
+    backgroundWorker: true;
+  };
+}>;
+
+export class ReplayTaskRunService extends BaseService {
+  public async call(existingTaskRun: TaskRun) {
+    const authenticatedEnvironment = await findEnvironmentById(
+      existingTaskRun.runtimeEnvironmentId
+    );
+    if (!authenticatedEnvironment) {
+      return;
+    }
+
+    logger.info("Replaying task run", {
+      taskRunId: existingTaskRun.id,
+      taskRunFriendlyId: existingTaskRun.friendlyId,
+    });
+
+    const payloadPacket = await conditionallyImportPacket({
+      data: existingTaskRun.payload,
+      dataType: existingTaskRun.payloadType,
+    });
+    const parsedPayload = await parsePacket(payloadPacket);
+
+    logger.info("Replaying task run payload", {
+      taskRunId: existingTaskRun.id,
+      taskRunFriendlyId: existingTaskRun.friendlyId,
+      payloadPacketType: payloadPacket.dataType,
+    });
+
+    const triggerTaskService = new TriggerTaskService();
+    return await triggerTaskService.call(existingTaskRun.taskIdentifier, authenticatedEnvironment, {
+      payload: parsedPayload,
+      options: {
+        queue: {
+          name: existingTaskRun.queue,
+        },
+        concurrencyKey: existingTaskRun.concurrencyKey ?? undefined,
+        test: existingTaskRun.isTest,
+      },
+    });
+  }
+}

apps/webapp/app/v3/services/restoreCheckpoint.server.ts

@@ -3,6 +3,7 @@ import { logger } from "~/services/logger.server";
 import { socketIo } from "../handleSocketIo.server";
 import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server";
 import { BaseService } from "./baseService.server";
+import { Machine } from "@trigger.dev/core/v3";
 
 const RESTORABLE_RUN_STATUSES: TaskRunStatus[] = ["WAITING_TO_RESUME"];
 const RESTORABLE_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["PAUSED"];
@@ -30,8 +31,14 @@ export class RestoreCheckpointService extends BaseService {
             attempt: {
               select: {
                 status: true,
+                backgroundWorkerTask: {
+                  select: {
+                    machineConfig: true,
+                  },
+                },
               },
             },
+            runtimeEnvironment: true,
           },
         },
       },
@@ -63,17 +70,34 @@ export class RestoreCheckpointService extends BaseService {
       return;
     }
 
+    const { machineConfig } = checkpoint.attempt.backgroundWorkerTask;
+    const machine = Machine.safeParse(machineConfig ?? {});
+
+    if (!machine.success) {
+      logger.error("Failed to parse machine config", {
+        attemptId: checkpoint.attemptId,
+        machineConfig: checkpoint.attempt.backgroundWorkerTask.machineConfig,
+      });
+      return;
+    }
+
     const eventService = new CreateCheckpointRestoreEventService(this._prisma);
     await eventService.restore({ checkpointId: checkpoint.id });
 
     socketIo.providerNamespace.emit("RESTORE", {
       version: "v1",
-      checkpointId: checkpoint.id,
-      runId: checkpoint.runId,
       type: checkpoint.type,
       location: checkpoint.location,
       reason: checkpoint.reason ?? undefined,
       imageRef: checkpoint.imageRef,
+      machine: machine.data,
+      // identifiers
+      checkpointId: checkpoint.id,
+      envId: checkpoint.runtimeEnvironment.id,
+      envType: checkpoint.runtimeEnvironment.type,
+      orgId: checkpoint.runtimeEnvironment.organizationId,
+      projectId: checkpoint.runtimeEnvironment.projectId,
+      runId: checkpoint.runId,
     });
 
     return checkpoint;

apps/webapp/app/v3/services/resumeAttempt.server.ts

@@ -6,7 +6,7 @@ import {
 } from "@trigger.dev/core/v3";
 import { $transaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { socketIo } from "../handleSocketIo.server";
 import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
 import { BaseService } from "./baseService.server";

apps/webapp/app/v3/services/resumeBatchRun.server.ts

@@ -1,6 +1,6 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { workerQueue } from "~/services/worker.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

apps/webapp/app/v3/services/resumeTaskDependency.server.ts

@@ -1,6 +1,6 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { workerQueue } from "~/services/worker.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -1,5 +1,4 @@
 import {
-  PRIMARY_VARIANT,
   SemanticInternalAttributes,
   TriggerTaskRequestBody,
   packetRequiresOffloading,
@@ -10,10 +9,9 @@ import { $transaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { eventRepository } from "../eventRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
-import { BaseService } from "./baseService.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { uploadToObjectStore } from "../r2.server";
-import { logger } from "~/services/logger.server";
+import { BaseService } from "./baseService.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;

apps/webapp/app/v3/sharedSocketConnection.ts

@@ -10,7 +10,56 @@ import { Evt } from "evt";
 import { randomUUID } from "node:crypto";
 import { logger } from "~/services/logger.server";
 import { SharedQueueConsumer } from "./marqs/sharedQueueConsumer.server";
-import { DisconnectReason, Namespace, Socket } from "socket.io";
+import type { DisconnectReason, Namespace, Socket } from "socket.io";
+import { ROOT_CONTEXT, Span, SpanKind, trace } from "@opentelemetry/api";
+import { env } from "~/env.server";
+
+const tracer = trace.getTracer("sharedQueueConsumerPool");
+
+interface SharedQueueConsumerPoolOptions {
+  sender: ZodMessageSender<typeof serverWebsocketMessages>;
+  poolSize: number;
+}
+
+class SharedQueueConsumerPool {
+  #consumers: SharedQueueConsumer[];
+  #span: Span;
+
+  constructor(opts: SharedQueueConsumerPoolOptions) {
+    this.#span = tracer.startSpan(
+      "SharedQueueConsumerPool()",
+      {
+        kind: SpanKind.CONSUMER,
+        attributes: {
+          "pool.size": opts.poolSize,
+        },
+      },
+      ROOT_CONTEXT
+    );
+
+    const spanContext = trace.setSpan(ROOT_CONTEXT, this.#span);
+
+    this.#consumers = Array(opts.poolSize)
+      .fill(null)
+      .map(
+        () =>
+          new SharedQueueConsumer(opts.sender, {
+            interval: env.SHARED_QUEUE_CONSUMER_INTERVAL_MS,
+            nextTickInterval: env.SHARED_QUEUE_CONSUMER_NEXT_TICK_INTERVAL_MS,
+            parentContext: spanContext,
+          })
+      );
+  }
+
+  async start() {
+    await Promise.allSettled(this.#consumers.map((consumer) => consumer.start()));
+  }
+
+  async stop() {
+    await Promise.allSettled(this.#consumers.map((consumer) => consumer.stop()));
+    this.#span.end();
+  }
+}
 
 interface SharedSocketConnectionOptions {
   namespace: Namespace<
@@ -22,6 +71,7 @@ interface SharedSocketConnectionOptions {
     MessageCatalogToSocketIoEvents<typeof serverWebsocketMessages>
   >;
   logger?: StructuredLogger;
+  poolSize?: number;
 }
 
 export class SharedSocketConnection {
@@ -29,8 +79,9 @@ export class SharedSocketConnection {
   public onClose: Evt<DisconnectReason> = new Evt();
 
   private _sender: ZodMessageSender<typeof serverWebsocketMessages>;
-  private _sharedConsumer: SharedQueueConsumer;
+  private _sharedQueueConsumerPool: SharedQueueConsumerPool;
   private _messageHandler: ZodMessageHandler<typeof clientWebsocketMessages>;
+  private _defaultPoolSize = 10;
 
   constructor(opts: SharedSocketConnectionOptions) {
     this.id = randomUUID();
@@ -50,9 +101,13 @@ export class SharedSocketConnection {
       },
     });
 
-    this._sharedConsumer = new SharedQueueConsumer(this._sender, {
-      interval: 100,
-      nextTickInterval: 1000,
+    logger.log("Starting SharedQueueConsumer pool", {
+      poolSize: opts.poolSize ?? this._defaultPoolSize,
+    });
+
+    this._sharedQueueConsumerPool = new SharedQueueConsumerPool({
+      poolSize: opts.poolSize ?? this._defaultPoolSize,
+      sender: this._sender,
     });
 
     opts.socket.on("disconnect", this.#handleClose.bind(this));
@@ -62,7 +117,7 @@ export class SharedSocketConnection {
       schema: clientWebsocketMessages,
       messages: {
         READY_FOR_TASKS: async (payload) => {
-          this._sharedConsumer.start();
+          this._sharedQueueConsumerPool.start();
         },
         BACKGROUND_WORKER_DEPRECATED: async (payload) => {
           // await this._sharedConsumer.deprecateBackgroundWorker(payload.backgroundWorkerId);
@@ -89,7 +144,7 @@ export class SharedSocketConnection {
   }
 
   async #handleClose(ev: DisconnectReason) {
-    await this._sharedConsumer.stop();
+    await this._sharedQueueConsumerPool.stop();
 
     this.onClose.post(ev);
   }

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

@@ -9,7 +9,7 @@ import {
   trace,
 } from "@opentelemetry/api";
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
-import { registerInstrumentations } from "@opentelemetry/instrumentation";
+import { InstrumentationOption, registerInstrumentations } from "@opentelemetry/instrumentation";
 import { ExpressInstrumentation } from "@opentelemetry/instrumentation-express";
 import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";
 import { Resource } from "@opentelemetry/resources";
@@ -23,12 +23,15 @@ import {
   TraceIdRatioBasedSampler,
 } from "@opentelemetry/sdk-trace-base";
 import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
-import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
+import { SEMRESATTRS_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
 import { PrismaInstrumentation } from "@prisma/instrumentation";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { singleton } from "~/utils/singleton";
 import { LoggerSpanExporter } from "./telemetry/loggerExporter.server";
+
+export const SEMINTATTRS_FORCE_RECORDING = "forceRecording";
+
 class CustomWebappSampler implements Sampler {
   constructor(private readonly _baseSampler: Sampler) {}
 
@@ -44,15 +47,26 @@ class CustomWebappSampler implements Sampler {
     const parentContext = trace.getSpanContext(context);
 
     // Exclude Prisma spans (adjust this logic as needed for your use case)
-    if (
-      !parentContext &&
-      ((attributes && attributes["model"] && attributes["method"]) || name.includes("prisma"))
-    ) {
+    if (!parentContext && name.includes("prisma")) {
       return { decision: SamplingDecision.NOT_RECORD };
     }
 
+    // If the span has the forceRecording attribute, always record it
+    if (attributes[SEMINTATTRS_FORCE_RECORDING]) {
+      return { decision: SamplingDecision.RECORD_AND_SAMPLED };
+    }
+
     // For all other spans, defer to the base sampler
-    return this._baseSampler.shouldSample(context, traceId, name, spanKind, attributes, links);
+    const result = this._baseSampler.shouldSample(
+      context,
+      traceId,
+      name,
+      spanKind,
+      attributes,
+      links
+    );
+
+    return result;
   }
 
   toString(): string {
@@ -65,29 +79,48 @@ export const tracer = singleton("tracer", getTracer);
 function getTracer() {
   diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ERROR);
 
+  const samplingRate = 1.0 / Math.max(parseInt(env.INTERNAL_OTEL_TRACE_SAMPLING_RATE, 10), 1);
+
   const provider = new NodeTracerProvider({
     forceFlushTimeoutMillis: 500,
     resource: new Resource({
-      [SemanticResourceAttributes.SERVICE_NAME]: "trigger.dev",
+      [SEMRESATTRS_SERVICE_NAME]: env.SERVICE_NAME,
     }),
     sampler: new ParentBasedSampler({
-      root: new CustomWebappSampler(
-        new TraceIdRatioBasedSampler(env.APP_ENV === "development" ? 1.0 : 0.05)
-      ), // 5% sampling
-    }), // 5% sampling
+      root: new CustomWebappSampler(new TraceIdRatioBasedSampler(samplingRate)),
+    }),
   });
 
-  if (env.OTLP_EXPORTER_TRACES_URL) {
+  if (env.INTERNAL_OTEL_TRACE_EXPORTER_URL) {
     const exporter = new OTLPTraceExporter({
-      url: env.OTLP_EXPORTER_TRACES_URL,
+      url: env.INTERNAL_OTEL_TRACE_EXPORTER_URL,
       timeoutMillis: 1000,
+      headers:
+        env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_NAME &&
+        env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_VALUE
+          ? {
+              [env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_NAME]:
+                env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_VALUE,
+            }
+          : undefined,
     });
 
-    provider.addSpanProcessor(new BatchSpanProcessor(exporter));
+    provider.addSpanProcessor(
+      new BatchSpanProcessor(exporter, {
+        maxExportBatchSize: 512,
+        scheduledDelayMillis: 1000,
+        exportTimeoutMillis: 30000,
+        maxQueueSize: 2048,
+      })
+    );
 
-    console.log(`⚡ Tracer: OTLP exporter enabled to ${env.OTLP_EXPORTER_TRACES_URL}`);
+    console.log(
+      `🔦 Tracer: OTLP exporter enabled to ${env.INTERNAL_OTEL_TRACE_EXPORTER_URL} (sampling = ${samplingRate})`
+    );
   } else {
-    if (env.LOG_TELEMETRY === "true") {
+    if (env.INTERNAL_OTEL_TRACE_LOGGING_ENABLED === "1") {
+      console.log(`🔦 Tracer: Logger exporter enabled`);
+
       const loggerExporter = new LoggerSpanExporter();
 
       provider.addSpanProcessor(new SimpleSpanProcessor(loggerExporter));
@@ -96,13 +129,18 @@ function getTracer() {
 
   provider.register();
 
+  let instrumentations: InstrumentationOption[] = [
+    new HttpInstrumentation(),
+    new ExpressInstrumentation(),
+  ];
+
+  if (env.INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED === "1") {
+    instrumentations.push(new PrismaInstrumentation());
+  }
+
   registerInstrumentations({
     tracerProvider: provider,
-    instrumentations: [
-      new HttpInstrumentation(),
-      new ExpressInstrumentation(),
-      new PrismaInstrumentation(),
-    ],
+    instrumentations,
   });
 
   return provider.getTracer("trigger.dev", "3.0.0.dp.1");

apps/webapp/package.json

@@ -19,7 +19,8 @@
     "generate:sourcemaps": "remix build --sourcemap",
     "clean:sourcemaps": "run-s clean:sourcemaps:*",
     "clean:sourcemaps:public": "rimraf ./build/**/*.map",
-    "clean:sourcemaps:build": "rimraf ./public/build/**/*.map"
+    "clean:sourcemaps:build": "rimraf ./public/build/**/*.map",
+    "test": "vitest"
   },
   "eslintIgnore": [
     "/node_modules",
@@ -140,6 +141,7 @@
     "remix-auth-github": "^1.6.0",
     "remix-typedjson": "0.3.1",
     "remix-utils": "^7.1.0",
+    "seedrandom": "^3.0.5",
     "semver": "^7.5.0",
     "simple-oauth2": "^5.0.0",
     "simplur": "^3.0.1",
@@ -186,6 +188,7 @@
     "@types/react": "18.2.69",
     "@types/react-collapse": "^5.0.4",
     "@types/react-dom": "18.2.7",
+    "@types/seedrandom": "^3.0.8",
     "@types/semver": "^7.3.13",
     "@types/simple-oauth2": "^5.0.4",
     "@types/slug": "^5.0.3",
@@ -214,7 +217,8 @@
     "tailwindcss": "3.4.1",
     "ts-node": "^10.7.0",
     "tsconfig-paths": "^3.14.1",
-    "typescript": "^5.1.6"
+    "typescript": "^5.1.6",
+    "vitest": "^1.4.0"
   },
   "engines": {
     "node": ">=16.0.0"

apps/webapp/server.ts

@@ -120,6 +120,10 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
       )}`
     );
 
+    socket.on("error", (err) => {
+      console.error("Connection upgrade error:", err);
+    });
+
     const url = new URL(req.url ?? "", "http://localhost");
 
     // Upgrade socket.io connection
@@ -133,6 +137,7 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
 
     // Only upgrade the connecting if the path is `/ws`
     if (url.pathname !== "/ws") {
+      // Setting the socket.destroy() error param causes an error event to be emitted which needs to be handled with socket.on("error") to prevent uncaught exceptions.
       socket.destroy(
         new Error(
           "Cannot connect because of invalid path: Please include `/ws` in the path of your upgrade request."

apps/webapp/test/marqs.test.ts

@@ -0,0 +1,263 @@
+import { SimpleWeightedChoiceStrategy } from "../app/v3/marqs/priorityStrategy.server";
+
+describe("SimpleWeightedChoiceStrategy", () => {
+  it("should use a weighted random choice algorithm to choose a queue", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual("queue3");
+  });
+
+  it("should filter out queues if any capacity is full", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 10, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 10, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection).toEqual({ range: [3, 6], selectionId: expect.any(String) });
+
+    // Now pass some queues that have some capacity
+    const chosenQueue2 = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue2).toEqual("queue3");
+
+    const nextSelection2 = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection2).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+
+  it("should adjust the next filter range only if passed the maximum number of queues", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 10, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+
+  it("should adjust the next candidate range ONLY for the matching parent queue", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue2");
+
+    expect(nextSelection).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+
+    const nextSelection2 = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection2).toEqual({ range: [3, 6], selectionId: expect.any(String) });
+
+    const chosenQueue2 = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue2",
+      "selectionId"
+    );
+
+    expect(chosenQueue2).toEqual("queue3");
+
+    const nextSelection3 = await stategy.nextCandidateSelection("parentQueue2");
+
+    expect(nextSelection3).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+});