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.

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.

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.

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.

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

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.

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.

-- 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 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:

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). 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 — the 3-layer code structure that hosts the worker loops
• State Machine — RequestState/NodeState transitions and UNKNOWN-handling
• Movement Lifecycle — how a movement traverses the workers in order
• Executor Protocol — the preflight/prepare/submit/observe/recover contract workers call into
• Runtime Workers Reference — per-worker implementation depth
• Data Model — movement_request_nodes, worker_heartbeats, callback_outbox schemas