The pipeline

How it works in one picture sketched a single pipeline. This page opens it up: which task does what, which crate implements it, where the bounded queues sit, and (the part that only reading crates/blockwatcher-core/src/pipeline/ settles) exactly what is shared across a deployment and what is private to one network.

flowchart LR
    rpc["RPC endpoints<br/>(external chains)"]
    sources["Sources<br/>evm-rpc · evm-mempool"]
    sinks["Sinks<br/>webhook · script · log"]
    storage["Storage<br/>checkpoints · dead letters · resources"]
    api["REST API"]
    metrics["Metrics"]
    engine["engine<br/>bounded channels · checkpoints"]

    subgraph pipeline["Engine pipeline"]
        direction LR
        decoder["Decoder"]
        matcher["Matcher<br/>predicates"]
        gate["Gate<br/>threshold · max_once"]
        decoder --> matcher
        matcher --> gate
    end

    rpc --> sources
    sources -->|"decode and match"| decoder
    gate --> sinks
    api -->|"manages resources"| storage
    storage <--> engine
    engine -->|"drives"| pipeline
    engine -.->|"reports"| metrics

    classDef module fill:none,stroke:#a9a3e3
    classDef core fill:none,stroke:#8a8d86,stroke-dasharray: 5 5

    class rpc,sources,sinks,decoder,matcher,gate module
    class engine,api,storage,metrics core
classDef dim fill:none,stroke:#999999,color:#999999,opacity:0.35
classDef focus fill:#ffd43b,stroke:#d9480f,stroke-width:3px,color:#1a1a1a
class rpc,sources,sinks,storage,api,metrics,decoder,matcher,gate dim
class engine focus
linkStyle 6,7,8 stroke:#d9480f,stroke-width:3px
click sources "selectors.html"
click decoder "chain-agnosticism.html"
click matcher "predicates.html"
click gate "gates.html"
click sinks "delivery.html"
click storage "resources.html"
click api "../reference/http-api.html"
click metrics "../reference/observability.html"
click engine "pipeline.html"

Key takeaways

• spawn turns one network’s PipelineSpec into four task kinds (source, processor, sink worker, checkpoint writer) linked end to end by channels.
• The decoder is shared per chain and the matcher is one instance for the whole engine; only the source, monitor set, and sink workers are per-network.
• The two bounded channels never drop a message: a full channel stalls the pipeline at the point of congestion rather than losing anything.
• A raw event registers with the progress tracker exactly once, after dispatch, for however many matches it produced, even zero.
• After a predicate-true decode, an optional per-monitor gate may Retain or Discard the hit with outstanding 0, or Emit a match/digest. Holding a hit does not stall the checkpoint.
• A cursor only ever persists once every completion guard the event handed out, across every monitor and sink it touched, has completed.

One pipeline per network

Engine::start builds one PipelineSpec per stored network and spawns it once, and every later network write goes through the same spawn call by way of a restart (crates/blockwatcher-core/src/pipeline/mod.rs). A PipelineSpec bundles that network’s already-constructed source, its chain’s decoder, the engine’s one matcher instance, the network’s compiled monitor set, and every sink any of those monitors names (crates/blockwatcher-core/src/pipeline/mod.rs). spawn turns that bundle into the task kinds below, wired together by channels it owns end to end.

flowchart LR
    src["Source task<br/>blockwatcher-evm"] -->|"events channel<br/>(bounded)"| proc
    proc["Processor task<br/>decode + match + gate<br/>blockwatcher-core + blockwatcher-evm + blockwatcher-expr"]
    proc -->|"sink channel<br/>(bounded, per sink)"| sw["Sink worker task<br/>blockwatcher-core"]
    proc -.->|register cursor| prog["Progress tracker<br/>blockwatcher-core"]
    sw -.->|complete / dead-letter| prog
    prog -->|checkpoint| cpw["Checkpoint writer task<br/>blockwatcher-core"]
    sw -->|on exhaustion| dl[("dead letters<br/>blockwatcher-storage")]
    cpw --> store[("checkpoint<br/>blockwatcher-storage")]

    classDef task fill:none,stroke:#a9a3e3
    class src,proc,sw,cpw task

Each task kind, and the crate doing the real work inside it:

• Source task: one per network, running whatever module the network’s source.module names (evm-rpc or evm-mempool, both in crates/blockwatcher-evm). It owns its own notion of position, its cursor, and must hand events to the next stage in non-decreasing cursor order, the contract Processor::run states for its receiver; nothing downstream reorders what it reads (crates/blockwatcher-core/src/pipeline/processor.rs).
• Processor task: one per network, decoding, matching, and gating in the same loop. It calls the network chain’s Decoder (the evm decoder in crates/blockwatcher-evm, the only one shipped) once per active, unpaused monitor’s selector, then evaluates that monitor’s compiled predicate through the engine’s single Matcher instance (blockwatcher-expr’s expr matcher, unless a deployment swaps it); see Processor::process_one (crates/blockwatcher-core/src/pipeline/processor.rs). Gate arithmetic lives in pipeline/gate.rs, not inlined in processor.rs.
• Sink worker task: one per sink a network’s monitors reference, running the sink module (webhook, script, or log) named by that sink’s SinkDef. It owns delivery retry and backoff and dead-lettering, identical regardless of which module is attached, per SinkWorker’s own doc comment (crates/blockwatcher-core/src/pipeline/sink_worker.rs).
• Checkpoint writer task: one per network, persisting whatever the progress tracker publishes and refusing to persist anything older than what it already wrote this run, in CheckpointWriter’s persist method (crates/blockwatcher-core/src/pipeline/checkpoint_writer.rs).

The Progress tracker (crates/blockwatcher-core/src/progress.rs) is not a task; it is a shared, lock-protected structure the processor registers events into and sink workers complete guards against, read by the checkpoint writer. It is covered in full on Delivery guarantees; this page treats it as the thing that turns “every dispatched match done” into “safe to persist this cursor.”

What is shared, and what is per-network

The pipeline diagram makes every task look network-scoped, but two of the module families it touches — the decoder and the matcher — are actually shared engine-wide, constructed once at boot and reused rather than rebuilt per network or per pipeline restart:

The two easiest to get wrong: the decoder is per chain, not per network (two evm networks decode through the exact same Arc<dyn Decoder>), and the matcher is a single engine-wide instance, not something a monitor or a network picks; blockwatcher.toml names it once for the entire process, as Engine’s matcher field doc comment states (crates/blockwatcher-core/src/engine.rs). A sink’s module instance, by contrast, is rebuilt every time a pipeline is (re)constructed: it is not shared even across two restarts of the same network, though this is invisible in practice because a sink module carries no state beyond its config.

Backpressure: two bounded channels, no drops

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 blockwatcher.toml’s engine section and applied to every pipeline alike, EngineConfig’s fields, read by spawn (crates/blockwatcher-core/src/config.rs, crates/blockwatcher-core/src/pipeline/mod.rs). Neither channel ever drops a message when full. A source’s send and the processor’s dispatch are plain awaited .send() calls, in Processor::process_one (crates/blockwatcher-core/src/pipeline/processor.rs); when a channel is at capacity, that await simply doesn’t resolve until a slot opens, which means a slow sink worker’s channel filling up stalls the processor’s dispatch loop, which in turn stalls the source’s own send: the whole pipeline pauses at the first point of congestion rather than losing anything. The two channels close only when their sending side is dropped, which is what a graceful shutdown or restart uses to drain: cancel the source, its sender drops, the processor’s receive loop ends and drops the sink senders, and each sink worker drains whatever is already queued before it exits, the cascade spawn’s own doc comment states (crates/blockwatcher-core/src/pipeline/mod.rs).

Escalation, abort, and the storage fence

Every drain (a whole-engine shutdown, and a single network’s stop or restart through the control plane) is bounded by drain_deadline_ms. Missing that deadline escalates: a hard_cancel signal fires, tasks that are still running get a brief grace window to exit on their own, and anything still not done after that is stopped with a forced abort() (crates/blockwatcher-core/src/engine/drain.rs). An escalation like this is always loud (a warning naming the network) and always counted (blockwatcher_pipelines_aborted_total).

An abort() can leave more behind than a stopped task: a storage backend whose operations run independent of the caller that started them (sqlite’s own connection lives behind a tokio::task::spawn_blocking call, for example) keeps running whatever it was asked to do at the moment of the abort, and that call still lands later, on its own schedule. Without accounting for it, a pipeline restarted right after such an escalation could read its resume point, or its gate journal, before that residual write lands, then start delivering against what it read, only for the residual write to land afterward and overwrite what the restart’s own work just persisted: a gap, not a duplicate.

Storage::quiesce closes that window: after any escalated drain, the engine awaits it, bounded by its own budget, before treating the network as safely stopped. It resolves once every storage operation an aborted task had in flight has actually completed. One function, stop_pipeline, is where every single-network stop and restart drains through, so this quiescing, its budget, and its counting all live in that one place regardless of which control-plane path called it. Three paths read what it decides before touching storage themselves, and all three refuse in the same shape when it times out, leaving the network abandoned (visible in /status the same way a failed crash-restart already is) until an operator retries or the process restarts:

• a restart, before re-reading the checkpoint and gate outbox it would otherwise redeliver or overwrite against;
• a gate envelope change or a gated monitor’s delete, before dropping that monitor’s held journal, which a residual write could otherwise resurrect after the drop; and
• a reorg invalidation, before pruning gate holds, retracting journaled deliveries, or rewinding the checkpoint, any of which a residual write could otherwise race.

Every one of these refusals is counted under blockwatcher_pipelines_quiesce_timeout_total exactly once, at stop_pipeline itself, no matter which of the three paths called it or how many times. A timeout is not forgiven just because a later attempt on the same network finds no pipeline left running to escalate: the engine remembers the debt and re-attempts the same quiesce, under the same budget, on that later attempt too, refusing again for as long as the residual write stays outstanding and clearing the debt only once some attempt actually waits it out.

A whole-engine shutdown facing the same timeout only warns and counts it instead of refusing anything, since the process is exiting either way and a later process starts with its own storage connections, which the residual write from this one cannot reach.

From one raw event to zero or more deliveries

Inside Processor::process_one, run once per raw event (crates/blockwatcher-core/src/pipeline/processor.rs):

1. An event addressed to a different network is dropped and counted (misrouted) rather than processed: a pipeline never touches an event that isn’t its own, because attributing it would corrupt this pipeline’s own cursor ordering.
2. The current monitor set is snapshotted once for the whole event, not once per monitor, so a hot-swapped monitor set landing mid-event can never split that event’s processing between an old and a new set.
3. For every active, unpaused monitor, the decoder runs against that monitor’s compiled selector. Zero, one, or several decoded events can result (an EVM log’s Transfer and a batched transaction’s several decoded calls are both possible); each one that decodes is evaluated against the monitor’s predicate (or accepted outright if it has none).
4. Every monitor that accepts a decoded event offers that occurrence to its gate (or, with gate omitted, treats it as Emit of this hit). Retain / Discard mint nothing. Emit mints one match per chosen journal index (one SinkEvent::Match, or one Digest if two or more) and queues one work item per sink that monitor’s actions name.
5. Once every pair from this one raw event is known, the event is registered with the progress tracker for exactly that many outstanding completions: registered exactly once per event, after dispatch, never before and never twice. A gate that retained or discarded adds zero to that count for that monitor (crates/blockwatcher-core/src/pipeline/processor.rs).

An event that matched nothing (no selector, or predicate false, or every gate quiet) still gets registered, with zero outstanding, which is what lets its cursor advance immediately rather than wait on deliveries that were never dispatched.

Delivery, retry, and the checkpoint’s dependency on it

Each sink worker pulls one work item at a time and calls deliver_with_retry (crates/blockwatcher-core/src/pipeline/delivery.rs): a transient or rate-limited failure sleeps a doubling backoff and tries again, up to the sink’s configured max_attempts; a permanent or range-narrower failure skips straight to giving up. Giving up means recording a dead letter in storage (retried forever on its own backoff if that write itself fails), because a checkpoint may only move past a loss storage has durably recorded, never past one it merely intends to record, the ordering SinkWorker::dead_letter’s own doc comment states (crates/blockwatcher-core/src/pipeline/sink_worker.rs). Only once delivery succeeds or the dead letter is durably written does the sink worker complete its guard. The checkpoint writer, in turn, only ever persists a cursor once every guard the processor handed out for it (across every monitor and every sink that event’s matches touched) has completed, via Progress and the CompletionGuard it hands out (crates/blockwatcher-core/src/progress.rs). Delivery guarantees covers this mechanism itself in depth, along with the one shipped source, evm-mempool, whose cursor is not a chain position and so cannot make the same promise across a crash.