perf(db): stop the events recent pass at a ceiling for regular-kind reads (#9157)

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<Result<Vec<StoredEvent
     matches_nothing.then(|| Ok(vec![]))
 }
 
-/// The floor to split `q` at ([`crate::store::recent_pass`]), or `None` to
-/// read it whole. Lookups by id, `e` tag or `d` tag, and fetch-everything
+/// Where to split `q` ([`crate::store::recent_pass`]), or `None` to read it
+/// whole. Lookups by id, `e` tag or `d` tag, and fetch-everything
 /// reads (`max_limit`, the COUNT fallback) match a fixed set of rows wherever
 /// they sit, so a recent pass rarely fills their page. An `OFFSET` skips rows
 /// on both sides of the floor, so its split point is unknown.
-fn recent_pass_floor(q: &EventQuery, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
+fn recent_pass_split(
+    q: &EventQuery,
+    now: DateTime<Utc>,
+) -> Option<crate::store::recent_pass::Split> {
     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<Utc>) -> Option<DateTime<Utc>
     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<String> {
+            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<String> = 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<StoredEvent>| rows.into_iter().map(|r| r.event.id).collect::<Vec<_>>();
-        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.

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<Utc>,
+    /// The recent pass reads below it, when set. No row the read can match
+    /// sits at or above it.
+    pub(crate) ceiling: Option<DateTime<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<Utc>),
+    /// `created_at >= floor`, and `created_at < ceiling` when set.
+    Recent(Split),
     /// `created_at < floor`.
     Older(DateTime<Utc>),
 }
@@ -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<String> {
-        let (op, floor) = match self {
-            Range::All => return None,
-            Range::Recent(floor) => (">=", floor),
-            Range::Older(floor) => ("<", floor),
+        let bound = |op: &str, at: DateTime<Utc>| {
+            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<Utc>) -> DateTime<Utc> {
-    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::<Utc>::MIN_UTC)
+/// 00:00 UTC on the first day of the month `offset` months from `at`'s.
+fn month_start(at: DateTime<Utc>, offset: i32) -> DateTime<Utc> {
+    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::<Utc>::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<Utc>) -> DateTime<Utc> {
+    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<Utc>) -> DateTime<Utc> {
+    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<Utc>,
     kinds: Option<&[i32]>,
     since: Option<DateTime<Utc>>,
     until: Option<DateTime<Utc>>,
-) -> Option<DateTime<Utc>> {
+) -> Option<Split> {
     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<bool> = None;
+        const CEILED: Option<bool> = Some(true);
+        const OPEN: Option<bool> = 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<DateTime<Utc>>,
             Option<DateTime<Utc>>,
-            bool,
+            Option<bool>,
         );
         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}");
             }

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<Vec<i32>> =
         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)

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)