Struct Reactor 

pub struct Reactor { /* private fields */ }

The reactor — the composition’s wired reactive halves, constructed from the substrate engine handle + the registry write side + the cached provider (the assembly-level pieces; the wave-8 composition entry unifies the whole backend’s construction).

The registry write side is held for the wake runner’s and the reconcile sweep’s refresh reads (rows_since) — the pair the engine-twin adapters implement, so the reactor wiring is engine-generic (it ran unchanged over all three engines in POC-R).

Implementations

impl Reactor

pub fn new( store: Arc<dyn Store>, registry: Arc<dyn RegistryWrite>, provider: Arc<CachedRegistryProvider>, ) -> Self

Construct the reactor from the assembly’s pieces.

pub fn store(&self) -> &Arc<dyn Store>

The substrate engine handle (the wave-7 maintenance chain rides the same handle).

pub fn registry(&self) -> &Arc<dyn RegistryWrite>

The registry write side (the refresh source).

pub fn provider(&self) -> &Arc<CachedRegistryProvider>

The cached provider (the invalidation target).

pub async fn announce(&self, event: &RegistryEvent) -> AnnounceReport

The announce composition — the one call a writing side makes after its durable row-write transaction commits: publish (EVENTS_STREAM, keyed by repo id), notify (WAKE_CHANNEL), and enqueue (MAINTENANCE_QUEUE, a pack-housekeeping job for the mutated repo) — each leg independent, each fail-soft. The legs’ outcomes are report data (AnnounceReport); this call never fails, and any leg that failed is closed by the reconcile sweep.

A suppression (the process dying between the durable commit and this call, or mid-legs) is the same posture — the sweep closes the window in one pass. On postgres the notify leg is load-bearing (the module docs’ asymmetry): its loss leaves caches stale until the reconcile tick.

pub async fn enqueue(&self, job: &MaintenanceJob) -> Result<i64>

The enqueue leg as a direct producer API — one housekeeping job on MAINTENANCE_QUEUE. The consumer side (the claim → leader lock → dispatch chain) is wave 7’s; the job’s payload (MaintenanceJob) is that worker’s dispatch vocabulary.

pub async fn reconcile( &self, applied: &AppliedWatermark, ) -> Result<Vec<RegistryRow>, RegistryError>

The reconcile sweep — the closure for any lost post-commit effect: rows_since(applied_watermark) → snapshot install → watermark advance. Idempotent by construction (pure reads, a monotone watermark — the second pass is empty). Driven by the scheduler tick once the maintenance chain’s consumer side lands (wave 7), callable on demand. Returns the rows it caught up on (empty = nothing missed). A failed sweep changes nothing — the watermark does not advance and the previous snapshot serves; the next wake, write, or tick retries.

pub async fn run_wake_runner(&self, stop: StopToken) -> Result<()>

The wake runner — the invalidation trigger’s delivery leg: listen([WAKE_CHANNEL]) → one whole-table refresh (CachedRegistryProvider::refresh_from) per wake. Coalesced bursts collapse engine-side (the sqlite watcher’s burst coalescing; per-notify on postgres) and one refresh carries a whole burst — the count of refreshes is correct by the snapshot swap. There is no polling at this layer: no commits → no refreshes (the refresh count is stable across idle windows). Refresh failures leave the previous snapshot serving (the next wake, write, or tick retries).

Returns when the source closes — a closed receiver never reopens; recovery is a fresh listen (the runner attached again; the reconcile tick is the staleness bound in between).

pub async fn run_stream_consumer( &self, consumer: &str, applied: &AppliedWatermark, stop: StopToken, on_event: impl FnMut(RegistryEvent, StreamEvent) + Send, ) -> Result<()>

The stream consumer — the durable change feed’s consumer side under the alkstore cursor model: subscribe(consumer) attaches at the consumer’s stored checkpoint (absent = 0 = the log head — a fresh consumer replays from the head; a restarted one resumes after the saved checkpoint); each observation is checkpointed after the apply (monotone saves — saves cannot regress); per-seq idempotency is applied as the consumer’s rule — an event whose seq is at or below the applied watermark is checkpointed past without re-invoking the callback.

on_event receives the decoded event and its raw feed position. An undecodable payload is checkpointed past (a foreign producer must not wedge the cursor). A storage failure backs off and retries — the feed is durable, silence loses nothing. Returns on the receiver’s terminal close (recovery is a fresh re-attach — the saved checkpoint resumes) or a stopped token; only a checkpoint save failure fails the run (the durable cursor is the restart boundary — losing it silently would widen the replay), and re-attaching retries it from the last landed checkpoint.