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

# Runtime Workers

> Background workers — dispatcher, observer, recovery, and callback workers

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

```python theme={null}
# 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**:

1. Calls `func(session)` in an infinite loop
2. The function returns the number of items processed (int)
3. On success the `consecutive_errors` counter is reset
4. Waits for `interval + random.uniform(-jitter, jitter)` seconds (minimum 0.1s)
5. 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()`.

```python theme={null}
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:

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

```sql theme={null}
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

```mermaid theme={null}
flowchart TB
    READY["READY"] -->|Request APPROVED → EXECUTING on first dispatch| PREPARING
    PREPARING --> BurnCheck["(if cctp_burn)<br/>validate_cctp_burn_caller_alignment()<br/>failure → preflight_result.next_state (FAILED etc.)"]
    BurnCheck --> Preflight["executor.preflight(context)<br/>null → continue<br/>non-null → transition to next_state then stop"]
    Preflight --> Prepare["executor.prepare(context)<br/>prepared_action == null → FAILED<br/>present → persist Artifact"]
    Prepare --> Sign{signing_required?}
    Sign -->|true| AS["AWAITING_SIGNATURE<br/>no signer_binding → FAILED<br/>signer.sign(sign_request)<br/>obtain sign_result"]
    Sign -->|false| SUBMITTING
    AS --> SUBMITTING
    SUBMITTING --> Started["set started_at, increment attempt_no"]
    Started --> Submit["executor.submit(context, prepared_action, **submit_kwargs)<br/>update provider_state, provider_refs, provider_ref_id<br/>save generated_artifacts"]
    Submit --> Branch{submit_result.next_state}
    Branch -->|SUBMITTED| ObsAfter["observation by observer afterwards"]
    Branch -->|COMPLETED| Advance["advance_after_completion()<br/>activate successor nodes"]
    Branch -->|FAILED/UNKNOWN| Apply["apply_request_state_from_nodes()"]
    Branch -->|OBSERVING| ObsAfter2["observation by observer afterwards"]
    ObsAfter --> Commit[commit]
    Advance --> Commit
    Apply --> Commit
    ObsAfter2 --> Commit
```

### ExecutionContext construction

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

```python theme={null}
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.

```python theme={null}
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.

```sql theme={null}
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.

```sql theme={null}
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.

```python theme={null}
# 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

```mermaid theme={null}
flowchart TB
    Start["SUBMITTED / OBSERVING"] --> NextCheck["next_observe_at check<br/>(not yet due → skip)"]
    NextCheck --> Build["build_execution_context(session, node)"]
    Build --> Timeout{"timeout due and enforcement enabled?"}
    Timeout -->|yes| TimeoutUnknown["UNKNOWN(OBSERVATION_TIMEOUT)<br/>no executor call; derive request state"]
    Timeout -->|no| Observe["executor.observe(context)"]
    Observe --> Update["update provider_state, provider_refs<br/>last_observed_at = now, observe_count += 1<br/>save proof artifact<br/>save generated_artifacts"]
    Update --> Branch{result.next_state}
    Branch -->|COMPLETED| Comp["node → COMPLETED<br/>advance_after_completion()<br/>successor nodes → READY (successor_keys)<br/>update current_frontier<br/>enqueue frontier_advanced callback<br/>(if no successors) derive_request_state()"]
    Branch -->|OBSERVING| Obs["node → OBSERVING (preserved)<br/>compute/update next_observe_at"]
    Branch -->|FAILED| Fail["node → FAILED<br/>apply_request_state_from_nodes()"]
    Branch -->|UNKNOWN| Unk["node → UNKNOWN<br/>apply_request_state_from_nodes()<br/>(if derived is null) set the frontier to this node"]
    Branch -->|other| Other["transition to that state"]
```

### advance\_after\_completion() details

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

```python theme={null}
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.

```
1. all nodes COMPLETED/SKIPPED -> RequestState.COMPLETED
2. FAILED/UNKNOWN node exists + side-effect completed node exists -> MANUAL_INTERVENTION
3. FAILED node exists (no side-effects) -> FAILED
4. READY node with `manual_*` prefix exists + no currently-executing node -> WAITING_MANUAL_ACTION
5. 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'`.

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

### Recovery pipeline

```mermaid theme={null}
flowchart TB
    Start[UNKNOWN] --> Build["build_execution_context(session, node)"]
    Build --> Recover["executor.recover(context)"]
    Recover --> Update["update provider_state, provider_refs<br/>save proof artifact<br/>save generated_artifacts"]
    Update --> Branch{result.next_state}
    Branch -->|SUBMITTED| Sub["back to observer"]
    Branch -->|COMPLETED| Comp["direct transition,<br/>advance is observer's responsibility"]
    Branch -->|FAILED| Fail["apply_request_state_from_nodes()"]
    Branch -->|other| Other["transition to that state"]
```

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

```sql theme={null}
SELECT movement_callback_outbox.*
FROM movement_callback_outbox
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 5
FOR UPDATE SKIP LOCKED
```

### Send flow

```mermaid theme={null}
flowchart TB
    Start["pending record"] --> Due["next_attempt_at not yet due → skip"]
    Due --> Post["httpx.AsyncClient.post(callback_url, json=payload)"]
    Post --> Result{result}
    Result -->|success 2xx| Sent["status → 'sent'<br/>attempts += 1"]
    Result -->|failure| Fail["attempts += 1<br/>last_error = str(exc)"]
    Fail --> Check{attempts >= max_attempts?}
    Check -->|yes| Dlq["status → 'dlq'"]
    Check -->|no| Retry["next_attempt_at = now + 2^min(attempts,6)s<br/>(exponential backoff, capped 64s)"]
    Sent --> Commit[commit]
    Dlq --> Commit
    Retry --> Commit
```

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

```text theme={null}
{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

```sql theme={null}
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

```mermaid theme={null}
flowchart TB
    Start["PENDING_APPROVAL (expired)"] --> A["approval_status → 'rejected'"]
    A --> B["request_state → EXPIRED"]
    B --> C["enqueue request_expired callback"]
    C --> D[commit]
```

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

```mermaid theme={null}
flowchart TB
    Iter["iterate in-memory executors/signers"] --> Call["call executor.health()"]
    Call --> R{result}
    R -->|success| Up["health_state = 'up'"]
    R -->|exception| Down["health_state = 'down'"]
    Up --> Lookup["look up executor_key/signer_key in DB"]
    Down --> Lookup
    Lookup --> Found{found?}
    Found -->|missing| Create["create new record"]
    Found -->|present| Update["update"]
    Create --> Update2["update health_state, last_health_checked_at<br/>(EVM signers) validate observed identity, then save health"]
    Update --> Update2
    Update2 --> Commit[commit]
```

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

```mermaid theme={null}
flowchart TB
    Iter["iterate balance_fetcher_registry"] --> Fetch["fetcher.fetch_balances()"]
    Fetch --> R{result}
    R -->|success| Append["append BalanceSnapshot row"]
    R -->|exception| Rollback["per-venue rollback +<br/>continue to next fetcher"]
    Append --> Delete["delete rows older than retention cutoff"]
    Rollback --> Delete
    Delete --> Commit[commit]
```

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

```mermaid theme={null}
flowchart TB
    Create["API: POST /v3/movements (create)"]
    Pending["MovementRequest: PENDING_APPROVAL<br/>MovementRequestNodes: all BLOCKED"]
    Approve["API: POST /v3/movements/{id}/approve"]
    Approved["MovementRequest: APPROVED<br/>frontier nodes: BLOCKED → READY"]
    Create --> Pending --> Approve --> Approved

    subgraph Dispatcher["node_dispatcher (every 2s)"]
        D1["Pick READY node<br/>READY → PREPARING<br/>Request: APPROVED → EXECUTING"]
        D2["preflight() / prepare() / (sign() if needed)"]
        D3["PREPARING → SUBMITTING"]
        D4["submit()"]
        D5["SUBMITTING → SUBMITTED"]
        D1 --> D2 --> D3 --> D4 --> D5
    end
    Approved --> Dispatcher

    subgraph Observer["node_observer (every 5s)"]
        O1["Pick SUBMITTED/OBSERVING node<br/>check next_observe_at"]
        O2["timeout check (when enabled)<br/>then observe()"]
        O3{result}
        O4["OBSERVING<br/>set next_observe_at"]
        O5["COMPLETED<br/>advance_after_completion()<br/>unlock successor (BLOCKED → READY)<br/>update frontier"]
        O6{successors?}
        O7["back to dispatcher"]
        O8["derive_request_state()"]
        O9["COMPLETED (request) /<br/>FAILED (request)"]
        O1 --> O2 --> O3
        O3 --> O4
        O3 --> O5 --> O6
        O6 -->|Has| O7
        O6 -->|No| O8 --> O9
    end
    D5 --> Observer

    subgraph Recovery["node_recovery (every 60s)"]
        R1["Pick UNKNOWN nodes<br/>executor.recover()"]
        R2["→ SUBMITTED (back to observer)<br/>→ FAILED (request fails)"]
        R1 --> R2
    end
    Observer -.UNKNOWN nodes.-> Recovery

    subgraph Callback["callback_dispatcher (every 3s)"]
        C1["Pick pending callbacks<br/>HTTP POST to callback_url<br/>→ sent / dlq"]
    end
    Observer -.callbacks enqueued at each state change.-> Callback
    Dispatcher -.callbacks enqueued.-> Callback
    Recovery -.callbacks enqueued.-> Callback
```

***

## Concurrency guarantees

### skip\_locked

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

```python theme={null}
.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

```mermaid theme={null}
flowchart TB
    API["API (approve)"] -->|creates READY nodes| Dispatcher[dispatcher]
    Dispatcher -->|SUBMITTED nodes| Observer[observer]
    Observer -->|COMPLETED → unlocks successor → READY| Dispatcher
    Observer -->|UNKNOWN → handed to recovery| Recovery[recovery]
    Recovery -->|SUBMITTED back to observer| Observer
    Recovery -->|FAILED terminal| Terminal["(terminal)"]
    Dispatcher -.every state change → enqueue_callback().-> CB[callback_dispatcher]
    Observer -.every state change → enqueue_callback().-> CB
    Recovery -.every state change → enqueue_callback().-> CB
    CB -->|HTTP POST| External["(external receiver)"]
```

1. **API approve** -> frontier nodes transition to `READY`
2. **dispatcher** -> drives `READY` nodes through to `SUBMITTED`
3. **observer** -> polls `SUBMITTED`/`OBSERVING` nodes, transitions them to `COMPLETED` or `UNKNOWN`
4. **observer (advance)** -> activates successor nodes of completed nodes to `READY`, which the dispatcher then picks up
5. **recovery** -> transitions `UNKNOWN` nodes to `SUBMITTED` (re-observed by observer) or `FAILED`
6. **callback\_dispatcher** -> sends callbacks generated by every state change to external systems

***

## Related documents

* [Data Model Reference](/reference/infrastructure/data-model) — table/state enumeration details
* [Bootstrap & Configuration](/reference/infrastructure/bootstrap) — worker interval settings, executor registration
* [V3 API Endpoints](/reference/api/v3-endpoints) — approve/reject/resume/retry/cancel endpoints