From 74abe2f0fe1a5341c8c7637b79569dd59176be27 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 5 Nov 2026 16:03:15 -0500 Subject: [PATCH] perf(db): split newest-first events reads at a month floor so Postgres prunes partitions (#9091) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Newest-first `events` page reads are now split at the start of the previous UTC month, so Postgres can prune the older partitions whenever the newest ones fill the page. Results are unchanged. **Why.** `events` is range-partitioned by month, and every partition carries the full index set. A page read with no `created_at` lower bound can't prune, so Postgres plans and locks every partition and every partition index: about 200 relations per read with 19 partitions. A backend has 16 fast-path lock slots, and every lock past those goes through the shared lock table. On a production-shaped load rig at pool 200, active backends spent 36% of their time waiting on `LWLock:LockManager` and 1.6% on disk. **What changes.** `query_events_on` (REQ history, bridge `/query`, mentions, rosters, notices) and `get_channel_window_on` (channel windows) now read in up to two passes on the same connection, through one helper (`store/recent_pass.rs`): 1. **Recent pass:** the same read plus `created_at >= `, where the floor is the start of the previous UTC month. When it returns a full page, that page is the answer, because every row at or above the floor sorts before every row below it. 2. **Older pass,** only when the recent pass comes up short: `created_at < `, limited to the rows still missing, appended after the recent rows. The ranges don't overlap, so no row is read twice and the page is exact. For channel windows the limit includes the +1 probe row, so `has_more` stays exact. **One snapshot, one deadline.** The two passes run in one `REPEATABLE READ, READ ONLY` transaction (`ReadTransaction`), so both read the same snapshot. Under `READ COMMITTED`, a replacement committing between the passes (a replaceable event, an artifact head) could move the live row from below the floor to above it, and it would land in neither pass. Inside a transaction the passes run in a savepoint instead; the only callers that read inside a transaction are routed replica reads, which are already `REPEATABLE READ`. Each pass after the first runs under whatever is left of the 20 s read deadline, so a split read never gets more time than an unsplit one. Single-statement reads (COUNT, unsplit pages) keep their old path, now through the same type. Channel windows now take the read deadline too. On a routed reader that is the reader's session deadline. On the writer they used to run under the writer's own session timeout (2 minutes), and now get the 20 s read deadline, like every other event read there. The floor is a SQL literal, not a bind parameter, so Postgres prunes when it plans the statement, cached generic plans included. It changes once a month, so the statement cache holds one statement per pass and shape. `#p` reads put the same bound on `event_mentions.event_created_at`. These reads are not split: - Reads whose own bounds already prune (`since` at or after the floor) or fall entirely before the floor (`until` before it). - `OFFSET` reads, since the offset spans both sides of the floor. - Lookups that match a fixed set of rows wherever they sit: `ids`, `#e`, `#d`, artifact `d` tags. - Fetch-everything reads (`max_limit`, the COUNT fallback). - Reads limited to replaceable or addressable kinds. Those are latest-state rows, often old and few, so a recent pass would rarely fill the page. No protocol, schema, index or client change. **Not covered:** COUNT and the per-row aux `#e` reads, neither of which has a page to fill. The recent pass also has no upper bound, because rows dated in the future must stay visible. So partitions ahead of the current month (the manager keeps 6) and the catch-all are still planned. ## Evidence A production-shaped load rig (6,600 desktop clients connecting cold at 100/s, against a 19-partition `events` table), comparing `main` with this branch. Each row is one arm, so expect run-to-run noise of about ±10%. | Arm | Postgres exec time (top statements) | Errors | Relation locks outside fast path (avg) | |---|---|---|---| | Pool 200 + channel-first index, `main` | 13,217 s | 4.75% | 9,481 | | Pool 200 + channel-first index, this PR | 10,931 s | 3.68% | 4,988 | | Default pool, no extra index, `main` | 6,705 s | 5.96% | — | | Default pool, no extra index, this PR | 6,145 s | 5.15% | — | | Default pool + index, `main` (repeat) | 1,070 s | 0.99% | — | | Default pool + index, this PR (repeat) | 1,013 s | 0.93% | — | Pages that fill above the floor stop planning old partitions, and that halves the locks taken outside the fast path. `LWLock:LockManager` stays at about 41% of active backend samples at pool 200, though. The dominant remaining cost there is the cold unread repair: a multi-channel read with no `since` and a page that never fills. That read needs its own fix (a lookback bound or a better plan) and is out of scope here. An earlier version re-ran the whole read after a short recent pass. That doubled the work on non-filling reads: errors went from 4.75% to 11.88% at pool 200. This version reads only the missing older rows. ## Tests - `recent_pass::tests`: floor arithmetic (including the year wrap), the predicate literals, an eligibility table, and an in-memory exactness test that drives the passes at every limit and pins which passes run with which limits. - `e_tag_filter_shape_tests::split_passes_bound_created_at_with_a_literal` / `recent_pass_skips_lookups`: SQL shape (both the `e.` and `m.` sides for `#p`, no bound on unsplit reads or COUNT) and the exclusions, including offset. - `postgres_tests::split_page_reads_match_the_unsplit_read` (buzz-db event): rows on both sides of the floor. A full recent page is answered alone; short pages at limits 4, 5 and 6 add exactly the missing older rows; offset reads across the floor are not split; a cursor below the floor works. It reads `pg_prepared_statements` to show which passes ran. With another session locking a partition only the older pass plans (and a short `lock_timeout`), a full recent page still answers and a short one returns `Err` rather than a partial page. - `postgres_tests::channel_window_split_matches_the_unsplit_read` (buzz-db thread): the same checks for channel windows, including `has_more` when the recent pass has no probe row, and an `Err` when the older pass fails. - `split_read_passes_share_one_snapshot` (event, both deadline modes) and `channel_window_passes_share_one_snapshot` (thread): another session holds a lock that only the older pass needs, so the read stops between its passes. A replacement then commits that deletes a row below the floor and makes its replacement live above it. The page must still be the first pass's snapshot: the replaced row is present and its replacement is absent. - `read_statements_share_one_deadline`: per-read, session and inside-a-transaction reads. The second statement runs under less than the full deadline, and finishing the read restores the connection's own timeout. - `channel_window_passes_share_one_deadline` (thread): through the production writer entry. The recent pass waits 8 s on a lock and the older pass is blocked for good. The window must be cancelled within the one shared deadline, not a full deadline after its older pass starts. - `channel_scoped_mention_reads_share_one_prepared_statement` now expects one recent and one older statement, shared by every channel count. Mutation-checked. Each of these mutants fails a new Postgres test: dropping the older pass, giving the older pass the full limit, removing the split, splitting offset reads, dropping the window's probe row from the split limit, removing the window's split, swallowing an older-pass error in either path, opening the split read `READ COMMITTED` (events and windows), opening it with a plain `BEGIN`, giving every pass the full deadline, and running a window with no read deadline. ## What got simpler Nothing got simpler overall. This adds one small module (`store/recent_pass.rs`) shared by both read paths. The single-statement deadline helper is now a thin wrapper over `ReadTransaction`, so the begin, savepoint and tighten logic lives in one place. The alternative, a `since` in every client plus a new index per read shape, means more code in more places and does nothing for app versions already shipped. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-db/src/store/event.rs | 618 +++++++++++++++--- crates/buzz-db/src/store/mod.rs | 2 + crates/buzz-db/src/store/recent_pass.rs | 263 ++++++++ crates/buzz-db/src/store/thread.rs | 451 ++++++++++++- crates/buzz-db/src/test_support.rs | 3 + .../src/test_support/recent_pass_tests.rs | 134 ++++ 6 files changed, 1379 insertions(+), 92 deletions(-) create mode 100644 crates/buzz-db/src/store/recent_pass.rs create mode 100644 crates/buzz-db/src/test_support/recent_pass_tests.rs diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 3f5c22bd99..011ccc02b7 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -17,6 +17,7 @@ use buzz_core::{CommunityId, StoredEvent}; use buzz_datastore_tracing::datastore_span; use crate::error::{DbError, Result}; +use crate::store::recent_pass::Range; use crate::{AdmittedTx, Db}; // Compatibility exports preserve the pre-extraction public event-store paths. @@ -582,11 +583,32 @@ pub(crate) async fn query_events_on( // Thread-reply reads must be planned with their root: a cached generic // plan sorts every reply of a large thread. Tag matches likewise. let persistent = !has_tag_values(q) && root.is_none(); - let mut qb = match root { - Some(root) => build_thread_reply_sql(q, root), - None => build_query_events_sql(q), + let limit = page_limit(q); + // Recent-first (see `recent_pass`): the newest partitions answer a page + // that fills there; older rows are read only for what is still missing. + // A thread-reply read is a `#e` lookup, which is never split. + let rows = match (root, recent_pass_floor(q, Utc::now())) { + (Some(root), _) => { + let mut qb = build_thread_reply_sql(q, root); + fetch_with_read_deadline(conn, qb.build().persistent(persistent), deadline).await? + } + (None, None) => { + let mut qb = build_query_events_sql(q, Range::All, limit); + fetch_with_read_deadline(conn, qb.build().persistent(persistent), deadline).await? + } + (None, Some(floor)) => { + let mut read = ReadTransaction::begin(conn, deadline).await?; + let mut rows = Vec::new(); + let mut pass = Some((Range::Recent(floor), limit)); + while let Some((range, pass_limit)) = pass { + let mut qb = build_query_events_sql(q, range, pass_limit); + rows.extend(read.fetch(qb.build().persistent(persistent)).await?); + pass = crate::store::recent_pass::next(range, rows.len(), limit); + } + read.finish().await?; + rows + } }; - let rows = fetch_with_read_deadline(conn, &mut qb, persistent, deadline).await?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -625,6 +647,25 @@ pub(crate) fn answer_without_db(q: &EventQuery) -> Option) -> Option> { + let lookup = q.offset.is_some_and(|o| o > 0) + || q.ids.is_some() + || q.e_tags.is_some() + || q.d_tag.is_some() + || q.d_tags.is_some() + || q.d_tag_values.is_some() + || q.max_limit.is_some(); + if lookup { + return None; + } + crate::store::recent_pass::floor_for(now, q.kinds.as_deref(), q.since, q.until) +} + /// The `LIMIT` [`query_events_on`] applies: the requested page, clamped. fn page_limit(q: &EventQuery) -> i64 { q.limit @@ -632,9 +673,14 @@ fn page_limit(q: &EventQuery) -> i64 { .min(q.max_limit.unwrap_or(DEFAULT_MAX_PAGE_LIMIT)) } -/// Generic SQL for [`query_events_on`], after [`answer_without_db`]. -fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { - let limit_val = page_limit(q); +/// Generic SQL for [`query_events_on`], after [`answer_without_db`]: the rows +/// of `q` inside `range` (see [`crate::store::recent_pass`]), at most +/// `limit_val`. +fn build_query_events_sql( + q: &EventQuery, + range: Range, + limit_val: i64, +) -> QueryBuilder { let offset_val = q.offset.unwrap_or(0); // e-tag reads filter inside a MATERIALIZED CTE and sort/limit outside it. @@ -656,7 +702,7 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { FROM events e \ INNER JOIN event_mentions m ON ", ); - push_mention_join(&mut b, q, p_hex); + push_mention_join(&mut b, q, p_hex, range); b } else { let mut b = QueryBuilder::new(if e_tag_fence { @@ -676,6 +722,9 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { // Use unqualified column names when no join, qualified when joined. let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; push_event_filters(&mut qb, col_prefix, q); + if let Some(bound) = range.predicate(&format!("{col_prefix}created_at")) { + qb.push(bound); + } // Composite ordering for deterministic pagination across ALL callers of // query_events (WebSocket REQ, REST endpoints, canvas, notes, etc.). @@ -910,7 +959,7 @@ async fn is_recorded_reply( // One id can carry a reply row and a stub with a fallback timestamp; // prefer the row that answers the question. .push(" ORDER BY (tm.root_event_id IS NOT NULL) DESC, (e.id IS NOT NULL) DESC LIMIT 1"); - let rows = fetch_with_read_deadline(conn, &mut qb, true, deadline).await?; + let rows = fetch_with_read_deadline(conn, qb.build().persistent(true), deadline).await?; let Some(row) = rows.first() else { return Ok(false); }; @@ -968,7 +1017,12 @@ fn build_thread_reply_sql(q: &EventQuery, root: Vec) -> QueryBuilder, q: &EventQuery, p_hex: &str) { +fn push_mention_join( + b: &mut QueryBuilder, + q: &EventQuery, + p_hex: &str, + range: Range, +) { b.push( "e.community_id = m.community_id AND e.id = m.event_id \ AND e.created_at = m.event_created_at \ @@ -982,6 +1036,9 @@ fn push_mention_join(b: &mut QueryBuilder, q: &EventQuery, p_hex if let Some(s) = q.since { b.push(" AND m.event_created_at >= ").push_bind(s); } + if let Some(bound) = range.predicate("m.event_created_at") { + b.push(bound); + } if let Some(u) = q.until { b.push(" AND m.event_created_at <= ").push_bind(u); } @@ -998,11 +1055,11 @@ fn push_mention_join(b: &mut QueryBuilder, q: &EventQuery, p_hex } } -/// [`READ_STATEMENT_TIMEOUT`] as a literal, so [`BEGIN_WITH_READ_DEADLINE_SQL`] -/// can splice it into a static statement. -macro_rules! read_statement_timeout { +/// [`READ_STATEMENT_TIMEOUT`] in seconds, as a literal, so +/// [`BEGIN_WITH_READ_DEADLINE_SQL`] can splice it into a static statement. +macro_rules! read_statement_timeout_secs { () => { - "20s" + 20 }; } @@ -1012,8 +1069,14 @@ macro_rules! read_statement_timeout { /// once no client can still be waiting for it. It only ever tightens the /// effective `statement_timeout`: a shorter operator or caller cap is kept, /// and a disabled (`0`) or longer one is lowered to 20 s. The limit is per -/// statement, not per request. -pub const READ_STATEMENT_TIMEOUT: &str = read_statement_timeout!(); +/// read, not per request: the passes of a split read share one (see +/// [`ReadTransaction`]). +pub const READ_STATEMENT_TIMEOUT: &str = concat!(read_statement_timeout_secs!(), "s"); + +/// [`READ_STATEMENT_TIMEOUT`] as a duration, for the budget a split read's +/// later passes run under. +pub(crate) const READ_DEADLINE: std::time::Duration = + std::time::Duration::from_secs(read_statement_timeout_secs!()); /// Lower `statement_timeout` to `$1` unless it is already tighter; `$2` /// selects transaction-local (`true`) or session (`false`) scope. @@ -1022,17 +1085,22 @@ pub(crate) const TIGHTEN_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('state OR current_setting('statement_timeout')::interval > $1::text::interval \ THEN $1::text ELSE current_setting('statement_timeout') END, $2)"; -/// `BEGIN` plus the [`TIGHTEN_STATEMENT_TIMEOUT_SQL`] tighten to +/// Opens a [`ReadTransaction`]: `REPEATABLE READ, READ ONLY`, so every +/// statement in it reads one snapshot. +const BEGIN_READ_SQL: &str = "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY"; + +/// [`BEGIN_READ_SQL`] plus the [`TIGHTEN_STATEMENT_TIMEOUT_SQL`] tighten to /// [`READ_STATEMENT_TIMEOUT`], transaction-local, as one simple-query message: /// one round trip instead of two. const BEGIN_WITH_READ_DEADLINE_SQL: &str = concat!( - "BEGIN; SELECT set_config('statement_timeout', \ + "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY; \ + SELECT set_config('statement_timeout', \ CASE WHEN current_setting('statement_timeout')::interval = interval '0' \ OR current_setting('statement_timeout')::interval > interval '", - read_statement_timeout!(), - "' THEN '", - read_statement_timeout!(), - "' ELSE current_setting('statement_timeout') END, true)" + read_statement_timeout_secs!(), + "s' THEN '", + read_statement_timeout_secs!(), + "s' ELSE current_setting('statement_timeout') END, true)" ); /// How an event read runs under [`READ_STATEMENT_TIMEOUT`]. @@ -1062,44 +1130,107 @@ pub async fn tighten_session_read_statement_timeout( .map(|_| ()) } -/// Run an event read under [`READ_STATEMENT_TIMEOUT`] without changing the -/// deadline for anything else on `conn`. Inside an open transaction the read -/// takes a savepoint, so a failed or cancelled read leaves the transaction -/// usable and the read's locks are released when it ends. With -/// [`ReadDeadline::PerRead`] the savepoint tightens the deadline, and an -/// autocommit connection opens a transaction whose `BEGIN` carries the -/// transaction-local tighten. Rolling either back (the read has no writes to -/// keep) reverts the tighten. -async fn fetch_with_read_deadline( +/// The statements of one event read, in one transaction that reads one +/// snapshot under one deadline. +/// +/// A read split at the recent-pass floor (see `recent_pass`) runs its passes +/// as separate statements, and they must see the same rows: under +/// `READ COMMITTED` a replacement that commits between them (a replaceable +/// event, an artifact head) can move a live row from below the floor to +/// above it, out of both passes. Outside a transaction this opens +/// `REPEATABLE READ, READ ONLY`, whose snapshot every statement reads. Inside +/// one it takes a savepoint, so a failed or cancelled statement leaves the +/// transaction usable and its locks are released when the read ends; the +/// callers that read inside a transaction are routed replica reads, already +/// `REPEATABLE READ`. +/// +/// The read runs under [`READ_STATEMENT_TIMEOUT`] without changing the +/// deadline for anything else on the connection, and its statements share +/// it: each after the first runs under what the earlier ones left. +/// [`ReadDeadline::PerRead`] tightens the deadline in the transaction or +/// savepoint. Rolling it back (the read has no writes to keep) reverts every +/// tighten. +pub(crate) struct ReadTransaction<'c> { + tx: sqlx::Transaction<'c, sqlx::Postgres>, + started: std::time::Instant, + statements: usize, +} + +impl<'c> ReadTransaction<'c> { + /// Open the read on `conn`. + pub(crate) async fn begin( + conn: &'c mut sqlx::PgConnection, + deadline: ReadDeadline, + ) -> Result { + use sqlx::Connection as _; + let per_read = deadline == ReadDeadline::PerRead; + let tx = if conn.is_in_transaction() { + let mut tx = conn.begin().await?; + if per_read { + tighten_read_deadline(&mut tx, READ_STATEMENT_TIMEOUT).await?; + } + tx + } else if per_read { + conn.begin_with(BEGIN_WITH_READ_DEADLINE_SQL).await? + } else { + conn.begin_with(BEGIN_READ_SQL).await? + }; + Ok(Self { + tx, + started: std::time::Instant::now(), + statements: 0, + }) + } + + /// Run one statement of the read. + pub(crate) async fn fetch( + &mut self, + query: sqlx::query::Query<'_, sqlx::Postgres, sqlx::postgres::PgArguments>, + ) -> Result> { + if self.statements > 0 { + let left = READ_DEADLINE + .saturating_sub(self.started.elapsed()) + .max(std::time::Duration::from_millis(1)); + tighten_read_deadline(&mut self.tx, &format!("{}ms", left.as_millis())).await?; + } + self.statements += 1; + Ok(query.fetch_all(&mut *self.tx).await?) + } + + /// End the read, rolling back everything it set. + pub(crate) async fn finish(self) -> Result<()> { + Ok(self.tx.rollback().await?) + } +} + +/// Lower the transaction-local `statement_timeout` to `timeout`. +async fn tighten_read_deadline( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + timeout: &str, +) -> Result<()> { + sqlx::query(TIGHTEN_STATEMENT_TIMEOUT_SQL) + .bind(timeout) + .bind(true) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Run a single-statement event read as a [`ReadTransaction`], except on an +/// autocommit session already under the deadline, where the statement alone +/// is the read. +pub(crate) async fn fetch_with_read_deadline( conn: &mut sqlx::PgConnection, - qb: &mut QueryBuilder, - persistent: bool, + query: sqlx::query::Query<'_, sqlx::Postgres, sqlx::postgres::PgArguments>, deadline: ReadDeadline, ) -> Result> { use sqlx::Connection as _; - let mut tx = match (conn.is_in_transaction(), deadline) { - (false, ReadDeadline::Session) => { - return Ok(qb.build().persistent(persistent).fetch_all(conn).await?); - } - (false, ReadDeadline::PerRead) => conn.begin_with(BEGIN_WITH_READ_DEADLINE_SQL).await?, - (true, _) => { - let mut tx = conn.begin().await?; - if deadline == ReadDeadline::PerRead { - sqlx::query(TIGHTEN_STATEMENT_TIMEOUT_SQL) - .bind(READ_STATEMENT_TIMEOUT) - .bind(true) - .execute(&mut *tx) - .await?; - } - tx - } - }; - let rows = qb - .build() - .persistent(persistent) - .fetch_all(&mut *tx) - .await?; - tx.rollback().await?; + if !conn.is_in_transaction() && deadline == ReadDeadline::Session { + return Ok(query.fetch_all(conn).await?); + } + let mut read = ReadTransaction::begin(conn, deadline).await?; + let rows = read.fetch(query).await?; + read.finish().await?; Ok(rows) } @@ -1336,7 +1467,7 @@ pub(crate) async fn count_events_on( let mut qb = count_events_sql(q); // Same transaction-local deadline as `query_events_on`, so a COUNT cannot // stall indefinitely either. - let row = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q), deadline) + let row = fetch_with_read_deadline(conn, qb.build().persistent(!has_tag_values(q)), deadline) .await? .into_iter() .next() @@ -1353,7 +1484,7 @@ fn count_events_sql(q: &EventQuery) -> QueryBuilder { "SELECT COUNT(*) as cnt FROM events e \ INNER JOIN event_mentions m ON ", ); - push_mention_join(&mut b, q, p_hex); + push_mention_join(&mut b, q, p_hex, Range::All); b } else { let mut b = QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE community_id = "); @@ -3161,7 +3292,7 @@ mod e_tag_filter_shape_tests { let mut q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); q.kinds = Some(vec![7]); q.e_tags = Some((0..188).map(|i| format!("{i:064x}")).collect()); - let qb = build_query_events_sql(&q); + let qb = build_query_events_sql(&q, Range::All, page_limit(&q)); let sql = qb.sql().as_str().to_owned(); assert!(sql.starts_with("WITH m AS MATERIALIZED (SELECT ")); @@ -3215,7 +3346,10 @@ mod e_tag_filter_shape_tests { let mut q = routed(); change(&mut q); assert!(thread_reply_root(&q).is_none(), "{label}"); - let sql = build_query_events_sql(&q).sql().as_str().to_owned(); + let sql = build_query_events_sql(&q, Range::All, page_limit(&q)) + .sql() + .as_str() + .to_owned(); assert!(!sql.contains("thread_metadata"), "{label}"); } } @@ -3223,10 +3357,83 @@ mod e_tag_filter_shape_tests { #[test] fn non_e_tag_query_is_not_fenced() { let q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); - let qb = build_query_events_sql(&q); + let qb = build_query_events_sql(&q, Range::All, page_limit(&q)); assert!(!qb.sql().as_str().contains("MATERIALIZED")); } + /// Each pass bounds `created_at` with a literal so Postgres prunes + /// partitions while planning; a `#p` read bounds the mention side too. + /// An unsplit read and the COUNT carry no bound. + #[test] + fn split_passes_bound_created_at_with_a_literal() { + let floor = chrono::TimeZone::with_ymd_and_hms(&Utc, 2026, 10, 1, 0, 0, 0).unwrap(); + let lit = "'2026-10-01T00:00:00Z'::timestamptz"; + let sql = |q: &EventQuery, range| { + build_query_events_sql(q, range, 7) + .sql() + .as_str() + .to_owned() + }; + let mut q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); + q.kinds = Some(vec![9]); + let recent = sql(&q, Range::Recent(floor)); + assert!( + recent.contains(&format!(" AND created_at >= {lit}")), + "{recent}" + ); + let older = sql(&q, Range::Older(floor)); + assert!( + older.contains(&format!(" AND created_at < {lit}")), + "{older}" + ); + let whole = sql(&q, Range::All); + assert!(!whole.contains("::timestamptz"), "{whole}"); + + q.p_tag_hex = Some("ab".repeat(32)); + for (range, op) in [(Range::Recent(floor), ">="), (Range::Older(floor), "<")] { + let mention = sql(&q, range); + for column in ["e.created_at", "m.event_created_at"] { + let bound = format!(" AND {column} {op} {lit}"); + assert!(mention.contains(&bound), "{mention}"); + } + } + let count = count_events_sql(&q).sql().as_str().to_owned(); + assert!(!count.contains("::timestamptz"), "{count}"); + } + + /// Which reads are split: page reads are; offset reads, lookups that + /// match a fixed row set and fetch-everything reads are not. + #[test] + fn recent_pass_skips_lookups() { + let now = Utc::now(); + let base = || { + let mut q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); + q.kinds = Some(vec![9]); + q + }; + assert!(recent_pass_floor(&base(), now).is_some(), "page read"); + let mut first_page = base(); + first_page.offset = Some(0); + assert!(recent_pass_floor(&first_page, now).is_some(), "offset 0"); + type Narrow = fn(&mut EventQuery); + let lookups: [(&str, Narrow); 7] = [ + ("offset", |q| q.offset = Some(1)), + ("ids", |q| q.ids = Some(vec![vec![0; 32]])), + ("e tags", |q| q.e_tags = Some(vec!["00".repeat(32)])), + ("d tag", |q| q.d_tag = Some("x".into())), + ("d tags", |q| q.d_tags = Some(vec!["x".into()])), + ("artifact d tags", |q| { + q.d_tag_values = Some(vec!["x".into()]) + }), + ("max limit", |q| q.max_limit = Some(10_000)), + ]; + for (label, set) in lookups { + let mut q = base(); + set(&mut q); + assert!(recent_pass_floor(&q, now).is_none(), "{label}"); + } + } + /// `#p` reads bound the mention side (bb-public runaway REQ): the join /// carries the event's `created_at`, and `since`/`until` also bound /// `m.event_created_at`, for both the page SELECT and the COUNT. @@ -3238,7 +3445,10 @@ mod e_tag_filter_shape_tests { q.since = Some(Utc::now()); q.until = Some(Utc::now()); q.kinds = Some(vec![9]); - let select = build_query_events_sql(&q).sql().as_str().to_owned(); + let select = build_query_events_sql(&q, Range::All, page_limit(&q)) + .sql() + .as_str() + .to_owned(); let count = count_events_sql(&q).sql().as_str().to_owned(); for (label, sql) in [("select", select), ("count", count)] { assert!( @@ -3259,7 +3469,10 @@ mod e_tag_filter_shape_tests { q.since = None; q.until = None; q.kinds = None; - let unbounded = build_query_events_sql(&q).sql().as_str().to_owned(); + let unbounded = build_query_events_sql(&q, Range::All, page_limit(&q)) + .sql() + .as_str() + .to_owned(); assert!(!unbounded.contains("m.event_created_at >=")); assert!(!unbounded.contains("m.event_created_at <=")); assert!(!unbounded.contains("m.event_kind")); @@ -3278,7 +3491,10 @@ mod e_tag_filter_shape_tests { q.channel_ids = Some((0..n).map(|_| uuid::Uuid::new_v4()).collect()); q.channel_ids_include_global = include_global; ( - build_query_events_sql(&q).sql().as_str().to_owned(), + build_query_events_sql(&q, Range::All, page_limit(&q)) + .sql() + .as_str() + .to_owned(), count_events_sql(&q).sql().as_str().to_owned(), ) }; @@ -3432,10 +3648,12 @@ mod postgres_tests { .await .expect("mention read"); } + // Each read is a recent pass (empty here) and then the older pass: one + // statement of each, shared by every channel count. assert_eq!( prepared_like(&mut conn, "%JOIN event_mentions%").await, - 1, - "one statement for every channel count" + 2, + "one recent and one older statement for every channel count" ); drop(conn); @@ -3501,13 +3719,250 @@ mod postgres_tests { } } + /// A newest-first page read split at the floor returns the rows the + /// unsplit read returns: the recent pass alone when it fills the page, + /// otherwise followed by the older pass for just the missing rows. + /// Prepared statements show which passes ran. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn split_page_reads_match_the_unsplit_read() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let channel = make_test_channel(&pool, community_uuid, None).await; + let db = crate::Db::from_pool(pool.clone()); + let author = Keys::generate(); + let now = Utc::now().timestamp() as u64; + let floor = crate::store::recent_pass::floor_at(Utc::now()).timestamp() as u64; + // Three rows inside the recent window, two before it, newest first. + let mut newest_first = Vec::new(); + for at in [ + now - 10, + now - 20, + now - 30, + floor - 86_400, + floor - 2 * 86_400, + ] { + let event = EventBuilder::new(Kind::Custom(9), format!("at {at}")) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&author) + .expect("sign"); + db.insert_event(community, &event, Some(channel)) + .await + .expect("insert"); + newest_first.push(event.id); + } + + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::Connection::clear_cached_statements(&mut *conn) + .await + .expect("clear prepared statements"); + let read = |limit: i64, offset: i64| EventQuery { + kinds: Some(vec![9]), + channel_id: Some(channel), + limit: Some(limit), + offset: Some(offset), + ..EventQuery::for_community(community) + }; + let ids = |rows: Vec| rows.into_iter().map(|r| r.event.id).collect::>(); + let recent = "%FROM events WHERE%created_at >= '%::timestamptz%"; + let older = "%FROM events WHERE%created_at < '%::timestamptz%"; + let all = "%FROM events WHERE%"; + + // Full recent page: answered by the recent pass alone. + let page = query_events_on(&mut conn, &read(3, 0), ReadDeadline::PerRead) + .await + .expect("recent page"); + assert_eq!(ids(page), newest_first[..3]); + assert_eq!(prepared_like(&mut conn, recent).await, 1, "recent pass ran"); + assert_eq!(prepared_like(&mut conn, all).await, 1, "nothing else ran"); + + // Short recent pages: the older pass adds only the missing rows. + for limit in [4, 5, 6] { + let page = query_events_on(&mut conn, &read(limit, 0), ReadDeadline::PerRead) + .await + .expect("split page"); + let expected = &newest_first[..newest_first.len().min(limit as usize)]; + assert_eq!(ids(page), expected, "limit {limit}"); + } + assert_eq!(prepared_like(&mut conn, older).await, 1, "older pass ran"); + assert_eq!(prepared_like(&mut conn, all).await, 2, "no unsplit read"); + + // An offset spans the floor, so the read is not split. + let page = query_events_on(&mut conn, &read(2, 1), ReadDeadline::PerRead) + .await + .expect("offset page"); + assert_eq!(ids(page), newest_first[1..3]); + let page = query_events_on(&mut conn, &read(3, 2), ReadDeadline::PerRead) + .await + .expect("offset page across the floor"); + assert_eq!(ids(page), newest_first[2..5]); + assert_eq!(prepared_like(&mut conn, all).await, 3, "unsplit read ran"); + + // A cursor below the floor is not split either: still exact. + let mut below = read(5, 0); + below.until = Some(DateTime::from_timestamp((floor - 1) as i64, 0).expect("until")); + let page = query_events_on(&mut conn, &below, ReadDeadline::PerRead) + .await + .expect("cursor page"); + assert_eq!(ids(page), newest_first[3..]); + + // A failed older pass fails the read; it never passes the recent rows + // off as a short page. Another session locks the partition of a row + // from before every dated monthly, which only the older pass plans + // (the recent rows' partition can hold rows below the floor too). + let ancient = EventBuilder::new(Kind::Custom(9), "ancient") + .custom_created_at(nostr::Timestamp::from(1_748_736_000)) // 2025-06-01 + .sign_with_keys(&author) + .expect("sign"); + db.insert_event(community, &ancient, Some(channel)) + .await + .expect("insert ancient"); + let partition_of = |id: nostr::EventId| { + let pool = pool.clone(); + async move { + sqlx::query_scalar::<_, String>( + "SELECT tableoid::regclass::text FROM events WHERE id = $1", + ) + .bind(id.as_bytes().to_vec()) + .fetch_one(&pool) + .await + .expect("partition") + } + }; + let old_partition = partition_of(ancient.id).await; + assert_ne!(old_partition, partition_of(newest_first[0]).await); + let mut locker = pool.begin().await.expect("begin locker"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "LOCK TABLE {old_partition} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(&mut *locker) + .await + .expect("lock old partition"); + sqlx::query("SET lock_timeout = '200ms'") + .execute(&mut *conn) + .await + .expect("set lock_timeout"); + let page = query_events_on(&mut conn, &read(3, 0), ReadDeadline::PerRead) + .await + .expect("a full recent page never touches the old partition"); + assert_eq!(ids(page), newest_first[..3]); + let failed = query_events_on(&mut conn, &read(5, 0), ReadDeadline::PerRead).await; + assert!( + failed.is_err(), + "a failed older pass must fail the read, got {:?} rows", + failed.map(|rows| rows.len()) + ); + locker.rollback().await.expect("release lock"); + sqlx::query("RESET lock_timeout") + .execute(&mut *conn) + .await + .expect("reset lock_timeout"); + } + + /// The passes of a split read share one snapshot: a replacement that + /// commits between them neither drops the row it replaces nor shows the + /// row it adds. Both deadline modes open the read the same way. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn split_read_passes_share_one_snapshot() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + for deadline in [ReadDeadline::PerRead, ReadDeadline::Session] { + let channel = make_test_channel(&pool, community_uuid, None).await; + let q = EventQuery { + kinds: Some(vec![9]), + channel_id: Some(channel), + limit: Some(10), + ..EventQuery::for_community(community) + }; + let (page, expected) = crate::test_support::replace_between_passes( + &pool, + community, + channel, + move |mut conn| async move { + query_events_on(&mut conn, &q, deadline) + .await + .expect("split read") + .into_iter() + .map(|stored| stored.event.id) + .collect::>() + }, + ) + .await; + assert_eq!(page, expected, "{deadline:?}"); + } + } + + /// The statements of one read share its deadline: each after the first + /// runs under what the earlier ones left, and ending the read restores + /// the connection's own. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_statements_share_one_deadline() { + use sqlx::Connection as _; + let pool = setup_pool().await; + let timeout_ms = || { + sqlx::query( + "SELECT (EXTRACT(EPOCH FROM current_setting('statement_timeout')::interval) \ + * 1000)::float8", + ) + }; + let full = READ_DEADLINE.as_millis() as f64; + for case in ["per read", "session", "inside a transaction"] { + let mut conn = pool.acquire().await.expect("acquire").detach(); + let deadline = match case { + "session" => { + tighten_session_read_statement_timeout(&mut conn) + .await + .expect("session deadline"); + ReadDeadline::Session + } + _ => ReadDeadline::PerRead, + }; + let mut outer = None; + let conn: &mut sqlx::PgConnection = if case == "inside a transaction" { + outer = Some(conn.begin().await.expect("begin outer")); + outer.as_mut().expect("outer") + } else { + &mut conn + }; + let before: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut *conn) + .await + .expect("show timeout"); + let mut read = ReadTransaction::begin(&mut *conn, deadline) + .await + .expect("begin read"); + let first: f64 = read.fetch(timeout_ms()).await.expect("first")[0].get(0); + let second: f64 = read.fetch(timeout_ms()).await.expect("second")[0].get(0); + read.finish().await.expect("finish"); + assert_eq!( + first, full, + "{case}: the first statement gets the whole deadline" + ); + assert!( + (1.0..full).contains(&second), + "{case}: the second gets what is left, got {second} ms" + ); + let after: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut *conn) + .await + .expect("show timeout"); + assert_eq!(after, before, "{case}: restored"); + drop(outer); + } + } + /// Effective `statement_timeout` observed *inside* the read deadline, and /// the value restored on `conn` afterwards. async fn deadline_inside_and_after(conn: &mut sqlx::PgConnection) -> (String, String) { let mut qb = QueryBuilder::new("SELECT current_setting('statement_timeout')"); - let rows = fetch_with_read_deadline(conn, &mut qb, true, ReadDeadline::PerRead) - .await - .expect("deadline read"); + let rows = + fetch_with_read_deadline(conn, qb.build().persistent(true), ReadDeadline::PerRead) + .await + .expect("deadline read"); let inside: String = rows[0].get(0); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *conn) @@ -3969,7 +4424,12 @@ mod postgres_tests { .expect("set session timeout"); // Autocommit connection (writer path). let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_read_deadline(&mut conn, &mut qb, true, ReadDeadline::PerRead).await; + let err = fetch_with_read_deadline( + &mut conn, + qb.build().persistent(true), + ReadDeadline::PerRead, + ) + .await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *conn) @@ -3981,7 +4441,9 @@ mod postgres_tests { // the failed savepoint must roll back and leave the outer tx usable. let mut tx = sqlx::Connection::begin(&mut *conn).await.expect("begin"); let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::PerRead).await; + let err = + fetch_with_read_deadline(&mut tx, qb.build().persistent(true), ReadDeadline::PerRead) + .await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *tx) @@ -4013,9 +4475,10 @@ mod postgres_tests { let mut qb = QueryBuilder::new( "SELECT current_setting('statement_timeout'), (SELECT count(*) FROM events)", ); - let rows = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::Session) - .await - .expect("session read"); + let rows = + fetch_with_read_deadline(&mut tx, qb.build().persistent(true), ReadDeadline::Session) + .await + .expect("session read"); let inside: String = rows[0].get(0); assert_eq!(inside, "1min", "Session must not tighten"); let held: i64 = sqlx::query_scalar( @@ -4028,9 +4491,10 @@ mod postgres_tests { assert_eq!(held, 0, "a Session read must release its locks"); let mut qb = QueryBuilder::new("SELECT current_setting('statement_timeout')"); - let rows = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::PerRead) - .await - .expect("per-read read"); + let rows = + fetch_with_read_deadline(&mut tx, qb.build().persistent(true), ReadDeadline::PerRead) + .await + .expect("per-read read"); let inside: String = rows[0].get(0); assert_eq!(inside, READ_STATEMENT_TIMEOUT, "PerRead tightens"); tx.rollback().await.expect("rollback"); diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1ff20aa376..b5578cb796 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -28,6 +28,8 @@ pub mod event; pub(crate) mod event_follow_up; /// Per-community NIP-FI issuer bindings. pub mod nip_fi_binding; +/// Recent-first passes for newest-first `events` page reads. +pub(crate) mod recent_pass; /// User-specific deletion submission and inventory preparation. pub mod user_deletion; /// Operator-global user restrictions and identity-lock primitives. diff --git a/crates/buzz-db/src/store/recent_pass.rs b/crates/buzz-db/src/store/recent_pass.rs new file mode 100644 index 0000000000..d8a14646e1 --- /dev/null +++ b/crates/buzz-db/src/store/recent_pass.rs @@ -0,0 +1,263 @@ +//! Recent-first newest-first page reads over the partitioned `events` table. +//! +//! `events` is range-partitioned by month on `created_at`, and every partition +//! carries the full index set. A read with no `created_at` lower bound cannot +//! prune partitions, so Postgres plans and locks every partition and every +//! partition index: about 200 relations. Only 16 fit a backend's fast-path +//! lock slots; the rest go through the shared lock table, whose partition +//! locks every backend contends on (`LWLock:LockManager`). +//! +//! A newest-first page (`ORDER BY created_at DESC … LIMIT n`) usually lies in +//! the newest partitions, so it is split at a floor ([`floor_for`], [`next`]), +//! the start of the previous UTC month: +//! +//! 1. The *recent* pass reads `created_at >= floor`. When it returns `n` rows, +//! they are the page: every row at or above the floor sorts before every +//! row below it. +//! 2. Otherwise the *older* pass reads `created_at < floor` for the rows +//! still missing, and they follow the recent rows. The two ranges are +//! disjoint, so no row is read twice and the page is exact, provided both +//! passes read one snapshot (the stores run them in one `REPEATABLE READ` +//! transaction, `ReadTransaction`). +//! +//! The floor is a SQL literal, so Postgres prunes when it plans the statement, +//! cached generic plans included. It changes once a month, so the +//! prepared-statement cache keeps one statement per pass and shape. + +use buzz_core::kind::{is_parameterized_replaceable, is_replaceable}; +use chrono::{DateTime, Datelike, TimeZone, Utc}; + +/// The `created_at` range one pass of a page read covers. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Range { + /// Every row: the read is not split. + All, + /// `created_at >= floor`. + Recent(DateTime), + /// `created_at < floor`. + Older(DateTime), +} + +impl Range { + /// The range as a SQL predicate on `column` (` AND …`), or `None` for + /// [`Range::All`]. + /// + /// The floor is a literal, not a bind: Postgres prunes partitions on a + /// literal while it plans, including in a cached generic plan, while a + /// generic plan with a bind parameter prunes only at execution, after it + /// has locked every partition. The text is chrono's fixed-format + /// rendering of a UTC timestamp, digits and separators only, so it + /// carries no caller input. + pub(crate) fn predicate(self, column: &str) -> Option { + let (op, floor) = match self { + Range::All => return None, + Range::Recent(floor) => (">=", floor), + Range::Older(floor) => ("<", floor), + }; + Some(format!( + " AND {column} {op} '{}'::timestamptz", + floor.format("%Y-%m-%dT%H:%M:%SZ") + )) + } +} + +/// The split point at `now`: 00:00 UTC on the first day of the previous +/// month, so the recent pass reads the current and previous monthly +/// partitions and the ones ahead of them. +pub(crate) fn floor_at(now: DateTime) -> DateTime { + let (year, month) = match now.month() { + 1 => (now.year() - 1, 12), + m => (now.year(), m - 1), + }; + Utc.with_ymd_and_hms(year, month, 1, 0, 0, 0) + .single() + .unwrap_or(DateTime::::MIN_UTC) +} + +/// The floor to split a page read at, or `None` to read it whole: its own +/// bounds already exclude everything older than the floor or everything +/// newer, or its rows are latest-state records that rarely fill a page above +/// the floor, so a split would only add a second statement. +/// +/// `kinds` are the read's kinds (`None`: any); `since`/`until` its bounds. +pub(crate) fn floor_for( + now: DateTime, + kinds: Option<&[i32]>, + since: Option>, + until: Option>, +) -> Option> { + let floor = floor_at(now); + // Replaceable and addressable rows are latest-state records, often months + // old, and few per author. + let latest_state_only = kinds.is_some_and(|ks| { + !ks.is_empty() + && ks.iter().all(|&k| { + u32::try_from(k).is_ok_and(|k| is_replaceable(k) || is_parameterized_replaceable(k)) + }) + }); + let since_prunes = since.is_some_and(|s| s >= floor); + let until_before_floor = until.is_some_and(|u| u < floor); + (!latest_state_only && !since_prunes && !until_before_floor).then_some(floor) +} + +/// The pass to run after `range` once the read holds `found` of its `limit` +/// rows, with that pass's limit: the older pass for the rows still missing +/// after a short recent pass, otherwise none. `limit` is the SQL `LIMIT`. +pub(crate) fn next(range: Range, found: usize, limit: i64) -> Option<(Range, i64)> { + let Range::Recent(floor) = range else { + return None; + }; + let found = i64::try_from(found).unwrap_or(i64::MAX); + (found < limit).then(|| (Range::Older(floor), limit - found)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn at(y: i32, m: u32, d: u32) -> DateTime { + Utc.with_ymd_and_hms(y, m, d, 12, 34, 56).single().unwrap() + } + + fn midnight(y: i32, m: u32, d: u32) -> DateTime { + Utc.with_ymd_and_hms(y, m, d, 0, 0, 0).single().unwrap() + } + + #[test] + fn floor_is_the_start_of_the_previous_month() { + assert_eq!(floor_at(at(2026, 11, 4)), midnight(2026, 10, 1)); + assert_eq!(floor_at(at(2026, 11, 30)), midnight(2026, 10, 1)); + assert_eq!(floor_at(at(2027, 1, 1)), midnight(2026, 12, 1)); + assert_eq!(floor_at(at(2026, 3, 31)), midnight(2026, 2, 1)); + } + + #[test] + fn predicates_are_fixed_utc_literals() { + let floor = midnight(2026, 10, 1); + assert_eq!(Range::All.predicate("e.created_at"), None); + assert_eq!( + Range::Recent(floor).predicate("e.created_at").as_deref(), + Some(" AND e.created_at >= '2026-10-01T00:00:00Z'::timestamptz") + ); + assert_eq!( + Range::Older(floor).predicate("created_at").as_deref(), + Some(" AND created_at < '2026-10-01T00:00:00Z'::timestamptz") + ); + } + + /// Which reads are split: (kinds, since, until) → split? + #[test] + fn split_eligibility() { + let now = at(2026, 11, 4); + let floor = floor_at(now); + let day = chrono::Duration::days(1); + // (label, kinds, since, until, split) + type Case<'a> = ( + &'a str, + Option<&'a [i32]>, + Option>, + Option>, + bool, + ); + let cases: &[Case] = &[ + ("unbounded messages", Some(&[9, 40002]), None, None, true), + ("any kind", None, None, None, true), + ( + "since older than the floor", + Some(&[9]), + Some(floor - day), + None, + true, + ), + ( + "since at the floor already prunes", + Some(&[9]), + Some(floor), + None, + false, + ), + ( + "recent since already prunes", + Some(&[9]), + Some(now - day), + None, + false, + ), + ("until at the floor", Some(&[9]), None, Some(floor), true), + ( + "until before the floor", + Some(&[9]), + None, + Some(floor - day), + false, + ), + ("replaceable only", Some(&[0, 10002]), None, None, false), + ("addressable only", Some(&[39002]), None, None, false), + ( + "mixed with a message kind", + Some(&[39002, 9]), + None, + None, + true, + ), + ( + "negative kind is not replaceable", + Some(&[-1]), + None, + None, + true, + ), + ]; + for &(label, kinds, since, until, split) in cases { + let expected = split.then_some(floor); + assert_eq!(floor_for(now, kinds, since, until), expected, "{label}"); + } + } + + /// The passes against an in-memory table of `created_at` values, newest + /// first, driven the way the stores drive them: the newest `limit` rows, + /// in order, for every limit, split or not, reading each row at most once. + #[test] + fn passes_return_the_newest_rows_and_read_each_row_once() { + let day = chrono::Duration::days(1); + let floor = midnight(2026, 10, 1); + // Three at or above the floor (one exactly at it), two below. + let table = [ + floor + day * 3, + floor + day, + floor, + floor - day, + floor - day * 9, + ]; + for limit in 0..=table.len() as i64 + 1 { + let expected: Vec<_> = table.iter().copied().take(limit as usize).collect(); + for split in [None, Some(floor)] { + let mut passes = Vec::new(); + let mut rows = Vec::new(); + let mut pass = Some((split.map_or(Range::All, Range::Recent), limit)); + while let Some((range, n)) = pass { + passes.push((range, n)); + rows.extend( + table + .iter() + .copied() + .filter(|&t| match range { + Range::All => true, + Range::Recent(f) => t >= f, + Range::Older(f) => t < f, + }) + .take(n as usize), + ); + pass = next(range, rows.len(), limit); + } + assert_eq!(rows, expected, "split {split:?}, limit {limit}"); + let expected_passes = match split { + None => vec![(Range::All, limit)], + Some(f) if limit <= 3 => vec![(Range::Recent(f), limit)], + Some(f) => vec![(Range::Recent(f), limit), (Range::Older(f), limit - 3)], + }; + assert_eq!(passes, expected_passes, "split {split:?}, limit {limit}"); + } + } + } +} diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 7c617aba6a..d5407cbf75 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -9,6 +9,8 @@ use chrono::{DateTime, Utc}; use sqlx::{PgPool, Row}; use uuid::Uuid; +use crate::store::recent_pass::Range; + use buzz_datastore_tracing::datastore_span; use buzz_core::CommunityId; @@ -580,6 +582,7 @@ pub async fn get_channel_window( limit, cursor, kind_filter, + crate::event::ReadDeadline::PerRead, ) .await } @@ -588,7 +591,8 @@ pub async fn get_channel_window( /// runs the page (and its participants batch) on the exact reader connection /// whose heartbeat observation proved coverage (the proof is /// connection-local; a different pooled session may sit at a different -/// replay position). +/// replay position). The page runs under the event read deadline, shared by +/// its passes (see `ReadTransaction`). pub(crate) async fn get_channel_window_on( conn: &mut sqlx::PgConnection, community_id: CommunityId, @@ -596,6 +600,7 @@ pub(crate) async fn get_channel_window_on( limit: u32, cursor: Option<(DateTime, Vec)>, kind_filter: Option<&[u32]>, + deadline: crate::event::ReadDeadline, ) -> Result { let mut param_idx = 3u32; // $1 is community_id, $2 is channel_id let mut sql = String::from( @@ -651,20 +656,58 @@ pub(crate) async fn get_channel_window_on( } } - sql.push_str(&format!( - " ORDER BY e.created_at DESC, e.id ASC LIMIT ${param_idx}" - )); - - let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)) - .bind(community_id.as_uuid()) - .bind(channel_id); - if let Some((ts, id)) = &cursor { - q = q.bind(*ts).bind(id.clone()); - } + // Recent-first (see `recent_pass`): the newest partitions answer a window + // that fills there (with its +1 probe row); older rows are read only for + // what is still missing. + let kinds: Option> = + kind_filter.map(|ks| ks.iter().filter_map(|&k| i32::try_from(k).ok()).collect()); + let floor = crate::store::recent_pass::floor_for( + Utc::now(), + kinds.as_deref(), + None, + cursor.as_ref().map(|(ts, _)| *ts), + ); // The +1 probe row is the server-internal has_more evidence. - q = q.bind(limit as i64 + 1); - - let mut db_rows = q.fetch_all(&mut *conn).await?; + let probe_limit = i64::from(limit) + 1; + let pass_query = |range: Range, pass_limit: i64| { + let mut pass_sql = sql.clone(); + if let Some(bound) = range.predicate("e.created_at") { + pass_sql.push_str(&bound); + } + pass_sql.push_str(&format!( + " ORDER BY e.created_at DESC, e.id ASC LIMIT ${param_idx}" + )); + let mut q = sqlx::query(sqlx::AssertSqlSafe(pass_sql)) + .bind(community_id.as_uuid()) + .bind(channel_id); + if let Some((ts, id)) = &cursor { + q = q.bind(*ts).bind(id.clone()); + } + q.bind(pass_limit) + }; + let mut db_rows = match floor { + None => { + crate::event::fetch_with_read_deadline( + &mut *conn, + pass_query(Range::All, probe_limit), + deadline, + ) + .await? + } + Some(floor) => { + // One snapshot and one deadline for both passes (see + // `ReadTransaction`). + let mut read = crate::event::ReadTransaction::begin(conn, deadline).await?; + let mut rows = Vec::new(); + let mut pass = Some((Range::Recent(floor), probe_limit)); + while let Some((range, pass_limit)) = pass { + rows.extend(read.fetch(pass_query(range, pass_limit)).await?); + pass = crate::store::recent_pass::next(range, rows.len(), probe_limit); + } + read.finish().await?; + rows + } + }; let has_more = db_rows.len() > limit as usize; db_rows.truncate(limit as usize); @@ -1092,6 +1135,7 @@ impl Db { limit, cursor.clone(), kind_filter, + crate::event::ReadDeadline::Session, ) .await { @@ -1164,7 +1208,7 @@ mod postgres_tests { use super::*; use crate::{ channel::{ChannelType, ChannelVisibility}, - event::{insert_event_with_thread_metadata, ThreadMetadataParams}, + event::{insert_event_with_thread_metadata, ReadDeadline, ThreadMetadataParams}, }; use nostr::{EventBuilder, Keys, Kind}; @@ -2252,6 +2296,383 @@ mod postgres_tests { assert!(page2.next_cursor.is_none(), "no cursor past the last row"); } + /// The window is split at the floor: a recent pass that fills the page + /// and its +1 probe row answers alone; otherwise the older pass reads the + /// missing rows, so `has_more` still sees rows before the floor. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_window_split_matches_the_unsplit_read() { + use nostr::Timestamp; + + let pool = setup_pool().await; + let author = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + &format!("window-recent-{}", Uuid::new_v4()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + let now = Utc::now().timestamp() as u64; + let floor = crate::store::recent_pass::floor_at(Utc::now()).timestamp() as u64; + // Two rows inside the recent window and two before it, newest first. + let mut newest_first = Vec::new(); + for at in [now - 10, now - 20, floor - 86_400, floor - 2 * 86_400] { + let event = EventBuilder::new(Kind::Custom(9), format!("at {at}")) + .custom_created_at(Timestamp::from(at)) + .sign_with_keys(&author) + .expect("sign"); + insert_root(&pool, community, channel.id, &event).await; + newest_first.push(event.id); + } + + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::Connection::clear_cached_statements(&mut *conn) + .await + .expect("clear prepared statements"); + let prepared = |like: &'static str| { + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM pg_prepared_statements WHERE statement LIKE $1", + ) + .bind(like) + }; + let recent = "%FROM events e%created_at >= '%::timestamptz%"; + let older = "%FROM events e%created_at < '%::timestamptz%"; + let all = "%FROM events e%"; + let ids = |w: &ChannelWindow| { + w.rows + .iter() + .map(|r| r.stored_event.event.id) + .collect::>() + }; + + // Limit 1: two recent rows fill the page and its probe row. + let window = get_channel_window_on( + &mut conn, + community, + channel.id, + 1, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("recent window"); + assert_eq!(ids(&window), newest_first[..1]); + assert!(window.has_more); + let recent_count = prepared(recent).fetch_one(&mut *conn).await.expect("count"); + let all_count = prepared(all).fetch_one(&mut *conn).await.expect("count"); + assert_eq!((recent_count, all_count), (1, 1), "recent pass alone"); + + // Limit 2: the recent pass has no probe row, so the older pass reads + // it and `has_more` sees the rows before the floor. + let window = get_channel_window_on( + &mut conn, + community, + channel.id, + 2, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("split window"); + assert_eq!(ids(&window), newest_first[..2]); + assert!(window.has_more, "older rows lie before the floor"); + let older_count = prepared(older).fetch_one(&mut *conn).await.expect("count"); + let all_count = prepared(all).fetch_one(&mut *conn).await.expect("count"); + assert_eq!((older_count, all_count), (1, 2), "older pass ran"); + + // The whole channel, then a cursor page below the floor. + let window = get_channel_window_on( + &mut conn, + community, + channel.id, + 4, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("whole window"); + assert_eq!(ids(&window), newest_first); + assert!(!window.has_more); + let cursor = get_channel_window_on( + &mut conn, + community, + channel.id, + 2, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("first page") + .next_cursor + .expect("cursor"); + let window = get_channel_window_on( + &mut conn, + community, + channel.id, + 2, + Some(cursor), + None, + ReadDeadline::PerRead, + ) + .await + .expect("older page"); + assert_eq!(ids(&window), newest_first[2..]); + assert!(!window.has_more); + + // A failed older pass fails the window; it never reports the recent + // rows as the whole channel. Another session locks the partition of a + // root from before every dated monthly, which only the older pass + // plans (the recent rows' partition can hold rows below the floor too). + let ancient = EventBuilder::new(Kind::Custom(9), "ancient") + .custom_created_at(Timestamp::from(1_748_736_000)) // 2025-06-01 + .sign_with_keys(&author) + .expect("sign"); + insert_root(&pool, community, channel.id, &ancient).await; + let partition_of = |id: nostr::EventId| { + let pool = pool.clone(); + async move { + sqlx::query_scalar::<_, String>( + "SELECT tableoid::regclass::text FROM events WHERE id = $1", + ) + .bind(id.as_bytes().to_vec()) + .fetch_one(&pool) + .await + .expect("partition") + } + }; + let old_partition = partition_of(ancient.id).await; + assert_ne!(old_partition, partition_of(newest_first[0]).await); + let mut locker = pool.begin().await.expect("begin locker"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "LOCK TABLE {old_partition} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(&mut *locker) + .await + .expect("lock old partition"); + sqlx::query("SET lock_timeout = '200ms'") + .execute(&mut *conn) + .await + .expect("set lock_timeout"); + let window = get_channel_window_on( + &mut conn, + community, + channel.id, + 1, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("a full recent window never touches the old partition"); + assert_eq!(ids(&window), newest_first[..1]); + let failed = get_channel_window_on( + &mut conn, + community, + channel.id, + 2, + None, + None, + ReadDeadline::PerRead, + ) + .await; + assert!( + failed.is_err(), + "a failed older pass must fail the window, got {:?}", + failed.map(|w| (w.rows.len(), w.has_more)) + ); + locker.rollback().await.expect("release lock"); + sqlx::query("RESET lock_timeout") + .execute(&mut *conn) + .await + .expect("reset lock_timeout"); + } + + /// The window's passes share one snapshot: a replacement that commits + /// between them neither drops the row it replaces nor shows the row it + /// adds. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_window_passes_share_one_snapshot() { + let pool = setup_pool().await; + let author = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + &format!("window-snapshot-{}", Uuid::new_v4()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + let channel_id = channel.id; + let (window, expected) = crate::test_support::replace_between_passes( + &pool, + community, + channel_id, + move |mut conn| async move { + get_channel_window_on( + &mut conn, + community, + channel_id, + 10, + None, + None, + ReadDeadline::PerRead, + ) + .await + .expect("split window") + .rows + .iter() + .map(|row| row.stored_event.event.id) + .collect::>() + }, + ) + .await; + assert_eq!(window, expected); + } + + /// A split window's passes share one read deadline. Another session holds + /// the newest partition for a while, so the recent pass spends part of + /// the deadline waiting, and holds the oldest partition, which only the + /// older pass plans, for good. The window must be cancelled once the whole + /// deadline has passed, not a full deadline after its older pass starts. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_window_passes_share_one_deadline() { + window_cancelled_within_the_read_deadline(Some(std::time::Duration::from_secs(8))).await; + } + + /// A writer window's first pass already runs under the read deadline, not + /// the writer session's longer timeout: with the newest partition held + /// for good, the window is cancelled in its recent pass. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn writer_window_first_pass_runs_under_the_read_deadline() { + window_cancelled_within_the_read_deadline(None).await; + } + + /// Read a split window through the production writer entry, whose own + /// session timeout is longer than the read deadline, while other sessions + /// lock the newest partition (released after `hold`, or never) and the + /// oldest one, which only the older pass plans (never released). The + /// window must be cancelled within the read deadline of its start. + async fn window_cancelled_within_the_read_deadline(hold: Option) { + let pool = setup_pool().await; + let author = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + &format!("window-deadline-{}", Uuid::new_v4()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + let db = crate::Db::from_pool(pool.clone()); + let mut partitions = Vec::new(); + for at in [Utc::now().timestamp() as u64 - 10, 1_748_736_000] { + let event = EventBuilder::new(Kind::Custom(9), format!("at {at}")) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&author) + .expect("sign"); + db.insert_event(community, &event, Some(channel.id)) + .await + .expect("insert"); + partitions.push( + sqlx::query_scalar::<_, String>( + "SELECT tableoid::regclass::text FROM events WHERE id = $1", + ) + .bind(event.id.as_bytes().to_vec()) + .fetch_one(&pool) + .await + .expect("partition"), + ); + } + let [newest, oldest] = <[String; 2]>::try_from(partitions).expect("two partitions"); + assert_ne!(newest, oldest); + let lock = |partition: String| { + let pool = pool.clone(); + async move { + let mut tx = pool.begin().await.expect("begin locker"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "LOCK TABLE {partition} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(&mut *tx) + .await + .expect("lock"); + tx + } + }; + let mut newest_lock = lock(newest.clone()).await; + let oldest_lock = lock(oldest).await; + + let started = std::time::Instant::now(); + let window_pool = pool.clone(); + let mut task = tokio::spawn(async move { + get_channel_window(&window_pool, community, channel.id, 10, None, None).await + }); + let waiting = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_locks \ + WHERE relation = $1::text::regclass AND NOT granted)", + ) + .bind(&newest) + .fetch_one(&mut *newest_lock) + .await + .expect("inspect the recent pass"); + if blocked { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!(waiting.is_ok(), "the recent pass never waited on the lock"); + // Spend part of the deadline in the recent pass, then let it finish. + if let Some(hold) = hold { + tokio::time::sleep(hold).await; + newest_lock + .rollback() + .await + .expect("release the newest partition"); + } + + let margin = std::time::Duration::from_secs(4); + let result = + tokio::time::timeout(crate::event::READ_DEADLINE + margin * 4, &mut task).await; + let elapsed = started.elapsed(); + drop(oldest_lock); + let Ok(result) = result else { + task.abort(); + panic!("the window was never cancelled"); + }; + let err = result + .expect("window task") + .expect_err("the window must be cancelled"); + assert!( + err.is_statement_cancelled(), + "cancelled by the deadline: {err}" + ); + assert!( + elapsed < crate::event::READ_DEADLINE + margin, + "the window got more than one read deadline; took {elapsed:?}" + ); + } + /// Rows with replies carry a thread summary (counts + batched /// participants); rows without replies carry none. This is the join that /// lets the GUI render thread affordances without a per-root fan-out. diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index f124315680..e5caa9473b 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -619,3 +619,6 @@ impl PausingProxy { self.next_session.load(std::sync::atomic::Ordering::SeqCst) } } + +mod recent_pass_tests; +pub(crate) use recent_pass_tests::replace_between_passes; diff --git a/crates/buzz-db/src/test_support/recent_pass_tests.rs b/crates/buzz-db/src/test_support/recent_pass_tests.rs new file mode 100644 index 0000000000..1be67a8dd7 --- /dev/null +++ b/crates/buzz-db/src/test_support/recent_pass_tests.rs @@ -0,0 +1,134 @@ +//! The recent-pass snapshot fixture. It writes `events` partitions directly, +//! so it lives in a standalone test module. + +/// Commit a row replacement while a read split at the recent-pass floor (see +/// `recent_pass`) is between its passes. +/// +/// Seeds `channel` with two kind:9 rows above the floor, `old` below it, +/// `ancient` in the partition before every dated monthly, and `new`: dated +/// now, soft-deleted until the replacement. Another session locks the +/// ancient partition, which only the older pass plans, so `read` stops +/// between its passes. The replacement then commits in one transaction, as +/// every replacement does: it deletes `old` and makes `new` live. The lock is +/// released and `read` finishes. +/// +/// Returns `read`'s result and the page its first pass's snapshot holds: +/// `[recent, recent, old, ancient]`. +pub(crate) async fn replace_between_passes( + pool: &sqlx::PgPool, + community: buzz_core::CommunityId, + channel: uuid::Uuid, + read: F, +) -> (T, Vec) +where + F: FnOnce(sqlx::pool::PoolConnection) -> Fut, + Fut: std::future::Future + Send + 'static, + T: Send + 'static, +{ + use nostr::{EventBuilder, Keys, Kind, Timestamp}; + + let db = crate::Db::from_pool(pool.clone()); + let author = Keys::generate(); + let now = chrono::Utc::now(); + let floor = crate::store::recent_pass::floor_at(now).timestamp() as u64; + let now = now.timestamp() as u64; + let insert = |at: u64, label: &str| { + let event = EventBuilder::new(Kind::Custom(9), format!("{label} at {at}")) + .custom_created_at(Timestamp::from(at)) + .sign_with_keys(&author) + .expect("sign"); + let db = db.clone(); + async move { + db.insert_event(community, &event, Some(channel)) + .await + .expect("insert"); + event.id + } + }; + let snapshot_page = vec![ + insert(now - 10, "recent").await, + insert(now - 20, "recent").await, + insert(floor - 86_400, "old").await, + insert(1_748_736_000, "ancient").await, // 2025-06-01 + ]; + let new = insert(now - 5, "new").await; + let partition_of = |id: nostr::EventId| async move { + sqlx::query_scalar::<_, String>("SELECT tableoid::regclass::text FROM events WHERE id = $1") + .bind(id.as_bytes().to_vec()) + .fetch_one(pool) + .await + .expect("partition") + }; + let old = snapshot_page[2]; + let (old_partition, ancient_partition, new_partition) = ( + partition_of(old).await, + partition_of(snapshot_page[3]).await, + partition_of(new).await, + ); + assert_ne!( + old_partition, ancient_partition, + "the replacement must not wait on the lock" + ); + assert_ne!( + new_partition, ancient_partition, + "the replacement must not wait on the lock" + ); + sqlx::query("UPDATE events SET deleted_at = now() WHERE id = $1") + .bind(new.as_bytes().to_vec()) + .execute(pool) + .await + .expect("hide the replacement"); + + let mut locker = pool.begin().await.expect("begin locker"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "LOCK TABLE {ancient_partition} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(&mut *locker) + .await + .expect("lock the ancient partition"); + let conn = pool.acquire().await.expect("acquire reader"); + let mut task = tokio::spawn(read(conn)); + let waiting = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE relation = $1::text::regclass AND NOT granted)", + ) + .bind(&ancient_partition) + .fetch_one(&mut *locker) + .await + .expect("inspect the older pass"); + if blocked { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + if waiting.is_err() { + task.abort(); + panic!("the read never reached its older pass"); + } + + let mut replace = pool.begin().await.expect("begin replacement"); + for (partition, deleted_at, id) in [ + (&old_partition, "now()", old), + (&new_partition, "NULL", new), + ] { + sqlx::query(sqlx::AssertSqlSafe(format!( + "UPDATE {partition} SET deleted_at = {deleted_at} WHERE id = $1" + ))) + .bind(id.as_bytes().to_vec()) + .execute(&mut *replace) + .await + .expect("replace"); + } + replace.commit().await.expect("commit replacement"); + locker.rollback().await.expect("release the lock"); + + let result = tokio::time::timeout(std::time::Duration::from_secs(10), &mut task).await; + if result.is_err() { + task.abort(); + panic!("the read did not finish after the lock was released"); + } + (result.unwrap().expect("read task"), snapshot_page) +}