From 19d829756aecc587d5ec70cc8af5898ed697f7de Mon Sep 17 00:00:00 2001 From: Mathias Hartveit Date: Mon, 5 Oct 2026 18:45:57 +0200 Subject: [PATCH 1/2] perf(gateway): read a group event transaction once for all recipients wire_group_events copied the cached game_group_event table once per recipient with a session. Now one snapshot per source database holds the transaction's rows grouped by recipient, and each recipient slices its rows from it. The insert-order guarantee and the per-database cursor stay. Co-Authored-By: Claude Sonnet 5.5 --- gateway/src/stdb/world_view.rs | 153 +++++++++++++++++++++++++++++++-- 1 file changed, 144 insertions(+), 9 deletions(-) diff --git a/gateway/src/stdb/world_view.rs b/gateway/src/stdb/world_view.rs index d6c489a3e..9b63b044d 100644 --- a/gateway/src/stdb/world_view.rs +++ b/gateway/src/stdb/world_view.rs @@ -1478,10 +1478,10 @@ fn wire_insert_live( }); } -/// [`wire_insert_live`] for `game_group_event` from `source`. The dispatcher also gets a read of -/// the cached table, which already holds the whole transaction when the callback runs. It runs -/// the read only for a recipient with a session, once per transaction, because the SDK copies -/// every cached row to iterate. +/// [`wire_insert_live`] for `game_group_event` from `source`. The dispatcher also gets the rows of +/// the row's transaction for its recipient, sliced from one [`GroupEventSnapshot`] per source +/// database. The table read behind the snapshot happens once per transaction, and only when some +/// recipient with a session needs it, because the SDK copies every cached row to iterate. fn wire_group_events( db: &RemoteTables, label: &'static str, @@ -1490,13 +1490,14 @@ fn wire_group_events( source: GroupEventSource, ) { let view = view.clone(); + let snapshot = GroupEventSnapshot::default(); db.game_group_event().on_insert(move |ctx, row| { if is_initial_apply(&ctx.event) { return; } guarded(label, || { group_event_appeared(&view, &coord, source, row, || { - ctx.db.game_group_event().iter().collect() + snapshot.recipient_rows(row, || ctx.db.game_group_event().iter().collect()) }); }); }); @@ -2802,9 +2803,9 @@ impl GroupEventCursor { /// The SDK fires the inserts of one transaction in hash order, and the client must hear them /// in insert order: a filled Meeting Stone party's `COMPLETE` comes before `SETQUEUE(0, NONE)`. /// So the first callback of a transaction takes every row of that transaction for this - /// recipient from `cached`, which reads the table that already holds the whole transaction, and - /// the later callbacks find theirs taken without reading it. Rows of one transaction share - /// `created_at`. + /// recipient from `cached`, which slices the transaction's rows out of a + /// [`GroupEventSnapshot`], and the later callbacks find theirs taken without calling it. Rows + /// of one transaction share `created_at`. fn take( &self, source: GroupEventSource, @@ -2833,6 +2834,75 @@ impl GroupEventCursor { } } +/// The `game_group_event` rows of the latest transaction seen from one source database, grouped by +/// recipient. Every recipient of the transaction slices its rows from the same copy. +#[derive(Default)] +pub(crate) struct GroupEventSnapshot(Mutex>); + +struct TransactionRows { + created_at: spacetimedb_sdk::Timestamp, + by_recipient: HashMap>, +} + +impl TransactionRows { + /// The rows of `row`'s transaction in `table`. `row` joins them when the table lacks it. + fn collect(row: &GroupEvent, table: Vec) -> Self { + let mut by_recipient: HashMap> = HashMap::new(); + let mut found = false; + for cached in table { + if cached.created_at != row.created_at { + continue; + } + found |= cached.id == row.id; + by_recipient + .entry(cached.recipient_guid) + .or_default() + .push(cached); + } + if !found { + by_recipient + .entry(row.recipient_guid) + .or_default() + .push(row.clone()); + } + Self { + created_at: row.created_at, + by_recipient, + } + } + + fn holds(&self, row: &GroupEvent) -> bool { + self.created_at == row.created_at + && self + .by_recipient + .get(&row.recipient_guid) + .is_some_and(|rows| rows.iter().any(|held| held.id == row.id)) + } +} + +impl GroupEventSnapshot { + /// The rows of `row`'s transaction addressed to `row`'s recipient, in no particular order. + /// + /// The first call for a transaction runs `read_table`, which must return the cached table + /// holding the whole transaction. Every later call for the same transaction reuses that + /// copy. A row the held transaction lacks starts a new one, so two transactions with the + /// same `created_at` never share a copy. + fn recipient_rows( + &self, + row: &GroupEvent, + read_table: impl FnOnce() -> Vec, + ) -> Vec { + let mut held = self.0.lock().unwrap_or_else(|p| p.into_inner()); + if !held.as_ref().is_some_and(|rows| rows.holds(row)) { + *held = Some(TransactionRows::collect(row, read_table())); + } + held.as_ref() + .and_then(|rows| rows.by_recipient.get(&row.recipient_guid)) + .cloned() + .unwrap_or_default() + } +} + /// A shard mirror delta is the point at which viewer-relative Loot Tag flags can be projected from /// current membership without racing the realm event that requested the mirror write. fn group_member_mirror_changed( @@ -6022,7 +6092,7 @@ mod account_claim_relay_tests { #[cfg(test)] mod group_event_order_tests { - use super::{GroupEvent, GroupEventCursor, GroupEventSource}; + use super::{GroupEvent, GroupEventCursor, GroupEventSnapshot, GroupEventSource}; const RECIPIENT: u64 = 7; @@ -6096,6 +6166,71 @@ mod group_event_order_tests { [3] ); } + + /// A Raid roster list writes rows for every member in one transaction. The SDK fires an + /// insert callback per row; the table is read once, and every member still gets its own rows. + #[test] + fn one_transaction_reads_the_table_once_for_every_recipient() { + let members = 40_u64; + let transaction: Vec = (0..members) + .flat_map(|member| { + let guid = 100 + member; + [ + event(1000 + member * 2, guid, 200), + event(1001 + member * 2, guid, 200), + ] + }) + .collect(); + let mut table = vec![event(5, 100, 100)]; + table.extend(transaction.iter().cloned()); + + let snapshot = GroupEventSnapshot::default(); + let reads = std::cell::Cell::new(0); + let mut relayed: Vec<(u64, Vec)> = Vec::new(); + for row in transaction.iter().rev() { + let cursor = GroupEventCursor::default(); + let rows = cursor.take(GroupEventSource::RealmCore, row, || { + snapshot.recipient_rows(row, || { + reads.set(reads.get() + 1); + table.clone() + }) + }); + relayed.push((row.recipient_guid, ids(&rows))); + } + + assert_eq!(reads.get(), 1); + for member in 0..members { + let first = 1000 + member * 2; + assert!(relayed.contains(&(100 + member, vec![first, first + 1]))); + } + } + + /// A row the held transaction lacks starts the next one, even with the same timestamp. + #[test] + fn the_next_transaction_reads_the_table_again() { + let table = [ + event(10, RECIPIENT, 200), + event(11, RECIPIENT, 200), + event(12, RECIPIENT, 200), + ]; + let snapshot = GroupEventSnapshot::default(); + let reads = std::cell::Cell::new(0); + let recipient_rows = |row: &GroupEvent, visible: usize| { + snapshot.recipient_rows(row, || { + reads.set(reads.get() + 1); + table[..visible].to_vec() + }) + }; + + let first = recipient_rows(&table[0], 2); + let again = recipient_rows(&table[1], 2); + let next = recipient_rows(&table[2], 3); + + assert_eq!(reads.get(), 2); + assert_eq!(ids(&first), [10, 11]); + assert_eq!(ids(&again), [10, 11]); + assert_eq!(ids(&next), [10, 11, 12]); + } } // =============================================================================================== From 2bb13e09d687139c782889e91ffc7021fbd5c0bb Mon Sep 17 00:00:00 2001 From: Mathias Hartveit Date: Mon, 5 Oct 2026 19:11:38 +0200 Subject: [PATCH 2/2] refactor(gateway): keep transaction selection in GroupEventCursor only The snapshot now holds just the table read grouped by recipient guid, owned by the callback closure with no lock. GroupEventSnapshot::take composes it with the cursor, and both wire_group_events and the test call that one path. Co-Authored-By: Claude Sonnet 5.5 --- gateway/src/stdb/world_view.rs | 165 +++++++++++++-------------------- 1 file changed, 67 insertions(+), 98 deletions(-) diff --git a/gateway/src/stdb/world_view.rs b/gateway/src/stdb/world_view.rs index 9b63b044d..924ff7d6b 100644 --- a/gateway/src/stdb/world_view.rs +++ b/gateway/src/stdb/world_view.rs @@ -1478,9 +1478,9 @@ fn wire_insert_live( }); } -/// [`wire_insert_live`] for `game_group_event` from `source`. The dispatcher also gets the rows of -/// the row's transaction for its recipient, sliced from one [`GroupEventSnapshot`] per source -/// database. The table read behind the snapshot happens once per transaction, and only when some +/// [`wire_insert_live`] for `game_group_event` from `source`. The dispatcher also gets a read of +/// the cached table, which already holds the whole transaction when the callback runs. One +/// [`GroupEventSnapshot`] per source database runs the read once per transaction, and only when a /// recipient with a session needs it, because the SDK copies every cached row to iterate. fn wire_group_events( db: &RemoteTables, @@ -1490,14 +1490,14 @@ fn wire_group_events( source: GroupEventSource, ) { let view = view.clone(); - let snapshot = GroupEventSnapshot::default(); + let mut snapshot = GroupEventSnapshot::default(); db.game_group_event().on_insert(move |ctx, row| { if is_initial_apply(&ctx.event) { return; } guarded(label, || { - group_event_appeared(&view, &coord, source, row, || { - snapshot.recipient_rows(row, || ctx.db.game_group_event().iter().collect()) + group_event_appeared(&view, &coord, source, row, &mut snapshot, || { + ctx.db.game_group_event().iter().collect() }); }); }); @@ -2754,7 +2754,8 @@ fn group_event_appeared( coord: &Coordinator, source: GroupEventSource, row: &GroupEvent, - cached: impl FnOnce() -> Vec, + snapshot: &mut GroupEventSnapshot, + read_table: impl FnOnce() -> Vec, ) { let Some(session) = view.session_of_owner(row.recipient_guid) else { return; @@ -2765,7 +2766,7 @@ fn group_event_appeared( if !super::subscriptions::private_recipient_audience(row.recipient_guid, viewer.self_guid) { return; } - let rows = viewer.group_events.take(source, row, cached); + let rows = snapshot.take(&viewer.group_events, source, row, read_table); if rows.is_empty() { return; } @@ -2803,9 +2804,9 @@ impl GroupEventCursor { /// The SDK fires the inserts of one transaction in hash order, and the client must hear them /// in insert order: a filled Meeting Stone party's `COMPLETE` comes before `SETQUEUE(0, NONE)`. /// So the first callback of a transaction takes every row of that transaction for this - /// recipient from `cached`, which slices the transaction's rows out of a - /// [`GroupEventSnapshot`], and the later callbacks find theirs taken without calling it. Rows - /// of one transaction share `created_at`. + /// recipient from `cached`, which reads the table that already holds the whole transaction, and + /// the later callbacks find theirs taken without calling it. Rows of one transaction share + /// `created_at`. fn take( &self, source: GroupEventSource, @@ -2834,72 +2835,40 @@ impl GroupEventCursor { } } -/// The `game_group_event` rows of the latest transaction seen from one source database, grouped by -/// recipient. Every recipient of the transaction slices its rows from the same copy. +/// The `game_group_event` table as one source database last cached it, grouped by recipient guid. +/// Every recipient of a transaction takes its rows from the same read. #[derive(Default)] -pub(crate) struct GroupEventSnapshot(Mutex>); - -struct TransactionRows { - created_at: spacetimedb_sdk::Timestamp, - by_recipient: HashMap>, -} - -impl TransactionRows { - /// The rows of `row`'s transaction in `table`. `row` joins them when the table lacks it. - fn collect(row: &GroupEvent, table: Vec) -> Self { - let mut by_recipient: HashMap> = HashMap::new(); - let mut found = false; - for cached in table { - if cached.created_at != row.created_at { - continue; - } - found |= cached.id == row.id; - by_recipient - .entry(cached.recipient_guid) - .or_default() - .push(cached); - } - if !found { - by_recipient - .entry(row.recipient_guid) - .or_default() - .push(row.clone()); - } - Self { - created_at: row.created_at, - by_recipient, - } - } - - fn holds(&self, row: &GroupEvent) -> bool { - self.created_at == row.created_at - && self - .by_recipient - .get(&row.recipient_guid) - .is_some_and(|rows| rows.iter().any(|held| held.id == row.id)) - } -} +struct GroupEventSnapshot(HashMap>); impl GroupEventSnapshot { - /// The rows of `row`'s transaction addressed to `row`'s recipient, in no particular order. + /// [`GroupEventCursor::take`] over the recipient's rows of the snapshot. /// - /// The first call for a transaction runs `read_table`, which must return the cached table - /// holding the whole transaction. Every later call for the same transaction reuses that - /// copy. A row the held transaction lacks starts a new one, so two transactions with the - /// same `created_at` never share a copy. - fn recipient_rows( - &self, + /// `read_table` runs only when the recipient's group lacks `row`, which is the first callback + /// of each transaction. The cache already holds the whole transaction by then, so the later + /// callbacks reuse the read. A row the table lacks reads it again. + fn take( + &mut self, + cursor: &GroupEventCursor, + source: GroupEventSource, row: &GroupEvent, read_table: impl FnOnce() -> Vec, ) -> Vec { - let mut held = self.0.lock().unwrap_or_else(|p| p.into_inner()); - if !held.as_ref().is_some_and(|rows| rows.holds(row)) { - *held = Some(TransactionRows::collect(row, read_table())); - } - held.as_ref() - .and_then(|rows| rows.by_recipient.get(&row.recipient_guid)) - .cloned() - .unwrap_or_default() + cursor.take(source, row, || { + let held = self + .0 + .get(&row.recipient_guid) + .is_some_and(|rows| rows.iter().any(|held| held.id == row.id)); + if !held { + self.0.clear(); + for cached in read_table() { + self.0 + .entry(cached.recipient_guid) + .or_default() + .push(cached); + } + } + self.0.get(&row.recipient_guid).cloned().unwrap_or_default() + }) } } @@ -6167,8 +6136,9 @@ mod group_event_order_tests { ); } - /// A Raid roster list writes rows for every member in one transaction. The SDK fires an - /// insert callback per row; the table is read once, and every member still gets its own rows. + /// A Raid roster list writes two rows for each of 40 members in one transaction. The SDK + /// fires an insert callback per row; the table is read once, and every member relays both of + /// its rows once, oldest first. #[test] fn one_transaction_reads_the_table_once_for_every_recipient() { let members = 40_u64; @@ -6184,28 +6154,28 @@ mod group_event_order_tests { let mut table = vec![event(5, 100, 100)]; table.extend(transaction.iter().cloned()); - let snapshot = GroupEventSnapshot::default(); + let mut snapshot = GroupEventSnapshot::default(); + let cursors: Vec = + (0..members).map(|_| GroupEventCursor::default()).collect(); let reads = std::cell::Cell::new(0); - let mut relayed: Vec<(u64, Vec)> = Vec::new(); + let mut relayed: Vec>> = vec![Vec::new(); members as usize]; for row in transaction.iter().rev() { - let cursor = GroupEventCursor::default(); - let rows = cursor.take(GroupEventSource::RealmCore, row, || { - snapshot.recipient_rows(row, || { - reads.set(reads.get() + 1); - table.clone() - }) + let member = (row.recipient_guid - 100) as usize; + let rows = snapshot.take(&cursors[member], GroupEventSource::RealmCore, row, || { + reads.set(reads.get() + 1); + table.clone() }); - relayed.push((row.recipient_guid, ids(&rows))); + relayed[member].push(ids(&rows)); } assert_eq!(reads.get(), 1); - for member in 0..members { - let first = 1000 + member * 2; - assert!(relayed.contains(&(100 + member, vec![first, first + 1]))); + for (member, calls) in relayed.iter().enumerate() { + let first = 1000 + member as u64 * 2; + assert_eq!(calls, &[vec![first, first + 1], vec![]]); } } - /// A row the held transaction lacks starts the next one, even with the same timestamp. + /// A row the snapshot lacks starts the next transaction, even with the same timestamp. #[test] fn the_next_transaction_reads_the_table_again() { let table = [ @@ -6213,23 +6183,22 @@ mod group_event_order_tests { event(11, RECIPIENT, 200), event(12, RECIPIENT, 200), ]; - let snapshot = GroupEventSnapshot::default(); + let mut snapshot = GroupEventSnapshot::default(); + let cursor = GroupEventCursor::default(); let reads = std::cell::Cell::new(0); - let recipient_rows = |row: &GroupEvent, visible: usize| { - snapshot.recipient_rows(row, || { - reads.set(reads.get() + 1); - table[..visible].to_vec() - }) + let mut take = |row: &GroupEvent, visible: usize| { + ids( + &snapshot.take(&cursor, GroupEventSource::RealmCore, row, || { + reads.set(reads.get() + 1); + table[..visible].to_vec() + }), + ) }; - let first = recipient_rows(&table[0], 2); - let again = recipient_rows(&table[1], 2); - let next = recipient_rows(&table[2], 3); - + assert_eq!(take(&table[0], 2), [10, 11]); + assert!(take(&table[1], 2).is_empty()); + assert_eq!(take(&table[2], 3), [12]); assert_eq!(reads.get(), 2); - assert_eq!(ids(&first), [10, 11]); - assert_eq!(ids(&again), [10, 11]); - assert_eq!(ids(&next), [10, 11, 12]); } }