diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 6e6d467d092..e365e785c85 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -785,9 +785,10 @@ async fn handle_active_audio_connection( &roster_ctrl_tx, ) => { teardown_remote_huddle(cause, channel_id, &reader_cancel, &fence); + None } _ = reader_cancel.cancelled() => { - crate::audio::join::send_clean_close(&mut stream, fenced, &pubkey).await; + crate::audio::join::send_clean_close(&mut stream, fenced, &pubkey).await } } }) @@ -861,9 +862,11 @@ async fn handle_active_audio_connection( let _ = forward_task.await; // The reader task owns the owner control stream; joining it here guarantees // its clean-close (or teardown) completes before connection cleanup returns. - if let Some(reader_task) = reader_task { - let _ = reader_task.await; - } + let remote_removal_revision = if let Some(reader_task) = reader_task { + reader_task.await.ok().flatten() + } else { + None + }; // The owner teardown watcher is cancelled by `cancel.cancel()` above (or has // already fired); join it so it settles before cleanup. if let Some(owner_teardown_task) = owner_teardown_task { @@ -882,9 +885,7 @@ async fn handle_active_audio_connection( let removal_revision = if remote_session.is_none() { removal.as_ref().map(|(delta, _)| delta.revision) } else { - // The ingress mirror's local revision is not the owner's authoritative - // ordering. Omit it rather than publishing a plausible-but-wrong value. - None + remote_removal_revision }; let should_auto_end = removal.as_ref().map(|(_, ended)| *ended).unwrap_or(false); diff --git a/crates/buzz-relay/src/audio/join.rs b/crates/buzz-relay/src/audio/join.rs index c003db9d8c3..81bea385318 100644 --- a/crates/buzz-relay/src/audio/join.rs +++ b/crates/buzz-relay/src/audio/join.rs @@ -880,11 +880,29 @@ pub enum HuddleControlMsg { /// `owner_mismatch`, `future_generation`). reason: RegisterRejection, }, - /// Non-owner → owner: the local client left; drop its remote peer. + /// Non-owner → owner: the local client left; drop its remote peer. Each + /// client owns a distinct control stream, so this can only address that + /// stream's registered admission. UnregisterPeer { /// Pubkey of the departing client. pubkey: String, }, + /// Owner → non-owner: confirms the authoritative roster revision assigned + /// when the requested admission was removed. + /// + /// Keep this variant append-only: `HuddleControlMsg` is postcard encoded, + /// so preserving every older discriminant lets a newer owner acknowledge an + /// unregister from an older ingress. The older ingress ignores the unknown + /// reply as non-terminal traffic. A newer ingress paired with an older owner + /// completes when that owner sends its existing `Goodbye` or closes the + /// stream. Either direction degrades to a revisionless LEFT without + /// breaking the control stream. + PeerUnregistered { + /// Pubkey the unregister request named. + pubkey: String, + /// Owner-monotonic revision for the removal. + roster_revision: u64, + }, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -1120,17 +1138,15 @@ impl HuddleControlAcceptor { session_id: Uuid, generation: u64, peer_id: Uuid, - ) { - let Some(room) = self.rooms.get(community, session_id) else { - return; - }; - let Some((delta, should_end)) = room.remove_peer_and_check_ended(peer_id) else { - return; - }; + ) -> Option { + let room = self.rooms.get(community, session_id)?; + let (delta, should_end) = room.remove_peer_and_check_ended(peer_id)?; + let revision = delta.revision; broadcast_peer_left(&room, delta, session_id); if should_end && self.rooms.cleanup_if_empty(community, session_id) { self.owners.release(session_id, generation); } + Some(revision) } /// Serve register/unregister frames for one non-owner pod's stream. @@ -1330,12 +1346,24 @@ impl HuddleControlAcceptor { HuddleControlMsg::UnregisterPeer { pubkey } => { if let Some(peer_id) = registered.remove(&pubkey) { if let Some(community_id) = stream_community { - self.remove_remote_peer( + if let Some(roster_revision) = self.remove_remote_peer( CommunityId::from_uuid(community_id), session_id, fenced.generation, peer_id, - ); + ) { + stream + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control( + &HuddleControlMsg::PeerUnregistered { + pubkey, + roster_revision, + }, + )?, + }) + .await?; + } } } } @@ -1356,6 +1384,7 @@ impl HuddleControlAcceptor { // Owner→non-owner replies never arrive on the owner's accept // side; a peer sending one is a protocol violation. HuddleControlMsg::PeerRegistered { .. } + | HuddleControlMsg::PeerUnregistered { .. } | HuddleControlMsg::RosterSnapshot { .. } | HuddleControlMsg::RosterDelta { .. } | HuddleControlMsg::RegisterRejected { .. } => { @@ -1655,7 +1684,12 @@ pub async fn read_owner_control( return HuddleTeardownCause::StreamClosed; } } - Ok(_) => {} + Ok(HuddleControlMsg::PeerRegistered { .. }) + | Ok(HuddleControlMsg::PeerUnregistered { .. }) + | Ok(HuddleControlMsg::RegisterRejected { .. }) + | Ok(HuddleControlMsg::RegisterPeer { .. }) + | Ok(HuddleControlMsg::UnregisterPeer { .. }) + | Ok(HuddleControlMsg::RosterResync) => {} Err(e) => debug!(owner_stream_error = %e, "invalid huddle owner control"), }, Ok(Some(_)) => continue, @@ -1839,12 +1873,18 @@ impl RemoteHuddleSession { } /// Unregister the client from the owner and close the control stream cleanly. +/// Returns the owner's authoritative removal revision when the acknowledgement +/// arrives before the bounded teardown deadline. /// /// Called on a *local-client-initiated* disconnect (the reader task's cancel /// branch), so the owner drops the remote peer and stops fanning media back. -/// Best-effort: teardown never blocks connection cleanup, and a `Goodbye` -/// already received from the owner makes this a no-op the owner ignores. -pub async fn send_clean_close(stream: &mut MeshStream, fenced: FencedHeader, pubkey: &str) { +/// Best-effort: teardown never blocks indefinitely, and a `Goodbye` already +/// received from the owner makes this a no-op the owner ignores. +pub async fn send_clean_close( + stream: &mut MeshStream, + fenced: FencedHeader, + pubkey: &str, +) -> Option { if let Ok(payload) = encode_control(&HuddleControlMsg::UnregisterPeer { pubkey: pubkey.to_string(), }) { @@ -1852,6 +1892,28 @@ pub async fn send_clean_close(stream: &mut MeshStream, fenced: FencedHeader, pub .send_frame(MeshStreamFrame::Data { fenced, payload }) .await; } + let roster_revision = tokio::time::timeout(Duration::from_secs(1), async { + loop { + match stream.recv_frame().await { + Ok(Some(MeshStreamFrame::Data { payload, .. })) => { + if let Ok(HuddleControlMsg::PeerUnregistered { + pubkey: removed, + roster_revision, + }) = decode_control(&payload) + { + if removed == pubkey { + return Some(roster_revision); + } + } + } + Ok(Some(_)) => continue, + Ok(None) | Err(_) => return None, + } + } + }) + .await + .ok() + .flatten(); let _ = stream .send_frame(MeshStreamFrame::Goodbye { fenced, @@ -1859,6 +1921,7 @@ pub async fn send_clean_close(stream: &mut MeshStream, fenced: FencedHeader, pub }) .await; let _ = stream.finish(); + roster_revision } /// Build the media datagram a non-owner ships to the owner for one client @@ -2158,6 +2221,10 @@ mod tests { epoch: 0, }), }, + HuddleControlMsg::PeerUnregistered { + pubkey: "abc123".into(), + roster_revision: 2, + }, HuddleControlMsg::RosterResync, HuddleControlMsg::RegisterRejected { pubkey: "abc123".into(), @@ -2376,6 +2443,81 @@ mod tests { served.await.unwrap().unwrap(); } + #[tokio::test] + async fn unregister_peer_acknowledges_owner_removal_revision() { + let owner_rt = rt(1); + let from = rt(2); + let session_id = Uuid::new_v4(); + let fenced = fenced_owned_by(owner_rt, session_id); + let rooms = Arc::new(AudioRoomManager::new()); + let acceptor = HuddleControlAcceptor::new( + Arc::clone(&rooms), + Arc::new(NullTransport) as Arc, + Arc::new(FakeDir::default()), + owner_rt, + Arc::new(HuddleOwnerRegistry::new()), + ); + let (owner_stream, mut client) = stream_pair(); + let hello = huddle_hello(from, fenced); + let served = + tokio::spawn(async move { acceptor.accept_inbound(from, hello, owner_stream).await }); + + client + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::RegisterPeer { + community_id: *community().as_uuid(), + pubkey: "client-a".into(), + protocol_version: 2, + }) + .unwrap(), + }) + .await + .unwrap(); + let registered = match client.recv_frame().await.unwrap().unwrap() { + MeshStreamFrame::Data { payload, .. } => decode_control(&payload).unwrap(), + other => panic!("expected PeerRegistered Data, got {other:?}"), + }; + assert!(matches!( + registered, + HuddleControlMsg::PeerRegistered { ref pubkey, .. } if pubkey == "client-a" + )); + + client + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::UnregisterPeer { + pubkey: "client-a".into(), + }) + .unwrap(), + }) + .await + .unwrap(); + let acknowledged = loop { + match client.recv_frame().await.unwrap().unwrap() { + MeshStreamFrame::Data { payload, .. } => { + let message = decode_control(&payload).unwrap(); + if matches!(message, HuddleControlMsg::PeerUnregistered { .. }) { + break message; + } + } + other => panic!("expected unregister acknowledgement Data, got {other:?}"), + } + }; + assert_eq!( + acknowledged, + HuddleControlMsg::PeerUnregistered { + pubkey: "client-a".into(), + roster_revision: 2, + } + ); + + client.finish().unwrap(); + drop(client); + served.await.unwrap().unwrap(); + assert!(rooms.get(community(), session_id).is_none()); + } + #[tokio::test] async fn abnormal_control_stream_close_fans_out_remote_leave() { let owner_rt = rt(1); @@ -3172,28 +3314,41 @@ mod tests { ); } - /// The client-initiated clean close emits `UnregisterPeer` then - /// `Goodbye(SessionEnded)` on the owner's control stream, in that order. + /// The client-initiated clean close emits `UnregisterPeer`, waits for the + /// owner's revision acknowledgement, then sends `Goodbye(SessionEnded)`. #[tokio::test] - async fn clean_close_sends_unregister_then_goodbye() { + async fn clean_close_returns_owner_removal_revision() { let fenced = fenced_owned_by(rt(2), Uuid::new_v4()); let (mut owner, mut client) = stream_pair(); - send_clean_close(&mut client, fenced, "client-a").await; + let close = + tokio::spawn(async move { send_clean_close(&mut client, fenced, "client-a").await }); match owner.recv_frame().await.unwrap().unwrap() { MeshStreamFrame::Data { payload, .. } => assert_eq!( decode_control(&payload).unwrap(), HuddleControlMsg::UnregisterPeer { - pubkey: "client-a".into() + pubkey: "client-a".into(), } ), other => panic!("expected UnregisterPeer Data, got {other:?}"), } + owner + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::PeerUnregistered { + pubkey: "client-a".into(), + roster_revision: 2, + }) + .unwrap(), + }) + .await + .unwrap(); match owner.recv_frame().await.unwrap().unwrap() { MeshStreamFrame::Goodbye { reason, .. } => { assert_eq!(reason, GoodbyeReason::SessionEnded) } other => panic!("expected Goodbye(SessionEnded), got {other:?}"), } + assert_eq!(close.await.unwrap(), Some(2)); } } diff --git a/desktop/src/features/huddle/lib/huddlePresence.test.mjs b/desktop/src/features/huddle/lib/huddlePresence.test.mjs index 9432c0a8ffb..1132167e76b 100644 --- a/desktop/src/features/huddle/lib/huddlePresence.test.mjs +++ b/desktop/src/features/huddle/lib/huddlePresence.test.mjs @@ -147,6 +147,31 @@ test("orders same-second reconnect events by roster revision", () => { assert.equal(result.has(BOB), true); }); +test("orders same-second remote leave after its join by roster revision", () => { + const result = reconstructHuddlePresence( + [ + event({ id: "1", kind: 48100, createdAt: 1 }), + participantEvent({ + id: "z", + kind: 48101, + admissionId: "remote", + rosterRevision: 1, + createdAt: 2, + }), + participantEvent({ + id: "a", + kind: 48102, + admissionId: "remote", + rosterRevision: 2, + createdAt: 2, + }), + ], + RELAY, + ); + + assert.equal(result.has(BOB), false); +}); + test("ignores an older replay for the same admission", () => { const tracker = new HuddlePresenceTracker(RELAY); tracker.apply(event({ id: "1", kind: 48100 })); diff --git a/desktop/src/features/user-status/hooks.test.mjs b/desktop/src/features/user-status/hooks.test.mjs index ae9b383a924..23c1c44c5d5 100644 --- a/desktop/src/features/user-status/hooks.test.mjs +++ b/desktop/src/features/user-status/hooks.test.mjs @@ -6,6 +6,7 @@ import { fetchUserStatusLookup, readCurrentUserStatusLookup, USER_STATUS_AUTHOR_CHUNK_SIZE, + USER_STATUS_FETCH_CONCURRENCY, } from "./hooks.ts"; function statusEvent(pubkey, createdAt = 1) { @@ -49,6 +50,40 @@ test("does not let a pending history result replace a newer live status", async assert.equal(lookup[pubkey]?.eventId, "live"); }); +test("preserves a cached status updated while authoritative history is pending", async () => { + const pubkey = "a".repeat(64); + const cached = { + [pubkey]: { + text: "Cached", + emoji: "💬", + updatedAt: 1, + eventId: "cached", + }, + }; + let current = cached; + let resolveHistory; + const lookupPromise = fetchUserStatusLookup( + [pubkey], + () => new Promise((resolve) => (resolveHistory = resolve)), + () => current, + ); + await new Promise((resolve) => setImmediate(resolve)); + current = { + [pubkey]: { + text: "Live", + emoji: "💬", + updatedAt: 2, + eventId: "live", + }, + }; + resolveHistory([]); + + const lookup = await lookupPromise; + + assert.equal(lookup[pubkey]?.text, "Live"); + assert.equal(lookup[pubkey]?.eventId, "live"); +}); + test("seeds overlapping status lookups with the newest cached version", () => { const pubkey = "a".repeat(64); const queryClient = new QueryClient(); @@ -76,6 +111,57 @@ test("seeds overlapping status lookups with the newest cached version", () => { queryClient.clear(); }); +test("clears an unchanged cached status absent from authoritative history", async () => { + const pubkey = "a".repeat(64); + const cached = { + [pubkey]: { + text: "Deleted elsewhere", + emoji: "💬", + updatedAt: 1, + eventId: "cached", + }, + }; + + const lookup = await fetchUserStatusLookup( + [pubkey], + async () => [], + () => cached, + ); + + assert.equal(lookup[pubkey], null); +}); + +test("bounds concurrent status page fetches", async () => { + const pubkeys = Array.from( + { + length: + USER_STATUS_AUTHOR_CHUNK_SIZE * (USER_STATUS_FETCH_CONCURRENCY + 1), + }, + (_, index) => index.toString(16).padStart(64, "0"), + ); + let inFlight = 0; + let peakInFlight = 0; + const releases = []; + + const lookupPromise = fetchUserStatusLookup(pubkeys, async () => { + inFlight += 1; + peakInFlight = Math.max(peakInFlight, inFlight); + await new Promise((resolve) => releases.push(resolve)); + inFlight -= 1; + return []; + }); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(peakInFlight, USER_STATUS_FETCH_CONCURRENCY); + while (releases.length) { + for (const release of releases.splice(0)) { + release(); + } + await new Promise((resolve) => setImmediate(resolve)); + } + await lookupPromise; +}); + test("fetches every current status when the author set exceeds one relay page", async () => { const pubkeys = Array.from( { length: USER_STATUS_AUTHOR_CHUNK_SIZE + 7 }, diff --git a/desktop/src/features/user-status/hooks.ts b/desktop/src/features/user-status/hooks.ts index 1c900ebf8a7..955938d1ee6 100644 --- a/desktop/src/features/user-status/hooks.ts +++ b/desktop/src/features/user-status/hooks.ts @@ -7,6 +7,7 @@ import { } from "@tanstack/react-query"; import type { UserStatusInput } from "@/features/user-status/types"; +import { collectWithConcurrency } from "@/shared/api/concurrency"; import { relayClient } from "@/shared/api/relayClient"; import type { RelayEvent, @@ -50,6 +51,20 @@ function newerUserStatus( : candidate; } +function sameUserStatus( + left: UserStatusCacheEntry | undefined, + right: UserStatusCacheEntry | undefined, +): boolean { + if (!left || !right) return left == null && right == null; + return ( + left.text === right.text && + left.emoji === right.emoji && + left.updatedAt === right.updatedAt && + left.eventId === right.eventId && + left.expiresAt === right.expiresAt + ); +} + function hiddenUserStatus(version: UserStatusVersion): UserStatus { return { text: "", @@ -196,6 +211,7 @@ export const USER_STATUS_REFETCH_INTERVAL_MS = 120_000; * The live subscription (setQueriesData) is the primary freshness path. */ export const USER_STATUS_FOCUS_STALE_TIME_MS = 5 * 60_000; export const USER_STATUS_AUTHOR_CHUNK_SIZE = 1_000; +export const USER_STATUS_FETCH_CONCURRENCY = 4; /** Focus-refetch policy for the user-status query; consumed by focusRefetchPolicy.test.mjs. */ export const userStatusFocusRefetchPolicy = { @@ -223,21 +239,27 @@ export async function fetchUserStatusLookup( normalizedAuthors.slice(index, index + USER_STATUS_AUTHOR_CHUNK_SIZE), ); } - const pages = await Promise.all( - chunks.map((authors) => + const cachedBeforeFetch = readCurrentLookup(); + const pages = await collectWithConcurrency( + chunks, + USER_STATUS_FETCH_CONCURRENCY, + (authors) => fetchEvents({ kinds: [KIND_USER_STATUS], authors, "#d": ["general"], limit: authors.length, }), - ), ); const currentLookup = readCurrentLookup(); const lookup: UserStatusLookup = {}; for (const pubkey of normalizedAuthors) { - lookup[pubkey] = currentLookup[pubkey] ?? null; + const current = currentLookup[pubkey]; + const cached = cachedBeforeFetch[pubkey]; + lookup[pubkey] = !sameUserStatus(current, cached) + ? (current ?? null) + : null; } const latestEvents = new Map(); for (const event of pages.flat()) {