Update lockfile

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

.changeset/brown-spies-burn.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+cli v3: increase otel force flush timeout to 30s from 500ms

.changeset/hot-buckets-behave.md

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

.changeset/itchy-chairs-itch.md

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

.changeset/new-pants-beg.md

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

.changeset/plenty-ducks-beam.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Output stderr logs on dev worker failure

.changeset/pre.json

@@ -50,6 +50,7 @@
     "big-tomatoes-deliver",
     "blue-pumas-whisper",
     "breezy-gorillas-mate",
+    "brown-spies-burn",
     "chilled-hornets-move",
     "clean-pianos-listen",
     "clever-apes-collect",
@@ -68,7 +69,9 @@
     "funny-swans-destroy",
     "gorgeous-gorillas-compete",
     "green-bags-wink",
+    "hot-buckets-behave",
     "hot-fishes-retire",
+    "itchy-chairs-itch",
     "khaki-apricots-design",
     "khaki-poems-lay",
     "late-icons-lie",
@@ -85,11 +88,13 @@
     "mighty-camels-joke",
     "mighty-flowers-train",
     "nasty-jars-pump",
+    "new-pants-beg",
     "new-rivers-tell",
     "nice-bulldogs-turn",
     "ninety-pets-travel",
     "odd-poets-own",
     "pink-pumas-rhyme",
+    "plenty-ducks-beam",
     "polite-ducks-switch",
     "polite-rockets-matter",
     "poor-flowers-cross",
@@ -100,6 +105,7 @@
     "rich-kangaroos-unite",
     "rotten-beers-refuse",
     "rotten-dryers-exercise",
+    "rude-toys-compare",
     "selfish-ducks-sort",
     "shaggy-spoons-taste",
     "sharp-emus-compare",
@@ -107,14 +113,18 @@
     "shiny-coats-cry",
     "silly-suits-switch",
     "six-ligers-exist",
+    "sixty-insects-watch",
     "slow-buses-own",
+    "slow-sloths-retire",
     "smart-needles-move",
     "smart-olives-eat",
     "spicy-lamps-smoke",
+    "spicy-terms-bow",
     "strange-ghosts-matter",
     "strange-sheep-pull",
     "strong-lemons-add",
     "strong-owls-know",
+    "stupid-adults-sniff",
     "stupid-bulldogs-applaud",
     "sweet-lizards-press",
     "swift-dragons-peel",
@@ -130,8 +140,11 @@
     "tiny-doors-type",
     "tiny-elephants-scream",
     "tricky-bulldogs-heal",
+    "tricky-keys-attack",
     "tricky-ladybugs-unite",
     "two-pumas-wait",
-    "warm-planes-taste"
+    "warm-olives-provide",
+    "warm-planes-taste",
+    "young-snails-sell"
   ]
 }

.changeset/rude-toys-compare.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Capture and display stderr on index failures

.changeset/sixty-insects-watch.md

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

.changeset/slow-sloths-retire.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/core-apps": patch
+"@trigger.dev/core": patch
+---
+
+- Fix uncaught provider exception
+- Remove unused provider messages

.changeset/spicy-terms-bow.md

@@ -0,0 +1,9 @@
+---
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+---
+
+- Fix init command SDK pinning
+- Show --api-url / -a flag where needed
+- CLI now also respects `TRIGGER_TELEMETRY_DISABLED`
+- Dedicated docker checkpoint test function

.changeset/stupid-adults-sniff.md

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

.changeset/tricky-keys-attack.md

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

.changeset/warm-olives-provide.md

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

.changeset/young-snails-sell.md

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

.env.example

@@ -25,6 +25,8 @@ DEV_OTEL_BATCH_PROCESSING_ENABLED="0"
 # OPTIONAL VARIABLES
 # This is used for validating emails that are allowed to log in. Every email that do not match this regex will be rejected.
 # WHITELISTED_EMAILS="authorized@yahoo\.com|authorized@gmail\.com"
+# Accounts with these emails will get global admin rights. This grants access to the admin UI.
+# ADMIN_EMAILS="admin@example\.com|another-admin@example\.com"
 # This is used for logging in via GitHub. You can leave these commented out if you don't want to use GitHub for authentication.
 # AUTH_GITHUB_CLIENT_ID=
 # AUTH_GITHUB_CLIENT_SECRET=

.github/workflows/publish-docker.yml

@@ -39,11 +39,25 @@ jobs:
             exit 1
           fi
           echo "::set-output name=version::${IMAGE_TAG}"
+
       - name: 🔢 Get the commit hash
         id: get_commit
         run: |
           echo ::set-output name=sha_short::$(echo ${{ github.sha }} | cut -c1-7)
 
+      - name: 📛 Set the tags
+        id: set_tags
+        run: |
+          ref_without_tag=ghcr.io/triggerdotdev/trigger.dev
+          image_tags=$ref_without_tag:${{ steps.get_version.outputs.version }}
+
+          # if it's a versioned tag, also tag it as latest
+          if [[ "${{ github.ref_name }}" == v.docker.* ]]; then
+            image_tags=$image_tags,$ref_without_tag:latest
+          fi
+          
+          echo "IMAGE_TAGS=${image_tags}" >> "$GITHUB_OUTPUT"
+
       - name: 🐙 Login to GitHub Container Registry
         uses: docker/login-action@v2
         with:
@@ -56,6 +70,5 @@ jobs:
         with:
           file: ./docker/Dockerfile
           platforms: linux/amd64,linux/arm64
-          tags: |
-            ghcr.io/triggerdotdev/trigger.dev:${{ steps.get_version.outputs.version }}
+          tags: ${{ steps.set_tags.outputs.IMAGE_TAGS }}
           push: true

.github/workflows/publish-infra.yml

@@ -1,6 +1,7 @@
 name: "🚢 Publish Infra Images"
 
 on:
+  workflow_call:
   push:
     tags:
       - "infra-dev-*"
@@ -29,9 +30,6 @@ permissions:
   packages: write
   contents: read
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
-
 env:
   AWS_REGION: us-east-1
 
@@ -39,7 +37,7 @@ jobs:
   build:
     strategy:
       matrix:
-        package: [coordinator, kubernetes-provider]
+        package: [coordinator, docker-provider, kubernetes-provider]
     runs-on: buildjet-16vcpu-ubuntu-2204
     env:
       DOCKER_BUILDKIT: "1"
@@ -48,20 +46,40 @@ jobs:
 
       - name: Generate image reference
         id: prep
-        # WARNING: This step expects the workflow to have been triggered by a specific tag format of: infra-${env}-*
         run: |
-          env=$(echo ${{ github.ref_name }} | cut -d- -f2)
-          sha=${GITHUB_SHA::7}
-          ts=$(date +%s)
+          # set image repo
           if [[ "${{ matrix.package }}" == *-provider ]]; then
-            provider_type=$(echo ${{ matrix.package }} | cut -d- -f1)
+            provider_type=$(echo "${{ matrix.package }}" | cut -d- -f1)
             repository=provider/${provider_type}
           else
-            repository=${{ matrix.package }}
+            repository="${{ matrix.package }}"
           fi
-          echo "IMAGE_TAG=${env}-${sha}-${ts}" >> "$GITHUB_OUTPUT"
           echo "REPOSITORY=${repository}" >> "$GITHUB_OUTPUT"
 
+          # set image tag
+          if [[ "${{ github.ref_type }}" == "tag" ]]; then
+            if [[ "${{ github.ref_name }}" == infra-*-* ]]; then
+              env=$(echo ${{ github.ref_name }} | cut -d- -f2)
+              sha=$(echo ${{ github.sha }} | head -c7)
+              ts=$(date +%s)
+              image_tag=${env}-${sha}-${ts}
+            elif [[ "${{ github.ref_name }}" == v.docker.* ]]; then
+              version="${GITHUB_REF_NAME#v.docker.}"
+              image_tag="v${version}"
+            elif [[ "${{ github.ref_name }}" == build-* ]]; then
+              image_tag="${GITHUB_REF_NAME#build-}"
+            else
+              echo "Invalid tag: ${{ github.ref_name }}"
+              exit 1
+            fi
+          elif [[ "${{ github.ref_name }}" == "main" ]]; then
+            image_tag="main"
+          else
+            echo "Invalid reference: ${{ github.ref }}"
+            exit 1
+          fi
+          echo "IMAGE_TAG=${image_tag}" >> "$GITHUB_OUTPUT"
+
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
@@ -92,3 +110,12 @@ jobs:
           REGISTRY: ghcr.io/triggerdotdev
           REPOSITORY: ${{ steps.prep.outputs.REPOSITORY }}
           IMAGE_TAG: ${{ steps.prep.outputs.IMAGE_TAG }}
+
+      - name: 🐙 Push 'latest' to GitHub Container Registry
+        if: startsWith(github.ref_name, 'v.docker.')
+        run: |
+          docker tag infra_image $REGISTRY/$REPOSITORY:latest
+          docker push $REGISTRY/$REPOSITORY:latest
+        env:
+          REGISTRY: ghcr.io/triggerdotdev
+          REPOSITORY: ${{ steps.prep.outputs.REPOSITORY }}

.github/workflows/publish.yml

@@ -57,3 +57,8 @@ jobs:
     needs: [typecheck, units]
     uses: ./.github/workflows/publish-docker.yml
     secrets: inherit
+
+  publish-infra:
+    needs: [typecheck, units]
+    uses: ./.github/workflows/publish-infra.yml
+    secrets: inherit

apps/coordinator/Containerfile

@@ -1,19 +1,19 @@
 # syntax=docker/dockerfile:labs
 
-FROM node:18-bullseye-slim@sha256:a4edd54dcfdcacc8a4100fee71498e8671d99556a1acf5614539214a70092426 AS node-18
+FROM node:20-bookworm-slim@sha256:72f2f046a5f8468db28730b990b37de63ce93fd1a72a40f531d6aa82afdf0d46 AS node-20
 
 WORKDIR /app
 
-FROM node-18 AS pruner
+FROM node-20 AS pruner
 
 COPY --chown=node:node . .
 RUN npx -q turbo@1.10.9 prune --scope=coordinator --docker
 RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' +
 
-FROM node-18 AS base
+FROM node-20 AS base
 
 RUN apt-get update \
-  && apt-get install -y buildah ca-certificates dumb-init \
+  && apt-get install -y buildah ca-certificates dumb-init docker.io \
   && rm -rf /var/lib/apt/lists/*
 
 COPY --chown=node:node .gitignore .gitignore

apps/coordinator/src/backoff.ts

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

apps/coordinator/src/index.ts

@@ -1,5 +1,6 @@
 import { createServer } from "node:http";
-import { $ } from "execa";
+import fs from "node:fs/promises";
+import { $, type ExecaChildProcess } from "execa";
 import { nanoid } from "nanoid";
 import { Server } from "socket.io";
 import {
@@ -11,7 +12,8 @@ import {
 } from "@trigger.dev/core/v3";
 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 { HttpReply, getTextBody, SimpleLogger, testDockerCheckpoint } from "@trigger.dev/core-apps";
+import { ExponentialBackoff } from "./backoff";
 
 import { collectDefaultMetrics, register, Gauge } from "prom-client";
 collectDefaultMetrics();
@@ -19,8 +21,29 @@ collectDefaultMetrics();
 const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || 8020);
 const NODE_NAME = process.env.NODE_NAME || "coordinator";
 const DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS = 30_000;
+const CHAOS_MONKEY_ENABLED = !!process.env.CHAOS_MONKEY_ENABLED;
+
+const FORCE_CHECKPOINT_SIMULATION = ["1", "true"].includes(
+  process.env.FORCE_CHECKPOINT_SIMULATION ?? "true"
+);
+const DISABLE_CHECKPOINT_SUPPORT = ["1", "true"].includes(
+  process.env.DISABLE_CHECKPOINT_SUPPORT ?? "false"
+);
+const SIMULATE_PUSH_FAILURE = ["1", "true"].includes(process.env.SIMULATE_PUSH_FAILURE ?? "false");
+const SIMULATE_PUSH_FAILURE_SECONDS = parseInt(
+  process.env.SIMULATE_PUSH_FAILURE_SECONDS ?? "300",
+  10
+);
+const SIMULATE_CHECKPOINT_FAILURE = ["1", "true"].includes(
+  process.env.SIMULATE_CHECKPOINT_FAILURE ?? "false"
+);
+const SIMULATE_CHECKPOINT_FAILURE_SECONDS = parseInt(
+  process.env.SIMULATE_CHECKPOINT_FAILURE_SECONDS ?? "300",
+  10
+);
 
 const REGISTRY_HOST = process.env.REGISTRY_HOST || "localhost:5000";
+const REGISTRY_NAMESPACE = process.env.REGISTRY_NAMESPACE || "trigger";
 const CHECKPOINT_PATH = process.env.CHECKPOINT_PATH || "/checkpoints";
 const REGISTRY_TLS_VERIFY = process.env.REGISTRY_TLS_VERIFY === "false" ? "false" : "true";
 
@@ -32,6 +55,10 @@ const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?
 
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 
+if (CHAOS_MONKEY_ENABLED) {
+  logger.log("🍌 Chaos monkey enabled");
+}
+
 type CheckpointerInitializeReturn = {
   canCheckpoint: boolean;
   willSimulate: boolean;
@@ -44,11 +71,52 @@ type CheckpointAndPushOptions = {
   deploymentVersion: string;
 };
 
+type CheckpointAndPushResult =
+  | { success: true; checkpoint: CheckpointData }
+  | {
+      success: false;
+      reason?: "CANCELED" | "DISABLED" | "ERROR" | "IN_PROGRESS" | "NO_SUPPORT" | "SKIP_RETRYING";
+    };
+
 type CheckpointData = {
   location: string;
   docker: boolean;
 };
 
+function isExecaChildProcess(maybeExeca: unknown): maybeExeca is Awaited<ExecaChildProcess> {
+  return typeof maybeExeca === "object" && maybeExeca !== null && "escapedCommand" in maybeExeca;
+}
+
+async function getFileSize(filePath: string): Promise<number> {
+  try {
+    const stats = await fs.stat(filePath);
+    return stats.size;
+  } catch (error) {
+    console.error("Error getting file size:", error);
+    return -1;
+  }
+}
+
+async function getParsedFileSize(filePath: string) {
+  const sizeInBytes = await getFileSize(filePath);
+
+  let message = `Size in bytes: ${sizeInBytes}`;
+
+  if (sizeInBytes > 1024 * 1024) {
+    const sizeInMB = (sizeInBytes / 1024 / 1024).toFixed(2);
+    message = `Size in MB (rounded): ${sizeInMB}`;
+  } else if (sizeInBytes > 1024) {
+    const sizeInKB = (sizeInBytes / 1024).toFixed(2);
+    message = `Size in KB (rounded): ${sizeInKB}`;
+  }
+
+  return {
+    path: filePath,
+    sizeInBytes,
+    message,
+  };
+}
+
 class Checkpointer {
   #initialized = false;
   #canCheckpoint = false;
@@ -56,73 +124,63 @@ 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 }) {}
 
-  async initialize(): Promise<CheckpointerInitializeReturn> {
+  async init(): Promise<CheckpointerInitializeReturn> {
     if (this.#initialized) {
-      return this.#getInitializeReturn();
+      return this.#getInitReturn(this.#canCheckpoint);
     }
 
     this.#logger.log(`${this.#dockerMode ? "Docker" : "Kubernetes"} mode`);
 
     if (this.#dockerMode) {
-      try {
-        await $`criu --version`;
-      } catch (error) {
-        this.#logger.error("No checkpoint support: Missing CRIU binary");
-        this.#logger.error("Will simulate instead");
-        this.#canCheckpoint = false;
-        this.#initialized = true;
+      const testCheckpoint = await testDockerCheckpoint();
 
-        return this.#getInitializeReturn();
+      if (testCheckpoint.ok) {
+        return this.#getInitReturn(true);
       }
 
-      try {
-        await $`docker checkpoint`;
-      } catch (error) {
-        this.#logger.error(
-          "No checkpoint support: Docker needs to have experimental features enabled"
-        );
-        this.#logger.error("Will simulate instead");
-        this.#canCheckpoint = false;
-        this.#initialized = true;
-
-        return this.#getInitializeReturn();
-      }
+      this.#logger.error(testCheckpoint.message, testCheckpoint.error ?? "");
+      return this.#getInitReturn(false);
     } else {
       try {
         await $`buildah login --get-login ${REGISTRY_HOST}`;
       } catch (error) {
         this.#logger.error(`No checkpoint support: Not logged in to registry ${REGISTRY_HOST}`);
-        this.#canCheckpoint = false;
-        this.#initialized = true;
-
-        return this.#getInitializeReturn();
+        return this.#getInitReturn(false);
       }
     }
 
-    this.#logger.log(
-      `Full checkpoint support${
-        this.#dockerMode && this.opts.forceSimulate ? " with forced simulation enabled." : "!"
-      }`
-    );
-
-    this.#initialized = true;
-    this.#canCheckpoint = true;
-
-    return this.#getInitializeReturn();
+    return this.#getInitReturn(true);
   }
 
-  #getInitializeReturn(): CheckpointerInitializeReturn {
+  #getInitReturn(canCheckpoint: boolean): CheckpointerInitializeReturn {
+    this.#initialized = true;
+    this.#canCheckpoint = canCheckpoint;
+
+    if (canCheckpoint) {
+      this.#logger.log("Full checkpoint support!");
+    }
+
+    const willSimulate = this.#dockerMode && (!this.#canCheckpoint || this.opts.forceSimulate);
+
+    if (willSimulate) {
+      this.#logger.log("Simulation mode enabled. Containers will be paused, not checkpointed.", {
+        forceSimulate: this.opts.forceSimulate,
+      });
+    }
+
     return {
-      canCheckpoint: this.#canCheckpoint,
-      willSimulate: this.#dockerMode && (!this.#canCheckpoint || this.opts.forceSimulate),
+      canCheckpoint,
+      willSimulate,
     };
   }
 
   #getImageRef(projectRef: string, deploymentVersion: string, shortCode: string) {
-    return `${REGISTRY_HOST}/trigger/${projectRef}:${deploymentVersion}.prod-${shortCode}`;
+    return `${REGISTRY_HOST}/${REGISTRY_NAMESPACE}/${projectRef}:${deploymentVersion}.prod-${shortCode}`;
   }
 
   #getExportLocation(projectRef: string, deploymentVersion: string, shortCode: string) {
@@ -139,7 +197,7 @@ class Checkpointer {
     const start = performance.now();
     logger.log(`checkpointAndPush() start`, { start, opts });
 
-    const result = await this.#checkpointAndPush(opts);
+    const result = await this.#checkpointAndPushWithBackoff(opts);
 
     const end = performance.now();
     logger.log(`checkpointAndPush() end`, {
@@ -147,17 +205,33 @@ class Checkpointer {
       end,
       diff: end - start,
       opts,
-      success: !!result,
+      success: result.success,
     });
 
-    return result;
+    if (!result.success) {
+      return;
+    }
+
+    return result.checkpoint;
   }
 
   isCheckpointing(runId: string) {
-    return this.#abortControllers.has(runId);
+    return this.#abortControllers.has(runId) || this.#waitingForRetry.has(runId);
   }
 
   cancelCheckpoint(runId: string): boolean {
+    // If the last checkpoint failed, pretend we canceled it
+    // This ensures tasks don't wait for external resume messages to continue
+    if (this.#hasFailedCheckpoint(runId)) {
+      this.#clearFailedCheckpoint(runId);
+      return true;
+    }
+
+    if (this.#waitingForRetry.has(runId)) {
+      this.#waitingForRetry.delete(runId);
+      return true;
+    }
+
     const controller = this.#abortControllers.get(runId);
 
     if (!controller) {
@@ -171,29 +245,138 @@ 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;
+        }
+
+        if (result.reason === "SKIP_RETRYING") {
+          this.#logger.log("Skipping retrying", { runId });
+          return result;
+        }
+
+        continue;
+      } catch (error) {
+        this.#logger.error("Checkpoint error", {
+          retry,
+          runId,
+          delay,
+          error: error instanceof Error ? error.message : error,
+        });
+      }
+    }
+
+    this.#logger.error(`Checkpoint failed after exponential backoff`, {
+      runId,
+      leaveRunning,
+      projectRef,
+      deploymentVersion,
+    });
+    this.#failCheckpoint(runId, "ERROR");
+
+    return { success: false, reason: "ERROR" };
+  }
+
   async #checkpointAndPush({
     runId,
     leaveRunning = true, // This mirrors kubernetes behaviour more accurately
     projectRef,
     deploymentVersion,
-  }: CheckpointAndPushOptions): Promise<CheckpointData | undefined> {
-    await this.initialize();
+  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
+    await this.init();
+
+    const options = {
+      runId,
+      leaveRunning,
+      projectRef,
+      deploymentVersion,
+    };
 
     if (!this.#dockerMode && !this.#canCheckpoint) {
       this.#logger.error("No checkpoint support. Simulation requires docker.");
-      return;
+      return { success: false, reason: "NO_SUPPORT" };
     }
 
     if (this.#abortControllers.has(runId)) {
-      logger.error("Checkpoint procedure already in progress", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-      });
-      return;
+      logger.error("Checkpoint procedure already in progress", { options });
+      return { success: false, reason: "IN_PROGRESS" };
+    }
+
+    // This is a new checkpoint, clear any last failure for this run
+    this.#clearFailedCheckpoint(runId);
+
+    if (DISABLE_CHECKPOINT_SUPPORT) {
+      this.#logger.error("Checkpoint support disabled", { options });
+      return { success: false, reason: "DISABLED" };
     }
 
     const controller = new AbortController();
@@ -201,19 +384,44 @@ class Checkpointer {
 
     const $$ = $({ signal: controller.signal });
 
-    try {
-      const shortCode = nanoid(8);
-      const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
-      const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
+    const shortCode = nanoid(8);
+    const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
+    const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
 
-      this.#logger.log("Checkpointing:", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-      });
+    const cleanup = async () => {
+      if (this.#dockerMode) {
+        return;
+      }
+
+      try {
+        await $`rm ${exportLocation}`;
+        this.#logger.log("Deleted checkpoint archive", { exportLocation });
+
+        await $`buildah rmi ${imageRef}`;
+        this.#logger.log("Deleted checkpoint image", { imageRef });
+      } catch (error) {
+        this.#logger.error("Failure during checkpoint cleanup", { exportLocation, error });
+      }
+    };
+
+    try {
+      if (CHAOS_MONKEY_ENABLED) {
+        console.log("🍌 Chaos monkey wreaking havoc");
+
+        const random = Math.random();
+
+        if (random < 0.33) {
+          // Fake long checkpoint duration
+          await $$`sleep 300`;
+        } else if (random < 0.66) {
+          // Fake checkpoint error
+          await $$`false`;
+        } else {
+          // no-op
+        }
+      }
+
+      this.#logger.log("Checkpointing:", { options });
 
       const containterName = this.#getRunContainerName(runId);
 
@@ -224,6 +432,13 @@ class Checkpointer {
             this.#logger.log("Simulating checkpoint");
             this.#logger.debug(await $$`docker pause ${containterName}`);
           } else {
+            if (SIMULATE_CHECKPOINT_FAILURE) {
+              if (performance.now() < SIMULATE_CHECKPOINT_FAILURE_SECONDS * 1000) {
+                this.#logger.error("Simulating checkpoint failure", { options });
+                throw new Error("SIMULATE_CHECKPOINT_FAILURE");
+              }
+            }
+
             if (leaveRunning) {
               this.#logger.debug(
                 await $$`docker checkpoint create --leave-running ${containterName} ${exportLocation}`
@@ -234,9 +449,9 @@ class Checkpointer {
               );
             }
           }
-        } catch (error: any) {
-          this.#logger.error(error.stderr);
-          return;
+        } catch (error) {
+          this.#logger.error("Failed while creating docker checkpoint", { exportLocation });
+          throw error;
         }
 
         this.#logger.log("checkpoint created:", {
@@ -245,14 +460,18 @@ class Checkpointer {
         });
 
         return {
-          location: exportLocation,
-          docker: true,
+          success: true,
+          checkpoint: {
+            location: exportLocation,
+            docker: true,
+          },
         };
       }
 
       // Create checkpoint (CRI)
       if (!this.#canCheckpoint) {
-        throw new Error("No checkpoint support in kubernetes mode.");
+        this.#logger.error("No checkpoint support in kubernetes mode.");
+        return { success: false, reason: "SKIP_RETRYING" };
       }
 
       const containerId = this.#logger.debug(
@@ -263,57 +482,109 @@ class Checkpointer {
       );
 
       if (!containerId.stdout) {
-        throw new Error("could not find container id");
+        this.#logger.error("could not find container id", { options, containterName });
+        return { success: false, reason: "SKIP_RETRYING" };
       }
 
+      const start = performance.now();
+
+      if (SIMULATE_CHECKPOINT_FAILURE) {
+        if (performance.now() < SIMULATE_CHECKPOINT_FAILURE_SECONDS * 1000) {
+          this.#logger.error("Simulating checkpoint failure", { options });
+          throw new Error("SIMULATE_CHECKPOINT_FAILURE");
+        }
+      }
+
+      // Create checkpoint
       this.#logger.debug(await $$`crictl checkpoint --export=${exportLocation} ${containerId}`);
+      const postCheckpoint = performance.now();
+
+      // Print checkpoint size
+      const size = await getParsedFileSize(exportLocation);
+      this.#logger.log("checkpoint archive created", { size, options });
 
       // Create image from checkpoint
       const container = this.#logger.debug(await $$`buildah from scratch`);
+      const postFrom = performance.now();
+
       this.#logger.debug(await $$`buildah add ${container} ${exportLocation} /`);
+      const postAdd = performance.now();
+
       this.#logger.debug(
         await $$`buildah config --annotation=io.kubernetes.cri-o.annotations.checkpoint.name=counter ${container}`
       );
+      const postConfig = performance.now();
+
       this.#logger.debug(await $$`buildah commit ${container} ${imageRef}`);
+      const postCommit = performance.now();
+
       this.#logger.debug(await $$`buildah rm ${container}`);
+      const postRm = performance.now();
+
+      if (SIMULATE_PUSH_FAILURE) {
+        if (performance.now() < SIMULATE_PUSH_FAILURE_SECONDS * 1000) {
+          this.#logger.error("Simulating push failure", { options });
+          throw new Error("SIMULATE_PUSH_FAILURE");
+        }
+      }
 
       // Push checkpoint image
       this.#logger.debug(await $$`buildah push --tls-verify=${REGISTRY_TLS_VERIFY} ${imageRef}`);
+      const postPush = performance.now();
 
-      this.#logger.log("Checkpointed and pushed image to:", { location: imageRef });
+      const perf = {
+        "crictl checkpoint": postCheckpoint - start,
+        "buildah from": postFrom - postCheckpoint,
+        "buildah add": postAdd - postFrom,
+        "buildah config": postConfig - postAdd,
+        "buildah commit": postCommit - postConfig,
+        "buildah rm": postRm - postCommit,
+        "buildah push": postPush - postRm,
+      };
 
-      try {
-        await $$`rm ${exportLocation}`;
-        this.#logger.log("Deleted checkpoint archive", { exportLocation });
-
-        // Disabled for now as this will increase restore time by having to pull the image again
-        // await $`buildah rmi ${imageRef}`;
-        // this.#logger.log("Deleted checkpoint image", { imageRef });
-      } catch (error) {
-        this.#logger.error("Failed during checkpoint cleanup", { exportLocation });
-        this.#logger.debug(error);
-      }
+      this.#logger.log("Checkpointed and pushed image to:", { location: imageRef, perf });
 
       return {
-        location: imageRef,
-        docker: false,
+        success: true,
+        checkpoint: {
+          location: imageRef,
+          docker: false,
+        },
       };
     } catch (error) {
-      this.#logger.error("checkpoint failed", {
-        options: {
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-        },
-        error,
-      });
-      return;
+      if (isExecaChildProcess(error)) {
+        if (error.isCanceled) {
+          this.#logger.error("Checkpoint canceled", { options, error });
+
+          return { success: false, reason: "CANCELED" };
+        }
+
+        this.#logger.error("Checkpoint command error", { options, error });
+
+        return { success: false, reason: "ERROR" };
+      }
+
+      this.#logger.error("Unhandled checkpoint error", { options, error });
+
+      return { success: false, reason: "ERROR" };
     } finally {
       this.#abortControllers.delete(runId);
+      await cleanup();
     }
   }
 
+  #failCheckpoint(runId: string, error: unknown) {
+    this.#failedCheckpoints.set(runId, error);
+  }
+
+  #clearFailedCheckpoint(runId: string) {
+    this.#failedCheckpoints.delete(runId);
+  }
+
+  #hasFailedCheckpoint(runId: string) {
+    return this.#failedCheckpoints.has(runId);
+  }
+
   #getRunContainerName(suffix: string) {
     return `task-run-${suffix}`;
   }
@@ -321,7 +592,7 @@ class Checkpointer {
 
 class TaskCoordinator {
   #httpServer: ReturnType<typeof createServer>;
-  #checkpointer = new Checkpointer({ forceSimulate: true });
+  #checkpointer = new Checkpointer({ forceSimulate: FORCE_CHECKPOINT_SIMULATION });
 
   #prodWorkerNamespace: ZodNamespace<
     typeof ProdWorkerToCoordinatorMessages,
@@ -345,7 +616,7 @@ class TaskCoordinator {
     private host = "0.0.0.0"
   ) {
     this.#httpServer = this.#createHttpServer();
-    this.#checkpointer.initialize();
+    this.#checkpointer.init();
     this.#delayThresholdInMs = this.#getDelayThreshold();
 
     if (process.env.DELAY_THRESHOLD_IN_MS) {
@@ -442,6 +713,30 @@ class TaskCoordinator {
 
           taskSocket.emit("REQUEST_ATTEMPT_CANCELLATION", message);
         },
+        REQUEST_RUN_CANCELLATION: async (message) => {
+          const taskSocket = await this.#getRunSocket(message.runId);
+
+          if (!taskSocket) {
+            logger.log("Socket for run not found", {
+              runId: message.runId,
+            });
+            return;
+          }
+
+          this.#checkpointer.cancelCheckpoint(message.runId);
+
+          if (message.delayInMs) {
+            taskSocket.emit("REQUEST_EXIT", {
+              version: "v2",
+              delayInMs: message.delayInMs,
+            });
+          } else {
+            // If there's no delay, assume the worker doesn't support non-v1 messages
+            taskSocket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+          }
+        },
         READY_FOR_RETRY: async (message) => {
           const taskSocket = await this.#getRunSocket(message.runId);
 
@@ -528,6 +823,20 @@ class TaskCoordinator {
       onConnection: async (socket, handler, sender) => {
         const logger = new SimpleLogger(`[prod-worker][${socket.id}]`);
 
+        const crashRun = async (error: { name: string; message: string; stack?: string }) => {
+          try {
+            this.#platformSocket?.send("RUN_CRASHED", {
+              version: "v1",
+              runId: socket.data.runId,
+              error,
+            });
+          } finally {
+            socket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+          }
+        };
+
         const checkpointInProgress = () => {
           return this.#checkpointableTasks.has(socket.data.runId);
         };
@@ -596,8 +905,9 @@ class TaskCoordinator {
             if (!executionAck) {
               logger.error("no execution ack", { runId: socket.data.runId });
 
-              socket.emit("REQUEST_EXIT", {
-                version: "v1",
+              await crashRun({
+                name: "ReadyForExecutionError",
+                message: "No execution ack",
               });
 
               return;
@@ -606,8 +916,9 @@ class TaskCoordinator {
             if (!executionAck.success) {
               logger.error("failed to get execution payload", { runId: socket.data.runId });
 
-              socket.emit("REQUEST_EXIT", {
-                version: "v1",
+              await crashRun({
+                name: "ReadyForExecutionError",
+                message: "Failed to get execution payload",
               });
 
               return;
@@ -621,6 +932,62 @@ class TaskCoordinator {
             socket.data.attemptFriendlyId = executionAck.payload.execution.attempt.id;
           } catch (error) {
             logger.error("Error", { error });
+
+            await crashRun({
+              name: "ReadyForExecutionError",
+              message:
+                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            });
+
+            return;
+          }
+        });
+
+        socket.on("READY_FOR_LAZY_ATTEMPT", async (message) => {
+          logger.log("[READY_FOR_LAZY_ATTEMPT]", message);
+
+          try {
+            const lazyAttempt = await this.#platformSocket?.sendWithAck("READY_FOR_LAZY_ATTEMPT", {
+              ...message,
+              envId: socket.data.envId,
+            });
+
+            if (!lazyAttempt) {
+              logger.error("no lazy attempt ack", { runId: socket.data.runId });
+
+              await crashRun({
+                name: "ReadyForLazyAttemptError",
+                message: "No lazy attempt ack",
+              });
+
+              return;
+            }
+
+            if (!lazyAttempt.success) {
+              logger.error("failed to get lazy attempt payload", { runId: socket.data.runId });
+
+              await crashRun({
+                name: "ReadyForLazyAttemptError",
+                message: "Failed to get lazy attempt payload",
+              });
+
+              return;
+            }
+
+            socket.emit("EXECUTE_TASK_RUN_LAZY_ATTEMPT", {
+              version: "v1",
+              lazyPayload: lazyAttempt.lazyPayload,
+            });
+          } catch (error) {
+            logger.error("Error", { error });
+
+            await crashRun({
+              name: "ReadyForLazyAttemptError",
+              message:
+                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            });
+
+            return;
           }
         });
 
@@ -666,7 +1033,7 @@ class TaskCoordinator {
             return;
           }
 
-          const { canCheckpoint, willSimulate } = await this.#checkpointer.initialize();
+          const { canCheckpoint, willSimulate } = await this.#checkpointer.init();
 
           const willCheckpointAndRestore = canCheckpoint || willSimulate;
 
@@ -714,6 +1081,19 @@ class TaskCoordinator {
           }
         });
 
+        socket.on("TASK_RUN_FAILED_TO_RUN", async ({ completion }) => {
+          logger.log("completed task", { completionId: completion.id });
+
+          this.#platformSocket?.send("TASK_RUN_FAILED_TO_RUN", {
+            version: "v1",
+            completion,
+          });
+
+          socket.emit("REQUEST_EXIT", {
+            version: "v1",
+          });
+        });
+
         socket.on("READY_FOR_CHECKPOINT", async (message) => {
           logger.log("[READY_FOR_CHECKPOINT]", message);
 
@@ -750,7 +1130,7 @@ class TaskCoordinator {
             return;
           }
 
-          const { canCheckpoint, willSimulate } = await this.#checkpointer.initialize();
+          const { canCheckpoint, willSimulate } = await this.#checkpointer.init();
 
           const willCheckpointAndRestore = canCheckpoint || willSimulate;
 
@@ -804,7 +1184,7 @@ class TaskCoordinator {
         socket.on("WAIT_FOR_TASK", async (message, callback) => {
           logger.log("[WAIT_FOR_TASK]", message);
 
-          const { canCheckpoint, willSimulate } = await this.#checkpointer.initialize();
+          const { canCheckpoint, willSimulate } = await this.#checkpointer.init();
 
           const willCheckpointAndRestore = canCheckpoint || willSimulate;
 
@@ -846,7 +1226,7 @@ class TaskCoordinator {
         socket.on("WAIT_FOR_BATCH", async (message, callback) => {
           logger.log("[WAIT_FOR_BATCH]", message);
 
-          const { canCheckpoint, willSimulate } = await this.#checkpointer.initialize();
+          const { canCheckpoint, willSimulate } = await this.#checkpointer.init();
 
           const willCheckpointAndRestore = canCheckpoint || willSimulate;
 
@@ -890,7 +1270,7 @@ class TaskCoordinator {
           logger.log("[INDEX_TASKS]", message);
 
           const workerAck = await this.#platformSocket?.sendWithAck("CREATE_WORKER", {
-            version: "v1",
+            version: "v2",
             projectRef: socket.data.projectRef,
             envId: socket.data.envId,
             deploymentId: message.deploymentId,
@@ -899,6 +1279,7 @@ class TaskCoordinator {
               packageVersion: message.packageVersion,
               tasks: message.tasks,
             },
+            supportsLazyAttempts: message.version !== "v1" && message.supportsLazyAttempts,
           });
 
           if (!workerAck) {
@@ -917,6 +1298,34 @@ class TaskCoordinator {
             error: message.error,
           });
         });
+
+        socket.on("CREATE_TASK_RUN_ATTEMPT", async (message, callback) => {
+          logger.log("[CREATE_TASK_RUN_ATTEMPT]", message);
+
+          const createAttempt = await this.#platformSocket?.sendWithAck("CREATE_TASK_RUN_ATTEMPT", {
+            runId: message.runId,
+            envId: socket.data.envId,
+          });
+
+          if (!createAttempt?.success) {
+            logger.debug("no ack while creating attempt", message);
+            callback({ success: false });
+            return;
+          }
+
+          socket.data.attemptFriendlyId = createAttempt.executionPayload.execution.attempt.id;
+
+          callback({
+            success: true,
+            executionPayload: createAttempt.executionPayload,
+          });
+        });
+
+        socket.on("UNRECOVERABLE_ERROR", async (message) => {
+          logger.log("[UNRECOVERABLE_ERROR]", message);
+
+          await crashRun(message.error);
+        });
       },
       onDisconnect: async (socket, handler, sender, logger) => {
         this.#platformSocket?.send("LOG", {
@@ -928,13 +1337,16 @@ class TaskCoordinator {
         TASK_HEARTBEAT: async (message) => {
           this.#platformSocket?.send("TASK_HEARTBEAT", message);
         },
+        TASK_RUN_HEARTBEAT: async (message) => {
+          this.#platformSocket?.send("TASK_RUN_HEARTBEAT", message);
+        },
       },
     });
 
     return provider;
   }
 
-  #cancelCheckpoint(runId: string) {
+  #cancelCheckpoint(runId: string): boolean {
     const checkpointWait = this.#checkpointableTasks.get(runId);
 
     if (checkpointWait) {
@@ -945,6 +1357,8 @@ class TaskCoordinator {
     // Cancel checkpointing procedure
     const checkpointCanceled = this.#checkpointer.cancelCheckpoint(runId);
 
+    logger.log("cancelCheckpoint()", { runId, checkpointCanceled });
+
     return checkpointCanceled;
   }
 

apps/docker-provider/.env.example

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

apps/docker-provider/Containerfile

@@ -1,16 +1,47 @@
-# syntax=docker/dockerfile:labs
-
-FROM node:18-slim AS base
-
-RUN apt-get update \
-  && apt-get install -y dumb-init
-
-FROM base
+FROM node:20-alpine@sha256:7a91aa397f2e2dfbfcdad2e2d72599f374e0b0172be1d86eeb73f1d33f36a4b2 AS node-20-alpine
 
 WORKDIR /app
 
-COPY --chown=node dist/index.mjs /app/
+FROM node-20-alpine AS pruner
+
+COPY --chown=node:node . .
+RUN npx -q turbo@1.10.9 prune --scope=docker-provider --docker
+RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' +
+
+FROM node-20-alpine AS base
+
+RUN apk add --no-cache dumb-init docker
+
+COPY --chown=node:node .gitignore .gitignore
+COPY --from=pruner --chown=node:node /app/out/json/ .
+COPY --from=pruner --chown=node:node /app/out/pnpm-lock.yaml ./pnpm-lock.yaml
+COPY --from=pruner --chown=node:node /app/out/pnpm-workspace.yaml ./pnpm-workspace.yaml
+
+FROM base AS dev-deps
+RUN corepack enable
+ENV NODE_ENV development
+
+RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm fetch --no-frozen-lockfile
+RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm install --ignore-scripts --no-frozen-lockfile
+
+FROM base AS builder
+RUN corepack enable
+
+COPY --from=pruner --chown=node:node /app/out/full/ .
+COPY --from=dev-deps --chown=node:node /app/ .
+COPY --chown=node:node turbo.json turbo.json
+
+RUN pnpm run -r --filter docker-provider build:bundle
+
+FROM base AS runner
+
+RUN corepack enable
+ENV NODE_ENV production
+
+COPY --from=builder --chown=node:node /app/apps/docker-provider/dist/index.mjs ./index.mjs
 
 EXPOSE 8000
 
-ENTRYPOINT [ "/usr/bin/dumb-init", "--", "/usr/local/bin/node", "/app/index.mjs" ]
+USER node
+
+CMD [ "/usr/bin/dumb-init", "--", "/usr/local/bin/node", "./index.mjs" ]

apps/docker-provider/src/index.ts

@@ -6,6 +6,8 @@ import {
   TaskOperationsRestoreOptions,
   TaskOperationsCreateOptions,
   TaskOperationsIndexOptions,
+  isExecaChildProcess,
+  testDockerCheckpoint,
 } from "@trigger.dev/core-apps";
 import { setTimeout } from "node:timers/promises";
 import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
@@ -13,75 +15,68 @@ import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
 const MACHINE_NAME = process.env.MACHINE_NAME || "local";
 const COORDINATOR_PORT = process.env.COORDINATOR_PORT || 8020;
 const COORDINATOR_HOST = process.env.COORDINATOR_HOST || "127.0.0.1";
+
 const OTEL_EXPORTER_OTLP_ENDPOINT =
   process.env.OTEL_EXPORTER_OTLP_ENDPOINT || "http://0.0.0.0:4318";
 
+const FORCE_CHECKPOINT_SIMULATION = ["1", "true"].includes(
+  process.env.FORCE_CHECKPOINT_SIMULATION ?? "true"
+);
+
 const logger = new SimpleLogger(`[${MACHINE_NAME}]`);
 
-type InitializeReturn = {
+type TaskOperationsInitReturn = {
   canCheckpoint: boolean;
   willSimulate: boolean;
 };
 
-function isExecaChildProcess(maybeExeca: unknown): maybeExeca is Awaited<ExecaChildProcess> {
-  return typeof maybeExeca === "object" && maybeExeca !== null && "escapedCommand" in maybeExeca;
-}
-
 class DockerTaskOperations implements TaskOperations {
   #initialized = false;
   #canCheckpoint = false;
 
   constructor(private opts = { forceSimulate: false }) {}
 
-  async #initialize(): Promise<InitializeReturn> {
+  async init(): Promise<TaskOperationsInitReturn> {
     if (this.#initialized) {
-      return this.#getInitializeReturn();
+      return this.#getInitReturn(this.#canCheckpoint);
     }
 
     logger.log("Initializing task operations");
 
-    if (this.opts.forceSimulate) {
-      logger.log("Forced simulation enabled. Will simulate regardless of checkpoint support.");
+    const testCheckpoint = await testDockerCheckpoint();
+
+    if (testCheckpoint.ok) {
+      return this.#getInitReturn(true);
     }
 
-    try {
-      await $`criu --version`;
-    } catch (error) {
-      logger.error("No checkpoint support: Missing CRIU binary. Will simulate instead.");
-      this.#canCheckpoint = false;
-      this.#initialized = true;
-
-      return this.#getInitializeReturn();
-    }
-
-    try {
-      await $`docker checkpoint`;
-    } catch (error) {
-      logger.error("No checkpoint support: Docker needs to have experimental features enabled");
-      logger.error("Will simulate instead");
-      this.#canCheckpoint = false;
-      this.#initialized = true;
-
-      return this.#getInitializeReturn();
-    }
-
-    logger.log("Full checkpoint support!");
-
-    this.#initialized = true;
-    this.#canCheckpoint = true;
-
-    return this.#getInitializeReturn();
+    logger.error(testCheckpoint.message, testCheckpoint.error);
+    return this.#getInitReturn(false);
   }
 
-  #getInitializeReturn(): InitializeReturn {
+  #getInitReturn(canCheckpoint: boolean): TaskOperationsInitReturn {
+    this.#initialized = true;
+    this.#canCheckpoint = canCheckpoint;
+
+    if (canCheckpoint) {
+      logger.log("Full checkpoint support!");
+    }
+
+    const willSimulate = !canCheckpoint || this.opts.forceSimulate;
+
+    if (willSimulate) {
+      logger.log("Simulation mode enabled. Containers will be paused, not checkpointed.", {
+        forceSimulate: this.opts.forceSimulate,
+      });
+    }
+
     return {
-      canCheckpoint: this.#canCheckpoint,
-      willSimulate: !this.#canCheckpoint || this.opts.forceSimulate,
+      canCheckpoint,
+      willSimulate,
     };
   }
 
   async index(opts: TaskOperationsIndexOptions) {
-    await this.#initialize();
+    await this.init();
 
     const containerName = this.#getIndexContainerName(opts.shortCode);
 
@@ -90,41 +85,27 @@ class DockerTaskOperations implements TaskOperations {
       port: COORDINATOR_PORT,
     });
 
-    try {
-      logger.debug(
-        await execa("docker", [
-          "run",
-          "--network=host",
-          "--rm",
-          `--env=INDEX_TASKS=true`,
-          `--env=TRIGGER_SECRET_KEY=${opts.apiKey}`,
-          `--env=TRIGGER_API_URL=${opts.apiUrl}`,
-          `--env=TRIGGER_ENV_ID=${opts.envId}`,
-          `--env=OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT}`,
-          `--env=POD_NAME=${containerName}`,
-          `--env=COORDINATOR_HOST=${COORDINATOR_HOST}`,
-          `--env=COORDINATOR_PORT=${COORDINATOR_PORT}`,
-          `--name=${containerName}`,
-          `${opts.imageRef}`,
-        ])
-      );
-    } catch (error: any) {
-      if (!isExecaChildProcess(error)) {
-        throw error;
-      }
-
-      logger.error("Index failed:", {
-        opts,
-        exitCode: error.exitCode,
-        escapedCommand: error.escapedCommand,
-        stdout: error.stdout,
-        stderr: error.stderr,
-      });
-    }
+    logger.debug(
+      await execa("docker", [
+        "run",
+        "--network=host",
+        "--rm",
+        `--env=INDEX_TASKS=true`,
+        `--env=TRIGGER_SECRET_KEY=${opts.apiKey}`,
+        `--env=TRIGGER_API_URL=${opts.apiUrl}`,
+        `--env=TRIGGER_ENV_ID=${opts.envId}`,
+        `--env=OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT}`,
+        `--env=POD_NAME=${containerName}`,
+        `--env=COORDINATOR_HOST=${COORDINATOR_HOST}`,
+        `--env=COORDINATOR_PORT=${COORDINATOR_PORT}`,
+        `--name=${containerName}`,
+        `${opts.imageRef}`,
+      ])
+    );
   }
 
   async create(opts: TaskOperationsCreateOptions) {
-    await this.#initialize();
+    await this.init();
 
     const containerName = this.#getRunContainerName(opts.runId);
 
@@ -160,7 +141,7 @@ class DockerTaskOperations implements TaskOperations {
   }
 
   async restore(opts: TaskOperationsRestoreOptions) {
-    await this.#initialize();
+    await this.init();
 
     const containerName = this.#getRunContainerName(opts.runId);
 
@@ -189,7 +170,7 @@ class DockerTaskOperations implements TaskOperations {
   }
 
   async delete(opts: { runId: string }) {
-    await this.#initialize();
+    await this.init();
 
     const containerName = this.#getRunContainerName(opts.runId);
     await this.#sendPreStop(containerName);
@@ -198,7 +179,7 @@ class DockerTaskOperations implements TaskOperations {
   }
 
   async get(opts: { runId: string }) {
-    await this.#initialize();
+    await this.init();
 
     logger.log("noop: get");
   }
@@ -278,7 +259,7 @@ class DockerTaskOperations implements TaskOperations {
 }
 
 const provider = new ProviderShell({
-  tasks: new DockerTaskOperations({ forceSimulate: true }),
+  tasks: new DockerTaskOperations({ forceSimulate: FORCE_CHECKPOINT_SIMULATION }),
   type: "docker",
 });
 

apps/kubernetes-provider/Containerfile

@@ -1,14 +1,14 @@
-FROM node:18-alpine@sha256:ca9f6cb0466f9638e59e0c249d335a07c867cd50c429b5c7830dda1bed584649 AS node-18-alpine
+FROM node:20-alpine@sha256:7a91aa397f2e2dfbfcdad2e2d72599f374e0b0172be1d86eeb73f1d33f36a4b2 AS node-20-alpine
 
 WORKDIR /app
 
-FROM node-18-alpine AS pruner
+FROM node-20-alpine AS pruner
 
 COPY --chown=node:node . .
 RUN npx -q turbo@1.10.9 prune --scope=kubernetes-provider --docker
 RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' +
 
-FROM node-18-alpine AS base
+FROM node-20-alpine AS base
 
 RUN apk add --no-cache dumb-init
 

apps/kubernetes-provider/src/index.ts

@@ -47,6 +47,10 @@ class KubernetesTaskOperations implements TaskOperations {
     this.#k8sApi = this.#createK8sApi();
   }
 
+  async init() {
+    // noop
+  }
+
   async index(opts: TaskOperationsIndexOptions) {
     await this.#createJob(
       {
@@ -212,7 +216,7 @@ class KubernetesTaskOperations implements TaskOperations {
             },
             {
               name: "populate-taskinfo",
-              image: "docker.io/library/busybox",
+              image: "registry.digitalocean.com/trigger/busybox",
               imagePullPolicy: "IfNotPresent",
               command: ["/bin/sh", "-c"],
               args: ["printenv COORDINATOR_HOST | tee /etc/taskinfo/coordinator-host"],
@@ -316,6 +320,9 @@ class KubernetesTaskOperations implements TaskOperations {
         {
           name: "registry-trigger",
         },
+        {
+          name: "registry-trigger-failover",
+        },
       ],
       nodeSelector: {
         nodetype: "worker",

apps/proxy/package.json

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

apps/proxy/src/index.ts

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

apps/proxy/src/rateLimit.ts

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

apps/proxy/src/rateLimiter.ts

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

apps/proxy/wrangler.toml

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

apps/webapp/app/components/SetupCommands.tsx

@@ -8,6 +8,7 @@ import {
 } from "./primitives/ClientTabs";
 import { ClipboardField } from "./primitives/ClipboardField";
 import { Paragraph } from "./primitives/Paragraph";
+import { useAppOrigin } from "~/hooks/useAppOrigin";
 
 export function InitCommand({ appOrigin, apiKey }: { appOrigin: string; apiKey: string }) {
   return (
@@ -133,9 +134,38 @@ export function TriggerDevStep({ extra }: { extra?: string }) {
 // Trigger.dev version 3 setup commands
 const v3PackageTag = "beta";
 
+function getApiUrlArg() {
+  const appOrigin = useAppOrigin();
+
+  let apiUrl: string | undefined = undefined;
+
+  switch (appOrigin) {
+    case "https://cloud.trigger.dev":
+      // don't display the arg, use the CLI default
+      break;
+    case "https://test-cloud.trigger.dev":
+      apiUrl = "https://test-api.trigger.dev";
+      break;
+    case "https://internal.trigger.dev":
+      apiUrl = "https://internal-api.trigger.dev";
+      break;
+    default:
+      apiUrl = appOrigin;
+      break;
+  }
+
+  return apiUrl ? `-a ${apiUrl}` : undefined;
+}
+
 export function InitCommandV3() {
   const project = useProject();
   const projectRef = project.ref;
+
+  const apiUrlArg = getApiUrlArg();
+
+  const initCommandParts = [`trigger.dev@${v3PackageTag}`, "init", `-p ${projectRef}`, apiUrlArg];
+  const initCommand = initCommandParts.filter(Boolean).join(" ");
+
   return (
     <ClientTabs defaultValue="npm">
       <ClientTabsList>
@@ -148,7 +178,7 @@ export function InitCommandV3() {
           variant="primary/medium"
           iconButton
           className="mb-4"
-          value={`npx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
+          value={`npx ${initCommand}`}
         />
       </ClientTabsContent>
       <ClientTabsContent value={"pnpm"}>
@@ -156,7 +186,7 @@ export function InitCommandV3() {
           variant="primary/medium"
           iconButton
           className="mb-4"
-          value={`pnpm dlx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
+          value={`pnpm dlx ${initCommand}`}
         />
       </ClientTabsContent>
       <ClientTabsContent value={"yarn"}>
@@ -164,7 +194,7 @@ export function InitCommandV3() {
           variant="primary/medium"
           iconButton
           className="mb-4"
-          value={`yarn dlx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
+          value={`yarn dlx ${initCommand}`}
         />
       </ClientTabsContent>
     </ClientTabs>

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

@@ -20,6 +20,17 @@ export function DeploymentError({ errorData }: DeploymentErrorProps) {
           maxLines={20}
         />
       )}
+      {errorData.stderr && (
+        <>
+          <DeploymentErrorHeader title="Error logs:" />
+          <CodeBlock
+            showCopyButton={false}
+            showLineNumbers={false}
+            code={errorData.stderr}
+            maxLines={20}
+          />
+        </>
+      )}
     </div>
   );
 }

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

@@ -0,0 +1,55 @@
+import { ArrowPathIcon } from "@heroicons/react/20/solid";
+import { Form, useNavigation } from "@remix-run/react";
+import { Button } from "~/components/primitives/Buttons";
+import {
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+} from "~/components/primitives/Dialog";
+
+type RollbackDeploymentDialogProps = {
+  projectId: string;
+  deploymentShortCode: string;
+  redirectPath: string;
+};
+
+export function RollbackDeploymentDialog({
+  projectId,
+  deploymentShortCode,
+  redirectPath,
+}: RollbackDeploymentDialogProps) {
+  const navigation = useNavigation();
+
+  const formAction = `/resources/${projectId}/deployments/${deploymentShortCode}/rollback`;
+  const isLoading = navigation.formAction === formAction;
+
+  return (
+    <DialogContent key="rollback">
+      <DialogHeader>Roll back to this deployment?</DialogHeader>
+      <DialogDescription>
+        This deployment will become the default for all future runs. Tasks triggered but not
+        included in this deploy will remain queued until you roll back to or create a new deployment
+        with these tasks included.
+      </DialogDescription>
+      <DialogFooter>
+        <Form
+          action={`/resources/${projectId}/deployments/${deploymentShortCode}/rollback`}
+          method="post"
+        >
+          <Button
+            type="submit"
+            name="redirectUrl"
+            value={redirectPath}
+            variant="primary/small"
+            LeadingIcon={isLoading ? "spinner-white" : ArrowPathIcon}
+            disabled={isLoading}
+            shortcut={{ modifiers: ["meta"], key: "enter" }}
+          >
+            {isLoading ? "Rolling back..." : "Roll back deployment"}
+          </Button>
+        </Form>
+      </DialogFooter>
+    </DialogContent>
+  );
+}

apps/webapp/app/consts.ts

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

apps/webapp/app/env.server.ts

@@ -27,15 +27,17 @@ const EnvironmentSchema = z.object({
     .string()
     .refine(isValidRegex, "WHITELISTED_EMAILS must be a valid regex.")
     .optional(),
+  ADMIN_EMAILS: z.string().refine(isValidRegex, "ADMIN_EMAILS must be a valid regex.").optional(),
   REMIX_APP_PORT: z.string().optional(),
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ENV: z.string().default(process.env.NODE_ENV),
   SERVICE_NAME: z.string().default("trigger.dev webapp"),
   SECRET_STORE: SecretStoreOptionsSchema.default("DATABASE"),
-  POSTHOG_PROJECT_KEY: z.string().optional(),
+  POSTHOG_PROJECT_KEY: z.string().default("phc_LFH7kJiGhdIlnO22hTAKgHpaKhpM8gkzWAFvHmf5vfS"),
   TELEMETRY_TRIGGER_API_KEY: z.string().optional(),
   TELEMETRY_TRIGGER_API_URL: z.string().optional(),
+  TRIGGER_TELEMETRY_DISABLED: z.string().optional(),
   HIGHLIGHT_PROJECT_ID: z.string().optional(),
   AUTH_GITHUB_CLIENT_ID: z.string().optional(),
   AUTH_GITHUB_CLIENT_SECRET: z.string().optional(),
@@ -100,6 +102,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 +117,7 @@ const EnvironmentSchema = z.object({
   CONTAINER_REGISTRY_USERNAME: z.string().optional(),
   CONTAINER_REGISTRY_PASSWORD: z.string().optional(),
   DEPLOY_REGISTRY_HOST: z.string().optional(),
+  DEPLOY_REGISTRY_NAMESPACE: z.string().default("trigger"),
   OBJECT_STORE_BASE_URL: z.string().optional(),
   OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
   OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
@@ -164,6 +171,24 @@ const EnvironmentSchema = z.object({
   ALERT_RESEND_API_KEY: z.string().optional(),
 
   MAX_SEQUENTIAL_INDEX_FAILURE_COUNT: z.coerce.number().default(96),
+
+  LOOPS_API_KEY: z.string().optional(),
+  MARQS_DISABLE_REBALANCING: z.coerce.boolean().default(false),
+
+  VERBOSE_GRAPHILE_LOGGING: z.string().default("false"),
+  V2_MARQS_ENABLED: z.string().default("0"),
+  V2_MARQS_CONSUMER_POOL_ENABLED: z.string().default("0"),
+  V2_MARQS_CONSUMER_POOL_SIZE: z.coerce.number().int().default(10),
+  V2_MARQS_CONSUMER_POLL_INTERVAL_MS: z.coerce.number().int().default(1000),
+  V2_MARQS_QUEUE_SELECTION_COUNT: z.coerce.number().int().default(36),
+  V2_MARQS_VISIBILITY_TIMEOUT_MS: z.coerce
+    .number()
+    .int()
+    .default(60 * 1000 * 15),
+  V2_MARQS_DEFAULT_ENV_CONCURRENCY: z.coerce.number().int().default(100),
+  V2_MARQS_VERBOSE: z.string().default("0"),
+  V3_MARQS_CONCURRENCY_MONITOR_ENABLED: z.string().default("0"),
+  V2_MARQS_CONCURRENCY_MONITOR_ENABLED: z.string().default("0"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/features.server.ts

@@ -7,20 +7,28 @@ export type TriggerFeatures = {
   alertsEnabled: boolean;
 };
 
-// If the request host is cloud.trigger.dev then we are on the managed cloud
-// or if env.NODE_ENV is development
-export function featuresForRequest(request: Request): TriggerFeatures {
-  const url = requestUrl(request);
-
-  const isManagedCloud =
-    url.host === "cloud.trigger.dev" ||
-    url.host === "test-cloud.trigger.dev" ||
-    url.host === "internal.trigger.dev" ||
-    process.env.CLOUD_ENV === "development";
+function isManagedCloud(host: string): boolean {
+  return (
+    host === "cloud.trigger.dev" ||
+    host === "test-cloud.trigger.dev" ||
+    host === "internal.trigger.dev" ||
+    process.env.CLOUD_ENV === "development"
+  );
+}
 
+function featuresForHost(host: string): TriggerFeatures {
   return {
-    isManagedCloud,
+    isManagedCloud: isManagedCloud(host),
     v3Enabled: env.V3_ENABLED === "true",
     alertsEnabled: env.ALERT_FROM_EMAIL !== undefined && env.ALERT_RESEND_API_KEY !== undefined,
   };
 }
+
+export function featuresForRequest(request: Request): TriggerFeatures {
+  const url = requestUrl(request);
+  return featuresForUrl(url);
+}
+
+export function featuresForUrl(url: URL): TriggerFeatures {
+  return featuresForHost(url.host);
+}

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

@@ -245,6 +245,7 @@ export async function revokeInvite({
   const invite = await prisma.orgMemberInvite.delete({
     where: {
       id: inviteId,
+      organizationId: org.id,
     },
     select: {
       email: true,

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

@@ -8,10 +8,10 @@ import type {
 import { customAlphabet } from "nanoid";
 import slug from "slug";
 import { prisma, PrismaClientOrTransaction } from "~/db.server";
-import { createProject } from "./project.server";
 import { generate } from "random-words";
 import { createApiKeyForEnv, createPkApiKeyForEnv, envSlug } from "./api-key.server";
 import { env } from "~/env.server";
+import { featuresForUrl } from "~/features.server";
 
 export type { Organization };
 
@@ -52,6 +52,8 @@ export async function createOrganization(
     );
   }
 
+  const features = featuresForUrl(new URL(env.APP_ORIGIN));
+
   const organization = await prisma.organization.create({
     data: {
       title,
@@ -64,6 +66,7 @@ export async function createOrganization(
           role: "ADMIN",
         },
       },
+      v3Enabled: features.v3Enabled && !features.isManagedCloud,
     },
     include: {
       members: true,

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/models/user.server.ts

@@ -47,12 +47,21 @@ export async function findOrCreateMagicLinkUser(
     },
   });
 
+  const adminEmailRegex = env.ADMIN_EMAILS ? new RegExp(env.ADMIN_EMAILS) : undefined;
+  const makeAdmin = adminEmailRegex ? adminEmailRegex.test(input.email) : false;
+
   const user = await prisma.user.upsert({
     where: {
       email: input.email,
     },
-    update: { email: input.email },
-    create: { email: input.email, authenticationMethod: "MAGIC_LINK" },
+    update: {
+      email: input.email,
+    },
+    create: {
+      email: input.email,
+      authenticationMethod: "MAGIC_LINK",
+      admin: makeAdmin, // only on create, to prevent automatically removing existing admins
+    },
   });
 
   return {

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,13 +62,16 @@ 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;
     jobKeyMode?: "replace" | "preserve_run_at" | "unsafe_dedupe";
     flags?: string[];
-    handler: (payload: z.infer<TConsumerSchema[K]>, job: GraphileJob) => Promise<void>;
+    handler: (
+      payload: z.infer<TConsumerSchema[K]>,
+      job: GraphileJob,
+      helpers: JobHelpers
+    ) => Promise<void>;
   };
 };
 
@@ -75,11 +84,17 @@ export type ZodRecurringTasks = {
   [key: string]: {
     match: string;
     options?: CronItemOptions;
-    handler: (payload: RecurringTaskPayload, job: GraphileJob) => Promise<void>;
+    handler: (
+      payload: RecurringTaskPayload,
+      job: GraphileJob,
+      helpers: JobHelpers
+    ) => Promise<void>;
   };
 };
 
-export type ZodWorkerEnqueueOptions = TaskSpec & {
+type ZodTaskSpec = Omit<TaskSpec, "queueName">;
+
+export type ZodWorkerEnqueueOptions = ZodTaskSpec & {
   tx?: PrismaClientOrTransaction;
 };
 
@@ -162,12 +177,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 +265,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 +292,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 +330,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);
 
@@ -298,12 +338,6 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       }
     }
 
-    logger.debug("Enqueuing worker task", {
-      identifier,
-      payload,
-      spec,
-    });
-
     const { job, durationInMs } = await this.#addJob(
       identifier as string,
       payload,
@@ -345,17 +379,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 +479,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),
     });
   }
 
@@ -569,7 +583,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       },
       async (span) => {
         try {
-          await task.handler(payload, job);
+          await task.handler(payload, job, helpers);
         } catch (error) {
           if (error instanceof Error) {
             span.recordException(error);
@@ -650,7 +664,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       },
       async (span) => {
         try {
-          await recurringTask.handler(payload._cron, job);
+          await recurringTask.handler(payload._cron, job, helpers);
         } catch (error) {
           if (error instanceof Error) {
             span.recordException(error);

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

@@ -10,15 +10,12 @@ import { User } from "~/models/user.server";
 import { z } from "zod";
 import { projectPath } from "~/utils/pathBuilder";
 import { JobRunStatus } from "@trigger.dev/database";
+import { BasePresenter } from "./v3/basePresenter.server";
 
 export type ProjectJob = Awaited<ReturnType<JobListPresenter["call"]>>[0];
 
-export class JobListPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
+export class JobListPresenter extends BasePresenter {
+  
 
   public async call({
     userId,
@@ -39,7 +36,7 @@ export class JobListPresenter {
       ? { some: { integration: { slug: integrationSlug } } }
       : {};
 
-    const jobs = await this.#prismaClient.job.findMany({
+    const jobs = await this._replica.job.findMany({
       select: {
         id: true,
         slug: true,
@@ -106,7 +103,7 @@ export class JobListPresenter {
     }[];
 
     if (jobs.length > 0) {
-      latestRuns = await this.#prismaClient.$queryRaw<
+      latestRuns = await this._replica.$queryRaw<
         {
           createdAt: Date;
           status: JobRunStatus;

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

@@ -11,13 +11,10 @@ import { User } from "~/models/user.server";
 import { z } from "zod";
 import { projectPath } from "~/utils/pathBuilder";
 import { Job } from "@trigger.dev/database";
+import { BasePresenter } from "./v3/basePresenter.server";
 
-export class JobPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
+export class JobPresenter extends BasePresenter {
+  
 
   public async call({
     userId,
@@ -30,7 +27,7 @@ export class JobPresenter {
     projectSlug: Project["slug"];
     organizationSlug: Organization["slug"];
   }) {
-    const job = await this.#prismaClient.job.findFirst({
+    const job = await this._replica.job.findFirst({
       select: {
         id: true,
         slug: true,

apps/webapp/app/presenters/OrgBillingPlanPresenter.ts

@@ -1,14 +1,7 @@
-import { PrismaClient, prisma } from "~/db.server";
-import { logger } from "~/services/logger.server";
 import { BillingService } from "../services/billing.server";
+import { BasePresenter } from "./v3/basePresenter.server";
 
-export class OrgBillingPlanPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
+export class OrgBillingPlanPresenter extends BasePresenter {
   public async call({ slug, isManagedCloud }: { slug: string; isManagedCloud: boolean }) {
     const billingPresenter = new BillingService(isManagedCloud);
     const plans = await billingPresenter.getPlans();
@@ -17,7 +10,7 @@ export class OrgBillingPlanPresenter {
       return;
     }
 
-    const organization = await this.#prismaClient.organization.findFirst({
+    const organization = await this._replica.organization.findFirst({
       where: {
         slug,
       },
@@ -27,7 +20,7 @@ export class OrgBillingPlanPresenter {
       return;
     }
 
-    const maxConcurrency = await this.#prismaClient.$queryRaw<
+    const maxConcurrency = await this._replica.$queryRaw<
       { organization_id: string; max_concurrent_runs: BigInt }[]
     >`WITH events AS (
       SELECT

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

@@ -1,17 +1,12 @@
 import { estimate } from "@trigger.dev/billing";
-import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema } from "~/db.server";
 import { featuresForRequest } from "~/features.server";
 import { BillingService } from "~/services/billing.server";
+import { BasePresenter } from "./v3/basePresenter.server";
 
-export class OrgUsagePresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
+export class OrgUsagePresenter extends BasePresenter {
   public async call({ userId, slug, request }: { userId: string; slug: string; request: Request }) {
-    const organization = await this.#prismaClient.organization.findFirst({
+    const organization = await this._replica.organization.findFirst({
       where: {
         slug,
         members: {
@@ -27,7 +22,7 @@ export class OrgUsagePresenter {
     }
 
     // Get count of runs since the start of the current month
-    const runsCount = await this.#prismaClient.jobRun.count({
+    const runsCount = await this._replica.jobRun.count({
       where: {
         organizationId: organization.id,
         createdAt: {
@@ -48,7 +43,7 @@ export class OrgUsagePresenter {
     // ]
     // This will be used to generate the chart on the usage page
     // Use prisma queryRaw for this since prisma doesn't support grouping by month
-    const monthlyRunsDataRaw = await this.#prismaClient.$queryRaw<
+    const monthlyRunsDataRaw = await this._replica.$queryRaw<
       {
         month: string;
         count: number;
@@ -64,7 +59,7 @@ export class OrgUsagePresenter {
     const monthlyRunsDataDisplay = fillInMissingRunMonthlyData(monthlyRunsData, 6);
 
     // Max concurrency each day over past 30 days
-    const concurrencyChartRawData = await this.#prismaClient.$queryRaw<
+    const concurrencyChartRawData = await this._replica.$queryRaw<
       { day: Date; max_concurrent_runs: BigInt }[]
     >`
       WITH time_boundaries AS (
@@ -115,7 +110,7 @@ export class OrgUsagePresenter {
       concurrencyChartRawData
     );
 
-    const dailyRunsRawData = await this.#prismaClient.$queryRaw<
+    const dailyRunsRawData = await this._replica.$queryRaw<
       { day: Date; runs: BigInt }[]
     >`SELECT date_trunc('day', "createdAt") as day, COUNT(*) as runs FROM ${sqlDatabaseSchema}."JobRun" WHERE "organizationId" = ${organization.id} AND "createdAt" >= NOW() - INTERVAL '30 days' AND "internal" = FALSE GROUP BY day`;
 

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

@@ -7,6 +7,7 @@ import {
 } from "~/components/runs/RunStatuses";
 import { PrismaClient, prisma } from "~/db.server";
 import { getUsername } from "~/utils/username";
+import { BasePresenter } from "./v3/basePresenter.server";
 
 type RunListOptions = {
   userId: string;
@@ -27,12 +28,8 @@ const DEFAULT_PAGE_SIZE = 20;
 
 export type RunList = Awaited<ReturnType<RunListPresenter["call"]>>;
 
-export class RunListPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
+export class RunListPresenter extends BasePresenter {
+  
 
   public async call({
     userId,
@@ -53,7 +50,7 @@ export class RunListPresenter {
     const directionMultiplier = direction === "forward" ? 1 : -1;
 
     // Find the organization that the user is a member of
-    const organization = await this.#prismaClient.organization.findFirstOrThrow({
+    const organization = await this._replica.organization.findFirstOrThrow({
       select: {
         id: true,
       },
@@ -64,7 +61,7 @@ export class RunListPresenter {
     });
 
     // Find the project scoped to the organization
-    const project = await this.#prismaClient.project.findFirstOrThrow({
+    const project = await this._replica.project.findFirstOrThrow({
       select: {
         id: true,
       },
@@ -75,7 +72,7 @@ export class RunListPresenter {
     });
 
     const job = jobSlug
-      ? await this.#prismaClient.job.findFirstOrThrow({
+      ? await this._replica.job.findFirstOrThrow({
           where: {
             slug: jobSlug,
             projectId: project.id,
@@ -84,10 +81,10 @@ export class RunListPresenter {
       : undefined;
 
     const event = eventId
-      ? await this.#prismaClient.eventRecord.findUnique({ where: { id: eventId } })
+      ? await this._replica.eventRecord.findUnique({ where: { id: eventId } })
       : undefined;
 
-    const runs = await this.#prismaClient.jobRun.findMany({
+    const runs = await this._replica.jobRun.findMany({
       select: {
         id: true,
         number: true,

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

@@ -7,6 +7,9 @@ import { getUsername } from "~/utils/username";
 
 const pageSize = 20;
 
+export type DeploymentList = Awaited<ReturnType<DeploymentListPresenter["call"]>>;
+export type DeploymentListItem = DeploymentList["deployments"][0];
+
 export class DeploymentListPresenter {
   #prismaClient: PrismaClient;
 
@@ -136,6 +139,8 @@ LIMIT ${pageSize} OFFSET ${pageSize * (page - 1)};`;
           deployedAt: deployment.deployedAt,
           tasksCount: deployment.tasksCount ? Number(deployment.tasksCount) : null,
           label: label?.label,
+          isCurrent: label?.label === "current",
+          isDeployed: deployment.status === "DEPLOYED",
           environment: {
             id: environment.id,
             type: environment.type,

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

@@ -17,6 +17,7 @@ export type ErrorData = {
   name: string;
   message: string;
   stack?: string;
+  stderr?: string;
 };
 
 export class DeploymentPresenter {
@@ -177,17 +178,20 @@ export class DeploymentPresenter {
             name: parsedErrorData.data.name,
             message: parsedErrorData.data.message,
             stack: createTaskMetadataFailedErrorStack(parsedError.data),
+            stderr: parsedErrorData.data.stderr,
           };
         } else {
           return {
             name: parsedErrorData.data.name,
             message: parsedErrorData.data.message,
+            stderr: parsedErrorData.data.stderr,
           };
         }
       } else {
         return {
           name: parsedErrorData.data.name,
           message: parsedErrorData.data.message,
+          stderr: parsedErrorData.data.stderr,
         };
       }
     }
@@ -196,6 +200,7 @@ export class DeploymentPresenter {
       name: parsedErrorData.data.name,
       message: parsedErrorData.data.message,
       stack: parsedErrorData.data.stack,
+      stderr: parsedErrorData.data.stderr,
     };
   }
 }

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,8 +1,10 @@
 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";
+import { BasePresenter } from "./basePresenter.server";
 
 type ScheduleListOptions = {
   projectId: string;
@@ -33,12 +35,7 @@ export type ScheduleListItem = {
 export type ScheduleList = Awaited<ReturnType<ScheduleListPresenter["call"]>>;
 export type ScheduleListAppliedFilters = ScheduleList["filters"];
 
-export class ScheduleListPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
+export class ScheduleListPresenter extends BasePresenter {
 
   public async call({
     userId,
@@ -53,7 +50,7 @@ export class ScheduleListPresenter {
       tasks !== undefined || environments !== undefined || (search !== undefined && search !== "");
 
     // Find the project scoped to the organization
-    const project = await this.#prismaClient.project.findFirstOrThrow({
+    const project = await this._replica.project.findFirstOrThrow({
       select: {
         id: true,
         environments: {
@@ -81,7 +78,7 @@ export class ScheduleListPresenter {
     });
 
     //get all possible scheduled tasks
-    const possibleTasks = await this.#prismaClient.backgroundWorkerTask.findMany({
+    const possibleTasks = await this._replica.backgroundWorkerTask.findMany({
       distinct: ["slug"],
       where: {
         projectId: project.id,
@@ -92,7 +89,7 @@ export class ScheduleListPresenter {
     //do this here to protect against SQL injection
     search = search && search !== "" ? `%${search}%` : undefined;
 
-    const totalCount = await this.#prismaClient.taskSchedule.count({
+    const totalCount = await this._replica.taskSchedule.count({
       where: {
         projectId: project.id,
         taskIdentifier: tasks ? { in: tasks } : undefined,
@@ -134,7 +131,7 @@ export class ScheduleListPresenter {
       },
     });
 
-    const rawSchedules = await this.#prismaClient.taskSchedule.findMany({
+    const rawSchedules = await this._replica.taskSchedule.findMany({
       select: {
         id: true,
         friendlyId: true,
@@ -198,7 +195,7 @@ export class ScheduleListPresenter {
 
     const latestRuns =
       rawSchedules.length > 0
-        ? await this.#prismaClient.$queryRaw<{ scheduleId: string; createdAt: Date }[]>`
+        ? await this._replica.$queryRaw<{ scheduleId: string; createdAt: Date }[]>`
     SELECT t."scheduleId", t."createdAt"
     FROM (
       SELECT "scheduleId", MAX("createdAt") as "LatestRun"
@@ -233,14 +230,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 +242,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,12 +8,18 @@ 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,
+  onlyDevEnvironments,
+  exceptDevEnvironments,
+  sortEnvironments,
+} from "~/utils/environmentSort";
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
 import { TaskRunStatus } from "~/database-types";
+import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
 
 export type Task = {
   slug: string;
@@ -72,6 +78,9 @@ export class TaskListPresenter extends BasePresenter {
       },
     });
 
+    const devEnvironments = onlyDevEnvironments(project.environments);
+    const nonDevEnvironments = exceptDevEnvironments(project.environments);
+
     const tasks = await this._replica.$queryRaw<
       {
         id: string;
@@ -83,10 +92,21 @@ export class TaskListPresenter extends BasePresenter {
         triggerSource: TaskTriggerSource;
       }[]
     >`
-    WITH workers AS (
+    WITH non_dev_workers AS (
+      SELECT wd."workerId" AS id
+      FROM ${sqlDatabaseSchema}."WorkerDeploymentPromotion" wdp
+      INNER JOIN ${sqlDatabaseSchema}."WorkerDeployment" wd
+        ON wd.id = wdp."deploymentId"
+      WHERE wdp."environmentId" IN (${Prisma.join(nonDevEnvironments.map((e) => e.id))})
+        AND wdp."label" = ${CURRENT_DEPLOYMENT_LABEL}
+    ),
+    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(devEnvironments).map((e) => e.id)
+      )})
+        OR id IN (SELECT id FROM non_dev_workers)
       ORDER BY "runtimeEnvironmentId", "createdAt" DESC
     )
     SELECT tasks.id, slug, "filePath", "exportName", "triggerSource", tasks."runtimeEnvironmentId", tasks."createdAt"
@@ -119,7 +139,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

@@ -3,7 +3,8 @@ import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { TestSearchParams } from "~/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test/route";
 import { sortEnvironments } from "~/utils/environmentSort";
 import { createSearchParams } from "~/utils/searchParams";
-import { getUsername } from "~/utils/username";
+import { findCurrentWorkerDeployment } from "~/v3/models/workerDeployment.server";
+import { BasePresenter } from "./basePresenter.server";
 
 type TaskListOptions = {
   userId: string;
@@ -15,16 +16,10 @@ export type TaskList = Awaited<ReturnType<TestPresenter["call"]>>;
 export type TaskListItem = NonNullable<TaskList["tasks"]>[0];
 export type SelectedEnvironment = NonNullable<TaskList["selectedEnvironment"]>;
 
-export class TestPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
+export class TestPresenter extends BasePresenter {
   public async call({ userId, projectSlug, url }: TaskListOptions) {
     // Find the project scoped to the organization
-    const project = await this.#prismaClient.project.findFirstOrThrow({
+    const project = await this._replica.project.findFirstOrThrow({
       select: {
         id: true,
         environments: {
@@ -36,9 +31,12 @@ export class TestPresenter {
           where: {
             OR: [
               {
-                orgMember: null,
+                type: {
+                  in: ["PREVIEW", "STAGING", "PRODUCTION"],
+                },
               },
               {
+                type: "DEVELOPMENT",
                 orgMember: {
                   userId,
                 },
@@ -82,31 +80,8 @@ export class TestPresenter {
       };
     }
 
-    //get all possible tasks
-    const tasks = await this.#prismaClient.$queryRaw<
-      {
-        id: string;
-        version: string;
-        taskIdentifier: string;
-        filePath: string;
-        exportName: string;
-        friendlyId: string;
-        triggerSource: TaskTriggerSource;
-      }[]
-    >`WITH workers AS (
-      SELECT 
-            bw.*,
-            ROW_NUMBER() OVER(ORDER BY string_to_array(bw.version, '.')::int[] DESC) AS rn
-      FROM 
-            ${sqlDatabaseSchema}."BackgroundWorker" bw
-      WHERE "runtimeEnvironmentId" = ${matchingEnvironment.id}
-    ),
-    latest_workers AS (SELECT * FROM workers WHERE rn = 1)
-    SELECT bwt.id, version, slug as "taskIdentifier", "filePath", "exportName", bwt."friendlyId", bwt."triggerSource"
-    FROM latest_workers
-    JOIN ${sqlDatabaseSchema}."BackgroundWorkerTask" bwt ON bwt."workerId" = latest_workers.id
-    ORDER BY bwt."exportName" ASC;
-    `;
+    const isDev = matchingEnvironment.type === "DEVELOPMENT";
+    const tasks = await this.#getTasks(matchingEnvironment.id, isDev);
 
     return {
       hasSelectedEnvironment: true as const,
@@ -115,8 +90,7 @@ export class TestPresenter {
       tasks: tasks.map((task) => {
         return {
           id: task.id,
-          version: task.version,
-          taskIdentifier: task.taskIdentifier,
+          taskIdentifier: task.slug,
           filePath: task.filePath,
           exportName: task.exportName,
           friendlyId: task.friendlyId,
@@ -125,4 +99,35 @@ export class TestPresenter {
       }),
     };
   }
+
+  async #getTasks(envId: string, isDev: boolean) {
+    if (isDev) {
+      return await this._replica.$queryRaw<
+        {
+          id: string;
+          version: string;
+          slug: string;
+          filePath: string;
+          exportName: string;
+          friendlyId: string;
+          triggerSource: TaskTriggerSource;
+        }[]
+      >`WITH workers AS (
+          SELECT 
+                bw.*,
+                ROW_NUMBER() OVER(ORDER BY string_to_array(bw.version, '.')::int[] DESC) AS rn
+          FROM 
+                ${sqlDatabaseSchema}."BackgroundWorker" bw
+          WHERE "runtimeEnvironmentId" = ${envId}
+        ),
+        latest_workers AS (SELECT * FROM workers WHERE rn = 1)
+        SELECT bwt.id, version, slug, "filePath", "exportName", bwt."friendlyId", bwt."triggerSource"
+        FROM latest_workers
+        JOIN ${sqlDatabaseSchema}."BackgroundWorkerTask" bwt ON bwt."workerId" = latest_workers.id
+        ORDER BY bwt."exportName" ASC;`;
+    } else {
+      const currentDeployment = await findCurrentWorkerDeployment(envId);
+      return currentDeployment?.worker?.tasks ?? [];
+    }
+  }
 }

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.deployments/route.tsx

@@ -1,7 +1,6 @@
-import { CommandLineIcon, ServerIcon } from "@heroicons/react/20/solid";
-import { Outlet, useParams } from "@remix-run/react";
+import { ArrowPathIcon, CommandLineIcon, ServerIcon } from "@heroicons/react/20/solid";
+import { Outlet, useLocation, useParams } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
-import { TerminalIcon, TerminalSquareIcon } from "lucide-react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { z } from "zod";
 import { BlankstateInstructions } from "~/components/BlankstateInstructions";
@@ -9,8 +8,9 @@ import { UserAvatar } from "~/components/UserProfilePhoto";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
 import { Badge } from "~/components/primitives/Badge";
-import { LinkButton } from "~/components/primitives/Buttons";
+import { Button, LinkButton } from "~/components/primitives/Buttons";
 import { DateTime } from "~/components/primitives/DateTime";
+import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
 import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
 import { PaginationControls } from "~/components/primitives/Pagination";
 import { Paragraph } from "~/components/primitives/Paragraph";
@@ -24,24 +24,26 @@ import {
   TableBlankRow,
   TableBody,
   TableCell,
-  TableCellChevron,
+  TableCellMenu,
   TableHeader,
   TableHeaderCell,
   TableRow,
 } from "~/components/primitives/Table";
 import { TextLink } from "~/components/primitives/TextLink";
 import { DeploymentStatus } from "~/components/runs/v3/DeploymentStatus";
+import { RollbackDeploymentDialog } from "~/components/runs/v3/RollbackDeploymentDialog";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { useUser } from "~/hooks/useUser";
-import { DeploymentListPresenter } from "~/presenters/v3/DeploymentListPresenter.server";
+import {
+  DeploymentListItem,
+  DeploymentListPresenter,
+} from "~/presenters/v3/DeploymentListPresenter.server";
 import { requireUserId } from "~/services/session.server";
-import { cn } from "~/utils/cn";
 import {
   ProjectParamSchema,
   docsPath,
   v3DeploymentPath,
-  v3DeploymentsPath,
   v3EnvironmentVariablesPath,
 } from "~/utils/pathBuilder";
 import { createSearchParams } from "~/utils/searchParams";
@@ -166,7 +168,7 @@ export default function Page() {
                                 "–"
                               )}
                             </TableCell>
-                            <TableCellChevron to={path} />
+                            <DeploymentActionsCell deployment={deployment} path={path} />
                           </TableRow>
                         );
                       })
@@ -240,3 +242,35 @@ function CreateDeploymentInstructions() {
     </MainCenteredContainer>
   );
 }
+
+function DeploymentActionsCell({
+  deployment,
+  path,
+}: {
+  deployment: DeploymentListItem;
+  path: string;
+}) {
+  const location = useLocation();
+  const project = useProject();
+
+  if (deployment.isCurrent || !deployment.isDeployed) return <TableCell to={path}>{""}</TableCell>;
+
+  return (
+    <TableCellMenu isSticky>
+      {!deployment.isCurrent && deployment.isDeployed && (
+        <Dialog>
+          <DialogTrigger asChild>
+            <Button variant="small-menu-item" LeadingIcon={ArrowPathIcon}>
+              Rollback
+            </Button>
+          </DialogTrigger>
+          <RollbackDeploymentDialog
+            projectId={project.id}
+            deploymentShortCode={deployment.shortCode}
+            redirectPath={`${location.pathname}${location.search}`}
+          />
+        </Dialog>
+      )}
+    </TableCellMenu>
+  );
+}

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

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

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

@@ -0,0 +1,31 @@
+import { LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  const details = await marqs?.getSharedQueueDetails();
+
+  return json(details);
+}

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

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

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

@@ -0,0 +1,45 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { CreateTaskRunAttemptService } from "~/v3/services/createTaskRunAttempt.server";
+
+const ParamsSchema = z.object({
+  /* This is the run friendly ID */
+  runParam: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { runParam } = parsed.data;
+
+  const service = new CreateTaskRunAttemptService();
+
+  try {
+    const { execution } = await service.call(runParam, authenticationResult.environment);
+
+    return json(execution, { status: 200 });
+  } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: error.status ?? 422 });
+    }
+
+    return json(
+      { error: error instanceof Error ? error.message : "Internal Server Error" },
+      { status: 500 }
+    );
+  }
+}

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

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

apps/webapp/app/routes/api.v2.events.$eventId.ts

@@ -2,7 +2,7 @@ import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { GetEvent } from "@trigger.dev/core";
 import { z } from "zod";
-import { prisma } from "~/db.server";
+import { $replica } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { apiCors } from "~/utils/apiCors";
 
@@ -59,7 +59,7 @@ function toJSON(eventRecord: FoundEventRecord): GetEvent {
 type FoundEventRecord = NonNullable<Awaited<ReturnType<typeof findEventRecord>>>;
 
 async function findEventRecord(eventId: string, environmentId: string) {
-  return await prisma.eventRecord.findUnique({
+  return await $replica.eventRecord.findUnique({
     select: {
       eventId: true,
       name: true,

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/routes/resources.$projectId.deployments.$deploymentShortCode.rollback.ts

@@ -0,0 +1,93 @@
+import { parse } from "@conform-to/zod";
+import { ActionFunction, json } from "@remix-run/node";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
+import { logger } from "~/services/logger.server";
+import { requireUserId } from "~/services/session.server";
+import { RollbackDeploymentService } from "~/v3/services/rollbackDeployment.server";
+
+export const rollbackSchema = z.object({
+  redirectUrl: z.string(),
+});
+
+const ParamSchema = z.object({
+  projectId: z.string(),
+  deploymentShortCode: z.string(),
+});
+
+export const action: ActionFunction = async ({ request, params }) => {
+  const userId = await requireUserId(request);
+  const { projectId, deploymentShortCode } = ParamSchema.parse(params);
+
+  console.log("projectId", projectId);
+  console.log("deploymentShortCode", deploymentShortCode);
+
+  const formData = await request.formData();
+  const submission = parse(formData, { schema: rollbackSchema });
+
+  if (!submission.value) {
+    return json(submission);
+  }
+
+  try {
+    const project = await prisma.project.findUnique({
+      where: {
+        id: projectId,
+        organization: {
+          members: {
+            some: {
+              userId,
+            },
+          },
+        },
+      },
+    });
+
+    if (!project) {
+      return redirectWithErrorMessage(submission.value.redirectUrl, request, "Project not found");
+    }
+
+    const deployment = await prisma.workerDeployment.findUnique({
+      where: {
+        projectId_shortCode: {
+          projectId: project.id,
+          shortCode: deploymentShortCode,
+        },
+      },
+    });
+
+    if (!deployment) {
+      return redirectWithErrorMessage(
+        submission.value.redirectUrl,
+        request,
+        "Deployment not found"
+      );
+    }
+
+    const rollbackService = new RollbackDeploymentService();
+    await rollbackService.call(deployment);
+
+    return redirectWithSuccessMessage(
+      submission.value.redirectUrl,
+      request,
+      "Rolled back deployment"
+    );
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Failed to roll back deployment", {
+        error: {
+          name: error.name,
+          message: error.message,
+          stack: error.stack,
+        },
+      });
+      submission.error = { runParam: error.message };
+      return json(submission);
+    } else {
+      logger.error("Failed to roll back deployment", { error });
+      submission.error = { runParam: JSON.stringify(error) };
+      return json(submission);
+    }
+  }
+};

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

@@ -1,5 +1,5 @@
 import { BillingClient, SetPlanBody } from "@trigger.dev/billing";
-import { PrismaClient, prisma } from "~/db.server";
+import { $replica, PrismaClient, PrismaReplicaClient, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { organizationBillingPath } from "~/utils/pathBuilder";
@@ -7,9 +7,11 @@ import { organizationBillingPath } from "~/utils/pathBuilder";
 export class BillingService {
   #billingClient: BillingClient | undefined;
   #prismaClient: PrismaClient;
+  #replica: PrismaReplicaClient;
 
-  constructor(isManagedCloud: boolean, prismaClient: PrismaClient = prisma) {
+  constructor(isManagedCloud: boolean, prismaClient: PrismaClient = prisma, replica: PrismaReplicaClient = $replica) {
     this.#prismaClient = prismaClient;
+    this.#replica = replica;
     if (isManagedCloud && process.env.BILLING_API_URL && process.env.BILLING_API_KEY) {
       this.#billingClient = new BillingClient({
         url: process.env.BILLING_API_URL,
@@ -35,7 +37,7 @@ export class BillingService {
       firstDayOfNextMonth.setMonth(firstDayOfNextMonth.getMonth() + 1);
       firstDayOfNextMonth.setHours(0, 0, 0, 0);
 
-      const currentRunCount = await this.#prismaClient.jobRun.count({
+      const currentRunCount = await this.#replica.jobRun.count({
         where: {
           organizationId: orgId,
           createdAt: {

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

@@ -6,6 +6,12 @@ import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { workerQueue } from "../worker.server";
 
+class AlreadyDeliveredError extends Error {
+  constructor() {
+    super("Event already delivered");
+  }
+}
+
 export class DeliverEventService {
   #prismaClient: PrismaClientOrTransaction;
 
@@ -14,81 +20,111 @@ export class DeliverEventService {
   }
 
   public async call(id: string) {
-    await $transaction(
-      this.#prismaClient,
-      async (tx) => {
-        const eventRecord = await tx.eventRecord.findUniqueOrThrow({
-          where: {
-            id,
-          },
-          include: {
-            environment: {
-              include: {
-                organization: true,
-                project: true,
+    try {
+      await $transaction(
+        this.#prismaClient,
+        async (tx) => {
+          const eventRecord = await tx.eventRecord.findUniqueOrThrow({
+            where: {
+              id,
+            },
+            include: {
+              environment: {
+                include: {
+                  organization: true,
+                  project: true,
+                },
               },
             },
-          },
-        });
+          });
 
-        const possibleEventDispatchers = await tx.eventDispatcher.findMany({
-          where: {
-            environmentId: eventRecord.environmentId,
-            event: {
-              has: eventRecord.name,
+          if (eventRecord.deliveredAt) {
+            logger.debug("Event already delivered", {
+              eventRecord: eventRecord.id,
+            });
+
+            return;
+          }
+
+          const possibleEventDispatchers = await tx.eventDispatcher.findMany({
+            where: {
+              environmentId: eventRecord.environmentId,
+              event: {
+                has: eventRecord.name,
+              },
+              source: eventRecord.source,
+              enabled: true,
+              manual: false,
             },
-            source: eventRecord.source,
-            enabled: true,
-            manual: false,
-          },
-        });
+          });
 
-        logger.debug("Found possible event dispatchers", {
-          possibleEventDispatchers,
-          eventRecord: eventRecord.id,
-        });
-
-        const matchingEventDispatchers = possibleEventDispatchers.filter((eventDispatcher) =>
-          this.#evaluateEventRule(eventDispatcher, eventRecord)
-        );
-
-        if (matchingEventDispatchers.length === 0) {
-          logger.debug("No matching event dispatchers", {
+          logger.debug("Found possible event dispatchers", {
+            possibleEventDispatchers,
             eventRecord: eventRecord.id,
           });
 
-          return;
-        }
+          const matchingEventDispatchers = possibleEventDispatchers.filter((eventDispatcher) =>
+            this.#evaluateEventRule(eventDispatcher, eventRecord)
+          );
 
-        logger.debug("Found matching event dispatchers", {
-          matchingEventDispatchers,
-          eventRecord: eventRecord.id,
-        });
+          if (matchingEventDispatchers.length === 0) {
+            logger.debug("No matching event dispatchers", {
+              eventRecord: eventRecord.id,
+            });
 
-        await Promise.all(
-          matchingEventDispatchers.map((eventDispatcher) =>
-            workerQueue.enqueue(
-              "events.invokeDispatcher",
-              {
-                id: eventDispatcher.id,
-                eventRecordId: eventRecord.id,
-              },
-              { tx }
+            return;
+          }
+
+          logger.debug("Found matching event dispatchers", {
+            matchingEventDispatchers,
+            eventRecord: eventRecord.id,
+          });
+
+          await Promise.all(
+            matchingEventDispatchers.map((eventDispatcher) =>
+              workerQueue.enqueue(
+                "events.invokeDispatcher",
+                {
+                  id: eventDispatcher.id,
+                  eventRecordId: eventRecord.id,
+                },
+                { tx }
+              )
             )
-          )
-        );
+          );
 
-        await tx.eventRecord.update({
-          where: {
-            id: eventRecord.id,
-          },
-          data: {
-            deliveredAt: new Date(),
-          },
+          // Optimistically mark the event as delivered
+          const lockedRecord = await tx.eventRecord.updateMany({
+            where: {
+              id: eventRecord.id,
+              deliveredAt: null,
+            },
+            data: {
+              deliveredAt: new Date(),
+            },
+          });
+
+          if (lockedRecord.count === 0) {
+            //this means we've already delivered it, because there were no records with deliveredAt = null
+            //by throwing it will rollback the transaction, stopping the queue from processing the event again
+            throw new AlreadyDeliveredError();
+          }
+        },
+        { timeout: 10000 }
+      );
+    }
+    catch (error) {
+      if (error instanceof AlreadyDeliveredError) {
+        logger.debug("Event already delivered, AlreadyDeliveredError", {
+          eventRecord: id,
         });
-      },
-      { timeout: 10000 }
-    );
+
+        //we swallow the error because we don't want to retry
+        return;
+      }
+
+      throw error;
+    }
   }
 
   #evaluateEventRule(dispatcher: EventDispatcher, eventRecord: EventRecord): boolean {

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: {
@@ -92,6 +110,14 @@ export class IngestSendEvent {
           },
         });
 
+        if (existingEventLog?.deliveredAt) {
+          logger.debug("Event already delivered", {
+            eventRecordId: existingEventLog.id,
+            deliveredAt: existingEventLog.deliveredAt,
+          });
+          return existingEventLog;
+        }
+
         const eventLog = await (existingEventLog
           ? this.updateEvent({ tx, existingEventLog, reqEvent: event, deliverAt })
           : this.createEvent({
@@ -106,6 +132,32 @@ export class IngestSendEvent {
 
         return eventLog;
       });
+
+      if (!createdEvent) return;
+
+      if (createdEvent.deliveredAt) {
+        logger.debug("Event already delivered", {
+          eventRecordId: createdEvent.id,
+          deliveredAt: createdEvent.deliveredAt,
+        });
+        //return the event if it was already delivered, don't enqueue it again
+        return createdEvent;
+      }
+
+      //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 +203,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, eventLog);
-
     return eventLog;
   }
 
@@ -177,8 +227,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, updatedEventLog);
-
     return updatedEventLog;
   }
 

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

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

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

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

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

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

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

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

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

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

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

@@ -16,20 +16,17 @@ import {
   supportsFeature,
 } from "@trigger.dev/core";
 import { BloomFilter } from "@trigger.dev/core-backend";
-import {
-  ConcurrencyLimitGroup,
-  JobRun,
-  JobVersion,
-  RuntimeEnvironment,
-} from "@trigger.dev/database";
+import { ConcurrencyLimitGroup, Job, JobRun, JobVersion } from "@trigger.dev/database";
 import { generateErrorMessage } from "zod-error";
 import { eventRecordToApiJson } from "~/api.server";
 import {
+  MAX_JOB_RUN_EXECUTION_COUNT,
   MAX_RUN_CHUNK_EXECUTION_LIMIT,
   MAX_RUN_YIELDED_EXECUTIONS,
   RUN_CHUNK_EXECUTION_BUFFER,
 } from "~/consts";
 import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { env } from "~/env.server";
 import { detectResponseIsTimeout } from "~/models/endpoint.server";
 import { isRunCompleted } from "~/models/jobRun.server";
 import { resolveRunConnections } from "~/models/runConnection.server";
@@ -37,18 +34,19 @@ 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];
+type FoundTask = NonNullable<Awaited<ReturnType<typeof getCompletedTasksForRun>>>[number];
 
 // We need to limit the cached tasks to not be too large >3.5MB when serialized
 const TOTAL_CACHED_TASK_BYTE_LIMIT = 3500000;
@@ -83,6 +81,14 @@ export class PerformRunExecutionV3Service {
   }
 
   public async call(input: PerformRunExecutionV3Input, driftInMs: number = 0) {
+    logger.debug("PerformRunExecutionV3Service.call", { input, driftInMs });
+
+    if (Array.isArray(input.id)) {
+      logger.error("PerformRunExecutionV3Service.call: input.id is an array", { input });
+
+      throw new Error("input.id must be a string");
+    }
+
     const run = await findRun(this.#prismaClient, input.id);
 
     if (!run) {
@@ -95,9 +101,10 @@ export class PerformRunExecutionV3Service {
   static async enqueue(
     run: JobRun & {
       version: JobVersion & {
-        environment: RuntimeEnvironment;
+        environment: AuthenticatedEnvironment;
         concurrencyLimitGroup?: ConcurrencyLimitGroup | null;
       };
+      job: Job;
     },
     priority: RunExecutionPriority,
     tx: PrismaClientOrTransaction,
@@ -106,27 +113,49 @@ export class PerformRunExecutionV3Service {
       skipRetrying?: boolean;
     } = {}
   ) {
-    return await executionWorker.enqueue(
-      "performRunExecutionV3",
-      {
-        id: run.id,
-        reason: "EXECUTE_JOB",
-      },
-      {
-        tx,
-        runAt: options.runAt,
-        jobKey: `job_run:EXECUTE_JOB:${run.id}`,
-        maxAttempts: options.skipRetrying ? env.DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS : undefined,
-        flags: executionRateLimiter?.flagsForRun(run, run.version) ?? [],
-        priority: priority === "initial" ? 0 : -1,
+    if (marqsv2 && run.version.environment.organization.v2MarqsEnabled) {
+      let queue = `job/${run.job.slug}`;
+
+      if (run.version.concurrencyLimitGroup) {
+        queue = `group/${run.version.concurrencyLimitGroup.name}`;
       }
-    );
+
+      const runAt =
+        priority === "initial" ? options.runAt ?? new Date() : run.startedAt ?? run.createdAt;
+
+      await marqsv2.enqueueMessage(
+        run.version.environment,
+        queue,
+        run.id,
+        { runId: run.id, attempt: 1 },
+        undefined,
+        runAt.getTime()
+      );
+    } else {
+      return await executionWorker.enqueue(
+        "performRunExecutionV3",
+        {
+          id: run.id,
+          reason: "EXECUTE_JOB",
+        },
+        {
+          tx,
+          runAt: options.runAt,
+          jobKey: `job_run:EXECUTE_JOB:${run.id}`,
+          maxAttempts: options.skipRetrying ? env.DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS : undefined,
+          flags: executionRateLimiter?.flagsForRun(run, run.version) ?? [],
+          priority: priority === "initial" ? 0 : -1,
+        }
+      );
+    }
   }
 
   static async dequeue(run: JobRun, tx: PrismaClientOrTransaction) {
     await executionWorker.dequeue(`job_run:EXECUTE_JOB:${run.id}`, {
       tx,
     });
+
+    await marqsv2?.acknowledgeMessage(run.id);
   }
 
   async #executeJob(run: FoundRun, input: PerformRunExecutionV3Input, driftInMs: number = 0) {
@@ -135,12 +164,58 @@ export class PerformRunExecutionV3Service {
         return;
       }
 
+      if (!run.organization.runsEnabled) {
+        return await this.#failRunExecution(this.#prismaClient, run, {
+          message: `Unable to execute run.`,
+        });
+      }
+
       if (!run.endpoint.url) {
         return await this.#failRunExecution(this.#prismaClient, run, {
           message: `Endpoint has no URL set`,
         });
       }
 
+      if (run.version.status === "DISABLED") {
+        return await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Job version ${run.version.version} is disabled, aborting run.`,
+          },
+          "ABORTED"
+        );
+      }
+
+      // If the execution duration is greater than the maximum execution time, we need to fail the run
+      if (run.executionDuration >= run.organization.maximumExecutionTimePerRunInMs) {
+        await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Execution timed out after ${
+              run.organization.maximumExecutionTimePerRunInMs / 1000
+            } seconds`,
+          },
+          "TIMED_OUT",
+          0
+        );
+        return;
+      }
+
+      if (run.executionCount >= MAX_JOB_RUN_EXECUTION_COUNT) {
+        await this.#failRunExecution(
+          this.#prismaClient,
+          run,
+          {
+            message: `Execution timed out after ${run.executionCount} executions`,
+          },
+          "TIMED_OUT",
+          0
+        );
+        return;
+      }
+
       const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
       const event = eventRecordToApiJson(run.event);
 
@@ -154,11 +229,14 @@ export class PerformRunExecutionV3Service {
         });
       }
 
+      const taskCount = await getTaskCountForRun(this.#prismaClient, run.id);
+      const tasks = await getCompletedTasksForRun(this.#prismaClient, run.id);
+
       const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext);
 
       const executionBody = await this.#createExecutionBody(
         run,
-        run.tasks,
+        tasks,
         startedAt,
         false,
         connections.auth,
@@ -207,6 +285,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: {
@@ -347,7 +429,8 @@ export class PerformRunExecutionV3Service {
               this.#prismaClient,
               run,
               input,
-              durationInMs
+              durationInMs,
+              taskCount
             );
           } else {
             return await this.#failRunExecutionWithRetry(
@@ -994,7 +1077,8 @@ export class PerformRunExecutionV3Service {
     prisma: PrismaClientOrTransaction,
     run: FoundRun,
     input: PerformRunExecutionV3Input,
-    durationInMs: number
+    durationInMs: number,
+    existingTaskCount: number
   ) {
     await $transaction(prisma, async (tx) => {
       const executionDuration = run.executionDuration + durationInMs;
@@ -1015,31 +1099,25 @@ export class PerformRunExecutionV3Service {
         return;
       }
 
-      const runWithLatestTask = await tx.jobRun.findUniqueOrThrow({
-        where: {
-          id: run.id,
-        },
-        select: {
-          tasks: {
-            select: {
-              id: true,
-              name: true,
-              status: true,
-              displayKey: true,
-            },
-            take: 1,
-            orderBy: { createdAt: "desc" },
-          },
-          _count: {
-            select: {
-              tasks: true,
-            },
-          },
-        },
-      });
+      const newTaskCount = await getTaskCountForRun(tx, run.id);
 
-      if (runWithLatestTask._count.tasks === run._count.tasks) {
-        const latestTask = runWithLatestTask.tasks[0];
+      if (newTaskCount === existingTaskCount) {
+        const latestTask = await tx.task.findFirst({
+          select: {
+            id: true,
+            name: true,
+            status: true,
+            displayKey: true,
+          },
+          where: {
+            runId: run.id,
+            status: "RUNNING",
+          },
+          orderBy: {
+            createdAt: "desc",
+          },
+          take: 1,
+        });
 
         const cause =
           latestTask?.status === "RUNNING"
@@ -1183,6 +1261,35 @@ function prepareNoOpTasksBloomFilter(possibleTasks: FoundTask[]): string {
   return filter.serialize();
 }
 
+async function getTaskCountForRun(prisma: PrismaClientOrTransaction, runId: string) {
+  return await prisma.task.count({
+    where: {
+      runId,
+    },
+  });
+}
+
+async function getCompletedTasksForRun(prisma: PrismaClientOrTransaction, runId: string) {
+  return await prisma.task.findMany({
+    where: {
+      runId,
+      status: "COMPLETED",
+    },
+    select: {
+      id: true,
+      idempotencyKey: true,
+      status: true,
+      noop: true,
+      output: true,
+      outputIsUndefined: true,
+      parentId: true,
+    },
+    orderBy: {
+      id: "asc",
+    },
+  });
+}
+
 async function findRun(prisma: PrismaClientOrTransaction, id: string) {
   return await prisma.jobRun.findUnique({
     where: { id },
@@ -1207,25 +1314,6 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
           },
         },
       },
-      tasks: {
-        where: {
-          status: {
-            in: ["COMPLETED"],
-          },
-        },
-        select: {
-          id: true,
-          idempotencyKey: true,
-          status: true,
-          noop: true,
-          output: true,
-          outputIsUndefined: true,
-          parentId: true,
-        },
-        orderBy: {
-          id: "asc",
-        },
-      },
       event: true,
       version: {
         include: {
@@ -1238,11 +1326,6 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
           recipientMethod: "ENDPOINT",
         },
       },
-      _count: {
-        select: {
-          tasks: true,
-        },
-      },
     },
   });
 }

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

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

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

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

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

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

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

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

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

@@ -7,6 +7,7 @@ import type { Organization } from "~/models/organization.server";
 import type { Project } from "~/models/project.server";
 import type { User } from "~/models/user.server";
 import { singleton } from "~/utils/singleton";
+import { loopsClient } from "./loops.server";
 
 type Options = {
   postHogApiKey?: string;
@@ -21,6 +22,11 @@ class Telemetry {
   #triggerClient: TriggerClient | undefined = undefined;
 
   constructor({ postHogApiKey, trigger }: Options) {
+    if (env.TRIGGER_TELEMETRY_DISABLED !== undefined) {
+      console.log("📉 Telemetry disabled");
+      return;
+    }
+
     if (postHogApiKey) {
       this.#posthogClient = new PostHog(postHogApiKey, { host: "https://eu.posthog.com" });
     } else {
@@ -39,18 +45,19 @@ class Telemetry {
 
   user = {
     identify: ({ user, isNewUser }: { user: User; isNewUser: boolean }) => {
-      if (this.#posthogClient === undefined) return;
-      this.#posthogClient.identify({
-        distinctId: user.id,
-        properties: {
-          email: user.email,
-          name: user.name,
-          authenticationMethod: user.authenticationMethod,
-          admin: user.admin,
-          createdAt: user.createdAt,
-          isNewUser,
-        },
-      });
+      if (this.#posthogClient) {
+        this.#posthogClient.identify({
+          distinctId: user.id,
+          properties: {
+            email: user.email,
+            name: user.name,
+            authenticationMethod: user.authenticationMethod,
+            admin: user.admin,
+            createdAt: user.createdAt,
+            isNewUser,
+          },
+        });
+      }
       if (isNewUser) {
         this.#capture({
           userId: user.id,
@@ -64,6 +71,12 @@ class Telemetry {
           },
         });
 
+        loopsClient?.userCreated({
+          userId: user.id,
+          email: user.email,
+          name: user.name,
+        });
+
         this.#triggerClient?.sendEvent({
           name: "user.created",
           payload: {

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

@@ -4,7 +4,21 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { ZodWorker } from "~/platform/zodWorker.server";
+import { eventRepository } from "~/v3/eventRepository.server";
+import { RequeueTaskRunService } from "~/v3/requeueTaskRun.server";
+import { DeliverAlertService } from "~/v3/services/alerts/deliverAlert.server";
+import { PerformDeploymentAlertsService } from "~/v3/services/alerts/performDeploymentAlerts.server";
+import { PerformTaskAttemptAlertsService } from "~/v3/services/alerts/performTaskAttemptAlerts.server";
+import { PerformBulkActionService } from "~/v3/services/bulk/performBulkAction.server";
+import { ExecuteTasksWaitingForDeployService } from "~/v3/services/executeTasksWaitingForDeploy";
 import { IndexDeploymentService } from "~/v3/services/indexDeployment.server";
+import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
+import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
+import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
+import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
+import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
+import { TriggerScheduledTaskService } from "~/v3/services/triggerScheduledTask.server";
+import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
 import { ExpireDispatcherService } from "./dispatchers/expireDispatcher.server";
 import { InvokeEphemeralDispatcherService } from "./dispatchers/invokeEphemeralEventDispatcher.server";
 import { sendEmail } from "./email.server";
@@ -30,20 +44,8 @@ import { DeliverWebhookRequestService } from "./sources/deliverWebhookRequest.se
 import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
 import { ProcessCallbackTimeoutService } from "./tasks/processCallbackTimeout.server";
 import { ResumeTaskService } from "./tasks/resumeTask.server";
-import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
-import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
-import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
-import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
-import { eventRepository } from "~/v3/eventRepository.server";
-import { ExecuteTasksWaitingForDeployService } from "~/v3/services/executeTasksWaitingForDeploy";
-import { TriggerScheduledTaskService } from "~/v3/services/triggerScheduledTask.server";
-import { PerformTaskAttemptAlertsService } from "~/v3/services/alerts/performTaskAttemptAlerts.server";
-import { DeliverAlertService } from "~/v3/services/alerts/deliverAlert.server";
-import { PerformDeploymentAlertsService } from "~/v3/services/alerts/performDeploymentAlerts.server";
-import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
-import { PerformBulkActionService } from "~/v3/services/bulk/performBulkAction.server";
-import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
-import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
+import { MarqsConcurrencyMonitor } from "~/v3/marqs/concurrencyMonitor.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -158,6 +160,15 @@ const workerCatalog = {
   "v3.performBulkActionItem": z.object({
     bulkActionItemId: z.string(),
   }),
+  "v3.requeueTaskRun": z.object({
+    runId: z.string(),
+  }),
+  "v3.retryAttempt": z.object({
+    runId: z.string(),
+  }),
+  "v2.requeueMessage": z.object({
+    runId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -247,14 +258,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();
 
@@ -288,6 +299,19 @@ function getWorkerQueue() {
           await eventRepository.truncateEvents();
         },
       },
+      "marqs.v3.queueConcurrencyMonitor": {
+        // run every 5 minutes
+        match: "*/5 * * * *",
+        handler: async (payload, job, helpers) => {
+          await MarqsConcurrencyMonitor.initiateV3Monitoring(helpers.abortSignal);
+        },
+      },
+      "marqs.v2.queueConcurrencyMonitor": {
+        match: "*/5 * * * *", // run every 5 minutes
+        handler: async (payload, job, helpers) => {
+          await MarqsConcurrencyMonitor.initiateV2Monitoring(helpers.abortSignal);
+        },
+      },
     },
     tasks: {
       "events.invokeDispatcher": {
@@ -301,7 +325,7 @@ function getWorkerQueue() {
       },
       "events.deliverScheduled": {
         priority: 0, // smaller number = higher priority
-        maxAttempts: 5,
+        maxAttempts: 8,
         handler: async ({ id, payload }, job) => {
           const service = new DeliverScheduledEventService();
 
@@ -327,7 +351,7 @@ function getWorkerQueue() {
         },
       },
       activateSource: {
-        priority: 10, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, graphileJob) => {
           const service = new ActivateSourceService();
@@ -353,9 +377,8 @@ function getWorkerQueue() {
         },
       },
       deliverHttpSourceRequest: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 14,
-        queueName: (payload) => `sources:${payload.id}`,
         handler: async (payload, job) => {
           const service = new DeliverHttpSourceRequestService();
 
@@ -363,9 +386,8 @@ function getWorkerQueue() {
         },
       },
       deliverWebhookRequest: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 14,
-        queueName: (payload) => `webhooks:${payload.id}`,
         handler: async (payload, job) => {
           const service = new DeliverWebhookRequestService();
 
@@ -391,14 +413,14 @@ function getWorkerQueue() {
         },
       },
       scheduleEmail: {
-        priority: 100,
+        priority: 0,
         maxAttempts: 3,
         handler: async (payload, job) => {
           await sendEmail(payload);
         },
       },
       indexEndpoint: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           const service = new IndexEndpointService();
@@ -406,7 +428,7 @@ function getWorkerQueue() {
         },
       },
       performEndpointIndexing: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           const service = new PerformEndpointIndexService();
@@ -423,7 +445,7 @@ function getWorkerQueue() {
         },
       },
       refreshOAuthToken: {
-        priority: 8, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 7,
         handler: async (payload, job) => {
           await integrationAuthRepository.refreshConnection({
@@ -432,7 +454,7 @@ function getWorkerQueue() {
         },
       },
       probeEndpoint: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 1,
         handler: async (payload, job) => {
           const service = new ProbeEndpointService();
@@ -447,7 +469,7 @@ function getWorkerQueue() {
         },
       },
       deliverRunSubscriptions: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 5,
         handler: async (payload, job) => {
           const service = new DeliverRunSubscriptionsService();
@@ -456,7 +478,7 @@ function getWorkerQueue() {
         },
       },
       deliverRunSubscription: {
-        priority: 1, // smaller number = higher priority
+        priority: 0, // smaller number = higher priority
         maxAttempts: 13,
         handler: async (payload, job) => {
           const service = new DeliverRunSubscriptionService();
@@ -474,7 +496,7 @@ function getWorkerQueue() {
         },
       },
       expireDispatcher: {
-        priority: 10,
+        priority: 0,
         maxAttempts: 3,
         handler: async (payload) => {
           const service = new ExpireDispatcherService();
@@ -600,6 +622,33 @@ function getWorkerQueue() {
           await service.performBulkActionItem(payload.bulkActionItemId);
         },
       },
+      "v3.requeueTaskRun": {
+        priority: 0,
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new RequeueTaskRunService();
+
+          await service.call(payload.runId);
+        },
+      },
+      "v3.retryAttempt": {
+        priority: 0,
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new RetryAttemptService();
+
+          return await service.call(payload.runId);
+        },
+      },
+      "v2.requeueMessage": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new RequeueV2Message();
+
+          await service.call(payload.runId);
+        },
+      },
     },
   });
 }
@@ -614,7 +663,7 @@ function getExecutionWorkerQueue() {
       pollInterval: env.EXECUTION_WORKER_POLL_INTERVAL,
       noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
       schema: env.WORKER_SCHEMA,
-      maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY,
+      maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY + 1,
     },
     shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: executionWorkerCatalog,
@@ -668,7 +717,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,40 @@ 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;
+  });
+}
+
+export function onlyDevEnvironments<T extends FilterableEnvironment>(environments: T[]): T[] {
+  return environments.filter((e) => e.type === "DEVELOPMENT");
+}
+
+export function exceptDevEnvironments<T extends FilterableEnvironment>(environments: T[]): T[] {
+  return environments.filter((e) => e.type !== "DEVELOPMENT");
+}

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

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

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

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

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

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

apps/webapp/app/v3/friendlyIdentifiers.ts

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

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

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

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

@@ -0,0 +1,216 @@
+import { Logger } from "@trigger.dev/core-backend";
+import { Redis } from "ioredis";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { MarQS, marqs as marqsv3 } from "./index.server";
+import { env } from "~/env.server";
+import { marqsv2 } from "./v2.server";
+
+export type MarqsConcurrencyMonitorOptions = {
+  dryRun?: boolean;
+  abortSignal?: AbortSignal;
+};
+
+export interface MarqsConcurrencyResolveCompletedRunsCallback {
+  (candidateRunIds: string[]): Promise<Array<{ id: string }>>;
+}
+
+export class MarqsConcurrencyMonitor {
+  private _logger: Logger;
+
+  constructor(
+    private marqs: MarQS,
+    private callback: MarqsConcurrencyResolveCompletedRunsCallback,
+    private options: MarqsConcurrencyMonitorOptions = {}
+  ) {
+    this._logger = logger.child({
+      component: "marqs",
+      operation: "concurrencyMonitor",
+      dryRun: this.dryRun,
+      marqs: marqs.name,
+    });
+  }
+
+  get dryRun() {
+    return typeof this.options.dryRun === "boolean" ? this.options.dryRun : false;
+  }
+
+  get keys() {
+    return this.marqs.keys;
+  }
+
+  get signal() {
+    return this.options.abortSignal;
+  }
+
+  public async call() {
+    this._logger.debug("[MarqsConcurrencyMonitor] Initiating monitoring");
+
+    const stats = {
+      streamCallbacks: 0,
+      processedKeys: 0,
+    };
+
+    const { stream, redis } = this.marqs.queueConcurrencyScanStream(10, () => {
+      this._logger.debug("[MarqsConcurrencyMonitor] stream closed", {
+        stats,
+      });
+    });
+
+    stream.on("data", async (keys) => {
+      stream.pause();
+
+      if (this.signal?.aborted) {
+        stream.destroy();
+        return;
+      }
+
+      stats.streamCallbacks++;
+
+      const uniqueKeys = Array.from(new Set<string>(keys));
+
+      if (uniqueKeys.length === 0) {
+        stream.resume();
+        return;
+      }
+
+      this._logger.debug("[MarqsConcurrencyMonitor] correcting queues concurrency", {
+        keys: uniqueKeys,
+      });
+
+      stats.processedKeys += uniqueKeys.length;
+
+      await Promise.all(uniqueKeys.map((key) => this.#processKey(key, redis))).finally(() => {
+        stream.resume();
+      });
+    });
+  }
+
+  async #processKey(key: string, redis: Redis) {
+    key = this.keys.stripKeyPrefix(key);
+    const orgKey = this.keys.orgCurrentConcurrencyKeyFromQueue(key);
+    const envKey = this.keys.envCurrentConcurrencyKeyFromQueue(key);
+
+    // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
+    const runIds = await redis.sunion(orgKey, envKey, key);
+
+    if (runIds.length === 0) {
+      return;
+    }
+
+    const perfNow = performance.now();
+
+    const completeRuns = await this.callback(runIds);
+
+    const durationMs = performance.now() - perfNow;
+
+    const completedRunIds = completeRuns.map((run) => run.id);
+
+    if (completedRunIds.length === 0) {
+      this._logger.debug("[MarqsConcurrencyMonitor] no completed runs found", {
+        key,
+        orgKey,
+        envKey,
+        runIds,
+        durationMs,
+      });
+
+      return;
+    }
+
+    this._logger.debug("[MarqsConcurrencyMonitor] removing completed runs from queue", {
+      key,
+      orgKey,
+      envKey,
+      completedRunIds,
+      durationMs,
+    });
+
+    if (this.dryRun) {
+      return;
+    }
+
+    const pipeline = redis.pipeline();
+
+    pipeline.srem(key, ...completedRunIds);
+    pipeline.srem(orgKey, ...completedRunIds);
+    pipeline.srem(envKey, ...completedRunIds);
+
+    try {
+      await pipeline.exec();
+    } catch (e) {
+      this._logger.error("[MarqsConcurrencyMonitor] error removing completed runs from queue", {
+        key,
+        orgKey,
+        envKey,
+        completedRunIds,
+        error: e,
+      });
+    }
+  }
+
+  static async initiateV3Monitoring(abortSignal?: AbortSignal) {
+    if (!marqsv3) {
+      return;
+    }
+
+    const instance = new MarqsConcurrencyMonitor(
+      marqsv3,
+      (runIds) =>
+        prisma.taskRun.findMany({
+          select: { id: true },
+          where: {
+            id: {
+              in: runIds,
+            },
+            status: {
+              in: [
+                "CANCELED",
+                "COMPLETED_SUCCESSFULLY",
+                "COMPLETED_WITH_ERRORS",
+                "CRASHED",
+                "SYSTEM_FAILURE",
+                "INTERRUPTED",
+              ],
+            },
+          },
+        }),
+      { dryRun: env.V3_MARQS_CONCURRENCY_MONITOR_ENABLED === "0", abortSignal }
+    );
+
+    await instance.call();
+  }
+
+  static async initiateV2Monitoring(abortSignal?: AbortSignal) {
+    if (!marqsv2) {
+      return;
+    }
+
+    const instance = new MarqsConcurrencyMonitor(
+      marqsv2,
+      (runIds) =>
+        prisma.jobRun.findMany({
+          select: { id: true },
+          where: {
+            id: {
+              in: runIds,
+            },
+            status: {
+              in: [
+                "CANCELED",
+                "SUCCESS",
+                "FAILURE",
+                "TIMED_OUT",
+                "ABORTED",
+                "CANCELED",
+                "INVALID_PAYLOAD",
+              ],
+            },
+          },
+        }),
+      { dryRun: env.V2_MARQS_CONCURRENCY_MONITOR_ENABLED === "0", abortSignal }
+    );
+
+    await instance.call();
+  }
+}

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

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

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

@@ -1,4 +1,12 @@
-import { Span, SpanKind, SpanOptions, context, propagation, trace } from "@opentelemetry/api";
+import {
+  Span,
+  SpanKind,
+  SpanOptions,
+  Tracer,
+  context,
+  propagation,
+  trace,
+} from "@opentelemetry/api";
 import {
   SEMATTRS_MESSAGE_ID,
   SEMATTRS_MESSAGING_OPERATION,
@@ -13,21 +21,20 @@ import { singleton } from "~/utils/singleton";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";
 import { AsyncWorker } from "./asyncWorker.server";
 import { MarQSShortKeyProducer } from "./marqsKeyProducer.server";
-import { SimpleWeightedChoiceStrategy } from "./priorityStrategy.server";
+import { SimpleWeightedChoiceStrategy } from "./simpleWeightedPriorityStrategy.server";
 import {
   MarQSKeyProducer,
   MarQSQueuePriorityStrategy,
   MessagePayload,
   QueueCapacities,
   QueueRange,
+  VisibilityTimeoutStrategy,
 } from "./types";
-
-const tracer = trace.getTracer("marqs");
+import { V3VisibilityTimeout } from "./v3VisibilityTimeout.server";
 
 const KEY_PREFIX = "marqs:";
 
 const constants = {
-  SHARED_QUEUE: "sharedQueue",
   MESSAGE_VISIBILITY_TIMEOUT_QUEUE: "msgVisibilityTimeout",
 } as const;
 
@@ -39,6 +46,8 @@ const SemanticAttributes = {
 };
 
 export type MarQSOptions = {
+  name: string;
+  tracer: Tracer;
   redis: RedisOptions;
   defaultEnvConcurrency: number;
   defaultOrgConcurrency: number;
@@ -48,6 +57,9 @@ export type MarQSOptions = {
   keysProducer: MarQSKeyProducer;
   queuePriorityStrategy: MarQSQueuePriorityStrategy;
   envQueuePriorityStrategy: MarQSQueuePriorityStrategy;
+  visibilityTimeoutStrategy: VisibilityTimeoutStrategy;
+  enableRebalancing?: boolean;
+  verbose?: boolean;
 };
 
 /**
@@ -66,12 +78,18 @@ export class MarQS {
     this.keys = options.keysProducer;
     this.queuePriorityStrategy = options.queuePriorityStrategy;
 
-    // Spawn options.workers workers to requeue visible messages
-    this.#startRequeuingWorkers();
     this.#startRebalanceWorkers();
     this.#registerCommands();
   }
 
+  get name() {
+    return this.options.name;
+  }
+
+  get tracer() {
+    return this.options.tracer;
+  }
+
   public async updateQueueConcurrencyLimits(
     env: AuthenticatedEnvironment,
     queue: string,
@@ -80,6 +98,10 @@ export class MarQS {
     return this.redis.set(this.keys.queueConcurrencyLimitKey(env, queue), concurrency);
   }
 
+  public async removeQueueConcurrencyLimits(env: AuthenticatedEnvironment, queue: string) {
+    return this.redis.del(this.keys.queueConcurrencyLimitKey(env, queue));
+  }
+
   public async updateEnvConcurrencyLimits(env: AuthenticatedEnvironment) {
     await this.#callUpdateGlobalConcurrencyLimits({
       envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
@@ -209,7 +231,8 @@ export class MarQS {
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
           this.options.envQueuePriorityStrategy,
-          (queue) => this.#calculateMessageQueueCapacities(queue)
+          (queue) => this.#calculateMessageQueueCapacities(queue),
+          env.id
         );
 
         if (!messageQueue) {
@@ -243,8 +266,9 @@ export class MarQS {
             [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
           });
         } else {
-          logger.error("Failed to read message, undoing the dequeueing of the message", {
+          logger.error(`Failed to read message, undoing the dequeueing of the message`, {
             messageData,
+            service: this.name,
           });
 
           await this.#callAcknowledgeMessage({
@@ -259,6 +283,11 @@ export class MarQS {
           });
         }
 
+        await this.options.visibilityTimeoutStrategy.heartbeat(
+          messageData.messageId,
+          this.visibilityTimeoutInMs
+        );
+
         return message;
       },
       {
@@ -272,20 +301,52 @@ export class MarQS {
     );
   }
 
+  public async getSharedQueueDetails() {
+    const parentQueue = this.keys.sharedQueueKey();
+
+    const { range } = await this.queuePriorityStrategy.nextCandidateSelection(
+      parentQueue,
+      "getSharedQueueDetails"
+    );
+    const queues = await this.#getChildQueuesWithScores(parentQueue, range);
+
+    const queuesWithScores = await this.#calculateQueueScores(queues, (queue) =>
+      this.#calculateMessageQueueCapacities(queue)
+    );
+
+    // We need to priority shuffle here to ensure all workers aren't just working on the highest priority queue
+    const choice = this.queuePriorityStrategy.chooseQueue(
+      queuesWithScores,
+      parentQueue,
+      "getSharedQueueDetails",
+      range
+    );
+
+    return {
+      selectionId: "getSharedQueueDetails",
+      queues,
+      queuesWithScores,
+      nextRange: range,
+      queueCount: queues.length,
+      queueChoice: choice,
+    };
+  }
+
   /**
    * Dequeue a message from the shared queue (this should be used in production environments)
    */
-  public async dequeueMessageInSharedQueue() {
+  public async dequeueMessageInSharedQueue(consumerId: string) {
     return this.#trace(
       "dequeueMessageInSharedQueue",
       async (span) => {
-        const parentQueue = constants.SHARED_QUEUE;
+        const parentQueue = this.keys.sharedQueueKey();
 
         // Read the parent queue for matching queues
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
           this.options.queuePriorityStrategy,
-          (queue) => this.#calculateMessageQueueCapacities(queue)
+          (queue) => this.#calculateMessageQueueCapacities(queue),
+          consumerId
         );
 
         if (!messageQueue) {
@@ -321,6 +382,11 @@ export class MarQS {
           });
         }
 
+        await this.options.visibilityTimeoutStrategy.heartbeat(
+          messageData.messageId,
+          this.visibilityTimeoutInMs
+        );
+
         return message;
       },
       {
@@ -350,6 +416,8 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callAcknowledgeMessage({
           parentQueue: message.parentQueue,
           messageKey: this.keys.messageKey(messageId),
@@ -415,6 +483,8 @@ export class MarQS {
           return;
         }
 
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callAcknowledgeMessage({
           parentQueue: oldMessage.parentQueue,
           messageKey: this.keys.messageKey(messageId),
@@ -444,27 +514,40 @@ export class MarQS {
     fn: (span: Span) => Promise<T>,
     options?: SpanOptions & { sampleRate?: number }
   ): Promise<T> {
-    return tracer.startActiveSpan(name, options ?? {}, async (span) => {
-      try {
-        return await fn(span);
-      } catch (e) {
-        if (e instanceof Error) {
-          span.recordException(e);
-        } else {
-          span.recordException(new Error(String(e)));
-        }
+    return this.tracer.startActiveSpan(
+      name,
+      {
+        ...options,
+        attributes: {
+          ...options?.attributes,
+        },
+      },
+      async (span) => {
+        try {
+          return await fn(span);
+        } catch (e) {
+          if (e instanceof Error) {
+            span.recordException(e);
+          } else {
+            span.recordException(new Error(String(e)));
+          }
 
-        throw e;
-      } finally {
-        span.end();
+          throw e;
+        } finally {
+          span.end();
+        }
       }
-    });
+    );
   }
 
   /**
    * Negative acknowledge a message, which will requeue the message
    */
-  public async nackMessage(messageId: string, retryAt: number = Date.now()) {
+  public async nackMessage(
+    messageId: string,
+    retryAt: number = Date.now(),
+    updates?: Record<string, unknown>
+  ) {
     return this.#trace(
       "nackMessage",
       async (span) => {
@@ -481,6 +564,12 @@ export class MarQS {
           [SemanticAttributes.PARENT_QUEUE]: message.parentQueue,
         });
 
+        if (updates) {
+          await this.replaceMessage(messageId, updates, retryAt, true);
+        }
+
+        await this.options.visibilityTimeoutStrategy.cancelHeartbeat(messageId);
+
         await this.#callNackMessage({
           messageKey: this.keys.messageKey(messageId),
           messageQueue: message.queue,
@@ -506,16 +595,11 @@ export class MarQS {
 
   // This should increment by the number of seconds, but with a max value of Date.now() + visibilityTimeoutInMs
   public async heartbeatMessage(messageId: string, seconds: number = 30) {
-    await this.#callHeartbeatMessage({
-      visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-      messageId,
-      milliseconds: seconds * 1000,
-      maxVisibilityTimeout: Date.now() + this.visibilityTimeoutInMs,
-    });
+    await this.options.visibilityTimeoutStrategy.heartbeat(messageId, seconds * 1000);
   }
 
   get visibilityTimeoutInMs() {
-    return this.options.visibilityTimeoutInMs ?? 300000;
+    return this.options.visibilityTimeoutInMs ?? 300000; // 5 minutes
   }
 
   async readMessage(messageId: string) {
@@ -531,9 +615,10 @@ export class MarQS {
         const message = MessagePayload.safeParse(JSON.parse(rawMessage));
 
         if (!message.success) {
-          logger.error("Failed to parse message", {
+          logger.error(`[${this.name}] Failed to parse message`, {
             messageId,
             error: message.error,
+            service: this.name,
           });
 
           return;
@@ -555,13 +640,15 @@ export class MarQS {
   async #getRandomQueueFromParentQueue(
     parentQueue: string,
     queuePriorityStrategy: MarQSQueuePriorityStrategy,
-    calculateCapacities: (queue: string) => Promise<QueueCapacities>
+    calculateCapacities: (queue: string) => Promise<QueueCapacities>,
+    consumerId: string
   ) {
     return this.#trace(
       "getRandomQueueFromParentQueue",
       async (span) => {
-        const { range, selectionId } = await queuePriorityStrategy.nextCandidateSelection(
-          parentQueue
+        const { range } = await queuePriorityStrategy.nextCandidateSelection(
+          parentQueue,
+          consumerId
         );
 
         const queues = await this.#getChildQueuesWithScores(parentQueue, range);
@@ -572,7 +659,8 @@ export class MarQS {
         const choice = this.queuePriorityStrategy.chooseQueue(
           queuesWithScores,
           parentQueue,
-          selectionId
+          consumerId,
+          range
         );
 
         span.setAttributes({
@@ -585,6 +673,28 @@ export class MarQS {
         span.setAttribute("nextRange.count", range.count);
         span.setAttribute("queueCount", queues.length);
 
+        if (this.options.verbose) {
+          if (typeof choice === "string") {
+            logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
+              queues,
+              queuesWithScores,
+              nextRange: range,
+              queueCount: queues.length,
+              queueChoice: choice,
+              consumerId,
+            });
+          } else {
+            logger.debug(`[${this.name}] getRandomQueueFromParentQueue`, {
+              queues,
+              queuesWithScores,
+              nextRange: range,
+              queueCount: queues.length,
+              noQueueChoice: true,
+              consumerId,
+            });
+          }
+        }
+
         if (typeof choice !== "string") {
           span.setAttribute("noQueueChoice", true);
 
@@ -619,6 +729,7 @@ export class MarQS {
           queue: queue.value,
           capacities: await calculateCapacities(queue.value),
           age: now - queue.score,
+          size: await this.redis.zcard(queue.value),
         };
       })
     );
@@ -663,6 +774,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);
@@ -673,64 +788,31 @@ export class MarQS {
     }
   }
 
-  #startRequeuingWorkers() {
-    // Start a new worker to requeue visible messages
-    for (let i = 0; i < this.options.workers; i++) {
-      const worker = new AsyncWorker(this.#requeueVisibleMessages.bind(this), 1000);
+  queueConcurrencyScanStream(count: number = 100, onEndCallback?: () => void) {
+    const pattern = this.keys.queueCurrentConcurrencyScanPattern();
 
-      this.#requeueingWorkers.push(worker);
+    logger.debug("Starting queue concurrency scan stream", {
+      pattern,
+      component: "marqs",
+      operation: "queueConcurrencyScanStream",
+      service: this.name,
+      count,
+    });
 
-      worker.start();
-    }
-  }
+    const redis = this.redis.duplicate();
 
-  async #requeueVisibleMessages() {
-    // Remove any of the messages from the timeoutQueue that have expired
-    const messages = await this.redis.zrangebyscore(
-      constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-      0,
-      Date.now(),
-      "LIMIT",
-      0,
-      10
-    );
+    const stream = redis.scanStream({
+      match: pattern,
+      type: "set",
+      count,
+    });
 
-    if (messages.length === 0) {
-      return;
-    }
+    stream.on("end", () => {
+      onEndCallback?.();
+      redis.quit();
+    });
 
-    for (let i = 0; i < messages.length; i++) {
-      const message = messages[i];
-
-      const messageData = await this.redis.get(this.keys.messageKey(message));
-
-      if (!messageData) {
-        // The message has been removed for some reason (TTL, etc.), so we should remove it from the timeout queue
-        await this.redis.zrem(constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE, message);
-
-        continue;
-      }
-
-      const parsedMessage = MessagePayload.safeParse(JSON.parse(messageData));
-
-      if (!parsedMessage.success) {
-        await this.redis.zrem(constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE, message);
-
-        continue;
-      }
-
-      await this.#callNackMessage({
-        messageKey: this.keys.messageKey(message),
-        messageQueue: parsedMessage.data.queue,
-        parentQueue: parsedMessage.data.parentQueue,
-        concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
-        visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-        messageId: parsedMessage.data.messageId,
-        messageScore: parsedMessage.data.timestamp,
-      });
-    }
+    return { stream, redis };
   }
 
   async #rebalanceParentQueues() {
@@ -748,6 +830,7 @@ export class MarQS {
         pattern,
         component: "marqs",
         operation: "rebalanceParentQueues",
+        service: this.name,
       });
 
       stream.on("data", async (keys) => {
@@ -759,6 +842,7 @@ export class MarQS {
           component: "marqs",
           operation: "rebalanceParentQueues",
           parentQueues: uniqueKeys,
+          service: this.name,
         });
 
         Promise.all(
@@ -808,6 +892,7 @@ export class MarQS {
           childQueuesWithScores,
           component: "marqs",
           operation: "rebalanceParentQueues",
+          service: this.name,
         });
 
         await Promise.all(
@@ -836,6 +921,7 @@ export class MarQS {
   async #callEnqueueMessage(message: MessagePayload) {
     logger.debug("Calling enqueueMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.enqueueMessage(
@@ -873,7 +959,6 @@ export class MarQS {
     const result = await this.redis.dequeueMessage(
       messageQueue,
       parentQueue,
-      visibilityQueue,
       concurrencyLimitKey,
       envConcurrencyLimitKey,
       orgConcurrencyLimitKey,
@@ -881,7 +966,6 @@ export class MarQS {
       envCurrentConcurrencyKey,
       orgCurrentConcurrencyKey,
       messageQueue,
-      String(this.options.visibilityTimeoutInMs ?? 300000), // 5 minutes
       String(Date.now()),
       String(this.options.defaultEnvConcurrency),
       String(this.options.defaultOrgConcurrency)
@@ -893,6 +977,7 @@ export class MarQS {
 
     logger.debug("Dequeue message result", {
       result,
+      service: this.name,
     });
 
     if (result.length !== 2) {
@@ -908,6 +993,7 @@ export class MarQS {
   async #callReplaceMessage(message: MessagePayload) {
     logger.debug("Calling replaceMessage", {
       messagePayload: message,
+      service: this.name,
     });
 
     return this.redis.replaceMessage(
@@ -944,6 +1030,7 @@ export class MarQS {
       orgConcurrencyKey,
       messageId,
       parentQueue,
+      service: this.name,
     });
 
     return this.redis.acknowledgeMessage(
@@ -990,6 +1077,7 @@ export class MarQS {
       visibilityQueue,
       messageId,
       messageScore,
+      service: this.name,
     });
 
     return this.redis.nackMessage(
@@ -1007,25 +1095,6 @@ export class MarQS {
     );
   }
 
-  #callHeartbeatMessage({
-    visibilityQueue,
-    messageId,
-    milliseconds,
-    maxVisibilityTimeout,
-  }: {
-    visibilityQueue: string;
-    messageId: string;
-    milliseconds: number;
-    maxVisibilityTimeout: number;
-  }) {
-    return this.redis.heartbeatMessage(
-      visibilityQueue,
-      messageId,
-      String(milliseconds),
-      String(maxVisibilityTimeout)
-    );
-  }
-
   async #callCalculateMessageCapacities({
     currentConcurrencyKey,
     currentEnvConcurrencyKey,
@@ -1109,6 +1178,7 @@ export class MarQS {
         currentScore,
         rebalanceResult,
         operation: "rebalanceParentQueueChild",
+        service: this.name,
       });
     }
 
@@ -1145,25 +1215,23 @@ end
     });
 
     this.redis.defineCommand("dequeueMessage", {
-      numberOfKeys: 9,
+      numberOfKeys: 8,
       lua: `
--- Keys: childQueue, parentQueue, visibilityQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
+-- Keys: childQueue, parentQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local childQueue = KEYS[1]
 local parentQueue = KEYS[2]
-local visibilityQueue = KEYS[3]
-local concurrencyLimitKey = KEYS[4]
-local envConcurrencyLimitKey = KEYS[5]
-local orgConcurrencyLimitKey = KEYS[6]
-local currentConcurrencyKey = KEYS[7]
-local envCurrentConcurrencyKey = KEYS[8]
-local orgCurrentConcurrencyKey = KEYS[9]
+local concurrencyLimitKey = KEYS[3]
+local envConcurrencyLimitKey = KEYS[4]
+local orgConcurrencyLimitKey = KEYS[5]
+local currentConcurrencyKey = KEYS[6]
+local envCurrentConcurrencyKey = KEYS[7]
+local orgCurrentConcurrencyKey = KEYS[8]
 
--- Args: childQueueName, visibilityQueue, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
+-- Args: childQueueName, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
 local childQueueName = ARGV[1]
-local visibilityTimeout = tonumber(ARGV[2])
-local currentTime = tonumber(ARGV[3])
-local defaultEnvConcurrencyLimit = ARGV[4]
-local defaultOrgConcurrencyLimit = ARGV[5]
+local currentTime = tonumber(ARGV[2])
+local defaultEnvConcurrencyLimit = ARGV[3]
+local defaultOrgConcurrencyLimit = ARGV[4]
 
 -- Check current org concurrency against the limit
 local orgCurrentConcurrency = tonumber(redis.call('SCARD', orgCurrentConcurrencyKey) or '0')
@@ -1199,11 +1267,9 @@ end
 
 local messageId = messages[1]
 local messageScore = tonumber(messages[2])
-local timeoutScore = currentTime + visibilityTimeout
 
 -- Move message to timeout queue and update concurrency
 redis.call('ZREM', childQueue, messageId)
-redis.call('ZADD', visibilityQueue, timeoutScore, messageId)
 redis.call('SADD', currentConcurrencyKey, messageId)
 redis.call('SADD', envCurrentConcurrencyKey, messageId)
 redis.call('SADD', orgCurrentConcurrencyKey, messageId)
@@ -1269,7 +1335,7 @@ else
     redis.call('ZADD', parentQueue, earliestMessage[2], messageQueueName)
 end
 
--- Remove the message from the timeout queue
+-- Remove the message from the timeout queue (deprecated, will eventually remove this)
 redis.call('ZREM', visibilityQueue, messageId)
 
 -- Update the concurrency keys
@@ -1297,20 +1363,18 @@ local messageId = ARGV[2]
 local currentTime = tonumber(ARGV[3])
 local messageScore = tonumber(ARGV[4])
 
--- Check to see if the message is still in the visibilityQueue
-local messageVisibility = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
-
-if messageVisibility == 0 then
-    return
-end
-
 -- Update the concurrency keys
 redis.call('SREM', concurrencyKey, messageId)
 redis.call('SREM', envConcurrencyKey, messageId)
 redis.call('SREM', orgConcurrencyKey, messageId)
 
--- Remove the message from the timeout queue
-redis.call('ZREM', visibilityQueue, messageId)
+-- Check to see if the message is still in the visibilityQueue
+local messageVisibility = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
+
+if messageVisibility > 0 then
+-- Remove the message from the timeout queue (deprecated, will eventually remove this)
+    redis.call('ZREM', visibilityQueue, messageId)
+end
 
 -- Enqueue the message into the queue
 redis.call('ZADD', childQueueKey, messageScore, messageId)
@@ -1337,12 +1401,16 @@ local milliseconds = tonumber(ARGV[2])
 local maxVisibilityTimeout = tonumber(ARGV[3])
 
 -- Get the current visibility timeout
-local currentVisibilityTimeout = tonumber(redis.call('ZSCORE', visibilityQueue, messageId)) or 0
+local zscoreResult = redis.call('ZSCORE', visibilityQueue, messageId)
 
-if currentVisibilityTimeout == 0 then
+-- If there's no currentVisibilityTimeout, return and do not execute ZADD
+if zscoreResult == false then
     return
 end
 
+local currentVisibilityTimeout = tonumber(zscoreResult)
+
+
 -- Calculate the new visibility timeout
 local newVisibilityTimeout = math.min(currentVisibilityTimeout + milliseconds * 1000, maxVisibilityTimeout)
 
@@ -1445,7 +1513,6 @@ declare module "ioredis" {
     dequeueMessage(
       childQueue: string,
       parentQueue: string,
-      visibilityQueue: string,
       concurrencyLimitKey: string,
       envConcurrencyLimitKey: string,
       orgConcurrencyLimitKey: string,
@@ -1453,7 +1520,6 @@ declare module "ioredis" {
       envCurrentConcurrencyKey: string,
       orgCurrentConcurrencyKey: string,
       childQueueName: string,
-      visibilityTimeout: string,
       currentTime: string,
       defaultEnvConcurrencyLimit: string,
       defaultOrgConcurrencyLimit: string,
@@ -1548,7 +1614,10 @@ function getMarQSClient() {
       };
 
       return new MarQS({
+        name: "marqs",
+        tracer: trace.getTracer("marqs"),
         keysProducer: new MarQSShortKeyProducer(KEY_PREFIX),
+        visibilityTimeoutStrategy: new V3VisibilityTimeout(),
         queuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 36 }),
         envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
         workers: 1,
@@ -1556,6 +1625,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

@@ -13,12 +13,16 @@ const constants = {
 } as const;
 
 export class MarQSShortKeyProducer implements MarQSKeyProducer {
-  constructor(private _prefix: string) {}
+  constructor(private _prefix: string) { }
 
   sharedQueueScanPattern() {
     return `${this._prefix}*${constants.SHARED_QUEUE}`;
   }
 
+  queueCurrentConcurrencyScanPattern() {
+    return `${this._prefix}${constants.ORG_PART}:*:${constants.ENV_PART}:*:queue:*:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
   stripKeyPrefix(key: string): string {
     if (key.startsWith(this._prefix)) {
       return key.slice(this._prefix.length);
@@ -58,6 +62,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 });
+  }
+}