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. 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( 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 + } +}