Release 3.0.0-beta.46

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

.changeset/fast-colts-relax.md

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

.changeset/friendly-walls-repair.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add sox and audiowaveform binaries to worker images

.changeset/mighty-eggs-grab.md

@@ -0,0 +1,30 @@
+---
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Tasks should now be much more robust and resilient to reconnects during crucial operations and other failure scenarios.
+
+Task runs now have to signal checkpointable state prior to ALL checkpoints. This ensures flushing always happens.
+
+All important socket.io RPCs will now be retried with backoff. Actions relying on checkpoints will be replayed if we haven't been checkpointed and restored as expected, e.g. after reconnect.
+
+Other changes:
+
+- Fix retry check in shared queue
+- Fix env var sync spinner
+- Heartbeat between retries
+- Fix retry prep
+- Fix prod worker no tasks detection
+- Fail runs above `MAX_TASK_RUN_ATTEMPTS`
+- Additional debug logs in all places
+- Prevent crashes due to failed socket schema parsing
+- Remove core-apps barrel
+- Upgrade socket.io-client to fix an ACK memleak
+- Additional index failure logs
+- Prevent message loss during reconnect
+- Prevent burst of heartbeats on reconnect
+- Prevent crash on failed cleanup
+- Handle at-least-once lazy execute message delivery
+- Handle uncaught entry point exceptions

.changeset/modern-stingrays-end.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+v3: Trigger delayed runs and reschedule them

.changeset/nervous-planets-sparkle.md

@@ -0,0 +1,9 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Improve non-zero exit code error messages
+- Detect OOM conditions within worker child processes
+- Internal errors can have optional stack traces
+- Docker provider can be set to enforce machine presets

.changeset/polite-pots-walk.md

@@ -0,0 +1,7 @@
+---
+"trigger.dev": patch
+---
+
+Add e2e fixtures corresponding to past issues
+Implement e2e suite parallelism
+Enhance log level for specific e2e suite messages

.changeset/pre.json

@@ -65,8 +65,10 @@
     "eight-pumas-float",
     "eleven-paws-join",
     "famous-boats-tease",
+    "fast-colts-relax",
     "few-students-share",
     "five-toes-destroy",
+    "friendly-walls-repair",
     "funny-swans-destroy",
     "gorgeous-gorillas-compete",
     "green-bags-wink",
@@ -88,9 +90,12 @@
     "lovely-drinks-flash",
     "many-ligers-pump",
     "mighty-camels-joke",
+    "mighty-eggs-grab",
     "mighty-flowers-train",
     "mighty-parrots-sin",
+    "modern-stingrays-end",
     "nasty-jars-pump",
+    "nervous-planets-sparkle",
     "new-pants-beg",
     "new-rivers-tell",
     "nice-bulldogs-turn",
@@ -100,9 +105,11 @@
     "plenty-ducks-beam",
     "polite-ducks-switch",
     "polite-pears-grow",
+    "polite-pots-walk",
     "polite-rockets-matter",
     "poor-flowers-cross",
     "purple-garlics-shop",
+    "purple-spiders-care",
     "rare-lamps-promise",
     "rare-roses-float",
     "real-planets-stare",
@@ -117,9 +124,11 @@
     "sharp-zebras-serve",
     "shiny-coats-cry",
     "silly-buses-obey",
+    "silly-forks-kiss",
     "silly-suits-switch",
     "silver-doors-juggle",
     "six-ligers-exist",
+    "six-rats-hunt",
     "sixty-insects-watch",
     "slow-buses-own",
     "slow-kiwis-hide",
@@ -127,6 +136,7 @@
     "smart-needles-move",
     "smart-olives-eat",
     "sour-pugs-teach",
+    "spicy-frogs-remain",
     "spicy-lamps-smoke",
     "spicy-terms-bow",
     "strange-ghosts-matter",
@@ -140,10 +150,13 @@
     "sweet-lizards-press",
     "swift-dragons-peel",
     "tall-bees-wave",
+    "tall-masks-repeat",
+    "tame-apricots-clap",
     "tame-guests-know",
     "tender-moose-tell",
     "tender-oranges-rhyme",
     "tender-turkeys-compete",
+    "thick-carrots-sneeze",
     "thin-parents-heal",
     "thirty-islands-kiss",
     "tidy-balloons-suffer",
@@ -154,6 +167,7 @@
     "tricky-bulldogs-heal",
     "tricky-keys-attack",
     "tricky-ladybugs-unite",
+    "twelve-knives-notice",
     "two-pumas-wait",
     "violet-clocks-notice",
     "warm-olives-provide",

.changeset/purple-spiders-care.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Await file watcher cleanup in dev

.changeset/silly-forks-kiss.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+v3: Copy over more of the project's package.json keys into the deployed package.json (support for custom config like zenstack)

.changeset/six-rats-hunt.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Improved ESM module require error detection logic

.changeset/spicy-frogs-remain.md

@@ -0,0 +1,10 @@
+---
+"trigger.dev": patch
+---
+
+- Prevent downgrades during update check and advise to upgrade CLI
+- Detect bun and use npm instead
+- During init, fail early and advise if not a TypeScript project
+- During init, allow specifying custom package manager args
+- Add links to dev worker started message
+- Fix links in unsupported terminals

.changeset/tall-masks-repeat.md

@@ -0,0 +1,9 @@
+---
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Fix artifact detection logs
+- Fix OOM detection and error messages
+- Add test link to cli deployment completion

.changeset/tame-apricots-clap.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+v3: postInstall config option now replaces the postinstall script found in package.json

.changeset/thick-carrots-sneeze.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+v3: Include presigned urls for downloading large payloads and outputs when using runs.retrieve

.changeset/twelve-knives-notice.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+v3: fix missing init output in task run function when no middleware is defined

.env.example

@@ -71,7 +71,7 @@ COORDINATOR_SECRET=coordinator-secret # generate the actual secret with `openssl
 # OBJECT_STORE_BASE_URL="https://{bucket}.{accountId}.r2.cloudflarestorage.com"
 # OBJECT_STORE_ACCESS_KEY_ID=
 # OBJECT_STORE_SECRET_ACCESS_KEY=
-# RUNTIME_WAIT_THRESHOLD_IN_MS=10000
+# CHECKPOINT_THRESHOLD_IN_MS=10000
 
 # These control the server-side internal telemetry
 # INTERNAL_OTEL_TRACE_EXPORTER_URL=<URL to send traces to>

.github/workflows/e2e.yml

@@ -23,7 +23,7 @@ jobs:
           fetch-depth: 0
 
       - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v2.2.4
+        uses: pnpm/action-setup@v4
         with:
           version: 8.15.5
 
@@ -63,7 +63,7 @@ jobs:
           submodules: recursive
 
       - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v2.2.4
+        uses: pnpm/action-setup@v4
         with:
           version: 8.15.5
 

.github/workflows/publish.yml

@@ -49,11 +49,11 @@ jobs:
     uses: ./.github/workflows/unit-tests.yml
     secrets: inherit
 
-  e2e:
-    uses: ./.github/workflows/e2e.yml
-    with:
-      package: cli-v3
-    secrets: inherit
+  # e2e:
+  #   uses: ./.github/workflows/e2e.yml
+  #   with:
+  #     package: cli-v3
+  #   secrets: inherit
 
   publish:
     needs: [typecheck, units]

.github/workflows/release.yml

@@ -28,7 +28,7 @@ jobs:
           fetch-depth: 0
 
       - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v2.2.4
+        uses: pnpm/action-setup@v4
         with:
           version: 8.15.5
 

.github/workflows/typecheck.yml

@@ -12,7 +12,7 @@ jobs:
           fetch-depth: 0
 
       - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v2.2.4
+        uses: pnpm/action-setup@v4
         with:
           version: 8.15.5
 

.github/workflows/unit-tests.yml

@@ -12,7 +12,7 @@ jobs:
           fetch-depth: 0
 
       - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v2.2.4
+        uses: pnpm/action-setup@v4
         with:
           version: 8.15.5
 

apps/coordinator/package.json

@@ -21,8 +21,7 @@
     "execa": "^8.0.1",
     "nanoid": "^5.0.6",
     "prom-client": "^15.1.0",
-    "socket.io": "4.7.4",
-    "socket.io-client": "4.7.4"
+    "socket.io": "4.7.4"
   },
   "devDependencies": {
     "@types/node": "^18",

apps/coordinator/src/chaosMonkey.ts

@@ -0,0 +1,95 @@
+import type { Execa$ } from "execa";
+import { setTimeout as timeout } from "node:timers/promises";
+
+class ChaosMonkeyError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "ChaosMonkeyError";
+  }
+}
+
+export class ChaosMonkey {
+  private chaosEventRate = 0.2;
+  private delayInSeconds = 45;
+
+  constructor(private enabled = false) {
+    if (this.enabled) {
+      console.log("🍌 Chaos monkey enabled");
+    }
+  }
+
+  static Error = ChaosMonkeyError;
+
+  enable() {
+    this.enabled = true;
+    console.log("🍌 Chaos monkey enabled");
+  }
+
+  disable() {
+    this.enabled = false;
+    console.log("🍌 Chaos monkey disabled");
+  }
+
+  async call({
+    $,
+    throwErrors = true,
+    addDelays = true,
+  }: {
+    $?: Execa$<string>;
+    throwErrors?: boolean;
+    addDelays?: boolean;
+  } = {}) {
+    if (!this.enabled) {
+      return;
+    }
+
+    const random = Math.random();
+
+    if (random > this.chaosEventRate) {
+      // Don't interfere with normal operation
+      return;
+    }
+
+    const chaosEvents: Array<() => Promise<any>> = [];
+
+    if (addDelays) {
+      chaosEvents.push(async () => {
+        console.log("🍌 Chaos monkey: Add delay");
+
+        if ($) {
+          await $`sleep ${this.delayInSeconds}`;
+        } else {
+          await timeout(this.delayInSeconds * 1000);
+        }
+      });
+    }
+
+    if (throwErrors) {
+      chaosEvents.push(async () => {
+        console.log("🍌 Chaos monkey: Throw error");
+
+        if ($) {
+          await $`false`;
+        } else {
+          throw new ChaosMonkey.Error("🍌 Chaos monkey: Throw error");
+        }
+      });
+    }
+
+    if (chaosEvents.length === 0) {
+      console.error("🍌 Chaos monkey: No events selected");
+      return;
+    }
+
+    const randomIndex = Math.floor(Math.random() * chaosEvents.length);
+
+    const chaosEvent = chaosEvents[randomIndex];
+
+    if (!chaosEvent) {
+      console.error("🍌 Chaos monkey: No event found");
+      return;
+    }
+
+    await chaosEvent();
+  }
+}

apps/coordinator/src/checkpointer.ts

@@ -0,0 +1,587 @@
+import { ExponentialBackoff } from "@trigger.dev/core-apps/backoff";
+import { isExecaChildProcess, testDockerCheckpoint } from "@trigger.dev/core-apps/checkpoints";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
+import { $ } from "execa";
+import { nanoid } from "nanoid";
+import fs from "node:fs/promises";
+import { ChaosMonkey } from "./chaosMonkey";
+
+type CheckpointerInitializeReturn = {
+  canCheckpoint: boolean;
+  willSimulate: boolean;
+};
+
+type CheckpointAndPushOptions = {
+  runId: string;
+  leaveRunning?: boolean;
+  projectRef: string;
+  deploymentVersion: string;
+  shouldHeartbeat?: boolean;
+};
+
+type CheckpointAndPushResult =
+  | { success: true; checkpoint: CheckpointData }
+  | {
+      success: false;
+      reason?: "CANCELED" | "DISABLED" | "ERROR" | "IN_PROGRESS" | "NO_SUPPORT" | "SKIP_RETRYING";
+    };
+
+type CheckpointData = {
+  location: string;
+  docker: boolean;
+};
+
+type CheckpointerOptions = {
+  dockerMode: boolean;
+  forceSimulate: boolean;
+  heartbeat: (runId: string) => void;
+  registryHost?: string;
+  registryNamespace?: string;
+  registryTlsVerify?: boolean;
+  disableCheckpointSupport?: boolean;
+  checkpointPath?: string;
+  simulateCheckpointFailure?: boolean;
+  simulateCheckpointFailureSeconds?: number;
+  simulatePushFailure?: boolean;
+  simulatePushFailureSeconds?: number;
+  chaosMonkey?: ChaosMonkey;
+};
+
+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,
+  };
+}
+
+export class Checkpointer {
+  #initialized = false;
+  #canCheckpoint = false;
+  #dockerMode: boolean;
+
+  #logger = new SimpleLogger("[checkptr]");
+  #abortControllers = new Map<string, AbortController>();
+  #failedCheckpoints = new Map<string, unknown>();
+  #waitingForRetry = new Set<string>();
+
+  private registryHost: string;
+  private registryNamespace: string;
+  private registryTlsVerify: boolean;
+
+  private disableCheckpointSupport: boolean;
+  private checkpointPath: string;
+
+  private simulateCheckpointFailure: boolean;
+  private simulateCheckpointFailureSeconds: number;
+  private simulatePushFailure: boolean;
+  private simulatePushFailureSeconds: number;
+
+  private chaosMonkey: ChaosMonkey;
+
+  constructor(private opts: CheckpointerOptions) {
+    this.#dockerMode = opts.dockerMode;
+
+    this.registryHost = opts.registryHost ?? "localhost:5000";
+    this.registryNamespace = opts.registryNamespace ?? "trigger";
+    this.registryTlsVerify = opts.registryTlsVerify ?? true;
+
+    this.disableCheckpointSupport = opts.disableCheckpointSupport ?? false;
+    this.checkpointPath = opts.checkpointPath ?? "/checkpoints";
+
+    this.simulateCheckpointFailure = opts.simulateCheckpointFailure ?? false;
+    this.simulateCheckpointFailureSeconds = opts.simulateCheckpointFailureSeconds ?? 300;
+    this.simulatePushFailure = opts.simulatePushFailure ?? false;
+    this.simulatePushFailureSeconds = opts.simulatePushFailureSeconds ?? 300;
+
+    this.chaosMonkey = opts.chaosMonkey ?? new ChaosMonkey(!!process.env.CHAOS_MONKEY_ENABLED);
+  }
+
+  async init(): Promise<CheckpointerInitializeReturn> {
+    if (this.#initialized) {
+      return this.#getInitReturn(this.#canCheckpoint);
+    }
+
+    this.#logger.log(`${this.#dockerMode ? "Docker" : "Kubernetes"} mode`);
+
+    if (this.#dockerMode) {
+      const testCheckpoint = await testDockerCheckpoint();
+
+      if (testCheckpoint.ok) {
+        return this.#getInitReturn(true);
+      }
+
+      this.#logger.error(testCheckpoint.message, testCheckpoint.error ?? "");
+      return this.#getInitReturn(false);
+    } else {
+      try {
+        await $`buildah login --get-login ${this.registryHost}`;
+      } catch (error) {
+        this.#logger.error(`No checkpoint support: Not logged in to registry ${this.registryHost}`);
+        return this.#getInitReturn(false);
+      }
+    }
+
+    return this.#getInitReturn(true);
+  }
+
+  #getInitReturn(canCheckpoint: boolean): CheckpointerInitializeReturn {
+    this.#canCheckpoint = canCheckpoint;
+
+    if (canCheckpoint) {
+      if (!this.#initialized) {
+        this.#logger.log("Full checkpoint support!");
+      }
+    }
+
+    this.#initialized = true;
+
+    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,
+      willSimulate,
+    };
+  }
+
+  #getImageRef(projectRef: string, deploymentVersion: string, shortCode: string) {
+    return `${this.registryHost}/${this.registryNamespace}/${projectRef}:${deploymentVersion}.prod-${shortCode}`;
+  }
+
+  #getExportLocation(projectRef: string, deploymentVersion: string, shortCode: string) {
+    const basename = `${projectRef}-${deploymentVersion}-${shortCode}`;
+
+    if (this.#dockerMode) {
+      return basename;
+    } else {
+      return `${this.checkpointPath}/${basename}.tar`;
+    }
+  }
+
+  async checkpointAndPush(opts: CheckpointAndPushOptions): Promise<CheckpointData | undefined> {
+    const start = performance.now();
+    this.#logger.log(`checkpointAndPush() start`, { start, opts });
+
+    let interval: NodeJS.Timer | undefined;
+
+    if (opts.shouldHeartbeat) {
+      interval = setInterval(() => {
+        this.#logger.log("Sending heartbeat", { runId: opts.runId });
+        this.opts.heartbeat(opts.runId);
+      }, 20_000);
+    }
+
+    try {
+      const result = await this.#checkpointAndPushWithBackoff(opts);
+
+      const end = performance.now();
+      this.#logger.log(`checkpointAndPush() end`, {
+        start,
+        end,
+        diff: end - start,
+        opts,
+        success: result.success,
+      });
+
+      if (!result.success) {
+        return;
+      }
+
+      return result.checkpoint;
+    } finally {
+      if (opts.shouldHeartbeat) {
+        clearInterval(interval);
+      }
+    }
+  }
+
+  isCheckpointing(runId: string) {
+    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) {
+      this.#logger.debug("Nothing to cancel", { runId });
+      return false;
+    }
+
+    controller.abort("cancelCheckpointing()");
+    this.#abortControllers.delete(runId);
+
+    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<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 { success: false, reason: "NO_SUPPORT" };
+    }
+
+    if (this.isCheckpointing(runId)) {
+      this.#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 (this.disableCheckpointSupport) {
+      this.#logger.error("Checkpoint support disabled", { options });
+      return { success: false, reason: "DISABLED" };
+    }
+
+    const controller = new AbortController();
+    this.#abortControllers.set(runId, controller);
+
+    const $$ = $({ signal: controller.signal });
+
+    const shortCode = nanoid(8);
+    const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
+    const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
+
+    const cleanup = async () => {
+      if (this.#dockerMode) {
+        return;
+      }
+
+      try {
+        await $`rm ${exportLocation}`;
+        this.#logger.log("Deleted checkpoint archive", { exportLocation });
+
+        await $`buildah rmi ${imageRef}`;
+        this.#logger.log("Deleted checkpoint image", { imageRef });
+      } catch (error) {
+        this.#logger.error("Failure during checkpoint cleanup", { exportLocation, error });
+      }
+    };
+
+    try {
+      await this.chaosMonkey.call({ $: $$ });
+
+      this.#logger.log("Checkpointing:", { options });
+
+      const containterName = this.#getRunContainerName(runId);
+
+      // Create checkpoint (docker)
+      if (this.#dockerMode) {
+        try {
+          if (this.opts.forceSimulate || !this.#canCheckpoint) {
+            this.#logger.log("Simulating checkpoint");
+            this.#logger.debug(await $$`docker pause ${containterName}`);
+          } else {
+            if (this.simulateCheckpointFailure) {
+              if (performance.now() < this.simulateCheckpointFailureSeconds * 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}`
+              );
+            } else {
+              this.#logger.debug(
+                await $$`docker checkpoint create ${containterName} ${exportLocation}`
+              );
+            }
+          }
+        } catch (error) {
+          this.#logger.error("Failed while creating docker checkpoint", { exportLocation });
+          throw error;
+        }
+
+        this.#logger.log("checkpoint created:", {
+          runId,
+          location: exportLocation,
+        });
+
+        return {
+          success: true,
+          checkpoint: {
+            location: exportLocation,
+            docker: true,
+          },
+        };
+      }
+
+      // Create checkpoint (CRI)
+      if (!this.#canCheckpoint) {
+        this.#logger.error("No checkpoint support in kubernetes mode.");
+        return { success: false, reason: "SKIP_RETRYING" };
+      }
+
+      const containerId = this.#logger.debug(
+        // @ts-expect-error
+        await $$`crictl ps`
+          .pipeStdout($$({ stdin: "pipe" })`grep ${containterName}`)
+          .pipeStdout($$({ stdin: "pipe" })`cut -f1 ${"-d "}`)
+      );
+
+      if (!containerId.stdout) {
+        this.#logger.error("could not find container id", { options, containterName });
+        return { success: false, reason: "SKIP_RETRYING" };
+      }
+
+      const start = performance.now();
+
+      if (this.simulateCheckpointFailure) {
+        if (performance.now() < this.simulateCheckpointFailureSeconds * 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 (this.simulatePushFailure) {
+        if (performance.now() < this.simulatePushFailureSeconds * 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=${String(this.registryTlsVerify)} ${imageRef}`
+      );
+      const postPush = performance.now();
+
+      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,
+      };
+
+      this.#logger.log("Checkpointed and pushed image to:", { location: imageRef, perf });
+
+      return {
+        success: true,
+        checkpoint: {
+          location: imageRef,
+          docker: false,
+        },
+      };
+    } catch (error) {
+      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}`;
+  }
+}

apps/docker-provider/package.json

@@ -18,8 +18,7 @@
   "dependencies": {
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/core-apps": "workspace:*",
-    "execa": "^8.0.1",
-    "socket.io-client": "^4.7.4"
+    "execa": "^8.0.1"
   },
   "devDependencies": {
     "@types/node": "^18.19.8",

apps/docker-provider/src/index.ts

@@ -1,14 +1,13 @@
 import { $, type ExecaChildProcess, execa } from "execa";
 import {
-  SimpleLogger,
-  TaskOperations,
   ProviderShell,
-  TaskOperationsRestoreOptions,
+  TaskOperations,
   TaskOperationsCreateOptions,
   TaskOperationsIndexOptions,
-  isExecaChildProcess,
-  testDockerCheckpoint,
-} from "@trigger.dev/core-apps";
+  TaskOperationsRestoreOptions,
+} from "@trigger.dev/core-apps/provider";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
+import { isExecaChildProcess, testDockerCheckpoint } from "@trigger.dev/core-apps/checkpoints";
 import { setTimeout } from "node:timers/promises";
 import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
 
@@ -54,13 +53,16 @@ class DockerTaskOperations implements TaskOperations {
   }
 
   #getInitReturn(canCheckpoint: boolean): TaskOperationsInitReturn {
-    this.#initialized = true;
     this.#canCheckpoint = canCheckpoint;
 
     if (canCheckpoint) {
-      logger.log("Full checkpoint support!");
+      if (!this.#initialized) {
+        logger.log("Full checkpoint support!");
+      }
     }
 
+    this.#initialized = true;
+
     const willSimulate = !canCheckpoint || this.opts.forceSimulate;
 
     if (willSimulate) {
@@ -109,22 +111,27 @@ class DockerTaskOperations implements TaskOperations {
 
     const containerName = this.#getRunContainerName(opts.runId);
 
+    const runArgs = [
+      "run",
+      "--network=host",
+      "--detach",
+      `--env=TRIGGER_ENV_ID=${opts.envId}`,
+      `--env=TRIGGER_RUN_ID=${opts.runId}`,
+      `--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}`,
+    ];
+
+    if (process.env.ENFORCE_MACHINE_PRESETS) {
+      runArgs.push(`--cpus=${opts.machine.cpu}`, `--memory=${opts.machine.memory}G`);
+    }
+
+    runArgs.push(`${opts.image}`);
+
     try {
-      logger.debug(
-        await execa("docker", [
-          "run",
-          "--network=host",
-          "--detach",
-          `--env=TRIGGER_ENV_ID=${opts.envId}`,
-          `--env=TRIGGER_RUN_ID=${opts.runId}`,
-          `--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.image}`,
-        ])
-      );
+      logger.debug(await execa("docker", runArgs));
     } catch (error) {
       if (!isExecaChildProcess(error)) {
         throw error;

apps/kubernetes-provider/package.json

@@ -19,8 +19,7 @@
     "@kubernetes/client-node": "^0.20.0",
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/core-apps": "workspace:*",
-    "p-queue": "^8.0.1",
-    "socket.io-client": "^4.7.4"
+    "p-queue": "^8.0.1"
   },
   "devDependencies": {
     "dotenv": "^16.4.2",

apps/kubernetes-provider/src/index.ts

@@ -1,12 +1,12 @@
 import * as k8s from "@kubernetes/client-node";
 import {
   ProviderShell,
-  SimpleLogger,
   TaskOperations,
   TaskOperationsCreateOptions,
   TaskOperationsIndexOptions,
   TaskOperationsRestoreOptions,
-} from "@trigger.dev/core-apps";
+} from "@trigger.dev/core-apps/provider";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
 import {
   MachinePreset,
   PostStartCauses,
@@ -16,13 +16,21 @@ import {
 import { randomUUID } from "crypto";
 import { TaskMonitor } from "./taskMonitor";
 import { PodCleaner } from "./podCleaner";
+import { UptimeHeartbeat } from "./uptimeHeartbeat";
 
 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 UPTIME_HEARTBEAT_URL = process.env.UPTIME_HEARTBEAT_URL;
+const UPTIME_INTERVAL_SECONDS = Number(process.env.UPTIME_INTERVAL_SECONDS || "60");
+const UPTIME_MAX_PENDING_RUNS = Number(process.env.UPTIME_MAX_PENDING_RUNS || "25");
+const UPTIME_MAX_PENDING_INDECES = Number(process.env.UPTIME_MAX_PENDING_INDECES || "10");
+const UPTIME_MAX_PENDING_ERRORS = Number(process.env.UPTIME_MAX_PENDING_ERRORS || "10");
+
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 logger.log(`running in ${RUNTIME_ENV} mode`);
 
@@ -529,27 +537,28 @@ provider.listen();
 
 const taskMonitor = new TaskMonitor({
   runtimeEnv: RUNTIME_ENV,
-  onIndexFailure: async (deploymentId, failureInfo) => {
-    logger.log("Indexing failed", { deploymentId, failureInfo });
+  onIndexFailure: async (deploymentId, details) => {
+    logger.log("Indexing failed", { deploymentId, details });
 
     try {
       provider.platformSocket.send("INDEXING_FAILED", {
         deploymentId,
         error: {
-          name: `Crashed with exit code ${failureInfo.exitCode}`,
-          message: failureInfo.reason,
-          stack: failureInfo.logs,
+          name: `Crashed with exit code ${details.exitCode}`,
+          message: details.reason,
+          stack: details.logs,
         },
+        overrideCompletion: details.overrideCompletion,
       });
     } catch (error) {
       logger.error(error);
     }
   },
-  onRunFailure: async (runId, failureInfo) => {
-    logger.log("Run failed:", { runId, failureInfo });
+  onRunFailure: async (runId, details) => {
+    logger.log("Run failed:", { runId, details });
 
     try {
-      provider.platformSocket.send("WORKER_CRASHED", { runId, ...failureInfo });
+      provider.platformSocket.send("WORKER_CRASHED", { runId, ...details });
     } catch (error) {
       logger.error(error);
     }
@@ -565,3 +574,19 @@ const podCleaner = new PodCleaner({
 });
 
 podCleaner.start();
+
+if (UPTIME_HEARTBEAT_URL) {
+  const uptimeHeartbeat = new UptimeHeartbeat({
+    runtimeEnv: RUNTIME_ENV,
+    namespace: "default",
+    intervalInSeconds: UPTIME_INTERVAL_SECONDS,
+    pingUrl: UPTIME_HEARTBEAT_URL,
+    maxPendingRuns: UPTIME_MAX_PENDING_RUNS,
+    maxPendingIndeces: UPTIME_MAX_PENDING_INDECES,
+    maxPendingErrors: UPTIME_MAX_PENDING_ERRORS,
+  });
+
+  uptimeHeartbeat.start();
+} else {
+  logger.log("Uptime heartbeat is disabled, set UPTIME_HEARTBEAT_URL to enable.");
+}

apps/kubernetes-provider/src/podCleaner.ts

@@ -1,5 +1,5 @@
 import * as k8s from "@kubernetes/client-node";
-import { SimpleLogger } from "@trigger.dev/core-apps";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
 
 type PodCleanerOptions = {
   runtimeEnv: "local" | "kubernetes";

apps/kubernetes-provider/src/taskMonitor.ts

@@ -1,25 +1,20 @@
 import * as k8s from "@kubernetes/client-node";
-import { SimpleLogger } from "@trigger.dev/core-apps";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
+import { EXIT_CODE_ALREADY_HANDLED, EXIT_CODE_CHILD_NONZERO } from "@trigger.dev/core-apps/process";
 import { setTimeout } from "timers/promises";
 import PQueue from "p-queue";
+import type { Prettify } from "@trigger.dev/core/v3";
 
-type IndexFailureHandler = (
-  deploymentId: string,
-  failureInfo: {
-    exitCode: number;
-    reason: string;
-    logs: string;
-  }
-) => Promise<any>;
+type FailureDetails = Prettify<{
+  exitCode: number;
+  reason: string;
+  logs: string;
+  overrideCompletion: boolean;
+}>;
 
-type RunFailureHandler = (
-  runId: string,
-  failureInfo: {
-    exitCode: number;
-    reason: string;
-    logs: string;
-  }
-) => Promise<any>;
+type IndexFailureHandler = (deploymentId: string, details: FailureDetails) => Promise<any>;
+
+type RunFailureHandler = (runId: string, details: FailureDetails) => Promise<any>;
 
 type TaskMonitorOptions = {
   runtimeEnv: "local" | "kubernetes";
@@ -144,8 +139,10 @@ export class TaskMonitor {
     const containerState = this.#getContainerStateSummary(containerStatus.state);
     const exitCode = containerState.exitCode ?? -1;
 
-    // We use this special exit code to signal any errors were already handled elsewhere
-    if (exitCode === 111) {
+    if (exitCode === EXIT_CODE_ALREADY_HANDLED) {
+      this.#logger.debug("Ignoring pod failure, already handled by worker", {
+        podName,
+      });
       return;
     }
 
@@ -162,6 +159,7 @@ export class TaskMonitor {
 
     let reason = rawReason || "Unknown error";
     let logs = rawLogs || "";
+    let overrideCompletion = false;
 
     switch (rawReason) {
       case "Error":
@@ -181,7 +179,10 @@ export class TaskMonitor {
         }
         break;
       case "OOMKilled":
-        reason = "Out of memory! Try increasing the memory on this task.";
+        overrideCompletion = true;
+        reason = `${
+          exitCode === EXIT_CODE_CHILD_NONZERO ? "Child process" : "Parent process"
+        } ran out of memory! Try choosing a machine preset with more memory for this task.`;
         break;
       default:
         break;
@@ -191,7 +192,8 @@ export class TaskMonitor {
       exitCode,
       reason,
       logs,
-    };
+      overrideCompletion,
+    } satisfies FailureDetails;
 
     const app = pod.metadata?.labels?.app;
 

apps/kubernetes-provider/src/uptimeHeartbeat.ts

@@ -0,0 +1,272 @@
+import * as k8s from "@kubernetes/client-node";
+import { SimpleLogger } from "@trigger.dev/core-apps/logger";
+
+type UptimeHeartbeatOptions = {
+  runtimeEnv: "local" | "kubernetes";
+  pingUrl: string;
+  namespace?: string;
+  intervalInSeconds?: number;
+  maxPendingRuns?: number;
+  maxPendingIndeces?: number;
+  maxPendingErrors?: number;
+  leadingEdge?: boolean;
+};
+
+export class UptimeHeartbeat {
+  private enabled = false;
+  private namespace: string;
+
+  private intervalInSeconds: number;
+  private maxPendingRuns: number;
+  private maxPendingIndeces: number;
+  private maxPendingErrors: number;
+
+  private leadingEdge = true;
+
+  private logger = new SimpleLogger("[UptimeHeartbeat]");
+  private k8sClient: {
+    core: k8s.CoreV1Api;
+    kubeConfig: k8s.KubeConfig;
+  };
+
+  constructor(private opts: UptimeHeartbeatOptions) {
+    this.namespace = opts.namespace ?? "default";
+
+    this.intervalInSeconds = opts.intervalInSeconds ?? 60;
+    this.maxPendingRuns = opts.maxPendingRuns ?? 25;
+    this.maxPendingIndeces = opts.maxPendingIndeces ?? 10;
+    this.maxPendingErrors = opts.maxPendingErrors ?? 10;
+
+    this.k8sClient = this.#createK8sClient();
+  }
+
+  #createK8sClient() {
+    const kubeConfig = new k8s.KubeConfig();
+
+    if (this.opts.runtimeEnv === "local") {
+      kubeConfig.loadFromDefault();
+    } else if (this.opts.runtimeEnv === "kubernetes") {
+      kubeConfig.loadFromCluster();
+    } else {
+      throw new Error(`Unsupported runtime environment: ${this.opts.runtimeEnv}`);
+    }
+
+    return {
+      core: kubeConfig.makeApiClient(k8s.CoreV1Api),
+      kubeConfig: kubeConfig,
+    };
+  }
+
+  #isRecord(candidate: unknown): candidate is Record<string, unknown> {
+    if (typeof candidate !== "object" || candidate === null) {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+  #logK8sError(err: unknown, debugOnly = false) {
+    if (debugOnly) {
+      this.logger.debug("K8s API Error", err);
+    } else {
+      this.logger.error("K8s API Error", err);
+    }
+  }
+
+  #handleK8sError(err: unknown) {
+    if (!this.#isRecord(err) || !this.#isRecord(err.body)) {
+      this.#logK8sError(err);
+      return;
+    }
+
+    this.#logK8sError(err, true);
+
+    if (typeof err.body.message === "string") {
+      this.#logK8sError({ message: err.body.message });
+      return;
+    }
+
+    this.#logK8sError({ body: err.body });
+  }
+
+  async #getPods(opts: {
+    namespace: string;
+    fieldSelector?: string;
+    labelSelector?: string;
+  }): Promise<Array<k8s.V1Pod> | undefined> {
+    const listReturn = await this.k8sClient.core
+      .listNamespacedPod(
+        opts.namespace,
+        undefined, // pretty
+        undefined, // allowWatchBookmarks
+        undefined, // _continue
+        opts.fieldSelector,
+        opts.labelSelector,
+        this.maxPendingRuns * 2, // limit
+        undefined, // resourceVersion
+        undefined, // resourceVersionMatch
+        undefined, // sendInitialEvents
+        this.intervalInSeconds, // timeoutSeconds,
+        undefined // watch
+      )
+      .catch(this.#handleK8sError.bind(this));
+
+    return listReturn?.body.items;
+  }
+
+  async #getPendingIndeces(): Promise<Array<k8s.V1Pod> | undefined> {
+    return await this.#getPods({
+      namespace: this.namespace,
+      fieldSelector: "status.phase=Pending",
+      labelSelector: "app=task-index",
+    });
+  }
+
+  async #getPendingTasks(): Promise<Array<k8s.V1Pod> | undefined> {
+    return await this.#getPods({
+      namespace: this.namespace,
+      fieldSelector: "status.phase=Pending",
+      labelSelector: "app=task-run",
+    });
+  }
+
+  #countPods(pods: Array<k8s.V1Pod>): number {
+    return pods.length;
+  }
+
+  #filterPendingPods(
+    pods: Array<k8s.V1Pod>,
+    waitingReason: "CreateContainerError" | "RunContainerError"
+  ): Array<k8s.V1Pod> {
+    return pods.filter((pod) => {
+      const containerStatus = pod.status?.containerStatuses?.[0];
+      return containerStatus?.state?.waiting?.reason === waitingReason;
+    });
+  }
+
+  async #sendPing() {
+    this.logger.log("Sending ping");
+
+    const start = Date.now();
+    const controller = new AbortController();
+
+    const timeoutMs = (this.intervalInSeconds * 1000) / 2;
+
+    const fetchTimeout = setTimeout(() => {
+      controller.abort();
+    }, timeoutMs);
+
+    try {
+      const response = await fetch(this.opts.pingUrl, {
+        signal: controller.signal,
+      });
+
+      if (!response.ok) {
+        this.logger.error("Failed to send ping, response not OK", {
+          status: response.status,
+        });
+        return;
+      }
+
+      const elapsedMs = Date.now() - start;
+      this.logger.log("Ping sent", { elapsedMs });
+    } catch (error) {
+      if (error instanceof DOMException && error.name === "AbortError") {
+        this.logger.log("Ping timeout", { timeoutSeconds: timeoutMs });
+        return;
+      }
+
+      this.logger.error("Failed to send ping", error);
+    } finally {
+      clearTimeout(fetchTimeout);
+    }
+  }
+
+  async #heartbeat() {
+    this.logger.log("Performing heartbeat");
+
+    const start = Date.now();
+
+    const pendingTasks = await this.#getPendingTasks();
+
+    if (!pendingTasks) {
+      this.logger.error("Failed to get pending tasks");
+      return;
+    }
+
+    const totalPendingTasks = this.#countPods(pendingTasks);
+
+    const pendingIndeces = await this.#getPendingIndeces();
+
+    if (!pendingIndeces) {
+      this.logger.error("Failed to get pending indeces");
+      return;
+    }
+
+    const totalPendingIndeces = this.#countPods(pendingIndeces);
+
+    const elapsedMs = Date.now() - start;
+
+    this.logger.log("Finished heartbeat checks", { elapsedMs });
+
+    if (totalPendingTasks > this.maxPendingRuns) {
+      this.logger.log("Too many pending tasks, skipping heartbeat", { totalPendingTasks });
+      return;
+    }
+
+    if (totalPendingIndeces > this.maxPendingIndeces) {
+      this.logger.log("Too many pending indeces, skipping heartbeat", { totalPendingIndeces });
+      return;
+    }
+
+    const totalCreateContainerErrors = this.#countPods(
+      this.#filterPendingPods(pendingTasks, "CreateContainerError")
+    );
+    const totalRunContainerErrors = this.#countPods(
+      this.#filterPendingPods(pendingTasks, "RunContainerError")
+    );
+
+    if (totalCreateContainerErrors + totalRunContainerErrors > this.maxPendingErrors) {
+      this.logger.log("Too many pending tasks with errors, skipping heartbeat", {
+        totalRunContainerErrors,
+        totalCreateContainerErrors,
+      });
+      return;
+    }
+
+    await this.#sendPing();
+
+    this.logger.log("Heartbeat done", { totalPendingTasks, elapsedMs });
+  }
+
+  async start() {
+    this.enabled = true;
+    this.logger.log("Starting");
+
+    if (this.leadingEdge) {
+      await this.#heartbeat();
+    }
+
+    const heartbeat = setInterval(async () => {
+      if (!this.enabled) {
+        clearInterval(heartbeat);
+        return;
+      }
+
+      try {
+        await this.#heartbeat();
+      } catch (error) {
+        this.logger.error("Error while heartbeating", error);
+      }
+    }, this.intervalInSeconds * 1000);
+  }
+
+  async stop() {
+    if (!this.enabled) {
+      return;
+    }
+
+    this.enabled = false;
+    this.logger.log("Shutting down..");
+  }
+}

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

@@ -146,7 +146,6 @@ function FilterMenu(props: RunFiltersProps) {
 
   const filterTrigger = (
     <SelectTrigger
-      autoFocus
       icon={
         <div className="flex size-4 items-center justify-center">
           <ListFilterIcon className="size-3.5" />

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

@@ -3,10 +3,12 @@ import {
   BoltSlashIcon,
   BugAntIcon,
   CheckCircleIcon,
+  ClockIcon,
   FireIcon,
   NoSymbolIcon,
   PauseCircleIcon,
   RectangleStackIcon,
+  TrashIcon,
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { TaskRunStatus } from "@trigger.dev/database";
@@ -16,6 +18,7 @@ import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
 export const allTaskRunStatuses = [
+  "DELAYED",
   "WAITING_FOR_DEPLOY",
   "PENDING",
   "EXECUTING",
@@ -28,10 +31,12 @@ export const allTaskRunStatuses = [
   "PAUSED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 export const filterableTaskRunStatuses = [
   "WAITING_FOR_DEPLOY",
+  "DELAYED",
   "PENDING",
   "EXECUTING",
   "RETRYING_AFTER_FAILURE",
@@ -42,9 +47,11 @@ export const filterableTaskRunStatuses = [
   "CRASHED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
+  DELAYED: "Task has been delayed and is waiting to be executed",
   PENDING: "Task is waiting to be executed",
   WAITING_FOR_DEPLOY: "Task needs to be deployed first to start executing",
   EXECUTING: "Task is currently being executed",
@@ -57,9 +64,10 @@ const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
   SYSTEM_FAILURE: "Task has failed due to a system failure",
   PAUSED: "Task has been paused by the user",
   CRASHED: "Task has crashed and won't be retried",
+  EXPIRED: "Task has surpassed its ttl and won't be executed",
 };
 
-export const QUEUED_STATUSES: TaskRunStatus[] = ["PENDING", "WAITING_FOR_DEPLOY"];
+export const QUEUED_STATUSES: TaskRunStatus[] = ["PENDING", "WAITING_FOR_DEPLOY", "DELAYED"];
 
 export const RUNNING_STATUSES: TaskRunStatus[] = [
   "EXECUTING",
@@ -74,6 +82,7 @@ export const FINISHED_STATUSES: TaskRunStatus[] = [
   "INTERRUPTED",
   "SYSTEM_FAILURE",
   "CRASHED",
+  "EXPIRED",
 ];
 
 export function descriptionForTaskRunStatus(status: TaskRunStatus): string {
@@ -109,6 +118,8 @@ export function TaskRunStatusIcon({
   className: string;
 }) {
   switch (status) {
+    case "DELAYED":
+      return <ClockIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "PENDING":
       return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "WAITING_FOR_DEPLOY":
@@ -133,6 +144,8 @@ export function TaskRunStatusIcon({
       return <BugAntIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "CRASHED":
       return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
+    case "EXPIRED":
+      return <TrashIcon className={cn(runStatusClassNameColor(status), className)} />;
 
     default: {
       assertNever(status);
@@ -143,6 +156,7 @@ export function TaskRunStatusIcon({
 export function runStatusClassNameColor(status: TaskRunStatus): string {
   switch (status) {
     case "PENDING":
+    case "DELAYED":
       return "text-charcoal-500";
     case "WAITING_FOR_DEPLOY":
       return "text-amber-500";
@@ -154,6 +168,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
     case "PAUSED":
       return "text-amber-300";
     case "CANCELED":
+    case "EXPIRED":
       return "text-charcoal-500";
     case "INTERRUPTED":
       return "text-error";
@@ -173,6 +188,8 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
 
 export function runStatusTitle(status: TaskRunStatus): string {
   switch (status) {
+    case "DELAYED":
+      return "Delayed";
     case "PENDING":
       return "Queued";
     case "WAITING_FOR_DEPLOY":
@@ -197,6 +214,8 @@ export function runStatusTitle(status: TaskRunStatus): string {
       return "System failure";
     case "CRASHED":
       return "Crashed";
+    case "EXPIRED":
+      return "Expired";
     default: {
       assertNever(status);
     }

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

@@ -118,6 +118,8 @@ export function TaskRunsTable({
           <TableHeaderCell>Duration</TableHeaderCell>
           <TableHeaderCell>Test</TableHeaderCell>
           <TableHeaderCell>Created at</TableHeaderCell>
+          <TableHeaderCell>Delayed until</TableHeaderCell>
+          <TableHeaderCell>TTL</TableHeaderCell>
           <TableHeaderCell>
             <span className="sr-only">Go to page</span>
           </TableHeaderCell>
@@ -125,7 +127,7 @@ export function TaskRunsTable({
       </TableHeader>
       <TableBody>
         {total === 0 && !hasFilters ? (
-          <TableBlankRow colSpan={9}>
+          <TableBlankRow colSpan={10}>
             {!isLoading && <NoRuns title="No runs found" />}
           </TableBlankRow>
         ) : runs.length === 0 ? (
@@ -187,6 +189,10 @@ export function TaskRunsTable({
                 <TableCell to={path}>
                   {run.createdAt ? <DateTime date={run.createdAt} /> : "–"}
                 </TableCell>
+                <TableCell to={path}>
+                  {run.delayUntil ? <DateTime date={run.delayUntil} /> : "–"}
+                </TableCell>
+                <TableCell to={path}>{run.ttl ?? "–"}</TableCell>
                 <RunActionsCell run={run} path={path} />
               </TableRow>
             );

apps/webapp/app/database-types.ts

@@ -40,6 +40,8 @@ export const TaskRunStatus = {
   COMPLETED_WITH_ERRORS: "COMPLETED_WITH_ERRORS",
   SYSTEM_FAILURE: "SYSTEM_FAILURE",
   CRASHED: "CRASHED",
+  DELAYED: "DELAYED",
+  EXPIRED: "EXPIRED",
 } as const satisfies Record<TaskRunStatusType, TaskRunStatusType>;
 
 export const JobRunStatus = {

apps/webapp/app/env.server.ts

@@ -151,7 +151,7 @@ const EnvironmentSchema = z.object({
   PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
   PROD_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
 
-  RUNTIME_WAIT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000),
+  CHECKPOINT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000),
 
   // Internal OTEL environment variables
   INTERNAL_OTEL_TRACE_EXPORTER_URL: z.string().optional(),
@@ -206,6 +206,9 @@ const EnvironmentSchema = z.object({
   USAGE_OPEN_METER_BASE_URL: z.string().optional(),
   EVENT_LOOP_MONITOR_ENABLED: z.string().default("1"),
   MAXIMUM_LIVE_RELOADING_EVENTS: z.coerce.number().int().default(1000),
+  MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
+  TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB
+  TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/models/api-key.server.ts

@@ -84,7 +84,9 @@ export function createPkApiKeyForEnv(envType: RuntimeEnvironment["type"]) {
   return `pk_${envSlug(envType)}_${apiKeyId(20)}`;
 }
 
-export function envSlug(environmentType: RuntimeEnvironment["type"]) {
+export type EnvSlug = "dev" | "stg" | "prod" | "prev";
+
+export function envSlug(environmentType: RuntimeEnvironment["type"]): EnvSlug {
   switch (environmentType) {
     case "DEVELOPMENT": {
       return "dev";
@@ -100,3 +102,7 @@ export function envSlug(environmentType: RuntimeEnvironment["type"]) {
     }
   }
 }
+
+export function isEnvSlug(maybeSlug: string): maybeSlug is EnvSlug {
+  return ["dev", "stg", "prod", "prev"].includes(maybeSlug);
+}

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

@@ -118,6 +118,7 @@ export function batchTaskRunItemStatusForRunStatus(
     case TaskRunStatus.COMPLETED_WITH_ERRORS:
     case TaskRunStatus.SYSTEM_FAILURE:
     case TaskRunStatus.CRASHED:
+    case TaskRunStatus.EXPIRED:
       return BatchTaskRunItemStatus.FAILED;
     case TaskRunStatus.PENDING:
     case TaskRunStatus.WAITING_FOR_DEPLOY:
@@ -125,6 +126,7 @@ export function batchTaskRunItemStatusForRunStatus(
     case TaskRunStatus.RETRYING_AFTER_FAILURE:
     case TaskRunStatus.EXECUTING:
     case TaskRunStatus.PAUSED:
+    case TaskRunStatus.DELAYED:
       return BatchTaskRunItemStatus.PENDING;
     default:
       assertNever(status);

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

@@ -1,11 +1,9 @@
-import { z } from "zod";
 import {
   Direction,
   FilterableEnvironment,
   FilterableStatus,
   filterableStatuses,
 } from "~/components/runs/RunStatuses";
-import { PrismaClient, prisma } from "~/db.server";
 import { getUsername } from "~/utils/username";
 import { BasePresenter } from "./v3/basePresenter.server";
 
@@ -29,8 +27,6 @@ const DEFAULT_PAGE_SIZE = 20;
 export type RunList = Awaited<ReturnType<RunListPresenter["call"]>>;
 
 export class RunListPresenter extends BasePresenter {
-  
-
   public async call({
     userId,
     eventId,

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

@@ -1,27 +1,52 @@
-import { User } from "@trigger.dev/database";
 import { ScheduleMetadataSchema } from "@trigger.dev/core";
-import { PrismaClient, prisma } from "~/db.server";
+import { User } from "@trigger.dev/database";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { calculateNextScheduledEvent } from "~/services/schedules/nextScheduledEvent.server";
+import { BasePresenter } from "./v3/basePresenter.server";
 
-export class ScheduledTriggersPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
+const DEFAULT_PAGE_SIZE = 20;
 
+export class ScheduledTriggersPresenter extends BasePresenter {
   public async call({
     userId,
     projectSlug,
     organizationSlug,
+    direction = "forward",
+    pageSize = DEFAULT_PAGE_SIZE,
+    cursor,
   }: {
     userId: User["id"];
     projectSlug: Project["slug"];
     organizationSlug: Organization["slug"];
+    direction?: "forward" | "backward";
+    pageSize?: number;
+    cursor?: string;
   }) {
-    const scheduled = await this.#prismaClient.scheduleSource.findMany({
+    const organization = await this._replica.organization.findFirstOrThrow({
+      select: {
+        id: true,
+      },
+      where: {
+        slug: organizationSlug,
+        members: { some: { userId } },
+      },
+    });
+
+    // Find the project scoped to the organization
+    const project = await this._replica.project.findFirstOrThrow({
+      select: {
+        id: true,
+      },
+      where: {
+        slug: projectSlug,
+        organizationId: organization.id,
+      },
+    });
+
+    const directionMultiplier = direction === "forward" ? 1 : -1;
+
+    const scheduled = await this._replica.scheduleSource.findMany({
       select: {
         id: true,
         key: true,
@@ -50,23 +75,50 @@ export class ScheduledTriggersPresenter {
               },
             },
           ],
-          organization: {
-            slug: organizationSlug,
-            members: {
-              some: {
-                userId,
-              },
-            },
-          },
-          project: {
-            slug: projectSlug,
-          },
+          projectId: project.id,
         },
       },
+      orderBy: [{ id: "desc" }],
+      //take an extra record to tell if there are more
+      take: directionMultiplier * (pageSize + 1),
+      //skip the cursor if there is one
+      skip: cursor ? 1 : 0,
+      cursor: cursor
+        ? {
+            id: cursor,
+          }
+        : undefined,
     });
 
+    const hasMore = scheduled.length > pageSize;
+
+    //get cursors for next and previous pages
+    let next: string | undefined;
+    let previous: string | undefined;
+    switch (direction) {
+      case "forward":
+        previous = cursor ? scheduled.at(0)?.id : undefined;
+        if (hasMore) {
+          next = scheduled[pageSize - 1]?.id;
+        }
+        break;
+      case "backward":
+        if (hasMore) {
+          previous = scheduled[1]?.id;
+          next = scheduled[pageSize]?.id;
+        } else {
+          next = scheduled[pageSize - 1]?.id;
+        }
+        break;
+    }
+
+    const scheduledToReturn =
+      direction === "backward" && hasMore
+        ? scheduled.slice(1, pageSize + 1)
+        : scheduled.slice(0, pageSize);
+
     return {
-      scheduled: scheduled.map((s) => {
+      scheduled: scheduledToReturn.map((s) => {
         const schedule = ScheduleMetadataSchema.parse(s.schedule);
         const nextEventTimestamp = s.active
           ? calculateNextScheduledEvent(schedule, s.lastEventTimestamp)
@@ -78,6 +130,10 @@ export class ScheduledTriggersPresenter {
           nextEventTimestamp,
         };
       }),
+      pagination: {
+        next,
+        previous,
+      },
     };
   }
 }

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

@@ -12,6 +12,7 @@ import {
 import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { generatePresignedUrl } from "~/v3/r2.server";
 import { BasePresenter } from "./basePresenter.server";
 
 export class ApiRetrieveRunPresenter extends BasePresenter {
@@ -44,7 +45,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       }
 
       let $payload: any;
+      let $payloadPresignedUrl: string | undefined;
       let $output: any;
+      let $outputPresignedUrl: string | undefined;
 
       if (showSecretDetails) {
         const payloadPacket = await conditionallyImportPacket({
@@ -52,7 +55,19 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
           dataType: taskRun.payloadType,
         });
 
-        $payload = await parsePacket(payloadPacket);
+        if (
+          payloadPacket.dataType === "application/store" &&
+          typeof payloadPacket.data === "string"
+        ) {
+          $payloadPresignedUrl = await generatePresignedUrl(
+            env.project.externalRef,
+            env.slug,
+            payloadPacket.data,
+            "GET"
+          );
+        } else {
+          $payload = await parsePacket(payloadPacket);
+        }
 
         if (taskRun.status === "COMPLETED_SUCCESSFULLY") {
           const completedAttempt = taskRun.attempts.find(
@@ -65,7 +80,19 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
               dataType: completedAttempt.outputType,
             });
 
-            $output = await parsePacket(outputPacket);
+            if (
+              outputPacket.dataType === "application/store" &&
+              typeof outputPacket.data === "string"
+            ) {
+              $outputPresignedUrl = await generatePresignedUrl(
+                env.project.externalRef,
+                env.slug,
+                outputPacket.data,
+                "GET"
+              );
+            } else {
+              $output = await parsePacket(outputPacket);
+            }
           }
         }
       }
@@ -84,9 +111,14 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         finishedAt: ApiRetrieveRunPresenter.isStatusFinished(apiStatus)
           ? taskRun.updatedAt
           : undefined,
+        delayedUntil: taskRun.delayUntil ?? undefined,
         payload: $payload,
+        payloadPresignedUrl: $payloadPresignedUrl,
         output: $output,
+        outputPresignedUrl: $outputPresignedUrl,
         isTest: taskRun.isTest,
+        ttl: taskRun.ttl ?? undefined,
+        expiredAt: taskRun.expiredAt ?? undefined,
         schedule: taskRun.schedule
           ? {
               id: taskRun.schedule.friendlyId,
@@ -142,6 +174,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
 
   static apiStatusFromRunStatus(status: TaskRunStatus): RunStatus {
     switch (status) {
+      case "DELAYED": {
+        return "DELAYED";
+      }
       case "WAITING_FOR_DEPLOY": {
         return "WAITING_FOR_DEPLOY";
       }
@@ -176,6 +211,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       case "COMPLETED_WITH_ERRORS": {
         return "FAILED";
       }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
       default: {
         assertNever(status);
       }
@@ -183,7 +221,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
   }
 
   static apiBooleanHelpersFromRunStatus(status: RunStatus) {
-    const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY";
+    const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY" || status === "DELAYED";
     const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN";
     const isCompleted =
       status === "COMPLETED" ||

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

@@ -209,7 +209,10 @@ export class ApiRunListPresenter extends BasePresenter {
           updatedAt: new Date(run.updatedAt),
           startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
           finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
+          delayedUntil: run.delayUntil ? new Date(run.delayUntil) : undefined,
           isTest: run.isTest,
+          ttl: run.ttl ?? undefined,
+          expiredAt: run.expiredAt ? new Date(run.expiredAt) : undefined,
           env: {
             id: run.environment.id,
             name: run.environment.slug,
@@ -233,6 +236,8 @@ export class ApiRunListPresenter extends BasePresenter {
 
   static apiStatusToRunStatuses(status: RunStatus): TaskRunStatus[] | TaskRunStatus {
     switch (status) {
+      case "DELAYED":
+        return "DELAYED";
       case "WAITING_FOR_DEPLOY": {
         return "WAITING_FOR_DEPLOY";
       }
@@ -266,6 +271,9 @@ export class ApiRunListPresenter extends BasePresenter {
       case "FAILED": {
         return "COMPLETED_WITH_ERRORS";
       }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
       default: {
         assertNever(status);
       }

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

@@ -158,10 +158,13 @@ export class RunListPresenter extends BasePresenter {
         createdAt: Date;
         startedAt: Date | null;
         lockedAt: Date | null;
+        delayUntil: Date | null;
         updatedAt: Date;
         isTest: boolean;
         spanId: string;
         idempotencyKey: string | null;
+        ttl: string | null;
+        expiredAt: Date | null;
       }[]
     >`
     SELECT
@@ -174,11 +177,14 @@ export class RunListPresenter extends BasePresenter {
     tr.status AS status,
     tr."createdAt" AS "createdAt",
     tr."startedAt" AS "startedAt",
+    tr."delayUntil" AS "delayUntil",
     tr."lockedAt" AS "lockedAt",
     tr."updatedAt" AS "updatedAt",
     tr."isTest" AS "isTest",
     tr."spanId" AS "spanId",
-    tr."idempotencyKey" AS "idempotencyKey"
+    tr."idempotencyKey" AS "idempotencyKey",
+    tr."ttl" AS "ttl",
+    tr."expiredAt" AS "expiredAt"
   FROM
     ${sqlDatabaseSchema}."TaskRun" tr
   LEFT JOIN
@@ -283,6 +289,7 @@ export class RunListPresenter extends BasePresenter {
           createdAt: run.createdAt.toISOString(),
           updatedAt: run.updatedAt.toISOString(),
           startedAt: startedAt ? startedAt.toISOString() : undefined,
+          delayUntil: run.delayUntil ? run.delayUntil.toISOString() : undefined,
           hasFinished,
           finishedAt: hasFinished ? run.updatedAt.toISOString() : undefined,
           isTest: run.isTest,
@@ -294,6 +301,8 @@ export class RunListPresenter extends BasePresenter {
           isCancellable: isCancellableRunStatus(run.status),
           environment: displayableEnvironment(environment, userId),
           idempotencyKey: run.idempotencyKey ? run.idempotencyKey : undefined,
+          ttl: run.ttl ? run.ttl : undefined,
+          expiredAt: run.expiredAt ? run.expiredAt.toISOString() : undefined,
         };
       }),
       pagination: {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx

@@ -2,6 +2,8 @@ import { NoSymbolIcon } from "@heroicons/react/20/solid";
 import { CheckCircleIcon, XCircleIcon } from "@heroicons/react/24/solid";
 import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { z } from "zod";
+import { ListPagination } from "~/components/ListPagination";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { DateTime } from "~/components/primitives/DateTime";
 import { LabelValueStack } from "~/components/primitives/LabelValueStack";
@@ -17,30 +19,38 @@ import {
   TableRow,
 } from "~/components/primitives/Table";
 import { TextLink } from "~/components/primitives/TextLink";
-import { useOrganization } from "~/hooks/useOrganizations";
-import { useProject } from "~/hooks/useProject";
+import { DirectionSchema } from "~/components/runs/RunStatuses";
 import { ScheduledTriggersPresenter } from "~/presenters/ScheduledTriggersPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { ProjectParamSchema, docsPath } from "~/utils/pathBuilder";
 
+const SearchSchema = z.object({
+  cursor: z.string().optional(),
+  direction: DirectionSchema.optional(),
+});
+
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
   const { organizationSlug, projectParam } = ProjectParamSchema.parse(params);
 
+  const url = new URL(request.url);
+  const s = Object.fromEntries(url.searchParams.entries());
+  const searchParams = SearchSchema.parse(s);
+
   const presenter = new ScheduledTriggersPresenter();
   const data = await presenter.call({
     userId,
     organizationSlug,
     projectSlug: projectParam,
+    direction: searchParams.direction,
+    cursor: searchParams.cursor,
   });
 
   return typedjson(data);
 };
 
-export default function Integrations() {
-  const { scheduled } = useTypedLoaderData<typeof loader>();
-  const organization = useOrganization();
-  const project = useProject();
+export default function Route() {
+  const { scheduled, pagination } = useTypedLoaderData<typeof loader>();
 
   return (
     <>
@@ -49,6 +59,10 @@ export default function Integrations() {
         expression or an interval.
       </Paragraph>
 
+      {scheduled.length > 0 && (
+        <ListPagination list={{ pagination }} className="mt-2 justify-end" />
+      )}
+
       <Table containerClassName="mt-4">
         <TableHeader>
           <TableRow>

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

@@ -23,6 +23,7 @@ import { Paragraph } from "~/components/primitives/Paragraph";
 import { Property, PropertyTable } from "~/components/primitives/PropertyTable";
 import {
   Table,
+  TableBlankRow,
   TableBody,
   TableCell,
   TableHeader,
@@ -180,6 +181,14 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
   }
 };
 
+function PlaceholderText({ title }: { title: string }) {
+  return (
+    <div className="flex items-center justify-center">
+      <Paragraph className="w-auto">{title}</Paragraph>
+    </div>
+  );
+}
+
 export default function Page() {
   const { schedule } = useTypedLoaderData<typeof loader>();
   const location = useLocation();
@@ -252,18 +261,30 @@ export default function Page() {
                   </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>
-                    </TableRow>
-                  ))}
+                  {schedule.active ? (
+                    schedule.nextRuns.length ? (
+                      schedule.nextRuns.map((run, index) => (
+                        <TableRow key={index}>
+                          {!isUtc && (
+                            <TableCell>
+                              <DateTime date={run} timeZone={schedule.timezone} />
+                            </TableCell>
+                          )}
+                          <TableCell>
+                            <DateTime date={run} timeZone="UTC" />
+                          </TableCell>
+                        </TableRow>
+                      ))
+                    ) : (
+                      <TableBlankRow colSpan={1}>
+                        <PlaceholderText title="You found a bug" />
+                      </TableBlankRow>
+                    )
+                  ) : (
+                    <TableBlankRow colSpan={1}>
+                      <PlaceholderText title="Schedule disabled" />
+                    </TableBlankRow>
+                  )}
                 </TableBody>
               </Table>
             </div>

apps/webapp/app/routes/api.v1.packets.$.ts

@@ -1,10 +1,8 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { env } from "~/env.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { logger } from "~/services/logger.server";
-import { r2 } from "~/v3/r2.server";
+import { generatePresignedUrl } from "~/v3/r2.server";
 
 const ParamsSchema = z.object({
   "*": z.string(),
@@ -26,34 +24,19 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const parsedParams = ParamsSchema.parse(params);
   const filename = parsedParams["*"];
 
-  if (!env.OBJECT_STORE_BASE_URL) {
-    return json({ error: "Object store base URL is not set" }, { status: 500 });
-  }
-
-  if (!r2) {
-    return json({ error: "Object store credentials are not set" }, { status: 500 });
-  }
-
-  const url = new URL(env.OBJECT_STORE_BASE_URL);
-  url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`;
-  url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes
-
-  const signed = await r2.sign(
-    new Request(url, {
-      method: "PUT",
-    }),
-    {
-      aws: { signQuery: true },
-    }
+  const presignedUrl = await generatePresignedUrl(
+    authenticationResult.environment.project.externalRef,
+    authenticationResult.environment.slug,
+    filename,
+    "PUT"
   );
 
-  logger.debug("Generated presigned URL", {
-    url: signed.url,
-    headers: Object.fromEntries(signed.headers),
-  });
+  if (!presignedUrl) {
+    return json({ error: "Failed to generate presigned URL" }, { status: 500 });
+  }
 
   // Caller can now use this URL to upload to that object.
-  return json({ presignedUrl: signed.url });
+  return json({ presignedUrl });
 }
 
 export async function loader({ request, params }: ActionFunctionArgs) {
@@ -67,35 +50,17 @@ export async function loader({ request, params }: ActionFunctionArgs) {
   const parsedParams = ParamsSchema.parse(params);
   const filename = parsedParams["*"];
 
-  if (!env.OBJECT_STORE_BASE_URL) {
-    return json({ error: "Object store base URL is not set" }, { status: 500 });
-  }
-
-  if (!r2) {
-    return json({ error: "Object store credentials are not set" }, { status: 500 });
-  }
-
-  const url = new URL(env.OBJECT_STORE_BASE_URL);
-  url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`;
-  url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes
-
-  const signed = await r2.sign(
-    new Request(url, {
-      method: request.method,
-    }),
-    {
-      aws: { signQuery: true },
-    }
+  const presignedUrl = await generatePresignedUrl(
+    authenticationResult.environment.project.externalRef,
+    authenticationResult.environment.slug,
+    filename,
+    "GET"
   );
 
-  logger.debug("Generated presigned URL", {
-    url: signed.url,
-    headers: Object.fromEntries(signed.headers),
-  });
+  if (!presignedUrl) {
+    return json({ error: "Failed to generate presigned URL" }, { status: 500 });
+  }
 
-  const getUrl = new URL(url.href);
-  getUrl.searchParams.delete("X-Amz-Expires");
-
-  // Caller can now use this URL to upload to that object.
-  return json({ presignedUrl: signed.url });
+  // Caller can now use this URL to fetch that object.
+  return json({ presignedUrl });
 }

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server.ts

@@ -3,6 +3,7 @@ import { PrismaClientOrTransaction, prisma } from "~/db.server";
 import { taskWithAttemptsToServerTask } from "~/models/task.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { startActiveSpan } from "~/v3/tracer.server";
 
 export class CompleteRunTaskService {
   #prismaClient: PrismaClientOrTransaction;
@@ -17,76 +18,81 @@ export class CompleteRunTaskService {
     id: string,
     taskBody: CompleteTaskBodyOutput
   ): Promise<ServerTask | undefined> {
-    const existingTask = await this.#prismaClient.task.findUnique({
-      where: {
-        id,
-      },
-      include: {
-        run: true,
-        attempts: {
-          where: {
-            status: "PENDING",
-          },
-          orderBy: {
-            number: "desc",
-          },
-          take: 1,
+    return startActiveSpan("CompleteRunTaskService.call", async (span) => {
+      span.setAttribute("runId", runId);
+      span.setAttribute("taskId", id);
+
+      const existingTask = await this.#prismaClient.task.findUnique({
+        where: {
+          id,
+        },
+        include: {
+          run: true,
+          attempts: {
+            where: {
+              status: "PENDING",
+            },
+            orderBy: {
+              number: "desc",
+            },
+            take: 1,
+          },
         },
-      },
-    });
-
-    if (!existingTask) {
-      return;
-    }
-
-    if (existingTask.runId !== runId) {
-      return;
-    }
-
-    if (existingTask.run.environmentId !== environment.id) {
-      return;
-    }
-
-    if (
-      existingTask.status === "COMPLETED" ||
-      existingTask.status === "ERRORED" ||
-      existingTask.status === "CANCELED"
-    ) {
-      logger.debug("Task already completed", {
-        existingTask,
       });
 
-      return taskWithAttemptsToServerTask(existingTask);
-    }
+      if (!existingTask) {
+        return;
+      }
 
-    if (existingTask.attempts.length === 1) {
-      await this.#prismaClient.taskAttempt.update({
+      if (existingTask.runId !== runId) {
+        return;
+      }
+
+      if (existingTask.run.environmentId !== environment.id) {
+        return;
+      }
+
+      if (
+        existingTask.status === "COMPLETED" ||
+        existingTask.status === "ERRORED" ||
+        existingTask.status === "CANCELED"
+      ) {
+        logger.debug("Task already completed", {
+          taskId: id,
+        });
+
+        return taskWithAttemptsToServerTask(existingTask);
+      }
+
+      if (existingTask.attempts.length === 1) {
+        await this.#prismaClient.taskAttempt.update({
+          where: {
+            id: existingTask.attempts[0].id,
+          },
+          data: {
+            status: "COMPLETED",
+          },
+        });
+      }
+
+      const updatedTask = await this.#prismaClient.task.update({
         where: {
-          id: existingTask.attempts[0].id,
+          id,
         },
         data: {
           status: "COMPLETED",
+          output: taskBody.output as any,
+          outputIsUndefined: typeof taskBody.output === "undefined",
+          completedAt: new Date(),
+          outputProperties: taskBody.properties,
+        },
+        include: {
+          attempts: true,
+          run: true,
         },
       });
-    }
 
-    const updatedTask = await this.#prismaClient.task.update({
-      where: {
-        id,
-      },
-      data: {
-        status: "COMPLETED",
-        output: taskBody.output as any,
-        outputIsUndefined: typeof taskBody.output === "undefined",
-        completedAt: new Date(),
-        outputProperties: taskBody.properties,
-      },
-      include: {
-        attempts: true,
-        run: true,
-      },
+      return taskWithAttemptsToServerTask(updatedTask);
     });
-
-    return taskWithAttemptsToServerTask(updatedTask);
   }
 }

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/route.ts

@@ -9,8 +9,10 @@ import {
 import { z } from "zod";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { logger } from "~/services/logger.server";
 import { CompleteRunTaskService } from "./CompleteRunTaskService.server";
+import { startActiveSpan } from "~/v3/tracer.server";
+import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server";
+import { FailRunTaskService } from "../api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -44,46 +46,51 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Invalid headers" }, { status: 400 });
   }
 
+  // Check the content size of the request and make sure it's not too large
+  const contentLength = request.headers.get("content-length");
+
+  if (!contentLength || parseInt(contentLength) > 3 * 1024 * 1024) {
+    const service = new FailRunTaskService();
+
+    await service.call(authenticatedEnv, runId, id, {
+      error: {
+        message: "Task output is too large. The limit is 3MB",
+      },
+    });
+
+    return json({ error: "Task output is too large. The limit is 3MB" }, { status: 413 });
+  }
+
   const { "trigger-version": triggerVersion } = headers.data;
 
   // Now parse the request body
-  const anyBody = await request.json();
-
-  logger.debug("CompleteRunTaskService.call() request body", {
-    body: anyBody,
-    runId,
-    id,
-  });
+  const anyBody = await parseRequestJsonAsync(request, { runId });
 
   if (triggerVersion === API_VERSIONS.SERIALIZED_TASK_OUTPUT) {
-    const body = CompleteTaskBodyV2InputSchema.safeParse(anyBody);
+    const body = await startActiveSpan("CompleteTaskBodyV2InputSchema.safeParse()", async () => {
+      return CompleteTaskBodyV2InputSchema.safeParse(anyBody);
+    });
 
     if (!body.success) {
       return json({ error: "Invalid request body" }, { status: 400 });
     }
 
-    // Make sure the length of the output is less than 3MB
-    if (body.data.output && body.data.output.length > 3 * 1024 * 1024) {
-      return json({ error: "Output must be less than 3MB" }, { status: 400 });
-    }
-
     return await completeRunTask(authenticatedEnv, runId, id, {
       ...body.data,
       output: body.data.output ? (JSON.parse(body.data.output) as any) : undefined,
     });
   } else {
-    const body = CompleteTaskBodyInputSchema.safeParse(anyBody);
+    const body = await startActiveSpan("CompleteTaskBodyInputSchema.safeParse()", async () => {
+      return CompleteTaskBodyInputSchema.omit({ output: true }).safeParse(anyBody);
+    });
 
     if (!body.success) {
       return json({ error: "Invalid request body" }, { status: 400 });
     }
 
-    // Make sure the length of the output is less than 3MB
-    if (JSON.stringify(body.data.output).length > 3 * 1024 * 1024) {
-      return json({ error: "Output must be less than 3MB" }, { status: 400 });
-    }
+    const output = (anyBody as any).output;
 
-    return await completeRunTask(authenticatedEnv, runId, id, body.data);
+    return await completeRunTask(authenticatedEnv, runId, id, { ...body.data, output });
   }
 }
 
@@ -98,12 +105,6 @@ async function completeRunTask(
   try {
     const task = await service.call(environment, runId, id, taskBody);
 
-    logger.debug("CompleteRunTaskService.call() response body", {
-      runId,
-      id,
-      task,
-    });
-
     if (!task) {
       return json({ message: "Task not found" }, { status: 404 });
     }

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server.ts

@@ -57,10 +57,6 @@ export class FailRunTaskService {
         existingTask.status === "ERRORED" ||
         existingTask.status === "CANCELED"
       ) {
-        logger.debug("Task already completed", {
-          existingTask,
-        });
-
         return existingTask;
       }
 

apps/webapp/app/routes/api.v1.runs.$runId.tasks/route.ts

@@ -6,6 +6,8 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { RunTaskService } from "~/services/tasks/runTask.server";
 import { ChangeRequestLazyLoadedCachedTasks } from "./ChangeRequestLazyLoadedCachedTasks.server";
+import { startActiveSpan } from "~/v3/tracer.server";
+import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -17,6 +19,8 @@ const HeadersSchema = z.object({
   "x-cached-tasks-cursor": z.string().optional().nullable(),
 });
 
+const BodySchema = RunTaskBodyOutputSchema.omit({ params: true });
+
 export async function action({ request, params }: ActionFunctionArgs) {
   // Ensure this is a POST request
   if (request.method.toUpperCase() !== "POST") {
@@ -44,18 +48,26 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   const { runId } = ParamsSchema.parse(params);
 
+  const contentLength = request.headers.get("content-length");
+
+  if (!contentLength || parseInt(contentLength) > 3 * 1024 * 1024) {
+    return json({ error: "Request body too large" }, { status: 413 });
+  }
+
   // Now parse the request body
-  const anyBody = await request.json();
+  const anyBody = await parseRequestJsonAsync(request, { runId });
 
-  logger.debug("RunTaskService.call() request body", {
-    body: anyBody,
-    runId,
-    idempotencyKey,
-    triggerVersion,
-    cachedTasksCursor,
-  });
-
-  const body = RunTaskBodyOutputSchema.safeParse(anyBody);
+  const body = await startActiveSpan(
+    "BodySchema.safeParse",
+    async () => {
+      return BodySchema.safeParse(anyBody);
+    },
+    {
+      attributes: {
+        runId,
+      },
+    }
+  );
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
@@ -64,12 +76,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const service = new RunTaskService();
 
   try {
-    const task = await service.call(runId, idempotencyKey, body.data);
-
-    logger.debug("RunTaskService.call() response body", {
-      runId,
-      idempotencyKey,
-      task,
+    const task = await service.call(runId, idempotencyKey, {
+      ...body.data,
+      params: (anyBody as any).params,
     });
 
     if (!task) {
@@ -84,7 +93,6 @@ export async function action({ request, params }: ActionFunctionArgs) {
       logger.debug(
         "RunTaskService.call() response migrating with ChangeRequestLazyLoadedCachedTasks",
         {
-          responseBody,
           cachedTasksCursor,
         }
       );

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

@@ -0,0 +1,85 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { RescheduleRunRequestBody } from "@trigger.dev/core/v3/schemas";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { RescheduleTaskRunService } from "~/v3/services/rescheduleTaskRun.server";
+
+const ParamsSchema = z.object({
+  runParam: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { runParam } = parsed.data;
+
+  const taskRun = await prisma.taskRun.findUnique({
+    where: {
+      friendlyId: runParam,
+      runtimeEnvironmentId: authenticationResult.environment.id,
+    },
+  });
+
+  if (!taskRun) {
+    return json({ error: "Run not found" }, { status: 404 });
+  }
+
+  const anyBody = await request.json();
+
+  const body = RescheduleRunRequestBody.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new RescheduleTaskRunService();
+
+  try {
+    const updatedRun = await service.call(taskRun, body.data);
+
+    if (!updatedRun) {
+      return json({ error: "An unknown error occurred" }, { status: 500 });
+    }
+
+    const presenter = new ApiRetrieveRunPresenter();
+    const result = await presenter.call(
+      updatedRun.friendlyId,
+      authenticationResult.environment,
+      true
+    );
+
+    if (!result) {
+      return json({ error: "Run not found" }, { status: 404 });
+    }
+
+    return json(result);
+  } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: 400 });
+    } else if (error instanceof Error) {
+      return json({ error: error.message }, { status: 500 });
+    } else {
+      return json({ error: "An unknown error occurred" }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts

@@ -7,6 +7,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { BatchTriggerTaskService } from "~/v3/services/batchTriggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { env } from "~/env.server";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -43,6 +44,12 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   const { taskId } = ParamsSchema.parse(params);
 
+  const contentLength = request.headers.get("content-length");
+
+  if (!contentLength || parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) {
+    return json({ error: "Request body too large" }, { status: 413 });
+  }
+
   // Now parse the request body
   const anyBody = await request.json();
 

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -2,9 +2,12 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { TriggerTaskRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
+import { env } from "~/env.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server";
 import { TriggerTaskService } from "~/v3/services/triggerTask.server";
+import { startActiveSpan } from "~/v3/tracer.server";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -32,6 +35,12 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Invalid or Missing API key" }, { status: 401 });
   }
 
+  const contentLength = request.headers.get("content-length");
+
+  if (!contentLength || parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) {
+    return json({ error: "Request body too large" }, { status: 413 });
+  }
+
   const rawHeaders = Object.fromEntries(request.headers);
 
   const headers = HeadersSchema.safeParse(rawHeaders);
@@ -52,9 +61,11 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const { taskId } = ParamsSchema.parse(params);
 
   // Now parse the request body
-  const anyBody = await request.json();
+  const anyBody = await parseRequestJsonAsync(request, { taskId });
 
-  const body = TriggerTaskRequestBody.safeParse(anyBody);
+  const body = await startActiveSpan("TriggerTaskRequestBody.safeParse()", async (span) => {
+    return TriggerTaskRequestBody.safeParse(anyBody);
+  });
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
@@ -76,17 +87,23 @@ export async function action({ request, params }: ActionFunctionArgs) {
       idempotencyKey,
       triggerVersion,
       headers: Object.fromEntries(request.headers),
-      body: body.data,
+      options: body.data.options,
       isFromWorker,
       traceContext,
     });
 
-    const run = await service.call(taskId, authenticationResult.environment, body.data, {
-      idempotencyKey: idempotencyKey ?? undefined,
-      triggerVersion: triggerVersion ?? undefined,
-      traceContext,
-      spanParentAsLink: spanParentAsLink === 1,
-    });
+    const run = await service.call(
+      taskId,
+      authenticationResult.environment,
+      { ...body.data },
+      // { ...body.data, payload: (anyBody as any).payload },
+      {
+        idempotencyKey: idempotencyKey ?? undefined,
+        triggerVersion: triggerVersion ?? undefined,
+        traceContext,
+        spanParentAsLink: spanParentAsLink === 1,
+      }
+    );
 
     if (!run) {
       return json({ error: "Task not found" }, { status: 404 });

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

@@ -33,6 +33,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const taskRun = await prisma.taskRun.findUnique({
     where: {
       friendlyId: runParam,
+      runtimeEnvironmentId: authenticationResult.environment.id,
     },
   });
 

apps/webapp/app/routes/projects.v3.$projectRef.runs.ts

@@ -0,0 +1,74 @@
+import { LoaderFunctionArgs, redirect } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { EnvSlug, isEnvSlug } from "~/models/api-key.server";
+import { requireUserId } from "~/services/session.server";
+
+const ParamsSchema = z.object({
+  projectRef: z.string(),
+});
+
+export async function loader({ params, request }: LoaderFunctionArgs) {
+  const userId = await requireUserId(request);
+
+  const { projectRef } = ParamsSchema.parse(params);
+
+  const project = await prisma.project.findFirst({
+    where: {
+      externalRef: projectRef,
+      organization: {
+        members: {
+          some: {
+            userId,
+          },
+        },
+      },
+    },
+    include: {
+      organization: true,
+    },
+  });
+
+  if (!project) {
+    return new Response("Project not found", { status: 404 });
+  }
+
+  const url = new URL(request.url);
+  const envSlug = url.searchParams.get("envSlug");
+
+  // Get the environment from the slug
+  if (envSlug && isEnvSlug(envSlug)) {
+    const env = await getEnvFromSlug(project.id, userId, envSlug);
+
+    if (env) {
+      url.searchParams.set("environments", env.id);
+    }
+
+    url.searchParams.delete("envSlug");
+  }
+
+  return redirect(
+    `/orgs/${project.organization.slug}/projects/v3/${project.slug}/runs${url.search}`
+  );
+}
+
+async function getEnvFromSlug(projectId: string, userId: string, envSlug: EnvSlug) {
+  if (envSlug === "dev") {
+    return await prisma.runtimeEnvironment.findFirst({
+      where: {
+        projectId,
+        slug: envSlug,
+        orgMember: {
+          userId,
+        },
+      },
+    });
+  }
+
+  return await prisma.runtimeEnvironment.findFirst({
+    where: {
+      projectId,
+      slug: envSlug,
+    },
+  });
+}

apps/webapp/app/routes/projects.v3.$projectRef.test.ts

@@ -0,0 +1,40 @@
+import { LoaderFunctionArgs, redirect } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { requireUserId } from "~/services/session.server";
+
+const ParamsSchema = z.object({
+  projectRef: z.string(),
+});
+
+export async function loader({ params, request }: LoaderFunctionArgs) {
+  const userId = await requireUserId(request);
+
+  const validatedParams = ParamsSchema.parse(params);
+
+  const project = await prisma.project.findFirst({
+    where: {
+      externalRef: validatedParams.projectRef,
+      organization: {
+        members: {
+          some: {
+            userId,
+          },
+        },
+      },
+    },
+    include: {
+      organization: true,
+    },
+  });
+
+  if (!project) {
+    return new Response("Not found", { status: 404 });
+  }
+
+  const url = new URL(request.url);
+
+  return redirect(
+    `/orgs/${project.organization.slug}/projects/v3/${project.slug}/test${url.search}`
+  );
+}

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -33,7 +33,13 @@ import { redirectWithErrorMessage } from "~/models/message.server";
 import { Span, SpanPresenter } from "~/presenters/v3/SpanPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { cn } from "~/utils/cn";
-import { v3RunPath, v3RunSpanPath, v3SpanParamsSchema, v3TraceSpanPath } from "~/utils/pathBuilder";
+import {
+  v3RunDownloadLogsPath,
+  v3RunPath,
+  v3RunSpanPath,
+  v3SpanParamsSchema,
+  v3TraceSpanPath,
+} from "~/utils/pathBuilder";
 import { SpanLink } from "~/v3/eventRepository.server";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
@@ -256,6 +262,15 @@ function RunActionButtons({ span }: { span: Span }) {
   if (span.isPartial) {
     return (
       <Dialog>
+        <LinkButton
+          to={v3RunDownloadLogsPath({ friendlyId: runParam })}
+          LeadingIcon={CloudArrowDownIcon}
+          variant="tertiary/medium"
+          target="_blank"
+          download
+        >
+          Download logs
+        </LinkButton>
         <DialogTrigger asChild>
           <Button variant="danger/medium" LeadingIcon={StopCircleIcon}>
             Cancel run
@@ -276,6 +291,15 @@ function RunActionButtons({ span }: { span: Span }) {
 
   return (
     <Dialog>
+      <LinkButton
+        to={v3RunDownloadLogsPath({ friendlyId: runParam })}
+        LeadingIcon={CloudArrowDownIcon}
+        variant="tertiary/medium"
+        target="_blank"
+        download
+      >
+        Download logs
+      </LinkButton>
       <DialogTrigger asChild>
         <Button variant="tertiary/medium" LeadingIcon={ArrowPathIcon}>
           Replay run

apps/webapp/app/routes/resources.packets.$environmentId.$.ts

@@ -2,9 +2,8 @@ import { LoaderFunctionArgs } from "@remix-run/node";
 import { basename } from "node:path";
 import { z } from "zod";
 import { prisma } from "~/db.server";
-import { env } from "~/env.server";
 import { requireUserId } from "~/services/session.server";
-import { r2 } from "~/v3/r2.server";
+import { generatePresignedRequest } from "~/v3/r2.server";
 
 const ParamSchema = z.object({
   environmentId: z.string(),
@@ -35,27 +34,17 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
-  if (!env.OBJECT_STORE_BASE_URL) {
-    return new Response("Object store base URL is not set", { status: 500 });
-  }
-
-  if (!r2) {
-    return new Response("Object store credentials are not set", { status: 500 });
-  }
-
-  const url = new URL(env.OBJECT_STORE_BASE_URL);
-  url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`;
-  url.searchParams.set("X-Amz-Expires", "30"); // 30 seconds
-
-  const signed = await r2.sign(
-    new Request(url, {
-      method: "GET",
-    }),
-    {
-      aws: { signQuery: true },
-    }
+  const signed = await generatePresignedRequest(
+    environment.project.externalRef,
+    environment.slug,
+    filename,
+    "GET"
   );
 
+  if (!signed) {
+    return new Response("Failed to generate presigned URL", { status: 500 });
+  }
+
   const response = await fetch(signed.url, {
     headers: signed.headers,
   });
@@ -64,7 +53,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     status: 200,
     headers: {
       "Content-Type": "application/octet-stream",
-      "Content-Disposition": `attachment; filename="${basename(url.pathname)}"`,
+      "Content-Disposition": `attachment; filename="${basename(filename)}"`,
     },
   });
 }

apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts

@@ -0,0 +1,112 @@
+import { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { requireUserId } from "~/services/session.server";
+import { v3RunParamsSchema } from "~/utils/pathBuilder";
+import {
+  PreparedEvent,
+  RunPreparedEvent,
+  eventRepository,
+  getDateFromNanoseconds,
+} from "~/v3/eventRepository.server";
+import { createGzip } from "zlib";
+import { Readable } from "stream";
+import { formatDurationMilliseconds } from "@trigger.dev/core/v3/utils/durations";
+
+export async function loader({ params, request }: LoaderFunctionArgs) {
+  const userId = await requireUserId(request);
+  const parsedParams = v3RunParamsSchema.pick({ runParam: true }).parse(params);
+
+  const run = await prisma.taskRun.findFirst({
+    where: {
+      friendlyId: parsedParams.runParam,
+      project: {
+        organization: {
+          members: {
+            some: {
+              userId,
+            },
+          },
+        },
+      },
+    },
+  });
+
+  if (!run) {
+    return new Response("Not found", { status: 404 });
+  }
+
+  const runEvents = await eventRepository.getRunEvents(run.friendlyId);
+
+  // Create a Readable stream from the runEvents array
+  const readable = new Readable({
+    read() {
+      runEvents.forEach((event) => {
+        try {
+          this.push(formatRunEvent(event) + "\n");
+        } catch {}
+      });
+      this.push(null); // End of stream
+    },
+  });
+
+  // Create a gzip transform stream
+  const gzip = createGzip();
+
+  // Pipe the readable stream into the gzip stream
+  const compressedStream = readable.pipe(gzip);
+
+  // Return the response with the compressed stream
+  return new Response(compressedStream as any, {
+    status: 200,
+    headers: {
+      "Content-Type": "application/octet-stream",
+      "Content-Disposition": `attachment; filename="${parsedParams.runParam}.log"`,
+      "Content-Encoding": "gzip",
+    },
+  });
+}
+
+function formatRunEvent(event: RunPreparedEvent): string {
+  const entries = [];
+  const parts: string[] = [];
+
+  parts.push(getDateFromNanoseconds(event.startTime).toISOString());
+
+  if (event.taskSlug) {
+    parts.push(event.taskSlug);
+  }
+
+  parts.push(event.level);
+  parts.push(event.message);
+
+  if (event.level === "TRACE") {
+    parts.push(`(${formatDurationMilliseconds(event.duration / 1_000_000)})`);
+  }
+
+  entries.push(parts.join(" "));
+
+  if (event.events) {
+    for (const subEvent of event.events) {
+      if (subEvent.name === "exception") {
+        const subEventParts: string[] = [];
+
+        subEventParts.push(subEvent.time as unknown as string);
+
+        if (event.taskSlug) {
+          subEventParts.push(event.taskSlug);
+        }
+
+        subEventParts.push(subEvent.name);
+        subEventParts.push((subEvent.properties as any).exception.message);
+
+        if ((subEvent.properties as any).exception.stack) {
+          subEventParts.push((subEvent.properties as any).exception.stack);
+        }
+
+        entries.push(subEventParts.join(" "));
+      }
+    }
+  }
+
+  return entries.join("\n");
+}

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

@@ -136,7 +136,7 @@ export class EndpointApi {
     };
   }
 
-  async executeJobRequest(options: RunJobBody) {
+  async executeJobRequest(options: RunJobBody, timeoutInMs?: number) {
     const startTimeInMs = performance.now();
 
     const response = await safeFetch(this.url, {
@@ -147,8 +147,18 @@ export class EndpointApi {
         "x-trigger-action": "EXECUTE_JOB",
       },
       body: JSON.stringify(options),
+      signal: timeoutInMs ? AbortSignal.timeout(timeoutInMs) : undefined,
     });
 
+    if (response) {
+      logger.debug("executeJobRequest() response from endpoint", {
+        status: response.status,
+        headers: Object.fromEntries(response.headers.entries()),
+      });
+    } else {
+      logger.debug("executeJobRequest() no response from endpoint");
+    }
+
     return {
       response,
       parser: RunJobResponseSchema,
@@ -434,7 +444,10 @@ async function safeFetch(url: string, options: RequestInit) {
   } catch (error) {
     logger.debug("Error while trying to connect to endpoint", {
       url,
-      error,
+      error:
+        error instanceof Error
+          ? { name: error.name, message: error.message, stack: error.stack }
+          : String(error),
     });
   }
 }

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

@@ -269,7 +269,10 @@ export class PerformRunExecutionV3Service {
 
       // TODO: add the ability to abort the execution from any server using Redis pub/sub
       const { response, parser, errorParser, headersParser, durationInMs } =
-        await client.executeJobRequest(executionBody);
+        await client.executeJobRequest(
+          executionBody,
+          run.environment.type === "DEVELOPMENT" ? 60_000 * 5 : undefined
+        );
 
       await createExecutionEvent({
         eventType: "finish",
@@ -929,6 +932,25 @@ export class PerformRunExecutionV3Service {
     executionCount: number = 1
   ) {
     await $transaction(this.#prismaClient, async (tx) => {
+      const service = new CompleteRunTaskService(tx);
+
+      const task = await service.call(run.environment, run.id, data.id, {
+        properties: data.properties,
+        output: data.output ? (JSON.parse(data.output) as any) : undefined,
+      });
+
+      if (!task || task.status === "ERRORED") {
+        return await this.#failRunExecution(
+          tx,
+          run,
+          {
+            message: task ? `Task '${task.name}' failed to complete` : "Task failed to complete",
+          },
+          "FAILURE",
+          durationInMs
+        );
+      }
+
       await tx.jobRun.update({
         where: {
           id: run.id,
@@ -958,13 +980,6 @@ export class PerformRunExecutionV3Service {
         },
       });
 
-      const service = new CompleteRunTaskService(tx);
-
-      await service.call(run.environment, run.id, data.id, {
-        properties: data.properties,
-        output: data.output ? (JSON.parse(data.output) as any) : undefined,
-      });
-
       await ResumeRunService.enqueue(run, tx);
     });
   }

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

@@ -6,6 +6,7 @@ import { taskWithAttemptsToServerTask } from "~/models/task.server";
 import { generateSecret } from "~/services/sources/utils.server";
 import { ulid } from "~/services/ulid.server";
 import { taskOperationWorker, workerQueue } from "~/services/worker.server";
+import { startActiveSpan } from "~/v3/tracer.server";
 
 export class RunTaskService {
   #prismaClient: PrismaClient;
@@ -19,142 +20,154 @@ export class RunTaskService {
     idempotencyKey: string,
     taskBody: RunTaskBodyOutput
   ): Promise<ServerTask | undefined> {
-    const delayUntilInFuture = taskBody.delayUntil
-      ? taskBody.delayUntil.getTime() > Date.now()
-      : false;
-    const callbackEnabled = taskBody.callback?.enabled ?? false;
+    return startActiveSpan("RunTaskService.call", async (span) => {
+      span.setAttribute("runId", runId);
 
-    // First
-    const existingTask = await this.#handleExistingTask(
-      runId,
-      idempotencyKey,
-      taskBody,
-      delayUntilInFuture,
-      callbackEnabled
-    );
+      const delayUntilInFuture = taskBody.delayUntil
+        ? taskBody.delayUntil.getTime() > Date.now()
+        : false;
+      const callbackEnabled = taskBody.callback?.enabled ?? false;
 
-    if (existingTask) {
-      return taskWithAttemptsToServerTask(existingTask);
-    }
+      // First
+      const existingTask = await this.#handleExistingTask(
+        runId,
+        idempotencyKey,
+        taskBody,
+        delayUntilInFuture,
+        callbackEnabled
+      );
 
-    const run = await this.#prismaClient.jobRun.findUnique({
-      where: {
-        id: runId,
-      },
-      select: {
-        status: true,
-        forceYieldImmediately: true,
-      },
-    });
+      if (existingTask) {
+        span.setAttribute("taskId", existingTask.id);
 
-    if (!run) throw new Error("Run not found");
-
-    const runConnection = taskBody.connectionKey
-      ? await this.#prismaClient.runConnection.findUnique({
-          where: {
-            runId_key: {
-              runId,
-              key: taskBody.connectionKey,
-            },
-          },
-          select: {
-            id: true,
-          },
-        })
-      : undefined;
-
-    const results = await $transaction(this.#prismaClient, async (tx) => {
-      // If task.delayUntil is set and is in the future, we'll set the task's status to "WAITING", else set it to RUNNING
-      let status: TaskStatus;
-
-      if (run.status === "CANCELED") {
-        status = "CANCELED";
-      } else {
-        status =
-          delayUntilInFuture || callbackEnabled
-            ? "WAITING"
-            : taskBody.noop
-            ? "COMPLETED"
-            : "RUNNING";
+        return taskWithAttemptsToServerTask(existingTask);
       }
 
-      const taskId = ulid();
-      const callbackUrl = callbackEnabled
-        ? `${env.APP_ORIGIN}/api/v1/tasks/${taskId}/callback/${generateSecret(12)}`
+      const run = await this.#prismaClient.jobRun.findUnique({
+        where: {
+          id: runId,
+        },
+        select: {
+          status: true,
+          forceYieldImmediately: true,
+        },
+      });
+
+      if (!run) throw new Error("Run not found");
+
+      const runConnection = taskBody.connectionKey
+        ? await this.#prismaClient.runConnection.findUnique({
+            where: {
+              runId_key: {
+                runId,
+                key: taskBody.connectionKey,
+              },
+            },
+            select: {
+              id: true,
+            },
+          })
         : undefined;
 
-      const task = await tx.task.create({
-        data: {
-          id: taskId,
-          idempotencyKey,
-          displayKey: taskBody.displayKey,
-          runConnectionId: runConnection ? runConnection.id : undefined,
-          icon: taskBody.icon,
-          runId,
-          parentId: taskBody.parentId,
-          name: taskBody.name ?? "Task",
-          description: taskBody.description,
-          status,
-          startedAt: new Date(),
-          completedAt: status === "COMPLETED" || status === "CANCELED" ? new Date() : undefined,
-          noop: taskBody.noop,
-          delayUntil: taskBody.delayUntil,
-          params: taskBody.params ?? undefined,
-          properties: this.#filterProperties(taskBody.properties) ?? undefined,
-          redact: taskBody.redact ?? undefined,
-          operation: taskBody.operation,
-          callbackUrl,
-          style: taskBody.style ?? { style: "normal" },
-          childExecutionMode: taskBody.parallel ? "PARALLEL" : "SEQUENTIAL",
-        },
-      });
+      const results = await $transaction(
+        this.#prismaClient,
+        async (tx) => {
+          // If task.delayUntil is set and is in the future, we'll set the task's status to "WAITING", else set it to RUNNING
+          let status: TaskStatus;
 
-      const taskAttempt = await tx.taskAttempt.create({
-        data: {
-          number: 1,
-          taskId: task.id,
-          status: "PENDING",
-        },
-      });
+          if (run.status === "CANCELED") {
+            status = "CANCELED";
+          } else {
+            status =
+              delayUntilInFuture || callbackEnabled
+                ? "WAITING"
+                : taskBody.noop
+                ? "COMPLETED"
+                : "RUNNING";
+          }
 
-      if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
-        // We need to schedule the operation
-        await taskOperationWorker.enqueue(
-          "performTaskOperation",
-          {
-            id: task.id,
-          },
-          { tx, runAt: task.delayUntil ?? undefined, jobKey: `operation:${task.id}` }
-        );
-      } else if (task.status === "WAITING" && callbackUrl && taskBody.callback) {
-        if (taskBody.callback.timeoutInSeconds > 0) {
-          // We need to schedule the callback timeout
-          await workerQueue.enqueue(
-            "processCallbackTimeout",
-            {
-              id: task.id,
+          const taskId = ulid();
+          const callbackUrl = callbackEnabled
+            ? `${env.APP_ORIGIN}/api/v1/tasks/${taskId}/callback/${generateSecret(12)}`
+            : undefined;
+
+          const task = await tx.task.create({
+            data: {
+              id: taskId,
+              idempotencyKey,
+              displayKey: taskBody.displayKey,
+              runConnectionId: runConnection ? runConnection.id : undefined,
+              icon: taskBody.icon,
+              runId,
+              parentId: taskBody.parentId,
+              name: taskBody.name ?? "Task",
+              description: taskBody.description,
+              status,
+              startedAt: new Date(),
+              completedAt: status === "COMPLETED" || status === "CANCELED" ? new Date() : undefined,
+              noop: taskBody.noop,
+              delayUntil: taskBody.delayUntil,
+              params: taskBody.params ?? undefined,
+              properties: this.#filterProperties(taskBody.properties) ?? undefined,
+              redact: taskBody.redact ?? undefined,
+              operation: taskBody.operation,
+              callbackUrl,
+              style: taskBody.style ?? { style: "normal" },
+              childExecutionMode: taskBody.parallel ? "PARALLEL" : "SEQUENTIAL",
             },
-            {
-              tx,
-              runAt: new Date(Date.now() + taskBody.callback.timeoutInSeconds * 1000),
-              jobKey: `process-callback:${task.id}`,
+          });
+
+          span.setAttribute("taskId", task.id);
+
+          const taskAttempt = await tx.taskAttempt.create({
+            data: {
+              number: 1,
+              taskId: task.id,
+              status: "PENDING",
+            },
+          });
+
+          if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
+            // We need to schedule the operation
+            await taskOperationWorker.enqueue(
+              "performTaskOperation",
+              {
+                id: task.id,
+              },
+              { tx, runAt: task.delayUntil ?? undefined, jobKey: `operation:${task.id}` }
+            );
+          } else if (task.status === "WAITING" && callbackUrl && taskBody.callback) {
+            if (taskBody.callback.timeoutInSeconds > 0) {
+              // We need to schedule the callback timeout
+              await workerQueue.enqueue(
+                "processCallbackTimeout",
+                {
+                  id: task.id,
+                },
+                {
+                  tx,
+                  runAt: new Date(Date.now() + taskBody.callback.timeoutInSeconds * 1000),
+                  jobKey: `process-callback:${task.id}`,
+                }
+              );
             }
-          );
-        }
+          }
+
+          return { task, taskAttempt };
+        },
+        { timeout: 10000 }
+      );
+
+      if (!results) {
+        return;
       }
 
-      return { task, taskAttempt };
+      const { task, taskAttempt } = results;
+
+      return task
+        ? taskWithAttemptsToServerTask({ ...task, attempts: [taskAttempt], run })
+        : undefined;
     });
-
-    if (!results) {
-      return;
-    }
-
-    const { task, taskAttempt } = results;
-
-    return task
-      ? taskWithAttemptsToServerTask({ ...task, attempts: [taskAttempt], run })
-      : undefined;
   }
 
   async #handleExistingTask(

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

@@ -47,6 +47,8 @@ import { ResumeTaskService } from "./tasks/resumeTask.server";
 import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
 import { MarqsConcurrencyMonitor } from "~/v3/marqs/concurrencyMonitor.server";
 import { reportUsageEvent } from "~/v3/openMeter.server";
+import { EnqueueDelayedRunService } from "~/v3/services/enqueueDelayedRun.server";
+import { ExpireEnqueuedRunService } from "~/v3/services/expireEnqueuedRun.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -177,6 +179,12 @@ const workerCatalog = {
     }),
     additionalData: z.record(z.any()).optional(),
   }),
+  "v3.enqueueDelayedRun": z.object({
+    runId: z.string(),
+  }),
+  "v3.expireRun": z.object({
+    runId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -672,6 +680,24 @@ function getWorkerQueue() {
           });
         },
       },
+      "v3.enqueueDelayedRun": {
+        priority: 0,
+        maxAttempts: 8,
+        handler: async (payload, job) => {
+          const service = new EnqueueDelayedRunService();
+
+          return await service.call(payload.runId);
+        },
+      },
+      "v3.expireRun": {
+        priority: 0,
+        maxAttempts: 8,
+        handler: async (payload, job) => {
+          const service = new ExpireEnqueuedRunService();
+
+          return await service.call(payload.runId);
+        },
+      },
     },
   });
 }

apps/webapp/app/utils/parseRequestJson.server.ts

@@ -0,0 +1,29 @@
+import { Attributes } from "@opentelemetry/api";
+import { startActiveSpan } from "~/v3/tracer.server";
+
+export async function parseRequestJsonAsync(
+  request: Request,
+  attributes?: Attributes
+): Promise<unknown> {
+  return await startActiveSpan(
+    "parseRequestJsonAsync()",
+    async (span) => {
+      span.setAttribute("content-length", parseInt(request.headers.get("content-length") ?? "0"));
+      span.setAttribute("content-type", request.headers.get("content-type") ?? "application/json");
+      span.setAttribute("experiment.async", false);
+
+      const rawText = await startActiveSpan("request.text()", async () => {
+        return await request.text();
+      });
+
+      if (rawText.length === 0) {
+        return;
+      }
+
+      return JSON.parse(rawText);
+    },
+    {
+      attributes,
+    }
+  );
+}

apps/webapp/app/utils/pathBuilder.ts

@@ -375,6 +375,10 @@ export function v3RunPath(organization: OrgForPath, project: ProjectForPath, run
   return `${v3RunsPath(organization, project)}/${run.friendlyId}`;
 }
 
+export function v3RunDownloadLogsPath(run: v3RunForPath) {
+  return `/resources/runs/${run.friendlyId}/logs/download`;
+}
+
 export function v3RunSpanPath(
   organization: OrgForPath,
   project: ProjectForPath,

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -740,7 +740,7 @@ async function resolveBuiltInProdVariables(runtimeEnvironment: RuntimeEnvironmen
     },
     {
       key: "TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS",
-      value: String(env.RUNTIME_WAIT_THRESHOLD_IN_MS),
+      value: String(env.CHECKPOINT_THRESHOLD_IN_MS),
     },
     {
       key: "TRIGGER_ORG_ID",

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

@@ -129,6 +129,10 @@ export type PreparedEvent = Omit<QueriedEvent, "events" | "style" | "duration">
   style: TaskEventStyle;
 };
 
+export type RunPreparedEvent = PreparedEvent & {
+  taskSlug?: string;
+};
+
 export type SpanLink =
   | {
       type: "run";
@@ -340,7 +344,7 @@ export class EventRepository {
     });
   }
 
-  async queryIncompleteEvents(queryOptions: QueryOptions) {
+  async queryIncompleteEvents(queryOptions: QueryOptions, allowCompleteDuplicate = false) {
     // First we will find all the events that match the query options (selecting minimal data).
     const taskEvents = await this.readReplica.taskEvent.findMany({
       where: queryOptions,
@@ -358,6 +362,10 @@ export class EventRepository {
       // If the event is cancelled, it is not incomplete
       if (event.isCancelled) return false;
 
+      if (allowCompleteDuplicate) {
+        return true;
+      }
+
       // There must not be another complete event with the same spanId
       const hasCompleteDuplicate = taskEvents.some(
         (otherEvent) =>
@@ -400,13 +408,19 @@ export class EventRepository {
         orderBy: {
           startTime: "asc",
         },
+        take: env.MAXIMUM_TRACE_SUMMARY_VIEW_COUNT,
       });
 
       let preparedEvents: Array<PreparedEvent> = [];
+      let rootSpanId: string | undefined;
       const eventsBySpanId = new Map<string, PreparedEvent>();
 
       for (const event of events) {
         preparedEvents.push(prepareEvent(event));
+
+        if (!rootSpanId && !event.parentId) {
+          rootSpanId = event.spanId;
+        }
       }
 
       for (const event of preparedEvents) {
@@ -424,6 +438,8 @@ export class EventRepository {
 
       preparedEvents = Array.from(eventsBySpanId.values());
 
+      const spansBySpanId = new Map<string, SpanSummary>();
+
       const spans = preparedEvents.map((event) => {
         const ancestorCancelled = isAncestorCancelled(eventsBySpanId, event.spanId);
         const duration = calculateDurationIfAncestorIsCancelled(
@@ -432,7 +448,7 @@ export class EventRepository {
           event.duration
         );
 
-        return {
+        const span = {
           recordId: event.id,
           id: event.spanId,
           parentId: event.parentId ?? undefined,
@@ -451,14 +467,17 @@ export class EventRepository {
             environmentType: event.environmentType,
           },
         };
+
+        spansBySpanId.set(event.spanId, span);
+
+        return span;
       });
 
-      const rootSpanId = events.find((event) => !event.parentId);
       if (!rootSpanId) {
         return;
       }
 
-      const rootSpan = spans.find((span) => span.id === rootSpanId.spanId);
+      const rootSpan = spansBySpanId.get(rootSpanId);
 
       if (!rootSpan) {
         return;
@@ -471,85 +490,259 @@ export class EventRepository {
     });
   }
 
+  public async getRunEvents(runId: string): Promise<RunPreparedEvent[]> {
+    return await startActiveSpan("getRunEvents", async (span) => {
+      const events = await this.readReplica.taskEvent.findMany({
+        select: {
+          id: true,
+          spanId: true,
+          parentId: true,
+          runId: true,
+          idempotencyKey: true,
+          message: true,
+          style: true,
+          startTime: true,
+          duration: true,
+          isError: true,
+          isPartial: true,
+          isCancelled: true,
+          level: true,
+          events: true,
+          environmentType: true,
+          taskSlug: true,
+        },
+        where: {
+          runId,
+          isPartial: false,
+        },
+        orderBy: {
+          startTime: "asc",
+        },
+      });
+
+      let preparedEvents: Array<PreparedEvent> = [];
+
+      for (const event of events) {
+        preparedEvents.push(prepareEvent(event));
+      }
+
+      return preparedEvents;
+    });
+  }
+
   // A Span can be cancelled if it is partial and has a parent that is cancelled
   // And a span's duration, if it is partial and has a cancelled parent, is the time between the start of the span and the time of the cancellation event of the parent
   public async getSpan(spanId: string, traceId: string) {
-    const traceSummary = await this.getTraceSummary(traceId);
+    return await startActiveSpan("getSpan", async (s) => {
+      const spanEvent = await this.#getSpanEvent(spanId);
 
-    const span = traceSummary?.spans.find((span) => span.id === spanId);
+      if (!spanEvent) {
+        return;
+      }
 
-    if (!span) {
-      return;
-    }
+      const preparedEvent = prepareEvent(spanEvent);
 
-    const fullEvent = await this.readReplica.taskEvent.findUnique({
-      where: {
-        id: span.recordId,
-      },
-    });
+      const span = await this.#createSpanFromEvent(preparedEvent);
 
-    if (!fullEvent) {
-      return;
-    }
+      const output = rehydrateJson(spanEvent.output);
+      const payload = rehydrateJson(spanEvent.payload);
 
-    const output = rehydrateJson(fullEvent.output);
-    const payload = rehydrateJson(fullEvent.payload);
+      const show = rehydrateShow(spanEvent.properties);
 
-    const show = rehydrateShow(fullEvent.properties);
+      const properties = sanitizedAttributes(spanEvent.properties);
 
-    const properties = sanitizedAttributes(fullEvent.properties);
+      const messagingEvent = SpanMessagingEvent.optional().safeParse(
+        (properties as any)?.messaging
+      );
 
-    const messagingEvent = SpanMessagingEvent.optional().safeParse((properties as any)?.messaging);
+      const links: SpanLink[] = [];
 
-    const links: SpanLink[] = [];
-
-    if (messagingEvent.success && messagingEvent.data) {
-      if (messagingEvent.data.message && "id" in messagingEvent.data.message) {
-        if (messagingEvent.data.message.id.startsWith("run_")) {
-          links.push({
-            type: "run",
-            icon: "runs",
-            title: `Run ${messagingEvent.data.message.id}`,
-            runId: messagingEvent.data.message.id,
-          });
+      if (messagingEvent.success && messagingEvent.data) {
+        if (messagingEvent.data.message && "id" in messagingEvent.data.message) {
+          if (messagingEvent.data.message.id.startsWith("run_")) {
+            links.push({
+              type: "run",
+              icon: "runs",
+              title: `Run ${messagingEvent.data.message.id}`,
+              runId: messagingEvent.data.message.id,
+            });
+          }
         }
       }
-    }
 
-    const backLinks = fullEvent.links as any as Link[] | undefined;
+      const backLinks = spanEvent.links as any as Link[] | undefined;
 
-    if (backLinks && backLinks.length > 0) {
-      backLinks.forEach((l) => {
-        const title = String(
-          l.attributes?.[SemanticInternalAttributes.LINK_TITLE] ?? "Triggered by"
-        );
+      if (backLinks && backLinks.length > 0) {
+        backLinks.forEach((l) => {
+          const title = String(
+            l.attributes?.[SemanticInternalAttributes.LINK_TITLE] ?? "Triggered by"
+          );
 
-        links.push({
-          type: "span",
-          icon: "trigger",
-          title,
-          traceId: l.context.traceId,
-          spanId: l.context.spanId,
+          links.push({
+            type: "span",
+            icon: "trigger",
+            title,
+            traceId: l.context.traceId,
+            spanId: l.context.spanId,
+          });
         });
-      });
+      }
+
+      const spanEvents = transformEvents(
+        preparedEvent.events,
+        spanEvent.metadata as Attributes,
+        spanEvent.environmentType === "DEVELOPMENT"
+      );
+
+      return {
+        ...spanEvent,
+        ...span.data,
+        payload,
+        output,
+        properties,
+        events: spanEvents,
+        show,
+        links,
+      };
+    });
+  }
+
+  async #createSpanFromEvent(event: PreparedEvent) {
+    return await startActiveSpan("createSpanFromEvent", async (s) => {
+      let ancestorCancelled = false;
+      let duration = event.duration;
+
+      if (!event.isCancelled && event.isPartial) {
+        await this.#walkSpanAncestors(event, (ancestorEvent, level) => {
+          if (level >= 8) {
+            return { stop: true };
+          }
+
+          if (ancestorEvent.isCancelled) {
+            ancestorCancelled = true;
+
+            // We need to get the cancellation time from the cancellation span event
+            const cancellationEvent = ancestorEvent.events.find(
+              (event) => event.name === "cancellation"
+            );
+
+            if (cancellationEvent) {
+              duration = calculateDurationFromStart(event.startTime, cancellationEvent.time);
+            }
+
+            return { stop: true };
+          }
+
+          return { stop: false };
+        });
+      }
+
+      const span = {
+        recordId: event.id,
+        id: event.spanId,
+        parentId: event.parentId ?? undefined,
+        runId: event.runId,
+        idempotencyKey: event.idempotencyKey,
+        data: {
+          message: event.message,
+          style: event.style,
+          duration,
+          isError: event.isError,
+          isPartial: ancestorCancelled ? false : event.isPartial,
+          isCancelled: event.isCancelled === true ? true : event.isPartial && ancestorCancelled,
+          startTime: getDateFromNanoseconds(event.startTime),
+          level: event.level,
+          events: event.events,
+          environmentType: event.environmentType,
+        },
+      };
+
+      return span;
+    });
+  }
+
+  async #walkSpanAncestors(
+    event: PreparedEvent,
+    callback: (event: PreparedEvent, level: number) => { stop: boolean }
+  ) {
+    const parentId = event.parentId;
+    if (!parentId) {
+      return;
     }
 
-    const events = transformEvents(
-      span.data.events,
-      fullEvent.metadata as Attributes,
-      traceSummary?.rootSpan.data.environmentType === "DEVELOPMENT"
-    );
+    await startActiveSpan("walkSpanAncestors", async (s) => {
+      let parentEvent = await this.#getSpanEvent(parentId);
+      let level = 1;
 
-    return {
-      ...fullEvent,
-      ...span.data,
-      payload,
-      output,
-      properties,
-      events,
-      show,
-      links,
-    };
+      while (parentEvent) {
+        const preparedParentEvent = prepareEvent(parentEvent);
+
+        const result = callback(preparedParentEvent, level);
+
+        if (result.stop) {
+          return;
+        }
+
+        if (!preparedParentEvent.parentId) {
+          return;
+        }
+
+        parentEvent = await this.#getSpanEvent(preparedParentEvent.parentId);
+
+        level++;
+      }
+    });
+  }
+
+  async #getSpanAncestors(event: PreparedEvent, levels = 1): Promise<Array<PreparedEvent>> {
+    if (levels >= 8) {
+      return [];
+    }
+
+    if (!event.parentId) {
+      return [];
+    }
+
+    const parentEvent = await this.#getSpanEvent(event.parentId);
+
+    if (!parentEvent) {
+      return [];
+    }
+
+    const preparedParentEvent = prepareEvent(parentEvent);
+
+    if (!preparedParentEvent.parentId) {
+      return [preparedParentEvent];
+    }
+
+    const moreAncestors = await this.#getSpanAncestors(preparedParentEvent, levels + 1);
+
+    return [preparedParentEvent, ...moreAncestors];
+  }
+
+  async #getSpanEvent(spanId: string) {
+    return await startActiveSpan("getSpanEvent", async (s) => {
+      const events = await this.readReplica.taskEvent.findMany({
+        where: {
+          spanId,
+        },
+        orderBy: {
+          startTime: "asc",
+        },
+      });
+
+      let finalEvent: TaskEvent | undefined;
+
+      for (const event of events) {
+        if (event.isPartial && finalEvent) {
+          continue;
+        }
+
+        finalEvent = event;
+      }
+
+      return finalEvent;
+    });
   }
 
   public async recordEvent(message: string, options: TraceEventOptions) {
@@ -929,6 +1122,7 @@ export function createExceptionPropertiesFromError(error: TaskRunError): Excepti
       return {
         type: "Internal error",
         message: [error.code, error.message].filter(Boolean).join(": "),
+        stacktrace: error.stackTrace,
       };
     }
     case "STRING_ERROR": {
@@ -1224,7 +1418,7 @@ function getNowInNanoseconds(): bigint {
   return BigInt(new Date().getTime() * 1_000_000);
 }
 
-function getDateFromNanoseconds(nanoseconds: bigint) {
+export function getDateFromNanoseconds(nanoseconds: bigint) {
   return new Date(Number(nanoseconds) / 1_000_000);
 }
 

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

@@ -9,14 +9,19 @@ import { BaseService } from "./services/baseService.server";
 const FAILABLE_TASK_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "PENDING", "WAITING_FOR_DEPLOY"];
 
 export class FailedTaskRunService extends BaseService {
-  public async call(runFriendlyId: string, completion: TaskRunFailedExecutionResult) {
+  public async call(anyRunId: string, completion: TaskRunFailedExecutionResult) {
+    const isFriendlyId = anyRunId.startsWith("run_");
+
     const taskRun = await this._prisma.taskRun.findUnique({
-      where: { friendlyId: runFriendlyId },
+      where: {
+        friendlyId: isFriendlyId ? anyRunId : undefined,
+        id: !isFriendlyId ? anyRunId : undefined,
+      },
     });
 
     if (!taskRun) {
       logger.error("[FailedTaskRunService] Task run not found", {
-        runFriendlyId,
+        anyRunId,
         completion,
       });
 

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

@@ -1,5 +1,6 @@
 import {
   ClientToSharedQueueMessages,
+  CoordinatorSocketData,
   CoordinatorToPlatformMessages,
   PlatformToCoordinatorMessages,
   PlatformToProviderMessages,
@@ -78,6 +79,7 @@ function createCoordinatorNamespace(io: Server) {
     authToken: env.COORDINATOR_SECRET,
     clientMessages: CoordinatorToPlatformMessages,
     serverMessages: PlatformToCoordinatorMessages,
+    socketData: CoordinatorSocketData,
     handlers: {
       READY_FOR_EXECUTION: async (message) => {
         const payload = await sharedQueueTasks.getLatestExecutionPayloadFromRun(
@@ -238,6 +240,45 @@ function createCoordinatorNamespace(io: Server) {
         }
       },
     },
+    onConnection: async (socket, handler, sender, logger) => {
+      if (socket.data.supportsDynamicConfig) {
+        socket.emit("DYNAMIC_CONFIG", {
+          version: "v1",
+          checkpointThresholdInMs: env.CHECKPOINT_THRESHOLD_IN_MS,
+        });
+      }
+    },
+    postAuth: async (socket, next, logger) => {
+      function setSocketDataFromHeader(
+        dataKey: keyof typeof socket.data,
+        headerName: string,
+        required: boolean = true
+      ) {
+        const value = socket.handshake.headers[headerName];
+
+        if (value) {
+          socket.data[dataKey] = Array.isArray(value) ? value[0] : value;
+          return;
+        }
+
+        if (required) {
+          logger.error("missing required header", { headerName });
+          throw new Error("missing header");
+        }
+      }
+
+      try {
+        setSocketDataFromHeader("supportsDynamicConfig", "x-supports-dynamic-config", false);
+      } catch (error) {
+        logger.error("setSocketDataFromHeader error", { error });
+        socket.disconnect(true);
+        return;
+      }
+
+      logger.debug("success", socket.data);
+
+      next();
+    },
   });
 
   return coordinator.namespace;
@@ -267,7 +308,7 @@ function createProviderNamespace(io: Server) {
         try {
           const service = new DeploymentIndexFailed();
 
-          await service.call(message.deploymentId, message.error);
+          await service.call(message.deploymentId, message.error, message.overrideCompletion);
         } catch (e) {
           logger.error("Error while indexing", { error: e });
         }

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

@@ -51,11 +51,24 @@ export class MarqsConcurrencyMonitor {
       processedKeys: 0,
     };
 
-    const { stream, redis } = this.marqs.queueConcurrencyScanStream(10, () => {
-      this._logger.debug("[MarqsConcurrencyMonitor] stream closed", {
-        stats,
-      });
-    });
+    const { stream, redis } = this.marqs.queueConcurrencyScanStream(
+      10,
+      () => {
+        this._logger.debug("[MarqsConcurrencyMonitor] stream closed", {
+          stats,
+        });
+      },
+      (error) => {
+        this._logger.debug("[MarqsConcurrencyMonitor] stream error", {
+          stats,
+          error: {
+            name: error.name,
+            message: error.message,
+            stack: error.stack,
+          },
+        });
+      }
+    );
 
     stream.on("data", async (keys) => {
       stream.pause();
@@ -80,9 +93,11 @@ export class MarqsConcurrencyMonitor {
 
       stats.processedKeys += uniqueKeys.length;
 
-      await Promise.all(uniqueKeys.map((key) => this.#processKey(key, redis))).finally(() => {
-        stream.resume();
-      });
+      await Promise.allSettled(uniqueKeys.map((key) => this.#processKey(key, redis))).finally(
+        () => {
+          stream.resume();
+        }
+      );
     });
   }
 
@@ -91,8 +106,20 @@ export class MarqsConcurrencyMonitor {
     const orgKey = this.keys.orgCurrentConcurrencyKeyFromQueue(key);
     const envKey = this.keys.envCurrentConcurrencyKeyFromQueue(key);
 
-    // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
-    const runIds = await redis.sunion(orgKey, envKey, key);
+    let runIds: string[] = [];
+
+    try {
+      // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
+      runIds = await redis.sunion(orgKey, envKey, key);
+    } catch (e) {
+      this._logger.error("[MarqsConcurrencyMonitor] error during sunion", {
+        key,
+        orgKey,
+        envKey,
+        runIds,
+        error: e,
+      });
+    }
 
     if (runIds.length === 0) {
       return;

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

@@ -788,7 +788,11 @@ export class MarQS {
     }
   }
 
-  queueConcurrencyScanStream(count: number = 100, onEndCallback?: () => void) {
+  queueConcurrencyScanStream(
+    count: number = 100,
+    onEndCallback?: () => void,
+    onErrorCallback?: (error: Error) => void
+  ) {
     const pattern = this.keys.queueCurrentConcurrencyScanPattern();
 
     logger.debug("Starting queue concurrency scan stream", {
@@ -812,6 +816,11 @@ export class MarQS {
       redis.quit();
     });
 
+    stream.on("error", (error) => {
+      onErrorCallback?.(error);
+      redis.quit();
+    });
+
     return { stream, redis };
   }
 
@@ -1625,7 +1634,7 @@ function getMarQSClient() {
         defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
         defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
         visibilityTimeoutInMs: 120 * 1000, // 2 minutes,
-        enableRebalancing: false,
+        enableRebalancing: !env.MARQS_DISABLE_REBALANCING,
       });
     } else {
       console.warn(

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

@@ -268,6 +268,7 @@ export class SharedQueueConsumer {
     // TODO: For every ACK, decide what should be done with the existing run and attempts. Make sure to check the current statuses first.
 
     switch (messageBody.data.type) {
+      // MARK: EXECUTE
       case "EXECUTE": {
         const existingTaskRun = await prisma.taskRun.findUnique({
           where: {
@@ -478,7 +479,7 @@ export class SharedQueueConsumer {
           ? lockedTaskRun.attempts[0].number + 1
           : 1;
 
-        const isRetry = nextAttemptNumber > 1;
+        const isRetry = lockedTaskRun.status === "WAITING_TO_RESUME" && nextAttemptNumber > 1;
 
         try {
           if (messageBody.data.checkpointEventId) {
@@ -493,6 +494,8 @@ export class SharedQueueConsumer {
               logger.error("Failed to restore checkpoint", {
                 queueMessage: message.data,
                 messageId: message.messageId,
+                runStatus: lockedTaskRun.status,
+                isRetry,
               });
 
               await this.#ackAndDoMoreWork(message.messageId);
@@ -503,8 +506,18 @@ export class SharedQueueConsumer {
           }
 
           if (!deployment.worker.supportsLazyAttempts) {
-            const service = new CreateTaskRunAttemptService();
-            await service.call(lockedTaskRun.friendlyId, undefined, false);
+            try {
+              const service = new CreateTaskRunAttemptService();
+              await service.call(lockedTaskRun.friendlyId, undefined, false);
+            } catch (error) {
+              logger.error("Failed to create task run attempt for outdate worker", {
+                error,
+                taskRun: lockedTaskRun.id,
+              });
+
+              await this.#ackAndDoMoreWork(message.messageId);
+              return;
+            }
           }
 
           if (isRetry) {
@@ -568,6 +581,7 @@ export class SharedQueueConsumer {
 
         break;
       }
+      // MARK: DEP RESUME
       // Resume after dependency completed with no remaining retries
       case "RESUME": {
         if (messageBody.data.checkpointEventId) {
@@ -728,6 +742,11 @@ export class SharedQueueConsumer {
         }
 
         try {
+          logger.debug("Broadcasting RESUME_AFTER_DEPENDENCY", {
+            runId: resumableAttempt.taskRunId,
+            attemptId: resumableAttempt.id,
+          });
+
           // The attempt should still be running so we can broadcast to all coordinators to resume immediately
           socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
             version: "v1",
@@ -752,6 +771,7 @@ export class SharedQueueConsumer {
 
         break;
       }
+      // MARK: DURATION RESUME
       // Resume after duration-based wait
       case "RESUME_AFTER_DURATION": {
         try {
@@ -785,6 +805,7 @@ export class SharedQueueConsumer {
 
         break;
       }
+      // MARK: FAIL
       // Fail for whatever reason, usually runs that have been resumed but stopped heartbeating
       case "FAIL": {
         const existingTaskRun = await prisma.taskRun.findUnique({
@@ -1122,6 +1143,11 @@ class SharedQueueTasks {
       },
       include: {
         lockedBy: true,
+        _count: {
+          select: {
+            attempts: true,
+          },
+        },
       },
     });
 
@@ -1143,6 +1169,7 @@ class SharedQueueTasks {
       runId: run.friendlyId,
       messageId: run.id,
       isTest: run.isTest,
+      attemptCount: run._count.attempts,
     } satisfies TaskRunExecutionLazyAttemptPayload;
   }
 

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

@@ -3,6 +3,7 @@ import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
+import { startActiveSpan } from "./tracer.server";
 
 export const r2 = singleton("r2", initializeR2);
 
@@ -23,30 +24,87 @@ export async function uploadToObjectStore(
   contentType: string,
   environment: AuthenticatedEnvironment
 ): Promise<string> {
-  if (!r2) {
-    throw new Error("Object store credentials are not set");
+  return await startActiveSpan("uploadToObjectStore()", async (span) => {
+    if (!r2) {
+      throw new Error("Object store credentials are not set");
+    }
+
+    if (!env.OBJECT_STORE_BASE_URL) {
+      throw new Error("Object store base URL is not set");
+    }
+
+    span.setAttributes({
+      projectRef: environment.project.externalRef,
+      environmentSlug: environment.slug,
+      filename: filename,
+    });
+
+    const url = new URL(env.OBJECT_STORE_BASE_URL);
+    url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`;
+
+    logger.debug("Uploading to object store", { url: url.href });
+
+    const response = await r2.fetch(url.toString(), {
+      method: "PUT",
+      headers: {
+        "Content-Type": contentType,
+      },
+      body: data,
+    });
+
+    if (!response.ok) {
+      throw new Error(`Failed to upload output to ${url}: ${response.statusText}`);
+    }
+
+    return url.href;
+  });
+}
+
+export async function generatePresignedRequest(
+  projectRef: string,
+  envSlug: string,
+  filename: string,
+  method: "PUT" | "GET" = "PUT"
+) {
+  if (!env.OBJECT_STORE_BASE_URL) {
+    return;
   }
 
-  if (!env.OBJECT_STORE_BASE_URL) {
-    throw new Error("Object store base URL is not set");
+  if (!r2) {
+    return;
   }
 
   const url = new URL(env.OBJECT_STORE_BASE_URL);
-  url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`;
+  url.pathname = `/packets/${projectRef}/${envSlug}/${filename}`;
+  url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes
 
-  logger.debug("Uploading to object store", { url: url.href });
+  const signed = await r2.sign(
+    new Request(url, {
+      method,
+    }),
+    {
+      aws: { signQuery: true },
+    }
+  );
 
-  const response = await r2.fetch(url.toString(), {
-    method: "PUT",
-    headers: {
-      "Content-Type": contentType,
-    },
-    body: data,
+  logger.debug("Generated presigned URL", {
+    url: signed.url,
+    headers: Object.fromEntries(signed.headers),
+    projectRef,
+    envSlug,
+    filename,
   });
 
-  if (!response.ok) {
-    throw new Error(`Failed to upload output to ${url}: ${response.statusText}`);
-  }
-
-  return url.href;
+  return signed;
+}
+
+export async function generatePresignedUrl(
+  projectRef: string,
+  envSlug: string,
+  filename: string,
+  method: "PUT" | "GET" = "PUT"
+) {
+  const signed = await generatePresignedRequest(projectRef, envSlug, filename, method);
+
+  return signed?.url;
 }

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

@@ -48,6 +48,7 @@ export class RequeueTaskRunService extends BaseService {
 
         break;
       }
+      case "DELAYED":
       case "WAITING_FOR_DEPLOY": {
         logger.debug("[RequeueTaskRunService] Removing task run from queue", { taskRun });
 
@@ -68,6 +69,7 @@ export class RequeueTaskRunService extends BaseService {
       case "CRASHED":
       case "COMPLETED_WITH_ERRORS":
       case "COMPLETED_SUCCESSFULLY":
+      case "EXPIRED":
       case "CANCELED": {
         logger.debug("[RequeueTaskRunService] Task run is completed", { taskRun });
 

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

@@ -344,6 +344,7 @@ export class CompleteAttemptService extends BaseService {
     supportsLazyAttempts?: boolean
   ) {
     if (checkpointEventId || !supportsLazyAttempts) {
+      // Workers without lazy attempt support always need to go through the queue, which is where the attempt is created
       // We have to replace a potential RESUME with EXECUTE to correctly retry the attempt
       return await marqs?.replaceMessage(
         run.id,
@@ -355,8 +356,9 @@ export class CompleteAttemptService extends BaseService {
         retryTimestamp
       );
     } else {
-      // There's no checkpoint so the worker is still running and waiting for a retry message
-      // It supports lazy attempts so we can bypass the queue and send the message directly to the worker
+      // There's no checkpoint and the worker supports lazy attempts
+      // This means the worker is still running and waiting for a retry message
+      // It supports lazy attempts so we can bypass the queue and send the message directly to it
       RetryAttemptService.enqueue(run.id, this._prisma, new Date(retryTimestamp));
     }
   }

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

@@ -13,6 +13,7 @@ export type CrashTaskRunServiceOptions = {
   logs?: string;
   crashAttempts?: boolean;
   crashedAt?: Date;
+  overrideCompletion?: boolean;
 };
 
 export class CrashTaskRunService extends BaseService {
@@ -36,7 +37,7 @@ export class CrashTaskRunService extends BaseService {
     }
 
     // Make sure the task run is in a crashable state
-    if (!isCrashableRunStatus(taskRun.status)) {
+    if (!opts.overrideCompletion && !isCrashableRunStatus(taskRun.status)) {
       logger.error("Task run is not in a crashable state", { runId, status: taskRun.status });
       return;
     }
@@ -74,9 +75,12 @@ export class CrashTaskRunService extends BaseService {
       },
     });
 
-    const inProgressEvents = await eventRepository.queryIncompleteEvents({
-      runId: taskRun.friendlyId,
-    });
+    const inProgressEvents = await eventRepository.queryIncompleteEvents(
+      {
+        runId: taskRun.friendlyId,
+      },
+      options?.overrideCompletion
+    );
 
     logger.debug("Crashing in-progress events", {
       inProgressEvents: inProgressEvents.map((event) => event.id),

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

@@ -7,6 +7,8 @@ import { BaseService, ServiceValidationError } from "./baseService.server";
 import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
 import { machinePresetFromConfig } from "../machinePresets.server";
 import { workerQueue } from "~/services/worker.server";
+import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
+import { CrashTaskRunService } from "./crashTaskRun.server";
 
 export class CreateTaskRunAttemptService extends BaseService {
   public async call(
@@ -93,6 +95,17 @@ export class CreateTaskRunAttemptService extends BaseService {
 
       const nextAttemptNumber = taskRun.attempts[0] ? taskRun.attempts[0].number + 1 : 1;
 
+      if (nextAttemptNumber > MAX_TASK_RUN_ATTEMPTS) {
+        const service = new CrashTaskRunService(this._prisma);
+        await service.call(taskRun.id, {
+          reason: taskRun.lockedBy.worker.supportsLazyAttempts
+            ? "Max attempts reached."
+            : "Max attempts reached. Please upgrade your CLI and SDK.",
+        });
+
+        throw new ServiceValidationError("Max attempts reached", 400);
+      }
+
       const taskRunAttempt = await $transaction(this._prisma, async (tx) => {
         const taskRunAttempt = await tx.taskRunAttempt.create({
           data: {

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

@@ -18,7 +18,8 @@ export class DeploymentIndexFailed extends BaseService {
       message: string;
       stack?: string;
       stderr?: string;
-    }
+    },
+    overrideCompletion = false
   ) {
     const isFriendlyId = maybeFriendlyId.startsWith("deployment_");
 
@@ -38,6 +39,15 @@ export class DeploymentIndexFailed extends BaseService {
     }
 
     if (FINAL_DEPLOYMENT_STATUSES.includes(deployment.status)) {
+      if (overrideCompletion) {
+        logger.error("No support for overriding final deployment statuses just yet", {
+          id: deployment.id,
+          status: deployment.status,
+          previousError: deployment.errorData,
+          incomingError: error,
+        });
+      }
+
       logger.error("Worker deployment already in final state", {
         id: deployment.id,
         status: deployment.status,

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

@@ -0,0 +1,72 @@
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { BaseService } from "./baseService.server";
+import { parseNaturalLanguageDuration } from "./triggerTask.server";
+import { workerQueue } from "~/services/worker.server";
+import { $transaction } from "~/db.server";
+
+export class EnqueueDelayedRunService extends BaseService {
+  public async call(runId: string) {
+    const run = await this._prisma.taskRun.findUnique({
+      where: {
+        id: runId,
+      },
+      include: {
+        runtimeEnvironment: {
+          include: {
+            organization: true,
+            project: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      logger.debug("Could not find delayed run to enqueue", {
+        runId,
+      });
+
+      return;
+    }
+
+    if (run.status !== "DELAYED") {
+      logger.debug("Delayed run cannot be enqueued because it's not in DELAYED status", {
+        run,
+      });
+
+      return;
+    }
+
+    await $transaction(this._prisma, async (tx) => {
+      await tx.taskRun.update({
+        where: {
+          id: run.id,
+        },
+        data: {
+          status: "PENDING",
+          queuedAt: new Date(),
+        },
+      });
+
+      if (run.ttl) {
+        const expireAt = parseNaturalLanguageDuration(run.ttl);
+
+        if (expireAt) {
+          await workerQueue.enqueue(
+            "v3.expireRun",
+            { runId: run.id },
+            { tx, runAt: expireAt, jobKey: `v3.expireRun.${run.id}` }
+          );
+        }
+      }
+    });
+
+    await marqs?.enqueueMessage(
+      run.runtimeEnvironment,
+      run.queue,
+      run.id,
+      { type: "EXECUTE", taskIdentifier: run.taskIdentifier },
+      run.concurrencyKey ?? undefined
+    );
+  }
+}

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

@@ -0,0 +1,53 @@
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { BaseService } from "./baseService.server";
+
+export class ExpireEnqueuedRunService extends BaseService {
+  public async call(runId: string) {
+    const run = await this._prisma.taskRun.findUnique({
+      where: {
+        id: runId,
+      },
+      include: {
+        runtimeEnvironment: {
+          include: {
+            organization: true,
+            project: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      logger.debug("Could not find enqueued run to expire", {
+        runId,
+      });
+
+      return;
+    }
+
+    if (run.status !== "PENDING") {
+      logger.debug("Run cannot be expired because it's not in PENDING status", {
+        run,
+      });
+
+      return;
+    }
+
+    logger.debug("Expiring enqueued run", {
+      run,
+    });
+
+    await this._prisma.taskRun.update({
+      where: {
+        id: run.id,
+      },
+      data: {
+        status: "EXPIRED",
+        expiredAt: new Date(),
+      },
+    });
+
+    await marqs?.acknowledgeMessage(run.id);
+  }
+}

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

@@ -0,0 +1,39 @@
+import { TaskRun } from "@trigger.dev/database";
+import { BaseService, ServiceValidationError } from "./baseService.server";
+import { RescheduleRunRequestBody } from "@trigger.dev/core/v3";
+import { parseDelay } from "./triggerTask.server";
+import { $transaction } from "~/db.server";
+import { workerQueue } from "~/services/worker.server";
+
+export class RescheduleTaskRunService extends BaseService {
+  public async call(taskRun: TaskRun, body: RescheduleRunRequestBody) {
+    if (taskRun.status !== "DELAYED") {
+      throw new ServiceValidationError("Cannot reschedule a run that is not delayed");
+    }
+
+    const delay = await parseDelay(body.delay);
+
+    if (!delay) {
+      throw new ServiceValidationError(`Invalid delay: ${body.delay}`);
+    }
+
+    return await $transaction(this._prisma, async (tx) => {
+      const updatedRun = await tx.taskRun.update({
+        where: {
+          id: taskRun.id,
+        },
+        data: {
+          delayUntil: delay,
+        },
+      });
+
+      await workerQueue.enqueue(
+        "v3.enqueueDelayedRun",
+        { runId: taskRun.id },
+        { tx, runAt: delay, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` }
+      );
+
+      return updatedRun;
+    });
+  }
+}

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

@@ -12,6 +12,7 @@ import { socketIo } from "../handleSocketIo.server";
 import { SharedQueueMessageBody, sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
 import { BaseService } from "./baseService.server";
 import { TaskRunAttempt } from "@trigger.dev/database";
+import { isFinalRunStatus } from "../taskStatus";
 
 export class ResumeAttemptService extends BaseService {
   public async call(
@@ -80,10 +81,11 @@ export class ResumeAttemptService extends BaseService {
         return;
       }
 
-      if (attempt.taskRun.status !== "WAITING_TO_RESUME") {
+      if (isFinalRunStatus(attempt.taskRun.status)) {
         logger.error("Run is not resumable", {
           attemptId: attempt.id,
           runId: attempt.taskRunId,
+          status: attempt.taskRun.status,
         });
         return;
       }

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

@@ -4,13 +4,15 @@ import {
   TriggerTaskRequestBody,
   packetRequiresOffloading,
 } from "@trigger.dev/core/v3";
-import { prisma } from "~/db.server";
+import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { autoIncrementCounter } from "~/services/autoIncrementCounter.server";
+import { workerQueue } from "~/services/worker.server";
 import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { eventRepository } from "../eventRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { uploadToObjectStore } from "../r2.server";
+import { startActiveSpan } from "../tracer.server";
 import { BaseService } from "./baseService.server";
 
 export type TriggerTaskServiceOptions = {
@@ -34,6 +36,12 @@ export class TriggerTaskService extends BaseService {
       span.setAttribute("taskId", taskId);
 
       const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey;
+      const delayUntil = await parseDelay(body.options?.delay);
+
+      const ttl =
+        typeof body.options?.ttl === "number"
+          ? stringifyDuration(body.options?.ttl)
+          : body.options?.ttl ?? (environment.type === "DEVELOPMENT" ? "10m" : undefined);
 
       const existingRun = idempotencyKey
         ? await this._prisma.taskRun.findUnique({
@@ -48,9 +56,19 @@ export class TriggerTaskService extends BaseService {
 
       if (existingRun && existingRun.taskIdentifier === taskId) {
         span.setAttribute("runId", existingRun.friendlyId);
+
         return existingRun;
       }
 
+      const runFriendlyId = generateFriendlyId("run");
+
+      const payloadPacket = await this.#handlePayloadPacket(
+        body.payload,
+        body.options?.payloadType ?? "application/json",
+        runFriendlyId,
+        environment
+      );
+
       return await eventRepository.traceEvent(
         taskId,
         {
@@ -75,15 +93,6 @@ export class TriggerTaskService extends BaseService {
           immediate: true,
         },
         async (event, traceContext) => {
-          const runFriendlyId = generateFriendlyId("run");
-
-          const payloadPacket = await this.#handlePayloadPacket(
-            body.payload,
-            body.options?.payloadType ?? "application/json",
-            runFriendlyId,
-            environment
-          );
-
           const run = await autoIncrementCounter.incrementInTransaction(
             `v3-run:${environment.id}:${taskId}`,
             async (num, tx) => {
@@ -111,7 +120,7 @@ export class TriggerTaskService extends BaseService {
 
               const taskRun = await tx.taskRun.create({
                 data: {
-                  status: "PENDING",
+                  status: delayUntil ? "DELAYED" : "PENDING",
                   number: num,
                   friendlyId: runFriendlyId,
                   runtimeEnvironmentId: environment.id,
@@ -128,6 +137,9 @@ export class TriggerTaskService extends BaseService {
                   concurrencyKey: body.options?.concurrencyKey,
                   queue: queueName,
                   isTest: body.options?.test ?? false,
+                  delayUntil,
+                  queuedAt: delayUntil ? undefined : new Date(),
+                  ttl,
                 },
               });
 
@@ -214,6 +226,26 @@ export class TriggerTaskService extends BaseService {
                 }
               }
 
+              if (taskRun.delayUntil) {
+                await workerQueue.enqueue(
+                  "v3.enqueueDelayedRun",
+                  { runId: taskRun.id },
+                  { tx, runAt: delayUntil, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` }
+                );
+              }
+
+              if (!taskRun.delayUntil && taskRun.ttl) {
+                const expireAt = parseNaturalLanguageDuration(taskRun.ttl);
+
+                if (expireAt) {
+                  await workerQueue.enqueue(
+                    "v3.expireRun",
+                    { runId: taskRun.id },
+                    { tx, runAt: expireAt, jobKey: `v3.expireRun.${taskRun.id}` }
+                  );
+                }
+              }
+
               return taskRun;
             },
             async (_, tx) => {
@@ -237,13 +269,15 @@ export class TriggerTaskService extends BaseService {
           }
 
           // We need to enqueue the task run into the appropriate queue. This is done after the tx completes to prevent a race condition where the task run hasn't been created yet by the time we dequeue.
-          await marqs?.enqueueMessage(
-            environment,
-            run.queue,
-            run.id,
-            { type: "EXECUTE", taskIdentifier: taskId },
-            body.options?.concurrencyKey
-          );
+          if (run.status === "PENDING") {
+            await marqs?.enqueueMessage(
+              environment,
+              run.queue,
+              run.id,
+              { type: "EXECUTE", taskIdentifier: taskId },
+              body.options?.concurrencyKey
+            );
+          }
 
           return run;
         }
@@ -257,26 +291,31 @@ export class TriggerTaskService extends BaseService {
     pathPrefix: string,
     environment: AuthenticatedEnvironment
   ) {
-    const packet = this.#createPayloadPacket(payload, payloadType);
+    return await startActiveSpan("handlePayloadPacket()", async (span) => {
+      const packet = this.#createPayloadPacket(payload, payloadType);
 
-    if (!packet.data) {
-      return packet;
-    }
+      if (!packet.data) {
+        return packet;
+      }
 
-    const { needsOffloading, size } = packetRequiresOffloading(packet);
+      const { needsOffloading, size } = packetRequiresOffloading(
+        packet,
+        env.TASK_PAYLOAD_OFFLOAD_THRESHOLD
+      );
 
-    if (!needsOffloading) {
-      return packet;
-    }
+      if (!needsOffloading) {
+        return packet;
+      }
 
-    const filename = `${pathPrefix}/payload.json`;
+      const filename = `${pathPrefix}/payload.json`;
 
-    await uploadToObjectStore(filename, packet.data, packet.dataType, environment);
+      await uploadToObjectStore(filename, packet.data, packet.dataType, environment);
 
-    return {
-      data: filename,
-      dataType: "application/store",
-    };
+      return {
+        data: filename,
+        dataType: "application/store",
+      };
+    });
   }
 
   #createPayloadPacket(payload: any, payloadType: string): IOPacket {
@@ -291,3 +330,104 @@ export class TriggerTaskService extends BaseService {
     return { dataType: payloadType };
   }
 }
+
+export async function parseDelay(value?: string | Date): Promise<Date | undefined> {
+  if (!value) {
+    return;
+  }
+
+  if (value instanceof Date) {
+    return value;
+  }
+
+  try {
+    const date = new Date(value);
+
+    // Check if the date is valid
+    if (isNaN(date.getTime())) {
+      return parseNaturalLanguageDuration(value);
+    }
+
+    if (date.getTime() <= Date.now()) {
+      return;
+    }
+
+    return date;
+  } catch (error) {
+    return parseNaturalLanguageDuration(value);
+  }
+}
+
+export function parseNaturalLanguageDuration(duration: string): Date | undefined {
+  const regexPattern = /^(\d+w)?(\d+d)?(\d+h)?(\d+m)?(\d+s)?$/;
+
+  const result: Date = new Date();
+  let hasMatch = false;
+
+  const elements = duration.match(regexPattern);
+  if (elements) {
+    if (elements[1]) {
+      const weeks = Number(elements[1].slice(0, -1));
+      if (weeks >= 0) {
+        result.setDate(result.getDate() + 7 * weeks);
+        hasMatch = true;
+      }
+    }
+    if (elements[2]) {
+      const days = Number(elements[2].slice(0, -1));
+      if (days >= 0) {
+        result.setDate(result.getDate() + days);
+        hasMatch = true;
+      }
+    }
+    if (elements[3]) {
+      const hours = Number(elements[3].slice(0, -1));
+      if (hours >= 0) {
+        result.setHours(result.getHours() + hours);
+        hasMatch = true;
+      }
+    }
+    if (elements[4]) {
+      const minutes = Number(elements[4].slice(0, -1));
+      if (minutes >= 0) {
+        result.setMinutes(result.getMinutes() + minutes);
+        hasMatch = true;
+      }
+    }
+    if (elements[5]) {
+      const seconds = Number(elements[5].slice(0, -1));
+      if (seconds >= 0) {
+        result.setSeconds(result.getSeconds() + seconds);
+        hasMatch = true;
+      }
+    }
+  }
+
+  if (hasMatch) {
+    return result;
+  }
+
+  return undefined;
+}
+
+function stringifyDuration(seconds: number): string | undefined {
+  if (seconds <= 0) {
+    return;
+  }
+
+  const units = {
+    w: Math.floor(seconds / 604800),
+    d: Math.floor((seconds % 604800) / 86400),
+    h: Math.floor((seconds % 86400) / 3600),
+    m: Math.floor((seconds % 3600) / 60),
+    s: Math.floor(seconds % 60),
+  };
+
+  // Filter the units having non-zero values and join them
+  const result: string = Object.entries(units)
+    .filter(([unit, val]) => val != 0)
+    .map(([unit, val]) => `${val}${unit}`)
+    .join("");
+
+  return result;
+}

apps/webapp/app/v3/taskStatus.ts

@@ -1,6 +1,7 @@
 import type { TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 
 export const CANCELLABLE_RUN_STATUSES: TaskRunStatus[] = [
+  "DELAYED",
   "PENDING",
   "WAITING_FOR_DEPLOY",
   "EXECUTING",
@@ -38,6 +39,7 @@ export const FINAL_RUN_STATUSES: TaskRunStatus[] = [
   "COMPLETED_WITH_ERRORS",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ];
 export const FINAL_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["CANCELED", "COMPLETED", "FAILED"];
 

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

@@ -117,7 +117,7 @@ function getTracer() {
   const samplingRate = 1.0 / Math.max(parseInt(env.INTERNAL_OTEL_TRACE_SAMPLING_RATE, 10), 1);
 
   const provider = new NodeTracerProvider({
-    forceFlushTimeoutMillis: 5000,
+    forceFlushTimeoutMillis: 15_000,
     resource: new Resource({
       [SEMRESATTRS_SERVICE_NAME]: env.SERVICE_NAME,
     }),
@@ -129,7 +129,7 @@ function getTracer() {
   if (env.INTERNAL_OTEL_TRACE_EXPORTER_URL) {
     const exporter = new OTLPTraceExporter({
       url: env.INTERNAL_OTEL_TRACE_EXPORTER_URL,
-      timeoutMillis: 10_000,
+      timeoutMillis: 15_000,
       headers:
         env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_NAME &&
         env.INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADER_VALUE

docs/mint.json

@@ -6,7 +6,7 @@
     "/v3-openapi.yaml"
   ],
   "versions": [
-    "v3 (Developer Preview)",
+    "v3",
     "v2"
   ],
   "api": {
@@ -22,15 +22,19 @@
   },
   "favicon": "/images/favicon.png",
   "colors": {
-    "primary": "#818cf8",
-    "light": "#818cf8",
-    "dark": "#818cf8",
+    "primary": "#A8FF53",
+    "light": "#A8FF53",
+    "dark": "#A8FF53",
     "anchors": {
       "from": "#fff",
       "to": "#eee"
+    },
+    "background": {
+      "light": "#fff",
+      "dark": "#121317"
     }
   },
-  "backgroundImage": "/images/docs-background.png",
+  "theme": "quill",
   "modeToggle": {
     "default": "dark",
     "isHidden": true
@@ -50,36 +54,6 @@
       "url": "https://discord.gg/kA47vcd8P6"
     }
   ],
-  "primaryTab": {
-    "name": "Home"
-  },
-  "tabs": [
-    {
-      "name": "Integrations",
-      "url": "integrations",
-      "version": "v2"
-    },
-    {
-      "name": "Reference",
-      "url": "sdk",
-      "version": "v2"
-    },
-    {
-      "name": "Examples",
-      "url": "https://trigger.dev/apis",
-      "version": "v2"
-    },
-    {
-      "name": "v3 Developer Preview",
-      "url": "https://trigger.dev/docs/v3",
-      "version": "v2"
-    },
-    {
-      "name": "v2",
-      "url": "https://trigger.dev/docs/documentation",
-      "version": "v3 (Developer Preview)"
-    }
-  ],
   "redirects": [
     {
       "source": "/documentation/quickstart",
@@ -101,14 +75,14 @@
   "navigation": [
     {
       "group": "",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/introduction"
       ]
     },
     {
       "group": "Getting Started",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/quick-start",
         "v3/upgrading-from-v2",
@@ -119,7 +93,7 @@
     },
     {
       "group": "Fundamentals",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/trigger-folder",
         "v3/tasks-overview",
@@ -137,7 +111,7 @@
     },
     {
       "group": "Development",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/cli-dev",
         "v3/run-tests"
@@ -145,7 +119,7 @@
     },
     {
       "group": "Deployment",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/deploy-environment-variables",
         "v3/cli-deploy",
@@ -160,7 +134,7 @@
     },
     {
       "group": "Writing tasks",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/writing-tasks-introduction",
         "v3/logging",
@@ -185,7 +159,7 @@
     },
     {
       "group": "API reference",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/management/overview",
         {
@@ -201,7 +175,8 @@
             "v3/management/runs/list",
             "v3/management/runs/retrieve",
             "v3/management/runs/replay",
-            "v3/management/runs/cancel"
+            "v3/management/runs/cancel",
+            "v3/management/runs/reschedule"
           ]
         },
         {
@@ -238,7 +213,7 @@
     },
     {
       "group": "Open source",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/github-repo",
         "v3/open-source-self-hosting",
@@ -247,7 +222,7 @@
     },
     {
       "group": "Troubleshooting",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/troubleshooting",
         "v3/troubleshooting-alerts",
@@ -258,7 +233,7 @@
     },
     {
       "group": "Help",
-      "version": "v3 (Developer Preview)",
+      "version": "v3",
       "pages": [
         "v3/community",
         "v3/help-slack",

docs/v3-openapi.yaml

@@ -269,6 +269,7 @@ paths:
 
   "/api/v1/timezones":
     get:
+      security: []
       operationId: get_timezones_v1
       summary: Get all supported timezones
       description: Get all supported timezones that schedule tasks support.
@@ -424,6 +425,71 @@ paths:
 
             await runs.cancel("run_1234");
 
+  "/api/v1/runs/{runId}/reschedule":
+    parameters:
+      - $ref: "#/components/parameters/runId"
+    post:
+      operationId: reschedule_run_v1
+      summary: Rescheduled a delayed run
+      description: Updates a delayed run with a new delay. Only valid when the run is in the DELAYED state.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              "$ref": "#/components/schemas/RescheduleRunRequestBody"
+      responses:
+        "200":
+          description: Successful request
+          content:
+            application/json:
+              schema:
+                "$ref": "#/components/schemas/RetrieveRunResponse"
+        "400":
+          description: Invalid request
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  error:
+                    type: string
+                    enum:
+                      - Invalid or missing run ID
+                      - Failed to create new run
+        "401":
+          description: Unauthorized request
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  error:
+                    type: string
+                    enum:
+                      - Invalid or Missing API key
+        "404":
+          description: Resource not found
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  error:
+                    type: string
+                    enum:
+                      - Run not found
+      tags:
+        - runs
+      security:
+        - secretKey: []
+      x-codeSamples:
+        - lang: typescript
+          source: |-
+            import { runs } from "@trigger.dev/sdk/v3";
+
+            const handle = await runs.reschedule("run_1234", { delay: new Date("2024-06-29T20:45:56.340Z") });
+
   "/api/v3/runs/{runId}":
     parameters:
       - $ref: "#/components/parameters/runId"
@@ -1681,6 +1747,18 @@ components:
           type: string
           example: slack_123456
       required: ["name", "value"]
+    RescheduleRunRequestBody:
+      type: object
+      properties:
+        delay:
+          oneOf:
+            - type: string
+              description: The duration to delay the run by. The duration should be in the format of `1d`, `6h`, `10m`, `11s`, etc.
+              example: 1hr
+            - type: string
+              format: date-time
+              description: The Date to delay the run until, e.g. `new Date()` or `"2024-06-25T15:45:26Z"`
+              example: 2024-06-25T15:45:26Z
     RetrieveRunResponse:
       type: object
       required:
@@ -1699,6 +1777,7 @@ components:
           type: string
           description: The status of the run
           enum:
+            - DELAYED
             - WAITING_FOR_DEPLOY
             - QUEUED
             - EXECUTING
@@ -1722,10 +1801,18 @@ components:
           type: object
           description: The payload that was sent to the task. Will be omitted if the request was made with a Public API key
           example: { "foo": "bar" }
+        payloadPresignedUrl:
+          type: string
+          description: The presigned URL to download the payload. Will only be included if the payload is too large to be included in the response. Expires in 5 minutes.
+          example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
         output:
           type: object
           description: The output of the run. Will be omitted if the request was made with a Public API key
           example: { "foo": "bar" }
+        outputPresignedUrl:
+          type: string
+          description: The presigned URL to download the output. Will only be included if the output is too large to be included in the response. Expires in 5 minutes.
+          example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
         idempotencyKey:
           type: string
           description: The idempotency key used to prevent creating duplicate runs, if provided
@@ -1748,6 +1835,10 @@ components:
           type: string
           format: date-time
           description: The time the run finished
+        delayedUntil:
+          type: string
+          format: date-time
+          description: If the run was triggered with a delay, this will be the time the run will be enqueued to execute
         schedule:
           type: object
           description: The schedule that triggered the run. Will be omitted if the run was not triggered by a schedule

docs/v3/introduction.mdx

@@ -15,7 +15,7 @@ Trigger.dev v3 makes it easy to write reliable long-running tasks without timeou
 - We run your tasks with no timeouts. You don't have to manage any infrastructure (unless you [self-host](/v3/open-source-self-hosting)). Workers are automatically scaled and managed for you.
 - We provide a multi-tenant queue that is used when triggering tasks.
 - We provide an SDK and CLI for writing tasks in your existing codebase, inside [/trigger folders](/v3/trigger-folder).
-- We provide different types of tasks: [regular](/v3/tasks-regular), [scheduled](/v3/tasks-scheduled), [zod](/v3/tasks-zod), [webhooks](/v3/tasks-webhooks).
+- We provide different types of tasks: [regular](/v3/tasks-regular) and [scheduled](/v3/tasks-scheduled).
 - We provide a dashboard for monitoring, debugging, and managing your tasks.
 
 We're [open source](https://github.com/triggerdotdev/trigger.dev) and you can choose to use the [Trigger.dev Cloud](https://cloud.trigger.dev) or [Self-host Trigger.dev](/v3/open-source-self-hosting) on your own infrastructure.

docs/v3/limits.mdx

@@ -37,3 +37,13 @@ If you add them dynamically using code make sure you add a `deduplicationKey` so
 If you're creating schedules for your user you will definitely need to request more schedules from us.
 
 <Snippet file="v3/soft-limit.mdx" />
+
+## Task payloads and outputs
+
+| Limit                  | Details                                        |
+| ---------------------- | ---------------------------------------------- |
+| Single trigger payload | Must not exceed 10MB                           |
+| Batch trigger payload  | The total of all payloads must not exceed 10MB |
+| Task outputs           | Must not exceed 10MB                           |
+
+Payloads and outputs that exceed 512KB will be offloaded to object storage and a presigned URL will be provided to download the data when calling `runs.retrieve`. You don't need to do anything to handle this in your tasks however, as we will transparently upload/download these during operation.

docs/v3/management/runs/reschedule.mdx

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

docs/v3/open-source-self-hosting.mdx

@@ -163,7 +163,7 @@ docker login -u <your_dockerhub_username>
 5. You can now deploy v3 projects using the CLI with these flags:
 
 ```bash
-trigger deploy --self-hosted --push
+npx trigger.dev@beta deploy --self-hosted --push
 ```
 
 ## Part 2: Split services

docs/v3/tasks-overview.mdx

@@ -4,7 +4,7 @@ sidebarTitle: "Tasks"
 description: "Tasks are functions that can run for a long time and provide strong resilience to failure."
 ---
 
-There are different types of tasks including [regular tasks](/v3/tasks-regular), [scheduled tasks](/v3/tasks-scheduled), [zod tasks](/v3/tasks-zod) and [webhook tasks](/v3/tasks-webhooks).
+There are different types of tasks including [regular tasks](/v3/tasks-regular) and [scheduled tasks](/v3/tasks-scheduled).
 
 ## Hello world task and how to trigger it
 

docs/v3/triggering.mdx

@@ -536,3 +536,177 @@ export async function create() {
 ```
 
 </CodeGroup>
+
+## Large Payloads
+
+We recommend keeping your task payloads as small as possible. We currently have a hard limit on task payloads above 10MB.
+
+If your payload size is larger than 512KB, instead of saving the payload to the database, we will upload it to an S3-compatible object store and store the URL in the database.
+
+When your task runs, we automatically download the payload from the object store and pass it to your task function. We also will return to you a `payloadPresignedUrl` from the `runs.retrieve` SDK function so you can download the payload if needed:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+const run = await runs.retrieve(handle);
+
+if (run.payloadPresignedUrl) {
+  const response = await fetch(run.payloadPresignedUrl);
+  const payload = await response.json();
+
+  console.log("Payload", payload);
+}
+```
+
+<Note>
+  We also use this same system for dealing with large task outputs, and subsequently will return a
+  corresponding `outputPresignedUrl`. Task outputs are limited to 100MB.
+</Note>
+
+If you need to pass larger payloads, you'll need to upload the payload to your own storage and pass a URL to the file in the payload instead. For example, uploading to S3 and then sending a presigned URL that expires in URL:
+
+<CodeGroup>
+
+```ts /yourServer.ts
+import { myTask } from "./trigger/myTasks";
+import { s3Client, getSignedUrl, PutObjectCommand, GetObjectCommand } from "./s3";
+import { createReadStream } from "node:fs";
+
+// Upload file to S3
+await s3Client.send(
+  new PutObjectCommand({
+    Bucket: "my-bucket",
+    Key: "myfile.json",
+    Body: createReadStream("large-payload.json"),
+  })
+);
+
+// Create presigned URL
+const presignedUrl = await getSignedUrl(
+  s3Client,
+  new GetObjectCommand({
+    Bucket: "my-bucket",
+    Key: "my-file.json",
+  }),
+  {
+    expiresIn: 3600, // expires in 1 hour
+  }
+);
+
+// Now send the URL to the task
+const handle = await myTask.trigger({
+  url: presignedUrl,
+});
+```
+
+```ts /trigger/myTasks.ts
+import { task } from "@trigger.dev/sdk/v3";
+
+export const myTask = task({
+  id: "my-task",
+  run: async (payload: { url: string }) => {
+    // Download the file from the URL
+    const response = await fetch(payload.url);
+    const data = await response.json();
+
+    // Do something with the data
+  },
+});
+```
+
+</CodeGroup>
+
+### Batch Triggering
+
+When using `batchTrigger` or `batchTriggerAndWait`, the total size of all payloads cannot exceed 10MB. This means if you are doing a batch of 100 runs, each payload should be less than 100KB.
+
+## Delayed runs
+
+When you want to trigger a task now, but have it run at a later time, you can use the `delay` option:
+
+```ts
+// Delay the task run by 1 hour
+await myTask.trigger({ some: "data" }, { delay: "1h" });
+// Delay the task run by 88 seconds
+await myTask.trigger({ some: "data" }, { delay: "88s" });
+// Delay the task run by 1 hour and 52 minutes and 18 seconds
+await myTask.trigger({ some: "data" }, { delay: "1h52m18s" });
+// Delay until a specific time
+await myTask.trigger({ some: "data" }, { delay: "2024-12-01T00:00:00" });
+// Delay using a Date object
+await myTask.trigger({ some: "data" }, { delay: new Date(Date.now() + 1000 * 60 * 60) });
+```
+
+Runs that are delayed and have not been enqueued yet will display in the dashboard with a "Delayed" status:
+
+![Delayed run in the dashboard](/images/v3/delayed-runs.png)
+
+<Note>
+  Delayed runs will be enqueued at the time specified, and will run as soon as possible after that
+  time, just as a normally triggered run would.
+</Note>
+
+You can cancel a delayed run using the `runs.cancel` SDK function:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+await runs.cancel("run_1234");
+```
+
+You can also reschedule a delayed run using the `runs.reschedule` SDK function:
+
+```ts
+import { runs } from "@trigger.dev/sdk/v3";
+
+// The delay option here takes the same format as the trigger delay option
+await runs.reschedule("run_1234", { delay: "1h" });
+```
+
+The `delay` option is also available when using `batchTrigger`:
+
+```ts
+await myTask.batchTrigger([{ payload: { some: "data" }, options: { delay: "1h" } }]);
+```
+
+## TTL
+
+You can set a TTL (time to live) when triggering a task, which will automatically expire the run if it hasn't started within the specified time. This is useful for ensuring that a run doesn't get stuck in the queue for too long.
+
+<Note>
+  All runs in development have a default `ttl` of 10 minutes. You can disable this by setting the
+  `ttl` option.
+</Note>
+
+```ts
+import { myTask } from "./trigger/myTasks";
+
+// Expire the run if it hasn't started within 1 hour
+await myTask.trigger({ some: "data" }, { ttl: "1h" });
+
+// If you specify a number, it will be treated as seconds
+await myTask.trigger({ some: "data" }, { ttl: 3600 }); // 1 hour
+```
+
+When a run is expired, it will be marked as "Expired" in the dashboard:
+
+![Expired runs in the dashboard](/images/v3/expired-runs.png)
+
+### Delayed runs and TTL
+
+When you use both `delay` and `ttl`, the TTL will start counting down from the time the run is enqueued, not from the time the run is triggered.
+
+So for example, when using the following code:
+
+```ts
+await myTask.trigger({ some: "data" }, { delay: "10m", ttl: "1h" });
+```
+
+The timeline would look like this:
+
+1. The run is created at 12:00:00
+2. The run is enqueued at 12:10:00
+3. The TTL starts counting down from 12:10:00
+4. If the run hasn't started by 13:10:00, it will be expired
+
+For this reason, the `ttl` option only accepts durations and not absolute timestamps.

docs/v3/troubleshooting.mdx

@@ -47,6 +47,27 @@ The first thing to do is to try again. If that fails then join [our Discord](htt
 
 Usually there will be some useful guidance below this message. If you can't figure out what's going wrong then join [our Discord](https://trigger.dev/discord) and create a Help forum post with a link to your deployment.
 
+## Project setup issues
+
+### `The requested module 'node:events' does not provide an export named 'addAbortListener'`
+If you see this error it means you're not a supported version of Node:
+
+```
+SyntaxError: The requested module 'node:events' does not provide an export named 'addAbortListener'
+at ModuleJob._instantiate (node:internal/modules/esm/module_job:123:21)
+at async ModuleJob.run (node:internal/modules/esm/module_job:189:5)
+
+Node.js v19.9.0
+```
+
+You need to be on at least these minor versions:
+
+| Version | Minimum |
+| ----- | ------- |
+| 18   | 18.16+  |
+| 20   | 20.11+  |
+| 21   | 21.0+  |
+
 ## Runtime issues
 
 ### `Environment variable not found:`

integrations/airtable/CHANGELOG.md

@@ -1,5 +1,35 @@
 # @trigger.dev/airtable
 
+## 3.0.0-beta.46
+
+### Patch Changes
+
+- @trigger.dev/integration-kit@3.0.0-beta.46
+- @trigger.dev/sdk@3.0.0-beta.46
+
+## 3.0.0-beta.45
+
+### Patch Changes
+
+- Updated dependencies [0e77e7ef7]
+  - @trigger.dev/sdk@3.0.0-beta.45
+  - @trigger.dev/integration-kit@3.0.0-beta.45
+
+## 3.0.0-beta.44
+
+### Patch Changes
+
+- @trigger.dev/integration-kit@3.0.0-beta.44
+- @trigger.dev/sdk@3.0.0-beta.44
+
+## 3.0.0-beta.43
+
+### Patch Changes
+
+- Updated dependencies [34ca7667d]
+  - @trigger.dev/sdk@3.0.0-beta.43
+  - @trigger.dev/integration-kit@3.0.0-beta.43
+
 ## 3.0.0-beta.42
 
 ### Patch Changes

integrations/airtable/package.json

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

integrations/github/CHANGELOG.md

@@ -1,5 +1,35 @@
 # @trigger.dev/github
 
+## 3.0.0-beta.46
+
+### Patch Changes
+
+- @trigger.dev/integration-kit@3.0.0-beta.46
+- @trigger.dev/sdk@3.0.0-beta.46
+
+## 3.0.0-beta.45
+
+### Patch Changes
+
+- Updated dependencies [0e77e7ef7]
+  - @trigger.dev/sdk@3.0.0-beta.45
+  - @trigger.dev/integration-kit@3.0.0-beta.45
+
+## 3.0.0-beta.44
+
+### Patch Changes
+
+- @trigger.dev/integration-kit@3.0.0-beta.44
+- @trigger.dev/sdk@3.0.0-beta.44
+
+## 3.0.0-beta.43
+
+### Patch Changes
+
+- Updated dependencies [34ca7667d]
+  - @trigger.dev/sdk@3.0.0-beta.43
+  - @trigger.dev/integration-kit@3.0.0-beta.43
+
 ## 3.0.0-beta.42
 
 ### Patch Changes