mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 10 Wiki Activity

fix(core): per-ticket rejoin bootstrap so room detours don't strand a rejoiner (A8)

Browse Source 
Found by manual 2-machine exploration: desktop creates room A, dopedart joins;
desktop leaves A, creates room B, leaves B, then rejoins A — and dopedart (still
in A) isn't reconnected until dopedart itself leaves and rejoins.

Root cause: `known_peers` (the A8 rejoin-bootstrap memory) was a single flat set
cleared on ANY join-ticket change. The detour through room B wiped the memory of
room A's peers, so rejoining A — whose ticket names the creator itself as host —
produced an EMPTY bootstrap (`compute_bootstrap` drops self), leaving the desktop
isolated in the gossip topic with no one to dial. dopedart never re-dials a peer
that's already a (now-departed) neighbor, so they never relink.

Fix: key `known_peers` by room ticket (`HashMap<ticket, HashMap<peer, addr>>`)
and stop clearing it. The event loop records peers under its room's ticket; a
join pulls bootstrap targets from that ticket's bucket. Revisiting a room after a
detour now still remembers its peers. Pre-existing bug (logic unchanged by the
B1 refactor); reproduces on pre-B1 main too.

256 lib tests green, clippy --all-targets clean. NOT yet 2-machine field-verified
— the exact A→B→A scenario is the gate (dopedart is up; verifying next).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-06-15 16:10:28 -04:00
co-authored by Claude Opus 4.8
parent fdab4e03a2
commit af482d9666
1 changed files with 46 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/mod.rs
+46 -27
View File
@@ -535,15 +535,16 @@ async fn run_core_loop(
error: identity_error.clone(), error: identity_error.clone(),
}) })
.await; .await;
// Peers seen in the current/most-recent room, retained ACROSS leave so a // Peers seen per room, keyed by the room's join ticket, retained ACROSS leaves
// rejoin can bootstrap to them. This is the fix for A8: the room creator's own // so a rejoin can bootstrap to them. This is the fix for A8: the room creator's
// ticket lists only themselves as host, so on rejoin these retained peers are // own ticket lists only themselves as host, so on rejoin these retained peers
// their only dial targets. Reset when the joined ticket changes (different / // are their only dial targets. Keyed by TICKET (not one flat set) so that
// brand-new room — see the Join handler). Keyed by id so updates refresh addr. // revisiting a room after a detour through another one still remembers the
let known_peers: Arc<std::sync::Mutex<HashMap<EndpointId, EndpointAddr>>> = // first room's peers — the old single-set version cleared them on any ticket
// change, so an A→B→A bounce stranded the rejoiner with an empty bootstrap.
// Inner map keyed by peer id so updates refresh the address.
let known_peers: Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>> =
Arc::new(std::sync::Mutex::new(HashMap::new())); Arc::new(std::sync::Mutex::new(HashMap::new()));
// The ticket of the room we last (re)joined, to detect room changes above.
let mut last_join_ticket: Option<String> = None;
let audio_backend = Arc::new(PipeWireBackend::new()); let audio_backend = Arc::new(PipeWireBackend::new());
@@ -687,18 +688,16 @@ async fn run_core_loop(
avatar: current_avatar.clone(), avatar: current_avatar.clone(),
}; };
// Retain peers across leave so a rejoin can dial them (A8). A // Snapshot THIS room's retained peers (by ticket) as extra bootstrap
// different ticket (or a fresh "create", which mints a random // targets so a rejoin can dial them (A8) — including after a detour
// ticket) shares no peers with the last room, so reset; rejoining // through another room, since the per-ticket archive isn't cleared.
// the SAME ticket keeps the set — that's what gives the room // Resolution rides the persistent address book.
// creator a dial target on rejoin. Then snapshot it as the extra let extra_bootstrap: Vec<EndpointAddr> = known_peers
// bootstrap set (resolution rides the persistent address book). .lock()
if last_join_ticket.as_deref() != Some(ticket_str.as_str()) { .unwrap()
known_peers.lock().unwrap().clear(); .get(&ticket_str)
} .map(|peers| peers.values().cloned().collect())
last_join_ticket = Some(ticket_str.clone()); .unwrap_or_default();
let extra_bootstrap: Vec<EndpointAddr> =
known_peers.lock().unwrap().values().cloned().collect();
crate::log_msg(&format!("Attempting room_state.join with self_state={:?}, extra_bootstrap={:?}", self_state, extra_bootstrap.iter().map(|a| a.id).collect::<Vec<_>>())); crate::log_msg(&format!("Attempting room_state.join with self_state={:?}, extra_bootstrap={:?}", self_state, extra_bootstrap.iter().map(|a| a.id).collect::<Vec<_>>()));
if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await { if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await {
@@ -1047,6 +1046,9 @@ async fn run_core_loop(
let multitrack_events = multitrack.clone(); let multitrack_events = multitrack.clone();
let is_multitrack_events = is_multitrack.clone(); let is_multitrack_events = is_multitrack.clone();
let known_peers_events = known_peers.clone(); let known_peers_events = known_peers.clone();
// The ticket of the room this event loop serves, so peer add/remove
// updates the right per-ticket bucket in `known_peers` (A8 archive).
let ticket_events = ticket_str.clone();
let event_task = tokio::spawn(async move { let event_task = tokio::spawn(async move {
while let Some(event) = room_events.recv().await { while let Some(event) = room_events.recv().await {
match event { match event {
@@ -1059,8 +1061,14 @@ async fn run_core_loop(
// Hand over the full address so reconnects can dial // Hand over the full address so reconnects can dial
// it directly rather than via the gossip lookup. // it directly rather than via the gossip lookup.
transport_events.connect_peer(state.addr.clone()).await; transport_events.connect_peer(state.addr.clone()).await;
// Retain this peer as a future rejoin bootstrap target (A8). // Retain this peer under this room's ticket as a
known_peers_events.lock().unwrap().insert(peer_id, state.addr.clone()); // future rejoin bootstrap target (A8).
known_peers_events
.lock()
.unwrap()
.entry(ticket_events.clone())
.or_default()
.insert(peer_id, state.addr.clone());
// If a multitrack recording is live, give this peer // If a multitrack recording is live, give this peer
// its own stem track (silence-padded back to t=0). // its own stem track (silence-padded back to t=0).
if is_multitrack_events.load(Ordering::Relaxed) if is_multitrack_events.load(Ordering::Relaxed)
@@ -1076,9 +1084,14 @@ async fn run_core_loop(
cancel_grace_timer(&grace_timers_events, &peer_id); cancel_grace_timer(&grace_timers_events, &peer_id);
seen_connected_events.lock().unwrap().remove(&peer_id); seen_connected_events.lock().unwrap().remove(&peer_id);
// Graceful leave: drop them as a rejoin dial target // Graceful leave: drop them as a rejoin dial target
// (a transient PeerConnectionLost deliberately does // for this room (a transient PeerConnectionLost
// NOT, so we can still re-dial a peer who's still up). // deliberately does NOT, so we can still re-dial a
known_peers_events.lock().unwrap().remove(&peer_id); // peer who's still up).
if let Some(peers) =
known_peers_events.lock().unwrap().get_mut(&ticket_events)
{
peers.remove(&peer_id);
}
transport_events.disconnect_peer(peer_id).await; transport_events.disconnect_peer(peer_id).await;
jitter_events.lock().await.remove(&peer_id); jitter_events.lock().await.remove(&peer_id);
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await; let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
@@ -1092,8 +1105,14 @@ async fn run_core_loop(
// re-records the same address. // re-records the same address.
cancel_grace_timer(&grace_timers_events, &peer_id); cancel_grace_timer(&grace_timers_events, &peer_id);
transport_events.connect_peer(state.addr.clone()).await; transport_events.connect_peer(state.addr.clone()).await;
// Refresh the retained rejoin target with the fresh addr (A8). // Refresh this room's retained rejoin target with the
known_peers_events.lock().unwrap().insert(peer_id, state.addr.clone()); // fresh addr (A8).
known_peers_events
.lock()
.unwrap()
.entry(ticket_events.clone())
.or_default()
.insert(peer_id, state.addr.clone());
let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await; let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
} }
RoomEvent::ChatMessage { from, name, text, .. } => { RoomEvent::ChatMessage { from, name, text, .. } => {