SQLite engine: streams — StreamHandle, extent-guarded reads, trim, durable subscribe (task sqlite-engine-streams)

alkstore-sqlite/src/store/stream_tests.rs

@@ -0,0 +1,741 @@
+//! The stream mechanism's acceptance tests (`sqlite-engine-streams`):
+//! publish/read/save/get/trim/subscribe all work against a temp-file
+//! store, the extent guard, exact-boundary trim with surviving offsets
+//! unrenumbered and negative horizons deleting nothing, monotone
+//! saves across the direct and receiver forms, replay + wake-driven
+//! delivery + watcher-death close on subscribe, key round-trip, the
+//! `stream` field carrying the contract name, and entry-point
+//! validation (ADR-015; ADR-019 §1/§6; ADR-021 §5; ADR-023 §2).
+
+use std::path::PathBuf;
+use std::time::Duration;
+
+use alkstore::{Error, EventReceiver, Store};
+use serde_json::{Value, json};
+
+use crate::store::{open, open_store};
+
+fn temp_dir(tag: &str) -> PathBuf {
+    let d = std::env::temp_dir().join(format!(
+        "alkstore-stream-{tag}-{}-{:?}",
+        std::process::id(),
+        std::thread::current().id()
+    ));
+    let _ = std::fs::remove_dir_all(&d);
+    std::fs::create_dir_all(&d).unwrap();
+    d
+}
+
+fn cleanup(dir: &PathBuf) {
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+fn temp_path(tag: &str) -> PathBuf {
+    temp_dir(tag).join("store.db")
+}
+
+async fn until<F: Fn() -> bool>(tag: &str, f: F) {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    while !f() {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "{tag}: condition never held"
+        );
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+}
+
+/// Wait for the next event with a bounded deadline — no tight-race
+/// assertions (the suite's stability posture).
+async fn must_recv_event(receiver: &mut dyn EventReceiver, tag: &str) -> alkstore::StreamEvent {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    loop {
+        match receiver.try_recv() {
+            Ok(Some(event)) => return event,
+            Ok(None) => {}
+            Err(e) => panic!("{tag}: receiver errored instead of idling: {e}"),
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "{tag}: event never arrived"
+        );
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+}
+
+/// Acceptance: full `StreamHandle` over a temp-file store — publish
+/// returns ascending offsets, reads decode into `StreamEvent` with the
+/// `stream` field carrying the contract name (not `topic`), keys
+/// round-trip exactly (`None` vs `Some`), consumers absent from the
+/// table read offset 0, offsets save and inspect (the monotone save's
+/// composition pinned with the direct forms below), and the ordering
+/// is `offset ASC` (global FIFO, ADR-015 §4).
+#[tokio::test(flavor = "multi_thread")]
+async fn publish_read_save_get_round_trip_end_to_end() {
+    let dir = temp_dir("round-trip");
+    let path = temp_path("round-trip");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    let stream = store.stream("events").await.unwrap();
+    assert_eq!(stream.name(), "events");
+
+    let o1 = stream.publish(json!({"n": 1})).await.unwrap();
+    let o2 = stream
+        .publish_with_key(Some("k".to_string()), json!({"n": 2}))
+        .await
+        .unwrap();
+    let o3 = stream.publish(json!({"n": 3})).await.unwrap();
+    assert!(o1 < o2 && o2 < o3, "offsets ascend per publish");
+
+    let page = stream.read_since(0, 10).await.unwrap();
+    assert_eq!(page.len(), 3, "the full stream reads back");
+    assert_eq!(page[0].offset, o1);
+    assert_eq!(page[0].stream, "events", "the stream field is the name");
+    assert_eq!(page[0].key, None, "plain publish carries no key");
+    assert_eq!(page[1].offset, o2);
+    assert_eq!(page[1].key, Some("k".to_string()));
+    assert_eq!(page[2].offset, o3);
+    assert_eq!(page[2].key, None);
+    for (i, event) in page.iter().enumerate() {
+        let decoded: Value = event.payload_as().unwrap();
+        assert_eq!(decoded, json!({"n": i as i64 + 1}));
+    }
+
+    // Cursor reads: offset > cursor, ASC.
+    let tail = stream.read_since(o1, 10).await.unwrap();
+    assert_eq!(tail.len(), 2);
+    assert_eq!(tail[0].offset, o2);
+
+    // Consumers absent from the table read 0 (the pre-save state).
+    let from_absent = stream
+        .read_from_consumer("fresh-consumer", 10)
+        .await
+        .unwrap();
+    assert_eq!(from_absent.len(), 3, "absent consumer reads from 0");
+    assert_eq!(
+        stream.get_offset("fresh-consumer").await.unwrap(),
+        0,
+        "get_offset of an absent consumer = 0"
+    );
+
+    // Save + inspect; reads from the consumer's checkpoint.
+    stream.save_offset("c", o2).await.unwrap();
+    assert_eq!(stream.get_offset("c").await.unwrap(), o2);
+    let from_c = stream.read_from_consumer("c", 10).await.unwrap();
+    assert_eq!(from_c.len(), 1);
+    assert_eq!(from_c[0].offset, o3);
+
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: the extent guard (ADR-023 §2) — `read_since` /
+/// `read_from_consumer` with `limit <= 0` return the empty `Vec` (never
+/// the whole stream, which SQLite's `LIMIT -1` dialect would yield),
+/// at the trait-impl entry.
+#[tokio::test(flavor = "multi_thread")]
+async fn extent_guard_reads_limit_zero_and_negative_yield_empty() {
+    let dir = temp_dir("extent-guard");
+    let store = open_store(
+        temp_path("extent-guard").to_str().unwrap(),
+        Default::default(),
+    )
+    .unwrap();
+    let stream = store.stream("s").await.unwrap();
+    stream.publish(json!({"n": 1})).await.unwrap();
+    stream.publish(json!({"n": 2})).await.unwrap();
+    stream.save_offset("c", 6).await.unwrap();
+
+    for limit in [0i64, -1, i64::MIN] {
+        let page = stream.read_since(0, limit).await.unwrap();
+        assert!(
+            page.is_empty(),
+            "read_since(limit={limit}) must read nothing"
+        );
+        let page = stream.read_from_consumer("c", limit).await.unwrap();
+        assert!(
+            page.is_empty(),
+            "read_from_consumer(limit={limit}) must read nothing"
+        );
+    }
+
+    // The stream was never handed out in bulk: the second event is
+    // still readable with a positive extent.
+    assert_eq!(stream.read_since(0, 10).await.unwrap().len(), 2);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: `trim_to` is exact-boundary (`offset <= horizon`),
+/// returns the deleted count, a negative horizon deletes nothing
+/// (boundary args are total, ADR-023 §2 — no guard, pass through),
+/// and surviving offsets are never renumbered (gaps legal).
+#[tokio::test(flavor = "multi_thread")]
+async fn trim_to_trims_the_exact_boundary_and_never_renumbers() {
+    let dir = temp_dir("trim");
+    let store = open_store(temp_path("trim").to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("s").await.unwrap();
+
+    let mut offsets = Vec::new();
+    for i in 0..5 {
+        offsets.push(stream.publish(json!({"n": i})).await.unwrap());
+    }
+
+    // Negative horizon: deletes nothing, idempotently (no guard).
+    assert_eq!(
+        stream.trim_to(-7).await.unwrap(),
+        0,
+        "a negative horizon deletes nothing"
+    );
+    assert_eq!(stream.read_since(0, 10).await.unwrap().len(), 5);
+
+    // Zero horizon is a no-op on this store (offsets start above 0 on
+    // the autoincrement counter — also nothing to delete).
+    assert_eq!(stream.trim_to(0).await.unwrap(), 0);
+
+    // Exact boundary: horizon = offsets[1] deletes exactly rows
+    // offsets[0..=1] (two rows), keeping offsets[2..].
+    let deleted = stream.trim_to(offsets[1]).await.unwrap();
+    assert_eq!(deleted, 2, "trim deletes offset <= horizon exactly");
+
+    let survivors = stream.read_since(0, 10).await.unwrap();
+    assert_eq!(survivors.len(), 3);
+    let survivor_offsets: Vec<i64> = survivors.iter().map(|e| e.offset).collect();
+    assert_eq!(
+        survivor_offsets,
+        offsets[2..],
+        "surviving offsets are never renumbered — gaps are legal"
+    );
+
+    // Reads from a trimmed-away region resume at the horizon's first
+    // remaining row (fewer/no rows, never an error).
+    let resumed = stream.read_since(offsets[0], 10).await.unwrap();
+    assert_eq!(resumed.len(), 3);
+
+    // Trim is idempotent below the surviving set.
+    assert_eq!(stream.trim_to(offsets[1]).await.unwrap(), 0);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: monotone saves (ADR-019 §6) — a direct save below the
+/// stored checkpoint is a silent no-op (`Ok(())`, offset unchanged);
+/// direct and receiver forms compose without regression.
+#[tokio::test(flavor = "multi_thread")]
+async fn offset_saves_are_monotone_across_forms() {
+    let dir = temp_dir("monotone");
+    let store = open_store(temp_path("monotone").to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("s").await.unwrap();
+
+    let o1 = stream.publish(json!({"n": 1})).await.unwrap();
+    let o2 = stream.publish(json!({"n": 2})).await.unwrap();
+    assert!(o1 < o2);
+
+    stream.save_offset("c", o2).await.unwrap();
+    // Regression: silent no-op, not an error (the act-differently rule).
+    stream.save_offset("c", o1).await.unwrap();
+    assert_eq!(
+        stream.get_offset("c").await.unwrap(),
+        o2,
+        "a save below the stored checkpoint does not rewind the cursor"
+    );
+
+    stream.save_offset("c", o2).await.unwrap();
+    assert_eq!(stream.get_offset("c").await.unwrap(), o2);
+
+    // The receiver form drives the same op: subscribing a fresh
+    // consumer, reading to tail, saving through the receiver — then a
+    // direct regression save cannot undo it.
+    let mut rx = store
+        .stream("s")
+        .await
+        .unwrap()
+        .subscribe("r")
+        .await
+        .unwrap();
+    let last = must_recv_event(&mut *rx, "monotone-receiver").await;
+    rx.save_offset().unwrap();
+    assert_eq!(rx.offset(), last.offset);
+    let stored = store
+        .stream("s")
+        .await
+        .unwrap()
+        .get_offset("r")
+        .await
+        .unwrap();
+    assert_eq!(stored, last.offset, "the receiver save checkpointed");
+
+    stream.save_offset("r", o1).await.unwrap();
+    let after = store
+        .stream("s")
+        .await
+        .unwrap()
+        .get_offset("r")
+        .await
+        .unwrap();
+    assert_eq!(
+        after, last.offset,
+        "the direct form cannot regress the receiver's"
+    );
+
+    // Receiver save before any yield is a no-op (the stored checkpoint
+    // untouched — ADR-021 §5). A fresh subscription over a consumer
+    // already at a stored offset: nothing yielded yet, save, unchanged.
+    let fresh_stream = store.stream("s").await.unwrap();
+    let mut fresh = fresh_stream.subscribe("pre-yield").await.unwrap();
+    fresh.save_offset().unwrap();
+    assert_eq!(fresh.offset(), 0);
+    assert_eq!(fresh_stream.get_offset("pre-yield").await.unwrap(), 0);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: `subscribe` replays from the *stored* offset (a
+/// consumer with a prior checkpoint resumes there, not from 0), then
+/// delivers post-attach events wake-driven — events published after
+/// attach arrive through the receiver.
+#[tokio::test(flavor = "multi_thread")]
+async fn subscribe_replays_from_stored_offset_then_delivers_wake_driven() {
+    let dir = temp_dir("subscribe");
+    let path = temp_path("subscribe");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    let pre = store.stream("s").await.unwrap();
+    let o1 = pre.publish(json!({"n": 1})).await.unwrap();
+    let o2 = pre.publish(json!({"n": 2})).await.unwrap();
+    let o3 = pre.publish(json!({"n": 3})).await.unwrap();
+    drop(pre);
+
+    // A stored checkpoint below the tail: replay starts there (o2 is
+    // the first event after the checkpoint).
+    let handle = store.stream("s").await.unwrap();
+    handle.save_offset("c", o1).await.unwrap();
+    let mut rx = handle.subscribe("c").await.unwrap();
+    drop(handle);
+
+    let first = must_recv_event(&mut *rx, "subscribe-replay").await;
+    assert_eq!(
+        first.offset, o2,
+        "replay resumes past the stored checkpoint"
+    );
+    let second = must_recv_event(&mut *rx, "subscribe-replay").await;
+    assert_eq!(second.offset, o3);
+    assert_eq!(second.stream, "s");
+
+    // Post-attach publishes arrive wake-driven — no polling needed.
+    let late = store
+        .stream("s")
+        .await
+        .unwrap()
+        .publish_with_key(Some("late".to_string()), json!({"n": 4}))
+        .await
+        .unwrap();
+    let wake_event = must_recv_event(&mut *rx, "subscribe-wake").await;
+    assert_eq!(wake_event.offset, late);
+    assert_eq!(wake_event.key.as_deref(), Some("late"));
+
+    // Receiver checkpoint composes with the substrate's monotone upsert.
+    rx.save_offset().unwrap();
+    let stored = store
+        .stream("s")
+        .await
+        .unwrap()
+        .get_offset("c")
+        .await
+        .unwrap();
+    assert_eq!(stored, wake_event.offset);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: key round-trips exactly — `None` vs `Some` preserved on
+/// every read form (direct page and receiver delivery), and the
+/// `stream` field carries the contract name (the `topic` column is
+/// never surfaced).
+#[tokio::test(flavor = "multi_thread")]
+async fn keys_round_trip_exactly_and_stream_field_carries_the_name() {
+    let dir = temp_dir("keys");
+    let store = open_store(temp_path("keys").to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("the-name").await.unwrap();
+
+    let plain = stream.publish(json!({})).await.unwrap();
+    let keyed = stream
+        .publish_with_key(Some("k".to_string()), json!({}))
+        .await
+        .unwrap();
+
+    let page = stream.read_since(0, 10).await.unwrap();
+    fn find(events: &[alkstore::StreamEvent], offset: i64) -> Option<&alkstore::StreamEvent> {
+        events.iter().find(|e| e.offset == offset)
+    }
+    assert_eq!(find(&page, plain).unwrap().key, None);
+    assert_eq!(find(&page, keyed).unwrap().key, Some("k".to_string()));
+    assert!(page.iter().all(|e| e.stream == "the-name"));
+
+    let mut rx = store
+        .stream("the-name")
+        .await
+        .unwrap()
+        .subscribe("c")
+        .await
+        .unwrap();
+    let e1 = must_recv_event(&mut *rx, "keys-receiver").await;
+    let e2 = must_recv_event(&mut *rx, "keys-receiver").await;
+    assert_eq!(e1.key, None);
+    assert_eq!(e2.key, Some("k".to_string()));
+    assert_eq!(e1.stream, "the-name");
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: watcher death closes the subscription terminally —
+/// `recv() -> None` (never reopens); `try_recv` reports `Err(Closed)`.
+/// Driven through store close (the death-guard clears every
+/// subscriber feed — the bridge exits, the consumer sees the close).
+#[tokio::test(flavor = "multi_thread")]
+async fn watcher_death_closes_the_subscription_terminally() {
+    let dir = temp_dir("death");
+    let store = open_store(temp_path("death").to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("s").await.unwrap();
+    stream.publish(json!({"n": 1})).await.unwrap();
+    let mut rx = stream.subscribe("c").await.unwrap();
+
+    let event = must_recv_event(&mut *rx, "death").await;
+    assert_eq!(event.stream, "s");
+
+    store.close();
+    let none = rx.recv().await;
+    assert!(
+        none.is_none(),
+        "watcher death closes the receiver: got {none:?}"
+    );
+    assert!(
+        matches!(rx.try_recv(), Err(Error::Closed)),
+        "after close, try_recv is Err(Closed) — not idle"
+    );
+    assert!(rx.recv().await.is_none(), "a closed receiver never reopens");
+    cleanup(&dir);
+}
+
+/// Acceptance: a subscribe attached while the stream is at tail idles
+/// (`Ok(None)`), then delivers when the next publish lands.
+#[tokio::test(flavor = "multi_thread")]
+async fn subscribe_at_tail_idles_then_delivers() {
+    let dir = temp_dir("tail-idle");
+    let store = open_store(temp_path("tail-idle").to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("s").await.unwrap();
+
+    let mut rx = stream.subscribe("c").await.unwrap();
+    tokio::time::sleep(Duration::from_millis(150)).await;
+    assert_eq!(rx.try_recv().unwrap(), None, "at tail, the receiver idles");
+
+    let o = stream.publish(json!({"n": 1})).await.unwrap();
+    let event = must_recv_event(&mut *rx, "tail-idle").await;
+    assert_eq!(event.offset, o);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: two consumers subscribe independently — each drains a
+/// copy of the stream from its own position; one's receiver save never
+/// gates the other's delivery (saved checkpoints never gate reads).
+#[tokio::test(flavor = "multi_thread")]
+async fn subscribers_are_independent_copies() {
+    let dir = temp_dir("two-consumers");
+    let store = open_store(
+        temp_path("two-consumers").to_str().unwrap(),
+        Default::default(),
+    )
+    .unwrap();
+    let stream = store.stream("s").await.unwrap();
+    let o1 = stream.publish(json!({"n": 1})).await.unwrap();
+    let o2 = stream.publish(json!({"n": 2})).await.unwrap();
+
+    let handle = store.stream("s").await.unwrap();
+    let mut a = handle.subscribe("a").await.unwrap();
+    let mut b = handle.subscribe("b").await.unwrap();
+    drop(handle);
+
+    // a drains and checkpoints both; b must still receive both (a's
+    // position is a's, not shared).
+    let _ = must_recv_event(&mut *a, "two-a-1").await;
+    let _ = must_recv_event(&mut *a, "two-a-2").await;
+    a.save_offset().unwrap();
+
+    let b1 = must_recv_event(&mut *b, "two-b-1").await;
+    let b2 = must_recv_event(&mut *b, "two-b-2").await;
+    assert_eq!(b1.offset, o1);
+    assert_eq!(b2.offset, o2);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: `recv()`'s `Err` arm carries `Database` only
+/// (ADR-021 §5) and close is terminal. Driving the transient-failure
+/// arm against a healthy engine is not directly reproducible (a
+/// healthy SELECT on a WAL file does not fail); the arm's shape is
+/// pinned here by construction — the receiver yields whatever the
+/// bridge read produced, and this test pins the observable close
+/// semantics across drop (dropping the receiver's channel source).
+/// The compile-time shape (`Option<Result<StreamEvent>>`) plus the
+/// `Database`-only doc rule are the pinned surface; this test pins the
+/// runtime close arm over the drop path.
+#[tokio::test(flavor = "multi_thread")]
+async fn receiver_drop_closes_its_own_subscription() {
+    let dir = temp_dir("receiver-drop");
+    let store = open_store(
+        temp_path("receiver-drop").to_str().unwrap(),
+        Default::default(),
+    )
+    .unwrap();
+    let stream = store.stream("s").await.unwrap();
+    stream.publish(json!({})).await.unwrap();
+
+    let baseline = store.watcher.subscriber_count();
+    assert_eq!(baseline, 0, "no subscriptions before any subscribe");
+
+    {
+        let handle = store.stream("s").await.unwrap();
+        let rx = handle.subscribe("c").await.unwrap();
+        drop(rx);
+    }
+
+    // The receiver's Drop unsubscribed eagerly.
+    until("receiver-drop", || {
+        store.watcher.subscriber_count() == baseline
+    })
+    .await;
+    assert_eq!(
+        store.watcher.subscriber_count(),
+        baseline,
+        "receiver drop must not leak subscriptions"
+    );
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: entry-point validation — stream names validate as
+/// shared-namespace kinds at the constructor; consumers as local names
+/// (non-empty only; reserved prefixes legal for consumers). Validation
+/// fires before any round trip: rejected constructors yield no handle,
+/// and rejected consumer names never touch the tables.
+#[tokio::test(flavor = "multi_thread")]
+async fn entry_points_validate_stream_names_and_consumers() {
+    let dir = temp_dir("validation");
+    let path = temp_path("validation");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    // Stream constructors: empty/whitespace → InvalidName; reserved
+    // prefix → ReservedName.
+    for name in ["", "   ", "__alkstore_x"] {
+        let handle = match store.stream(name).await {
+            Err(e) => e,
+            Ok(_) => panic!("stream({name:?}) must reject by validation"),
+        };
+        match handle {
+            Error::InvalidName { name: n } => assert_eq!(name, n),
+            Error::ReservedName { name: n } => assert_eq!(name, n),
+            other => panic!("stream({name:?}) must reject by validation, got {other}"),
+        }
+    }
+
+    let stream = store.stream("s").await.unwrap();
+
+    // Empty-`Some` keys are InvalidName (the tx path's rule).
+    let err = stream
+        .publish_with_key(Some(String::new()), json!({}))
+        .await
+        .unwrap_err();
+    match err {
+        Error::InvalidName { name } => assert_eq!(name, ""),
+        other => panic!("empty-Some key must be InvalidName, got {other}"),
+    }
+    let err = stream
+        .publish_with_key(Some("   ".to_string()), json!({}))
+        .await
+        .unwrap_err();
+    assert!(matches!(err, Error::InvalidName { .. }));
+
+    // Consumers: non-empty only — reserved prefixes are legal.
+    for consumer in ["", "   "] {
+        let err = stream.read_from_consumer(consumer, 5).await.unwrap_err();
+        assert!(matches!(err, Error::InvalidName { .. }));
+        let err = stream.save_offset(consumer, 1).await.unwrap_err();
+        assert!(matches!(err, Error::InvalidName { .. }));
+        let err = stream.get_offset(consumer).await.unwrap_err();
+        assert!(matches!(err, Error::InvalidName { .. }));
+        assert!(stream.subscribe(consumer).await.is_err());
+    }
+
+    // A legal-but-reserved consumer name works (non-empty only).
+    let o = stream.publish(json!({"n": 9})).await.unwrap();
+    stream.save_offset("__alkstore_local", o).await.unwrap();
+    assert_eq!(stream.get_offset("__alkstore_local").await.unwrap(), o);
+
+    // The rejected consumer saves never touched the table.
+    assert_eq!(stream.get_offset("never-saved").await.unwrap(), 0);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: `StreamHandle` ops on a closed store fail closed with
+/// `Error::Database` (the engine-wide closed-store posture); a
+/// subscribe on a closed store fails too.
+#[tokio::test(flavor = "multi_thread")]
+async fn closed_store_fails_stream_ops() {
+    let dir = temp_dir("closed-store");
+    let store = open_store(
+        temp_path("closed-store").to_str().unwrap(),
+        Default::default(),
+    )
+    .unwrap();
+    let stream = store.stream("s").await.unwrap();
+    stream.publish(json!({"n": 1})).await.unwrap();
+    let stored_handle = store.stream("s").await.unwrap();
+    store.close();
+
+    let err = stream.publish(json!({})).await.unwrap_err();
+    assert!(matches!(err, Error::Database(_)), "publish: {err}");
+    let err = stream.read_since(0, 5).await.unwrap_err();
+    assert!(matches!(err, Error::Database(_)), "read_since: {err}");
+    let err = stream.read_from_consumer("c", 5).await.unwrap_err();
+    assert!(
+        matches!(err, Error::Database(_)),
+        "read_from_consumer: {err}"
+    );
+    let err = stream.save_offset("c", 1).await.unwrap_err();
+    assert!(matches!(err, Error::Database(_)), "save_offset: {err}");
+    let err = stream.get_offset("c").await.unwrap_err();
+    assert!(matches!(err, Error::Database(_)), "get_offset: {err}");
+    let err = stream.trim_to(10).await.unwrap_err();
+    assert!(matches!(err, Error::Database(_)), "trim_to: {err}");
+    assert!(stored_handle.subscribe("c").await.is_err());
+
+    // New constructors on a closed store fail too.
+    let err = match store.stream("s").await {
+        Err(e) => e,
+        Ok(_) => panic!("stream on a closed store must fail closed"),
+    };
+    assert!(matches!(err, Error::Database(_)));
+    cleanup(&dir);
+}
+
+/// Acceptance: the durable subscription survives store restart — a
+/// fresh open over the same file resumes from the saved checkpoint
+/// (durable consumption; offsets are positions in the durable log).
+#[tokio::test(flavor = "multi_thread")]
+async fn subscription_state_survives_store_restart() {
+    let dir = temp_dir("restart");
+    let path = temp_path("restart");
+
+    let saved;
+    {
+        let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+        let stream = store.stream("s").await.unwrap();
+        let o1 = stream.publish(json!({"n": 1})).await.unwrap();
+        let _ = stream.publish(json!({"n": 2})).await.unwrap();
+        stream.save_offset("c", o1).await.unwrap();
+        saved = stream.get_offset("c").await.unwrap();
+        store.close();
+    }
+
+    {
+        let store = open(path.to_str().unwrap(), Default::default()).unwrap();
+        let stream = store.stream("s").await.unwrap();
+        // Checkpoint durable across the restart.
+        assert_eq!(stream.get_offset("c").await.unwrap(), saved);
+        // Replay resumes from it — the second event is the first read.
+        let page = stream.read_from_consumer("c", 10).await.unwrap();
+        assert_eq!(page.len(), 1, "the resumed read starts past the checkpoint");
+        let saved_offset = saved;
+        // A fresh subscription replays from the saved offset.
+        let mut rx = stream.subscribe("c").await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        match rx.try_recv().unwrap() {
+            Some(event) => assert_eq!(event.offset, saved_offset + 1),
+            None => panic!("the resumed subscription must replay past the checkpoint"),
+        }
+        drop(rx);
+        drop(stream);
+        drop(store);
+    }
+    cleanup(&dir);
+}
+/// Acceptance: the tx seam and the handle compose — a `publish_tx`
+/// inside a committed business transaction is visible to the handle's
+/// reads and wakes a live subscriber; a rolled-back `publish_tx` ghosts
+/// (no read, no wake).
+#[tokio::test(flavor = "multi_thread")]
+async fn tx_publishes_compose_with_the_handle() {
+    let dir = temp_dir("tx-compose");
+    let path = temp_path("tx-compose");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let mut rx = store
+        .stream("s")
+        .await
+        .unwrap()
+        .subscribe("c")
+        .await
+        .unwrap();
+
+    let mut tx = store.begin_tx().await.unwrap();
+    let tx_offset = tx.publish_tx("s", json!({"via": "tx"})).await.unwrap();
+    tx.commit().await.unwrap();
+
+    let event = must_recv_event(&mut *rx, "tx-compose").await;
+    assert_eq!(event.offset, tx_offset);
+    assert_eq!(event.payload_as::<Value>().unwrap(), json!({"via": "tx"}));
+
+    // Rollback: the publish ghosts.
+    let mut tx = store.begin_tx().await.unwrap();
+    tx.publish_tx("s", json!({"via": "ghost"})).await.unwrap();
+    drop(tx);
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    while rx.try_recv().unwrap().is_some() {}
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert_eq!(
+        rx.try_recv().unwrap(),
+        None,
+        "a rolled-back publish must not deliver"
+    );
+
+    // The handle read confirms the table state.
+    let stream = store.stream("s").await.unwrap();
+    let page = stream.read_since(0, 10).await.unwrap();
+    assert_eq!(page.len(), 1, "the ghost never landed");
+    assert_eq!(page[0].offset, tx_offset);
+    store.close();
+    cleanup(&dir);
+}
+
+/// Acceptance: the receiver's `read_since` is offset-anchored and
+/// independent of its position (saved checkpoints do not gate reads —
+/// replay is a read concern); extent-guarded like the direct form.
+#[tokio::test(flavor = "multi_thread")]
+async fn receiver_read_since_is_offset_anchored_and_extent_guarded() {
+    let dir = temp_dir("receiver-read");
+    let store = open_store(
+        temp_path("receiver-read").to_str().unwrap(),
+        Default::default(),
+    )
+    .unwrap();
+    let stream = store.stream("s").await.unwrap();
+    let o1 = stream.publish(json!({"n": 1})).await.unwrap();
+    let o2 = stream.publish(json!({"n": 2})).await.unwrap();
+
+    let mut rx = stream.subscribe("c").await.unwrap();
+    let _ = must_recv_event(&mut *rx, "receiver-read").await;
+    let _ = must_recv_event(&mut *rx, "receiver-read").await;
+    rx.save_offset().unwrap();
+    assert_eq!(rx.offset(), o2);
+
+    // The receiver replays from any offset regardless of its position.
+    let replay = rx.read_since(o1 - 1, 10).await.unwrap();
+    assert_eq!(replay.len(), 2);
+    assert_eq!(replay[0].offset, o1);
+    // Extent guard on the receiver form too.
+    assert_eq!(rx.read_since(0, 0).await.unwrap().len(), 0);
+    assert_eq!(rx.read_since(0, -3).await.unwrap().len(), 0);
+    store.close();
+    cleanup(&dir);
+}