diff --git a/crates/lanspread-peer/src/call_to_play.rs b/crates/lanspread-peer/src/call_to_play.rs index 50d5959..6b9ca83 100644 --- a/crates/lanspread-peer/src/call_to_play.rs +++ b/crates/lanspread-peer/src/call_to_play.rs @@ -25,6 +25,8 @@ use crate::peer_db::PeerEndpointGeneration; pub(crate) const MAX_CALL_TO_PLAY_AUTHORS: usize = 64; pub(crate) const LOCAL_TERMINAL_EVENT_RESERVE: usize = 1; pub(crate) const LOCAL_TERMINAL_BYTE_RESERVE: usize = 512; +const MAX_CALL_TO_PLAY_CALLS_PER_AUTHOR: usize = 128; +const MAX_REMOTE_CALL_TO_PLAY_EVENTS: usize = 16_384; const EXPIRED_RETENTION_MS: i64 = 5 * 60_000; const TERMINAL_RETENTION_MS: i64 = 15 * 60_000; @@ -177,6 +179,9 @@ pub(crate) enum CallToPlayValidationError { NonMonotonicAuthorHistory, NonMonotonicCallHistory(CallId), ActionAfterTerminal(CallId), + TooManyCalls { + maximum: usize, + }, } impl fmt::Display for CallToPlayValidationError { @@ -221,6 +226,12 @@ impl fmt::Display for CallToPlayValidationError { "Call-to-Play call {call_id} has an action after termination" ) } + Self::TooManyCalls { maximum } => { + write!( + formatter, + "Call-to-Play author exceeds the {maximum}-call limit" + ) + } } } } @@ -338,6 +349,9 @@ pub(crate) enum ObserveRemoteAuthorOutcome { generation_rebound: bool, }, InvalidAbsent(CallToPlayValidationError), + BudgetLimited { + session_changed: bool, + }, AtCapacity, RejectedLocalIdentity, } @@ -345,7 +359,10 @@ pub(crate) enum ObserveRemoteAuthorOutcome { impl ObserveRemoteAuthorOutcome { #[must_use] pub(crate) const fn view_changed(&self) -> bool { - matches!(self, Self::Applied { .. } | Self::InvalidCleared(_)) + matches!( + self, + Self::Applied { .. } | Self::BudgetLimited { .. } | Self::InvalidCleared(_) + ) } } @@ -529,6 +546,22 @@ impl CallToPlayStore { } }; + let needs_budget = self.remote.get(&author_id).is_none_or(|current| { + current.runtime_session_id != runtime_session_id + || snapshot.revision > current.snapshot.revision + }); + let budget_limited = + needs_budget && !self.remote_event_budget_allows(author_id, snapshot.events.len()); + let snapshot = if budget_limited { + CallToPlayAuthorSnapshot { + revision: snapshot.revision, + display_name: snapshot.display_name, + events: Vec::new(), + } + } else { + snapshot + }; + if let Some(current) = self.remote.get_mut(&author_id) { let session_changed = current.runtime_session_id != runtime_session_id; if session_changed || snapshot.revision > current.snapshot.revision { @@ -537,7 +570,11 @@ impl CallToPlayStore { runtime_session_id, snapshot, }; - return ObserveRemoteAuthorOutcome::Applied { session_changed }; + return if budget_limited { + ObserveRemoteAuthorOutcome::BudgetLimited { session_changed } + } else { + ObserveRemoteAuthorOutcome::Applied { session_changed } + }; } let generation_rebound = current.endpoint_generation != endpoint_generation; @@ -563,11 +600,30 @@ impl CallToPlayStore { snapshot, }, ); - ObserveRemoteAuthorOutcome::Applied { - session_changed: true, + if budget_limited { + ObserveRemoteAuthorOutcome::BudgetLimited { + session_changed: true, + } + } else { + ObserveRemoteAuthorOutcome::Applied { + session_changed: true, + } } } + fn remote_event_budget_allows(&self, author_id: PeerId, incoming: usize) -> bool { + let current = self + .remote + .values() + .map(|slice| slice.snapshot.events.len()) + .sum::(); + let replaced = self + .remote + .get(&author_id) + .map_or(0, |slice| slice.snapshot.events.len()); + current.saturating_sub(replaced).saturating_add(incoming) <= MAX_REMOTE_CALL_TO_PLAY_EVENTS + } + /// Clears every remote author and returns the resulting full, local-only /// publication. /// @@ -640,6 +696,14 @@ impl CallToPlayStore { self.remote.len() } + #[cfg(test)] + fn remote_event_count(&self) -> usize { + self.remote + .values() + .map(|slice| slice.snapshot.events.len()) + .sum() + } + fn publish_local_at( &mut self, intent: CallToPlayLocalIntent, @@ -960,6 +1024,7 @@ fn validate_author_snapshot( let mut event_ids = HashSet::with_capacity(snapshot.events.len()); let mut creator_calls = HashMap::::new(); + let mut creator_call_count = 0_usize; if snapshot .events .windows(2) @@ -978,8 +1043,14 @@ fn validate_author_snapshot( call_id: event.call_id, }); } - if let CallToPlayAction::Create { deadline, .. } = event.action - && creator_calls + if let CallToPlayAction::Create { deadline, .. } = event.action { + creator_call_count += 1; + if creator_call_count > MAX_CALL_TO_PLAY_CALLS_PER_AUTHOR { + return Err(CallToPlayValidationError::TooManyCalls { + maximum: MAX_CALL_TO_PLAY_CALLS_PER_AUTHOR, + }); + } + if creator_calls .insert( event.call_id, CreatorValidationState { @@ -990,8 +1061,9 @@ fn validate_author_snapshot( }, ) .is_some() - { - return Err(CallToPlayValidationError::DuplicateCreate(event.call_id)); + { + return Err(CallToPlayValidationError::DuplicateCreate(event.call_id)); + } } } @@ -2437,6 +2509,91 @@ mod tests { assert_eq!(receipt.revision, 1); } + #[test] + fn one_author_cannot_publish_unbounded_creator_roots() { + let author = peer(1); + let generations = endpoint_generations(1); + let events = (0..=MAX_CALL_TO_PLAY_CALLS_PER_AUTHOR) + .map(|index| { + let index = u128::try_from(index).expect("test index fits in nonce"); + create_event( + author, + CallId::new(author, call_nonce(index + 1_000)), + index + 1, + NOW + i64::try_from(index).expect("test index fits in timestamp"), + NOW + 60_000, + ) + }) + .collect(); + let prepared = prepare( + author, + generations[0], + session(2), + snapshot(1, "Alice", events), + ); + + assert_eq!( + prepared.validation_error(), + Some(&CallToPlayValidationError::TooManyCalls { + maximum: MAX_CALL_TO_PLAY_CALLS_PER_AUTHOR, + }) + ); + } + + fn maximal_author_history(author: PeerId) -> Vec { + let call_id = CallId::new(author, call_nonce(1)); + let mut events = Vec::with_capacity(MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR); + events.push(create_event(author, call_id, 1, NOW, NOW + 60_000)); + events.extend((2..=MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR as u128).map(|id| { + action_event( + call_id, + id, + NOW + i64::try_from(id).expect("event id fits in timestamp"), + CallToPlayAction::Rsvp, + ) + })); + events + } + + #[test] + fn aggregate_remote_event_budget_keeps_a_revision_watermark() { + let generation = endpoint_generations(1)[0]; + let mut store = store(peer(200)); + for seed in 1..=4 { + assert!(matches!( + store.observe_prepared_remote(prepare( + peer(seed), + generation, + session(seed), + snapshot(u64::from(seed), "Peer", maximal_author_history(peer(seed)),), + )), + ObserveRemoteAuthorOutcome::Applied { .. } + )); + } + assert_eq!(store.remote_event_count(), MAX_REMOTE_CALL_TO_PLAY_EVENTS); + + let fifth = peer(5); + assert_eq!( + store.observe_prepared_remote(prepare( + fifth, + generation, + session(5), + snapshot(5, "Fifth", maximal_author_history(fifth)), + )), + ObserveRemoteAuthorOutcome::BudgetLimited { + session_changed: true, + } + ); + assert_eq!(store.remote_event_count(), MAX_REMOTE_CALL_TO_PLAY_EVENTS); + assert_eq!( + store + .remote_author_state(fifth) + .expect("budget-limited author should retain a watermark") + .revision, + 5, + ); + } + #[test] fn full_view_is_deterministic_and_wholly_recomputed() { let local = peer(3); diff --git a/crates/lanspread-peer/src/services/remote_state.rs b/crates/lanspread-peer/src/services/remote_state.rs index ba6afc2..539ac6b 100644 --- a/crates/lanspread-peer/src/services/remote_state.rs +++ b/crates/lanspread-peer/src/services/remote_state.rs @@ -287,6 +287,11 @@ fn log_call_to_play_outcome(peer_id: PeerId, outcome: &ObserveRemoteAuthorOutcom ObserveRemoteAuthorOutcome::AtCapacity => { log::warn!("Call-to-Play author limit reached; ignoring {peer_id}"); } + ObserveRemoteAuthorOutcome::BudgetLimited { .. } => { + log::warn!( + "Call-to-Play aggregate event budget reached; retaining an empty revision watermark for {peer_id}" + ); + } ObserveRemoteAuthorOutcome::RejectedLocalIdentity => { log::warn!("Rejected remote Call-to-Play snapshot for local identity {peer_id}"); }