From 8ae35c5c92694af7488bbf6a2140c27aed9d0798 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 5 Nov 2026 20:22:57 -0500 Subject: [PATCH] perf(db): stop the events recent pass at a ceiling for regular-kind reads (#9157) --- crates/buzz-db/src/store/event.rs | 181 ++++++++++-- crates/buzz-db/src/store/recent_pass.rs | 265 +++++++++++++----- crates/buzz-db/src/store/thread.rs | 13 +- .../src/test_support/recent_pass_tests.rs | 6 +- 4 files changed, 363 insertions(+), 102 deletions(-) diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 18b3bcea22..d3c09338b9 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -588,7 +588,7 @@ pub(crate) async fn query_events_on( // 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())) { + let rows = match (root, recent_pass_split(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? @@ -597,10 +597,10 @@ pub(crate) async fn query_events_on( 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)) => { + (None, Some(split)) => { let mut read = ReadTransaction::begin(conn, deadline).await?; let mut rows = Vec::new(); - let mut pass = Some((Range::Recent(floor), limit)); + let mut pass = Some((Range::Recent(split), 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?); @@ -648,12 +648,15 @@ pub(crate) fn answer_without_db(q: &EventQuery) -> Option) -> Option> { +fn recent_pass_split( + q: &EventQuery, + now: DateTime, +) -> Option { let lookup = q.offset.is_some_and(|o| o > 0) || q.ids.is_some() || q.e_tags.is_some() @@ -664,7 +667,7 @@ fn recent_pass_floor(q: &EventQuery, now: DateTime) -> Option if lookup { return None; } - crate::store::recent_pass::floor_for(now, q.kinds.as_deref(), q.since, q.until) + crate::store::recent_pass::split_for(now, q.kinds.as_deref(), q.since, q.until) } /// The `LIMIT` [`query_events_on`] applies: the requested page, clamped. @@ -3037,13 +3040,19 @@ mod e_tag_filter_shape_tests { 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. + /// Each pass bounds `created_at` with literals so Postgres prunes + /// partitions while planning: the recent pass on both sides, the older + /// pass below the floor. 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 at = |m| chrono::TimeZone::with_ymd_and_hms(&Utc, 2026, m, 1, 0, 0, 0).unwrap(); + let split = crate::store::recent_pass::Split { + floor: at(10), + ceiling: Some(at(12)), + }; + let floor = "'2026-10-01T00:00:00Z'::timestamptz"; + let ceiling = "'2026-12-01T00:00:00Z'::timestamptz"; let sql = |q: &EventQuery, range| { build_query_events_sql(q, range, 7) .sql() @@ -3052,25 +3061,34 @@ mod e_tag_filter_shape_tests { }; let mut q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); q.kinds = Some(vec![9]); - let recent = sql(&q, Range::Recent(floor)); + let recent = sql(&q, Range::Recent(split)); + for bound in [ + format!(" AND created_at >= {floor}"), + format!(" AND created_at < {ceiling}"), + ] { + assert!(recent.contains(&bound), "{recent}"); + } + let older = sql(&q, Range::Older(split.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.contains(&format!(" AND created_at < {floor}")), "{older}" ); + assert!(!older.contains(ceiling), "{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 bounds = [ + (Range::Recent(split), vec![(">=", floor), ("<", ceiling)]), + (Range::Older(split.floor), vec![("<", floor)]), + ]; + for (range, ops) in bounds { 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}"); + for &(op, lit) in &ops { + let bound = format!(" AND {column} {op} {lit}"); + assert!(mention.contains(&bound), "{mention}"); + } } } let count = count_events_sql(&q).sql().as_str().to_owned(); @@ -3087,10 +3105,10 @@ mod e_tag_filter_shape_tests { q.kinds = Some(vec![9]); q }; - assert!(recent_pass_floor(&base(), now).is_some(), "page read"); + assert!(recent_pass_split(&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"); + assert!(recent_pass_split(&first_page, now).is_some(), "offset 0"); type Narrow = fn(&mut EventQuery); let lookups: [(&str, Narrow); 7] = [ ("offset", |q| q.offset = Some(1)), @@ -3106,7 +3124,7 @@ mod e_tag_filter_shape_tests { for (label, set) in lookups { let mut q = base(); set(&mut q); - assert!(recent_pass_floor(&q, now).is_none(), "{label}"); + assert!(recent_pass_split(&q, now).is_none(), "{label}"); } } @@ -3630,6 +3648,106 @@ mod postgres_tests { assert_eq!(ids(page), expected[..7]); } + /// The recent pass of a regular-kind read locks only the partitions + /// between its floor and ceiling, even as a cached generic plan. Without + /// the ceiling (a read that may match latest-state kinds) it also locks + /// the monthlies made ahead of time and the open-ended last partition. + /// Mutation: drop the ceiling from `Range::predicate`, or set it for + /// every split read → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn recent_pass_locks_only_the_partitions_below_its_ceiling() { + use crate::store::recent_pass::{ceiling_at, split_for}; + let pool = setup_pool().await; + let db = crate::Db::from_pool(pool.clone()); + // The layout a running relay keeps: dated monthlies through the + // months ahead, then an open-ended catch-all. + db.maintain_partitions( + crate::partition::PARTITION_MANAGER_MONTHS_AHEAD, + crate::partition::PartitionMaintenancePolicy { + create_enabled: true, + advance_enabled: true, + }, + ) + .await + .expect("maintain partitions"); + 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 now = Utc::now(); + let ceiled = split_for(now, Some(&[9]), None, None).expect("split"); + let open = split_for(now, Some(&[9, 39002]), None, None).expect("split"); + assert_eq!(ceiled.ceiling, Some(ceiling_at(now))); + assert_eq!(open.ceiling, None); + + // One row at each edge of the ceiled range names its partitions. + let author = Keys::generate(); + let mut expected = std::collections::BTreeSet::new(); + let last = ceiling_at(now) - chrono::Duration::seconds(1); + for at in [ceiled.floor, now, last] { + let event = EventBuilder::new(Kind::Custom(9), format!("at {at}")) + .custom_created_at(nostr::Timestamp::from(at.timestamp() as u64)) + .sign_with_keys(&author) + .expect("sign"); + db.insert_event(community, &event, Some(channel)) + .await + .expect("insert"); + let partition: String = + sqlx::query_scalar("SELECT tableoid::regclass::text FROM events WHERE id = $1") + .bind(event.id.as_bytes().to_vec()) + .fetch_one(&pool) + .await + .expect("partition"); + expected.insert(partition); + } + + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::query("SET plan_cache_mode = force_generic_plan") + .execute(&mut *conn) + .await + .expect("force generic plans"); + let q = EventQuery { + kinds: Some(vec![9]), + channel_id: Some(channel), + ..EventQuery::for_community(community) + }; + async fn locked( + conn: &mut sqlx::PgConnection, + q: &EventQuery, + split: crate::store::recent_pass::Split, + ) -> std::collections::BTreeSet { + let mut tx = sqlx::Connection::begin(conn).await.expect("begin"); + let mut qb = build_query_events_sql(q, Range::Recent(split), 10); + let rows = qb + .build() + .persistent(true) + .fetch_all(&mut *tx) + .await + .expect("recent pass"); + assert_eq!(rows.len(), 3, "the recent pass reads every row"); + let locked: Vec = sqlx::query_scalar( + "SELECT c.relname::text FROM pg_locks l \ + JOIN pg_class c ON c.oid = l.relation \ + JOIN pg_inherits i ON i.inhrelid = c.oid \ + WHERE l.pid = pg_backend_pid() \ + AND i.inhparent = 'events'::regclass", + ) + .fetch_all(&mut *tx) + .await + .expect("locked partitions"); + tx.rollback().await.expect("rollback"); + locked.into_iter().collect() + } + let ceiled_locks = locked(&mut conn, &q, ceiled).await; + assert_eq!(ceiled_locks, expected, "partitions below the ceiling"); + let open_locks = locked(&mut conn, &q, open).await; + assert!( + open_locks.is_superset(&ceiled_locks) && open_locks.len() > ceiled_locks.len(), + "without the ceiling the pass also locks the later partitions: {open_locks:?}" + ); + } + /// 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. @@ -3676,8 +3794,17 @@ mod postgres_tests { ..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%"; + // The recent pass of this kind:9 read also stops at the ceiling, so + // the older pass is told apart by its `<` bound at the floor. + let floor_lit = + crate::store::recent_pass::floor_at(Utc::now()).format("%Y-%m-%dT%H:%M:%SZ"); + let ceiling_lit = + crate::store::recent_pass::ceiling_at(Utc::now()).format("%Y-%m-%dT%H:%M:%SZ"); + let recent = &format!( + "%FROM events WHERE%created_at >= '{floor_lit}'::timestamptz \ + AND created_at < '{ceiling_lit}'::timestamptz%" + ); + let older = &format!("%FROM events WHERE%created_at < '{floor_lit}'::timestamptz%"); let all = "%FROM events WHERE%"; // Full recent page: answered by the recent pass alone. diff --git a/crates/buzz-db/src/store/recent_pass.rs b/crates/buzz-db/src/store/recent_pass.rs index d8a14646e1..96d7c72035 100644 --- a/crates/buzz-db/src/store/recent_pass.rs +++ b/crates/buzz-db/src/store/recent_pass.rs @@ -8,32 +8,59 @@ //! 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 newest partitions, so it is split ([`split_for`], [`next`]) at a floor, //! 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. +//! 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. A read of regular kinds only also stops at a +//! ceiling ([`ceiling_at`]), so it does not plan the empty monthly +//! partitions made ahead of time or the open-ended last one. //! 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. +//! The bounds are SQL literals, so Postgres prunes when it plans the +//! statement, cached generic plans included. They change about 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}; +use chrono::{DateTime, Datelike, Duration, TimeZone, Utc}; + +/// How far ahead of a relay's clock a regular-kind row's `created_at` can be, +/// with margin. +/// +/// Ingest rejects events more than 900 s ahead of the accepting relay's clock +/// (`MAX_TIMESTAMP_DRIFT_SECS` in `buzz-relay`'s ingest handler), and the +/// regular-kind events the relay signs itself are stamped with its clock. The +/// margin over 900 s covers the clock difference between the relay that wrote +/// a row and the one reading it. It costs one extra planned partition in the +/// last hour of a month. +/// +/// Replaceable and addressable rows have no such bound: the relay stamps its +/// own replacements one second past the live row when that is later than +/// `now`, so a coordinate replaced faster than once a second runs ahead. +pub(crate) const MAX_FUTURE: Duration = Duration::hours(1); + +/// Where a split page read divides: see the module docs. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct Split { + /// The recent pass reads at or above it; the older pass below it. + pub(crate) floor: DateTime, + /// The recent pass reads below it, when set. No row the read can match + /// sits at or above it. + pub(crate) ceiling: Option>, +} /// 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`, and `created_at < ceiling` when set. + Recent(Split), /// `created_at < floor`. Older(DateTime), } @@ -42,73 +69,98 @@ 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 + /// The bounds are literals, not binds: 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), + let bound = |op: &str, at: DateTime| { + format!( + " AND {column} {op} '{}'::timestamptz", + at.format("%Y-%m-%dT%H:%M:%SZ") + ) }; - Some(format!( - " AND {column} {op} '{}'::timestamptz", - floor.format("%Y-%m-%dT%H:%M:%SZ") - )) + match self { + Range::All => None, + Range::Recent(split) => { + let ceiling = split.ceiling.map(|c| bound("<", c)); + Some(bound(">=", split.floor) + ceiling.as_deref().unwrap_or("")) + } + Range::Older(floor) => Some(bound("<", floor)), + } } } -/// 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) +/// 00:00 UTC on the first day of the month `offset` months from `at`'s. +fn month_start(at: DateTime, offset: i32) -> DateTime { + let months = at.year() * 12 + at.month0() as i32 + offset; + Utc.with_ymd_and_hms( + months.div_euclid(12), + months.rem_euclid(12) as u32 + 1, + 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 +/// The floor at `now`: 00:00 UTC on the first day of the previous month, so +/// the recent pass reads the current and previous monthly partitions. +pub(crate) fn floor_at(now: DateTime) -> DateTime { + month_start(now, -1) +} + +/// The ceiling at `now`: 00:00 UTC on the first day of the month after +/// `now` + [`MAX_FUTURE`], so the recent pass of a regular-kind read plans +/// only the previous and current monthly partitions (and the next one in the +/// last hour of a month). +pub(crate) fn ceiling_at(now: DateTime) -> DateTime { + month_start(now + MAX_FUTURE, 1) +} + +/// Where to split a page read, 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. +/// the floor, so a split would only add a second statement. The recent pass +/// has a ceiling only when every kind is regular (see [`MAX_FUTURE`]). /// /// `kinds` are the read's kinds (`None`: any); `since`/`until` its bounds. -pub(crate) fn floor_for( +pub(crate) fn split_for( now: DateTime, kinds: Option<&[i32]>, since: Option>, until: Option>, -) -> 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 latest_state = |k: i32| { + u32::try_from(k).is_ok_and(|k| is_replaceable(k) || is_parameterized_replaceable(k)) + }; + let kinds = kinds.filter(|ks| !ks.is_empty()); + let latest_state_only = kinds.is_some_and(|ks| ks.iter().all(|&k| latest_state(k))); + let regular_only = kinds.is_some_and(|ks| !ks.iter().any(|&k| latest_state(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) + (!latest_state_only && !since_prunes && !until_before_floor).then(|| Split { + floor, + ceiling: regular_only.then(|| ceiling_at(now)), + }) } /// 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 { + let Range::Recent(split) = range else { return None; }; let found = i64::try_from(found).unwrap_or(i64::MAX); - (found < limit).then(|| (Range::Older(floor), limit - found)) + (found < limit).then(|| (Range::Older(split.floor), limit - found)) } #[cfg(test)] @@ -131,12 +183,57 @@ mod tests { assert_eq!(floor_at(at(2026, 3, 31)), midnight(2026, 2, 1)); } + /// The ceiling is the start of the next month, or of the one after once + /// `now` + [`MAX_FUTURE`] reaches the next month: a row the relay can + /// still accept is always below it. + #[test] + fn ceiling_is_the_start_of_the_next_month_after_max_future() { + let ceiling = ceiling_at; + assert_eq!(ceiling(at(2026, 11, 4)), midnight(2026, 12, 1)); + assert_eq!(ceiling(midnight(2026, 11, 1)), midnight(2026, 12, 1)); + assert_eq!(ceiling(at(2026, 12, 31)), midnight(2027, 1, 1)); + let month_end = midnight(2026, 12, 1); + let second = Duration::seconds(1); + assert_eq!( + ceiling(month_end - MAX_FUTURE - second), + month_end, + "before the margin" + ); + assert_eq!( + ceiling(month_end - MAX_FUTURE), + midnight(2027, 1, 1), + "inside the margin" + ); + assert_eq!( + ceiling(midnight(2026, 12, 31) + Duration::hours(23)), + midnight(2027, 2, 1) + ); + for now in [at(2026, 11, 4), month_end - second, midnight(2026, 12, 31)] { + assert!(ceiling(now) > now + MAX_FUTURE, "{now}"); + } + } + #[test] fn predicates_are_fixed_utc_literals() { let floor = midnight(2026, 10, 1); + let split = Split { + floor, + ceiling: Some(midnight(2026, 12, 1)), + }; assert_eq!(Range::All.predicate("e.created_at"), None); assert_eq!( - Range::Recent(floor).predicate("e.created_at").as_deref(), + Range::Recent(split).predicate("e.created_at").as_deref(), + Some( + " AND e.created_at >= '2026-10-01T00:00:00Z'::timestamptz \ + AND e.created_at < '2026-12-01T00:00:00Z'::timestamptz" + ) + ); + let open = Split { + ceiling: None, + ..split + }; + assert_eq!( + Range::Recent(open).predicate("e.created_at").as_deref(), Some(" AND e.created_at >= '2026-10-01T00:00:00Z'::timestamptz") ); assert_eq!( @@ -145,72 +242,87 @@ mod tests { ); } - /// Which reads are split: (kinds, since, until) → split? + /// Which reads are split, and which recent passes stop at the ceiling: + /// (kinds, since, until) → `WHOLE` (not split), `CEILED` or `OPEN`. #[test] fn split_eligibility() { + const WHOLE: Option = None; + const CEILED: Option = Some(true); + const OPEN: Option = Some(false); let now = at(2026, 11, 4); let floor = floor_at(now); - let day = chrono::Duration::days(1); - // (label, kinds, since, until, split) + let day = Duration::days(1); + // (label, kinds, since, until, outcome) type Case<'a> = ( &'a str, Option<&'a [i32]>, Option>, Option>, - bool, + Option, ); let cases: &[Case] = &[ - ("unbounded messages", Some(&[9, 40002]), None, None, true), - ("any kind", None, None, None, true), + ("unbounded messages", Some(&[9, 40002]), None, None, CEILED), + ("any kind", None, None, None, OPEN), + ("no kinds listed", Some(&[]), None, None, OPEN), ( "since older than the floor", Some(&[9]), Some(floor - day), None, - true, + CEILED, ), ( "since at the floor already prunes", Some(&[9]), Some(floor), None, - false, + WHOLE, ), ( "recent since already prunes", Some(&[9]), Some(now - day), None, - false, + WHOLE, ), - ("until at the floor", Some(&[9]), None, Some(floor), true), + ("until at the floor", Some(&[9]), None, Some(floor), CEILED), ( "until before the floor", Some(&[9]), None, Some(floor - day), - false, + WHOLE, ), - ("replaceable only", Some(&[0, 10002]), None, None, false), - ("addressable only", Some(&[39002]), None, None, false), + ("replaceable only", Some(&[0, 10002]), None, None, WHOLE), + ("addressable only", Some(&[39002]), None, None, WHOLE), ( - "mixed with a message kind", + "mixed with an addressable kind", Some(&[39002, 9]), None, None, - true, + OPEN, + ), + ( + "mixed with a replaceable kind", + Some(&[9, 13534]), + None, + None, + OPEN, ), ( "negative kind is not replaceable", Some(&[-1]), None, None, - true, + CEILED, ), ]; - for &(label, kinds, since, until, split) in cases { - let expected = split.then_some(floor); - assert_eq!(floor_for(now, kinds, since, until), expected, "{label}"); + for &(label, kinds, since, until, outcome) in cases { + let expected = outcome.map(|ceiled| Split { + floor, + ceiling: ceiled.then(|| ceiling_at(now)), + }); + assert_eq!(split_for(now, kinds, since, until), expected, "{label}"); } } @@ -219,19 +331,31 @@ mod tests { /// 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 day = Duration::days(1); let floor = midnight(2026, 10, 1); - // Three at or above the floor (one exactly at it), two below. + let ceiling = midnight(2026, 12, 1); + // Four at or above the floor (one exactly at it, one a second below + // the ceiling), two below. No row sits at or above the ceiling. let table = [ + ceiling - Duration::seconds(1), floor + day * 3, floor + day, floor, floor - day, floor - day * 9, ]; + let recent = 4; 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 ceiled = Split { + floor, + ceiling: Some(ceiling), + }; + let open = Split { + ceiling: None, + ..ceiled + }; + for split in [None, Some(ceiled), Some(open)] { let mut passes = Vec::new(); let mut rows = Vec::new(); let mut pass = Some((split.map_or(Range::All, Range::Recent), limit)); @@ -243,7 +367,7 @@ mod tests { .copied() .filter(|&t| match range { Range::All => true, - Range::Recent(f) => t >= f, + Range::Recent(s) => s.floor <= t && s.ceiling.is_none_or(|c| t < c), Range::Older(f) => t < f, }) .take(n as usize), @@ -253,8 +377,11 @@ mod tests { 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)], + Some(s) if limit <= recent => vec![(Range::Recent(s), limit)], + Some(s) => vec![ + (Range::Recent(s), limit), + (Range::Older(s.floor), limit - recent), + ], }; 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 f78786d758..5ae76ec594 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -661,7 +661,7 @@ pub(crate) async fn get_channel_window_on( // 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( + let split = crate::store::recent_pass::split_for( Utc::now(), kinds.as_deref(), None, @@ -685,7 +685,7 @@ pub(crate) async fn get_channel_window_on( } q.bind(pass_limit) }; - let mut db_rows = match floor { + let mut db_rows = match split { None => { crate::event::fetch_with_read_deadline( &mut *conn, @@ -694,12 +694,12 @@ pub(crate) async fn get_channel_window_on( ) .await? } - Some(floor) => { + Some(split) => { // 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)); + let mut pass = Some((Range::Recent(split), 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); @@ -2628,8 +2628,11 @@ mod postgres_tests { let waiting = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let blocked: bool = sqlx::query_scalar( + // `pg_locks` spans every database, and the per-test + // databases are clones that share relation OIDs. "SELECT EXISTS(SELECT 1 FROM pg_locks \ - WHERE relation = $1::text::regclass AND NOT granted)", + WHERE relation = $1::text::regclass AND NOT granted \ + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()))", ) .bind(&newest) .fetch_one(&mut *newest_lock) diff --git a/crates/buzz-db/src/test_support/recent_pass_tests.rs b/crates/buzz-db/src/test_support/recent_pass_tests.rs index 1be67a8dd7..db13662866 100644 --- a/crates/buzz-db/src/test_support/recent_pass_tests.rs +++ b/crates/buzz-db/src/test_support/recent_pass_tests.rs @@ -91,7 +91,11 @@ where 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)", + // `pg_locks` spans every database, and the per-test databases + // are clones that share relation OIDs: match this one only. + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE relation = $1::text::regclass \ + AND NOT granted \ + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()))", ) .bind(&ancient_partition) .fetch_one(&mut *locker)