jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 06:21:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): never auto-end a huddle on drain; hold empty rooms 20s

Browse Source 
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>
This commit is contained in:
Meli
2026-08-22 19:02:19 -04:00
parent 3f9250d713
commit 9d139889d0
2 changed files with 562 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

315
crates/buzz-relay/src/audio/handler.rs
View File

@@ -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<Arc<AppState>>,
@@ -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<u64>) {
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<CancellationToken>,
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<AppState>,
tenant: TenantContext,
channel_id: Uuid,
parent_id_for_event: Uuid,
last_leaver_pubkey: String,
room: Arc<crate::audio::room::Room>,
idle: crate::audio::room::IdleGeneration,
owner_lost: Option<CancellationToken>,
owner_generation: Option<u64>,
) {
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");
}
}

315
crates/buzz-relay/src/audio/room.rs
View File

@@ -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<u8>,
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<IdleGeneration>)> {
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"
);
}
}