blockwatcher-core

blockwatcher-core is blockwatcher’s engine: it assembles a running pipeline out of a network’s already-constructed source, its chain’s decoder, and the engine’s one matcher; tracks every dispatched match until it is either delivered or dead-lettered; decides the checkpoint that tracking makes safe to persist; and is the one path (ControlHandle) by which a resource write, delete, pause, dry run, or dead-letter replay reaches a running deployment. It is the largest crate in the workspace, and this is its largest page.

Its production dependencies are exactly the names scripts/check-dep-graph.sh’s ALLOW_BLOCKWATCHER_CORE entry lists:

• blockwatcher-types: the vocabulary every resource, event, and identifier this crate moves is expressed in
• blockwatcher-ports: the port traits spawn’s tasks drive (Source, Decoder, Matcher, Gate, Sink, Storage)
• serde: the derive machinery behind every (de)serializable shape in the crate
• serde_json: every resource’s on-the-wire and in-storage JSON representation
• thiserror: the derive behind EngineError (error.rs)
• tokio (macros/rt/time features): the async runtime every task, channel, and timer in the crate runs on
• tokio-util: CancellationToken, the primitive spawn’s whole cancellation topology (cancel, hard_cancel, drain_token) is built from
• futures: FutureExt::catch_unwind, used by the source task wrapper to catch a panicking source without taking the pipeline down with it
• metrics: the facade every named constant in metrics.rs emits through
• tracing: every structured log line in the crate

No chain SDK, HTTP client, or storage driver appears anywhere in this list or its transitive tree. blockwatcher-core also carries no family exemption at all in that same script’s family_exemptions_for (family_exemptions_for in scripts/check-dep-graph.sh): nothing forbidden may appear anywhere in its transitive tree under any named carve-out. See Architecture decisions § Chain knowledge stays out of the core and Workspace map § Verifying the rings for the rule this crate is one example of.

The pipeline and Delivery guarantees already describe this crate’s pipeline at an operator’s level: one task per stage, the backpressure it enforces, the guarantee a checkpoint carries. This page goes one level deeper: every file inside pipeline/ and control/ mapped to what it owns, the exact channel and cancellation-token wiring spawn builds, and the checkpoint writer’s own retry state machine. Where the two levels overlap, this page links to the concept page rather than restating it.

Key takeaways

• blockwatcher-core is the engine: it assembles one pipeline per network from a source, a decoder, the one matcher, and an optional per-monitor gate, and is the only path (ControlHandle) through which a resource write reaches a running deployment.
• No chain SDK, HTTP client, or storage driver appears anywhere in its dependency tree, and it carries no family exemption at all.
• It is the largest crate in the workspace, depending only on blockwatcher-types and blockwatcher-ports among workspace crates.
• This page goes one level deeper than the pipeline and delivery-guarantees concept pages: every file inside pipeline/ and control/, the exact channel and cancellation wiring spawn builds, and the checkpoint writer’s own retry state machine.

Responsibilities

• Assembles and runs one pipeline per network: a single function, pipeline::spawn, constructs every channel and cancellation token a pipeline needs and then starts a source task, a processor task, one sink worker task per sink any of that network’s monitors reference, and a checkpoint writer task on top of them (crates/blockwatcher-core/src/pipeline/mod.rs). See Pipeline workers below for how every file in pipeline/ maps to one of those tasks.
• Owns the whole control plane: Engine::start/Engine::shutdown for boot validation and a deadline-bounded drain, and ControlHandle for every resource write, delete, pause/resume, dry run, checkpoint skip, and dead-letter replay a running deployment accepts (engine.rs, control/). See The control plane.
• Tracks every dispatched match until it is delivered or dead-lettered, and derives from that the one cursor a network’s pipeline may safely persist, through Progress/CompletionGuard (progress.rs) and CheckpointWriter (pipeline/checkpoint_writer.rs). See Progress and checkpointing.
• Compiles a monitor’s selectors and predicate, and a chain’s specs, against the live decoder and the engine’s one matcher instance, once at write time, never on the event path (compile.rs, compiled.rs).
• Names every module family a config can select from and reports an actionable refusal, listing every registered alternative, for a name nothing registered (catalog.rs).
• Fans every pipeline event out to two independent observability destinations: the atomic PipelineCounters a status read serves with no cardinality cost (counters.rs), and the metrics facade constants an exporter may or may not be listening to (metrics.rs). See blockwatcher-metrics for the side of that pair which actually exports anything.
• Reads and type-validates every stored resource before anything downstream ever sees it (resources.rs), and reports a point-in-time snapshot of every running, paused, or abandoned pipeline (status.rs).

Not this crate’s job: deciding what a chain’s wire format means, or implementing any source, decoder, sink, or storage module: blockwatcher-evm, blockwatcher-storage, and blockwatcher-sinks each own exactly one, behind the port traits blockwatcher-ports declares; running the predicate language itself (blockwatcher-expr: this crate only calls Matcher::compile/matches through the port); exposing anything over HTTP (blockwatcher-api wraps ControlHandle in routes) or Prometheus (blockwatcher-metrics installs the recorder this crate’s metrics.rs module writes into, and serves the scrape endpoint); defining the port traits it drives or the vocabulary types it moves (blockwatcher-ports, blockwatcher-types).

Key types and traits

Pipeline workers

pipeline/ holds one file per pipeline stage plus the glue between them: nothing in the directory is shared infrastructure unrelated to running one network’s pipeline:

progress.rs sits one directory up from all of them, deliberately: Progress is not a task, it is a shared, lock-protected structure the processor registers events into and sink workers complete guards against. See Progress and checkpointing.

Pipeline worker topology

spawn (pipeline/mod.rs) builds every channel, watch, and cancellation token a pipeline needs before it spawns a single task, and every task below is handed only the subset it actually touches:

flowchart TD
    handle["PipelineHandle<br/>(held by Engine)"]

    subgraph one["one spawn() call"]
        src["Source task<br/>ctx.events (Sender)<br/>ctx.interest / ctx.status (watch)<br/>ctx.cancel"]
        proc["Processor task<br/>reads events_rx<br/>reads set_rx (watch)<br/>writes Progress.begin<br/>writes sink senders"]
        sw["Sink worker task<br/>(one per sink)<br/>reads its own WorkItem receiver<br/>completes guards via Progress<br/>cancel = hard_cancel"]
        cpw["Checkpoint writer task<br/>reads checkpoint_rx (watch)<br/>reads drain_token<br/>cancel = hard_cancel"]
        prog[("Progress<br/>shared struct, not a task")]
    end

    src -->|"events_tx: mpsc(event_channel_capacity)<br/>pipeline/mod.rs"| proc
    proc -->|"sink_tx: mpsc(sink_channel_capacity)<br/>one per sink, pipeline/mod.rs"| sw
    proc -->|"begin(cursor, outstanding)"| prog
    sw -->|"guard.complete()"| prog
    prog -->|"checkpoint_rx: watch&lt;Option&lt;Checkpoint&gt;&gt;"| cpw
    cpw -->|"put_checkpoint"| store[("checkpoint<br/>blockwatcher-storage")]

    handle -->|"set_tx.send_replace<br/>(hot swap)"| proc
    handle -->|"interest_tx.send_replace<br/>(hot swap)"| src
    src -->|"status_tx.send_replace"| handle

    parent(("parent_cancel<br/>engine root")) -.->|".child_token()"| src
    hard(("hard_cancel<br/>independent root,<br/>only a deadline<br/>escalation fires it")) -.-> sw
    hard -.-> cpw
    proc -.->|"Arc&lt;DrainSignal&gt; canary<br/>dropped on task exit"| drain{{"drain_token"}}
    sw -.->|"canary dropped<br/>on every worker's exit"| drain
    drain -.->|"cancelled() once<br/>every canary has dropped"| cpw

    classDef token fill:none,stroke:#e39a3a
    class parent,hard,drain token

Two details that are easy to miss reading the concept-level diagram alone:

• The bounded mpsc channels are the only queues in this diagram. Every other arrow is a tokio::sync::watch (publishes only the latest value, never queues several) or a direct call against Progress’s own mutex-guarded state. See Backpressure below.
• hard_cancel is not a descendant of parent_cancel, and the source task never holds it at all. Only the sink workers and the checkpoint writer check it; only a deadline escalation inside drain_pipeline (engine/drain.rs, see Boot, restart, and shutdown) ever fires it. If the pipeline’s own ordinary shutdown signal (cancel) triggered it directly instead, a sink worker or the checkpoint writer would give up on work still sitting in a bounded channel that an un-escalated drain, given the time it is normally owed, would have gone on to flush, per spawn’s own comment (pipeline/mod.rs).
• The drain-signal canary is held by the processor and every sink worker, never by the source. DrainSignal’s Drop impl fires drain_token once every clone of it has dropped (checkpoint_writer.rs); the checkpoint writer’s own Progress clone, and PipelineHandle’s, would otherwise keep Progress’s watch channel open forever, since neither ever drops for the life of the handle. Excluding the source is deliberate too: a source exiting has nothing to do with whether the processor and sink workers still have buffered work to finish draining.

Backpressure: the bounded channels

Between source and processor sits one tokio::sync::mpsc channel of capacity event_channel_capacity (default 256); between processor and each sink worker sits one of capacity sink_channel_capacity (default 64), both configured once in EngineConfig (config.rs) and created inside spawn (pipeline/mod.rs). These are the only bounded channels a pipeline has. Nothing else in the topology above can exert backpressure: Progress’s internal VecDeque<Slot> (progress.rs) has no bound at all (it grows with however many events are in flight and shrinks as their guards complete), and every watch channel (set_tx/set_rx, interest_tx/interest_rx, status_tx/status_rx, checkpoint_rx) holds at most one pending value by construction: a send_replace overwrites whatever was there, it never queues a second one behind it. A hot-swapped monitor set or a republished checkpoint can never itself pile up; only the mpsc channels above can.

Both channel operations are ordinary tokio::sync::mpsc::Sender::send calls with nothing wrapped around them: no try_send, no timeout, no fallback path (the sink dispatch call site is processor.rs). That is the entire mechanism: a full channel just makes that one .await pend. The pipeline § Backpressure: two bounded channels, no drops walks through where that pending call propagates to (the processor’s whole dispatch loop, then the source’s own send) and Delivery guarantees covers why a stall is the accepted outcome rather than a drop. This page’s own contribution is the inventory above: the bounded points it names are the only ones in the entire topology, and their configured capacities are the only backpressure knobs EngineConfig exposes: nothing else in spawn’s wiring can ever be the thing that is full.

Progress and checkpointing

Delivery guarantees already names Progress as the mechanism deciding when a cursor becomes safe to persist; this section is the data structure and algorithm underneath that decision. Progress’s state (State, progress.rs, held inside Inner alongside the watch Sender, progress.rs) is a VecDeque<Slot> behind one Mutex: each Slot (progress.rs) carries one raw event’s cursor, its source_state, and a remaining count of matches dispatched from it that have not yet completed. begin (progress.rs) pushes one Slot per raw event, in the order the processor calls it, and nothing ever removes a slot except from the front. advance_locked (progress.rs) is the entire algorithm on top of that queue: walk forward from the front for as long as Slot::remaining is zero, drop every slot that check passes, and, only if at least one slot was dropped, call send_replace with the cursor of the last one. Running the walk and the send_replace inside the same critical section is what rules out two concurrent completions racing each other into the watch channel in the wrong order; a version that unlocked between the two could have one thread publish cursor 3 and a second thread, scheduled just behind it, publish cursor 1 afterward, and nothing would ever notice or correct it since the slots behind both are already gone.

CompletionGuard::complete (progress.rs) is Slot::remaining -= 1 followed by the same advance_locked call; dropping a guard without calling complete (a worker task dying mid-delivery, most notably) leaves that slot’s remaining nonzero forever, which permanently stalls the prefix at that slot rather than expiring or timing out. Delivery guarantees § The contract, precisely covers what that guarantee means to a consumer; the mechanism above is what it compiles down to.

CheckpointWriter (checkpoint_writer.rs) is the other half: a task per network that watches Progress’s publish channel and persists whatever it sees, forever retrying a failed write against the same value. Persisting whatever the watch currently holds, rather than a queued history of values, is safe specifically because of how advance_locked publishes: since every publish happens while holding the same lock a later publish must also acquire, a later value can never be an earlier cursor than one already sent. One rule this task adds on top of Progress’s own monotonicity: persist (checkpoint_writer.rs) tracks last_persisted within this writer instance’s own lifetime and refuses (counts, logs, and skips) any checkpoint older than that, independent of what Progress would otherwise publish. That guard exists for a source that violates its own non-decreasing-cursor contract mid-run; it is explicitly not enforced across a restart, because a fresh writer’s first persist is unconditional: the mechanism Boot, restart, and shutdown below relies on to make a legitimate rewind (a positively detected reorg, a resumed process) possible at all.

stateDiagram-v2
    [*] --> Waiting
    Waiting --> Waiting: sample tick (every 1s)<br/>gauge IN_FLIGHT_EVENTS
    Waiting --> CheckRegression: checkpoint_rx.changed()<br/>or drain fires<br/>(borrow_and_update reads the latest value)
    CheckRegression --> Waiting: cursor behind last_persisted<br/>refused, counted<br/>(CHECKPOINT_REGRESSIONS_REFUSED)
    CheckRegression --> Persisting: cursor at or past last_persisted
    Persisting --> Persisted: storage.put_checkpoint Ok<br/>last_persisted = cursor
    Persisted --> Waiting: loop, unless drain already fired
    Persisted --> [*]: drain fired, nothing left to flush
    Persisting --> Backoff: storage.put_checkpoint Err<br/>counted (CHECKPOINT_WRITE_FAILED)
    Backoff --> Persisting: backoff elapses<br/>(doubles, capped at max_backoff_ms)
    Backoff --> Abandoned: hard_cancel fires<br/>(deadline escalation only)
    Abandoned --> [*]
    Waiting --> [*]: drain fired,<br/>nothing pending to flush

Two states are worth naming precisely against the source: CheckRegression is persist’s own comparison against last_persisted (checkpoint_writer.rs), and it is the only place a value is discarded rather than eventually written: a discard here is never silent, it increments PipelineCounters::checkpoint_regressions_refused and the matching metrics.rs constant, and logs at error level naming both cursors. Abandoned is reachable only from Backoff, and only because hard_cancel fired: the writer never gives up on a value it has already decided to persist for any other reason, including the pipeline’s own ordinary cancel token, which this task does not even hold a reference to.

The control plane

blockwatcher-core’s own module comments draw the same distinction this section does: control/ is “the caller-facing half — it owns the validate/compile/ persist ordering, the optimistic concurrency, and the mutation lock” (engine/control.rs:4-6), while engine/control.rs (a different file, one level up from control/) is “the engine-side half,” owning what happens to the running pipeline once a mutation has already validated and persisted its change (engine/control.rs:1-2).

control/: the caller-facing half

One file per method family on ControlHandle:

Every mutating method here follows the same order, stated once in control/mod.rs’s own module doc comment and never varied: “validate references, compile or construct whatever needs it, persist with optimistic concurrency, then update whatever is actually running — in that order, always, for every mutating method, writes and deletes alike” (control/mod.rs:65-69). A write or delete that has already persisted is treated as done regardless of what happens next: whatever running state it still has to touch (a republished monitor set, a restarted pipeline) is attempted best-effort, and a failure there is logged with the resource and reason rather than turned into an error response for a resource storage already durably holds.

Pause is persisted

Both pause surfaces, ControlHandle::pause_monitor/resume_monitor and pause_network/resume_network (control/mod.rs), persist the pause to storage before touching anything running, and every path that builds a monitor set or boots a pipeline reads that persisted state back:

• Pausing a monitor writes its id to storage’s paused-monitor set (ControlHandle::set_paused, control/mod.rs), then republishes the network’s running monitor set through build_monitor_set with that same persisted set (Engine::republish_paused_monitors, engine/control.rs). The persist is the request: a pipeline that is not running, or a republish that fails, is logged and left for the next rebuild, which reads the persisted set from scratch and always applies it — the same path a network, sink, or spec write’s restart takes, and the same one boot itself seeds from (engine/boot.rs). Deleting a monitor clears its pause row with it (control/deletes.rs).
• Pausing a network persists the pause (Storage::set_paused), marks it in Engine.paused_networks (a HashSet<NetworkId>, engine.rs) so a concurrent SourceExit cannot race a restart back into existence, then stops its pipeline through the same bounded drain a restart uses. Resuming clears the persisted row and restarts the pipeline unless one is already running for it; a restart that then fails is logged rather than returned, since the persist already succeeded (resume_network, control/mod.rs). Boot reads the persisted set before constructing any pipeline and never spawns one for a paused network at all, intersected against the networks boot actually built so a pause row that outlived its network cannot render a status entry for one that no longer exists (engine.rs’s constructor). The restart supervisor consults the same in-memory mark (is_paused) and the delete tombstone (is_tombstoned) before ever scheduling a restart of its own, so an operator’s pause is never silently undone by a source exiting and the supervisor bringing it back (engine/supervisor.rs). The tombstone is a HashMap of network id to the sink ids the last pipeline for that network was spawned with, so a late Invalidated after delete still retracts only those sinks. Deleting a network clears its pause row with it.

A monitor’s pause additionally drops out of the source’s merged interest hint the moment it is paused: build_monitor_set excludes a paused monitor’s selector from the interest union it hands the decoder (compile.rs), so pausing a monitor can only ever shrink what its network’s source is asked to fetch, never leave it fetching data nothing will read.

Boot, restart, and shutdown

engine.rs plus its engine/ submodules are the other side of the control plane: what an engine does on its own behalf (boot, shutdown, restart supervision) rather than what a caller’s mutation asks of it.

Two things a reader coming from pipeline/mod.rs’s spawn should carry into this section: restart_pipeline’s replacement pipeline is built to completion before the existing one is touched, so a fallible step (a storage error, a module that no longer constructs) always leaves whatever was running before completely untouched; and the replacement’s checkpoint is deliberately read after the old pipeline has drained, never before. A pre-drain read would miss whatever further completions land while the old pipeline winds down, so the replacement would come up already behind its own predecessor and, on its first persist, push the stored cursor backward (engine/control.rs, doc comment). This is the one place in the whole crate a restart is allowed to make a checkpoint “rewind” in CheckpointWriter’s sense above: the fresh writer’s first persist is unconditional, precisely so this legitimate case is not caught by the same guard that refuses a source’s own contract violation.

Neighbours

blockwatcher-core depends on, in production (the same list named in full above, and no chain SDK, HTTP client, or storage driver anywhere in this list or its transitive tree (see the exemption-free guarantee described above):

• blockwatcher-types
• blockwatcher-ports
• serde
• serde_json
• thiserror
• tokio (macros/rt/time features)
• tokio-util
• futures
• metrics
• tracing

and, in [dev-dependencies] only:

• blockwatcher-ports (fakes feature): FakeSource, FakeDecoder, FakeMatcher, FakeSink, MemoryStorage, FlakyStorage; every test in this crate runs against these, never a real module
• blockwatcher-testkit: RecordingMetrics, used to assert on what a pipeline flow emits through the metrics facade
• async-trait: implementing Source/Sink test doubles (PanickingSource, CancelThenOkSource, SlowSink) that blockwatcher-ports::fakes doesn’t already provide
• tokio (rt-multi-thread/test-util features): multi-threaded and paused-time async tests

The following crates depend on it directly (per the dependency table):

• blockwatcher-api: wraps every ControlHandle method in an HTTP route
• blockwatcher-embed: calls Engine::start for in-process hosts and for the binary’s own boot
• blockwatcher (binary): owns the process’s own shutdown signal, and calls Engine::validate for its offline config-check mode

Reading the source

1. Start at lib.rs: the module list and the full re-export surface; every name in the key types table above is pub re-exported here (CheckpointWriter is the one type this page covers in depth that is not: mod pipeline; has no pub, so nothing outside blockwatcher-core can name it, pub struct on the type itself notwithstanding).
2. error.rs: EngineError, every variant naming the identity of what it happened to; read this before anything else, since almost every fallible function in the crate returns it.
3. config.rs: EngineConfig and SourceRestart, and their defaults (event_channel_capacity: 256, sink_channel_capacity: 64, drain_deadline_ms: 10_000, journal_depth: 1024).
4. catalog.rs: ModuleCatalog, and the family! macro that generates one register/lookup pair per port family so registering, say, a sink factory as a source is a compile error.
5. compiled.rs, then compile.rs: CompiledMonitor/MonitorSet, and the three write-time functions that build them: compile_monitor (selectors, predicate, and gate), build_monitor_set (read its doc comment on field-union poisoning closely: one un-introspectable monitor’s predicate makes the whole network’s referenced-fields hint unknown, on purpose), and spec_set_for_chain.
6. resources.rs: Resources/Versioned<T>, and why a listing (load) fails whole on one bad record while a by-id read (load_one) fails only the lookup that names it.
7. counters.rs, then metrics.rs: PipelineCounters/CounterSnapshot, then the named Prometheus constants and the emission helpers (count, count_sink, count_gate, gauge) every pipeline stage calls alongside its matching atomic increment. Gate metric names (blockwatcher_gate_hits_total and the rest) are the same strings Observability lists. Invalidate, retract, and journal-gap counters live here too; they have no PipelineCounters twin. See blockwatcher-metrics § From a pipeline event to a scrape line for where the second half of that pair actually goes.
8. progress.rs: Progress/CompletionGuard/Slot/advance_locked; read this before pipeline/checkpoint_writer.rs, which depends on everything it guarantees.
9. pipeline/mod.rs, then processor.rs, sink_worker.rs, delivery.rs, checkpoint_writer.rs, in that order. See Pipeline workers above for what each one owns.
10. engine.rs: Engine, EngineDeps, ShutdownReport, and the mod declarations for its own submodules; read Engine::start and Engine::shutdown here before descending into engine/boot.rs, engine/control.rs, engine/drain.rs, engine/supervisor.rs, and engine/invalidate.rs. See Boot, restart, and shutdown above for what each one owns.
11. control/mod.rs, then writes.rs, deletes.rs, inspect.rs, skip.rs, dead_letters.rs. See The control plane above for what each one owns, and read mod.rs’s module doc comment first: it states the validate-persist-swap ordering and the post-persist contract every one of the other files holds to.
12. status.rs: SourceStatusView, QueueDepth, PipelineStatus, EngineStatus, and the small pure functions (head_of, lag_between, queue_depth_with_capacity) Engine::status_snapshot composes them from.