Embedding blockwatcher in a host process

The blockwatcher binary is one way to run the engine: it owns the CLI, instance config, HTTP control plane, metrics listener, and SIGTERM/SIGINT drain. A host that already has a process — a ledger-sync worker, a service that applies matches to its own store — can run the same engine in-process without standing any of that up.

That path is blockwatcher-embed. It is the one place compiled-in module families are registered (crates/blockwatcher-embed/src/catalog.rs), and it re-exports the engine types a host needs to boot and stop. The binary is a client of the same crate for catalog construction; it does not keep a second registrar.

flowchart LR
    subgraph host["host process"]
        embed["blockwatcher-embed<br/>build_catalog · Engine::start"]
        ports["host port impls<br/>Sink · Storage · Source<br/>Decoder · Matcher · Gate"]
        stop["host-owned shutdown"]
        embed --> ports
        stop --> embed
    end

    subgraph bin["blockwatcher binary"]
        embed2["blockwatcher-embed<br/>(same catalog + start)"]
        extra["CLI · config · seed<br/>API · metrics · SIGTERM"]
        extra --> embed2
    end

Key takeaways

• Depend on blockwatcher-embed, not on the blockwatcher binary crate, to run the engine in-process.
• build_catalog is the one registration path; Engine::start boots from a catalog, a storage backend, and an EngineConfig.
• A host can supply an implementation of any of the six ports, not just Sink: register it on the catalog before Engine::start, or, for storage, hand it straight to EngineDeps.
• A host sink implements Sink::deliver(&SinkEvent) in-process and must treat both Match and Retracted as at-least-once (idempotent on match_id).
• Each port’s own obligations are the same in-process as in a shipped module; Extending blockwatcher § What each port asks of a module states them per port.
• The host owns shutdown: call Engine::shutdown. Embed does not wire SIGTERM or SIGINT.
• Embed does not serve the REST API or the Prometheus scrape endpoint. Those stay process concerns of the binary.

The host recipe

Add blockwatcher-embed to the host’s Cargo.toml. Default features (evm, expr, sinks) mirror the binary: they fold the same module families into the catalog. Storage (memory, sqlite) and gate modules (threshold, max_once) are always registered. A host-supplied gate uses the same compile/on_hit contract and still must not open sqlite. A host that implements its own Sink also depends on blockwatcher-ports and blockwatcher-types.

Boot is four steps: build the catalog, construct storage, start the engine, stop it yourself.

#![allow(unused)]
fn main() {
use blockwatcher_embed::{build_catalog, Engine, EngineConfig, EngineDeps};

let catalog = build_catalog()?;
let storage_factory = catalog.storage("memory")?;
let storage = storage_factory(serde_json::json!({}))
    .await
    .expect("memory storage constructs from empty config");

let config: EngineConfig = serde_json::from_value(serde_json::json!({
    "matcher": { "module": "expr", "config": {} }
}))?;

let engine = Engine::start(EngineDeps {
    storage,
    catalog,
    config,
})
.await?;

// Host-owned shutdown: embed does not install signal handlers.
let _report = engine.shutdown().await;
}

EngineConfig has no overall default because it refuses to guess a matcher module. Deserializing a JSON object that names expr (when that feature is on) fills every other tunable from core’s own defaults, including journal_depth = 1024. See Configuration reference § [engine].

Resources (networks, specs, sinks, monitors) still live in storage. A host either writes them through ControlHandle (re-exported from embed) or persists them into the backend before Engine::start, the same shapes the binary’s seed path writes. Embed does not load a seed directory of its own.

Any of the six ports can be the host’s own

A sink is the common case, not the only one. build_catalog returns a catalog the host still owns until EngineDeps takes it by value, and every register_* method on it is public, so a host can add its own module to any port family before start:

Two things do not change by being in-process. Each port’s contract is identical to a shipped module’s, so an in-process source still owes the engine non-decreasing cursor order and an in-process matcher still owes it a bounded compile; Extending blockwatcher § What each port asks of a module states those obligations per port, and they are worth reading before writing the impl rather than after. And a duplicate name within one family is EngineError::DuplicateModule from register_*, reported rather than panicked precisely so a host with no main of its own can recover.

A host that supplies its own module for a family usually also wants to stop linking the shipped one. blockwatcher-embed’s default features are evm, expr, and sinks; turning them off with default-features = false sheds those families, leaving storage (always registered) and whatever the host registers itself. What a build can select is then exactly what it linked, so a name the host never registered is refused at write or boot time listing only the names actually present.

An in-process sink

Shipped sink modules (webhook, script, log) still work: register a sink resource that names one of them, the same as in the binary. A host that wants to apply and undo by MatchId in the same process implements the Sink port itself and registers it on the catalog before start.

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use blockwatcher_ports::{Sink, SinkError};
use blockwatcher_types::SinkEvent;

struct LedgerSink;

#[async_trait]
impl Sink for LedgerSink {
    async fn deliver(&self, event: &SinkEvent) -> Result<(), SinkError> {
        match event {
            SinkEvent::Match(m) => {
                // Apply by m.id. At-least-once: a duplicate id is the same
                // occurrence, not a new one.
                let _ = m;
                Ok(())
            }
            SinkEvent::Retracted { match_id } => {
                // Undo by match_id. Also at-least-once: must be idempotent.
                let _ = match_id;
                Ok(())
            }
        }
    }
}
}

(Sink is one method, crates/blockwatcher-ports/src/sink.rs. The engine owns retry, backoff, and dead-lettering; a sink that retries internally is a bug, the same contract as every shipped module.)

Retracted is how a consumer undoes work that a deep invalidate proved was on a dead fork. Register the impl on the catalog before start, under a name a sink resource can select:

#![allow(unused)]
fn main() {
let mut catalog = build_catalog()?;
catalog.register_sink("ledger", |_config| {
    Box::pin(async move { Ok(Arc::new(LedgerSink) as Arc<dyn Sink>) })
})?;
}

A stored sink resource with "module": "ledger" then constructs this type the same way a catalog sink does. Delivery guarantees covers when retracts fire, the bounded journal they read from, and the ordering that every retract for a sink finishes (delivered or dead-lettered) before any post-restart Match from that network is offered. A host that ignores Retracted will keep orphan rows it can no longer identify by id.

Classify failures through SinkError::Delivery { message, class: ErrorClass::… } exactly as a shipped sink does. The engine’s retry loop reads that class; it does not know this sink is in-process.

Host-owned storage

Storage is the one port a host does not have to register at all. EngineDeps.storage is an Arc<dyn Storage>, so a host that already has a database hands its own backend straight in, and every resource, checkpoint, dead letter, and journal row lands in the host’s store instead of a sqlite file beside the process:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use blockwatcher_ports::Storage;

let storage: Arc<dyn Storage> = Arc::new(LedgerStore::new(pool));

let engine = Engine::start(EngineDeps {
    storage,
    catalog: build_catalog()?,
    config,
})
.await?;
}

catalog.register_storage exists for symmetry, but nothing in the engine resolves storage by name; only the binary does, from [storage] in blockwatcher.toml. A host registers a storage module only if its own configuration should be able to name one.

Storage is also the widest port and the only one with a shared behavioural contract. Take blockwatcher-testkit as a dev-dependency and run exercise_storage_contract against the host backend: it is the same sequence memory and sqlite pass, so passing it is what makes the host’s store equivalent to them rather than merely compiling (blockwatcher-testkit § The storage contract). The rules it checks, optimistic concurrency on every write, unpaginated resource listings, and pruning the journal inside record_delivery, are stated in blockwatcher-storage § The Storage port contract.

A host-owned source

A host that already streams a chain, or that reads from an internal bus a shipped source knows nothing about, implements Source and registers it under a name a network resource can select:

#![allow(unused)]
fn main() {
use blockwatcher_ports::{Source, SourceError};

catalog.register_source("ledger-feed", |config| {
    Box::pin(async move {
        let config: FeedConfig =
            serde_json::from_value(config).map_err(|e| SourceError::InvalidConfig {
                message: format!("invalid ledger-feed config: {e}"),
            })?;
        Ok(Arc::new(LedgerFeed::new(config)) as Arc<dyn Source>)
    })
})?;
}

A stored network resource with "source": { "module": "ledger-feed", … } then gets one instance per network, exactly as evm-rpc would. The obligations are unchanged by the source being local: events go out through ctx.events in non-decreasing cursor order from a single loop, ctx.cancel means return SourceOutcome::Ended, a detected rewind is SourceOutcome::Invalidated { from } rather than an error, and the cursor mapping is the host’s to choose and document. scan and confirmed_tip default to Unsupported, so a streaming-only feed implements neither.

blockwatcher-testkit’s recv and stop are worth pulling in as a dev-dependency here too: they drive the source under a deadline and fail a run that ignores its cancellation token instead of hanging the host’s test suite.

A host-owned decoder

A decoder is the one module a host registers but never names. The engine constructs every registered decoder at boot with an empty config object and indexes them by the ChainKind each chain() returns, so a host decoder becomes reachable purely by storing Spec resources whose chain matches:

#![allow(unused)]
fn main() {
use blockwatcher_ports::{Decoder, SpecError};

catalog.register_decoder("ledger-wire", |_config| {
    Box::pin(async move { Ok(Arc::new(LedgerDecoder) as Arc<dyn Decoder>) })
})?;
}

Two registered decoders claiming the same ChainKind is EngineError::DuplicateChainDecoder at boot, so a host adding a decoder for a chain the build already carries should drop that family’s feature first. The compile-time and hot-path obligations, resolving only the specs a selector names, matching declared field names to the shapes actually decoded, and keeping decode output order deterministic, are where a host decoder most often goes quietly wrong; Extending blockwatcher § Decoder states each with the failure it produces.

A host-owned matcher

A host with its own predicate language implements Matcher, registers it, and names it in EngineConfig instead of expr:

#![allow(unused)]
fn main() {
catalog.register_matcher("host-lang", |_config| {
    Box::pin(async move { Ok(Arc::new(HostMatcher) as Arc<dyn Matcher>) })
})?;

let config: EngineConfig = serde_json::from_value(serde_json::json!({
    "matcher": { "module": "host-lang", "config": {} }
}))?;
}

One matcher runs per process, which is why EngineConfig has no overall default and refuses to guess one. A host that replaces expr should build embed without the expr feature; DEFAULT_MATCHER exists only under that feature, and nothing else fills the field. The obligation most easily missed is that compile takes operator-controlled source: the engine bounds the length it passes in, but the module must bound the work and memory one call can consume, because a compile that can be driven to exhaust a stack or heap is a write-path denial of service.

What embed is not

• Not a second engine API. Engine::start, Engine::shutdown, and ControlHandle are the same types the binary calls. Embed does not wrap them.
• Not a signal handler. The host decides when to drain. Engine::shutdown is the same deadline-bounded drain the binary maps onto its exit codes; without a host call, nothing asks the engine to stop.
• Not the HTTP or metrics stack. blockwatcher-api and blockwatcher-metrics are not dependencies of embed. A host that wants those listeners still runs the binary, or wires those crates itself.
• Not a second registration path for workspace modules. A module that ships in the workspace still becomes selectable through one get_all() line in its own crate, which build_catalog folds, and the binary must not grow its own catalog beside embed’s. A host registering a type it owns is the separate case above: it adds to the catalog build_catalog returned rather than replacing how that catalog is built.

Extending blockwatcher walks a shipped module through registration step by step and states what each of the six ports asks of the module behind it, in § What each port asks of a module; its § Extending to a new network covers the chain-family case, which a host reaches by registering a Source and a Decoder together rather than by adding a crate to the workspace. blockwatcher-embed is the crate page: features, allowlist, and what it re-exports.