feat(call-to-play)!: coordinate game sessions across peers

Implement the launcher design as a production peer-to-peer feature. Call to
Play actions are immutable, validated events broadcast over the existing QUIC
control channel, deduplicated in a bounded in-memory history, and exchanged in
Hello/HelloAck so late joiners reconstruct current calls.

Add the Tauri bridge and modular launcher surfaces for play-now and scheduled
calls, check-in, readiness buffers, role-aware controls, chat, tickers, and
actual caller launch. A deterministic frontend reducer derives presentation
state from replicated history. Extend the JSONL peer harness with publish/list
commands and a three-peer live-delivery and late-join scenario.

This intentionally raises the only supported wire protocol from version 5 to
version 6; older builds are not supported. Document the transport architecture
and exclude generated peer-test state from Docker build contexts.

Test Plan:
- `just fmt` -- passed
- `just clippy` -- passed
- `just test` -- passed
- `just frontend-test` -- passed, 20 tests
- `just build` -- passed
- `just peer-cli-tests S2 S48` -- passed
- `git diff --cached --check` -- passed
This commit is contained in:
2026-07-21 22:30:11 +02:00
parent 8f151e38b4
commit 0f53bc4b78
31 changed files with 3032 additions and 21 deletions
+15
View File
@@ -44,6 +44,21 @@ When a peer is discovered:
- Any message updates `last_seen`.
- Pings run only when idle (or on a longer interval), not every 5 seconds.
- Library updates are pushed as deltas, debounced and coalesced.
- Call to Play actions are broadcast as immutable, uniquely identified events.
### Call to Play replication
Call to Play is transient peer-session state rather than database state. The
peer keeps a bounded event history, deduplicated by event ID. A local action is
applied to that history, sent to the UI, and broadcast to every currently known
peer. An incoming live event is applied once and sent to the UI without being
rebroadcast, which prevents forwarding loops.
`Hello` and `HelloAck` include each side's event history. This lets peers that
join after a call was created reconstruct the same nominations, responses,
RSVPs, chat, and terminal actions. The launcher reducer sorts the event stream
deterministically and derives deadlines and check-in phases from timestamps.
There is deliberately no compatibility path for older protocol versions.
### 4) Shutdown
+225
View File
@@ -0,0 +1,225 @@
//! Replicated event history for Call to Play coordination.
use std::collections::HashSet;
use lanspread_proto::{CallToPlayAction, CallToPlayEvent};
use tokio::sync::mpsc::UnboundedSender;
use crate::{PeerEvent, context::Ctx, events, network::send_call_to_play_events};
const MAX_EVENTS: usize = 4_096;
const MAX_ID_CHARS: usize = 128;
const MAX_GAME_ID_CHARS: usize = 256;
const MAX_USERNAME_CHARS: usize = 24;
const MAX_MESSAGE_CHARS: usize = 500;
#[derive(Debug, Default)]
pub(crate) struct CallToPlayStore {
events: Vec<CallToPlayEvent>,
event_ids: HashSet<String>,
}
impl CallToPlayStore {
pub(crate) fn snapshot(&self) -> Vec<CallToPlayEvent> {
self.events.clone()
}
pub(crate) fn insert(&mut self, event: CallToPlayEvent) -> Result<bool, &'static str> {
validate_event(&event)?;
if self.event_ids.contains(&event.id) {
return Ok(false);
}
if self.events.len() >= MAX_EVENTS {
return Err("Call to Play event history is full");
}
self.event_ids.insert(event.id.clone());
self.events.push(event);
Ok(true)
}
pub(crate) fn insert_all(&mut self, events: Vec<CallToPlayEvent>) -> Vec<CallToPlayEvent> {
let mut accepted = Vec::new();
for event in events {
match self.insert(event.clone()) {
Ok(true) => accepted.push(event),
Ok(false) => {}
Err(err) => log::warn!("Ignoring invalid Call to Play event {}: {err}", event.id),
}
}
accepted
}
}
pub(crate) async fn publish(
ctx: &Ctx,
tx_notify_ui: &UnboundedSender<PeerEvent>,
event: CallToPlayEvent,
) {
match ctx.call_to_play.write().await.insert(event.clone()) {
Ok(false) => return,
Err(err) => {
log::warn!("Rejecting local Call to Play event {}: {err}", event.id);
return;
}
Ok(true) => {}
}
events::send(
tx_notify_ui,
PeerEvent::CallToPlayEvents(vec![event.clone()]),
);
let peer_addresses = ctx.peer_game_db.read().await.get_peer_addresses();
ctx.task_tracker.spawn(async move {
let deliveries = peer_addresses.into_iter().map(|peer_addr| {
let event = event.clone();
async move {
if let Err(err) = send_call_to_play_events(peer_addr, vec![event]).await {
log::warn!("Failed to send Call to Play event to {peer_addr}: {err}");
}
}
});
futures::future::join_all(deliveries).await;
});
}
fn validate_event(event: &CallToPlayEvent) -> Result<(), &'static str> {
validate_nonempty(&event.id, MAX_ID_CHARS, "invalid event id")?;
validate_nonempty(&event.call_id, MAX_ID_CHARS, "invalid call id")?;
validate_nonempty(&event.actor, MAX_USERNAME_CHARS, "invalid actor")?;
if event.at <= 0 {
return Err("invalid event timestamp");
}
match &event.action {
CallToPlayAction::Create {
game_id,
max_players,
scheduled_for,
deadline,
} => {
validate_nonempty(game_id, MAX_GAME_ID_CHARS, "invalid game id")?;
if !(2..=64).contains(max_players) {
return Err("max players must be between 2 and 64");
}
if *deadline <= event.at {
return Err("deadline must be after creation");
}
if scheduled_for.is_some_and(|scheduled| scheduled != *deadline) {
return Err("scheduled call deadline must match its start time");
}
}
CallToPlayAction::Respond { ready_at } => {
if ready_at.is_some_and(|ready| ready < event.at) {
return Err("ready time cannot be before the response");
}
}
CallToPlayAction::SendMessage { message_id, text } => {
validate_nonempty(message_id, MAX_ID_CHARS, "invalid message id")?;
validate_nonempty(text, MAX_MESSAGE_CHARS, "invalid message")?;
}
CallToPlayAction::AddTime { deadline } => {
if *deadline <= event.at {
return Err("extended deadline must be in the future");
}
}
CallToPlayAction::Rsvp
| CallToPlayAction::Leave
| CallToPlayAction::Cancel
| CallToPlayAction::Start => {}
}
Ok(())
}
fn validate_nonempty(
value: &str,
max_chars: usize,
error: &'static str,
) -> Result<(), &'static str> {
if value.trim().is_empty() || value.chars().count() > max_chars {
return Err(error);
}
Ok(())
}
#[cfg(test)]
mod tests {
use lanspread_proto::{CallToPlayAction, CallToPlayEvent};
use super::CallToPlayStore;
fn create_event(id: &str) -> CallToPlayEvent {
CallToPlayEvent {
id: id.to_string(),
call_id: "call-1".to_string(),
actor: "Alice".to_string(),
at: 1_000,
action: CallToPlayAction::Create {
game_id: "game-1".to_string(),
max_players: 4,
scheduled_for: None,
deadline: 61_000,
},
}
}
#[test]
fn deduplicates_events_without_reordering_history() {
let mut store = CallToPlayStore::default();
assert!(
store
.insert(create_event("event-1"))
.expect("valid event should be inserted")
);
assert!(
!store
.insert(create_event("event-1"))
.expect("duplicate valid event should be accepted")
);
assert!(
store
.insert(create_event("event-2"))
.expect("valid event should be inserted")
);
let ids = store
.snapshot()
.into_iter()
.map(|event| event.id)
.collect::<Vec<_>>();
assert_eq!(ids, ["event-1", "event-2"]);
}
#[test]
fn rejects_malformed_events() {
let mut store = CallToPlayStore::default();
let mut event = create_event("event-1");
event.action = CallToPlayAction::SendMessage {
message_id: "message-1".to_string(),
text: " ".to_string(),
};
assert_eq!(store.insert(event), Err("invalid message"));
assert!(store.snapshot().is_empty());
}
#[test]
fn merge_returns_only_new_valid_events() {
let mut store = CallToPlayStore::default();
store
.insert(create_event("event-1"))
.expect("valid event should be inserted");
let mut invalid = create_event("invalid");
invalid.actor.clear();
let accepted = store.insert_all(vec![
create_event("event-1"),
invalid,
create_event("event-2"),
]);
assert_eq!(accepted, [create_event("event-2")]);
}
}
+5
View File
@@ -10,6 +10,7 @@ use crate::{
PeerEvent,
StreamInstallProvider,
Unpacker,
call_to_play::CallToPlayStore,
events,
library::LocalLibraryState,
peer_db::PeerGameDB,
@@ -51,6 +52,7 @@ pub struct Ctx {
pub shutdown: CancellationToken,
pub task_tracker: TaskTracker,
pub active_outbound_transfers: OutboundTransfers,
pub call_to_play: Arc<RwLock<CallToPlayStore>>,
}
/// Context for peer connection handling.
@@ -69,6 +71,7 @@ pub struct PeerCtx {
pub shutdown: CancellationToken,
pub task_tracker: TaskTracker,
pub active_outbound_transfers: OutboundTransfers,
pub call_to_play: Arc<RwLock<CallToPlayStore>>,
}
impl std::fmt::Debug for PeerCtx {
@@ -113,6 +116,7 @@ impl Ctx {
shutdown,
task_tracker,
active_outbound_transfers,
call_to_play: Arc::new(RwLock::new(CallToPlayStore::default())),
}
}
@@ -135,6 +139,7 @@ impl Ctx {
shutdown: self.shutdown.clone(),
task_tracker: self.task_tracker.clone(),
active_outbound_transfers: self.active_outbound_transfers.clone(),
call_to_play: self.call_to_play.clone(),
}
}
}
+2
View File
@@ -6,6 +6,7 @@ use crate::state_paths::peer_id_path;
pub const FEATURE_LIBRARY_DELTA: &str = "library-delta-v1";
pub const FEATURE_LIBRARY_SNAPSHOT: &str = "library-snapshot-v1";
pub const FEATURE_CALL_TO_PLAY: &str = "call-to-play-v1";
pub fn load_or_create_peer_id(state_dir: &Path) -> eyre::Result<String> {
let path = peer_id_path(state_dir);
@@ -28,5 +29,6 @@ pub fn default_features() -> Vec<String> {
vec![
FEATURE_LIBRARY_DELTA.to_string(),
FEATURE_LIBRARY_SNAPSHOT.to_string(),
FEATURE_CALL_TO_PLAY.to_string(),
]
}
+15
View File
@@ -12,6 +12,7 @@
// Module declarations
// =============================================================================
mod call_to_play;
mod config;
mod context;
mod download;
@@ -46,6 +47,7 @@ pub use config::{CHUNK_SIZE, MAX_RETRY_COUNT};
pub use error::PeerError;
pub use install::{UnpackFuture, Unpacker};
use lanspread_db::db::{Game, GameCatalog, GameFileDescription};
pub use lanspread_proto::{CallToPlayAction, CallToPlayEvent};
pub use migration::{MigrationReport, migrate_legacy_state};
pub use peer_db::{
MajorityValidationResult,
@@ -159,6 +161,8 @@ pub enum PeerEvent {
ActiveOperationsChanged {
active_operations: Vec<ActiveOperation>,
},
/// New or requested Call to Play events in replication order.
CallToPlayEvents(Vec<CallToPlayEvent>),
/// A required peer runtime component failed.
RuntimeFailed {
component: PeerRuntimeComponent,
@@ -259,6 +263,10 @@ pub enum PeerCommand {
GetPeerCount,
/// Connect directly to a peer address without waiting for mDNS discovery.
ConnectPeer(SocketAddr),
/// Publish one local Call to Play action to this peer and the LAN.
PublishCallToPlay(CallToPlayEvent),
/// Request the complete in-memory Call to Play history.
GetCallToPlayEvents,
}
/// Optional startup settings for non-GUI callers and tests.
@@ -489,6 +497,13 @@ async fn handle_peer_commands(
PeerCommand::ConnectPeer(addr) => {
handle_connect_peer_command(ctx, tx_notify_ui, addr).await;
}
PeerCommand::PublishCallToPlay(event) => {
call_to_play::publish(ctx, tx_notify_ui, event).await;
}
PeerCommand::GetCallToPlayEvents => {
let events = ctx.call_to_play.read().await.snapshot();
events::send(tx_notify_ui, PeerEvent::CallToPlayEvents(events));
}
}
}
}
+8 -1
View File
@@ -9,7 +9,7 @@ use bytes::BytesMut;
use futures::{SinkExt, StreamExt};
use if_addrs::{IfAddr, Interface, get_if_addrs};
use lanspread_db::db::GameFileDescription;
use lanspread_proto::{Hello, HelloAck, LibraryDelta, Message, Request, Response};
use lanspread_proto::{CallToPlayEvent, Hello, HelloAck, LibraryDelta, Message, Request, Response};
use s2n_quic::{
Client as QuicClient,
Connection,
@@ -173,6 +173,13 @@ pub async fn send_goodbye(peer_addr: SocketAddr, peer_id: String) -> eyre::Resul
send_oneway_request(peer_addr, Request::Goodbye { peer_id }).await
}
pub async fn send_call_to_play_events(
peer_addr: SocketAddr,
events: Vec<CallToPlayEvent>,
) -> eyre::Result<()> {
send_oneway_request(peer_addr, Request::CallToPlayEvents { events }).await
}
/// Requests game file details from a peer.
pub async fn request_game_details_from_peer(
peer_addr: SocketAddr,
+128 -5
View File
@@ -8,6 +8,7 @@ use tokio::sync::{RwLock, mpsc::UnboundedSender};
use crate::{
PeerEvent,
call_to_play::CallToPlayStore,
context::{Ctx, PeerCtx},
events,
identity::default_features,
@@ -24,6 +25,7 @@ pub(crate) struct HandshakeCtx {
peer_game_db: Arc<RwLock<PeerGameDB>>,
tx_notify_ui: UnboundedSender<PeerEvent>,
catalog: Arc<RwLock<GameCatalog>>,
call_to_play: Arc<RwLock<CallToPlayStore>>,
}
impl HandshakeCtx {
@@ -35,6 +37,7 @@ impl HandshakeCtx {
peer_game_db: ctx.peer_game_db.clone(),
tx_notify_ui: tx_notify_ui.clone(),
catalog: ctx.catalog.clone(),
call_to_play: ctx.call_to_play.clone(),
}
}
@@ -46,6 +49,7 @@ impl HandshakeCtx {
peer_game_db: ctx.peer_game_db.clone(),
tx_notify_ui: ctx.tx_notify_ui.clone(),
catalog: ctx.catalog.clone(),
call_to_play: ctx.call_to_play.clone(),
}
}
}
@@ -58,28 +62,36 @@ async fn required_listen_addr(
}
pub(super) async fn build_hello_ack(ctx: &PeerCtx) -> eyre::Result<HelloAck> {
let library_guard = ctx.local_library.read().await;
let listen_addr = required_listen_addr(&ctx.local_peer_addr).await?;
let library = build_library_snapshot(&library_guard);
let library = {
let library_guard = ctx.local_library.read().await;
build_library_snapshot(&library_guard)
};
let call_to_play_events = ctx.call_to_play.read().await.snapshot();
Ok(HelloAck {
peer_id: ctx.peer_id.as_ref().clone(),
proto_ver: PROTOCOL_VERSION,
listen_addr,
library,
features: default_features(),
call_to_play_events,
})
}
async fn build_hello_from_state(ctx: &HandshakeCtx) -> eyre::Result<Hello> {
let library_guard = ctx.local_library.read().await;
let listen_addr = required_listen_addr(&ctx.local_peer_addr).await?;
let library = build_library_snapshot(&library_guard);
let library = {
let library_guard = ctx.local_library.read().await;
build_library_snapshot(&library_guard)
};
let call_to_play_events = ctx.call_to_play.read().await.snapshot();
Ok(Hello {
peer_id: ctx.peer_id.as_ref().clone(),
proto_ver: PROTOCOL_VERSION,
listen_addr,
library,
features: default_features(),
call_to_play_events,
})
}
@@ -114,6 +126,13 @@ pub(crate) async fn perform_handshake_with_peer(
let _ = ctx.peer_game_db.write().await.remove_peer(expected);
}
merge_call_to_play_events(
&ctx.call_to_play,
&ctx.tx_notify_ui,
ack.call_to_play_events,
)
.await;
let record_addr = ack.listen_addr;
let upsert = record_remote_library(
&ctx.peer_game_db,
@@ -149,6 +168,12 @@ pub(super) async fn accept_inbound_hello(
}
let addr = hello.listen_addr;
merge_call_to_play_events(
&ctx.call_to_play,
&ctx.tx_notify_ui,
hello.call_to_play_events,
)
.await;
let handshake_ctx = HandshakeCtx::from_peer_ctx(ctx);
let upsert = record_remote_library(
&ctx.peer_game_db,
@@ -165,6 +190,17 @@ pub(super) async fn accept_inbound_hello(
build_hello_ack(ctx).await
}
async fn merge_call_to_play_events(
store: &Arc<RwLock<CallToPlayStore>>,
tx_notify_ui: &UnboundedSender<PeerEvent>,
incoming: Vec<lanspread_proto::CallToPlayEvent>,
) {
let accepted = store.write().await.insert_all(incoming);
if !accepted.is_empty() {
events::send(tx_notify_ui, PeerEvent::CallToPlayEvents(accepted));
}
}
pub(super) fn spawn_library_resync(
ctx: HandshakeCtx,
peer_addr: SocketAddr,
@@ -212,7 +248,15 @@ mod tests {
};
use lanspread_db::db::GameCatalog;
use lanspread_proto::{Availability, GameSummary, Hello, LibrarySnapshot, PROTOCOL_VERSION};
use lanspread_proto::{
Availability,
CallToPlayAction,
CallToPlayEvent,
GameSummary,
Hello,
LibrarySnapshot,
PROTOCOL_VERSION,
};
use tokio::sync::{RwLock, mpsc};
use tokio_util::{sync::CancellationToken, task::TaskTracker};
@@ -248,6 +292,7 @@ mod tests {
peer_game_db,
tx_notify_ui,
catalog: Arc::new(RwLock::new(GameCatalog::empty())),
call_to_play: Arc::new(RwLock::new(crate::call_to_play::CallToPlayStore::default())),
}
}
@@ -264,6 +309,21 @@ mod tests {
}
}
fn call_to_play_event() -> CallToPlayEvent {
CallToPlayEvent {
id: "event-1".to_string(),
call_id: "call-1".to_string(),
actor: "Alice".to_string(),
at: 1_000,
action: CallToPlayAction::Create {
game_id: "game".to_string(),
max_players: 4,
scheduled_for: None,
deadline: 61_000,
},
}
}
#[tokio::test]
async fn outbound_hello_requires_local_listener_addr() {
let ctx = test_handshake_ctx(None);
@@ -304,6 +364,22 @@ mod tests {
assert_eq!(hello.library.games[0].id, "game");
}
#[tokio::test]
async fn outbound_hello_carries_call_to_play_history() {
let ctx = test_handshake_ctx(Some(addr([10, 66, 0, 2], 40000)));
ctx.call_to_play
.write()
.await
.insert(call_to_play_event())
.expect("valid event should be inserted");
let hello = build_hello_from_state(&ctx)
.await
.expect("listener address is present");
assert_eq!(hello.call_to_play_events, [call_to_play_event()]);
}
#[tokio::test]
async fn inbound_hello_applies_remote_library_snapshot() {
let peer_game_db = Arc::new(RwLock::new(PeerGameDB::new()));
@@ -335,6 +411,7 @@ mod tests {
games: vec![summary("remote-game")],
},
features: Vec::new(),
call_to_play_events: Vec::new(),
};
let ack = accept_inbound_hello(&peer_ctx, None, hello)
@@ -406,6 +483,7 @@ mod tests {
games: vec![summary("self-game")],
},
features: Vec::new(),
call_to_play_events: Vec::new(),
};
let ack = accept_inbound_hello(&peer_ctx, None, self_hello)
@@ -422,4 +500,49 @@ mod tests {
"self hello must emit no peer discovery events"
);
}
#[tokio::test]
async fn inbound_hello_merges_call_to_play_history_once() {
let peer_game_db = Arc::new(RwLock::new(PeerGameDB::new()));
let ctx = Ctx::new(
peer_game_db,
"local-peer".to_string(),
PathBuf::new(),
PathBuf::new(),
Arc::new(NoopUnpacker),
CancellationToken::new(),
TaskTracker::new(),
Arc::new(RwLock::new(GameCatalog::empty())),
Arc::new(RwLock::new(HashMap::new())),
Arc::new(crate::NoopStreamInstallProvider),
);
*ctx.local_peer_addr.write().await = Some(addr([127, 0, 0, 1], 4000));
let (tx_notify_ui, mut rx_notify_ui) = mpsc::unbounded_channel();
let peer_ctx = ctx.to_peer_ctx(tx_notify_ui);
let remote_addr = addr([127, 0, 0, 1], 5000);
let hello = Hello {
peer_id: "remote-peer".to_string(),
proto_ver: PROTOCOL_VERSION,
listen_addr: remote_addr,
library: LibrarySnapshot {
library_rev: 0,
games: Vec::new(),
},
features: Vec::new(),
call_to_play_events: vec![call_to_play_event(), call_to_play_event()],
};
accept_inbound_hello(&peer_ctx, None, hello)
.await
.expect("current protocol hello should be accepted");
assert_eq!(
ctx.call_to_play.read().await.snapshot(),
[call_to_play_event()]
);
assert!(matches!(
rx_notify_ui.recv().await,
Some(PeerEvent::CallToPlayEvents(events)) if events == [call_to_play_event()]
));
}
}
@@ -90,6 +90,16 @@ async fn dispatch_request(
handle_library_delta(ctx, peer_id, delta).await;
framed_tx
}
Request::CallToPlayEvents { events: incoming } => {
let accepted = ctx.call_to_play.write().await.insert_all(incoming);
if !accepted.is_empty() {
events::send(
&ctx.tx_notify_ui,
crate::PeerEvent::CallToPlayEvents(accepted),
);
}
framed_tx
}
Request::GetGame { id } => handle_get_game(ctx, id, framed_tx).await,
Request::GetGameFileData(desc) => handle_file_data_request(ctx, desc, framed_tx).await,
Request::GetGameFileChunk {