From 9d139889d0e8a2287db05c79faaa2a33c6f65e23 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 19:02:19 -0400 Subject: [PATCH] fix(relay): never auto-end a huddle on drain; hold empty rooms 20s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A relay drain (SIGTERM or owner-drain) cancelled every local owner client at once, so the last teardown emptied the room and took the archive + 48103 path: every pod restart ended every huddle it owned. The same path also ended a huddle the instant its last client blipped, so a reconnect found the channel archived. The leave path now distinguishes a drain-driven teardown (the owner draining token fired, or `shutting_down` / `owners.is_draining()` is set, covering single-pod mode where there is no mesh token) and only releases the lease and drops the empty room, leaving the channel alive for rejoiners to re-acquire through Redis. Ordinary last-leaver departures no longer end the room atomically. `remove_peer_and_check_idle` captures an `IdleGeneration` (the admission count) under the guard lock; a spawned task waits `ROOM_EMPTY_GRACE` and calls `end_if_idle`, which sets `ended` only if the room is still empty and no admission happened in between. A rejoin during the window fences the stale observation even if the rejoiner has already left again, and two leavers who both observed empty cannot both archive. Owner-loss or a drain that begins mid-window aborts the end; owner-loss wins a tie with the timer. The fence only holds if the rejoiner lands on the same `Room`. A pre-last leaver's `cleanup_if_empty` (or a failed pre-admission join) could run after the last leaver's idle observation and evict the empty room from the manager; the rejoiner would then get a fresh `Room` and the stale grace task would archive underneath it. The idle observation now sets `idle_hold` on the room and `cleanup_if_empty` refuses to evict while it is set. The hold survives `end_if_idle` so the room stays registered while `archive_channel` is in flight — a joiner in that gap meets `ended` on the same room instead of admitting into a fresh one whose pre-join DB check raced the archive. The hold clears on admission or via the fenced `release_idle_hold`, called after the archive resolves (success or rollback) and on the drain/abort paths that abandon the window. Tests: room idle generation (last-leaver-only, end-once, rejoin fences stale), manager eviction (concurrent-leaver ordering pins the room and the rejoiner fences the stale end; released hold permits eviction, stale release does not; ended room stays pinned and refuses joiners until the archive resolves; failed archive reopens and releases), and paused-time grace outcomes (holds through window, rejoin, no-rejoin ends, owner-loss aborts, mid-window drain aborts). Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- crates/buzz-relay/src/audio/handler.rs | 315 +++++++++++++++++++++---- crates/buzz-relay/src/audio/room.rs | 315 +++++++++++++++++++++++-- 2 files changed, 562 insertions(+), 68 deletions(-) diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index de8f1e1459..2c2631a292 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -60,6 +60,11 @@ const MAX_MISSED_PONGS: u8 = 3; /// Auth timeout. const AUTH_TIMEOUT: Duration = Duration::from_secs(5); +/// How long an emptied huddle room stays open for a rejoin before it +/// auto-ends. Covers a laptop sleep/wake or a client-side reconnect without +/// ending the huddle for everyone who was about to come back. +const ROOM_EMPTY_GRACE: Duration = Duration::from_secs(20); + /// WebSocket upgrade handler for `/huddle/:channel_id/audio`. pub async fn ws_audio_handler( State(state): State>, @@ -790,6 +795,10 @@ async fn handle_active_audio_connection( // the local generation floor so the fresh generation is accepted. The cause // distinction is carried on the remote control streams; locally the action // is the same WS teardown. Silent on ordinary client leave. + // Clones survive the teardown watcher's move so the leave path below can + // tell a drain-driven teardown from an ordinary leave. + let owner_lost_after_leave = owner_lost.clone(); + let owner_draining_after_leave = owner_draining.clone(); let owner_teardown_task = if owner_lost.is_some() || owner_draining.is_some() { let fence = Arc::clone( &state @@ -862,14 +871,15 @@ async fn handle_active_audio_connection( let _ = owner_teardown_task.await; } - // Atomic owner remove + end check: remove_peer_and_check_ended holds the - // AdmissionGuard lock across index recycling AND the is_empty + ended=true - // check. Ingress mirrors never archive authoritative huddle state; they + // Atomic owner remove + idle observation: remove_peer_and_check_idle holds + // the AdmissionGuard lock across index recycling AND the is_empty check, + // capturing the admission generation so a later end_if_idle cannot race a + // rejoin. Ingress mirrors never archive authoritative huddle state; they // remove locally and let the owner decide room lifetime. let removal = if remote_session.is_some() { - room.remove_peer(peer_id).map(|delta| (delta, false)) + room.remove_peer(peer_id).map(|delta| (delta, None)) } else { - room.remove_peer_and_check_ended(peer_id) + room.remove_peer_and_check_idle(peer_id) }; let removal_revision = if remote_session.is_none() { removal.as_ref().map(|(delta, _)| delta.revision) @@ -878,7 +888,7 @@ async fn handle_active_audio_connection( // ordering. Omit it rather than publishing a plausible-but-wrong value. None }; - let should_auto_end = removal.as_ref().map(|(_, ended)| *ended).unwrap_or(false); + let idle = removal.as_ref().and_then(|(_, idle)| *idle); if remote_session.is_none() { if let Some((delta, _)) = removal { @@ -916,45 +926,51 @@ async fn handle_active_audio_connection( ) .await; - let room_emptied; - if should_auto_end { - info!(channel_id = %channel_id, "audio room empty — auto-ending huddle"); - - match state - .db - .archive_channel(tenant.community(), channel_id) - .await - { - Err(e) => { - warn!(channel_id = %channel_id, "auto-archive failed, huddle stays alive: {e}"); - room.clear_ended(); - room_emptied = false; - } - Ok(()) => { - room_emptied = state - .audio_rooms - .cleanup_if_empty(tenant.community(), channel_id); - - emit_participant_event( - &state, - &tenant, - channel_id, - parent_id_for_event, - ParticipantLifecycle { - kind: Kind::Custom(48103), - participant_pubkey: &pubkey_hex, - roster_revision: None, - admission_id: None, - }, - ) - .await; - } + // A relay drain (SIGTERM / owner-drain) tore every local client down at + // once. That empties the room without anyone choosing to leave, so it must + // never end the huddle: release the lease so rejoiners re-acquire through + // Redis, and leave the channel alive. Ordinary last-leaver departures get a + // grace window before the room ends so a reconnecting client keeps its + // huddle. + let draining = owner_draining_after_leave.is_some_and(|token| token.is_cancelled()) + || relay_is_draining(&state); + let room_emptied = match idle { + Some(idle) if !draining => { + info!( + channel_id = %channel_id, + grace_secs = ROOM_EMPTY_GRACE.as_secs(), + "audio room empty — holding huddle open for rejoin" + ); + tokio::spawn(end_room_after_grace( + Arc::clone(&state), + tenant.clone(), + channel_id, + parent_id_for_event, + pubkey_hex.clone(), + Arc::clone(&room), + idle, + owner_lost_after_leave, + owner_generation, + )); + // The grace task now owns room cleanup and lease release. + false } - } else { - room_emptied = state + Some(idle) => { + info!(channel_id = %channel_id, "audio room emptied by relay drain — huddle stays alive"); + room.release_idle_hold(idle); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id) + } + // Another peer was present under the same lock that removed this one + // (or this is an ingress mirror, which never observes idle). Cleanup is + // safe from any departure: a pending grace window pins the room in the + // manager, so a slow pre-last leaver cannot detach a room whose end is + // still in flight. + None => state .audio_rooms - .cleanup_if_empty(tenant.community(), channel_id); - } + .cleanup_if_empty(tenant.community(), channel_id), + }; // Owner path: release this room's lease when the room empties, so a new // owner can acquire and the renewer stops cleanly (silent, not owner-loss). @@ -963,9 +979,7 @@ async fn handle_active_audio_connection( // is a no-op for the stale generation and leaves the live renewer running. // Only the last leaver empties the room, so exactly one release fires. if room_emptied { - if let (Some(mesh), Some(generation)) = (state.mesh(), owner_generation) { - mesh.owners.release(channel_id, generation); - } + release_owner_lease(&state, channel_id, owner_generation); } info!( @@ -975,6 +989,134 @@ async fn handle_active_audio_connection( ); } +/// Whether this runtime is shutting down or draining its huddle ownership. +/// `shutting_down` flips on SIGTERM before the mesh watcher propagates it to +/// `owners.drain_all()`, and is the only signal in single-pod mode. +fn relay_is_draining(state: &AppState) -> bool { + state.shutting_down.load(Ordering::Relaxed) + || state.mesh().is_some_and(|mesh| mesh.owners.is_draining()) +} + +fn release_owner_lease(state: &AppState, channel_id: Uuid, owner_generation: Option) { + if let (Some(mesh), Some(generation)) = (state.mesh(), owner_generation) { + mesh.owners.release(channel_id, generation); + } +} + +/// Outcome of waiting out the empty-room grace window. +#[derive(Debug, PartialEq, Eq)] +enum IdleOutcome { + /// The grace window elapsed with no rejoin; the room is now `ended` and the + /// caller must archive + emit 48103. + Ended, + /// A peer was admitted during the window (whether or not it has since + /// left). That peer's own departure owns the next lifecycle decision. + Rejoined, + /// The lease was lost or the relay began draining before the window + /// elapsed. The room must not end; another owner may hold it now. + Aborted, +} + +/// Wait `grace`, then end the room if it is still idle. `owner_lost` aborts the +/// wait; `draining` is re-polled when the timer fires so a drain that began +/// mid-window (single-pod mode has no token) cannot end the huddle. +async fn await_room_idle( + room: &crate::audio::room::Room, + idle: crate::audio::room::IdleGeneration, + grace: Duration, + owner_lost: Option, + draining: impl Fn() -> bool, +) -> IdleOutcome { + let lost = async { + match owner_lost { + Some(token) => token.cancelled().await, + None => std::future::pending().await, + } + }; + tokio::select! { + // Owner loss wins a tie with the timer: another pod may own the room. + biased; + _ = lost => IdleOutcome::Aborted, + _ = tokio::time::sleep(grace) => { + if draining() { + IdleOutcome::Aborted + } else if room.end_if_idle(idle) { + IdleOutcome::Ended + } else { + IdleOutcome::Rejoined + } + } + } +} + +#[allow(clippy::too_many_arguments)] +async fn end_room_after_grace( + state: Arc, + tenant: TenantContext, + channel_id: Uuid, + parent_id_for_event: Uuid, + last_leaver_pubkey: String, + room: Arc, + idle: crate::audio::room::IdleGeneration, + owner_lost: Option, + owner_generation: Option, +) { + let outcome = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, owner_lost, || { + relay_is_draining(&state) + }) + .await; + let room_emptied = match outcome { + IdleOutcome::Rejoined => return, + IdleOutcome::Aborted => { + info!(channel_id = %channel_id, "audio room grace aborted — huddle stays alive"); + room.release_idle_hold(idle); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id) + } + IdleOutcome::Ended => { + info!(channel_id = %channel_id, "audio room stayed empty — auto-ending huddle"); + match state + .db + .archive_channel(tenant.community(), channel_id) + .await + { + Err(e) => { + warn!(channel_id = %channel_id, "auto-archive failed, huddle stays alive: {e}"); + room.clear_ended(); + room.release_idle_hold(idle); + return; + } + Ok(()) => { + // The hold outlived `end_if_idle` so no joiner could land + // on a replacement room while the archive was in flight. + room.release_idle_hold(idle); + let emptied = state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); + emit_participant_event( + &state, + &tenant, + channel_id, + parent_id_for_event, + ParticipantLifecycle { + kind: Kind::Custom(48103), + participant_pubkey: &last_leaver_pubkey, + roster_revision: None, + admission_id: None, + }, + ) + .await; + emptied + } + } + } + }; + if room_emptied { + release_owner_lease(&state, channel_id, owner_generation); + } +} + /// React to a non-owner huddle teardown signal read off the owner's control /// stream: cancel the connection (which drives the client's WS to close so it /// rejoins) and forget the local generation floor for this session. @@ -1691,4 +1833,85 @@ mod tests { "oversized messages must be rejected by the WebSocket parser before the handler sees them" ); } + + fn idle_room() -> (crate::audio::room::Room, crate::audio::room::IdleGeneration) { + let room = crate::audio::room::Room::new( + buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()), + Uuid::new_v4(), + ); + let (peer, ..) = room.add_peer("alice".into(), 3).expect("admit"); + let (_, idle) = room.remove_peer_and_check_idle(peer).expect("peer existed"); + (room, idle.expect("last leaver observes idle")) + } + + /// The grace window holds the room open; once it elapses with no rejoin the + /// room ends and refuses further admission. + #[tokio::test(start_paused = true)] + async fn empty_room_ends_only_after_grace_elapses() { + let (room, idle) = idle_room(); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, || false); + tokio::pin!(wait); + + tokio::time::advance(ROOM_EMPTY_GRACE - Duration::from_millis(1)).await; + assert!( + futures_util::poll!(&mut wait).is_pending(), + "room must stay open for the whole grace window" + ); + let (bob, ..) = room + .add_peer("bob".into(), 3) + .expect("rejoin is admitted during grace"); + room.remove_peer(bob); + + tokio::time::advance(Duration::from_millis(1)).await; + assert_eq!( + wait.await, + IdleOutcome::Rejoined, + "a rejoin during grace fences out the stale observation" + ); + assert!(room.add_peer("carol".into(), 3).is_ok(), "room never ended"); + } + + #[tokio::test(start_paused = true)] + async fn empty_room_with_no_rejoin_ends_after_grace() { + let (room, idle) = idle_room(); + let outcome = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, || false).await; + assert_eq!(outcome, IdleOutcome::Ended); + assert!(matches!( + room.add_peer("bob".into(), 3), + Err(crate::audio::room::AdmissionError::Ended) + )); + } + + /// Owner-loss during the window aborts without ending: the room now belongs + /// to whichever pod re-acquires the lease. + #[tokio::test(start_paused = true)] + async fn owner_loss_during_grace_aborts_without_ending() { + let (room, idle) = idle_room(); + let lost = CancellationToken::new(); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, Some(lost.clone()), || false); + tokio::pin!(wait); + + tokio::time::advance(Duration::from_secs(1)).await; + lost.cancel(); + assert_eq!(wait.await, IdleOutcome::Aborted); + assert!(room.add_peer("bob".into(), 3).is_ok(), "room was not ended"); + } + + /// A drain that begins mid-window (single-pod mode has no drain token, only + /// `shutting_down`) must not end the huddle when the timer fires. + #[tokio::test(start_paused = true)] + async fn drain_during_grace_aborts_without_ending() { + let (room, idle) = idle_room(); + let draining = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let flag = Arc::clone(&draining); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, move || { + flag.load(Ordering::Relaxed) + }); + tokio::pin!(wait); + + tokio::time::advance(Duration::from_secs(5)).await; + draining.store(true, Ordering::Relaxed); + assert_eq!(wait.await, IdleOutcome::Aborted); + assert!(room.add_peer("bob".into(), 3).is_ok(), "room was not ended"); + } } diff --git a/crates/buzz-relay/src/audio/room.rs b/crates/buzz-relay/src/audio/room.rs index d2849f3e0b..0c14a45413 100644 --- a/crates/buzz-relay/src/audio/room.rs +++ b/crates/buzz-relay/src/audio/room.rs @@ -117,6 +117,14 @@ pub type IndexedPeerAdmission = ( u64, ); +/// Snapshot of a room's admission history taken the moment it was observed +/// empty. Passed back to [`Room::end_if_idle`] after the grace window: if any +/// peer admitted in between, the snapshot is stale and the end is refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct IdleGeneration { + admissions: u64, +} + /// Reason a peer was refused entry to a room. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum AdmissionError { @@ -179,6 +187,18 @@ struct AdmissionGuard { /// this behavior. pinned_version: Option, roster_revision: u64, + /// Count of successful admissions. A room-empty observation captures this + /// value; [`Room::end_if_idle`] only ends the room if no admission has + /// happened since, so a rejoin during the empty-room grace window fences + /// out the stale auto-end without any explicit timer cancellation. + admissions: u64, + /// The last peer left and an [`IdleGeneration`] was handed out: a grace + /// window may be pending against *this* `Room`. While set, the manager + /// must not drop the room — a rejoiner has to land on the same `Room` so + /// its admission fences the pending [`Room::end_if_idle`]. Stays set + /// through `end_if_idle` while the archive is in flight; cleared by + /// admission or by [`Room::release_idle_hold`]. + idle_hold: bool, } impl AdmissionGuard { @@ -190,6 +210,8 @@ impl AdmissionGuard { ended: false, pinned_version: None, roster_revision: 0, + admissions: 0, + idle_hold: false, } } @@ -325,6 +347,8 @@ impl Room { protocol_version: requested_version, }, ); + g.admissions += 1; + g.idle_hold = false; g.roster_revision = g.roster_revision.wrapping_add(1); let revision = g.roster_revision; let delta = RosterDelta { @@ -386,6 +410,8 @@ impl Room { protocol_version: requested_version, }, ); + g.admissions += 1; + g.idle_hold = false; g.roster_revision = g.roster_revision.wrapping_add(1); let revision = g.roster_revision; let delta = RosterDelta { @@ -425,12 +451,16 @@ impl Room { Some(delta) } - /// Remove a peer AND atomically check if the room should end. - /// If the room is now empty, sets `ended = true` under the same lock - /// acquisition that removes the peer — no window for a concurrent - /// `add_peer` to sneak in between removal and the ended flag. - /// Returns `(roster_delta, should_auto_end)`. - pub fn remove_peer_and_check_ended(&self, peer_id: Uuid) -> Option<(RosterDelta, bool)> { + /// Remove a peer AND atomically observe whether it left the room empty. + /// Returns `(roster_delta, idle)` where `idle` is `Some` only for the + /// departure that emptied a not-yet-ended room. The [`IdleGeneration`] + /// captures the admission count under the same lock acquisition that + /// removed the peer, so the caller can wait out a grace window and then + /// call [`Self::end_if_idle`] without racing a concurrent `add_peer`. + pub fn remove_peer_and_check_idle( + &self, + peer_id: Uuid, + ) -> Option<(RosterDelta, Option)> { let mut g = self.guard.lock().ok()?; let (_, peer) = self.peers.remove(&peer_id)?; let peer_index = peer.peer_index; @@ -445,18 +475,60 @@ impl Room { epoch: peer.epoch, }), }; - // Only the first task to see empty + !ended wins the auto-end. - // This prevents duplicate archive/48103 when two peers disconnect - // simultaneously and both see is_empty() == true. - let should_end = if !g.ended && self.peers.is_empty() { - g.ended = true; - true - } else { - false - }; + let idle = (!g.ended && self.peers.is_empty()).then_some(IdleGeneration { + admissions: g.admissions, + }); + g.idle_hold |= idle.is_some(); let _ = self.roster_tx.send(delta.clone()); drop(g); - Some((delta, should_end)) + Some((delta, idle)) + } + + /// End the room if it is still idle: empty, not already ended, and no + /// admission has happened since `idle` was observed. Sets `ended = true` + /// under the admission lock so no `add_peer` can sneak in after the check. + /// Returns `true` exactly once per idle generation — two leavers who both + /// observed the same empty room cannot both end it, and a rejoin (even one + /// that has since left again) fences out the stale observation. + /// + /// The idle hold is retained: the caller still has to archive the channel, + /// and until that resolves the room must stay registered so a concurrent + /// joiner meets `ended` on this `Room` instead of a fresh one whose + /// pre-join DB check can race the archive. Call + /// [`Self::release_idle_hold`] once the archive has succeeded, or + /// [`Self::clear_ended`] + `release_idle_hold` if it failed. + pub fn end_if_idle(&self, idle: IdleGeneration) -> bool { + let Ok(mut g) = self.guard.lock() else { + return false; + }; + if g.ended || g.admissions != idle.admissions || !self.peers.is_empty() { + return false; + } + g.ended = true; + true + } + + /// Release the hold taken by the idle observation `idle`, so the manager + /// may evict the room — either because the grace window was abandoned + /// (drain, owner loss) or because the end it guarded has fully resolved. + /// Fenced like [`Self::end_if_idle`]: a later admission owns its own hold, + /// which a stale release must not lift. + pub fn release_idle_hold(&self, idle: IdleGeneration) { + if let Ok(mut g) = self.guard.lock() { + if g.admissions == idle.admissions { + g.idle_hold = false; + } + } + } + + /// True when the room may be dropped from the manager: no peers, and no + /// grace window or in-flight end pending that a rejoiner would need to + /// land on. Both are read under the admission lock so a concurrent + /// `add_peer` cannot slip a peer in between the two checks. + fn is_evictable(&self) -> bool { + self.guard + .lock() + .is_ok_and(|g| !g.idle_hold && self.peers.is_empty()) } /// Fan-out a binary frame to all peers except the sender. Protocol v3 @@ -621,10 +693,16 @@ impl AudioRoomManager { Some(room) } - /// Remove the room if it has no peers. Returns `true` if the room was removed. + /// Remove the room if it has no peers and no pending grace window. + /// Returns `true` if the room was removed. + /// + /// A room whose last peer just left stays registered until its grace + /// window resolves ([`Room::end_if_idle`] / [`Room::release_idle_hold`]): + /// evicting it early would hand a rejoiner a fresh `Room` that the pending + /// end cannot see, and the stale end would then archive the live huddle. pub fn cleanup_if_empty(&self, community_id: CommunityId, channel_id: Uuid) -> bool { self.rooms - .remove_if(&(community_id, channel_id), |_, room| room.is_empty()) + .remove_if(&(community_id, channel_id), |_, room| room.is_evictable()) .is_some() } } @@ -787,11 +865,13 @@ mod tests { let (peer_id, _, _, _, _, _) = room1 .add_peer("alice".to_string(), 2) .expect("first peer admits"); - // Last peer leaves and ends the room atomically. - let (_, ended) = room1 - .remove_peer_and_check_ended(peer_id) + // Last peer leaves; the idle observation ends the room after grace. + let (_, idle) = room1 + .remove_peer_and_check_idle(peer_id) .expect("peer existed"); - assert!(ended, "single-peer room should end on its last departure"); + let idle = idle.expect("single-peer room should be idle on its last departure"); + assert!(room1.end_if_idle(idle), "idle room ends"); + room1.release_idle_hold(idle); assert!(manager.cleanup_if_empty(community_id, channel_id)); // Next joiner with a different version on the same channel id gets a @@ -993,4 +1073,195 @@ mod tests { // And the room state must be unchanged. assert_eq!(room.peers.len(), MAX_PEERS_PER_ROOM); } + + /// Only the departure that empties the room observes idleness; the + /// observation then ends the room exactly once. + #[test] + fn idle_observed_only_by_last_leaver_and_ends_once() { + let room = fresh_room(); + let (alice, ..) = room.add_peer("alice".into(), 2).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 2).unwrap(); + + let (_, idle) = room.remove_peer_and_check_idle(alice).unwrap(); + assert!(idle.is_none(), "room still has bob"); + let (_, idle) = room.remove_peer_and_check_idle(bob).unwrap(); + let idle = idle.expect("bob emptied the room"); + + assert!(room.end_if_idle(idle)); + assert!( + !room.end_if_idle(idle), + "second end on the same generation is a no-op" + ); + assert!(matches!( + room.add_peer("carol".into(), 2), + Err(AdmissionError::Ended) + )); + } + + /// A rejoin during the grace window fences out the stale idle observation, + /// even if the rejoiner has already left again — its own departure owns the + /// next lifecycle decision. + #[test] + fn rejoin_during_grace_fences_stale_idle() { + let room = fresh_room(); + let (alice, ..) = room.add_peer("alice".into(), 2).unwrap(); + let (_, idle) = room.remove_peer_and_check_idle(alice).unwrap(); + let stale = idle.unwrap(); + + let (alice_again, ..) = room + .add_peer("alice".into(), 2) + .expect("room is not ended during grace"); + assert!(!room.end_if_idle(stale), "occupied room never ends"); + + let (_, idle) = room.remove_peer_and_check_idle(alice_again).unwrap(); + let fresh = idle.expect("alice emptied the room again"); + assert!( + !room.end_if_idle(stale), + "stale generation cannot end a room a later admission touched" + ); + assert!(room.end_if_idle(fresh), "the fresh observation ends it"); + } + + /// Mari's concurrent-leaver ordering: Alice removes first (sees Bob, gets + /// no idle), Bob removes last (gets the idle generation), then Alice's + /// delayed `cleanup_if_empty` runs. The registry must keep the room pinned + /// so a rejoiner lands on the same `Room` and fences Bob's pending end; + /// otherwise the stale end would archive underneath a live replacement. + #[test] + fn pending_grace_pins_room_in_manager_across_concurrent_leavers() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 3).unwrap(); + + let (_, alice_idle) = room.remove_peer_and_check_idle(alice).unwrap(); + assert!(alice_idle.is_none(), "Bob is still present"); + let (_, bob_idle) = room.remove_peer_and_check_idle(bob).unwrap(); + let bob_idle = bob_idle.expect("Bob emptied the room"); + + // Alice's cleanup arrives late, after Bob's idle observation. + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "an empty room with a pending grace window must not be evicted" + ); + let rejoin_room = manager.get_or_create(community_id, channel_id); + assert!( + Arc::ptr_eq(&room, &rejoin_room), + "rejoin must land on the room the grace task holds" + ); + let (carol, ..) = rejoin_room.add_peer("carol".into(), 3).unwrap(); + assert!( + !room.end_if_idle(bob_idle), + "Carol's admission fences Bob's stale end" + ); + + // Carol leaves; her observation owns the lifecycle and ends the room. + let (_, carol_idle) = room.remove_peer_and_check_idle(carol).unwrap(); + let carol_idle = carol_idle.unwrap(); + assert!(room.end_if_idle(carol_idle)); + room.release_idle_hold(carol_idle); + assert!(manager.cleanup_if_empty(community_id, channel_id)); + assert!(manager.get(community_id, channel_id).is_none()); + } + + /// Mari's archive-in-flight ordering: the grace window expires and + /// `end_if_idle` succeeds, but the archive write has not resolved yet. A + /// delayed `cleanup_if_empty` must not detach the ended room, and a joiner + /// arriving in that gap must meet `ended` on the same `Room` rather than + /// admit into a fresh one whose pre-join DB check raced the archive. + #[test] + fn ended_room_stays_pinned_until_archive_resolves() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let idle = room.remove_peer_and_check_idle(alice).unwrap().1.unwrap(); + assert!(room.end_if_idle(idle)); + + // Archive is in flight: cleanup cannot detach, and a joiner lands on + // the ended room and is refused. + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "an ended room with its archive in flight must stay registered" + ); + let joiner_room = manager.get_or_create(community_id, channel_id); + assert!(Arc::ptr_eq(&room, &joiner_room)); + assert!(matches!( + joiner_room.add_peer("bob".into(), 3), + Err(AdmissionError::Ended) + )); + + // Archive succeeded: release, then the manager may evict. + room.release_idle_hold(idle); + assert!(manager.cleanup_if_empty(community_id, channel_id)); + assert!(manager.get(community_id, channel_id).is_none()); + } + + /// Archive failure rolls the end back: the room reopens, the hold is + /// released, and the next joiner gets a live room again. + #[test] + fn failed_archive_reopens_room_and_releases_hold() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let idle = room.remove_peer_and_check_idle(alice).unwrap().1.unwrap(); + assert!(room.end_if_idle(idle)); + + room.clear_ended(); + room.release_idle_hold(idle); + assert!( + manager.cleanup_if_empty(community_id, channel_id), + "a reopened, empty, unheld room is evictable" + ); + let next = manager.get_or_create(community_id, channel_id); + assert!(next.add_peer("bob".into(), 3).is_ok(), "fresh room admits"); + } + + /// Drain variant: the last leaver's idle observation is abandoned (the + /// huddle must outlive the pod), so it releases its hold and the room is + /// evicted. A pre-last teardown's cleanup racing ahead of that release + /// still cannot detach the room, and a stale release after a rejoin cannot + /// lift the rejoiner's hold. + #[test] + fn released_hold_allows_eviction_but_stale_release_does_not() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 3).unwrap(); + + assert!(room.remove_peer_and_check_idle(alice).unwrap().1.is_none()); + let drained = room.remove_peer_and_check_idle(bob).unwrap().1.unwrap(); + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "pre-last cleanup must wait for the drain-owned release" + ); + + // Rejoin before the release lands: the rejoiner owns a fresh hold. + let (carol, ..) = room.add_peer("carol".into(), 3).unwrap(); + let fresh = room.remove_peer_and_check_idle(carol).unwrap().1.unwrap(); + room.release_idle_hold(drained); + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "a stale release must not lift a later observation's hold" + ); + + room.release_idle_hold(fresh); + assert!( + manager.cleanup_if_empty(community_id, channel_id), + "released hold on an empty room permits eviction" + ); + let next = manager.get_or_create(community_id, channel_id); + assert!(!Arc::ptr_eq(&room, &next), "evicted room is replaced"); + assert!( + next.add_peer("dave".into(), 2).is_ok(), + "fresh room, no pin" + ); + } }