diff --git a/docs/contacts-plan.md b/docs/contacts-plan.md index e8a1c06..91c5c5b 100644 --- a/docs/contacts-plan.md +++ b/docs/contacts-plan.md @@ -133,14 +133,35 @@ address book), though a lightweight mutual-add is optional polish. stranger. **Loopback integration test PASSED over real iroh endpoints** (friend gets Pong+room; stranger gets unusable reply); run with `cargo test -- --ignored presence_net`. -- **STILL deferred (next session, needs care + 2 machines):** - - **The endpoint-lifecycle fork (decide first):** a second always-on endpoint - shares our node id with the per-join room endpoint (possible relay/identity - collision) vs. refactoring to ONE persistent endpoint that hosts friends + - room. Spike/decide before wiring `serve` live. - - Spawn `serve` outside the room session; the rate-limited outbound ping - scheduler (app open + slow refresh); report our current gathering via - `restamp`; render friend status + a **room-name tag + Join** entry (overlaps P5). +- **Endpoint-lifecycle fork — DECIDED 2026-06-15 via a throwaway spike: option (b), + a SINGLE persistent endpoint.** The spike bound two endpoints sharing one + `SecretKey` (one node id), each accepting a different ALPN, and probed both from a + third endpoint. Result: **every inbound connection landed on ONE endpoint** (the + first-bound), and connections for the other ALPN failed at the QUIC handshake with + *"error 120: peer doesn't support any known protocol"* (the connection physically + reached the wrong instance, which doesn't speak that ALPN). So **one node id = + exactly one reachable endpoint instance** — option (a) (a second always-on + friends endpoint sharing our id) is impossible, not merely risky. This is a hard + handshake-layer collision, confirmed on one machine (no cross-network leg needed; + which instance "wins" is just bind-order). **⇒ build (b):** one persistent + endpoint bound once at startup, hosting friends-control + gossip + audio via a + single persistent `Router`; a room "join" becomes subscribe-a-gossip-topic + + spawn-audio-tasks (not rebuild-everything); `NetworkMode` changes require a full + endpoint rebuild (acceptable — already "applies on next join"). +- **B1 (persistent network stack) — DONE 2026-06-15, tests-green (⚠️ 2-machine + field test pending; on branch `w7-b1-persistent-endpoint`, not merged).** A + persistent `NetStack` (endpoint + gossip + `Router`) is built once at startup and + reused across calls; a new persistent `AudioRouter` (`src/network/iroh_impl.rs`) + delegates inbound audio links to the active session's `Shared` (`bind` on join, + `clear` on leave), so the single router/endpoint outlive any room session. Join + no longer rebuilds the endpoint; network-mode/identity changes rebuild the stack + when idle, else defer to the next Leave/Join. This is the prerequisite for wiring + the live listener (below). +- **STILL deferred (B2, after B1 merges):** + - Spawn `serve` on the persistent endpoint (via the Router's FRIENDS_ALPN + handler), outside the room session; the rate-limited outbound ping scheduler + (app open + slow refresh); report our current gathering via `restamp`; render + friend status + a **room-name tag + Join** entry (overlaps P5). ### P5 — Recents + UI — Medium Local recents list (cosmetic room tags). Friends-list UI with status dots diff --git a/src/core/mod.rs b/src/core/mod.rs index f3397f3..489c590 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -6,14 +6,14 @@ use crate::codec::{AudioEncoder, opus_impl::OpusEncoder}; use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES}; use crate::network::{ NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket, - iroh_impl::IrohTransport, + iroh_impl::{IrohTransport, AudioRouter}, gossip::IrohGossipState, }; use crate::core::messages::{CoreCommand, UiEvent}; use crate::config::{NetworkMode, RecordingMode}; use crate::audio::multitrack::MultitrackRecorder; -use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, endpoint::presets, protocol::Router}; +use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router}; use iroh_gossip::net::Gossip; use tokio::sync::{mpsc, Mutex}; use std::collections::{HashMap, HashSet}; @@ -306,8 +306,6 @@ impl ConnEventHandler { } struct ActiveSession { - endpoint: Endpoint, - router: Router, room_state: Arc, capture_thread: std::thread::JoinHandle<()>, datagram_task: tokio::task::JoinHandle<()>, @@ -368,9 +366,10 @@ impl ActiveSession { self.transport.leave().await; crate::log_msg("Room left"); - crate::log_msg("Shutting down router..."); - let _ = tokio::time::timeout(std::time::Duration::from_secs(1), self.router.shutdown()).await; - crate::log_msg("Router shut down"); + // NOTE: the router + endpoint are persistent (owned by the NetStack), so we + // deliberately do NOT shut them down here — only this session's peer links + // (closed by `transport.leave()` above) and tasks are torn down. The caller + // clears the persistent `audio_router` so stray inbound links are dropped. crate::log_msg("Joining capture thread..."); let _ = self.capture_thread.join(); @@ -378,6 +377,101 @@ impl ActiveSession { } } +/// The app's persistent network stack: ONE endpoint + gossip + router, built once +/// and kept alive for the whole `run_core_loop` lifetime (rebuilt only on a network +/// -mode or identity change). Room sessions come and go on top of this — `join` +/// subscribes a gossip topic + binds a per-session transport to `audio_router`, +/// `leave` drops them — but the endpoint stays reachable between calls. That's the +/// W7 prerequisite: the friends presence listener needs an endpoint up while idle, +/// and a node id can only have ONE live endpoint instance (proven by the dual- +/// endpoint spike — see `docs/contacts-plan.md` P4). +struct NetStack { + endpoint: Endpoint, + gossip: Gossip, + router: Router, + /// The persistent inbound-audio handler on `router`; per-join we bind the + /// active session's transport into it, and clear it on leave. + audio_router: AudioRouter, + /// In-memory address book (ticket + gossip fed), shared with every session. + memory_lookup: iroh::address_lookup::memory::MemoryLookup, +} + +impl NetStack { + async fn shutdown(self) { + crate::log_msg("Shutting down network stack (router + endpoint)..."); + let _ = tokio::time::timeout(Duration::from_secs(1), self.router.shutdown()).await; + self.endpoint.close().await; + crate::log_msg("Network stack shut down"); + } +} + +/// Build the persistent network stack for the given identity + relay/discovery +/// posture. Binds the endpoint (the per-`NetworkMode` build hoisted out of the old +/// per-join path), spawns one gossip instance + one router accepting gossip and +/// audio, and kicks off `online()` in the background so app launch isn't blocked on +/// the relay handshake. +async fn build_net_stack( + secret_key: SecretKey, + network_mode: NetworkMode, + memory_lookup: iroh::address_lookup::memory::MemoryLookup, +) -> Result { + // Build the endpoint per the configured relay/discovery posture. All postures + // keep the in-memory address lookup (fed by tickets and gossip); they differ in + // whether n0's relay and DNS presence beacon are used. `Minimal` sets only the + // mandatory crypto provider and deliberately omits the n0 DNS publish/resolve. + let endpoint = match network_mode { + NetworkMode::N0Full => { + Endpoint::builder(presets::N0) + .secret_key(secret_key.clone()) + .address_lookup(memory_lookup.clone()) + .bind() + .await + } + NetworkMode::RelayNoDiscovery => { + Endpoint::builder(presets::Minimal) + .secret_key(secret_key.clone()) + .relay_mode(RelayMode::Default) + .address_lookup(memory_lookup.clone()) + .bind() + .await + } + NetworkMode::DirectOnly => { + Endpoint::builder(presets::Minimal) + .secret_key(secret_key.clone()) + .relay_mode(RelayMode::Disabled) + .address_lookup(memory_lookup.clone()) + .bind() + .await + } + }?; + + // Bring the endpoint online in the background so launch isn't blocked on the + // relay handshake; joins/sends just work once it's up (online() is idempotent). + let ep = endpoint.clone(); + tokio::spawn(async move { ep.online().await }); + + // One gossip instance for all rooms; sessions subscribe/unsubscribe topics on + // it. Frame budget raised above the 4 KB default so a hard-capped custom avatar + // (W4) can ride presence inline — all peers must use the same value. + let gossip = Gossip::builder() + .max_message_size(65536) + .spawn(endpoint.clone()); + + let audio_router = AudioRouter::new(); + let router = Router::builder(endpoint.clone()) + .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone()) + .accept(b"peerspeak-audio", audio_router.clone()) + .spawn(); + + Ok(NetStack { + endpoint, + gossip, + router, + audio_router, + memory_lookup, + }) +} + /// Finalize and clear the active recording, if any, emitting `RecordingStopped`. /// No-op when not recording. Called on stop, room leave, and room switch so a /// recording is always closed cleanly (its WAV size fields patched). @@ -493,6 +587,28 @@ async fn run_core_loop( // Standalone capture-only mic meter, live only when no session exists. let mut mic_monitor: Option = None; + // The persistent network stack (endpoint + gossip + router), built once at + // startup and kept alive for the app's lifetime. Room sessions ride on top of + // it (subscribe a topic + bind the audio router on join, clear on leave); it's + // rebuilt only when the network mode or identity changes. Moving `memory_lookup` + // in — all later access is via `net.memory_lookup`. + let mut net = match build_net_stack(secret_key.clone(), network_mode, memory_lookup).await { + Ok(stack) => stack, + Err(e) => { + // Only a local socket bind can fail here (the relay handshake is + // backgrounded), so this is fatal to networking — surface it rather + // than dying silently. + let _ = ui_tx + .send(UiEvent::Error(format!("Failed to start networking: {e}"))) + .await; + return Err(e); + } + }; + // Set when a network-mode / identity change arrives mid-call; the stack is + // rebuilt on the next Leave (or before the next Join), preserving the old + // "applies on next join" semantics while keeping the endpoint up while idle. + let mut net_rebuild_pending = false; + while let Some(cmd) = cmd_rx.recv().await { match cmd { CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation, avatar } => { @@ -503,10 +619,22 @@ async fn run_core_loop( // capture/mixer feeders are about to stop. stop_recording(&recorder, &is_recording, &multitrack, &is_multitrack, &ui_tx).await; - // Clean up any existing session + // Clean up any existing session FIRST (this calls `transport.leave()` + // on the current endpoint), before any stack rebuild closes it. if let Some(session) = active_session.take() { crate::log_msg("Shutting down existing active session"); session.shutdown(audio_backend.clone()).await; + net.audio_router.clear(); + } + + // If a network-mode / identity change was deferred while a call was + // active, rebuild the persistent stack now — after the old session is + // gone, before the new one binds — so this join uses the new posture. + if net_rebuild_pending { + let lookup = net.memory_lookup.clone(); + net.shutdown().await; + net = build_net_stack(secret_key.clone(), network_mode, lookup).await?; + net_rebuild_pending = false; } // Release a standalone mic monitor if running — it shares the @@ -514,43 +642,12 @@ async fn run_core_loop( // call claims it. (Safe here: any session was just shut down.) stop_mic_monitor(&audio_backend, mic_monitor.take()); - // Build the endpoint per the configured relay/discovery posture. - // All postures keep the in-memory address lookup (fed by tickets - // and gossip); they differ in whether n0's relay and DNS presence - // beacon are used. `Minimal` sets only the mandatory crypto - // provider and deliberately omits the n0 DNS publish/resolve. - let bind_result = match network_mode { - NetworkMode::N0Full => { - Endpoint::builder(presets::N0) - .secret_key(secret_key.clone()) - .address_lookup(memory_lookup.clone()) - .bind() - .await - } - NetworkMode::RelayNoDiscovery => { - Endpoint::builder(presets::Minimal) - .secret_key(secret_key.clone()) - .relay_mode(RelayMode::Default) - .address_lookup(memory_lookup.clone()) - .bind() - .await - } - NetworkMode::DirectOnly => { - Endpoint::builder(presets::Minimal) - .secret_key(secret_key.clone()) - .relay_mode(RelayMode::Disabled) - .address_lookup(memory_lookup.clone()) - .bind() - .await - } - }; - let endpoint = match bind_result { - Ok(ep) => ep, - Err(e) => { - let _ = ui_tx.send(UiEvent::Error(format!("Failed to bind endpoint: {}", e))).await; - continue; - } - }; + // Use the app's persistent endpoint (bound once at startup); the + // per-join endpoint build is gone. A node id has exactly one live + // endpoint instance (W7 spike), so the friends listener and room + // audio necessarily share this one. online() is idempotent — it just + // ensures we're relay-reachable before joining the gossip swarm. + let endpoint = net.endpoint.clone(); endpoint.online().await; // Determine target ticket @@ -566,27 +663,17 @@ async fn run_core_loop( ticket_str }; - // Initialize Gossip and Transport - // Raise the gossip frame budget above the 4 KB default so a - // hard-capped custom avatar (W4) can ride presence inline. All - // peers must use the same value (already a breaking build req from - // the avatar field). 64 KB leaves generous headroom over our cap. - let gossip = Gossip::builder() - .max_message_size(65536) - .spawn(endpoint.clone()); - let (transport, audio_proto) = IrohTransport::new(endpoint.clone()); - let transport = Arc::new(transport); - - // Start Router - let router = iroh::protocol::Router::builder(endpoint.clone()) - .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone()) - .accept(b"peerspeak-audio", audio_proto) - .spawn(); + // Per-session transport over the persistent endpoint, bound to the + // persistent audio router so this call's inbound audio links route + // to it (cleared on leave). Gossip + router are persistent on the + // NetStack; the session just subscribes its topic below. + let transport = Arc::new(IrohTransport::new(endpoint.clone())); + net.audio_router.bind(&transport); let room_state = Arc::new(IrohGossipState::new( endpoint.clone(), - gossip.clone(), - memory_lookup.clone(), + net.gossip.clone(), + net.memory_lookup.clone(), secret_key.clone(), )); @@ -617,7 +704,7 @@ async fn run_core_loop( if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await { crate::log_msg(&format!("Error room_state.join failed: {:?}", e)); let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await; - let _ = router.shutdown().await; + net.audio_router.clear(); continue; } crate::log_msg("Joined room successfully via room_state"); @@ -663,7 +750,7 @@ async fn run_core_loop( if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) { let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await; let _ = room_state.leave().await; - let _ = router.shutdown().await; + net.audio_router.clear(); continue; } @@ -675,7 +762,7 @@ async fn run_core_loop( let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await; let _ = audio_backend.stop(); let _ = room_state.leave().await; - let _ = router.shutdown().await; + net.audio_router.clear(); continue; } @@ -1062,8 +1149,6 @@ async fn run_core_loop( }); let session = ActiveSession { - endpoint: endpoint.clone(), - router, room_state: room_state.clone(), capture_thread, datagram_task, @@ -1098,8 +1183,18 @@ async fn run_core_loop( current_sharing = None; if let Some(session) = active_session.take() { session.shutdown(audio_backend.clone()).await; + // Stop routing inbound audio links — the endpoint/router stay up. + net.audio_router.clear(); let _ = ui_tx.send(UiEvent::RoomLeft).await; } + // Apply any network-mode / identity change that was deferred while we + // were in the call (rebuild while idle keeps the endpoint reachable). + if net_rebuild_pending { + let lookup = net.memory_lookup.clone(); + net.shutdown().await; + net = build_net_stack(secret_key.clone(), network_mode, lookup).await?; + net_rebuild_pending = false; + } } CoreCommand::ToggleMute => { @@ -1111,7 +1206,7 @@ async fn run_core_loop( let self_state = PeerState { name: current_name.clone(), is_muted: new_state, - addr: session.endpoint.addr(), + addr: net.endpoint.addr(), sharing: current_sharing.clone(), avatar: current_avatar.clone(), }; @@ -1127,7 +1222,7 @@ async fn run_core_loop( let self_state = PeerState { name: current_name.clone(), is_muted: is_muted.load(Ordering::Relaxed), - addr: session.endpoint.addr(), + addr: net.endpoint.addr(), sharing: current_sharing.clone(), avatar: current_avatar.clone(), }; @@ -1204,16 +1299,28 @@ async fn run_core_loop( CoreCommand::SetNetworkMode(mode) => { network_mode = mode; + // Rebuild the persistent stack to the new posture immediately if + // idle; if a call is active, defer to the next Leave/Join so the + // live call isn't disrupted (preserves "applies on next join"). + if active_session.is_none() { + let lookup = net.memory_lookup.clone(); + net.shutdown().await; + net = build_net_stack(secret_key.clone(), network_mode, lookup).await?; + } else { + net_rebuild_pending = true; + } } CoreCommand::RegenerateIdentity => { - // Mint + persist a fresh identity, discarding the old one. Takes - // effect on the NEXT join (the endpoint is rebuilt with this key - // then) — consistent with SetNetworkMode's "applies on next join." + // Mint + persist a fresh identity, discarding the old one. The + // persistent endpoint is rebuilt with the new key (now if idle, else + // on the next Leave/Join) — consistent with "applies on next join." + let mut regenerated = false; match crate::identity::regenerate() { Ok(key) => { secret_key = key; identity_error = None; + regenerated = true; crate::log_msg("identity: regenerated to a fresh persistent id"); } Err(e) => { @@ -1223,6 +1330,18 @@ async fn run_core_loop( crate::log_msg(&format!("identity: regenerate failed: {e:#}")); } } + // Re-bind the endpoint under the new id so our node id actually + // changes. Only on a successful regenerate (a failed one left the + // key unchanged, so a rebuild would be pointless churn). + if regenerated { + if active_session.is_none() { + let lookup = net.memory_lookup.clone(); + net.shutdown().await; + net = build_net_stack(secret_key.clone(), network_mode, lookup).await?; + } else { + net_rebuild_pending = true; + } + } let _ = ui_tx .send(UiEvent::IdentityStatus { node_id: secret_key.public().to_string(), @@ -1362,7 +1481,7 @@ async fn run_core_loop( let self_state = PeerState { name: current_name.clone(), is_muted: is_muted.load(Ordering::Relaxed), - addr: session.endpoint.addr(), + addr: net.endpoint.addr(), sharing: Some(ticket), avatar: current_avatar.clone(), }; @@ -1387,7 +1506,7 @@ async fn run_core_loop( let self_state = PeerState { name: current_name.clone(), is_muted: is_muted.load(Ordering::Relaxed), - addr: session.endpoint.addr(), + addr: net.endpoint.addr(), sharing: None, avatar: current_avatar.clone(), }; diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index 470d120..20ddf04 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -276,25 +276,57 @@ async fn obtain_conn( } } -#[derive(Clone)] -pub struct AudioProtocol { - shared: Arc, +/// The persistent audio protocol handler. Registered ONCE on the app's single +/// long-lived [`Router`](iroh::protocol::Router), it delegates each inbound audio +/// connection to whatever session's [`Shared`] is currently bound, or drops it +/// when idle (no active call). This is what lets the endpoint + router outlive any +/// individual room session — the W7 friends listener needs the endpoint reachable +/// *between* calls — while the per-peer connection state stays per-session. +#[derive(Clone, Default)] +pub struct AudioRouter { + /// The active session's transport state, swapped in on join and cleared on + /// leave. `None` means "no call" → inbound audio links are dropped. + current: Arc>>>, } -impl std::fmt::Debug for AudioProtocol { +impl std::fmt::Debug for AudioRouter { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("AudioProtocol").finish_non_exhaustive() + f.debug_struct("AudioRouter").finish_non_exhaustive() } } -impl iroh::protocol::ProtocolHandler for AudioProtocol { +impl AudioRouter { + pub fn new() -> Self { + Self::default() + } + + /// Route inbound audio connections to `transport`'s session (called on join). + pub fn bind(&self, transport: &IrohTransport) { + *self.current.lock().unwrap() = Some(transport.shared.clone()); + } + + /// Stop routing — drop inbound audio links until the next [`bind`](Self::bind) + /// (called on leave). The endpoint/router stay alive; we just have nowhere to + /// hand a fresh audio connection. + pub fn clear(&self) { + *self.current.lock().unwrap() = None; + } +} + +impl iroh::protocol::ProtocolHandler for AudioRouter { fn accept( &self, connection: Connection, ) -> impl std::future::Future> + Send { let peer_id = connection.remote_id(); - let shared = self.shared.clone(); + // Snapshot the current session (clone the Arc) and release the lock before + // any await — the guard isn't held across suspension points. + let shared = self.current.lock().unwrap().clone(); async move { + // No active call → nothing to route this inbound link to; drop it. + let Some(shared) = shared else { + return Ok(()); + }; // Deterministic roles: the lower id dials, the higher id accepts. // If we're the dialer for this peer we never consume inbound links // (we make our own), so just let this one drop. @@ -321,7 +353,11 @@ pub struct IrohTransport { } impl IrohTransport { - pub fn new(endpoint: Endpoint) -> (Self, AudioProtocol) { + /// Build a per-session transport over the app's persistent `endpoint`. The + /// inbound audio handler is the persistent [`AudioRouter`] (registered once on + /// the router) — call [`AudioRouter::bind`] with this transport to route this + /// session's inbound links to it. + pub fn new(endpoint: Endpoint) -> Self { let (incoming_tx, incoming_rx) = mpsc::channel(1000); let (conn_events_tx, conn_events_rx) = mpsc::channel(100); let self_id = endpoint.id(); @@ -337,14 +373,11 @@ impl IrohTransport { conn_events_tx, }); - let protocol = AudioProtocol { shared: shared.clone() }; - let transport = Self { + Self { shared, incoming_rx: tokio::sync::Mutex::new(Some(incoming_rx)), conn_events_rx: tokio::sync::Mutex::new(Some(conn_events_rx)), - }; - - (transport, protocol) + } } /// Tear down for an intentional leave/quit: close every live connection with diff --git a/src/presence_net.rs b/src/presence_net.rs index 2d5df55..354865a 100644 --- a/src/presence_net.rs +++ b/src/presence_net.rs @@ -14,11 +14,13 @@ //! //! **Deferred to a later slice (the hard integration):** spawning `serve` on a //! persistent endpoint that lives OUTSIDE the per-join room session, and the -//! outbound ping scheduler. There's a real fork there — a second always-on -//! endpoint would share our node id with the per-join room endpoint (possible -//! relay/identity collision), vs. refactoring to a single persistent endpoint. -//! That decision wants care + a 2-machine check, so it's intentionally NOT made -//! here; this module works against whatever `Endpoint` it's handed. +//! outbound ping scheduler. The endpoint fork was DECIDED 2026-06-15 by a +//! throwaway spike: a **single persistent endpoint** (option b), NOT a second +//! always-on endpoint sharing our id (option a). The spike proved two endpoints +//! sharing one `SecretKey` can't coexist — inbound connections all land on one +//! instance and the other's ALPN fails the QUIC handshake ("error 120"). So this +//! `serve` will run as the FRIENDS_ALPN handler on the app's one persistent +//! `Router`; it still works against whatever `Endpoint` it's handed. use crate::presence::ControlMsg; use anyhow::{Context, Result, bail}; diff --git a/tests/reconnect_eviction.rs b/tests/reconnect_eviction.rs index c3dfaca..db3d952 100644 --- a/tests/reconnect_eviction.rs +++ b/tests/reconnect_eviction.rs @@ -30,8 +30,7 @@ async fn make_transport() -> Arc { .bind() .await .expect("bind endpoint"); - let (transport, _audio_proto) = IrohTransport::new(endpoint); - Arc::new(transport) + Arc::new(IrohTransport::new(endpoint)) } /// A peer identity with no live endpoint — all we need to key the handler's maps. diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs index a3a9036..29beb78 100644 --- a/tests/transport_loopback.rs +++ b/tests/transport_loopback.rs @@ -24,7 +24,7 @@ use peerspeak::codec::AudioEncoder; use peerspeak::codec::opus_impl::OpusEncoder; use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer}; use peerspeak::network::{ConnEvent, NetworkTransport}; -use peerspeak::network::iroh_impl::IrohTransport; +use peerspeak::network::iroh_impl::{AudioRouter, IrohTransport}; const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; @@ -110,14 +110,18 @@ async fn spawn_node_with_key(secret: iroh::SecretKey) -> Node { .await .expect("bind endpoint"); - let (transport, audio_proto) = IrohTransport::new(endpoint.clone()); + let transport = Arc::new(IrohTransport::new(endpoint.clone())); + // Mirror production: a persistent AudioRouter bound to this session's transport + // is what the router accepts inbound audio links on. + let audio_router = AudioRouter::new(); + audio_router.bind(&transport); let router = Router::builder(endpoint.clone()) - .accept(AUDIO_ALPN, audio_proto) + .accept(AUDIO_ALPN, audio_router) .spawn(); Node { endpoint, - transport: Arc::new(transport), + transport, _router: router, lookup, }