From 0ada1d37077f701507855fe58a4e78ac4e345f30 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 5 Nov 2026 18:25:56 -0500 Subject: [PATCH] buzz-db: read explicit #h sets one channel at a time (#9110) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Explicit multi-channel `#h` reads whose channels × page limit fit a budget of 8,192 now read each channel through the channel index and keep the newest `n` rows overall. Before, Postgres was left to plan `channel_id = ANY($set) ORDER BY created_at DESC LIMIT n` from a row estimate. Stacked on #9091 (both change `build_query_events_sql`). It retargets to `main` when that merges. ## Problem The desktop app's cold unread repair reads every joined channel in chunks, with `kinds [9, 40002, 40008]`, `include_aux`, `limit 500` and no `since`. It never fills its page. For that shape, Postgres's custom plans (the first executions of each prepared statement on a connection) overestimate how many rows a quiet channel set matches. They walk the partition's primary key backward and test every community row against the set. The inflated cost also turns on JIT. On a load-test database with about 940k events in one partition (`EXPLAIN ANALYZE`, custom plan): | Channels in the set | Plan | Rows returned | Time | |---|---|---|---| | 3 | channel index | 4 | 0.5 ms | | 6 | channel index, plus JIT | 254 | 54 ms (1.2 ms scan) | | 15 | pkey walk, 940k rows filtered | 427 | 546 ms | | 48 | pkey walk, 299k rows filtered | 863 | 197 ms | | 128 | channel index | 500 | 5 ms | Generic plans pick the channel index, but every connection runs custom plans first. A client-side `since` would not avoid the walk: it still walks the whole window. It would also change unread semantics, and it does nothing for app versions already shipped. ## Change `build_query_events_sql` handles plain explicit `#h` reads like this: ```sql SELECT e.* FROM (SELECT DISTINCT unnest($set::uuid[]) AS ch) c CROSS JOIN LATERAL ( SELECT … FROM events WHERE community_id = $1 … AND channel_id = c.ch … ORDER BY created_at DESC, id ASC LIMIT $n ) e ORDER BY e.created_at DESC, e.id ASC LIMIT $n OFFSET 0 ``` - **Same page as the set read:** every row of the overall top `n` is also in its own channel's top `n`. - **Bounded candidates:** at most distinct channels × `n` rows reach the outer sort, whatever the statistics say. How many index entries each channel scans still depends on the other predicates, such as `kinds`. - **Same SQL for any number of channels:** the channel set stays one array bind. - **Repeated ids:** `DISTINCT` reads a repeated id once. - **Per pass:** #9091's recent and older passes each use this shape. These keep today's SQL: scopes that include global events (`channel_id IS NULL OR ANY`), `#p` mention joins, e-tag fences, offset pages, the COUNT, and sets whose channels × page limit exceed the budget. ### Candidate budget A per-channel read fetches up to `n` rows from every channel, where the set read stops once the page fills. So busy channel sets cost more. On a synthetic community (920k events; 128 busy channels with 1,000 recent messages and 1,000 reactions each, 400 more busy channels, 128 quiet channels with 3 messages each), the repair shape (`kinds [9, 40002, 40008]`, recent pass), JIT off: | Set × limit | Set read, custom plan | Set read, generic plan | Per-channel | |---|---|---|---| | 128 busy × 500 | 2.3 ms | 238 ms | 114 ms | | 20 busy + 108 quiet × 500 | 19 ms | 33 ms | 18 ms | | 128 quiet × 500 | 473 ms | 0.7 ms | 1.3 ms | | 15 busy × 500 | 22 ms | 23 ms | 13 ms | | 15 quiet × 500 | 0.2 ms | 0.2 ms | 0.3 ms | | 128 busy × 100 | 0.7 ms | | 17 ms | | 128 quiet × 100 | 141 ms | | 1.2 ms | With JIT on (the default), the per-channel custom plan for 128 channels × 500 crossed the JIT thresholds: about 0.9 s for the busy set and 0.75 s for the quiet one, nearly all compile time. At 15 × 500 it stayed under them. So the per-channel read runs only while channels × page limit ≤ 8,192 (`PER_CHANNEL_CANDIDATE_BUDGET`, the clamped page limit). That bounds it at about 8k candidate rows, where it was still faster than the set read here. Larger sets keep the set read, as on `main`. Either way there are two SQL texts, so plan caching holds. The desktop repair's full 128-channel chunks at `limit 500` (64,000) are over the budget and keep the set read. Only its last, partial chunk (16 channels or fewer) and other small explicit sets read per channel. The load-rig numbers below predate the budget and don't show its effect. On the same data, the per-channel form took 0.5 ms (6 channels), 0.9 ms (15) and 4.2 ms (128). Load rig, cold desktop reconnect scenario (every client runs the unread repair), #9091 alone vs this PR on top: | | Pool default: #9091 | Pool default: + this PR | Pool 200: #9091 | Pool 200: + this PR | |---|---|---|---|---| | Repair read mean | 302 ms | 41 ms | 1,990 ms | 93 ms | | Repair read total DB time | 1,547 s | 245 s | 9,622 s | 561 s | | Repair reads served | 5,121 | 5,915 | 4,834 | 6,011 | | Client op failures | 23,475 | 10,966 | 28,797 | 10,829 | | Unread-load errors | 1,018 | 113 | 1,434 | 3 | | Relay pool-timeout log lines | 8,762 | 824 | 13,860 | 409 | The rig stays overloaded in every arm. Most remaining failures are the per-connection concurrent-request limit and connect/auth under load, not this read. ## Tests - `per_channel_reads_match_the_set_read` (Postgres) covers: - every limit from 1 to total+1, including a page held entirely by one channel - newer rows of another kind or another channel taking no slot - a repeated channel id and an id tie across channels - keyset paging across the tie, and `since` - rows below the recent-pass floor - a prepared-statement check that the read ran per channel and no `ANY` read ran - `per_channel_scope_eligibility` covers which reads keep the current SQL, including the budget edge (16 × 512 in, 17 × 512 and 128 × 500 out) and the clamped page limit. - `channel_scope_sql_does_not_depend_on_channel_count` now also asserts the per-channel shape for explicit `#h` pages. ## What got simpler Not much. This adds one SQL shape to the events query builder, behind one eligibility function and one budget. In exchange, the planner's row estimate no longer decides whether a small multi-channel read scans a whole partition. The channel predicate is decided in one place: `push_event_filters` no longer pushes the channel scope, and each caller (the page read and the thread route) pushes its own. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-db/src/store/event.rs | 230 +++++++++++++++++++++++++++++- 1 file changed, 225 insertions(+), 5 deletions(-) diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 7913ff46d2..2ee6b30a52 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -689,6 +689,7 @@ fn build_query_events_sql( // every community row against the id set; the fence keeps it on the GIN // tags index. let e_tag_fence = q.e_tags.is_some(); + let per_channel = per_channel_scope(q); let mut qb: QueryBuilder = if let Some(ref p_hex) = q.p_tag_hex { // Join against event_mentions for #p-filtered queries (indexed). let mut b = QueryBuilder::new(if e_tag_fence { @@ -710,6 +711,11 @@ fn build_query_events_sql( } else { "" }); + if let Some(ids) = per_channel { + b.push("SELECT e.* FROM (SELECT DISTINCT unnest(") + .push_bind(ids.to_vec()) + .push("::uuid[]) AS ch) c CROSS JOIN LATERAL ("); + } b.push( "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ FROM events WHERE community_id = ", @@ -722,6 +728,12 @@ fn build_query_events_sql( // 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); + // A per-channel read scopes each lateral to one channel of the set. + if per_channel.is_some() { + qb.push(" AND channel_id = c.ch"); + } else { + push_channel_scope(&mut qb, col_prefix, q); + } if let Some(bound) = range.predicate(&format!("{col_prefix}created_at")) { qb.push(bound); } @@ -734,6 +746,10 @@ fn build_query_events_sql( // degrades, add a composite index like `(pubkey, kind, created_at DESC, id ASC)`. if e_tag_fence { qb.push(") SELECT * FROM m ORDER BY created_at DESC, id ASC LIMIT "); + } else if per_channel.is_some() { + qb.push(" ORDER BY created_at DESC, id ASC LIMIT ") + .push_bind(limit_val) + .push(") e ORDER BY e.created_at DESC, e.id ASC LIMIT "); } else { qb.push(format!( " ORDER BY {col_prefix}created_at DESC, {col_prefix}id ASC LIMIT " @@ -745,8 +761,8 @@ fn build_query_events_sql( } /// AND every [`EventQuery`] predicate except the community, the `deleted_at` -/// test and the `#p` join, on columns qualified by `col_prefix` (`""` or -/// `"e."`). +/// test, the channel scope (`push_channel_scope`) and the `#p` join, on +/// columns qualified by `col_prefix` (`""` or `"e."`). fn push_event_filters(qb: &mut QueryBuilder, col_prefix: &str, q: &EventQuery) { // Generic reads return only current artifact revisions, including delete // tombstones; explicit revision IDs also read earlier revisions. @@ -763,8 +779,6 @@ fn push_event_filters(qb: &mut QueryBuilder, col_prefix: &str, q qb.push(format!(" AND {col_prefix}channel_id IS NULL")); } - push_channel_scope(qb, col_prefix, q); - if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) { qb.push(format!(" AND {col_prefix}kind IN (")); let mut sep = qb.separated(", "); @@ -1000,12 +1014,53 @@ fn build_thread_reply_sql(q: &EventQuery, root: Vec) -> QueryBuilder Option<&[Uuid]> { + let ids = q.channel_ids.as_deref().filter(|ids| !ids.is_empty())?; + let eligible = !q.channel_ids_include_global + && q.channel_id.is_none() + && !q.global_only + && q.p_tag_hex.is_none() + && q.e_tags.is_none() + && q.offset.unwrap_or(0) == 0 + && usize::try_from(page_limit(q)) + .is_ok_and(|n| ids.len().saturating_mul(n) <= PER_CHANNEL_CANDIDATE_BUDGET); + eligible.then_some(ids) +} + /// `#p` reads join `event_mentions` by the full event key, including /// `created_at`: mention rows copy the event's `created_at` at insert, so the /// equality is exact and turns each joined row into one primary-key probe on @@ -3495,6 +3550,8 @@ mod e_tag_filter_shape_tests { /// change the SQL text, or each count is its own statement that is planned /// again on every call. Holds for the `#p` join and plain reads, for /// access scope and explicit `#h`, in both the page SELECT and the COUNT. + /// A plain explicit `#h` page within the candidate budget unnests the + /// bind and reads each channel; over it, the page is the set read. #[test] fn channel_scope_sql_does_not_depend_on_channel_count() { let sql = |p_tag: bool, include_global: bool, n: usize| { @@ -3515,7 +3572,19 @@ mod e_tag_filter_shape_tests { for include_global in [true, false] { let one = sql(p_tag, include_global, 1); assert_eq!(one, sql(p_tag, include_global, 40)); - for text in [&one.0, &one.1] { + let per_channel = !p_tag && !include_global; + assert_eq!( + one.0.contains("unnest($") && one.0.contains("channel_id = c.ch"), + per_channel, + "{}", + one.0 + ); + let set_reads = if per_channel { + vec![&one.1] + } else { + vec![&one.0, &one.1] + }; + for text in set_reads { assert!(text.contains("channel_id = ANY($"), "{text}"); assert!(!text.contains("channel_id IN ("), "{text}"); assert_eq!( @@ -3527,6 +3596,46 @@ mod e_tag_filter_shape_tests { } } } + + /// Only a plain explicit `#h` page within the candidate budget reads + /// channel by channel: global events, a `#p` join, an e-tag fence, an + /// offset, an empty set or a set × limit over the budget keep their own + /// SQL. + #[test] + fn per_channel_scope_eligibility() { + let base = || EventQuery { + channel_ids: Some(vec![uuid::Uuid::new_v4(), uuid::Uuid::new_v4()]), + channel_ids_include_global: false, + ..EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())) + }; + assert!(per_channel_scope(&base()).is_some()); + let mut offset_zero = base(); + offset_zero.offset = Some(0); + assert!(per_channel_scope(&offset_zero).is_some()); + let at_budget = |channels: usize, limit: i64| EventQuery { + channel_ids: Some((0..channels).map(|_| uuid::Uuid::new_v4()).collect()), + limit: Some(limit), + ..base() + }; + assert!(per_channel_scope(&at_budget(16, 512)).is_some()); + assert!(per_channel_scope(&at_budget(17, 512)).is_none()); + assert!(per_channel_scope(&at_budget(128, 500)).is_none()); + // The clamped page limit counts, not the requested one. + assert!(per_channel_scope(&at_budget(8, 5_000)).is_some()); + type Edit = fn(&mut EventQuery); + let others: [(&str, Edit); 5] = [ + ("global", |q| q.channel_ids_include_global = true), + ("#p", |q| q.p_tag_hex = Some("ab".repeat(32))), + ("#e", |q| q.e_tags = Some(vec!["cd".repeat(32)])), + ("offset", |q| q.offset = Some(1)), + ("empty", |q| q.channel_ids = Some(Vec::new())), + ]; + for (label, set) in others { + let mut q = base(); + set(&mut q); + assert!(per_channel_scope(&q).is_none(), "{label}"); + } + } } #[cfg(test)] @@ -3732,6 +3841,117 @@ mod postgres_tests { } } + /// An explicit channel set is read one channel at a time and returns the + /// set read's page: newest first across channels with id ties broken + /// ascending, one channel holding the whole page, other kinds and other + /// channels taking no slot, a repeated id read once, keyset cursors and + /// `since`, and older rows the recent pass leaves missing. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn per_channel_reads_match_the_set_read() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let mut channels = Vec::new(); + for _ in 0..4 { + channels.push(make_test_channel(&pool, community_uuid, None).await); + } + let (busy, quiet, empty, outside) = (channels[0], channels[1], channels[2], channels[3]); + 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; + let mut matching = Vec::new(); + let mut insert = |channel: Uuid, kind: u16, at: u64, matches: bool| { + let event = + EventBuilder::new(Kind::Custom(kind), format!("{kind} at {at} in {channel}")) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&author) + .expect("sign"); + if matches { + matching.push((at, event.id)); + } + (channel, event) + }; + let mut events = Vec::new(); + // The busy channel alone holds the newest five rows. + for i in 0..5 { + events.push(insert(busy, 9, now - 10 - i, true)); + } + // Newer rows of another kind or another channel take no slot. + events.push(insert(busy, 7, now - 1, false)); + events.push(insert(outside, 9, now - 2, false)); + // A tie across channels, then older rows on both sides of the floor. + events.push(insert(busy, 9, now - 100, true)); + events.push(insert(quiet, 9, now - 100, true)); + events.push(insert(quiet, 9, now - 200, true)); + events.push(insert(busy, 9, floor - 86_400, true)); + events.push(insert(quiet, 9, floor - 2 * 86_400, true)); + for (channel, event) in &events { + db.insert_event(community, event, Some(*channel)) + .await + .expect("insert"); + } + matching.sort_by(|a, b| { + b.0.cmp(&a.0) + .then_with(|| a.1.as_bytes().cmp(b.1.as_bytes())) + }); + let expected: Vec<_> = matching.iter().map(|(_, id)| *id).collect(); + + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::Connection::clear_cached_statements(&mut *conn) + .await + .expect("clear prepared statements"); + let read = |limit: i64| EventQuery { + kinds: Some(vec![9]), + channel_ids: Some(vec![busy, quiet, empty, busy]), + channel_ids_include_global: false, + limit: Some(limit), + ..EventQuery::for_community(community) + }; + let ids = |rows: Vec| rows.into_iter().map(|r| r.event.id).collect::>(); + + for limit in 1..=expected.len() as i64 + 1 { + let page = query_events_on(&mut conn, &read(limit), ReadDeadline::PerRead) + .await + .expect("page"); + let want = &expected[..expected.len().min(limit as usize)]; + assert_eq!(ids(page), want, "limit {limit}"); + } + assert!( + prepared_like(&mut conn, "%CROSS JOIN LATERAL%channel_id = c.ch%").await >= 1, + "the set was read per channel" + ); + assert_eq!( + prepared_like(&mut conn, "%channel_id = ANY(%").await, + 0, + "no set read ran" + ); + + // Keyset pages walk the same order, across the tie and the floor. + let first = query_events_on(&mut conn, &read(6), ReadDeadline::PerRead) + .await + .expect("first page"); + let last = first.last().expect("full first page").clone(); + let mut next = read(6); + next.until = Some( + DateTime::from_timestamp(last.event.created_at.as_secs() as i64, 0).expect("until"), + ); + next.before_id = Some(last.event.id.as_bytes().to_vec()); + let page = query_events_on(&mut conn, &next, ReadDeadline::PerRead) + .await + .expect("next page"); + assert_eq!(ids(first), expected[..6]); + assert_eq!(ids(page), expected[6..]); + + let mut since = read(10); + since.since = Some(DateTime::from_timestamp((now - 100) as i64, 0).expect("since")); + let page = query_events_on(&mut conn, &since, ReadDeadline::PerRead) + .await + .expect("since page"); + assert_eq!(ids(page), expected[..7]); + } + /// 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.