Fix authoritative status and huddle removal ordering

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);
 

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<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
         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<u64> {
+        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<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
                 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<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
                 // 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,26 +1873,55 @@ 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<u64> {
     if let Ok(payload) = encode_control(&HuddleControlMsg::UnregisterPeer {
         pubkey: pubkey.to_string(),
     }) {
         let _ = stream
             .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,
             reason: HUDDLE_SESSION_ENDED,
         })
         .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<dyn RelayPeerTransport>,
+            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));
     }
 }

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 }));