test(sfu): in-process media-forwarding integration test (rust/tests/forwarding.rs)

Closes the coverage gap the README flagged as "Next": two real str0m clients
connect to the sidecar over loopback UDP (full ICE/DTLS/SRTP + the data-channel
renegotiation the browser does), peer A sends Opus RTP, and we assert peer B
RECEIVES the forwarded media. Pure `cargo test`, ~0.8s, no browser — proves the
SFU data plane (accept offer → forward without decode) end to end.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

README.md

@@ -219,7 +219,7 @@ Membership lives in a swappable `RoomStore`. The default `ArrayRoomStore` keeps
 
 ## Status
 
-**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, full per-participant **call recording**, and **backend-hosted rooms on the Rust SFU sidecar** (server-terminated ICE/DTLS/SRTP, forwarding without decode, native Ogg/Opus recording, mute/kick/stats/usage events, plug-and-play binary management) — all tested, including a real PHP↔Rust round-trip that spawns the sidecar from PHPUnit. Next: the duplex AI bridge inside the SFU data plane (today `OpenAiRealtimeBridge` covers turn-based AI participants) and browser E2E automation.
+**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, full per-participant **call recording**, and **backend-hosted rooms on the Rust SFU sidecar** (server-terminated ICE/DTLS/SRTP, forwarding without decode, native Ogg/Opus recording, mute/kick/stats/usage events, plug-and-play binary management) — all tested, including a real PHP↔Rust round-trip that spawns the sidecar from PHPUnit and an **in-process media-forwarding test** (`rust/tests/forwarding.rs`: two real str0m clients over loopback UDP, one sends Opus, the other receives the forwarded media — full ICE/DTLS/SRTP + data-channel renegotiation, no browser). Next: the duplex AI bridge inside the SFU data plane (today `OpenAiRealtimeBridge` covers turn-based AI participants).
 
 ## Architecture
 

rust/tests/forwarding.rs

@@ -0,0 +1,217 @@
+//! Media-forwarding integration test — the coverage the README flagged as "Next".
+//!
+//! Boots the real sidecar and connects TWO real str0m clients to it over loopback
+//! UDP (full ICE/DTLS/SRTP, and the data-channel renegotiation the browser does).
+//! Peer A sends Opus RTP; we assert peer B RECEIVES the forwarded media. No
+//! browser — the "clients" are str0m `Rtc` instances driven over real sockets, so
+//! this proves the SFU data plane (accept offer → forward without decode) end to
+//! end and runs in plain `cargo test`.
+
+use std::io::{BufRead, BufReader, Write};
+use std::net::{IpAddr, UdpSocket};
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use str0m::change::{SdpAnswer, SdpOffer, SdpPendingOffer};
+use str0m::channel::ChannelId;
+use str0m::media::{Direction, MediaKind, MediaTime, Mid};
+use str0m::net::{Protocol, Receive};
+use str0m::{Candidate, Event, Input, Output, Rtc};
+
+use blax_webrtc_sfu::server::{self, Config};
+
+fn start_sidecar(socket: PathBuf) -> thread::JoinHandle<()> {
+    let config = Config {
+        socket,
+        udp_ip: IpAddr::from([127, 0, 0, 1]),
+        udp_port: 0,
+        public_ip: Some(IpAddr::from([127, 0, 0, 1])),
+    };
+    thread::spawn(move || server::run(config).expect("sidecar to run"))
+}
+
+fn control_connect(socket: &PathBuf) -> UnixStream {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if let Ok(s) = UnixStream::connect(socket) {
+            return s;
+        }
+        assert!(Instant::now() < deadline, "control socket never came up");
+        thread::sleep(Duration::from_millis(20));
+    }
+}
+
+/// add_peer over the control socket (what PHP does); returns the answer SDP.
+fn add_peer(
+    reader: &mut impl BufRead,
+    writer: &mut impl Write,
+    id: u64,
+    room: &str,
+    peer: &str,
+    offer: &str,
+) -> String {
+    let req = serde_json::json!({"id": id, "cmd": "add_peer", "room": room, "peer": peer, "offer": offer});
+    writeln!(writer, "{req}").unwrap();
+    writer.flush().unwrap();
+    let mut line = String::new();
+    reader.read_line(&mut line).unwrap();
+    let v: serde_json::Value = serde_json::from_str(&line).unwrap();
+    assert_eq!(v["ok"], true, "add_peer failed: {v}");
+    v["answer"].as_str().expect("answer sdp").to_string()
+}
+
+struct Peer {
+    rtc: Rtc,
+    socket: UdpSocket,
+    audio_mid: Option<Mid>,
+    cid: Option<ChannelId>,
+    connected: bool,
+    media_recv: usize,
+}
+
+impl Peer {
+    /// A str0m client offering a data channel (+ an audio track when `with_audio`).
+    fn new(with_audio: bool) -> (Peer, String, SdpPendingOffer) {
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+        socket.set_read_timeout(Some(Duration::from_millis(3))).unwrap();
+        let addr = socket.local_addr().unwrap();
+
+        let mut rtc = Rtc::builder().build();
+        // No OS candidate gathering in str0m: add our host candidate so the offer
+        // carries it (the SFU also learns us from inbound STUN).
+        rtc.add_local_candidate(Candidate::host(addr, "udp").unwrap());
+
+        let mut change = rtc.sdp_api();
+        let audio_mid =
+            with_audio.then(|| change.add_media(MediaKind::Audio, Direction::SendRecv, None, None));
+        change.add_channel("data".to_string());
+        let (offer, pending) = change.apply().expect("offer");
+
+        (
+            Peer { rtc, socket, audio_mid, cid: None, connected: false, media_recv: 0 },
+            offer.to_sdp_string(),
+            pending,
+        )
+    }
+
+    fn accept_answer(&mut self, pending: SdpPendingOffer, answer_sdp: &str) {
+        let answer = SdpAnswer::from_sdp_string(answer_sdp).expect("parse answer");
+        self.rtc.sdp_api().accept_answer(pending, answer).expect("accept answer");
+    }
+
+    /// Drain all pending output (send Transmits, handle events); return next timeout.
+    fn poll(&mut self) -> Instant {
+        loop {
+            if !self.rtc.is_alive() {
+                return Instant::now() + Duration::from_millis(50);
+            }
+            match self.rtc.poll_output().expect("poll_output") {
+                Output::Transmit(t) => {
+                    let _ = self.socket.send_to(&t.contents, t.destination);
+                }
+                Output::Timeout(t) => return t,
+                Output::Event(e) => self.on_event(e),
+            }
+        }
+    }
+
+    fn on_event(&mut self, e: Event) {
+        match e {
+            Event::Connected => self.connected = true,
+            Event::ChannelOpen(cid, _) => self.cid = Some(cid),
+            Event::ChannelData(cd) => self.on_channel_data(&cd.data),
+            Event::MediaData(_) => self.media_recv += 1,
+            _ => {}
+        }
+    }
+
+    /// Data-channel renegotiation: the SFU offers a peer's track over the DC as
+    /// others join; accept it and answer on the same channel (mirrors the browser).
+    fn on_channel_data(&mut self, data: &[u8]) {
+        let Ok(offer) = serde_json::from_slice::<SdpOffer>(data) else { return };
+        let Ok(answer) = self.rtc.sdp_api().accept_offer(offer) else { return };
+        if let Some(cid) = self.cid {
+            if let Some(mut ch) = self.rtc.channel(cid) {
+                let json = serde_json::to_string(&answer).unwrap();
+                let _ = ch.write(false, json.as_bytes());
+            }
+        }
+    }
+
+    fn recv(&mut self) {
+        let mut buf = vec![0u8; 2000];
+        loop {
+            match self.socket.recv_from(&mut buf) {
+                Ok((n, source)) => {
+                    let dest = self.socket.local_addr().unwrap();
+                    if let Ok(contents) = (&buf[..n]).try_into() {
+                        let _ = self.rtc.handle_input(Input::Receive(
+                            Instant::now(),
+                            Receive { proto: Protocol::Udp, source, destination: dest, contents },
+                        ));
+                    }
+                }
+                Err(_) => break, // WouldBlock / read timeout
+            }
+        }
+        let _ = self.rtc.handle_input(Input::Timeout(Instant::now()));
+    }
+
+    /// Write one Opus packet on the audio track (payload is opaque — the SFU
+    /// forwards without decoding, so B receives these bytes verbatim).
+    fn write_audio(&mut self, seq: u64) {
+        let Some(mid) = self.audio_mid else { return };
+        let Some(w) = self.rtc.writer(mid) else { return };
+        let pt = w.payload_params().find(|p| p.spec().codec.is_audio()).map(|p| p.pt());
+        if let Some(pt) = pt {
+            let _ = w.write(pt, Instant::now(), MediaTime::from_micros(seq * 20_000), vec![0xA5u8; 80]);
+        }
+    }
+}
+
+#[test]
+fn two_clients_forward_audio() {
+    let socket = std::env::temp_dir().join(format!("blax-sfu-fwd-{}.sock", std::process::id()));
+    let _ = std::fs::remove_file(&socket);
+    let _sidecar = start_sidecar(socket.clone());
+
+    let ctrl = control_connect(&socket);
+    let mut reader = BufReader::new(ctrl.try_clone().unwrap());
+    let mut writer = ctrl;
+
+    // A sends audio; B (data channel only) receives A's track via renegotiation.
+    let (mut a, a_offer, a_pending) = Peer::new(true);
+    let a_answer = add_peer(&mut reader, &mut writer, 1, "room", "alice", &a_offer);
+    a.accept_answer(a_pending, &a_answer);
+
+    let (mut b, b_offer, b_pending) = Peer::new(false);
+    let b_answer = add_peer(&mut reader, &mut writer, 2, "room", "bob", &b_offer);
+    b.accept_answer(b_pending, &b_answer);
+
+    let deadline = Instant::now() + Duration::from_secs(20);
+    let mut seq = 0u64;
+    while Instant::now() < deadline {
+        a.poll();
+        a.recv();
+        b.poll();
+        b.recv();
+        if a.connected {
+            a.write_audio(seq);
+            seq += 1;
+        }
+        if b.media_recv >= 5 {
+            break;
+        }
+        thread::sleep(Duration::from_millis(3));
+    }
+
+    assert!(a.connected, "peer A never reached Connected");
+    assert!(b.connected, "peer B never reached Connected");
+    assert!(
+        b.media_recv >= 5,
+        "peer B did not receive forwarded media (received {} packets)",
+        b.media_recv
+    );
+}