Release 3.0.0-beta.39

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

.changeset/beige-pears-explode.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add an e2e suite to test compiling with v3 CLI.

.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/hot-wasps-sin.md

@@ -0,0 +1,7 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+v3: Usage tracking

.changeset/itchy-chairs-itch.md

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

.changeset/mighty-parrots-sin.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+v3: Remove aggressive otel flush timeouts in dev/prod

.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

@@ -46,10 +46,12 @@
   "changesets": [
     "afraid-sheep-joke",
     "angry-eagles-trade",
+    "beige-pears-explode",
     "beige-pens-dance",
     "big-tomatoes-deliver",
     "blue-pumas-whisper",
     "breezy-gorillas-mate",
+    "brown-spies-burn",
     "chilled-hornets-move",
     "clean-pianos-listen",
     "clever-apes-collect",
@@ -68,7 +70,10 @@
     "funny-swans-destroy",
     "gorgeous-gorillas-compete",
     "green-bags-wink",
+    "hot-buckets-behave",
     "hot-fishes-retire",
+    "hot-wasps-sin",
+    "itchy-chairs-itch",
     "khaki-apricots-design",
     "khaki-poems-lay",
     "late-icons-lie",
@@ -84,12 +89,15 @@
     "many-ligers-pump",
     "mighty-camels-joke",
     "mighty-flowers-train",
+    "mighty-parrots-sin",
     "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,21 +108,29 @@
     "rich-kangaroos-unite",
     "rotten-beers-refuse",
     "rotten-dryers-exercise",
+    "rude-toys-compare",
     "selfish-ducks-sort",
+    "serious-hats-rest",
     "shaggy-spoons-taste",
     "sharp-emus-compare",
     "sharp-zebras-serve",
     "shiny-coats-cry",
     "silly-suits-switch",
+    "silver-doors-juggle",
     "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",
+    "strong-phones-smoke",
+    "stupid-adults-sniff",
     "stupid-bulldogs-applaud",
     "sweet-lizards-press",
     "swift-dragons-peel",
@@ -122,6 +138,7 @@
     "tame-guests-know",
     "tender-moose-tell",
     "tender-oranges-rhyme",
+    "tender-turkeys-compete",
     "thin-parents-heal",
     "thirty-islands-kiss",
     "tidy-balloons-suffer",
@@ -130,8 +147,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/serious-hats-rest.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+v3: [prod] force flush timeout should be 1s

.changeset/silver-doors-juggle.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Make deduplicationKey required when creating/updating a schedule

.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/strong-phones-smoke.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+fix: allow command login to read api url from cli args

.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/tender-turkeys-compete.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Added timezone support to schedules

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

@@ -1,9 +1,53 @@
-name: "🧪 E2E Tests"
+name: "E2E"
 on:
   workflow_call:
+    inputs:
+      package:
+        description: The identifier of the job to run
+        default: webapp
+        required: false
+        type: string
 jobs:
-  e2e:
-    name: "🧪 E2E Tests"
+  cli-v3:
+    name: "🧪 CLI v3 tests"
+    if: inputs.package == 'cli-v3' || inputs.package == ''
+    runs-on: buildjet-8vcpu-ubuntu-2204
+    strategy:
+      fail-fast: false
+      matrix:
+        package-manager: ["npm", "pnpm", "yarn"]
+    steps:
+      - name: ⬇️ Checkout repo
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: ⎔ Setup pnpm
+        uses: pnpm/action-setup@v2.2.4
+        with:
+          version: 8.15.5
+
+      - name: ⎔ Setup node
+        uses: buildjet/setup-node@v3
+        with:
+          node-version: 20.11.1
+          cache: "pnpm"
+
+      - name: 📥 Download deps
+        run: pnpm install --frozen-lockfile --filter trigger.dev...
+
+      - name: 🔧 Build v3 cli monorepo dependencies
+        run: pnpm run build --filter trigger.dev^...
+
+      - name: 🔧 Build worker template files
+        run: pnpm --filter trigger.dev run build:workers
+
+      - name: Run E2E Tests
+        run: |
+          PM=${{ matrix.package-manager }} pnpm --filter trigger.dev run test:e2e
+  webapp:
+    name: "🧪 Webapp tests"
+    if: inputs.package == 'webapp' || inputs.package == ''
     runs-on: buildjet-16vcpu-ubuntu-2204
     steps:
       - name: 🐳 Login to Docker Hub

.github/workflows/pr_checks.yml

@@ -29,4 +29,6 @@ jobs:
 
   # e2e:
   #   uses: ./.github/workflows/e2e.yml
+  #   with:
+  #     package: webapp
   #   secrets: inherit

.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

@@ -49,11 +49,18 @@ jobs:
     uses: ./.github/workflows/unit-tests.yml
     secrets: inherit
 
-  # e2e:
-  #   uses: ./.github/workflows/e2e.yml
-  #   secrets: inherit
+  e2e:
+    uses: ./.github/workflows/e2e.yml
+    with:
+      package: cli-v3
+    secrets: inherit
 
   publish:
     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

@@ -7,7 +7,12 @@ import {
   TaskOperationsIndexOptions,
   TaskOperationsRestoreOptions,
 } from "@trigger.dev/core-apps";
-import { Machine, PostStartCauses, PreStopCauses, EnvironmentType } from "@trigger.dev/core/v3";
+import {
+  MachinePreset,
+  PostStartCauses,
+  PreStopCauses,
+  EnvironmentType,
+} from "@trigger.dev/core/v3";
 import { randomUUID } from "crypto";
 import { TaskMonitor } from "./taskMonitor";
 import { PodCleaner } from "./podCleaner";
@@ -16,6 +21,7 @@ const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local";
 const NODE_NAME = process.env.NODE_NAME || "local";
 const OTEL_EXPORTER_OTLP_ENDPOINT =
   process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318";
+const POD_CLEANER_INTERVAL_SECONDS = Number(process.env.POD_CLEANER_INTERVAL_SECONDS || "300");
 
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 logger.log(`running in ${RUNTIME_ENV} mode`);
@@ -47,6 +53,10 @@ class KubernetesTaskOperations implements TaskOperations {
     this.#k8sApi = this.#createK8sApi();
   }
 
+  async init() {
+    // noop
+  }
+
   async index(opts: TaskOperationsIndexOptions) {
     await this.#createJob(
       {
@@ -212,7 +222,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 +326,9 @@ class KubernetesTaskOperations implements TaskOperations {
         {
           name: "registry-trigger",
         },
+        {
+          name: "registry-trigger-failover",
+        },
       ],
       nodeSelector: {
         nodetype: "worker",
@@ -391,10 +404,10 @@ class KubernetesTaskOperations implements TaskOperations {
     };
   }
 
-  #getResourcesFromMachineConfig(config: Machine): ComputeResources {
+  #getResourcesFromMachineConfig(preset: MachinePreset): ComputeResources {
     return {
-      cpu: `${config.cpu}`,
-      memory: `${config.memory}G`,
+      cpu: `${preset.cpu}`,
+      memory: `${preset.memory}G`,
     };
   }
 
@@ -548,7 +561,7 @@ taskMonitor.start();
 const podCleaner = new PodCleaner({
   runtimeEnv: RUNTIME_ENV,
   namespace: "default",
-  intervalInSeconds: 300,
+  intervalInSeconds: POD_CLEANER_INTERVAL_SECONDS,
 });
 
 podCleaner.start();

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/primitives/DateTime.tsx

@@ -6,6 +6,7 @@ type DateTimeProps = {
   timeZone?: string;
   includeSeconds?: boolean;
   includeTime?: boolean;
+  showTimezone?: boolean;
 };
 
 export const DateTime = ({
@@ -13,6 +14,7 @@ export const DateTime = ({
   timeZone,
   includeSeconds = true,
   includeTime = true,
+  showTimezone = false,
 }: DateTimeProps) => {
   const locales = useLocales();
 
@@ -42,7 +44,12 @@ export const DateTime = ({
     );
   }, [locales, includeSeconds, realDate]);
 
-  return <Fragment>{formattedDateTime.replace(/\s/g, String.fromCharCode(32))}</Fragment>;
+  return (
+    <Fragment>
+      {formattedDateTime.replace(/\s/g, String.fromCharCode(32))}
+      {showTimezone ? ` (${timeZone ?? "UTC"})` : null}
+    </Fragment>
+  );
 };
 
 export function formatDateTime(

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

@@ -8,6 +8,7 @@ import { ShortcutDefinition, useShortcutKeys } from "~/hooks/useShortcutKeys";
 import { cn } from "~/utils/cn";
 import { ShortcutKey } from "./ShortcutKey";
 import { ChevronDown } from "lucide-react";
+import { MatchSorterOptions, matchSorter } from "match-sorter";
 
 const sizes = {
   small: {
@@ -75,7 +76,10 @@ export interface SelectProps<TValue extends string | string[], TItem>
   showHeading?: boolean;
   items?: TItem[] | Section<TItem>[];
   empty?: React.ReactNode;
-  filter?: (item: ItemFromSection<TItem>, search: string, title?: string) => boolean;
+  filter?:
+    | boolean
+    | MatchSorterOptions<TItem>
+    | ((item: ItemFromSection<TItem>, search: string, title?: string) => boolean);
   children:
     | React.ReactNode
     | ((
@@ -129,18 +133,44 @@ export function Select<TValue extends string | string[], TItem>({
     if (!items) return [];
     if (!searchValue || !filter) return items;
 
+    if (typeof filter === "function") {
+      if (isSection(items)) {
+        return items
+          .map((section) => ({
+            ...section,
+            items: section.items.filter((item) =>
+              filter(item as ItemFromSection<TItem>, searchValue, section.title)
+            ),
+          }))
+          .filter((section) => section.items.length > 0);
+      }
+
+      return items.filter((item) => filter(item as ItemFromSection<TItem>, searchValue));
+    }
+
+    if (typeof filter === "boolean" && filter) {
+      if (isSection(items)) {
+        return items
+          .map((section) => ({
+            ...section,
+            items: matchSorter(section.items, searchValue),
+          }))
+          .filter((section) => section.items.length > 0);
+      }
+
+      return matchSorter(items, searchValue);
+    }
+
     if (isSection(items)) {
       return items
         .map((section) => ({
           ...section,
-          items: section.items.filter((item) =>
-            filter(item as ItemFromSection<TItem>, searchValue, section.title)
-          ),
+          items: matchSorter(section.items, searchValue, filter),
         }))
         .filter((section) => section.items.length > 0);
     }
 
-    return items.filter((item) => filter(item as ItemFromSection<TItem>, searchValue));
+    return matchSorter(items, searchValue, filter);
   }, [searchValue, items]);
 
   const enableItemShortcuts = allowItemShortcuts && matches.length === items?.length;

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/components/scheduled/timezones.tsx

@@ -0,0 +1,63 @@
+import { useVirtualizer } from "@tanstack/react-virtual";
+import { useRef } from "react";
+import { SelectItem } from "../primitives/Select";
+
+export function TimezoneList({ timezones }: { timezones: string[] }) {
+  const parentRef = useRef<HTMLDivElement>(null);
+
+  const rowVirtualizer = useVirtualizer({
+    count: timezones.length,
+    getScrollElement: () => parentRef.current,
+    estimateSize: () => 28,
+  });
+
+  return (
+    <div
+      ref={parentRef}
+      className="max-h-[calc(min(480px,var(--popover-available-height))-2.35rem)] overflow-y-auto overscroll-contain scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600"
+    >
+      <div
+        style={{
+          height: `${rowVirtualizer.getTotalSize()}px`,
+          width: "100%",
+          position: "relative",
+        }}
+      >
+        {rowVirtualizer.getVirtualItems().map((virtualItem) => (
+          <TimezoneCell
+            key={virtualItem.key}
+            size={virtualItem.size}
+            start={virtualItem.start}
+            timezone={timezones[virtualItem.index]}
+          />
+        ))}
+      </div>
+    </div>
+  );
+}
+
+function TimezoneCell({
+  timezone,
+  size,
+  start,
+}: {
+  timezone: string;
+  size: number;
+  start: number;
+}) {
+  return (
+    <SelectItem
+      value={timezone}
+      style={{
+        position: "absolute",
+        top: 0,
+        left: 0,
+        width: "100%",
+        height: `${size}px`,
+        transform: `translateY(${start}px)`,
+      }}
+    >
+      {timezone}
+    </SelectItem>
+  );
+}

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

@@ -1,5 +1,5 @@
-import { z } from "zod";
 import { SecretStoreOptionsSchema } from "./services/secrets/secretStoreOptionsSchema.server";
+import { z } from "zod";
 import { isValidRegex } from "./utils/regex";
 import { isValidDatabaseUrl } from "./utils/db";
 
@@ -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,39 @@ 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"),
+  /* Usage settings */
+  USAGE_EVENT_URL: z.string().optional(),
+  PROD_USAGE_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),
+
+  CENTS_PER_HOUR_MICRO: z.coerce.number().default(0),
+  CENTS_PER_HOUR_SMALL_1X: z.coerce.number().default(0),
+  CENTS_PER_HOUR_SMALL_2X: z.coerce.number().default(0),
+  CENTS_PER_HOUR_MEDIUM_1X: z.coerce.number().default(0),
+  CENTS_PER_HOUR_MEDIUM_2X: z.coerce.number().default(0),
+  CENTS_PER_HOUR_LARGE_1X: z.coerce.number().default(0),
+  CENTS_PER_HOUR_LARGE_2X: z.coerce.number().default(0),
+  BASE_RUN_COST_IN_CENTS: z.coerce.number().default(0),
+
+  USAGE_OPEN_METER_API_KEY: z.string().optional(),
+  USAGE_OPEN_METER_BASE_URL: z.string().optional(),
 });
 
 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/endpoint.server.ts

@@ -27,6 +27,7 @@ export function detectResponseIsTimeout(rawBody: string, response?: Response) {
 
   return (
     isResponseVercelTimeout(response) ||
+    isResponseCloudfrontTimeout(response) ||
     isResponseDenoDeployTimeout(rawBody, response) ||
     isResponseCloudflareTimeout(rawBody, response)
   );
@@ -50,3 +51,7 @@ function isResponseVercelTimeout(response: Response) {
 function isResponseDenoDeployTimeout(rawBody: string, response: Response) {
   return response.status === 502 && rawBody.includes("TIME_LIMIT");
 }
+
+function isResponseCloudfrontTimeout(response: Response) {
+  return response.status === 504 && typeof response.headers.get("x-amz-cf-id") === "string";
+}

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.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,7 @@
 import { RuntimeEnvironmentType } from "@trigger.dev/database";
 import { PrismaClient, prisma } from "~/db.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
+import { getTimezones } from "~/utils/timezones.server";
 
 type EditScheduleOptions = {
   userId: string;
@@ -67,24 +69,13 @@ 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 {
       possibleTasks: possibleTasks.map((task) => task.slug),
       possibleEnvironments,
+      possibleTimezones: getTimezones(),
       schedule: await this.#getExistingSchedule(friendlyId, possibleEnvironments),
     };
   }
@@ -102,6 +93,7 @@ export class EditSchedulePresenter {
         externalId: true,
         deduplicationKey: true,
         userProvidedDeduplicationKey: true,
+        timezone: true,
         taskIdentifier: true,
         instances: {
           select: {

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,
       },
     });
 
@@ -152,10 +156,12 @@ export class RunListPresenter extends BasePresenter {
         runtimeEnvironmentId: string;
         status: TaskRunStatus;
         createdAt: Date;
+        startedAt: Date | null;
         lockedAt: Date | null;
         updatedAt: Date;
         isTest: boolean;
         spanId: string;
+        idempotencyKey: string | null;
       }[]
     >`
     SELECT
@@ -167,10 +173,12 @@ export class RunListPresenter extends BasePresenter {
     tr."runtimeEnvironmentId" AS "runtimeEnvironmentId",
     tr.status AS status,
     tr."createdAt" AS "createdAt",
+    tr."startedAt" AS "startedAt",
     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 +218,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}`
@@ -265,12 +274,15 @@ export class RunListPresenter extends BasePresenter {
 
         const hasFinished = FINISHED_STATUSES.includes(run.status);
 
+        const startedAt = run.startedAt ?? run.lockedAt;
+
         return {
           id: run.id,
           friendlyId: run.runFriendlyId,
           number: Number(run.number),
           createdAt: run.createdAt.toISOString(),
-          startedAt: run.lockedAt ? run.lockedAt.toISOString() : undefined,
+          updatedAt: run.updatedAt.toISOString(),
+          startedAt: startedAt ? startedAt.toISOString() : undefined,
           hasFinished,
           finishedAt: hasFinished ? run.updatedAt.toISOString() : undefined,
           isTest: run.isTest,
@@ -280,7 +292,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;
@@ -20,6 +22,7 @@ export type ScheduleListItem = {
   userProvidedDeduplicationKey: boolean;
   cron: string;
   cronDescription: string;
+  timezone: string;
   externalId: string | null;
   nextRun: Date;
   lastRun: Date | undefined;
@@ -33,13 +36,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,
     projectId,
@@ -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: {
@@ -74,14 +71,25 @@ export class ScheduleListPresenter {
             },
           },
         },
+        organization: {
+          select: {
+            maximumSchedulesLimit: true,
+          },
+        },
       },
       where: {
         id: projectId,
       },
     });
 
+    const schedulesCount = await this._prisma.taskSchedule.count({
+      where: {
+        projectId,
+      },
+    });
+
     //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 +100,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 +142,7 @@ export class ScheduleListPresenter {
       },
     });
 
-    const rawSchedules = await this.#prismaClient.taskSchedule.findMany({
+    const rawSchedules = await this._replica.taskSchedule.findMany({
       select: {
         id: true,
         friendlyId: true,
@@ -143,6 +151,7 @@ export class ScheduleListPresenter {
         userProvidedDeduplicationKey: true,
         generatorExpression: true,
         generatorDescription: true,
+        timezone: true,
         externalId: true,
         instances: {
           select: {
@@ -198,7 +207,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"
@@ -221,10 +230,11 @@ export class ScheduleListPresenter {
         userProvidedDeduplicationKey: schedule.userProvidedDeduplicationKey,
         cron: schedule.generatorExpression,
         cronDescription: schedule.generatorDescription,
+        timezone: schedule.timezone,
         active: schedule.active,
         externalId: schedule.externalId,
         lastRun: latestRun?.createdAt,
-        nextRun: calculateNextScheduledTimestamp(schedule.generatorExpression),
+        nextRun: calculateNextScheduledTimestamp(schedule.generatorExpression, schedule.timezone),
         environments: schedule.instances.map((instance) => {
           const environment = project.environments.find((env) => env.id === instance.environmentId);
           if (!environment) {
@@ -233,14 +243,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,16 +255,13 @@ 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,
+      limits: {
+        used: schedulesCount,
+        limit: project.organization.maximumSchedulesLimit,
+      },
       filters: {
         tasks,
         environments,

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);
@@ -291,7 +311,7 @@ export class TaskListPresenter extends BasePresenter {
     >`    
     SELECT 
       tr."taskIdentifier", 
-      AVG(EXTRACT(EPOCH FROM (tr."updatedAt" - tr."lockedAt"))) as duration
+      AVG(EXTRACT(EPOCH FROM (tr."updatedAt" - COALESCE(tr."startedAt", tr."lockedAt")))) as duration
       FROM 
       ${sqlDatabaseSchema}."TaskRun" as tr
     WHERE 

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

@@ -6,6 +6,7 @@ import {
   TaskTriggerSource,
 } from "@trigger.dev/database";
 import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
+import { getTimezones } from "~/utils/timezones.server";
 import { getUsername } from "~/utils/username";
 
 type TestTaskOptions = {
@@ -37,6 +38,7 @@ export type TestTask =
   | {
       triggerSource: "SCHEDULED";
       task: Task;
+      possibleTimezones: string[];
       runs: ScheduledRun[];
     };
 
@@ -61,6 +63,7 @@ export type ScheduledRun = Omit<RawRun, "number" | "payload"> & {
     timestamp: Date;
     lastTimestamp?: Date;
     externalId?: string;
+    timezone: string;
   };
 };
 
@@ -168,9 +171,11 @@ export class TestTaskPresenter {
           ),
         };
       case "SCHEDULED":
+        const possibleTimezones = getTimezones();
         return {
           triggerSource: "SCHEDULED",
           task: taskWithEnvironment,
+          possibleTimezones,
           runs: (
             await Promise.all(
               latestRuns.map(async (r) => {
@@ -195,6 +200,9 @@ export class TestTaskPresenter {
 
 async function getScheduleTaskRunPayload(run: RawRun) {
   const payload = await parsePacket({ data: run.payload, dataType: run.payloadType });
+  if (!payload.timezone) {
+    payload.timezone = "UTC";
+  }
   const parsed = ScheduledTaskPayload.safeParse(payload);
   return parsed;
 }

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

@@ -1,7 +1,8 @@
+import { ScheduleObject } from "@trigger.dev/core/v3";
 import { PrismaClient, prisma } from "~/db.server";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { nextScheduledTimestamps } from "~/v3/utils/calculateNextSchedule.server";
 import { RunListPresenter } from "./RunListPresenter.server";
-import { ScheduleObject } from "@trigger.dev/core/v3";
 
 type ViewScheduleOptions = {
   userId?: string;
@@ -23,13 +24,14 @@ export class ViewSchedulePresenter {
         friendlyId: true,
         generatorExpression: true,
         generatorDescription: true,
+        timezone: true,
         externalId: true,
         deduplicationKey: true,
         userProvidedDeduplicationKey: true,
         taskIdentifier: true,
         project: {
           select: {
-            slug: true,
+            id: true,
           },
         },
         instances: {
@@ -38,6 +40,7 @@ export class ViewSchedulePresenter {
               select: {
                 id: true,
                 type: true,
+                slug: true,
                 orgMember: {
                   select: {
                     user: {
@@ -66,12 +69,13 @@ export class ViewSchedulePresenter {
     }
 
     const nextRuns = schedule.active
-      ? nextScheduledTimestamps(schedule.generatorExpression, new Date(), 5)
+      ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5)
       : [];
 
     const runPresenter = new RunListPresenter(this.#prismaClient);
+
     const { runs } = await runPresenter.call({
-      projectSlug: schedule.project.slug,
+      projectId: schedule.project.id,
       scheduleId: schedule.id,
       pageSize: 5,
     });
@@ -79,27 +83,14 @@ export class ViewSchedulePresenter {
     return {
       schedule: {
         ...schedule,
+        timezone: schedule.timezone,
         cron: schedule.generatorExpression,
         cronDescription: schedule.generatorDescription,
         nextRuns,
         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);
         }),
       },
     };
@@ -116,6 +107,7 @@ export class ViewSchedulePresenter {
         expression: result.schedule.cron,
         description: result.schedule.cronDescription,
       },
+      timezone: result.schedule.timezone,
       externalId: result.schedule.externalId ?? undefined,
       deduplicationKey: result.schedule.userProvidedDeduplicationKey
         ? result.schedule.deduplicationKey ?? undefined

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/_app.orgs.$organizationSlug.projects.v3.$projectParam.schedules.$scheduleParam/route.tsx

@@ -185,7 +185,8 @@ export default function Page() {
   const location = useLocation();
   const organization = useOrganization();
   const project = useProject();
-  const user = useUser();
+
+  const isUtc = schedule.timezone === "UTC";
 
   return (
     <div className="grid h-full max-h-full grid-rows-[2.5rem_1fr_3.25rem] overflow-hidden bg-background-bright">
@@ -210,6 +211,7 @@ export default function Page() {
                   <Paragraph variant="small">{schedule.cronDescription}</Paragraph>
                 </div>
               </Property>
+              <Property label="Timezone">{schedule.timezone}</Property>
               <Property label="Environments">
                 <EnvironmentLabels size="small" environments={schedule.environments} />
               </Property>
@@ -245,19 +247,21 @@ export default function Page() {
               <Table>
                 <TableHeader>
                   <TableRow>
+                    {!isUtc && <TableHeaderCell>{schedule.timezone}</TableHeaderCell>}
                     <TableHeaderCell>UTC</TableHeaderCell>
-                    <TableHeaderCell>Local time</TableHeaderCell>
                   </TableRow>
                 </TableHeader>
                 <TableBody>
                   {schedule.nextRuns.map((run, index) => (
                     <TableRow key={index}>
+                      {!isUtc && (
+                        <TableCell>
+                          <DateTime date={run} timeZone={schedule.timezone} />
+                        </TableCell>
+                      )}
                       <TableCell>
                         <DateTime date={run} timeZone="UTC" />
                       </TableCell>
-                      <TableCell>
-                        <DateTime date={run} />
-                      </TableCell>
                     </TableRow>
                   ))}
                 </TableBody>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.schedules.edit.$scheduleParam/route.tsx

@@ -21,7 +21,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
 };
 
 export default function Page() {
-  const { schedule, possibleTasks, possibleEnvironments, showGenerateField } =
+  const { schedule, possibleTasks, possibleEnvironments, possibleTimezones, showGenerateField } =
     useTypedLoaderData<typeof loader>();
 
   return (
@@ -29,6 +29,7 @@ export default function Page() {
       schedule={schedule}
       possibleTasks={possibleTasks}
       possibleEnvironments={possibleEnvironments}
+      possibleTimezones={possibleTimezones}
       showGenerateField={showGenerateField}
     />
   );

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

@@ -20,7 +20,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
 };
 
 export default function Page() {
-  const { schedule, possibleTasks, possibleEnvironments, showGenerateField } =
+  const { schedule, possibleTasks, possibleEnvironments, possibleTimezones, showGenerateField } =
     useTypedLoaderData<typeof loader>();
 
   return (
@@ -29,6 +29,7 @@ export default function Page() {
       possibleTasks={possibleTasks}
       possibleEnvironments={possibleEnvironments}
       showGenerateField={showGenerateField}
+      possibleTimezones={possibleTimezones}
     />
   );
 }

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

@@ -4,12 +4,21 @@ import { Outlet, useLocation, useParams } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { BlankstateInstructions } from "~/components/BlankstateInstructions";
+import { Feedback } from "~/components/Feedback";
 import { AdminDebugTooltip } from "~/components/admin/debugTooltip";
 import { InlineCode } from "~/components/code/InlineCode";
-import { EnvironmentLabel, EnvironmentLabels } from "~/components/environments/EnvironmentLabel";
+import { EnvironmentLabels } from "~/components/environments/EnvironmentLabel";
 import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
-import { LinkButton } from "~/components/primitives/Buttons";
+import { Button, LinkButton } from "~/components/primitives/Buttons";
 import { DateTime } from "~/components/primitives/DateTime";
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTrigger,
+} from "~/components/primitives/Dialog";
 import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
 import { PaginationControls } from "~/components/primitives/Pagination";
 import { Paragraph } from "~/components/primitives/Paragraph";
@@ -78,6 +87,7 @@ export default function Page() {
     possibleEnvironments,
     hasFilters,
     filters,
+    limits,
     currentPage,
     totalPages,
   } = useTypedLoaderData<typeof loader>();
@@ -107,15 +117,43 @@ export default function Page() {
             </PropertyTable>
           </AdminDebugTooltip>
 
-          <LinkButton
-            LeadingIcon={PlusIcon}
-            to={`${v3NewSchedulePath(organization, project)}${location.search}`}
-            variant="primary/small"
-            shortcut={{ key: "n" }}
-            disabled={possibleTasks.length === 0 || isShowingNewPane}
-          >
-            New schedule
-          </LinkButton>
+          {limits.used >= limits.limit ? (
+            <Dialog>
+              <DialogTrigger asChild>
+                <Button
+                  LeadingIcon={PlusIcon}
+                  variant="primary/small"
+                  shortcut={{ key: "n" }}
+                  disabled={possibleTasks.length === 0 || isShowingNewPane}
+                >
+                  New schedule
+                </Button>
+              </DialogTrigger>
+              <DialogContent>
+                <DialogHeader>You've exceeded your limit</DialogHeader>
+                <DialogDescription>
+                  You've used {limits.used}/{limits.limit} of your schedules. You can request more
+                  schedules.
+                </DialogDescription>
+                <DialogFooter>
+                  <Feedback
+                    button={<Button variant="primary/medium">Request more</Button>}
+                    defaultValue="help"
+                  />
+                </DialogFooter>
+              </DialogContent>
+            </Dialog>
+          ) : (
+            <LinkButton
+              LeadingIcon={PlusIcon}
+              to={`${v3NewSchedulePath(organization, project)}${location.search}`}
+              variant="primary/small"
+              shortcut={{ key: "n" }}
+              disabled={possibleTasks.length === 0 || isShowingNewPane}
+            >
+              New schedule
+            </LinkButton>
+          )}
         </PageAccessories>
       </NavBar>
       <PageBody scrollable={false}>
@@ -142,7 +180,21 @@ export default function Page() {
                 </div>
 
                 <SchedulesTable schedules={schedules} hasFilters={hasFilters} />
-                <div className="mt-2 justify-end">
+                <div className="mt-2 justify-between">
+                  <Paragraph variant="extra-small" className="mt-3">
+                    <span className={limits.used >= limits.limit ? "text-warning" : ""}>
+                      You've used {limits.used}/{limits.limit} of your schedules.
+                    </span>{" "}
+                    <Feedback
+                      button={
+                        <button className=" text-secondary transition hover:text-indigo-400">
+                          Request more
+                        </button>
+                      }
+                      defaultValue="help"
+                    />
+                    .
+                  </Paragraph>
                   <PaginationControls currentPage={currentPage} totalPages={totalPages} />
                 </div>
               </div>
@@ -236,12 +288,13 @@ function SchedulesTable({
         <TableRow>
           <TableHeaderCell>ID</TableHeaderCell>
           <TableHeaderCell>Task ID</TableHeaderCell>
+          <TableHeaderCell>External ID</TableHeaderCell>
           <TableHeaderCell>CRON</TableHeaderCell>
           <TableHeaderCell hiddenLabel>CRON description</TableHeaderCell>
-          <TableHeaderCell>External ID</TableHeaderCell>
+          <TableHeaderCell>Timezone</TableHeaderCell>
+          <TableHeaderCell>Next run</TableHeaderCell>
+          <TableHeaderCell>Last run</TableHeaderCell>
           <TableHeaderCell>Deduplication key</TableHeaderCell>
-          <TableHeaderCell>Next run (UTC)</TableHeaderCell>
-          <TableHeaderCell>Last run (UTC)</TableHeaderCell>
           <TableHeaderCell>Environments</TableHeaderCell>
           <TableHeaderCell>Enabled</TableHeaderCell>
         </TableRow>
@@ -262,6 +315,9 @@ function SchedulesTable({
                 <TableCell to={path} className={cellClass}>
                   {schedule.taskIdentifier}
                 </TableCell>
+                <TableCell to={path} className={cellClass}>
+                  {schedule.externalId ? schedule.externalId : "–"}
+                </TableCell>
                 <TableCell to={path} className={cellClass}>
                   {schedule.cron}
                 </TableCell>
@@ -269,17 +325,21 @@ function SchedulesTable({
                   {schedule.cronDescription}
                 </TableCell>
                 <TableCell to={path} className={cellClass}>
-                  {schedule.externalId ? schedule.externalId : "–"}
+                  {schedule.timezone}
+                </TableCell>
+                <TableCell to={path} className={cellClass}>
+                  <DateTime date={schedule.nextRun} timeZone={schedule.timezone} />
+                </TableCell>
+                <TableCell to={path} className={cellClass}>
+                  {schedule.lastRun ? (
+                    <DateTime date={schedule.lastRun} timeZone={schedule.timezone} />
+                  ) : (
+                    "–"
+                  )}
                 </TableCell>
                 <TableCell to={path} className={cellClass}>
                   {schedule.userProvidedDeduplicationKey ? schedule.deduplicationKey : "–"}
                 </TableCell>
-                <TableCell to={path} className={cellClass}>
-                  <DateTime date={schedule.nextRun} timeZone="utc" />
-                </TableCell>
-                <TableCell to={path} className={cellClass}>
-                  {schedule.lastRun ? <DateTime date={schedule.lastRun} timeZone="utc" /> : "–"}
-                </TableCell>
                 <TableCell to={path} className={cellClass}>
                   <EnvironmentLabels environments={schedule.environments} size="small" />
                 </TableCell>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test.tasks.$taskParam/route.tsx

@@ -26,8 +26,10 @@ import {
   ResizablePanel,
   ResizablePanelGroup,
 } from "~/components/primitives/Resizable";
+import { Select } from "~/components/primitives/Select";
 import { TextLink } from "~/components/primitives/TextLink";
 import { TaskRunStatusCombo } from "~/components/runs/v3/TaskRunStatus";
+import { TimezoneList } from "~/components/scheduled/timezones";
 import { redirectBackWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
 import {
   ScheduledRun,
@@ -95,7 +97,13 @@ export default function Page() {
       return <StandardTaskForm task={result.task} runs={result.runs} />;
     }
     case "SCHEDULED": {
-      return <ScheduledTaskForm task={result.task} runs={result.runs} />;
+      return (
+        <ScheduledTaskForm
+          task={result.task}
+          runs={result.runs}
+          possibleTimezones={result.possibleTimezones}
+        />
+      );
     }
   }
 }
@@ -215,12 +223,21 @@ function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: Standa
   );
 }
 
-function ScheduledTaskForm({ task, runs }: { task: TestTask["task"]; runs: ScheduledRun[] }) {
+function ScheduledTaskForm({
+  task,
+  runs,
+  possibleTimezones,
+}: {
+  task: TestTask["task"];
+  runs: ScheduledRun[];
+  possibleTimezones: string[];
+}) {
   const lastSubmission = useActionData();
   const [selectedCodeSampleId, setSelectedCodeSampleId] = useState(runs.at(0)?.id);
   const [timestampValue, setTimestampValue] = useState<Date | undefined>();
   const [lastTimestampValue, setLastTimestampValue] = useState<Date | undefined>();
   const [externalIdValue, setExternalIdValue] = useState<string | undefined>();
+  const [timezoneValue, setTimezoneValue] = useState<string>("UTC");
 
   //set initial values
   useEffect(() => {
@@ -233,11 +250,20 @@ function ScheduledTaskForm({ task, runs }: { task: TestTask["task"]; runs: Sched
     setTimestampValue(initialRun.payload.timestamp);
     setLastTimestampValue(initialRun.payload.lastTimestamp);
     setExternalIdValue(initialRun.payload.externalId);
+    setTimezoneValue(initialRun.payload.timezone);
   }, [selectedCodeSampleId]);
 
   const [
     form,
-    { timestamp, lastTimestamp, externalId, triggerSource, taskIdentifier, environmentId },
+    {
+      timestamp,
+      lastTimestamp,
+      externalId,
+      triggerSource,
+      taskIdentifier,
+      environmentId,
+      timezone,
+    },
   ] = useForm({
     id: "test-task-scheduled",
     // TODO: type this
@@ -314,6 +340,30 @@ function ScheduledTaskForm({ task, runs }: { task: TestTask["task"]; runs: Sched
                 </Hint>
                 <FormError id={lastTimestamp.errorId}>{lastTimestamp.error}</FormError>
               </InputGroup>
+              <InputGroup>
+                <Label htmlFor={timezone.id}>Timezone</Label>
+                <Select
+                  {...conform.select(timezone)}
+                  placeholder="Select a timezone"
+                  defaultValue={timezoneValue}
+                  value={timezoneValue}
+                  setValue={(e) => {
+                    if (Array.isArray(e)) return;
+                    setTimezoneValue(e);
+                  }}
+                  items={possibleTimezones}
+                  filter={{ keys: [(item) => item.replace(/\//g, " ").replace(/_/g, " ")] }}
+                  dropdownIcon
+                  variant="tertiary/medium"
+                >
+                  {(matches) => <TimezoneList timezones={matches} />}
+                </Select>
+                <Hint>
+                  The Timestamp and Last timestamp are in UTC so this just changes the timezone
+                  string that comes through in the payload.
+                </Hint>
+                <FormError id={timezone.errorId}>{timezone.error}</FormError>
+              </InputGroup>
               <InputGroup>
                 <Label required={false} htmlFor={externalId.id}>
                   External ID

apps/webapp/app/routes/_app.timezones/route.tsx

@@ -0,0 +1,35 @@
+import { Link } from "@remix-run/react";
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { LogoIcon } from "~/components/LogoIcon";
+import { Header1 } from "~/components/primitives/Headers";
+import { Paragraph } from "~/components/primitives/Paragraph";
+import { getTimezones } from "~/utils/timezones.server";
+
+export const loader = async ({ request }: LoaderFunctionArgs) => {
+  return typedjson({
+    timezones: getTimezones(),
+  });
+};
+
+export default function Page() {
+  const { timezones } = useTypedLoaderData<typeof loader>();
+  return (
+    <div className="grid grid-rows-[2.5rem,1fr]">
+      <div className="flex items-center border-b border-b-grid-dimmed px-3">
+        <Link to="/">
+          <LogoIcon className="relative -top-px mr-2 h-4 w-4 min-w-[1rem]" />
+        </Link>
+      </div>
+      <div className="overflow-y-auto p-8 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
+        <Header1 spacing>Supported timezones</Header1>
+        <Paragraph spacing>We support these timezones when creating a schedule.</Paragraph>
+        <ul className="">
+          {timezones.map((timezone) => (
+            <li key={timezone}>{timezone}</li>
+          ))}
+        </ul>
+      </div>
+    </div>
+  );
+}

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.deployments.ts

@@ -1,5 +1,8 @@
 import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
-import { InitializeDeploymentRequestBody, InitializeDeploymentResponseBody } from "@trigger.dev/core/v3";
+import {
+  InitializeDeploymentRequestBody,
+  InitializeDeploymentResponseBody,
+} from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -37,13 +40,11 @@ export async function action({ request, params }: ActionFunctionArgs) {
     contentHash: deployment.contentHash,
     shortCode: deployment.shortCode,
     version: deployment.version,
-    externalBuildData: deployment.externalBuildData as InitializeDeploymentResponseBody["externalBuildData"],
+    externalBuildData:
+      deployment.externalBuildData as InitializeDeploymentResponseBody["externalBuildData"],
     imageTag,
-    registryHost: env.DEPLOY_REGISTRY_HOST
-  }
+    registryHost: env.DEPLOY_REGISTRY_HOST,
+  };
 
-  return json(
-    responseBody,
-    { status: 200 }
-  );
+  return json(responseBody, { status: 200 });
 }

apps/webapp/app/routes/api.v1.projects.$projectRef.envvars.$slug.$name.ts

@@ -123,7 +123,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
   const repository = new EnvironmentVariablesRepository();
 
-  const variables = await repository.getEnvironment(environment.project.id, environment.id, true);
+  const variables = await repository.getEnvironment(environment.project.id, environment.id);
 
   const environmentVariable = variables.find((v) => v.key === parsedParams.data.name);
 

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

@@ -80,7 +80,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
   const repository = new EnvironmentVariablesRepository();
 
-  const variables = await repository.getEnvironment(environment.project.id, environment.id, true);
+  const variables = await repository.getEnvironment(environment.project.id, environment.id);
 
   return json(variables.map((variable) => ({ name: variable.key, value: variable.value })));
 }

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

@@ -2,7 +2,7 @@ import { LoaderFunctionArgs, json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
+import { resolveVariablesForEnvironment } from "~/v3/environmentVariables/environmentVariablesRepository.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -41,9 +41,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return json({ error: "Project not found" }, { status: 404 });
   }
 
-  const repository = new EnvironmentVariablesRepository();
-
-  const variables = await repository.getEnvironmentVariables(project.id, authenticatedEnv.id);
+  const variables = await resolveVariablesForEnvironment(authenticatedEnv);
 
   return json({
     variables: variables.reduce((acc: Record<string, string>, variable) => {

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.v1.schedules.$scheduleId.ts

@@ -79,9 +79,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
           friendlyId: parsedParams.data.scheduleId,
           taskIdentifier: body.data.task,
           cron: body.data.cron,
+          timezone: body.data.timezone,
           environments: [authenticationResult.environment.id],
           externalId: body.data.externalId,
-          deduplicationKey: body.data.deduplicationKey,
         };
 
         const schedule = await service.call(authenticationResult.environment.projectId, options);
@@ -95,6 +95,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
             expression: schedule.cron,
             description: schedule.cronDescription,
           },
+          timezone: schedule.timezone,
           externalId: schedule.externalId ?? undefined,
           deduplicationKey: schedule.deduplicationKey,
           environments: schedule.environments,

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

@@ -43,6 +43,7 @@ export async function action({ request }: ActionFunctionArgs) {
       environments: [authenticationResult.environment.id],
       externalId: body.data.externalId,
       deduplicationKey: body.data.deduplicationKey,
+      timezone: body.data.timezone,
     };
 
     const schedule = await service.call(authenticationResult.environment.projectId, options);
@@ -56,6 +57,7 @@ export async function action({ request }: ActionFunctionArgs) {
         expression: schedule.cron,
         description: schedule.cronDescription,
       },
+      timezone: schedule.timezone,
       externalId: schedule.externalId ?? undefined,
       deduplicationKey: schedule.deduplicationKey,
       environments: schedule.environments,
@@ -111,6 +113,7 @@ export async function loader({ request }: LoaderFunctionArgs) {
         expression: schedule.cron,
         description: schedule.cronDescription,
       },
+      timezone: schedule.timezone,
       deduplicationKey: schedule.userProvidedDeduplicationKey
         ? schedule.deduplicationKey
         : undefined,

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

@@ -0,0 +1,28 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { apiCors } from "~/utils/apiCors";
+import { getTimezones } from "~/utils/timezones.server";
+
+const SearchParamsSchema = z.object({
+  excludeUtc: z.preprocess((value) => value === "true", z.boolean()).default(false),
+});
+
+export async function loader({ request }: LoaderFunctionArgs) {
+  if (request.method.toUpperCase() === "OPTIONS") {
+    return apiCors(request, json({}));
+  }
+
+  const rawSearchParams = new URL(request.url).searchParams;
+  const params = SearchParamsSchema.safeParse(Object.fromEntries(rawSearchParams.entries()));
+
+  if (!params.success) {
+    return apiCors(
+      request,
+      json({ error: "Invalid request parameters", issues: params.error.issues }, { status: 400 })
+    );
+  }
+
+  const timezones = getTimezones(!params.data.excludeUtc);
+  return apiCors(request, json({ timezones }));
+}

apps/webapp/app/routes/api.v1.usage.ingest.ts

@@ -0,0 +1,97 @@
+import { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { MachinePresetName } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { validateJWTTokenAndRenew } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { workerQueue } from "~/services/worker.server";
+import { machinePresetFromName } from "~/v3/machinePresets.server";
+import { reportUsageEvent } from "~/v3/openMeter.server";
+
+const JWTPayloadSchema = z.object({
+  environment_id: z.string(),
+  org_id: z.string(),
+  project_id: z.string(),
+  run_id: z.string(),
+  machine_preset: z.string(),
+});
+
+const BodySchema = z.object({
+  durationMs: z.number(),
+});
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const jwtResult = await validateJWTTokenAndRenew(request, JWTPayloadSchema);
+
+  if (!jwtResult) {
+    return { status: 401, body: "Unauthorized" };
+  }
+
+  const rawJson = await request.json();
+
+  const json = BodySchema.safeParse(rawJson);
+
+  if (!json.success) {
+    logger.error("Failed to parse request body", { rawJson });
+
+    return { status: 400, body: "Bad Request" };
+  }
+
+  const preset = machinePresetFromName(jwtResult.payload.machine_preset as MachinePresetName);
+
+  logger.debug("[/api/v1/usage/ingest] Reporting usage", { jwtResult, json: json.data, preset });
+
+  if (json.data.durationMs > 0) {
+    const costInCents = json.data.durationMs * preset.centsPerMs;
+
+    await prisma.taskRun.update({
+      where: {
+        id: jwtResult.payload.run_id,
+      },
+      data: {
+        usageDurationMs: {
+          increment: json.data.durationMs,
+        },
+        costInCents: {
+          increment: json.data.durationMs * preset.centsPerMs,
+        },
+      },
+    });
+
+    try {
+      await reportUsageEvent({
+        source: "webapp",
+        type: "usage",
+        subject: jwtResult.payload.org_id,
+        data: {
+          durationMs: json.data.durationMs,
+          costInCents: String(costInCents),
+        },
+      });
+    } catch (e) {
+      logger.error("Failed to report usage event, enqueing v3.reportUsage", { error: e });
+
+      await workerQueue.enqueue("v3.reportUsage", {
+        orgId: jwtResult.payload.org_id,
+        data: {
+          costInCents: String(costInCents),
+        },
+        additionalData: {
+          durationMs: json.data.durationMs,
+        },
+      });
+    }
+  }
+
+  return new Response(null, {
+    status: 200,
+    headers: {
+      "x-trigger-jwt": jwtResult.jwt,
+    },
+  });
+}

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/routes/resources.orgs.$organizationSlug.projects.$projectParam.schedules.new/route.tsx

@@ -3,9 +3,10 @@ import { parse } from "@conform-to/zod";
 import { CheckIcon, XMarkIcon } from "@heroicons/react/20/solid";
 import { Form, useActionData, useLocation, useNavigation } from "@remix-run/react";
 import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { useVirtualizer } from "@tanstack/react-virtual";
 import { parseExpression } from "cron-parser";
 import cronstrue from "cronstrue";
-import { useState } from "react";
+import { useRef, useState } from "react";
 import {
   environmentTextClassName,
   environmentTitle,
@@ -42,6 +43,7 @@ import { ProjectParamSchema, docsPath, v3SchedulesPath } from "~/utils/pathBuild
 import { CronPattern, UpsertSchedule } from "~/v3/schedules";
 import { UpsertTaskScheduleService } from "~/v3/services/upsertTaskSchedule.server";
 import { AIGeneratedCronField } from "../resources.orgs.$organizationSlug.projects.$projectParam.schedules.new.natural-language";
+import { TimezoneList } from "~/components/scheduled/timezones";
 
 const cronFormat = `*    *    *    *    *
 ┬    ┬    ┬    ┬    ┬
@@ -117,9 +119,12 @@ export function UpsertScheduleForm({
   schedule,
   possibleTasks,
   possibleEnvironments,
+  possibleTimezones,
   showGenerateField,
 }: EditableScheduleElements & { showGenerateField: boolean }) {
   const lastSubmission = useActionData();
+  const [selectedTimezone, setSelectedTimezone] = useState<string>(schedule?.timezone ?? "UTC");
+  const isUtc = selectedTimezone === "UTC";
   const [cronPattern, setCronPattern] = useState<string>(schedule?.cron ?? "");
   const navigation = useNavigation();
   const isLoading = navigation.state !== "idle";
@@ -127,18 +132,20 @@ export function UpsertScheduleForm({
   const project = useProject();
   const location = useLocation();
 
-  const [form, { taskIdentifier, cron, externalId, environments, deduplicationKey }] = useForm({
-    id: "create-schedule",
-    // TODO: type this
-    lastSubmission: lastSubmission as any,
-    shouldRevalidate: "onSubmit",
-    onValidate({ formData }) {
-      return parse(formData, { schema: UpsertSchedule });
-    },
-  });
+  const [form, { taskIdentifier, cron, timezone, externalId, environments, deduplicationKey }] =
+    useForm({
+      id: "create-schedule",
+      // TODO: type this
+      lastSubmission: lastSubmission as any,
+      shouldRevalidate: "onSubmit",
+      onValidate({ formData }) {
+        return parse(formData, { schema: UpsertSchedule });
+      },
+    });
 
   let cronPatternResult: CronPatternResult | undefined = undefined;
   let nextRuns: Date[] | undefined = undefined;
+
   if (cronPattern !== "") {
     const result = CronPattern.safeParse(cronPattern);
 
@@ -149,7 +156,10 @@ export function UpsertScheduleForm({
       };
     } else {
       try {
-        const expression = parseExpression(cronPattern, { utc: true });
+        const expression = parseExpression(
+          cronPattern,
+          isUtc ? { utc: true } : { tz: selectedTimezone }
+        );
         cronPatternResult = {
           isValid: true,
           description: cronstrue.toString(cronPattern),
@@ -195,6 +205,7 @@ export function UpsertScheduleForm({
                 items={possibleTasks}
                 filter={(task, search) => task.toLowerCase().includes(search.toLowerCase())}
                 dropdownIcon
+                variant="tertiary/medium"
               >
                 {(matches) => (
                   <>
@@ -241,25 +252,52 @@ export function UpsertScheduleForm({
                 <ValidCronMessage isValid={false} message={cronPatternResult.error} />
               )}
             </InputGroup>
+            <InputGroup>
+              <Label htmlFor={timezone.id}>Timezone</Label>
+              <Select
+                {...conform.select(timezone)}
+                placeholder="Select a timezone"
+                defaultValue={selectedTimezone}
+                value={selectedTimezone}
+                setValue={(e) => {
+                  if (Array.isArray(e)) return;
+                  setSelectedTimezone(e);
+                }}
+                items={possibleTimezones}
+                filter={{ keys: [(item) => item.replace(/\//g, " ").replace(/_/g, " ")] }}
+                dropdownIcon
+                variant="tertiary/medium"
+              >
+                {(matches) => <TimezoneList timezones={matches} />}
+              </Select>
+              <Hint>
+                {isUtc
+                  ? "UTC will not change with daylight savings time."
+                  : "This will automatically adjust for daylight savings time."}
+              </Hint>
+              <FormError id={timezone.errorId}>{timezone.error}</FormError>
+            </InputGroup>
             {nextRuns !== undefined && (
               <div className="flex flex-col gap-1">
                 <Header3>Next 5 runs</Header3>
                 <Table>
                   <TableHeader>
                     <TableRow>
+                      {!isUtc && <TableHeaderCell>{selectedTimezone}</TableHeaderCell>}
                       <TableHeaderCell>UTC</TableHeaderCell>
-                      <TableHeaderCell>Local time</TableHeaderCell>
                     </TableRow>
                   </TableHeader>
                   <TableBody>
                     {nextRuns.map((run, index) => (
                       <TableRow key={index}>
+                        {!isUtc && (
+                          <TableCell>
+                            <DateTime date={run} timeZone={selectedTimezone} />
+                          </TableCell>
+                        )}
                         <TableCell>
                           <DateTime date={run} timeZone="UTC" />
                         </TableCell>
-                        <TableCell>
-                          <DateTime date={run} />
-                        </TableCell>
                       </TableRow>
                     ))}
                   </TableBody>

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

@@ -1,6 +1,7 @@
 import { Prettify } from "@trigger.dev/core";
 import { z } from "zod";
 import {
+  RuntimeEnvironment,
   findEnvironmentByApiKey,
   findEnvironmentByPublicApiKey,
 } from "~/models/runtimeEnvironment.server";
@@ -12,6 +13,9 @@ import {
 import { prisma } from "~/db.server";
 import { json } from "@remix-run/server-runtime";
 import { findProjectByRef } from "~/models/project.server";
+import { SignJWT, jwtVerify, errors } from "jose";
+import { env } from "~/env.server";
+import { logger } from "./logger.server";
 
 type Optional<T, K extends keyof T> = Prettify<Omit<T, K> & Partial<Pick<T, K>>>;
 
@@ -209,3 +213,127 @@ export async function authenticatedEnvironmentForAuthentication(
     }
   }
 }
+
+const JWT_SECRET = new TextEncoder().encode(env.SESSION_SECRET);
+const JWT_ALGORITHM = "HS256";
+const DEFAULT_JWT_EXPIRATION_IN_MS = 1000 * 60 * 60; // 1 hour
+
+export async function generateJWTTokenForEnvironment(
+  environment: RuntimeEnvironment,
+  payload: Record<string, string>
+) {
+  const jwt = await new SignJWT({
+    environment_id: environment.id,
+    org_id: environment.organizationId,
+    project_id: environment.projectId,
+    ...payload,
+  })
+    .setProtectedHeader({ alg: JWT_ALGORITHM })
+    .setIssuedAt()
+    .setIssuer("https://id.trigger.dev")
+    .setAudience("https://api.trigger.dev")
+    .setExpirationTime(calculateJWTExpiration())
+    .sign(JWT_SECRET);
+
+  return jwt;
+}
+
+export async function validateJWTTokenAndRenew<T extends z.ZodTypeAny>(
+  request: Request,
+  payloadSchema: T
+): Promise<{ payload: z.infer<T>; jwt: string } | undefined> {
+  try {
+    const jwt = request.headers.get("x-trigger-jwt");
+
+    if (!jwt) {
+      logger.debug("Missing JWT token in request", {
+        headers: Object.fromEntries(request.headers),
+      });
+
+      return;
+    }
+
+    const { payload: rawPayload } = await jwtVerify(jwt, JWT_SECRET, {
+      issuer: "https://id.trigger.dev",
+      audience: "https://api.trigger.dev",
+    });
+
+    const payload = payloadSchema.safeParse(rawPayload);
+
+    if (!payload.success) {
+      logger.error("Failed to validate JWT", { payload: rawPayload, issues: payload.error.issues });
+
+      return;
+    }
+
+    const renewedJwt = await renewJWTToken(payload.data);
+
+    return {
+      payload: payload.data,
+      jwt: renewedJwt,
+    };
+  } catch (error) {
+    if (error instanceof errors.JWTExpired) {
+      // Now we need to try and renew the token using the API key auth
+      const authenticatedEnv = await authenticateApiRequest(request);
+
+      if (!authenticatedEnv) {
+        logger.error("Failed to renew JWT token, missing or invalid Authorization header", {
+          error: error.message,
+        });
+
+        return;
+      }
+
+      const payload = payloadSchema.safeParse(error.payload);
+
+      if (!payload.success) {
+        logger.error("Failed to parse jwt payload after expired", {
+          payload: error.payload,
+          issues: payload.error.issues,
+        });
+
+        return;
+      }
+
+      const renewedJwt = await generateJWTTokenForEnvironment(authenticatedEnv.environment, {
+        ...payload.data,
+      });
+
+      logger.debug("Renewed JWT token from Authorization header API Key", {
+        environment: authenticatedEnv.environment,
+        payload: payload.data,
+      });
+
+      return {
+        payload: payload.data,
+        jwt: renewedJwt,
+      };
+    }
+
+    logger.error("Failed to validate JWT token", { error });
+  }
+}
+
+async function renewJWTToken(payload: Record<string, string>) {
+  const jwt = await new SignJWT(payload)
+    .setProtectedHeader({ alg: JWT_ALGORITHM })
+    .setIssuedAt()
+    .setIssuer("https://id.trigger.dev")
+    .setAudience("https://api.trigger.dev")
+    .setExpirationTime(calculateJWTExpiration())
+    .sign(JWT_SECRET);
+
+  return jwt;
+}
+
+function calculateJWTExpiration() {
+  if (env.PROD_USAGE_HEARTBEAT_INTERVAL_MS) {
+    return (
+      (Date.now() + Math.max(DEFAULT_JWT_EXPIRATION_IN_MS, env.PROD_USAGE_HEARTBEAT_INTERVAL_MS)) /
+      1000
+    );
+  }
+
+  return (Date.now() + DEFAULT_JWT_EXPIRATION_IN_MS) / 1000;
+}

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
@@ -196,11 +145,13 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
     "/api/internal/stripe_webhooks",
     "/api/v1/authorization-code",
     "/api/v1/token",
+    "/api/v1/usage/ingest",
     /^\/api\/v1\/tasks\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/tasks/$id/callback/$secret
     /^\/api\/v1\/runs\/[^\/]+\/tasks\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/runs/$runId/tasks/$id/callback/$secret
     /^\/api\/v1\/http-endpoints\/[^\/]+\/env\/[^\/]+\/[^\/]+$/, // /api/v1/http-endpoints/$httpEndpointId/env/$envType/$shortcode
     /^\/api\/v1\/sources\/http\/[^\/]+$/, // /api/v1/sources/http/$id
     /^\/api\/v1\/endpoints\/[^\/]+\/[^\/]+\/index\/[^\/]+$/, // /api/v1/endpoints/$environmentId/$endpointSlug/index/$indexHookIdentifier
+    "/api/v1/timezones",
   ],
   log: {
     rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",