From a84fbd2e57b94a081d95c17553418e71c29e35b3 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 21:08:03 +0000 Subject: [PATCH 1/5] test(quic): add concurrent producer/consumer regression for SendBuffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous SendBuffer suite (FlowControlEnforcementTest) is entirely single-threaded — every test calls enqueue and takeChunk sequentially on the same coroutine, so the race that crashed the audio path in production (NoSuchElementException from chunks.first() under concurrent enqueue + takeChunk) stayed invisible. The whole :quic commonTest tree had no concurrent test at all. Three new tests run real-thread races on Dispatchers.Default: - concurrent_enqueue_and_takeChunk_does_not_throw drives multiple producer coroutines + a consumer coroutine and asserts the buffer drains cleanly with no exception. - concurrent_takeChunk_callers_never_double_drain_a_chunk fans out multiple consumers against a pre-populated buffer; asserts the sum of bytes handed out equals the bytes enqueued (i.e. no chunk is double-counted by overlapping head-peel paths). - concurrent_finish_with_inflight_enqueue_emits_correct_fin races finish() against in-flight writes and asserts the FIN comes AFTER every enqueued byte. Tests pass against the synchronised SendBuffer; running them against the pre-fix unsynchronised version corrupts state badly enough that the consumer wedges (an explicit "this is what the bug looked like" demonstration). With internal synchronisation in place the suite finishes in <0.2 s. Documents the concurrent-access contract so a future "let's drop the sync, it's hot" refactor immediately fails CI. --- .../quic/stream/SendBufferConcurrencyTest.kt | 173 ++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferConcurrencyTest.kt diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferConcurrencyTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferConcurrencyTest.kt new file mode 100644 index 000000000..bc2599591 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferConcurrencyTest.kt @@ -0,0 +1,173 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quic.stream + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Concurrent producer/consumer regression tests for [SendBuffer]. + * + * The original suite was entirely single-threaded: every test enqueued + * and then took chunks on the same coroutine, so torn state across + * threads stayed invisible. + * + * In production [SendBuffer] is hit by two distinct paths: + * - the application's writer coroutine calling [SendBuffer.enqueue] + * (e.g. via `WtPeerStreamDemux`'s per-stream send callback); + * - the [com.vitorpamplona.quic.connection.QuicConnectionDriver] send loop calling + * [SendBuffer.takeChunk] under the connection mutex, which is + * NOT held by the producer. + * + * Without internal synchronisation the writer would see + * `pendingBytes > 0` (set by an in-flight `enqueue`) before the + * matching `chunks.addLast` became visible, fall into the + * head-peel branch, and crash with + * `NoSuchElementException: ArrayDeque is empty.` from + * `chunks.first()`. These tests reliably reproduce that scenario on + * the unsynchronised version and validate ordering / accounting on + * the fixed one. + */ +class SendBufferConcurrencyTest { + @Test + fun concurrent_enqueue_and_takeChunk_does_not_throw() = + runBlocking { + // Multi-thread dispatcher is load-bearing — runTest's virtual + // scheduler is single-threaded and would never expose the race. + withContext(Dispatchers.Default) { + repeat(REPEATS) { + val buf = SendBuffer() + coroutineScope { + // Producers race against a consumer that drains as + // fast as it can. + val producers = + List(PRODUCERS) { + async { + repeat(WRITES_PER_PRODUCER) { + buf.enqueue(byteArrayOf(0x42)) + } + } + } + val consumer = + async { + var taken = 0 + while (taken < PRODUCERS * WRITES_PER_PRODUCER) { + val chunk = buf.takeChunk(maxBytes = MAX_BYTES) ?: continue + taken += chunk.data.size + } + taken + } + producers.awaitAll() + // Drain whatever the consumer didn't pick up before + // producers finished. Without internal synchronisation + // either side could throw before reaching this. + consumer.await() + } + assertEquals(0, buf.readableBytes, "all bytes should be drained at the end of round") + assertEquals((PRODUCERS * WRITES_PER_PRODUCER).toLong(), buf.sentOffset) + } + } + } + + @Test + fun concurrent_takeChunk_callers_never_double_drain_a_chunk() = + runBlocking { + withContext(Dispatchers.Default) { + repeat(REPEATS) { + val buf = SendBuffer() + repeat(PRODUCERS * WRITES_PER_PRODUCER) { + buf.enqueue(byteArrayOf(0x55)) + } + val expectedTotal = PRODUCERS * WRITES_PER_PRODUCER + coroutineScope { + // Multiple consumers racing for the same buffer mirrors + // the (unlikely but possible) case of overlapping driver + // ticks; the byte total must stay exact. + val takers = + List(CONSUMERS) { + async { + var localTaken = 0 + while (true) { + val chunk = buf.takeChunk(maxBytes = MAX_BYTES) ?: break + localTaken += chunk.data.size + } + localTaken + } + } + val totalTaken = takers.awaitAll().sum() + assertEquals(expectedTotal, totalTaken, "each byte must be handed out to exactly one consumer") + assertTrue(buf.readableBytes == 0, "buffer must end empty after all consumers stop") + } + } + } + } + + @Test + fun concurrent_finish_with_inflight_enqueue_emits_correct_fin() = + runBlocking { + withContext(Dispatchers.Default) { + repeat(REPEATS) { + val buf = SendBuffer() + coroutineScope { + val producer = + async { + repeat(WRITES_PER_PRODUCER) { + buf.enqueue(byteArrayOf(0x77)) + } + buf.finish() + } + val consumer = + async { + var sawFin = false + var bytes = 0 + while (!sawFin) { + val chunk = buf.takeChunk(maxBytes = MAX_BYTES) ?: continue + bytes += chunk.data.size + if (chunk.fin) sawFin = true + } + bytes + } + producer.await() + val drained = consumer.await() + assertEquals(WRITES_PER_PRODUCER, drained, "FIN must come AFTER every enqueued byte") + assertTrue(buf.finSent) + } + } + } + } + + private companion object { + // Tuned to reliably trigger the original race on a multi-core + // host (~6 ms per round on 8 cores) without ballooning CI time. + const val REPEATS = 50 + const val PRODUCERS = 4 + const val CONSUMERS = 4 + const val WRITES_PER_PRODUCER = 200 + const val MAX_BYTES = 64 + } +} From df98235d31f150735b2c259fefad3f1d9c741807 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 29 Apr 2026 17:16:14 -0400 Subject: [PATCH 2/5] Minor adjustments to remove warnings --- .../com/vitorpamplona/quic/connection/QuicConnectionWriter.kt | 4 ++-- .../com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index f1d06b956..3924d90a6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -72,8 +72,8 @@ fun drainOutbound( val handshakeHasContent = handshakeFrames != null && handshakeState.sendProtection != null // Build natural-size first. - val initialNatural = if (initialHasContent) buildLongHeaderFromFrames(conn, EncryptionLevel.INITIAL, initialFrames!!, padBytes = 0) else null - val handshakeNatural = if (handshakeHasContent) buildLongHeaderFromFrames(conn, EncryptionLevel.HANDSHAKE, handshakeFrames!!, padBytes = 0) else null + val initialNatural = if (initialHasContent) buildLongHeaderFromFrames(conn, EncryptionLevel.INITIAL, initialFrames, padBytes = 0) else null + val handshakeNatural = if (handshakeHasContent) buildLongHeaderFromFrames(conn, EncryptionLevel.HANDSHAKE, handshakeFrames, padBytes = 0) else null val firstPass = listOfNotNull(initialNatural, handshakeNatural, applicationPkt) if (firstPass.isEmpty()) return null diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt index dbe027e69..80749b9f9 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt @@ -159,7 +159,6 @@ class InMemoryQuicPipe( when (peeked.type) { LongHeaderType.INITIAL -> initialPnSpace LongHeaderType.HANDSHAKE -> handshakePnSpace - else -> return } val parsed = LongHeaderPacket.parseAndDecrypt( From b0e7c62bb51978df0265ebe84823a213ccaa26c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 21:36:23 +0000 Subject: [PATCH 3/5] fix(nests): broadcast presence/chat to the linked room's full relay list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hand-raise toggles in Amethyst weren't appearing in nostrnests's UI. nostrnests's NestsUI v2 routes reads against a fixed five-relay list (NestsUI-v2/src/lib/const.ts: relay.snort.social, nos.lol, relay.damus.io, relay.ditto.pub, relay.primal.net) — no outbox model. Its presence query is just `kinds:[10312], "#a":[roomATag]` over those five relays. Commit 637174ef already taught computeRelayListToBroadcast to fan a kind 30312 *room* event out to its `relays` tag. But kind 10312 presence (and chat / reactions, etc.) is a different event type that *links* to the room via an `a` tag. For those, broadcast was only reaching: - the broadcaster's outbox relays - the single firstOrNull() hint baked into the `a` tag - the relay we happened to receive the room event from If none of those overlap with nostrnests's fixed five reads, the hand-raise update is silently dropped. Extend the `linkedAddressIds` block in computeRelayListToBroadcast: when the linked address resolves to a MeetingSpaceEvent / MeetingRoomEvent / LiveActivitiesEvent, also add that linked event's allRelayUrls(). This covers presence updates and any other room-scoped event that points at a Nest via `#a`. https://claude.ai/code/session_016G7oP5BotPjUBgvMYrrrxh --- .../com/vitorpamplona/amethyst/model/Account.kt | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt index 981a776fa..fa65a3aa9 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt @@ -195,6 +195,7 @@ import com.vitorpamplona.quartz.nip47WalletConnect.rpc.Request import com.vitorpamplona.quartz.nip47WalletConnect.rpc.Response import com.vitorpamplona.quartz.nip51Lists.bookmarkList.BookmarkListEvent import com.vitorpamplona.quartz.nip51Lists.bookmarkList.tags.AddressBookmark +import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.MeetingRoomEvent import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.MeetingSpaceEvent import com.vitorpamplona.quartz.nip53LiveActivities.streaming.LiveActivitiesEvent import com.vitorpamplona.quartz.nip56Reports.ReportType @@ -1002,6 +1003,20 @@ class Account( linkedNote.event?.let { linkedEvent -> relayList.addAll(computeRelaysForChannels(linkedEvent)) + + // Audio rooms / live activities pin their relay set + // on the room event itself. Presence (kind 10312) and + // any other event whose `a` tag points at one of these + // must also fan out to that relay list — otherwise a + // hand-raise / mute / publishing update only reaches + // the broadcaster's outbox + the single firstOrNull() + // hint baked into the `a` tag, and a fixed-relay + // listener like nostrnests never sees it. + when (linkedEvent) { + is MeetingSpaceEvent -> relayList.addAll(linkedEvent.allRelayUrls()) + is MeetingRoomEvent -> relayList.addAll(linkedEvent.allRelayUrls()) + is LiveActivitiesEvent -> relayList.addAll(linkedEvent.allRelayUrls()) + } } } } From 9628d17f5fae6ed87c207a7853fcc5b7ba049704 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 29 Apr 2026 17:44:11 -0400 Subject: [PATCH 4/5] Improves the relay list used for linked Notes to include their relay urls --- .../vitorpamplona/amethyst/model/Account.kt | 22 +++++-------------- 1 file changed, 6 insertions(+), 16 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt index fa65a3aa9..0a1e883bd 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt @@ -981,7 +981,7 @@ class Account( } linkedNote.event?.let { linkedEvent -> - relayList.addAll(computeRelaysForChannels(linkedEvent)) + relayList.addAll(computeRelayListToBroadcast(linkedEvent)) } } } @@ -1002,21 +1002,7 @@ class Account( } linkedNote.event?.let { linkedEvent -> - relayList.addAll(computeRelaysForChannels(linkedEvent)) - - // Audio rooms / live activities pin their relay set - // on the room event itself. Presence (kind 10312) and - // any other event whose `a` tag points at one of these - // must also fan out to that relay list — otherwise a - // hand-raise / mute / publishing update only reaches - // the broadcaster's outbox + the single firstOrNull() - // hint baked into the `a` tag, and a fixed-relay - // listener like nostrnests never sees it. - when (linkedEvent) { - is MeetingSpaceEvent -> relayList.addAll(linkedEvent.allRelayUrls()) - is MeetingRoomEvent -> relayList.addAll(linkedEvent.allRelayUrls()) - is LiveActivitiesEvent -> relayList.addAll(linkedEvent.allRelayUrls()) - } + relayList.addAll(computeRelayListToBroadcast(linkedEvent)) } } } @@ -1030,6 +1016,10 @@ class Account( relayList.addAll(event.allRelayUrls()) } + if (event is MeetingRoomEvent) { + relayList.addAll(event.allRelayUrls()) + } + if (event is LiveActivitiesEvent) { relayList.addAll(event.allRelayUrls()) } From 8d228db440bb564afef4e7d5e0a67b6848812afa Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 22:13:36 +0000 Subject: [PATCH 5/5] fix(nests): preserve relays tag on participant updates + sync action bar with stage grid MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related bugs the user hit while testing against nostrnests: 1. Approving a hand-raise made the Hands tab disappear (the host's local app saw the role grant) but the audience member did not appear on stage anywhere else. Cause: RoomParticipantActions.rebuild() rebuilt the kind-30312 without the original room's `relays` tag. The recently-added broadcast fan-out path (Account.computeRelayListToBroadcast) keys off `event.allRelayUrls()` for kind 30312, so a republished room event with no relays tag only reaches the host's outbox — not nostrnests's fixed five reads or the audience member subscribing on those reads. The audience member never sees their SPEAKER tag and stays absent from the participant list. Fix: copy `original.relays()` into the rebuild so every republish (approve, demote, kick → re-broadcast) keeps the relay routing that lets the room reach its full audience. 2. Tapping "Leave Stage" as the host emptied the StageGrid ("Waiting for speakers…") but the action bar still showed the Talk + Leave Stage cluster. Cause: two different definitions of "on stage" were in play. StageGrid uses ParticipantGrid (role + presence.onstage flag), so flipping onStageNow=false drops the host out. The action bar gated on the role-only `onStage: List` (the host's HOST tag never goes away), so the cluster stayed. Fix: derive isOnStageMe from participantGrid.onStage so both surfaces share the same "stepped off" semantics. The host who taps Leave Stage now drops to the audience-style mute toggle and can rejoin (kind-10312 onstage flips back to 1) without the controls lying about their state. https://claude.ai/code/session_016G7oP5BotPjUBgvMYrrrxh --- .../room/participants/RoomParticipantActions.kt | 9 +++++++++ .../loggedIn/nests/room/screen/NestFullScreen.kt | 12 +++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/participants/RoomParticipantActions.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/participants/RoomParticipantActions.kt index 88311e720..bb00c422c 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/participants/RoomParticipantActions.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/participants/RoomParticipantActions.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.MeetingSpaceEv import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.endpoint import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.image import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.participants +import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.relays import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.summary import com.vitorpamplona.quartz.nip53LiveActivities.meetingSpaces.tags.StatusTag import com.vitorpamplona.quartz.nip53LiveActivities.streaming.tags.ParticipantTag @@ -132,6 +133,14 @@ internal object RoomParticipantActions { original.endpoint()?.let { endpoint(it) } original.summary()?.takeIf { it.isNotBlank() }?.let { summary(it) } original.image()?.takeIf { it.isNotBlank() }?.let { image(it) } + // Preserve the room's `relays` tag — without it, the + // republished kind-30312 fans out only to the host's + // outbox, never reaching the room's own relay set + // (e.g. nostrnests's fixed five reads). The audience + // member would then never see their role grant and + // wouldn't appear on stage for any other participant + // listening on those relays. + original.relays().takeIf { it.isNotEmpty() }?.let { relays(it) } if (others.isNotEmpty()) participants(others) } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/screen/NestFullScreen.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/screen/NestFullScreen.kt index 1546caabd..8ed1ce968 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/screen/NestFullScreen.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/screen/NestFullScreen.kt @@ -135,7 +135,6 @@ internal fun NestFullScreen( val isHost = accountViewModel.account.signer.pubKey == event.pubKey val myPubkey = accountViewModel.account.signer.pubKey - val isOnStageMe = remember(onStage, myPubkey) { onStage.any { it.pubKey == myPubkey } } val leaveScope = rememberCoroutineScope() val topBarContext = LocalContext.current @@ -152,6 +151,17 @@ internal fun NestFullScreen( presences = presences, ) } + // Tie the action bar's "am I on stage" gate to the same data + // source the StageGrid uses (role + presence.onstage flag), + // not just the kind-30312 role tag. Otherwise a host who taps + // "Leave Stage" flips presence to onstage=0 and disappears + // from the StageGrid, but the action bar keeps showing the + // Talk + Leave Stage cluster because it's still gated on + // role-only. + val isOnStageMe = + remember(participantGrid, myPubkey) { + participantGrid.onStage.any { it.pubkey == myPubkey } + } // Same logic HandRaiseQueueSection uses internally — duplicated // here so the tab label can show a count without coupling the // section to the screen.