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/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-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/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,15 +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",
     "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"
+    "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/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/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

.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/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";
@@ -334,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);
@@ -356,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");
@@ -629,7 +659,7 @@ class TaskCoordinator {
             return;
           }
 
-          if (completion.retry.delay < 10_000) {
+          if (completion.retry.delay < this.#delayThresholdInMs) {
             completeWithoutCheckpoint(false);
             return;
           }

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

@@ -9,6 +9,7 @@ import {
 } 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";
@@ -24,6 +25,10 @@ type Namespace = {
   };
 };
 
+type ComputeResources = {
+  [K in "cpu" | "memory" | "ephemeral-storage"]?: string;
+};
+
 class KubernetesTaskOperations implements TaskOperations {
   #namespace: Namespace;
   #k8sApi: {
@@ -55,25 +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",
-                env: opts.envId,
-                envtype: this.#envTypeToLabelValue(opts.envType),
-                org: opts.orgId,
-                project: opts.projectId,
+                deployment: opts.deploymentId,
               },
             },
             spec: {
-              restartPolicy: "Never",
-              imagePullSecrets: [
-                {
-                  name: "registry-trigger",
-                },
-              ],
-              nodeSelector: {
-                nodetype: "worker",
-              },
+              ...this.#defaultPodSpec,
               containers: [
                 {
                   name: this.#getIndexContainerName(opts.shortCode),
@@ -83,12 +78,13 @@ class KubernetesTaskOperations implements TaskOperations {
                       containerPort: 8000,
                     },
                   ],
-                  // resources: {
-                  //   limits: {
-                  //     cpu: "100m",
-                  //     memory: "50Mi",
-                  //   },
-                  // },
+                  resources: {
+                    limits: {
+                      cpu: "250m",
+                      memory: "0.5G",
+                      "ephemeral-storage": "2Gi",
+                    },
+                  },
                   lifecycle: {
                     preStop: {
                       exec: {
@@ -97,10 +93,7 @@ class KubernetesTaskOperations implements TaskOperations {
                     },
                   },
                   env: [
-                    {
-                      name: "DEBUG",
-                      value: "true",
-                    },
+                    ...this.#getSharedEnv(opts.envId),
                     {
                       name: "INDEX_TASKS",
                       value: "true",
@@ -113,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",
-                        },
-                      },
-                    },
                   ],
                 },
               ],
@@ -167,26 +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",
-            env: opts.envId,
-            envtype: this.#envTypeToLabelValue(opts.envType),
-            org: opts.orgId,
-            project: opts.projectId,
             run: opts.runId,
           },
         },
         spec: {
-          restartPolicy: "Never",
-          imagePullSecrets: [
-            {
-              name: "registry-trigger",
-            },
-          ],
-          nodeSelector: {
-            nodetype: "worker",
-          },
+          ...this.#defaultPodSpec,
           containers: [
             {
               name: this.#getRunContainerName(opts.runId),
@@ -197,14 +143,15 @@ class KubernetesTaskOperations implements TaskOperations {
                 },
               ],
               resources: {
-                limits: this.#getResourcesFromMachineConfig(opts.machine),
+                requests: {
+                  ...this.#defaultResourceRequests,
+                },
+                limits: {
+                  ...this.#defaultResourceLimits,
+                  ...this.#getResourcesFromMachineConfig(opts.machine),
+                },
               },
               lifecycle: {
-                postStart: {
-                  exec: {
-                    command: this.#getLifecycleCommand("postStart", "create"),
-                  },
-                },
                 preStop: {
                   exec: {
                     command: this.#getLifecycleCommand("preStop", "terminate"),
@@ -212,54 +159,11 @@ class KubernetesTaskOperations implements TaskOperations {
                 },
               },
               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: [
                 {
@@ -288,27 +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",
-            env: opts.envId,
-            envtype: this.#envTypeToLabelValue(opts.envType),
-            org: opts.orgId,
-            project: opts.projectId,
             run: opts.runId,
             checkpoint: opts.checkpointId,
           },
         },
         spec: {
-          restartPolicy: "Never",
-          imagePullSecrets: [
-            {
-              name: "registry-trigger",
-            },
-          ],
-          nodeSelector: {
-            nodetype: "worker",
-          },
+          ...this.#defaultPodSpec,
           initContainers: [
             {
               name: "pull-base-image",
@@ -348,7 +241,13 @@ class KubernetesTaskOperations implements TaskOperations {
                 },
               ],
               resources: {
-                limits: this.#getResourcesFromMachineConfig(opts.machine),
+                requests: {
+                  ...this.#defaultResourceRequests,
+                },
+                limits: {
+                  ...this.#defaultResourceLimits,
+                  ...this.#getResourcesFromMachineConfig(opts.machine),
+                },
               },
               lifecycle: {
                 postStart: {
@@ -406,7 +305,90 @@ class KubernetesTaskOperations implements TaskOperations {
     }
   }
 
-  #getResourcesFromMachineConfig(config: Machine) {
+  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`,
@@ -417,7 +399,19 @@ class KubernetesTaskOperations implements TaskOperations {
     type: THookType,
     cause: THookType extends "postStart" ? PostStartCauses : PreStopCauses
   ) {
-    return ["/bin/sh", "-c", `sleep 1; wget -q -O- 127.0.0.1:8000/${type}?cause=${cause}`];
+    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) {
@@ -516,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>
@@ -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/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,8 +88,6 @@ const EnvironmentSchema = z.object({
 
   //v3
   V3_ENABLED: z.string().default("false"),
-  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(),
@@ -101,6 +102,9 @@ const EnvironmentSchema = z.object({
   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(),
@@ -114,6 +118,17 @@ const EnvironmentSchema = z.object({
   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
@@ -122,7 +137,7 @@ const EnvironmentSchema = z.object({
   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_SAMPING_RATE: z.string().default("20"),
+  INTERNAL_OTEL_TRACE_SAMPLING_RATE: z.string().default("20"),
   INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED: z.string().default("0"),
 });
 

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/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.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.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,

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/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -477,6 +477,46 @@ export class EnvironmentVariablesRepository implements Repository {
         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,
@@ -255,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

@@ -20,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(),
@@ -57,6 +61,7 @@ export type SharedQueueConsumerOptions = {
   traceTimeoutSeconds?: number;
   nextTickInterval?: number;
   interval?: number;
+  parentContext?: Context;
 };
 
 export class SharedQueueConsumer {
@@ -83,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,
     };
   }
 
@@ -139,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) {
@@ -187,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;
     }
 
@@ -199,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(
@@ -212,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();
@@ -251,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);
@@ -381,6 +374,7 @@ export class SharedQueueConsumer {
             lockedById: backgroundTask.id,
           },
           include: {
+            runtimeEnvironment: true,
             attempts: {
               take: 1,
               orderBy: { number: "desc" },
@@ -411,7 +405,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
               name: lockedTaskRun.queue,
             },
           },
@@ -437,7 +431,7 @@ export class SharedQueueConsumer {
             backgroundWorkerTaskId: backgroundTask.id,
             status: "PENDING" as const,
             queueId: queue.id,
-            runtimeEnvironmentId: environment.id,
+            runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
           },
           include: {
             backgroundWorkerTask: true,
@@ -493,10 +487,10 @@ export class SharedQueueConsumer {
                 machine: machine.data,
                 // identifiers
                 id: taskRunAttempt.id,
-                envId: environment.id,
-                envType: environment.type,
-                orgId: environment.organizationId,
-                projectId: environment.projectId,
+                envId: lockedTaskRun.runtimeEnvironment.id,
+                envType: lockedTaskRun.runtimeEnvironment.type,
+                orgId: lockedTaskRun.runtimeEnvironment.organizationId,
+                projectId: lockedTaskRun.runtimeEnvironment.projectId,
                 runId: taskRunAttempt.taskRunId,
               },
             });
@@ -625,7 +619,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: resumableAttempt.runtimeEnvironmentId,
               name: resumableRun.queue,
             },
           },
@@ -754,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/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);
 

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) {
@@ -105,7 +131,24 @@ export async function createBackgroundTasks(
         },
       });
 
-      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: {
@@ -115,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,
@@ -130,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

@@ -56,6 +56,7 @@ export class IndexDeploymentService extends BaseService {
         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/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

@@ -30,6 +30,8 @@ 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) {}
 
@@ -49,8 +51,22 @@ class CustomWebappSampler implements Sampler {
       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 {
@@ -63,16 +79,16 @@ 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_SAMPING_RATE, 10), 1);
+  const samplingRate = 1.0 / Math.max(parseInt(env.INTERNAL_OTEL_TRACE_SAMPLING_RATE, 10), 1);
 
   const provider = new NodeTracerProvider({
     forceFlushTimeoutMillis: 500,
     resource: new Resource({
-      [SEMRESATTRS_SERVICE_NAME]: "trigger.dev",
+      [SEMRESATTRS_SERVICE_NAME]: env.SERVICE_NAME,
     }),
     sampler: new ParentBasedSampler({
-      root: new CustomWebappSampler(new TraceIdRatioBasedSampler(samplingRate)), // 5% sampling
-    }), // 5% sampling
+      root: new CustomWebappSampler(new TraceIdRatioBasedSampler(samplingRate)),
+    }),
   });
 
   if (env.INTERNAL_OTEL_TRACE_EXPORTER_URL) {
@@ -92,13 +108,15 @@ function getTracer() {
     provider.addSpanProcessor(
       new BatchSpanProcessor(exporter, {
         maxExportBatchSize: 512,
-        scheduledDelayMillis: 200,
+        scheduledDelayMillis: 1000,
         exportTimeoutMillis: 30000,
         maxQueueSize: 2048,
       })
     );
 
-    console.log(`🔦 Tracer: OTLP exporter enabled to ${env.INTERNAL_OTEL_TRACE_EXPORTER_URL}`);
+    console.log(
+      `🔦 Tracer: OTLP exporter enabled to ${env.INTERNAL_OTEL_TRACE_EXPORTER_URL} (sampling = ${samplingRate})`
+    );
   } else {
     if (env.INTERNAL_OTEL_TRACE_LOGGING_ENABLED === "1") {
       console.log(`🔦 Tracer: Logger exporter enabled`);

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) });
+  });
+});

apps/webapp/tsconfig.json

@@ -2,6 +2,7 @@
   "exclude": ["./cypress", "./cypress.config.ts"],
   "include": ["remix.env.d.ts", "**/*.ts", "**/*.tsx"],
   "compilerOptions": {
+    "types": ["vitest/globals"],
     "lib": ["DOM", "DOM.Iterable", "ES2019"],
     "isolatedModules": true,
     "esModuleInterop": true,

apps/webapp/vitest.config.ts

@@ -0,0 +1,8 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+  test: {
+    include: ["test/**/*.test.ts"],
+    globals: true,
+  },
+});

docs/_snippets/quickstart-whats-next.mdx

@@ -11,7 +11,7 @@
   >
     Learn more about how Trigger.dev works and how it can help you.
   </Card>
-  <Card title="Examples" icon="slot-machine" href="https:/trigger.dev/apis">
+  <Card title="Examples" icon="slot-machine" href="https://trigger.dev/apis">
     Find code examples for many popular APIs. These can be copied / modified for use in your own
     projects.
   </Card>

docs/_snippets/v3/step-cli-dev.mdx

@@ -0,0 +1,21 @@
+<Step title="Run the CLI `dev` command">
+
+The CLI `dev` command runs a server for your tasks. It will watches for changes in your `/trigger` directory and communicates with the Trigger.dev platform to register your tasks, perform runs, and send data back and forth.
+
+<CodeGroup>
+
+```bash npm
+npx trigger.dev@beta dev
+```
+
+```bash pnpm
+pnpm dlx trigger.dev@beta dev
+```
+
+```bash yarn
+yarn dlx trigger.dev@beta dev
+```
+
+</CodeGroup>
+
+</Step>

docs/_snippets/v3/step-cli-init.mdx

@@ -0,0 +1,30 @@
+<Step title="Run the CLI `init` command">
+
+The easiest way to get started it to use the CLI. It will add Trigger.dev to your existing project, create a `/trigger` folder and give you an example task.
+
+Run this command in the root of your project to get started:
+
+<CodeGroup>
+
+```bash npm
+npx trigger.dev@beta init
+```
+
+```bash pnpm
+pnpm dlx trigger.dev@beta init
+```
+
+```bash yarn
+yarn dlx trigger.dev@beta init
+```
+
+</CodeGroup>
+
+It will do a few things:
+
+1. Log you into the CLI if you're not already logged in.
+2. Create a `trigger.config.ts` file in the root of your project.
+3. Ask where you'd like to create the `/trigger` directory.
+4. Create the `/trigger` directory with an example task, `/trigger/example.[ts/js]`.
+
+</Step>

docs/_snippets/v3/step-run-test.mdx

@@ -0,0 +1,9 @@
+<Step title="Perform a test run using the dashboard">
+
+The CLI `dev` command spits out various useful URLs. Right now we want to visit the Test page it provided.
+
+You should see our Example task in the list, select it. Most tasks have a "payload" which you enter in the JSON editor, but our example task doesn't need any input.
+
+Press the "Run test" button.
+
+</Step>

docs/_snippets/v3/step-view-run.mdx

@@ -0,0 +1,7 @@
+<Step title="View your run">
+
+Congratulations, you should see the run page which will live reload showing you the current state of the run.
+
+If you go back to your terminal you'll see that the dev command also shows tasks that are running and links to the run log.
+
+</Step>

docs/documentation/introduction.mdx

@@ -1,9 +1,14 @@
 ---
-title: "Getting Started: Introduction"
+title: "v2: Introduction"
 sidebarTitle: "Introduction"
-description: "Welcome to the Trigger.dev documentation."
+description: "Welcome to the Trigger.dev (v2) documentation."
 ---
 
+<Warning>
+  These are the docs for Trigger.dev v2. You probably want [the v3
+  docs](https://trigger.dev/docs/v3).
+</Warning>
+
 Trigger.dev is an open source framework for creating long-running Jobs directly in your app with API Integrations, webhooks, scheduling and delays. You can reliably run Jobs that wouldn't normally work in serverless environments (like Vercel) because of timeouts.
 
 You can use [Trigger.dev Cloud](https://cloud.trigger.dev) or [Self-host Trigger.dev](/documentation/guides/self-hosting) on your own infrastructure.

docs/mint.json

@@ -1,7 +1,8 @@
 {
   "$schema": "https://mintlify.com/schema.json",
   "name": "Trigger.dev",
-  "openapi": ["/openapi.yml"],
+  "openapi": ["/openapi.yml", "/v3-openapi.json"],
+  "versions": ["v3 (Developer Preview)", "v2"],
   "logo": {
     "dark": "/logo/dark.png",
     "light": "/logo/light.png",
@@ -37,22 +38,34 @@
       "url": "https://discord.gg/kA47vcd8P6"
     }
   ],
+  "primaryTab": {
+    "name": "Home"
+  },
   "tabs": [
     {
       "name": "Integrations",
-      "url": "integrations"
+      "url": "integrations",
+      "version": "v2"
     },
     {
       "name": "Reference",
-      "url": "sdk"
+      "url": "sdk",
+      "version": "v2"
     },
     {
       "name": "Examples",
-      "url": "https://trigger.dev/apis"
+      "url": "https://trigger.dev/apis",
+      "version": "v2"
     },
     {
       "name": "v3 Developer Preview",
-      "url": "v3"
+      "url": "https://trigger.dev/docs/v3",
+      "version": "v2"
+    },
+    {
+      "name": "v2",
+      "url": "https://trigger.dev/docs",
+      "version": "v3 (Developer Preview)"
     }
   ],
   "redirects": [
@@ -76,6 +89,7 @@
   "navigation": [
     {
       "group": "Getting Started",
+      "version": "v2",
       "pages": [
         "documentation/introduction",
         {
@@ -107,6 +121,7 @@
     },
     {
       "group": "Concepts",
+      "version": "v2",
       "pages": [
         "documentation/concepts/what-is-triggerdotdev",
         "documentation/concepts/limits",
@@ -142,6 +157,7 @@
     },
     {
       "group": "Guides",
+      "version": "v2",
       "pages": [
         {
           "group": "Frameworks",
@@ -226,6 +242,7 @@
     },
     {
       "group": "Community & Support",
+      "version": "v2",
       "pages": [
         "documentation/get-help",
         "documentation/faq",
@@ -430,24 +447,29 @@
     },
     {
       "group": "Overview",
+      "version": "v2",
       "pages": ["examples/introduction"]
     },
     {
       "group": "",
+      "version": "v3 (Developer Preview)",
       "pages": ["v3/introduction"]
     },
     {
       "group": "Getting Started",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/quick-start",
         "v3/upgrading-from-v2",
         "v3/changelog",
         "v3/feature-matrix",
-        "v3/limits"
+        "v3/known-issues",
+        "v3/limits-performance"
       ]
     },
     {
       "group": "Fundamentals",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/trigger-folder",
         "v3/tasks-overview",
@@ -456,15 +478,18 @@
         {
           "group": "Task types",
           "pages": ["v3/tasks-regular", "v3/tasks-scheduled", "v3/tasks-zod", "v3/tasks-webhooks"]
-        }
+        },
+        "v3/trigger-config"
       ]
     },
     {
       "group": "Development",
+      "version": "v3 (Developer Preview)",
       "pages": ["v3/cli-dev", "v3/run-tests"]
     },
     {
       "group": "Deployment",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/deploy-environment-variables",
         "v3/cli-deploy",
@@ -477,6 +502,7 @@
     },
     {
       "group": "Writing tasks",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/writing-tasks-introduction",
         "v3/logging",
@@ -506,6 +532,7 @@
     },
     {
       "group": "Dashboard",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/dashboard-overview",
         "v3/dashboard-runs",
@@ -515,6 +542,7 @@
     },
     {
       "group": "API reference",
+      "version": "v3 (Developer Preview)",
       "pages": [
         {
           "group": "Functions",
@@ -559,11 +587,21 @@
             "v3/reference-cli-who-am-i"
           ]
         },
-        "v3/reference-trigger-config"
+        {
+          "group": "Management API",
+          "pages": [
+            "v3/management-start-run",
+            "v3/management-get-run",
+            "v3/management-get-runs",
+            "v3/management-replay-run",
+            "v3/management-cancel-run"
+          ]
+        }
       ]
     },
     {
       "group": "Architecture",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/architecture-how-it-works",
         "v3/architecture-multi-tenant-queue",
@@ -572,10 +610,12 @@
     },
     {
       "group": "Open source",
+      "version": "v3 (Developer Preview)",
       "pages": ["v3/github-repo", "v3/open-source-self-hosting", "v3/open-source-contributing"]
     },
     {
       "group": "Help",
+      "version": "v3 (Developer Preview)",
       "pages": [
         "v3/help-faqs",
         "v3/community",
@@ -589,10 +629,5 @@
     "twitter": "https://twitter.com/triggerdotdev",
     "github": "https://github.com/triggerdotdev",
     "linkedin": "https://www.linkedin.com/company/triggerdotdev"
-  },
-  "analytics": {
-    "posthog": {
-      "apiKey": "phc_9aSDbJCaDUMdZdHxxMPTvcj7A9fsl3mCgM1RBPmPsl7"
-    }
   }
 }

docs/v3-openapi.json

@@ -0,0 +1,216 @@
+{
+  "openapi": "3.1.0",
+  "info": {
+    "title": "Trigger.dev v3 REST API",
+    "description": "The REST API lets you trigger and manage runs on Trigger.dev. You can trigger a run, get the status of a run, and get the results of a run. ",
+    "version": "2024-04"
+  },
+  "servers": [
+    {
+      "url": "https://api.trigger.dev",
+      "description": "Trigger.dev API"
+    }
+  ],
+  "paths": {
+    "/api/v1/runs/{run_id}/replay": {
+      "post": {
+        "description": "Creates a new run with the same payload and options as the original run.",
+        "parameters": [
+          {
+            "in": "path",
+            "name": "run_id",
+            "required": true,
+            "schema": {
+              "type": "string"
+            },
+            "description": "The ID of an existing run. When you trigger a run you will get an id in the response."
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "id": {
+                      "type": "string",
+                      "description": "The ID of the new run."
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Invalid request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Invalid or missing run ID", "Failed to create new run"]
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Unauthorized request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Invalid or Missing API key"]
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "404": {
+            "description": "Resource not found",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Run not found"]
+                    }
+                  }
+                }
+              }
+            }
+          }
+        },
+        "tags": ["run"],
+        "security": [{ "bearerAuth": [] }],
+        "operationId": "replay_run_v1",
+        "x-codeSamples": [
+          {
+            "lang": "typescript",
+            "source": "const handle = await runs.replay(existingRun.id);"
+          },
+          {
+            "lang": "sh",
+            "source": "curl --request POST \\\n\t--url https://api.trigger.dev/api/v1/runs/{run_id}/replay \\\n\t--header 'Authorization: Bearer <token>'"
+          }
+        ]
+      }
+    },
+    "/api/v1/runs/{run_id}/cancel": {
+      "post": {
+        "description": "Cancels a run.",
+        "parameters": [
+          {
+            "in": "path",
+            "name": "run_id",
+            "required": true,
+            "schema": {
+              "type": "string"
+            },
+            "description": "The ID of an existing run. When you trigger a run you will get an id in the response."
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "message": {
+                      "type": "string",
+                      "description": "Confirmation message that the run was canceled."
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Invalid request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Invalid or missing run ID", "Failed to create new run"]
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Unauthorized request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Invalid or Missing API key"]
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "404": {
+            "description": "Resource not found",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "error": {
+                      "type": "string",
+                      "enum": ["Run not found"]
+                    }
+                  }
+                }
+              }
+            }
+          }
+        },
+        "tags": ["run"],
+        "security": [{ "bearerAuth": [] }],
+        "operationId": "replay_run_v1",
+        "x-codeSamples": [
+          {
+            "lang": "typescript",
+            "source": "await runs.cancel(existingRun.id);"
+          },
+          {
+            "lang": "sh",
+            "source": "curl --request POST \\\n\t--url https://api.trigger.dev/api/v1/runs/{run_id}/cancel \\\n\t--header 'Authorization: Bearer <token>'"
+          }
+        ]
+      }
+    }
+  },
+  "components": {
+    "securitySchemes": {
+      "bearerAuth": {
+        "type": "http",
+        "scheme": "bearer",
+        "description": "Use your Secret API key in the form 'Bearer <SECRET KEY>' (without the quotation marks)"
+      }
+    }
+  },
+  "security": [{ "bearerAuth": [] }]
+}

docs/v3/errors-retrying.mdx

@@ -7,12 +7,12 @@ When an uncaught error is thrown inside your task, that task attempt will fail.
 
 You can configure retrying in two ways:
 
-1. In your [trigger.config file](/v3/reference-trigger-config) you can set the default retrying behavior for all tasks.
+1. In your [trigger.config file](/v3/trigger-config) you can set the default retrying behavior for all tasks.
 2. On each task you can set the retrying behavior.
 
 <Note>
   By default when you create your project using the CLI init command we disabled retrying in the DEV
-  environment. You can enable it in your [trigger.config file](/v3/reference-trigger-config).
+  environment. You can enable it in your [trigger.config file](/v3/trigger-config).
 </Note>
 
 ## A simple example with OpenAI

docs/v3/feature-matrix.mdx

@@ -3,8 +3,24 @@ title: "Feature matrix"
 description: "What features are currently available in the Developer Preview"
 ---
 
-| Feature                                | Description                                | Status |
-| -------------------------------------- | ------------------------------------------ | ------ |
-| [Regular tasks](/v3/tasks-regular)     | A task that can be triggered from anywhere | ✅     |
-| [Scheduled tasks](/v3/tasks-scheduled) | A task that can be triggered on a schedule | ⏳     |
-| [Webhook tasks](v3/tasks-webhooks)     | A task that can be triggered by a webhook  | ⏳     |
+| Feature                                                                             | Description                                        | Status |
+| ----------------------------------------------------------------------------------- | -------------------------------------------------- | ------ |
+| [Regular tasks](/v3/tasks-regular)                                                  | A task that can be triggered from anywhere         | ✅     |
+| [Triggering](/v3/triggering)                                                        | Triggering and batch triggering tasks              | ✅     |
+| [Testing from the dashboard](/v3/run-tests)                                         | Test your tasks from the dashboard                 | ✅     |
+| [Queues and concurrency controls](/v3/queue-concurrency)                            | Queues and concurrency controls                    | ✅     |
+| [Per-tenant queuing](/v3/queue-concurrency#concurrency-keys-and-per-tenant-queuing) | Separate queues for each of your users             | ✅     |
+| [Reattempts and retrying](/v3/errors-retrying)                                      | Write reliable tasks using retries                 | ✅     |
+| [Atomic versioning](/v3/versioning)                                                 | Each deploy creates a new version                  | ✅     |
+| [Deploy via CLI](/v3/cli-deploy)                                                    | Deploy from the command line                       | ✅     |
+| [Deploy via GitHub Actions](/v3/github-actions)                                     | Deploy using GitHub Actions                        | ✅     |
+| [Scheduled tasks](/v3/tasks-scheduled)                                              | A task that can be triggered on a schedule         | ⏳     |
+| [Zod tasks](/v3/tasks-zod)                                                          | Define tasks using Zod schemas                     | ⏳     |
+| [Webhook tasks](/v3/tasks-webhooks)                                                 | A task that can be triggered by a webhook          | ⏳     |
+| Full text search of runs                                                            | Find a run by searching the payload and output     | ⏳     |
+| Logs view with search                                                               | All logs view with filtering and full text search  | ⏳     |
+| Alerts                                                                              | Add alerts in the UI for errors and queue backlogs | ⏳     |
+| Notifications                                                                       | Send data to your web app from a run               | ⏳     |
+| Rollbacks                                                                           | Easily rollback changes when errors happen         | ⏳     |
+
+[Let us know](https://trigger.dev/discord) what we should prioritize and what we are missing.

docs/v3/known-issues.mdx

@@ -0,0 +1,12 @@
+---
+title: "Known issues"
+description: "Please let us know if you encounter an issue not on this list."
+---
+
+#### `pnpm` is not currently supported
+
+By default, pnpm has a more complicated package structure than npm. If you see this error when running the CLI then you're experiencing this bug:
+
+```bash
+✘ [ERROR] Could not resolve "@trigger.dev/core/v3"
+```

docs/v3/limits-performance.mdx

@@ -1,5 +1,5 @@
 ---
-title: "Limits"
+title: "Limits & Performance"
 ---
 
 <Snippet file="incomplete-docs.mdx" />

docs/v3/logging.mdx

@@ -38,18 +38,20 @@ Tracing is a way to follow the flow of your code. It's very useful for debugging
 
 Trigger.dev uses OpenTelemetry tracing under the hood. With automatic tracing for many things like task triggering, task attempts, HTTP requests, and more.
 
-### Automatic instrumentation
-
 | Name          | Description                      |
 | ------------- | -------------------------------- |
 | Task triggers | Task triggers.                   |
 | Task attempts | Task attempts.                   |
 | HTTP requests | HTTP requests made by your code. |
-| OpenAI        | OpenAI SDK calls.                |
 
-We want to provide automatic instrumentation for as many things as possible. Please do [request any automatic instrumentation](https://github.com/triggerdotdev/trigger.dev/issues/new?template=instrumentation_request.yml) you would like to see.
+### Adding instrumentations
 
-## Add custom traces
+![The run log
+  ](/images/v3/auto-instrumentation.png)
+
+You can [add instrumentations](/v3/trigger-config#instrumentations). The Prisma one above will automatically trace all Prisma queries.
+
+### Add custom traces
 
 If you want to add custom traces to your code, you can use the `logger.trace` function. It will create a new OTEL trace and you can set attributes on it.
 

docs/v3/management-cancel-run.mdx

@@ -0,0 +1,4 @@
+---
+title: "Cancel run"
+openapi: "v3-openapi POST /api/v1/runs/{run_id}/cancel"
+---