ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(peer): bound remote Call-to-Play state

Browse Source 
Wire-level author limits still allowed one peer to publish thousands of creator
roots and a Sybil set to retain hundreds of thousands of aggregate events.
Those valid snapshots were repeatedly projected for the desktop.

Limit one author to 128 creator roots and all retained remote history to 16,384
events. Over-budget revisions replace that author's slice with an empty
watermark, which frees memory and prevents liveness from pulling the rejected
snapshot every five seconds. Local author capacity remains independent.

Test Plan:
- `just test` -- passed outside the sandbox; 498 peer tests and all workspace
  targets passed.
- `just fmt` -- Rust formatting completed; the recipe then hit the pre-existing
  generated security-report Markdown-lint failures.
- `git diff --cached --check` -- passed.
This commit is contained in:
ddidderr
2026-09-12 12:26:55 +02:00
parent 0189622085
commit 30663fea34
2 changed files with 170 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/src/call_to_play.rs
+165 -8
View File
@@ -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::<usize>();
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::<CallId, CreatorValidationState>::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<CallToPlayAuthorEvent> {
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);
crates/lanspread-peer/src/services/remote_state.rs
+5
View File
@@ -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}");
}