Module server 

Shared server utility functions between all dispatcher components

Structs

InFlightSet

The dispatcher’s in-flight (progress) set: the dispatched-but-unfinished tasks, keyed by task id. Phase 4 of the dispatcher rationalization replaces the contended Arc<Mutex<HashMap<i64, TaskProgress>>> with a sharded, lock-free [DashMap] — so the ventilator’s lease (Self::insert), the sink’s return (Self::remove), and the reaper sweep (Self::take_expired) no longer serialise on one global lock — plus an AtomicUsize size counter so the per-request backpressure check (in_flight_saturated) reads the size in O(1) without locking or scanning the map.

RateLimitedLog

Rate-limited logging for high-frequency, low-value events — the dispatcher’s discarded messages (malformed replies/requests, unknown service names, unknown task ids). The dispatcher’s request and sink loops would otherwise warn! one line per skipped message; under a sustained flood (a hostile or buggy peer spamming bad frames) even leveled logging serialises a synchronous write per event and slows the real pipeline — a self-inflicted throughput-DoS (KNOWN_ISSUES D-11). This aggregates instead: it counts events and signals an emit at most once per interval (plus the very first event, so a problem is visible immediately), carrying the suppressed count. So a flood of any size costs O(1) log I/O per interval, not O(flood) — counted, not narrated — while a genuine trickle is still surfaced. Per-thread (no sharing/locking); the clock read per event is a cheap monotonic Instant::now().

ReapSummary

Tally of one reaping pass — the dispatcher’s re-lease / dead-letter health signal (Arm 8 observability). Returned by reap_expired_into so the ventilator can log it without the reaper itself reaching for tracing.

Enums

ExpiredOutcome

The fate of a timed-out in-flight task, decided by classify_expired.

Constants

DONE_QUEUE_CAPACITY

Capacity of the bounded done (results-pending-persist) channel between the producers (sink + ventilator reaper) and the finalize thread. A full channel blocks the producer — that is the backpressure (a slow DB makes the sink wait, which backs up the ZMQ PULL, which backs up the workers), replacing the old DONE_QUEUE_HARD_LIMIT panic-then-OOM backstop with a graceful, loss-free hand-off (a bounded channel blocks rather than drops). Phase 1 of the dispatcher rationalization (docs/archive/DISPATCHER_RATIONALIZATION.md).

MAX_DISPATCH_RETRIES

The maximum number of dispatch retries before a perpetually-incomplete task is given up on. A task re-dispatched this many times that still never returns a result is treated as a hard failure (Fatal) rather than retried forever.

PROGRESS_QUEUE_HARD_LIMIT

Hard ceiling on the in-flight (progress) set. Reaching it means backpressure (crate::config::DispatcherConfig::max_in_flight) failed to hold the line, so we fail fast (panic → process abort → external restart) rather than exhaust memory — the dispatcher’s intentional fail-fast design. Backpressure must engage below this (asserted in tests).

Statics

SHUTDOWN_REQUESTED

Graceful-shutdown request flag (O-1, orchestration). Set by the SIGTERM/SIGINT handler the dispatcher binary installs; the ventilator checks it each loop iteration and, when set, signals completion (dispatch_complete) and returns — so the manager drains the in-flight set and flushes the finalize batch before exiting, rather than the supervisor hard-killing in-flight work. ONLY a planned stop uses this; unexpected failures still fail-fast (panic → abort). Production-only — bounded test runs never install the handler, so this stays false and the dispatch path is byte-for-byte unchanged.

Functions

apply_tcp_keepalive

Applies TCP keepalive to a worker-facing ZMQ socket (the ventilator’s ROUTER, the sink’s PULL), the “stable” half of a remote worker interface: keepalive keeps idle worker connections alive across NAT/firewall idle-timeouts (an idle mapping is otherwise silently dropped, dropping the worker from the fleet until it reconnects) and lets the OS reap a dead peer’s route. It does not affect task-recovery correctness — the lease reaper is that net — so idle_seconds <= 0 safely disables it (OS default). Set before bind so accepted connections inherit it.

classify_expired

Decides what to do with an in-flight task that timed out: retry it (until MAX_DISPATCH_RETRIES dispatches) or give up and report it Fatal.

get_sandbox_id

Getter for a corpus’s sandbox id from the shared SandboxCache, with no DB access — the sink’s read on the result path. An absent entry yields None (treat as an ordinary corpus); the ventilator always memoises a task’s corpus on dispatch, before its result can return, so the entry is present by the time the sink looks.

get_service

Getter for a Service from the shared ServiceCache, with no DB access (a None entry means a prior lookup found no such service).

get_sync_sandbox_id

Memoised getter for a corpus’s sandbox id, populating the shared SandboxCache from the backend on a miss (the ventilator’s DB-holding counterpart to get_sandbox_id). One lookup per corpus_id ever dispatched — bounded by the corpus count, not the task count.

get_sync_service

Memoized getter for a Service record from the backend, populating the shared ServiceCache on a miss. The or_insert_with holds only the relevant DashMap shard during the one-time DB lookup (vs. the old whole-map Mutex), so concurrent lookups of other services are unblocked.

in_flight_saturated

Whether the in-flight set is saturated and the ventilator should apply backpressure (stop leasing new work and mock-reply). Saturation is inclusive: at the threshold we already hold back, keeping the set bounded below PROGRESS_QUEUE_HARD_LIMIT.

install_shutdown_handlers

Installs SIGTERM + SIGINT handlers that request a graceful drain-and-stop. Call once from the dispatcher binary’s main (never in bounded test runs).

mark_done_batch

Persists a batch of finished reports to the Task store, with bounded retry on a transient DB failure. The finalize thread drains the batch off the bounded done channel and hands it here in one mark_done call (amortizing the round-trip). On exhausted retries it returns Err, which the finalize thread propagates into a panic → the manager aborts the whole dispatcher — the intended fail-fast on a DB runaway (the channel design replaces the old mutex-poisoning that achieved the same propagation). A crash here loses nothing: the tasks remain Queued and recover on restart.

reap_expired_into

Reaps timed-out in-flight tasks and routes each to its own service’s dispatch queue (a retry) or the done queue (a Fatal). Decoupled from the refetch path so the in-flight set drains even under sustained backpressure (KNOWN_ISSUES D-6); routing by task.service_id rather than the requesting service fixes the latent cross-service requeue bug. Returns a ReapSummary of what it did (re-leases vs dead-letters) for the health log.

send_done

Hands a finished report off to the finalize thread over the bounded done channel. A full channel blocks the producer (backpressure — see DONE_QUEUE_CAPACITY). An Err means the finalize receiver is gone (the thread died) — the report can’t be persisted, but its task stays Queued and is recovered on restart, and the manager’s supervision aborts the dispatcher, so we only log.

shutdown_requested

Whether a graceful shutdown has been requested (set by install_shutdown_handlers’s handler).

Type Aliases

SandboxCache

The dispatcher’s sandbox cache: a corpus_id → Option<sandbox_id> memo, the corpus-keyed twin of ServiceCache. Some(id) means the corpus is a sandbox (Arm 5) whose result archives are name-scoped by its own id; None means an ordinary corpus. Sandbox-ness is immutable per corpus (a corpus is born sandbox-or-not and never flips), so a one-time DB lookup per corpus_id is correct forever — no invalidation. The ventilator memoises it on dispatch (it holds a Backend), so the sink can scope a result’s output path with a lock-free read — no per-result DB hit — keeping a sandbox rerun from overwriting the parent’s archives (KNOWN_ISSUES F-6).

ServiceCache

The dispatcher’s service cache: a service_name → Option<Service> memo shared by the ventilator (populated on a cache miss) and the sink (read on every result), so a Service row is looked up from the DB at most once per name. Phase 4 of the dispatcher rationalization replaces the old Arc<Mutex<HashMap>> with a sharded DashMap, so this near-static, read-mostly cache is no longer behind a single global lock contended on every dispatch.