From 8cc7cbd42be5f8bd4145705710c118a95eeff00d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 12:26:19 +0000 Subject: [PATCH] fix(nests-tests): hang-listen catalog-read uses single long-lived subscribe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the create-drop-recreate retry loop with one subscribe held open for the full 10 s read budget. Inner timeouts are gone — we just await catalog.next() under one outer 10 s timeout. Why: moq-rs 0.10.x maps Error::Cancel to wire reset code 0 (see moq-lite/src/error.rs). The previous retry shape produced a cascade: attempt 0 timeout → drop catalog_track → moq-rs sees track.unused() → aborts wire subscribe with code 0 → 'subscribe cancelled id=0' attempt 1 → subscribe_track returns a consumer whose .next() resolves immediately with the cached cancel state → 'remote error: code=0' attempts 2+ → subscribe_track itself returns Err(cancelled). 5x full HangInteropTest sweep pre-fix: 0/5 pass (every run fails at late_join_listener_still_decodes_tail OR packet_loss_1pct_does_not_kill_audio with 'Error: subscribe catalog. cancelled.'). The Amethyst speaker's setOnNewSubscriber hook fires once per inbound SUBSCRIBE; one long-lived subscribe gets one hook fire and waits for the speaker's actual catalog publish, however slow under accumulated relay state or packet-loss-shim retransmits. Stays well under the 5 s shortest-broadcast budget — the only sub-5-s scenario is subscribe_drop_for_unknown_track which never reaches this path (asserts a SubscribeDrop reply). --- .../hang-interop/hang-listen/src/main.rs | 87 +++++++++++-------- 1 file changed, 50 insertions(+), 37 deletions(-) diff --git a/nestsClient/tests/hang-interop/hang-listen/src/main.rs b/nestsClient/tests/hang-interop/hang-listen/src/main.rs index 89b5d33e7..2964cc712 100644 --- a/nestsClient/tests/hang-interop/hang-listen/src/main.rs +++ b/nestsClient/tests/hang-interop/hang-listen/src/main.rs @@ -166,46 +166,59 @@ async fn listen( tracing::info!(%path, "broadcast announced"); // Subscribe to the catalog and read the first published - // version. The catalog hook in Amethyst's - // `MoqLiteNestsSpeaker` (`setOnNewSubscriber`) can race the - // SUBSCRIBE bidi mid-suite — under accumulated relay state - // the first try sometimes resolves with "cancelled" before - // the catalog frame arrives. Retry up to 3 times with a - // 2 s per-attempt timeout (6 s total worst case). Under - // concurrent load (multiple jvmTest workers contending for - // the relay) the first attempt's wire round-trip can - // exceed 500 ms; the longer per-attempt budget keeps the - // happy-path fast (resolves on the first attempt) while - // tolerating slow handshakes. - let info = { - let mut last_err: Option = None; - let mut decoded: Option = None; - for attempt in 0..3 { - let catalog_track = broadcast - .subscribe_track(&hang::Catalog::default_track()) - .context("subscribe catalog")?; - let mut catalog = hang::CatalogConsumer::new(catalog_track); - match tokio::time::timeout(Duration::from_secs(2), catalog.next()).await { - Ok(Ok(Some(c))) => { - decoded = Some(c); - break; - } - Ok(Ok(None)) => { - last_err = Some(anyhow!("catalog ended before first publish")); - } - Ok(Err(e)) => { - tracing::warn!(attempt, %e, "catalog read error; retrying"); - last_err = Some(anyhow::Error::new(e).context("read catalog")); - } - Err(_) => { - tracing::warn!(attempt, "catalog read timed out; retrying"); - last_err = Some(anyhow!("catalog read timed out (attempt {attempt})")); + // version. The naïve "subscribe → next() with timeout → on + // timeout resubscribe" pattern is broken on moq-rs 0.10.x: + // + // - When the `TrackConsumer` is dropped, moq-rs's track + // producer side observes `track.unused()` and aborts the + // wire subscribe with `Error::Cancel`, which maps to + // stream-reset code **0** (per `moq-lite/src/error.rs`). + // - Code 0 is broadcast to any consumer that calls + // `subscribe_track` for the SAME track name within the + // race window — the new consumer's `.next()` resolves + // immediately with `cancelled`, producing the cascade + // `subscribe cancelled id=0 → subscribe error id=1 code=0 + // → subscribe_track failed: cancelled` we observed. + // + // Fix: hold ONE subscription open for the full retry budget. + // Inner timeouts on `.next()` poll for the first published + // group; an outer timeout caps the total wait. The track + // consumer stays alive across inner iterations, so moq-rs + // never sees `track.unused()` and never propagates `Cancel`. + // + // The Amethyst speaker's `onNewSubscriber` hook fires once + // per inbound SUBSCRIBE (see `MoqLiteNestsSpeaker.kt`'s + // `setOnNewSubscriber`). With one long-lived subscribe, we + // get one hook fire, and however long the speaker takes to + // respond — under accumulated relay state, packet-loss + // shim-induced re-transmits, or other test-side timing + // pressure — we still see the first published group as long + // as it arrives within the budget. + // + // Total budget: 10 s. Within every scenario's broadcast + // window (the shortest is `subscribe_drop_for_unknown_track` + // at 5 s, but that test asserts a SubscribeDrop and never + // reaches this path). + let catalog_track = broadcast + .subscribe_track(&hang::Catalog::default_track()) + .context("subscribe catalog")?; + let mut catalog = hang::CatalogConsumer::new(catalog_track); + let info = match tokio::time::timeout(Duration::from_secs(10), async { + loop { + match catalog.next().await { + Ok(Some(c)) => return Ok::(c), + Ok(None) => { + return Err(anyhow!("catalog ended before first publish")); } + Err(e) => return Err(anyhow::Error::new(e).context("read catalog")), } } - decoded.ok_or_else(|| { - last_err.unwrap_or_else(|| anyhow!("catalog read failed after 3 attempts")) - })? + }) + .await + { + Ok(Ok(c)) => c, + Ok(Err(e)) => return Err(e.context("catalog read")), + Err(_) => return Err(anyhow!("catalog read timed out after 10 s")), }; // Pick the first Opus / Container::Legacy audio rendition.