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/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",
@@ -117,6 +119,7 @@
     "strange-sheep-pull",
     "strong-lemons-add",
     "strong-owls-know",
+    "stupid-adults-sniff",
     "stupid-bulldogs-applaud",
     "sweet-lizards-press",
     "swift-dragons-peel",
@@ -136,6 +139,7 @@
     "tricky-ladybugs-unite",
     "two-pumas-wait",
     "warm-olives-provide",
-    "warm-planes-taste"
+    "warm-planes-taste",
+    "young-snails-sell"
   ]
 }

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

@@ -13,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();
@@ -25,6 +26,21 @@ 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";
@@ -54,6 +70,10 @@ 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;
@@ -101,6 +121,7 @@ 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 }) {}
 
@@ -184,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`, {
@@ -192,7 +213,7 @@ class Checkpointer {
       end,
       diff: end - start,
       opts,
-      success: !!result,
+      success: result.success,
     });
 
     if (!result.success) {
@@ -203,7 +224,7 @@ class Checkpointer {
   }
 
   isCheckpointing(runId: string) {
-    return this.#abortControllers.has(runId);
+    return this.#abortControllers.has(runId) || this.#waitingForRetry.has(runId);
   }
 
   cancelCheckpoint(runId: string): boolean {
@@ -214,6 +235,11 @@ class Checkpointer {
       return true;
     }
 
+    if (this.#waitingForRetry.has(runId)) {
+      this.#waitingForRetry.delete(runId);
+      return true;
+    }
+
     const controller = this.#abortControllers.get(runId);
 
     if (!controller) {
@@ -227,14 +253,108 @@ 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<
-    { success: true; checkpoint: CheckpointData } | { success: false; reason?: "CANCELED" }
-  > {
+  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
     await this.initialize();
 
     const options = {
@@ -246,22 +366,47 @@ class Checkpointer {
 
     if (!this.#dockerMode && !this.#canCheckpoint) {
       this.#logger.error("No checkpoint support. Simulation requires docker.");
-      return { success: false };
+      return { success: false, reason: "NO_SUPPORT" };
     }
 
     if (this.#abortControllers.has(runId)) {
       logger.error("Checkpoint procedure already in progress", { options });
-      return { success: false };
+      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();
     this.#abortControllers.set(runId, controller);
 
     const $$ = $({ signal: controller.signal });
 
+    const shortCode = nanoid(8);
+    const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
+    const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
+
+    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");
@@ -279,10 +424,6 @@ class Checkpointer {
         }
       }
 
-      const shortCode = nanoid(8);
-      const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
-      const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
-
       this.#logger.log("Checkpointing:", { options });
 
       const containterName = this.#getRunContainerName(runId);
@@ -294,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}`
@@ -341,6 +489,13 @@ class Checkpointer {
 
       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();
@@ -367,6 +522,13 @@ class Checkpointer {
       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();
@@ -383,17 +545,6 @@ class Checkpointer {
 
       this.#logger.log("Checkpointed and pushed image to:", { location: imageRef, perf });
 
-      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("Failed during checkpoint cleanup", { exportLocation });
-        throw error;
-      }
-
       return {
         success: true,
         checkpoint: {
@@ -409,19 +560,17 @@ class Checkpointer {
           return { success: false, reason: "CANCELED" };
         }
 
-        // Everything that's not a cancellation is a failure
-        this.#failCheckpoint(runId, error);
         this.#logger.error("Checkpoint command error", { options, error });
 
-        return { success: false };
+        return { success: false, reason: "ERROR" };
       }
 
-      this.#failCheckpoint(runId, error);
       this.#logger.error("Unhandled checkpoint error", { options, error });
 
-      return { success: false };
+      return { success: false, reason: "ERROR" };
     } finally {
       this.#abortControllers.delete(runId);
+      await cleanup();
     }
   }
 
@@ -575,6 +724,8 @@ class TaskCoordinator {
             return;
           }
 
+          this.#checkpointer.cancelCheckpoint(message.runId);
+
           if (message.delayInMs) {
             taskSocket.emit("REQUEST_EXIT", {
               version: "v2",
@@ -782,6 +933,14 @@ 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;
           }
         });
 
@@ -822,6 +981,14 @@ class TaskCoordinator {
             });
           } catch (error) {
             logger.error("Error", { error });
+
+            await crashRun({
+              name: "ReadyForLazyAttemptError",
+              message:
+                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            });
+
+            return;
           }
         });
 
@@ -1191,6 +1358,8 @@ class TaskCoordinator {
     // Cancel checkpointing procedure
     const checkpointCanceled = this.#checkpointer.cancelCheckpoint(runId);
 
+    logger.log("cancelCheckpoint()", { runId, checkpointCanceled });
+
     return checkpointCanceled;
   }
 

apps/kubernetes-provider/src/index.ts

@@ -316,6 +316,9 @@ class KubernetesTaskOperations implements TaskOperations {
         {
           name: "registry-trigger",
         },
+        {
+          name: "registry-trigger-failover",
+        },
       ],
       nodeSelector: {
         nodetype: "worker",

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(),
@@ -166,6 +171,20 @@ const EnvironmentSchema = z.object({
   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/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.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/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,12 +16,7 @@ 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 {
@@ -31,6 +26,7 @@ import {
   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";
@@ -38,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];
@@ -96,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,
@@ -107,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) {
@@ -136,6 +156,12 @@ 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`,
@@ -248,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/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,22 +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 { RequeueTaskRunService } from "~/v3/requeueTaskRun.server";
-import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
+import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -166,6 +165,9 @@ const workerCatalog = {
   "v3.retryAttempt": z.object({
     runId: z.string(),
   }),
+  "v2.requeueMessage": z.object({
+    runId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -255,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();
 
@@ -309,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();
 
@@ -335,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();
@@ -361,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();
 
@@ -371,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();
 
@@ -399,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();
@@ -414,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();
@@ -431,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({
@@ -440,7 +440,7 @@ function getWorkerQueue() {
         },
       },
       probeEndpoint: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 1,
         handler: async (payload, job) => {
           const service = new ProbeEndpointService();
@@ -455,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();
@@ -464,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();
@@ -482,7 +482,7 @@ function getWorkerQueue() {
         },
       },
       expireDispatcher: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 3,
         handler: async (payload) => {
           const service = new ExpireDispatcherService();
@@ -626,6 +626,15 @@ function getWorkerQueue() {
           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);
+        },
+      },
     },
   });
 }
@@ -640,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,
@@ -694,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/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

@@ -87,6 +87,7 @@ function createCoordinatorNamespace(io: Server) {
         );
 
         if (!payload) {
+          logger.error("Failed to retrieve execution payload", message);
           return { success: false };
         } else {
           return { success: true, payload };
@@ -106,6 +107,12 @@ function createCoordinatorNamespace(io: Server) {
 
           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 };
         }
       },
@@ -152,7 +159,13 @@ function createCoordinatorNamespace(io: Server) {
 
           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 };
         }
       },
@@ -179,7 +192,10 @@ function createCoordinatorNamespace(io: Server) {
 
           return { success: true, executionPayload: payload };
         } catch (error) {
-          logger.error("Error while creating attempt", { error });
+          logger.error("Error while creating attempt", {
+            runId: message.runId,
+            error,
+          });
           return { success: false };
         }
       },
@@ -188,8 +204,11 @@ function createCoordinatorNamespace(io: Server) {
           const service = new DeploymentIndexFailed();
 
           await service.call(message.deploymentId, message.error);
-        } catch (e) {
-          logger.error("Error while processing index failure", { error: e });
+        } catch (error) {
+          logger.error("Error while processing index failure", {
+            deploymentId: message.deploymentId,
+            error,
+          });
         }
       },
       RUN_CRASHED: async (message) => {
@@ -200,8 +219,11 @@ function createCoordinatorNamespace(io: Server) {
             reason: `${message.error.name}: ${message.error.message}`,
             logs: message.error.stack,
           });
-        } catch (e) {
-          logger.error("Error while processing run failure", { error: e });
+        } catch (error) {
+          logger.error("Error while processing run failure", {
+            runId: message.runId,
+            error,
+          });
         }
       },
     },

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,22 +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";
-import { RequeueTaskRunService } from "../requeueTaskRun.server";
-
-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;
 
@@ -40,6 +46,8 @@ const SemanticAttributes = {
 };
 
 export type MarQSOptions = {
+  name: string;
+  tracer: Tracer;
   redis: RedisOptions;
   defaultEnvConcurrency: number;
   defaultOrgConcurrency: number;
@@ -49,6 +57,9 @@ export type MarQSOptions = {
   keysProducer: MarQSKeyProducer;
   queuePriorityStrategy: MarQSQueuePriorityStrategy;
   envQueuePriorityStrategy: MarQSQueuePriorityStrategy;
+  visibilityTimeoutStrategy: VisibilityTimeoutStrategy;
+  enableRebalancing?: boolean;
+  verbose?: boolean;
 };
 
 /**
@@ -73,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,
@@ -81,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),
@@ -210,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) {
@@ -244,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({
@@ -260,9 +285,9 @@ export class MarQS {
           });
         }
 
-        await RequeueTaskRunService.enqueue(
+        await this.options.visibilityTimeoutStrategy.heartbeat(
           messageData.messageId,
-          new Date(Date.now() + this.visibilityTimeoutInMs)
+          this.visibilityTimeoutInMs
         );
 
         return message;
@@ -279,10 +304,11 @@ export class MarQS {
   }
 
   public async getSharedQueueDetails() {
-    const parentQueue = constants.SHARED_QUEUE;
+    const parentQueue = this.keys.sharedQueueKey();
 
-    const { range, selectionId } = await this.queuePriorityStrategy.nextCandidateSelection(
-      parentQueue
+    const { range } = await this.queuePriorityStrategy.nextCandidateSelection(
+      parentQueue,
+      "getSharedQueueDetails"
     );
     const queues = await this.#getChildQueuesWithScores(parentQueue, range);
 
@@ -294,11 +320,12 @@ export class MarQS {
     const choice = this.queuePriorityStrategy.chooseQueue(
       queuesWithScores,
       parentQueue,
-      selectionId
+      "getSharedQueueDetails",
+      range
     );
 
     return {
-      selectionId,
+      selectionId: "getSharedQueueDetails",
       queues,
       queuesWithScores,
       nextRange: range,
@@ -310,17 +337,18 @@ export class MarQS {
   /**
    * 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) {
@@ -356,6 +384,11 @@ export class MarQS {
           });
         }
 
+        await this.options.visibilityTimeoutStrategy.heartbeat(
+          messageData.messageId,
+          this.visibilityTimeoutInMs
+        );
+
         return message;
       },
       {
@@ -385,7 +418,7 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
-        await RequeueTaskRunService.dequeue(messageId);
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
 
         await this.#callAcknowledgeMessage({
           parentQueue: message.parentQueue,
@@ -452,7 +485,7 @@ export class MarQS {
           return;
         }
 
-        await RequeueTaskRunService.dequeue(messageId);
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
 
         await this.#callAcknowledgeMessage({
           parentQueue: oldMessage.parentQueue,
@@ -483,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) => {
@@ -520,7 +566,11 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
-        await RequeueTaskRunService.dequeue(messageId);
+        if (updates) {
+          await this.replaceMessage(messageId, updates, retryAt, true);
+        }
+
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
 
         await this.#callNackMessage({
           messageKey: this.keys.messageKey(messageId),
@@ -547,15 +597,7 @@ 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) {
-    // We are still calling this for backwards compatibility, but we should be using the v3.requeueTaskRun job
-    await this.#callHeartbeatMessage({
-      visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-      messageId,
-      milliseconds: seconds * 1000,
-      maxVisibilityTimeout: Date.now() + this.visibilityTimeoutInMs,
-    });
-
-    await RequeueTaskRunService.enqueue(messageId, new Date(Date.now() + seconds * 1000));
+    await this.options.visibilityTimeoutStrategy.heartbeat(messageId, seconds * 1000);
   }
 
   get visibilityTimeoutInMs() {
@@ -575,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;
@@ -599,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);
@@ -616,7 +661,8 @@ export class MarQS {
         const choice = this.queuePriorityStrategy.chooseQueue(
           queuesWithScores,
           parentQueue,
-          selectionId
+          consumerId,
+          range
         );
 
         span.setAttributes({
@@ -629,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);
 
@@ -663,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),
         };
       })
     );
@@ -707,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);
@@ -792,6 +865,7 @@ export class MarQS {
         pattern,
         component: "marqs",
         operation: "rebalanceParentQueues",
+        service: this.name,
       });
 
       stream.on("data", async (keys) => {
@@ -803,6 +877,7 @@ export class MarQS {
           component: "marqs",
           operation: "rebalanceParentQueues",
           parentQueues: uniqueKeys,
+          service: this.name,
         });
 
         Promise.all(
@@ -852,6 +927,7 @@ export class MarQS {
           childQueuesWithScores,
           component: "marqs",
           operation: "rebalanceParentQueues",
+          service: this.name,
         });
 
         await Promise.all(
@@ -880,6 +956,7 @@ export class MarQS {
   async #callEnqueueMessage(message: MessagePayload) {
     logger.debug("Calling enqueueMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.enqueueMessage(
@@ -935,6 +1012,7 @@ export class MarQS {
 
     logger.debug("Dequeue message result", {
       result,
+      service: this.name,
     });
 
     if (result.length !== 2) {
@@ -950,6 +1028,7 @@ export class MarQS {
   async #callReplaceMessage(message: MessagePayload) {
     logger.debug("Calling replaceMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.replaceMessage(
@@ -986,6 +1065,7 @@ export class MarQS {
       orgConcurrencyKey,
       messageId,
       parentQueue,
+      service: this.name,
     });
 
     return this.redis.acknowledgeMessage(
@@ -1032,6 +1112,7 @@ export class MarQS {
       visibilityQueue,
       messageId,
       messageScore,
+      service: this.name,
     });
 
     return this.redis.nackMessage(
@@ -1049,28 +1130,6 @@ export class MarQS {
     );
   }
 
-  /**
-   * @deprecated This is being replaced by the v3.requeueTaskRun graphile worker job
-   */
-  #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,
@@ -1154,6 +1213,7 @@ export class MarQS {
         currentScore,
         rebalanceResult,
         operation: "rebalanceParentQueueChild",
+        service: this.name,
       });
     }
 
@@ -1589,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,
@@ -1597,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

@@ -90,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>,
@@ -101,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
@@ -235,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);

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/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/bulk/performBulkAction.server.ts

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

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

@@ -179,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/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/triggerTask.server.ts

@@ -4,12 +4,12 @@ import {
   TriggerTaskRequestBody,
   packetRequiresOffloading,
 } from "@trigger.dev/core/v3";
-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, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { uploadToObjectStore } from "../r2.server";
 import { BaseService } from "./baseService.server";
 
@@ -84,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 = sanitizeQueueName(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;
@@ -244,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/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"
----

docs/v3/management-retrieve-run.mdx

@@ -1,4 +0,0 @@
----
-title: "Retrieve run"
-openapi: "v3-openapi GET /api/v3/runs/{run_id}"
----

docs/v3/management-start-run.mdx

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

docs/v3/management/overview.mdx

@@ -0,0 +1,220 @@
+---
+title: Overview & Authentication
+sidebarTitle: Overview & Authentication
+description: Using the Trigger.dev v3 management API
+---
+
+## Installation
+
+The management API is available through the same `@trigger.dev/sdk` package used in defining and triggering tasks. If you have already installed the package in your project, you can skip this step.
+
+<Note>Make sure you use the `beta` tag when installing, as v3 is still in Developer Preview.</Note>
+
+<CodeGroup>
+
+```bash npm
+npm i @trigger.dev/sdk@beta
+```
+
+```bash pnpm
+pnpm add @trigger.dev/sdk@beta
+```
+
+```bash yarn
+yarn add @trigger.dev/sdk@beta
+```
+
+</CodeGroup>
+
+## Usage
+
+All `v3` functionality is provided through the `@trigger.dev/sdk/v3` module. You can import the entire module or individual resources as needed.
+
+```ts
+import { configure, runs } from "@trigger.dev/sdk/v3";
+
+configure({
+  // this is the default and if the `TRIGGER_SECRET_KEY` environment variable is set, can omit calling configure
+  secretKey: process.env["TRIGGER_SECRET_KEY"],
+});
+
+async function main() {
+  const runs = await runs.list({
+    limit: 10,
+    status: ["COMPLETED"],
+  });
+}
+
+main().catch(console.error);
+```
+
+## Authentication
+
+There are two methods of authenticating with the management API: using a secret key associated with a specific environment in a project (`secretKey`), or using a personal access token (`personalAccessToken`). Both methods should only be used in a backend server, as they provide full access to the project.
+
+<Info>
+  Support for client-side authentication is coming soon to v3 but is not available at the time of
+  writing.
+</Info>
+
+Certain API functions work with both authentication methods, but require different arguments depending on the method used. For example, the `runs.list` function can be called using either a `secretKey` or a `personalAccessToken`, but the `projectRef` argument is required when using a `personalAccessToken`:
+
+```ts
+import { configure, runs } from "@trigger.dev/sdk/v3";
+
+// Using secretKey authentication
+configure({
+  secretKey: process.env["TRIGGER_SECRET_KEY"], // starts with tr_dev_ or tr_prod_
+});
+
+function secretKeyExample() {
+  return runs.list({
+    limit: 10,
+    status: ["COMPLETED"],
+  });
+}
+
+// Using personalAccessToken authentication
+configure({
+  secretKey: process.env["TRIGGER_ACCESS_TOKEN"], // starts with tr_pat_
+});
+
+function personalAccessTokenExample() {
+  // Notice the projectRef argument is required when using a personalAccessToken
+  return runs.list("prof_1234", {
+    limit: 10,
+    status: ["COMPLETED"],
+    projectRef: "tr_proj_1234567890",
+  });
+}
+```
+
+<Accordion title="View endpoint support">
+  Consult the following table to see which endpoints support each authentication method.
+
+| Endpoint               | Secret key | Personal Access Token |
+| ---------------------- | ---------- | --------------------- |
+| `runs.list`            | ✅         | ✅                    |
+| `runs.retrieve`        | ✅         |                       |
+| `runs.cancel`          | ✅         |                       |
+| `runs.replay`          | ✅         |                       |
+| `envvars.list`         | ✅         | ✅                    |
+| `envvars.retrieve`     | ✅         | ✅                    |
+| `envvars.upload`       | ✅         | ✅                    |
+| `envvars.create`       | ✅         | ✅                    |
+| `envvars.update`       | ✅         | ✅                    |
+| `envvars.del`          | ✅         | ✅                    |
+| `schedules.list`       | ✅         |                       |
+| `schedules.create`     | ✅         |                       |
+| `schedules.retrieve`   | ✅         |                       |
+| `schedules.update`     | ✅         |                       |
+| `schedules.activate`   | ✅         |                       |
+| `schedules.deactivate` | ✅         |                       |
+| `schedules.del`        | ✅         |                       |
+
+</Accordion>
+
+### Secret key
+
+Secret key authentication scopes the API access to a specific environment in a project, and works with certain endpoints. You can read our [API Keys guide](/v3/apikeys) for more information.
+
+### Personal Access Token (PAT)
+
+A PAT is a token associated with a specific user, and gives access to all the orgs, projects, and environments that the user has access to. You can identify a PAT by the `tr_pat_` prefix. Because a PAT does not scope access to a specific environment, you must provide the `projectRef` argument when using a PAT (and sometimes the environment as well).
+
+For example, when uploading environment variables using a PAT, you must provide the `projectRef` and `environment` arguments:
+
+```ts
+import { configure, envvars } from "@trigger.dev/sdk/v3";
+
+configure({
+  secretKey: process.env["TRIGGER_ACCESS_TOKEN"], // starts with tr_pat_
+});
+
+await envvars.upload("proj_1234", "dev", {
+  variables: {
+    MY_ENV_VAR: "MY_ENV_VAR_VALUE",
+  },
+  override: true,
+});
+```
+
+## Handling errors
+
+When the SDK method is unable to connect to the API server, or the API server returns a non-successful response, the SDK will throw an `ApiError` that you can catch and handle:
+
+```ts
+import { runs, APIError } from "@trigger.dev/sdk/v3";
+
+async function main() {
+  try {
+    const run = await runs.retrieve("run_1234");
+  } catch (error) {
+    if (error instanceof ApiError) {
+      console.error(`API error: ${error.status}, ${error.headers}, ${error.body}`);
+    } else {
+      console.error(`Unknown error: ${error.message}`);
+    }
+  }
+}
+```
+
+## Auto-pagination
+
+All list endpoints in the management API support auto-pagination.
+You can use `for await … of` syntax to iterate through items across all pages:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+async function fetchAllRuns() {
+  const allRuns = [];
+
+  for await (const run of runs.list({ limit: 10 })) {
+    allRuns.push(run);
+  }
+
+  return allRuns;
+}
+```
+
+You can also use helpers on the return value from any `list` method to get the next/previous page of results:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+async function main() {
+  let page = await runs.list({ limit: 10 });
+
+  for (const run of page.data) {
+    console.log(run);
+  }
+
+  while (page.hasNextPage()) {
+    page = await page.getNextPage();
+    // ... do something with the next page
+  }
+}
+```
+
+## Advanced usage
+
+### Accessing raw HTTP responses
+
+All API methods return a `Promise` subclass `ApiPromise` that includes helpers for accessing the underlying HTTP response:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+async function main() {
+  const { data: run, response: raw } = await runs.retrieve("run_1234").withResponse();
+
+  console.log(raw.status);
+  console.log(raw.headers);
+
+  const response = await runs.retrieve("run_1234").asResponse(); // Returns a Response object
+
+  console.log(response.status);
+  console.log(response.headers);
+}
+```

docs/v3/management/projects/runs.mdx

@@ -0,0 +1,4 @@
+---
+title: "List runs"
+openapi: "v3-openapi GET /api/v1/projects/{projectRef}/runs"
+---

docs/v3/management/runs/cancel.mdx

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

docs/v3/management/runs/list.mdx

@@ -0,0 +1,4 @@
+---
+title: "List runs"
+openapi: "v3-openapi GET /api/v1/runs"
+---

docs/v3/management/runs/replay.mdx

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

docs/v3/management/runs/retrieve.mdx

@@ -0,0 +1,4 @@
+---
+title: "Retrieve run"
+openapi: "v3-openapi GET /api/v3/runs/{runId}"
+---

docs/v3/trigger-config.mdx

@@ -91,6 +91,22 @@ export const config: TriggerConfig = {
 };
 ```
 
+There is a [huge library of instrumentations](https://opentelemetry.io/ecosystem/registry/?language=js) you can easily add to your project like this.
+
+Some ones we recommend:
+
+| Package                                 | Description                                                                                                              |
+| --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ |
+| `@opentelemetry/instrumentation-undici` | Logs all fetch calls (inc. Undici fetch)                                                                                 |
+| `@opentelemetry/instrumentation-fs`     | Logs all file system calls                                                                                               |
+| `@opentelemetry/instrumentation-http`   | Logs all HTTP calls                                                                                                      |
+| `@prisma/instrumentation`               | Logs all Prisma calls, you need to [enable tracing](https://github.com/prisma/prisma/tree/main/packages/instrumentation) |
+| `@traceloop/instrumentation-openai`     | Logs all OpenAI calls                                                                                                    |
+
+## Syncing environment variables
+
+You can sync environment variables from another service using the `resolveEnvVars` function. [Read the docs](/v3/deploy-environment-variables#sync-env-vars-from-another-service) for more information.
+
 ## ESM-only packages
 
 We'll let you know when running the CLI dev command if this is a problem. Some packages are ESM-only so they don't work directly from CJS when using Node.js. In that case you need to add them to the `dependenciesToBundle` array in your `trigger.config.ts` file.
@@ -262,4 +278,4 @@ export const taskThatUsesDecorators = task({
 
 ## Troubleshooting
 
-If you have an issue with bundling let us know on [Discord](https://trigger.dev/discord) or [via email](https://trigger.dev/contact).
+If you have an issue with bundling checkout our [troubleshooting guide](/v3/troubleshooting).

docs/v3/troubleshooting.mdx

@@ -0,0 +1,83 @@
+---
+title: "Common problems"
+description: "Some common problems you might experience and their solutions"
+---
+
+## Deployment
+
+Running the [trigger.dev deploy] command builds and deploys your code. Sometimes there can be issues building your code.
+
+You can run the deploy command with `--log-level debug` at the end. This will spit out a lot of information about the deploy. If you can't figure out the problem from the information below please join [our Discord](https://trigger.dev/discord) and create a help forum post. Do NOT share the extended debug logs publicly as they might reveal private information about your project.
+
+Here are some common problems and their solutions:
+
+### `Typecheck failed, aborting deployment`
+
+We typecheck your code before deploying. If the typecheck fails, the deployment is aborted. You should see logs with details about the typecheck failure.
+
+You can skip typechecking, by adding the `--skip-typecheck` flag when calling deploy.
+
+### `Error: Cannot find module 'X'`
+
+This errors occurs if we can't figure out how to automatically import some code. You can fix this by adding it to the `dependenciesToBundle` array in the [trigger.config file](/v3/trigger-config).
+
+Like this:
+
+```ts trigger.config.ts
+import type { TriggerConfig } from "@trigger.dev/sdk/v3";
+
+export const config: TriggerConfig = {
+  //..other stuff
+  //either regex or strings of package names
+  dependenciesToBundle: [/@sindresorhus/, "escape-string-regexp"],
+};
+```
+
+### `Failed to build project image: Error building image`
+
+There should be a link below the error message to the full build logs on your machine. Take a look at these to see what went wrong. Join [our Discord](https://trigger.dev/discord) and you share it privately with us if you can't figure out what's going wrong. Do NOT share these publicly as the verbose logs might reveal private information about your project.
+
+### `Deployment timed out`
+
+The last stage of deployment is to run it on our servers – we register the new versions of your tasks with the dashboard during this step. We allow 3 mins for this to succeed or fail. If it fails then you'll see this error.
+
+The first thing to do is to try again. If that fails then join [our Discord](https://trigger.dev/discord) and create a Help forum post with a link to your deployment.
+
+### `Deployment encountered an error`
+
+Usually there will be some useful guidance below this message. If you can't figure out what's going wrong then join [our Discord](https://trigger.dev/discord) and create a Help forum post with a link to your deployment.
+
+## Runtime issues
+
+### `Environment variable not found:`
+
+Your code is deployed separately from the rest of your app(s) so you need to make sure that you set any environment variables you use in your tasks in the Trigger.dev dashboard. [Read the guide](/v3/deploy-environment-variables).
+
+### `Error: @prisma/client did not initialize yet.`
+
+Prisma uses code generation to create the client from your schema file. This means you need to add a bit of config so we can generate this file before your tasks run: [read the guide](/v3/trigger-config#prisma-and-other-generators).
+
+### When triggering subtasks the parent task finishes too soon
+
+Make sure that you always use `await` when you call `trigger`, `triggerAndWait`, `batchTrigger`, and `batchTriggerAndWait`. If you don't then it's likely the task(s) won't be triggered because the calling function process can be terminated before the networks calls are sent.
+
+## Framework specific issues
+
+### NestJS swallows all errors/exceptions
+
+If you're using NestJS and you add code like this into your tasks you will prevent any errors from being surfaced:
+
+```ts
+export const simplestTask = task({
+  id: "nestjs-example",
+  run: async (payload) => {
+    //by doing this you're swallowing any errors
+    const app = await NestFactory.createApplicationContext(AppModule);
+    await app.init();
+
+    //etc...
+  },
+});
+```
+
+NestJS has a global exception filter that catches all errors and swallows them, so we can't receive them. Our current recommendation is to not use NestJS inside your tasks. If you're a NestJS user you can still use Trigger.dev but just don't use NestJS inside your tasks like this.

integrations/airtable/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @trigger.dev/airtable
 
+## 3.0.0-beta.36
+
+### Patch Changes
+
+- Updated dependencies [51bb4c887]
+- Updated dependencies [ba71f959e]
+  - @trigger.dev/sdk@3.0.0-beta.36
+  - @trigger.dev/integration-kit@3.0.0-beta.36
+
 ## 3.0.0-beta.35
 
 ### Patch Changes

integrations/airtable/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/airtable",
-  "version": "3.0.0-beta.35",
+  "version": "3.0.0-beta.36",
   "description": "Trigger.dev integration for airtable",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,8 +25,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.35",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.35",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.36",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.36",
     "airtable": "^0.12.1",
     "zod": "3.22.3"
   },

integrations/github/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @trigger.dev/github
 
+## 3.0.0-beta.36
+
+### Patch Changes
+
+- Updated dependencies [51bb4c887]
+- Updated dependencies [ba71f959e]
+  - @trigger.dev/sdk@3.0.0-beta.36
+  - @trigger.dev/integration-kit@3.0.0-beta.36
+
 ## 3.0.0-beta.35
 
 ### Patch Changes

integrations/github/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/github",
-  "version": "3.0.0-beta.35",
+  "version": "3.0.0-beta.36",
   "description": "The official GitHub integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -30,8 +30,8 @@
     "@octokit/request-error": "^5.0.1",
     "@octokit/webhooks": "^12.0.10",
     "octokit": "^3.1.2",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.35",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.35",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.36",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.36",
     "zod": "3.22.3"
   },
   "engines": {

integrations/linear/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @trigger.dev/linear
 
+## 3.0.0-beta.36
+
+### Patch Changes
+
+- Updated dependencies [51bb4c887]
+- Updated dependencies [ba71f959e]
+  - @trigger.dev/sdk@3.0.0-beta.36
+  - @trigger.dev/integration-kit@3.0.0-beta.36
+
 ## 3.0.0-beta.35
 
 ### Patch Changes

integrations/linear/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/linear",
-  "version": "3.0.0-beta.35",
+  "version": "3.0.0-beta.36",
   "description": "Trigger.dev integration for @linear/sdk",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
   },
   "dependencies": {
     "@linear/sdk": "^8.0.0",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.35",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.35",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.36",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.36",
     "zod": "3.22.3"
   },
   "engines": {