blockwatcher-testkit

blockwatcher-testkit is shared test scaffolding: a recording metrics recorder and a handful of small harnesses that let a module’s tests drive a port the same way every other module’s tests do, instead of each crate growing its own copy of “wait for this condition” or “assert the storage contract holds.” Its whole lib.rs doc comment states the constraint the rest of this page is built on: “Dev-dependency only — the dependency gate rejects any production edge onto this crate by name” (crates/blockwatcher-testkit/src/lib.rs:1-2). See Workspace map § Verifying the rings for exactly how scripts/check-dep-graph.sh checks that by name rather than folding it into the ordinary allowlist rule.

Its production dependencies:

• blockwatcher-types: RawEvent (source_run.rs) and the resource/checkpoint/ dead-letter vocabulary storage_contract.rs writes and reads
• blockwatcher-ports: Storage/StorageError (storage_contract.rs) and SourceError (source_run.rs), the port traits this crate’s harnesses drive
• metrics: the Recorder/Counter/Gauge/Key types recorder.rs implements against
• serde_json: the JSON values storage_contract.rs’s exercise writes and compares
• tokio (time/rt features): the timers wait.rs and source_run.rs await on
• tokio-util: CancellationToken, the type source_run.rs’s stop cancels and asserts the Source port’s cancellation contract against

No [dev-dependencies] section exists in crates/blockwatcher-testkit/Cargo.toml at all: this crate’s own unit-tested modules (endpoint_url.rs, recorder.rs) exercise themselves with nothing beyond what production already pulls in.

Key takeaways

• blockwatcher-testkit is shared test scaffolding: a recording metrics recorder and small harnesses that let a module’s tests drive a port the same way every other module’s tests do.
• It is dev-dependency only; the dependency gate rejects any production edge onto this crate by name.
• Its harnesses take a real or fake port implementation as an argument and drive it; they never substitute for a port directly.

Responsibilities

Small files, each owning exactly one harness:

• endpoint_url.rs: publish_endpoint_url, which writes a URL into a freshly claimed environment variable and hands back the env:NAME reference an evm-rpc endpoint’s url_secret expects, so a test can point a source at a mock node without writing that node’s ephemeral URL into seed JSON.
• recorder.rs: RecordingMetrics, a metrics::Recorder that records every counter emission and every counter/gauge registration instead of exporting them, so a test can assert an emission fired with the labels it claims, or that a module registered the metrics it documents even when a code path never drove them.
• source_run.rs: recv and stop, for driving a running Source from a test: taking its next event with a bounded wait, and cancelling it while asserting the port’s cancellation contract (a cancelled run returns promptly and returns Ok(()), never an error, never a panic).
• storage_contract.rs: exercise_storage_contract, the one behavioral contract every Storage implementation in the workspace, the ports fake included, must pass, plus dead_letter, a fixture builder for the DeadLetter value that exercise and any other test needs.
• wait.rs: until, the bounded poll every suite that waits on a condition shares, so a broken property fails its own test rather than hanging the suite.

Not this crate’s job: implementing a real module, deciding what a port’s contract means (blockwatcher-ports defines the traits and error enums this crate’s harnesses drive; this crate only exercises them), or substituting for a port directly. blockwatcher-ports’ fakes feature (FakeSource, FakeDecoder, FakeMatcher, FakeSink, MemoryStorage, FlakyStorage) and testing feature (mockall-generated mocks) are the in-memory implementations and unit-level substitutes a test builds against; see blockwatcher-ports § Responsibilities. This crate’s own harnesses take a real or fake port implementation as an argument or a type parameter and drive it, rather than being one.

Key types and functions

The metrics recorder

RecordingMetrics installs thread-locally via metrics::set_default_local_recorder, which is why its own doc comment warns that tests using it must run on a current-thread tokio runtime: “the default #[tokio::test] flavor” (recorder.rs:1-9). An emission from a worker thread of a multi-thread runtime never reaches it. Its Recorder impl (recorder.rs:94-124) is deliberately asymmetric across the three metric kinds metrics defines:

#![allow(unused)]
fn main() {
impl Recorder for RecordingMetrics {
    fn describe_counter(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
    fn describe_gauge(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
    fn describe_histogram(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}

    fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> Counter {
        let name = key.name().to_string();
        let labels = labels_of(key);
        let mut state = self.state.lock().expect("recording metrics lock");
        state.registrations.push((name.clone(), labels.clone()));
        drop(state);
        Counter::from_arc(Arc::new(CounterHandle {
            name,
            labels,
            state: Arc::clone(&self.state),
        }))
    }

    fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> Gauge {
        self.state
            .lock()
            .expect("recording metrics lock")
            .registrations
            .push((key.name().to_string(), labels_of(key)));
        Gauge::noop()
    }

    fn register_histogram(&self, _: &Key, _: &Metadata<'_>) -> Histogram {
        Histogram::noop()
    }
}
}

Only a counter’s handle actually records a value (CounterHandle::increment/absolute, recorder.rs): a gauge’s registration is captured, but every gauge handle this recorder ever hands back is Gauge::noop(), and every histogram (registration included) is dropped entirely. count’s own doc comment names the one caveat that follows from folding absolute into increment: “mixing increment and absolute on one counter yields a sum, not the counter’s true value — this double aggregates every event, it does not track counter semantics” (recorder.rs:46-51).

blockwatcher-core (pipeline/mod.rs, tests/engine/restart.rs), blockwatcher-rpc (src/pool.rs), blockwatcher-sinks (tests/sinks/engine_delivery.rs), and blockwatcher-evm (tests/mempool_loop.rs, tests/reorg_and_failover.rs) each import RecordingMetrics to assert on what their own pipeline or pool emits through the metrics facade, in place of a real Prometheus backend.

Driving a source: recv and stop

Both functions in source_run.rs exist because the two obvious alternatives (a bare recv().await and a bare abort()) can hang a whole suite instead of failing one test. recv wraps the receiver in a 10-second tokio::time::timeout (source_run.rs); stop cancels the supplied token, then asserts, within the same deadline, that the Source’s run future returns and returns exactly Ok(()) (source_run.rs:31-41):

#![allow(unused)]
fn main() {
pub async fn stop(
    cancel: CancellationToken,
    handle: tokio::task::JoinHandle<Result<(), SourceError>>,
) {
    cancel.cancel();
    tokio::time::timeout(DEADLINE, handle)
        .await
        .expect("run did not return promptly after cancel")
        .expect("run task panicked")
        .expect("run returned an error instead of Ok(()) after cancel");
}
}

blockwatcher-evm’s mempool_loop.rs, reorg_and_failover.rs, and source_loop.rs all import both functions to drive evm-rpc and evm-mempool sources against a mock node.

The storage contract

exercise_storage_contract takes a factory (Fn() -> Future<Output = Arc<dyn Storage>>) rather than one instance, and calls it once per section (storage_contract.rs), because its own module doc comment states the precondition every section depends on: “fresh must return an empty store on every call” (storage_contract.rs:1-3). The sections it runs, in order, cover optimistic-concurrency create/update/delete and its rejection paths, ResourceKind namespacing and sorted listing, per-network checkpoint round-tripping, dead-letter insertion-order paging, and, its most pointed section, values and cursors that a typed SQL column could not hold at all: a u64 above 2^53, non-ASCII text, a nested JSON null, and a checkpoint cursor at u64::MAX, “bigger than i64::MAX, the ceiling of SQLite’s signed 64-bit INTEGER type” (storage_contract.rs:317-324). blockwatcher-storage’s tests/contract.rs runs this same exercise against the ports fake, its own in-memory module, and its sqlite module in turn, so all three are proven equivalent by a shared test rather than by shared code; see blockwatcher-storage § Neighbours for that crate’s side of the same import.

Bounded waiting: until

wait.rs’s own doc comment states the shared budget: a 10ms poll interval against a 60-second deadline, “generous, because it covers a loaded machine that may be compiling at the same time” (wait.rs:1-13). The wait between polls is a real tokio::time::sleep, never a yield_now spin, which is what lets a #[tokio::test(start_paused = true)] test’s virtual clock auto-advance past whatever the condition is waiting on (wait.rs). blockwatcher-api’s tests/api/helpers.rs, blockwatcher-core’s tests/engine/helpers.rs, and the blockwatcher binary’s tests/boot.rs and tests/escalation_seam.rs each wrap this same until behind their own crate-local convenience function rather than calling it inline at every call site.

blockwatcher-e2e needs the identical shape but cannot depend on this crate at all (its own allowlist entry is empty, see blockwatcher-e2e), so its tests/e2e/harness.rs hand-writes a second until rather than importing this one, quoting a captured process’s stderr into its panic message the way a black-box scenario needs to. publish_endpoint_url is duplicated the same way, for the same reason, in tests/e2e/harness.rs’s write_chain_seed (crates/blockwatcher-e2e/tests/e2e/harness.rs, doc comment).

Neighbours

Every consumer reaches for a different subset of this crate’s harnesses:

flowchart LR
    testkit["blockwatcher-testkit"] --> core["blockwatcher-core<br/>RecordingMetrics, until"]
    testkit --> api["blockwatcher-api<br/>until"]
    testkit --> rpc["blockwatcher-rpc<br/>RecordingMetrics"]
    testkit --> sinks["blockwatcher-sinks<br/>RecordingMetrics"]
    testkit --> storage["blockwatcher-storage<br/>exercise_storage_contract"]
    testkit --> evm["blockwatcher-evm<br/>RecordingMetrics, recv, stop"]
    testkit --> bin["blockwatcher binary<br/>publish_endpoint_url, until"]

blockwatcher-testkit depends on, in production:

• blockwatcher-types
• blockwatcher-ports
• metrics
• serde_json
• tokio (time/rt features)
• tokio-util

The following crates depend on it, every edge under [dev-dependencies] only (per the dependency table, and checked by name rather than by the general allowlist rule, per Workspace map § Verifying the rings):

• blockwatcher-core: RecordingMetrics (pipeline/mod.rs’s own tests, tests/engine/restart.rs) and until (tests/engine/helpers.rs)
• blockwatcher-api: until (tests/api/helpers.rs)
• blockwatcher-rpc: RecordingMetrics (src/pool.rs’s own tests)
• blockwatcher-sinks: RecordingMetrics (tests/sinks/engine_delivery.rs)
• blockwatcher-storage: dead_letter and exercise_storage_contract (tests/contract.rs)
• blockwatcher-evm: RecordingMetrics, recv, stop (tests/mempool_loop.rs, tests/reorg_and_failover.rs, tests/source_loop.rs)
• blockwatcher (binary): publish_endpoint_url and until (tests/boot.rs, tests/escalation_seam.rs)

Reading the source

1. lib.rs: the module list, the crate’s whole re-export surface (one name or pair per module), and the doc comment stating the dev-dependency-only constraint the gate enforces by name.
2. wait.rs: until, the smallest file and the one every other harness’s own tests lean on indirectly through the pattern it establishes.
3. source_run.rs: recv, then stop; read the module doc comment first, since both functions exist to turn a possible hang into a bounded failure.
4. storage_contract.rs: exercise_storage_contract’s section calls, then each section function in the order it calls them; dead_letter at the bottom of the file is the one fixture builder the sections above it also share.
5. recorder.rs: RecordingMetrics, CounterHandle, and the Recorder impl; the module doc comment states the thread-local caveat before anything else.
6. endpoint_url.rs: publish_endpoint_url; short enough to read in one pass, and its doc comment is also the fullest explanation in the crate of why an ephemeral mock node’s URL has to travel through the environment rather than through seed JSON.