From a32b6d624801de2a91b1dcd5bb4659771d5ea643 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 22:39:33 +0000 Subject: [PATCH] =?UTF-8?q?test(nests):=20T16=20Phase=203=20=E2=80=94=20ud?= =?UTF-8?q?p-loss-shim=20+=20I9=20+=20I5=20hot-swap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - **udp-loss-shim** body: tokio UDP loopback that drops a configurable fraction of datagrams. Single-tenant (one client at a time) — moq-lite is connection-multiplexed by source port, so 1:1 forwarding is enough. - **I9** (`packet_loss_1pct_does_not_kill_audio`): routes the Kotlin speaker through the shim with `--loss-rate 0.01`; asserts the decoded PCM has ≥ 80% expected sample count and the 440 Hz tone survives. moq-lite groups are reliable streams so retransmits absorb the loss. - **I5** (`speaker_hot_swap_does_not_crash`): drives the reconnecting-speaker with `tokenRefreshAfterMs = 2_500` to force a hot-swap mid-broadcast. The reference hang-listen is single-shot subscribe (it doesn't re-subscribe on broadcast re-announce), so it captures only the pre-swap chunk; the test asserts the speaker survives without corrupting active uni streams (≥ 1 s of audio + FFT peak at 440 Hz on the captured chunk). The "no audible gap" property the spec calls for is an Amethyst-listener concern (handles re-announce transparently); a Phase 3 follow-up would exercise that path end-to-end through `connectNestsListener`. I7 (publisher reconnect on the Rust side, ref→A) is the mirror image and would need hang-publish to take a "--reconnect-after-ms" flag, plus the Amethyst listener path through the harness — also Phase 3 follow-up. https://claude.ai/code/session_01ERJPUYfdLPwZ99pr5EcEcV --- cli/hang-interop/Cargo.lock | 47 ++++- cli/hang-interop/udp-loss-shim/Cargo.toml | 13 +- cli/hang-interop/udp-loss-shim/src/main.rs | 143 +++++++++++++- .../interop/native/HangInteropTest.kt | 187 ++++++++++++++++-- 4 files changed, 361 insertions(+), 29 deletions(-) diff --git a/cli/hang-interop/Cargo.lock b/cli/hang-interop/Cargo.lock index a291032cd..5ff1b3205 100644 --- a/cli/hang-interop/Cargo.lock +++ b/cli/hang-interop/Cargo.lock @@ -555,7 +555,7 @@ checksum = "4e7f34442dbe69c60fe8eaf58a8cafff81a1f278816d8ab4db255b3bef4ac3c4" dependencies = [ "getrandom 0.3.4", "libm", - "rand", + "rand 0.9.4", "siphasher", ] @@ -1268,7 +1268,7 @@ dependencies = [ "conducer", "futures", "num_enum", - "rand", + "rand 0.9.4", "serde", "thiserror 2.0.18", "tokio", @@ -1331,7 +1331,7 @@ dependencies = [ "moq-lite", "parking_lot", "quinn", - "rand", + "rand 0.9.4", "rcgen", "reqwest", "rustls", @@ -1671,7 +1671,7 @@ dependencies = [ "fastbloom", "getrandom 0.3.4", "lru-slab", - "rand", + "rand 0.9.4", "ring", "rustc-hash", "rustls", @@ -1713,14 +1713,35 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rand" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" +dependencies = [ + "libc", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + [[package]] name = "rand" version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ - "rand_chacha", - "rand_core", + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", ] [[package]] @@ -1730,7 +1751,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", ] [[package]] @@ -2639,7 +2669,10 @@ version = "0.0.1" dependencies = [ "anyhow", "clap", + "rand 0.8.6", "tokio", + "tracing", + "tracing-subscriber", ] [[package]] diff --git a/cli/hang-interop/udp-loss-shim/Cargo.toml b/cli/hang-interop/udp-loss-shim/Cargo.toml index 10b0bc7dc..711fa9c94 100644 --- a/cli/hang-interop/udp-loss-shim/Cargo.toml +++ b/cli/hang-interop/udp-loss-shim/Cargo.toml @@ -5,7 +5,15 @@ edition.workspace = true publish.workspace = true license.workspace = true -# Phase 1: stub. Phase 3 wires this for the I9 packet-loss scenario. +# UDP loopback that drops a configurable fraction of datagrams. +# Used by the I9 packet-loss interop scenario: +# +# client (--server-bind 0.0.0.0:0) → udp-loss-shim --listen X +# → moq-relay --upstream Y +# +# The shim is a single-tenant relay (one client at a time) — moq-lite +# is on QUIC which is connection-multiplexed by the client's source +# port, so we forward 1:1. [[bin]] name = "udp-loss-shim" @@ -15,3 +23,6 @@ path = "src/main.rs" anyhow.workspace = true clap.workspace = true tokio.workspace = true +rand = "0.8" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/cli/hang-interop/udp-loss-shim/src/main.rs b/cli/hang-interop/udp-loss-shim/src/main.rs index 2364fbb52..d2f5d803b 100644 --- a/cli/hang-interop/udp-loss-shim/src/main.rs +++ b/cli/hang-interop/udp-loss-shim/src/main.rs @@ -1,27 +1,152 @@ -//! udp-loss-shim — UDP loopback that drops a configurable fraction of -//! datagrams, used by the I9 packet-loss scenario. **Phase 1 stub.** +//! udp-loss-shim — UDP loopback that drops a configurable fraction +//! of datagrams. Used by the I9 packet-loss cross-stack interop +//! scenario. See +//! `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`. +//! +//! Topology: +//! +//! client → `--listen ` (this binary) → `--upstream ` (moq-relay) +//! +//! The shim picks one client (the first peer that sends to its +//! listen socket), forwards datagrams in both directions +//! 1:1 modulo the loss roll, and exits when the parent test +//! kills it. moq-lite is on QUIC which is connection-multiplexed +//! by the client's source port, so single-tenant forwarding is +//! enough for the test scenarios. -use anyhow::Result; +use std::net::SocketAddr; +use std::sync::Arc; + +use anyhow::{Context, Result}; use clap::Parser; +use tokio::net::UdpSocket; +use tokio::sync::Mutex; #[derive(Parser, Debug)] -#[command(name = "udp-loss-shim", about = "UDP packet-loss shim (Phase 1 stub)")] +#[command(name = "udp-loss-shim", about = "UDP loopback with configurable packet loss")] struct Args { + /// Address to listen on (the client connects here). #[arg(long)] listen: String, + + /// Upstream address to forward to (moq-relay's UDP port). #[arg(long)] upstream: String, + + /// Fraction of datagrams to drop, 0.0–1.0. Applied independently + /// to each direction. #[arg(long, default_value_t = 0.0)] loss_rate: f32, } #[tokio::main] async fn main() -> Result<()> { + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_writer(std::io::stderr) + .try_init(); + let args = Args::parse(); - eprintln!( - "udp-loss-shim Phase-1 stub — listen={} upstream={} loss_rate={}", - args.listen, args.upstream, args.loss_rate + anyhow::ensure!( + (0.0..=1.0).contains(&args.loss_rate), + "loss-rate must be in 0.0..=1.0, got {}", + args.loss_rate ); - eprintln!("Phase 3 will implement actual UDP forwarding. Exiting cleanly."); - Ok(()) + + let listen_addr: SocketAddr = args.listen.parse().context("parse --listen")?; + let upstream_addr: SocketAddr = args.upstream.parse().context("parse --upstream")?; + + // Listen socket — accepts datagrams from the client. + let listen_sock = Arc::new( + UdpSocket::bind(listen_addr) + .await + .with_context(|| format!("bind --listen {listen_addr}"))?, + ); + // Upstream socket — talks to moq-relay. Bound to an ephemeral + // port; relay's outbound path back replies to whatever source + // port we picked. + let upstream_sock = Arc::new( + UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))) + .await + .context("bind upstream socket")?, + ); + upstream_sock + .connect(upstream_addr) + .await + .with_context(|| format!("connect upstream {upstream_addr}"))?; + + tracing::info!( + listen = %listen_addr, + upstream = %upstream_addr, + loss_rate = args.loss_rate, + "udp-loss-shim ready" + ); + + // Track the client's source address. moq-lite's QUIC client + // doesn't use connection migration in our test setup, so the + // first peer that sends to us is the only client we care about. + let client_addr: Arc>> = Arc::new(Mutex::new(None)); + + // Direction 1: client → upstream (with loss). + let loss = args.loss_rate; + let listen_clone = listen_sock.clone(); + let upstream_clone = upstream_sock.clone(); + let client_clone = client_addr.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 65_535]; + loop { + let (n, src) = match listen_clone.recv_from(&mut buf).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(%e, "listen recv error; exiting"); + return; + } + }; + // Latch the client address on first packet. + { + let mut c = client_clone.lock().await; + if c.is_none() { + tracing::info!(%src, "client latched"); + *c = Some(src); + } + } + if rand::random::() < loss { + tracing::trace!(bytes = n, %src, "drop client→upstream"); + continue; + } + if let Err(e) = upstream_clone.send(&buf[..n]).await { + tracing::warn!(%e, "upstream send failed"); + } + } + }); + + // Direction 2: upstream → client (with loss). + let loss = args.loss_rate; + let upstream_clone = upstream_sock.clone(); + let listen_clone = listen_sock.clone(); + let client_clone = client_addr.clone(); + let mut buf = [0u8; 65_535]; + loop { + let n = match upstream_clone.recv(&mut buf).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(%e, "upstream recv error; exiting"); + return Ok(()); + } + }; + if rand::random::() < loss { + tracing::trace!(bytes = n, "drop upstream→client"); + continue; + } + let dst = match *client_clone.lock().await { + Some(addr) => addr, + None => { + tracing::trace!(bytes = n, "upstream sent before client latched; ignoring"); + continue; + } + }; + if let Err(e) = listen_clone.send_to(&buf[..n], dst).await { + tracing::warn!(%e, %dst, "listen send_to failed"); + } + } } diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/HangInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/HangInteropTest.kt index d7cba6997..c88b5fcaf 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/HangInteropTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/HangInteropTest.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.nestsclient.audio.PcmAssertions import com.vitorpamplona.nestsclient.audio.SineWaveAudioCapture import com.vitorpamplona.nestsclient.buildRelayConnectTarget import com.vitorpamplona.nestsclient.connectNestsSpeaker +import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory @@ -384,6 +385,88 @@ class HangInteropTest { PcmAssertions.assertFftPeak(tailWindow, expectedHz = 440.0, halfWindowHz = 5.0) } + /** + * I9 — 1% packet loss via the `udp-loss-shim` between the + * Kotlin speaker's UDP socket and the relay. Asserts the + * decoded PCM still has ≥ 80 % of the expected sample count + * and the 440 Hz peak survives — moq-lite groups are + * reliable streams (`bestEffort=false`), so lost bytes get + * retransmitted and the listener still observes the whole + * tone with mild jitter. + */ + @Test + fun packet_loss_1pct_does_not_kill_audio() = + runBlocking { + val out = + runSpeakerToHangListen( + speakerSeconds = 5, + captureFirstFrame = false, + udpLossRate = 0.01f, + ) + val pcm = readFloat32Pcm(out.pcmFile) + val expected = 5.0 * AudioFormat.SAMPLE_RATE_HZ + assertTrue( + pcm.size >= expected * 0.80, + "expected ≥ 80% of $expected samples under 1% packet loss, " + + "got ${pcm.size} (${"%.1f".format(pcm.size / expected * 100)} %)", + ) + val warmup = AudioFormat.SAMPLE_RATE_HZ / 25 + val analysed = pcm.copyOfRange(warmup, pcm.size) + PcmAssertions.assertFftPeak(analysed, expectedHz = 440.0, halfWindowHz = 5.0) + } + + /** + * I5 — speaker hot-swap mid-broadcast (proactive JWT refresh). + * Drives `connectReconnectingNestsSpeaker` with a 2.5 s + * `tokenRefreshAfterMs` so the speaker recycles its session + * mid-broadcast. + * + * **Limitation:** the reference `hang-listen` is single-shot + * — it reads the catalog once and subscribes once. When the + * speaker hot-swaps, the relay unannounces the old broadcast + * and announces a new one; hang-listen's audio subscription + * dies and it doesn't re-subscribe. So the listener captures + * only the pre-hot-swap chunk (~2.5 s of 5 s). + * + * The Amethyst production listener handles re-announce + * transparently (`ReconnectingNestsListener.kt`), so the + * "no audio gap" assertion the spec calls for is a property + * of the Amethyst path, not the hang-listen path. This test + * therefore asserts only: + * + * - the speaker survives the hot-swap (no crash, got some + * audio), + * - the FFT peak on the captured pre-swap chunk is still + * at 440 Hz (the swap doesn't corrupt active uni + * streams). + * + * The "no audible gap" assertion belongs to a Phase 3 + * follow-up that exercises this scenario through the + * Amethyst Kotlin LISTENER instead of `hang-listen`. + */ + @Test + fun speaker_hot_swap_does_not_crash() = + runBlocking { + val out = + runSpeakerToHangListen( + speakerSeconds = 5, + captureFirstFrame = false, + hotSwapAfterMs = 2_500L, + ) + val pcm = readFloat32Pcm(out.pcmFile) + // Got SOMETHING (≥ 1 s of audio) — speaker didn't + // crash on the hot-swap. + assertTrue( + pcm.size >= AudioFormat.SAMPLE_RATE_HZ, + "expected ≥ 1 s of audio across the hot-swap, got ${pcm.size} samples", + ) + // The captured chunk's tone is still 440 Hz — + // spectral integrity intact. + val warmup = AudioFormat.SAMPLE_RATE_HZ / 25 + val analysed = pcm.copyOfRange(warmup, pcm.size) + PcmAssertions.assertFftPeak(analysed, expectedHz = 440.0, halfWindowHz = 5.0) + } + /** * Rust↔Rust round-trip: pure-Rust through our harness. * Validates the cargo workspace + relay config + `moq-lite-03` @@ -478,15 +561,67 @@ private suspend fun runSpeakerToHangListen( listenerLateJoinDelayMs: Long = 150L, muteWindowMs: ClosedRange? = null, captureFirstFrame: Boolean, + /** + * If non-null, route the Kotlin speaker's UDP through a + * `udp-loss-shim` subprocess that drops this fraction of + * datagrams (0.0..=1.0). The shim listens on a fresh + * ephemeral port and forwards to the harness's relay; the + * speaker's `endpoint` is rewritten to the shim port. The + * hang-listen subprocess still connects directly to the + * relay (no loss), so any frame deficit is attributable to + * the speaker→relay leg. + */ + udpLossRate: Float? = null, + /** + * If non-null, drive the speaker through + * `connectReconnectingNestsSpeaker` with this + * `tokenRefreshAfterMs`, forcing a session recycle (hot-swap) + * mid-broadcast. Default uses the simple non-reconnecting + * speaker — the I1/I2/I3/I8/I10/I11 scenarios don't need + * the reconnect orchestrator. + */ + hotSwapAfterMs: Long? = null, ): HangListenOutput { val harness = NativeMoqRelayHarness.shared() val signer: NostrSigner = NostrSignerInternal(KeyPair()) val pubkey = signer.pubKey + + // If a loss rate is requested, spin up a udp-loss-shim + // between the speaker and the relay. The speaker connects + // through the shim (lossy); hang-listen still connects to + // the relay directly so any frame deficit is attributable + // to the lossy leg. + val (relayHostForSpeaker, relayPortForSpeaker, lossShimProc) = + if (udpLossRate != null) { + val shimPort = java.net.ServerSocket(0).use { it.localPort } + val (relayHost, relayPort) = harness.loopbackHostPort() + val proc = + ProcessBuilder( + harness.udpLossShimBin().toString(), + "--listen", + "127.0.0.1:$shimPort", + "--upstream", + "$relayHost:$relayPort", + "--loss-rate", + udpLossRate.toString(), + ).redirectErrorStream(true) + .also { it.environment()["RUST_LOG"] = "info" } + .start() + // Tiny breathing room for the shim's listen socket + // to bind before the speaker's QUIC handshake hits. + Thread.sleep(200) + Triple("127.0.0.1", shimPort, proc) + } else { + val (h, p) = harness.loopbackHostPort() + Triple(h, p, null) + } + val speakerEndpoint = "https://$relayHostForSpeaker:$relayPortForSpeaker" + val room = NestsRoomConfig( authBaseUrl = "", - endpoint = harness.relayUrl, + endpoint = speakerEndpoint, hostPubkey = pubkey, roomId = "rt-${UUID.randomUUID()}", ) @@ -518,17 +653,44 @@ private suspend fun runSpeakerToHangListen( lateinit var listenProc: Process try { val speaker = - connectNestsSpeaker( - httpClient = StaticTokenNestsClient, - transport = transport, - scope = pumpScope, - room = room, - signer = signer, - speakerPubkeyHex = pubkey, - captureFactory = { SineWaveAudioCapture(freqHz = 440) }, - encoderFactory = { JvmOpusEncoder() }, - framesPerGroup = 5, - ) + if (hotSwapAfterMs != null) { + connectReconnectingNestsSpeaker( + httpClient = StaticTokenNestsClient, + transport = transport, + scope = pumpScope, + room = room, + signer = signer, + speakerPubkeyHex = pubkey, + captureFactory = { SineWaveAudioCapture(freqHz = 440) }, + encoderFactory = { JvmOpusEncoder() }, + tokenRefreshAfterMs = hotSwapAfterMs, + connector = { + connectNestsSpeaker( + httpClient = StaticTokenNestsClient, + transport = transport, + scope = pumpScope, + room = room, + signer = signer, + speakerPubkeyHex = pubkey, + captureFactory = { SineWaveAudioCapture(freqHz = 440) }, + encoderFactory = { JvmOpusEncoder() }, + framesPerGroup = 5, + ) + }, + ) + } else { + connectNestsSpeaker( + httpClient = StaticTokenNestsClient, + transport = transport, + scope = pumpScope, + room = room, + signer = signer, + speakerPubkeyHex = pubkey, + captureFactory = { SineWaveAudioCapture(freqHz = 440) }, + encoderFactory = { JvmOpusEncoder() }, + framesPerGroup = 5, + ) + } val handle = speaker.startBroadcasting() delay(listenerLateJoinDelayMs) @@ -575,6 +737,7 @@ private suspend fun runSpeakerToHangListen( speaker.close() } finally { pumpScope.coroutineContext[Job]?.cancel() + lossShimProc?.destroy() } val exited = listenProc.waitFor(15, TimeUnit.SECONDS)