From 54f667f5bd777402b54854e044e2bf954d50a644 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 15 May 2026 08:50:22 -0400 Subject: [PATCH] fix(nests): unstick reconnecting speaker + listener interop on real relays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- nestsClient/build.gradle.kts | 29 +++++++++++++++ .../HotSwappablePublisherSource.kt | 21 +++++++++++ .../nestsclient/MoqLiteNestsSpeaker.kt | 31 ++++++++++++++++ .../nestsclient/ReconnectingNestsListener.kt | 25 +++++++++++++ .../nestsclient/ReconnectingNestsSpeaker.kt | 8 +++++ .../nestsclient/audio/JvmOpusEncoder.kt | 36 +++++++++++++++++-- .../nestsclient/interop/NostrNestsHarness.kt | 14 +++++++- 7 files changed, 160 insertions(+), 4 deletions(-) diff --git a/nestsClient/build.gradle.kts b/nestsClient/build.gradle.kts index 1c1e8b52b..3cb05ae46 100644 --- a/nestsClient/build.gradle.kts +++ b/nestsClient/build.gradle.kts @@ -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().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 diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/HotSwappablePublisherSource.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/HotSwappablePublisherSource.kt index 1a895d238..7d7eb0626 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/HotSwappablePublisherSource.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/HotSwappablePublisherSource.kt @@ -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 diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 74c0766af..132bbccab 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -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 diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index c4e6b3602..a2e894fd3 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -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 diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index 8cfd57665..df0679e27 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -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 diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/audio/JvmOpusEncoder.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/audio/JvmOpusEncoder.kt index 2f6554b93..ed49fcadc 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/audio/JvmOpusEncoder.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/audio/JvmOpusEncoder.kt @@ -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 } } diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsHarness.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsHarness.kt index 5aae2cabc..2029b2c14 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsHarness.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsHarness.kt @@ -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.