blockwatcher-storage

blockwatcher-storage implements the Storage port twice over: versioned, optimistic-concurrency resources, per-network checkpoints, an append-only dead-letter queue, persisted operator pause, a bounded delivery journal, and the gate hit journal (gate_hits / gate_meta), backed by an in-memory map or a durable sqlite file (crates/blockwatcher-storage/src/lib.rs). It is a module crate: the heavy driver (rusqlite) is quarantined here, and nothing above the Storage port ever has to know which backend a running instance chose.

Delivery guarantees § Dead letters already states the operator-facing consequence of the choice between backends: sqlite is what makes a dead-letter queue survive a restart at all, memory does not. This page covers the mechanics underneath that: the Storage contract in full, each backend’s concrete data structures, sqlite’s schema and migration approach, and exactly how optimistic concurrency and dead-letter payload absence are implemented: verified against the schema and the SQL itself, not restated at the operator level.

Key takeaways

• blockwatcher-storage implements the Storage port twice over: an in-memory backend with no persistence, and a durable sqlite-backed one.
• The heavy driver (rusqlite) is quarantined in this crate; nothing above the Storage port ever has to know which backend a running instance chose.
• sqlite is what makes a dead-letter queue survive a restart at all; memory does not.

Responsibilities

• Implement the full Storage port over an in-memory HashMap-backed structure with no persistence at all: MemoryStorage (memory.rs), including the delivery journal as a per-network Vec.
• Implement the full Storage port over one sqlite file and one guarded connection, durable across a restart: SqliteStorage (sqlite.rs).
• Enforce optimistic concurrency on every resource write and delete: a mismatched expected version is a typed VersionConflict, never a silent overwrite, in both backends (memory.rs, sqlite.rs).
• Apply put_batch’s create-only, all-or-nothing contract atomically in both backends, overriding the port’s own non-atomic default loop (memory.rs, sqlite.rs).
• Register both backends under the module names "memory" and "sqlite" so a composition root’s catalog can select either by config (registry.rs).
• (sqlite only) Create its schema idempotently on first open, migrate schema version 1 → 2 (delivery journal table), 2 → 3 (persisted pause), and 3 → 4 (gate_hits / gate_meta) in place, refuse to open a file written by a newer schema version, and serialize every connection access through one mutex inside spawn_blocking so the async runtime never blocks on file I/O: open_blocking (sqlite.rs).

Not this crate’s job: declaring the Storage trait itself, or any of the types it moves (Checkpoint, DeadLetter, VersionedRecord, ResourceKind): those are blockwatcher-ports and blockwatcher-types; deciding when a checkpoint is safe to persist, or retrying a failed persist: blockwatcher-core’s CheckpointWriter owns that policy and calls this crate’s backends only to execute the write it already decided to make; deciding that a dead letter should be discarded outright, singly or in bulk, rather than replayed: that is ControlHandle::discard_dead_letter/ discard_dead_letters in blockwatcher-core, which this crate’s delete_dead_letter/delete_dead_letters only execute; pruning the delivery journal: backends self-prune behind journal_depth on each record_delivery, and there is no operator prune API; picking which backend a deployment runs: that is blockwatcher.toml, read by blockwatcher-core’s module catalog.

Key types and traits

Neither backend defines its own error type: both return blockwatcher_ports::StorageError (VersionConflict, AlreadyExists, NotFound, Backend { message, class }, InvalidConfig), and neither backend’s Config struct is pub (reachable only through its ModuleRegistry::factory), documented via its own registry_examples/*.json file rather than a public type.

The Storage port contract

Every method below is declared once, in crates/blockwatcher-ports/src/storage.rs, and both backends implement every one of them:

Every method on the trait belongs to one of these categories: resources (keyed by ResourceKind × id, with a u64 version for optimistic concurrency), checkpoints (keyed by NetworkId), dead letters (keyed by NetworkId + MatchId), pause (keyed by PauseTarget + id), the delivery journal (keyed by NetworkId + MatchId, ordered by cursor), and the gate journal (gate_hits / gate_meta, keyed by pipeline + monitor). Memory, sqlite, and FlakyStorage all implement it; exercise_storage_contract covers the gate methods too.

Both backends persist the identical categories; only where they land differs:

flowchart LR
    subgraph categories["categories"]
        res["resources<br/>kind + id keyed, versioned"]
        chk["checkpoints<br/>one cursor per network"]
        dl["dead letters<br/>append-only queue per network"]
        pause["pause<br/>target + id keyed"]
        journal["delivery journal<br/>match ids in a cursor window"]
    end
    categories --> mem["MemoryStorage<br/>HashMap, gone on restart"]
    categories --> sql["SqliteStorage<br/>one file, survives restart"]

The memory backend

MemoryStorage holds one Mutex<State> (memory.rs), where State is one plain Rust collection per category and nothing else (memory.rs):

#![allow(unused)]
fn main() {
#[derive(Default)]
struct State {
    records: HashMap<(ResourceKind, String), (u64, serde_json::Value)>,
    checkpoints: HashMap<NetworkId, Checkpoint>,
    /// Appended to, never reordered: operators triage dead letters in the
    /// order delivery gave up on them.
    dead_letters: HashMap<NetworkId, Vec<DeadLetter>>,
    /// Per-pipeline delivered match ids, newest write prunes by primary.
    journal: HashMap<NetworkId, Vec<(Cursor, MatchId)>>,
    /// Present means paused; a resume removes the entry.
    paused: BTreeSet<(PauseTarget, String)>,
}
}

Every method is a lock, a map operation, and an unlock, with no I/O anywhere in the file (module doc, memory.rs). Optimistic concurrency is a plain match over (expected_version, current) in put (memory.rs); put_batch pre-validates every entry against both existing state and the rest of the batch before writing any of it, under one lock acquisition, which makes it genuinely atomic here despite the trait’s own default being a non-atomic loop (memory.rs). Dead letters are a real, ordered Vec<DeadLetter> per network (append-only except for delete_dead_letter/delete_dead_letters and update_dead_letter’s targeted mutations, memory.rs), not a stub: paging (skip/take), removal, and in-place replacement all behave exactly as the port contract specifies. Nothing here is written to disk; a process restart loses every record, checkpoint, dead letter, and pause, matching the crate’s own trade-off note (memory.rs).

The sqlite backend

Schema

One CREATE TABLE/CREATE INDEX batch per schema version (SCHEMA_V1, SCHEMA_V2, SCHEMA_V3, SCHEMA_V4 in sqlite.rs), with SCHEMA_VERSION 4:

CREATE TABLE resources ( … );
CREATE TABLE checkpoints ( … );
CREATE TABLE dead_letters ( … );
CREATE INDEX dead_letters_pipeline ON dead_letters (pipeline, seq);

CREATE TABLE delivery_journal (
  pipeline         TEXT NOT NULL,
  cursor_primary   INTEGER NOT NULL,
  cursor_secondary INTEGER NOT NULL,
  match_id         TEXT NOT NULL,
  PRIMARY KEY (pipeline, match_id)
);
CREATE INDEX delivery_journal_pipeline_primary
  ON delivery_journal (pipeline, cursor_primary);

CREATE TABLE pauses (
  target TEXT NOT NULL,
  id     TEXT NOT NULL,
  PRIMARY KEY (target, id)
);

CREATE TABLE gate_hits (
  pipeline         TEXT NOT NULL,
  monitor          TEXT NOT NULL,
  cursor_primary   INTEGER NOT NULL,
  cursor_secondary INTEGER NOT NULL,
  event_index      INTEGER NOT NULL,
  event_ts         INTEGER NOT NULL,
  event_json       TEXT NOT NULL,
  PRIMARY KEY (pipeline, monitor, cursor_primary, cursor_secondary, event_index)
);
CREATE INDEX gate_hits_pipeline_primary ON gate_hits (pipeline, cursor_primary);
CREATE TABLE gate_meta (
  pipeline      TEXT NOT NULL,
  monitor       TEXT NOT NULL,
  last_emit_ts  INTEGER,
  aux           BLOB,
  PRIMARY KEY (pipeline, monitor)
);

Each table stores its whole record as one JSON TEXT blob (value, checkpoint, entry) alongside only the columns a query needs to filter or order by, pauses excepted: a pause carries no payload beyond its own existence, so target and id are the entire row and presence alone means paused. resources keys on (kind, id); checkpoints keys on pipeline; pauses keys on (target, id); dead_letters gets an autoincrementing seq that is both its primary key and, via the dead_letters_pipeline index, the ordering list_dead_letters pages by: arrival order, for free, from the column sqlite already maintains.

Migration approach

A single idempotent function run once at open (open_blocking, sqlite.rs): MIGRATIONS is an ordered array holding each version’s schema batch ([SCHEMA_V1, SCHEMA_V2, SCHEMA_V3, SCHEMA_V4]), and SCHEMA_VERSION is simply its length, so adding a version and bumping the version number can never drift apart from each other. Opening reads PRAGMA user_version and slices MIGRATIONS from that index onward: a fresh file (0) runs every step, a 1 file runs only SCHEMA_V2 onward, a 2 file runs only SCHEMA_V3 onward, a 3 file runs only SCHEMA_V4, and a file already at SCHEMA_VERSION runs nothing. Every pending step executes inside one transaction, PRAGMA user_version is set to SCHEMA_VERSION inside that same transaction, and the whole thing commits together, so a crash mid-migration can never leave the file at its old version with some of the new tables already present, a state this backend would otherwise treat as fresh and fail to re-create. A version outside 0..=SCHEMA_VERSION (including a negative value, which this backend never writes) refuses to open at all, before any write, naming the found version (sqlite.rs), a refusal pinned by a test that diffs the file’s bytes before and after it to prove nothing was touched.

A future bump means appending one more schema batch to MIGRATIONS, not introducing a new subsystem or a new arm to a hand-written match.

Optimistic concurrency, precisely

put’s update path (sqlite.rs:256-276) is a single conditional UPDATE:

UPDATE resources SET version = version + 1, value = ?4
WHERE kind = ?1 AND id = ?2 AND version = ?3

If rows_affected comes back 0, a follow-up SELECT version tells the two possible causes apart: no row at all (NotFound) versus a row that exists at a different version (VersionConflict { expected, actual }). delete (sqlite.rs) is the identical pattern with a conditional DELETE. A create (expected_version: None) is a plain INSERT, whose rusqlite::Error for a primary-key collision is mapped to AlreadyExists (sqlite.rs). put_batch serializes every entry to JSON before opening a transaction (so a serde failure never leaves one open), then runs one INSERT per entry inside it, dropping the transaction uncommitted on any failure (sqlite.rs), which is what makes it atomic where the port’s own default loop is not.

Dead letters and checkpoint provenance are inside the blob, not a column

record_dead_letter/update_dead_letter serialize the whole DeadLetter struct (match_id, monitor, sink, cursor, attempts, reason, and the optional payload) into the single entry TEXT column (sqlite.rs). A query that needs to find one letter by match_id reaches into that blob with sqlite’s own json_extract, rather than a dedicated column:

DELETE FROM dead_letters
WHERE pipeline = ?1 AND json_extract(entry, '$.match_id') = ?2

(sqlite.rs:454-480, comment noting this scan is accepted at triage-scale, with a dedicated index left for whichever deployment first proves it necessary). Because DeadLetter.payload: Option<SinkEvent> is #[serde(default, skip_serializing_if = "Option::is_none")] (crates/blockwatcher-types/src/event.rs), a dead letter recorded before replay support existed has no SQL NULL to check at all: its entry blob simply has no "payload" key, and serde_json::from_str fills that gap back in as None on read via #[serde(default)]. A payload that is a tagged SinkEvent or a legacy bare Match object both load; see Delivery guarantees § What a sink receives. The 409 checkpoint_provenance check the HTTP API documents (which module wrote a checkpoint) works the same way: Checkpoint.module: Option<String> lives inside the checkpoints.checkpoint JSON blob (sqlite.rs), and this crate stores and returns it opaquely: the module comparison itself happens one layer up, in the engine that reads the decoded Checkpoint back out.

Cursor representation

Cursor { primary: u64, secondary: u64 } never gets its own columns in either backend: in MemoryStorage it is a plain, unserialized struct field inside the in-memory Checkpoint; in SqliteStorage it exists only as the nested {"primary": N, "secondary": N} object inside the whole Checkpoint’s serialized JSON blob, the same shape the HTTP API’s checkpoint field shows on the wire.

Connection handling

SqliteStorage holds Arc<Mutex<rusqlite::Connection>>: one connection, not a pool (sqlite.rs). Every port method funnels through a shared with_conn helper that clones the Arc, locks the mutex, and runs the blocking sqlite call inside tokio::task::spawn_blocking, so the async runtime is never blocked on file I/O even though the connection itself serializes every access (sqlite.rs). with_conn also takes a read guard on a tokio::sync::RwLock<()> fence before spawning, and moves the guard into the blocking closure rather than merely holding it across the .await: if the caller awaiting with_conn is aborted, the closure keeps running on the blocking pool regardless, and the guard travels with it, so it still marks the call as in flight until the closure actually returns. SqliteStorage’s Storage::quiesce override takes the fence’s write side and immediately drops it, which a fair, write-preferring RwLock resolves only once every read guard taken before it has been dropped: this is what lets a caller that just escalated a drain to abort() wait out whatever storage call that abort left running, rather than risk a successor racing it. PRAGMA journal_mode=WAL is set at open for a file-backed database (skipped for :memory:, which cannot run WAL), and busy_timeout is set from the configured busy_timeout_ms (default 5_000) to soften contention from an external reader, not to support two blockwatcher processes writing the same file, which this backend does not support at all (sqlite.rs).

Neighbours

blockwatcher-storage depends on, in production:

• blockwatcher-types: vocabulary crate (Checkpoint, DeadLetter, MatchId, NetworkId, PauseTarget, ResourceKind, VersionedRecord)
• blockwatcher-ports: the Storage trait this crate implements twice
• async-trait: required to implement the async fn-bearing Storage trait
• serde: (de)serialization derive for each backend’s Config
• serde_json: the JSON representation every stored value, checkpoint, and dead letter serializes to
• tokio (rt feature): spawn_blocking for the sqlite backend’s connection access
• rusqlite: the sqlite driver itself, this crate’s one check-dep-graph.sh family exemption

and, in [dev-dependencies] only:

• blockwatcher-ports (fakes feature): blockwatcher_ports::fakes::MemoryStorage, run through the same shared contract test as this crate’s own two backends, for parity
• blockwatcher-testkit: dead_letter (a fixture builder) and exercise_storage_contract, the shared behavioral-contract suite both backends are proven against
• tokio (macros/rt/time/test-util features): async tests
• tempfile: a real filesystem path for sqlite’s file-backed tests

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

• blockwatcher-embed: folds both backends into the catalog build_catalog returns
• blockwatcher (binary): reaches storage both directly and through embed

Reading the source

1. Start at lib.rs: three pub mod declarations and nothing else; the module doc names the categories this crate persists.
2. registry.rs: storages::get_all, and the self-verifying test that constructs every registered backend from its own documented example config.
3. memory.rs: MemoryStorage, State, and every Storage method in file order; read this first, since every method here is the simplest possible correct implementation of the same contract sqlite.rs implements durably.
4. sqlite.rs: SCHEMA_V1/SCHEMA_VERSION, open_blocking (the migration logic), then every Storage method; read the module doc comment first for the single-writer trade-off this backend accepts.
5. tests/contract.rs (workspace-shared, via blockwatcher-testkit): the one suite both backends (and the ports fakes::MemoryStorage) run through, proving they satisfy an identical contract by shared test rather than shared code.