diff --git a/desktop/src-tauri/src/huddle/playout.rs b/desktop/src-tauri/src/huddle/playout.rs index 4371d8a1313..5bfce3adad5 100644 --- a/desktop/src-tauri/src/huddle/playout.rs +++ b/desktop/src-tauri/src/huddle/playout.rs @@ -33,7 +33,7 @@ use tokio_util::sync::CancellationToken; use super::human_floor::HumanFloor; use super::jitter::{PeerJitterBuffer, SAMPLE_RATE_HZ}; use super::relay_api::{WsStream, REMOTE_SPEECH_THRESHOLD}; -use super::wire::{FrameHeader, FLAG_DTX, V2_HEADER_LEN}; +use super::wire::{parse_relay_frame, FLAG_DTX}; /// Speaker-tick window for emitting `huddle-active-speakers`. Active set is /// cleared each tick — peers that didn't send a frame in the last window are @@ -149,18 +149,11 @@ fn is_agent_peer( }) } -/// Whether `peer_idx` is currently occupied at exactly `epoch`, per the -/// authoritative roster. A frame is deliverable only when both match: an index -/// absent from the roster is stale, and a slot reused by a later occupant has -/// advanced its epoch, so a departed occupant's in-flight frame is fenced -/// rather than mis-attributed to the new occupant. A legacy relay omits the -/// epoch, which degrades to `0` on both sides, making the fence a no-op. -fn is_current_occupant( - peer_idx: u8, - epoch: u8, - index_to_epoch: &std::collections::HashMap, -) -> bool { - index_to_epoch.get(&peer_idx) == Some(&epoch) +/// Whether `peer_idx` is currently occupied per the authoritative roster. +/// Protocol v2 media carries only the peer index, so roster presence is the +/// strongest routing boundary available until the relay supports v3 epochs. +fn is_current_occupant(peer_idx: u8, index_to_epoch: &std::collections::HashMap) -> bool { + index_to_epoch.contains_key(&peer_idx) } fn same_occupancy( @@ -196,7 +189,7 @@ fn f32_samples_to_le_bytes(samples: &[f32]) -> Vec { /// One remote peer's slot: jitter buffer + dedicated rodio Player. /// /// Per-frame seq/timestamp come from the v2 wire header (sender-authored). -/// The relay forwards `peer_index | epoch | header | opus_bytes` opaquely; we +/// The relay forwards `peer_index | header | opus_bytes` opaquely; we /// parse the header here and pass the sender's own monotonic seq + 48 kHz media /// timestamp into NetEq. struct PeerSlot { @@ -422,22 +415,19 @@ pub(crate) async fn run_playout_recv_loop( msg = ws_rx.next() => { match msg { Some(Ok(WsMsg::Binary(data))) => { - // Wire shape (v2): [peer_index: u8][epoch: u8][header: 8 bytes][opus payload...] - // The minimum size is 2 (peer_index + epoch) + 8 (header) + ≥1 Opus byte. - if data.len() <= 2 + V2_HEADER_LEN { + // Wire shape (v2): [peer_index: u8][header: 8 bytes][opus payload...] + // The minimum size is 1 (peer index) + 8 (header) + ≥1 Opus byte. + let Some((peer_idx, header, opus_bytes)) = parse_relay_frame(&data) else { + eprintln!( + "buzz-desktop: dropping malformed v2 audio relay frame ({} bytes)", + data.len(), + ); continue; - } - let peer_idx = data[0]; - let epoch = data[1]; - // Fence the peer-index reuse race: a frame authored by a - // departed occupant that arrives after its index is - // reassigned carries the old epoch. Drop it rather than - // mis-attribute stale audio (and the new occupant's - // human/agent STT policy) to whoever grabbed the index. - // An index absent from the roster is also stale. A slot - // with no known epoch (legacy relay) degrades to 0 on - // both sides, so the fence is a no-op there. - if !is_current_occupant(peer_idx, epoch, &index_to_epoch) { + }; + // Protocol v2 has no media epoch. Drop frames for slots + // absent from the control roster; delayed frames after + // an index is reassigned cannot be fenced until v3. + if !is_current_occupant(peer_idx, &index_to_epoch) { continue; } // Suppress only an agent stream synthesized and @@ -446,21 +436,6 @@ pub(crate) async fn run_playout_recv_loop( if is_locally_synthesized_peer(peer_idx, &local_tts_publishers) { continue; } - let after_idx = &data[2..]; - let Some((header, opus_bytes)) = FrameHeader::parse(after_idx) - else { - // Malformed v2 frame: header parse only fails when - // the slice is too short, which `if data.len() <= ...` - // already guards. Defensive log + drop. - eprintln!( - "buzz-desktop: dropping malformed audio frame from peer {peer_idx} ({} bytes)", - data.len(), - ); - continue; - }; - if opus_bytes.is_empty() { - continue; - } let is_dtx = (header.flags & FLAG_DTX) != 0; // Only count non-DTX arrivals toward the UI's // active-speaker set. DTX/comfort packets are emitted @@ -771,34 +746,16 @@ mod tests { ); } - /// Causal regression for the peer-index reuse race (Jude's blocking - /// finding): a frame authored by a departed occupant that arrives after - /// its slot is reassigned to a new occupant carries the stale epoch and - /// must be fenced, never mis-attributed to the new occupant. #[test] - fn stale_epoch_frame_is_fenced_after_its_index_is_reused() { + fn v2_media_is_routed_only_for_current_roster_indices() { let mut index_to_epoch = std::collections::HashMap::new(); - // Slot 3 first occupied at epoch 0. index_to_epoch.insert(3_u8, 0_u8); assert!( - is_current_occupant(3, 0, &index_to_epoch), + is_current_occupant(3, &index_to_epoch), "current occupant's frame is delivered" ); - - // The occupant departs and a new peer reuses slot 3 at epoch 1. - index_to_epoch.insert(3, 1); - assert!( - !is_current_occupant(3, 0, &index_to_epoch), - "in-flight frame from the departed occupant (epoch 0) is fenced" - ); - assert!( - is_current_occupant(3, 1, &index_to_epoch), - "the new occupant's frame (epoch 1) is delivered" - ); - - // A frame for an index absent from the roster is stale. assert!( - !is_current_occupant(9, 0, &index_to_epoch), + !is_current_occupant(9, &index_to_epoch), "frame for an unoccupied index is dropped" ); } diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 20a2be57652..190397aa054 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -114,6 +114,9 @@ async fn connect_authenticated_audio_socket( "type": "auth", "event": event_json, "parent_channel_id": parent_channel_id, + // Use the released v2 contract while deployed relays remain capped at + // v2. Relay-to-client media therefore has a one-byte peer-index prefix; + // see huddle::wire for the compatibility tradeoff. "protocol_version": super::wire::PROTOCOL_VERSION, }); ws_tx diff --git a/desktop/src-tauri/src/huddle/wire.rs b/desktop/src-tauri/src/huddle/wire.rs index 518377a60b0..bcf9c007c2d 100644 --- a/desktop/src-tauri/src/huddle/wire.rs +++ b/desktop/src-tauri/src/huddle/wire.rs @@ -7,25 +7,18 @@ //! //! No per-frame metadata; receiver synthesizes sequence/timestamp on arrival. //! Kept for backward compatibility — relay still admits v1 clients into -//! v1-pinned rooms — but new clients always speak v3. +//! v1-pinned rooms — but new clients speak v2 while deployed relays remain +//! capped at the released v2 contract. //! -//! ## v2 (released) +//! ## v2 (compatibility contract) //! //! Client → relay: `` //! Relay → client: `` //! -//! ## v3 (this commit) -//! -//! Client → relay: `` -//! Relay → client: `` -//! -//! The relay prefixes each forwarded frame with the sender's stable -//! `peer_index` and the current occupancy `epoch` of that index. The epoch -//! advances each time a slot is reused by a new occupant, so a client can -//! fence a frame authored by a departed occupant that arrives after its index -//! is reassigned — it carries the stale epoch and is dropped rather than -//! mis-attributed. The client's own send path is unaffected: it emits only -//! `
` and the relay stamps the prefix. +//! Protocol v2 does not carry v3's occupancy epoch in media frames. The +//! control-plane roster still resets decoder and playout state when an index is +//! reassigned, but v2 cannot fence a delayed packet from the previous occupant +//! after that reassignment. //! //! Header layout (8 bytes, network byte order, big-endian): //! @@ -44,13 +37,13 @@ //! * `level_dbov` is client-authored telemetry. The relay parses it for //! logging/active-speaker hints, clamps invalid values into range, and //! **never** uses it for trust decisions (admission, moderation, etc.). -//! * Negotiation lives in the WS auth message (`protocol_version: 3`), not +//! * Negotiation lives in the WS auth message (`protocol_version: 2`), not //! in any bit of `flags`. Mixed-version rooms are rejected at the relay //! with `upgrade_required`. /// Wire protocol version this client speaks. Bumped only when the frame /// layout itself changes; the relay tracks pinned per-room. -pub const PROTOCOL_VERSION: u8 = 3; +pub const PROTOCOL_VERSION: u8 = 2; /// Length of the v2 per-frame header in bytes. pub const V2_HEADER_LEN: usize = 8; @@ -135,6 +128,19 @@ impl FrameHeader { } } +/// Parse a complete relay-to-client v2 frame. +/// +/// The released v2 contract has exactly one relay-authored prefix byte: the +/// sender's peer index. A non-empty Opus payload must follow the fixed header. +pub fn parse_relay_frame(bytes: &[u8]) -> Option<(u8, FrameHeader, &[u8])> { + let (&peer_index, framed_audio) = bytes.split_first()?; + let (header, opus_payload) = FrameHeader::parse(framed_audio)?; + if opus_payload.is_empty() { + return None; + } + Some((peer_index, header, opus_payload)) +} + /// Compute a dBov audio level for a normalized f32 PCM frame. /// /// "dBov" is RMS expressed in dB relative to full scale (where full scale = @@ -218,6 +224,41 @@ mod tests { assert_eq!(tail, b"opus-bytes"); } + #[test] + fn relay_frame_uses_the_v2_one_byte_peer_prefix() { + let header = FrameHeader { + seq: 0x0102, + ts_48k: 960, + level_dbov: -20, + flags: 0, + }; + let mut frame = vec![7]; + frame.extend_from_slice(&header.encode()); + frame.extend_from_slice(b"opus"); + + let (peer_index, parsed_header, opus_payload) = + parse_relay_frame(&frame).expect("valid v2 relay frame"); + assert_eq!(peer_index, 7); + assert_eq!(parsed_header, header); + assert_eq!(opus_payload, b"opus"); + } + + #[test] + fn relay_frame_rejects_a_missing_opus_payload() { + let mut frame = vec![7]; + frame.extend_from_slice( + &FrameHeader { + seq: 1, + ts_48k: 960, + level_dbov: -20, + flags: 0, + } + .encode(), + ); + + assert!(parse_relay_frame(&frame).is_none()); + } + /// Bytes in big-endian network order, matching Max's spec. This pins /// the byte layout against accidental endianness changes. #[test]