From 14d82b9f9e221d64f6213638ae789cfa2ba77990 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 16:48:19 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(feed):=20one=20shared=20reader=20behin?= =?UTF-8?q?d=20the=20change=20stream=20(xdvg.2,=20event-push=20=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before, every /changes/stream connection woke on every commit, crossed the writer barrier and read its own page: N followers cost N store reads per transaction. Now one reader per store (src/server/feed_hub.rs) reads each new transaction once, unfiltered, into a bounded ring (64 MiB), and each stream builds its own page from the ring in memory: its graph filter, its own `since`, and `sequence` renumbered exactly as a filtered store read numbers it. Serving costs no store read; an idle stream costs nothing. - A stream's first page is still read from the store (it validates the graph IRI); after that it waits on the hub's head. - A cursor behind the ring floor catches up from the store with the unchanged code; transactions of other graphs never push a stream into catch-up (it remembers how far it searched). - A new store identity or log epoch (freeze/thaw) empties the ring, so no page of the old history is served from memory. - A wake with nothing new costs one id lookup, never a page read. - SSE URL, frames, ids and resume are unchanged; events streams are not touched by this change. Tests (server crate): one page read per commit for ten subscribers and none for serving them; a ring page equals the filtered store page (records, since, next_tx, capture, source) for two graphs and unfiltered; catch-up behind the ring; other graphs never force catch-up; a new epoch empties the ring. Rollback: revert; streams go back to per-connection reads. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/server/feed_hub.rs | 274 +++++++++++++++++++++++++++++++++++ src/server/feed_hub_tests.rs | 203 ++++++++++++++++++++++++++ src/server/feed_stream.rs | 49 ++++++- src/server/handle.rs | 4 + src/server/tests.rs | 1 + 5 files changed, 528 insertions(+), 3 deletions(-) create mode 100644 src/server/feed_hub.rs create mode 100644 src/server/feed_hub_tests.rs diff --git a/src/server/feed_hub.rs b/src/server/feed_hub.rs new file mode 100644 index 000000000..0d2b242c3 --- /dev/null +++ b/src/server/feed_hub.rs @@ -0,0 +1,274 @@ +//! The change feed's shared reader (docs/design/event-push.md ยง2). +//! +//! Before the hub, every `/changes/stream` connection woke on every commit, +//! crossed the writer barrier and read its own page: N followers cost N store +//! reads per transaction. The hub reads each new transaction ONCE, unfiltered, +//! into a bounded in-memory ring, and every subscriber builds its own page from +//! the ring in memory: its graph filter, its own `since`, the records renumbered +//! exactly as a filtered store read numbers them. A subscriber does no store +//! read while it is served from the ring, and an idle one costs nothing. +//! +//! The ring covers every transaction in `(floor, head]`. A subscriber behind +//! the floor (a new one resuming from an old cursor, or one that fell behind a +//! trim) is answered `CatchUp` and pages the store itself, exactly as before the +//! hub, until it reaches the ring. When the store's identity or log epoch +//! changes (a freeze or thaw re-mints it), the ring is emptied, so nobody is +//! served a page of the old history from memory. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration; + +use serde_json::{Value, json}; +use tokio::sync::watch; + +use crate::{SharedStore, StoreHandle, base::blocking}; + +/// Transactions read per store page. +const PAGE_TXS: usize = 64; +/// Bytes of records the ring holds before it trims its oldest transactions. +const RING_BYTES: usize = 64 * 1024 * 1024; + +/// One transaction as the hub read it: every record, in every graph. +struct Entry { + tx: i64, + records: Arc>, + watermark_tx: i64, + watermark_timestamp: Option, + bytes: usize, +} + +struct Ring { + /// False until the reader's first read: every subscriber catches up. + ready: bool, + floor: i64, + head: i64, + source: Value, + entries: VecDeque, + bytes: usize, +} + +/// What the hub can do for a subscriber at its cursor. +pub(crate) enum Serve { + /// The next page, built from the ring. + Page(Value), + /// Nothing in scope after the cursor yet; wait for the head to move. + Wait, + /// The cursor is behind the ring: page the store directly. + CatchUp, +} + +pub(crate) struct Hub { + ring: Mutex, + head: watch::Sender, + /// Change pages read by the hub, for the one-read-per-transaction control. + pub(crate) reads: AtomicU64, +} + +impl Hub { + /// The store's hub, started on first use. + pub(crate) fn of(store: &SharedStore) -> Arc { + store + .changes_hub + .get_or_init(|| { + let hub = Arc::new(Self { + ring: Mutex::new(Ring { + ready: false, + floor: 0, + head: 0, + source: Value::Null, + entries: VecDeque::new(), + bytes: 0, + }), + head: watch::channel(0).0, + reads: AtomicU64::new(0), + }); + tokio::spawn(read_loop(Arc::downgrade(store), hub.clone())); + hub + }) + .clone() + } + + /// A receiver that changes whenever the ring's head moves. + pub(crate) fn watch(&self) -> watch::Receiver { + self.head.subscribe() + } + + /// The next page for a subscriber at `cursor` with an optional graph IRI. + /// `scanned` is how far this subscriber has already searched the ring for + /// its graph, so transactions of other graphs never push it into catch-up. + pub(crate) fn serve(&self, cursor: i64, graph: Option<&str>, scanned: &mut i64) -> Serve { + let Ok(ring) = self.ring.lock() else { + return Serve::CatchUp; + }; + let from = cursor.max(*scanned); + if !ring.ready || from < ring.floor { + return Serve::CatchUp; + } + for entry in ring.entries.iter().filter(|e| e.tx > from) { + let records: Vec = match graph { + None => entry.records.as_ref().clone(), + Some(iri) => entry + .records + .iter() + .filter(|r| r["graph"].as_str() == Some(iri)) + .enumerate() + .map(|(sequence, r)| { + let mut r = r.clone(); + r["sequence"] = json!(sequence); + r + }) + .collect(), + }; + if records.is_empty() { + continue; + } + return Serve::Page(json!({ + "records": records, + "since": cursor, + "next_tx": entry.tx, + "watermark_tx": entry.watermark_tx, + "watermark_timestamp": entry.watermark_timestamp, + "capture": "old_and_new_values", + "source": ring.source, + })); + } + *scanned = ring.head; + Serve::Wait + } +} + +/// The hub's one reader: read every new transaction after the head, then +/// wait for the next commit. Ends when the store is gone. +async fn read_loop(store: Weak, hub: Arc) { + let Some(mut hints) = store.upgrade().map(|s| s.commit_wake.subscribe()) else { + return; + }; + hints.borrow_and_update(); + loop { + let Some(shared) = store.upgrade() else { + return; + }; + let reader = hub.clone(); + let read = blocking(move || { + // Let the hinted commit finish before reading it. + drop(shared.writer.lock()); + read_new(&shared, &reader); + Ok(()) + }) + .await; + if read.is_err() { + // Unreadable: empty the ring so subscribers page the store, which + // reports the failure to each of them. + if let Ok(mut ring) = hub.ring.lock() { + ring.ready = false; + ring.entries.clear(); + ring.bytes = 0; + } + } + if tokio::time::timeout(Duration::from_secs(3600), hints.changed()) + .await + .is_ok_and(|r| r.is_err()) + { + return; + } + } +} + +/// Read every transaction after the ring's head into the ring. +fn read_new(store: &StoreHandle, hub: &Hub) { + let read = store.read(); + let Ok(source) = read.store_id().and_then(|id| Ok((id, read.log_epoch()?))) else { + return; + }; + let source = json!({"store_id": source.0, "log_epoch": source.1}); + let mut moved = false; + { + let Ok(mut ring) = hub.ring.lock() else { + return; + }; + if !ring.ready || ring.source != source { + // First read, or a different history: start empty at the head. + let Ok(head) = read.latest_tx_id() else { + return; + }; + ring.entries.clear(); + ring.bytes = 0; + ring.floor = head; + ring.head = head; + ring.source = source.clone(); + ring.ready = true; + moved = true; + } + } + // A wake with nothing new (a hint for a commit already read, or one that + // wrote no facts) costs one id lookup, never a page read. + let Ok(latest) = read.latest_tx_id() else { + return; + }; + loop { + let head = match hub.ring.lock() { + Ok(ring) => ring.head, + Err(_) => return, + }; + if latest <= head { + break; + } + hub.reads.fetch_add(1, Ordering::Relaxed); + let Ok(page) = read.changes_after( + head, + PAGE_TXS, + quipu::store::changes::Capture::OldAndNewValues, + None, + ) else { + return; + }; + if page.records.is_empty() { + break; + } + // A short page reached the head: no second read to learn that. + let short = page.next_tx == page.watermark_tx; + let mut by_tx: std::collections::BTreeMap> = Default::default(); + for record in page.records { + by_tx + .entry(record["tx"].as_i64().unwrap_or_default()) + .or_default() + .push(record); + } + let short = short || by_tx.len() < PAGE_TXS; + let Ok(mut ring) = hub.ring.lock() else { + return; + }; + for (tx, records) in by_tx { + let bytes = records.iter().map(|r| r.to_string().len()).sum(); + ring.entries.push_back(Entry { + tx, + records: Arc::new(records), + watermark_tx: page.watermark_tx, + watermark_timestamp: page.watermark_timestamp.clone(), + bytes, + }); + ring.bytes += bytes; + ring.head = tx; + } + while ring.bytes > RING_BYTES { + let Some(old) = ring.entries.pop_front() else { + break; + }; + ring.bytes -= old.bytes; + ring.floor = old.tx; + } + moved = true; + if short { + break; + } + } + if moved && let Ok(ring) = hub.ring.lock() { + hub.head.send_replace(ring.head); + } +} + +#[cfg(test)] +#[path = "feed_hub_tests.rs"] +mod tests; diff --git a/src/server/feed_hub_tests.rs b/src/server/feed_hub_tests.rs new file mode 100644 index 000000000..5cb398a35 --- /dev/null +++ b/src/server/feed_hub_tests.rs @@ -0,0 +1,203 @@ +//! The change feed's shared reader: one store read per committed transaction +//! whatever the number of subscribers, pages identical to a filtered store +//! read, catch-up behind the ring, and an emptied ring on a new log epoch. +use std::sync::Arc; +use std::time::Duration; + +use quipu::store::changes::Capture; +use quipu::store::{Datum, Store}; +use quipu::types::{Op, Value}; +use serde_json::Value as Json; + +use super::{Hub, Serve}; +use crate::{SharedStore, StoreHandle}; + +const T: &str = "2026-10-24T00:00:00Z"; +const BOARD: &str = "urn:graph:board"; +const OTHER: &str = "urn:graph:other"; + +fn write(shared: &SharedStore, graph: &str, entity: &str) { + let mut store = shared.lock(); + let datum = Datum { + entity: store.intern(entity).unwrap(), + attribute: store.intern("urn:p:title").unwrap(), + value: Value::Str(format!("title of {entity}")), + valid_from: T.into(), + valid_to: None, + op: Op::Assert, + }; + // Reuse a registered graph: graph_create commits (and hints) on its own, + // which would make one write two commits. + let g = match store.registered_graph_id(graph).unwrap() { + Some(g) => g, + None => store.graph_create(graph).unwrap(), + }; + store.transact_to_graph(&[datum], T, None, None, g).unwrap(); +} + +fn shared() -> SharedStore { + let shared: SharedStore = Arc::new(StoreHandle::writer_only(Store::open_in_memory().unwrap())); + write(&shared, BOARD, "urn:seed"); + write(&shared, OTHER, "urn:x:seed"); + shared +} + +/// Wait until the hub's ring head reaches the store's latest transaction. +async fn settled(shared: &SharedStore, hub: &Hub) { + let latest = shared.read().latest_tx_id().unwrap(); + let mut head = hub.watch(); + for _ in 0..200 { + if *head.borrow_and_update() >= latest { + return; + } + let _ = tokio::time::timeout(Duration::from_millis(50), head.changed()).await; + } + panic!("hub head never reached {latest}"); +} + +/// Wait until the hub's reader has gone quiet: no read for 100 ms. A write can +/// commit more than one transaction, and a late hint wakes one more (empty) +/// read; measuring before it ends would charge it to the wrong step. +async fn quiet(hub: &Hub) { + let reads = || hub.reads.load(std::sync::atomic::Ordering::Relaxed); + let mut last = reads(); + loop { + tokio::time::sleep(Duration::from_millis(100)).await; + let now = reads(); + if now == last { + return; + } + last = now; + } +} + +/// The page fields that do not depend on when the page was read. +fn stable(page: &Json) -> Json { + serde_json::json!({ + "records": page["records"], "since": page["since"], "next_tx": page["next_tx"], + "capture": page["capture"], "source": page["source"], + }) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn ten_subscribers_cost_one_store_read_per_commit() { + let shared = shared(); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + quiet(&hub).await; + let mut cursors = [shared.read().latest_tx_id().unwrap(); 10]; + let mut scanned = [0; 10]; + let reads = || hub.reads.load(std::sync::atomic::Ordering::Relaxed); + for i in 0..5 { + let before = reads(); + write(&shared, BOARD, &format!("urn:item:{i}")); + settled(&shared, &hub).await; + quiet(&hub).await; + assert_eq!( + reads() - before, + 1, + "commit {i}: one store read, whatever the number of subscribers" + ); + // Serving all ten subscribers reads nothing. + for (cursor, scanned) in cursors.iter_mut().zip(scanned.iter_mut()) { + let Serve::Page(page) = hub.serve(*cursor, Some(BOARD), scanned) else { + panic!("subscriber not served from the ring"); + }; + *cursor = page["next_tx"].as_i64().unwrap(); + } + assert_eq!(reads() - before, 1); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_ring_page_equals_the_filtered_store_page() { + let shared = shared(); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + let start = shared.read().latest_tx_id().unwrap(); + write(&shared, OTHER, "urn:x:1"); + write(&shared, BOARD, "urn:item:1"); + write(&shared, OTHER, "urn:x:2"); + settled(&shared, &hub).await; + for graph in [Some(BOARD), Some(OTHER), None] { + let mut cursor = start; + let mut scanned = 0; + loop { + let served = hub.serve(cursor, graph, &mut scanned); + let store = shared.read(); + let g = graph.map(|iri| store.lookup(iri).unwrap().unwrap()); + let want = store + .changes_after(cursor, 1, Capture::OldAndNewValues, g) + .unwrap() + .to_json(); + match served { + Serve::Page(page) => { + assert_eq!(stable(&page), stable(&want), "graph {graph:?} at {cursor}"); + cursor = page["next_tx"].as_i64().unwrap(); + } + Serve::Wait => { + assert!(want["records"].as_array().unwrap().is_empty()); + break; + } + Serve::CatchUp => panic!("a cursor inside the ring must not catch up"), + } + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_cursor_behind_the_ring_catches_up_from_the_store() { + let shared = shared(); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + let mut scanned = 0; + // The seed transaction predates the hub: the ring starts after it. + assert!(matches!( + hub.serve(0, Some(BOARD), &mut scanned), + Serve::CatchUp + )); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn other_graphs_never_push_a_subscriber_into_catch_up() { + let shared = shared(); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + let cursor = shared.read().latest_tx_id().unwrap(); + let mut scanned = 0; + for i in 0..3 { + write(&shared, OTHER, &format!("urn:x:{i}")); + settled(&shared, &hub).await; + assert!(matches!( + hub.serve(cursor, Some(BOARD), &mut scanned), + Serve::Wait + )); + } + assert_eq!(scanned, shared.read().latest_tx_id().unwrap()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_new_log_epoch_empties_the_ring() { + let shared = shared(); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + let cursor = shared.read().latest_tx_id().unwrap(); + write(&shared, BOARD, "urn:item:1"); + settled(&shared, &hub).await; + let mut scanned = 0; + assert!(matches!( + hub.serve(cursor, Some(BOARD), &mut scanned), + Serve::Page(_) + )); + shared + .lock() + .mint_log_epoch("test: history removed") + .unwrap(); + write(&shared, BOARD, "urn:item:2"); + settled(&shared, &hub).await; + let mut scanned = 0; + assert!( + matches!(hub.serve(cursor, Some(BOARD), &mut scanned), Serve::CatchUp), + "a page of the old epoch was served from memory" + ); +} diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index fa39bd4b2..a2e6bd4a5 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -26,6 +26,9 @@ use super::{ feed, }; +#[path = "feed_hub.rs"] +pub(crate) mod feed_hub; + #[derive(serde::Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct Params { @@ -45,6 +48,12 @@ struct Cursor { terminal: bool, /// Frames of one chunked transaction still to send, in order. pending: VecDeque, + /// A change stream is served from the shared reader (`feed_hub`) once + /// its first page has been read from the store (which validates the + /// graph); it then waits on the hub's head instead of its own hint. + hub: Option<(Arc, watch::Receiver)>, + /// How far the hub's ring has been searched for this stream's graph. + scanned: i64, _slot: OwnedSemaphorePermit, } @@ -106,6 +115,8 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool wait: false, terminal: false, pending: VecDeque::new(), + hub: None, + scanned: 0, _slot: slot, }; let output = unfold(cursor, |mut cursor| async move { @@ -116,7 +127,13 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool return None; } loop { - if cursor.wait { + if cursor.wait + && let Some((_, head)) = cursor.hub.as_mut() + { + // The hub crosses the writer barrier and reads once for every + // stream; this one only waits for the ring's head to move. + let _ = tokio::time::timeout(Duration::from_secs(3600), head.changed()).await; + } else if cursor.wait { // Hourly metadata reconcile covers external writers that did // not install this serving connection's hook. Heartbeats do // not query the store. No per-connection polling loop. @@ -145,7 +162,23 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool cursor, )); } - let page = if cursor.changes { + if let Some((_, head)) = cursor.hub.as_mut() { + head.borrow_and_update(); + } + let served = match &cursor.hub { + Some((hub, _)) if cursor.changes => hub.serve( + cursor.offset, + cursor.params.graph.as_deref(), + &mut cursor.scanned, + ), + _ => feed_hub::Serve::CatchUp, + }; + let page = if let feed_hub::Serve::Page(body) = served { + Ok(axum::Json(body)) + } else if matches!(served, feed_hub::Serve::Wait) { + cursor.wait = true; + continue; + } else if cursor.changes { let params = serde_json::from_value( json!({"since": cursor.offset, "limit": 1, "capture": "old_and_new_values", "graph": cursor.params.graph}), ); @@ -199,6 +232,14 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool )); } }; + // The first store page validated this stream's graph: from here + // on it is served from the shared reader. + let attached = cursor.changes && cursor.hub.is_none(); + if attached { + let hub = feed_hub::Hub::of(&cursor.store); + let head = hub.watch(); + cursor.hub = Some((hub, head)); + } let key = if cursor.changes { "next_tx" } else { @@ -209,7 +250,9 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool .and_then(Value::as_i64) .unwrap_or(cursor.offset); if next <= cursor.offset { - cursor.wait = true; + // A stream that just attached asks the hub before it waits, so + // a commit the hub read meanwhile is not slept through. + cursor.wait = !attached; continue; } let data = body.to_string(); diff --git a/src/server/handle.rs b/src/server/handle.rs index 6639aad01..c16071c8f 100644 --- a/src/server/handle.rs +++ b/src/server/handle.rs @@ -32,6 +32,8 @@ pub(crate) type SharedStore = Arc; pub(crate) struct StoreHandle { /// Coalesced hints only; durable rows remain in the existing SQLite logs. pub(crate) commit_wake: tokio::sync::watch::Sender<()>, + /// The change feed's shared reader, started by the first change stream. + pub(crate) changes_hub: std::sync::OnceLock>, pub(crate) graph_metrics: super::graph_metrics::GraphMetrics, pub(crate) writer: FairMutex, pub(crate) readers: ReadPool, @@ -154,6 +156,7 @@ impl StoreHandle { let commit_wake = Self::commit_wake_for(&store); Self { commit_wake, + changes_hub: std::sync::OnceLock::new(), graph_metrics: super::graph_metrics::GraphMetrics::new(db_path), vector_reads_pooled: store.has_sqlite_vector_backend(), embedding_provider: store.embedding_provider(), @@ -195,6 +198,7 @@ impl StoreHandle { let commit_wake = Self::commit_wake_for(&store); Self { commit_wake, + changes_hub: std::sync::OnceLock::new(), graph_metrics: super::graph_metrics::GraphMetrics::new(":memory:"), vector_reads_pooled: store.has_sqlite_vector_backend(), embedding_provider: store.embedding_provider(), diff --git a/src/server/tests.rs b/src/server/tests.rs index 7f00df7de..495388f3d 100644 --- a/src/server/tests.rs +++ b/src/server/tests.rs @@ -1039,6 +1039,7 @@ pub(super) fn pooled_handle_with_provider( } let handle = super::StoreHandle { commit_wake: super::StoreHandle::commit_wake_for(&store), + changes_hub: std::sync::OnceLock::new(), graph_metrics: super::graph_metrics::GraphMetrics::new(&path), embedding_provider: store.embedding_provider(), search_config: store.search_config().clone(), From 77204004371057e77a52e34d767d2bb33325af03 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 20:41:40 -0400 Subject: [PATCH 2/2] fix(feed): a quiet graph behind the ring floor stops reading the store Review finding on #821 (non-author review): a scoped subscriber that resumes behind the hub's ring floor, for example at its graph's last transaction while other graphs moved the store on, read an empty page from the store. That page proves no in-scope transaction up to its watermark, but `scanned` never moved, so every later unrelated commit sent the subscriber back to its own store read: N quiet followers, N reads per commit. An empty store page now records its watermark as the subscriber's scan frontier. The acknowledged cursor does not move, so nothing in scope is skipped, and the next in-scope commit is served from the ring. Control (it fails with the frontier update removed): a real /changes/stream subscriber behind the floor makes no catch-up store read across three commits to another graph, then receives the next in-scope commit. Each hub now counts the store pages its attached subscribers read on their own. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/server/feed_hub.rs | 3 ++ src/server/feed_hub_tests.rs | 79 ++++++++++++++++++++++++++++++++++++ src/server/feed_stream.rs | 14 +++++++ 3 files changed, 96 insertions(+) diff --git a/src/server/feed_hub.rs b/src/server/feed_hub.rs index 0d2b242c3..0d2773e35 100644 --- a/src/server/feed_hub.rs +++ b/src/server/feed_hub.rs @@ -64,6 +64,8 @@ pub(crate) struct Hub { head: watch::Sender, /// Change pages read by the hub, for the one-read-per-transaction control. pub(crate) reads: AtomicU64, + /// Store pages read by attached subscribers catching up on their own. + pub(crate) catchup_reads: AtomicU64, } impl Hub { @@ -83,6 +85,7 @@ impl Hub { }), head: watch::channel(0).0, reads: AtomicU64::new(0), + catchup_reads: AtomicU64::new(0), }); tokio::spawn(read_loop(Arc::downgrade(store), hub.clone())); hub diff --git a/src/server/feed_hub_tests.rs b/src/server/feed_hub_tests.rs index 5cb398a35..ebc053031 100644 --- a/src/server/feed_hub_tests.rs +++ b/src/server/feed_hub_tests.rs @@ -201,3 +201,82 @@ async fn a_new_log_epoch_empties_the_ring() { "a page of the old epoch was served from memory" ); } + +/// A real `/changes/stream` subscriber; its frames, as `(event, id)`. +fn subscribe( + shared: &SharedStore, + since: i64, + graph: &str, +) -> tokio::sync::mpsc::Receiver<(String, String)> { + use futures::StreamExt; + let params = + serde_json::from_value(serde_json::json!({"since": since, "graph": graph})).unwrap(); + let (frames, received) = tokio::sync::mpsc::channel(64); + let shared = shared.clone(); + tokio::spawn(async move { + let response = super::super::changes_stream( + axum::extract::State(shared), + axum::http::HeaderMap::new(), + axum::extract::Query(params), + ) + .await; + let mut body = response.into_body().into_data_stream(); + let mut buffer = String::new(); + while let Some(Ok(chunk)) = body.next().await { + buffer.push_str(std::str::from_utf8(&chunk).unwrap()); + while let Some(end) = buffer.find("\n\n") { + let block: String = buffer.drain(..end + 2).collect(); + let field = |name: &str| { + block + .lines() + .find_map(|l| l.strip_prefix(name)) + .unwrap_or_default() + .to_string() + }; + let event = field("event: "); + if !event.is_empty() && frames.send((event, field("id: "))).await.is_err() { + return; + } + } + } + }); + received +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_quiet_graph_behind_the_ring_floor_stops_reading_the_store() { + // BOARD's last transaction predates the hub; OTHER moves the store past + // it, so the ring's floor is above the subscriber's cursor. + let shared = shared(); + let board_last = shared.read().latest_tx_id().unwrap() - 1; + write(&shared, OTHER, "urn:x:before"); + let hub = Hub::of(&shared); + settled(&shared, &hub).await; + let mut frames = subscribe(&shared, board_last, BOARD); + quiet(&hub).await; + tokio::time::sleep(Duration::from_millis(200)).await; + let catchup = || hub.catchup_reads.load(std::sync::atomic::Ordering::Relaxed); + let before = catchup(); + + // Commits in other graphs only: served from the ring, no store read. + for i in 0..3 { + write(&shared, OTHER, &format!("urn:x:{i}")); + settled(&shared, &hub).await; + } + quiet(&hub).await; + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + catchup(), + before, + "unrelated commits sent the subscriber to the store" + ); + + // And nothing in scope is lost: the next BOARD commit arrives. + write(&shared, BOARD, "urn:item:after"); + let board_tx = shared.read().latest_tx_id().unwrap(); + let frame = tokio::time::timeout(Duration::from_secs(5), frames.recv()) + .await + .expect("the BOARD commit never arrived") + .unwrap(); + assert_eq!(frame, ("quipu.changes".to_string(), board_tx.to_string())); +} diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index a2e6bd4a5..8972bd2e3 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -179,6 +179,10 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool cursor.wait = true; continue; } else if cursor.changes { + if let Some((hub, _)) = &cursor.hub { + hub.catchup_reads + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } let params = serde_json::from_value( json!({"since": cursor.offset, "limit": 1, "capture": "old_and_new_values", "graph": cursor.params.graph}), ); @@ -250,6 +254,16 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool .and_then(Value::as_i64) .unwrap_or(cursor.offset); if next <= cursor.offset { + // An empty store page proves no in-scope transaction up to its + // watermark: the stream need not search below it again, so a + // quiet graph behind the ring floor reaches the ring instead of + // re-reading the store on every unrelated commit. The cursor + // itself does not move. + if cursor.changes + && let Some(watermark) = body.get("watermark_tx").and_then(Value::as_i64) + { + cursor.scanned = cursor.scanned.max(watermark); + } // A stream that just attached asks the hub before it waits, so // a commit the hub read meanwhile is not slept through. cursor.wait = !attached;