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

test(nests): one-way latency + 5 % packet-loss path in audio benchmark

Browse Source 
Expands `AudioLatencyComparisonTest` from pure inter-arrival jitter
into a real one-way latency measurement against the same
`hang-publish` Rust reference, AND adds a loss-path variant gated
through `udp-loss-shim`.

`hang-publish` (Rust sidecar):
  - new `--log-send-times` flag. When set, the publisher prints
    `SEND frame=<N> send_t_us=<UNIX_EPOCH_MICROS>` to stdout for
    every frame, captured with `SystemTime::now()` immediately
    before `frame.encode(group)`. Lock + flush per line because a
    piped Rust stdout is fully-buffered by default and the JVM-side
    parser needs a live stream. Default off so existing scenarios
    (the I7 reconnect test in particular) keep their stderr-only
    RUST_LOG output untouched.

Kotlin test:
  - `TimestampingOpusEncoder` wraps `JvmOpusEncoder` and stamps an
    `Instant.now()` epoch-micros value per `encode(...)` call into a
    shared map keyed by a monotonic frame counter. encode runs
    immediately before `publisher.send` inside
    `NestMoqLiteBroadcaster`, so the captured timestamp is the
    closest cross-stack anchor we have to "moment the frame entered
    the publisher's outbound buffer" — same anchor the Rust side
    uses.
  - One-way latency = arrival epoch-micros − send epoch-micros per
    matched frame. Both sides read `CLOCK_REALTIME` on linux/macOS,
    so values can be compared directly without a sync handshake when
    the two processes share a host.
  - Frame matching uses `MoqObject.objectId` directly. The listener
    layer already synthesises objectId as a per-SUBSCRIPTION
    monotonic counter, so it equals the publisher's absolute frame
    index; a previous draft of this test did `groupId * 5 + objectId`
    and silently aligned against the wrong frames, producing
    negative latencies.
  - New `under_5pct_packet_loss_pacing_and_one_way_latency` scenario.
    Spawns TWO `udp-loss-shim` instances (one per publisher: the
    shim latches a single client on first datagram, so a shared shim
    would silently swallow the second publisher's traffic), each
    forwarding 1:1 to the moq-relay's UDP port modulo a 5 %
    bidirectional drop rate. The listener stays on the direct path
    so any latency growth comes from publisher-side retransmit /
    ack feedback, not listener-side loss.
  - Both Kotlin and Rust pinned to `FRAMES_PER_GROUP = 5` (Rust's
    default; Kotlin's default is 50). Matched so the per-group
    uni-stream open/close cost is the same on both sides — the
    comparison is about implementation, not about which stack picked
    which group size.

Observed (10 s window, localhost, MacBook Pro M2):

  ===== Audio publisher comparison: clean (loss=0%) =====
  Kotlin speaker     ttf=137 ms  inter-arrival p50/95/99/max=20.17/20.67/20.88/21.16 ms
                                  one-way        p50/95/99/max=80.60/81.31/81.74/83.91 ms
  Rust hang-publish  ttf=325 ms  inter-arrival p50/95/99/max=20.00/21.18/21.49/21.95 ms
                                  one-way        p50/95/99/max=100.32/101.45/102.04/168.58 ms
  ===== Audio publisher comparison: loss-5pct (loss=5%) =====
  Kotlin speaker                  one-way        p50/95/99/max=80.50/81.55/82.31/86.43 ms
  Rust hang-publish               one-way        p50/95/99/max=100.44/119.88/122.16/158.98 ms

  Reading: Kotlin matches Rust on the clean path within ~20 ms
  baseline (and actually sits below it), and absorbs 5 % loss with
  only +0.6 ms p99 growth vs Rust's +20 ms. The 80-100 ms baseline
  is moq-lite's group-buffer floor with 5-frames/group — protocol,
  not stack.

Asserts: each side delivers >= 80 % of expected frames on the clean
path, >= 60 % under loss, and clean-path median inter-arrival sits
in [15, 35] ms. Tail percentiles are printed, not gated.

Gated by `-DnestsHangInterop=true`; both sidecars (hang-publish,
udp-loss-shim) come from `interopBuildSidecars`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
2026-05-15 09:27:25 -04:00
parent e18e6c8399
commit 9fd8cddebf
2 changed files with 610 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/AudioLatencyComparisonTest.kt
+558 -257
View File
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.NestsClient
import com.vitorpamplona.nestsclient.NestsListenerState
import com.vitorpamplona.nestsclient.NestsRoomConfig
import com.vitorpamplona.nestsclient.audio.JvmOpusEncoder
import com.vitorpamplona.nestsclient.audio.OpusEncoder
import com.vitorpamplona.nestsclient.audio.SineWaveAudioCapture
import com.vitorpamplona.nestsclient.connectNestsSpeaker
import com.vitorpamplona.nestsclient.connectReconnectingNestsListener
@@ -42,322 +43,561 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import org.junit.Test
import java.net.DatagramSocket
import java.time.Instant
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong
import kotlin.test.assertTrue
/**
* Side-by-side **inter-arrival jitter** measurement for two moq-lite
* audio publishers feeding the same relay through the same listener:
* Side-by-side **pacing + one-way latency** measurement for two
* moq-lite audio publishers feeding the same relay through the same
* listener:
*
* - the Amethyst Kotlin speaker (`connectNestsSpeaker` →
* `MoqLiteNestsSpeaker` → `NestMoqLiteBroadcaster` → moq-lite uni
* streams),
* - the upstream reference `hang-publish` Rust binary
* (`nestsClient/tests/hang-interop/hang-publish`).
* (`nestsClient/tests/hang-interop/hang-publish`), invoked with
* `--log-send-times` so each frame stamps a wall-clock send time
* onto stdout immediately before entering the moq-lite group
* producer.
*
* Both publish identical-shape tracks (440 Hz mono Opus, 20 ms frames,
* 32 kbps) at the same wall-clock cadence into the same `moq-relay`
* the harness brought up; a single Kotlin listener subscribes to
* BOTH speaker pubkeys so the receive path is identical for both
* sides — the variable under test is the **publisher stack's pacing
* fidelity**.
* Both publish identical-shape tracks (440 Hz mono Opus, 32 kbps,
* 20 ms frames, **5 frames per group on both sides** — Kotlin's
* default is 50, overridden here so the per-group QUIC uni-stream
* open/close cost is matched between the two stacks). The single
* Kotlin listener subscribes to BOTH speaker pubkeys so the receive
* path is identical for both sides — the variable under test is the
* **publisher stack's pacing fidelity** AND, with the matched
* send-time stamps, **one-way latency** from "frame entered the
* publisher's outbound buffer" to "frame surfaced through the
* listener's `objects` flow".
*
* Send-time capture, both sides:
* - Kotlin: a [TimestampingOpusEncoder] wrapping [JvmOpusEncoder]
* records `Instant.now().toMicros()` per `encode(...)` call,
* keyed by an internal monotonic frame counter. `encode` is
* called once per PCM frame, immediately before the encoded
* bytes are queued onto the publisher's outbound channel — that
* is the same "moment the frame entered the outbound buffer"
* anchor the Rust side uses.
* - Rust: `--log-send-times` makes hang-publish print one
* `SEND frame=<N> send_t_us=<US>` line per frame to stdout,
* captured with `SystemTime::now()` immediately before
* `frame.encode(group)`. Both stamps come from
* `clock_gettime(CLOCK_REALTIME)` on linux/macOS, so the two
* processes share a clock without a sync handshake.
*
* Frame-to-frame pairing uses `MoqObject.groupId` /
* `MoqObject.objectId` to derive the publisher's absolute frame
* index (`groupId * FRAMES_PER_GROUP + objectId`), so out-of-order
* group delivery — a real possibility under packet loss / parallel
* uni streams — doesn't break the matching.
*
* What we measure (per side):
* - `frames` — how many objects reached the listener inside the
* [DURATION_MS] window. Floor of 80 % of the theoretical maximum
* `(DURATION_MS / FRAME_MS)` proves the publisher actually got
* audio onto the wire and through the relay; values close to the
* ceiling mean no frames were dropped at the publisher's `send`
* site or stalled in the relay's egress.
* - `medianInterArrivalMs`, `p95`, `p99`, `maxInterArrivalMs` —
* consecutive-object delta at the listener. Real-time audio
* wants this tight around `FRAME_MS = 20`; the spread is the
* publisher stack's queue jitter (mic → encoder → publisher.send
* → uni-stream open → QUIC writer → wire), since the relay-side
* fan-out and the listener-side decode are constant.
* - `timeToFirstFrameMs` — wall clock from publisher startup to
* the first frame at the listener. Includes one-time costs
* (handshake, ANNOUNCE / SUBSCRIBE fan-out, first uni-stream
* open). Informative, NOT directly comparable across stacks
* (the Rust side pays a process-spawn that the JVM side doesn't),
* so it's reported but not asserted on.
* - `frames` — delivered objects in the [DURATION_MS] window.
* QUIC streams are reliable, so frame *count* should stay
* close to the theoretical max even under loss; loss instead
* surfaces as a fat **latency tail** from retransmission.
* - Inter-arrival jitter at the listener (`p50` / `p95` / `p99`
* / `max`) — pacing fidelity, independent of the publisher's
* wall clock.
* - **One-way latency** (`p50` / `p95` / `p99` / `max`) —
* `arrival_t - send_t` per matched frame, in ms. This is the
* real-time-audio number; "is our stack as fast as the Rust
* reference" answered directly.
* - `timeToFirstFrameMs` — wallclock from publisher startup to
* first frame at the listener. Informative, includes one-time
* handshake / ANNOUNCE / SUBSCRIBE costs; not directly
* comparable across stacks (Rust pays a process-spawn the JVM
* side doesn't).
*
* What this test does NOT measure:
* - End-to-end glass-to-ear latency — both publishers run their
* own pacing loops; we don't have a synchronised `send_t` from
* `hang-publish` because adding one would require modifying the
* Rust sidecar. Jitter at the listener captures pacing fidelity
* without needing send-side timestamps, which is the part the
* **publisher stack** actually controls.
* - Recovery under packet loss / congestion — both sides run on
* a fast localhost link with no simulator interposed. For loss
* behaviour see the `quic-network-simulator`-based runs in
* `quic/interop/`.
* Two scenarios:
* - `clean_path_*` — direct UDP between publisher and relay,
* localhost link, no loss.
* - `with_5pct_loss_*` — publisher → `udp-loss-shim` → relay.
* 5 % bidirectional packet loss. The listener stays on the
* direct path so any latency growth must come from the
* publisher-side retransmit / ack feedback loop the loss
* induces.
*
* Asserts:
* - Each side delivers ≥ 80 % of the expected frame count
* (catches a publisher that silently stalled).
* - Each side's median inter-arrival sits in `[15, 35] ms`
* (catches a stack that's not pacing at all and just bursts).
* Tail percentiles + `timeToFirstFrameMs` are printed for
* diagnostics but not gated — a CI-grade jitter SLO would have to
* be tuned to the test host and isn't this test's job.
* Asserts (loose, host-tolerant):
* - each side delivers ≥ 80 % of expected frames on the clean
* path, ≥ 60 % under 5 % loss (catches a stalled publisher
* without flapping on a loaded CI host),
* - each side's median inter-arrival sits in `[15, 35] ms` on
* the clean path (catches a no-pacing burst-mode publisher).
*
* Gated by `-DnestsHangInterop=true` because it needs the
* cargo-built `hang-publish` sidecar; same opt-in as every other
* test in `interop/native/`.
* Tail percentiles + one-way latency are printed for human
* inspection. The whole point of the test is the COMPARISON: as
* long as the Kotlin numbers stay close to the Rust numbers in
* every column, the publisher stack is competitive. A regression
* — Kotlin's p99 ballooning to 3× Rust's, say — would be
* immediately visible in the printed output.
*
* Gated by `-DnestsHangInterop=true`; needs the cargo-built
* sidecars (`hang-publish`, `udp-loss-shim`) the
* `interopBuildSidecars` task produces.
*/
class AudioLatencyComparisonTest {
@Test
fun kotlin_speaker_jitter_matches_hang_publish_within_real_time_budget() =
runBlocking {
val harness = NativeMoqRelayHarness.shared()
fun clean_path_pacing_and_one_way_latency() = runComparison(scenario = "clean", lossRate = 0.0f)
// Shared room — both publishers ANNOUNCE inside the same
// moq-lite namespace, each claiming a distinct broadcast
// suffix (their own pubkey). One Kotlin listener subscribes
// to both so the receive path is identical and any jitter
// delta must come from the publisher side.
val hostSigner: NostrSigner = NostrSignerInternal(KeyPair())
val kotlinSpeakerSigner: NostrSigner = NostrSignerInternal(KeyPair())
val rustSpeakerSigner: NostrSigner = NostrSignerInternal(KeyPair())
val kotlinPubkey = kotlinSpeakerSigner.pubKey
val rustPubkey = rustSpeakerSigner.pubKey
@Test
fun under_5pct_packet_loss_pacing_and_one_way_latency() = runComparison(scenario = "loss-5pct", lossRate = 0.05f)
val room =
NestsRoomConfig(
authBaseUrl = "<unused-public-relay>",
endpoint = harness.relayUrl,
hostPubkey = hostSigner.pubKey,
roomId = "lat-${UUID.randomUUID()}",
/**
* Single-scenario driver. Both `@Test` entry points are thin
* wrappers so JUnit reports them as separate cases (and the
* NativeMoqRelayHarness shutdown hook + sidecar reuse stay
* shared across them).
*/
private fun runComparison(
scenario: String,
lossRate: Float,
) = runBlocking {
val harness = NativeMoqRelayHarness.shared()
val hostSigner: NostrSigner = NostrSignerInternal(KeyPair())
val kotlinSpeakerSigner: NostrSigner = NostrSignerInternal(KeyPair())
val rustSpeakerSigner: NostrSigner = NostrSignerInternal(KeyPair())
val kotlinPubkey = kotlinSpeakerSigner.pubKey
val rustPubkey = rustSpeakerSigner.pubKey
val room =
NestsRoomConfig(
authBaseUrl = "<unused-public-relay>",
endpoint = harness.relayUrl,
hostPubkey = hostSigner.pubKey,
roomId = "lat-$scenario-${UUID.randomUUID()}",
)
val pumpScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
val activeShims = mutableListOf<Process>()
var publishProc: Process? = null
var rustLogScraper: Job? = null
var kotlinSpeaker: com.vitorpamplona.nestsclient.NestsSpeaker? = null
try {
val transport =
QuicWebTransportFactory(
parentScope = pumpScope,
certificateValidator = PermissiveCertificateValidator(),
)
val pumpScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
try {
val transport =
QuicWebTransportFactory(
parentScope = pumpScope,
certificateValidator = PermissiveCertificateValidator(),
)
// ---- Listener first. Use the RECONNECTING wrapper so
// its inner re-issuance pump retries SUBSCRIBE with
// exponential backoff (100 → 200 → 400 → 800 → 1 000 ms)
// — without that, the plain listener's
// `subscribeSpeaker` against an unannounced broadcast
// hits "subscribe stream FIN before reply" the moment
// the relay sees no producer. With it, subscribing
// BEFORE either publisher is up is safe; the retry
// catches whichever ANNOUNCE arrives first.
val listener =
connectReconnectingNestsListener(
httpClient = LatencyTestNestsClient,
transport = transport,
scope = pumpScope,
room = room,
signer = hostSigner,
tokenRefreshAfterMs = 0L,
)
withTimeoutOrNull(5_000L) {
listener.state.first { it is NestsListenerState.Connected }
} ?: error("listener never reached Connected within 5 s")
val kotlinSub = listener.subscribeSpeaker(kotlinPubkey)
val rustSub = listener.subscribeSpeaker(rustPubkey)
val kotlinArrivals = ArrayList<Long>(EXPECTED_FRAMES + 64)
val rustArrivals = ArrayList<Long>(EXPECTED_FRAMES + 64)
val kotlinCollect =
pumpScope.launch {
kotlinSub.objects.collect { kotlinArrivals += System.nanoTime() }
}
val rustCollect =
pumpScope.launch {
rustSub.objects.collect { rustArrivals += System.nanoTime() }
}
val publishProc: Process
val kotlinSpeaker =
try {
// Start the Rust process FIRST since process-spawn
// is the slow setup. Capture its t0 right before
// start() and the Kotlin t0 right before
// connectNestsSpeaker; the time-to-first-frame
// numbers below subtract these.
val rustT0 = System.nanoTime()
publishProc =
ProcessBuilder(
harness.hangPublishBin().toString(),
"--relay-url",
"${harness.relayUrl}/${room.moqNamespace()}",
"--broadcast",
rustPubkey,
"--track-name",
"audio/data",
"--duration",
(DURATION_MS / 1_000L).toString(),
"--freq-hz",
"440",
).redirectErrorStream(true)
.also { it.environment()["RUST_LOG"] = "info" }
.start()
// Kotlin speaker. `connectNestsSpeaker` suspends
// through the WebTransport handshake + moq-lite
// SETUP; `startBroadcasting` opens the audio and
// catalog publishers and is the moment "frames
// can start flowing" — capture t0 around that.
val kotlinT0 = System.nanoTime()
val speaker =
connectNestsSpeaker(
httpClient = LatencyTestNestsClient,
transport = transport,
scope = pumpScope,
room = room,
signer = kotlinSpeakerSigner,
speakerPubkeyHex = kotlinPubkey,
captureFactory = { SineWaveAudioCapture(freqHz = 440) },
encoderFactory = { JvmOpusEncoder() },
)
speaker.startBroadcasting()
publisherStartNanos = PublisherStart(kotlinT0, rustT0)
speaker
} catch (t: Throwable) {
kotlinCollect.cancel()
rustCollect.cancel()
pumpScope.coroutineContext[Job]?.cancel()
throw t
}
try {
delay(DURATION_MS + COLLECTION_HEADROOM_MS)
} finally {
runCatching { kotlinSpeaker.close() }
runCatching { publishProc.destroy() }
// Stop the collectors AFTER the publishers — late
// frames sit in QUIC stream buffers and arrive a few
// ms after speaker.close, so cancelling first would
// truncate the tail of the sample.
delay(200)
kotlinCollect.cancel()
rustCollect.cancel()
// ---- Loss path setup. When [lossRate] > 0, stand up two
// udp-loss-shim instances — one per publisher — each
// forwarding 1:1 to the moq-relay's UDP port modulo the
// configured drop fraction. Two shims (not one) because
// the shim latches a single client on first datagram, so a
// shared shim would silently swallow whichever publisher
// spoke second. The listener stays on the direct relay
// path: we want loss between publisher and relay only.
val relayUdpPort = parseUdpPortFrom(harness.relayUrl)
val (kotlinEndpoint, rustEndpoint) =
if (lossRate > 0f) {
val (kotlinShim, kotlinShimPort) =
spawnLossShim(harness, relayUdpPort, lossRate, "kotlin")
val (rustShim, rustShimPort) =
spawnLossShim(harness, relayUdpPort, lossRate, "rust")
activeShims += kotlinShim
activeShims += rustShim
"https://127.0.0.1:$kotlinShimPort" to "https://127.0.0.1:$rustShimPort"
} else {
harness.relayUrl to harness.relayUrl
}
val starts =
checkNotNull(publisherStartNanos) { "publisher t0 should have been captured" }
val kotlinStats =
JitterStats.compute(label = "Kotlin speaker", startNanos = starts.kotlin, arrivals = kotlinArrivals)
val rustStats =
JitterStats.compute(label = "Rust hang-publish", startNanos = starts.rust, arrivals = rustArrivals)
// Print so a CI run captures the comparison even when the
// test passes — that's the whole point of the test (it's
// diagnostic, not regression-gating).
println("===== Audio publisher pacing comparison =====")
println(kotlinStats)
println(rustStats)
println("=============================================")
assertTrue(
kotlinStats.frames >= MIN_FRAMES_ACCEPTED,
"Kotlin speaker delivered only ${kotlinStats.frames} frames, expected " +
"≥ $MIN_FRAMES_ACCEPTED of $EXPECTED_FRAMES — publisher stalled or " +
"the relay dropped the audio track. Stats:\n$kotlinStats",
// ---- Listener first. Use the RECONNECTING wrapper so its
// inner re-issuance pump retries SUBSCRIBE with exponential
// backoff (100 → 200 → 400 → 800 → 1 000 ms) — without it,
// a SUBSCRIBE against an unannounced broadcast hits
// "subscribe stream FIN before reply" and never recovers.
// With it, subscribing BEFORE either publisher is up is
// safe; the retry catches whichever ANNOUNCE arrives first.
val listener =
connectReconnectingNestsListener(
httpClient = LatencyTestNestsClient,
transport = transport,
scope = pumpScope,
room = room,
signer = hostSigner,
tokenRefreshAfterMs = 0L,
)
assertTrue(
rustStats.frames >= MIN_FRAMES_ACCEPTED,
"Rust hang-publish delivered only ${rustStats.frames} frames, expected " +
"≥ $MIN_FRAMES_ACCEPTED of $EXPECTED_FRAMES — sidecar process never " +
"made it to steady state. Stats:\n$rustStats",
withTimeoutOrNull(5_000L) {
listener.state.first { it is NestsListenerState.Connected }
} ?: error("listener never reached Connected within 5 s")
val kotlinSub = listener.subscribeSpeaker(kotlinPubkey)
val rustSub = listener.subscribeSpeaker(rustPubkey)
val kotlinArrivals = ConcurrentHashMap<Long, FrameArrival>()
val rustArrivals = ConcurrentHashMap<Long, FrameArrival>()
// `MoqLiteNestsListener` synthesises [MoqObject.objectId] as
// a per-SUBSCRIPTION monotonic counter (not group-relative),
// so it already maps directly to the publisher's absolute
// frame index — no `groupId * FRAMES_PER_GROUP` arithmetic
// needed, and doing it anyway would skip every frame past
// the first group on each side and produce nonsensical
// negative one-way latencies from misaligned matching.
val kotlinCollect =
pumpScope.launch {
kotlinSub.objects.collect { obj ->
kotlinArrivals[obj.objectId] = FrameArrival(System.nanoTime(), nowEpochMicros())
}
}
val rustCollect =
pumpScope.launch {
rustSub.objects.collect { obj ->
rustArrivals[obj.objectId] = FrameArrival(System.nanoTime(), nowEpochMicros())
}
}
// Start Rust process FIRST: spawn is the slow setup, and
// its --log-send-times stdout drives a background scraper
// that we need running before frames begin to flow.
val rustT0 = System.nanoTime()
val rustSendTimesUs = ConcurrentHashMap<Long, Long>()
publishProc =
ProcessBuilder(
harness.hangPublishBin().toString(),
"--relay-url",
"$rustEndpoint/${room.moqNamespace()}",
"--broadcast",
rustPubkey,
"--track-name",
"audio/data",
"--duration",
(DURATION_MS / 1_000L).toString(),
"--freq-hz",
"440",
"--log-send-times",
).also { it.environment()["RUST_LOG"] = "info" }
// Stdout is the SEND-line channel; keep stderr
// separate so RUST_LOG messages don't interleave
// with the parser's per-frame state.
.redirectErrorStream(false)
.start()
rustLogScraper =
pumpScope.launch(Dispatchers.IO) {
publishProc!!.inputStream.bufferedReader().use { reader ->
reader.lineSequence().forEach { line ->
val match = SEND_LINE_REGEX.matchEntire(line) ?: return@forEach
val frame = match.groupValues[1].toLong()
val sendUs = match.groupValues[2].toLong()
rustSendTimesUs[frame] = sendUs
}
}
}
// Kotlin speaker. Override framesPerGroup to match the Rust
// sidecar's default (5) — the test is about publisher-stack
// fidelity, not about which stack ships which default group
// size, so matching the knob explicitly makes the
// comparison apples-to-apples.
val kotlinT0 = System.nanoTime()
val kotlinSendTimesUs = ConcurrentHashMap<Long, Long>()
val frameCounter = AtomicLong(0L)
val kotlinPublishRoom = room.copy(endpoint = kotlinEndpoint)
kotlinSpeaker =
connectNestsSpeaker(
httpClient = LatencyTestNestsClient,
transport = transport,
scope = pumpScope,
room = kotlinPublishRoom,
signer = kotlinSpeakerSigner,
speakerPubkeyHex = kotlinPubkey,
captureFactory = { SineWaveAudioCapture(freqHz = 440) },
encoderFactory = { TimestampingOpusEncoder(JvmOpusEncoder(), frameCounter, kotlinSendTimesUs) },
framesPerGroup = FRAMES_PER_GROUP.toInt(),
)
kotlinSpeaker.startBroadcasting()
delay(DURATION_MS + COLLECTION_HEADROOM_MS)
// Tear down publishers BEFORE collectors so late frames
// (sitting in QUIC stream buffers — they arrive a few ms
// after speaker.close) still land in `kotlinArrivals` /
// `rustArrivals`. Cancelling the collectors first would
// truncate the tail of the sample.
runCatching { kotlinSpeaker.close() }
runCatching { publishProc.destroy() }
delay(300)
kotlinCollect.cancel()
rustCollect.cancel()
rustLogScraper.cancel()
val kotlinStats =
FrameStats.compute(
label = "Kotlin speaker",
startNanos = kotlinT0,
arrivals = kotlinArrivals,
sendTimesUs = kotlinSendTimesUs,
)
val rustStats =
FrameStats.compute(
label = "Rust hang-publish",
startNanos = rustT0,
arrivals = rustArrivals,
sendTimesUs = rustSendTimesUs,
)
// Print so a CI run captures the comparison even when the
// test passes — that's the whole point of the test (it's
// diagnostic, not regression-gating).
println("===== Audio publisher comparison: $scenario (loss=${"%.0f".format(lossRate * 100)}%) =====")
println(kotlinStats)
println(rustStats)
println("=".repeat(70))
val frameFloor =
if (lossRate > 0f) MIN_FRAMES_UNDER_LOSS else MIN_FRAMES_CLEAN_PATH
assertTrue(
kotlinStats.frames >= frameFloor,
"Kotlin speaker delivered only ${kotlinStats.frames} frames in the " +
"$scenario scenario, expected ≥ $frameFloor of $EXPECTED_FRAMES — " +
"publisher stalled or the relay dropped the audio track. " +
"Stats:\n$kotlinStats",
)
assertTrue(
rustStats.frames >= frameFloor,
"Rust hang-publish delivered only ${rustStats.frames} frames in the " +
"$scenario scenario, expected ≥ $frameFloor of $EXPECTED_FRAMES — " +
"sidecar process never made it to steady state. Stats:\n$rustStats",
)
// Inter-arrival pacing is only meaningful on the clean
// path. Under packet loss, the sender's pacing intent is
// preserved but retransmissions land in bursts, so the
// listener sees clusters even though the publisher paced
// perfectly. Assert pacing on clean only; report it for
// loss but don't gate on it.
if (lossRate == 0f) {
assertTrue(
kotlinStats.medianInterArrivalMs in PACING_MEDIAN_OK_MS,
"Kotlin speaker median inter-arrival ${kotlinStats.medianInterArrivalMs} ms " +
"is outside the sane pacing window $PACING_MEDIAN_OK_MS — implies the " +
"publisher's not pacing at all (just bursting frames into the relay).",
"outside the sane window $PACING_MEDIAN_OK_MS — the publisher's " +
"not pacing at all (just bursting frames into the relay).",
)
assertTrue(
rustStats.medianInterArrivalMs in PACING_MEDIAN_OK_MS,
"Rust hang-publish median inter-arrival ${rustStats.medianInterArrivalMs} ms " +
"is outside the sane pacing window $PACING_MEDIAN_OK_MS — sidecar's " +
"pacing loop is broken (or the relay is queueing).",
"outside the sane window $PACING_MEDIAN_OK_MS — sidecar's pacing " +
"loop is broken (or the relay is queueing).",
)
} finally {
pumpScope.coroutineContext[Job]?.cancel()
}
} finally {
// Always best-effort tear down sidecars + scope, even when
// the test bails partway (assertion / listener-never-
// Connected / sidecar failed to spawn). pumpScope cancel
// cascades to the collectors and the rust-log scraper.
runCatching { kotlinSpeaker?.close() }
runCatching { publishProc?.destroy() }
pumpScope.coroutineContext[Job]?.cancel()
activeShims.forEach { runCatching { it.destroy() } }
}
}
// Captured inside the try because Kotlin doesn't let us write
// `val (kotlinT0, rustT0) = …` across the try boundary.
private var publisherStartNanos: PublisherStart? = null
/**
* Find a free UDP port by binding ephemeral and immediately
* releasing; the kernel briefly reserves the same port for our
* caller via [SO_REUSEADDR]-style handoff. Brief race window
* before the shim binds it back — fine for tests, would NOT be
* fine for production.
*/
private fun pickFreeUdpPort(): Int = DatagramSocket(0).use { it.localPort }
private data class PublisherStart(
val kotlin: Long,
val rust: Long,
/**
* Spawn a `udp-loss-shim` instance binding an ephemeral local
* UDP port, forwarding to [upstreamPort] (moq-relay's UDP
* listener), with [lossRate] applied independently to each
* direction. Returns the process AND the bound listen port so
* the caller can construct the publisher's HTTPS URL.
*
* Blocks briefly for the shim to bind; without a delay, the
* publisher's first packet would race the shim's listen socket
* and land in the kernel's drop counter.
*/
private fun spawnLossShim(
harness: NativeMoqRelayHarness,
upstreamPort: Int,
lossRate: Float,
tag: String,
): Pair<Process, Int> {
val listenPort = pickFreeUdpPort()
val proc =
ProcessBuilder(
harness.udpLossShimBin().toString(),
"--listen",
"127.0.0.1:$listenPort",
"--upstream",
"127.0.0.1:$upstreamPort",
"--loss-rate",
lossRate.toString(),
).redirectErrorStream(true)
.also { it.environment()["RUST_LOG"] = "info" }
.start()
// Give the shim ~150 ms to bind. The shim logs "udp-loss-shim ready"
// at INFO once it's listening; tailing for that line would be
// more deterministic, but adds a reader thread per shim for a
// one-shot signal — a fixed sleep is fine for tests.
Thread.sleep(150)
check(proc.isAlive) {
"udp-loss-shim ($tag) exited immediately on listen=$listenPort upstream=$upstreamPort " +
"loss=$lossRate; stdout=${proc.inputStream.bufferedReader().readText()}"
}
return proc to listenPort
}
/**
* Pull the UDP port out of the harness's `https://127.0.0.1:<port>`
* URL. The harness's moq-relay listens on both TCP (HTTP/WebTransport
* handshake — irrelevant here) and UDP (QUIC) on the same port, so
* the URL's port suffix is what udp-loss-shim's `--upstream` needs.
*/
private fun parseUdpPortFrom(relayUrl: String): Int {
val match = Regex("https://[^:/]+:(\\d+)").find(relayUrl)
return checkNotNull(match) { "could not parse port from relay URL '$relayUrl'" }
.groupValues[1]
.toInt()
}
/**
* Wall-clock microseconds since `UNIX_EPOCH`, matching what the
* Rust sidecar's `SystemTime::now().duration_since(UNIX_EPOCH)`
* emits. `Instant.now()` on the JVM reads `CLOCK_REALTIME`,
* which is the same source — the two values can be compared
* directly when both processes run on the same host.
*/
private fun nowEpochMicros(): Long {
val i = Instant.now()
return i.epochSecond * 1_000_000L + i.nano / 1_000L
}
/**
* Captured arrival point for a single frame. `nanoTime` is the
* JVM-monotonic anchor used for inter-arrival deltas (it doesn't
* jump on NTP step). `epochMicros` is wall-clock, used to pair
* with the publisher's send time and compute one-way latency.
*/
private data class FrameArrival(
val nanoTime: Long,
val epochMicros: Long,
)
/**
* Per-publisher pacing summary. Inter-arrival deltas are computed
* BETWEEN consecutive received objects (skipping the first one,
* which has no predecessor), so a count of `N` arrivals yields
* `N - 1` deltas. Times stored as nanoseconds during collection
* (single allocation per frame) and converted to ms at compute
* time so the on-path collector loop stays allocation-light.
* Per-publisher pacing + latency summary. Inter-arrival deltas
* are computed BETWEEN consecutive received frames in
* publisher-frame order (so out-of-order delivery doesn't skew
* the result); one-way latency is `arrival_us - send_us` per
* matched frame, in ms.
*/
private data class JitterStats(
private data class FrameStats(
val label: String,
val frames: Int,
val matchedFrames: Int,
val timeToFirstFrameMs: Double,
val medianInterArrivalMs: Double,
val p95InterArrivalMs: Double,
val p99InterArrivalMs: Double,
val maxInterArrivalMs: Double,
val medianLatencyMs: Double,
val p95LatencyMs: Double,
val p99LatencyMs: Double,
val maxLatencyMs: Double,
) {
override fun toString(): String =
"%-18s frames=%4d ttf=%6.1f ms inter-arrival: p50=%5.2f p95=%5.2f p99=%5.2f max=%6.2f ms"
.format(
label,
frames,
timeToFirstFrameMs,
medianInterArrivalMs,
p95InterArrivalMs,
p99InterArrivalMs,
maxInterArrivalMs,
)
(
"%-18s frames=%4d ttf=%6.1f ms" +
" inter-arrival p50/95/99/max=%5.2f/%5.2f/%5.2f/%6.2f ms" +
" one-way p50/95/99/max=%6.2f/%6.2f/%6.2f/%7.2f ms (n=%d)"
).format(
label,
frames,
timeToFirstFrameMs,
medianInterArrivalMs,
p95InterArrivalMs,
p99InterArrivalMs,
maxInterArrivalMs,
medianLatencyMs,
p95LatencyMs,
p99LatencyMs,
maxLatencyMs,
matchedFrames,
)
companion object {
fun compute(
label: String,
startNanos: Long,
arrivals: List<Long>,
): JitterStats {
arrivals: Map<Long, FrameArrival>,
sendTimesUs: Map<Long, Long>,
): FrameStats {
if (arrivals.isEmpty()) {
return JitterStats(label, 0, Double.NaN, Double.NaN, Double.NaN, Double.NaN, Double.NaN)
}
val ttfMs = (arrivals.first() - startNanos) / 1_000_000.0
if (arrivals.size < 2) {
return JitterStats(label, arrivals.size, ttfMs, Double.NaN, Double.NaN, Double.NaN, Double.NaN)
}
val deltasMs = DoubleArray(arrivals.size - 1)
for (i in 1 until arrivals.size) {
deltasMs[i - 1] = (arrivals[i] - arrivals[i - 1]) / 1_000_000.0
return FrameStats(
label = label,
frames = 0,
matchedFrames = 0,
timeToFirstFrameMs = Double.NaN,
medianInterArrivalMs = Double.NaN,
p95InterArrivalMs = Double.NaN,
p99InterArrivalMs = Double.NaN,
maxInterArrivalMs = Double.NaN,
medianLatencyMs = Double.NaN,
p95LatencyMs = Double.NaN,
p99LatencyMs = Double.NaN,
maxLatencyMs = Double.NaN,
)
}
// Inter-arrival deltas — sort by publisher frame index
// so out-of-order delivery (parallel uni streams, loss
// retransmits) doesn't distort the histogram.
val sortedArrivals =
arrivals.entries.sortedBy { it.key }.map { it.value }
val ttfMs = (sortedArrivals.first().nanoTime - startNanos) / 1_000_000.0
val deltasMs =
if (sortedArrivals.size < 2) {
DoubleArray(0)
} else {
DoubleArray(sortedArrivals.size - 1).also { d ->
for (i in 1 until sortedArrivals.size) {
d[i - 1] = (sortedArrivals[i].nanoTime - sortedArrivals[i - 1].nanoTime) / 1_000_000.0
}
}
}
deltasMs.sort()
return JitterStats(
// One-way latency — only computable for frames with
// both a send_t AND an arrival_t. The intersection
// size is the `matchedFrames` count we report.
val latenciesMs =
arrivals.entries
.mapNotNull { (frame, arr) ->
val sendUs = sendTimesUs[frame] ?: return@mapNotNull null
(arr.epochMicros - sendUs) / 1_000.0
}.toDoubleArray()
latenciesMs.sort()
return FrameStats(
label = label,
frames = arrivals.size,
matchedFrames = latenciesMs.size,
timeToFirstFrameMs = ttfMs,
medianInterArrivalMs = percentile(deltasMs, 0.50),
p95InterArrivalMs = percentile(deltasMs, 0.95),
p99InterArrivalMs = percentile(deltasMs, 0.99),
maxInterArrivalMs = deltasMs.last(),
maxInterArrivalMs = if (deltasMs.isEmpty()) Double.NaN else deltasMs.last(),
medianLatencyMs = percentile(latenciesMs, 0.50),
p95LatencyMs = percentile(latenciesMs, 0.95),
p99LatencyMs = percentile(latenciesMs, 0.99),
maxLatencyMs = if (latenciesMs.isEmpty()) Double.NaN else latenciesMs.last(),
)
}
/**
* Linear-interpolated percentile (R-7 / "type 7" — the
* default in NumPy, R, and Excel's PERCENTILE). Input
* default in NumPy, R, and Excel's `PERCENTILE`). Input
* MUST already be sorted ascending; we sort the array
* before this is called.
* before this is called. Returns `NaN` on empty input
* rather than throwing, so the [FrameStats] dump for a
* fully-failed publisher still renders.
*/
private fun percentile(
sorted: DoubleArray,
@@ -375,10 +615,46 @@ class AudioLatencyComparisonTest {
}
/**
* Bypass the NIP-98 auth handshake — the harness's moq-relay runs
* with `--auth-public ""`, which grants any path without a JWT.
* Same shape as `ReverseStaticTokenNestsClient` in the other
* native-relay tests, but they're file-private so we can't share.
* Delegating [OpusEncoder] that records `Instant.now()` per
* `encode` call into [sendTimesUs], keyed by an internal
* monotonic frame counter shared across the broadcaster's
* lifetime. `encode` is invoked once per PCM frame, immediately
* before the encoded Opus bytes are queued onto the publisher's
* outbound channel by `NestMoqLiteBroadcaster` — so the
* captured timestamp lines up with the "moment the frame
* entered the publisher's outbound buffer" anchor the Rust
* sidecar uses.
*/
private class TimestampingOpusEncoder(
private val inner: OpusEncoder,
private val frameCounter: AtomicLong,
private val sendTimesUs: MutableMap<Long, Long>,
) : OpusEncoder {
override fun encode(pcm: ShortArray): ByteArray {
val frame = frameCounter.getAndIncrement()
sendTimesUs[frame] = nowEpochMicrosStatic()
return inner.encode(pcm)
}
override fun release() = inner.release()
companion object {
// Static-scope copy so the encoder doesn't pull the
// enclosing test class in (constructed BY the
// broadcaster, lifetime independent of the @Test method).
private fun nowEpochMicrosStatic(): Long {
val i = Instant.now()
return i.epochSecond * 1_000_000L + i.nano / 1_000L
}
}
}
/**
* Bypass the NIP-98 auth handshake — the harness's moq-relay
* runs with `--auth-public ""`, which grants any path without a
* JWT. Same shape as the existing `ReverseStaticTokenNestsClient`
* in the other native-relay tests; they're file-private so we
* can't share.
*/
private object LatencyTestNestsClient : NestsClient {
override suspend fun mintToken(
@@ -389,10 +665,27 @@ class AudioLatencyComparisonTest {
}
companion object {
/** Matches Rust `hang-publish`'s `--log-send-times` output line. */
private val SEND_LINE_REGEX = Regex("""^SEND frame=(\d+) send_t_us=(\d+)$""")
/** moq-lite / Opus frame cadence; matches `hang-publish`'s `FRAME_DURATION_US`. */
private const val FRAME_MS = 20L
/** Measurement window. 10 s = 500 frames per side is plenty for stable P95/P99. */
/**
* Frames per moq-lite group, matched on both sides.
* `hang-publish` hard-codes 5; we override Kotlin's default
* (50) to 5 here so each side opens/closes a uni stream at
* the same cadence — the per-group transport cost is a
* publisher-stack property and matching it makes the rest
* of the comparison meaningful.
*
* `Long` so the `groupId * FRAMES_PER_GROUP + objectId`
* arithmetic in the collector lambdas matches
* `MoqObject.groupId`'s `Long` type without casts.
*/
private const val FRAMES_PER_GROUP = 5L
/** Measurement window. 10 s = 500 frames per side is plenty for stable p95/p99. */
private const val DURATION_MS = 10_000L
/** Extra wall-clock after publishers stop, so the listener drains in-flight uni streams. */
@@ -400,15 +693,23 @@ class AudioLatencyComparisonTest {
private const val EXPECTED_FRAMES = (DURATION_MS / FRAME_MS).toInt()
/** 80 % delivery — sanity floor for "publisher actually got onto the wire". */
private const val MIN_FRAMES_ACCEPTED = (EXPECTED_FRAMES * 0.8).toInt()
/** Clean-path delivery floor: 80 % of expected. */
private const val MIN_FRAMES_CLEAN_PATH = (EXPECTED_FRAMES * 0.8).toInt()
/**
* Median inter-arrival should sit very close to [FRAME_MS] = 20 ms.
* The window catches a publisher that's just bursting frames
* (median would collapse to a fraction of a ms) without being
* so tight it fails on a loaded CI host. Tail percentiles are
* NOT asserted — they're reported for human inspection only.
* Loss-path delivery floor: 60 % of expected. QUIC streams
* are reliable so the count should still be close to 100 %
* (loss surfaces as latency, not drops), but lowering the
* floor leaves headroom for a slow-start retransmit storm
* that eats into the test window.
*/
private const val MIN_FRAMES_UNDER_LOSS = (EXPECTED_FRAMES * 0.6).toInt()
/**
* Median inter-arrival should sit very close to [FRAME_MS] = 20 ms
* on the clean path. The window catches a publisher that's
* just bursting frames (median would collapse) without being
* so tight it false-fails on a loaded CI host.
*/
private val PACING_MEDIAN_OK_MS = 15.0..35.0
}
nestsClient/tests/hang-interop/hang-publish/src/main.rs
+52 -1
View File
@@ -7,7 +7,8 @@
//! the Rust → Amethyst direction. See
//! `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`.
use std::time::Duration;
use std::io::Write;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::{Context, anyhow};
use bytes::Bytes;
@@ -87,6 +88,25 @@ struct Args {
/// exercise the Kotlin listener's publisher-cycle handling.
#[arg(long, default_value_t = 0)]
reconnect_after_ms: u64,
/// If set, emit one line per frame to stdout, immediately before
/// the frame is handed to the moq-lite group producer:
///
/// `SEND frame=<N> send_t_us=<UNIX_EPOCH_MICROS>`
///
/// Used by `AudioLatencyComparisonTest` to pair publisher send
/// time with the listener's arrival time and compute one-way
/// latency. `send_t_us` is the wall-clock microseconds since
/// `UNIX_EPOCH` (CLOCK_REALTIME on linux / macOS), which matches
/// what the JVM's `Instant.now()` reads on the listener side —
/// the two clocks share the same source when both processes run
/// on the same host.
///
/// Disabled by default so existing interop scenarios (I7, etc.)
/// keep their plain stderr-only RUST_LOG output and don't have
/// to filter through 50 frame-tag lines per second on stdout.
#[arg(long, default_value_t = false)]
log_send_times: bool,
}
#[tokio::main]
@@ -364,6 +384,37 @@ async fn publish_cycle(
}
let g = group.as_mut().expect("group always Some after init");
if args.log_send_times {
// Read SystemTime AS LATE AS POSSIBLE before the moq-lite
// group write so the captured `send_t_us` reflects "moment
// the frame entered the publisher's outbound buffer". The
// group write is non-blocking (the per-stream send queue
// does the actual QUIC work asynchronously) so this is the
// closest deterministic anchor we have to "send time" from
// application code. Unwrap on the duration_since: the
// system clock running before UNIX_EPOCH would be a
// catastrophic environment failure, not a test condition.
let send_t_us = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system clock predates UNIX_EPOCH")
.as_micros();
// Bypass tracing — RUST_LOG goes to stderr by config and
// the test parses stdout. Lock + flush per line because
// Rust's `Stdout` is fully-buffered when not a TTY (which
// is exactly our case here — `redirectErrorStream(true)`
// on the JVM side makes stdout a pipe), so a passive
// `writeln!` would batch frames into ~8 KB chunks and the
// parser would see them in bursts instead of as a live
// stream. Per-line flush is cheap (≤ 50 lines/s/publisher).
let stdout = std::io::stdout();
let mut handle = stdout.lock();
let _ = writeln!(
handle,
"SEND frame={} send_t_us={}",
*frame_no, send_t_us
);
let _ = handle.flush();
}
frame.encode(g).context("encode hang frame into group")?;
frames_in_group += 1;
if frames_in_group == FRAMES_PER_GROUP {