> ## Documentation Index
> Fetch the complete documentation index at: https://jephalabs.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Worker Orchestration

> Why QTG uses Postgres as its scheduler instead of an external orchestrator like Airflow

# Worker Orchestration: Postgres-as-Scheduler

## The design question

QTG executes multi-step movements with **per-step dependencies**, **at-least-once retries**, and **long-running observation windows** (a CCTP burn might wait 20 minutes for Circle's attestation). Tools like Apache Airflow exist for exactly this shape of problem.

So why doesn't QTG use Airflow?

Because the **database already knows everything an orchestrator would need to know** — what ran, what's pending, what's stuck, what depends on what. Layering a separate scheduler on top means **two systems must agree on truth** instead of one.

QTG's design choice: **the database IS the scheduler.** Workers poll it; nothing pushes work at them. There is no central dispatch process, no broker, no message queue, no DAG engine.

```mermaid theme={null}
flowchart LR
    DB[("PostgreSQL<br/>State of every<br/>movement + node")]
    W1["dispatcher<br/>(picks READY nodes)"]
    W2["observer<br/>(polls SUBMITTED/OBSERVING)"]
    W3["recovery<br/>(picks UNKNOWN)"]
    W4["callback<br/>(picks unsent outbox)"]

    W1 -->|"SELECT...FOR UPDATE<br/>SKIP LOCKED"| DB
    W2 -->|"unlocked ID list →<br/>per-ID FOR UPDATE SKIP LOCKED claim"| DB
    W3 -->|"SELECT...FOR UPDATE<br/>SKIP LOCKED"| DB
    W4 -->|"SELECT...FOR UPDATE<br/>SKIP LOCKED"| DB

    DB -.->|state mutation| W1
    DB -.->|state mutation| W2
    DB -.->|state mutation| W3
    DB -.->|state mutation| W4

    style DB fill:#1f6feb,color:#fff
```

Every worker is a **loop**: ask the DB "is there anything to do?", do one batch of work, commit the new state, sleep, repeat.

## Comparison with Apache Airflow

Airflow and QTG solve overlapping problems, but the architecture is inverted:

|                                | Airflow                                            | QTG                                                                                                           |
| ------------------------------ | -------------------------------------------------- | ------------------------------------------------------------------------------------------------------------- |
| **Where does the DAG live?**   | Python files parsed at scheduler tick              | Compiled `movement_request_nodes` rows + `movement_plan_edges` rows                                           |
| **What triggers a task?**      | Scheduler scans DAG, pushes to executor            | Worker queries DB for nodes in `READY` state                                                                  |
| **What carries state?**        | Metastore (Postgres) + scheduler in-memory queue   | Postgres alone — workers are stateless                                                                        |
| **How are retries scheduled?** | Scheduler tracks `next_retry_dt` per task instance | Node row carries `next_observe_at` / `retry_after_seconds`; worker filter is `WHERE now() >= next_observe_at` |
| **Concurrency control**        | Pool slots in scheduler + worker pool sizes        | Worker-specific row claims; safety depends on each claim's complete transaction lifetime                      |
| **Fault tolerance**            | Scheduler HA, executor heartbeats                  | Multiple worker replicas + DB row locks                                                                       |
| **Add a new task type**        | New Operator class + Operator registry             | New executor key + node\_config schema                                                                        |

The trade-off is straightforward: Airflow gives you a UI, calendar scheduling, and an enormous ecosystem of operators. QTG gives up that breadth in exchange for **one fewer system in the trust path of a USDC transfer**. A movement either advances by changing rows in Postgres, or it doesn't advance — there's no third system that can lie about it.

<Info>
  This isn't a claim that Airflow is the wrong tool in general. For a daily ETL pipeline, Airflow is excellent. The choice here is shaped by QTG's specific blast-radius: an undetected scheduler/DB disagreement would mean the system thinks a transfer happened when it didn't, or vice versa. Reducing to one source of truth is worth the lost ergonomics.
</Info>

## The worker loop pattern

Every QTG worker shares the same loop skeleton. One tick of that loop for the process-selected network does this:

1. Open a per-network DB session.
2. Write the worker's heartbeat row and commit it.
3. Run the worker's batch of work, returning how many rows it processed.
4. On success, reset the consecutive-error counter; on failure, back off exponentially.
5. Sleep for a jittered interval, then repeat.

Four things to notice:

1. **Heartbeat first, work second.** Even if a worker finds nothing to do, the heartbeat row is updated — operators can tell the worker is alive.
2. **Process-selected network.** Each loop defaults to `settings.network_mode` and opens a session bound to that network's schema. Mainnet and testnet deployments share **no worker session** — a testnet worker cannot address mainnet rows through its selected session.
3. **Bounded backoff on failure.** Consecutive errors trigger exponential backoff to a 60-second ceiling. The loop never wedges or fast-loops on a persistent failure.
4. **Jittered sleep.** `interval ± jitter` prevents synchronized waves of workers hammering the DB at the same instant.

## Two worker tiers: core and sidecar

Workers fall into two tiers, distinguished by what happens when they crash:

* **Core workers** — `node_dispatcher`, `node_observer`, `node_recovery`.
* **Sidecar workers** — `callback_dispatcher`, `expiration_checker`, `executor_health`.

```mermaid theme={null}
flowchart TB
    subgraph CORE["Core workers (fail-fast)"]
        D["node_dispatcher<br/>READY → SUBMITTED"]
        O["node_observer<br/>SUBMITTED/OBSERVING → terminal"]
        R["node_recovery<br/>UNKNOWN → terminal"]
    end
    subgraph SIDE["Sidecar workers (isolated)"]
        C["callback_dispatcher"]
        E["expiration_checker"]
        H["executor_health"]
    end

    CRASH1["Core crash<br/>↓<br/>tear down ALL workers<br/>(supervisor exits)"]
    CRASH2["Sidecar crash<br/>↓<br/>log + others stay alive"]

    D -.crash.-> CRASH1
    O -.crash.-> CRASH1
    R -.crash.-> CRASH1
    C -.crash.-> CRASH2
    E -.crash.-> CRASH2
    H -.crash.-> CRASH2

    style CORE fill:#ffe0b2
    style SIDE fill:#e3f2fd
    style CRASH1 fill:#ffcdd2
    style CRASH2 fill:#fff9c4
```

The rationale:

* A movement cannot make forward progress if `node_dispatcher` is dead. Letting the system run "mostly working" hides the failure. **Fail-fast.**
* A failed `callback_dispatcher` only stops external notifications — the movement itself still advances, and the operator can address the callback failure separately. **Isolated.**

Both tiers share a single `asyncio.TaskGroup` so external cancellation (SIGTERM) cleanly shuts down everything.

## The six primary workers

| Worker                | Polls for                                                            | Transitions                                      | Loop owner |
| --------------------- | -------------------------------------------------------------------- | ------------------------------------------------ | ---------- |
| `node_dispatcher`     | nodes in `READY` whose parents are all terminal                      | `READY → SUBMITTING → SUBMITTED`                 | core       |
| `node_observer`       | nodes in `SUBMITTED`/`OBSERVING` with `next_observe_at ≤ now()`      | `SUBMITTED/OBSERVING → COMPLETED/FAILED/UNKNOWN` | core       |
| `node_recovery`       | nodes in `UNKNOWN` or stale `SUBMITTING`                             | `UNKNOWN → COMPLETED/FAILED` via `recover()`     | core       |
| `callback_dispatcher` | rows in `callback_outbox` with `next_attempt_at ≤ now()`             | enqueued → sent/failed                           | sidecar    |
| `expiration_checker`  | `movement_requests` with `expires_at ≤ now()` and non-terminal state | → `EXPIRED`                                      | sidecar    |
| `executor_health`     | every registered executor                                            | refreshes `executor_health` table                | sidecar    |

Auxiliary loops include `evm_nonce_reaper`, `ccip_stranded_reaper`, `audit_outbox_promoter`, `capital_transfer_followup`, `template_proposal_expiration`, `verify_drift_worker`, and `balance_snapshot` — see [Runtime Workers reference](/reference/workers/runtime-workers) for the full per-worker breakdown.

## The core mechanism: SELECT ... FOR UPDATE SKIP LOCKED

The key Postgres primitive is **`SELECT ... FOR UPDATE SKIP LOCKED`** — row-level
skip-locked claiming. It is safe only when a worker holds the right claim rows
for its complete mutation lifetime; a one-time batch lock that is released by an
earlier per-row commit does not protect later in-memory candidates.

What it does:

1. The query takes a row-level lock on every returned row.
2. Any row that's **already locked by another transaction is silently skipped** — the query never blocks on it.
3. The lock is released on transaction commit/rollback.

This gives workers whose complete claim contracts are correct **safe concurrent
replicas** with no scheduler or coordinator:

* Two `node_dispatcher` instances querying the same table see disjoint sets of rows.
* A crashed worker's lock dies with its transaction — the row becomes pickable by the next worker immediately.
* The DB does all the work; the workers are stateless.

Without `SKIP LOCKED`, you'd need either a scheduler to partition work or an external lock service. With it, **horizontal scaling is "run more workers"** and nothing else.

<Tip>
  `SKIP LOCKED` is not a blanket horizontal-safety guarantee. Each state-changing
  worker must prove its own lock targets and claim lifetime. The observer locks a
  node and its request through observation and commit; recovery has a separate,
  reconcile-only contract. The observer-to-recovery ownership follow-up is tracked
  separately, not redesigned here.
</Tip>

## Pacing: next\_observe\_at and the polling rate problem

Long-running bridges (CCIP, Stargate, CCTP) finish in **minutes to tens of minutes**. Polling every second is wasteful; polling every minute leaves money on the table when a fast path completes in 90 seconds.

QTG sidesteps fixed intervals by writing a **per-node `next_observe_at`** column.
The observer uses two stages: an advisory unlocked ID list, followed by a locked
per-ID claim that repeats every eligibility predicate.

```sql theme={null}
-- advisory, unlocked ID list
SELECT movement_request_nodes.id
FROM movement_request_nodes
JOIN movement_requests ON movement_requests.id = movement_request_nodes.request_id
WHERE node_state IN ('SUBMITTED', 'OBSERVING')
  AND movement_requests.request_state = 'EXECUTING'
  AND (next_observe_at IS NULL OR next_observe_at <= NOW())
ORDER BY movement_requests.created_at
LIMIT :max_nodes;

-- per-ID locked claim repeats the predicates
SELECT ...
FROM movement_request_nodes
JOIN movement_requests ON movement_requests.id = movement_request_nodes.request_id
WHERE movement_request_nodes.id = :node_id
  AND node_state IN ('SUBMITTED', 'OBSERVING')
  AND movement_requests.request_state = 'EXECUTING'
  AND (next_observe_at IS NULL OR next_observe_at <= NOW())
FOR UPDATE OF movement_request_nodes, movement_requests SKIP LOCKED;
```

When the worker is scoped to one movement, both stages also add
`movement_requests.id = :movement_id`; an advisory ID never bypasses that
scope at claim time.

The joined per-ID claim runs inside a savepoint. PostgreSQL can lock the node
tuple before `SKIP LOCKED` rejects its locked request tuple; a claim miss rolls
back only that savepoint so the partial tuple lock does not survive into the
next candidate. A successful claim releases the savepoint while its node and
request locks remain held by the outer per-node transaction through provider
work and commit. This is lock hygiene for a multi-table claim, not a lease or a
separate claim transaction.

Each observe round computes the **next** poll time based on:

* Protocol-specific baselines (e.g. CCIP lane latency, Stargate RPC latency).
* An adaptive pacing phase — one of `pre_baseline`, `at_baseline`, `over_baseline`, or `fatal` — derived from how long the node has been observing relative to its baseline.
* The executor's `retry_after_seconds` hint when it returns `UNKNOWN`.

This pushes pacing decisions **into the data** instead of the worker. A fast lane polls every few seconds; a slow lane polls every few minutes; the worker code doesn't care. To change the pacing of a lane, you change a config (or a registry row), not a worker.

## At-least-once + idempotency: how recovery is safe

The dispatcher and the observer use the same correctness model that durable queue systems use:

1. **Mark intent in DB before side effect.** The dispatcher writes a per-lane intent row **before** signing or broadcasting the transaction.
2. **Side effect.** Submit the transaction.
3. **Mark completion in DB after side effect.** Update the intent row with `send_tx_hash`, etc.

If the worker crashes between (2) and (3), the next worker sees an intent in `SUBMITTING` with no `send_tx_hash`. The recovery worker's job is to **figure out which side of (2) the crash happened on** by reading the chain — scanning for the orphan transaction that step (2) may or may not have broadcast.

Two pieces of node state codify this:

* `NodeState.UNKNOWN` — the observer couldn't determine the outcome, so the node must go through recovery rather than a blind retry.
* `has_side_effect` — `True` for any node that changed external state.

When a node is `UNKNOWN` **and** `has_side_effect=True`, the dispatcher refuses to retry it blindly. Only the recovery path can move it forward — and recovery is implemented per-executor so it knows what evidence to look for (an attestation, an `OFTReceived` log, a Bybit withdrawal status).

This is the same safety contract that distinguishes "exactly-once" databases from "at-least-once" queues: **the system retries, but the retry is idempotent because the executor knows how to recognize past success.**

See [State Machine](/concepts/state-machine#why-unknown-state-is-special) for the full UNKNOWN-handling rules.

## Heartbeats and dead-worker detection

Every worker upserts a `worker_heartbeats` row on every tick, atomically via Postgres's `INSERT ... ON CONFLICT DO UPDATE` keyed on `worker_name`. Each upsert records `last_tick_at`, `expected_interval_seconds`, and a `live` status.

The `expected_interval_seconds` column is the contract. Operators (or another monitoring worker) can query:

```sql theme={null}
SELECT worker_name, last_tick_at, expected_interval_seconds
FROM worker_heartbeats
WHERE NOW() - last_tick_at > expected_interval_seconds * 2
```

— any row returned is a stalled worker. The heartbeat write is best-effort: if it fails, the worker logs a warning but keeps running. **A failing heartbeat must never break the worker.**

## Network class isolation

A subtle but important detail: every worker tick defaults to the process-selected `NetworkClass` and opens a session bound to that network's schema.

This isolates each network's schema (`mainnet` vs `testnet` Postgres schemas, see [Data Model](/reference/infrastructure/data-model)). A testnet drift worker cannot reach mainnet tables; a mainnet observer cannot reach testnet tables. The isolation is enforced by `schema_translate_map` plus the session's network-class binding, while process scope comes from `settings.network_mode`.

Loops such as `evm_nonce_reaper` also default to one task for the process-selected network. Running both network classes requires separate process deployments or an explicit administrative/test target list; each task then keeps its own loop and heartbeat row.

## What's NOT in the system

A useful way to internalize the design is to enumerate what QTG **does not** have:

* **No message broker** (Kafka, RabbitMQ, SQS) — workers poll Postgres.
* **No scheduler process** (Airflow, Temporal, Argo) — workers run their own loops.
* **No worker registry service** — the `worker_heartbeats` table is the registry.
* **No retry queue** — `next_observe_at` is the retry queue, sorted by Postgres on every query.
* **No DAG engine** — the DAG is `movement_request_nodes` + `movement_plan_edges` rows; it's compiled by the template compiler and consumed by the dispatcher.

Each absence is a deliberate trade: less surface area, fewer trust boundaries, simpler operator mental model — at the cost of less ergonomic UI, less general scheduling, and no built-in cron.

## Operator gate: MG\_WORKERS\_ENABLED

The whole worker subsystem is gated by **one boolean**: `MG_WORKERS_ENABLED`.

* `MG_WORKERS_ENABLED=true` — full runtime. Workers tick, movements progress, transfers happen.
* `MG_WORKERS_ENABLED=false` — API-only mode. The FastAPI app responds, you can create movements and approve them, but they sit at `APPROVED` forever. Every API endpoint stays idempotent.

This is the development mode and the "operator is debugging" mode. Drills like `run_ccip_send_drill` refuse to proceed without the gate enabled. The same gate is what lets you safely freeze a misbehaving system: flip the env var, restart the API process, and nothing advances until you flip it back.

## Adding a new periodic worker

To add a worker, you provide three things conceptually:

1. **A batch-of-work function** that takes a DB session, does one unit of work, and returns the number of rows it processed.
2. **A worker entry** that exposes that function as a named runtime worker.
3. **A bootstrap registration** that schedules it on the shared loop with an interval and jitter, and assigns it to a tier.

The new worker inherits everything on this page automatically:

* Heartbeat upsert.
* Network-class isolation.
* Exponential backoff on consecutive errors.
* Clean SIGTERM shutdown across all workers.
* Skip-locked concurrency, as long as its work-claiming query uses row-level skip-locked claiming.

The only design question for you is "core or sidecar?" — i.e. is this worker's failure something that should take down the whole runtime, or something to log and keep going?

## Related Docs

* [Architecture](/concepts/architecture) — the 3-layer code structure that hosts the worker loops
* [State Machine](/concepts/state-machine) — RequestState/NodeState transitions and UNKNOWN-handling
* [Movement Lifecycle](/concepts/movement-lifecycle) — how a movement traverses the workers in order
* [Executor Protocol](/concepts/executor-protocol) — the preflight/prepare/submit/observe/recover contract workers call into
* [Runtime Workers Reference](/reference/workers/runtime-workers) — per-worker implementation depth
* [Data Model](/reference/infrastructure/data-model) — `movement_request_nodes`, `worker_heartbeats`, `callback_outbox` schemas