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]