fix(nests): unstick reconnecting speaker + listener interop on real relays
Two production-path bugs surfaced during a full interop sweep (`./gradlew :nestsClient:jvmTest -DnestsInterop=true` + `-DnestsHangInterop=true`) plus the toolchain / harness friction that was hiding them. Production fixes: - ReconnectingNestsSpeaker's hot-swap path opens publishers via `openPublisherForHotSwap` instead of `startBroadcasting`, so the underlying `MoqLiteNestsSpeaker`'s state machine never made the Connected -> Broadcasting transition. The reconnect wrapper mirrors that state, so callers waiting on Broadcasting (the VM, `connectReconnectingNestsSpeaker` interop tests) hung on Connected forever. Adds `HotSwappablePublisherSource.reportBroadcasting(isMuted)`, implemented in `MoqLiteNestsSpeaker`, and the hot-swap pump now calls it each iteration so a JWT-refreshed session re-enters Broadcasting too. - The stale-group filter on listener reconnect assumes monotonic group lineage across publisher session-restarts (true for ReconnectingNestsSpeaker, which seeds each new publisher with the prior `nextSequence`; false for external publishers like kixelated's hang-publish reference, which mints a fresh state on every reconnect and restarts at 0). Without compensation, the watermark from a session-1 max ~18 dropped the entire post-restart stream from session 2. The collect now resets the watermark on the first object of a new subscription when its group id arrives well below the current watermark — a publisher-restart signal that a relay-cache replay (which carries exactly the prior max) wouldn't produce. Test harness fixes — these are what blocked the suites from running at all on a clean Apple-Silicon box: - NostrNestsHarness: the cloned `docker-compose-moq.yml` declares no `depends_on`, so on a cold `up -d` moq-relay raced moq-auth's Node boot, got `Connection refused` on its JWKS GET, exited with no retry, and every later QUIC handshake failed with "read loop exited (socket closed or peer closed)". Gate on moq-auth's /health first, then idempotent `up -d moq-relay` so the relay always boots against a live auth sidecar. - nestsClient/build.gradle.kts: cargo builds inside the hang-interop task panic under CMake 4.x because `audiopus_sys` / rustls' aws-lc-sys ship a `CMakeLists.txt` that predates CMake 4's minimum-version floor. Set `CMAKE_POLICY_VERSION_MINIMUM=3.5` on each cargo Exec. - JvmOpusEncoder + the hang-interop Test task: opus-java 1.1.1's bundled `natives/darwin/libopus.dylib` is x86_64-only, so its in-jar loader reports unsupported on Apple Silicon. Probe a small set of canonical system libopus locations (`brew install opus` on macOS, distro paths on linux) and `System.load` by absolute path so the symbols are in the process when JNA's lazy `Opus.INSTANCE` init falls back to RTLD_DEFAULT. Gradle also threads `/opt/homebrew/opt/opus/lib` onto `jna.library.path` for completeness; both paths are no-ops where the in-jar binary already loads. After these: - :nestsClient:jvmTest -DnestsInterop=true — 312/312, 0 fail - :nestsClient:jvmTest -DnestsHangInterop=true — 312/312, 0 fail - quic/interop/run-matrix.sh -s aioquic — 19 passed, 0 failed, 3 unsupported (E/V2/CM are documented amethyst gaps). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
@@ -164,6 +164,10 @@ val interopInstallMoqRelay by tasks.registering(Exec::class) {
|
||||
"--root", hangInteropCacheDir.asFile.absolutePath,
|
||||
"--locked",
|
||||
)
|
||||
// See `interopBuildSidecars` — a transitive `*-sys` crate (rustls'
|
||||
// aws-lc-sys) builds a C library via CMake whose CMakeLists.txt
|
||||
// predates CMake 4's minimum-version floor.
|
||||
environment("CMAKE_POLICY_VERSION_MINIMUM", "3.5")
|
||||
val installed =
|
||||
hangInteropCacheDir.dir("bin").file(
|
||||
if (org.gradle.internal.os.OperatingSystem.current().isWindows) "moq-relay.exe" else "moq-relay",
|
||||
@@ -184,6 +188,9 @@ val interopInstallMoqTokenCli by tasks.registering(Exec::class) {
|
||||
"--root", hangInteropCacheDir.asFile.absolutePath,
|
||||
"--locked",
|
||||
)
|
||||
// See `interopBuildSidecars` — CMake 4 vs. a stale bundled
|
||||
// CMakeLists.txt in a transitive `*-sys` crate.
|
||||
environment("CMAKE_POLICY_VERSION_MINIMUM", "3.5")
|
||||
val installed =
|
||||
hangInteropCacheDir.dir("bin").file(
|
||||
if (org.gradle.internal.os.OperatingSystem.current().isWindows) "moq-token-cli.exe" else "moq-token-cli",
|
||||
@@ -199,6 +206,13 @@ val interopBuildSidecars by tasks.registering(Exec::class) {
|
||||
group = "interop"
|
||||
workingDir = hangInteropDir.asFile
|
||||
commandLine("cargo", "build", "--release")
|
||||
// `audiopus_sys` bundles libopus, whose CMakeLists.txt declares
|
||||
// `cmake_minimum_required(VERSION <3.5)`. CMake 4.x removed
|
||||
// compatibility with that, so the build script panics. CMake reads
|
||||
// this env var (added in 3.31) as the floor policy version, which
|
||||
// lets the stale libopus config run unchanged. Harmless on older
|
||||
// CMake that predates CMake 4's removal.
|
||||
environment("CMAKE_POLICY_VERSION_MINIMUM", "3.5")
|
||||
// Track only manifests + sources; the `target/` subtree is the
|
||||
// output, including it as an input would mark the task always
|
||||
// out-of-date.
|
||||
@@ -227,6 +241,21 @@ tasks.withType<Test>().configureEach {
|
||||
val cargoBin = hangInteropCacheDir.dir("bin").asFile
|
||||
systemProperty("nestsHangInteropSidecarsDir", sidecarRelease.absolutePath)
|
||||
systemProperty("nestsHangInteropCargoBinDir", cargoBin.absolutePath)
|
||||
// `club.minnced:opus-java` 1.1.1 ships only an x86_64 darwin dylib,
|
||||
// so its in-jar loader reports unsupported on Apple Silicon. Hand
|
||||
// JNA an extra search directory so a Homebrew-installed arm64
|
||||
// libopus (`brew install opus`, /opt/homebrew/opt/opus/lib) wins
|
||||
// over the bundled binary. Append rather than replace to preserve
|
||||
// any caller-supplied path; the existing system search still covers
|
||||
// other platforms. Silently skipped if Homebrew isn't installed —
|
||||
// tests on Intel mac / linux take the in-jar path unchanged.
|
||||
val brewOpusLib = file("/opt/homebrew/opt/opus/lib")
|
||||
if (brewOpusLib.exists()) {
|
||||
val existing = System.getProperty("jna.library.path")
|
||||
val combined =
|
||||
if (existing.isNullOrEmpty()) brewOpusLib.absolutePath else "$existing:${brewOpusLib.absolutePath}"
|
||||
systemProperty("jna.library.path", combined)
|
||||
}
|
||||
// Per-method moq-relay trace log dir for the routing-race
|
||||
// investigation (plan 2026-05-07-moq-relay-routing-investigation.md).
|
||||
// Off by default; opt in via -DnestsHangInteropTraceRelay=true so a
|
||||
|
||||
+21
@@ -56,6 +56,27 @@ internal interface HotSwappablePublisherSource {
|
||||
startSequence: Long = 0L,
|
||||
): MoqLitePublisherHandle
|
||||
|
||||
/**
|
||||
* Flip the speaker's outward state to [NestsSpeakerState.Broadcasting].
|
||||
*
|
||||
* The hot-swap pump owns its own long-lived broadcaster and opens
|
||||
* publishers via [openPublisherForHotSwap] instead of going through
|
||||
* [MoqLiteNestsSpeaker.startBroadcasting], so the speaker's state
|
||||
* machine never sees the Connected → Broadcasting transition that
|
||||
* `startBroadcasting` performs. Without this call the reconnect
|
||||
* wrapper — which mirrors the underlying speaker's state — stays
|
||||
* stuck on Connected forever and callers waiting on Broadcasting
|
||||
* (the production VM, interop tests) time out.
|
||||
*
|
||||
* Called once per session iteration after the audio publisher is
|
||||
* installed, so a freshly-swapped session re-enters Broadcasting
|
||||
* too. No-op when the speaker is already terminal.
|
||||
*
|
||||
* @param isMuted the user's current mute intent, replayed onto the
|
||||
* new state so the swap doesn't reset the mute indicator.
|
||||
*/
|
||||
fun reportBroadcasting(isMuted: Boolean)
|
||||
|
||||
/**
|
||||
* Surface a broadcast-pipeline terminal failure (e.g. sustained
|
||||
* `publisher.send` errors past
|
||||
|
||||
+31
@@ -229,6 +229,37 @@ class MoqLiteNestsSpeaker internal constructor(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* [HotSwappablePublisherSource] implementation. The hot-swap pump
|
||||
* drives publishing through [openPublisherForHotSwap] rather than
|
||||
* [startBroadcasting], so it has to flip the state machine into
|
||||
* Broadcasting itself. Connected → Broadcasting on first call;
|
||||
* re-applies the mute intent if already Broadcasting (a session
|
||||
* swap re-enters this with the new session still in Connected, so
|
||||
* the common case is the Connected branch). No-op once terminal.
|
||||
*/
|
||||
override fun reportBroadcasting(isMuted: Boolean) {
|
||||
val current = mutableState.value
|
||||
mutableState.value =
|
||||
when (current) {
|
||||
is NestsSpeakerState.Connected -> {
|
||||
NestsSpeakerState.Broadcasting(
|
||||
room = current.room,
|
||||
negotiatedMoqVersion = current.negotiatedMoqVersion,
|
||||
isMuted = isMuted,
|
||||
)
|
||||
}
|
||||
|
||||
is NestsSpeakerState.Broadcasting -> {
|
||||
current.copy(isMuted = isMuted)
|
||||
}
|
||||
|
||||
else -> {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Called from the broadcaster's `onTerminalFailure` callback (off
|
||||
* the speaker's coroutine). Transitions the speaker to `Failed` so
|
||||
|
||||
+25
@@ -458,8 +458,33 @@ private class ReconnectingHandle(
|
||||
var emitted = 0L
|
||||
var droppedStale = 0L
|
||||
var handleHighestGroup = -1L
|
||||
var firstObjectOnHandle = true
|
||||
try {
|
||||
handle.objects.collect { obj ->
|
||||
if (firstObjectOnHandle) {
|
||||
firstObjectOnHandle = false
|
||||
// Publisher-side session restart detection.
|
||||
// The watermark assumes monotonic group
|
||||
// lineage across publisher sessions, which
|
||||
// holds for ReconnectingNestsSpeaker (it
|
||||
// seeds each new publisher with the prior
|
||||
// nextSequence). External publishers — the
|
||||
// kixelated/hang-publish reference, any
|
||||
// moq-lite publisher that mints a fresh
|
||||
// state per reconnect — restart at 0. If
|
||||
// the first group on a fresh subscription
|
||||
// arrives well below the watermark, the
|
||||
// publisher restarted, NOT the relay
|
||||
// replaying its cache (a replay carries
|
||||
// exactly the prior max, not a value
|
||||
// many groups below). Reset the watermark
|
||||
// so the entire post-restart stream isn't
|
||||
// dropped as stale.
|
||||
val watermark = priorGroupWatermark.get()
|
||||
if (watermark > 0L && obj.groupId < watermark - 1L) {
|
||||
priorGroupWatermark.set(-1L)
|
||||
}
|
||||
}
|
||||
if (obj.groupId <= priorGroupWatermark.get()) {
|
||||
// Stale group re-served from the relay
|
||||
// cache after a re-subscribe — drop so
|
||||
|
||||
+8
@@ -598,6 +598,14 @@ private class ReissuingBroadcastHandle(
|
||||
if (old != null) runCatching { old.close() }
|
||||
}
|
||||
|
||||
// The audio publisher is now live on this session. Flip the
|
||||
// underlying speaker's state machine into Broadcasting — the
|
||||
// hot-swap path bypasses `startBroadcasting`, so without this
|
||||
// the speaker (and the reconnect wrapper that mirrors it) would
|
||||
// sit on Connected forever. Done every iteration so a swapped
|
||||
// session — which arrives in Connected — re-enters Broadcasting.
|
||||
hotSwap.reportBroadcasting(desiredMuted)
|
||||
|
||||
// Catalog track on the new session. Keeps the broadcast
|
||||
// discoverable to standards-aligned moq-lite watchers (the
|
||||
// kixelated/moq browser reference) across JWT refresh — without
|
||||
|
||||
+33
-3
@@ -99,10 +99,40 @@ class JvmOpusEncoder(
|
||||
if (nativesLoaded) return
|
||||
synchronized(loadLock) {
|
||||
if (nativesLoaded) return
|
||||
check(OpusLibrary.isSupportedPlatform()) {
|
||||
"club.minnced:opus-java natives not available for this platform"
|
||||
if (OpusLibrary.isSupportedPlatform()) {
|
||||
check(OpusLibrary.loadFromJar()) { "OpusLibrary.loadFromJar() returned false" }
|
||||
} else {
|
||||
// opus-java 1.1.1's bundled `natives/darwin/libopus.dylib`
|
||||
// is x86_64-only, so `isSupportedPlatform()` returns
|
||||
// false on Apple Silicon (and on any host the jar
|
||||
// doesn't ship a binary for). Fall back to a
|
||||
// system-installed libopus: probe a small set of
|
||||
// canonical locations and `System.load` it directly
|
||||
// so its symbols are in the process. JNA's lazy
|
||||
// [Opus.INSTANCE] init falls back to RTLD_DEFAULT
|
||||
// for symbol lookup when `jna.library.path` doesn't
|
||||
// resolve the bare name, so preloading by absolute
|
||||
// path is the most reliable cross-host approach
|
||||
// (`brew install opus` on macOS, `apt install libopus0`
|
||||
// / `yum install opus` on linux). After the load, touch
|
||||
// [Opus.INSTANCE] so any remaining linkage issue
|
||||
// surfaces here rather than at the first encode call.
|
||||
val candidates =
|
||||
listOf(
|
||||
"/opt/homebrew/opt/opus/lib/libopus.dylib",
|
||||
"/usr/local/opt/opus/lib/libopus.dylib",
|
||||
"/usr/lib/x86_64-linux-gnu/libopus.so.0",
|
||||
"/usr/lib/aarch64-linux-gnu/libopus.so.0",
|
||||
"/usr/lib64/libopus.so.0",
|
||||
)
|
||||
val found = candidates.firstOrNull { java.io.File(it).exists() }
|
||||
checkNotNull(found) {
|
||||
"club.minnced:opus-java natives not available for this platform " +
|
||||
"and no system libopus found at any of: $candidates"
|
||||
}
|
||||
System.load(found)
|
||||
Opus.INSTANCE
|
||||
}
|
||||
check(OpusLibrary.loadFromJar()) { "OpusLibrary.loadFromJar() returned false" }
|
||||
nativesLoaded = true
|
||||
}
|
||||
}
|
||||
|
||||
+13
-1
@@ -195,7 +195,6 @@ class NostrNestsHarness private constructor(
|
||||
runDocker(workDir, "up", "-d")
|
||||
try {
|
||||
waitForPort("127.0.0.1", AUTH_HOST_PORT, PORT_READY_TIMEOUT_MS)
|
||||
waitForPort("127.0.0.1", MOQ_HOST_PORT, PORT_READY_TIMEOUT_MS)
|
||||
// moq-auth's Node runtime opens the listen socket
|
||||
// before its handlers are wired, so the first POST that
|
||||
// arrives in the gap can RST-on-write. Wait for /health
|
||||
@@ -203,6 +202,19 @@ class NostrNestsHarness private constructor(
|
||||
// pipeline is live and avoids the SocketException that
|
||||
// otherwise hits the first test of the second class.
|
||||
waitForHealth("http://127.0.0.1:$AUTH_HOST_PORT/health", PORT_READY_TIMEOUT_MS)
|
||||
// moq-relay fetches its JWK set from moq-auth at startup
|
||||
// and exits — with no retry — if that GET is refused.
|
||||
// The compose file declares no `depends_on`, so on a cold
|
||||
// `up -d` moq-relay races moq-auth's Node boot, loses,
|
||||
// logs "failed to GET JWKS … Connection refused", and
|
||||
// dies; every later QUIC handshake then fails with
|
||||
// "read loop exited (socket closed or peer closed)".
|
||||
// Now that moq-auth is confirmed healthy, (re)start
|
||||
// moq-relay: this is a no-op if it survived the race and
|
||||
// a clean restart if it didn't, so it always comes up
|
||||
// against a live auth sidecar.
|
||||
runDocker(workDir, "up", "-d", "moq-relay")
|
||||
waitForPort("127.0.0.1", MOQ_HOST_PORT, PORT_READY_TIMEOUT_MS)
|
||||
} catch (t: Throwable) {
|
||||
// Capture container state + recent logs BEFORE tearing
|
||||
// down so the failure message is actually actionable.
|
||||
|
||||
Reference in New Issue
Block a user