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

Downgrade desktop Huddles to audio protocol v2 (#6610)

Browse Source 
## Summary
- negotiate Huddle audio protocol v2 on desktop
- decode the released one-byte peer-index prefix
- retain roster-driven playout resets and document the missing v3 epoch
fence

## Testing
- `just desktop-tauri-fmt-check`
- `just desktop-tauri-clippy`
- `just desktop-tauri-test`

Signed-off-by: kenny lopez <klopez4212@gmail.com>
This commit is contained in:
klopez4212
2026-08-23 16:24:51 +01:00
committed by GitHub
parent e236329413
commit 0720f5380c
3 changed files with 82 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
desktop/src-tauri/src/huddle/playout.rs
View File

@@ -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<u8, u8>,
) -> 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<u8, u8>) -> bool {
index_to_epoch.contains_key(&peer_idx)
}
fn same_occupancy(
@@ -196,7 +189,7 @@ fn f32_samples_to_le_bytes(samples: &[f32]) -> Vec<u8> {
/// 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"
);
}

3
desktop/src-tauri/src/huddle/relay_api.rs
View File

@@ -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

73
desktop/src-tauri/src/huddle/wire.rs
View File

@@ -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: `<header: [u8; 8]><opus_bytes>`
//! Relay → client: `<peer_index: u8><header: [u8; 8]><opus_bytes>`
//!
//! ## v3 (this commit)
//!
//! Client → relay: `<header: [u8; 8]><opus_bytes>`
//! Relay → client: `<peer_index: u8><epoch: u8><header: [u8; 8]><opus_bytes>`
//!
//! 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
//! `<header><opus_bytes>` 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]