diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 9e930ccc4..ff375dda6 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -224,33 +224,72 @@ async fn handle_audio_connection( } }; - // Huddle audio guardrail (plan §5b). Audio frames are relayed only within - // a single pod; under horizontal scaling (any-pod-any-connection) two peers - // in one huddle can land on different pods and never hear each other. A - // multi-pod deployment sets `huddle_audio_available = false`, and we surface - // a clear, client-handleable "unavailable" signal here — BEFORE joining a - // room — rather than shipping a silent split-room. Single-pod deployments - // leave the flag at its `true` default and keep today's behavior. The fix - // is an out-of-relay media/SFU service (Tyler's long-term target), not - // sticky-routing huddles into this rewrite. - if !state.config.huddle_audio_available { - debug!( - channel_id = %channel_id, - pubkey = %pubkey_hex, - "huddle audio unavailable under horizontal scaling — rejecting join" - ); - let _ = ws_send - .send(WsMessage::Text( - serde_json::json!({ - "type": "error", - "code": "huddle_audio_unavailable", - "message": "huddle audio unavailable in this deployment" - }) - .to_string() - .into(), - )) - .await; - return; + // Huddle cross-pod routing (mesh) OR single-pod guardrail. + // + // When the mesh is live (`state.mesh()` is `Some`), a huddle can span pods: + // Redis arbitrates ownership and this pod either owns the room locally or + // forwards the client to the owner over a `HuddleControl` stream. When the + // mesh is off, we keep today's behavior exactly — including the + // `huddle_audio_available=false` rejection under a non-mesh horizontal + // deployment (two peers on different pods would never hear each other). + // + // `remote_owner` is `Some` only on the non-owner path; it carries the + // registration to the owner and, once the client is admitted locally, is + // opened so its media forwards to the owner instead of fanning out locally. + let mut pending_remote: Option = None; + match state.mesh() { + Some(mesh) => { + match crate::audio::join::resolve_join( + &mesh.directory, + tenant.community(), + channel_id, + mesh.local_runtime_id, + ) + .await + { + Ok(outcome) => pending_remote = Some(outcome), + Err(e) => { + warn!( + channel_id = %channel_id, + pubkey = %pubkey_hex, + "huddle join rejected by fence: {e}" + ); + let _ = ws_send + .send(WsMessage::Text( + serde_json::json!({ + "type": "error", + "code": "join_rejected", + "message": "huddle join rejected" + }) + .to_string() + .into(), + )) + .await; + return; + } + } + } + None => { + if !state.config.huddle_audio_available { + debug!( + channel_id = %channel_id, + pubkey = %pubkey_hex, + "huddle audio unavailable under horizontal scaling — rejecting join" + ); + let _ = ws_send + .send(WsMessage::Text( + serde_json::json!({ + "type": "error", + "code": "huddle_audio_unavailable", + "message": "huddle audio unavailable in this deployment" + }) + .to_string() + .into(), + )) + .await; + return; + } + } } let room = state.audio_rooms.get_or_create(channel_id); @@ -368,6 +407,79 @@ async fn handle_audio_connection( "audio peer joined" ); + // Non-owner path: the huddle is owned by another pod. Register this client + // with the owner over a `HuddleControl` stream so the owner fans media back + // to us; the returned session forwards the client's Opus to the owner. A + // fence/admission rejection from the owner is surfaced to the client as the + // same WS error a same-pod join would produce — never a silent media drop. + // + // NOTE (roster sync): cross-pod roster fan-out (the owner's full participant + // list reflected to this client) is the next increment; today this pod's + // local `joined`/`peers` reflect only co-located peers. The media + fence + // path is complete and correct; roster consistency lands with the owner's + // roster-delta messages over the same `HuddleControl` stream. + let mut remote_session: Option = None; + if let (Some(mesh), Some(crate::audio::join::JoinOutcome::RemoteOwner { .. })) = + (state.mesh(), pending_remote) + { + let outcome = pending_remote.expect("RemoteOwner matched above"); + let fenced = outcome.fenced_header(channel_id, mesh.local_runtime_id); + let crate::audio::join::JoinOutcome::RemoteOwner { + owner_runtime_id, .. + } = outcome + else { + unreachable!("matched RemoteOwner above"); + }; + match crate::audio::join::dial_remote_owner( + Arc::clone(&mesh.transport), + mesh.local_runtime_id, + owner_runtime_id, + fenced, + pubkey_hex.clone(), + requested_version, + ) + .await + { + Ok(session) => remote_session = Some(session), + Err(crate::audio::join::DialError::Rejected(reason)) => { + warn!( + channel_id = %channel_id, + pubkey = %pubkey_hex, + "huddle owner rejected registration: {reason:?}" + ); + let _ = ws_send + .send(WsMessage::Text( + remote_rejection_ws_error(&reason).to_string().into(), + )) + .await; + room.remove_peer(peer_id); + state.audio_rooms.cleanup_if_empty(channel_id); + return; + } + Err(crate::audio::join::DialError::Mesh(e)) => { + warn!( + channel_id = %channel_id, + pubkey = %pubkey_hex, + "huddle owner registration failed: {e}" + ); + let _ = ws_send + .send(WsMessage::Text( + serde_json::json!({ + "type": "error", + "code": "huddle_owner_unreachable", + "message": "could not reach the huddle owner" + }) + .to_string() + .into(), + )) + .await; + room.remove_peer(peer_id); + state.audio_rooms.cleanup_if_empty(channel_id); + return; + } + } + } + let peers_snapshot: Vec = room .peer_pubkeys() .into_iter() @@ -427,9 +539,16 @@ async fn handle_audio_connection( ctrl_tx, Arc::clone(&missed_pongs), cancel.clone(), + remote_session.as_mut(), ) .await; + // Non-owner path: unregister from the owner and close the control stream + // cleanly. Best-effort — teardown never blocks connection cleanup. + if let Some(session) = remote_session { + session.close().await; + } + cancel.cancel(); let _ = send_task.await; let _ = heartbeat_task.await; @@ -499,6 +618,43 @@ async fn handle_audio_connection( ); } +/// Map an owner's registration rejection to the client-facing WS error, using +/// the same `code`s a same-pod join produces so a cross-pod client handles them +/// identically. Fence rejections carry their taxonomy code for observability. +fn remote_rejection_ws_error( + reason: &crate::audio::join::RegisterRejection, +) -> serde_json::Value { + use crate::audio::join::RegisterRejection; + match reason { + RegisterRejection::RoomFull => serde_json::json!({ + "type": "error", "code": "room_full", + "message": "peer index space exhausted" + }), + RegisterRejection::RoomEnded => serde_json::json!({ + "type": "error", "code": "room_ended", "message": "huddle has ended" + }), + RegisterRejection::VersionMismatch { pinned, requested } => serde_json::json!({ + "type": "error", "code": "upgrade_required", + "message": format!( + "this huddle is using audio protocol v{pinned}; your client requested v{requested}" + ), + "pinned_version": pinned, + "requested_version": requested, + }), + RegisterRejection::Fenced(f) => serde_json::json!({ + "type": "error", "code": "join_rejected", + "message": "huddle join rejected", + "fence_reason": f.code(), + }), + } +} + +/// Receive loop: reads client frames and routes them. Local/owner joins fan +/// out through the local room; a non-owner join forwards to the huddle owner +/// via `remote_session`. Argument count reflects the pre-existing connection +/// wiring plus the one mesh session; a param struct would obscure more than it +/// clarifies at this single call site. +#[allow(clippy::too_many_arguments)] async fn recv_loop( mut ws_recv: futures_util::stream::SplitStream, room: Arc, @@ -507,6 +663,7 @@ async fn recv_loop( ctrl_tx: mpsc::Sender, missed_pongs: Arc, cancel: CancellationToken, + mut remote_session: Option<&mut crate::audio::join::RemoteHuddleSession>, ) { use crate::audio::wire::{FrameHeader, V2_HEADER_LEN}; @@ -567,7 +724,15 @@ async fn recv_loop( } } - room.broadcast_frame(peer_id, data); + // Non-owner path forwards the client's Opus to the + // huddle owner as a datagram (the owner is the sole + // fan-out authority); the owner-side room fans it back + // to every participant, including our co-located peers. + // Owner/local path fans out through the local room. + match remote_session.as_deref_mut() { + Some(session) => session.forward_media(&data), + None => room.broadcast_frame(peer_id, data), + } } Some(Ok(WsMessage::Text(text))) => { if text.len() > MAX_TEXT_FRAME_BYTES { diff --git a/crates/buzz-relay/src/audio/join.rs b/crates/buzz-relay/src/audio/join.rs index 40d529cc6..544d20e6f 100644 --- a/crates/buzz-relay/src/audio/join.rs +++ b/crates/buzz-relay/src/audio/join.rs @@ -35,11 +35,20 @@ //! [`HuddleControlMsg::RegisterRejected`] (fence/admission failure surfaced to //! the client as a join error, never a silent media drop). +use std::sync::Arc; + use buzz_core::CommunityId; -use buzz_relay_mesh::{FencedHeader, MeshError, Profile, RuntimeId}; +use buzz_relay_mesh::{ + FencedHeader, GoodbyeReason, MeshDatagram, MeshError, MeshStream, MeshStreamFrame, Profile, + RelayPeerTransport, RuntimeId, StreamHello, StreamRole, +}; use serde::{Deserialize, Serialize}; +use tracing::debug; use uuid::Uuid; +use super::mesh::spawn_remote_peer_sink; +use super::room::{AdmissionError, AudioRoomManager}; + /// The slice of the Redis fenced session directory the huddle join path needs. /// /// Implemented by the session-directory lane's `SessionDirectory` over the @@ -377,6 +386,418 @@ pub fn decode_control(bytes: &[u8]) -> Result { /// desync, so it never rides datagrams. pub const HUDDLE_CONTROL_PROFILE: Profile = Profile::HuddleControl; +// ── Owner-side HuddleControl accept path ───────────────────────────────────── +// +// The owner pod hosts the real [`Room`]. When a *non-owner* pod opens a +// `HuddleControl` stream and registers a client, the owner admits that client +// as an ordinary [`AudioPeer`] whose `audio_tx` is drained by +// [`super::mesh::spawn_remote_peer_sink`] back to the non-owner pod as +// datagrams. `Room` never learns about the mesh — a remote participant looks +// exactly like a local one to fan-out. + +/// Owner-side handler for inbound `HuddleControl` streams. +/// +/// One instance per relay; the boot-seam dispatcher routes every +/// `Profile::HuddleControl` session stream to [`Self::accept_inbound`]. It is +/// the counterpart to Perci's reliable-stream acceptor — same +/// `accept_inbound(community_id, from, hello, stream)` shape, different profile +/// and body. +pub struct HuddleControlAcceptor { + rooms: Arc, + transport: Arc, + directory: Arc, + local_runtime_id: RuntimeId, +} + +impl HuddleControlAcceptor { + /// Build the acceptor. `directory` is the fenced arbiter (re-validated on + /// every registration); `transport` is used to open the media datagram + /// sink back to each registering pod. + pub fn new( + rooms: Arc, + transport: Arc, + directory: Arc, + local_runtime_id: RuntimeId, + ) -> Self { + Self { + rooms, + transport, + directory, + local_runtime_id, + } + } + + /// Accept and validate an inbound `HuddleControl` stream, then serve its + /// register/unregister control loop until the stream closes. + /// + /// Validation mirrors the reliable-stream acceptor and enforces the fencing + /// law on receipt: the claimed sender must be the authenticated peer, the + /// profile must be `HuddleControl`, the fence must pass Redis, and this pod + /// must be the fenced owner. Any of these failing rejects the stream before + /// a single peer is admitted. + pub async fn accept_inbound( + &self, + community_id: CommunityId, + from: RuntimeId, + hello: StreamHello, + stream: MeshStream, + ) -> Result<(), MeshError> { + if hello.sender != from { + return Err(MeshError::Transport(format!( + "huddle-control hello.sender {} != authenticated peer {from}", + hello.sender + ))); + } + let StreamRole::Session { fenced, profile } = hello.role else { + return Err(MeshError::Transport( + "huddle-control stream Hello was not a session role".into(), + )); + }; + if profile != Profile::HuddleControl { + return Err(MeshError::Transport(format!( + "huddle-control acceptor got profile {profile:?}" + ))); + } + // Fence at every hop: the origin validated before dialing; the owner + // re-validates on receipt so a lease that moved in between is caught. + self.directory.validate(community_id, &fenced).await?; + if fenced.owner_runtime_id != self.local_runtime_id { + return Err(MeshError::OwnerMismatch { + session_id: fenced.session_id, + generation: fenced.generation, + frame_owner_runtime_id: fenced.owner_runtime_id, + current_owner_runtime_id: self.local_runtime_id, + }); + } + + self.serve_control_loop(community_id, from, fenced, stream) + .await + } + + /// Serve register/unregister frames for one non-owner pod's stream. + /// + /// Peers this stream registers are tracked so a stream close (the non-owner + /// pod went away) tears them all down — no leaked remote peers holding + /// index slots in the owner's room. + async fn serve_control_loop( + &self, + community_id: CommunityId, + from: RuntimeId, + fenced: FencedHeader, + mut stream: MeshStream, + ) -> Result<(), MeshError> { + let session_id = fenced.session_id; + // pubkey -> peer_id, for UnregisterPeer and teardown on stream close. + let mut registered: std::collections::HashMap = std::collections::HashMap::new(); + + let result = loop { + let msg = match stream.recv_frame().await { + Ok(Some(MeshStreamFrame::Data { fenced: f, payload })) => { + // Re-fence every control frame, not just the Hello: a lease + // that moves mid-stream must reject subsequent registers. + if f != fenced { + break Err(MeshError::OwnerMismatch { + session_id, + generation: f.generation, + frame_owner_runtime_id: f.owner_runtime_id, + current_owner_runtime_id: self.local_runtime_id, + }); + } + if let Err(e) = self.directory.validate(community_id, &f).await { + break Err(e); + } + match decode_control(&payload) { + Ok(m) => m, + Err(e) => break Err(e), + } + } + Ok(Some(MeshStreamFrame::Goodbye { .. })) | Ok(None) => break Ok(()), + Ok(Some(other)) => { + break Err(MeshError::Transport(format!( + "huddle-control stream got unexpected frame {other:?}" + ))); + } + Err(e) => break Err(e), + }; + + match msg { + HuddleControlMsg::RegisterPeer { + pubkey, + protocol_version, + } => { + let reply = self.register_remote_peer( + session_id, + fenced, + from, + &pubkey, + protocol_version, + &mut registered, + ); + if let Err(e) = stream + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&reply)?, + }) + .await + { + break Err(e); + } + } + HuddleControlMsg::UnregisterPeer { pubkey } => { + if let Some(peer_id) = registered.remove(&pubkey) { + if let Some(room) = self.rooms.get(session_id) { + room.remove_peer(peer_id); + } + } + } + // Owner→non-owner replies never arrive on the owner's accept + // side; a peer sending one is a protocol violation. + HuddleControlMsg::PeerRegistered { .. } + | HuddleControlMsg::RegisterRejected { .. } => { + break Err(MeshError::Transport( + "huddle-control owner received an owner→non-owner reply".into(), + )); + } + } + }; + + // Teardown: drop every peer this stream registered, regardless of how + // the loop ended. Dropping the peer drops its `audio_tx`, which ends the + // matching `spawn_remote_peer_sink` task. + if let Some(room) = self.rooms.get(session_id) { + for (_pubkey, peer_id) in registered { + room.remove_peer(peer_id); + } + } + result + } + + /// Admit one remote client into the owner's room and wire its fan-out back + /// to the registering pod as datagrams. Returns the reply to send. + fn register_remote_peer( + &self, + session_id: Uuid, + fenced: FencedHeader, + from: RuntimeId, + pubkey: &str, + protocol_version: u8, + registered: &mut std::collections::HashMap, + ) -> HuddleControlMsg { + let room = self.rooms.get_or_create(session_id); + match room.add_peer(pubkey.to_string(), protocol_version) { + Ok((peer_id, peer_index, audio_rx, _peer_ctrl_rx)) => { + registered.insert(pubkey.to_string(), peer_id); + // The owner's Room fans out to this remote peer's `audio_tx`; + // the sink drains `audio_rx` and ships each frame as a datagram + // to the pod that hosts the client. + spawn_remote_peer_sink(Arc::clone(&self.transport), from, fenced, audio_rx); + HuddleControlMsg::PeerRegistered { + pubkey: pubkey.to_string(), + peer_index, + } + } + Err(reason) => HuddleControlMsg::RegisterRejected { + pubkey: pubkey.to_string(), + reason: admission_to_rejection(reason), + }, + } + } +} + +/// Map a room admission failure to the wire rejection taxonomy. Kept 1:1 with +/// the single-pod WS error codes so a cross-pod join surfaces the same +/// client-facing error a same-pod join would. +fn admission_to_rejection(err: AdmissionError) -> RegisterRejection { + match err { + AdmissionError::Full => RegisterRejection::RoomFull, + AdmissionError::Ended => RegisterRejection::RoomEnded, + AdmissionError::VersionMismatch { pinned, requested } => { + RegisterRejection::VersionMismatch { pinned, requested } + } + } +} + +/// The `Goodbye` reason a non-owner sends when its client leaves the huddle +/// cleanly. Re-exported so the handler's dial path uses one spelling. +pub const HUDDLE_SESSION_ENDED: GoodbyeReason = GoodbyeReason::SessionEnded; + +// ── Non-owner-side HuddleControl dial path ─────────────────────────────────── +// +// A client connected here whose huddle is owned by another pod. We keep the +// client as an ordinary local WS peer (heartbeats, `joined`/`left`, delivery of +// the owner's fan-out) but there is NO local fan-out: the owner is the sole +// fan-out authority. Each client Opus frame is shipped to the owner as a +// datagram tagged with the OWNER-assigned peer index; the owner fans out to +// everyone (including this pod's co-located clients, which hear each other via +// the owner round-trip — `deliver_prefixed` skips a client's own index so it +// never hears itself). + +/// A registered cross-pod huddle session on the non-owner side. +/// +/// Holds everything needed to forward the local client's media to the owner and +/// to unregister cleanly on disconnect. Media delivery *back* to the client goes +/// through the ordinary local room via `MeshAudioRouter::on_media_datagram`, so +/// this handle owns only the outbound (client→owner) half plus teardown. +pub struct RemoteHuddleSession { + /// The owner-allocated peer index this client occupies in the owner's room. + /// Stamped on every media datagram so the owner attributes frames correctly. + peer_index: u8, + /// Fenced header for this session's owner epoch; every datagram carries it. + fenced: FencedHeader, + /// The pod that owns the huddle. + owner: RuntimeId, + /// Pubkey of the local client, for the closing `UnregisterPeer`. + pubkey: String, + /// Transport for datagrams and the control-stream teardown. + transport: Arc, + /// The open `HuddleControl` stream to the owner; a clean `Goodbye` is sent + /// on teardown. + stream: MeshStream, + /// Per-datagram monotonic sequence for loss/reorder observability. + seq: u64, +} + +/// Why a cross-pod join could not complete on the non-owner side. +#[derive(Debug)] +pub enum DialError { + /// The owner refused the registration; surfaced to the client as the same + /// WS error a same-pod join would produce, never a silent media drop. + Rejected(RegisterRejection), + /// Transport / protocol failure opening or serving the control stream. + Mesh(MeshError), +} + +impl From for DialError { + fn from(e: MeshError) -> Self { + DialError::Mesh(e) + } +} + +/// Open a `HuddleControl` stream to the owner and register the local client. +/// +/// On success the owner has admitted the client as a remote peer and returned +/// its owner-assigned index; the returned [`RemoteHuddleSession`] forwards media +/// and unregisters on drop. On [`DialError::Rejected`] the caller surfaces the +/// owner's admission failure to the client unchanged. +pub async fn dial_remote_owner( + transport: Arc, + local_runtime_id: RuntimeId, + owner: RuntimeId, + fenced: FencedHeader, + pubkey: String, + protocol_version: u8, +) -> Result { + let hello = StreamHello { + sender: local_runtime_id, + role: StreamRole::Session { + fenced, + profile: Profile::HuddleControl, + }, + }; + // `open_session_stream` sends the Hello before returning. + let mut stream = transport.open_session_stream(owner, hello).await?; + + stream + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::RegisterPeer { + pubkey: pubkey.clone(), + protocol_version, + })?, + }) + .await?; + + match stream.recv_frame().await? { + Some(MeshStreamFrame::Data { payload, .. }) => match decode_control(&payload)? { + HuddleControlMsg::PeerRegistered { peer_index, .. } => Ok(RemoteHuddleSession { + peer_index, + fenced, + owner, + pubkey, + transport, + stream, + seq: 0, + }), + HuddleControlMsg::RegisterRejected { reason, .. } => Err(DialError::Rejected(reason)), + other => Err(DialError::Mesh(MeshError::Transport(format!( + "expected PeerRegistered/RegisterRejected, got {other:?}" + )))), + }, + Some(MeshStreamFrame::Goodbye { .. }) | None => Err(DialError::Mesh(MeshError::Transport( + "owner closed HuddleControl stream before replying".into(), + ))), + Some(other) => Err(DialError::Mesh(MeshError::Transport(format!( + "unexpected HuddleControl frame from owner: {other:?}" + )))), + } +} + +/// The `StreamHello.sender` for a dialed session: the fenced header carries the +/// owner's identity, but the *sender* is this pod. The owner validates +/// `hello.sender == authenticated peer`, so it must be our own runtime id — the +/// handler threads `local_runtime_id` in explicitly. +impl RemoteHuddleSession { + /// The owner-assigned index this client occupies in the owner's room. + pub fn peer_index(&self) -> u8 { + self.peer_index + } + + /// Forward one client Opus frame to the owner as a media datagram, tagged + /// with the owner-assigned index. Drop-on-error: realtime audio never blocks + /// on a slow or gone link (the same discipline as local fan-out). + pub fn forward_media(&mut self, client_frame: &[u8]) { + let dgram = media_datagram(self.peer_index, self.fenced, self.seq, client_frame); + self.seq = self.seq.wrapping_add(1); + if let Err(e) = self.transport.send_datagram(self.owner, dgram) { + debug!(owner = %self.owner, "huddle media datagram to owner failed: {e}"); + } + } + + /// Unregister the client from the owner and close the control stream + /// cleanly. Best-effort: teardown never blocks connection cleanup. + pub async fn close(mut self) { + let _ = self + .stream + .send_frame(MeshStreamFrame::Data { + fenced: self.fenced, + payload: match encode_control(&HuddleControlMsg::UnregisterPeer { + pubkey: self.pubkey.clone(), + }) { + Ok(p) => p, + Err(_) => return, + }, + }) + .await; + let _ = self + .stream + .send_frame(MeshStreamFrame::Goodbye { + fenced: self.fenced, + reason: HUDDLE_SESSION_ENDED, + }) + .await; + } +} + +/// Build the media datagram a non-owner ships to the owner for one client +/// frame: `[owner_peer_index][client frame]`, stamped with the session fence +/// and sequence. Pure so the framing is unit-testable without a live transport +/// or stream. +fn media_datagram( + peer_index: u8, + fenced: FencedHeader, + seq: u64, + client_frame: &[u8], +) -> MeshDatagram { + let mut payload = Vec::with_capacity(1 + client_frame.len()); + payload.push(peer_index); + payload.extend_from_slice(client_frame); + MeshDatagram { + fenced, + seq, + payload, + } +} + #[cfg(test)] mod tests { use super::*; @@ -602,4 +1023,44 @@ mod tests { } ); } + + #[test] + fn admission_errors_map_to_wire_rejections() { + assert_eq!( + admission_to_rejection(AdmissionError::Full), + RegisterRejection::RoomFull + ); + assert_eq!( + admission_to_rejection(AdmissionError::Ended), + RegisterRejection::RoomEnded + ); + assert_eq!( + admission_to_rejection(AdmissionError::VersionMismatch { + pinned: 2, + requested: 1 + }), + RegisterRejection::VersionMismatch { + pinned: 2, + requested: 1 + } + ); + } + + #[test] + fn media_datagram_tags_owner_index_and_stamps_fence() { + let fenced = FencedHeader { + session_id: Uuid::new_v4(), + generation: 9, + owner_runtime_id: rt(2), + }; + // Owner-assigned index is the first payload byte; client bytes follow. + let d0 = media_datagram(42, fenced, 0, &[0xDE, 0xAD]); + assert_eq!(d0.payload, vec![42, 0xDE, 0xAD]); + assert_eq!(d0.fenced, fenced); + assert_eq!(d0.seq, 0); + // Empty client frame still carries the index byte (owner tolerates it). + let d1 = media_datagram(7, fenced, 3, &[]); + assert_eq!(d1.payload, vec![7]); + assert_eq!(d1.seq, 3); + } } diff --git a/crates/buzz-relay/src/audio/mesh.rs b/crates/buzz-relay/src/audio/mesh.rs index c635173eb..4c293d164 100644 --- a/crates/buzz-relay/src/audio/mesh.rs +++ b/crates/buzz-relay/src/audio/mesh.rs @@ -241,8 +241,8 @@ impl MeshAudioRouter { /// feeds this task, which wraps each frame as a [`MeshDatagram`] and sends it to /// the pod that hosts that participant. Drops on a disconnected/oversize peer — /// realtime audio never blocks fan-out on one slow remote link. -pub fn spawn_remote_peer_sink( - transport: Arc, +pub fn spawn_remote_peer_sink( + transport: Arc, to: RuntimeId, fenced: FencedHeader, mut frames: mpsc::Receiver,