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) +}