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

fix: drop stale relay-replayed groups on Nests listener reconnect

Browse Source 
On every re-subscribe (listener reconnect or publisher-cycle), the
moq-lite relay re-serves its cached latest group from the first frame.
The wrapper forwarded it straight into the decoder, so in a fully-muted
room the same old clip looped once per reconnect. Track the highest
group sequence delivered by prior subscriptions and drop any group not
strictly newer, mirroring kixelated/hang's Container.Consumer.#run.

https://claude.ai/code/session_01G9h2dzkEj6Y2F1Yr2kCojp
This commit is contained in:
Claude
2026-05-14 18:43:53 +00:00
parent d8d077b4e7
commit 39f2b939c3
2 changed files with 144 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
+43 -3
View File
@@ -45,6 +45,7 @@ import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
/**
@@ -334,6 +335,29 @@ private class ReconnectingHandle(
)
val liveHandleRef = AtomicReference<SubscribeHandle?>(null)
// Cross-subscription replay guard. moq-lite publishers carry
// their group lineage forward monotonically across their own
// session recycles (ReconnectingNestsSpeaker.runHotSwapIteration
// seeds each new publisher with the prior one's nextSequence),
// and on every re-subscribe — listener-side reconnect OR the
// inner publisher-cycle loop — the relay re-serves its cached
// latest group from the first frame. Without a drop filter that
// stale group is replayed straight into the decoder; in a
// fully-muted room, where no newer group is ever produced, it's
// the same ~1 s clip looping once per reconnect. Mirrors
// kixelated/hang's `Container.Consumer.#run`, which drops any
// group not strictly newer than what's already been consumed.
//
// Watermark = highest group sequence delivered by a PRIOR
// underlying subscription. It only advances when a subscription
// ends (the `finally` below), so frames of the *current* group
// are never dropped mid-stream — a live group is always newer
// than the frozen watermark. Cost: a reconnect mid-group drops
// the tail of that one in-progress group (≤ ~1 s of audio),
// which is the right trade for live audio and strictly better
// than replaying already-played speech.
val priorGroupWatermark = AtomicLong(-1L)
// Re-subscribe pump. Two re-issue triggers, layered:
//
// 1. Listener session swap (outer collectLatest) — fires
@@ -432,15 +456,31 @@ private class ReconnectingHandle(
liveHandleRef.set(handle)
Log.d("NestRx") { "wrapper: handle attached id=${handle.subscribeId}, starting collect" }
var emitted = 0L
var droppedStale = 0L
var handleHighestGroup = -1L
try {
handle.objects.collect {
handle.objects.collect { obj ->
if (obj.groupId <= priorGroupWatermark.get()) {
// Stale group re-served from the relay
// cache after a re-subscribe — drop so
// it never reaches the decoder.
droppedStale += 1
return@collect
}
if (obj.groupId > handleHighestGroup) handleHighestGroup = obj.groupId
emitted += 1
frames.emit(it)
frames.emit(obj)
}
} finally {
if (liveHandleRef.get() === handle) liveHandleRef.set(null)
// Advance the watermark only now the subscription
// is done, so the next re-subscribe drops every
// group up to and including the last one we saw.
if (handleHighestGroup > priorGroupWatermark.get()) {
priorGroupWatermark.set(handleHighestGroup)
}
}
Log.w("NestRx") { "wrapper: handle objects flow ENDED id=${handle.subscribeId} emitted=$emitted — re-issuing after ${RESUBSCRIBE_BACKOFF_MS}ms" }
Log.w("NestRx") { "wrapper: handle objects flow ENDED id=${handle.subscribeId} emitted=$emitted droppedStale=$droppedStale — re-issuing after ${RESUBSCRIBE_BACKOFF_MS}ms" }
// Brief backoff so a permanently-gone
// publisher doesn't tight-loop the relay
// with re-subscribes. 100 ms stays well
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt
+101 -4
View File
@@ -137,10 +137,13 @@ class ReconnectingNestsListenerTest {
}
}
private fun frame(payload: ByteArray): MoqObject =
private fun frame(
payload: ByteArray,
groupId: Long = 0L,
): MoqObject =
MoqObject(
trackAlias = 1L,
groupId = 0L,
groupId = groupId,
objectId = 0L,
publisherPriority = 0,
payload = payload,
@@ -232,7 +235,7 @@ class ReconnectingNestsListenerTest {
first.frames.subscriptionCount.first { it > 0 }
}
first.frames.emit(frame(byteArrayOf(0x01)))
first.frames.emit(frame(byteArrayOf(0x01), groupId = 0L))
// Wait for FRAME1 to traverse pump → wrapper.frames →
// consumer collector. Without this sync the next
@@ -261,7 +264,11 @@ class ReconnectingNestsListenerTest {
second.frames.subscriptionCount.first { it > 0 }
}
second.frames.emit(frame(byteArrayOf(0x02)))
// Post-reconnect audio lands in a newer group — moq-lite
// publishers carry their group lineage forward
// monotonically across recycles, so the second session's
// first real frame is group 1, not a replay of group 0.
second.frames.emit(frame(byteArrayOf(0x02), groupId = 1L))
val result = collected.await()
assertEquals(2, result.size)
@@ -283,6 +290,96 @@ class ReconnectingNestsListenerTest {
}
}
/**
* Regression: after a reconnect the relay re-serves its cached
* latest group from the first frame. moq-lite publishers carry
* group sequences forward monotonically across their own recycles,
* so a re-served group has a sequence we've already consumed — the
* wrapper must drop it instead of replaying already-played audio
* into the decoder. In a fully-muted room (no newer group ever
* produced) the un-dropped replay loops the same clip once per
* reconnect.
*/
@Test
fun reconnect_does_not_replay_already_consumed_group() =
runBlocking {
val scope = CoroutineScope(SupervisorJob())
try {
val first = ScriptedListener(room)
val second = ScriptedListener(room)
val listenersInOrder = mutableListOf(first, second)
val reconnecting =
connectReconnectingNestsListener(
httpClient = httpClient,
transport = transport,
scope = scope,
room = room,
signer = signer,
policy = NestsReconnectPolicy(initialDelayMs = 1L),
connector = { listenersInOrder.removeAt(0) },
)
withTimeout(5_000L) {
reconnecting.state.first { it is NestsListenerState.Connected }
}
val handle = reconnecting.subscribeSpeaker("speaker-pubkey")
val consumerSubscribed = CompletableDeferred<Unit>()
val consumerProgress = MutableStateFlow(0)
@Suppress("UNCHECKED_CAST")
val objectsAsShared = handle.objects as SharedFlow<MoqObject>
// Expect exactly two frames: the original group-0 frame
// and the genuinely-new group-1 frame. The group-0 frame
// re-served by the second session must NOT appear.
val collected =
scope.async {
withTimeout(5_000L) {
objectsAsShared
.onSubscription { consumerSubscribed.complete(Unit) }
.take(2)
.onEach { consumerProgress.value += 1 }
.toList()
}
}
withTimeout(5_000L) { consumerSubscribed.await() }
withTimeout(5_000L) {
first.frames.subscriptionCount.first { it > 0 }
}
first.frames.emit(frame(byteArrayOf(0x01), groupId = 0L))
withTimeout(5_000L) {
consumerProgress.first { it >= 1 }
}
first.fail("scripted-disconnect")
withTimeout(5_000L) {
second.frames.subscriptionCount.first { it > 0 }
}
// The relay re-serves the cached latest group (0) from
// the start — this is the stale replay that must be
// dropped.
second.frames.emit(frame(byteArrayOf(0x02), groupId = 0L))
// ...followed by genuinely new audio in the next group.
second.frames.emit(frame(byteArrayOf(0x03), groupId = 1L))
val result = collected.await()
assertEquals(2, result.size)
assertEquals(0x01.toByte(), result[0].payload[0])
// 0x02 (re-served group 0) dropped; 0x03 (group 1) kept.
assertEquals(0x03.toByte(), result[1].payload[0])
handle.unsubscribe()
reconnecting.close()
} finally {
scope.cancel()
}
}
@Test
fun unsubscribe_before_session_swap_releases_handle() =
runBlocking {