Runtime Workers Reference

Source files: src/qtg/interfaces/workers/, src/qtg/application/services/dispatch.py, observe.py, recover.py

Defines the qtg v3 background worker architecture. When MG_WORKERS_ENABLED=true, the runtime supervises 8 always-on workers plus 9 conditional workers (each gated by its own flag) that handle DAG execution, observation, recovery, callbacks, expiration, template-proposal expiry, audit promotion, health checks, nonce/stranded-message reaping, capital-transfer follow-up, balance snapshots, and registry/CCIP drift detection.

Worker Architecture

loop_runner pattern

All workers run through the shared run_loop() runner.

# src/qtg/interfaces/workers/loop_runner.py
async def run_loop(
    name: str,
    func: Callable[[AsyncSession], Awaitable[int]],
    interval: float,
    jitter: float = 0,
) -> None:

Behavior:

Calls func(session) in an infinite loop
The function returns the number of items processed (int)
On success the consecutive_errors counter is reset
Waits for interval + random.uniform(-jitter, jitter) seconds (minimum 0.1s)
On exception, increments consecutive_errors

Error handling:

CancelledError: worker terminates (shutdown)
Other exceptions: log and continue execution
Consecutive errors >= MAX_CONSECUTIVE_ERRORS (5): exponential backoff applied

backoff = min(2^consecutive_errors, MAX_BACKOFF_SECONDS)  # max 60s

Session management: a new session is created via AsyncSessionLocal() on every loop iteration and auto-released when the block exits.

Worker registration / supervision

When MG_WORKERS_ENABLED=true, the lifespan() context manager in main.py runs a single supervision task via run_worker_groups().

runtime_supervision_task = asyncio.create_task(
    run_worker_groups(
        core_workers={
            "node_dispatcher": lambda: run_loop(...),
            "node_observer": lambda: run_loop(...),
            "node_recovery": lambda: run_loop(...),
            # joins the core group only when enabled (default True)
            "evm_nonce_reaper": ...,        # if settings.evm_nonce_reaper_enabled
            "ccip_stranded_reaper": ...,    # if settings.ccip_stranded_reaper_enabled
        },
        sidecar_workers={
            "callback_dispatcher": lambda: run_loop(...),
            "expiration_checker": lambda: run_loop(...),
            "template_proposal_expiration": lambda: run_loop(...),
            "executor_health": lambda: run_loop(...),
            "audit_outbox_promoter": lambda: run_loop(...),
            # conditional sidecars, each gated by its own flag:
            "capital_transfer_followup": ...,  # if settings.capital_transfer_followup_enabled
            "balance_snapshot": ...,           # if settings.balance_snapshot_enabled
            "ccip_drift": ...,                 # if settings.ccip_enabled + sidecar
            "verify_drift": ...,               # if settings.verify_drift_enabled
        },
    ),
    name="qtg.runtime_supervision",
)

Always-on vs conditional

8 always-on (run whenever MG_WORKERS_ENABLED=true):

node_dispatcher, node_observer, node_recovery (core group)
callback_dispatcher, expiration_checker, template_proposal_expiration, executor_health, audit_outbox_promoter (sidecar group)

9 conditional (each gated by its own flag):

Worker	Flag	Group
`evm_nonce_reaper`	`MG_EVM_NONCE_REAPER_ENABLED` (default `True`)	core (joins fail-fast group)
`ccip_stranded_reaper`	`MG_CCIP_STRANDED_REAPER_ENABLED` (default `True`)	core (joins fail-fast group)
`capital_transfer_followup`	`MG_CAPITAL_TRANSFER_FOLLOWUP_ENABLED`	sidecar
`notification_dispatcher`	`MG_NOTIFY_ENABLED`	sidecar
`balance_history_rollup`	`MG_BALANCE_HISTORY_ENABLED` (default `True`)	sidecar
`cex_transfer_ingest`	non-empty `MG_CEX_TRANSFER_INGEST_VENUES_CSV` (no boolean flag)	sidecar
`balance_snapshot`	`MG_BALANCE_SNAPSHOT_ENABLED`	sidecar
`ccip_drift`	`MG_CCIP_ENABLED` (+ sidecar client)	sidecar
`verify_drift`	`MG_VERIFY_DRIFT_ENABLED`	sidecar

Supervision semantics

core group
- node_dispatcher
- node_observer
- node_recovery
- evm_nonce_reaper (when enabled — joins the core fail-fast group)
- ccip_stranded_reaper (when enabled — joins the core fail-fast group)
sidecar group
- callback_dispatcher
- expiration_checker
- template_proposal_expiration
- executor_health
- audit_outbox_promoter
- capital_transfer_followup / balance_snapshot / ccip_drift / verify_drift (when enabled)

Current meaning:

If a core worker exits abnormally outside its loop, the entire core group fails fast. The two conditional reapers (evm_nonce_reaper, ccip_stranded_reaper) deliberately join this core group so their failure is treated as fatal, not silently swallowed.
If a sidecar worker crashes, it is logged and the remaining sidecar/core workers continue.
On app shutdown, the supervision task is cancelled and waits for a clean shutdown.
Retry / backoff / per-iteration error handling inside run_loop() is preserved as-is.

Worker summary table

Worker	Interval (default)	Function	Always-on?	Role
`node_dispatcher`	2s	`dispatch_ready_nodes`	always (core)	Feed READY nodes into the execution pipeline
`node_observer`	5s	`observe_active_nodes`	always (core)	Poll state of SUBMITTED/OBSERVING nodes
`node_recovery`	60s	`recover_unknown_nodes`	always (core)	Attempt recovery of UNKNOWN nodes
`callback_dispatcher`	3s	`dispatch_pending_callbacks`	always (sidecar)	Send entries from the callback outbox
`expiration_checker`	30s	`expire_pending_requests`	always (sidecar)	Handle approval expiration
`template_proposal_expiration`	30s	`expire_template_proposals`	always (sidecar)	Expire stale template proposals
`executor_health`	60s	`refresh_executor_health`	always (sidecar)	Health-check executors/signers
`audit_outbox_promoter`	1s	audit promoter tick	always (sidecar)	Promote `audit_outbox` rows into `audit_events`
`evm_nonce_reaper`	60s	EVM nonce reaper loops	if `MG_EVM_NONCE_REAPER_ENABLED` (core)	Reclaim stale EVM nonce reservations
`ccip_stranded_reaper`	60s	`make_ccip_stranded_reaper()`	if `MG_CCIP_STRANDED_REAPER_ENABLED` (core)	Reap stranded CCIP send intents
`capital_transfer_followup`	60s	`follow_up_pending_capital_transfers`	if `MG_CAPITAL_TRANSFER_FOLLOWUP_ENABLED` (sidecar)	Follow up pending capital transfers
`balance_snapshot`	30s	balance snapshot loops	if `MG_BALANCE_SNAPSHOT_ENABLED` (sidecar)	Store venue balance snapshots + retention cleanup
`ccip_drift`	(configurable)	`run_drift_worker`	if `MG_CCIP_ENABLED` + sidecar (sidecar)	Detect CCIP registry drift
`verify_drift`	21600s	verify-drift loops	if `MG_VERIFY_DRIFT_ENABLED` (sidecar)	Verify on-chain registry baselines against config

1. node_dispatcher

Interval: 2 seconds (MG_NODE_DISPATCH_INTERVAL_SECONDS) Finds nodes in the READY state and runs the preflight -> prepare -> (sign) -> submit pipeline. Processes one node at a time.

Lookup query (per-candidate claim)

Production dispatch_ready_nodes() no longer uses a single multi-candidate FOR UPDATE SKIP LOCKED join. That shape was shown to retain a speculative movement_request_nodes lock after skipping a request-side miss while claiming a later FIFO candidate in the same statement. Claim path:

Walk unlocked FIFO candidate IDs one at a time (no FOR UPDATE on the listing query; exclude already-tried IDs after a miss).
For each candidate, run a per-ID joined claim under a miss-only savepoint:

SELECT movement_request_nodes.*
FROM movement_request_nodes
JOIN movement_requests ON movement_requests.id = movement_request_nodes.request_id
WHERE movement_request_nodes.id = :candidate_id
  AND movement_request_nodes.node_state = 'READY'
  AND movement_requests.request_state IN ('APPROVED', 'EXECUTING')
FOR UPDATE OF movement_request_nodes, movement_requests SKIP LOCKED

On miss, roll back only that savepoint so partial locks from the skipped attempt are released; on hit, release the savepoint and keep locks through the existing outer dispatch transaction (no per-node outer commit split). Durable proof: tests/qtg/application/test_dispatch_pickup_concurrency.py. Observer↔recovery request-lock ownership is tracked as separate residual work.

SKIP LOCKED: prevents duplicate processing in multi-instance environments
FIFO order: older requests come first, ordered by created_at

Dispatch pipeline

ExecutionContext construction

The build_execution_context() function assembles the context required to execute a node.

ExecutionContext(
    protocol_version='1.0',
    request_id=str(request.id),
    request_node_id=str(request_node.id),
    compiled_plan_hash=request.compiled_plan_hash,
    template_key=template.template_key,
    template_version=plan_version.version,
    node_key=request_node.node_key,
    node_kind=plan_node.node_kind,
    action_type=plan_node.action_type,
    attempt_no=request_node.attempt_no + 1,
    intent=dict(request.intent),
    input_params=dict(request.input_params),
    resolved_bindings={...},
    risk_controls=dict(plan_version.risk_controls),
    prior_artifacts=[...],           # artifacts of direct-predecessor completed nodes
    node_config={...},               # plan_node.config after top-level $ref resolution
    provider_context={...},          # flat provider_refs of direct-predecessor completed nodes
    timeout_policy=dict(plan_node.timeout_policy),
)

provider_context is a dict that flat-merges the provider_refs of the current node’s direct-predecessor nodes in the COMPLETED state. Through this, successor nodes can reference predecessor txids, withdrawal_ids, etc. node_config is the result of resolving the template’s top-level $ref:node_key.field expressions before handing it to the executor. At runtime a synthetic namespaced ref map is built from every completed node’s provider_refs, but those values are used only for $ref resolution and are not placed in provider_context. Therefore the executor still sees a flat provider_context as before. If a $ref cannot be resolved during build_execution_context(), ConfigRefResolutionError is raised and the dispatcher transitions the node to FAILED, recording error_code = CONFIG_REF_UNRESOLVED.

SignRequest construction

This is the request passed to the signer for nodes that require signing.

SignRequest(
    protocol_version='1.0',
    request_id=context.request_id,
    request_node_id=context.request_node_id,
    signing_intent=SigningIntent(
        action=prepared_action.action_type,
        asset=str(context.intent.get('asset', '')),
        amount=str(context.intent.get('amount', ...)),
        destination=str(destination_value),
        max_fee_usd=None,
        chain_family=str(context.input_params.get('chain_family', 'evm')),
        allowed_payload_hash=prepared_action.payload_hash,
    ),
    payload_format=prepared_action.payload_format,
    payload=prepared_action.payload or '',
    payload_hash=prepared_action.payload_hash,
    metadata=context.input_params,
)

2. node_observer

Interval: 5 seconds (MG_NODE_OBSERVE_INTERVAL_SECONDS) Polls nodes in the SUBMITTED or OBSERVING state to detect state changes. Processes up to 10 nodes per pass.

Lookup query (per-node claim)

The observer works in two phases so a slow provider observation on one node never holds a lock that blocks the rest of the batch: 1. Advisory, unlocked candidate snapshot — list up to 10 due candidate IDs. No rows are locked here.

SELECT movement_request_nodes.id
FROM movement_request_nodes
JOIN movement_requests ON movement_requests.id = movement_request_nodes.request_id
WHERE movement_request_nodes.node_state IN ('SUBMITTED', 'OBSERVING')
  AND movement_requests.request_state = 'EXECUTING'
  AND (movement_request_nodes.next_observe_at IS NULL
       OR movement_request_nodes.next_observe_at <= now())
ORDER BY movement_requests.created_at ASC
LIMIT 10

2. Per-node claim — for each candidate ID, re-select the node and its request with FOR UPDATE ... SKIP LOCKED (locking both rows), still filtered on the same due predicate so a node another worker already advanced is not re-claimed.

SELECT movement_request_nodes.*, movement_requests.*
FROM movement_request_nodes
JOIN movement_requests ON movement_requests.id = movement_request_nodes.request_id
WHERE movement_request_nodes.id = :node_id
  AND movement_request_nodes.node_state IN ('SUBMITTED', 'OBSERVING')
  AND movement_requests.request_state = 'EXECUTING'
  AND (movement_request_nodes.next_observe_at IS NULL
       OR movement_request_nodes.next_observe_at <= now())
FOR UPDATE OF movement_request_nodes, movement_requests SKIP LOCKED

The claim runs inside a savepoint: PostgreSQL can lock the node side of the joined tuple before SKIP LOCKED rejects an already-locked request side, so a miss rolls the savepoint back and releases those partial locks. A hit keeps the node + request locks for the whole observe lifetime (through provider observation and commit).

Observation interval control

next_observe_at is maintained per node to prevent unnecessary polling.

# skip if the current time is before next_observe_at
if node.next_observe_at and _as_utc(node.next_observe_at) > datetime.now(UTC):
    continue

Default observation intervals (DEFAULT_OBSERVE_INTERVALS):

action_type	Interval (seconds)
`cex_withdrawal_status`	5
`cex_deposit_status`	10
`destination_chain_receive_observe`	10
`destination_chain_finality_observe`	15
`protocol_observe`	15
`cctp_attestation`	15
`cctp_mint_status`	5
`ccip_delivery`	60
`debridge_fulfillment`	30
(other)	10

If the executor returns retry_after_seconds, that value takes precedence.

Observation timeout enforcement

timeout_policy.observe_seconds is enforced only when MG_OBSERVE_TIMEOUT_ENFORCEMENT_ENABLED=true; the setting defaults to false. When a due node has exceeded its positive observe_seconds since started_at, the observer does not call the executor. It sets the node to UNKNOWN(error_code=OBSERVATION_TIMEOUT), then normal request-state derivation escalates a lane with a completed side effect to MANUAL_INTERVENTION and enqueues the existing request_attention callback. The timeout never marks the node FAILED; its reservation remains preserved for operator handling. The check is intentionally after the next_observe_at gate. A timeout can therefore be detected up to one backoff interval late; retry hints are clamped to MAX_BACKOFF_SECONDS=600, so the bounded delay is at most 600 seconds. This is acceptable for a 1800–3600-second paging signal, not a precise real-time deadline. timeout_policy is not included in the graph hash or compiled-plan hash. A post-approval DB edit to timeout_policy.observe_seconds therefore does not trigger hash-drift detection. Its blast radius is paging timing only, but this is still an operational limitation: audit DB template observe_seconds values before globally enabling enforcement. Existing seed templates already declare observe_seconds, so enabling the flag without that audit can activate timeouts on existing live lanes.

Observe pipeline

advance_after_completion() details

When a node reaches COMPLETED, successor nodes are activated along the DAG.

async def advance_after_completion(session, request, completed_node, nodes):
    successor_keys = await get_successor_keys(session, completed_node.plan_node_id)

    if not successor_keys:
        # last node: declare full completion
        await apply_request_state_from_nodes(session, request, nodes, ...)
        return

    for node in nodes:
        if node.node_key in successor_keys and node.node_state == NodeState.BLOCKED:
            # BLOCKED -> READY
            await set_node_state(session, node, NodeState.READY, ...)

    request.current_frontier = successor_keys
    await enqueue_callback(session, request, event_type='frontier_advanced', ...)

derive_request_state() details

Aggregates all node states to derive the request state.

all nodes COMPLETED/SKIPPED -> RequestState.COMPLETED
FAILED/UNKNOWN node exists + side-effect completed node exists -> MANUAL_INTERVENTION
FAILED node exists (no side-effects) -> FAILED
READY node with `manual_*` prefix exists + no currently-executing node -> WAITING_MANUAL_ACTION
otherwise -> None (no change)

3. node_recovery

Interval: 60 seconds (MG_NODE_RECOVERY_INTERVAL_SECONDS) Recovers nodes in UNKNOWN, plus nodes stuck in SUBMITTING past their lane’s grace cutoff, via the executor’s recover() method. The batch is MG_NODE_RECOVERY_BATCH_SIZE (default 10) split across four arms in priority order — CCIP, Stargate, CEX, then everything else — each taking max(1, batch // 4). Keep the batch at 4 or more, or the later arms starve.

Lookup query

Each arm selects its own lane, ordered by movement_request_nodes.updated_at ASC with FOR UPDATE ... SKIP LOCKED, and every arm requires movement_requests.request_state = 'EXECUTING'.

A movement that has left EXECUTING is no longer swept. An UNKNOWN node inside a MANUAL_INTERVENTION movement will not be auto-recovered — and that is exactly the state a completed side-effect node alongside an UNKNOWN node produces. Resume the movement to EXECUTING first, or reconcile it by hand.

Recovery pipeline

UNKNOWN is the state of “we sent an execution request but cannot determine the result”. The executor’s recover() method queries the provider to determine the actual state.

4. callback_dispatcher

Interval: 3 seconds (MG_CALLBACK_DISPATCH_INTERVAL_SECONDS) Pulls callbacks in the pending state from the callback outbox and sends them via HTTP POST. Processes up to 5 per pass.

Lookup query

SELECT movement_callback_outbox.*
FROM movement_callback_outbox
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 5
FOR UPDATE SKIP LOCKED

Send flow

Callback authentication (v3)

The body contains the pure payload; auth metadata is passed via headers. Headers:

X-QTG-Callback-Timestamp
X-QTG-Callback-Nonce
X-QTG-Callback-Signature-Version: v3
X-QTG-Callback-Signature

Signature message:

{timestamp}
{nonce}
{sha256(body_bytes)}

MovementCallbackOutbox reuses the same callback_timestamp / callback_nonce on retries. As a result the receiver can make a deterministic replay decision for the same callback delivery attempt.

Retry backoff

On a failed delivery, next_attempt_at is set to exponential backoff: now + 2 ** min(attempts, 6) seconds — i.e. 2, 4, 8, 16, 32, then capped at 64s for all later attempts. (This is not a flat 5s retry.) When attempts reaches max_attempts (default 5, MG_CALLBACK_MAX_ATTEMPTS), the row transitions to dlq.

5. expiration_checker

Interval: 30 seconds (MG_EXPIRATION_CHECK_INTERVAL_SECONDS) Processes expiration of requests in PENDING_APPROVAL whose expires_at has passed. Processes up to 20 per pass.

Lookup query

SELECT movement_requests.*
FROM movement_requests
WHERE request_state = 'PENDING_APPROVAL'
  AND expires_at IS NOT NULL
  AND expires_at < now()
LIMIT 20
FOR UPDATE SKIP LOCKED

Processing flow

The approval TTL is MG_APPROVAL_TTL_SECONDS (default 300 seconds = 5 minutes).

6. executor_health

Interval: 60 seconds (MG_EXECUTOR_HEALTH_INTERVAL_SECONDS) Calls the health() method on every in-memory registered executor and signer and persists health state to the DB.

Processing flow

If the executor/signer does not implement health(), health_state = 'unknown' is preserved. For signer identity, this worker is the sole writer of observed_signer_address: it validates a nonzero EVM health().signer_address value, stores the canonical lowercase observation, and clears the observation when the runtime signer disappears and the DB row becomes orphan. It never writes expected_signer_address; the admin enrollment endpoint PUT /v3/signers/{signer_key}/expected-identity is the sole expected-identity writer. The worker records health separately from identity posture, which is derived when the roster is read. The signer-identity contract deliberately stops at this expected-versus-observed registry evidence. It does not enforce movement-time or rotation-time drift; that enforcement is a separate follow-up.

7. balance_snapshot (optional)

Interval: 30 seconds (MG_BALANCE_SNAPSHOT_INTERVAL_SECONDS) Activation conditions:

MG_WORKERS_ENABLED=true
MG_BALANCE_SNAPSHOT_ENABLED=true

Iterates the registered balance_fetcher_registry and stores per-venue balances append-only in the balance_snapshots table. The current default implementation is GatewayBalanceFetcher, which records the unified USDC balance read from the Gateway API as a single venue_key="gateway" snapshot.

Processing flow

Operational meaning

Baseline data source for GET /v3/balances
Uses the same append-only schema as fresh=true on-demand fetches
Stale judgment is computed at lookup time (balance_stale_threshold_seconds); the worker is responsible for snapshot storage and retention

8. template_proposal_expiration (always-on)

Interval: 30 seconds (MG_TEMPLATE_PROPOSAL_EXPIRATION_INTERVAL_SECONDS) Expires stale template_proposals rows whose review/approval window has elapsed (expire_template_proposals). Keeps the proposal queue clean so abandoned proposals do not linger as actionable.

9. audit_outbox_promoter (always-on)

Interval: 1 second Copies rows from the audit_outbox table into audit_events on a tight tick so forensic queries see successful mutations promptly. This is the read-side projection half of the audit outbox pattern: mutating routes write to audit_outbox transactionally, and this worker promotes them to the durable audit_events log.

10. evm_nonce_reaper (conditional, core)

Flag: MG_EVM_NONCE_REAPER_ENABLED (default True) Interval: 60 seconds (MG_EVM_NONCE_REAPER_INTERVAL_SECONDS) Reclaims stale EVM nonce reservations (evm_nonce_reservation) older than the age threshold so a worker that died mid-broadcast does not permanently strand a nonce. It is registered only when an EVM RPC client and EVM endpoints are available; otherwise a warning is logged and the worker is skipped. When enabled it joins the core fail-fast group — its failure is treated as fatal to the core group, not silently logged.

11. ccip_stranded_reaper (conditional, core)

Flag: MG_CCIP_STRANDED_REAPER_ENABLED (default True) Interval: 60 seconds (MG_CCIP_STRANDED_REAPER_INTERVAL_SECONDS) Reaps stranded CCIP send intents (ccip_send_intent) that have aged past the threshold without resolving. Like the nonce reaper, it joins the core fail-fast group when enabled.

12. capital_transfer_followup (conditional, sidecar)

Flag: MG_CAPITAL_TRANSFER_FOLLOWUP_ENABLED Interval: 60 seconds (MG_CAPITAL_TRANSFER_FOLLOWUP_INTERVAL_SECONDS) Follows up on pending capital_transfer_requests (follow_up_pending_capital_transfers) that have gone stale, advancing or surfacing them for operator attention.

13. verify_drift (conditional, sidecar)

Flag: MG_VERIFY_DRIFT_ENABLED Interval: 21600 seconds / 6 hours (MG_VERIFY_DRIFT_INTERVAL_SECONDS) Re-verifies on-chain registry baselines against configured values on a slow cadence, flagging drift between what is registered and what the chain reports. Runs per-chain with bounded RPC concurrency and per-lane timeouts.

14. ccip_drift (conditional, sidecar)

Flag: MG_CCIP_ENABLED (and a configured CCIP sidecar client) Interval: configurable (MG_CCIP_DRIFT_POLL_INTERVAL_SECONDS) Polls the CCIP sidecar for registry drift (run_drift_worker). Active only when CCIP is enabled and the sidecar client is available.

Worker heartbeats

The worker_heartbeats table records liveness for the supervised workers. Each process seeds heartbeat rows for its selected network class at startup (iterate_per_network(_seed_worker_heartbeats)), so separate network deployments keep separate heartbeat baselines. This gives operators a DB-visible signal of which workers are expected to be running.

State Progression Flow Through Workers

The ASCII diagram below shows the full flow of one movement request from creation through completion, progressing through the workers.

Concurrency guarantees

skip_locked

Each worker uses FOR UPDATE SKIP LOCKED to claim rows so no two workers process the same node.

.with_for_update(skip_locked=True)

The dispatcher, recovery, callback, and expiration workers lock rows directly in their pickup query. The observer instead lists due candidate IDs without a lock, then claims each node — and its request — one at a time with FOR UPDATE OF ... SKIP LOCKED inside a savepoint, so one node’s long provider observation never holds a lock that blocks the rest of the batch. In an environment with multiple worker instances (or concurrent execution within the same process) this means:

Rows already locked by another transaction are skipped
Concurrent processing is safe without deadlocks
The same node/request is not processed twice

Limits

Per-worker batch sizes:

Worker	Limit	Note
node_dispatcher	1	One at a time (serial execution guarantee)
node_observer	10	Observe multiple nodes concurrently
node_recovery	10	Recover multiple nodes concurrently
callback_dispatcher	5	Send multiple callbacks concurrently
expiration_checker	20	Expire multiple requests concurrently
executor_health	(all)	Every registered executor/signer

Ordering basis

Worker	ORDER BY	Meaning
node_dispatcher	`movement_requests.created_at ASC`	FIFO: older requests first
node_observer	`movement_requests.created_at ASC`	FIFO: older requests first
node_recovery	`movement_request_nodes.updated_at ASC`	Longest-idle UNKNOWN node first
callback_dispatcher	`movement_callback_outbox.created_at ASC`	FIFO: older callbacks first
expiration_checker	(none, limit only)	Any expired ones

Worker interaction

API approve -> frontier nodes transition to READY
dispatcher -> drives READY nodes through to SUBMITTED
observer -> polls SUBMITTED/OBSERVING nodes, transitions them to COMPLETED or UNKNOWN
observer (advance) -> activates successor nodes of completed nodes to READY, which the dispatcher then picks up
recovery -> transitions UNKNOWN nodes to SUBMITTED (re-observed by observer) or FAILED
callback_dispatcher -> sends callbacks generated by every state change to external systems

Data Model Reference — table/state enumeration details
Bootstrap & Configuration — worker interval settings, executor registration
V3 API Endpoints — approve/reject/resume/retry/cancel endpoints

​Runtime Workers Reference

​Worker Architecture

​loop_runner pattern

​Worker registration / supervision

​Always-on vs conditional

​Supervision semantics

​Worker summary table

​1. node_dispatcher

​Lookup query (per-candidate claim)

​Dispatch pipeline

​ExecutionContext construction

​SignRequest construction

​2. node_observer

​Lookup query (per-node claim)

​Observation interval control

​Observation timeout enforcement

​Observe pipeline

​advance_after_completion() details

​derive_request_state() details

​3. node_recovery

​Lookup query

​Recovery pipeline

​4. callback_dispatcher

​Lookup query

​Send flow

​Callback authentication (v3)

​Retry backoff

​5. expiration_checker

​Lookup query

​Processing flow

​6. executor_health

​Processing flow

​7. balance_snapshot (optional)

​Processing flow

​Operational meaning

​8. template_proposal_expiration (always-on)

​9. audit_outbox_promoter (always-on)

​10. evm_nonce_reaper (conditional, core)

​11. ccip_stranded_reaper (conditional, core)

​12. capital_transfer_followup (conditional, sidecar)

​13. verify_drift (conditional, sidecar)

​14. ccip_drift (conditional, sidecar)

​Worker heartbeats

​State Progression Flow Through Workers

​Concurrency guarantees

​skip_locked

​Limits

​Ordering basis

​Worker interaction

​Related documents

Runtime Workers Reference

Worker Architecture

loop_runner pattern

Worker registration / supervision

Always-on vs conditional

Supervision semantics

Worker summary table

1. node_dispatcher

Lookup query (per-candidate claim)

Dispatch pipeline

ExecutionContext construction

SignRequest construction

2. node_observer

Lookup query (per-node claim)

Observation interval control

Observation timeout enforcement

Observe pipeline

advance_after_completion() details

derive_request_state() details

3. node_recovery

Lookup query

Recovery pipeline

4. callback_dispatcher

Lookup query

Send flow

Callback authentication (v3)

Retry backoff

5. expiration_checker

Lookup query

Processing flow

6. executor_health

Processing flow

7. balance_snapshot (optional)

Processing flow

Operational meaning

8. template_proposal_expiration (always-on)

9. audit_outbox_promoter (always-on)

10. evm_nonce_reaper (conditional, core)

11. ccip_stranded_reaper (conditional, core)

12. capital_transfer_followup (conditional, sidecar)

13. verify_drift (conditional, sidecar)

14. ccip_drift (conditional, sidecar)

Worker heartbeats

State Progression Flow Through Workers

Concurrency guarantees

skip_locked

Limits

Ordering basis

Worker interaction

Related documents