test: add loopback integration test for the audio transport

Drives two real iroh endpoints on localhost (relay disabled, addresses
exchanged directly) through the full production path: reactive connect_peer
with lower-id dialing, sequenced broadcast, receive_datagrams, and
JitterBuffer decode at a realistic mixer cadence. Verifies datagrams arrive
attributed to the sender, the sequence header round-trips, and Opus decodes
to full 20ms frames -- without needing a mic, speakers, or the GUI.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/transport_loopback.rs

@@ -0,0 +1,144 @@
+//! End-to-end loopback test for the redesigned audio transport.
+//!
+//! Spins up two real iroh endpoints on localhost (relay disabled, addresses
+//! exchanged directly) and drives the actual production path: reactive
+//! `connect_peer`, sequenced `broadcast`, `receive_datagrams`, and the
+//! per-peer `JitterBuffer` decode. No microphone, speakers, or GUI required.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use bytes::Bytes;
+use iroh::address_lookup::memory::MemoryLookup;
+use iroh::endpoint::presets;
+use iroh::protocol::Router;
+use iroh::{Endpoint, RelayMode};
+use opus::{Application, Channels};
+
+use peerspeak::codec::AudioEncoder;
+use peerspeak::codec::opus_impl::OpusEncoder;
+use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer};
+use peerspeak::network::NetworkTransport;
+use peerspeak::network::iroh_impl::IrohTransport;
+
+const AUDIO_ALPN: &[u8] = b"peerspeak-audio";
+
+struct Node {
+    endpoint: Endpoint,
+    transport: Arc<IrohTransport>,
+    _router: Router,
+    lookup: MemoryLookup,
+}
+
+async fn spawn_node() -> Node {
+    let lookup = MemoryLookup::new();
+    let endpoint = Endpoint::builder(presets::Minimal)
+        .secret_key(iroh::SecretKey::generate())
+        // Direct-only: two endpoints on the same host reach each other via the
+        // addresses we seed into each other's lookup, with no external relay.
+        .relay_mode(RelayMode::Disabled)
+        .address_lookup(lookup.clone())
+        .bind()
+        .await
+        .expect("bind endpoint");
+
+    let (transport, audio_proto) = IrohTransport::new(endpoint.clone());
+    let router = Router::builder(endpoint.clone())
+        .accept(AUDIO_ALPN, audio_proto)
+        .spawn();
+
+    Node {
+        endpoint,
+        transport: Arc::new(transport),
+        _router: router,
+        lookup,
+    }
+}
+
+/// One real, decodable Opus packet for a 20ms mono frame, prefixed with the
+/// 4-byte little-endian sequence header the transport/jitter buffer expect.
+fn packet(enc: &mut OpusEncoder, seq: u32) -> Bytes {
+    let pcm: Vec<i16> = (0..FRAME_SAMPLES)
+        .map(|i| if i % 2 == 0 { 2000 } else { -2000 })
+        .collect();
+    let encoded = enc.encode(&pcm).unwrap();
+    let mut buf = Vec::with_capacity(4 + encoded.len());
+    buf.extend_from_slice(&seq.to_le_bytes());
+    buf.extend_from_slice(&encoded);
+    Bytes::from(buf)
+}
+
+#[tokio::test]
+async fn loopback_sequenced_audio_reaches_peer_and_decodes() {
+    let a = spawn_node().await;
+    let b = spawn_node().await;
+
+    // Seed each side with the other's full address so direct dialing works.
+    a.lookup.add_endpoint_info(b.endpoint.addr());
+    b.lookup.add_endpoint_info(a.endpoint.addr());
+
+    let a_id = a.endpoint.id();
+    let b_id = b.endpoint.id();
+
+    // Subscribe to incoming datagrams on B before any are sent.
+    let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B");
+
+    // Reactive connection setup, exactly as core does on peer-join. Calling on
+    // both sides is fine: the lower EndpointId dials, the higher accepts, and a
+    // single shared connection forms.
+    a.transport.connect_peer(b_id).await;
+    b.transport.connect_peer(a_id).await;
+
+    // Let the dial + accept registration settle.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // A sends 50 sequenced frames.
+    const N: u32 = 50;
+    let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
+    for seq in 0..N {
+        a.transport.broadcast(packet(&mut enc, seq));
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+
+    // Collect what B receives and feed it through a real jitter buffer,
+    // popping a frame per arrival to mirror the mixer's steady 20ms cadence
+    // (so the buffer stays shallow rather than overflowing its cap).
+    let mut jitter = JitterBuffer::new().unwrap();
+    let mut received = 0u32;
+    let mut decoded_frames = 0u32;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    // Loop ends when the channel closes or the deadline is hit (pattern stops matching).
+    while let Ok(Some((from, bytes))) = tokio::time::timeout_at(deadline, b_rx.recv()).await {
+        assert_eq!(from, a_id, "datagram should be attributed to sender A");
+        assert!(bytes.len() >= 4, "packet carries a sequence header");
+        let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
+        jitter.insert(seq, bytes[4..].to_vec());
+        received += 1;
+
+        if let Some(frame) = jitter.pop_frame() {
+            assert_eq!(frame.len(), FRAME_SAMPLES, "decoded frame is one 20ms frame");
+            decoded_frames += 1;
+        }
+        if received >= N {
+            break;
+        }
+    }
+
+    // Drain whatever remains buffered behind the playout delay.
+    while let Some(frame) = jitter.pop_frame() {
+        assert_eq!(frame.len(), FRAME_SAMPLES);
+        decoded_frames += 1;
+    }
+
+    // On localhost essentially nothing should be lost over a real QUIC datagram path.
+    assert!(
+        received >= N - 2,
+        "expected to receive ~{N} datagrams, got {received}"
+    );
+    // And nearly all received packets should decode to PCM (a few absorbed by
+    // the initial priming delay).
+    assert!(
+        decoded_frames >= N - 5,
+        "expected to decode ~{N} frames, got {decoded_frames} (received {received})"
+    );
+}