# BackgroundScheduler
Autonomous Pass executor backed by [`SharedRing`](./SHARED_RING.md)
+ [`HeartbeatTable`](./HEARTBEAT.md) +
[`FailoverWatchdog`](./FAILOVER.md) +
[`pass_registry`](./PASS_REGISTRY.md). Each process opens shared
submit + result rings, registers in the heartbeat table, and
drives one worker thread that drains the submit-ring, executes
the Pass via the closure registry, and pushes the result onto
the result-ring. Cross-thread + cross-process + disk durability
from the same MMF substrate.
> **The "cross-process Pass dispatch with auto-failover"
> primitive.** Submit at **219 ns** vs `mpsc::sync_channel`
> 204 ns (1.07x slower; tied). watchdog_scan at 18.91 ns for an
> 8-slot heartbeat (~2.36 ns per slot). try_recv at 254 ns
> (full submit -> worker -> result -> recv path). Architectural
> lever stacks: cross-process visibility AND durable ring (file
> IS the queue) AND auto-failover, at zero measurable cost over
> the in-process mpsc baseline.
**Constraints (read first):**
- **Native sidecar integration**: `BackgroundScheduler` carries a `HandshakeHeader` + `ObservationRing` and implements `subetha_sidecar::AdaptiveInstance`. `Submitter` / `ResultCollector` handles forward their observations through the underlying `SharedRing`'s sidecar (no separate wrapper-level ring). Wrap in `SidecarBox::new` to register with the global sidecar; raw `start()` returns the unregistered type unchanged.
- **Pass payload bounded by `MAX_ARG_LEN`** (46 bytes within
the 56-byte ring slot, minus header). Larger args need a
side-channel.
- **Wire format**: `[closure_id u32][result_token u32][arg_len
u16][args...]` for submit; `[result_token u32][status
u8][result_len u16][result...]` for result.
- **`result_token` is caller-supplied correlation ID**:
echoed in the result-ring payload so the originator matches
results to submissions.
- **Worker drops results when result-ring is full**: no
blocking on result-push (the worker stays unblocked even
when no consumer is reading).
- **Worker beats heartbeat every 100 iterations** of its
drain loop. `mark_in_flight` flips a bit during execution;
`clear_in_flight` clears it on completion.
- **`Drop` on the scheduler signals stop + joins the worker**.
- **Three MMF files**: submit ring + result ring + heartbeat
table. Pass paths; the scheduler creates-or-opens each.
- **Cross-process backed by MMF.**
---
## Table of contents
- [What it is](#what-it-is)
- [Wire format](#wire-format)
- [Worker loop](#worker-loop)
- [Bench evidence](#bench-evidence)
- [Worked examples](#worked-examples)
- [Use case patterns](#use-case-patterns)
- [Known limitations](#known-limitations)
- [Common pitfalls](#common-pitfalls)
- [References](#references)
---
## What it is
```text
Producer side Consumer side
+--------------------+ +----------------------+
| -> result_token | | -> SubmittedResult |
+----------|---------+ +----------^-----------+
| |
v |
+-----------------------+ +----------------------+
+-----------|-----------+ +----------^-----------+
| |
v |
+---------------------------+ |
| 2. mark_in_flight bit | |
| 3. pass_registry exec |--------------+
| 4. push to result ring |
| 5. clear in_flight bit |
| 6. heartbeat.beat |
+---------------------------+
|
v
+------------------+
| HeartbeatTable | <base>.hb.bin
+------------------+
^
|
+------------------+
| FailoverWatchdog | (coordinator process only)
| .scan() -> |
| ReclaimReport |
+------------------+
```
Three MMF files. Multiple processes can each `start()` against
the same paths; they share the rings and the heartbeat. One
process (typically the coordinator) periodically calls
`watchdog_scan()` to detect dead workers and reclaim their
in-flight bits.
---
## Wire format
### Submit ring slot (56 bytes)
```text
| offset 8 | arg_len (u16 LE) |
| offset 10 | args (variable bytes) |
```
`MAX_ARG_LEN = 46` (56 - 10).
### Result ring slot (56 bytes)
```text
| offset 5 | result_len (u16 LE) |
| offset 7 | result (variable) |
```
Status 0 = Ok, status 1 = Err (with stringified error). `MAX_RESULT_LEN = 49`.
---
## Worker loop
```text
loop:
if stop.load(Acquire): break
beat_counter += 1
if beat_counter % 100 == 0: heartbeat.beat(my_slot)
match submit_ring.try_pop:
Ok(slot):
pass, token = parse_submit_slot(slot)
bit = (token & 0x3F) as u8
heartbeat.mark_in_flight(my_slot, bit)
result = pass_registry::execute(&pass)
result_ring.try_push(encode_result_slot(token, result)).ok()
heartbeat.clear_in_flight(my_slot, bit)
Err(Empty):
sleep 100us # idle backoff
Err(_):
break
heartbeat.unregister(my_slot)
```
The worker is one autonomous thread per process. Multiple
workers across processes naturally load-balance via the
single shared submit ring (each `try_pop` is one CAS).
---
## Bench evidence
Bench harness: `crates/subetha-cxc/benches/scheduler.rs`. Captured
2026-06-02 on Windows 11 / Zen+ R7 2700, Criterion with
`--sample-size=15 --warm-up-time=1 --measurement-time=2`.
Workload: Pass with 4-byte args, echo closure (returns args).
Worker drains concurrently (pre-spawned at scheduler start).
| submit (encode + ring push + opp. recv) | 219.25 ns | 204.49 ns | tied (mmf 1.07x slower) |
| try_recv (encode + recv + decode + submit-feed) | 254.30 ns | n/a | full path |
| watchdog_scan (8-slot heartbeat) | 18.91 ns | n/a | ~2.36 ns/slot |
### Reading the trade-offs
1. **Submit is tied with mpsc::sync_channel.** Both contenders
pay: encode the Pass + push to queue + opportunistically
drain. The MMF substrate adds zero measurable cost over the
in-process std queue. The Vyukov MPMC ring + 56-byte slot
encode equals the cost of `mpsc::SyncSender::send` + Vec
allocation for the Pass.
2. **try_recv at 254 ns** measures the full submit -> worker
pickup -> execute -> result push -> recv loop. The
worker's idle-sleep of 100 µs means recv is bottlenecked on
the worker's wake-up cadence; under steady submission
pressure the worker stays awake.
3. **watchdog_scan at 18.91 ns** is the same scan pattern as
`failover.scan`: O(slots) atomic loads, dead count is free.
4. **The architectural lever is cross-process + durable +
failover.** mpsc::sync_channel is in-process only and has
no failover. The scheduler's ring file IS the persistent
queue.
### Rule 3b bench audit
- **Fair contender**: `mpsc::sync_channel<Pass>` is the std
in-process Pass-dispatch equivalent. Same payload shape, same
encode cost (Vec arg in Pass), same drain pattern.
- **Worker pre-spawned**: `BackgroundScheduler::start()` spawns
the worker thread once before the bench; no `thread::spawn`
inside `b.iter`. The mpsc baseline pre-spawns its drain
thread similarly.
- **Sizing**: ring 1024 slots (no overflow at criterion's iter
counts); heartbeat 8 slots (typical scheduler pool size).
- **MMF lifecycle managed**: scheduler created + ops + dropped +
3 files removed per bench function.
### What the numbers do NOT show
- **Cross-process Pass dispatch**: process A submits, process B's
worker drains and executes. The mpsc baseline cannot do this.
- **Crash recovery**: a worker process crashes mid-pass; the
submit ring's items are still in the durable file. A
restarted worker (or another worker process) picks them up.
- **Auto-failover via watchdog**: dead workers' heartbeat lapses;
the watchdog reclaims their in-flight bits; their in-progress
work is reassigned.
- **Multi-process load balancing**: N worker processes drain the
same submit ring; each `try_pop` is one CAS, scaling without
serialization.
---
## Worked examples
### Single-process worker pool
```rust
use subetha_cxc::{BackgroundScheduler, Pass, pass_registry};
// Register closures at startup.
pass_registry::register(0x1234, |args| Ok(args.to_vec()));
let sched = BackgroundScheduler::start(
"/tmp/sched-submit.bin",
"/tmp/sched-result.bin",
"/tmp/sched-hb.bin",
1024, // ring capacity
16, // heartbeat slots
).unwrap();
let submitter = sched.submitter();
let collector = sched.collector();
// Submit work.
let token = submitter.submit(&Pass {
closure_id: 0x1234,
args: vec![1, 2, 3, 4],
}).unwrap();
// Receive result (poll).
loop {
if let Ok(r) = collector.try_recv() {
assert_eq!(r.token, token);
break;
}
std::thread::sleep(std::time::Duration::from_micros(100));
}
```
### Cross-process worker pool with coordinator
```rust
// Worker process (multiple of these):
use subetha_cxc::BackgroundScheduler;
let _sched = BackgroundScheduler::start(
"/tmp/sched-submit.bin",
"/tmp/sched-result.bin",
"/tmp/sched-hb.bin",
1024, 16,
).unwrap();
// Worker thread runs autonomously; main thread continues.
std::thread::park(); // or do other coordinator work
// Coordinator process: submit + collect + run watchdog
let sched = BackgroundScheduler::start(
"/tmp/sched-submit.bin",
"/tmp/sched-result.bin",
"/tmp/sched-hb.bin",
1024, 16,
).unwrap();
let submitter = sched.submitter();
let collector = sched.collector();
loop {
submitter.submit(&Pass { closure_id: 0x1234, args: vec![] }).ok();
while let Ok(r) = collector.try_recv() { handle_result(r); }
let report = sched.watchdog_scan();
for (slot, snap) in &report.dead_slots {
eprintln!("dead worker pid={} slot={}", snap.pid, slot);
reassign(snap.in_flight_bitmap);
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
```
---
## Use case patterns
### Pattern: cross-process job queue with failover
Multiple worker processes share submit + result rings. Any
worker drains; a coordinator runs the watchdog. Crashed
workers' in-flight bits are reclaimed within `grace_epochs`.
### Pattern: durable RPC
The submit ring file IS the persistent queue. A coordinator
crash leaves submitted Pass items in the ring; a restarted
coordinator (or any other process) drains them and produces
results. No external queue service needed.
### Pattern: failure-tolerant batch dispatch
Submit N Pass items; track which `result_token`s have been
collected. On worker death (detected via watchdog),
re-submit the unhandled tokens. The result ring carries the
token in the payload so correlation is direct.
---
## Known limitations
- **Args capped at 46 bytes, result capped at 49 bytes**: same
as `SharedRing::PAYLOAD_BYTES = 56` minus header. Larger
payloads need pointer indirection or a side-channel.
- **Worker has 100 us idle-sleep**: bursty workloads pay up to
100 us latency on the first submit after idle. Sustained
submission keeps the worker awake.
- **Result dropped when result-ring is full**: the worker does
not block on result-push. Consumers must drain promptly.
- **One worker per scheduler instance**: scaling requires
multiple `BackgroundScheduler::start` calls (multiple worker
threads in one process, or one per process).
- **Worker beats heartbeat every 100 iterations** of its loop:
on idle, this is one beat every ~10 ms. Tune
`grace_epochs` accordingly.
- **No prioritization**: the submit ring is FIFO. For
priority dispatch, compose with
[PriorityFanout](./PRIORITY_FANOUT.md).
- **Cross-process backed by MMF.**
---
## Common pitfalls
- **Forgetting to register the closure in every participating
process.** The `pass_registry` is per-process; each worker
process must call `register(id, closure)` at startup for
every closure it intends to execute.
- **Not draining the result ring.** The worker drops results
when the result-ring is full; submissions still produce work
but their results are lost. The collector must drain
steadily.
- **Treating `result_token` as a sequence number.** It is a
correlation ID picked monotonically per submitter; different
submitters' tokens may interleave. Use it ONLY to match
results to submissions, not for ordering.
- **Holding the scheduler in a long-running scope without
draining.** The internal worker keeps running but no
external party reads results. Drop the scheduler explicitly
(or wrap it in a tight scope) to signal stop.
- **Wrapping the submitter/collector in a Mutex.** Pointless;
the underlying rings are lock-free MPMC.
---
## References
- Source: `crates/subetha-cxc/src/scheduler.rs` (739 lines, 8 unit
tests including submit-execute-result round-trip,
unknown-closure error path, args-too-large rejection,
heartbeat registration, and 16-message in-order delivery).
- Bench: `crates/subetha-cxc/benches/scheduler.rs` (submit,
try_recv, watchdog_scan vs `mpsc::sync_channel`).
- Underlying primitive: [SHARED_RING.md](./SHARED_RING.md) -
the Vyukov MPMC rings for submit + result.
- Underlying primitive: [HEARTBEAT.md](./HEARTBEAT.md) -
the heartbeat table workers register in.
- Underlying primitive: [PASS_REGISTRY.md](./PASS_REGISTRY.md) -
the in-process closure registry executed by the worker.
- Composes with: [FAILOVER.md](./FAILOVER.md) -
`watchdog_scan` calls `FailoverWatchdog::scan` on the
heartbeat table.
- Composes with: [PROGRESS_TASK.md](./PROGRESS_TASK.md) -
each Pass closure runs to completion; long-running passes
may track their own progress via a separate ProgressTask.