Daniel Sutton
0205feda39
Extracts every Postgres waitpoint and edge operation out of
`WaitpointSystem` into a `WaitpointCoordinator` seam with one Postgres
implementation, so a different coordination backend can be plugged in
later without any caller changing.
Pure refactor. Zero behaviour change, and zero test-file diffs — the
existing engine corpus is the characterisation test.
## What moved
`WaitpointCoordinator` (`waitpointCoordinator/types.ts`, declared with
`type`) has nine members: `clearRunBlockState`, `readRunBlockState`,
`registerBlocks`, `registerBlocksLockless`, `complete`,
`createDateTimeWaitpoint`, `createManualWaitpoint`,
`mintAssociatedWaitpointData`, `createAssociatedWaitpoint`.
`LegacyPostgresWaitpointCoordinator` implements them against the run-ops
store. Its dependencies are `{ runStore, prisma, logger }` only, so it
structurally cannot reach the run lock, the worker, or the event bus —
orchestration stays in `WaitpointSystem`, which keeps all ten public
signatures, all six `worker.enqueue` sites, the racepoints, the snapshot
transitions, and the event emissions.
Two register methods rather than one with a flag, so "the batch path
issues no extra query" is structural instead of conditional. Both share
one private edge-write helper.
## Six notes for reviewers — please read before "simplifying" any of
these
1. **`nanoid(24)` is called twice with different values on purpose**, in
each create path: once for the upsert `where` key, once for
`create.data`. Hoisting either to a shared constant makes the where-key
match the create-key, turning a guaranteed-miss upsert into a possible
update. In `createManualWaitpoint` both calls plus
`WaitpointId.generate()` stay *inside* the retry loop so each attempt
tries a fresh key.
2. **The two enqueue conditions are deliberately asymmetric.** DATETIME
enqueues `finishWaitpoint` unconditionally after a non-cached create,
with `availableAt: completedAfter`. MANUAL enqueues only when `timeout`
is set. That is existing behaviour, not an oversight. The coordinator
returns a discriminated union on `kind` rather than a boolean so the
enqueue is structurally unreachable on the cached path.
3. **One false clause was deleted from a moved comment.** The old
comment on the full-clear delete claimed the caller's `tx` is not
forwarded. The code does forward it, and `PostgresRunStore` uses `tx ??
this.prisma`, so a single store joins the caller's transaction — only
the routing store strips it. The rest of that comment is unchanged.
4. **The MANUAL timeout enqueue now sits outside the P2002 retry loop.**
Safe because the worker is Redis-backed and cannot raise
`Prisma.PrismaClientKnownRequestError`, so the loop never retried on it.
**If a Postgres-backed enqueue is ever swapped in, that equivalence
breaks silently.**
5. **The coordinator caches `runStore`/`prisma`/`logger` at
construction**, where the old code read `this.$.*` per call. Equivalent
only because nothing reassigns them: one assignment at
`engine/index.ts`, and the `resources` object is a `const` that is never
mutated.
6. **Two comments in other files are now stale and were left alone** —
`engine/index.ts` and `completeWaitpointCrossSeamGuard.test.ts` both
describe routing as the first statement of
`waitpointSystem.completeWaitpoint`. Both tests still pass, because that
guard sits in `index.ts` before the delegation. Left untouched to keep
this diff to three files.
## Preserved verbatim
The `unnest` edge CTE rather than a `Waitpoint` join; the pending count
as a separate statement after the edge write (READ COMMITTED needs its
own snapshot); completion's `findWaitpointOnPrimary` re-read through the
*resolved handle* while the blocked-run fan-out goes back through the
*router*; the residency and colocate hints, with colocation objects
built only in the Postgres arm and the count keeping its `runId`
argument; `ON CONFLICT DO NOTHING` and the `(taskRunId, waitpointId,
batchIndex)` multi-index edge semantics; the unread `batchId` select,
which rides inside two `logger.debug` payloads.
`internal-packages/run-store/` is untouched, so the CTE and the conflict
semantics never moved.
## Verification
| Check | Result |
| --- | --- |
| Engine corpus | 61/61 files, 353 passed, 1 skipped, **0 failed**
(baseline: 352 passed, 1 failed) |
| Test-file diffs | **empty** |
| `run-engine` typecheck | `tsc --noEmit -p tsconfig.build.json` exits 0
|
| `webapp` typecheck | 146 errors on this branch, **146 identical errors
at baseline** — pre-existing, none added |
The webapp typecheck does not pass. The failures are pre-existing
(`PrismaPg` not assignable to `never`; missing `@trigger.dev/rbac`
exports) and the sorted error lists are byte-identical to the merge
base, so this branch adds none — but the criterion is genuinely unmet
and needs a separate fix.
No changeset and no `.server-changes` note: internal refactor with no
user-visible change.
## Follow-ups this surfaced
- The dominant RUN waitpoint is still created outside the seam —
`buildRunAssociatedWaitpoint` now mints through the coordinator, but the
row is inserted nested inside `createRun`/`createFailedRun`. That needs
its own packet before a second backend lands, or the commonest waitpoint
gets split across two of them.
- `clearRunBlockState` overloads opposite outcomes on `undefined` versus
`[]`: `undefined` clears every edge, `[]` clears none. Both callers are
correct today; worth splitting when the file is next touched.
- A stray non-`.sql` entry in `internal-packages/clickhouse/schema/`
breaks every `containerTest` in the repo, because the testcontainers
migration reader `readFile`s every `readdir` entry without filtering
despite a comment claiming it filters. Hit this during setup; unrelated
to this change and left for a separate fix.
Trigger.dev Run Engine
The Run Engine process runs from triggering, to executing, retrying, and completing them.
It is responsible for:
- Creating, updating, and completing runs as they progress.
- Operating the run queue, including handling concurrency.
- Heartbeats which detects stalled runs and attempts to automatically recover them.
- Registering checkpoints which enable pausing/resuming of runs.
Glossary
- Platform: The main Trigger.dev API, dashboard, database. The Run Engine is part of the platform.
- Worker group: A group of workers that all pull from the same queue, e.g. "us-east-1", "my-self-hosted-workers".
- Worker: A worker is a 'server' that connects to the platform and receives runs.
- Supervisor: Pulls new runs from the queue, communicates with the platform, spins up new Deploy executors.
- Deploy container: Container that comes from a specific deploy from a user's project.
- Run controller: The code that manages running the task.
- Run executor: The actual task running.
- Worker: A worker is a 'server' that connects to the platform and receives runs.
Overview
╔═══════════════════════════════╗
║ ║░
║ Run triggered ║░
║ ║░
╚═══════════════════════════════╝░
___ ___ _ ░░░░░░░░░░░░░░░│░░░░░░░░░░░░░░░░░
| _ \_ _ _ _ | __|_ _ __ _(_)_ _ ___ │
╔══| / || | ' \ | _|| ' \/ _` | | ' \/ -_)═════════════════════════════╬══════════════════════════════════════╗
║ |_|_\\_,_|_||_| |___|_||_\__, |_|_||_\___| │ ║
║ |___/ │ ║
║ │ ║
║ ┌────────────────────────────────────── Has delay/debounce? ║
║ │ │ ║
║ Yes No ║
║ │ │ ║
║ ▼ ▼ ║
║ ╔═══════════════════════════════╗ ╔═══════════════════════════════╗ ║
║ ║ ║ Delay/ ║ ║ ║
║ ║ DELAYED ║◀────debounce────║ RUN_CREATED ║ ║
║ ║ ║ ║ ║ ║
║ ╚═══════════════════════════════╝ ╚═══════════════════════════════╝ ║
║ │ │ ║
║ │ │ ║
║ +===============================+ No delay/debounce ║
║ | | │ ║
║ | Redis Worker | │ ║
║ | | ▼ ║
║ +===============================+ ╔═══════════════════════════════╗ ║
║ │ ║ ║ ║
║ └───────────After delay──────────▶║ QUEUED ║◀────────────┐ ║
║ ║ ║ │ ║
║ ╚═══════════════════════════════╝ │ ║
║ ┌────All Waitpoints complete?─────┐ │ │ ║
║ │ │ │ │ ║
║ │ ▼ ▼ │ ║
║ ╔═══════════════════════════════╗ +===============================+ │ ║
║ ║ ║ | | Slow retry ║
║ ║ SUSPENDED ║ | Run Queue | │ ║
║ ║ ║ | | │ ║
║ ╚═══════════════════════════════╝ +===============================+ │ ║
Run not executing ║ ▲ │ ║
║ │ │ │ ║
═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ╬ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ╬ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ╬ ═ ═ ═ ═║═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═ ═
║ │ │ │ ║
Run maybe executing ║ │ │ ║ ╔═══════════════════════════════╗
║ │ │ │ ║ ║ ║░
║ │ Pulled from the queue ◀─────────────────┼────────╬───◈║ Dequeue a run ║░
║ │ │ │ ║ ║ ║░
║ │ ▼ │ ║ ╚═══════════════════════════════╝░
║ │ ╔═══════════════════════════════╗ │ ║ ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
║ │ ║ ║ │ ║
║ │ ║ PENDING_EXECUTING ║ │ ║
╔═══════════════════════════════╗ ║ │ ║ ║ │ ║
║ ║░ ║ │ ╚═══════════════════════════════╝ │ ║
║ Checkpoint created ║◈───╬───────────────────────┤ │ │ ║ ╔═══════════════════════════════╗
║ ║░ ║ │ ║ ║ ║░
╚═══════════════════════════════╝░ ║ │ ├─────────────────────────────┼────────╬───◈║ Start attempt ║░
░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ ║ │ │ ║ ║ ║░
║ │ ▼ │ ║ ╚═══════════════════════════════╝░
║ All Is executing on worker │ ║ ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
║ │ ┌────Waitpoints───┐ │ ┌─Quick retry │ ║
║ │ complete? │ │ │ │ │ ║
║ │ │ ▼ ▼ ▼ │ │ ║
║ ╔═══════════════════════════════╗ ╔═══════════════════════════════╗ │ │ ║
║ ║ ║ Hits a ║ ║ │ │ ║
║ ║ EXECUTING_WITH_WAITPOINTS ║◀───Waitpoint────║ EXECUTING ║ │ │ ║
║ ║ ║ ║ ║ │ │ ║
║ ╚═══════════════════════════════╝ ╚═══════════════════════════════╝ │ │ ║
║ │ │ │ ║ ╔═══════════════════════════════╗
║ │ │ ║ ║ ║░
║ ◀──────────────────────┼──────┼────────╬───◈║ Complete attempt ║░
║ │ │ │ ║ ║ ║░
║ │ │ │ ║ ╚═══════════════════════════════╝░
║ │ │ │ ║ ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
║ ├───────────────▶ Attempt failed ║
║ │ │ ║
║ Attempt success │ ║
║ │ All retries ║
║ ▼ used ║
╔═══════════════════════════════╗ ║ ╔═══════════════════════════════╗ │ ║
║ ║░ ║ ║ ║ │ ║
║ User cancels a run ║────╬──────────────▶ Is executing? ─────────── No ─────────▶║ FINISHED ║◀───────┘ ║
║ ║░ ║ │ ║ ║ ║
╚═══════════════════════════════╝░ ║ Yes ╚═══════════════════════════════╝ ║
░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ ║ │ ▲ ║
║ ▼ │ ║
║ ╔═══════════════════════════════╗ │ ║
║ ║ ║ │ ║
║ ║ PENDING_CANCEL ║─────────────────────────────────┘ ║
║ ║ ║ ║
║ ╚═══════════════════════════════╝ ║
║ ║
║ ║
║ ║
╚════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
Run locking
Many operations on the run are "atomic" in the sense that only a single operation can mutate them at a time. We use RedLock to create a distributed lock to ensure this. Postgres locking is not enough on its own because we have multiple API instances and Redis is used for the queue.
There are race conditions we need to deal with:
- When checkpointing the run continues to execute until the checkpoint has been stored. At the same time the run continues and the checkpoint can become irrelevant if the waitpoint is completed. Both can happen at the same time, so we must lock the run and protect against outdated checkpoints.
Run execution
The execution state of a run is stored in the TaskRunExecutionSnapshot table in Postgres. This is separate from the TaskRun status which is exposed to users via the dashboard and API.
The TaskRunExecutionSnapshot executionStatus is used to determine the execution status and is internal to the run engine. It is a log of events that impact run execution – the data is used to execute the run.
A common pattern we use is to read the current state and check that the passed in snapshotId matches the current snapshotId. If it doesn't, we know that the state has moved on. In the case of a checkpoint coming in, we know we can just ignore it.
We can also store invalid states by setting an error. These invalid states are purely used for debugging and are ignored for execution purposes.
Workers
A worker is a server that runs tasks.
In the dashboard under the "Regions" page, you can see all worker groups. You can set the default region there.
Then when triggering runs, you can override the region to use. The region is used internally to set the masterQueue that a run is placed in, this allows pulling runs only for that worker group.
Pulling from the queue
A worker will call the Trigger.dev API with it's region. For dev environments, we will pass the environment id.
Run Queue
This is a fair multi-tenant queue. It is designed to fairly select runs, respect concurrency limits, and have high throughput. It provides visibility into the current concurrency for the env, org, etc.
It has built-in reliability features:
- When nacking we increment the
attemptand if it continually fails we will move it to a Dead Letter Queue (DLQ). - If a run is in the DLQ you can redrive it.
Heartbeats
Heartbeats are used to determine if a run has become stalled. Depending on the current execution status, we do different things. For example, if the run has been dequeued but the attempt hasn't been started we requeue it.
Checkpoints
Checkpoints allow pausing an executing run and then resuming it later. This is an optimization to avoid wasted compute and is especially useful with "Waitpoints".
Waitpoints
A "Waitpoint" is something that can block a run from continuing:
A single Waitpoint can block many runs, the same waitpoint can only block a run once (there's a unique constraint). They block run execution from continuing until all of them are completed.
They can have output data associated with them, e.g. the finished run payload. That includes an error, e.g. a failed run.
There are currently three types:
RUNwhich gets completed when the associated run completes. Every run has anassociatedWaitpointthat matches the lifetime of the run.DATETIMEwhich gets completed when the datetime is reached.MANUALwhich gets completed when that event occurs.
Waitpoints can have an idempotencyKey which allows stops them from being created multiple times. This is especially useful for event waitpoints, where you don't want to create a new waitpoint for the same event twice.
wait.for() or wait.until()
Wait for a future time, then continue. We should add the option to pass an idempotencyKey so a second attempt doesn't wait again. By default it would wait again.
//Note if the idempotency key is a string, it will get prefixed with the run id.
//you can explicitly pass in an idempotency key created with the the global scope.
await wait.until(new Date("2022-01-01T00:00:00Z"), { idempotencyKey: "first-wait" });
await wait.until(new Date("2022-01-01T00:00:00Z"), { idempotencyKey: "second-wait" });
triggerAndWait() or batchTriggerAndWait()
Trigger and then wait for run(s) to finish. If the run fails it will still continue but with the errors so the developer can decide what to do.
The trigger delay option
When triggering a run and passing the delay option, we use a DATETIME waitpoint to block the run from starting.
wait.forRequest()
Wait until a request has been received at the URL that you are given. This is useful for pausing a run and then continuing it again when some external event occurs on another service. For example, Replicate have an API where they will callback when their work is complete.
wait.forToken(waitpointId)
A more advanced SDK which would require uses to explicitly create a waitpoint. We would also need createWaitpoint(), completeWaitpoint(), and failWaitpoint().
// Your backend
import { wait } from "@trigger.dev/sdk";
type ApprovalToken = {
status: "approved" | "rejected";
};
const waitpoint = await wait.createToken({ idempotencyKey: `purchase-${payload.cart.id}` });
const waitpoint = await wait.retrieveToken(waitpoint.id);
await wait.completeToken<ApprovalToken>(tokenId, {
status: "approved",
});
// /trigger/approval.ts
export const approvalFlow = task({
id: "approvalFlow",
run: async (payload) => {
//...do stuff
// This must be called inside a task run function
const result = await wait.forToken<ApprovalToken>(payload.tokenId);
if (result.ok) {
console.log("Token completed", result.output.status); // "approved" or "rejected"
} else {
console.log("Token timed out", result.error);
}
if (!result.ok) {
//...timeout
}
//...do more stuff
},
});
Run flow control
There are several ways to control when a run will execute (or not). Each of these should be configurable on a task, a named queue that is shared between tasks, and at trigger time including the ability to pass a key so you can have per-tenant controls.
Concurrency limits
When trigger is called the run is added to the queue. We only dequeue when the concurrency limit hasn't been exceeded for that task/queue.
Debouncing
When trigger is called, we prevent too many runs happening in a period by collapsing into a single run. This is done by discarding some runs in a period.
This is useful:
- To prevent too many runs happening in a short period.
We should mark the run as "DELAYED" with the correct delayUntil time. This will allow the user to see that the run is delayed and why.
Emitting events
The Run Engine emits events using its eventBus. This is used for runs completing, failing, or things that any workers should be aware of.
RunEngine System Architecture
The RunEngine is composed of several specialized systems that handle different aspects of task execution and management. Below is a diagram showing the relationships between these systems.
graph TD
RE[RunEngine]
DS[DequeueSystem]
RAS[RunAttemptSystem]
ESS[ExecutionSnapshotSystem]
WS[WaitpointSystem]
BS[BatchSystem]
ES[EnqueueSystem]
CS[CheckpointSystem]
DRS[DelayedRunSystem]
TS[TtlSystem]
WFS[WaitingForWorkerSystem]
%% Core Dependencies
RE --> DS
RE --> RAS
RE --> ESS
RE --> WS
RE --> BS
RE --> ES
RE --> CS
RE --> DRS
RE --> TS
RE --> WFS
%% System Dependencies
DS --> ESS
DS --> RAS
RAS --> ESS
RAS --> WS
RAS --> BS
WS --> ESS
WS --> ES
ES --> ESS
CS --> ESS
CS --> ES
DRS --> ES
WFS --> ES
TS --> WS
%% Shared Resources
subgraph Resources
PRI[(Prisma)]
LOG[Logger]
TRC[Tracer]
RQ[RunQueue]
RL[RunLocker]
EB[EventBus]
WRK[Worker]
RCQ[ReleaseConcurrencyQueue]
end
%% Resource Dependencies
RE -.-> Resources
DS & RAS & ESS & WS & BS & ES & CS & DRS & TS & WFS -.-> Resources
System Responsibilities
DequeueSystem
- Handles dequeuing of tasks from master queues
- Manages resource allocation and constraints
- Handles task deployment verification
RunAttemptSystem
- Manages run attempt lifecycle
- Handles success/failure scenarios
- Manages retries and cancellations
- Coordinates with other systems for run completion
ExecutionSnapshotSystem
- Creates and manages execution snapshots
- Tracks run state and progress
- Manages heartbeats for active runs
- Maintains execution history
WaitpointSystem
- Manages waitpoints for task synchronization
- Handles waitpoint completion
- Coordinates blocked runs
- Manages concurrency release
BatchSystem
- Manages batch operations
- Handles batch completion
- Coordinates batch-related task runs
EnqueueSystem
- Handles enqueueing of runs
- Manages run scheduling
- Coordinates with execution snapshots
Shared Resources
- Prisma: Database access
- Logger: Logging functionality
- Tracer: Tracing and monitoring
- RunQueue: Task queue management
- RunLocker: Run locking mechanism
- EventBus: Event communication
- Worker: Background task execution
- ReleaseConcurrencyQueue: Manages concurrency token release
Key Interactions
- RunEngine orchestrates all systems and manages shared resources
- DequeueSystem works with RunAttemptSystem for task execution
- RunAttemptSystem coordinates with WaitpointSystem and BatchSystem
- WaitpointSystem uses EnqueueSystem for run scheduling
- ExecutionSnapshotSystem is used by all other systems to track state
- All systems share common resources through the
SystemResourcesinterface