From 37420e26edad797d58b173439e76f335e16a9bd6 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 12 Sep 2026 13:06:10 +0200 Subject: [PATCH] fix(peer): coalesce full UI view publications Keep one queued and one replaceable pending snapshot for remote-library and Call-to-Play views while preserving lifecycle-event FIFO delivery. Generation barriers and fenced drains prevent stale nonempty views from crossing disable or acknowledgement boundaries. Test Plan: - just test - just clippy - focused burst, lifecycle ordering, fence, repeated-barrier, and stale-view tests - independent ordering review - git diff --check --- crates/lanspread-peer-cli/src/main.rs | 6 +- crates/lanspread-peer/src/context.rs | 17 +- .../src/download/orchestrator.rs | 6 +- crates/lanspread-peer/src/events.rs | 596 +++++++++++++++++- crates/lanspread-peer/src/handlers.rs | 208 +++--- crates/lanspread-peer/src/lib.rs | 23 +- .../lanspread-peer/src/network_generation.rs | 47 +- .../lanspread-peer/src/services/discovery.rs | 5 +- .../lanspread-peer/src/services/liveness.rs | 6 +- .../src/services/local_monitor.rs | 47 +- .../src/services/remote_state.rs | 32 +- .../lanspread-peer/src/services/state_sync.rs | 4 +- .../lanspread-peer/src/services/transfer.rs | 6 +- crates/lanspread-peer/src/startup.rs | 21 +- crates/lanspread-peer/src/stream_install.rs | 13 +- crates/lanspread-peer/src/transfer_status.rs | 10 +- .../src-tauri/src/lib.rs | 32 +- 17 files changed, 828 insertions(+), 251 deletions(-) diff --git a/crates/lanspread-peer-cli/src/main.rs b/crates/lanspread-peer-cli/src/main.rs index 21fc5ad..7190a2f 100644 --- a/crates/lanspread-peer-cli/src/main.rs +++ b/crates/lanspread-peer-cli/src/main.rs @@ -34,6 +34,7 @@ use lanspread_peer::{ PeerCommand, PeerEndpoint, PeerEvent, + PeerEventReceiver, PeerGameDB, PeerIdentity, PeerRuntimeComponent, @@ -44,6 +45,7 @@ use lanspread_peer::{ StreamInstallSettings, load_peer_identity, migrate_legacy_state, + peer_event_channel, start_peer_with_options, }; use lanspread_peer_cli::{ @@ -308,7 +310,7 @@ async fn main() -> eyre::Result<()> { let fixture_seeds = seed_fixtures(&args.games_dir, &args.fixtures)?; let migration = migrate_legacy_state(&args.games_dir, &args.state_dir).await; - let (tx_events, rx_events) = mpsc::unbounded_channel(); + let (tx_events, rx_events) = peer_event_channel(); let peer_game_db = Arc::new(RwLock::new(PeerGameDB::new())); let catalog_game_db = Arc::new(catalog_game_db); let active_outbound_transfers: OutboundTransfers = Arc::new(RwLock::new(HashMap::new())); @@ -865,7 +867,7 @@ async fn wait_peers(shared: &SharedState, count: usize, timeout: Duration) -> ey } async fn event_loop( - mut rx_events: mpsc::UnboundedReceiver, + mut rx_events: PeerEventReceiver, shared: Arc, writer: JsonlWriter, ) { diff --git a/crates/lanspread-peer/src/context.rs b/crates/lanspread-peer/src/context.rs index b0ac058..c5e6a2c 100644 --- a/crates/lanspread-peer/src/context.rs +++ b/crates/lanspread-peer/src/context.rs @@ -18,6 +18,7 @@ use tokio_util::{sync::CancellationToken, task::TaskTracker}; use crate::{ PeerEvent, + PeerEventSender, StreamInstallProvider, Unpacker, call_to_play::CallToPlayStore, @@ -45,7 +46,7 @@ const OUTBOUND_CHANGE_DIRTY: u8 = 2; #[derive(Debug)] pub struct OutboundTransferChange { state: Arc, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, } impl Drop for OutboundTransferChange { @@ -105,13 +106,13 @@ impl Drop for OutboundTransferChange { #[derive(Clone, Debug)] pub(crate) struct OutboundTransferNotifier { state: Arc, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, } impl OutboundTransferNotifier { fn new( state: Arc, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, ) -> Self { Self { state, @@ -242,7 +243,7 @@ impl NetworkServiceCtx { pub(crate) fn to_peer_ctx( &self, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, ) -> PeerCtx { self.core.to_peer_ctx(tx_notify_ui, self.shutdown.clone()) } @@ -271,7 +272,7 @@ pub struct PeerCtx { pub peer_id: PeerId, pub(crate) runtime_session_id: RuntimeSessionId, pub(crate) state_sync: StateSyncHandle, - pub tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + pub tx_notify_ui: PeerEventSender, pub stream_install_provider: Arc, pub shutdown: CancellationToken, pub active_outbound_transfers: OutboundTransfers, @@ -345,7 +346,7 @@ impl Ctx { /// Creates a `PeerCtx` from this context. pub fn to_peer_ctx( &self, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, shutdown: CancellationToken, ) -> PeerCtx { let outbound_transfer_notifier = OutboundTransferNotifier::new( @@ -459,7 +460,7 @@ mod tests { sync::{Arc, atomic::AtomicU8}, }; - use tokio::sync::{RwLock, mpsc}; + use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; use super::{OUTBOUND_CHANGE_IDLE, OperationGuard, OperationKind, OutboundTransferNotifier}; @@ -467,7 +468,7 @@ mod tests { #[test] fn outbound_transfer_churn_keeps_at_most_one_edge_queued() { - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let notifier = OutboundTransferNotifier::new(Arc::new(AtomicU8::new(OUTBOUND_CHANGE_IDLE)), tx); diff --git a/crates/lanspread-peer/src/download/orchestrator.rs b/crates/lanspread-peer/src/download/orchestrator.rs index 6e7157a..a5410db 100644 --- a/crates/lanspread-peer/src/download/orchestrator.rs +++ b/crates/lanspread-peer/src/download/orchestrator.rs @@ -9,7 +9,7 @@ use std::{ use futures::stream::FuturesUnordered; use lanspread_db::content_manifest::ContentId; use lanspread_proto::PeerEndpoint; -use tokio::sync::mpsc::UnboundedSender; +use crate::PeerEventSender; use tokio_util::sync::CancellationToken; use super::{ @@ -153,7 +153,7 @@ pub(crate) struct DownloadGameRequest<'a> { pub(crate) sources: &'a [PeerEndpoint], pub(crate) content_id: ContentId, pub(crate) quarantine: &'a ContentQuarantine, - pub(crate) tx_notify_ui: UnboundedSender, + pub(crate) tx_notify_ui: PeerEventSender, pub(crate) cancel_token: CancellationToken, pub(crate) quic: QuicConnector, } @@ -372,7 +372,7 @@ struct TransferContext<'a> { sources: &'a [PeerEndpoint], content_id: ContentId, quarantine: &'a ContentQuarantine, - tx_notify_ui: &'a UnboundedSender, + tx_notify_ui: &'a PeerEventSender, cancel_token: &'a CancellationToken, quic: &'a QuicConnector, version_buffer: Arc, diff --git a/crates/lanspread-peer/src/events.rs b/crates/lanspread-peer/src/events.rs index f4207e4..e597e17 100644 --- a/crates/lanspread-peer/src/events.rs +++ b/crates/lanspread-peer/src/events.rs @@ -2,15 +2,20 @@ use std::{ collections::{BTreeMap, HashMap}, - sync::Arc, + fmt, + sync::{Arc, Mutex, MutexGuard}, }; use lanspread_db::content_manifest::ContentId; -use tokio::sync::{RwLock, mpsc::UnboundedSender}; +use tokio::sync::{ + RwLock, + mpsc::{self, error::TryRecvError}, +}; use crate::{ ActiveOperation, ActiveOperationKind, + CallToPlayView, PeerEvent, RemoteGameAvailability, RemoteLibraryView, @@ -18,10 +23,431 @@ use crate::{ peer_db::PeerGameDB, }; -pub fn send(tx_notify_ui: &UnboundedSender, event: PeerEvent) { +/// Sender side of the peer-to-UI event transport. +/// +/// Lifecycle events are queued losslessly in send order. The two complete-view +/// variants keep one concrete queued delivery plus one replaceable latest +/// pending snapshot, so a slow UI cannot make complete snapshots accumulate. +pub struct PeerEventSender { + queue: mpsc::UnboundedSender, + views: Arc>, +} + +impl Clone for PeerEventSender { + fn clone(&self) -> Self { + Self { + queue: self.queue.clone(), + views: Arc::clone(&self.views), + } + } +} + +impl fmt::Debug for PeerEventSender { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PeerEventSender") + .field("closed", &self.is_closed()) + .finish_non_exhaustive() + } +} + +/// Receiver side of the peer-to-UI event transport. +pub struct PeerEventReceiver { + queue: mpsc::UnboundedReceiver, + views: Arc>, +} + +#[derive(Debug)] +pub struct PeerEventSendError { + kind: &'static str, +} + +impl PeerEventSendError { + #[must_use] + pub const fn kind(&self) -> &'static str { + self.kind + } +} + +impl fmt::Display for PeerEventSendError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "peer event channel closed while sending {}", + self.kind + ) + } +} + +impl std::error::Error for PeerEventSendError {} + +enum QueuedPeerEvent { + Lifecycle(PeerEvent), + RemoteLibrary(ViewWake), + CallToPlay(ViewWake), +} + +struct ViewWake { + queue: mpsc::UnboundedSender, +} + +struct QueueClosed; + +struct VersionedView { + generation: Arc<()>, + snapshot: T, +} + +struct CoalescedViews { + generation: Arc<()>, + remote_library: ViewSlot, + call_to_play: ViewSlot, +} + +struct ViewSlot { + marker_queued: bool, + queued_delivery: Option>, + pending_latest: Option>, +} + +impl Default for ViewSlot { + fn default() -> Self { + Self { + marker_queued: false, + queued_delivery: None, + pending_latest: None, + } + } +} + +impl Default for CoalescedViews { + fn default() -> Self { + Self { + generation: Arc::new(()), + remote_library: ViewSlot::default(), + call_to_play: ViewSlot::default(), + } + } +} + +/// Creates the event transport used by one peer runtime embedding. +#[must_use] +pub fn peer_event_channel() -> (PeerEventSender, PeerEventReceiver) { + let (queue_tx, queue_rx) = mpsc::unbounded_channel(); + let views = Arc::new(Mutex::new(CoalescedViews::default())); + ( + PeerEventSender { + queue: queue_tx, + views: Arc::clone(&views), + }, + PeerEventReceiver { + queue: queue_rx, + views, + }, + ) +} + +impl PeerEventSender { + /// Sends one event, coalescing only complete remote-library and Call-to-Play + /// projections. + pub fn send(&self, event: PeerEvent) -> Result<(), PeerEventSendError> { + let kind: &'static str = (&event).into(); + if self.queue.is_closed() { + return Err(PeerEventSendError { kind }); + } + let mut views = lock_views(&self.views); + let generation = Arc::clone(&views.generation); + let mut failed_lifecycle = None; + let result = match event { + PeerEvent::RemoteLibraryView(view) => enqueue_view( + &self.queue, + &generation, + &mut views.remote_library, + view, + QueuedPeerEvent::RemoteLibrary, + ), + PeerEvent::CallToPlayView(view) => enqueue_view( + &self.queue, + &generation, + &mut views.call_to_play, + view, + QueuedPeerEvent::CallToPlay, + ), + event => self + .queue + .send(QueuedPeerEvent::Lifecycle(event)) + .map_err(|error| { + failed_lifecycle = Some(error); + QueueClosed + }), + }; + drop(views); + drop(failed_lifecycle); + result.map_err(|_| PeerEventSendError { kind }) + } + + #[must_use] + pub fn is_closed(&self) -> bool { + self.queue.is_closed() + } + + /// Starts a new view-publication generation and atomically installs both + /// authoritative replacement views. + /// + /// The network manager calls this after the prior generation has drained and + /// before it publishes authoritative clear views. New clear deliveries can + /// therefore be queued ahead of `Disabled` even when stale markers remain in + /// the underlying FIFO. + pub(crate) fn publish_view_generation( + &self, + remote_library: RemoteLibraryView, + call_to_play: CallToPlayView, + ) -> Result<(), PeerEventSendError> { + if self.queue.is_closed() { + return Err(PeerEventSendError { + kind: "ViewGenerationBarrier", + }); + } + let mut views = lock_views(&self.views); + views.generation = Arc::new(()); + let generation = Arc::clone(&views.generation); + replace_view_at_barrier( + &self.queue, + &generation, + &mut views.remote_library, + remote_library, + QueuedPeerEvent::RemoteLibrary, + ) + .map_err(|_| PeerEventSendError { + kind: "RemoteLibraryView", + })?; + replace_view_at_barrier( + &self.queue, + &generation, + &mut views.call_to_play, + call_to_play, + QueuedPeerEvent::CallToPlay, + ) + .map_err(|_| PeerEventSendError { + kind: "CallToPlayView", + }) + } +} + +fn enqueue_view( + queue: &mpsc::UnboundedSender, + generation: &Arc<()>, + slot: &mut ViewSlot, + snapshot: T, + wrap: impl FnOnce(ViewWake) -> QueuedPeerEvent, +) -> Result<(), QueueClosed> { + let versioned = VersionedView { + generation: Arc::clone(generation), + snapshot, + }; + if slot.marker_queued { + slot.pending_latest = Some(versioned); + return Ok(()); + } + + slot.marker_queued = true; + slot.queued_delivery = Some(versioned); + slot.pending_latest = None; + let queued = wrap(ViewWake { + queue: queue.clone(), + }); + if queue.send(queued).is_ok() { + Ok(()) + } else { + slot.marker_queued = false; + slot.queued_delivery = None; + Err(QueueClosed) + } +} + +fn replace_view_at_barrier( + queue: &mpsc::UnboundedSender, + generation: &Arc<()>, + slot: &mut ViewSlot, + snapshot: T, + wrap: impl FnOnce(ViewWake) -> QueuedPeerEvent, +) -> Result<(), QueueClosed> { + slot.queued_delivery = Some(VersionedView { + generation: Arc::clone(generation), + snapshot, + }); + slot.pending_latest = None; + if slot.marker_queued { + return Ok(()); + } + + slot.marker_queued = true; + let queued = wrap(ViewWake { + queue: queue.clone(), + }); + if queue.send(queued).is_ok() { + Ok(()) + } else { + slot.marker_queued = false; + slot.queued_delivery = None; + Err(QueueClosed) + } +} + +impl PeerEventReceiver { + pub async fn recv(&mut self) -> Option { + while let Some(queued) = self.queue.recv().await { + if let Some(event) = resolve_queued_event(queued, &self.views) { + return Some(event); + } + } + None + } + + pub fn try_recv(&mut self) -> Result { + loop { + let queued = self.queue.try_recv()?; + if let Some(event) = resolve_queued_event(queued, &self.views) { + return Ok(event); + } + } + } + + /// Consumes exactly one item from the underlying FIFO. A stale view marker + /// resolves to `None`; lifecycle fences use this to drain an exact captured + /// queue prefix without consuming later traffic. + pub fn try_recv_queued(&mut self) -> Result, TryRecvError> { + self.queue + .try_recv() + .map(|queued| resolve_queued_event(queued, &self.views)) + } + + #[must_use] + pub fn len(&self) -> usize { + self.queue.len() + } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + /// Drains every event and coalesced snapshot admitted before this call + /// acquired the shared transport lock. Producers cannot replace pending + /// views until the captured prefix and both pending slots have been taken. + pub fn drain_ready(&mut self) -> Result, TryRecvError> { + let mut views = lock_views(&self.views); + let generation = Arc::clone(&views.generation); + let mut events = Vec::with_capacity(self.queue.len() + 2); + loop { + let queued = match self.queue.try_recv() { + Ok(queued) => queued, + Err(TryRecvError::Empty) => break, + Err(error) => { + drop(views); + return Err(error); + } + }; + match queued { + QueuedPeerEvent::Lifecycle(event) => events.push(event), + QueuedPeerEvent::RemoteLibrary(wake) => { + if let Some(event) = resolve_view_locked( + &wake, + &generation, + &mut views.remote_library, + QueuedPeerEvent::RemoteLibrary, + PeerEvent::RemoteLibraryView, + ) { + events.push(event); + } + } + QueuedPeerEvent::CallToPlay(wake) => { + if let Some(event) = resolve_view_locked( + &wake, + &generation, + &mut views.call_to_play, + QueuedPeerEvent::CallToPlay, + PeerEvent::CallToPlayView, + ) { + events.push(event); + } + } + } + } + Ok(events) + } +} + +fn resolve_queued_event( + queued: QueuedPeerEvent, + views: &Mutex, +) -> Option { + match queued { + QueuedPeerEvent::Lifecycle(event) => Some(event), + QueuedPeerEvent::RemoteLibrary(wake) => resolve_view( + &wake, + views, + |views| &mut views.remote_library, + QueuedPeerEvent::RemoteLibrary, + PeerEvent::RemoteLibraryView, + ), + QueuedPeerEvent::CallToPlay(wake) => resolve_view( + &wake, + views, + |views| &mut views.call_to_play, + QueuedPeerEvent::CallToPlay, + PeerEvent::CallToPlayView, + ), + } +} + +fn resolve_view( + wake: &ViewWake, + views: &Mutex, + slot: impl FnOnce(&mut CoalescedViews) -> &mut ViewSlot, + wrap_queue: impl FnOnce(ViewWake) -> QueuedPeerEvent, + wrap_event: impl FnOnce(T) -> PeerEvent, +) -> Option { + let mut views = lock_views(views); + let current_generation = Arc::clone(&views.generation); + let slot = slot(&mut views); + resolve_view_locked(wake, ¤t_generation, slot, wrap_queue, wrap_event) +} + +fn resolve_view_locked( + wake: &ViewWake, + current_generation: &Arc<()>, + slot: &mut ViewSlot, + wrap_queue: impl FnOnce(ViewWake) -> QueuedPeerEvent, + wrap_event: impl FnOnce(T) -> PeerEvent, +) -> Option { + let delivery = slot.queued_delivery.take(); + if let Some(pending) = slot.pending_latest.take() { + slot.queued_delivery = Some(pending); + let next = wrap_queue(ViewWake { + queue: wake.queue.clone(), + }); + if wake.queue.send(next).is_err() { + slot.marker_queued = false; + slot.queued_delivery = None; + } + } else { + slot.marker_queued = false; + } + delivery + .filter(|delivery| Arc::ptr_eq(&delivery.generation, current_generation)) + .map(|delivery| wrap_event(delivery.snapshot)) +} + +fn lock_views(views: &Mutex) -> MutexGuard<'_, CoalescedViews> { + views + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +pub fn send(tx_notify_ui: &PeerEventSender, event: PeerEvent) { if let Err(err) = tx_notify_ui.send(event) { - let kind: &'static str = (&err.0).into(); - log::error!("Failed to send {kind} event: channel closed"); + log::error!("Failed to send {} event: channel closed", err.kind()); } } @@ -40,7 +466,7 @@ pub(crate) fn active_operation_snapshot_from_map( } pub(crate) fn send_active_operations_snapshot( - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, active_operations: &HashMap, ) { send( @@ -53,7 +479,7 @@ pub(crate) fn send_active_operations_snapshot( pub(crate) async fn emit_active_operations( active_operations: &Arc>>, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) { let active_operations = active_operations.read().await; send_active_operations_snapshot(tx_notify_ui, &active_operations); @@ -71,16 +497,13 @@ fn active_operation_kind(operation: OperationKind) -> ActiveOperationKind { pub async fn emit_peer_game_list( peer_game_db: &Arc>, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) { let db = peer_game_db.read().await; send_remote_library_view_locked(&db, tx_notify_ui); } -fn send_remote_library_view_locked( - peer_game_db: &PeerGameDB, - tx_notify_ui: &UnboundedSender, -) { +fn send_remote_library_view_locked(peer_game_db: &PeerGameDB, tx_notify_ui: &PeerEventSender) { send( tx_notify_ui, PeerEvent::RemoteLibraryView(remote_library_view(peer_game_db)), @@ -113,7 +536,7 @@ pub(crate) fn remote_library_view(peer_game_db: &PeerGameDB) -> RemoteLibraryVie pub async fn emit_peer_count( peer_game_db: &Arc>, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) { let db = peer_game_db.read().await; let peer_count = db.peer_endpoints().len(); @@ -124,6 +547,151 @@ pub async fn emit_peer_count( mod tests { use super::*; + fn remote_view(game_id: &str) -> RemoteLibraryView { + RemoteLibraryView { + games: vec![RemoteGameAvailability { + game_id: game_id.to_owned(), + content_id: ContentId::from_bytes( + [u8::try_from(game_id.len()).expect("test game ID length should fit u8"); 32], + ), + peer_count: 1, + }], + } + } + + fn call_to_play_view(author_name: &str, seed: u8) -> CallToPlayView { + let author_id = crate::PeerId::from_bytes([seed; 32]); + CallToPlayView { + events: vec![crate::CallToPlayViewEvent { + id: crate::EventNonce::from_bytes([seed; 16]), + call_id: crate::CallId::new( + author_id, + lanspread_proto::CallNonce::from_bytes([seed; 16]), + ), + author_id, + author_name: author_name.to_owned(), + at: i64::from(seed), + action: crate::CallToPlayAction::Rsvp, + }], + } + } + + #[tokio::test] + async fn complete_views_keep_first_queued_and_only_latest_pending() { + let (tx, mut rx) = peer_event_channel(); + tx.send(PeerEvent::RemoteLibraryView(remote_view("first"))) + .expect("first remote view should enqueue"); + tx.send(PeerEvent::RemoteLibraryView(remote_view("middle"))) + .expect("middle remote view should coalesce"); + tx.send(PeerEvent::RemoteLibraryView(remote_view("latest"))) + .expect("latest remote view should replace pending"); + tx.send(PeerEvent::PeerCountUpdated(1)) + .expect("first lifecycle event should enqueue"); + tx.send(PeerEvent::PeerCountUpdated(2)) + .expect("second lifecycle event should enqueue"); + tx.send(PeerEvent::CallToPlayView(call_to_play_view("first", 1))) + .expect("first Call-to-Play view should enqueue"); + tx.send(PeerEvent::CallToPlayView(call_to_play_view("latest", 2))) + .expect("latest Call-to-Play view should become pending"); + + assert_eq!(rx.len(), 4, "pending snapshots must not occupy FIFO slots"); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::RemoteLibraryView(view)) if view.games[0].game_id == "first" + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::PeerCountUpdated(1)) + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::PeerCountUpdated(2)) + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::CallToPlayView(view)) if view.events[0].author_name == "first" + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::RemoteLibraryView(view)) if view.games[0].game_id == "latest" + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::CallToPlayView(view)) if view.events[0].author_name == "latest" + )); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + #[test] + fn ready_fence_drains_queued_and_pending_views_without_later_traffic() { + let (tx, mut rx) = peer_event_channel(); + tx.send(PeerEvent::RemoteLibraryView(remote_view("queued"))) + .expect("queued view should enqueue"); + tx.send(PeerEvent::PeerCountUpdated(7)) + .expect("lifecycle event should enqueue"); + tx.send(PeerEvent::RemoteLibraryView(remote_view("pending"))) + .expect("pending view should coalesce"); + + let drained = rx.drain_ready().expect("ready prefix should drain"); + assert_eq!(drained.len(), 3); + assert!(matches!( + &drained[0], + PeerEvent::RemoteLibraryView(view) if view.games[0].game_id == "queued" + )); + assert!(matches!(&drained[1], PeerEvent::PeerCountUpdated(7))); + assert!(matches!( + &drained[2], + PeerEvent::RemoteLibraryView(view) if view.games[0].game_id == "pending" + )); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + + tx.send(PeerEvent::PeerCountUpdated(8)) + .expect("later lifecycle traffic should enqueue independently"); + assert!(matches!(rx.try_recv(), Ok(PeerEvent::PeerCountUpdated(8)))); + } + + #[tokio::test] + async fn generation_barrier_discards_old_views_before_clear_and_disabled() { + let (tx, mut rx) = peer_event_channel(); + tx.send(PeerEvent::RemoteLibraryView(remote_view("old-first"))) + .expect("old remote view should enqueue"); + tx.send(PeerEvent::RemoteLibraryView(remote_view("old-pending"))) + .expect("old remote pending view should coalesce"); + tx.send(PeerEvent::CallToPlayView(call_to_play_view("old-first", 1))) + .expect("old Call-to-Play view should enqueue"); + tx.send(PeerEvent::CallToPlayView(call_to_play_view( + "old-pending", + 2, + ))) + .expect("old Call-to-Play pending view should coalesce"); + + tx.publish_view_generation(RemoteLibraryView::default(), CallToPlayView::default()) + .expect("clear views should enqueue in the new generation"); + tx.publish_view_generation(RemoteLibraryView::default(), CallToPlayView::default()) + .expect("repeated barriers should replace in place"); + assert_eq!(rx.len(), 2, "barriers must not accumulate view markers"); + tx.send(PeerEvent::LocalNetworkSharingStateChanged( + crate::LocalNetworkSharingState::Disabled, + )) + .expect("Disabled should remain a lifecycle event"); + + assert!(matches!( + rx.recv().await, + Some(PeerEvent::RemoteLibraryView(view)) if view.games.is_empty() + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::CallToPlayView(view)) if view.events.is_empty() + )); + assert!(matches!( + rx.recv().await, + Some(PeerEvent::LocalNetworkSharingStateChanged( + crate::LocalNetworkSharingState::Disabled + )) + )); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + #[tokio::test] async fn remote_library_view_is_enqueued_before_a_waiting_writer_can_commit() { let peer_game_db = Arc::new(RwLock::new(PeerGameDB::new())); @@ -135,7 +703,7 @@ mod tests { tokio::task::yield_now().await; assert!(!writer.is_finished(), "writer must wait for the read guard"); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = peer_event_channel(); send_remote_library_view_locked(&guard, &tx); assert!(matches!( rx.try_recv(), diff --git a/crates/lanspread-peer/src/handlers.rs b/crates/lanspread-peer/src/handlers.rs index 365235e..5638662 100644 --- a/crates/lanspread-peer/src/handlers.rs +++ b/crates/lanspread-peer/src/handlers.rs @@ -14,7 +14,7 @@ use lanspread_db::{ db::GameDB, }; use lanspread_proto::PeerEndpoint; -use tokio::sync::mpsc::UnboundedSender; +use crate::PeerEventSender; use tokio_util::sync::CancellationToken; #[cfg(test)] @@ -82,7 +82,7 @@ where async fn register_download_attempt( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, key: DownloadAttemptKey, cancellation: CancellationToken, ) -> DownloadAttemptStatus { @@ -202,7 +202,7 @@ impl fmt::Display for StreamDownloadError { } /// Handles the `ListGames` command. -pub async fn handle_list_games_command(ctx: &Ctx, tx_notify_ui: &UnboundedSender) { +pub async fn handle_list_games_command(ctx: &Ctx, tx_notify_ui: &PeerEventSender) { log::info!("ListGames command received"); events::emit_peer_game_list(&ctx.peer_game_db, tx_notify_ui).await; } @@ -211,7 +211,7 @@ pub async fn handle_list_games_command(ctx: &Ctx, tx_notify_ui: &UnboundedSender #[allow(clippy::too_many_lines)] pub async fn handle_download_game_files_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: String, install_after_download: bool, ) { @@ -494,7 +494,7 @@ pub async fn handle_download_game_files_command( fn finish_cached_download( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, attempt: &DownloadAttemptKey, install_after_download: bool, target: OperationTarget, @@ -547,7 +547,7 @@ async fn settle_download_completion( /// Handles the `InstallGame` command. pub async fn handle_install_game_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: String, ) { let games_folder = ctx.game_dir.read().await.clone(); @@ -575,7 +575,7 @@ async fn stream_install_target_is_ready(ctx: &Ctx, target: &OperationTarget) -> async fn begin_stream_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, attempt: &DownloadAttemptKey, ) -> bool { @@ -631,7 +631,7 @@ async fn begin_stream_install_operation( pub async fn handle_stream_install_game_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: String, settings: StreamInstallSettings, ) { @@ -742,7 +742,7 @@ pub async fn handle_stream_install_game_command( struct StreamInstallOperation { ctx: Ctx, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, target: OperationTarget, manifest: Arc, settings: StreamInstallSettings, @@ -768,7 +768,7 @@ async fn stream_install_sources( async fn select_stream_install_sources_or_finish( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, manifest: &CatalogContentManifest, download_guard: OperationGuard, @@ -798,7 +798,7 @@ async fn select_stream_install_sources_or_finish( /// Handles the `UninstallGame` command. pub async fn handle_uninstall_game_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: String, ) { let games_folder = ctx.game_dir.read().await.clone(); @@ -812,7 +812,7 @@ pub async fn handle_uninstall_game_command( pub async fn handle_remove_downloaded_game_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: String, ) { let games_folder = ctx.game_dir.read().await.clone(); @@ -826,7 +826,7 @@ pub async fn handle_remove_downloaded_game_command( pub async fn handle_cancel_download_command( ctx: &Ctx, - _tx_notify_ui: &UnboundedSender, + _tx_notify_ui: &PeerEventSender, id: String, ) { let signal = ctx.active_downloads.read().await.get(&id).cloned(); @@ -945,7 +945,7 @@ async fn run_stream_install_operation(operation: StreamInstallOperation) { async fn finish_stream_receive_error( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, download_guard: OperationGuard, download_attempt: &DownloadAttemptStatus, @@ -975,7 +975,7 @@ async fn finish_stream_receive_error( struct StreamInstallPromotionPreparation<'a> { ctx: &'a Ctx, - tx_notify_ui: &'a UnboundedSender, + tx_notify_ui: &'a PeerEventSender, target: &'a OperationTarget, transaction: install::StreamedInstallTransaction, settings: &'a StreamInstallSettings, @@ -1055,7 +1055,7 @@ async fn prepare_streamed_install_for_promotion( struct StreamInstallReceiveRequest<'a> { ctx: &'a Ctx, - tx_notify_ui: &'a UnboundedSender, + tx_notify_ui: &'a PeerEventSender, target: &'a OperationTarget, manifest: &'a Arc, sources: &'a [PeerEndpoint], @@ -1218,7 +1218,7 @@ async fn receive_streamed_install_from_peers( async fn finish_failed_stream_download( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, guard: OperationGuard, status: &DownloadAttemptStatus, @@ -1243,7 +1243,7 @@ async fn finish_failed_stream_download( struct StreamInstallCommit { ctx: Ctx, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, target: OperationTarget, transaction: install::StreamedInstallTransaction, manifest: Arc, @@ -1346,7 +1346,7 @@ fn promote_streamed_install( fn spawn_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: OperationTarget, ) { let ctx = ctx.clone(); @@ -1358,7 +1358,7 @@ fn spawn_install_operation( async fn run_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: OperationTarget, ) { let id = target.game_id().to_owned(); @@ -1442,7 +1442,7 @@ struct PreparedInstallOperation { async fn prepare_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, ) -> Option { let id = target.game_id(); @@ -1499,7 +1499,7 @@ async fn prepare_install_operation( async fn revalidate_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, expected_kind: OperationKind, ) -> Option { @@ -1522,7 +1522,7 @@ async fn revalidate_install_operation( async fn run_started_install_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: OperationTarget, prepared: PreparedInstallOperation, operation_guard: OperationGuard, @@ -1599,7 +1599,7 @@ async fn run_started_install_operation( async fn run_uninstall_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: OperationTarget, ) { let id = target.game_id().to_owned(); @@ -1696,7 +1696,7 @@ async fn run_uninstall_operation( async fn run_remove_downloaded_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: OperationTarget, ) { let id = target.game_id().to_owned(); @@ -1804,7 +1804,7 @@ enum BeginOperationResult { async fn begin_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, operation: OperationKind, ) -> BeginOperationResult { @@ -1820,7 +1820,7 @@ async fn begin_operation( async fn begin_operation_with_drain_timeout( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, operation: OperationKind, drain_timeout: Duration, @@ -2009,7 +2009,7 @@ async fn cancel_and_wait_for_outbound_transfers( async fn transition_download_to_install( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, id: &str, operation: OperationKind, ) -> bool { @@ -2042,7 +2042,7 @@ async fn transition_download_to_install( transitioned } -async fn end_operation(ctx: &Ctx, tx_notify_ui: &UnboundedSender, id: &str) { +async fn end_operation(ctx: &Ctx, tx_notify_ui: &PeerEventSender, id: &str) { if ctx.active_operations.write().await.remove(id).is_some() { events::emit_active_operations(&ctx.active_operations, tx_notify_ui).await; } @@ -2059,7 +2059,7 @@ async fn clear_active_download(ctx: &Ctx, attempt: &DownloadAttemptKey) { } fn send_download_failed( - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, attempt: &DownloadAttemptKey, reason: DownloadFailureReason, ) { @@ -2074,7 +2074,7 @@ fn send_download_failed( async fn end_download_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, attempt: &DownloadAttemptKey, ) { clear_active_download(ctx, attempt).await; @@ -2083,7 +2083,7 @@ async fn end_download_operation( async fn settle_target_state( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, ) -> eyre::Result<()> { install::recover_game_root( @@ -2097,7 +2097,7 @@ async fn settle_target_state( async fn settle_and_end_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, guard: OperationGuard, label: &str, @@ -2121,7 +2121,7 @@ async fn settle_and_end_operation( async fn settle_and_end_download( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, guard: OperationGuard, attempt: &DownloadAttemptKey, @@ -2142,7 +2142,7 @@ async fn settle_and_end_download( async fn finish_successful_download_after_refresh( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, guard: OperationGuard, status: &DownloadAttemptStatus, @@ -2164,7 +2164,7 @@ async fn finish_successful_download_after_refresh( async fn finish_failed_download_after_refresh( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, guard: OperationGuard, status: &DownloadAttemptStatus, @@ -2193,7 +2193,7 @@ fn catalog_contains(ctx: &Ctx, id: &str) -> bool { async fn begin_local_recovery( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, game_dir: &Path, force_empty_snapshot: bool, ) -> eyre::Result<()> { @@ -2225,7 +2225,7 @@ async fn begin_local_recovery( /// quarantined. `Err` is returned only before changing the configured root. pub async fn handle_set_game_dir_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, game_dir: PathBuf, ) -> Result { handle_set_game_dir_command_with_drain_timeout( @@ -2239,7 +2239,7 @@ pub async fn handle_set_game_dir_command( async fn handle_set_game_dir_command_with_drain_timeout( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, requested_game_dir: PathBuf, drain_timeout: Duration, ) -> Result { @@ -2344,14 +2344,14 @@ async fn handle_set_game_dir_command_with_drain_timeout( /// Loads the configured local library and announces the result. pub async fn load_local_library( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) -> eyre::Result<()> { load_local_library_with_policy(ctx, tx_notify_ui, LocalLibraryEventPolicy::OnChange).await } async fn load_local_library_with_policy( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, event_policy: LocalLibraryEventPolicy, ) -> eyre::Result<()> { let game_dir = { ctx.game_dir.read().await.clone() }; @@ -2385,7 +2385,7 @@ async fn load_local_library_with_policy( async fn scan_and_announce_local_library( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, game_dir: &Path, event_policy: LocalLibraryEventPolicy, recovery_failed_ids: &HashSet, @@ -2410,7 +2410,7 @@ async fn scan_and_announce_local_library( /// active-operation snapshot, while preserving freeze behavior for other games. async fn refresh_local_game_for_ending_operation( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, target: &OperationTarget, ) -> eyre::Result<()> { let catalog = ctx.catalog.catalog(); @@ -2456,7 +2456,7 @@ async fn active_operation_ids(ctx: &Ctx) -> HashSet { } /// Handles the `GetPeerCount` command. -pub async fn handle_get_peer_count_command(ctx: &Ctx, tx_notify_ui: &UnboundedSender) { +pub async fn handle_get_peer_count_command(ctx: &Ctx, tx_notify_ui: &PeerEventSender) { log::info!("GetPeerCount command received"); events::emit_peer_count(&ctx.peer_game_db, tx_notify_ui).await; } @@ -2464,7 +2464,7 @@ pub async fn handle_get_peer_count_command(ctx: &Ctx, tx_notify_ui: &UnboundedSe /// Connects to a peer directly, bypassing mDNS discovery. pub async fn handle_connect_peer_command( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, endpoint: PeerEndpoint, ) { log::info!("Direct connect command received for {}", endpoint.addr); @@ -2509,7 +2509,7 @@ pub async fn handle_connect_peer_command( /// Updates the local game database and announces changes to peers. pub async fn update_and_announce_games( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, scan: LocalLibraryScan, ) { let _ = update_and_announce_games_with_policy( @@ -2524,7 +2524,7 @@ pub async fn update_and_announce_games( async fn update_and_announce_games_with_policy( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, scan: LocalLibraryScan, event_policy: LocalLibraryEventPolicy, ending_operation_id: Option<&str>, @@ -2659,7 +2659,7 @@ mod tests { PeerId, RuntimeSessionId, }; - use tokio::sync::{RwLock, mpsc, oneshot}; + use tokio::sync::{RwLock, oneshot}; use tokio_util::{sync::CancellationToken, task::TaskTracker}; use super::*; @@ -2834,7 +2834,7 @@ mod tests { async fn register_test_download( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) -> (DownloadAttemptStatus, CancellationToken) { let cancellation = CancellationToken::new(); let status = register_download_attempt( @@ -2868,7 +2868,7 @@ mod tests { #[test] fn cancelled_download_owner_does_not_emit_failed_event() { - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let status = DownloadAttemptStatus::new( DownloadAttemptKey::next("game".to_owned()), CancellationToken::new(), @@ -2885,7 +2885,7 @@ mod tests { #[test] fn uncancelled_download_error_emits_failed_event() { - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let attempt = DownloadAttemptKey::next("game".to_owned()); send_download_failed(&tx, &attempt, DownloadFailureReason::OperationFailed); @@ -2959,7 +2959,7 @@ mod tests { ); seed_exact_download_ownership(&ctx, games.path(), content_id).await; let peer_db = ctx.peer_game_db.write().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); tokio::time::timeout( Duration::from_secs(1), @@ -2993,7 +2993,7 @@ mod tests { NetworkControl::disabled_for_test(), ); let peer_db = ctx.peer_game_db.write().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); tokio::time::timeout( Duration::from_secs(1), @@ -3033,7 +3033,7 @@ mod tests { NetworkControl::disabled_for_test(), ); let peer_db = ctx.peer_game_db.write().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); tokio::time::timeout( Duration::from_secs(1), @@ -3070,7 +3070,7 @@ mod tests { NetworkControl::disabled_for_test(), ); let peer_db = ctx.peer_game_db.write().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); tokio::time::timeout( Duration::from_secs(1), @@ -3100,7 +3100,7 @@ mod tests { .current_publication() .expect("initial Call-to-Play projection should load") .view; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let (reply_tx, reply_rx) = oneshot::channel(); tokio::time::timeout( @@ -3136,7 +3136,7 @@ mod tests { let games = TempDir::new("lanspread-handler-enabled-call-to-play"); let ctx = test_ctx(games.path().to_path_buf()); let store = ctx.call_to_play.write().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let (reply_tx, mut reply_rx) = oneshot::channel(); let task = tokio::spawn({ let ctx = ctx.clone(); @@ -3177,7 +3177,7 @@ mod tests { async fn streamed_install_without_extracted_catalog_manifest_fails_before_admission() { let games = TempDir::new("lanspread-handler-stream-capability"); let ctx = test_ctx(games.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); handle_stream_install_game_command( &ctx, @@ -3226,7 +3226,7 @@ mod tests { ) .await ); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); handle_download_game_files_command(&ctx, &tx, "game".to_string(), false).await; @@ -3264,7 +3264,7 @@ mod tests { .write() .await .insert("game".to_owned(), OperationKind::Downloading); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); handle_download_game_files_command(&ctx, &tx, "game".to_owned(), false).await; @@ -3401,7 +3401,7 @@ mod tests { games.path().to_path_buf(), catalog_bundle([("broken", "20250101"), ("healthy", "20250101")]), ); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let error = load_local_library(&ctx, &tx) .await @@ -3445,7 +3445,7 @@ mod tests { .expect("game-root symlink should be created"); let ctx = test_ctx(games.path().to_path_buf()); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let error = load_local_library(&ctx, &tx) .await @@ -3470,7 +3470,7 @@ mod tests { async fn operation_admission_rejects_recovering_and_failed_games() { let games = TempDir::new("lanspread-handler-recovery-admission"); let ctx = test_ctx(games.path().to_path_buf()); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); ctx.recovery_quarantine.begin(games.path().to_path_buf()); let target = operation_target(games.path()); @@ -3496,7 +3496,7 @@ mod tests { let games = TempDir::new("lanspread-handler-download-refresh-failure"); write_file(&games.game_root(), b"not a directory"); let ctx = test_ctx(games.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let (status, cancel) = register_test_download(&ctx, &tx).await; ctx.active_operations .write() @@ -3519,14 +3519,14 @@ mod tests { assert_no_event(&mut rx).await; } - async fn recv_event(rx: &mut mpsc::UnboundedReceiver) -> PeerEvent { + async fn recv_event(rx: &mut crate::PeerEventReceiver) -> PeerEvent { tokio::time::timeout(Duration::from_secs(1), rx.recv()) .await .expect("event should arrive") .expect("event channel should remain open") } - async fn assert_no_event(rx: &mut mpsc::UnboundedReceiver) { + async fn assert_no_event(rx: &mut crate::PeerEventReceiver) { assert!( tokio::time::timeout(Duration::from_millis(50), rx.recv()) .await @@ -3535,7 +3535,7 @@ mod tests { ); } - fn drain_events(rx: &mut mpsc::UnboundedReceiver) -> Vec { + fn drain_events(rx: &mut crate::PeerEventReceiver) -> Vec { let mut events = Vec::new(); while let Ok(event) = rx.try_recv() { events.push(event); @@ -3682,7 +3682,7 @@ mod tests { let state = TempDir::new("lanspread-handler-stream-retry-setup-state"); std::fs::create_dir_all(games.game_root().join("local")) .expect("installed tree should be created"); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let status = DownloadAttemptStatus::new( DownloadAttemptKey::next("game".to_owned()), CancellationToken::new(), @@ -3719,7 +3719,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); // 1. Initial scan: the game is ready and announced @@ -3845,7 +3845,7 @@ mod tests { .expect("catalog construction should defer manifest body parsing"), ); let ctx = test_ctx_with_catalog(games.path().to_path_buf(), catalog); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let scan = scan_local_library(games.path(), ctx.state_dir.as_ref(), ctx.catalog.catalog()) .await .expect("ready game should scan before manifest priming"); @@ -3884,7 +3884,7 @@ mod tests { ); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let current_scan = scan_local_library(current.path(), ctx.state_dir.as_ref(), catalog) @@ -3921,7 +3921,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let older_scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), catalog) .await @@ -3973,7 +3973,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let target = operation_target(temp.path()); assert_eq!( @@ -3997,7 +3997,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), ctx.catalog.catalog()) .await .expect("initial scan should succeed"); @@ -4057,7 +4057,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let target = operation_target(temp.path()); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), ctx.catalog.catalog()) .await @@ -4122,7 +4122,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), ctx.catalog.catalog()) .await .expect("initial scan should succeed"); @@ -4195,7 +4195,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), catalog) @@ -4221,7 +4221,7 @@ mod tests { write_file(&root.join("local").join("old.txt"), b"old"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), catalog) .await @@ -4256,7 +4256,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_install_operation(&ctx, &tx, operation_target(temp.path())).await; @@ -4281,7 +4281,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); let target = operation_target(temp.path()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let prepared = prepare_install_operation(&ctx, &tx, &target) .await @@ -4334,7 +4334,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); let target = operation_target(temp.path()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); assert!(stream_install_target_is_ready(&ctx, &target).await); assert_eq!( @@ -4383,7 +4383,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (prepare_tx, _prepare_rx) = mpsc::unbounded_channel(); + let (prepare_tx, _prepare_rx) = crate::peer_event_channel(); let (download_status, _download_cancel) = register_test_download(&ctx, &prepare_tx).await; ctx.active_operations .write() @@ -4394,7 +4394,7 @@ mod tests { .await .expect("downloaded game should be installable"); let read_guard = ctx.active_operations.read().await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let install_task = tokio::spawn({ let ctx = ctx.clone(); @@ -4445,7 +4445,7 @@ mod tests { async fn cancel_download_command_only_cancels_active_token() { let temp = TempDir::new("lanspread-handler-cancel-download"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let (_status, cancel) = register_test_download(&ctx, &tx).await; ctx.active_operations .write() @@ -4472,7 +4472,7 @@ mod tests { write_file(&root.join("local").join("old.txt"), b"old"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_install_operation(&ctx, &tx, operation_target(temp.path())).await; @@ -4497,7 +4497,7 @@ mod tests { write_file(&root.join("game.eti"), b"old archive"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_install_operation(&ctx, &tx, operation_target(temp.path())).await; assert_active_update( @@ -4560,7 +4560,7 @@ mod tests { write_file(&root.join("local").join("old.txt"), b"old"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_uninstall_operation(&ctx, &tx, operation_target(temp.path())).await; @@ -4592,7 +4592,7 @@ mod tests { &["game.eti"], ) .await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let scan = scan_local_library(temp.path(), ctx.state_dir.as_ref(), catalog) .await @@ -4630,7 +4630,7 @@ mod tests { write_file(&root.join("game.eti"), b"archive"); } let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); // On the current-thread test runtime, this ready lock acquisition does // not yield to the newly spawned command. Move the configured root @@ -4671,7 +4671,7 @@ mod tests { write_file(&root.join("local/payload.txt"), b"installed"); } let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); handle_uninstall_game_command(&ctx, &tx, "game".to_string()).await; *ctx.game_dir.write().await = next.path().to_path_buf(); @@ -4715,7 +4715,7 @@ mod tests { &["game.eti"], ) .await; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); handle_remove_downloaded_game_command(&ctx, &tx, "game".to_string()).await; *ctx.game_dir.write().await = next.path().to_path_buf(); @@ -4757,7 +4757,7 @@ mod tests { .write() .await .insert("game".to_string(), OperationKind::Downloading); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let error = handle_set_game_dir_command(&ctx, &tx, next.path().to_path_buf()) .await @@ -4778,7 +4778,7 @@ mod tests { symlink(current.path(), &alias).expect("same-root alias should be created"); let current = std::fs::canonicalize(current.path()).expect("root should canonicalize"); let ctx = test_ctx(current.clone()); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let accepted = handle_set_game_dir_command(&ctx, &tx, alias) .await @@ -4801,7 +4801,7 @@ mod tests { let current = std::fs::canonicalize(current.path()).expect("root should canonicalize"); let next = std::fs::canonicalize(next.path()).expect("new root should canonicalize"); let ctx = test_ctx(current); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let accepted = handle_set_game_dir_command(&ctx, &tx, alias.clone()) .await @@ -4817,7 +4817,7 @@ mod tests { let current = TempDir::new("lanspread-handler-invalid-dir-current"); let missing = current.path().join("missing"); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let error = handle_set_game_dir_command(&ctx, &tx, missing) .await @@ -4837,7 +4837,7 @@ mod tests { let candidate_path = std::fs::canonicalize(candidate.path()).expect("candidate root should canonicalize"); let ctx = test_ctx(current_path.clone()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let published = summary("game", "20250101", Availability::Ready); ctx.local_library .write() @@ -4884,7 +4884,7 @@ mod tests { let temp = TempDir::new("lanspread-handler-same-dir"); write_file(&temp.game_root().join(".version.ini.tmp"), b"tmp"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let first = CancellationToken::new(); let second = CancellationToken::new(); *ctx.active_outbound_transfers.write().await = HashMap::from([ @@ -4936,7 +4936,7 @@ mod tests { let temp = TempDir::new("lanspread-handler-same-dir-timeout"); write_file(&temp.game_root().join(".version.ini.tmp"), b"tmp"); let ctx = test_ctx(temp.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let token = CancellationToken::new(); ctx.active_outbound_transfers .write() @@ -4969,7 +4969,7 @@ mod tests { .write() .await .insert("game".to_string(), OperationKind::Downloading); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let error = handle_set_game_dir_command(&ctx, &tx, temp.path().to_path_buf()) .await @@ -4985,7 +4985,7 @@ mod tests { let next = TempDir::new("lanspread-handler-new-dir"); write_file(&next.game_root().join(".version.ini.tmp"), b"tmp"); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let first = CancellationToken::new(); let second = CancellationToken::new(); *ctx.active_outbound_transfers.write().await = HashMap::from([ @@ -5041,7 +5041,7 @@ mod tests { write_file(&root.join("version.ini"), b"20250101"); write_file(&root.join("game.eti"), b"archive"); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let scan = scan_local_library(current.path(), ctx.state_dir.as_ref(), catalog) @@ -5095,7 +5095,7 @@ mod tests { let current = TempDir::new("lanspread-handler-revision-exhausted-current"); let next = TempDir::new("lanspread-handler-revision-exhausted-next"); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let published = summary("game", "20250101", Availability::Ready); ctx.local_library .write() @@ -5138,7 +5138,7 @@ mod tests { std::fs::create_dir_all(next_root.join(".version.ini.tmp")) .expect("invalid recovery scratch directory should be created"); let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); assert_eq!( handle_set_game_dir_command_with_drain_timeout( @@ -5198,7 +5198,7 @@ mod tests { } let ctx = test_ctx(current.path().to_path_buf()); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let catalog = ctx.catalog.catalog(); let scan = scan_local_library(current.path(), ctx.state_dir.as_ref(), catalog) .await diff --git a/crates/lanspread-peer/src/lib.rs b/crates/lanspread-peer/src/lib.rs index 89f868f..7bf2f5c 100644 --- a/crates/lanspread-peer/src/lib.rs +++ b/crates/lanspread-peer/src/lib.rs @@ -87,7 +87,7 @@ pub use scoped_blocking::scoped_blocking; pub use scoped_process::{ScopedProcess, ScopedProcessOutput}; use tokio::sync::{ RwLock, - mpsc::{UnboundedReceiver, UnboundedSender}, + mpsc::UnboundedReceiver, oneshot, }; use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -134,6 +134,7 @@ pub use crate::{ StreamInstallProvider, }, }; +pub use events::{PeerEventReceiver, PeerEventSender, peer_event_channel}; // ============================================================================= // Public API types @@ -563,7 +564,7 @@ impl std::fmt::Debug for PeerStartOptions { #[allow(clippy::implicit_hasher)] pub fn start_peer( game_dir: impl Into, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, peer_game_db: Arc>, unpacker: Arc, catalog: Arc, @@ -582,7 +583,7 @@ pub fn start_peer( #[allow(clippy::implicit_hasher)] pub fn start_peer_with_options( game_dir: impl Into, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, peer_game_db: Arc>, unpacker: Arc, catalog: Arc, @@ -669,7 +670,7 @@ const fn identity_durability(persistence: &PeerIdentityPersistence) -> PeerIdent #[allow(clippy::too_many_arguments, clippy::implicit_hasher)] async fn run_peer( mut rx_control: UnboundedReceiver, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, peer_game_db: Arc>, peer_identity: Arc, game_dir: PathBuf, @@ -724,7 +725,7 @@ async fn run_peer( async fn handle_peer_commands( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, rx_control: &mut UnboundedReceiver, ) -> eyre::Result<()> { loop { @@ -805,7 +806,7 @@ async fn handle_peer_commands( async fn handle_apply_call_to_play_intent( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, intent: CallToPlayLocalIntent, display_name: String, reply: oneshot::Sender>, @@ -837,7 +838,7 @@ async fn handle_apply_call_to_play_intent( async fn handle_set_call_to_play_display_name( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, display_name: String, reply: oneshot::Sender>, ) { @@ -863,7 +864,7 @@ async fn handle_set_call_to_play_display_name( async fn handle_get_call_to_play_view( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, reply: Option>>, ) { let result = { @@ -999,7 +1000,7 @@ mod configured_game_dir_tests { let state = TempDir::new("lanspread-startup-state-root"); let alias = target.path().join("."); let expected = std::fs::canonicalize(target.path()).expect("target should canonicalize"); - let (events, _event_rx) = mpsc::unbounded_channel(); + let (events, _event_rx) = crate::peer_event_channel(); let mut handle = start_peer_with_options( alias, @@ -1044,7 +1045,7 @@ mod configured_game_dir_tests { let identity = Arc::new(loaded.identity); let expected_peer_id = identity.peer_id(); let (tx_control, rx_control) = mpsc::unbounded_channel(); - let (tx_events, mut rx_events) = mpsc::unbounded_channel(); + let (tx_events, mut rx_events) = crate::peer_event_channel(); let mut handle = crate::startup::spawn_peer_runtime( tx_control, @@ -1128,7 +1129,7 @@ mod configured_game_dir_tests { ) .expect("old-root intent should be valid"); write_intent(state.path(), "orphan", &intent).expect("old-root intent should be persisted"); - let (events, _event_rx) = mpsc::unbounded_channel(); + let (events, _event_rx) = crate::peer_event_channel(); let result = start_peer_with_options( candidate.path().to_path_buf(), diff --git a/crates/lanspread-peer/src/network_generation.rs b/crates/lanspread-peer/src/network_generation.rs index 0904658..a001b2c 100644 --- a/crates/lanspread-peer/src/network_generation.rs +++ b/crates/lanspread-peer/src/network_generation.rs @@ -20,6 +20,7 @@ use tokio_util::{ use crate::{ LocalNetworkSharingState, PeerEvent, + PeerEventSender, PeerRuntimeComponent, context::{Ctx, NetworkServiceCtx}, events, @@ -283,7 +284,7 @@ impl EnableFailure { struct GenerationStart { id: NetworkGenerationId, core: Ctx, - tx_notify_ui: mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, shutdown: CancellationToken, tasks: TaskTracker, failure_tx: mpsc::UnboundedSender, @@ -410,7 +411,7 @@ impl NetworkManager { pub(crate) async fn run( mut self, ctx: Ctx, - tx_notify_ui: mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, initially_enabled: bool, ) { let mut current = None; @@ -445,7 +446,7 @@ impl NetworkManager { async fn run_body( &mut self, ctx: &Ctx, - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, initially_enabled: bool, current: &mut Option, ) { @@ -495,7 +496,7 @@ impl NetworkManager { async fn enable( &mut self, ctx: &Ctx, - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, current: &mut Option, ) -> Result { if current.is_some() { @@ -585,7 +586,7 @@ impl NetworkManager { async fn start_generation( &self, ctx: &Ctx, - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, id: NetworkGenerationId, shutdown: &CancellationToken, tasks: &TaskTracker, @@ -673,7 +674,7 @@ impl NetworkManager { async fn disable( &self, ctx: &Ctx, - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, current: &mut Option, ) { let Some(generation) = current.as_mut() else { @@ -693,7 +694,7 @@ impl NetworkManager { async fn stop_generation( &self, ctx: &Ctx, - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, generation: &mut NetworkGeneration, ) { self.begin_stop_generation(generation); @@ -800,7 +801,7 @@ async fn clear_local_peer_addr(ctx: &Ctx) { *ctx.local_peer_addr.write().await = None; } -fn send_state(tx_notify_ui: &mpsc::UnboundedSender, state: LocalNetworkSharingState) { +fn send_state(tx_notify_ui: &PeerEventSender, state: LocalNetworkSharingState) { events::send( tx_notify_ui, PeerEvent::LocalNetworkSharingStateChanged(state), @@ -808,7 +809,7 @@ fn send_state(tx_notify_ui: &mpsc::UnboundedSender, state: LocalNetwo } fn report_failure( - tx_notify_ui: &mpsc::UnboundedSender, + tx_notify_ui: &PeerEventSender, component: PeerRuntimeComponent, error: String, ) { @@ -991,7 +992,7 @@ mod tests { } async fn wait_for_state( - events: &mut mpsc::UnboundedReceiver, + events: &mut crate::PeerEventReceiver, expected: LocalNetworkSharingState, ) { loop { @@ -1055,7 +1056,7 @@ mod tests { async fn startup_disabled_does_not_construct_a_network_generation() { let harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1071,7 +1072,7 @@ mod tests { async fn enable_reply_waits_for_readiness_and_atomic_admission() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1105,7 +1106,7 @@ mod tests { async fn disable_reply_waits_for_every_admitted_permit() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1188,7 +1189,7 @@ mod tests { async fn readiness_failure_drains_partial_generation_and_allows_reenable() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1223,7 +1224,7 @@ mod tests { async fn disable_interrupts_initial_enable_without_a_failure_diagnostic() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, true)); let pending_ready = harness .started @@ -1273,7 +1274,7 @@ mod tests { async fn current_required_failure_auto_disables_and_closes_admission() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; let (id, failures) = enable_fake_generation(&harness.control, &mut harness.started).await; @@ -1297,7 +1298,7 @@ mod tests { async fn root_shutdown_waits_for_held_permit_and_joins_runtime() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; enable_fake_generation(&harness.control, &mut harness.started).await; @@ -1321,7 +1322,7 @@ mod tests { async fn dropping_enable_reply_does_not_cancel_the_transition() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1362,7 +1363,7 @@ mod tests { async fn idempotent_stable_requests_do_not_restart_or_republish() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1388,7 +1389,7 @@ mod tests { async fn client_cleanup_error_is_diagnostic_but_off_ack_is_authoritative() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; enable_fake_generation(&harness.control, &mut harness.started).await; @@ -1423,7 +1424,7 @@ mod tests { let harness = test_harness(); harness.start_panics.store(true, Ordering::SeqCst); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1448,7 +1449,7 @@ mod tests { async fn stale_generation_failure_cannot_close_current_admission() { let mut harness = test_harness(); let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; @@ -1513,7 +1514,7 @@ mod tests { let mut harness = test_harness(); harness.manager.last_generation = u64::MAX; let (_temp, ctx) = test_ctx(harness.control.clone()); - let (event_tx, mut events) = mpsc::unbounded_channel(); + let (event_tx, mut events) = crate::peer_event_channel(); let manager_task = tokio::spawn(harness.manager.run(ctx.clone(), event_tx, false)); wait_for_state(&mut events, LocalNetworkSharingState::Disabled).await; diff --git a/crates/lanspread-peer/src/services/discovery.rs b/crates/lanspread-peer/src/services/discovery.rs index e0ef382..76bb6da 100644 --- a/crates/lanspread-peer/src/services/discovery.rs +++ b/crates/lanspread-peer/src/services/discovery.rs @@ -12,13 +12,14 @@ use futures::{StreamExt as _, stream::FuturesUnordered}; use lanspread_mdns::{LANSPREAD_SERVICE_TYPE, MdnsBrowser, MdnsService, MdnsServicePoll}; use lanspread_proto::{PROTOCOL_VERSION, PeerEndpoint, PeerId}; use tokio::sync::{ - mpsc::{self, UnboundedSender}, + mpsc, oneshot, }; use tokio_util::sync::CancellationToken; use crate::{ PeerEvent, + PeerEventSender, context::NetworkServiceCtx, events, services::{ @@ -188,7 +189,7 @@ impl Drop for DiscoveryWorker { /// Runs the peer discovery service using mDNS. #[allow(clippy::too_many_lines)] pub async fn run_peer_discovery( - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, ctx: NetworkServiceCtx, ) -> eyre::Result<()> { log::info!("Starting peer discovery task"); diff --git a/crates/lanspread-peer/src/services/liveness.rs b/crates/lanspread-peer/src/services/liveness.rs index da8118f..e1b58ca 100644 --- a/crates/lanspread-peer/src/services/liveness.rs +++ b/crates/lanspread-peer/src/services/liveness.rs @@ -3,10 +3,8 @@ use std::{sync::Arc, time::Duration}; use futures::{StreamExt as _, stream}; -use tokio::sync::mpsc::UnboundedSender; - use crate::{ - PeerEvent, + PeerEventSender, config::{PEER_PING_IDLE_SECS, PEER_PING_INTERVAL_SECS, peer_stale_timeout}, content_quarantine::ContentQuarantine, context::{NetworkServiceCtx, OperationKind}, @@ -22,7 +20,7 @@ const MAX_CONCURRENT_PINGS: usize = 8; /// uses `last_revision_check`; inbound and content traffic only affect /// `last_seen`, which remains the stale-pruning clock. pub async fn run_ping_service( - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, ctx: NetworkServiceCtx, ) -> eyre::Result<()> { log::info!( diff --git a/crates/lanspread-peer/src/services/local_monitor.rs b/crates/lanspread-peer/src/services/local_monitor.rs index 4418e28..e4e0c53 100644 --- a/crates/lanspread-peer/src/services/local_monitor.rs +++ b/crates/lanspread-peer/src/services/local_monitor.rs @@ -14,13 +14,13 @@ use std::{ use futures::FutureExt; use tokio::{ - sync::{RwLock, mpsc::UnboundedSender}, + sync::RwLock, task::{JoinError, JoinSet}, time::{Instant, MissedTickBehavior}, }; use crate::{ - PeerEvent, + PeerEventSender, config::{LOCAL_GAME_FALLBACK_SCAN_SECS, LOCAL_GAME_POLL_INTERVAL_SECS}, context::Ctx, game_paths::{is_download_protected_root_name, is_ignored_games_root_name}, @@ -67,10 +67,7 @@ struct RescanGate { } /// Monitors the local game directory for changes. -pub async fn run_local_game_monitor( - tx_notify_ui: UnboundedSender, - ctx: Ctx, -) -> eyre::Result<()> { +pub async fn run_local_game_monitor(tx_notify_ui: PeerEventSender, ctx: Ctx) -> eyre::Result<()> { log::info!("Starting polling-based local game directory monitor"); let mut snapshot = initial_poll_snapshot(&ctx).await; @@ -171,7 +168,7 @@ async fn initial_poll_snapshot(ctx: &Ctx) -> Option { async fn poll_local_game_changes( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, gate: &RescanGate, rescans: &mut JoinSet<()>, previous: &mut Option, @@ -328,7 +325,7 @@ fn changed_game_ids(previous: &PollSnapshot, current: &PollSnapshot) -> BTreeSet async fn queue_changed_games( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, gate: &RescanGate, rescans: &mut JoinSet<()>, changed_ids: BTreeSet, @@ -354,7 +351,7 @@ async fn queue_changed_games( async fn queue_rescan( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, gate: &RescanGate, rescans: &mut JoinSet<()>, id: String, @@ -376,12 +373,7 @@ async fn queue_rescan( }); } -async fn run_gated_rescan( - ctx: Ctx, - tx_notify_ui: UnboundedSender, - gate: RescanGate, - id: String, -) { +async fn run_gated_rescan(ctx: Ctx, tx_notify_ui: PeerEventSender, gate: RescanGate, id: String) { loop { gate.pending.write().await.remove(&id); @@ -418,7 +410,7 @@ async fn run_gated_rescan( gate.running.write().await.remove(&id); } -async fn run_fallback_scan(ctx: &Ctx, tx_notify_ui: &UnboundedSender) { +async fn run_fallback_scan(ctx: &Ctx, tx_notify_ui: &PeerEventSender) { let _admission = ctx.operation_admission.lock().await; let game_dir = ctx.game_dir.read().await.clone(); let catalog = ctx.catalog.catalog(); @@ -445,11 +437,12 @@ mod tests { }; use lanspread_db::content_manifest::CatalogBundle; - use tokio::sync::{RwLock, mpsc}; + use tokio::sync::RwLock; use tokio_util::{sync::CancellationToken, task::TaskTracker}; use super::*; use crate::{ + PeerEvent, UnpackFuture, Unpacker, context::OperationKind, @@ -512,9 +505,7 @@ mod tests { panic!("injected monitor loop panic"); } - async fn recv_local_update( - rx: &mut mpsc::UnboundedReceiver, - ) -> Vec { + async fn recv_local_update(rx: &mut crate::PeerEventReceiver) -> Vec { let event = tokio::time::timeout(Duration::from_secs(1), rx.recv()) .await .expect("local update event should arrive") @@ -604,7 +595,7 @@ mod tests { .insert("game".to_string(), OperationKind::Downloading); let rescan_gate = RescanGate::default(); let mut rescans = JoinSet::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); queue_changed_games( &ctx, @@ -635,7 +626,7 @@ mod tests { ); let rescan_gate = RescanGate::default(); let mut rescans = JoinSet::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let mut state = Some( capture_poll_snapshot(&ctx) .await @@ -669,7 +660,7 @@ mod tests { ); let gate = RescanGate::default(); let mut rescans = JoinSet::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let library_guard = ctx.local_library.write().await; queue_rescan(&ctx, &tx, &gate, &mut rescans, "game".to_string()).await; @@ -707,7 +698,7 @@ mod tests { ); let gate = RescanGate::default(); let mut rescans = JoinSet::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let admission = ctx.operation_admission.lock().await; queue_rescan(&ctx, &tx, &gate, &mut rescans, "game".to_string()).await; @@ -751,7 +742,7 @@ mod tests { ); let gate = RescanGate::default(); let mut rescans = JoinSet::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); let admission = ctx.operation_admission.lock().await; queue_rescan(&ctx, &tx, &gate, &mut rescans, "game".to_string()).await; @@ -792,7 +783,7 @@ mod tests { let temp = TempDir::new("lanspread-local-monitor-structured-shutdown"); let ctx = test_ctx(temp.path().to_path_buf(), empty_catalog_bundle()); let monitor_ctx = ctx.clone(); - let (tx, _rx) = mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let monitor = tokio::spawn(run_local_game_monitor(tx, monitor_ctx)); tokio::task::yield_now().await; @@ -813,7 +804,7 @@ mod tests { temp.path().to_path_buf(), catalog_bundle([("game", "20250101")]), ); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_fallback_scan(&ctx, &tx).await; @@ -837,7 +828,7 @@ mod tests { temp.path().to_path_buf(), catalog_bundle([("game", "20250101")]), ); - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); run_fallback_scan(&ctx, &tx).await; diff --git a/crates/lanspread-peer/src/services/remote_state.rs b/crates/lanspread-peer/src/services/remote_state.rs index 539ac6b..cf84afc 100644 --- a/crates/lanspread-peer/src/services/remote_state.rs +++ b/crates/lanspread-peer/src/services/remote_state.rs @@ -10,12 +10,13 @@ use lanspread_proto::{ PeerStateSnapshot, RuntimeSessionId, }; -use tokio::sync::{RwLock, mpsc::UnboundedSender}; +use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; use crate::{ CallToPlayView, PeerEvent, + PeerEventSender, call_to_play::{ CallToPlayPublication, CallToPlayStore, @@ -45,7 +46,7 @@ use crate::{ pub(crate) struct RemoteStateCtx { local_peer_id: PeerId, peer_game_db: Arc>, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, call_to_play: Arc>, quic: QuicConnector, cancellation: CancellationToken, @@ -56,7 +57,7 @@ impl RemoteStateCtx { #[must_use] pub(crate) fn from_network( ctx: &NetworkServiceCtx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) -> Self { Self { local_peer_id: ctx.peer_id, @@ -310,7 +311,7 @@ fn log_call_to_play_outcome(peer_id: PeerId, outcome: &ObserveRemoteAuthorOutcom fn enqueue_commit_transition( db: &PeerGameDB, call_to_play: &mut CallToPlayStore, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, state_sync: &StateSyncHandle, publication: PreparedCallToPlayPublication, endpoint: PeerEndpoint, @@ -351,7 +352,7 @@ fn enqueue_commit_transition( fn enqueue_final_views( db: &PeerGameDB, call_to_play: &mut CallToPlayStore, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, state_sync: &StateSyncHandle, publication: PreparedCallToPlayPublication, ) { @@ -368,11 +369,11 @@ fn enqueue_final_views( } /// Clears every remote projection after one network generation has fully -/// drained, then queues the authoritative empty replacement views while the +/// drained, then queues the authoritative replacement views while the /// database-to-Call-to-Play lock order is still held. pub(crate) async fn clear_remote_state_and_publish( ctx: &Ctx, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) { let mut db = ctx.peer_game_db.write().await; let mut call_to_play = ctx.call_to_play.write().await; @@ -383,11 +384,14 @@ pub(crate) async fn clear_remote_state_and_publish( events::send(tx_notify_ui, PeerEvent::PeerLost(endpoint)); } events::send(tx_notify_ui, PeerEvent::PeerCountUpdated(0)); - events::send( - tx_notify_ui, - PeerEvent::RemoteLibraryView(events::remote_library_view(&db)), - ); - enqueue_call_to_play_publication(publication, tx_notify_ui, &ctx.state_sync, true); + ctx.state_sync + .publish_call_to_play_revision(publication.local_revision); + if let Err(error) = tx_notify_ui.publish_view_generation( + events::remote_library_view(&db), + CallToPlayView::from(publication.view), + ) { + log::error!("Failed to publish cleared peer view generation: {error}"); + } if let Some(error) = preparation_error { log::warn!( @@ -398,7 +402,7 @@ pub(crate) async fn clear_remote_state_and_publish( fn enqueue_local_prune_if_changed( call_to_play: &mut CallToPlayStore, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, state_sync: &StateSyncHandle, publication: PreparedCallToPlayPublication, ) { @@ -411,7 +415,7 @@ fn enqueue_local_prune_if_changed( fn enqueue_call_to_play_publication( publication: CallToPlayPublication, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, state_sync: &StateSyncHandle, force_view: bool, ) { diff --git a/crates/lanspread-peer/src/services/state_sync.rs b/crates/lanspread-peer/src/services/state_sync.rs index 1584108..976f3a4 100644 --- a/crates/lanspread-peer/src/services/state_sync.rs +++ b/crates/lanspread-peer/src/services/state_sync.rs @@ -8,7 +8,7 @@ use tokio::sync::{Mutex, mpsc, watch}; use tokio_util::sync::CancellationToken; use crate::{ - PeerEvent, + PeerEventSender, context::NetworkServiceCtx, network::{send_call_to_play_changed, send_library_changed}, peer_db::PeerRevisionSnapshot, @@ -170,7 +170,7 @@ type FanoutFuture = Pin + Send>>; /// Runs the bounded pull scheduler and local hint fanout in one lexical scope. pub(crate) async fn run_state_sync( ctx: NetworkServiceCtx, - tx_notify_ui: tokio::sync::mpsc::UnboundedSender, + tx_notify_ui: PeerEventSender, cancellation: CancellationToken, ) -> eyre::Result<()> { let mut pinned_rx = ctx.state_sync.inbox.pinned_rx.lock().await; diff --git a/crates/lanspread-peer/src/services/transfer.rs b/crates/lanspread-peer/src/services/transfer.rs index 0000080..770729d 100644 --- a/crates/lanspread-peer/src/services/transfer.rs +++ b/crates/lanspread-peer/src/services/transfer.rs @@ -504,7 +504,7 @@ mod tests { CatalogContentManifestBody, CatalogExtractedEntry, }; - use tokio::sync::{RwLock, mpsc}; + use tokio::sync::RwLock; use tokio_util::task::TaskTracker; use super::*; @@ -611,7 +611,7 @@ mod tests { root: &Path, manifest: &CatalogContentManifest, provider: Arc, - ) -> (PeerCtx, mpsc::UnboundedReceiver) { + ) -> (PeerCtx, crate::PeerEventReceiver) { let catalog = Arc::new( CatalogBundle::from_manifests([manifest.clone()]) .expect("test catalog should be complete"), @@ -643,7 +643,7 @@ mod tests { availability: Availability::Ready, }, ); - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = crate::peer_event_channel(); (ctx.to_peer_ctx(tx, CancellationToken::new()), rx) } diff --git a/crates/lanspread-peer/src/startup.rs b/crates/lanspread-peer/src/startup.rs index 6782d26..632d682 100644 --- a/crates/lanspread-peer/src/startup.rs +++ b/crates/lanspread-peer/src/startup.rs @@ -23,9 +23,9 @@ use tokio_util::{sync::CancellationToken, task::TaskTracker}; use crate::{ PeerCommand, - PeerEvent, PeerId, PeerIdentityDurability, + PeerEventSender, PeerRuntimeComponent, StreamInstallProvider, Unpacker, @@ -284,7 +284,7 @@ pub(crate) enum SupervisionPolicy { pub(crate) fn spawn_peer_runtime( tx_control: UnboundedSender, rx_control: UnboundedReceiver, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, peer_game_db: Arc>, peer_identity: Arc, identity_durability: PeerIdentityDurability, @@ -380,11 +380,11 @@ async fn run_runtime_root( } } -pub(crate) fn spawn_core_services(ctx: &Ctx, tx_notify_ui: &UnboundedSender) { +pub(crate) fn spawn_core_services(ctx: &Ctx, tx_notify_ui: &PeerEventSender) { spawn_local_library_monitor(ctx, tx_notify_ui); } -fn spawn_local_library_monitor(ctx: &Ctx, tx_notify_ui: &UnboundedSender) { +fn spawn_local_library_monitor(ctx: &Ctx, tx_notify_ui: &PeerEventSender) { let ctx = ctx.clone(); let tx_notify_ui = tx_notify_ui.clone(); let task_tracker = ctx.task_tracker.clone(); @@ -408,7 +408,7 @@ fn spawn_local_library_monitor(ctx: &Ctx, tx_notify_ui: &UnboundedSender( task_tracker: &TaskTracker, shutdown: &CancellationToken, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, component: PeerRuntimeComponent, policy: SupervisionPolicy, mut make_service: F, @@ -505,13 +505,16 @@ fn spawn_supervised_service( #[cfg(test)] fn report_required_service_failure( - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, component: PeerRuntimeComponent, error: String, shutdown: &CancellationToken, ) { log::error!("{component:?} failed: {error}"); - crate::events::send(tx_notify_ui, PeerEvent::RuntimeFailed { component, error }); + crate::events::send( + tx_notify_ui, + crate::PeerEvent::RuntimeFailed { component, error }, + ); shutdown.cancel(); } @@ -611,7 +614,7 @@ mod tests { async fn required_service_failure_cancels_runtime_and_emits_event() { let tracker = TaskTracker::new(); let shutdown = CancellationToken::new(); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = crate::peer_event_channel(); spawn_supervised_service( &tracker, @@ -646,7 +649,7 @@ mod tests { async fn restart_service_restarts_until_shutdown() { let tracker = TaskTracker::new(); let shutdown = CancellationToken::new(); - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let attempts = Arc::new(AtomicUsize::new(0)); spawn_supervised_service( diff --git a/crates/lanspread-peer/src/stream_install.rs b/crates/lanspread-peer/src/stream_install.rs index 887304f..64105b8 100644 --- a/crates/lanspread-peer/src/stream_install.rs +++ b/crates/lanspread-peer/src/stream_install.rs @@ -36,7 +36,7 @@ use s2n_quic::{ use tokio::{ io::{AsyncRead, AsyncReadExt}, process::Command, - sync::{mpsc, mpsc::UnboundedSender}, + sync::mpsc, time::{self, Instant as TokioInstant, MissedTickBehavior}, }; use tokio_util::{ @@ -47,6 +47,7 @@ use tokio_util::{ use crate::{ DownloadProgress, PeerEvent, + PeerEventSender, install::root_eti_archives, network::connect_to_peer, path_validation::validate_game_file_path, @@ -1274,7 +1275,7 @@ impl StreamInstallReceiveState { game_id: &str, peer_endpoint: PeerEndpoint, content_id: ContentId, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) -> StreamInstallReceiveResult { match frame { StreamInstallFrame::ArchiveBegin { @@ -1381,7 +1382,7 @@ pub(crate) struct ReceiveStreamedInstallRequest<'a> { pub(crate) manifest: Arc, pub(crate) staging_dir: &'a Path, pub(crate) attempt: DownloadAttemptReporter, - pub(crate) tx_notify_ui: UnboundedSender, + pub(crate) tx_notify_ui: PeerEventSender, pub(crate) quic: &'a QuicConnector, pub(crate) cancel_token: CancellationToken, } @@ -1854,7 +1855,7 @@ impl IncomingFile { game_id: &str, peer_endpoint: PeerEndpoint, content_id: ContentId, - tx_notify_ui: &UnboundedSender, + tx_notify_ui: &PeerEventSender, ) -> StreamInstallReceiveResult<()> { if &self.relative_path != relative_path { return Err(StreamInstallReceiveError::integrity(eyre::eyre!( @@ -3051,7 +3052,7 @@ mod tests { .begin_archive(&canonical_path("a.eti"), 1) .expect("sender telemetry need not equal the catalog-owned total"); - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = crate::peer_event_channel(); let status = crate::transfer_status::DownloadAttemptStatus::new( crate::DownloadAttemptKey::next("game".to_owned()), CancellationToken::new(), @@ -3146,7 +3147,7 @@ mod tests { let payload = b"payload"; let endpoint = peer_endpoint(); let exact_content_id = content_id(); - let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel(); + let (event_tx, mut event_rx) = crate::peer_event_channel(); let mut accepted = IncomingFile::new( canonical_path("accepted.bin"), PathBuf::from("accepted.bin"), diff --git a/crates/lanspread-peer/src/transfer_status.rs b/crates/lanspread-peer/src/transfer_status.rs index d1aea98..396c99b 100644 --- a/crates/lanspread-peer/src/transfer_status.rs +++ b/crates/lanspread-peer/src/transfer_status.rs @@ -10,7 +10,7 @@ use std::{ }; use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; -use tokio::sync::mpsc::UnboundedSender; +use crate::PeerEventSender; use tokio_util::sync::CancellationToken; use crate::{DownloadProgress, PeerEvent, events}; @@ -140,7 +140,7 @@ struct AttemptRuntimeState { struct DownloadAttemptState { key: DownloadAttemptKey, cancellation: CancellationToken, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, runtime: Mutex, } @@ -153,7 +153,7 @@ impl DownloadAttemptStatus { pub(crate) fn new( key: DownloadAttemptKey, cancellation: CancellationToken, - tx_notify_ui: UnboundedSender, + tx_notify_ui: PeerEventSender, ) -> Self { Self { state: Arc::new(DownloadAttemptState { @@ -459,9 +459,9 @@ mod tests { fn attempt() -> ( DownloadAttemptStatus, - tokio::sync::mpsc::UnboundedReceiver, + crate::PeerEventReceiver, ) { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, rx) = crate::peer_event_channel(); ( DownloadAttemptStatus::new( DownloadAttemptKey::next("game".to_owned()), 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 5bbeaa7..20d261c 100644 --- a/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs +++ b/crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs @@ -35,6 +35,8 @@ use lanspread_peer::{ NoopStreamInstallProvider, PeerCommand, PeerEvent, + PeerEventReceiver, + PeerEventSender, PeerGameDB, PeerIdentity, PeerIdentityDurability, @@ -47,6 +49,7 @@ use lanspread_peer::{ UnpackFuture, Unpacker, migrate_legacy_state, + peer_event_channel, scoped_blocking, start_peer_with_options, }; @@ -687,7 +690,7 @@ struct InstallSettings { language: String, } -struct PeerEventTx(UnboundedSender); +struct PeerEventTx(PeerEventSender); #[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)] enum UiOperationKind { @@ -4086,7 +4089,7 @@ fn emit_game_id_event(app_handle: &AppHandle, event: &str, id: &str, label: &str fn spawn_peer_event_loop( app_handle: AppHandle, - mut rx_peer_event: UnboundedReceiver, + mut rx_peer_event: PeerEventReceiver, mut rx_ui_state: UnboundedReceiver, ) { let tasks = app_handle @@ -4228,6 +4231,7 @@ fn set_identity_diagnostic_in_loop( Ok(next) } +#[cfg(test)] fn take_exactly_queued( receiver: &mut UnboundedReceiver, count: usize, @@ -4243,15 +4247,17 @@ fn take_exactly_queued( async fn drain_queued_peer_events( app_handle: &AppHandle, - receiver: &mut UnboundedReceiver, + receiver: &mut PeerEventReceiver, ) -> Result<(), String> { // The core replies only after all events for the requested transition have - // been enqueued. Once this UI command wins the fair select, snapshot the - // peer queue and process that exact FIFO prefix. Later autonomous traffic - // remains for the normal event-loop turn and cannot contaminate this - // transition's acknowledgement boundary. - let count = receiver.len(); - let queued = take_exactly_queued(receiver, count)?; + // been enqueued. Once this UI command wins the fair select, hold the shared + // producer fence while draining every event already admitted, including a + // coalesced pending view. Later autonomous traffic remains for the normal + // event-loop turn and cannot contaminate this transition's acknowledgement + // boundary. + let queued = receiver + .drain_ready() + .map_err(|error| format!("peer-event queue changed while applying its fence: {error}"))?; for event in queued { handle_peer_event(app_handle, event).await; } @@ -4260,7 +4266,7 @@ async fn drain_queued_peer_events( async fn fence_queued_peer_events( app_handle: &AppHandle, - receiver: &mut UnboundedReceiver, + receiver: &mut PeerEventReceiver, ) -> Result { drain_queued_peer_events(app_handle, receiver).await?; current_local_network_sharing(app_handle.state::().inner()) @@ -4268,7 +4274,7 @@ async fn fence_queued_peer_events( async fn reset_game_transfer_status_in_loop( app_handle: &AppHandle, - receiver: &mut UnboundedReceiver, + receiver: &mut PeerEventReceiver, ) -> Result { // A successful game-root command has already caused core to enqueue its // preceding events. Drain that bounded prefix before terminalizing the old @@ -4290,7 +4296,7 @@ async fn reset_game_transfer_status_in_loop( async fn handle_ui_state_command( app_handle: &AppHandle, command: UiStateCommand, - peer_events: &mut UnboundedReceiver, + peer_events: &mut PeerEventReceiver, ) { match command { UiStateCommand::MutateSharing { mutation, reply } => { @@ -4629,7 +4635,7 @@ struct ProtocolMismatchSnapshot { #[cfg_attr(mobile, tauri::mobile_entry_point)] pub fn run() { // channel to receive events from the peer - let (tx_peer_event, rx_peer_event) = tokio::sync::mpsc::unbounded_channel::(); + let (tx_peer_event, rx_peer_event) = peer_event_channel(); let (tx_ui_state, rx_ui_state) = tokio::sync::mpsc::unbounded_channel::(); tauri::Builder::default()