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

feat(call-to-play): retain terminal outcomes

Browse Source 
Keep complete running and cancelled histories visible for fifteen minutes so
peers retain the roster, chat, and outcome long enough to understand what
happened. Compact them to terminal tombstones afterward without charging
settled calls against the active-history limit.

Model running and cancelled as durable read-only frontend states, exclude
them from active badges, prune retired raw events, and document the lifecycle.
Add peer scenario S49 to prove a late joiner reconstructs a terminal call with
its roster and chat intact.

Test Plan:
- just fmt
- just clippy
- just test
- just frontend-test
- just build
- just peer-cli-tests S48 S49
- python3 -m py_compile crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
- git diff --cached --check
This commit is contained in:
ddidderr
2026-07-23 18:03:57 +02:00
parent e5d70ae56f
commit 9c34efa705
13 changed files with 494 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/src/call_to_play.rs
+92 -21
View File
@@ -23,6 +23,7 @@ const MAX_GAME_ID_CHARS: usize = 256;
const MAX_USERNAME_CHARS: usize = 24;
const MAX_MESSAGE_CHARS: usize = 500;
const EXPIRED_RETENTION_MS: i64 = 5 * 60_000;
const TERMINAL_RETENTION_MS: i64 = 15 * 60_000;
#[derive(Debug, Default)]
pub(crate) struct CallToPlayStore {
@@ -120,6 +121,7 @@ impl CallToPlayStore {
.map(|event| event.call_id.clone())
.collect::<HashSet<_>>();
let retained_tombstones = terminal_tombstone_call_ids(&retained);
let retained_history = HistoryIndex::build(&retained);
let mut applicable = Vec::with_capacity(new_events.len());
let mut missing_call_ids = BTreeSet::new();
let mut obsolete = 0;
@@ -129,6 +131,13 @@ impl CallToPlayStore {
obsolete += 1;
continue;
}
if let Some(terminal) = retained_history.terminal_events.get(&event.call_id)
&& (matches!(event.action, CallToPlayAction::Create { .. })
|| (event.at, event.id.as_str()) > terminal.order_key())
{
obsolete += 1;
continue;
}
if !matches!(event.action, CallToPlayAction::Create { .. })
&& !rooted_calls.contains(event.call_id.as_str())
{
@@ -336,10 +345,13 @@ fn compact_history(events: &mut Vec<CallToPlayEvent>, now: i64) {
if expired_calls.contains(event.call_id.as_str()) {
return false;
}
index
.terminal_events
.get(&event.call_id)
.is_none_or(|terminal| event.id == terminal.event_id)
let Some(terminal) = index.terminal_events.get(&event.call_id) else {
return true;
};
if (event.at, event.id.as_str()) > terminal.order_key() {
return false;
}
now - terminal.at <= TERMINAL_RETENTION_MS || event.id == terminal.event_id
});
}
@@ -536,7 +548,13 @@ fn validate_nonempty(
mod tests {
use lanspread_proto::{CallToPlayAck, CallToPlayAction, CallToPlayEvent};
use super::{CallToPlayStore, MAX_EVENTS, MergeError, delivery_resync_reason};
use super::{
CallToPlayStore,
MAX_EVENTS,
MergeError,
TERMINAL_RETENTION_MS,
delivery_resync_reason,
};
const TEST_NOW: i64 = 8_000_000_000_000;
@@ -705,6 +723,11 @@ mod tests {
store
.merge_batch_at(vec![create_event("create"), start.clone()], TEST_NOW)
.expect("terminal history should merge");
let after_terminal_retention = start.at + TERMINAL_RETENTION_MS + 1;
assert_eq!(
store.snapshot_at(after_terminal_retention).as_slice(),
std::slice::from_ref(&start)
);
let stale = store
.merge_batch_at(
@@ -718,13 +741,13 @@ mod tests {
},
),
],
TEST_NOW,
after_terminal_retention,
)
.expect("stale history is an obsolete merge outcome");
assert!(stale.applied.is_empty());
assert_eq!(stale.obsolete, 2);
assert_eq!(store.snapshot_at(TEST_NOW), [start]);
assert_eq!(store.snapshot_at(after_terminal_retention), [start]);
}
#[test]
@@ -735,32 +758,31 @@ mod tests {
let merged = store
.merge_batch_at(vec![terminal.clone()], TEST_NOW)
.expect("terminal action should compact the full call");
.expect("terminal action should settle the full call");
assert_eq!(merged.applied.as_slice(), std::slice::from_ref(&terminal));
assert_eq!(store.snapshot_at(TEST_NOW), [terminal]);
assert_eq!(store.snapshot_at(TEST_NOW).len(), MAX_EVENTS + 1);
assert_eq!(
store.snapshot_at(terminal.at + TERMINAL_RETENTION_MS + 1),
[terminal]
);
}
}
#[test]
fn terminal_tombstones_do_not_consume_active_history_capacity() {
fn terminal_histories_do_not_consume_active_history_capacity() {
let mut store = full_active_store();
let mut terminal = action_event("call-1-start", "call-1", CallToPlayAction::Start);
terminal.at = TEST_NOW + 2_000;
store
.merge_batch_at(
vec![action_event(
"call-1-start",
"call-1",
CallToPlayAction::Start,
)],
TEST_NOW,
)
.expect("start should compact active history");
.merge_batch_at(vec![terminal], TEST_NOW)
.expect("start should settle active history");
let created = create_event_for("call-2", "call-2-create");
let merged = store
.merge_batch_at(vec![created.clone()], TEST_NOW)
.expect("terminal tombstone must not block a new active call");
.expect("terminal history must not block a new active call");
assert_eq!(merged.applied, [created]);
assert_eq!(store.snapshot_at(TEST_NOW).len(), 2);
assert_eq!(store.snapshot_at(TEST_NOW).len(), MAX_EVENTS + 2);
}
#[test]
@@ -804,6 +826,55 @@ mod tests {
assert!(store.snapshot_at(TEST_NOW).contains(&message));
}
#[test]
fn terminal_history_preserves_roster_and_chat_for_display_window() {
let mut store = CallToPlayStore::default();
let message = action_event(
"message",
"call-1",
CallToPlayAction::SendMessage {
message_id: "message-1".to_string(),
text: "Launching now".to_string(),
},
);
let terminal = action_event("terminal", "call-1", CallToPlayAction::Start);
let history = vec![
create_event("create"),
action_event("rsvp", "call-1", CallToPlayAction::Rsvp),
message,
terminal.clone(),
];
store
.merge_batch_at(history.clone(), TEST_NOW)
.expect("terminal history should merge");
let mut post_terminal = action_event(
"post-terminal",
"call-1",
CallToPlayAction::SendMessage {
message_id: "message-2".to_string(),
text: "Too late".to_string(),
},
);
post_terminal.at = terminal.at + 1;
let obsolete = store
.merge_batch_at(
vec![create_event("different-create"), post_terminal],
TEST_NOW,
)
.expect("terminal updates should be obsolete");
assert!(obsolete.applied.is_empty());
assert_eq!(obsolete.obsolete, 2);
assert_eq!(
store.snapshot_at(terminal.at + TERMINAL_RETENTION_MS),
history
);
assert_eq!(
store.snapshot_at(terminal.at + TERMINAL_RETENTION_MS + 1),
[terminal]
);
}
#[test]
fn only_failed_or_incomplete_deliveries_request_resync() {
assert!(delivery_resync_reason(Err(())).is_some());