From 4b7725db1674104944116add4f55d3b16caaa8dc Mon Sep 17 00:00:00 2001 From: ddidderr Date: Tue, 21 Jul 2026 22:48:19 +0200 Subject: [PATCH] fix(call-to-play): compact terminal histories The 4,096-event store retained every completed call forever and local commands only reported that they reached the queue. Once the bound was reached, GUI actions could therefore fail with no user-visible result. The CLI snapshot wait could also be satisfied by an unrelated live event. Keep the complete event and chat history for every active call so late joiners receive full context. When the creator starts or cancels a call, replace its history with a single terminal tombstone; this bounds retained payload while still healing peers that missed the live terminal action. Publish commands now reply with the actual store result, and CLI snapshots use a direct reply. Test Plan: - `just fmt` -- passed - `just clippy` -- passed - `just test` -- passed, including full-cap terminal compaction - `just build` -- passed - `just peer-cli-tests S48` -- passed - `git diff --cached --check` -- passed --- crates/lanspread-peer-cli/src/main.rs | 35 ++-- crates/lanspread-peer/ARCHITECTURE.md | 15 +- crates/lanspread-peer/src/call_to_play.rs | 149 ++++++++++++++++-- crates/lanspread-peer/src/lib.rs | 26 ++- .../src-tauri/src/lib.rs | 16 +- 5 files changed, 198 insertions(+), 43 deletions(-) diff --git a/crates/lanspread-peer-cli/src/main.rs b/crates/lanspread-peer-cli/src/main.rs index 52b268d..05a9e76 100644 --- a/crates/lanspread-peer-cli/src/main.rs +++ b/crates/lanspread-peer-cli/src/main.rs @@ -47,7 +47,7 @@ use lanspread_peer_cli::{ use serde_json::{Value, json}; use tokio::{ io::{AsyncBufReadExt, BufReader}, - sync::{Notify, RwLock, mpsc}, + sync::{Notify, RwLock, mpsc, oneshot}, }; #[derive(Debug)] @@ -103,7 +103,6 @@ struct CliState { unavailable_games: HashSet, downloads: HashMap, call_to_play_events: Vec, - call_to_play_generation: u64, } #[derive(Clone, serde::Serialize)] @@ -247,24 +246,25 @@ async fn handle_command( CliCommand::ListPeers => list_peers(shared).await, CliCommand::ListGames => list_games(shared).await, CliCommand::ListCallToPlay => { - let generation = shared.state.read().await.call_to_play_generation; - sender.send(PeerCommand::GetCallToPlayEvents)?; - tokio::time::timeout(Duration::from_secs(1), async { - loop { - if shared.state.read().await.call_to_play_generation > generation { - break; - } - shared.notify.notified().await; - } - }) - .await - .wrap_err("timed out waiting for Call to Play history")?; - let events = shared.state.read().await.call_to_play_events.clone(); + let (reply, result) = oneshot::channel(); + sender.send(PeerCommand::GetCallToPlayEvents { reply: Some(reply) })?; + let events = tokio::time::timeout(Duration::from_secs(1), result) + .await + .wrap_err("timed out waiting for Call to Play history")? + .wrap_err("peer stopped before returning Call to Play history")?; Ok(json!({ "events": events })) } CliCommand::PublishCallToPlay { event } => { - sender.send(PeerCommand::PublishCallToPlay(event.clone()))?; - Ok(json!({"queued": true, "event_id": event.id})) + let (reply, result) = oneshot::channel(); + sender.send(PeerCommand::PublishCallToPlay { + event: event.clone(), + reply, + })?; + result + .await + .wrap_err("peer stopped before publishing Call to Play event")? + .map_err(eyre::Report::msg)?; + Ok(json!({"published": true, "event_id": event.id})) } CliCommand::SetGameDir { path } => { sender.send(PeerCommand::SetGameDir(path.clone()))?; @@ -534,7 +534,6 @@ async fn update_state_from_event(shared: &SharedState, event: PeerEvent) -> (&'s .filter(|event| known.insert(event.id.clone())) .cloned(), ); - state.call_to_play_generation = state.call_to_play_generation.saturating_add(1); ("call-to-play-events", json!({ "events": events })) } PeerEvent::GotGameFiles { diff --git a/crates/lanspread-peer/ARCHITECTURE.md b/crates/lanspread-peer/ARCHITECTURE.md index 6bb9e5f..e71418e 100644 --- a/crates/lanspread-peer/ARCHITECTURE.md +++ b/crates/lanspread-peer/ARCHITECTURE.md @@ -49,10 +49,17 @@ When a peer is discovered: ### Call to Play replication Call to Play is transient peer-session state rather than database state. The -peer keeps a bounded event history, deduplicated by event ID. A local action is -applied to that history, sent to the UI, and broadcast to every currently known -peer. An incoming live event is applied once and sent to the UI without being -rebroadcast, which prevents forwarding loops. +peer keeps a bounded event history, deduplicated by event ID. Every event and +chat message remains in the snapshot for the full lifetime of an active call, +so a peer joining mid-call receives the complete context. A creator's `Start` +or `Cancel` event replaces that terminal call with one small tombstone; this +lets a peer that missed the live action heal on its next handshake without +retaining the inactive call's full history. Active calls are never partially +trimmed. If genuinely active history reaches the bound, local publishes return +an error to the caller instead of appearing to succeed. A +local action is applied to that history, sent to the UI, and broadcast to every +currently known peer. An incoming live event is applied once and sent to the UI +without being rebroadcast, which prevents forwarding loops. Actors are keyed by the peer's stable ID and carry a separate display name. The origin peer overwrites the actor ID on local actions, and live-event envelopes diff --git a/crates/lanspread-peer/src/call_to_play.rs b/crates/lanspread-peer/src/call_to_play.rs index 5c102d5..10cd12e 100644 --- a/crates/lanspread-peer/src/call_to_play.rs +++ b/crates/lanspread-peer/src/call_to_play.rs @@ -1,6 +1,6 @@ //! Replicated event history for Call to Play coordination. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use lanspread_proto::{CallToPlayAction, CallToPlayEvent}; use tokio::sync::mpsc::UnboundedSender; @@ -29,12 +29,16 @@ impl CallToPlayStore { if self.event_ids.contains(&event.id) { return Ok(false); } - if self.events.len() >= MAX_EVENTS { + + let event_id = event.id.clone(); + self.event_ids.insert(event_id.clone()); + self.events.push(event); + self.evict_terminal_calls(); + if self.events.len() > MAX_EVENTS { + self.events.retain(|event| event.id != event_id); + self.event_ids.remove(&event_id); return Err("Call to Play event history is full"); } - - self.event_ids.insert(event.id.clone()); - self.events.push(event); Ok(true) } @@ -49,19 +53,66 @@ impl CallToPlayStore { } accepted } + + fn evict_terminal_calls(&mut self) { + let mut creators = HashMap::::new(); + for event in &self.events { + if !matches!(event.action, CallToPlayAction::Create { .. }) { + continue; + } + let candidate = (event.at, event.id.clone(), event.actor_id.clone()); + creators + .entry(event.call_id.clone()) + .and_modify(|current| { + if (candidate.0, &candidate.1) < (current.0, ¤t.1) { + current.clone_from(&candidate); + } + }) + .or_insert(candidate); + } + + let mut terminal_events = HashMap::::new(); + for event in &self.events { + let Some((created_at, create_id, creator_id)) = creators.get(&event.call_id) else { + continue; + }; + if !matches!( + event.action, + CallToPlayAction::Cancel | CallToPlayAction::Start + ) || event.actor_id != *creator_id + || (event.at, &event.id) <= (*created_at, create_id) + { + continue; + } + let candidate = (event.at, event.id.clone()); + terminal_events + .entry(event.call_id.clone()) + .and_modify(|current| { + if (candidate.0, &candidate.1) < (current.0, ¤t.1) { + current.clone_from(&candidate); + } + }) + .or_insert(candidate); + } + self.events.retain(|event| { + terminal_events + .get(&event.call_id) + .is_none_or(|(_, terminal_id)| event.id == *terminal_id) + }); + } } pub(crate) async fn publish( ctx: &Ctx, tx_notify_ui: &UnboundedSender, mut event: CallToPlayEvent, -) { +) -> Result<(), &'static str> { event.actor_id.clone_from(ctx.peer_id.as_ref()); match ctx.call_to_play.write().await.insert(event.clone()) { - Ok(false) => return, + Ok(false) => return Ok(()), Err(err) => { log::warn!("Rejecting local Call to Play event {}: {err}", event.id); - return; + return Err(err); } Ok(true) => {} } @@ -87,6 +138,7 @@ pub(crate) async fn publish( }); futures::future::join_all(deliveries).await; }); + Ok(()) } fn validate_event(event: &CallToPlayEvent) -> Result<(), &'static str> { @@ -154,7 +206,7 @@ fn validate_nonempty( mod tests { use lanspread_proto::{CallToPlayAction, CallToPlayEvent}; - use super::CallToPlayStore; + use super::{CallToPlayStore, MAX_EVENTS}; fn create_event(id: &str) -> CallToPlayEvent { CallToPlayEvent { @@ -172,6 +224,17 @@ mod tests { } } + fn action_event(id: &str, call_id: &str, action: CallToPlayAction) -> CallToPlayEvent { + CallToPlayEvent { + id: id.to_string(), + call_id: call_id.to_string(), + actor_id: "peer-alice".to_string(), + actor_name: "Alice".to_string(), + at: 2_000, + action, + } + } + #[test] fn deduplicates_events_without_reordering_history() { let mut store = CallToPlayStore::default(); @@ -229,4 +292,72 @@ mod tests { assert_eq!(accepted, [create_event("event-2")]); } + + #[test] + fn terminal_action_evicts_the_whole_call_even_at_capacity() { + let mut store = CallToPlayStore::default(); + store + .insert(create_event("create")) + .expect("create should fit"); + for index in 1..MAX_EVENTS { + store + .insert(action_event( + &format!("event-{index}"), + "call-1", + CallToPlayAction::Rsvp, + )) + .expect("active history should fit through the cap"); + } + + assert!( + store + .insert(action_event("start", "call-1", CallToPlayAction::Start)) + .expect("terminal action should compact the full call") + ); + let snapshot = store.snapshot(); + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].id, "start"); + } + + #[test] + fn terminal_action_keeps_unrelated_active_calls() { + let mut store = CallToPlayStore::default(); + store + .insert(create_event("call-1-create")) + .expect("first call should fit"); + let mut other_create = create_event("call-2-create"); + other_create.call_id = "call-2".to_string(); + store.insert(other_create).expect("second call should fit"); + store + .insert(action_event("cancel", "call-1", CallToPlayAction::Cancel)) + .expect("cancel should compact the first call"); + + let snapshot = store.snapshot(); + assert_eq!(snapshot.len(), 2); + assert!(snapshot.iter().any(|event| event.id == "cancel")); + assert!(snapshot.iter().any(|event| event.call_id == "call-2")); + } + + #[test] + fn full_active_history_returns_an_error() { + let mut store = CallToPlayStore::default(); + store + .insert(create_event("create")) + .expect("create should fit"); + for index in 1..MAX_EVENTS { + store + .insert(action_event( + &format!("event-{index}"), + "call-1", + CallToPlayAction::Rsvp, + )) + .expect("active history should fit through the cap"); + } + + assert_eq!( + store.insert(action_event("overflow", "call-1", CallToPlayAction::Rsvp)), + Err("Call to Play event history is full") + ); + assert_eq!(store.snapshot().len(), MAX_EVENTS); + } } diff --git a/crates/lanspread-peer/src/lib.rs b/crates/lanspread-peer/src/lib.rs index 9061ff0..3ae9a01 100644 --- a/crates/lanspread-peer/src/lib.rs +++ b/crates/lanspread-peer/src/lib.rs @@ -60,6 +60,7 @@ pub use peer_db::{ use tokio::sync::{ RwLock, mpsc::{UnboundedReceiver, UnboundedSender}, + oneshot, }; use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -228,7 +229,7 @@ pub enum ActiveOperationKind { } /// Commands sent to the peer system from the UI. -#[derive(Clone, Debug)] +#[derive(Debug)] pub enum PeerCommand { /// Request a list of all available games. ListGames, @@ -264,9 +265,14 @@ pub enum PeerCommand { /// Connect directly to a peer address without waiting for mDNS discovery. ConnectPeer(SocketAddr), /// Publish one local Call to Play action to this peer and the LAN. - PublishCallToPlay(CallToPlayEvent), + PublishCallToPlay { + event: CallToPlayEvent, + reply: oneshot::Sender>, + }, /// Request the complete in-memory Call to Play history. - GetCallToPlayEvents, + GetCallToPlayEvents { + reply: Option>>, + }, } /// Optional startup settings for non-GUI callers and tests. @@ -497,12 +503,18 @@ async fn handle_peer_commands( PeerCommand::ConnectPeer(addr) => { handle_connect_peer_command(ctx, tx_notify_ui, addr).await; } - PeerCommand::PublishCallToPlay(event) => { - call_to_play::publish(ctx, tx_notify_ui, event).await; + PeerCommand::PublishCallToPlay { event, reply } => { + let result = call_to_play::publish(ctx, tx_notify_ui, event) + .await + .map_err(str::to_owned); + let _ = reply.send(result); } - PeerCommand::GetCallToPlayEvents => { + PeerCommand::GetCallToPlayEvents { reply } => { let events = ctx.call_to_play.read().await.snapshot(); - events::send(tx_notify_ui, PeerEvent::CallToPlayEvents(events)); + events::send(tx_notify_ui, PeerEvent::CallToPlayEvents(events.clone())); + if let Some(reply) = reply { + let _ = reply.send(events); + } } } } diff --git a/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs b/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs index 8d756c5..884201c 100644 --- a/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs +++ b/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs @@ -40,6 +40,7 @@ use tauri_plugin_shell::{ use tokio::sync::{ RwLock, mpsc::{UnboundedReceiver, UnboundedSender}, + oneshot, }; use tracing::{Event, Level, Metadata, Subscriber, field::Visit}; use tracing_subscriber::{ @@ -285,7 +286,10 @@ async fn request_call_to_play_events( return Ok(None); }; - if peer_ctrl.send(PeerCommand::GetCallToPlayEvents).is_err() { + if peer_ctrl + .send(PeerCommand::GetCallToPlayEvents { reply: None }) + .is_err() + { return Ok(None); } Ok(state.inner().local_peer_id.read().await.clone()) @@ -295,16 +299,18 @@ async fn request_call_to_play_events( async fn publish_call_to_play( event: CallToPlayEvent, state: tauri::State<'_, LanSpreadState>, -) -> tauri::Result { +) -> Result { let peer_ctrl = state.inner().peer_ctrl.read().await.clone(); let Some(peer_ctrl) = peer_ctrl else { log::warn!("Peer system not initialized yet"); return Ok(false); }; - Ok(peer_ctrl - .send(PeerCommand::PublishCallToPlay(event)) - .is_ok()) + let (reply, result) = oneshot::channel(); + peer_ctrl + .send(PeerCommand::PublishCallToPlay { event, reply }) + .map_err(|err| err.to_string())?; + result.await.map_err(|err| err.to_string())?.map(|()| true) } #[tauri::command]