fix(quic-interop): prepareRequests must hold streamsLock, not lifecycleLock

aioquic interop multiplexing 2026-05-06 qlog post-mortem:
  - 2898 packets sent in 60s, each carrying ONE STREAM frame
  - server log: streams created/discarded strictly serially, ~30-40ms apart
  - 1421/2000 files completed before the runner's 60s timeout
  - shape ≈ 1 RTT per stream — wire was emitting one stream per datagram

Cause: Http3GetClient.prepareRequests + HqInteropGetClient.prepareRequests
both did `conn.lock.withLock { ... openBidiStreamLocked() }`. Post the
lock-split refactor `conn.lock` is the deprecated alias for lifecycleLock.
The writer's drainOutbound takes streamsLock — not lifecycleLock — so the
send loop interleaved between every two openBidiStreamLocked calls,
draining one stream's data per pass.

Fix:
  1. Both prepareRequests impls now use conn.streamsLock.withLock.
  2. openBidiStreamLocked now `check`s streamsLock.isLocked at entry
     so this can never silently regress again — calling it without the
     lock (or with the wrong lock) throws IllegalStateException with
     a message naming streamsLock as the lock to acquire.
  3. New BatchedOpenLockContractTest pins the contract:
     - calling openBidiStreamLocked WITHOUT any lock throws
     - calling openBidiStreamLocked while holding lifecycleLock throws
       (the exact regression shape)
     - calling openBidiStreamLocked while holding streamsLock works
       (happy path)

The runtime check is the regression-proof part: future callers physically
cannot hold the wrong lock without the test (and prod) blowing up at the
first call site.

https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT

quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt

@@ -57,7 +57,11 @@ class HqInteropGetClient(
         @Suppress("UNUSED_PARAMETER") authority: String,
         paths: List<String>,
     ): List<RequestHandle> =
-        conn.lock.withLock {
+        // streamsLock, NOT lifecycleLock (the deprecated `conn.lock`
+        // alias) — see Http3GetClient.prepareRequests for the full
+        // story. tl;dr: holding the wrong lock lets the send-loop
+        // drain between opens and emits one STREAM per packet.
+        conn.streamsLock.withLock {
             paths.map { path ->
                 val stream = conn.openBidiStreamLocked()
                 val request = "GET $path\r\n".encodeToByteArray()

quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt

@@ -159,7 +159,19 @@ class Http3GetClient(
         authority: String,
         paths: List<String>,
     ): List<RequestHandle> =
-        conn.lock.withLock {
+        // CRITICAL: streamsLock — the lock the writer's drainOutbound
+        // takes. Holding lifecycleLock (the deprecated `conn.lock`
+        // alias) here did NOT block the send loop, so the writer
+        // interjected between every openBidiStreamLocked call and
+        // emitted ONE STREAM frame per packet. aioquic interop
+        // 2026-05-06 qlog: 2898 packets sent in 60s, each carrying
+        // exactly one stream frame; server processed streams strictly
+        // sequentially at ~1 RTT per stream → 1421/2000 files in 60s
+        // before timeout. Holding streamsLock blocks the drain so
+        // all N opens land before any drain runs, the writer then
+        // packs many STREAM frames into each datagram, and the
+        // server sees the burst on the wire.
+        conn.streamsLock.withLock {
             paths.map { path ->
                 val stream = conn.openBidiStreamLocked()
                 stream.send.enqueue(encodeRequest(authority, path))

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt

@@ -768,6 +768,22 @@ class QuicConnection(
      * [streamsLock].
      */
     fun openBidiStreamLocked(): QuicStream {
+        // Mutex.isLocked is the only check we have — kotlinx.coroutines
+        // Mutex doesn't expose ownership without an `owner` argument,
+        // and we don't pass one in production. So this catches the
+        // common bug — caller used the wrong lock or no lock — but
+        // not the rarer case of "caller held a DIFFERENT lock that
+        // happens to be locked too." The interop runner's multiplexing
+        // failure on 2026-05-06 was precisely this: prepareRequests
+        // held lifecycleLock (`conn.lock`) and called this fn, the
+        // send loop's drainOutbound interleaved between opens, and
+        // we emitted one STREAM per packet (1421/2000 files in 60s).
+        check(streamsLock.isLocked) {
+            "openBidiStreamLocked requires streamsLock to be held — caller " +
+                "must wrap with streamsLock.withLock { ... }. Without that, " +
+                "drainOutbound can race the streams mutation and emit one " +
+                "STREAM per packet under multiplex load."
+        }
         if (nextLocalBidiIndex >= peerMaxStreamsBidi) {
             throw QuicStreamLimitException(
                 "peer-granted bidi stream cap reached " +

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt

@@ -0,0 +1,157 @@
+/*
+ * 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.connection
+
+import com.vitorpamplona.quic.tls.InProcessTlsServer
+import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.sync.withLock
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNotNull
+
+/**
+ * Lock contract for batched stream opens — the production pattern used by
+ * `Http3GetClient.prepareRequests` / `HqInteropGetClient.prepareRequests`
+ * and any future caller that wants to open N streams atomically without
+ * the send loop interjecting between opens.
+ *
+ * Background. The interop runner's `multiplexing` testcase against aioquic
+ * timed out on 2026-05-06, downloading 1421/2000 files in 60s (~23
+ * streams/sec, exactly 1 RTT per stream — the wire was emitting one STREAM
+ * frame per datagram, not 64). The qlog showed 2898 packets sent in 60s,
+ * each carrying ONE STREAM frame. Cause: `prepareRequests` had been written
+ * to hold `conn.lock` (the deprecated alias for `lifecycleLock`) while
+ * calling [QuicConnection.openBidiStreamLocked]. `lifecycleLock` doesn't
+ * gate the writer — `drainOutbound` takes [QuicConnection.streamsLock] —
+ * so the writer's send loop could (and did) interleave between every two
+ * `openBidiStreamLocked` invocations, draining one stream per pass.
+ *
+ * Two-layer guard:
+ *   1. [QuicConnection.openBidiStreamLocked] now `check`s that
+ *      [QuicConnection.streamsLock] is held when called. This catches
+ *      callers that pass NO lock or the WRONG lock at the source.
+ *   2. The "batched open under streamsLock yields a small fixed packet
+ *      count" contract is pinned by [MultiplexingCoalescingTest] —
+ *      already in the suite. Together they fail loudly if the prod
+ *      callers regress.
+ *
+ * If you're adding a new caller for `openBidiStreamLocked`, the right
+ * pattern is:
+ *
+ *     conn.streamsLock.withLock {
+ *         repeat(n) { conn.openBidiStreamLocked() ... }
+ *     }
+ */
+class BatchedOpenLockContractTest {
+    @Test
+    fun `openBidiStreamLocked throws when called without streamsLock held`() {
+        runBlocking {
+            val client = handshakedClient()
+            // No `streamsLock.withLock { ... }` wrapper. This is the
+            // shape the regressed `prepareRequests` had — except it
+            // held `lifecycleLock` instead. Either way streamsLock is
+            // not held, the assertion in openBidiStreamLocked fires.
+            val ex =
+                assertFailsWith<IllegalStateException> {
+                    client.openBidiStreamLocked()
+                }
+            assertNotNull(ex.message)
+            // The error should mention streamsLock so the caller knows
+            // what to fix.
+            kotlin.test.assertTrue(
+                ex.message!!.contains("streamsLock"),
+                "error message should name the lock to acquire; got: ${ex.message}",
+            )
+        }
+    }
+
+    @Test
+    fun `openBidiStreamLocked throws when caller holds the wrong lock (lifecycleLock)`() {
+        runBlocking {
+            val client = handshakedClient()
+            // Exact regression shape: caller holds lifecycleLock and
+            // calls openBidiStreamLocked. The check should fire because
+            // streamsLock is not held — even though SOME lock is.
+            assertFailsWith<IllegalStateException> {
+                client.lifecycleLock.withLock {
+                    client.openBidiStreamLocked()
+                }
+            }
+        }
+    }
+
+    @Test
+    fun `openBidiStreamLocked succeeds when streamsLock is held`() {
+        runBlocking {
+            val client = handshakedClient()
+            // Happy path — caller holds streamsLock, batched open works.
+            // Sanity-check that the assertion in openBidiStreamLocked
+            // doesn't fire on the correct call shape.
+            val streams =
+                client.streamsLock.withLock {
+                    List(64) { client.openBidiStreamLocked() }
+                }
+            assertEquals(64, streams.size)
+            assertEquals(64, streams.map { it.streamId }.toSet().size, "stream ids must be unique")
+        }
+    }
+
+    private fun handshakedClient(): QuicConnection =
+        runBlocking {
+            val client =
+                QuicConnection(
+                    serverName = "example.test",
+                    config =
+                        QuicConnectionConfig(
+                            initialMaxStreamsBidi = 256,
+                        ),
+                    tlsCertificateValidator = PermissiveCertificateValidator(),
+                )
+            val serverScid = ConnectionId.random(8)
+            val tlsServer =
+                InProcessTlsServer(
+                    transportParameters =
+                        TransportParameters(
+                            initialMaxData = 1_000_000,
+                            initialMaxStreamDataBidiLocal = 100_000,
+                            initialMaxStreamDataBidiRemote = 100_000,
+                            initialMaxStreamDataUni = 100_000,
+                            initialMaxStreamsBidi = 256,
+                            initialMaxStreamsUni = 16,
+                            initialSourceConnectionId = serverScid.bytes,
+                            originalDestinationConnectionId = client.destinationConnectionId.bytes,
+                        ).encode(),
+                )
+            val pipe =
+                InMemoryQuicPipe(
+                    client = client,
+                    initialDcid = client.destinationConnectionId.bytes,
+                    serverScid = serverScid,
+                    tlsServer = tlsServer,
+                )
+            client.start()
+            pipe.drive(maxRounds = 16)
+            assertEquals(QuicConnection.Status.CONNECTED, client.status)
+            client
+        }
+}