nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(nests-tests): hang-listen catalog-read uses single long-lived subscribe

Browse Source 
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).
This commit is contained in:
Claude
2026-05-07 12:26:19 +00:00
parent c25736fb0f
commit 8cc7cbd42b
1 changed files with 50 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/tests/hang-interop/hang-listen/src/main.rs
+50 -37
View File
@@ -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<anyhow::Error> = None;
let mut decoded: Option<hang::Catalog> = 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::<hang::Catalog, anyhow::Error>(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.