diff --git a/gateway/src/stdb/world_view.rs b/gateway/src/stdb/world_view.rs index d6c489a3e..924ff7d6b 100644 --- a/gateway/src/stdb/world_view.rs +++ b/gateway/src/stdb/world_view.rs @@ -1479,9 +1479,9 @@ 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. +/// 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, label: &'static str, @@ -1490,12 +1490,13 @@ fn wire_group_events( source: GroupEventSource, ) { let view = view.clone(); + 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, || { + group_event_appeared(&view, &coord, source, row, &mut snapshot, || { ctx.db.game_group_event().iter().collect() }); }); @@ -2753,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; @@ -2764,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,7 +2805,7 @@ impl GroupEventCursor { /// 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 + /// the later callbacks find theirs taken without calling it. Rows of one transaction share /// `created_at`. fn take( &self, @@ -2833,6 +2835,43 @@ impl GroupEventCursor { } } +/// 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)] +struct GroupEventSnapshot(HashMap>); + +impl GroupEventSnapshot { + /// [`GroupEventCursor::take`] over the recipient's rows of the snapshot. + /// + /// `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 { + 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() + }) + } +} + /// 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 +6061,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 +6135,71 @@ mod group_event_order_tests { [3] ); } + + /// 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; + 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 mut snapshot = GroupEventSnapshot::default(); + let cursors: Vec = + (0..members).map(|_| GroupEventCursor::default()).collect(); + let reads = std::cell::Cell::new(0); + let mut relayed: Vec>> = vec![Vec::new(); members as usize]; + for row in transaction.iter().rev() { + 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[member].push(ids(&rows)); + } + + assert_eq!(reads.get(), 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 snapshot lacks starts the next transaction, 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 mut snapshot = GroupEventSnapshot::default(); + let cursor = GroupEventCursor::default(); + let reads = std::cell::Cell::new(0); + let mut take = |row: &GroupEvent, visible: usize| { + ids( + &snapshot.take(&cursor, GroupEventSource::RealmCore, row, || { + reads.set(reads.get() + 1); + table[..visible].to_vec() + }), + ) + }; + + 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); + } } // ===============================================================================================