Crate mnesis_store_testing

Expand description

The executable conformance kit for mnesis-store adapters — and the guide to writing one (issue #281).

Every store adapter (mnesis-inmemory, mnesis-fjall, mnesis-postgres, and any future one) implements the same seam: RawEventStore + WakeSource, optionally AtomicAppend and SnapshotStore. That seam has a contract that goes well beyond “the trait compiles” — inclusive vs. exclusive read bounds, optimistic-conflict rejection, subscription catch-up→live ordering, concurrent-writer linearizability. This crate pins that contract as runnable checks, organized into four cross-cutting categories (CLAUDE.md rule 7) plus two opt-in capability modules:

sequence — multi-step protocol: append/read round-trips, optimistic conflict + retry, $all ordering and resume, subscription catch-up-then-live.
boundary — defensive inputs: version gaps, wrong first version, metadata absent vs. present, max-length event types, prefix-colliding stream ids.
linearizability — concurrent writers: single-winner on a contended stream, all-land on distinct streams, wake-after-idle, the CaughtUp-boundary race.
lifecycle — close → reopen (persistent adapters only): events, the $all watermark, and conflict state all survive a reopen.
atomic (feature atomic-append) — AtomicAppend: multi-stream commits are all-or-nothing.
snapshot (feature snapshot) — SnapshotStore: state and position commit/hydrate atomically together, and a schema bump reads back Stale, never decode garbage.

The rest of this page is the writing-a-store-adapter guide. Sections 1–4 restate the seam’s contract in one place (the trait docs in mnesis-store remain the normative source; follow the links); section 5 shows how to prove an implementation with the kit; section 6 lists the pinned ambiguities that trip up new adapters. It assumes no knowledge of the shipped adapters — you never need to read mnesis-fjall or mnesis-postgres source.

§What you implement

One store type implements two mandatory traits.

RawEventStore — bytes in, bytes out. The adapter never sees typed events or codecs; the repository facade encodes into PendingEnvelopes before calling you. You supply:

type Error — your own error type, bound core::error::Error + Send + Sync + 'static. Keep it distinct from mnesis-store’s facade error types; the facade wraps yours, and a shared type would double-wrap.
type Stream — the per-stream read cursor: an owned, 'static, Send futures::Stream with Item = Result<PersistedEnvelope, Self::Error> (the EventStream marker bound).
type AllPosition — your store-local $all resume position: any Copy + Ord + Send + Sync + Debug + 'static type implementing AllPosition. A scalar sequence for an embedded store, a commit-ordered composite for a concurrent SQL store. It is never carried on the envelope; it rides only on $all items. mnesis-store ships no scalar impl, and the orphan rule blocks impl AllPosition for u64 in your crate — define a local newtype: struct MyPos(u64); impl AllPosition for MyPos {} (plus the derives the supertraits need).
type AllStream — the all-streams read cursor: an owned, 'static, Send futures::Stream with Item = Result<(Self::AllPosition, StreamKey, PersistedEnvelope), Self::Error> — every item is tagged with its position AND the origin StreamKey (stream attribution is a store guarantee — a $all consumer routes without decoding the payload). The per-stream read_stream does NOT stamp the id: there it is the query argument, so re-stamping it on every item would be redundant.
Make both stream types Unpin. The trait imposes no such bound, but the subscription path (Subscription) requires it.
Three methods — append, read_stream, read_all — whose contracts are the next two sections.

WakeSource — how live subscriptions learn that a commit landed. See “The wake contract” below.

Optional capability traits, each with its own kit module:

AtomicAppend (at mnesis_store::import::AtomicAppend, behind mnesis-store’s import feature) — commit several per-stream runs in one transaction, all-or-nothing; the primitive bulk import needs. Each write’s expected_version is validated against the target’s running head (counting earlier writes to the same target inside the batch); any mismatch aborts the whole transaction with AtomicAppendError::Conflict { index, actual }, and on any failure no write is applied. On success return the highest AllPosition the transaction committed across every stream (Option, None iff writes is empty) — the whole-batch read-your-writes token; unlike append, an empty writes is a legitimate no-op here, hence Option.
SnapshotStore<Vec<u8>, P> — atomic persistence of derived state plus the position it was folded to (hydrate / commit). Byte-level: S = Vec<u8>; typed state is a codec bridge upstream, not your concern. Two useful instantiations: P = Version (aggregate snapshots) and P = your AllPosition (projection checkpoints). hydrate returns the three-state Hydrated — Absent (never saved), Stale (saved under a different schema version; the caller rebuilds), Found (position + state). State and position commit together: the trait has no “save state alone”, and your implementation must persist the pair atomically so a half-write is unrepresentable.

Consumers never call you directly — they go through the Store<S> handle, repositories, and subscriptions, all generic over the seam. Implement the traits and everything above them works.

§Storing an event

append hands you a PendingBatch — a non-empty run of PendingEnvelopes (batch.first(), batch.last(), batch.iter(), batch.len() which is NonZeroUsize) — and reads must hand back PersistedEnvelopes. The supported recipe is the canonical wire frame: persist, per event, the Version (from PendingEnvelope::version()) plus the output of encode_frame:

let frame = mnesis_store::wire::encode_frame(
    env.schema_version_value(),
    &env.event_type_value(),
    &env.payload_value(),
    env.metadata_value().as_ref(),
)?; // EncodedFrame { value: Bytes, offsets: FrameOffsets }

Store frame.value (one contiguous Bytes buffer), frame.offsets, and the SchemaVersion; on read, rebuild with PersistedEnvelope::try_new (version, value, schema_version, offsets.event_type, offsets.payload, offsets.metadata). The frame lands the payload on a 16-byte boundary inside the buffer — an invariant zero-copy codecs (rkyv, POD) rely on. A custom storage layout is allowed, but then payload alignment and field re-validation are on you.

§The append contract

append(id, expected_version, envelopes) is optimistic concurrency:

expected_version is the stream head the caller last saw: None = a fresh stream with no events, Some(v) = the head is exactly v. Compare it against the stream’s actual current head; on mismatch return AppendError::Conflict carrying the stream id, the caller’s expectation, and the actual head — the caller reloads from actual and retries. The diagnostic id field is mnesis::ErrorId, built truncation-aware from the key’s Display: stream_id: ErrorId::from_display(id).
The head check and the event insertion must be one atomic step (a transaction, CAS, or a lock). A check-then-insert with a window between lets a concurrent writer slip in and corrupt the stream; the kit’s linearizability checks race real writers at exactly this seam.
Envelope versions must run strictly sequentially from expected_version + 1 (from 1 when None). A gap, duplicate, or out-of-order batch is rejected in the Conflict domain — and nothing lands: a rejected append leaves the store byte-identical, per-stream and $all alike. In that Conflict, expected is the caller’s stated expectation and actual is the store’s current head — the fields describe the head disagreement, never the malformed batch.
Stamp every accepted event with the next AllPosition: strictly monotonic across all streams in commit order, not required to be gapless — an aborted append may burn positions, and readers tolerate the gaps.
Return the AllPosition you stamped on the run’s LAST event. That is the read-your-writes token: a caller awaits its $all consumer reaching this position and then knows its own write is visible there. Return the position the committing transaction actually assigned — not a counter read back afterwards, which a concurrent append could make disagree. Because PendingBatch is non-empty there is always exactly one position to return; there is no write-nothing case (that is why the input is a PendingBatch, not a slice — the empty case is answered once, at the type, instead of by every adapter).
After the commit is durable — never before — fire your wake path (see “The wake contract”).

§The read contract

Two read methods, deliberately asymmetric.

read_stream(id, from) — a bounded scan of one stream:

from is inclusive: yield every event with version >= from, in ascending Version order, then terminate with None.
An absent stream is an empty stream, never an error.
After None the stream stays None (fused) — the kit polls again to prove it.
Internal batching/pagination is allowed and must be invisible; bounding resident memory is your concern (fjall, for instance, holds one lazy LSM cursor rather than fixed-size batches).

read_all(from: Option<AllPosition>) — a bounded scan across all streams:

from is exclusive: None = from the very beginning, Some(p) = strictly after p. Yield in ascending position order, each item tagged (position, stream key, envelope), then terminate with None when caught up.
Resume is Ord-based: the subscription loop reopens with the last position it delivered, and there is deliberately no successor function. Your scan must read “strictly greater than from” — tolerating gaps by scanning a range, never by stepping +1.

The asymmetry (inclusive Version vs. exclusive AllPosition) is intentional: a single stream’s versions are a gapless successor sequence, so the resume seam computes v + 1 itself and asks inclusively — while a composite $all position (e.g. postgres’s transaction-ordered pair) has no natural +1, so resume must be “strictly after what I saw”. Both reads serve the same strict-after resume; the difference is who computes the successor.

What a scan opened before a concurrent commit observes is adapter-unspecified — see “Contract notes”.

§The wake contract

A live subscription is a catch-up-then-park loop; the loop itself ships generically in mnesis-store and works for any adapter. Your half is WakeSource: two methods and one call-site discipline.

register(stream: Option<&[u8]>) — called once, synchronously, when a subscription opens (None registers for $all). Return a WakeRegistration that keeps wake-routing alive until dropped.
arm — returns an owned 'static future. Contract: the future captures a “seen point” at the moment arm is called and resolves once a wake is delivered after that point — a wake landing between arm and the .await must NOT be lost. The generic loop arms before its confirming re-scan whenever it thinks it is caught up; that ordering plus your arm-time capture is the entire lost-wakeup defense.
wake(stream) — call after every durable commit to stream, never before (a woken subscriber immediately re-reads and must see the data). A per-stream commit is also an $all event: $all observers must be woken too.
Spurious wakes are permitted — each costs one empty re-scan. Lost wakes are not — a lost wake is a subscription hung forever, and the kit’s check_wake_after_idle and check_caught_up_boundary_race exist to catch exactly that.

In-process adapters should not build this machinery: embed mnesis_wake::StreamNotifiers and delegate — the exact shape mnesis-inmemory and mnesis-fjall ship:

use std::sync::Arc;

use mnesis_store::wake::WakeSource;
use mnesis_wake::{NotifyError, StreamNotifiers, WakeReg};

struct MyStore {
    // ... your storage ...
    notifiers: Arc<StreamNotifiers>, // StreamNotifiers::new() -> Arc<StreamNotifiers>
}

impl WakeSource for MyStore {
    type Registration = WakeReg;
    type Error = NotifyError;

    fn register(&self, stream: Option<&[u8]>) -> Result<WakeReg, NotifyError> {
        self.notifiers.register(stream)
    }

    fn wake(&self, stream: &[u8]) {
        self.notifiers.wake(stream); // per-stream subscribers + the `$all` generation
    }
}

…and at the end of a successful non-empty append, after the commit is durable: self.notifiers.wake(id.as_ref()); (one call — it wakes the stream’s subscribers and bumps the store-wide $all generation). A distributed adapter implements the same two traits over its own signal (postgres: LISTEN/NOTIFY).

§Running the kit

An adapter proves conformance by invoking the conformance! macro (and the capability macros it needs) once, from one test file. Each generates one named #[tokio::test] per check, so nextest reports every contract rule as its own test — a failure names the exact rule that broke, not “some test in the suite.” Dependencies you’ll need: tokio with macros + rt-multi-thread (plus sync/time if your adapter uses tokio primitives), thiserror for your error enum (workspace rule), and mnesis-wake for the in-process WakeSource.

§The factory contract

Every macro takes a factory: Fn() -> Fut where Fut: Future<Output = (S, C)> + Send, S: RawEventStore + WakeSource. C is an adapter-chosen guard kept alive for the check’s duration — a TempDir for fjall, () for an adapter that owns its storage outright (in-memory, a connection pool). The factory is called once per generated test and must produce a fresh store each call; checks never share state across tests.

mnesis_store_testing::conformance! {
    factory: || async { (InMemoryStore::new(), ()) },
}

§Capability and lifecycle macros

conformance_atomic_append! — AtomicAppend checks; requires the atomic-append feature and an S: AtomicAppend factory.
conformance_snapshot! — SnapshotStore checks; requires the snapshot feature, an S: SnapshotStore<_, P> factory, and two pairs of ascending sample P positions: positions (ordinary values) and extremes (the representable edges, proving the position codec has no off-by-one at either end).
conformance_lifecycle! — close/reopen checks against the SAME backing storage; skipped entirely by in-memory adapters (nothing to reopen), run by every persistent adapter (fjall, postgres). Takes two closures: open (the usual factory shape) and reopen (Fn(S, C) -> Fut<Output = (S, C)>), which consumes the prior pair so it can drop the store before reopening the same storage.

§`skip_unless:` for environment-gated adapters

Every macro accepts an optional skip_unless: <fn() -> bool> that guards each generated test: when it returns false the test returns immediately (a vacuous pass, not a failure). mnesis-postgres uses this to skip the whole matrix when DATABASE_URL is unset locally, while still running for real under the nixosTest CI attribute that supplies a live database.

§Contract notes

Ambiguities pinned during this work, load-bearing for anyone writing a new adapter:

Read visibility under concurrent append is adapter-unspecified. Whether a reader started before a concurrent commit observes it is not part of the contract — FjallStore pins one snapshot at scan-open (repeatable-read), InMemoryStore re-reads live state on each refill. Both are conformant; the kit asserts eventual convergence (every committed event is observed once the reader catches up), never mid-flight visibility.
GlobalSeq / $all positions are strictly monotonic but never gapless. An aborted append may burn a position with no event landing there. Checks assert strict ordering, never contiguity.
Empty stream ids are a permitted adapter limitation. fjall/LSM rejects an empty key, so the kit only ever constructs non-empty stream ids — an adapter is not required to support the empty id.
Some(empty) metadata is unrepresentable by construction. The envelope’s metadata value type rejects a zero-length Some at construction (ValueError::MetadataEmpty) — the wire format reserves u32::MAX as the absent-metadata sentinel, so “empty but present” would collide with “absent”. The boundary check is check_metadata_absent_vs_present_distinct, not …_vs_empty_.
Public error enums are #[non_exhaustive]. AppendError, AtomicAppendError, and their siblings may grow variants without a major bump — match the variant you handle (Conflict) plus a wildcard arm, never exhaustively.

Re-exports§

pub use row::ConformanceRow;
pub use row::SubId;

Modules§

atomic: AtomicAppend capability conformance: several per-stream runs commit in ONE transaction — all land or none do.
boundary: Defensive Boundary conformance: inputs that violate the append protocol must be rejected cleanly and completely — nothing lands, nothing corrupts — and legal-but-extreme values must round-trip.
lifecycle: Lifecycle conformance (opt-in): close → reopen must preserve events, versions, and the $all position watermark. In-memory adapters have nothing to reopen and skip this module.
linearizability: Linearizability/Isolation conformance: genuinely-overlapping writers and a parked subscriber. Real overlap via tokio::spawn + Barrier (CLAUDE rule 8 — never sequential-then-check).
row: Shared test-data row, id type, and drive/drain helpers used by every conformance module.
sequence: Sequence/Protocol conformance: multi-step interactions on one store — append→read round-trips, optimistic-conflict protocol, $all ordering and resume, and the subscription catch-up→live protocol.
snapshot: SnapshotStore capability conformance: state and position commit and hydrate together; a schema change reads back as Stale, never as decode garbage.

Macros§

conformance: Run the full core conformance matrix (sequence + boundary + linearizability) against a store factory.
conformance_atomic_append: Run the AtomicAppend capability conformance (feature atomic-append).
conformance_lifecycle: Run the lifecycle conformance (persistent adapters only): open yields a fresh (store, ctx); reopen consumes both and reopens the SAME storage.
conformance_snapshot: Run the SnapshotStore capability conformance (feature snapshot).