Release 3.0.0-beta.36

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

.changeset/hot-buckets-behave.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Support triggering tasks with non-URL friendly characters in the ID

.changeset/itchy-chairs-itch.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix for calling trigger and passing a custom queue

.changeset/new-pants-beg.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Fix issue when using SDK in non-node environments by scoping the stream import with node:

.changeset/pre.json

@@ -68,7 +68,9 @@
     "funny-swans-destroy",
     "gorgeous-gorillas-compete",
     "green-bags-wink",
+    "hot-buckets-behave",
     "hot-fishes-retire",
+    "itchy-chairs-itch",
     "khaki-apricots-design",
     "khaki-poems-lay",
     "late-icons-lie",
@@ -85,6 +87,7 @@
     "mighty-camels-joke",
     "mighty-flowers-train",
     "nasty-jars-pump",
+    "new-pants-beg",
     "new-rivers-tell",
     "nice-bulldogs-turn",
     "ninety-pets-travel",
@@ -107,6 +110,7 @@
     "shiny-coats-cry",
     "silly-suits-switch",
     "six-ligers-exist",
+    "sixty-insects-watch",
     "slow-buses-own",
     "smart-needles-move",
     "smart-olives-eat",
@@ -115,6 +119,7 @@
     "strange-sheep-pull",
     "strong-lemons-add",
     "strong-owls-know",
+    "stupid-adults-sniff",
     "stupid-bulldogs-applaud",
     "sweet-lizards-press",
     "swift-dragons-peel",
@@ -130,8 +135,11 @@
     "tiny-doors-type",
     "tiny-elephants-scream",
     "tricky-bulldogs-heal",
+    "tricky-keys-attack",
     "tricky-ladybugs-unite",
     "two-pumas-wait",
-    "warm-planes-taste"
+    "warm-olives-provide",
+    "warm-planes-taste",
+    "young-snails-sell"
   ]
 }

.changeset/sixty-insects-watch.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Set the deploy timeout to 3mins from 1min

.changeset/stupid-adults-sniff.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Management SDK overhaul and adding the runs.list API

.changeset/tricky-keys-attack.md

@@ -0,0 +1,14 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Clear paused states before retry
+- Detect and handle unrecoverable worker errors
+- Remove checkpoints after successful push
+- Permanently switch to DO hosted busybox image
+- Fix IPC timeout issue, or at least handle it more gracefully
+- Handle checkpoint failures
+- Basic chaos monkey for checkpoint testing
+- Stack traces are back in the dashboard
+- Display final errors on root span

.changeset/warm-olives-provide.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Improve handling of IPC timeouts and fix checkpoint cancellation after failures

.changeset/young-snails-sell.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Increase cleanup IPC timeout

apps/coordinator/src/backoff.ts

@@ -0,0 +1,247 @@
+type ExponentialBackoffType = "NoJitter" | "FullJitter" | "EqualJitter";
+
+type ExponentialBackoffOptions = {
+  base: number;
+  factor: number;
+  min: number;
+  max: number;
+  maxRetries: number;
+  maxElapsed: number;
+};
+
+class StopRetrying extends Error {
+  constructor(message?: string) {
+    super(message);
+    this.name = "StopRetrying";
+  }
+}
+
+export class ExponentialBackoff {
+  #retries: number = 0;
+
+  #type: ExponentialBackoffType;
+  #base: number;
+  #factor: number;
+
+  #min: number;
+  #max: number;
+
+  #maxRetries: number;
+  #maxElapsed: number;
+
+  constructor(type?: ExponentialBackoffType, opts: Partial<ExponentialBackoffOptions> = {}) {
+    this.#type = type ?? "NoJitter";
+    this.#base = opts.base ?? 2;
+    this.#factor = opts.factor ?? 1;
+
+    this.#min = opts.min ?? -Infinity;
+    this.#max = opts.max ?? Infinity;
+
+    this.#maxRetries = opts.maxRetries ?? Infinity;
+    this.#maxElapsed = opts.maxElapsed ?? Infinity;
+  }
+
+  #clone() {
+    return new ExponentialBackoff(this.#type, {
+      base: this.#base,
+      factor: this.#factor,
+      min: this.#min,
+      max: this.#max,
+      maxRetries: this.#maxRetries,
+      maxElapsed: this.#maxElapsed,
+    });
+  }
+
+  type(type?: ExponentialBackoffType) {
+    if (typeof type !== "undefined") {
+      this.#type = type;
+    }
+    return this.#clone();
+  }
+
+  base(base?: number) {
+    if (typeof base !== "undefined") {
+      this.#base = base;
+    }
+    return this.#clone();
+  }
+
+  factor(factor?: number) {
+    if (typeof factor !== "undefined") {
+      this.#factor = factor;
+    }
+    return this.#clone();
+  }
+
+  min(min?: number) {
+    if (typeof min !== "undefined") {
+      this.#min = min;
+    }
+    return this.#clone();
+  }
+
+  max(max?: number) {
+    if (typeof max !== "undefined") {
+      this.#max = max;
+    }
+    return this.#clone();
+  }
+
+  maxRetries(maxRetries?: number) {
+    if (typeof maxRetries !== "undefined") {
+      this.#maxRetries = maxRetries;
+    }
+    return this.#clone();
+  }
+
+  maxElapsed(maxElapsed?: number) {
+    if (typeof maxElapsed !== "undefined") {
+      this.#maxElapsed = maxElapsed;
+    }
+    return this.#clone();
+  }
+
+  retries(retries?: number) {
+    if (typeof retries !== "undefined") {
+      if (retries > this.#maxRetries) {
+        console.error(
+          `Can't set retries ${retries} higher than maxRetries (${
+            this.#maxRetries
+          }), setting to maxRetries instead.`
+        );
+        this.#retries = this.#maxRetries;
+      } else {
+        this.#retries = retries;
+      }
+    }
+    return this.#clone();
+  }
+
+  async *retryAsync(maxRetries: number = this.#maxRetries ?? Infinity) {
+    let elapsed = 0;
+    let retry = 0;
+
+    while (retry <= maxRetries) {
+      const delay = this.delay(retry);
+      elapsed += delay;
+
+      if (elapsed > this.#maxElapsed) {
+        break;
+      }
+
+      yield {
+        delay: {
+          seconds: delay,
+          milliseconds: delay * 1000,
+        },
+        retry,
+      };
+
+      retry++;
+    }
+  }
+
+  async *[Symbol.asyncIterator]() {
+    yield* this.retryAsync();
+  }
+
+  delay(retries: number = this.#retries, jitter: boolean = true) {
+    if (retries > this.#maxRetries) {
+      console.error(
+        `Can't set retries ${retries} higher than maxRetries (${
+          this.#maxRetries
+        }), setting to maxRetries instead.`
+      );
+      retries = this.#maxRetries;
+    }
+
+    let delay = this.#factor * this.#base ** retries;
+
+    switch (this.#type) {
+      case "NoJitter": {
+        break;
+      }
+      case "FullJitter": {
+        if (!jitter) {
+          delay = 0;
+          break;
+        }
+
+        delay *= Math.random();
+        break;
+      }
+      case "EqualJitter": {
+        if (!jitter) {
+          delay *= 0.5;
+          break;
+        }
+
+        delay *= 0.5 * (1 + Math.random());
+        break;
+      }
+      default: {
+        throw new Error(`Unknown backoff type: ${this.#type}`);
+      }
+    }
+
+    delay = Math.min(delay, this.#max);
+    delay = Math.max(delay, this.#min);
+    delay = Math.round(delay);
+
+    return delay;
+  }
+
+  elapsed(retries: number = this.#retries, jitter: boolean = true) {
+    let elapsed = 0;
+
+    for (let i = 0; i <= retries; i++) {
+      elapsed += this.delay(i, jitter);
+    }
+
+    const total = elapsed;
+
+    let days = 0;
+    if (elapsed > 3600 * 24) {
+      days = Math.floor(elapsed / 3600 / 24);
+      elapsed -= days * 3600 * 24;
+    }
+
+    let hours = 0;
+    if (elapsed > 3600) {
+      hours = Math.floor(elapsed / 3600);
+      elapsed -= hours * 3600;
+    }
+
+    let minutes = 0;
+    if (elapsed > 60) {
+      minutes = Math.floor(elapsed / 60);
+      elapsed -= minutes * 60;
+    }
+
+    const seconds = elapsed;
+
+    return {
+      seconds,
+      minutes,
+      hours,
+      days,
+      total,
+    };
+  }
+
+  reset() {
+    this.#retries = 0;
+    return this;
+  }
+
+  next() {
+    this.#retries++;
+    return this.delay();
+  }
+
+  stop() {
+    throw new StopRetrying();
+  }
+
+  static StopRetrying = StopRetrying;
+}

apps/coordinator/src/index.ts

@@ -1,5 +1,6 @@
 import { createServer } from "node:http";
-import { $ } from "execa";
+import fs from "node:fs/promises";
+import { $, type ExecaChildProcess } from "execa";
 import { nanoid } from "nanoid";
 import { Server } from "socket.io";
 import {
@@ -12,6 +13,7 @@ import {
 import { ZodNamespace } from "@trigger.dev/core/v3/zodNamespace";
 import { ZodSocketConnection } from "@trigger.dev/core/v3/zodSocket";
 import { HttpReply, getTextBody, SimpleLogger } from "@trigger.dev/core-apps";
+import { ExponentialBackoff } from "./backoff";
 
 import { collectDefaultMetrics, register, Gauge } from "prom-client";
 collectDefaultMetrics();
@@ -19,6 +21,26 @@ 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 CHAOS_MONKEY_ENABLED = !!process.env.CHAOS_MONKEY_ENABLED;
+
+const FORCE_CHECKPOINT_SIMULATION = ["1", "true"].includes(
+  process.env.FORCE_CHECKPOINT_SIMULATION ?? "true"
+);
+const DISABLE_CHECKPOINT_SUPPORT = ["1", "true"].includes(
+  process.env.DISABLE_CHECKPOINT_SUPPORT ?? "false"
+);
+const SIMULATE_PUSH_FAILURE = ["1", "true"].includes(process.env.SIMULATE_PUSH_FAILURE ?? "false");
+const SIMULATE_PUSH_FAILURE_SECONDS = parseInt(
+  process.env.SIMULATE_PUSH_FAILURE_SECONDS ?? "300",
+  10
+);
+const SIMULATE_CHECKPOINT_FAILURE = ["1", "true"].includes(
+  process.env.SIMULATE_CHECKPOINT_FAILURE ?? "false"
+);
+const SIMULATE_CHECKPOINT_FAILURE_SECONDS = parseInt(
+  process.env.SIMULATE_CHECKPOINT_FAILURE_SECONDS ?? "300",
+  10
+);
 
 const REGISTRY_HOST = process.env.REGISTRY_HOST || "localhost:5000";
 const CHECKPOINT_PATH = process.env.CHECKPOINT_PATH || "/checkpoints";
@@ -32,6 +54,10 @@ const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?
 
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 
+if (CHAOS_MONKEY_ENABLED) {
+  logger.log("🍌 Chaos monkey enabled");
+}
+
 type CheckpointerInitializeReturn = {
   canCheckpoint: boolean;
   willSimulate: boolean;
@@ -44,11 +70,49 @@ type CheckpointAndPushOptions = {
   deploymentVersion: string;
 };
 
+type CheckpointAndPushResult =
+  | { success: true; checkpoint: CheckpointData }
+  | { success: false; reason?: "CANCELED" | "DISABLED" | "ERROR" | "IN_PROGRESS" | "NO_SUPPORT" };
+
 type CheckpointData = {
   location: string;
   docker: boolean;
 };
 
+function isExecaChildProcess(maybeExeca: unknown): maybeExeca is Awaited<ExecaChildProcess> {
+  return typeof maybeExeca === "object" && maybeExeca !== null && "escapedCommand" in maybeExeca;
+}
+
+async function getFileSize(filePath: string): Promise<number> {
+  try {
+    const stats = await fs.stat(filePath);
+    return stats.size;
+  } catch (error) {
+    console.error("Error getting file size:", error);
+    return -1;
+  }
+}
+
+async function getParsedFileSize(filePath: string) {
+  const sizeInBytes = await getFileSize(filePath);
+
+  let message = `Size in bytes: ${sizeInBytes}`;
+
+  if (sizeInBytes > 1024 * 1024) {
+    const sizeInMB = (sizeInBytes / 1024 / 1024).toFixed(2);
+    message = `Size in MB (rounded): ${sizeInMB}`;
+  } else if (sizeInBytes > 1024) {
+    const sizeInKB = (sizeInBytes / 1024).toFixed(2);
+    message = `Size in KB (rounded): ${sizeInKB}`;
+  }
+
+  return {
+    path: filePath,
+    sizeInBytes,
+    message,
+  };
+}
+
 class Checkpointer {
   #initialized = false;
   #canCheckpoint = false;
@@ -56,6 +120,8 @@ class Checkpointer {
 
   #logger = new SimpleLogger("[checkptr]");
   #abortControllers = new Map<string, AbortController>();
+  #failedCheckpoints = new Map<string, unknown>();
+  #waitingForRetry = new Set<string>();
 
   constructor(private opts = { forceSimulate: false }) {}
 
@@ -139,7 +205,7 @@ class Checkpointer {
     const start = performance.now();
     logger.log(`checkpointAndPush() start`, { start, opts });
 
-    const result = await this.#checkpointAndPush(opts);
+    const result = await this.#checkpointAndPushWithBackoff(opts);
 
     const end = performance.now();
     logger.log(`checkpointAndPush() end`, {
@@ -147,17 +213,33 @@ class Checkpointer {
       end,
       diff: end - start,
       opts,
-      success: !!result,
+      success: result.success,
     });
 
-    return result;
+    if (!result.success) {
+      return;
+    }
+
+    return result.checkpoint;
   }
 
   isCheckpointing(runId: string) {
-    return this.#abortControllers.has(runId);
+    return this.#abortControllers.has(runId) || this.#waitingForRetry.has(runId);
   }
 
   cancelCheckpoint(runId: string): boolean {
+    // If the last checkpoint failed, pretend we canceled it
+    // This ensures tasks don't wait for external resume messages to continue
+    if (this.#hasFailedCheckpoint(runId)) {
+      this.#clearFailedCheckpoint(runId);
+      return true;
+    }
+
+    if (this.#waitingForRetry.has(runId)) {
+      this.#waitingForRetry.delete(runId);
+      return true;
+    }
+
     const controller = this.#abortControllers.get(runId);
 
     if (!controller) {
@@ -171,29 +253,133 @@ class Checkpointer {
     return true;
   }
 
+  async #checkpointAndPushWithBackoff({
+    runId,
+    leaveRunning = true, // This mirrors kubernetes behaviour more accurately
+    projectRef,
+    deploymentVersion,
+  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
+    this.#logger.log("Checkpointing with backoff", {
+      runId,
+      leaveRunning,
+      projectRef,
+      deploymentVersion,
+    });
+
+    const backoff = new ExponentialBackoff()
+      .type("EqualJitter")
+      .base(3)
+      .max(3 * 3600)
+      .maxElapsed(48 * 3600);
+
+    for await (const { delay, retry } of backoff) {
+      try {
+        if (retry > 0) {
+          this.#logger.error("Retrying checkpoint", {
+            runId,
+            retry,
+            delay,
+          });
+
+          this.#waitingForRetry.add(runId);
+          await new Promise((resolve) => setTimeout(resolve, delay.milliseconds));
+
+          if (!this.#waitingForRetry.has(runId)) {
+            this.#logger.log("Checkpoint canceled while waiting for retry", { runId });
+            return { success: false, reason: "CANCELED" };
+          } else {
+            this.#waitingForRetry.delete(runId);
+          }
+        }
+
+        const result = await this.#checkpointAndPush({
+          runId,
+          leaveRunning,
+          projectRef,
+          deploymentVersion,
+        });
+
+        if (result.success) {
+          return result;
+        }
+
+        if (result.reason === "CANCELED") {
+          this.#logger.log("Checkpoint canceled, won't retry", { runId });
+          // Don't fail the checkpoint, as it was canceled
+          return result;
+        }
+
+        if (result.reason === "IN_PROGRESS") {
+          this.#logger.log("Checkpoint already in progress, won't retry", { runId });
+          this.#failCheckpoint(runId, result.reason);
+          return result;
+        }
+
+        if (result.reason === "NO_SUPPORT") {
+          this.#logger.log("No checkpoint support, won't retry", { runId });
+          this.#failCheckpoint(runId, result.reason);
+          return result;
+        }
+
+        if (result.reason === "DISABLED") {
+          this.#logger.log("Checkpoint support disabled, won't retry", { runId });
+          this.#failCheckpoint(runId, result.reason);
+          return result;
+        }
+
+        continue;
+      } catch (error) {
+        this.#logger.error("Checkpoint error", {
+          retry,
+          runId,
+          delay,
+          error: error instanceof Error ? error.message : error,
+        });
+      }
+    }
+
+    this.#logger.error(`Checkpoint failed after exponential backoff`, {
+      runId,
+      leaveRunning,
+      projectRef,
+      deploymentVersion,
+    });
+    this.#failCheckpoint(runId, "ERROR");
+
+    return { success: false, reason: "ERROR" };
+  }
+
   async #checkpointAndPush({
     runId,
     leaveRunning = true, // This mirrors kubernetes behaviour more accurately
     projectRef,
     deploymentVersion,
-  }: CheckpointAndPushOptions): Promise<CheckpointData | undefined> {
+  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
     await this.initialize();
 
+    const options = {
+      runId,
+      leaveRunning,
+      projectRef,
+      deploymentVersion,
+    };
+
     if (!this.#dockerMode && !this.#canCheckpoint) {
       this.#logger.error("No checkpoint support. Simulation requires docker.");
-      return;
+      return { success: false, reason: "NO_SUPPORT" };
     }
 
     if (this.#abortControllers.has(runId)) {
-      logger.error("Checkpoint procedure already in progress", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-      });
-      return;
+      logger.error("Checkpoint procedure already in progress", { options });
+      return { success: false, reason: "IN_PROGRESS" };
+    }
+
+    // This is a new checkpoint, clear any last failure for this run
+    this.#clearFailedCheckpoint(runId);
+
+    if (DISABLE_CHECKPOINT_SUPPORT) {
+      this.#logger.error("Checkpoint support disabled", { options });
+      return { success: false, reason: "DISABLED" };
     }
 
     const controller = new AbortController();
@@ -201,19 +387,44 @@ class Checkpointer {
 
     const $$ = $({ signal: controller.signal });
 
-    try {
-      const shortCode = nanoid(8);
-      const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
-      const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
+    const shortCode = nanoid(8);
+    const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
+    const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
 
-      this.#logger.log("Checkpointing:", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-      });
+    const cleanup = async () => {
+      if (this.#dockerMode) {
+        return;
+      }
+
+      try {
+        await $`rm ${exportLocation}`;
+        this.#logger.log("Deleted checkpoint archive", { exportLocation });
+
+        await $`buildah rmi ${imageRef}`;
+        this.#logger.log("Deleted checkpoint image", { imageRef });
+      } catch (error) {
+        this.#logger.error("Failure during checkpoint cleanup", { exportLocation, error });
+      }
+    };
+
+    try {
+      if (CHAOS_MONKEY_ENABLED) {
+        console.log("🍌 Chaos monkey wreaking havoc");
+
+        const random = Math.random();
+
+        if (random < 0.33) {
+          // Fake long checkpoint duration
+          await $$`sleep 300`;
+        } else if (random < 0.66) {
+          // Fake checkpoint error
+          await $$`false`;
+        } else {
+          // no-op
+        }
+      }
+
+      this.#logger.log("Checkpointing:", { options });
 
       const containterName = this.#getRunContainerName(runId);
 
@@ -224,6 +435,13 @@ class Checkpointer {
             this.#logger.log("Simulating checkpoint");
             this.#logger.debug(await $$`docker pause ${containterName}`);
           } else {
+            if (SIMULATE_CHECKPOINT_FAILURE) {
+              if (performance.now() < SIMULATE_CHECKPOINT_FAILURE_SECONDS * 1000) {
+                this.#logger.error("Simulating checkpoint failure", { options });
+                throw new Error("SIMULATE_CHECKPOINT_FAILURE");
+              }
+            }
+
             if (leaveRunning) {
               this.#logger.debug(
                 await $$`docker checkpoint create --leave-running ${containterName} ${exportLocation}`
@@ -234,9 +452,9 @@ class Checkpointer {
               );
             }
           }
-        } catch (error: any) {
-          this.#logger.error(error.stderr);
-          return;
+        } catch (error) {
+          this.#logger.error("Failed while creating docker checkpoint", { exportLocation });
+          throw error;
         }
 
         this.#logger.log("checkpoint created:", {
@@ -245,8 +463,11 @@ class Checkpointer {
         });
 
         return {
-          location: exportLocation,
-          docker: true,
+          success: true,
+          checkpoint: {
+            location: exportLocation,
+            docker: true,
+          },
         };
       }
 
@@ -266,54 +487,105 @@ class Checkpointer {
         throw new Error("could not find container id");
       }
 
+      const start = performance.now();
+
+      if (SIMULATE_CHECKPOINT_FAILURE) {
+        if (performance.now() < SIMULATE_CHECKPOINT_FAILURE_SECONDS * 1000) {
+          this.#logger.error("Simulating checkpoint failure", { options });
+          throw new Error("SIMULATE_CHECKPOINT_FAILURE");
+        }
+      }
+
+      // Create checkpoint
       this.#logger.debug(await $$`crictl checkpoint --export=${exportLocation} ${containerId}`);
+      const postCheckpoint = performance.now();
+
+      // Print checkpoint size
+      const size = await getParsedFileSize(exportLocation);
+      this.#logger.log("checkpoint archive created", { size, options });
 
       // Create image from checkpoint
       const container = this.#logger.debug(await $$`buildah from scratch`);
+      const postFrom = performance.now();
+
       this.#logger.debug(await $$`buildah add ${container} ${exportLocation} /`);
+      const postAdd = performance.now();
+
       this.#logger.debug(
         await $$`buildah config --annotation=io.kubernetes.cri-o.annotations.checkpoint.name=counter ${container}`
       );
+      const postConfig = performance.now();
+
       this.#logger.debug(await $$`buildah commit ${container} ${imageRef}`);
+      const postCommit = performance.now();
+
       this.#logger.debug(await $$`buildah rm ${container}`);
+      const postRm = performance.now();
+
+      if (SIMULATE_PUSH_FAILURE) {
+        if (performance.now() < SIMULATE_PUSH_FAILURE_SECONDS * 1000) {
+          this.#logger.error("Simulating push failure", { options });
+          throw new Error("SIMULATE_PUSH_FAILURE");
+        }
+      }
 
       // Push checkpoint image
       this.#logger.debug(await $$`buildah push --tls-verify=${REGISTRY_TLS_VERIFY} ${imageRef}`);
+      const postPush = performance.now();
 
-      this.#logger.log("Checkpointed and pushed image to:", { location: imageRef });
+      const perf = {
+        "crictl checkpoint": postCheckpoint - start,
+        "buildah from": postFrom - postCheckpoint,
+        "buildah add": postAdd - postFrom,
+        "buildah config": postConfig - postAdd,
+        "buildah commit": postCommit - postConfig,
+        "buildah rm": postRm - postCommit,
+        "buildah push": postPush - postRm,
+      };
 
-      try {
-        await $$`rm ${exportLocation}`;
-        this.#logger.log("Deleted checkpoint archive", { exportLocation });
-
-        // Disabled for now as this will increase restore time by having to pull the image again
-        // await $`buildah rmi ${imageRef}`;
-        // this.#logger.log("Deleted checkpoint image", { imageRef });
-      } catch (error) {
-        this.#logger.error("Failed during checkpoint cleanup", { exportLocation });
-        this.#logger.debug(error);
-      }
+      this.#logger.log("Checkpointed and pushed image to:", { location: imageRef, perf });
 
       return {
-        location: imageRef,
-        docker: false,
+        success: true,
+        checkpoint: {
+          location: imageRef,
+          docker: false,
+        },
       };
     } catch (error) {
-      this.#logger.error("checkpoint failed", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-        error,
-      });
-      return;
+      if (isExecaChildProcess(error)) {
+        if (error.isCanceled) {
+          this.#logger.error("Checkpoint canceled", { options, error });
+
+          return { success: false, reason: "CANCELED" };
+        }
+
+        this.#logger.error("Checkpoint command error", { options, error });
+
+        return { success: false, reason: "ERROR" };
+      }
+
+      this.#logger.error("Unhandled checkpoint error", { options, error });
+
+      return { success: false, reason: "ERROR" };
     } finally {
       this.#abortControllers.delete(runId);
+      await cleanup();
     }
   }
 
+  #failCheckpoint(runId: string, error: unknown) {
+    this.#failedCheckpoints.set(runId, error);
+  }
+
+  #clearFailedCheckpoint(runId: string) {
+    this.#failedCheckpoints.delete(runId);
+  }
+
+  #hasFailedCheckpoint(runId: string) {
+    return this.#failedCheckpoints.has(runId);
+  }
+
   #getRunContainerName(suffix: string) {
     return `task-run-${suffix}`;
   }
@@ -321,7 +593,7 @@ class Checkpointer {
 
 class TaskCoordinator {
   #httpServer: ReturnType<typeof createServer>;
-  #checkpointer = new Checkpointer({ forceSimulate: true });
+  #checkpointer = new Checkpointer({ forceSimulate: FORCE_CHECKPOINT_SIMULATION });
 
   #prodWorkerNamespace: ZodNamespace<
     typeof ProdWorkerToCoordinatorMessages,
@@ -442,6 +714,30 @@ class TaskCoordinator {
 
           taskSocket.emit("REQUEST_ATTEMPT_CANCELLATION", message);
         },
+        REQUEST_RUN_CANCELLATION: async (message) => {
+          const taskSocket = await this.#getRunSocket(message.runId);
+
+          if (!taskSocket) {
+            logger.log("Socket for run not found", {
+              runId: message.runId,
+            });
+            return;
+          }
+
+          this.#checkpointer.cancelCheckpoint(message.runId);
+
+          if (message.delayInMs) {
+            taskSocket.emit("REQUEST_EXIT", {
+              version: "v2",
+              delayInMs: message.delayInMs,
+            });
+          } else {
+            // If there's no delay, assume the worker doesn't support non-v1 messages
+            taskSocket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+          }
+        },
         READY_FOR_RETRY: async (message) => {
           const taskSocket = await this.#getRunSocket(message.runId);
 
@@ -528,6 +824,20 @@ class TaskCoordinator {
       onConnection: async (socket, handler, sender) => {
         const logger = new SimpleLogger(`[prod-worker][${socket.id}]`);
 
+        const crashRun = async (error: { name: string; message: string; stack?: string }) => {
+          try {
+            this.#platformSocket?.send("RUN_CRASHED", {
+              version: "v1",
+              runId: socket.data.runId,
+              error,
+            });
+          } finally {
+            socket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+          }
+        };
+
         const checkpointInProgress = () => {
           return this.#checkpointableTasks.has(socket.data.runId);
         };
@@ -596,8 +906,9 @@ class TaskCoordinator {
             if (!executionAck) {
               logger.error("no execution ack", { runId: socket.data.runId });
 
-              socket.emit("REQUEST_EXIT", {
-                version: "v1",
+              await crashRun({
+                name: "ReadyForExecutionError",
+                message: "No execution ack",
               });
 
               return;
@@ -606,8 +917,9 @@ class TaskCoordinator {
             if (!executionAck.success) {
               logger.error("failed to get execution payload", { runId: socket.data.runId });
 
-              socket.emit("REQUEST_EXIT", {
-                version: "v1",
+              await crashRun({
+                name: "ReadyForExecutionError",
+                message: "Failed to get execution payload",
               });
 
               return;
@@ -621,6 +933,62 @@ class TaskCoordinator {
             socket.data.attemptFriendlyId = executionAck.payload.execution.attempt.id;
           } catch (error) {
             logger.error("Error", { error });
+
+            await crashRun({
+              name: "ReadyForExecutionError",
+              message:
+                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            });
+
+            return;
+          }
+        });
+
+        socket.on("READY_FOR_LAZY_ATTEMPT", async (message) => {
+          logger.log("[READY_FOR_LAZY_ATTEMPT]", message);
+
+          try {
+            const lazyAttempt = await this.#platformSocket?.sendWithAck("READY_FOR_LAZY_ATTEMPT", {
+              ...message,
+              envId: socket.data.envId,
+            });
+
+            if (!lazyAttempt) {
+              logger.error("no lazy attempt ack", { runId: socket.data.runId });
+
+              await crashRun({
+                name: "ReadyForLazyAttemptError",
+                message: "No lazy attempt ack",
+              });
+
+              return;
+            }
+
+            if (!lazyAttempt.success) {
+              logger.error("failed to get lazy attempt payload", { runId: socket.data.runId });
+
+              await crashRun({
+                name: "ReadyForLazyAttemptError",
+                message: "Failed to get lazy attempt payload",
+              });
+
+              return;
+            }
+
+            socket.emit("EXECUTE_TASK_RUN_LAZY_ATTEMPT", {
+              version: "v1",
+              lazyPayload: lazyAttempt.lazyPayload,
+            });
+          } catch (error) {
+            logger.error("Error", { error });
+
+            await crashRun({
+              name: "ReadyForLazyAttemptError",
+              message:
+                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            });
+
+            return;
           }
         });
 
@@ -714,6 +1082,19 @@ class TaskCoordinator {
           }
         });
 
+        socket.on("TASK_RUN_FAILED_TO_RUN", async ({ completion }) => {
+          logger.log("completed task", { completionId: completion.id });
+
+          this.#platformSocket?.send("TASK_RUN_FAILED_TO_RUN", {
+            version: "v1",
+            completion,
+          });
+
+          socket.emit("REQUEST_EXIT", {
+            version: "v1",
+          });
+        });
+
         socket.on("READY_FOR_CHECKPOINT", async (message) => {
           logger.log("[READY_FOR_CHECKPOINT]", message);
 
@@ -890,7 +1271,7 @@ class TaskCoordinator {
           logger.log("[INDEX_TASKS]", message);
 
           const workerAck = await this.#platformSocket?.sendWithAck("CREATE_WORKER", {
-            version: "v1",
+            version: "v2",
             projectRef: socket.data.projectRef,
             envId: socket.data.envId,
             deploymentId: message.deploymentId,
@@ -899,6 +1280,7 @@ class TaskCoordinator {
               packageVersion: message.packageVersion,
               tasks: message.tasks,
             },
+            supportsLazyAttempts: message.version !== "v1" && message.supportsLazyAttempts,
           });
 
           if (!workerAck) {
@@ -917,6 +1299,34 @@ class TaskCoordinator {
             error: message.error,
           });
         });
+
+        socket.on("CREATE_TASK_RUN_ATTEMPT", async (message, callback) => {
+          logger.log("[CREATE_TASK_RUN_ATTEMPT]", message);
+
+          const createAttempt = await this.#platformSocket?.sendWithAck("CREATE_TASK_RUN_ATTEMPT", {
+            runId: message.runId,
+            envId: socket.data.envId,
+          });
+
+          if (!createAttempt?.success) {
+            logger.debug("no ack while creating attempt", message);
+            callback({ success: false });
+            return;
+          }
+
+          socket.data.attemptFriendlyId = createAttempt.executionPayload.execution.attempt.id;
+
+          callback({
+            success: true,
+            executionPayload: createAttempt.executionPayload,
+          });
+        });
+
+        socket.on("UNRECOVERABLE_ERROR", async (message) => {
+          logger.log("[UNRECOVERABLE_ERROR]", message);
+
+          await crashRun(message.error);
+        });
       },
       onDisconnect: async (socket, handler, sender, logger) => {
         this.#platformSocket?.send("LOG", {
@@ -928,13 +1338,16 @@ class TaskCoordinator {
         TASK_HEARTBEAT: async (message) => {
           this.#platformSocket?.send("TASK_HEARTBEAT", message);
         },
+        TASK_RUN_HEARTBEAT: async (message) => {
+          this.#platformSocket?.send("TASK_RUN_HEARTBEAT", message);
+        },
       },
     });
 
     return provider;
   }
 
-  #cancelCheckpoint(runId: string) {
+  #cancelCheckpoint(runId: string): boolean {
     const checkpointWait = this.#checkpointableTasks.get(runId);
 
     if (checkpointWait) {
@@ -945,6 +1358,8 @@ class TaskCoordinator {
     // Cancel checkpointing procedure
     const checkpointCanceled = this.#checkpointer.cancelCheckpoint(runId);
 
+    logger.log("cancelCheckpoint()", { runId, checkpointCanceled });
+
     return checkpointCanceled;
   }
 

apps/docker-provider/.env.example

@@ -4,6 +4,8 @@ PLATFORM_WS_PORT=3030
 PLATFORM_SECRET=provider-secret
 SECURE_CONNECTION=false
 
+OTEL_EXPORTER_OTLP_ENDPOINT=http://0.0.0.0:3030/otel
+
 # Use this if you are on macOS
 # COORDINATOR_HOST="host.docker.internal"
 # OTEL_EXPORTER_OTLP_ENDPOINT="http://host.docker.internal:4318"

apps/docker-provider/src/index.ts

@@ -13,9 +13,14 @@ import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
 const MACHINE_NAME = process.env.MACHINE_NAME || "local";
 const COORDINATOR_PORT = process.env.COORDINATOR_PORT || 8020;
 const COORDINATOR_HOST = process.env.COORDINATOR_HOST || "127.0.0.1";
+
 const OTEL_EXPORTER_OTLP_ENDPOINT =
   process.env.OTEL_EXPORTER_OTLP_ENDPOINT || "http://0.0.0.0:4318";
 
+const FORCE_CHECKPOINT_SIMULATION = ["1", "true"].includes(
+  process.env.FORCE_CHECKPOINT_SIMULATION ?? "true"
+);
+
 const logger = new SimpleLogger(`[${MACHINE_NAME}]`);
 
 type InitializeReturn = {
@@ -278,7 +283,7 @@ class DockerTaskOperations implements TaskOperations {
 }
 
 const provider = new ProviderShell({
-  tasks: new DockerTaskOperations({ forceSimulate: true }),
+  tasks: new DockerTaskOperations({ forceSimulate: FORCE_CHECKPOINT_SIMULATION }),
   type: "docker",
 });
 

apps/kubernetes-provider/src/index.ts

@@ -212,7 +212,7 @@ class KubernetesTaskOperations implements TaskOperations {
             },
             {
               name: "populate-taskinfo",
-              image: "docker.io/library/busybox",
+              image: "registry.digitalocean.com/trigger/busybox",
               imagePullPolicy: "IfNotPresent",
               command: ["/bin/sh", "-c"],
               args: ["printenv COORDINATOR_HOST | tee /etc/taskinfo/coordinator-host"],
@@ -316,6 +316,9 @@ class KubernetesTaskOperations implements TaskOperations {
         {
           name: "registry-trigger",
         },
+        {
+          name: "registry-trigger-failover",
+        },
       ],
       nodeSelector: {
         nodetype: "worker",

apps/proxy/package.json

@@ -7,9 +7,9 @@
     "dev": "wrangler dev"
   },
   "devDependencies": {
-    "@cloudflare/workers-types": "^4.20230419.0",
+    "@cloudflare/workers-types": "^4.20240512.0",
     "typescript": "^5.0.4",
-    "wrangler": "^3.0.0"
+    "wrangler": "^3.57.1"
   },
   "dependencies": {
     "@aws-sdk/client-sqs": "^3.445.0",

apps/proxy/src/index.ts

@@ -1,5 +1,7 @@
 import { queueEvent } from "./events/queueEvent";
 import { queueEvents } from "./events/queueEvents";
+import { applyRateLimit } from "./rateLimit";
+import { Ratelimit } from "./rateLimiter";
 
 export interface Env {
   /** The hostname needs to be changed to allow requests to pass to the Trigger.dev platform */
@@ -9,6 +11,8 @@ export interface Env {
   AWS_SQS_SECRET_ACCESS_KEY: string;
   AWS_SQS_QUEUE_URL: string;
   AWS_SQS_REGION: string;
+  //rate limiter
+  API_RATE_LIMITER: Ratelimit;
 }
 
 export default {
@@ -25,13 +29,13 @@ export default {
     switch (url.pathname) {
       case "/api/v1/events": {
         if (request.method === "POST") {
-          return queueEvent(request, env);
+          return applyRateLimit(request, env, () => queueEvent(request, env));
         }
         break;
       }
       case "/api/v1/events/bulk": {
         if (request.method === "POST") {
-          return queueEvents(request, env);
+          return applyRateLimit(request, env, () => queueEvents(request, env));
         }
         break;
       }

apps/proxy/src/rateLimit.ts

@@ -0,0 +1,46 @@
+import { Env } from "src";
+import { getApiKeyFromRequest } from "./apikey";
+import { json } from "./json";
+
+export async function applyRateLimit(
+  request: Request,
+  env: Env,
+  fn: () => Promise<Response>
+): Promise<Response> {
+  const apiKey = getApiKeyFromRequest(request);
+  if (apiKey) {
+    const result = await env.API_RATE_LIMITER.limit({ key: `apikey-${apiKey.apiKey}` });
+    const { success } = result;
+    console.log(`Rate limiter`, {
+      success,
+      key: `${apiKey.apiKey.substring(0, 12)}...`,
+    });
+    if (!success) {
+      //60s in the future
+      const reset = Date.now() + 60 * 1000;
+      const secondsUntilReset = Math.max(0, (reset - new Date().getTime()) / 1000);
+
+      return json(
+        {
+          title: "Rate Limit Exceeded",
+          status: 429,
+          type: "https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/429",
+          detail: `Rate limit exceeded. Retry in ${secondsUntilReset} seconds.`,
+          error: `Rate limit exceeded. Retry in ${secondsUntilReset} seconds.`,
+          reset,
+        },
+        {
+          status: 429,
+          headers: {
+            "x-ratelimit-reset": reset.toString(),
+          },
+        }
+      );
+    }
+  } else {
+    console.log(`Rate limiter: no API key for request`);
+  }
+
+  //call the original function
+  return fn();
+}

apps/proxy/src/rateLimiter.ts

@@ -0,0 +1,23 @@
+export interface Ratelimit {
+  /*
+   * The ratelimit function
+   * @param {RatelimitOptions} options
+   * @returns {Promise<RatelimitResponse>}
+   */
+  limit: (options: RatelimitOptions) => Promise<RatelimitResponse>;
+}
+
+export interface RatelimitOptions {
+  /*
+   * The key to identify the user, can be an IP address, user ID, etc.
+   */
+  key: string;
+}
+
+export interface RatelimitResponse {
+  /*
+   * The ratelimit success status
+   * @returns {boolean}
+   */
+  success: boolean;
+}

apps/proxy/wrangler.toml

@@ -1,7 +1,33 @@
 name = "proxy"
 main = "src/index.ts"
-compatibility_date = "2023-10-30"
+compatibility_date = "2024-05-13"
 compatibility_flags = [ "nodejs_compat" ]
 
 [env.staging]
-[env.prod]
+  # The rate limiting API is in open beta.
+  [[env.staging.unsafe.bindings]]
+  name = "API_RATE_LIMITER"
+  type = "ratelimit"
+  # An identifier you define, that is unique to your Cloudflare account.
+  # Must be an integer.
+  namespace_id = "1"
+
+  # Limit: the number of tokens allowed within a given period in a single
+  # Cloudflare location
+  # Period: the duration of the period, in seconds. Must be either 10 or 60
+  simple = { limit = 100, period = 60 }
+
+
+[env.prod]
+  # The rate limiting API is in open beta.
+  [[env.prod.unsafe.bindings]]
+  name = "API_RATE_LIMITER"
+  type = "ratelimit"
+  # An identifier you define, that is unique to your Cloudflare account.
+  # Must be an integer.
+  namespace_id = "2"
+
+  # Limit: the number of tokens allowed within a given period in a single
+  # Cloudflare location
+  # Period: the duration of the period, in seconds. Must be either 10 or 60
+  simple = { limit = 300, period = 60 }

apps/webapp/app/consts.ts

@@ -13,3 +13,4 @@ export const VERCEL_RESPONSE_TIMEOUT_STATUS_CODES = [408, 504];
 export const MAX_BATCH_TRIGGER_ITEMS = 100;
 export const MAX_TASK_RUN_ATTEMPTS = 250;
 export const BULK_ACTION_RUN_LIMIT = 250;
+export const MAX_JOB_RUN_EXECUTION_COUNT = 250;

apps/webapp/app/env.server.ts

@@ -100,6 +100,10 @@ const EnvironmentSchema = z.object({
   API_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
   API_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
 
+  //Ingesting event rate limit
+  INGEST_EVENT_RATE_LIMIT_WINDOW: z.string().default("60s"),
+  INGEST_EVENT_RATE_LIMIT_MAX: z.coerce.number().int().optional(),
+
   //v3
   V3_ENABLED: z.string().default("false"),
   PROVIDER_SECRET: z.string().default("provider-secret"),
@@ -111,6 +115,7 @@ const EnvironmentSchema = z.object({
   CONTAINER_REGISTRY_USERNAME: z.string().optional(),
   CONTAINER_REGISTRY_PASSWORD: z.string().optional(),
   DEPLOY_REGISTRY_HOST: z.string().optional(),
+  DEPLOY_REGISTRY_NAMESPACE: z.string().default("trigger"),
   OBJECT_STORE_BASE_URL: z.string().optional(),
   OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
   OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
@@ -164,6 +169,22 @@ const EnvironmentSchema = z.object({
   ALERT_RESEND_API_KEY: z.string().optional(),
 
   MAX_SEQUENTIAL_INDEX_FAILURE_COUNT: z.coerce.number().default(96),
+
+  LOOPS_API_KEY: z.string().optional(),
+  MARQS_DISABLE_REBALANCING: z.coerce.boolean().default(false),
+
+  VERBOSE_GRAPHILE_LOGGING: z.string().default("false"),
+  V2_MARQS_ENABLED: z.string().default("0"),
+  V2_MARQS_CONSUMER_POOL_ENABLED: z.string().default("0"),
+  V2_MARQS_CONSUMER_POOL_SIZE: z.coerce.number().int().default(10),
+  V2_MARQS_CONSUMER_POLL_INTERVAL_MS: z.coerce.number().int().default(1000),
+  V2_MARQS_QUEUE_SELECTION_COUNT: z.coerce.number().int().default(36),
+  V2_MARQS_VISIBILITY_TIMEOUT_MS: z.coerce
+    .number()
+    .int()
+    .default(60 * 1000 * 15),
+  V2_MARQS_DEFAULT_ENV_CONCURRENCY: z.coerce.number().int().default(100),
+  V2_MARQS_VERBOSE: z.string().default("0"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

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

@@ -124,6 +124,7 @@ type DisplayableInputEnvironment = Prisma.RuntimeEnvironmentGetPayload<{
   select: {
     id: true;
     type: true;
+    slug: true;
     orgMember: {
       select: {
         user: {
@@ -138,17 +139,24 @@ type DisplayableInputEnvironment = Prisma.RuntimeEnvironmentGetPayload<{
   };
 }>;
 
-export function displayableEnvironments(
+export function displayableEnvironment(
   environment: DisplayableInputEnvironment,
   userId: string | undefined
 ) {
+  let userName: string | undefined = undefined;
+
+  if (environment.type === "DEVELOPMENT") {
+    if (!environment.orgMember) {
+      userName = "Deleted";
+    } else if (environment.orgMember.user.id !== userId) {
+      userName = getUsername(environment.orgMember.user);
+    }
+  }
+
   return {
     id: environment.id,
     type: environment.type,
-    userName: environment.orgMember
-      ? environment.orgMember.user.id === userId
-        ? undefined
-        : getUsername(environment.orgMember.user)
-      : undefined,
+    slug: environment.slug,
+    userName,
   };
 }

apps/webapp/app/platform/zodWorker.server.ts

@@ -10,7 +10,12 @@ import type {
   TaskSpec,
   WorkerUtils,
 } from "graphile-worker";
-import { run as graphileRun, makeWorkerUtils, parseCronItems } from "graphile-worker";
+import {
+  run as graphileRun,
+  makeWorkerUtils,
+  parseCronItems,
+  Logger as GraphileLogger,
+} from "graphile-worker";
 import { SpanKind, trace } from "@opentelemetry/api";
 
 import omit from "lodash.omit";
@@ -19,6 +24,7 @@ import { $replica, PrismaClient, PrismaClientOrTransaction } from "~/db.server";
 import { PgListenService } from "~/services/db/pgListen.server";
 import { workerLogger as logger } from "~/services/logger.server";
 import { flattenAttributes } from "@trigger.dev/core/v3";
+import { env } from "~/env.server";
 
 const tracer = trace.getTracer("zodWorker", "3.0.0.dp.1");
 
@@ -56,7 +62,6 @@ const AddJobResultsSchema = z.array(GraphileJobSchema);
 
 export type ZodTasks<TConsumerSchema extends MessageCatalogSchema> = {
   [K in keyof TConsumerSchema]: {
-    queueName?: string | ((payload: z.infer<TConsumerSchema[K]>) => string);
     jobKey?: string | ((payload: z.infer<TConsumerSchema[K]>) => string | undefined);
     priority?: number;
     maxAttempts?: number;
@@ -79,7 +84,9 @@ export type ZodRecurringTasks = {
   };
 };
 
-export type ZodWorkerEnqueueOptions = TaskSpec & {
+type ZodTaskSpec = Omit<TaskSpec, "queueName">;
+
+export type ZodWorkerEnqueueOptions = ZodTaskSpec & {
   tx?: PrismaClientOrTransaction;
 };
 
@@ -162,12 +169,25 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
 
     this.#workerUtils = await makeWorkerUtils(this.#runnerOptions);
 
+    const graphileLogger = new GraphileLogger((scope) => {
+      return (level, message, meta) => {
+        if (env.VERBOSE_GRAPHILE_LOGGING !== "true") return;
+
+        logger.debug(`[graphile-worker][${this.#name}][${level}] ${message}`, {
+          scope,
+          meta,
+          workerName: this.#name,
+        });
+      };
+    });
+
     this.#runner = await graphileRun({
       ...this.#runnerOptions,
       noHandleSignals: true,
       taskList: this.#createTaskListFromTasks(),
       parsedCronItems,
       forbiddenFlags: this.#rateLimiter?.forbiddenFlags.bind(this.#rateLimiter),
+      logger: graphileLogger,
     });
 
     if (!this.#runner) {
@@ -237,6 +257,20 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       this.#logDebug("stop");
     });
 
+    this.#runner?.events.on("worker:getJob:error", ({ worker, error }) => {
+      this.#logDebug("worker:getJob:error", { workerId: worker.workerId, error });
+    });
+
+    this.#runner?.events.on("worker:getJob:start", ({ worker }) => {
+      if (env.VERBOSE_GRAPHILE_LOGGING !== "true") return;
+      this.#logDebug("worker:getJob:start", { workerId: worker.workerId });
+    });
+
+    this.#runner?.events.on("job:start", ({ worker, job }) => {
+      if (env.VERBOSE_GRAPHILE_LOGGING !== "true") return;
+      this.#logDebug("job:start", { workerId: worker.workerId, job });
+    });
+
     process.on("SIGTERM", this._handleSignal.bind(this));
     process.on("SIGINT", this._handleSignal.bind(this));
 
@@ -250,16 +284,18 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
 
     this.#shuttingDown = true;
 
+    this.#logDebug(
+      `Received ${signal}, shutting down zodWorker with timeout ${this.#shutdownTimeoutInMs}ms`
+    );
+
     if (this.#shutdownTimeoutInMs) {
       setTimeout(() => {
-        this.#logDebug("Shutdown timeout reached, exiting process");
+        this.#logDebug(`Shutdown timeout of ${this.#shutdownTimeoutInMs} reached, exiting process`);
 
         process.exit(0);
       }, this.#shutdownTimeoutInMs);
     }
 
-    this.#logDebug(`Received ${signal}, shutting down zodWorker...`);
-
     this.stop().finally(() => {
       this.#logDebug("zodWorker stopped");
     });
@@ -286,10 +322,6 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       ...optionsWithoutTx,
     };
 
-    if (typeof task.queueName === "function") {
-      spec.queueName = task.queueName(payload);
-    }
-
     if (typeof task.jobKey === "function") {
       const jobKey = task.jobKey(payload);
 
@@ -345,17 +377,15 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       `SELECT * FROM ${this.graphileWorkerSchema}.add_job(
           identifier => $1::text,
           payload => $2::json,
-          queue_name => $3::text,
-          run_at => $4::timestamptz,
-          max_attempts => $5::int,
-          job_key => $6::text,
-          priority => $7::int,
-          flags => $8::text[],
-          job_key_mode => $9::text
+          run_at => $3::timestamptz,
+          max_attempts => $4::int,
+          job_key => $5::text,
+          priority => $6::int,
+          flags => $7::text[],
+          job_key_mode => $8::text
         )`,
       identifier,
       JSON.stringify(payload),
-      spec.queueName || null,
       spec.runAt || null,
       spec.maxAttempts || null,
       spec.jobKey || null,
@@ -447,33 +477,15 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     return taskList;
   }
 
-  async #getQueueName(queueId: number | null) {
-    if (queueId === null) {
-      return;
-    }
-
-    const schema = z.array(z.object({ queue_name: z.string() }));
-
-    const rawQueueNameResults = await $replica.$queryRawUnsafe(
-      `SELECT queue_name FROM ${this.graphileWorkerSchema}._private_job_queues WHERE id = $1`,
-      queueId
-    );
-
-    const queueNameResults = schema.parse(rawQueueNameResults);
-
-    return queueNameResults[0]?.queue_name;
-  }
-
   async #rescheduleTask(payload: unknown, helpers: JobHelpers) {
     this.#logDebug("Rescheduling task", { payload, job: helpers.job });
 
     await this.enqueue(helpers.job.task_identifier, payload, {
-      runAt: new Date(Date.now() + 1000 * 10),
-      queueName: await this.#getQueueName(helpers.job.job_queue_id),
+      runAt: helpers.job.run_at,
       priority: helpers.job.priority,
       jobKey: helpers.job.key ?? undefined,
       flags: Object.keys(helpers.job.flags ?? []),
-      maxAttempts: helpers.job.max_attempts,
+      maxAttempts: helpers.job.max_attempts - (helpers.job.attempts - 1),
     });
   }
 

apps/webapp/app/presenters/ProjectPresenter.server.ts

@@ -1,6 +1,6 @@
 import { PrismaClient, prisma } from "~/db.server";
 import { Project } from "~/models/project.server";
-import { displayableEnvironments } from "~/models/runtimeEnvironment.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { User } from "~/models/user.server";
 import { sortEnvironments } from "~/utils/environmentSort";
 
@@ -86,7 +86,7 @@ export class ProjectPresenter {
       httpEndpointCount: project._count.httpEndpoints,
       environments: sortEnvironments(
         project.environments.map((environment) => ({
-          ...displayableEnvironments(environment, userId),
+          ...displayableEnvironment(environment, userId),
           userId: environment.orgMember?.user.id,
         }))
       ),

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

@@ -1,5 +1,15 @@
-import { AttemptStatus, RetrieveRunResponse, RunStatus, logger } from "@trigger.dev/core/v3";
-import { TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
+import {
+  AttemptStatus,
+  RetrieveRunResponse,
+  RunStatus,
+  SerializedError,
+  TaskRunError,
+  conditionallyImportPacket,
+  createJsonErrorObject,
+  logger,
+  parsePacket,
+} from "@trigger.dev/core/v3";
+import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { BasePresenter } from "./basePresenter.server";
@@ -23,6 +33,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
             },
           },
           lockedToVersion: true,
+          schedule: true,
         },
       });
 
@@ -32,14 +43,65 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         return undefined;
       }
 
+      let $payload: any;
+      let $output: any;
+
+      if (showSecretDetails) {
+        const payloadPacket = await conditionallyImportPacket({
+          data: taskRun.payload,
+          dataType: taskRun.payloadType,
+        });
+
+        $payload = await parsePacket(payloadPacket);
+
+        if (taskRun.status === "COMPLETED_SUCCESSFULLY") {
+          const completedAttempt = taskRun.attempts.find(
+            (a) => a.status === "COMPLETED" && typeof a.output !== null
+          );
+
+          if (completedAttempt && completedAttempt.output) {
+            const outputPacket = await conditionallyImportPacket({
+              data: completedAttempt.output,
+              dataType: completedAttempt.outputType,
+            });
+
+            $output = await parsePacket(outputPacket);
+          }
+        }
+      }
+
+      const apiStatus = ApiRetrieveRunPresenter.apiStatusFromRunStatus(taskRun.status);
+
       return {
         id: taskRun.friendlyId,
-        status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(taskRun.status),
+        status: apiStatus,
         taskIdentifier: taskRun.taskIdentifier,
         idempotencyKey: taskRun.idempotencyKey ?? undefined,
         version: taskRun.lockedToVersion ? taskRun.lockedToVersion.version : undefined,
         createdAt: taskRun.createdAt ?? undefined,
         updatedAt: taskRun.updatedAt ?? undefined,
+        startedAt: taskRun.lockedAt ?? undefined,
+        finishedAt: ApiRetrieveRunPresenter.isStatusFinished(apiStatus)
+          ? taskRun.updatedAt
+          : undefined,
+        payload: $payload,
+        output: $output,
+        isTest: taskRun.isTest,
+        schedule: taskRun.schedule
+          ? {
+              id: taskRun.schedule.friendlyId,
+              externalId: taskRun.schedule.externalId ?? undefined,
+              deduplicationKey: taskRun.schedule.userProvidedDeduplicationKey
+                ? taskRun.schedule.deduplicationKey
+                : undefined,
+              generator: {
+                type: "CRON",
+                expression: taskRun.schedule.generatorExpression,
+                description: taskRun.schedule.generatorDescription,
+              },
+            }
+          : undefined,
+        ...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(apiStatus),
         attempts: !showSecretDetails
           ? []
           : taskRun.attempts.map((a) => ({
@@ -49,34 +111,68 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
               updatedAt: a.updatedAt ?? undefined,
               startedAt: a.startedAt ?? undefined,
               completedAt: a.completedAt ?? undefined,
+              error: ApiRetrieveRunPresenter.apiErrorFromError(a.error),
             })),
       };
     });
   }
 
+  static apiErrorFromError(error: Prisma.JsonValue): SerializedError | undefined {
+    if (!error) {
+      return;
+    }
+
+    const errorData = TaskRunError.safeParse(error);
+
+    if (errorData.success) {
+      return createJsonErrorObject(errorData.data);
+    }
+  }
+
+  static isStatusFinished(status: RunStatus) {
+    return (
+      status === "COMPLETED" ||
+      status === "FAILED" ||
+      status === "CANCELED" ||
+      status === "INTERRUPTED" ||
+      status === "CRASHED" ||
+      status === "SYSTEM_FAILURE"
+    );
+  }
+
   static apiStatusFromRunStatus(status: TaskRunStatus): RunStatus {
     switch (status) {
-      case "WAITING_FOR_DEPLOY":
-      case "PENDING": {
-        return "PENDING";
+      case "WAITING_FOR_DEPLOY": {
+        return "WAITING_FOR_DEPLOY";
+      }
+      case "PENDING": {
+        return "QUEUED";
+      }
+      case "PAUSED":
+      case "WAITING_TO_RESUME": {
+        return "FROZEN";
+      }
+      case "RETRYING_AFTER_FAILURE": {
+        return "REATTEMPTING";
       }
-      case "RETRYING_AFTER_FAILURE":
       case "EXECUTING": {
         return "EXECUTING";
       }
-      case "WAITING_TO_RESUME":
-      case "PAUSED": {
-        return "PAUSED";
-      }
       case "CANCELED": {
         return "CANCELED";
       }
       case "COMPLETED_SUCCESSFULLY": {
         return "COMPLETED";
       }
-      case "SYSTEM_FAILURE":
-      case "INTERRUPTED":
-      case "CRASHED":
+      case "SYSTEM_FAILURE": {
+        return "SYSTEM_FAILURE";
+      }
+      case "INTERRUPTED": {
+        return "INTERRUPTED";
+      }
+      case "CRASHED": {
+        return "CRASHED";
+      }
       case "COMPLETED_WITH_ERRORS": {
         return "FAILED";
       }
@@ -86,6 +182,30 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
     }
   }
 
+  static apiBooleanHelpersFromRunStatus(status: RunStatus) {
+    const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY";
+    const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN";
+    const isCompleted =
+      status === "COMPLETED" ||
+      status === "CANCELED" ||
+      status === "FAILED" ||
+      status === "CRASHED" ||
+      status === "INTERRUPTED" ||
+      status === "SYSTEM_FAILURE";
+    const isFailed = isCompleted && status !== "COMPLETED";
+    const isSuccess = isCompleted && status === "COMPLETED";
+    const isCancelled = status === "CANCELED";
+
+    return {
+      isQueued,
+      isExecuting,
+      isCompleted,
+      isFailed,
+      isSuccess,
+      isCancelled,
+    };
+  }
+
   static apiStatusFromAttemptStatus(status: TaskRunAttemptStatus): AttemptStatus {
     switch (status) {
       case "PENDING": {

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

@@ -0,0 +1,274 @@
+import { ListRunResponse, ListRunResponseItem, RunStatus } from "@trigger.dev/core/v3";
+import { Project, RuntimeEnvironment, TaskRunStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
+import { z } from "zod";
+import { fromZodError } from "zod-validation-error";
+import { logger } from "~/services/logger.server";
+import { ApiRetrieveRunPresenter } from "./ApiRetrieveRunPresenter.server";
+import { RunListOptions, RunListPresenter } from "./RunListPresenter.server";
+import { BasePresenter } from "./basePresenter.server";
+
+const SearchParamsSchema = z.object({
+  "page[size]": z.coerce.number().int().positive().min(1).max(100).optional(),
+  "page[after]": z.string().optional(),
+  "page[before]": z.string().optional(),
+  "filter[status]": z
+    .string()
+    .optional()
+    .transform((value, ctx) => {
+      if (!value) {
+        return undefined;
+      }
+
+      const statuses = value.split(",");
+      const parsedStatuses = statuses.map((status) => RunStatus.safeParse(status));
+
+      if (parsedStatuses.some((result) => !result.success)) {
+        const invalidStatuses: string[] = [];
+
+        for (const [index, result] of parsedStatuses.entries()) {
+          if (!result.success) {
+            invalidStatuses.push(statuses[index]);
+          }
+        }
+
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: `Invalid status values: ${invalidStatuses.join(", ")}`,
+        });
+
+        return z.NEVER;
+      }
+
+      const $statuses = parsedStatuses
+        .map((result) => (result.success ? result.data : undefined))
+        .filter(Boolean);
+
+      return Array.from(new Set($statuses));
+    }),
+  "filter[env]": z
+    .string()
+    .optional()
+    .transform((value) => {
+      return value ? value.split(",") : undefined;
+    }),
+  "filter[taskIdentifier]": z
+    .string()
+    .optional()
+    .transform((value) => {
+      return value ? value.split(",") : undefined;
+    }),
+  "filter[version]": z
+    .string()
+    .optional()
+    .transform((value) => {
+      return value ? value.split(",") : undefined;
+    }),
+  "filter[bulkAction]": z.string().optional(),
+  "filter[schedule]": z.string().optional(),
+  "filter[isTest]": z
+    .string()
+    .optional()
+    .transform((value, ctx) => {
+      if (!value) {
+        return undefined;
+      }
+
+      if (value === "true") {
+        return true;
+      }
+
+      if (value === "false") {
+        return false;
+      }
+
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: `Invalid value for isTest: ${value}`,
+      });
+
+      return z.NEVER;
+    }),
+  "filter[createdAt][from]": z.coerce.date().optional(),
+  "filter[createdAt][to]": z.coerce.date().optional(),
+  "filter[createdAt][period]": z.string().optional(),
+});
+
+type SearchParamsSchema = z.infer<typeof SearchParamsSchema>;
+
+export class ApiRunListPresenter extends BasePresenter {
+  public async call(
+    project: Project,
+    searchParams: URLSearchParams,
+    environment?: RuntimeEnvironment
+  ): Promise<ListRunResponse> {
+    return this.trace("call", async (span) => {
+      const rawSearchParams = Object.fromEntries(searchParams.entries());
+      const $searchParams = SearchParamsSchema.safeParse(rawSearchParams);
+
+      if (!$searchParams.success) {
+        logger.error("Invalid search params", {
+          searchParams: rawSearchParams,
+          errors: $searchParams.error.errors,
+        });
+
+        throw fromZodError($searchParams.error);
+      }
+
+      logger.debug("Valid search params", { searchParams: $searchParams.data });
+
+      const options: RunListOptions = {
+        projectId: project.id,
+      };
+
+      // pagination
+      if ($searchParams.data["page[size]"]) {
+        options.pageSize = $searchParams.data["page[size]"];
+      }
+
+      if ($searchParams.data["page[after]"]) {
+        options.cursor = $searchParams.data["page[after]"];
+        options.direction = "forward";
+      }
+
+      if ($searchParams.data["page[before]"]) {
+        options.cursor = $searchParams.data["page[before]"];
+        options.direction = "backward";
+      }
+
+      // filters
+      if (environment) {
+        options.environments = [environment.id];
+      } else {
+        if ($searchParams.data["filter[env]"]) {
+          const environments = await this._prisma.runtimeEnvironment.findMany({
+            where: {
+              projectId: project.id,
+              slug: {
+                in: $searchParams.data["filter[env]"],
+              },
+            },
+          });
+
+          options.environments = environments.map((env) => env.id);
+        }
+      }
+
+      if ($searchParams.data["filter[status]"]) {
+        options.statuses = $searchParams.data["filter[status]"].flatMap((status) =>
+          ApiRunListPresenter.apiStatusToRunStatuses(status)
+        );
+      }
+
+      if ($searchParams.data["filter[taskIdentifier]"]) {
+        options.tasks = $searchParams.data["filter[taskIdentifier]"];
+      }
+
+      if ($searchParams.data["filter[version]"]) {
+        options.versions = $searchParams.data["filter[version]"];
+      }
+
+      if ($searchParams.data["filter[bulkAction]"]) {
+        options.bulkId = $searchParams.data["filter[bulkAction]"];
+      }
+
+      if ($searchParams.data["filter[schedule]"]) {
+        options.scheduleId = $searchParams.data["filter[schedule]"];
+      }
+
+      if ($searchParams.data["filter[createdAt][from]"]) {
+        options.from = $searchParams.data["filter[createdAt][from]"].getTime();
+      }
+
+      if ($searchParams.data["filter[createdAt][to]"]) {
+        options.to = $searchParams.data["filter[createdAt][to]"].getTime();
+      }
+
+      if ($searchParams.data["filter[createdAt][period]"]) {
+        options.period = $searchParams.data["filter[createdAt][period]"];
+      }
+
+      if (typeof $searchParams.data["filter[isTest]"] === "boolean") {
+        options.isTest = $searchParams.data["filter[isTest]"];
+      }
+
+      const presenter = new RunListPresenter();
+
+      logger.debug("Calling RunListPresenter", { options });
+
+      const results = await presenter.call(options);
+
+      const data: ListRunResponseItem[] = results.runs.map((run) => {
+        return {
+          id: run.friendlyId,
+          status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status),
+          taskIdentifier: run.taskIdentifier,
+          idempotencyKey: run.idempotencyKey,
+          version: run.version ?? undefined,
+          createdAt: new Date(run.createdAt),
+          updatedAt: new Date(run.updatedAt),
+          startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
+          finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
+          isTest: run.isTest,
+          env: {
+            id: run.environment.id,
+            name: run.environment.slug,
+            user: run.environment.userName,
+          },
+          ...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
+            ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status)
+          ),
+        };
+      });
+
+      return {
+        data,
+        pagination: {
+          next: results.pagination.next,
+          previous: results.pagination.previous,
+        },
+      };
+    });
+  }
+
+  static apiStatusToRunStatuses(status: RunStatus): TaskRunStatus[] | TaskRunStatus {
+    switch (status) {
+      case "WAITING_FOR_DEPLOY": {
+        return "WAITING_FOR_DEPLOY";
+      }
+      case "QUEUED": {
+        return "PENDING";
+      }
+      case "EXECUTING": {
+        return "EXECUTING";
+      }
+      case "REATTEMPTING": {
+        return "RETRYING_AFTER_FAILURE";
+      }
+      case "FROZEN": {
+        return ["PAUSED", "WAITING_TO_RESUME"];
+      }
+      case "CANCELED": {
+        return "CANCELED";
+      }
+      case "COMPLETED": {
+        return "COMPLETED_SUCCESSFULLY";
+      }
+      case "SYSTEM_FAILURE": {
+        return "SYSTEM_FAILURE";
+      }
+      case "INTERRUPTED": {
+        return "INTERRUPTED";
+      }
+      case "CRASHED": {
+        return "CRASHED";
+      }
+      case "FAILED": {
+        return "COMPLETED_WITH_ERRORS";
+      }
+      default: {
+        assertNever(status);
+      }
+    }
+  }
+}

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

@@ -1,5 +1,6 @@
 import { RuntimeEnvironmentType } from "@trigger.dev/database";
 import { PrismaClient, prisma } from "~/db.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 
 type EditScheduleOptions = {
   userId: string;
@@ -67,19 +68,7 @@ export class EditSchedulePresenter {
     });
 
     const possibleEnvironments = project.environments.map((environment) => {
-      let userName: undefined | string;
-      if (environment.orgMember) {
-        if (environment.orgMember.user.id !== userId) {
-          userName =
-            environment.orgMember.user.displayName ?? environment.orgMember.user.name ?? undefined;
-        }
-      }
-
-      return {
-        id: environment.id,
-        type: environment.type,
-        userName,
-      };
+      return displayableEnvironment(environment, userId);
     });
 
     return {

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

@@ -79,6 +79,19 @@ export class EnvironmentVariablesPresenter {
         project: {
           slug: projectSlug,
         },
+        OR: [
+          {
+            type: {
+              in: ["PREVIEW", "STAGING", "PRODUCTION"],
+            },
+          },
+          {
+            type: "DEVELOPMENT",
+            orgMember: {
+              userId,
+            },
+          },
+        ],
       },
     });
 

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

@@ -3,13 +3,13 @@ import parse from "parse-duration";
 import { Direction } from "~/components/runs/RunStatuses";
 import { FINISHED_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 import { sqlDatabaseSchema } from "~/db.server";
-import { displayableEnvironments } from "~/models/runtimeEnvironment.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { CANCELLABLE_STATUSES } from "~/v3/services/cancelTaskRun.server";
 import { BasePresenter } from "./basePresenter.server";
 
-type RunListOptions = {
+export type RunListOptions = {
   userId?: string;
-  projectSlug: string;
+  projectId: string;
   //filters
   tasks?: string[];
   versions?: string[];
@@ -20,6 +20,7 @@ type RunListOptions = {
   bulkId?: string;
   from?: number;
   to?: number;
+  isTest?: boolean;
   //pagination
   direction?: Direction;
   cursor?: string;
@@ -35,7 +36,7 @@ export type RunListAppliedFilters = RunList["filters"];
 export class RunListPresenter extends BasePresenter {
   public async call({
     userId,
-    projectSlug,
+    projectId,
     tasks,
     versions,
     statuses,
@@ -43,6 +44,7 @@ export class RunListPresenter extends BasePresenter {
     scheduleId,
     period,
     bulkId,
+    isTest,
     from,
     to,
     direction = "forward",
@@ -59,7 +61,9 @@ export class RunListPresenter extends BasePresenter {
       (period !== undefined && period !== "all") ||
       (bulkId !== undefined && bulkId !== "") ||
       from !== undefined ||
-      to !== undefined;
+      to !== undefined ||
+      (scheduleId !== undefined && scheduleId !== "") ||
+      typeof isTest === "boolean";
 
     // Find the project scoped to the organization
     const project = await this._replica.project.findFirstOrThrow({
@@ -85,7 +89,7 @@ export class RunListPresenter extends BasePresenter {
         },
       },
       where: {
-        slug: projectSlug,
+        id: projectId,
       },
     });
 
@@ -156,6 +160,7 @@ export class RunListPresenter extends BasePresenter {
         updatedAt: Date;
         isTest: boolean;
         spanId: string;
+        idempotencyKey: string | null;
       }[]
     >`
     SELECT
@@ -170,7 +175,8 @@ export class RunListPresenter extends BasePresenter {
     tr."lockedAt" AS "lockedAt",
     tr."updatedAt" AS "updatedAt",
     tr."isTest" AS "isTest",
-    tr."spanId" AS "spanId"
+    tr."spanId" AS "spanId",
+    tr."idempotencyKey" AS "idempotencyKey"
   FROM
     ${sqlDatabaseSchema}."TaskRun" tr
   LEFT JOIN
@@ -210,6 +216,7 @@ export class RunListPresenter extends BasePresenter {
           : Prisma.empty
       }
       ${scheduleId ? Prisma.sql`AND tr."scheduleId" = ${scheduleId}` : Prisma.empty}
+      ${typeof isTest === "boolean" ? Prisma.sql`AND tr."isTest" = ${isTest}` : Prisma.empty}
       ${
         periodMs
           ? Prisma.sql`AND tr."createdAt" >= NOW() - INTERVAL '1 millisecond' * ${periodMs}`
@@ -270,6 +277,7 @@ export class RunListPresenter extends BasePresenter {
           friendlyId: run.runFriendlyId,
           number: Number(run.number),
           createdAt: run.createdAt.toISOString(),
+          updatedAt: run.updatedAt.toISOString(),
           startedAt: run.lockedAt ? run.lockedAt.toISOString() : undefined,
           hasFinished,
           finishedAt: hasFinished ? run.updatedAt.toISOString() : undefined,
@@ -280,7 +288,8 @@ export class RunListPresenter extends BasePresenter {
           spanId: run.spanId,
           isReplayable: true,
           isCancellable: CANCELLABLE_STATUSES.includes(run.status),
-          environment: displayableEnvironments(environment, userId),
+          environment: displayableEnvironment(environment, userId),
+          idempotencyKey: run.idempotencyKey ? run.idempotencyKey : undefined,
         };
       }),
       pagination: {

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

@@ -1,6 +1,7 @@
 import { Prisma, RuntimeEnvironmentType } from "@trigger.dev/database";
 import { ScheduleListFilters } from "~/components/runs/v3/ScheduleFilters";
 import { PrismaClient, prisma, sqlDatabaseSchema } from "~/db.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { getUsername } from "~/utils/username";
 import { calculateNextScheduledTimestamp } from "~/v3/utils/calculateNextSchedule.server";
 
@@ -233,14 +234,7 @@ export class ScheduleListPresenter {
             );
           }
 
-          return {
-            id: instance.environmentId,
-            type: environment.type,
-            userName:
-              environment.orgMember?.user.id === userId
-                ? undefined
-                : getUsername(environment.orgMember?.user),
-          };
+          return displayableEnvironment(environment, userId);
         }),
       };
     });
@@ -252,14 +246,7 @@ export class ScheduleListPresenter {
       schedules,
       possibleTasks: possibleTasks.map((task) => task.slug),
       possibleEnvironments: project.environments.map((environment) => {
-        return {
-          id: environment.id,
-          type: environment.type,
-          userName:
-            environment.orgMember?.user.id === userId
-              ? undefined
-              : getUsername(environment.orgMember?.user),
-        };
+        return displayableEnvironment(environment, userId);
       }),
       hasFilters,
       filters: {

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

@@ -8,9 +8,9 @@ import { QUEUED_STATUSES, RUNNING_STATUSES } from "~/components/runs/v3/TaskRunS
 import { sqlDatabaseSchema } from "~/db.server";
 import type { Organization } from "~/models/organization.server";
 import type { Project } from "~/models/project.server";
-import { displayableEnvironments } from "~/models/runtimeEnvironment.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 import type { User } from "~/models/user.server";
-import { sortEnvironments } from "~/utils/environmentSort";
+import { filterOrphanedEnvironments, sortEnvironments } from "~/utils/environmentSort";
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
 import { TaskRunStatus } from "~/database-types";
@@ -86,7 +86,9 @@ export class TaskListPresenter extends BasePresenter {
     WITH workers AS (
       SELECT DISTINCT ON ("runtimeEnvironmentId") id, "runtimeEnvironmentId", version
       FROM ${sqlDatabaseSchema}."BackgroundWorker"
-      WHERE "runtimeEnvironmentId" IN (${Prisma.join(project.environments.map((e) => e.id))})
+      WHERE "runtimeEnvironmentId" IN (${Prisma.join(
+        filterOrphanedEnvironments(project.environments).map((e) => e.id)
+      )})
       ORDER BY "runtimeEnvironmentId", "createdAt" DESC
     )
     SELECT tasks.id, slug, "filePath", "exportName", "triggerSource", tasks."runtimeEnvironmentId", tasks."createdAt"
@@ -119,7 +121,7 @@ export class TaskListPresenter extends BasePresenter {
         existingTask.triggerSource = task.triggerSource;
       }
 
-      existingTask.environments.push(displayableEnvironments(environment, userId));
+      existingTask.environments.push(displayableEnvironment(environment, userId));
 
       //order the environments
       existingTask.environments = sortEnvironments(existingTask.environments);

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

@@ -36,9 +36,12 @@ export class TestPresenter {
           where: {
             OR: [
               {
-                orgMember: null,
+                type: {
+                  in: ["PREVIEW", "STAGING", "PRODUCTION"],
+                },
               },
               {
+                type: "DEVELOPMENT",
                 orgMember: {
                   userId,
                 },

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

@@ -2,6 +2,7 @@ import { PrismaClient, prisma } from "~/db.server";
 import { nextScheduledTimestamps } from "~/v3/utils/calculateNextSchedule.server";
 import { RunListPresenter } from "./RunListPresenter.server";
 import { ScheduleObject } from "@trigger.dev/core/v3";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 
 type ViewScheduleOptions = {
   userId?: string;
@@ -29,7 +30,7 @@ export class ViewSchedulePresenter {
         taskIdentifier: true,
         project: {
           select: {
-            slug: true,
+            id: true,
           },
         },
         instances: {
@@ -38,6 +39,7 @@ export class ViewSchedulePresenter {
               select: {
                 id: true,
                 type: true,
+                slug: true,
                 orgMember: {
                   select: {
                     user: {
@@ -70,8 +72,9 @@ export class ViewSchedulePresenter {
       : [];
 
     const runPresenter = new RunListPresenter(this.#prismaClient);
+
     const { runs } = await runPresenter.call({
-      projectSlug: schedule.project.slug,
+      projectId: schedule.project.id,
       scheduleId: schedule.id,
       pageSize: 5,
     });
@@ -85,21 +88,7 @@ export class ViewSchedulePresenter {
         runs,
         environments: schedule.instances.map((instance) => {
           const environment = instance.environment;
-          let userName: undefined | string;
-          if (environment.orgMember) {
-            if (environment.orgMember.user.id !== userId) {
-              userName =
-                environment.orgMember.user.displayName ??
-                environment.orgMember.user.name ??
-                undefined;
-            }
-          }
-
-          return {
-            id: environment.id,
-            type: environment.type,
-            userName,
-          };
+          return displayableEnvironment(environment, userId);
         }),
       },
     };

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

@@ -34,4 +34,26 @@ export abstract class BasePresenter {
       }
     );
   }
+
+  protected async trace<T>(trace: string, fn: (span: Span) => Promise<T>): Promise<T> {
+    return tracer.startActiveSpan(
+      `${this.constructor.name}.${trace}`,
+      { kind: SpanKind.SERVER },
+      async (span) => {
+        try {
+          return await fn(span);
+        } catch (e) {
+          if (e instanceof Error) {
+            span.recordException(e);
+          } else {
+            span.recordException(new Error(String(e)));
+          }
+
+          throw e;
+        } finally {
+          span.end();
+        }
+      }
+    );
+  }
 }

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

@@ -39,6 +39,7 @@ import { cn } from "~/utils/cn";
 import { ProjectParamSchema, v3ProjectPath, v3RunsPath, v3TestPath } from "~/utils/pathBuilder";
 import { ListPagination } from "../../components/ListPagination";
 import { BULK_ACTION_RUN_LIMIT } from "~/consts";
+import { findProjectBySlug } from "~/models/project.server";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
@@ -57,10 +58,16 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const { tasks, versions, statuses, environments, period, bulkId, from, to, cursor, direction } =
     TaskRunListSearchFilters.parse(s);
 
+  const project = await findProjectBySlug(organizationSlug, projectParam, userId);
+
+  if (!project) {
+    throw new Error("Project not found");
+  }
+
   const presenter = new RunListPresenter();
   const list = presenter.call({
     userId,
-    projectSlug: projectParam,
+    projectId: project.id,
     tasks,
     versions,
     statuses,

apps/webapp/app/routes/admin.api.v1.marqs.ts

@@ -0,0 +1,31 @@
+import { LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  // 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 details = await marqs?.getSharedQueueDetails();
+
+  return json(details);
+}

apps/webapp/app/routes/api.v1.projects.$projectRef.runs.ts

@@ -0,0 +1,62 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { ValidationError } from "zod-validation-error";
+import { findProjectByRef } from "~/models/project.server";
+import { ApiRunListPresenter } from "~/presenters/v3/ApiRunListPresenter.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { apiCors } from "~/utils/apiCors";
+
+const ParamsSchema = z.object({
+  projectRef: z.string(),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  if (request.method.toUpperCase() === "OPTIONS") {
+    return apiCors(request, json({}));
+  }
+
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return apiCors(request, json({ error: "Invalid or Missing API key" }, { status: 401 }));
+  }
+
+  const $params = ParamsSchema.safeParse(params);
+
+  if (!$params.success) {
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  const project = await findProjectByRef($params.data.projectRef, authenticationResult.userId);
+
+  if (!project) {
+    return json({ error: "Project not found" }, { status: 404 });
+  }
+
+  const url = new URL(request.url);
+
+  const presenter = new ApiRunListPresenter();
+
+  try {
+    const result = await presenter.call(project, url.searchParams);
+
+    if (!result) {
+      return apiCors(request, json({ data: [] }));
+    }
+
+    return apiCors(request, json(result));
+  } catch (error) {
+    if (error instanceof ValidationError) {
+      return apiCors(
+        request,
+        json({ error: "Query Error", details: error.details }, { status: 400 })
+      );
+    } else {
+      return apiCors(
+        request,
+        json({ error: error instanceof Error ? error.message : String(error) }, { status: 400 })
+      );
+    }
+  }
+}

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

@@ -0,0 +1,45 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { CreateTaskRunAttemptService } from "~/v3/services/createTaskRunAttempt.server";
+
+const ParamsSchema = z.object({
+  /* This is the run friendly ID */
+  runParam: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // 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;
+
+  const service = new CreateTaskRunAttemptService();
+
+  try {
+    const { execution } = await service.call(runParam, authenticationResult.environment);
+
+    return json(execution, { status: 200 });
+  } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: error.status ?? 422 });
+    }
+
+    return json(
+      { error: error instanceof Error ? error.message : "Internal Server Error" },
+      { status: 500 }
+    );
+  }
+}

apps/webapp/app/routes/api.v1.runs.ts

@@ -0,0 +1,52 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { ValidationError } from "zod-validation-error";
+import { ApiRunListPresenter } from "~/presenters/v3/ApiRunListPresenter.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { apiCors } from "~/utils/apiCors";
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  if (request.method.toUpperCase() === "OPTIONS") {
+    return apiCors(request, json({}));
+  }
+
+  const authenticationResult = await authenticateApiRequest(request, {
+    allowPublicKey: false,
+  });
+
+  if (!authenticationResult) {
+    return apiCors(request, json({ error: "Invalid or Missing API key" }, { status: 401 }));
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  const url = new URL(request.url);
+
+  const presenter = new ApiRunListPresenter();
+
+  try {
+    const result = await presenter.call(
+      authenticatedEnv.project,
+      url.searchParams,
+      authenticatedEnv
+    );
+
+    if (!result) {
+      return apiCors(request, json({ data: [] }));
+    }
+
+    return apiCors(request, json(result));
+  } catch (error) {
+    if (error instanceof ValidationError) {
+      return apiCors(
+        request,
+        json({ error: "Query Error", details: error.details }, { status: 400 })
+      );
+    } else {
+      return apiCors(
+        request,
+        json({ error: error instanceof Error ? error.message : String(error) }, { status: 400 })
+      );
+    }
+  }
+}

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

@@ -48,5 +48,5 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Internal Server Error" }, { status: 500 });
   }
 
-  return json({ message: "Run cancelled" }, { status: 200 });
+  return json({ id: runParam }, { status: 200 });
 }

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

@@ -1,45 +1,21 @@
 import { Ratelimit } from "@upstash/ratelimit";
 import { Request as ExpressRequest, Response as ExpressResponse, NextFunction } from "express";
-import Redis, { RedisOptions } from "ioredis";
+import { RedisOptions } from "ioredis";
 import { createHash } from "node:crypto";
 import { env } from "~/env.server";
 import { logger } from "./logger.server";
-
-function createRedisRateLimitClient(
-  redisOptions: RedisOptions
-): ConstructorParameters<typeof Ratelimit>[0]["redis"] {
-  const redis = new Redis(redisOptions);
-
-  return {
-    sadd: async <TData>(key: string, ...members: TData[]): Promise<number> => {
-      return redis.sadd(key, members as (string | number | Buffer)[]);
-    },
-    eval: <TArgs extends unknown[], TData = unknown>(
-      ...args: [script: string, keys: string[], args: TArgs]
-    ): Promise<TData> => {
-      const script = args[0];
-      const keys = args[1];
-      const argsArray = args[2];
-      return redis.eval(
-        script,
-        keys.length,
-        ...keys,
-        ...(argsArray as (string | Buffer | number)[])
-      ) as Promise<TData>;
-    },
-  };
-}
+import { Duration, Limiter, RateLimiter, createRedisRateLimitClient } from "./rateLimiter.server";
 
 type Options = {
+  redis?: RedisOptions;
+  keyPrefix: string;
+  pathMatchers: (RegExp | string)[];
+  pathWhiteList?: (RegExp | string)[];
+  limiter: Limiter;
   log?: {
     requests?: boolean;
     rejections?: boolean;
   };
-  redis: RedisOptions;
-  keyPrefix: string;
-  pathMatchers: (RegExp | string)[];
-  pathWhiteList?: (RegExp | string)[];
-  limiter: ConstructorParameters<typeof Ratelimit>[0]["limiter"];
 };
 
 //returns an Express middleware that rate limits using the Bearer token in the Authorization header
@@ -54,12 +30,12 @@ export function authorizationRateLimitMiddleware({
     requests: true,
   },
 }: Options) {
-  const rateLimiter = new Ratelimit({
-    redis: createRedisRateLimitClient(redis),
-    limiter: limiter,
-    ephemeralCache: new Map(),
-    analytics: false,
-    prefix: keyPrefix,
+  const rateLimiter = new RateLimiter({
+    redis,
+    keyPrefix,
+    limiter,
+    logSuccess: log.requests,
+    logFailure: log.rejections,
   });
 
   return async (req: ExpressRequest, res: ExpressResponse, next: NextFunction) => {
@@ -135,27 +111,9 @@ export function authorizationRateLimitMiddleware({
     res.set("x-ratelimit-reset", reset.toString());
 
     if (success) {
-      if (log.requests) {
-        logger.info(`RateLimiter (${keyPrefix}): under rate limit`, {
-          limit,
-          reset,
-          remaining,
-          hashedAuthorizationValue,
-        });
-      }
       return next();
     }
 
-    if (log.rejections) {
-      logger.warn(`RateLimiter (${keyPrefix}): rate limit exceeded`, {
-        limit,
-        reset,
-        remaining,
-        pending,
-        hashedAuthorizationValue,
-      });
-    }
-
     res.setHeader("Content-Type", "application/problem+json");
     const secondsUntilReset = Math.max(0, (reset - new Date().getTime()) / 1000);
     return res.status(429).send(
@@ -167,6 +125,7 @@ export function authorizationRateLimitMiddleware({
           detail: `Rate limit exceeded ${remaining}/${limit} requests remaining. Retry in ${secondsUntilReset} seconds.`,
           reset,
           limit,
+          remaining,
           secondsUntilReset,
           error: `Rate limit exceeded ${remaining}/${limit} requests remaining. Retry in ${secondsUntilReset} seconds.`,
         },
@@ -177,18 +136,8 @@ export function authorizationRateLimitMiddleware({
   };
 }
 
-type Duration = Parameters<typeof Ratelimit.slidingWindow>[1];
-
 export const apiRateLimiter = authorizationRateLimitMiddleware({
-  keyPrefix: "ratelimit:api",
-  redis: {
-    port: env.REDIS_PORT,
-    host: env.REDIS_HOST,
-    username: env.REDIS_USERNAME,
-    password: env.REDIS_PASSWORD,
-    enableAutoPipelining: true,
-    ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
-  },
+  keyPrefix: "api",
   limiter: Ratelimit.slidingWindow(env.API_RATE_LIMIT_MAX, env.API_RATE_LIMIT_WINDOW as Duration),
   pathMatchers: [/^\/api/],
   // Allow /api/v1/tasks/:id/callback/:secret

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

@@ -0,0 +1,95 @@
+import Redis, { RedisOptions } from "ioredis";
+import {
+  $transaction,
+  Prisma,
+  PrismaClientOrTransaction,
+  PrismaTransactionOptions,
+  prisma,
+} from "~/db.server";
+import { env } from "~/env.server";
+import { singleton } from "~/utils/singleton";
+
+export type AutoIncrementCounterOptions = {
+  redis: RedisOptions;
+};
+
+export class AutoIncrementCounter {
+  private _redis: Redis;
+
+  constructor(private options: AutoIncrementCounterOptions) {
+    this._redis = new Redis(options.redis);
+  }
+
+  async incrementInTransaction<T>(
+    key: string,
+    callback: (num: number, tx: PrismaClientOrTransaction) => Promise<T>,
+    backfiller?: (key: string, db: PrismaClientOrTransaction) => Promise<number | undefined>,
+    client: PrismaClientOrTransaction = prisma,
+    transactionOptions?: PrismaTransactionOptions
+  ): Promise<T | undefined> {
+    let performedIncrement = false;
+    let performedBackfill = false;
+
+    try {
+      return await $transaction(
+        client,
+        async (tx) => {
+          let newNumber = await this.#increment(key);
+
+          performedIncrement = true;
+
+          if (newNumber === 1 && backfiller) {
+            const backfilledNumber = await backfiller(key, tx);
+
+            if (backfilledNumber && backfilledNumber > 1) {
+              newNumber = backfilledNumber + 1;
+              await this._redis.set(key, newNumber);
+              performedBackfill = true;
+            }
+          }
+
+          return await callback(newNumber, tx);
+        },
+        transactionOptions
+      );
+    } catch (e) {
+      if (
+        e instanceof Prisma.PrismaClientKnownRequestError ||
+        e instanceof Prisma.PrismaClientUnknownRequestError ||
+        e instanceof Prisma.PrismaClientValidationError
+      ) {
+        if (performedIncrement && !performedBackfill) {
+          await this._redis.decr(key);
+        }
+      }
+
+      throw e;
+    }
+  }
+
+  async #increment(key: string): Promise<number> {
+    return await this._redis.incr(key);
+  }
+}
+
+export const autoIncrementCounter = singleton("auto-increment-counter", getAutoIncrementCounter);
+
+function getAutoIncrementCounter() {
+  if (!env.REDIS_HOST || !env.REDIS_PORT) {
+    throw new Error(
+      "Could not initialize auto-increment counter because process.env.REDIS_HOST and process.env.REDIS_PORT are required to be set. "
+    );
+  }
+
+  return new AutoIncrementCounter({
+    redis: {
+      keyPrefix: "auto-counter:",
+      port: env.REDIS_PORT,
+      host: env.REDIS_HOST,
+      username: env.REDIS_USERNAME,
+      password: env.REDIS_PASSWORD,
+      enableAutoPipelining: true,
+      ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
+  });
+}

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

@@ -124,6 +124,7 @@ export class EndpointApi {
         "x-trigger-action": "INDEX_ENDPOINT",
       },
       redirect: "manual",
+      signal: AbortSignal.timeout(5_000),
     });
 
     return {

apps/webapp/app/services/endpoints/recurringEndpointIndex.server.ts

@@ -27,7 +27,7 @@ export class RecurringEndpointIndexService {
         indexings: {
           none: {
             createdAt: {
-              gt: new Date(currentTimestamp - 10 * 60 * 1000),
+              gt: new Date(currentTimestamp - 60 * 60 * 1000),
             },
           },
         },

apps/webapp/app/services/events/ingestSendEvent.server.ts

@@ -4,6 +4,10 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { workerQueue } from "~/services/worker.server";
 import { logger } from "../logger.server";
 import { EventRecord, ExternalAccount } from "@trigger.dev/database";
+import { Duration, RateLimiter } from "../rateLimiter.server";
+import { Ratelimit } from "@upstash/ratelimit";
+import { env } from "~/env.server";
+import { singleton } from "~/utils/singleton";
 
 type UpdateEventInput = {
   tx: PrismaClientOrTransaction;
@@ -29,6 +33,20 @@ type EventSource = {
 
 const EVENT_UPDATE_THRESHOLD_WINDOW_IN_MSECS = 5 * 1000; // 5 seconds
 
+const rateLimiter = singleton("eventRateLimiter", getSharedRateLimiter);
+
+function getSharedRateLimiter() {
+  if (env.INGEST_EVENT_RATE_LIMIT_MAX) {
+    return new RateLimiter({
+      keyPrefix: "ingestsendevent",
+      limiter: Ratelimit.slidingWindow(
+        env.INGEST_EVENT_RATE_LIMIT_MAX,
+        env.INGEST_EVENT_RATE_LIMIT_WINDOW as Duration
+      ),
+    });
+  }
+}
+
 export class IngestSendEvent {
   #prismaClient: PrismaClientOrTransaction;
 
@@ -65,7 +83,7 @@ export class IngestSendEvent {
         return;
       }
 
-      return await $transaction(this.#prismaClient, async (tx) => {
+      const createdEvent = await $transaction(this.#prismaClient, async (tx) => {
         const externalAccount = options?.accountId
           ? await tx.externalAccount.upsert({
               where: {
@@ -106,6 +124,23 @@ export class IngestSendEvent {
 
         return eventLog;
       });
+
+      if (!createdEvent) return;
+
+      //rate limit
+      const result = await rateLimiter?.limit(environment.organizationId);
+      if (result && !result.success) {
+        logger.info("IngestSendEvent: Rate limit exceeded", {
+          eventRecordId: createdEvent.id,
+          organizationId: environment.organizationId,
+          reset: result.reset,
+          limit: result.limit,
+        });
+        return;
+      }
+
+      await this.enqueueWorkerEvent(this.#prismaClient, createdEvent);
+      return createdEvent;
     } catch (error) {
       const prismaError = PrismaErrorSchema.safeParse(error);
 
@@ -151,8 +186,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, eventLog);
-
     return eventLog;
   }
 
@@ -177,8 +210,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, updatedEventLog);
-
     return updatedEventLog;
   }
 

apps/webapp/app/services/jobs/registerJob.server.ts

@@ -6,11 +6,11 @@ import {
   assertExhaustive,
 } from "@trigger.dev/core";
 import type { Endpoint, Integration, Job, JobIntegration, JobVersion } from "@trigger.dev/database";
-import { DEFAULT_MAX_CONCURRENT_RUNS } from "~/consts";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
 import type { RuntimeEnvironment } from "~/models/runtimeEnvironment.server";
+import { putConcurrencyLimitGroup, putJobConcurrencyLimit } from "~/v3/marqs/v2.server";
 import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { logger } from "../logger.server";
 import { RegisterScheduleSourceService } from "../schedules/registerScheduleSource.server";
@@ -175,13 +175,17 @@ export class RegisterJobService {
 
     try {
       if (jobVersion.concurrencyLimitGroup) {
-        // Upsert the maxSize for the concurrency limit group
+        // Upsert the maxSize for the concurrency limit group (marqs v2)
+        await putConcurrencyLimitGroup(jobVersion.concurrencyLimitGroup, environment);
+
+        // Upsert the maxSize for the concurrency limit group (legacy)
         await executionRateLimiter?.putConcurrencyLimitGroup(
           jobVersion.concurrencyLimitGroup,
           environment
         );
       }
 
+      await putJobConcurrencyLimit(job, jobVersion, environment);
       await executionRateLimiter?.putJobVersionConcurrencyLimit(jobVersion, environment);
     } catch (error) {
       logger.error("Error setting concurrency limit", {

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

@@ -0,0 +1,75 @@
+import { env } from "~/env.server";
+import { logger } from "./logger.server";
+
+class LoopsClient {
+  constructor(private readonly apiKey: string) {}
+
+  async userCreated({
+    userId,
+    email,
+    name,
+  }: {
+    userId: string;
+    email: string;
+    name: string | null;
+  }) {
+    logger.info(`Loops send "sign-up" event`, { userId, email, name });
+    return this.#sendEvent({
+      email,
+      userId,
+      firstName: name?.split(" ").at(0),
+      eventName: "sign-up",
+    });
+  }
+
+  async #sendEvent({
+    email,
+    userId,
+    firstName,
+    eventName,
+    eventProperties,
+  }: {
+    email: string;
+    userId: string;
+    firstName?: string;
+    eventName: string;
+    eventProperties?: Record<string, string | number | boolean>;
+  }) {
+    const options = {
+      method: "POST",
+      headers: { Authorization: `Bearer ${this.apiKey}`, "Content-Type": "application/json" },
+      body: JSON.stringify({
+        email,
+        userId,
+        firstName,
+        eventName,
+        eventProperties,
+      }),
+    };
+
+    try {
+      const response = await fetch("https://app.loops.so/api/v1/events/send", options);
+
+      if (!response.ok) {
+        logger.error(`Loops sendEvent ${eventName} bad status`, { status: response.status });
+        return false;
+      }
+
+      const responseBody = (await response.json()) as any;
+
+      if (!responseBody.success) {
+        logger.error(`Loops sendEvent ${eventName} failed response`, {
+          message: responseBody.message,
+        });
+        return false;
+      }
+
+      return true;
+    } catch (error) {
+      logger.error(`Loops sendEvent ${eventName} failed`, { error });
+      return false;
+    }
+  }
+}
+
+export const loopsClient = env.LOOPS_API_KEY ? new LoopsClient(env.LOOPS_API_KEY) : null;

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

@@ -0,0 +1,105 @@
+import { Ratelimit } from "@upstash/ratelimit";
+import Redis, { RedisOptions } from "ioredis";
+import { env } from "~/env.server";
+import { logger } from "./logger.server";
+
+type Options = {
+  redis?: RedisOptions;
+  keyPrefix: string;
+  limiter: Limiter;
+  logSuccess?: boolean;
+  logFailure?: boolean;
+};
+
+export type Limiter = ConstructorParameters<typeof Ratelimit>[0]["limiter"];
+export type Duration = Parameters<typeof Ratelimit.slidingWindow>[1];
+export type RateLimitResponse = Awaited<ReturnType<Ratelimit["limit"]>>;
+
+export class RateLimiter {
+  #ratelimit: Ratelimit;
+
+  constructor(private readonly options: Options) {
+    const { redis, keyPrefix, limiter } = options;
+    const prefix = `ratelimit:${keyPrefix}`;
+    this.#ratelimit = new Ratelimit({
+      redis: createRedisRateLimitClient(
+        redis ?? {
+          port: env.REDIS_PORT,
+          host: env.REDIS_HOST,
+          username: env.REDIS_USERNAME,
+          password: env.REDIS_PASSWORD,
+          enableAutoPipelining: true,
+          ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+        }
+      ),
+      limiter,
+      ephemeralCache: new Map(),
+      analytics: false,
+      prefix,
+    });
+
+    logger.info(`RateLimiter (${keyPrefix}): initialized`, {
+      keyPrefix,
+      redisKeyspace: prefix,
+    });
+  }
+
+  async limit(identifier: string, rate = 1): Promise<RateLimitResponse> {
+    const result = this.#ratelimit.limit(identifier, { rate });
+    const { success, limit, reset, remaining } = await result;
+
+    if (success && this.options.logSuccess) {
+      logger.info(`RateLimiter (${this.options.keyPrefix}): under rate limit`, {
+        limit,
+        reset,
+        remaining,
+        identifier,
+      });
+    }
+
+    //log these by default
+    if (!success && this.options.logFailure !== false) {
+      logger.info(`RateLimiter (${this.options.keyPrefix}): rate limit exceeded`, {
+        limit,
+        reset,
+        remaining,
+        identifier,
+      });
+    }
+
+    return result;
+  }
+}
+
+export function createRedisRateLimitClient(
+  redisOptions: RedisOptions
+): ConstructorParameters<typeof Ratelimit>[0]["redis"] {
+  const redis = new Redis(redisOptions);
+
+  return {
+    sadd: async <TData>(key: string, ...members: TData[]): Promise<number> => {
+      return redis.sadd(key, members as (string | number | Buffer)[]);
+    },
+    hset: <TValue>(
+      key: string,
+      obj: {
+        [key: string]: TValue;
+      }
+    ): Promise<number> => {
+      return redis.hset(key, obj);
+    },
+    eval: <TArgs extends unknown[], TData = unknown>(
+      ...args: [script: string, keys: string[], args: TArgs]
+    ): Promise<TData> => {
+      const script = args[0];
+      const keys = args[1];
+      const argsArray = args[2];
+      return redis.eval(
+        script,
+        keys.length,
+        ...keys,
+        ...(argsArray as (string | Buffer | number)[])
+      ) as Promise<TData>;
+    },
+  };
+}

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

@@ -77,6 +77,7 @@ type RedisRunExecutionRateLimiterOptions = {
 };
 
 const FORBIDDEN_FLAG_KEY = "forbiddenFlags";
+const PAUSED_FLAG_KEY = "pausedFlags";
 const KEY_PREFIX = "tr:exec:";
 
 class RedisRunExecutionRateLimiter implements RunExecutionRateLimiter, ZodWorkerRateLimiter {
@@ -109,6 +110,10 @@ local currentSize = redis.call('ZCOUNT', setKey, timestamp - windowSize, timesta
 if currentSize < maxSize then
     redis.call('ZADD', setKey, timestamp, jobId)
 
+    if currentSize + 1 >= maxSize then
+        redis.call('SADD', forbiddenFlagsKey, forbiddenFlag)
+    end
+
     return true
 else
     redis.call('SADD', forbiddenFlagsKey, forbiddenFlag)
@@ -176,7 +181,7 @@ end
   }
 
   async forbiddenFlags(): Promise<string[]> {
-    return this.redis.smembers(FORBIDDEN_FLAG_KEY);
+    return this.redis.sunion(FORBIDDEN_FLAG_KEY, PAUSED_FLAG_KEY);
   }
 
   async putConcurrencyLimitGroup(
@@ -377,8 +382,8 @@ function getRateLimiter() {
               tls: {
                 checkServerIdentity: () => {
                   // disable TLS verification
-                  return undefined
-                }
+                  return undefined;
+                },
               },
               enableAutoPipelining: true,
             },
@@ -397,7 +402,7 @@ function getRateLimiter() {
           username: env.REDIS_USERNAME,
           password: env.REDIS_PASSWORD,
           enableAutoPipelining: true,
-          ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} })
+          ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
         },
         defaultConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
       });

apps/webapp/app/services/runs/continueRun.server.ts

@@ -1,4 +1,4 @@
-import { $transaction, Prisma, PrismaClient, prisma } from "~/db.server";
+import { Prisma, PrismaClient, prisma } from "~/db.server";
 import { ResumeRunService } from "./resumeRun.server";
 
 const RESUMABLE_STATUSES = [
@@ -18,50 +18,44 @@ export class ContinueRunService {
   }
 
   public async call({ runId }: { runId: string }) {
-    return await $transaction(
-      this.#prismaClient,
-      async (tx) => {
-        const run = await tx.jobRun.findUniqueOrThrow({
-          where: { id: runId },
-          include: {
-            environment: true,
-          },
-        });
-
-        if (!RESUMABLE_STATUSES.includes(run.status)) {
-          throw new Error("Run is not resumable");
-        }
-
-        // Delete any tasks that are errored
-        const erroredTasks = await tx.task.findMany({
-          where: {
-            runId: runId,
-            status: "ERRORED",
-          },
-        });
-
-        await tx.jobRun.update({
-          where: { id: runId },
-          data: {
-            status: "QUEUED",
-            queuedAt: new Date(),
-            startedAt: null,
-            completedAt: null,
-            output: Prisma.DbNull,
-            timedOutAt: null,
-            timedOutReason: null,
-          },
-        });
-
-        for (const task of erroredTasks) {
-          await tx.task.delete({
-            where: { id: task.id },
-          });
-        }
-
-        await ResumeRunService.enqueue(run, tx);
+    const run = await this.#prismaClient.jobRun.findUniqueOrThrow({
+      where: { id: runId },
+      include: {
+        environment: true,
       },
-      { timeout: 30_000 }
-    );
+    });
+
+    if (!RESUMABLE_STATUSES.includes(run.status)) {
+      throw new Error("Run is not resumable");
+    }
+
+    await this.#prismaClient.jobRun.update({
+      where: { id: runId },
+      data: {
+        status: "QUEUED",
+        queuedAt: new Date(),
+        startedAt: null,
+        completedAt: null,
+        output: Prisma.DbNull,
+        timedOutAt: null,
+        timedOutReason: null,
+      },
+    });
+
+    // Now we need to reset errored tasks to PENDING
+    await this.#prismaClient.task.updateMany({
+      where: {
+        runId: runId,
+        status: "ERRORED",
+      },
+      data: {
+        status: "RUNNING",
+        output: Prisma.DbNull,
+        completedAt: null,
+        startedAt: new Date(),
+      },
+    });
+
+    await ResumeRunService.enqueue(run, this.#prismaClient);
   }
 }

apps/webapp/app/services/runs/performRunExecutionV3.server.ts

@@ -16,20 +16,17 @@ import {
   supportsFeature,
 } from "@trigger.dev/core";
 import { BloomFilter } from "@trigger.dev/core-backend";
-import {
-  ConcurrencyLimitGroup,
-  JobRun,
-  JobVersion,
-  RuntimeEnvironment,
-} from "@trigger.dev/database";
+import { ConcurrencyLimitGroup, Job, JobRun, JobVersion } from "@trigger.dev/database";
 import { generateErrorMessage } from "zod-error";
 import { eventRecordToApiJson } from "~/api.server";
 import {
+  MAX_JOB_RUN_EXECUTION_COUNT,
   MAX_RUN_CHUNK_EXECUTION_LIMIT,
   MAX_RUN_YIELDED_EXECUTIONS,
   RUN_CHUNK_EXECUTION_BUFFER,
 } from "~/consts";
 import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { env } from "~/env.server";
 import { detectResponseIsTimeout } from "~/models/endpoint.server";
 import { isRunCompleted } from "~/models/jobRun.server";
 import { resolveRunConnections } from "~/models/runConnection.server";
@@ -37,15 +34,16 @@ import { prepareTasksForCaching, prepareTasksForCachingLegacy } from "~/models/t
 import { CompleteRunTaskService } from "~/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server";
 import { formatError } from "~/utils/formatErrors.server";
 import { safeJsonZodParse } from "~/utils/json";
+import { marqsv2 } from "~/v3/marqs/v2.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
 import { EndpointApi } from "../endpointApi.server";
 import { createExecutionEvent } from "../executions/createExecutionEvent.server";
 import { logger } from "../logger.server";
+import { executionRateLimiter } from "../runExecutionRateLimiter.server";
 import { ResumeTaskService } from "../tasks/resumeTask.server";
 import { executionWorker, workerQueue } from "../worker.server";
 import { forceYieldCoordinator } from "./forceYieldCoordinator.server";
 import { ResumeRunService } from "./resumeRun.server";
-import { executionRateLimiter } from "../runExecutionRateLimiter.server";
-import { env } from "~/env.server";
 
 type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
 type FoundTask = FoundRun["tasks"][number];
@@ -95,9 +93,10 @@ export class PerformRunExecutionV3Service {
   static async enqueue(
     run: JobRun & {
       version: JobVersion & {
-        environment: RuntimeEnvironment;
+        environment: AuthenticatedEnvironment;
         concurrencyLimitGroup?: ConcurrencyLimitGroup | null;
       };
+      job: Job;
     },
     priority: RunExecutionPriority,
     tx: PrismaClientOrTransaction,
@@ -106,27 +105,49 @@ export class PerformRunExecutionV3Service {
       skipRetrying?: boolean;
     } = {}
   ) {
-    return await executionWorker.enqueue(
-      "performRunExecutionV3",
-      {
-        id: run.id,
-        reason: "EXECUTE_JOB",
-      },
-      {
-        tx,
-        runAt: options.runAt,
-        jobKey: `job_run:EXECUTE_JOB:${run.id}`,
-        maxAttempts: options.skipRetrying ? env.DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS : undefined,
-        flags: executionRateLimiter?.flagsForRun(run, run.version) ?? [],
-        priority: priority === "initial" ? 0 : -1,
+    if (marqsv2 && run.version.environment.organization.v2MarqsEnabled) {
+      let queue = `job/${run.job.slug}`;
+
+      if (run.version.concurrencyLimitGroup) {
+        queue = `group/${run.version.concurrencyLimitGroup.name}`;
       }
-    );
+
+      const runAt =
+        priority === "initial" ? options.runAt ?? new Date() : run.startedAt ?? run.createdAt;
+
+      await marqsv2.enqueueMessage(
+        run.version.environment,
+        queue,
+        run.id,
+        { runId: run.id, attempt: 1 },
+        undefined,
+        runAt.getTime()
+      );
+    } else {
+      return await executionWorker.enqueue(
+        "performRunExecutionV3",
+        {
+          id: run.id,
+          reason: "EXECUTE_JOB",
+        },
+        {
+          tx,
+          runAt: options.runAt,
+          jobKey: `job_run:EXECUTE_JOB:${run.id}`,
+          maxAttempts: options.skipRetrying ? env.DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS : undefined,
+          flags: executionRateLimiter?.flagsForRun(run, run.version) ?? [],
+          priority: priority === "initial" ? 0 : -1,
+        }
+      );
+    }
   }
 
   static async dequeue(run: JobRun, tx: PrismaClientOrTransaction) {
     await executionWorker.dequeue(`job_run:EXECUTE_JOB:${run.id}`, {
       tx,
     });
+
+    await marqsv2?.acknowledgeMessage(run.id);
   }
 
   async #executeJob(run: FoundRun, input: PerformRunExecutionV3Input, driftInMs: number = 0) {
@@ -135,12 +156,58 @@ export class PerformRunExecutionV3Service {
         return;
       }
 
+      if (!run.organization.runsEnabled) {
+        return await this.#failRunExecution(this.#prismaClient, run, {
+          message: `Unable to execute run.`,
+        });
+      }
+
       if (!run.endpoint.url) {
         return await this.#failRunExecution(this.#prismaClient, run, {
           message: `Endpoint has no URL set`,
         });
       }
 
+      if (run.version.status === "DISABLED") {
+        return await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Job version ${run.version.version} is disabled, aborting run.`,
+          },
+          "ABORTED"
+        );
+      }
+
+      // If the execution duration is greater than the maximum execution time, we need to fail the run
+      if (run.executionDuration >= run.organization.maximumExecutionTimePerRunInMs) {
+        await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Execution timed out after ${
+              run.organization.maximumExecutionTimePerRunInMs / 1000
+            } seconds`,
+          },
+          "TIMED_OUT",
+          0
+        );
+        return;
+      }
+
+      if (run.executionCount >= MAX_JOB_RUN_EXECUTION_COUNT) {
+        await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Execution timed out after ${run.executionCount} executions`,
+          },
+          "TIMED_OUT",
+          0
+        );
+        return;
+      }
+
       const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
       const event = eventRecordToApiJson(run.event);
 
@@ -207,6 +274,10 @@ export class PerformRunExecutionV3Service {
 
       forceYieldCoordinator.deregisterRun(run.id);
 
+      if (marqsv2 && run.organization.v2MarqsEnabled) {
+        await marqsv2.acknowledgeMessage(run.id);
+      }
+
       //if the run has been canceled while it's being executed, we shouldn't do anything more
       const updatedRun = await this.#prismaClient.jobRun.findUnique({
         select: {

apps/webapp/app/services/runs/resumeRun.server.ts

@@ -147,9 +147,15 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
   return await prisma.jobRun.findUnique({
     where: { id },
     include: {
+      job: true,
       version: {
         include: {
-          environment: true,
+          environment: {
+            include: {
+              organization: true,
+              project: true,
+            },
+          },
           concurrencyLimitGroup: true,
         },
       },

apps/webapp/app/services/runs/startRun.server.ts

@@ -4,11 +4,11 @@ import {
   type IntegrationConnection,
 } from "@trigger.dev/database";
 import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server";
-import { $transaction, prisma } from "~/db.server";
+import { prisma } from "~/db.server";
+import { autoIncrementCounter } from "../autoIncrementCounter.server";
+import { logger } from "../logger.server";
 import { workerQueue } from "../worker.server";
 import { ResumeRunService } from "./resumeRun.server";
-import { createHash } from "node:crypto";
-import { logger } from "../logger.server";
 
 type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
 type RunConnectionsByKey = Awaited<ReturnType<typeof createRunConnections>>;
@@ -67,22 +67,14 @@ export class StartRunService {
           : undefined
       )
       .filter(Boolean);
-    const lockId = jobIdToLockId(run.jobId);
 
-    await $transaction(
-      this.#prismaClient,
-      async (tx) => {
-        const counter = await tx.jobCounter.upsert({
-          where: { jobId: run.jobId },
-          update: { lastNumber: { increment: 1 } },
-          create: { jobId: run.jobId, lastNumber: 1 },
-          select: { lastNumber: true },
-        });
-
-        const updatedRun = await this.#prismaClient.jobRun.update({
+    await autoIncrementCounter.incrementInTransaction(
+      `v2-run:${run.jobId}`,
+      async (num, tx) => {
+        const updatedRun = await tx.jobRun.update({
           where: { id },
           data: {
-            number: counter.lastNumber,
+            number: num,
             status: "QUEUED",
             queuedAt: new Date(),
             runConnections: {
@@ -93,7 +85,16 @@ export class StartRunService {
 
         await ResumeRunService.enqueue(updatedRun, tx);
       },
-      { timeout: 60000 }
+      async (_, tx) => {
+        const counter = await tx.jobCounter.findUnique({
+          where: { jobId: run.jobId },
+          select: { lastNumber: true },
+        });
+
+        return counter?.lastNumber;
+      },
+      this.#prismaClient,
+      { timeout: 10_000 }
     );
   }
 
@@ -242,8 +243,3 @@ async function createRunConnections(tx: PrismaClientOrTransaction, run: FoundRun
 function hasMissingConnections(runConnectionsByKey: RunConnectionsByKey) {
   return Object.values(runConnectionsByKey).some((connection) => connection.result === "missing");
 }
-
-function jobIdToLockId(jobId: string): number {
-  // Convert jobId to a unique lock identifier
-  return parseInt(createHash("sha256").update(jobId).digest("hex").slice(0, 8), 16);
-}

apps/webapp/app/services/sources/handleHttpSource.server.ts

@@ -61,7 +61,6 @@ export class HandleHttpSourceService {
             id: delivery.id,
           },
           {
-            queueName: `deliver:${triggerSource.id}`,
             tx,
             maxAttempts:
               triggerSource.environment.type === RuntimeEnvironmentType.DEVELOPMENT ? 1 : undefined,

apps/webapp/app/services/tasks/resumeTask.server.ts

@@ -51,6 +51,9 @@ export class ResumeTaskService {
     logger.debug("ResumeTaskService.call resuming run execution", {
       parent: task.parent,
       taskId: task.id,
+      runId: task.run.id,
+      org: task.run.organizationId,
+      environment: task.run.environmentId,
     });
 
     if (task.parent && task.parent.childExecutionMode === "PARALLEL") {

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

@@ -7,6 +7,7 @@ import type { Organization } from "~/models/organization.server";
 import type { Project } from "~/models/project.server";
 import type { User } from "~/models/user.server";
 import { singleton } from "~/utils/singleton";
+import { loopsClient } from "./loops.server";
 
 type Options = {
   postHogApiKey?: string;
@@ -39,18 +40,19 @@ class Telemetry {
 
   user = {
     identify: ({ user, isNewUser }: { user: User; isNewUser: boolean }) => {
-      if (this.#posthogClient === undefined) return;
-      this.#posthogClient.identify({
-        distinctId: user.id,
-        properties: {
-          email: user.email,
-          name: user.name,
-          authenticationMethod: user.authenticationMethod,
-          admin: user.admin,
-          createdAt: user.createdAt,
-          isNewUser,
-        },
-      });
+      if (this.#posthogClient) {
+        this.#posthogClient.identify({
+          distinctId: user.id,
+          properties: {
+            email: user.email,
+            name: user.name,
+            authenticationMethod: user.authenticationMethod,
+            admin: user.admin,
+            createdAt: user.createdAt,
+            isNewUser,
+          },
+        });
+      }
       if (isNewUser) {
         this.#capture({
           userId: user.id,
@@ -64,6 +66,12 @@ class Telemetry {
           },
         });
 
+        loopsClient?.userCreated({
+          userId: user.id,
+          email: user.email,
+          name: user.name,
+        });
+
         this.#triggerClient?.sendEvent({
           name: "user.created",
           payload: {

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

@@ -4,7 +4,21 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { ZodWorker } from "~/platform/zodWorker.server";
+import { eventRepository } from "~/v3/eventRepository.server";
+import { RequeueTaskRunService } from "~/v3/requeueTaskRun.server";
+import { DeliverAlertService } from "~/v3/services/alerts/deliverAlert.server";
+import { PerformDeploymentAlertsService } from "~/v3/services/alerts/performDeploymentAlerts.server";
+import { PerformTaskAttemptAlertsService } from "~/v3/services/alerts/performTaskAttemptAlerts.server";
+import { PerformBulkActionService } from "~/v3/services/bulk/performBulkAction.server";
+import { ExecuteTasksWaitingForDeployService } from "~/v3/services/executeTasksWaitingForDeploy";
 import { IndexDeploymentService } from "~/v3/services/indexDeployment.server";
+import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
+import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
+import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
+import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
+import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
+import { TriggerScheduledTaskService } from "~/v3/services/triggerScheduledTask.server";
+import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
 import { ExpireDispatcherService } from "./dispatchers/expireDispatcher.server";
 import { InvokeEphemeralDispatcherService } from "./dispatchers/invokeEphemeralEventDispatcher.server";
 import { sendEmail } from "./email.server";
@@ -30,20 +44,7 @@ import { DeliverWebhookRequestService } from "./sources/deliverWebhookRequest.se
 import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
 import { ProcessCallbackTimeoutService } from "./tasks/processCallbackTimeout.server";
 import { ResumeTaskService } from "./tasks/resumeTask.server";
-import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
-import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
-import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
-import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
-import { eventRepository } from "~/v3/eventRepository.server";
-import { ExecuteTasksWaitingForDeployService } from "~/v3/services/executeTasksWaitingForDeploy";
-import { TriggerScheduledTaskService } from "~/v3/services/triggerScheduledTask.server";
-import { PerformTaskAttemptAlertsService } from "~/v3/services/alerts/performTaskAttemptAlerts.server";
-import { DeliverAlertService } from "~/v3/services/alerts/deliverAlert.server";
-import { PerformDeploymentAlertsService } from "~/v3/services/alerts/performDeploymentAlerts.server";
-import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
-import { PerformBulkActionService } from "~/v3/services/bulk/performBulkAction.server";
-import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
-import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -158,6 +159,15 @@ const workerCatalog = {
   "v3.performBulkActionItem": z.object({
     bulkActionItemId: z.string(),
   }),
+  "v3.requeueTaskRun": z.object({
+    runId: z.string(),
+  }),
+  "v3.retryAttempt": z.object({
+    runId: z.string(),
+  }),
+  "v2.requeueMessage": z.object({
+    runId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -247,14 +257,14 @@ function getWorkerQueue() {
       pollInterval: env.WORKER_POLL_INTERVAL,
       noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
       schema: env.WORKER_SCHEMA,
-      maxPoolSize: env.WORKER_CONCURRENCY,
+      maxPoolSize: env.WORKER_CONCURRENCY + 1,
     },
     shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: workerCatalog,
     recurringTasks: {
       // Run this every 5 minutes
       autoIndexProductionEndpoints: {
-        match: "*/5 * * * *",
+        match: "*/30 * * * *",
         handler: async (payload, job) => {
           const service = new RecurringEndpointIndexService();
 
@@ -301,7 +311,7 @@ function getWorkerQueue() {
       },
       "events.deliverScheduled": {
         priority: 0, // smaller number = higher priority
-        maxAttempts: 5,
+        maxAttempts: 8,
         handler: async ({ id, payload }, job) => {
           const service = new DeliverScheduledEventService();
 
@@ -327,7 +337,7 @@ function getWorkerQueue() {
         },
       },
       activateSource: {
-        priority: 10, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, graphileJob) => {
           const service = new ActivateSourceService();
@@ -353,9 +363,8 @@ function getWorkerQueue() {
         },
       },
       deliverHttpSourceRequest: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 14,
-        queueName: (payload) => `sources:${payload.id}`,
         handler: async (payload, job) => {
           const service = new DeliverHttpSourceRequestService();
 
@@ -363,9 +372,8 @@ function getWorkerQueue() {
         },
       },
       deliverWebhookRequest: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 14,
-        queueName: (payload) => `webhooks:${payload.id}`,
         handler: async (payload, job) => {
           const service = new DeliverWebhookRequestService();
 
@@ -391,14 +399,14 @@ function getWorkerQueue() {
         },
       },
       scheduleEmail: {
-        priority: 100,
+        priority: 0,
         maxAttempts: 3,
         handler: async (payload, job) => {
           await sendEmail(payload);
         },
       },
       indexEndpoint: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           const service = new IndexEndpointService();
@@ -406,7 +414,7 @@ function getWorkerQueue() {
         },
       },
       performEndpointIndexing: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           const service = new PerformEndpointIndexService();
@@ -423,7 +431,7 @@ function getWorkerQueue() {
         },
       },
       refreshOAuthToken: {
-        priority: 8, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           await integrationAuthRepository.refreshConnection({
@@ -432,7 +440,7 @@ function getWorkerQueue() {
         },
       },
       probeEndpoint: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 1,
         handler: async (payload, job) => {
           const service = new ProbeEndpointService();
@@ -447,7 +455,7 @@ function getWorkerQueue() {
         },
       },
       deliverRunSubscriptions: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 5,
         handler: async (payload, job) => {
           const service = new DeliverRunSubscriptionsService();
@@ -456,7 +464,7 @@ function getWorkerQueue() {
         },
       },
       deliverRunSubscription: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 13,
         handler: async (payload, job) => {
           const service = new DeliverRunSubscriptionService();
@@ -474,7 +482,7 @@ function getWorkerQueue() {
         },
       },
       expireDispatcher: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 3,
         handler: async (payload) => {
           const service = new ExpireDispatcherService();
@@ -600,6 +608,33 @@ function getWorkerQueue() {
           await service.performBulkActionItem(payload.bulkActionItemId);
         },
       },
+      "v3.requeueTaskRun": {
+        priority: 0,
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new RequeueTaskRunService();
+
+          await service.call(payload.runId);
+        },
+      },
+      "v3.retryAttempt": {
+        priority: 0,
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new RetryAttemptService();
+
+          return await service.call(payload.runId);
+        },
+      },
+      "v2.requeueMessage": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new RequeueV2Message();
+
+          await service.call(payload.runId);
+        },
+      },
     },
   });
 }
@@ -614,7 +649,7 @@ function getExecutionWorkerQueue() {
       pollInterval: env.EXECUTION_WORKER_POLL_INTERVAL,
       noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
       schema: env.WORKER_SCHEMA,
-      maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY,
+      maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY + 1,
     },
     shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: executionWorkerCatalog,
@@ -668,7 +703,7 @@ function getTaskOperationWorkerQueue() {
       pollInterval: env.TASK_OPERATION_WORKER_POLL_INTERVAL,
       noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
       schema: env.WORKER_SCHEMA,
-      maxPoolSize: env.TASK_OPERATION_WORKER_CONCURRENCY,
+      maxPoolSize: env.TASK_OPERATION_WORKER_CONCURRENCY + 1,
     },
     shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: taskOperationWorkerCatalog,

apps/webapp/app/utils/environmentSort.ts

@@ -1,4 +1,4 @@
-import { RuntimeEnvironmentType } from "@trigger.dev/database";
+import { Prisma, RuntimeEnvironmentType } from "@trigger.dev/database";
 
 const environmentSortOrder: RuntimeEnvironmentType[] = [
   "DEVELOPMENT",
@@ -29,3 +29,32 @@ export function sortEnvironments<T extends SortType>(environments: T[]): T[] {
     return difference;
   });
 }
+
+type FilterableEnvironment =
+  | {
+      type: RuntimeEnvironmentType;
+      orgMemberId?: string;
+    }
+  | {
+      type: RuntimeEnvironmentType;
+      //intentionally vague so we can match anything
+      orgMember?: Record<string, any>;
+    };
+
+export function filterOrphanedEnvironments<T extends FilterableEnvironment>(
+  environments: T[]
+): T[] {
+  return environments.filter((environment) => {
+    if (environment.type !== "DEVELOPMENT") return true;
+
+    if ("orgMemberId" in environment) {
+      return !!environment.orgMemberId;
+    }
+
+    if ("orgMember" in environment) {
+      return !!environment.orgMember;
+    }
+
+    return false;
+  });
+}

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

@@ -54,7 +54,10 @@ export class AuthenticatedSocketConnection {
       schema: clientWebsocketMessages,
       messages: {
         READY_FOR_TASKS: async (payload) => {
-          await this._consumer.registerBackgroundWorker(payload.backgroundWorkerId);
+          await this._consumer.registerBackgroundWorker(
+            payload.backgroundWorkerId,
+            payload.inProgressRuns ?? []
+          );
         },
         BACKGROUND_WORKER_DEPRECATED: async (payload) => {
           await this._consumer.deprecateBackgroundWorker(payload.backgroundWorkerId);
@@ -69,10 +72,22 @@ export class AuthenticatedSocketConnection {
               );
               break;
             }
+            case "TASK_RUN_FAILED_TO_RUN": {
+              await this._consumer.taskRunFailed(
+                payload.backgroundWorkerId,
+                payload.data.completion
+              );
+
+              break;
+            }
             case "TASK_HEARTBEAT": {
               await this._consumer.taskHeartbeat(payload.backgroundWorkerId, payload.data.id);
               break;
             }
+            case "TASK_RUN_HEARTBEAT": {
+              await this._consumer.taskRunHeartbeat(payload.backgroundWorkerId, payload.data.id);
+              break;
+            }
           }
         },
       },

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

@@ -10,6 +10,7 @@ import {
   SpanEvents,
   SpanMessagingEvent,
   TaskEventStyle,
+  TaskRunError,
   correctErrorStackTrace,
   createPacketAttributesAsJson,
   flattenAttributes,
@@ -117,6 +118,7 @@ export type QueriedEvent = Prisma.TaskEventGetPayload<{
     isCancelled: true;
     level: true;
     events: true;
+    environmentType: true;
   };
 }>;
 
@@ -156,6 +158,7 @@ export type SpanSummary = {
     isPartial: boolean;
     isCancelled: boolean;
     level: NonNullable<CreatableEvent["level"]>;
+    environmentType: CreatableEventEnvironmentType;
   };
 };
 
@@ -165,6 +168,7 @@ export type UpdateEventOptions = {
   attributes: TraceAttributes;
   endTime?: Date;
   immediate?: boolean;
+  events?: SpanEvents;
 };
 
 export class EventRepository {
@@ -239,7 +243,7 @@ export class EventRepository {
       isCancelled: false,
       status: options?.attributes.isError ? "ERROR" : "OK",
       links: event.links ?? [],
-      events: event.events ?? [],
+      events: event.events ?? (options?.events as any) ?? [],
       duration: calculateDurationFromStart(event.startTime, options?.endTime),
       properties: event.properties as Attributes,
       metadata: event.metadata as Attributes,
@@ -386,6 +390,7 @@ export class EventRepository {
         isCancelled: true,
         level: true,
         events: true,
+        environmentType: true,
       },
       where: {
         traceId,
@@ -421,6 +426,7 @@ export class EventRepository {
           startTime: getDateFromNanoseconds(event.startTime),
           level: event.level,
           events: event.events,
+          environmentType: event.environmentType,
         },
       };
     });
@@ -505,7 +511,11 @@ export class EventRepository {
       });
     }
 
-    const events = transformEvents(span.data.events, fullEvent.metadata as Attributes);
+    const events = transformEvents(
+      span.data.events,
+      fullEvent.metadata as Attributes,
+      traceSummary?.rootSpan.data.environmentType === "DEVELOPMENT"
+    );
 
     return {
       ...fullEvent,
@@ -877,6 +887,36 @@ export function stripAttributePrefix(attributes: Attributes, prefix: string) {
   return result;
 }
 
+export function createExceptionPropertiesFromError(error: TaskRunError): ExceptionEventProperties {
+  switch (error.type) {
+    case "BUILT_IN_ERROR": {
+      return {
+        type: error.name,
+        message: error.message,
+        stacktrace: error.stackTrace,
+      };
+    }
+    case "CUSTOM_ERROR": {
+      return {
+        type: "Error",
+        message: error.raw,
+      };
+    }
+    case "INTERNAL_ERROR": {
+      return {
+        type: "Internal error",
+        message: [error.code, error.message].filter(Boolean).join(": "),
+      };
+    }
+    case "STRING_ERROR": {
+      return {
+        type: "Error",
+        message: error.raw,
+      };
+    }
+  }
+}
+
 /**
  * Filters out partial events from a batch of creatable events, excluding those that have a corresponding full event.
  * @param batch - The batch of creatable events to filter.
@@ -1097,16 +1137,16 @@ function removePrivateProperties(
   return result;
 }
 
-function transformEvents(events: SpanEvents, properties: Attributes): SpanEvents {
-  return (events ?? []).map((event) => transformEvent(event, properties));
+function transformEvents(events: SpanEvents, properties: Attributes, isDev: boolean): SpanEvents {
+  return (events ?? []).map((event) => transformEvent(event, properties, isDev));
 }
 
-function transformEvent(event: SpanEvent, properties: Attributes): SpanEvent {
+function transformEvent(event: SpanEvent, properties: Attributes, isDev: boolean): SpanEvent {
   if (isExceptionSpanEvent(event)) {
     return {
       ...event,
       properties: {
-        exception: transformException(event.properties.exception, properties),
+        exception: transformException(event.properties.exception, properties, isDev),
       },
     };
   }
@@ -1116,11 +1156,12 @@ function transformEvent(event: SpanEvent, properties: Attributes): SpanEvent {
 
 function transformException(
   exception: ExceptionEventProperties,
-  properties: Attributes
+  properties: Attributes,
+  isDev: boolean
 ): ExceptionEventProperties {
   const projectDirAttributeValue = properties[SemanticInternalAttributes.PROJECT_DIR];
 
-  if (typeof projectDirAttributeValue !== "string") {
+  if (projectDirAttributeValue !== undefined && typeof projectDirAttributeValue !== "string") {
     return exception;
   }
 
@@ -1129,6 +1170,7 @@ function transformException(
     stacktrace: exception.stacktrace
       ? correctErrorStackTrace(exception.stacktrace, projectDirAttributeValue, {
           removeFirstLine: true,
+          isDev,
         })
       : undefined,
   };

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

@@ -0,0 +1,66 @@
+import { TaskRunFailedExecutionResult } from "@trigger.dev/core/v3";
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+import { TaskRunStatus } from "@trigger.dev/database";
+import { createExceptionPropertiesFromError, eventRepository } from "./eventRepository.server";
+import { BaseService } from "./services/baseService.server";
+
+const FAILABLE_TASK_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "PENDING", "WAITING_FOR_DEPLOY"];
+
+export class FailedTaskRunService extends BaseService {
+  public async call(runFriendlyId: string, completion: TaskRunFailedExecutionResult) {
+    const taskRun = await this._prisma.taskRun.findUnique({
+      where: { friendlyId: runFriendlyId },
+    });
+
+    if (!taskRun) {
+      logger.error("[FailedTaskRunService] Task run not found", {
+        runFriendlyId,
+        completion,
+      });
+
+      return;
+    }
+
+    if (!FAILABLE_TASK_RUN_STATUSES.includes(taskRun.status)) {
+      logger.error("[FailedTaskRunService] Task run is not in a failable state", {
+        taskRun,
+        completion,
+      });
+
+      return;
+    }
+
+    // No more retries, we need to fail the task run
+    logger.debug("[FailedTaskRunService] Failing task run", { taskRun, completion });
+
+    await marqs?.acknowledgeMessage(taskRun.id);
+
+    // Now we need to "complete" the task run event/span
+    await eventRepository.completeEvent(taskRun.spanId, {
+      endTime: new Date(),
+      attributes: {
+        isError: true,
+      },
+      events: [
+        {
+          name: "exception",
+          time: new Date(),
+          properties: {
+            exception: createExceptionPropertiesFromError(completion.error),
+          },
+        },
+      ],
+    });
+
+    await this._prisma.taskRun.update({
+      where: {
+        id: taskRun.id,
+      },
+      data: {
+        status: "SYSTEM_FAILURE",
+      },
+    });
+  }
+}

apps/webapp/app/v3/friendlyIdentifiers.ts

@@ -2,6 +2,6 @@ import { customAlphabet } from "nanoid";
 
 const idGenerator = customAlphabet("123456789abcdefghijkmnopqrstuvwxyz", 21);
 
-export function generateFriendlyId(prefix: string) {
-  return `${prefix}_${idGenerator()}`;
+export function generateFriendlyId(prefix: string, size?: number) {
+  return `${prefix}_${idGenerator(size)}`;
 }

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

@@ -22,6 +22,7 @@ import { DeploymentIndexFailed } from "./services/deploymentIndexFailed.server";
 import { Redis } from "ioredis";
 import { createAdapter } from "@socket.io/redis-adapter";
 import { CrashTaskRunService } from "./services/crashTaskRun.server";
+import { CreateTaskRunAttemptService } from "./services/createTaskRunAttempt.server";
 
 export const socketIo = singleton("socketIo", initalizeIoServer);
 
@@ -86,11 +87,35 @@ function createCoordinatorNamespace(io: Server) {
         );
 
         if (!payload) {
+          logger.error("Failed to retrieve execution payload", message);
           return { success: false };
         } else {
           return { success: true, payload };
         }
       },
+      READY_FOR_LAZY_ATTEMPT: async (message) => {
+        try {
+          const payload = await sharedQueueTasks.getLazyAttemptPayload(
+            message.envId,
+            message.runId
+          );
+
+          if (!payload) {
+            logger.error("Failed to retrieve lazy attempt payload", message);
+            return { success: false, reason: "Failed to retrieve payload" };
+          }
+
+          return { success: true, lazyPayload: payload };
+        } catch (error) {
+          logger.error("Error while creating lazy attempt", {
+            runId: message.runId,
+            envId: message.envId,
+            totalCompletions: message.totalCompletions,
+            error,
+          });
+          return { success: false };
+        }
+      },
       READY_FOR_RESUME: async (message) => {
         const resumeAttempt = new ResumeAttemptService();
         await resumeAttempt.call(message);
@@ -103,9 +128,15 @@ function createCoordinatorNamespace(io: Server) {
           checkpoint: message.checkpoint,
         });
       },
+      TASK_RUN_FAILED_TO_RUN: async (message) => {
+        await sharedQueueTasks.taskRunFailed(message.completion);
+      },
       TASK_HEARTBEAT: async (message) => {
         await sharedQueueTasks.taskHeartbeat(message.attemptFriendlyId);
       },
+      TASK_RUN_HEARTBEAT: async (message) => {
+        await sharedQueueTasks.taskRunHeartbeat(message.runId);
+      },
       CHECKPOINT_CREATED: async (message) => {
         const createCheckpoint = new CreateCheckpointService();
         await createCheckpoint.call(message);
@@ -123,11 +154,48 @@ function createCoordinatorNamespace(io: Server) {
           const worker = await service.call(message.projectRef, environment, message.deploymentId, {
             localOnly: false,
             metadata: message.metadata,
+            supportsLazyAttempts: message.version !== "v1" && message.supportsLazyAttempts,
           });
 
           return { success: !!worker };
         } catch (error) {
-          logger.error("Error while creating worker", { error });
+          logger.error("Error while creating worker", {
+            error,
+            envId: message.envId,
+            projectRef: message.projectRef,
+            deploymentId: message.deploymentId,
+            version: message.version,
+          });
+          return { success: false };
+        }
+      },
+      CREATE_TASK_RUN_ATTEMPT: async (message) => {
+        try {
+          const environment = await findEnvironmentById(message.envId);
+
+          if (!environment) {
+            logger.error("Environment not found", { id: message.envId });
+            return { success: false, reason: "Environment not found" };
+          }
+
+          const service = new CreateTaskRunAttemptService();
+          const { attempt } = await service.call(message.runId, environment, false);
+
+          const payload = await sharedQueueTasks.getExecutionPayloadFromAttempt(attempt.id, true);
+
+          if (!payload) {
+            logger.error("Failed to retrieve payload after attempt creation", {
+              id: message.envId,
+            });
+            return { success: false, reason: "Failed to retrieve payload" };
+          }
+
+          return { success: true, executionPayload: payload };
+        } catch (error) {
+          logger.error("Error while creating attempt", {
+            runId: message.runId,
+            error,
+          });
           return { success: false };
         }
       },
@@ -136,8 +204,26 @@ function createCoordinatorNamespace(io: Server) {
           const service = new DeploymentIndexFailed();
 
           await service.call(message.deploymentId, message.error);
-        } catch (e) {
-          logger.error("Error while indexing", { error: e });
+        } catch (error) {
+          logger.error("Error while processing index failure", {
+            deploymentId: message.deploymentId,
+            error,
+          });
+        }
+      },
+      RUN_CRASHED: async (message) => {
+        try {
+          const service = new CrashTaskRunService();
+
+          await service.call(message.runId, {
+            reason: `${message.error.name}: ${message.error.message}`,
+            logs: message.error.stack,
+          });
+        } catch (error) {
+          logger.error("Error while processing run failure", {
+            runId: message.runId,
+            error,
+          });
         }
       },
     },

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

@@ -1,8 +1,10 @@
 import { Context, ROOT_CONTEXT, Span, SpanKind, context, trace } from "@opentelemetry/api";
 import {
   TaskRunExecution,
+  TaskRunExecutionLazyAttemptPayload,
   TaskRunExecutionPayload,
   TaskRunExecutionResult,
+  TaskRunFailedExecutionResult,
   serverWebsocketMessages,
 } from "@trigger.dev/core/v3";
 import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
@@ -14,16 +16,16 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
-import { generateFriendlyId } from "../friendlyIdentifiers";
-import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { CancelTaskRunService } from "../services/cancelTaskRun.server";
 import { CompleteAttemptService } from "../services/completeAttempt.server";
+import { CreateTaskRunAttemptService } from "../services/createTaskRunAttempt.server";
 import {
   SEMINTATTRS_FORCE_RECORDING,
   attributesFromAuthenticatedEnv,
   tracer,
 } from "../tracer.server";
 import { DevSubscriber, devPubSub } from "./devPubSub.server";
+import { FailedTaskRunService } from "../failedTaskRun.server";
 
 const MessageBody = z.discriminatedUnion("type", [
   z.object({
@@ -54,7 +56,6 @@ export class DevQueueConsumer {
   private _taskSuccesses: number = 0;
   private _currentSpan: Span | undefined;
   private _endSpanInNextIteration = false;
-  private _inProgressAttempts: Map<string, string> = new Map(); // Keys are task attempt friendly IDs, values are TaskRun ids/queue message ids
   private _inProgressRuns: Map<string, string> = new Map(); // Keys are task run friendly IDs, values are TaskRun internal ids/queue message ids
 
   constructor(
@@ -78,7 +79,7 @@ export class DevQueueConsumer {
     this._backgroundWorkers.delete(id);
   }
 
-  public async registerBackgroundWorker(id: string) {
+  public async registerBackgroundWorker(id: string, inProgressRuns: string[] = []) {
     const backgroundWorker = await prisma.backgroundWorker.findUnique({
       where: { friendlyId: id, runtimeEnvironmentId: this.env.id },
       include: {
@@ -96,7 +97,10 @@ export class DevQueueConsumer {
 
     this._backgroundWorkers.set(backgroundWorker.id, backgroundWorker);
 
-    logger.debug("Registered background worker", { backgroundWorker: backgroundWorker.id });
+    logger.debug("Registered background worker", {
+      backgroundWorker: backgroundWorker.id,
+      inProgressRuns,
+    });
 
     const subscriber = await devPubSub.subscribe(`backgroundWorker:${backgroundWorker.id}:*`);
 
@@ -113,6 +117,10 @@ export class DevQueueConsumer {
 
     this._backgroundWorkerSubscriber.set(backgroundWorker.id, subscriber);
 
+    for (const runId of inProgressRuns) {
+      this._inProgressRuns.set(runId, runId);
+    }
+
     // Start reading from the queue if we haven't already
     await this.#enable();
   }
@@ -122,15 +130,16 @@ export class DevQueueConsumer {
     completion: TaskRunExecutionResult,
     execution: TaskRunExecution
   ) {
-    this._inProgressAttempts.delete(execution.attempt.id);
-
     if (completion.ok) {
       this._taskSuccesses++;
     } else {
       this._taskFailures++;
     }
 
-    logger.debug("Task run completed", { taskRunCompletion: completion, execution });
+    logger.debug("[DevQueueConsumer] taskAttemptCompleted()", {
+      taskRunCompletion: completion,
+      execution,
+    });
 
     const service = new CompleteAttemptService();
     const result = await service.call({ completion, execution, env: this.env });
@@ -140,7 +149,24 @@ export class DevQueueConsumer {
     }
   }
 
+  public async taskRunFailed(workerId: string, completion: TaskRunFailedExecutionResult) {
+    this._taskFailures++;
+
+    logger.debug("[DevQueueConsumer] taskRunFailed()", { completion });
+
+    this._inProgressRuns.delete(completion.id);
+
+    const service = new FailedTaskRunService();
+
+    await service.call(completion.id, completion);
+  }
+
+  /**
+   * @deprecated Use `taskRunHeartbeat` instead
+   */
   public async taskHeartbeat(workerId: string, id: string, seconds: number = 60) {
+    logger.debug("[DevQueueConsumer] taskHeartbeat()", { id, seconds });
+
     const taskRunAttempt = await prisma.taskRunAttempt.findUnique({
       where: { friendlyId: id },
     });
@@ -152,6 +178,12 @@ export class DevQueueConsumer {
     await marqs?.heartbeatMessage(taskRunAttempt.taskRunId, seconds);
   }
 
+  public async taskRunHeartbeat(workerId: string, id: string, seconds: number = 60) {
+    logger.debug("[DevQueueConsumer] taskRunHeartbeat()", { id, seconds });
+
+    await marqs?.heartbeatMessage(id, seconds);
+  }
+
   public async stop(reason: string = "CLI disconnected") {
     if (!this._enabled) {
       return;
@@ -184,66 +216,23 @@ export class DevQueueConsumer {
   }
 
   async #cancelInProgressRunsAndAttempts(reason: string) {
-    const cancelAttemptService = new CancelAttemptService();
     const cancelTaskRunService = new CancelTaskRunService();
 
     const cancelledAt = new Date();
 
-    const inProgressAttempts = new Map(this._inProgressAttempts);
     const inProgressRuns = new Map(this._inProgressRuns);
 
-    this._inProgressAttempts.clear();
     this._inProgressRuns.clear();
 
-    const inProgressRunsWithNoInProgressAttempts: string[] = [];
-    const inProgressAttemptRunIds = new Set(inProgressAttempts.values());
-
-    for (const [runId, messageId] of inProgressRuns) {
-      if (!inProgressAttemptRunIds.has(messageId)) {
-        inProgressRunsWithNoInProgressAttempts.push(messageId);
-      }
-    }
-
     logger.debug("Cancelling in progress runs and attempts", {
-      attempts: Array.from(inProgressAttempts.keys()),
       runs: Array.from(inProgressRuns.keys()),
     });
 
-    for (const [attemptId, messageId] of inProgressAttempts) {
-      await this.#cancelInProgressAttempt(
-        attemptId,
-        messageId,
-        cancelAttemptService,
-        cancelledAt,
-        reason
-      );
-    }
-
-    for (const runId of inProgressRunsWithNoInProgressAttempts) {
+    for (const [_, runId] of inProgressRuns) {
       await this.#cancelInProgressRun(runId, cancelTaskRunService, cancelledAt, reason);
     }
   }
 
-  async #cancelInProgressAttempt(
-    attemptId: string,
-    messageId: string,
-    cancelAttemptService: CancelAttemptService,
-    cancelledAt: Date,
-    reason: string
-  ) {
-    logger.debug("Cancelling in progress attempt", { attemptId, messageId });
-
-    try {
-      await cancelAttemptService.call(attemptId, messageId, cancelledAt, reason, this.env);
-    } catch (e) {
-      logger.error("Failed to cancel in progress attempt", {
-        attemptId,
-        messageId,
-        error: e,
-      });
-    }
-  }
-
   async #cancelInProgressRun(
     runId: string,
     service: CancelTaskRunService,
@@ -252,16 +241,20 @@ export class DevQueueConsumer {
   ) {
     logger.debug("Cancelling in progress run", { runId });
 
-    const taskRun = await prisma.taskRun.findUnique({
-      where: { id: runId },
-    });
+    const taskRun = runId.startsWith("run_")
+      ? await prisma.taskRun.findUnique({
+          where: { friendlyId: runId },
+        })
+      : await prisma.taskRun.findUnique({
+          where: { id: runId },
+        });
 
     if (!taskRun) {
       return;
     }
 
     try {
-      await service.call(taskRun, { reason, cancelAttempts: false, cancelledAt });
+      await service.call(taskRun, { reason, cancelAttempts: true, cancelledAt });
     } catch (e) {
       logger.error("Failed to cancel in progress run", {
         runId,
@@ -474,141 +467,131 @@ export class DevQueueConsumer {
     }
 
     if (!this._enabled) {
+      logger.debug("Dev queue consumer is disabled", { env: this.env, queueMessage: message });
+
       await marqs?.nackMessage(message.messageId);
       return;
     }
 
-    const taskRunAttempt = await prisma.taskRunAttempt.create({
-      data: {
-        number: lockedTaskRun.attempts[0] ? lockedTaskRun.attempts[0].number + 1 : 1,
-        friendlyId: generateFriendlyId("attempt"),
-        taskRunId: lockedTaskRun.id,
-        startedAt: new Date(),
-        backgroundWorkerId: backgroundTask.workerId,
-        backgroundWorkerTaskId: backgroundTask.id,
-        status: "EXECUTING" as const,
-        queueId: queue.id,
-        runtimeEnvironmentId: this.env.id,
-      },
-    });
-
-    const execution: TaskRunExecution = {
-      task: {
-        id: backgroundTask.slug,
-        filePath: backgroundTask.filePath,
-        exportName: backgroundTask.exportName,
-      },
-      attempt: {
-        id: taskRunAttempt.friendlyId,
-        number: taskRunAttempt.number,
-        startedAt: taskRunAttempt.startedAt ?? taskRunAttempt.createdAt,
-        backgroundWorkerId: backgroundWorker.id,
-        backgroundWorkerTaskId: backgroundTask.id,
-        status: "EXECUTING" as const,
-      },
-      run: {
-        id: lockedTaskRun.friendlyId,
-        payload: lockedTaskRun.payload,
-        payloadType: lockedTaskRun.payloadType,
-        context: lockedTaskRun.context,
-        createdAt: lockedTaskRun.createdAt,
-        tags: lockedTaskRun.tags.map((tag) => tag.name),
-        isTest: lockedTaskRun.isTest,
-        idempotencyKey: lockedTaskRun.idempotencyKey ?? undefined,
-      },
-      queue: {
-        id: queue.friendlyId,
-        name: queue.name,
-      },
-      environment: {
-        id: this.env.id,
-        slug: this.env.slug,
-        type: this.env.type,
-      },
-      organization: {
-        id: this.env.organization.id,
-        slug: this.env.organization.slug,
-        name: this.env.organization.title,
-      },
-      project: {
-        id: this.env.project.id,
-        ref: this.env.project.externalRef,
-        slug: this.env.project.slug,
-        name: this.env.project.name,
-      },
-      batch:
-        lockedTaskRun.batchItems[0] && lockedTaskRun.batchItems[0].batchTaskRun
-          ? { id: lockedTaskRun.batchItems[0].batchTaskRun.friendlyId }
-          : undefined,
-    };
-
     const environmentRepository = new EnvironmentVariablesRepository();
     const variables = await environmentRepository.getEnvironmentVariables(
       this.env.project.id,
       this.env.id
     );
 
-    const payload: TaskRunExecutionPayload = {
-      execution,
-      traceContext: lockedTaskRun.traceContext as Record<string, unknown>,
-      environment: variables.reduce((acc: Record<string, string>, curr) => {
-        acc[curr.key] = curr.value;
-        return acc;
-      }, {}),
-    };
+    if (backgroundWorker.supportsLazyAttempts) {
+      const payload: TaskRunExecutionLazyAttemptPayload = {
+        traceContext: lockedTaskRun.traceContext as Record<string, unknown>,
+        environment: variables.reduce((acc: Record<string, string>, curr) => {
+          acc[curr.key] = curr.value;
+          return acc;
+        }, {}),
+        runId: lockedTaskRun.friendlyId,
+        messageId: lockedTaskRun.id,
+        isTest: lockedTaskRun.isTest,
+      };
 
-    try {
-      // TODO: send trace context down to the CLI
-      await this._sender.send("BACKGROUND_WORKER_MESSAGE", {
-        backgroundWorkerId: backgroundWorker.friendlyId,
-        data: {
-          type: "EXECUTE_RUNS",
-          payloads: [payload],
-        },
-      });
-
-      logger.debug("Saving the in progress attempt", {
-        taskRunAttempt: taskRunAttempt.id,
-        messageId: message.messageId,
-      });
-
-      this._inProgressAttempts.set(taskRunAttempt.friendlyId, message.messageId);
-      this._inProgressRuns.set(lockedTaskRun.friendlyId, message.messageId);
-    } catch (e) {
-      if (e instanceof Error) {
-        this._currentSpan?.recordException(e);
-      } else {
-        this._currentSpan?.recordException(new Error(String(e)));
-      }
-
-      this._endSpanInNextIteration = true;
-
-      // We now need to unlock the task run and delete the task run attempt
-      await prisma.$transaction([
-        prisma.taskRun.update({
-          where: {
-            id: lockedTaskRun.id,
-          },
+      try {
+        await this._sender.send("BACKGROUND_WORKER_MESSAGE", {
+          backgroundWorkerId: backgroundWorker.friendlyId,
           data: {
-            lockedAt: null,
-            lockedById: null,
-            status: "PENDING",
+            type: "EXECUTE_RUN_LAZY_ATTEMPT",
+            payload,
           },
-        }),
-        prisma.taskRunAttempt.delete({
-          where: {
-            id: taskRunAttempt.id,
+        });
+
+        logger.debug("Executing the run", {
+          messageId: message.messageId,
+        });
+
+        this._inProgressRuns.set(lockedTaskRun.friendlyId, message.messageId);
+      } catch (e) {
+        if (e instanceof Error) {
+          this._currentSpan?.recordException(e);
+        } else {
+          this._currentSpan?.recordException(new Error(String(e)));
+        }
+
+        this._endSpanInNextIteration = true;
+
+        // We now need to unlock the task run and delete the task run attempt
+        await prisma.$transaction([
+          prisma.taskRun.update({
+            where: {
+              id: lockedTaskRun.id,
+            },
+            data: {
+              lockedAt: null,
+              lockedById: null,
+              status: "PENDING",
+            },
+          }),
+        ]);
+
+        this._inProgressRuns.delete(lockedTaskRun.friendlyId);
+
+        // Finally we need to nack the message so it can be retried
+        await marqs?.nackMessage(message.messageId);
+      } finally {
+        setTimeout(() => this.#doWork(), 100);
+      }
+    } else {
+      const service = new CreateTaskRunAttemptService();
+      const { execution } = await service.call(lockedTaskRun.friendlyId, this.env);
+
+      const payload: TaskRunExecutionPayload = {
+        traceContext: lockedTaskRun.traceContext as Record<string, unknown>,
+        environment: variables.reduce((acc: Record<string, string>, curr) => {
+          acc[curr.key] = curr.value;
+          return acc;
+        }, {}),
+        execution,
+      };
+
+      try {
+        await this._sender.send("BACKGROUND_WORKER_MESSAGE", {
+          backgroundWorkerId: backgroundWorker.friendlyId,
+          data: {
+            type: "EXECUTE_RUNS",
+            payloads: [payload],
           },
-        }),
-      ]);
+        });
 
-      this._inProgressAttempts.delete(taskRunAttempt.friendlyId);
-      this._inProgressRuns.delete(lockedTaskRun.friendlyId);
+        logger.debug("Executing the run", {
+          messageId: message.messageId,
+        });
 
-      // Finally we need to nack the message so it can be retried
-      await marqs?.nackMessage(message.messageId);
-    } finally {
-      setTimeout(() => this.#doWork(), 100);
+        this._inProgressRuns.set(lockedTaskRun.friendlyId, message.messageId);
+      } catch (e) {
+        if (e instanceof Error) {
+          this._currentSpan?.recordException(e);
+        } else {
+          this._currentSpan?.recordException(new Error(String(e)));
+        }
+
+        this._endSpanInNextIteration = true;
+
+        // We now need to unlock the task run and delete the task run attempt
+        await prisma.$transaction([
+          prisma.taskRun.update({
+            where: {
+              id: lockedTaskRun.id,
+            },
+            data: {
+              lockedAt: null,
+              lockedById: null,
+              status: "PENDING",
+            },
+          }),
+        ]);
+
+        this._inProgressRuns.delete(lockedTaskRun.friendlyId);
+
+        // Finally we need to nack the message so it can be retried
+        await marqs?.nackMessage(message.messageId);
+      } finally {
+        setTimeout(() => this.#doWork(), 100);
+      }
     }
   }
 

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

@@ -1,4 +1,12 @@
-import { Span, SpanKind, SpanOptions, context, propagation, trace } from "@opentelemetry/api";
+import {
+  Span,
+  SpanKind,
+  SpanOptions,
+  Tracer,
+  context,
+  propagation,
+  trace,
+} from "@opentelemetry/api";
 import {
   SEMATTRS_MESSAGE_ID,
   SEMATTRS_MESSAGING_OPERATION,
@@ -13,21 +21,20 @@ import { singleton } from "~/utils/singleton";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";
 import { AsyncWorker } from "./asyncWorker.server";
 import { MarQSShortKeyProducer } from "./marqsKeyProducer.server";
-import { SimpleWeightedChoiceStrategy } from "./priorityStrategy.server";
+import { SimpleWeightedChoiceStrategy } from "./simpleWeightedPriorityStrategy.server";
 import {
   MarQSKeyProducer,
   MarQSQueuePriorityStrategy,
   MessagePayload,
   QueueCapacities,
   QueueRange,
+  VisibilityTimeoutStrategy,
 } from "./types";
-
-const tracer = trace.getTracer("marqs");
+import { V3VisibilityTimeout } from "./v3VisibilityTimeout.server";
 
 const KEY_PREFIX = "marqs:";
 
 const constants = {
-  SHARED_QUEUE: "sharedQueue",
   MESSAGE_VISIBILITY_TIMEOUT_QUEUE: "msgVisibilityTimeout",
 } as const;
 
@@ -39,6 +46,8 @@ const SemanticAttributes = {
 };
 
 export type MarQSOptions = {
+  name: string;
+  tracer: Tracer;
   redis: RedisOptions;
   defaultEnvConcurrency: number;
   defaultOrgConcurrency: number;
@@ -48,6 +57,9 @@ export type MarQSOptions = {
   keysProducer: MarQSKeyProducer;
   queuePriorityStrategy: MarQSQueuePriorityStrategy;
   envQueuePriorityStrategy: MarQSQueuePriorityStrategy;
+  visibilityTimeoutStrategy: VisibilityTimeoutStrategy;
+  enableRebalancing?: boolean;
+  verbose?: boolean;
 };
 
 /**
@@ -72,6 +84,14 @@ export class MarQS {
     this.#registerCommands();
   }
 
+  get name() {
+    return this.options.name;
+  }
+
+  get tracer() {
+    return this.options.tracer;
+  }
+
   public async updateQueueConcurrencyLimits(
     env: AuthenticatedEnvironment,
     queue: string,
@@ -80,6 +100,10 @@ export class MarQS {
     return this.redis.set(this.keys.queueConcurrencyLimitKey(env, queue), concurrency);
   }
 
+  public async removeQueueConcurrencyLimits(env: AuthenticatedEnvironment, queue: string) {
+    return this.redis.del(this.keys.queueConcurrencyLimitKey(env, queue));
+  }
+
   public async updateEnvConcurrencyLimits(env: AuthenticatedEnvironment) {
     await this.#callUpdateGlobalConcurrencyLimits({
       envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
@@ -209,7 +233,8 @@ export class MarQS {
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
           this.options.envQueuePriorityStrategy,
-          (queue) => this.#calculateMessageQueueCapacities(queue)
+          (queue) => this.#calculateMessageQueueCapacities(queue),
+          env.id
         );
 
         if (!messageQueue) {
@@ -243,8 +268,9 @@ export class MarQS {
             [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
           });
         } else {
-          logger.error("Failed to read message, undoing the dequeueing of the message", {
+          logger.error(`Failed to read message, undoing the dequeueing of the message`, {
             messageData,
+            service: this.name,
           });
 
           await this.#callAcknowledgeMessage({
@@ -259,6 +285,11 @@ export class MarQS {
           });
         }
 
+        await this.options.visibilityTimeoutStrategy.heartbeat(
+          messageData.messageId,
+          this.visibilityTimeoutInMs
+        );
+
         return message;
       },
       {
@@ -272,20 +303,52 @@ export class MarQS {
     );
   }
 
+  public async getSharedQueueDetails() {
+    const parentQueue = this.keys.sharedQueueKey();
+
+    const { range } = await this.queuePriorityStrategy.nextCandidateSelection(
+      parentQueue,
+      "getSharedQueueDetails"
+    );
+    const queues = await this.#getChildQueuesWithScores(parentQueue, range);
+
+    const queuesWithScores = await this.#calculateQueueScores(queues, (queue) =>
+      this.#calculateMessageQueueCapacities(queue)
+    );
+
+    // We need to priority shuffle here to ensure all workers aren't just working on the highest priority queue
+    const choice = this.queuePriorityStrategy.chooseQueue(
+      queuesWithScores,
+      parentQueue,
+      "getSharedQueueDetails",
+      range
+    );
+
+    return {
+      selectionId: "getSharedQueueDetails",
+      queues,
+      queuesWithScores,
+      nextRange: range,
+      queueCount: queues.length,
+      queueChoice: choice,
+    };
+  }
+
   /**
    * Dequeue a message from the shared queue (this should be used in production environments)
    */
-  public async dequeueMessageInSharedQueue() {
+  public async dequeueMessageInSharedQueue(consumerId: string) {
     return this.#trace(
       "dequeueMessageInSharedQueue",
       async (span) => {
-        const parentQueue = constants.SHARED_QUEUE;
+        const parentQueue = this.keys.sharedQueueKey();
 
         // Read the parent queue for matching queues
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
           this.options.queuePriorityStrategy,
-          (queue) => this.#calculateMessageQueueCapacities(queue)
+          (queue) => this.#calculateMessageQueueCapacities(queue),
+          consumerId
         );
 
         if (!messageQueue) {
@@ -321,6 +384,11 @@ export class MarQS {
           });
         }
 
+        await this.options.visibilityTimeoutStrategy.heartbeat(
+          messageData.messageId,
+          this.visibilityTimeoutInMs
+        );
+
         return message;
       },
       {
@@ -350,6 +418,8 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callAcknowledgeMessage({
           parentQueue: message.parentQueue,
           messageKey: this.keys.messageKey(messageId),
@@ -415,6 +485,8 @@ export class MarQS {
           return;
         }
 
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callAcknowledgeMessage({
           parentQueue: oldMessage.parentQueue,
           messageKey: this.keys.messageKey(messageId),
@@ -444,27 +516,40 @@ export class MarQS {
     fn: (span: Span) => Promise<T>,
     options?: SpanOptions & { sampleRate?: number }
   ): Promise<T> {
-    return tracer.startActiveSpan(name, options ?? {}, async (span) => {
-      try {
-        return await fn(span);
-      } catch (e) {
-        if (e instanceof Error) {
-          span.recordException(e);
-        } else {
-          span.recordException(new Error(String(e)));
-        }
+    return this.tracer.startActiveSpan(
+      name,
+      {
+        ...options,
+        attributes: {
+          ...options?.attributes,
+        },
+      },
+      async (span) => {
+        try {
+          return await fn(span);
+        } catch (e) {
+          if (e instanceof Error) {
+            span.recordException(e);
+          } else {
+            span.recordException(new Error(String(e)));
+          }
 
-        throw e;
-      } finally {
-        span.end();
+          throw e;
+        } finally {
+          span.end();
+        }
       }
-    });
+    );
   }
 
   /**
    * Negative acknowledge a message, which will requeue the message
    */
-  public async nackMessage(messageId: string, retryAt: number = Date.now()) {
+  public async nackMessage(
+    messageId: string,
+    retryAt: number = Date.now(),
+    updates?: Record<string, unknown>
+  ) {
     return this.#trace(
       "nackMessage",
       async (span) => {
@@ -481,6 +566,12 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
+        if (updates) {
+          await this.replaceMessage(messageId, updates, retryAt, true);
+        }
+
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callNackMessage({
           messageKey: this.keys.messageKey(messageId),
           messageQueue: message.queue,
@@ -506,16 +597,11 @@ export class MarQS {
 
   // This should increment by the number of seconds, but with a max value of Date.now() + visibilityTimeoutInMs
   public async heartbeatMessage(messageId: string, seconds: number = 30) {
-    await this.#callHeartbeatMessage({
-      visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-      messageId,
-      milliseconds: seconds * 1000,
-      maxVisibilityTimeout: Date.now() + this.visibilityTimeoutInMs,
-    });
+    await this.options.visibilityTimeoutStrategy.heartbeat(messageId, seconds * 1000);
   }
 
   get visibilityTimeoutInMs() {
-    return this.options.visibilityTimeoutInMs ?? 300000;
+    return this.options.visibilityTimeoutInMs ?? 300000; // 5 minutes
   }
 
   async readMessage(messageId: string) {
@@ -531,9 +617,10 @@ export class MarQS {
         const message = MessagePayload.safeParse(JSON.parse(rawMessage));
 
         if (!message.success) {
-          logger.error("Failed to parse message", {
+          logger.error(`[${this.name}] Failed to parse message`, {
             messageId,
             error: message.error,
+            service: this.name,
           });
 
           return;
@@ -555,13 +642,15 @@ export class MarQS {
   async #getRandomQueueFromParentQueue(
     parentQueue: string,
     queuePriorityStrategy: MarQSQueuePriorityStrategy,
-    calculateCapacities: (queue: string) => Promise<QueueCapacities>
+    calculateCapacities: (queue: string) => Promise<QueueCapacities>,
+    consumerId: string
   ) {
     return this.#trace(
       "getRandomQueueFromParentQueue",
       async (span) => {
-        const { range, selectionId } = await queuePriorityStrategy.nextCandidateSelection(
-          parentQueue
+        const { range } = await queuePriorityStrategy.nextCandidateSelection(
+          parentQueue,
+          consumerId
         );
 
         const queues = await this.#getChildQueuesWithScores(parentQueue, range);
@@ -572,7 +661,8 @@ export class MarQS {
         const choice = this.queuePriorityStrategy.chooseQueue(
           queuesWithScores,
           parentQueue,
-          selectionId
+          consumerId,
+          range
         );
 
         span.setAttributes({
@@ -585,6 +675,28 @@ export class MarQS {
         span.setAttribute("nextRange.count", range.count);
         span.setAttribute("queueCount", queues.length);
 
+        if (this.options.verbose) {
+          if (typeof choice === "string") {
+            logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
+              queues,
+              queuesWithScores,
+              nextRange: range,
+              queueCount: queues.length,
+              queueChoice: choice,
+              consumerId,
+            });
+          } else {
+            logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
+              queues,
+              queuesWithScores,
+              nextRange: range,
+              queueCount: queues.length,
+              noQueueChoice: true,
+              consumerId,
+            });
+          }
+        }
+
         if (typeof choice !== "string") {
           span.setAttribute("noQueueChoice", true);
 
@@ -619,6 +731,7 @@ export class MarQS {
           queue: queue.value,
           capacities: await calculateCapacities(queue.value),
           age: now - queue.score,
+          size: await this.redis.zcard(queue.value),
         };
       })
     );
@@ -663,6 +776,10 @@ export class MarQS {
   }
 
   #startRebalanceWorkers() {
+    if (!this.options.enableRebalancing) {
+      return;
+    }
+
     // Start a new worker to rebalance parent queues periodically
     for (let i = 0; i < this.options.workers; i++) {
       const worker = new AsyncWorker(this.#rebalanceParentQueues.bind(this), 60_000);
@@ -748,6 +865,7 @@ export class MarQS {
         pattern,
         component: "marqs",
         operation: "rebalanceParentQueues",
+        service: this.name,
       });
 
       stream.on("data", async (keys) => {
@@ -759,6 +877,7 @@ export class MarQS {
           component: "marqs",
           operation: "rebalanceParentQueues",
           parentQueues: uniqueKeys,
+          service: this.name,
         });
 
         Promise.all(
@@ -808,6 +927,7 @@ export class MarQS {
           childQueuesWithScores,
           component: "marqs",
           operation: "rebalanceParentQueues",
+          service: this.name,
         });
 
         await Promise.all(
@@ -836,6 +956,7 @@ export class MarQS {
   async #callEnqueueMessage(message: MessagePayload) {
     logger.debug("Calling enqueueMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.enqueueMessage(
@@ -873,7 +994,6 @@ export class MarQS {
     const result = await this.redis.dequeueMessage(
       messageQueue,
       parentQueue,
-      visibilityQueue,
       concurrencyLimitKey,
       envConcurrencyLimitKey,
       orgConcurrencyLimitKey,
@@ -881,7 +1001,6 @@ export class MarQS {
       envCurrentConcurrencyKey,
       orgCurrentConcurrencyKey,
       messageQueue,
-      String(this.options.visibilityTimeoutInMs ?? 300000), // 5 minutes
       String(Date.now()),
       String(this.options.defaultEnvConcurrency),
       String(this.options.defaultOrgConcurrency)
@@ -893,6 +1012,7 @@ export class MarQS {
 
     logger.debug("Dequeue message result", {
       result,
+      service: this.name,
     });
 
     if (result.length !== 2) {
@@ -908,6 +1028,7 @@ export class MarQS {
   async #callReplaceMessage(message: MessagePayload) {
     logger.debug("Calling replaceMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.replaceMessage(
@@ -944,6 +1065,7 @@ export class MarQS {
       orgConcurrencyKey,
       messageId,
       parentQueue,
+      service: this.name,
     });
 
     return this.redis.acknowledgeMessage(
@@ -990,6 +1112,7 @@ export class MarQS {
       visibilityQueue,
       messageId,
       messageScore,
+      service: this.name,
     });
 
     return this.redis.nackMessage(
@@ -1007,25 +1130,6 @@ export class MarQS {
     );
   }
 
-  #callHeartbeatMessage({
-    visibilityQueue,
-    messageId,
-    milliseconds,
-    maxVisibilityTimeout,
-  }: {
-    visibilityQueue: string;
-    messageId: string;
-    milliseconds: number;
-    maxVisibilityTimeout: number;
-  }) {
-    return this.redis.heartbeatMessage(
-      visibilityQueue,
-      messageId,
-      String(milliseconds),
-      String(maxVisibilityTimeout)
-    );
-  }
-
   async #callCalculateMessageCapacities({
     currentConcurrencyKey,
     currentEnvConcurrencyKey,
@@ -1109,6 +1213,7 @@ export class MarQS {
         currentScore,
         rebalanceResult,
         operation: "rebalanceParentQueueChild",
+        service: this.name,
       });
     }
 
@@ -1145,25 +1250,23 @@ end
     });
 
     this.redis.defineCommand("dequeueMessage", {
-      numberOfKeys: 9,
+      numberOfKeys: 8,
       lua: `
--- Keys: childQueue, parentQueue, visibilityQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
+-- Keys: childQueue, parentQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local childQueue = KEYS[1]
 local parentQueue = KEYS[2]
-local visibilityQueue = KEYS[3]
-local concurrencyLimitKey = KEYS[4]
-local envConcurrencyLimitKey = KEYS[5]
-local orgConcurrencyLimitKey = KEYS[6]
-local currentConcurrencyKey = KEYS[7]
-local envCurrentConcurrencyKey = KEYS[8]
-local orgCurrentConcurrencyKey = KEYS[9]
+local concurrencyLimitKey = KEYS[3]
+local envConcurrencyLimitKey = KEYS[4]
+local orgConcurrencyLimitKey = KEYS[5]
+local currentConcurrencyKey = KEYS[6]
+local envCurrentConcurrencyKey = KEYS[7]
+local orgCurrentConcurrencyKey = KEYS[8]
 
--- Args: childQueueName, visibilityQueue, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
+-- Args: childQueueName, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
 local childQueueName = ARGV[1]
-local visibilityTimeout = tonumber(ARGV[2])
-local currentTime = tonumber(ARGV[3])
-local defaultEnvConcurrencyLimit = ARGV[4]
-local defaultOrgConcurrencyLimit = ARGV[5]
+local currentTime = tonumber(ARGV[2])
+local defaultEnvConcurrencyLimit = ARGV[3]
+local defaultOrgConcurrencyLimit = ARGV[4]
 
 -- Check current org concurrency against the limit
 local orgCurrentConcurrency = tonumber(redis.call('SCARD', orgCurrentConcurrencyKey) or '0')
@@ -1199,11 +1302,9 @@ end
 
 local messageId = messages[1]
 local messageScore = tonumber(messages[2])
-local timeoutScore = currentTime + visibilityTimeout
 
 -- Move message to timeout queue and update concurrency
 redis.call('ZREM', childQueue, messageId)
-redis.call('ZADD', visibilityQueue, timeoutScore, messageId)
 redis.call('SADD', currentConcurrencyKey, messageId)
 redis.call('SADD', envCurrentConcurrencyKey, messageId)
 redis.call('SADD', orgCurrentConcurrencyKey, messageId)
@@ -1269,7 +1370,7 @@ else
     redis.call('ZADD', parentQueue, earliestMessage[2], messageQueueName)
 end
 
--- Remove the message from the timeout queue
+-- Remove the message from the timeout queue (deprecated, will eventually remove this)
 redis.call('ZREM', visibilityQueue, messageId)
 
 -- Update the concurrency keys
@@ -1297,20 +1398,18 @@ local messageId = ARGV[2]
 local currentTime = tonumber(ARGV[3])
 local messageScore = tonumber(ARGV[4])
 
--- Check to see if the message is still in the visibilityQueue
-local messageVisibility = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
-
-if messageVisibility == 0 then
-    return
-end
-
 -- Update the concurrency keys
 redis.call('SREM', concurrencyKey, messageId)
 redis.call('SREM', envConcurrencyKey, messageId)
 redis.call('SREM', orgConcurrencyKey, messageId)
 
--- Remove the message from the timeout queue
-redis.call('ZREM', visibilityQueue, messageId)
+-- Check to see if the message is still in the visibilityQueue
+local messageVisibility = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
+
+if messageVisibility > 0 then
+-- Remove the message from the timeout queue (deprecated, will eventually remove this)
+    redis.call('ZREM', visibilityQueue, messageId)
+end
 
 -- Enqueue the message into the queue
 redis.call('ZADD', childQueueKey, messageScore, messageId)
@@ -1337,12 +1436,16 @@ local milliseconds = tonumber(ARGV[2])
 local maxVisibilityTimeout = tonumber(ARGV[3])
 
 -- Get the current visibility timeout
-local currentVisibilityTimeout = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
+local zscoreResult = redis.call('ZSCORE', visibilityQueue, messageId)
 
-if currentVisibilityTimeout == 0 then
+-- If there's no currentVisibilityTimeout, return and do not execute ZADD
+if zscoreResult == false then
     return
 end
 
+local currentVisibilityTimeout = tonumber(zscoreResult)
+
+
 -- Calculate the new visibility timeout
 local newVisibilityTimeout = math.min(currentVisibilityTimeout + milliseconds * 1000, maxVisibilityTimeout)
 
@@ -1445,7 +1548,6 @@ declare module "ioredis" {
     dequeueMessage(
       childQueue: string,
       parentQueue: string,
-      visibilityQueue: string,
       concurrencyLimitKey: string,
       envConcurrencyLimitKey: string,
       orgConcurrencyLimitKey: string,
@@ -1453,7 +1555,6 @@ declare module "ioredis" {
       envCurrentConcurrencyKey: string,
       orgCurrentConcurrencyKey: string,
       childQueueName: string,
-      visibilityTimeout: string,
       currentTime: string,
       defaultEnvConcurrencyLimit: string,
       defaultOrgConcurrencyLimit: string,
@@ -1548,7 +1649,10 @@ function getMarQSClient() {
       };
 
       return new MarQS({
+        name: "marqs",
+        tracer: trace.getTracer("marqs"),
         keysProducer: new MarQSShortKeyProducer(KEY_PREFIX),
+        visibilityTimeoutStrategy: new V3VisibilityTimeout(),
         queuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 36 }),
         envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
         workers: 1,
@@ -1556,6 +1660,7 @@ function getMarQSClient() {
         defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
         defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
         visibilityTimeoutInMs: 120 * 1000, // 2 minutes,
+        enableRebalancing: !env.MARQS_DISABLE_REBALANCING,
       });
     } else {
       console.warn(

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

@@ -58,6 +58,10 @@ export class MarQSShortKeyProducer implements MarQSKeyProducer {
       ].join(":");
     }
 
+    return this.sharedQueueKey();
+  }
+
+  sharedQueueKey(): string {
     return constants.SHARED_QUEUE;
   }
 

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

@@ -1,156 +0,0 @@
-import { RedisOptions } from "ioredis";
-import {
-  MarQSQueuePriorityStrategy,
-  PriorityStrategyChoice,
-  QueueRange,
-  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: QueueRange; selectionId: string }> {
-    throw new Error("Method not implemented.");
-  }
-}
-
-export type SimpleWeightedChoiceStrategyOptions = {
-  queueSelectionCount: number;
-  randomSeed?: string;
-};
-
-export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
-  private _nextRangesByParentQueue: Map<string, QueueRange> = new Map();
-  private _randomGenerator = seedrandom(this.options.randomSeed);
-
-  constructor(private options: SimpleWeightedChoiceStrategyOptions) {}
-
-  private nextRangeForParentQueue(parentQueue: string): QueueRange {
-    return (
-      this._nextRangesByParentQueue.get(parentQueue) ?? {
-        offset: 0,
-        count: this.options.queueSelectionCount,
-      }
-    );
-  }
-
-  chooseQueue(
-    queues: QueueWithScores[],
-    parentQueue: string,
-    selectionId: string
-  ): PriorityStrategyChoice {
-    const filteredQueues = filterQueuesAtCapacity(queues);
-
-    if (queues.length === this.options.queueSelectionCount) {
-      const nextRangeForParentQueue = this.nextRangeForParentQueue(parentQueue);
-      const nextRange: QueueRange = nextRangeForParentQueue
-        ? {
-            offset: nextRangeForParentQueue.offset + this.options.queueSelectionCount,
-            count: this.options.queueSelectionCount,
-          }
-        : { offset: this.options.queueSelectionCount, count: this.options.queueSelectionCount };
-      // 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);
-    }
-
-    if (filteredQueues.length === 0) {
-      return { abort: true };
-    }
-
-    const queueWeights = this.#calculateQueueWeights(filteredQueues);
-
-    return weightedRandomChoice(queueWeights, this._randomGenerator());
-  }
-
-  async nextCandidateSelection(
-    parentQueue: string
-  ): Promise<{ range: QueueRange; 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/requeueV2Message.server.ts

@@ -0,0 +1,27 @@
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+import { BaseService } from "../services/baseService.server";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { workerQueue } from "~/services/worker.server";
+import { marqsv2 } from "./v2.server";
+
+export class RequeueV2Message extends BaseService {
+  public async call(runId: string) {
+    logger.debug("[RequeueV2Message] Requeueing task run", { runId });
+
+    marqsv2?.nackMessage(runId);
+  }
+
+  public static async enqueue(runId: string, runAt?: Date, tx?: PrismaClientOrTransaction) {
+    return await workerQueue.enqueue(
+      "v2.requeueMessage",
+      { runId },
+      { runAt, jobKey: `requeueV2Message:${runId}` }
+    );
+  }
+
+  public static async dequeue(runId: string, tx?: PrismaClientOrTransaction) {
+    return await workerQueue.dequeue(`requeueV2Message:${runId}`, { tx });
+  }
+}

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

@@ -5,6 +5,7 @@ import {
   ProdTaskRunExecutionPayload,
   TaskRunError,
   TaskRunExecution,
+  TaskRunExecutionLazyAttemptPayload,
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
@@ -33,6 +34,9 @@ import {
 import { RestoreCheckpointService } from "../services/restoreCheckpoint.server";
 import { SEMINTATTRS_FORCE_RECORDING, tracer } from "../tracer.server";
 import { CrashTaskRunService } from "../services/crashTaskRun.server";
+import { FailedTaskRunService } from "../failedTaskRun.server";
+import { CreateTaskRunAttemptService } from "../services/createTaskRunAttempt.server";
+import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
 
 const WithTraceContext = z.object({
   traceparent: z.string().optional(),
@@ -86,6 +90,7 @@ export class SharedQueueConsumer {
   private _currentSpan: Span | undefined;
   private _endSpanInNextIteration = false;
   private _tasks = sharedQueueTasks;
+  private _id: string;
 
   constructor(
     private _sender: ZodMessageSender<typeof serverWebsocketMessages>,
@@ -97,6 +102,8 @@ export class SharedQueueConsumer {
       nextTickInterval: options.nextTickInterval ?? 1000, // 1 second
       interval: options.interval ?? 100, // 100ms
     };
+
+    this._id = generateFriendlyId("shared-queue", 6);
   }
 
   // This method is called when a background worker is deprecated and will no longer be used unless a run is locked to it
@@ -231,7 +238,7 @@ export class SharedQueueConsumer {
     // When the task run completes, ack the message
     // Using a heartbeat mechanism, if the client keeps responding with a heartbeat, we'll keep the message processing and increase the visibility timeout.
 
-    const message = await marqs?.dequeueMessageInSharedQueue();
+    const message = await marqs?.dequeueMessageInSharedQueue(this._id);
 
     if (!message) {
       this.#doMoreWork(this._options.nextTickInterval);
@@ -260,6 +267,14 @@ export class SharedQueueConsumer {
           where: {
             id: message.messageId,
           },
+          include: {
+            lockedToVersion: {
+              include: {
+                deployment: true,
+                tasks: true,
+              },
+            },
+          },
         });
 
         if (!existingTaskRun) {
@@ -291,7 +306,7 @@ export class SharedQueueConsumer {
           (!retryingFromCheckpoint &&
             !EXECUTABLE_RUN_STATUSES.withoutCheckpoint.includes(existingTaskRun.status))
         ) {
-          logger.debug("Task run has invalid status for execution", {
+          logger.error("Task run has invalid status for execution", {
             queueMessage: message.data,
             messageId: message.messageId,
             taskRun: existingTaskRun.id,
@@ -299,6 +314,12 @@ export class SharedQueueConsumer {
             retryingFromCheckpoint,
           });
 
+          const service = new CrashTaskRunService();
+          await service.call(existingTaskRun.id, {
+            crashAttempts: true,
+            reason: `Invalid run status for execution: ${existingTaskRun.status}`,
+          });
+
           await this.#ackAndDoMoreWork(message.messageId);
           return;
         }
@@ -398,6 +419,7 @@ export class SharedQueueConsumer {
                 createdAt: "desc",
               },
             },
+            lockedBy: true,
           },
         });
 
@@ -443,39 +465,12 @@ export class SharedQueueConsumer {
           return;
         }
 
-        const taskRunAttempt = await prisma.taskRunAttempt.create({
-          data: {
-            number: lockedTaskRun.attempts[0] ? lockedTaskRun.attempts[0].number + 1 : 1,
-            friendlyId: generateFriendlyId("attempt"),
-            taskRunId: lockedTaskRun.id,
-            startedAt: new Date(),
-            backgroundWorkerId: backgroundTask.workerId,
-            backgroundWorkerTaskId: backgroundTask.id,
-            status: "PENDING" as const,
-            queueId: queue.id,
-            runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
-          },
-          include: {
-            backgroundWorkerTask: true,
-          },
-        });
+        const nextAttemptNumber = lockedTaskRun.attempts[0]
+          ? lockedTaskRun.attempts[0].number + 1
+          : 1;
 
-        const isRetry = taskRunAttempt.number > 1;
+        const isRetry = nextAttemptNumber > 1;
 
-        const { machineConfig } = taskRunAttempt.backgroundWorkerTask;
-        const machine = Machine.safeParse(machineConfig ?? {});
-
-        if (!machine.success) {
-          logger.error("Failed to parse machine config", {
-            queueMessage: message.data,
-            messageId: message.messageId,
-            attemptId: taskRunAttempt.id,
-            machineConfig,
-          });
-
-          await this.#ackAndDoMoreWork(message.messageId);
-          return;
-        }
         try {
           if (messageBody.data.checkpointEventId) {
             const restoreService = new RestoreCheckpointService();
@@ -494,12 +489,35 @@ export class SharedQueueConsumer {
               await this.#ackAndDoMoreWork(message.messageId);
               return;
             }
-          } else if (isRetry) {
+
+            break;
+          }
+
+          if (!deployment.worker.supportsLazyAttempts) {
+            const service = new CreateTaskRunAttemptService();
+            await service.call(lockedTaskRun.friendlyId, undefined, false);
+          }
+
+          if (isRetry) {
             socketIo.coordinatorNamespace.emit("READY_FOR_RETRY", {
               version: "v1",
-              runId: taskRunAttempt.taskRunId,
+              runId: lockedTaskRun.id,
             });
           } else {
+            const machineConfig = lockedTaskRun.lockedBy?.machineConfig;
+            const machine = Machine.safeParse(machineConfig ?? {});
+
+            if (!machine.success) {
+              logger.error("Failed to parse machine config", {
+                queueMessage: message.data,
+                messageId: message.messageId,
+                machineConfig,
+              });
+
+              await this.#ackAndDoMoreWork(message.messageId);
+              return;
+            }
+
             await this._sender.send("BACKGROUND_WORKER_MESSAGE", {
               backgroundWorkerId: deployment.worker.friendlyId,
               data: {
@@ -508,12 +526,12 @@ export class SharedQueueConsumer {
                 version: deployment.version,
                 machine: machine.data,
                 // identifiers
-                id: taskRunAttempt.id,
+                id: "placeholder", // TODO: Remove this completely in a future release
                 envId: lockedTaskRun.runtimeEnvironment.id,
                 envType: lockedTaskRun.runtimeEnvironment.type,
                 orgId: lockedTaskRun.runtimeEnvironment.organizationId,
                 projectId: lockedTaskRun.runtimeEnvironment.projectId,
-                runId: taskRunAttempt.taskRunId,
+                runId: lockedTaskRun.id,
               },
             });
           }
@@ -535,11 +553,7 @@ export class SharedQueueConsumer {
               data: {
                 lockedAt: null,
                 lockedById: null,
-              },
-            }),
-            prisma.taskRunAttempt.delete({
-              where: {
-                id: taskRunAttempt.id,
+                status: lockedTaskRun.status,
               },
             }),
           ]);
@@ -1096,7 +1110,50 @@ class SharedQueueTasks {
     return this.getExecutionPayloadFromAttempt(latestAttempt.id, setToExecuting, isRetrying);
   }
 
+  async getLazyAttemptPayload(
+    envId: string,
+    runId: string
+  ): Promise<TaskRunExecutionLazyAttemptPayload | undefined> {
+    const environment = await findEnvironmentById(envId);
+
+    if (!environment) {
+      logger.error("Environment not found", { id: envId });
+      return;
+    }
+
+    const run = await prisma.taskRun.findUnique({
+      where: {
+        id: runId,
+        runtimeEnvironmentId: environment.id,
+      },
+    });
+
+    if (!run) {
+      logger.error("Run not found", { id: runId, envId });
+      return;
+    }
+
+    const environmentRepository = new EnvironmentVariablesRepository();
+    const variables = await environmentRepository.getEnvironmentVariables(
+      environment.projectId,
+      environment.id
+    );
+
+    return {
+      traceContext: run.traceContext as Record<string, unknown>,
+      environment: variables.reduce((acc: Record<string, string>, curr) => {
+        acc[curr.key] = curr.value;
+        return acc;
+      }, {}),
+      runId: run.friendlyId,
+      messageId: run.id,
+      isTest: run.isTest,
+    } satisfies TaskRunExecutionLazyAttemptPayload;
+  }
+
   async taskHeartbeat(attemptFriendlyId: string, seconds: number = 60) {
+    logger.debug("[SharedQueueConsumer] taskHeartbeat()", { id: attemptFriendlyId, seconds });
+
     const taskRunAttempt = await prisma.taskRunAttempt.findUnique({
       where: { friendlyId: attemptFriendlyId },
     });
@@ -1107,6 +1164,20 @@ class SharedQueueTasks {
 
     await marqs?.heartbeatMessage(taskRunAttempt.taskRunId, seconds);
   }
+
+  async taskRunHeartbeat(runId: string, seconds: number = 60) {
+    logger.debug("[SharedQueueConsumer] taskRunHeartbeat()", { runId, seconds });
+
+    await marqs?.heartbeatMessage(runId, seconds);
+  }
+
+  public async taskRunFailed(completion: TaskRunFailedExecutionResult) {
+    logger.debug("[SharedQueueConsumer] taskRunFailed()", { completion });
+
+    const service = new FailedTaskRunService();
+
+    await service.call(completion.id, completion);
+  }
 }
 
 export const sharedQueueTasks = singleton("sharedQueueTasks", () => new SharedQueueTasks());

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

@@ -0,0 +1,127 @@
+import { RedisOptions } from "ioredis";
+import { nanoid } from "nanoid";
+import {
+  MarQSQueuePriorityStrategy,
+  PriorityStrategyChoice,
+  QueueRange,
+  QueueWithScores,
+} from "./types";
+
+export type SimpleWeightedChoiceStrategyOptions = {
+  queueSelectionCount: number;
+  randomSeed?: string;
+  excludeEnvCapacity?: boolean;
+};
+
+export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  private _nextRangesByParentQueue: Map<string, QueueRange> = new Map();
+
+  constructor(private options: SimpleWeightedChoiceStrategyOptions) {}
+
+  private nextRangeForParentQueue(parentQueue: string, consumerId: string): QueueRange {
+    return (
+      this._nextRangesByParentQueue.get(`${consumerId}:${parentQueue}`) ?? {
+        offset: 0,
+        count: this.options.queueSelectionCount,
+      }
+    );
+  }
+
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    consumerId: string,
+    previousRange: QueueRange
+  ): PriorityStrategyChoice {
+    const filteredQueues = filterQueuesAtCapacity(queues);
+
+    if (queues.length === this.options.queueSelectionCount) {
+      const nextRange: QueueRange = {
+        offset: previousRange.offset + this.options.queueSelectionCount,
+        count: this.options.queueSelectionCount,
+      };
+      // 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(`${consumerId}:${parentQueue}`, nextRange);
+    } else {
+      this._nextRangesByParentQueue.delete(`${consumerId}:${parentQueue}`);
+    }
+
+    if (filteredQueues.length === 0) {
+      return { abort: true };
+    }
+
+    const queueWeights = this.#calculateQueueWeights(filteredQueues);
+
+    return weightedRandomChoice(queueWeights);
+  }
+
+  async nextCandidateSelection(
+    parentQueue: string,
+    consumerId: string
+  ): Promise<{ range: QueueRange }> {
+    return {
+      range: this.nextRangeForParentQueue(parentQueue, consumerId),
+    };
+  }
+
+  #calculateQueueWeights(queues: QueueWithScores[]) {
+    const avgQueueSize = queues.reduce((acc, { size }) => acc + size, 0) / queues.length;
+    const avgMessageAge = queues.reduce((acc, { age }) => acc + age, 0) / queues.length;
+
+    return queues.map(({ capacities, age, queue, size }) => {
+      let totalWeight = 1;
+
+      if (size > avgQueueSize) {
+        totalWeight += Math.min(size / avgQueueSize, 4);
+      }
+
+      if (age > avgMessageAge) {
+        totalWeight += Math.min(age / avgMessageAge, 4);
+      }
+
+      return {
+        queue,
+        totalWeight: age,
+      };
+    });
+  }
+}
+
+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; totalWeight: number }>) {
+  const totalWeight = queues.reduce((acc, queue) => acc + queue.totalWeight, 0);
+  let randomNum = Math.random() * totalWeight;
+
+  for (const queue of queues) {
+    if (randomNum < queue.totalWeight) {
+      return queue.queue;
+    }
+
+    randomNum -= queue.totalWeight;
+  }
+
+  // If we get here, we should just return a random queue
+  return queues[Math.floor(Math.random() * queues.length)].queue;
+}
+
+export class NoopWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice {
+    return { abort: true };
+  }
+
+  nextCandidateSelection(parentQueue: string): Promise<{ range: QueueRange; selectionId: string }> {
+    return Promise.resolve({ range: { offset: 0, count: 0 }, selectionId: nanoid(24) });
+  }
+}

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

@@ -16,6 +16,7 @@ export type QueueWithScores = {
   queue: string;
   capacities: QueueCapacities;
   age: number;
+  size: number;
 };
 
 export type QueueRange = { offset: number; count: number };
@@ -26,6 +27,7 @@ export interface MarQSKeyProducer {
   orgConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
   queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string): string;
   envSharedQueueKey(env: AuthenticatedEnvironment): string;
+  sharedQueueKey(): string;
   sharedQueueScanPattern(): string;
   concurrencyLimitKeyFromQueue(queue: string): string;
   currentConcurrencyKeyFromQueue(queue: string): string;
@@ -52,14 +54,15 @@ export interface MarQSQueuePriorityStrategy {
    *
    * @param queues
    * @param parentQueue
-   * @param selectionId
+   * @param consumerId
    *
    * @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
+    consumerId: string,
+    previousRange: QueueRange
   ): PriorityStrategyChoice;
 
   /**
@@ -68,10 +71,11 @@ export interface MarQSQueuePriorityStrategy {
    * The `selectionId` is used to identify the selection and should be passed to chooseQueue
    *
    * @param parentQueue The parent queue that holds the candidate queues
+   * @param consumerId The consumerId that is making the request
    *
    * @returns The scores and the selectionId for the next candidate selection
    */
-  nextCandidateSelection(parentQueue: string): Promise<{ range: QueueRange; selectionId: string }>;
+  nextCandidateSelection(parentQueue: string, consumerId: string): Promise<{ range: QueueRange }>;
 }
 
 export const MessagePayload = z.object({
@@ -85,3 +89,8 @@ export const MessagePayload = z.object({
 });
 
 export type MessagePayload = z.infer<typeof MessagePayload>;
+
+export interface VisibilityTimeoutStrategy {
+  heartbeat(messageId: string, timeoutInMs: number): Promise<void>;
+  cancelHeartbeat(messageId: string): Promise<void>;
+}

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

@@ -0,0 +1,322 @@
+import { trace } from "@opentelemetry/api";
+import { RetryOptions, calculateNextRetryDelay } from "@trigger.dev/core/v3";
+import { ConcurrencyLimitGroup, Job, JobVersion } from "@trigger.dev/database";
+import { z } from "zod";
+import { env } from "~/env.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { PerformRunExecutionV3Service } from "~/services/runs/performRunExecutionV3.server";
+import { singleton } from "~/utils/singleton";
+import { generateFriendlyId } from "../friendlyIdentifiers";
+import { MarQS } from "./index.server";
+import { MarQSShortKeyProducer } from "./marqsKeyProducer.server";
+import { RequeueV2Message } from "./requeueV2Message.server";
+import {
+  NoopWeightedChoiceStrategy,
+  SimpleWeightedChoiceStrategy,
+} from "./simpleWeightedPriorityStrategy.server";
+import { VisibilityTimeoutStrategy } from "./types";
+
+const KEY_PREFIX = "marqsv2:";
+const SHARED_QUEUE_NAME = "sharedQueue";
+
+export class V2VisibilityTimeout implements VisibilityTimeoutStrategy {
+  async heartbeat(messageId: string, timeoutInMs: number): Promise<void> {
+    RequeueV2Message.enqueue(messageId, new Date(Date.now() + timeoutInMs));
+  }
+
+  async cancelHeartbeat(messageId: string): Promise<void> {
+    RequeueV2Message.dequeue(messageId);
+  }
+}
+
+export class MarQSV2KeyProducer extends MarQSShortKeyProducer {
+  constructor(prefix: string) {
+    super(prefix);
+  }
+
+  envSharedQueueKey(env: AuthenticatedEnvironment) {
+    return SHARED_QUEUE_NAME;
+  }
+
+  sharedQueueKey(): string {
+    return SHARED_QUEUE_NAME;
+  }
+}
+
+export const marqsv2 = singleton("marqsv2", getMarQSClient);
+
+function getMarQSClient() {
+  if (env.V2_MARQS_ENABLED === "0") {
+    return;
+  }
+
+  if (!env.REDIS_HOST || !env.REDIS_PORT) {
+    throw new Error(
+      "Could not initialize marqsv2 because process.env.REDIS_HOST and process.env.REDIS_PORT are required to be set. Trigger.dev v2 will not work without this."
+    );
+  }
+
+  const redisOptions = {
+    keyPrefix: KEY_PREFIX,
+    port: env.REDIS_PORT,
+    host: env.REDIS_HOST,
+    username: env.REDIS_USERNAME,
+    password: env.REDIS_PASSWORD,
+    enableAutoPipelining: true,
+    ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+  };
+
+  return new MarQS({
+    verbose: env.V2_MARQS_VERBOSE === "1",
+    name: "marqsv2",
+    tracer: trace.getTracer("marqsv2"),
+    visibilityTimeoutStrategy: new V2VisibilityTimeout(),
+    keysProducer: new MarQSV2KeyProducer(KEY_PREFIX),
+    queuePriorityStrategy: new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: env.V2_MARQS_QUEUE_SELECTION_COUNT,
+    }),
+    envQueuePriorityStrategy: new NoopWeightedChoiceStrategy(), // We don't use this in v2, since all queues go through the shared queue
+    workers: 0,
+    redis: redisOptions,
+    defaultEnvConcurrency: env.V2_MARQS_DEFAULT_ENV_CONCURRENCY, // this is so we aren't limited by the environment concurrency
+    defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
+    visibilityTimeoutInMs: env.V2_MARQS_VISIBILITY_TIMEOUT_MS, // 15 minutes
+    enableRebalancing: !env.MARQS_DISABLE_REBALANCING,
+  });
+}
+
+export type V2QueueConsumerOptions = {
+  pollInterval?: number;
+  retryOptions?: RetryOptions;
+};
+
+const MessageBody = z.object({
+  version: z.literal("v1").default("v1"),
+  runId: z.string(),
+  attempt: z.number().default(1),
+});
+
+export class V2QueueConsumer {
+  private _enabled = false;
+  private _pollInterval: number;
+  private _retryOptions: RetryOptions = {
+    maxAttempts: 3,
+    factor: 2,
+    minTimeoutInMs: 1000,
+    maxTimeoutInMs: 60000,
+    randomize: true,
+  };
+  private _id: string;
+
+  constructor(private _options: V2QueueConsumerOptions = {}) {
+    this._pollInterval = this._options.pollInterval || 1000;
+    this._retryOptions = {
+      ...this._retryOptions,
+      ...this._options.retryOptions,
+    };
+    this._id = generateFriendlyId("v2-consumer", 6);
+  }
+
+  async start(startDelay: number = 0) {
+    if (this._enabled) {
+      return;
+    }
+
+    this._enabled = true;
+
+    // Only putting this here once does not actually delay the start of the consumer (for some reason)
+    await new Promise((resolve) => setTimeout(resolve, startDelay));
+    await new Promise((resolve) => setTimeout(resolve, startDelay));
+
+    logger.debug(`[marqsv2] Starting V2QueueConsumer`, {
+      startDelay,
+    });
+
+    return this.#doWork().catch(console.error);
+  }
+
+  async stop() {
+    if (!this._enabled) {
+      return;
+    }
+
+    logger.debug("[marqsv2] Stopping V2QueueConsumer");
+
+    this._enabled = false;
+  }
+
+  async #doWork() {
+    if (!this._enabled) {
+      return;
+    }
+
+    await this.#doWorkInternal();
+  }
+
+  async #doWorkInternal() {
+    const message = await marqsv2?.dequeueMessageInSharedQueue(this._id);
+
+    if (!message) {
+      setTimeout(() => this.#doWork(), this._pollInterval);
+      return;
+    }
+
+    const messageBody = MessageBody.safeParse(message.data);
+
+    if (!messageBody.success) {
+      logger.error("[marqsv2] Failed to parse message", {
+        queueMessage: message.data,
+        error: messageBody.error,
+      });
+
+      await marqsv2?.acknowledgeMessage(message.messageId);
+
+      setTimeout(() => this.#doWork(), this._pollInterval);
+      return;
+    }
+
+    logger.debug("[V2QueueConsumer] Received message", {
+      messageData: messageBody.data,
+    });
+
+    try {
+      const service = new PerformRunExecutionV3Service();
+
+      await service.call({
+        id: messageBody.data.runId,
+        reason: "EXECUTE_JOB",
+        isRetry: false,
+        lastAttempt: false,
+      });
+    } catch (error) {
+      logger.error("[marqsv2] Failed to execute job", {
+        runId: messageBody.data.runId,
+        error,
+      });
+
+      const attempt = messageBody.data.attempt + 1;
+
+      const retryDelay = calculateNextRetryDelay(this._retryOptions, attempt);
+
+      if (!retryDelay) {
+        logger.error("[marqsv2] Job failed after max attempts", {
+          runId: messageBody.data.runId,
+          attempt,
+        });
+
+        await marqsv2?.acknowledgeMessage(message.messageId);
+      } else {
+        await marqsv2?.nackMessage(message.messageId, Date.now() + retryDelay, {
+          attempt,
+        });
+      }
+    } finally {
+      setTimeout(() => this.#doWork(), this._pollInterval);
+    }
+  }
+}
+
+interface V2QueueConsumerPoolOptions {
+  poolSize: number;
+  pollInterval: number;
+}
+
+class V2QueueConsumerPool {
+  #consumers: V2QueueConsumer[];
+  #shuttingDown: boolean = false;
+
+  constructor(private opts: V2QueueConsumerPoolOptions) {
+    this.#consumers = Array(opts.poolSize)
+      .fill(null)
+      .map((_, i) => new V2QueueConsumer({ pollInterval: opts.pollInterval }));
+
+    process.on("SIGTERM", this.#handleSignal.bind(this));
+    process.on("SIGINT", this.#handleSignal.bind(this));
+  }
+
+  async start() {
+    await Promise.allSettled(
+      this.#consumers.map((consumer, i) =>
+        consumer.start(i * (this.opts.pollInterval / this.opts.poolSize))
+      )
+    );
+  }
+
+  async stop() {
+    await Promise.allSettled(this.#consumers.map((consumer) => consumer.stop()));
+  }
+
+  async #handleSignal(signal: string) {
+    if (this.#shuttingDown) {
+      return;
+    }
+
+    this.#shuttingDown = true;
+
+    logger.debug(`[V2QueueConsumerPool] Received ${signal}, shutting down...`);
+
+    this.stop().finally(() => {
+      logger.debug("V2QueueConsumerPool shutdown");
+    });
+  }
+}
+
+export const v2QueueConsumerPool = singleton("v2QueueConsumerPool", initalizePool);
+
+async function initalizePool() {
+  if (env.V2_MARQS_ENABLED === "0") {
+    return;
+  }
+
+  if (env.V2_MARQS_CONSUMER_POOL_ENABLED === "0") {
+    return;
+  }
+
+  console.log(
+    `🎱 Initializing V2QueueConsumerPool (poolSize=${env.V2_MARQS_CONSUMER_POOL_SIZE}, pollInterval=${env.V2_MARQS_CONSUMER_POLL_INTERVAL_MS})`
+  );
+
+  const pool = new V2QueueConsumerPool({
+    poolSize: env.V2_MARQS_CONSUMER_POOL_SIZE,
+    pollInterval: env.V2_MARQS_CONSUMER_POLL_INTERVAL_MS,
+  });
+
+  await pool.start();
+
+  return pool;
+}
+
+export async function putConcurrencyLimitGroup(
+  concurrencyLimitGroup: ConcurrencyLimitGroup,
+  env: AuthenticatedEnvironment
+): Promise<void> {
+  logger.debug(`[marqsv2] Updating concurrency limit group`, {
+    concurrencyLimitGroup,
+    environment: env,
+  });
+
+  await marqsv2?.updateQueueConcurrencyLimits(
+    env,
+    `group/${concurrencyLimitGroup.name}`,
+    concurrencyLimitGroup.concurrencyLimit
+  );
+}
+
+export async function putJobConcurrencyLimit(
+  job: Job,
+  version: JobVersion,
+  env: AuthenticatedEnvironment
+): Promise<void> {
+  logger.debug(`[marqsv2] Updating job concurrency limit`, {
+    job,
+    version,
+    environment: env,
+  });
+
+  if (typeof version.concurrencyLimit === "number" && version.concurrencyLimit > 0) {
+    await marqsv2?.updateQueueConcurrencyLimits(env, `job/${job.slug}`, version.concurrencyLimit);
+  } else {
+    await marqsv2?.removeQueueConcurrencyLimits(env, `job/${job.slug}`);
+  }
+}

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

@@ -0,0 +1,12 @@
+import { RequeueTaskRunService } from "../requeueTaskRun.server";
+import { VisibilityTimeoutStrategy } from "./types";
+
+export class V3VisibilityTimeout implements VisibilityTimeoutStrategy {
+  async heartbeat(messageId: string, timeoutInMs: number): Promise<void> {
+    await RequeueTaskRunService.enqueue(messageId, new Date(Date.now() + timeoutInMs));
+  }
+
+  async cancelHeartbeat(messageId: string): Promise<void> {
+    await RequeueTaskRunService.dequeue(messageId);
+  }
+}

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

@@ -0,0 +1,95 @@
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+import assertNever from "assert-never";
+import { FailedTaskRunService } from "./failedTaskRun.server";
+import { BaseService } from "./services/baseService.server";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { workerQueue } from "~/services/worker.server";
+
+export class RequeueTaskRunService extends BaseService {
+  public async call(runId: string) {
+    const taskRun = await this._prisma.taskRun.findUnique({
+      where: { id: runId },
+    });
+
+    if (!taskRun) {
+      logger.error("[RequeueTaskRunService] Task run not found", {
+        runId,
+      });
+
+      return;
+    }
+
+    switch (taskRun.status) {
+      case "PENDING": {
+        logger.debug("[RequeueTaskRunService] Requeueing task run", { taskRun });
+
+        await marqs?.nackMessage(taskRun.id);
+
+        break;
+      }
+      case "EXECUTING":
+      case "RETRYING_AFTER_FAILURE": {
+        logger.debug("[RequeueTaskRunService] Failing task run", { taskRun });
+
+        const service = new FailedTaskRunService();
+
+        await service.call(taskRun.friendlyId, {
+          ok: false,
+          id: taskRun.friendlyId,
+          retry: undefined,
+          error: {
+            type: "INTERNAL_ERROR",
+            code: "TASK_RUN_HEARTBEAT_TIMEOUT",
+            message: "Did not receive a heartbeat from the worker in time",
+          },
+        });
+
+        break;
+      }
+      case "WAITING_FOR_DEPLOY": {
+        logger.debug("[RequeueTaskRunService] Removing task run from queue", { taskRun });
+
+        await marqs?.acknowledgeMessage(taskRun.id);
+
+        break;
+      }
+      case "WAITING_TO_RESUME":
+      case "PAUSED": {
+        logger.debug("[RequeueTaskRunService] Requeueing task run", { taskRun });
+
+        await marqs?.nackMessage(taskRun.id);
+
+        break;
+      }
+      case "SYSTEM_FAILURE":
+      case "INTERRUPTED":
+      case "CRASHED":
+      case "COMPLETED_WITH_ERRORS":
+      case "COMPLETED_SUCCESSFULLY":
+      case "CANCELED": {
+        logger.debug("[RequeueTaskRunService] Task run is completed", { taskRun });
+
+        await marqs?.acknowledgeMessage(taskRun.id);
+
+        break;
+      }
+      default: {
+        assertNever(taskRun.status);
+      }
+    }
+  }
+
+  public static async enqueue(runId: string, runAt?: Date, tx?: PrismaClientOrTransaction) {
+    return await workerQueue.enqueue(
+      "v3.requeueTaskRun",
+      { runId },
+      { runAt, jobKey: `requeueTaskRun:${runId}` }
+    );
+  }
+
+  public static async dequeue(runId: string, tx?: PrismaClientOrTransaction) {
+    return await workerQueue.dequeue(`requeueTaskRun:${runId}`, { tx });
+  }
+}

apps/webapp/app/v3/services/alerts/deliverAlert.server.ts

@@ -808,7 +808,6 @@ export class DeliverAlertService extends BaseService {
         tx,
         runAt: options?.runAt,
         jobKey: `deliverAlert:${alertId}`,
-        queueName: options?.queueName,
       }
     );
   }

apps/webapp/app/v3/services/alerts/performDeploymentAlerts.server.ts

@@ -58,9 +58,7 @@ export class PerformDeploymentAlertsService extends BaseService {
         },
       });
 
-      await DeliverAlertService.enqueue(alert.id, tx, {
-        queueName: `alert-channel:${alertChannel.id}`,
-      });
+      await DeliverAlertService.enqueue(alert.id, tx);
     });
   }
 

apps/webapp/app/v3/services/alerts/performTaskAttemptAlerts.server.ts

@@ -59,9 +59,7 @@ export class PerformTaskAttemptAlertsService extends BaseService {
         },
       });
 
-      await DeliverAlertService.enqueue(alert.id, tx, {
-        queueName: `alert-channel:${alertChannel.id}`,
-      });
+      await DeliverAlertService.enqueue(alert.id, tx);
     });
   }
 

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

@@ -34,7 +34,7 @@ export abstract class BaseService {
 }
 
 export class ServiceValidationError extends Error {
-  constructor(message: string) {
+  constructor(message: string, public status?: number) {
     super(message);
     this.name = "ServiceValidationError";
   }

apps/webapp/app/v3/services/bulk/performBulkAction.server.ts

@@ -88,7 +88,6 @@ export class PerformBulkActionService extends BaseService {
       },
       {
         jobKey: `performBulkActionItem:${bulkActionItemId}`,
-        queueName: `bulkActionItem:${groupId}`,
       }
     );
   }

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

@@ -24,9 +24,15 @@ const CANCELLABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = [
   "PENDING",
 ];
 
-type ExtendedTaskRunAttempt = Prisma.TaskRunAttemptGetPayload<{
+type ExtendedTaskRun = Prisma.TaskRunGetPayload<{
   include: {
     runtimeEnvironment: true;
+    lockedToVersion: true;
+  };
+}>;
+
+type ExtendedTaskRunAttempt = Prisma.TaskRunAttemptGetPayload<{
+  include: {
     backgroundWorker: true;
   };
 }>;
@@ -71,11 +77,10 @@ export class CancelTaskRunService extends BaseService {
           },
           include: {
             backgroundWorker: true,
-            runtimeEnvironment: true,
           },
         },
-        dependency: true,
         runtimeEnvironment: true,
+        lockedToVersion: true,
       },
     });
 
@@ -96,6 +101,7 @@ export class CancelTaskRunService extends BaseService {
     // Cancel any in progress attempts
     if (opts.cancelAttempts) {
       await this.#cancelPotentiallyRunningAttempts(cancelledTaskRun, cancelledTaskRun.attempts);
+      await this.#cancelRemainingRunWorkers(cancelledTaskRun);
     }
 
     return {
@@ -103,9 +109,12 @@ export class CancelTaskRunService extends BaseService {
     };
   }
 
-  async #cancelPotentiallyRunningAttempts(run: TaskRun, attempts: ExtendedTaskRunAttempt[]) {
+  async #cancelPotentiallyRunningAttempts(
+    run: ExtendedTaskRun,
+    attempts: ExtendedTaskRunAttempt[]
+  ) {
     for (const attempt of attempts) {
-      if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
+      if (run.runtimeEnvironment.type === "DEVELOPMENT") {
         // Signal the task run attempt to stop
         await devPubSub.publish(
           `backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
@@ -158,4 +167,19 @@ export class CancelTaskRunService extends BaseService {
       }
     }
   }
+
+  async #cancelRemainingRunWorkers(run: ExtendedTaskRun) {
+    if (run.runtimeEnvironment.type === "DEVELOPMENT") {
+      // Nothing to do
+      return;
+    }
+
+    // Broadcast cancel message to all coordinators
+    socketIo.coordinatorNamespace.emit("REQUEST_RUN_CANCELLATION", {
+      version: "v1",
+      runId: run.id,
+      // Give the attempts some time to exit gracefully. If the runs supports lazy attempts, it also supports exit delays.
+      delayInMs: run.lockedToVersion?.supportsLazyAttempts ? 5_000 : undefined,
+    });
+  }
 }

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

@@ -11,7 +11,7 @@ import { PrismaClientOrTransaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { safeJsonParse } from "~/utils/json";
-import { eventRepository } from "../eventRepository.server";
+import { createExceptionPropertiesFromError, eventRepository } from "../eventRepository.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
@@ -20,6 +20,7 @@ import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
 import { CreateCheckpointService } from "./createCheckpoint.server";
 import { TaskRun } from "@trigger.dev/database";
 import { PerformTaskAttemptAlertsService } from "./alerts/performTaskAttemptAlerts.server";
+import { RetryAttemptService } from "./retryAttempt.server";
 
 type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
 
@@ -57,6 +58,8 @@ export class CompleteAttemptService extends BaseService {
         },
       });
 
+      // No attempt, so there's no message to ACK
+
       return "COMPLETED";
     }
 
@@ -143,6 +146,8 @@ export class CompleteAttemptService extends BaseService {
         env
       );
 
+      // The cancel service handles ACK
+
       return "COMPLETED";
     }
 
@@ -173,7 +178,7 @@ export class CompleteAttemptService extends BaseService {
           properties: {
             retryAt: retryAt.toISOString(),
           },
-          runId: taskRunAttempt.taskRunId,
+          runId: taskRunAttempt.taskRun.friendlyId,
           style: {
             icon: "schedule-attempt",
           },
@@ -185,7 +190,10 @@ export class CompleteAttemptService extends BaseService {
         endTime: retryAt,
       });
 
-      logger.debug("Retrying", { taskRun: taskRunAttempt.taskRun.friendlyId });
+      logger.debug("Retrying", {
+        taskRun: taskRunAttempt.taskRun.friendlyId,
+        retry: completion.retry,
+      });
 
       await this._prisma.taskRun.update({
         where: {
@@ -203,7 +211,12 @@ export class CompleteAttemptService extends BaseService {
       }
 
       if (!checkpoint) {
-        await this.#enqueueRetry(taskRunAttempt.taskRun, completion.retry.timestamp);
+        await this.#retryAttempt(
+          taskRunAttempt.taskRun,
+          completion.retry.timestamp,
+          undefined,
+          taskRunAttempt.backgroundWorker.supportsLazyAttempts
+        );
         return "RETRIED";
       }
 
@@ -231,10 +244,12 @@ export class CompleteAttemptService extends BaseService {
           },
         });
 
+        await marqs?.acknowledgeMessage(taskRunAttempt.taskRunId);
+
         return "COMPLETED";
       }
 
-      await this.#enqueueRetry(
+      await this.#retryAttempt(
         taskRunAttempt.taskRun,
         completion.retry.timestamp,
         checkpointCreateResult.event.id
@@ -253,6 +268,15 @@ export class CompleteAttemptService extends BaseService {
         attributes: {
           isError: true,
         },
+        events: [
+          {
+            name: "exception",
+            time: new Date(),
+            properties: {
+              exception: createExceptionPropertiesFromError(completion.error),
+            },
+          },
+        ],
       });
 
       if (
@@ -310,17 +334,28 @@ export class CompleteAttemptService extends BaseService {
     }
   }
 
-  async #enqueueRetry(run: TaskRun, retryTimestamp: number, checkpointEventId?: string) {
-    // We have to replace a potential RESUME with EXECUTE to correctly retry the attempt
-    return await marqs?.replaceMessage(
-      run.id,
-      {
-        type: "EXECUTE",
-        taskIdentifier: run.taskIdentifier,
-        checkpointEventId: checkpointEventId,
-      },
-      retryTimestamp
-    );
+  async #retryAttempt(
+    run: TaskRun,
+    retryTimestamp: number,
+    checkpointEventId?: string,
+    supportsLazyAttempts?: boolean
+  ) {
+    if (checkpointEventId || !supportsLazyAttempts) {
+      // We have to replace a potential RESUME with EXECUTE to correctly retry the attempt
+      return await marqs?.replaceMessage(
+        run.id,
+        {
+          type: "EXECUTE",
+          taskIdentifier: run.taskIdentifier,
+          checkpointEventId: checkpointEventId,
+        },
+        retryTimestamp
+      );
+    } else {
+      // There's no checkpoint so the worker is still running and waiting for a retry message
+      // It supports lazy attempts so we can bypass the queue and send the message directly to the worker
+      RetryAttemptService.enqueue(run.id, this._prisma, new Date(retryTimestamp));
+    }
   }
 
   #generateMetadataAttributesForNextAttempt(execution: TaskRunExecution) {
@@ -353,6 +388,7 @@ async function findAttempt(prismaClient: PrismaClientOrTransaction, friendlyId:
     include: {
       taskRun: true,
       backgroundWorkerTask: true,
+      backgroundWorker: true,
     },
   });
 }

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

@@ -63,6 +63,7 @@ export class CreateBackgroundWorkerService extends BaseService {
           contentHash: body.metadata.contentHash,
           cliVersion: body.metadata.cliPackageVersion,
           sdkVersion: body.metadata.packageVersion,
+          supportsLazyAttempts: body.supportsLazyAttempts,
         },
       });
 
@@ -178,6 +179,8 @@ export async function createBackgroundTasks(
           taskQueue.name,
           taskQueue.concurrencyLimit
         );
+      } else {
+        await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
       }
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError) {

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

@@ -45,6 +45,7 @@ export class CreateDeployedBackgroundWorkerService extends BaseService {
           contentHash: body.metadata.contentHash,
           cliVersion: body.metadata.cliPackageVersion,
           sdkVersion: body.metadata.packageVersion,
+          supportsLazyAttempts: body.supportsLazyAttempts,
         },
       });
 

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

@@ -0,0 +1,213 @@
+import { TaskRunExecution } from "@trigger.dev/core/v3";
+import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { generateFriendlyId } from "../friendlyIdentifiers";
+import { BaseService, ServiceValidationError } from "./baseService.server";
+import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
+
+export class CreateTaskRunAttemptService extends BaseService {
+  public async call(
+    runId: string,
+    env?: AuthenticatedEnvironment,
+    setToExecuting = true
+  ): Promise<{
+    execution: TaskRunExecution;
+    run: TaskRun;
+    attempt: TaskRunAttempt;
+  }> {
+    const environment = env ?? (await getAuthenticatedEnvironmentFromRun(runId, this._prisma));
+
+    if (!environment) {
+      throw new ServiceValidationError("Environment not found", 404);
+    }
+
+    const isFriendlyId = runId.startsWith("run_");
+
+    return await this.traceWithEnv("call()", environment, async (span) => {
+      if (isFriendlyId) {
+        span.setAttribute("taskRunFriendlyId", runId);
+      } else {
+        span.setAttribute("taskRunId", runId);
+      }
+
+      const taskRun = await this._prisma.taskRun.findUnique({
+        where: {
+          id: !isFriendlyId ? runId : undefined,
+          friendlyId: isFriendlyId ? runId : undefined,
+          runtimeEnvironmentId: environment.id,
+        },
+        include: {
+          tags: true,
+          attempts: {
+            take: 1,
+            orderBy: {
+              number: "desc",
+            },
+          },
+          lockedBy: {
+            include: {
+              worker: true,
+            },
+          },
+          batchItems: {
+            include: {
+              batchTaskRun: true,
+            },
+          },
+        },
+      });
+
+      logger.debug("Creating a task run attempt", { taskRun });
+
+      if (!taskRun) {
+        throw new ServiceValidationError("Task run not found", 404);
+      }
+
+      span.setAttribute("taskRunId", taskRun.id);
+      span.setAttribute("taskRunFriendlyId", taskRun.friendlyId);
+
+      if (taskRun.status === "CANCELED") {
+        throw new ServiceValidationError("Task run is cancelled", 400);
+      }
+
+      if (!taskRun.lockedBy) {
+        throw new ServiceValidationError("Task run is not locked", 400);
+      }
+
+      const queue = await this._prisma.taskQueue.findUnique({
+        where: {
+          runtimeEnvironmentId_name: {
+            runtimeEnvironmentId: environment.id,
+            name: taskRun.queue,
+          },
+        },
+      });
+
+      if (!queue) {
+        throw new ServiceValidationError("Queue not found", 404);
+      }
+
+      const nextAttemptNumber = taskRun.attempts[0] ? taskRun.attempts[0].number + 1 : 1;
+
+      const taskRunAttempt = await $transaction(this._prisma, async (tx) => {
+        const taskRunAttempt = await tx.taskRunAttempt.create({
+          data: {
+            number: nextAttemptNumber,
+            friendlyId: generateFriendlyId("attempt"),
+            taskRunId: taskRun.id,
+            startedAt: new Date(),
+            backgroundWorkerId: taskRun.lockedBy!.worker.id,
+            backgroundWorkerTaskId: taskRun.lockedBy!.id,
+            status: setToExecuting ? "EXECUTING" : "PENDING",
+            queueId: queue.id,
+            runtimeEnvironmentId: environment.id,
+          },
+          include: {
+            backgroundWorker: true,
+            backgroundWorkerTask: true,
+          },
+        });
+
+        if (setToExecuting) {
+          await tx.taskRun.update({
+            where: {
+              id: taskRun.id,
+            },
+            data: {
+              status: "EXECUTING",
+            },
+          });
+        }
+
+        return taskRunAttempt;
+      });
+
+      if (!taskRunAttempt) {
+        logger.error("Failed to create task run attempt", { runId: taskRun.id, nextAttemptNumber });
+        throw new ServiceValidationError("Failed to create task run attempt", 500);
+      }
+
+      const execution: TaskRunExecution = {
+        task: {
+          id: taskRun.lockedBy.slug,
+          filePath: taskRun.lockedBy.filePath,
+          exportName: taskRun.lockedBy.exportName,
+        },
+        attempt: {
+          id: taskRunAttempt.friendlyId,
+          number: taskRunAttempt.number,
+          startedAt: taskRunAttempt.startedAt ?? taskRunAttempt.createdAt,
+          backgroundWorkerId: taskRun.lockedBy.worker.id,
+          backgroundWorkerTaskId: taskRun.lockedBy.id,
+          status: "EXECUTING" as const,
+        },
+        run: {
+          id: taskRun.friendlyId,
+          payload: taskRun.payload,
+          payloadType: taskRun.payloadType,
+          context: taskRun.context,
+          createdAt: taskRun.createdAt,
+          tags: taskRun.tags.map((tag) => tag.name),
+          isTest: taskRun.isTest,
+          idempotencyKey: taskRun.idempotencyKey ?? undefined,
+        },
+        queue: {
+          id: queue.friendlyId,
+          name: queue.name,
+        },
+        environment: {
+          id: environment.id,
+          slug: environment.slug,
+          type: environment.type,
+        },
+        organization: {
+          id: environment.organization.id,
+          slug: environment.organization.slug,
+          name: environment.organization.title,
+        },
+        project: {
+          id: environment.project.id,
+          ref: environment.project.externalRef,
+          slug: environment.project.slug,
+          name: environment.project.name,
+        },
+        batch:
+          taskRun.batchItems[0] && taskRun.batchItems[0].batchTaskRun
+            ? { id: taskRun.batchItems[0].batchTaskRun.friendlyId }
+            : undefined,
+      };
+
+      return {
+        execution,
+        run: taskRun,
+        attempt: taskRunAttempt,
+      };
+    });
+  }
+}
+
+async function getAuthenticatedEnvironmentFromRun(
+  friendlyId: string,
+  prismaClient?: PrismaClientOrTransaction
+) {
+  const taskRun = await (prismaClient ?? prisma).taskRun.findUnique({
+    where: {
+      friendlyId,
+    },
+    include: {
+      runtimeEnvironment: {
+        include: {
+          organization: true,
+          project: true,
+        },
+      },
+    },
+  });
+
+  if (!taskRun) {
+    return;
+  }
+
+  return taskRun?.runtimeEnvironment;
+}

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

@@ -62,12 +62,12 @@ export class IndexDeploymentService extends BaseService {
       logger.debug("Index ACK received", { responses });
 
       if (responses.length === 0) {
-        // timeout the deployment if 50 seconds have passed and the deployment is still not indexed
+        // timeout the deployment if 180 seconds have passed and the deployment is still not indexed
         await TimeoutDeploymentService.enqueue(
           deployment.id,
           "DEPLOYING",
           "Could not index deployment in time",
-          new Date(Date.now() + 50_000)
+          new Date(Date.now() + 180_000)
         );
       } else {
         const indexFailed = new DeploymentIndexFailed();

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

@@ -6,6 +6,7 @@ import { createRemoteImageBuild } from "../remoteImageBuilder.server";
 import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion";
 import { BaseService } from "./baseService.server";
 import { TimeoutDeploymentService } from "./timeoutDeployment.server";
+import { env } from "~/env.server";
 
 const nanoid = customAlphabet("1234567890abcdefghijklmnopqrstuvwxyz", 8);
 
@@ -64,7 +65,7 @@ export class InitializeDeploymentService extends BaseService {
         new Date(Date.now() + 180_000) // 3 minutes
       );
 
-      const imageTag = `trigger/${environment.project.externalRef}:${deployment.version}.${environment.slug}`;
+      const imageTag = `${env.DEPLOY_REGISTRY_NAMESPACE}/${environment.project.externalRef}:${deployment.version}.${environment.slug}`;
 
       return { deployment, imageTag };
     });

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

@@ -107,7 +107,7 @@ export class ResumeBatchRunService extends BaseService {
       {
         tx,
         runAt,
-        queueName: `resumeBatchRun-${batchRunId}`,
+        jobKey: `resumeBatchRun-${batchRunId}`,
       }
     );
   }

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

@@ -0,0 +1,39 @@
+import { BaseService } from "./baseService.server";
+import { logger } from "~/services/logger.server";
+import { socketIo } from "../handleSocketIo.server";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { workerQueue } from "~/services/worker.server";
+
+export class RetryAttemptService extends BaseService {
+  public async call(runId: string) {
+    const taskRun = await this._prisma.taskRun.findFirst({
+      where: {
+        id: runId,
+      },
+    });
+
+    if (!taskRun) {
+      logger.error("Task run not found", { runId });
+      return;
+    }
+
+    socketIo.coordinatorNamespace.emit("READY_FOR_RETRY", {
+      version: "v1",
+      runId,
+    });
+  }
+
+  static async enqueue(runId: string, tx: PrismaClientOrTransaction, runAt?: Date) {
+    return await workerQueue.enqueue(
+      "v3.retryAttempt",
+      {
+        runId,
+      },
+      {
+        tx,
+        runAt,
+        jobKey: `retryAttempt:${runId}`,
+      }
+    );
+  }
+}

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

@@ -4,13 +4,12 @@ import {
   TriggerTaskRequestBody,
   packetRequiresOffloading,
 } from "@trigger.dev/core/v3";
-import { nanoid } from "nanoid";
-import { createHash } from "node:crypto";
-import { $transaction } from "~/db.server";
+import { prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { autoIncrementCounter } from "~/services/autoIncrementCounter.server";
+import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { eventRepository } from "../eventRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "~/v3/marqs/index.server";
 import { uploadToObjectStore } from "../r2.server";
 import { BaseService } from "./baseService.server";
 
@@ -85,106 +84,151 @@ export class TriggerTaskService extends BaseService {
             environment
           );
 
-          const lockId = taskIdentifierToLockId(taskId);
+          const run = await autoIncrementCounter.incrementInTransaction(
+            `v3-run:${environment.id}:${taskId}`,
+            async (num, tx) => {
+              const lockedToBackgroundWorker = body.options?.lockToVersion
+                ? await tx.backgroundWorker.findUnique({
+                    where: {
+                      projectId_runtimeEnvironmentId_version: {
+                        projectId: environment.projectId,
+                        runtimeEnvironmentId: environment.id,
+                        version: body.options?.lockToVersion,
+                      },
+                    },
+                  })
+                : undefined;
 
-          const run = await $transaction(this._prisma, async (tx) => {
-            const lockedToBackgroundWorker = body.options?.lockToVersion
-              ? await tx.backgroundWorker.findUnique({
+              let queueName = sanitizeQueueName(body.options?.queue?.name ?? `task/${taskId}`);
+
+              // Check that the queuename is not an empty string
+              if (!queueName) {
+                queueName = sanitizeQueueName(`task/${taskId}`);
+              }
+
+              event.setAttribute("queueName", queueName);
+              span.setAttribute("queueName", queueName);
+
+              const taskRun = await tx.taskRun.create({
+                data: {
+                  status: "PENDING",
+                  number: num,
+                  friendlyId: runFriendlyId,
+                  runtimeEnvironmentId: environment.id,
+                  projectId: environment.projectId,
+                  idempotencyKey,
+                  taskIdentifier: taskId,
+                  payload: payloadPacket.data ?? "",
+                  payloadType: payloadPacket.dataType,
+                  context: body.context,
+                  traceContext: traceContext,
+                  traceId: event.traceId,
+                  spanId: event.spanId,
+                  lockedToVersionId: lockedToBackgroundWorker?.id,
+                  concurrencyKey: body.options?.concurrencyKey,
+                  queue: queueName,
+                  isTest: body.options?.test ?? false,
+                },
+              });
+
+              if (payloadPacket.data) {
+                if (
+                  payloadPacket.dataType === "application/json" ||
+                  payloadPacket.dataType === "application/super+json"
+                ) {
+                  event.setAttribute("payload", JSON.parse(payloadPacket.data) as any);
+                } else {
+                  event.setAttribute("payload", payloadPacket.data);
+                }
+
+                event.setAttribute("payloadType", payloadPacket.dataType);
+              }
+
+              event.setAttribute("runId", taskRun.friendlyId);
+              span.setAttribute("runId", taskRun.friendlyId);
+
+              if (body.options?.dependentAttempt) {
+                const dependentAttempt = await tx.taskRunAttempt.findUnique({
+                  where: { friendlyId: body.options.dependentAttempt },
+                });
+
+                if (dependentAttempt) {
+                  await tx.taskRunDependency.create({
+                    data: {
+                      taskRunId: taskRun.id,
+                      dependentAttemptId: dependentAttempt.id,
+                    },
+                  });
+                }
+              } else if (body.options?.dependentBatch) {
+                const dependentBatchRun = await tx.batchTaskRun.findUnique({
+                  where: { friendlyId: body.options.dependentBatch },
+                });
+
+                if (dependentBatchRun) {
+                  await tx.taskRunDependency.create({
+                    data: {
+                      taskRunId: taskRun.id,
+                      dependentBatchRunId: dependentBatchRun.id,
+                    },
+                  });
+                }
+              }
+
+              if (body.options?.queue) {
+                const concurrencyLimit = body.options.queue.concurrencyLimit
+                  ? Math.max(0, body.options.queue.concurrencyLimit)
+                  : null;
+                const taskQueue = await prisma.taskQueue.upsert({
                   where: {
-                    projectId_runtimeEnvironmentId_version: {
-                      projectId: environment.projectId,
+                    runtimeEnvironmentId_name: {
                       runtimeEnvironmentId: environment.id,
-                      version: body.options?.lockToVersion,
+                      name: queueName,
                     },
                   },
-                })
-              : undefined;
+                  update: {
+                    concurrencyLimit,
+                    rateLimit: body.options.queue.rateLimit,
+                  },
+                  create: {
+                    friendlyId: generateFriendlyId("queue"),
+                    name: queueName,
+                    concurrencyLimit,
+                    runtimeEnvironmentId: environment.id,
+                    projectId: environment.projectId,
+                    rateLimit: body.options.queue.rateLimit,
+                    type: "NAMED",
+                  },
+                });
 
-            const counter = await tx.taskRunNumberCounter.upsert({
-              where: {
-                taskIdentifier_environmentId: {
-                  taskIdentifier: taskId,
-                  environmentId: environment.id,
+                if (typeof taskQueue.concurrencyLimit === "number") {
+                  await marqs?.updateQueueConcurrencyLimits(
+                    environment,
+                    taskQueue.name,
+                    taskQueue.concurrencyLimit
+                  );
+                } else {
+                  await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
+                }
+              }
+
+              return taskRun;
+            },
+            async (_, tx) => {
+              const counter = await tx.taskRunNumberCounter.findUnique({
+                where: {
+                  taskIdentifier_environmentId: {
+                    taskIdentifier: taskId,
+                    environmentId: environment.id,
+                  },
                 },
-              },
-              update: { lastNumber: { increment: 1 } },
-              create: { taskIdentifier: taskId, environmentId: environment.id, lastNumber: 1 },
-              select: { lastNumber: true },
-            });
-
-            const queueName = body.options?.queue?.name ?? `task/${taskId}`;
-
-            event.setAttribute("queueName", queueName);
-            span.setAttribute("queueName", queueName);
-
-            const taskRun = await tx.taskRun.create({
-              data: {
-                status: "PENDING",
-                number: counter.lastNumber,
-                friendlyId: runFriendlyId,
-                runtimeEnvironmentId: environment.id,
-                projectId: environment.projectId,
-                idempotencyKey,
-                taskIdentifier: taskId,
-                payload: payloadPacket.data ?? "",
-                payloadType: payloadPacket.dataType,
-                context: body.context,
-                traceContext: traceContext,
-                traceId: event.traceId,
-                spanId: event.spanId,
-                lockedToVersionId: lockedToBackgroundWorker?.id,
-                concurrencyKey: body.options?.concurrencyKey,
-                queue: queueName,
-                isTest: body.options?.test ?? false,
-              },
-            });
-
-            if (payloadPacket.data) {
-              if (
-                payloadPacket.dataType === "application/json" ||
-                payloadPacket.dataType === "application/super+json"
-              ) {
-                event.setAttribute("payload", JSON.parse(payloadPacket.data) as any);
-              } else {
-                event.setAttribute("payload", payloadPacket.data);
-              }
-
-              event.setAttribute("payloadType", payloadPacket.dataType);
-            }
-
-            event.setAttribute("runId", taskRun.friendlyId);
-            span.setAttribute("runId", taskRun.friendlyId);
-
-            if (body.options?.dependentAttempt) {
-              const dependentAttempt = await tx.taskRunAttempt.findUnique({
-                where: { friendlyId: body.options.dependentAttempt },
+                select: { lastNumber: true },
               });
 
-              if (dependentAttempt) {
-                await tx.taskRunDependency.create({
-                  data: {
-                    taskRunId: taskRun.id,
-                    dependentAttemptId: dependentAttempt.id,
-                  },
-                });
-              }
-            } else if (body.options?.dependentBatch) {
-              const dependentBatchRun = await tx.batchTaskRun.findUnique({
-                where: { friendlyId: body.options.dependentBatch },
-              });
-
-              if (dependentBatchRun) {
-                await tx.taskRunDependency.create({
-                  data: {
-                    taskRunId: taskRun.id,
-                    dependentBatchRunId: dependentBatchRun.id,
-                  },
-                });
-              }
-            }
-
-            return taskRun;
-          });
+              return counter?.lastNumber;
+            },
+            this._prisma
+          );
 
           if (!run) {
             return;
@@ -245,8 +289,3 @@ export class TriggerTaskService extends BaseService {
     return { dataType: payloadType };
   }
 }
-
-function taskIdentifierToLockId(taskIdentifier: string): number {
-  // Convert taskIdentifier to a unique lock identifier
-  return parseInt(createHash("sha256").update(taskIdentifier).digest("hex").slice(0, 8), 16);
-}

apps/webapp/package.json

@@ -101,7 +101,7 @@
     "@trigger.dev/yalt": "workspace:*",
     "@types/pg": "8.6.6",
     "@uiw/react-codemirror": "^4.19.5",
-    "@upstash/ratelimit": "^1.0.1",
+    "@upstash/ratelimit": "^1.1.3",
     "@whatwg-node/fetch": "^0.9.14",
     "assert-never": "^1.2.1",
     "aws4fetch": "^1.0.18",

apps/webapp/test/marqs.test.ts

@@ -1,316 +0,0 @@
-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: { offset: 3, count: 3 },
-      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: { offset: 6, count: 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: { offset: 0, count: 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: 0, 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("queue1");
-
-    const nextSelection = await stategy.nextCandidateSelection("parentQueue2");
-
-    expect(nextSelection).toEqual({
-      range: { offset: 0, count: 3 },
-      selectionId: expect.any(String),
-    });
-
-    const nextSelection2 = await stategy.nextCandidateSelection("parentQueue");
-
-    expect(nextSelection2).toEqual({
-      range: { offset: 3, count: 3 },
-      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 },
-          },
-        },
-      ],
-      "parentQueue",
-      "selectionId"
-    );
-
-    expect(chosenQueue2).toEqual("queue2");
-
-    const nextSelection3 = await stategy.nextCandidateSelection("parentQueue");
-
-    expect(nextSelection3).toEqual({
-      range: { offset: 6, count: 3 },
-      selectionId: expect.any(String),
-    });
-
-    // Not passed 3 queues, so the range should be reset (we've reached the end)
-    const chosenQueue3 = 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 },
-          },
-        },
-      ],
-      "parentQueue",
-      "selectionId"
-    );
-
-    expect(chosenQueue3).toEqual("queue2");
-
-    const nextSelection4 = await stategy.nextCandidateSelection("parentQueue");
-
-    expect(nextSelection4).toEqual({
-      range: { offset: 0, count: 3 },
-      selectionId: expect.any(String),
-    });
-  });
-});

apps/webapp/test/placeholder.test.ts

@@ -0,0 +1,5 @@
+describe("Placeholder", () => {
+  it("should pass", () => {
+    expect(true).toBe(true);
+  });
+});

docs/mint.json

@@ -1,17 +1,11 @@
 {
   "$schema": "https://mintlify.com/schema.json",
   "name": "Trigger.dev",
-  "openapi": [
-    "/openapi.yml",
-    "/v3-openapi.yaml"
-  ],
-  "versions": [
-    "v3 (Developer Preview)",
-    "v2"
-  ],
+  "openapi": ["/openapi.yml", "/v3-openapi.yaml"],
+  "versions": ["v3 (Developer Preview)", "v2"],
   "api": {
     "playground": {
-      "mode": "hide"
+      "mode": "simple"
     },
     "maintainOrder": true
   },
@@ -102,19 +96,12 @@
     {
       "group": "",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/introduction"
-      ]
+      "pages": ["v3/introduction"]
     },
     {
       "group": "Getting Started",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/quick-start",
-        "v3/upgrading-from-v2",
-        "v3/changelog",
-        "v3/feature-matrix"
-      ]
+      "pages": ["v3/quick-start", "v3/upgrading-from-v2", "v3/changelog", "v3/feature-matrix"]
     },
     {
       "group": "Fundamentals",
@@ -126,10 +113,7 @@
         "v3/apikeys",
         {
           "group": "Task types",
-          "pages": [
-            "v3/tasks-regular",
-            "v3/tasks-scheduled"
-          ]
+          "pages": ["v3/tasks-regular", "v3/tasks-scheduled"]
         },
         "v3/trigger-config"
       ]
@@ -137,10 +121,7 @@
     {
       "group": "Development",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/cli-dev",
-        "v3/run-tests"
-      ]
+      "pages": ["v3/cli-dev", "v3/run-tests"]
     },
     {
       "group": "Deployment",
@@ -151,9 +132,7 @@
         "v3/github-actions",
         {
           "group": "Deployment integrations",
-          "pages": [
-            "v3/vercel-integration"
-          ]
+          "pages": ["v3/vercel-integration"]
         }
       ]
     },
@@ -186,35 +165,43 @@
       "group": "API reference",
       "version": "v3 (Developer Preview)",
       "pages": [
+        "v3/management/overview",
         {
           "group": "Runs API",
           "pages": [
-            "v3/management-retrieve-run",
-            "v3/management-replay-run",
-            "v3/management-cancel-run"
+            "v3/management/runs/list",
+            "v3/management/runs/retrieve",
+            "v3/management/runs/replay",
+            "v3/management/runs/cancel"
           ]
         },
         {
           "group": "Schedules API",
           "pages": [
-            "v3/management-list-schedules",
-            "v3/management-create-schedule",
-            "v3/management-retrieve-schedule",
-            "v3/management-update-schedule",
-            "v3/management-delete-schedule",
-            "v3/management-deactivate-schedule",
-            "v3/management-activate-schedule"
+            "v3/management/schedules/list",
+            "v3/management/schedules/create",
+            "v3/management/schedules/retrieve",
+            "v3/management/schedules/update",
+            "v3/management/schedules/delete",
+            "v3/management/schedules/deactivate",
+            "v3/management/schedules/activate"
           ]
         },
         {
           "group": "Env Vars API",
           "pages": [
-            "v3/management-envvars-list",
-            "v3/management-envvars-import",
-            "v3/management-envvars-create",
-            "v3/management-envvars-retrieve",
-            "v3/management-envvars-update",
-            "v3/management-envvars-delete"
+            "v3/management/envvars/list",
+            "v3/management/envvars/import",
+            "v3/management/envvars/create",
+            "v3/management/envvars/retrieve",
+            "v3/management/envvars/update",
+            "v3/management/envvars/delete"
+          ]
+        },
+        {
+          "group": "Projects API",
+          "pages": [
+            "v3/management/projects/runs"
           ]
         }
       ]
@@ -222,16 +209,13 @@
     {
       "group": "Open source",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/github-repo",
-        "v3/open-source-self-hosting",
-        "v3/open-source-contributing"
-      ]
+      "pages": ["v3/github-repo", "v3/open-source-self-hosting", "v3/open-source-contributing"]
     },
     {
       "group": "Troubleshooting",
       "version": "v3 (Developer Preview)",
       "pages": [
+        "v3/troubleshooting",
         "v3/troubleshooting-alerts",
         "v3/troubleshooting-uptime-status",
         "v3/troubleshooting-github-issues",
@@ -241,11 +225,7 @@
     {
       "group": "Help",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/community",
-        "v3/help-slack",
-        "v3/help-email"
-      ]
+      "pages": ["v3/community", "v3/help-slack", "v3/help-email"]
     },
     {
       "group": "Getting Started",
@@ -437,10 +417,7 @@
       "pages": [
         {
           "group": "Airtable",
-          "pages": [
-            "integrations/apis/airtable",
-            "integrations/apis/airtable-tasks"
-          ]
+          "pages": ["integrations/apis/airtable", "integrations/apis/airtable-tasks"]
         },
         {
           "group": "GitHub",
@@ -466,25 +443,16 @@
         },
         {
           "group": "Plain",
-          "pages": [
-            "integrations/apis/plain",
-            "integrations/apis/plain-tasks"
-          ]
+          "pages": ["integrations/apis/plain", "integrations/apis/plain-tasks"]
         },
         "integrations/apis/replicate",
         {
           "group": "SendGrid",
-          "pages": [
-            "integrations/apis/sendgrid",
-            "integrations/apis/sendgrid-tasks"
-          ]
+          "pages": ["integrations/apis/sendgrid", "integrations/apis/sendgrid-tasks"]
         },
         {
           "group": "Resend",
-          "pages": [
-            "integrations/apis/resend",
-            "integrations/apis/resend-tasks"
-          ]
+          "pages": ["integrations/apis/resend", "integrations/apis/resend-tasks"]
         },
         {
           "group": "Shopify",
@@ -496,10 +464,7 @@
         },
         {
           "group": "Slack",
-          "pages": [
-            "integrations/apis/slack",
-            "integrations/apis/slack-tasks"
-          ]
+          "pages": ["integrations/apis/slack", "integrations/apis/slack-tasks"]
         },
         "integrations/apis/stripe",
         {
@@ -525,9 +490,7 @@
             "sdk/triggerclient/constructor",
             {
               "group": "Instance properties",
-              "pages": [
-                "sdk/triggerclient/store"
-              ]
+              "pages": ["sdk/triggerclient/store"]
             },
             {
               "group": "Instance methods",
@@ -590,10 +553,7 @@
             "sdk/dynamictrigger/constructor",
             {
               "group": "Instance methods",
-              "pages": [
-                "sdk/dynamictrigger/register",
-                "sdk/dynamictrigger/unregister"
-              ]
+              "pages": ["sdk/dynamictrigger/register", "sdk/dynamictrigger/unregister"]
             }
           ]
         },
@@ -604,10 +564,7 @@
             "sdk/dynamicschedule/constructor",
             {
               "group": "Instance methods",
-              "pages": [
-                "sdk/dynamicschedule/register",
-                "sdk/dynamicschedule/unregister"
-              ]
+              "pages": ["sdk/dynamicschedule/register", "sdk/dynamicschedule/unregister"]
             }
           ]
         },
@@ -620,9 +577,7 @@
     {
       "group": "HTTP Reference",
       "version": "v2",
-      "pages": [
-        "sdk/api-reference/events/create-an-event"
-      ]
+      "pages": ["sdk/api-reference/events/create-an-event"]
     },
     {
       "group": "React SDK",
@@ -638,9 +593,7 @@
     {
       "group": "Overview",
       "version": "v2",
-      "pages": [
-        "examples/introduction"
-      ]
+      "pages": ["examples/introduction"]
     }
   ],
   "footerSocials": {
@@ -648,4 +601,4 @@
     "github": "https://github.com/triggerdotdev",
     "linkedin": "https://www.linkedin.com/company/triggerdotdev"
   }
-}
+}

docs/v3-openapi.yaml

@@ -39,7 +39,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -79,7 +79,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -114,7 +114,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -157,7 +157,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -189,7 +189,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -223,7 +223,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -257,7 +257,7 @@ paths:
       tags:
         - schedules
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -265,20 +265,14 @@ paths:
 
             const schedule = await schedules.activate(scheduleId);
 
-  "/api/v1/runs/{run_id}/replay":
+  "/api/v1/runs/{runId}/replay":
+    parameters:
+      - $ref: "#/components/parameters/runId"
     post:
       operationId: replay_run_v1
       summary: Replay a run
       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
@@ -325,9 +319,9 @@ paths:
                     enum:
                       - Run not found
       tags:
-        - run
+        - runs
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -335,19 +329,14 @@ paths:
 
             const handle = await runs.replay("run_1234");
 
-  "/api/v1/runs/{run_id}/cancel":
+  "/api/v2/runs/{runId}/cancel":
+    parameters:
+      - $ref: "#/components/parameters/runId"
     post:
       operationId: cancel_run_v1
-      description: Cancels a run.
       summary: Cancel 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.
+      description: Cancels an in-progress run. If the run is already completed, this
+        will have no effect.
       responses:
         "200":
           description: Successful request
@@ -356,9 +345,10 @@ paths:
               schema:
                 type: object
                 properties:
-                  message:
+                  id:
                     type: string
-                    description: Confirmation message that the run was canceled.
+                    description: The ID of the run that was canceled.
+                    example: run_1234
         "400":
           description: Invalid request
           content:
@@ -394,9 +384,9 @@ paths:
                     enum:
                       - Run not found
       tags:
-        - run
+        - runs
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
@@ -404,19 +394,14 @@ paths:
 
             await runs.cancel("run_1234");
 
-  "/api/v3/runs/{run_id}":
+  "/api/v3/runs/{runId}":
+    parameters:
+      - $ref: "#/components/parameters/runId"
     get:
       operationId: retrieve_run_v1
-      description: Retrieve a run
       summary: Retrieve 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.
+      description: |
+        Retrieve information about a run, including its status, payload, output, and attempts. If you authenticate with a Public API key, we will omit the payload and output fields for security reasons. 
       responses:
         "200":
           description: Successful request
@@ -460,13 +445,182 @@ paths:
       tags:
         - run
       security:
-        - apiKey: []
+        - secretKey: []
       x-codeSamples:
         - lang: typescript
           source: |-
             import { runs } from "@trigger.dev/sdk/v3";
 
-            await runs.retrieve("run_1234");
+            const result = await runs.retrieve("run_1234");
+
+            // We include boolean helpers to check the status of the run
+            // (isSuccess, isFailed, isCompleted, etc.)
+            if (result.isSuccess) {
+              console.log("Run was successful with output", result.output);
+            }
+
+            // You also have access to the run status that includes more granular information
+            console.log("Run status:", result.status);
+
+            // You can access the payload and output
+            console.log("Payload:", result.payload);
+            console.log("Output:", result.output);
+
+            // You can also access the attempts, which will give you information about errors (if they exist)
+            for (const attempt of result.attempts) {
+              if (attempt.status === "FAILED") {
+                console.log("Attempt failed with error:", attempt.error);
+              }
+            }
+
+  "/api/v1/runs":
+    get:
+      operationId: list_runs_v1
+      summary: List runs
+      description: List runs in a specific environment. You can filter the runs by status, created at, task identifier, version, and more.
+      parameters:
+        - $ref: "#/components/parameters/cursorPagination"
+        - $ref: "#/components/parameters/runsFilter"
+      responses:
+        "200":
+          description: Successful request
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ListRunsResult"
+        "400":
+          description: Invalid query parameters
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorWithDetailsResponse"
+        "401":
+          description: Unauthorized request
+      tags:
+        - runs
+      security:
+        - secretKey: []
+      x-codeSamples:
+        - lang: typescript
+          label: List runs
+          source: |-
+            import { runs } from "@trigger.dev/sdk/v3";
+
+            // Get the first page of runs
+            let page = await runs.list({ limit: 20 });
+
+            for (const run of page.data) {
+              console.log(`Run ID: ${run.id}, Status: ${run.status}`);
+            }
+
+            // Convenience methods are provided for manually paginating:
+            while (page.hasNextPage()) {
+              page = await page.getNextPage();
+              // Do something with the next page of runs
+            }
+
+            // Auto-paginate through all runs
+            const allRuns = [];
+
+            for await (const run of runs.list({ limit: 20 })) {
+              allRuns.push(run);
+            }
+        - lang: typescript
+          label: Filter runs
+          source: |-
+            import { runs } from "@trigger.dev/sdk/v3";
+
+            const response = await runs.list({
+              status: ["QUEUED", "EXECUTING"],
+              taskIdentifier: ["my-task", "my-other-task"],
+              from: new Date("2024-04-01T00:00:00Z"),
+              to: new Date(),
+            });
+
+            for (const run of response.data) {
+              console.log(`Run ID: ${run.id}, Status: ${run.status}`);
+            }
+
+  "/api/v1/projects/{projectRef}/runs":
+    parameters:
+      - $ref: "#/components/parameters/projectRef"
+    get:
+      operationId: list_project_runs_v1
+      summary: List project runs
+      description: List runs in a project, across multiple environments, using Personal Access Token auth. You can filter the runs by status, created at, task identifier, version, and more.
+      parameters:
+        - $ref: "#/components/parameters/cursorPagination"
+        - $ref: "#/components/parameters/runsFilterWithEnv"
+      responses:
+        "200":
+          description: Successful request
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ListRunsResult"
+        "400":
+          description: Invalid request
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorWithDetailsResponse"
+        "401":
+          description: Unauthorized request
+      tags:
+        - runs
+      security:
+        - personalAccessToken: []
+      x-codeSamples:
+        - lang: typescript
+          label: List runs
+          source: |-
+            import { runs, configure } from "@trigger.dev/sdk/v3";
+
+            configure({
+              secretKey: "tr_pat_1234" // always use an environment variable for this
+            });
+
+            // Get the first page of runs
+            let page = await runs.list("proj_1234", { limit: 20 });
+
+            for (const run of page.data) {
+              console.log(`Run ID: ${run.id}, Status: ${run.status}`);
+            }
+
+            // Convenience methods are provided for manually paginating:
+            while (page.hasNextPage()) {
+              page = await page.getNextPage();
+              // Do something with the next page of runs
+            }
+
+            // Auto-paginate through all runs
+            const allRuns = [];
+
+            for await (const run of runs.list("proj_1234", { limit: 20 })) {
+              allRuns.push(run);
+            }
+        - lang: typescript
+          label: Filter runs
+          source: |-
+            import { runs, configure } from "@trigger.dev/sdk/v3";
+
+            configure({
+              secretKey: "tr_pat_1234" // always use an environment variable for this
+            });
+
+            const response = await runs.list("proj_1234", {
+              env: ["prod", "staging"],
+              status: ["QUEUED", "EXECUTING"],
+              taskIdentifier: ["my-task", "my-other-task"],
+              from: new Date("2024-04-01T00:00:00Z"),
+              to: new Date(),
+            });
+
+            for (const run of response.data) {
+              console.log(`Run ID: ${run.id}, Status: ${run.status}`);
+            }
+      
+
 
   "/api/v1/projects/{projectRef}/envvars/{env}":
     parameters:
@@ -504,7 +658,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -577,7 +731,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -674,7 +828,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -772,7 +926,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -830,7 +984,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -889,7 +1043,7 @@ paths:
       tags:
         - envvars
       security:
-        - apiKey: []
+        - secretKey: []
         - personalAccessToken: []
       x-codeSamples:
         - lang: typescript
@@ -918,6 +1072,67 @@ paths:
 
 components:
   parameters:
+    runsFilterWithEnv:
+      in: query
+      name: filter
+      style: deepObject
+      explode: true
+      description: |
+        Use this parameter to filter the runs. You can filter by created at, environment, status, task identifier, and version.
+
+        For array fields, you can provide multiple values to filter by using a comma-separated list. For example, to get QUEUED and EXECUTING runs, you can use `filter[status]=QUEUED,EXECUTING`.
+
+        For object fields, you should use the "form" encoding style. For example, to filter by the period, you can use `filter[createdAt][period]=1d`.
+      schema:
+        allOf:
+          - $ref: "#/components/schemas/CommonRunsFilter"
+          - $ref: "#/components/schemas/EnvFilter"
+    runsFilter:
+      in: query
+      name: filter
+      style: deepObject
+      explode: true
+      description: |
+        Use this parameter to filter the runs. You can filter by created at, status, task identifier, and version.
+
+        For array fields, you can provide multiple values to filter by using a comma-separated list. For example, to get QUEUED and EXECUTING runs, you can use `filter[status]=QUEUED,EXECUTING`.
+
+        For object fields, you should use the "form" encoding style. For example, to filter by the period, you can use `filter[createdAt][period]=1d`.
+      schema:
+        $ref: "#/components/schemas/CommonRunsFilter"
+    cursorPagination:
+      in: query
+      name: page
+      style: deepObject
+      explode: true
+      description: |
+        Use this parameter to paginate the results. You can specify the number of runs per page, and the ID of the run to start the page after or before.
+
+        For object fields like `page`, you should use the "form" encoding style. For example, to get the next page of runs, you can use `page[after]=run_1234`.
+      schema:
+        type: object
+        properties:
+          size:
+            type: integer
+            maximum: 100
+            minimum: 10
+            default: 25
+            description: Number of runs per page. Maximum is 100.
+          after:
+            type: string
+            description: The ID of the run to start the page after. This will set the direction of the pagination to forward.
+          before:
+            type: string
+            description: The ID of the run to start the page before. This will set the direction of the pagination to backward.
+    runId:
+      in: path
+      name: runId
+      required: true
+      schema:
+        type: string
+      description: |
+        The ID of an run, starts with `run_`. The run ID will be returned when you trigger a run on a task.
+      example: run_1234
     projectRef:
       in: path
       name: projectRef
@@ -944,7 +1159,7 @@ components:
       description: The name of the environment variable.
       example: SLACK_API_KEY
   securitySchemes:
-    apiKey:
+    secretKey:
       type: http
       scheme: bearer
       description: |
@@ -974,6 +1189,173 @@ components:
         configure({ secretKey: "tr_pat_1234" });
         ```
   schemas:
+    EnvFilter:
+      type: object
+      properties:
+        env:
+          type: array
+          items:
+            type: string
+            description: The environment of the project
+            enum:
+              - dev
+              - staging
+              - prod
+    CommonRunsFilter:
+      type: object
+      properties:
+        createdAt:
+          type: object
+          properties:
+            from:
+              type: string
+              format: date-time
+              description: The start date to filter the runs by
+            to:
+              type: string
+              format: date-time
+              description: The end date to filter the runs by
+            period:
+              type: string
+              description: The period to filter the runs by
+              example: 1d
+        status:
+          type: array
+          items:
+            type: string
+            description: The status of the run
+            enum:
+              - WAITING_FOR_DEPLOY
+              - QUEUED
+              - EXECUTING
+              - REATTEMPTING
+              - FROZEN
+              - COMPLETED
+              - CANCELED
+              - FAILED
+              - CRASHED
+              - INTERRUPTED
+              - SYSTEM_FAILURE
+        taskIdentifier:
+          type: array
+          items:
+            type: string
+            description: The identifier of the task that was run
+        version:
+          type: array
+          items:
+            type: string
+            description: The version of the worker that executed the run
+
+        bulkAction:
+          type: string
+          description: The bulk action ID to filter the runs by
+          example: bulk_1234
+        schedule:
+          type: string
+          description: The schedule ID to filter the runs by
+          example: schedule_1234
+        isTest:
+          type: boolean
+          description: Whether the run is a test run or not
+          example: false
+    ListRunsResult:
+      type: object
+      properties:
+        data:
+          type: array
+          items:
+            "$ref": "#/components/schemas/ListRunItem"
+        pagination:
+          type: object
+          properties:
+            next:
+              type: string
+              description: The run ID to start the next page after. This should be used as the `page[after]` parameter in the next request.
+              example: run_1234
+            previous:
+              type: string
+              description: The run ID to start the previous page before. This should be used as the `page[before]` parameter in the next request.
+              example: run_5678
+    ListRunItem:
+      type: object
+      required:
+        - id
+        - status
+        - taskIdentifier
+        - createdAt
+        - updatedAt
+        - isTest
+        - env
+      properties:
+        id:
+          type: string
+          description: The unique ID of the run, prefixed with `run_`
+          example: run_1234
+        status:
+          type: string
+          description: The status of the run
+          enum:
+            - WAITING_FOR_DEPLOY
+            - QUEUED
+            - EXECUTING
+            - REATTEMPTING
+            - FROZEN
+            - COMPLETED
+            - CANCELED
+            - FAILED
+            - CRASHED
+            - INTERRUPTED
+            - SYSTEM_FAILURE
+        taskIdentifier:
+          type: string
+          description: The identifier of the task that was run
+          example: my-task
+        version:
+          type: string
+          example: 20240523.1
+          description: The version of the worker that executed the run
+        env:
+          type: object
+          description: The environment of the run
+          required:
+            - id
+            - name
+          properties:
+            id:
+              type: string
+              description: The unique ID of the environment
+              example: cl1234
+            name:
+              type: string
+              description: The name of the environment
+              example: dev
+            user:
+              type: string
+              description: If this is a dev environment, the username of the user represented by this environment
+              example: Anna
+        idempotencyKey:
+          type: string
+          description: The idempotency key used to prevent creating duplicate runs, if provided
+          example: idempotency_key_1234
+        isTest:
+          type: boolean
+          description: Whether the run is a test run or not
+          example: false
+        createdAt:
+          type: string
+          format: date-time
+        updatedAt:
+          type: string
+          format: date-time
+        startedAt:
+          type: string
+          format: date-time
+          description: The time the run started
+        finishedAt:
+          type: string
+          format: date-time
+          description: The time the run finished
     InvalidEnvVarsRequestResponse:
       type: object
       properties:
@@ -999,6 +1381,35 @@ components:
         error:
           type: string
       required: ["error"]
+    ErrorWithDetailsResponse:
+      type: object
+      properties:
+        error:
+          type: string
+          example: Query Error
+        details:
+          type: array
+          items:
+            type: object
+            required:
+              - code
+              - message
+            properties:
+              code:
+                type: string
+                description: The error code
+                example: custom
+              message:
+                type: string
+                description: The error message
+                example: "Invalid status values: FOOBAR"
+              path:
+                type: array
+                items:
+                  type: string
+                description: The relevant path in the request
+                example: ["filter[status]"]
+      required: ["error"]
     ListEnvironmentVariablesResponse:
       type: array
       items:
@@ -1032,27 +1443,95 @@ components:
       properties:
         id:
           type: string
+          description: The unique ID of the run, prefixed with `run_`
+          example: run_1234
         status:
           type: string
+          description: The status of the run
           enum:
-            - PENDING
+            - WAITING_FOR_DEPLOY
+            - QUEUED
             - EXECUTING
-            - PAUSED
+            - REATTEMPTING
+            - FROZEN
             - COMPLETED
-            - FAILED
             - CANCELED
+            - FAILED
+            - CRASHED
+            - INTERRUPTED
+            - SYSTEM_FAILURE
         taskIdentifier:
           type: string
-        idempotencyKey:
-          type: string
+          description: The identifier of the task that was run
+          example: my-task
         version:
           type: string
+          example: 20240523.1
+          description: The version of the worker that executed the run
+        payload:
+          type: object
+          description: The payload that was sent to the task. Will be omitted if the request was made with a Public API key
+          example: {"foo": "bar"}
+        output:
+          type: object
+          description: The output of the run. Will be omitted if the request was made with a Public API key
+          example: {"foo": "bar"}
+        idempotencyKey:
+          type: string
+          description: The idempotency key used to prevent creating duplicate runs, if provided
+          example: idempotency_key_1234
         createdAt:
           type: string
           format: date-time
         updatedAt:
           type: string
           format: date-time
+        isTest:
+          type: boolean
+          description: Whether the run is a test run or not
+          example: false
+        startedAt:
+          type: string
+          format: date-time
+          description: The time the run started
+        finishedAt:
+          type: string
+          format: date-time
+          description: The time the run finished
+        schedule:
+          type: object
+          description: The schedule that triggered the run. Will be omitted if the run was not triggered by a schedule
+          required:
+            - id
+            - generator
+          properties:
+            id:
+              type: string
+              description: The unique ID of the schedule, prefixed with `sched_`
+              example: sched_1234
+            externalId:
+              type: string
+              description: The external ID of the schedule. Can be anything that is useful to you (e.g., user ID, org ID, etc.)
+              example: user_1234
+            deduplicationKey:
+              type: string
+              description: The deduplication key used to prevent creating duplicate schedules
+              example: dedup_key_1234
+            generator:
+              type: object
+              properties:
+                type:
+                  type: string
+                  enum:
+                    - CRON
+                expression:
+                  type: string
+                  description: The cron expression used to generate the schedule
+                  example: 0 0 * * *
+                description:
+                  type: string
+                  description: The description of the generator in plain english
+                  example: Every day at midnight
         attempts:
           type: array
           items:
@@ -1065,6 +1544,8 @@ components:
             properties:
               id:
                 type: string
+                description: The unique ID of the attempt, prefixed with `attempt_`
+                example: attempt_1234
               status:
                 type: string
                 enum:
@@ -1074,6 +1555,8 @@ components:
                   - COMPLETED
                   - FAILED
                   - CANCELED
+              error:
+                $ref: "#/components/schemas/SerializedError"
               createdAt:
                 type: string
                 format: date-time
@@ -1174,3 +1657,17 @@ components:
           type: string
         userName:
           type: string
+    SerializedError:
+      type: object
+      required: 
+        - message
+      properties:
+        message:
+          type: string
+          example: Something went wrong
+        name:
+          type: string
+          example: Error
+        stackTrace:
+          type: string
+          example: "Error: Something went wrong"

docs/v3/deploy-environment-variables.mdx

@@ -7,7 +7,9 @@ An environment variable in Node.js is accessed in your code using `process.env.M
 
 We deploy your tasks and scale them up and down when they are triggered. So any environment variables you use in your tasks need to accessible to us so your code will run successfully.
 
-## Setting environment variables
+## In the dashboard
+
+### Setting environment variables
 
 <Steps>
 
@@ -28,7 +30,7 @@ We deploy your tasks and scale them up and down when they are triggered. So any
   locally.
 </Note>
 
-## Editing environment variables
+### Editing environment variables
 
 You can edit an environment variable's values. You cannot edit the key name, you must delete and create a new one.
 
@@ -44,7 +46,7 @@ You can edit an environment variable's values. You cannot edit the key name, you
 
 </Steps>
 
-## Deleting environment variables
+### Deleting environment variables
 
 <Warn>
   Environment variables are fetched and injected before a runs begins. So if you delete one you can
@@ -63,3 +65,113 @@ You can edit an environment variable's values. You cannot edit the key name, you
 </Step>
 
 </Steps>
+
+## In your code
+
+You can use our SDK to get and manipulate environment variables. You can also easily sync environment variables from another service into Trigger.dev.
+
+### Directly manipulating environment variables
+
+We have a complete set of SDK functions (and REST API) you can use to directly manipulate environment variables.
+
+| Function                                              | Description                                                 |
+| ----------------------------------------------------- | ----------------------------------------------------------- |
+| [envvars.list()](/v3/management/envvars/list)         | List all environment variables                              |
+| [envvars.upload()](/v3/management/envvars/import)     | Upload multiple env vars. You can override existing values. |
+| [envvars.create()](/v3/management/envvars/create)     | Create a new environment variable                           |
+| [envvars.retrieve()](/v3/management/envvars/retrieve) | Retrieve an environment variable                            |
+| [envvars.update()](/v3/management/envvars/update)     | Update a single environment variable                        |
+| [envvars.del()](/v3/management/envvars/delete)        | Delete a single environment variable                        |
+
+### Sync env vars from another service
+
+You could use the SDK functions above but it's much easier to use our `resolveEnvVars` function in your `trigger.config` file.
+
+In this example we're using env vars from Infisical.
+
+```ts /trigger.config.ts
+import type { TriggerConfig, ResolveEnvironmentVariablesFunction } from "@trigger.dev/sdk/v3";
+
+//This runs when you run the deploy command or the dev command
+export const resolveEnvVars: ResolveEnvironmentVariablesFunction = async ({
+  //the project ref (starting with "proj_")
+  projectRef,
+  //any existing env vars from a .env file or Trigger.dev
+  env,
+  //"dev", "staging", or "prod"
+  environment,
+}) => {
+  //the existing environment variables from Trigger.dev (or your local .env file)
+  if (env.INFISICAL_CLIENT_ID === undefined || env.INFISICAL_CLIENT_SECRET === undefined) {
+    //returning undefined won't modify the existing env vars
+    return;
+  }
+
+  const client = new InfisicalClient({
+    clientId: env.INFISICAL_CLIENT_ID,
+    clientSecret: env.INFISICAL_CLIENT_SECRET,
+  });
+
+  const secrets = await client.listSecrets({
+    environment,
+    projectId: env.INFISICAL_PROJECT_ID!,
+  });
+
+  return {
+    variables: secrets.map((secret) => ({
+      name: secret.secretKey,
+      value: secret.secretValue,
+    })),
+    // this defaults to true
+    // override: true,
+  };
+};
+
+//the rest of your config file
+export const config: TriggerConfig = {
+  project: "proj_1234567890",
+  //etc
+};
+```
+
+#### Local development
+
+When you [develop locally](/v3/cli-dev) `resolveEnvVars()` will inject the env vars from Infisical into your local `process.env`.
+
+#### Deploy
+
+When you run the [CLI deploy command](/v3/cli-deploy) directly or using [GitHub Actions](/v3/github-actions) it will sync the environment variables from Infisical to Trigger.dev. This means they'll appear on the Environment Variables page so you can confirm that it's worked.
+
+This means that you need to redeploy your Trigger.dev tasks if you change the environment variables in Infisical.
+
+### The variables return type
+
+You can return `variables` as an object with string keys and values, or an array of names + values.
+
+```ts
+return {
+  variables: {
+    MY_ENV_VAR: "my value",
+    MY_OTHER_ENV_VAR: "my other value",
+  },
+};
+```
+
+or
+
+```ts
+return {
+  variables: [
+    {
+      name: "MY_ENV_VAR",
+      value: "my value",
+    },
+    {
+      name: "MY_OTHER_ENV_VAR",
+      value: "my other value",
+    },
+  ],
+};
+```
+
+This should mean that for most secret services you won't need to convert the data into a different format.

docs/v3/github-actions.mdx

@@ -95,15 +95,24 @@ To set it in GitHub go to your repository, click on "Settings", "Secrets and var
 ## Version pinning
 
 The CLI and `@trigger.dev/*` package versions need to be in sync, otherwise there will be errors and unpredictable behavior. Hence, the `deploy` command will automatically fail during CI on any version mismatches.
+Tip: add the deploy command to your `package.json` file to keep versions managed in the same place. For example:
 
-To ensure a smooth CI experience you can pin the CLI version in the deploy step, like so:
+```json
+{
+  "scripts": {
+    "deploy:trigger-prod": "npx trigger.dev@3.0.0-beta.34 deploy",
+    "deploy:trigger": "npx trigger.dev@3.0.0-beta.34 deploy --env staging"
+  }
+}
+```
+Your workflow file will follow the version specified in the `package.json` script, like so:
 
 ```yaml .github/workflows/release-trigger.yml
 - name: 🚀 Deploy Trigger.dev
   env:
     TRIGGER_ACCESS_TOKEN: ${{ secrets.TRIGGER_ACCESS_TOKEN }}
   run: |
-    npx trigger.dev@3.0.0-beta.16 deploy
+    npm run deploy:trigger
 ```
 
 You should use the version you run locally during dev and manual deploy. The current version is displayed in the banner, but you can also check it by appending `--version` to any command.

docs/v3/management-cancel-run.mdx

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

docs/v3/management-get-run.mdx

@@ -1,6 +0,0 @@
----
-title: "Get run"
-description: "Get a run using the Task id."
----
-
-<Snippet file="incomplete-docs.mdx" />

docs/v3/management-get-runs.mdx

@@ -1,6 +0,0 @@
----
-title: "Get runs"
-description: "Get runs using a Task id."
----
-
-<Snippet file="incomplete-docs.mdx" />

docs/v3/management-replay-run.mdx

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