Merge branch 'claude/fix-hand-raise-sync-Yf4b4' of https://github.com/vitorpamplona/amethyst into claude/fix-hand-raise-sync-Yf4b4

2026-04-29 18:19:22 -04:00
parent 9b583d99ff 8d228db440
commit 997f52a31a
5 changed files with 195 additions and 4 deletions
@@ -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)
        }
    }
@@ -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.
@@ -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
@@ -159,7 +159,6 @@ class InMemoryQuicPipe(
                when (peeked.type) {
                    LongHeaderType.INITIAL -> initialPnSpace
                    LongHeaderType.HANDSHAKE -> handshakePnSpace
-                    else -> return
                }
            val parsed =
                LongHeaderPacket.parseAndDecrypt(
@@ -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
+    }
+}