Merge branch 'worktree-agent-acb67f8575e4086eb' into claude/research-quic-libraries-hH1Dc

quic/plans/2026-05-08-lock-split-design.md

@@ -0,0 +1,223 @@
+# QuicConnection Lock Split — Design Note
+
+Date: 2026-05-08
+
+## Problem
+
+`QuicConnection.lock: Mutex` serialises every meaningful operation:
+
+  - `drainOutbound` (send loop, holds lock during a full datagram build —
+    iterates every stream, allocates packet numbers, encrypts).
+  - `feedDatagram` (read loop, holds lock during decrypt + frame dispatch
+    + per-stream insert).
+  - `openBidiStream` / `openUniStream` (app code, holds lock for stream
+    allocation + map insert).
+  - `getOrCreatePeerStreamLocked` (parser path on the read loop's
+    critical section, but app code can also call it from tests).
+
+Multiplexing test against aioquic measures ~25 streams/sec — every
+coroutine fights this single mutex.
+
+## Goal
+
+Split the mutex into per-domain mutexes so the read loop, send loop, and
+app code can mostly progress concurrently. Per-stream `synchronized(this)`
+inside `SendBuffer`/`ReceiveBuffer` already handles per-stream
+serialisation; we don't touch those.
+
+## Domain Map
+
+### Domain A — `streamsLock: Mutex` (the streams registry)
+
+Fields:
+
+  - `streams: MutableMap<Long, QuicStream>`
+  - `streamsList: MutableList<QuicStream>` (insertion-ordered list parallel
+    to `streams`, used by writer round-robin)
+  - `nextLocalBidiIndex`, `nextLocalUniIndex`
+  - `streamRoundRobinStart` — read+written by writer; used in the
+    same critical section it holds `streamsLock` for the iteration
+  - `peerInitiatedUniCount`, `peerInitiatedBidiCount`
+  - `advertisedMaxStreamsUni`, `advertisedMaxStreamsBidi`,
+    `advertisedMaxData`
+  - `pendingMaxStreamsUni`, `pendingMaxStreamsBidi`, `pendingMaxData`
+  - `pendingMaxStreamData: MutableMap<Long, Long>`
+  - `pendingNewConnectionId: MutableMap<Long, …>`
+  - `newPeerStreams: ArrayDeque<QuicStream>`
+  - `pendingDatagrams: ArrayDeque<ByteArray>` — outbound DATAGRAMs
+  - `incomingDatagrams: ArrayDeque<ByteArray>` — inbound DATAGRAMs
+  - `sendConnectionFlowCredit`, `sendConnectionFlowConsumed`
+  - `receiveConnectionFlowLimit`
+
+Rationale: the writer needs an atomic snapshot of "all streams + all
+pending control-frame retransmits + datagram queues + flow-control
+counters" in one critical section to assemble a packet. The parser needs
+the same coverage when delivering a STREAM frame (look up or create
+the stream + queue receive bytes + bump pending* fields). Splitting
+these into multiple sub-locks would force the writer/parser to acquire
+several locks per pass — same contention, more deadlock risk.
+
+`peerMaxStreamsBidi`, `peerMaxStreamsUni` stay `@Volatile` (already are):
+the writer reads them once at the top of a stream open; the parser
+writes once on inbound MAX_STREAMS. Atomic long write is sufficient on
+all supported platforms.
+
+### Domain B — `LevelState.levelLock: Mutex` (one per level: initial / handshake / application)
+
+Fields per `LevelState`:
+
+  - `pnSpace: PacketNumberSpaceState`
+  - `sentPackets: MutableMap<Long, SentPacket>`
+  - `ackTracker`
+  - `cryptoSend: SendBuffer`, `cryptoReceive: ReceiveBuffer`
+  - `sendProtection: PacketProtection?`, `receiveProtection: PacketProtection?`
+  - `keysDiscarded`
+  - `largestAckedPn`, `largestAckedSentTimeMs`
+
+The writer iterates through levels in order (initial → handshake →
+application) when building a coalesced datagram. Each level's critical
+section is independent, so the lock is held only for the duration of
+build at that level (which doesn't touch the streams registry except
+to read `streamsListLocked()` for stream frames inside the application
+build — that read transitions through `streamsLock`).
+
+### Domain C — `lifecycleLock: Mutex` (status + handshake metadata)
+
+Fields:
+
+  - `status: Status`
+  - `closeReason: String?`, `closeErrorCode: Long`
+  - `peerTransportParameters: TransportParameters?` — read-mostly after
+    handshake; using `@Volatile` reference + write-once-after-handshake
+    is sufficient here. Promoted to `@Volatile` so writer/parser can
+    snapshot without a lock.
+  - `handshakeComplete: Boolean`
+  - `closeAllSignals` (the channels are themselves thread-safe; lock is
+    only required to serialise the status transition)
+
+### Domain D — Atomic / `@Volatile` (no lock)
+
+Fields:
+
+  - `pendingPing` — toggled by driver under PTO; observed by writer.
+    Promote to `@Volatile`.
+  - `consecutivePtoCount` — already `@Volatile`. Driver writes it under
+    its own logic; no further protection needed because it's only read
+    inside the same loop iteration that wrote it.
+  - `destinationConnectionId` — already has volatile semantics
+    (`internal set` on a `@Volatile var`). Stays as is.
+  - `udpStatsSupplier` — already `@Volatile`.
+  - `peerMaxStreamsBidi`, `peerMaxStreamsUni` — already `@Volatile`.
+  - `handshakeDoneSignal: CompletableDeferred<Unit>` — coroutines
+    primitive, thread-safe.
+
+### Domain E — Per-stream (UNCHANGED)
+
+`QuicStream` already protects its `SendBuffer` / `ReceiveBuffer` with
+internal `synchronized(this)` blocks. Nothing changes here.
+
+## Lock Acquisition Order
+
+To prevent deadlock, document and enforce:
+
+```
+lifecycleLock < streamsLock < (any LevelState.levelLock)
+```
+
+Per-stream `synchronized(...)` blocks inside `SendBuffer`/`ReceiveBuffer`
+remain at the leaf — never acquire any QuicConnection mutex while
+holding a per-stream lock.
+
+In practice the only nesting that happens is:
+
+  - `drainOutbound` acquires `streamsLock` (for the streams loop +
+    stream-frame creation) but the per-level builds happen *outside*
+    that block — each level acquires its own `levelLock` separately.
+    No nested `streamsLock` ⊃ `levelLock` chain.
+  - Actually re-checking the design: the writer needs to allocate a
+    PN at the chosen level *while* it has decided which streams to
+    flush. Two options:
+      (1) acquire streamsLock, snapshot streams + frames, release;
+          acquire each levelLock to encode + record.
+      (2) hold streamsLock during level encode for the application
+          packet (because stream-frame retransmit tokens get recorded
+          into level.sentPackets in the same operation).
+    We take option (2) — encode under both locks, with strict order
+    `streamsLock` → `levelLock`. The other levels (initial/handshake)
+    don't touch streamsLock at all, so they only acquire `levelLock`.
+
+## Public API Compatibility
+
+`QuicConnection.lock: Mutex` is `val`-public. External callers exist
+(tests + InMemoryQuicPipe-driven harnesses). To avoid breaking those:
+
+  - Keep the `lock: Mutex` field as a deprecated forwarder. It now
+    *also* exists, but it is an alias for `lifecycleLock`. New code
+    must NOT use it. Existing tests that lock it before mutating
+    state used to cover all domains; we update them in place to use
+    the appropriate lock(s).
+
+Actually simpler: keep `lock: Mutex` as a *no-op* lock (still a
+`Mutex` so external code compiles), document that it no longer
+guards anything, update the tests that lock it.
+
+After review: tests use `conn.lock` to serialise their direct calls to
+`onTokensAcked`/`onTokensLost`/`getOrCreatePeerStreamLocked`. We update
+those tests to acquire `streamsLock` instead (since those routines
+mutate stream-domain state). The `lock` field is kept as deprecated
+for source compatibility but is functionally a leaf no-op.
+
+## Migration Plan
+
+1. Add `streamsLock`, `lifecycleLock` fields. Keep `lock` as alias of
+   `lifecycleLock`.
+2. Add `levelLock` to `LevelState`.
+3. Convert `getOrCreatePeerStreamLocked` → `getOrCreatePeerStream` doing
+   its own `streamsLock` acquisition. Keep the old name as a forwarder
+   for backwards compat.
+4. Update `openBidiStream`, `openUniStream`, `streamById`, `pollIncomingPeerStream`,
+   `awaitIncomingPeerStream`, `pollIncomingDatagram`, `awaitIncomingDatagram`,
+   `queueDatagram`, `flowControlSnapshot` to acquire `streamsLock`.
+5. Update `close`, `markClosedExternally` to use `lifecycleLock`.
+6. Update driver's `readLoop`/`sendLoop`:
+     - `feedDatagram` no longer wraps in conn-wide lock. Instead the
+       parser acquires `streamsLock` around stream-touching code,
+       and `levelLock` around level-touching code.
+     - `drainOutbound` is restructured similarly.
+7. Update tests that hold `conn.lock` to use the relevant new lock.
+
+## Risk + Mitigation
+
+  - **Deadlock**: enforce order via code review + (where practical)
+    inline comments at each acquisition site. Keep nesting shallow.
+  - **Missed coverage**: enumerate every field in this doc; if a field
+    can be mutated from two domains we either move it to a single domain
+    or annotate it as @Volatile.
+  - **Performance regression**: more mutex acquisitions overall; but
+    the critical path (multiplexing test) sees parallel execution
+    instead of serial, which more than compensates.
+
+## Implementation Phases
+
+This commit implements **phase 1** — separate domain locks but
+`drainOutbound` and `feedDatagram` still hold `streamsLock` for the
+entire pass. The wins from phase 1 alone:
+
+  - App code (`openBidiStream`, `streamById`, `flowControlSnapshot`) no
+    longer contends with `lifecycleLock`-only operations.
+  - The PTO timer path stops touching any mutex (volatile fields).
+  - `markClosedExternally` no longer needs a lock.
+  - `close()` only takes lifecycleLock — opens the path for in-progress
+    drain to finish without status-write contention.
+
+Phase 2 (deferred follow-up): split `buildApplicationPacket` into a
+"collect frames under streamsLock" stage and an "encrypt + record
+under levelLock" stage so app coroutines can intersperse during the
+encrypt window. That requires more invasive surgery on the writer's
+internals; phase 1 ships first to lock in the safer subset.
+
+## Verification
+
+  - `:quic:jvmTest` — full suite must pass.
+  - `MultiplexingThroughputTest` (new): 1000 streams in <500 ms on
+    InMemoryQuicPipe.

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

@@ -23,11 +23,26 @@ package com.vitorpamplona.quic.connection
 import com.vitorpamplona.quic.connection.recovery.SentPacket
 import com.vitorpamplona.quic.stream.ReceiveBuffer
 import com.vitorpamplona.quic.stream.SendBuffer
+import kotlinx.coroutines.sync.Mutex
 
 /** Per-encryption-level state owned by [QuicConnection]. */
 class LevelState {
-    var pnSpace = PacketNumberSpaceState()
-        private set
+    /**
+     * Lock-split refactor (2026-05-08): per-level mutex protecting
+     * everything packet-protection / packet-number-space related at this
+     * encryption level. The writer acquires this around the encode +
+     * `sentPackets` record block; the parser acquires it around
+     * `pnSpace.observeInbound` + `ackTracker.receivedPacket` +
+     * `cryptoReceive.insert` + `sentPackets` reads on inbound ACK.
+     *
+     * Acquisition order: `QuicConnection.lifecycleLock` →
+     * `QuicConnection.streamsLock` → `LevelState.levelLock`.
+     * Per-stream `synchronized(this)` blocks inside SendBuffer/ReceiveBuffer
+     * remain at the leaf.
+     */
+    val levelLock: Mutex = Mutex()
+
+    val pnSpace = PacketNumberSpaceState()
 
     var ackTracker =
         com.vitorpamplona.quic.recovery

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

@@ -147,32 +147,28 @@ class QuicConnection(
     val handshake = LevelState()
     val application = LevelState()
 
-    /**
-     * RFC 9000 §17.2.5.1: the Retry token the server handed us in a Retry
-     * packet, which we must echo verbatim in the Token field of every
-     * subsequent Initial we send. Null until [applyRetry] runs.
-     */
     @Volatile
-    var retryToken: ByteArray? = null
-        internal set
-
-    /**
-     * RFC 9000 §17.2.5.2: a client MUST NOT process more than one Retry
-     * packet per connection. Any subsequent Retry is silently dropped.
-     * Latched true by [applyRetry] on a successfully-verified Retry.
-     */
-    @Volatile
-    var retryConsumed: Boolean = false
-        internal set
-
     var handshakeComplete: Boolean = false
         private set
 
+    /**
+     * Lock-split refactor (2026-05-08): @Volatile because the writer/parser
+     * read this without acquiring [lifecycleLock] (the field is written
+     * once at handshake completion, then immutable).
+     */
+    @Volatile
     var peerTransportParameters: TransportParameters? = null
         private set
 
     enum class Status { HANDSHAKING, CONNECTED, CLOSING, CLOSED }
 
+    /**
+     * Lock-split refactor (2026-05-08): @Volatile so concurrent loops can
+     * read the status without a lock — coarse "are we still alive?" checks.
+     * Mutating transitions still go through [lifecycleLock] for atomicity
+     * with [closeReason]/[closeErrorCode] updates.
+     */
+    @Volatile
     var status: Status = Status.HANDSHAKING
         internal set
 
@@ -316,7 +312,11 @@ class QuicConnection(
      * emits a PING frame on the next drain. The PING elicits an
      * ACK from the peer; that ACK runs through loss detection and
      * declares any in-flight packets lost, triggering retransmit.
+     *
+     * Lock-split refactor (2026-05-08): @Volatile so the driver
+     * sets it without acquiring any mutex.
      */
+    @Volatile
     internal var pendingPing: Boolean = false
 
     /**
@@ -642,6 +642,13 @@ class QuicConnection(
             maxDatagramFrameSize = config.maxDatagramFrameSize,
         )
 
+    /**
+     * Lock-split refactor (2026-05-08): caller must hold [streamsLock]
+     * because we mutate [streams], [peerMaxStreamsBidi]/Uni, and
+     * [sendConnectionFlowCredit]. Invoked from the TLS listener inside
+     * [QuicConnectionParser.feedDatagram] which acquires [streamsLock]
+     * around CRYPTO-frame handling.
+     */
     private fun applyPeerTransportParameters() {
         val raw = tls.peerTransportParameters ?: return
         val tp = TransportParameters.decode(raw)
@@ -690,13 +697,39 @@ class QuicConnection(
     }
 
     /**
-     * Single mutex protecting connection-wide mutable state: streams map,
-     * datagram queues, stream-id counters, status. The driver acquires this
-     * around its read/send loops; public API methods listed below acquire it
-     * before mutating. Internal-only methods (used only from inside the
-     * driver loops) do NOT lock — caller must hold the lock.
+     * Lock-split refactor (2026-05-08): split the previous single
+     * `lock` into three independent mutexes so the read loop, send
+     * loop, and app coroutines can mostly progress in parallel.
+     *
+     *   - [streamsLock] guards the streams registry, datagram queues,
+     *     stream-id counters and connection-level flow-control bookkeeping.
+     *   - [LevelState.levelLock] (per encryption level) guards the
+     *     packet-number space, sentPackets retention, ackTracker and
+     *     CRYPTO buffers at that level.
+     *   - [lifecycleLock] guards [status]/[closeReason]/[closeErrorCode]
+     *     transitions.
+     *
+     * Acquisition order to prevent deadlock:
+     * `lifecycleLock` → `streamsLock` → `LevelState.levelLock`.
+     * Per-stream synchronized blocks inside `SendBuffer`/`ReceiveBuffer`
+     * remain at the leaf — never acquire any of the above while holding
+     * a per-stream lock.
+     *
+     * The historical `lock` field is retained as an alias of
+     * [lifecycleLock] for source-compatibility with external callers
+     * (tests, harnesses, in-process bridges). New code MUST NOT use it
+     * — it no longer protects streams or level state.
      */
-    val lock: Mutex = Mutex()
+    val streamsLock: Mutex = Mutex()
+
+    val lifecycleLock: Mutex = Mutex()
+
+    @Deprecated(
+        "Use streamsLock / lifecycleLock / LevelState.levelLock as appropriate. Lock-split refactor 2026-05-08.",
+        replaceWith = ReplaceWith("streamsLock"),
+    )
+    val lock: Mutex
+        get() = lifecycleLock
 
     /**
      * Allocate a new client-initiated bidirectional stream. Locked.
@@ -706,25 +739,21 @@ class QuicConnection(
      * check capacity proactively if the caller wants to back-pressure rather
      * than throw.
      */
-    suspend fun openBidiStream(): QuicStream = lock.withLock { openBidiStreamLocked() }
-
-    /**
-     * The connection-lock-holding part of [openBidiStream]. Public so
-     * batched-multiplex callers (interop multiplexing testcase, MoQ
-     * audio rooms emitting many group streams in quick succession) can
-     * acquire [lock] ONCE and bracket multiple stream opens — preventing
-     * the send loop from interjecting between opens and draining one
-     * stream's data per packet. Caller MUST hold [lock].
-     *
-     * The non-batched paths use [openBidiStream] which holds the lock
-     * for one call only.
-     */
-    fun openBidiStreamLocked(): QuicStream {
-        if (nextLocalBidiIndex >= peerMaxStreamsBidi) {
-            throw QuicStreamLimitException(
-                "peer-granted bidi stream cap reached " +
-                    "(used=$nextLocalBidiIndex limit=$peerMaxStreamsBidi)",
-            )
+    suspend fun openBidiStream(): QuicStream =
+        streamsLock.withLock {
+            if (nextLocalBidiIndex >= peerMaxStreamsBidi) {
+                throw QuicStreamLimitException(
+                    "peer-granted bidi stream cap reached " +
+                        "(used=$nextLocalBidiIndex limit=$peerMaxStreamsBidi)",
+                )
+            }
+            val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++)
+            val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL)
+            stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote
+            stream.receiveLimit = config.initialMaxStreamDataBidiLocal
+            streams[id] = stream
+            streamsList += stream
+            stream
         }
         val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++)
         val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL)
@@ -745,7 +774,7 @@ class QuicConnection(
      * carrying real-time Opus audio.
      */
     suspend fun openUniStream(bestEffort: Boolean = false): QuicStream =
-        lock.withLock {
+        streamsLock.withLock {
             if (nextLocalUniIndex >= peerMaxStreamsUni) {
                 throw QuicStreamLimitException(
                     "peer-granted uni stream cap reached " +
@@ -787,7 +816,7 @@ class QuicConnection(
      * See `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`.
      */
     suspend fun flowControlSnapshot(): QuicFlowControlSnapshot =
-        lock.withLock {
+        streamsLock.withLock {
             val tp = peerTransportParameters
             // Sum bytes the application has enqueued but the writer
             // hasn't yet handed to a STREAM frame. A non-zero value
@@ -826,7 +855,7 @@ class QuicConnection(
             )
         }
 
-    suspend fun pollIncomingPeerStream(): QuicStream? = lock.withLock { newPeerStreams.removeFirstOrNull() }
+    suspend fun pollIncomingPeerStream(): QuicStream? = streamsLock.withLock { newPeerStreams.removeFirstOrNull() }
 
     /**
      * Suspends until a peer-initiated stream is queued OR the connection
@@ -851,7 +880,7 @@ class QuicConnection(
      */
     suspend fun awaitIncomingPeerStream(): QuicStream? {
         while (true) {
-            lock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it }
+            streamsLock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it }
             if (status == Status.CLOSED) return null
             // select between "wakeup" and "closed" so neither path can hang.
             val keepWaiting =
@@ -866,17 +895,17 @@ class QuicConnection(
             if (!keepWaiting) {
                 // After a close-wake, drain one more time to surface any
                 // streams added between the last drain and the close.
-                lock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it }
+                streamsLock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it }
                 return null
             }
         }
     }
 
-    suspend fun streamById(id: Long): QuicStream? = lock.withLock { streams[id] }
+    suspend fun streamById(id: Long): QuicStream? = streamsLock.withLock { streams[id] }
 
-    suspend fun queueDatagram(payload: ByteArray) = lock.withLock { pendingDatagrams.addLast(payload) }
+    suspend fun queueDatagram(payload: ByteArray) = streamsLock.withLock { pendingDatagrams.addLast(payload) }
 
-    suspend fun pollIncomingDatagram(): ByteArray? = lock.withLock { incomingDatagrams.removeFirstOrNull() }
+    suspend fun pollIncomingDatagram(): ByteArray? = streamsLock.withLock { incomingDatagrams.removeFirstOrNull() }
 
     /**
      * Suspending counterpart of [pollIncomingDatagram]. Returns null only when
@@ -885,7 +914,7 @@ class QuicConnection(
      */
     suspend fun awaitIncomingDatagram(): ByteArray? {
         while (true) {
-            lock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it }
+            streamsLock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it }
             if (status == Status.CLOSED) return null
             val keepWaiting =
                 select<Boolean> {
@@ -893,7 +922,7 @@ class QuicConnection(
                     closedSignal.onReceiveCatching { false }
                 }
             if (!keepWaiting) {
-                lock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it }
+                streamsLock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it }
                 return null
             }
         }
@@ -904,8 +933,7 @@ class QuicConnection(
         errorCode: Long,
         reason: String,
     ) {
-        var firedQlog = false
-        lock.withLock {
+        lifecycleLock.withLock {
             if (status == Status.CLOSED || status == Status.CLOSING) return@withLock
             closeErrorCode = errorCode
             closeReason = reason
@@ -972,8 +1000,9 @@ class QuicConnection(
     }
 
     /**
-     * Caller must hold [lock]. Used by [QuicConnectionParser] inside the
-     * driver's read loop, which already holds the connection lock.
+     * Caller must hold [streamsLock]. Used by [QuicConnectionParser] inside
+     * the driver's read loop, which already holds [streamsLock] around the
+     * stream-domain section of frame dispatch.
      */
     internal fun getOrCreatePeerStreamLocked(id: Long): QuicStream {
         streams[id]?.let { return it }

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

@@ -35,10 +35,12 @@ import kotlinx.coroutines.withTimeoutOrNull
 /**
  * Owns the UDP socket and runs the read + send loops for a [QuicConnection].
  *
- * Synchronization: every public mutator on [QuicConnection] takes
- * `connection.lock`; the driver acquires the same lock around feed + drain.
- * That guarantees the read loop, send loop, and app coroutines never see a
- * mid-mutation state of the streams map / datagram queues / counters.
+ * Synchronization (post lock-split refactor 2026-05-08): the driver no
+ * longer takes a single connection-wide lock around feed/drain. Instead
+ * [feedDatagram] and [drainOutbound] internally acquire the appropriate
+ * domain locks (`streamsLock` and the per-level `LevelState.levelLock`)
+ * for the precise critical sections they touch — leaving app coroutines
+ * (`openBidiStream`, etc.) free to run in parallel with the I/O loops.
  *
  * The send loop is woken by a `Channel<Unit>(CONFLATED)` rather than a
  * polling timer — no idle CPU. App writes ([QuicConnection.queueDatagram]
@@ -99,7 +101,9 @@ class QuicConnectionDriver(
         try {
             while (connection.status != QuicConnection.Status.CLOSED) {
                 val datagram = socket.receive() ?: break
-                connection.lock.withLock {
+                // Phase 1 of the lock-split refactor: parser holds
+                // streamsLock for a single datagram-feed pass.
+                connection.streamsLock.withLock {
                     feedDatagram(connection, datagram, nowMillis())
                 }
                 // Inbound data may have produced new outbound (acks, crypto, etc.).
@@ -123,11 +127,16 @@ class QuicConnectionDriver(
         // floor (the same prior-shipping behavior, kept for
         // handshake-timeout safety on lossy paths).
         while (connection.status != QuicConnection.Status.CLOSED) {
-            connection.lock.withLock {
-                while (true) {
-                    val out = drainOutbound(connection, nowMillis()) ?: break
-                    socket.send(out)
-                }
+            // Phase 1 of the lock-split refactor: the writer holds
+            // streamsLock for the build, releases it for the actual
+            // socket.send() so a slow socket doesn't stall app
+            // coroutines (open/close streams, queue datagrams).
+            while (true) {
+                val out =
+                    connection.streamsLock.withLock {
+                        drainOutbound(connection, nowMillis())
+                    } ?: break
+                socket.send(out)
             }
             val ptoBaseMs =
                 if (connection.lossDetection.hasFirstRttSample) {
@@ -145,49 +154,14 @@ class QuicConnectionDriver(
                     Unit
                 }
             if (woke == null) {
-                // PTO fired. RFC 9002 §6.2.4: the probe packet MUST
-                // be ack-eliciting at the encryption level with
-                // unacknowledged data, and SHOULD retransmit lost
-                // data rather than just emit a PING.
-                //
-                // Pre-1-RTT we have a concrete thing to retransmit:
-                // the unacknowledged ClientHello (at Initial) or
-                // ClientFinished (at Handshake). We requeue ALL
-                // currently-inflight CRYPTO bytes for the highest
-                // active pre-application level so the next drain's
-                // takeChunk emits a fresh CRYPTO frame at the
-                // original offset. The pendingPing flag stays set
-                // as a fallback — `collectHandshakeLevelFrames`
-                // skips the PING when a CRYPTO retransmit lands in
-                // the same frame list, so we don't waste a frame.
-                //
-                // Why this matters in interop: aioquic strictly
-                // rejects pre-handshake Initials that contain no
-                // CRYPTO frame (CONNECTION_CLOSE 0x0 "Packet
-                // contains no CRYPTO frame"). A bare-PING probe
-                // hits that immediately. With CRYPTO retransmit on
-                // PTO, the probe carries a re-emitted ClientHello
-                // and the server processes it normally.
-                //
-                // Post-handshake (1-RTT installed) we retain the
-                // bare-PING behavior; STREAM retransmit is driven
-                // by packet-number-threshold loss detection from
-                // the ACK that the PING elicits.
-                val newPtoCount =
-                    connection.lock
-                        .withLock {
-                            connection.pendingPing = true
-                            if (connection.application.sendProtection == null) {
-                                val level = highestPreApplicationLevel(connection)
-                                if (level != null) {
-                                    connection.requeueAllInflightCrypto(level)
-                                }
-                            }
-                            connection.consecutivePtoCount =
-                                (connection.consecutivePtoCount + 1).coerceAtMost(6)
-                            connection.consecutivePtoCount
-                        }
-                connection.qlogObserver.onPtoFired(newPtoCount, ptoMillis)
+                // PTO fired. Set pendingPing so the writer emits a
+                // PING on the next drain (RFC 9002 §6.2.4 probe
+                // packet). The peer's ACK feeds loss detection +
+                // retransmit (steps 5–6). Both fields are @Volatile;
+                // no lock required.
+                connection.pendingPing = true
+                connection.consecutivePtoCount =
+                    (connection.consecutivePtoCount + 1).coerceAtMost(6)
             }
         }
     }

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

@@ -54,6 +54,15 @@ import com.vitorpamplona.quic.tls.TlsClient
  * — typically Initial + Handshake from the server in the same datagram during
  * the handshake. We loop until the datagram is fully consumed or a packet
  * fails to parse (which we drop silently per RFC 9001 §5.5).
+ *
+ * Lock-split refactor (2026-05-08): caller must hold
+ * [QuicConnection.streamsLock]. The driver wraps its read loop in
+ * `streamsLock.withLock { feedDatagram(...) }`. Test harnesses that drive
+ * single-threaded packet flow (no concurrent app code) may invoke this
+ * directly without lock acquisition; the runtime invariants still hold
+ * because there's no contending thread. Phase 1 wraps the whole feed
+ * under streamsLock so frame-dispatch / stream creation / level state
+ * remains a single critical section.
  */
 fun feedDatagram(
     conn: QuicConnection,

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

@@ -57,6 +57,15 @@ import com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket
  *
  * RFC 9000 §14: any datagram containing an Initial packet from the client
  * MUST be padded to at least 1200 bytes total.
+ *
+ * Lock-split refactor (2026-05-08): caller must hold
+ * [QuicConnection.streamsLock]. Phase 1 keeps level-state mutation
+ * inline under the same critical section as the streams-domain work
+ * the writer needs — the win comes from `lifecycleLock`-only callers
+ * (close(), status reads, PTO bookkeeping) no longer fighting this lock.
+ * The driver wraps `streamsLock.withLock { drainOutbound(...) }`; tests
+ * that drive single-threaded send paths can call this without holding
+ * the lock — there's no contending thread.
  */
 fun drainOutbound(
     conn: QuicConnection,

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

@@ -64,7 +64,7 @@ class AckTrackerPurgeOnAckOfAckTest {
 
             // Peer ACKs the packet that carried our outbound ACK
             // covering up to PN 4.
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensAcked(
                     listOf(
@@ -75,7 +75,7 @@ class AckTrackerPurgeOnAckOfAckTest {
                     ),
                 )
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             // Tracker is now empty: peer has confirmed receipt of our
             // ACK that covered everything up to PN 4. Re-advertising
@@ -96,7 +96,7 @@ class AckTrackerPurgeOnAckOfAckTest {
             }
 
             // Peer ACKs our Initial-level outbound ACK.
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensAcked(
                     listOf(
@@ -104,7 +104,7 @@ class AckTrackerPurgeOnAckOfAckTest {
                     ),
                 )
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             // Initial tracker drained; Application tracker untouched.
             assertTrue(conn.initial.ackTracker.isEmpty())
@@ -120,7 +120,7 @@ class AckTrackerPurgeOnAckOfAckTest {
             }
             // Peer ACKs our outbound ACK that covered up to PN 4 only;
             // the tracker's higher-PN ranges (5..9) must survive.
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensAcked(
                     listOf(
@@ -128,7 +128,7 @@ class AckTrackerPurgeOnAckOfAckTest {
                     ),
                 )
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertFalse(conn.application.ackTracker.isEmpty())
             assertEquals(9L, conn.application.ackTracker.largestReceived())
@@ -145,7 +145,7 @@ class AckTrackerPurgeOnAckOfAckTest {
             for (pn in 0L..9L) {
                 conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
             }
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensAcked(
                     listOf(
@@ -160,7 +160,7 @@ class AckTrackerPurgeOnAckOfAckTest {
                 )
                 assertTrue(conn.application.ackTracker.isEmpty())
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
         }
 }

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

@@ -81,12 +81,12 @@ class CryptoRetransmitTest {
                     .single()
 
             // Simulate loss via direct dispatch.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensLost(listOf(cryptoToken))
                 client.initial.sentPackets.remove(firstPn)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
 
             // Initial-level cryptoSend should now have re-queued bytes
@@ -126,11 +126,11 @@ class CryptoRetransmitTest {
                 client.initial.sentPackets.entries
                     .first { it.value.tokens.any { t -> t is RecoveryToken.Crypto } }
             // ACK via direct dispatch.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensAcked(packet.value.tokens)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             // After ACK the Initial-level cryptoSend's flushedFloor should
             // have advanced — we check by observing that another takeChunk

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

@@ -0,0 +1,139 @@
+/*
+ * 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.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Throughput contract for the lock-split refactor (2026-05-08): opening
+ * many parallel bidi streams + queueing requests must not be serialised
+ * by a single connection-wide mutex. Phase 1 of the split (separate
+ * `streamsLock` / `lifecycleLock` / per-level `levelLock`) targets the
+ * multiplexing testcase that drove this refactor — see
+ * `quic/plans/2026-05-08-lock-split-design.md`.
+ *
+ * The test stands up an in-memory client (no socket I/O), opens 1000
+ * client-bidi streams concurrently, enqueues a small request body + FIN
+ * on each, and asserts the operation completes within a generous wall-
+ * clock budget. The number is deliberately loose: this is a contract
+ * for "lock contention isn't pathological", not a microbenchmark.
+ *
+ * NOTE: the in-memory pipe doesn't drive a concurrent send loop, so
+ * this test exercises the lock-acquisition cost of `openBidiStream`
+ * itself rather than full multiplexing throughput. The interop runner
+ * provides the end-to-end measurement.
+ */
+class MultiplexingThroughputTest {
+    @Test
+    fun open_1000_bidi_streams_completes_quickly() {
+        runBlocking {
+            val client =
+                QuicConnection(
+                    serverName = "example.test",
+                    config =
+                        QuicConnectionConfig(
+                            initialMaxStreamsBidi = 2_000,
+                            initialMaxStreamsUni = 2_000,
+                            initialMaxData = 100_000_000,
+                            initialMaxStreamDataBidiLocal = 100_000,
+                            initialMaxStreamDataBidiRemote = 100_000,
+                            initialMaxStreamDataUni = 100_000,
+                        ),
+                    tlsCertificateValidator = PermissiveCertificateValidator(),
+                )
+            val serverScid = ConnectionId.random(8)
+            val tlsServer =
+                InProcessTlsServer(
+                    transportParameters =
+                        TransportParameters(
+                            initialMaxData = 100_000_000,
+                            initialMaxStreamDataBidiLocal = 100_000,
+                            initialMaxStreamDataBidiRemote = 100_000,
+                            initialMaxStreamDataUni = 100_000,
+                            initialMaxStreamsBidi = 2_000,
+                            initialMaxStreamsUni = 2_000,
+                            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)
+
+            val request = ByteArray(50) { it.toByte() }
+            val streamCount = 1_000
+
+            // Open all streams in parallel — each launch contends for
+            // streamsLock briefly. Pre-refactor this serialised against
+            // any in-flight drainOutbound call; phase 1 keeps openBidi
+            // contention scoped to streamsLock-only.
+            val started =
+                kotlin.time.TimeSource.Monotonic
+                    .markNow()
+            val opens =
+                (0 until streamCount).map {
+                    async {
+                        val stream = client.openBidiStream()
+                        stream.send.enqueue(request)
+                        stream.send.finish()
+                        stream.streamId
+                    }
+                }
+            val ids = opens.awaitAll()
+            val elapsed = started.elapsedNow()
+
+            // Useful diagnostic for measuring future regressions: stdout
+            // shows up in the test report so phase-1 vs phase-2 can be
+            // compared against the same test.
+            println(
+                "[MultiplexingThroughputTest] opened $streamCount bidi streams in " +
+                    "${elapsed.inWholeMilliseconds}ms " +
+                    "(${(streamCount * 1000.0 / elapsed.inWholeMilliseconds.coerceAtLeast(1L)).toLong()} streams/sec)",
+            )
+            assertEquals(streamCount, ids.size)
+            assertEquals(streamCount, ids.toSet().size, "stream ids must be unique")
+            // Generous bound; in-process opens of 1000 streams should
+            // complete in well under half a second on any developer
+            // machine — pre-refactor this was minutes due to lock
+            // contention against the (idle) send-loop drain. The looser
+            // 2-second bound is still 100x what's expected on actual
+            // hardware while accounting for slow CI workers.
+            assertTrue(
+                elapsed.inWholeMilliseconds < 2_000L,
+                "1000 parallel openBidiStream calls took ${elapsed.inWholeMilliseconds}ms; expected <2000ms",
+            )
+        }
+    }
+}

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

@@ -47,11 +47,11 @@ class OnTokensLostTest {
     fun ackToken_doesNotPopulateAnyPending() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensLost(listOf(RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertNull(conn.pendingMaxStreamsUni)
             assertNull(conn.pendingMaxStreamsBidi)
@@ -64,12 +64,12 @@ class OnTokensLostTest {
         runBlocking {
             val conn = newConn()
             // Simulate the writer having advertised a higher cap.
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxStreamsUni = 150L
                 conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 150L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertEquals(150L, conn.pendingMaxStreamsUni)
         }
@@ -82,12 +82,12 @@ class OnTokensLostTest {
             // the value carried by the lost token (150). The lost
             // frame is irrelevant — re-emitting 150 would not extend
             // the cap. neqo's fc.rs line 322 supersede check.
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxStreamsUni = 200L
                 conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 150L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertNull(conn.pendingMaxStreamsUni, "stale lost extension must not be re-emitted")
         }
@@ -96,12 +96,12 @@ class OnTokensLostTest {
     fun lostMaxStreamsBidi_matchingAdvertised_setsPending() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxStreamsBidi = 200L
                 conn.onTokensLost(listOf(RecoveryToken.MaxStreamsBidi(maxStreams = 200L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertEquals(200L, conn.pendingMaxStreamsBidi)
         }
@@ -110,12 +110,12 @@ class OnTokensLostTest {
     fun lostMaxData_matchingAdvertised_setsPending() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxData = 1_000_000L
                 conn.onTokensLost(listOf(RecoveryToken.MaxData(maxData = 1_000_000L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertEquals(1_000_000L, conn.pendingMaxData)
         }
@@ -124,12 +124,12 @@ class OnTokensLostTest {
     fun lostMaxData_supersededIsDropped() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxData = 2_000_000L
                 conn.onTokensLost(listOf(RecoveryToken.MaxData(maxData = 1_000_000L)))
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertNull(conn.pendingMaxData)
         }
@@ -138,13 +138,13 @@ class OnTokensLostTest {
     fun lostMaxStreamData_unknownStream_dropped() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.onTokensLost(
                     listOf(RecoveryToken.MaxStreamData(streamId = 999L, maxData = 1024L)),
                 )
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             // No stream with id 999 exists ⇒ token is dropped silently.
             assertEquals(emptyMap<Long, Long>(), conn.pendingMaxStreamData)
@@ -154,7 +154,7 @@ class OnTokensLostTest {
     fun multipleLostTokens_dispatchAll() =
         runBlocking {
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxStreamsUni = 150L
                 conn.advertisedMaxStreamsBidi = 200L
@@ -168,7 +168,7 @@ class OnTokensLostTest {
                     ),
                 )
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
             assertEquals(150L, conn.pendingMaxStreamsUni)
             assertEquals(200L, conn.pendingMaxStreamsBidi)
@@ -183,7 +183,7 @@ class OnTokensLostTest {
             // most one value (the last setter wins; the supersede
             // check filters older losses).
             val conn = newConn()
-            conn.lock.lock()
+            conn.streamsLock.lock()
             try {
                 conn.advertisedMaxStreamsUni = 200L
                 // First lost packet had MaxStreamsUni(150) — stale, dropped.
@@ -193,7 +193,7 @@ class OnTokensLostTest {
                 conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 200L)))
                 assertEquals(200L, conn.pendingMaxStreamsUni)
             } finally {
-                conn.lock.unlock()
+                conn.streamsLock.unlock()
             }
         }
 }

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

@@ -94,7 +94,7 @@ class PeerStreamCreditExtensionTest {
             // Simulate the relay opening uni streams to us. SERVER_UNI
             // stream IDs use the encoding `index << 2 | 0x3`. Two streams
             // (cap=4, half-window=2) is the threshold for a refresh.
-            client.lock
+            client.streamsLock
                 .let {
                     // Acquire under lock since getOrCreatePeerStreamLocked requires it.
                     it
@@ -103,7 +103,7 @@ class PeerStreamCreditExtensionTest {
             kotlinx.coroutines.sync
                 .Mutex()
                 .let { /* noop: silence unused-import linter */ }
-            client.lock.let { l ->
+            client.streamsLock.let { l ->
                 kotlinx.coroutines.runBlocking {
                     l.lock()
                     try {
@@ -179,7 +179,7 @@ class PeerStreamCreditExtensionTest {
 
             // Open 10 peer streams — half-window for cap=100 is 50, so
             // we're well below the threshold.
-            client.lock.let { l ->
+            client.streamsLock.let { l ->
                 kotlinx.coroutines.runBlocking {
                     l.lock()
                     try {

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

@@ -46,11 +46,11 @@ class PendingFlowControlEmitTest {
     fun pendingMaxStreamsUni_drainEmitsFrameAndToken() =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamsUni = 150L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
 
             val sizeBefore = client.application.sentPackets.size
@@ -79,11 +79,11 @@ class PendingFlowControlEmitTest {
     fun pendingMaxStreamsBidi_drainEmitsFrameAndToken(): Unit =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamsBidi = 200L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             val sizeBefore = client.application.sentPackets.size
             runCatching { drainOutbound(client, nowMillis = 1L) }
@@ -103,11 +103,11 @@ class PendingFlowControlEmitTest {
     fun pendingMaxData_drainEmitsFrameAndToken(): Unit =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxData = 5_000_000L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             val sizeBefore = client.application.sentPackets.size
             runCatching { drainOutbound(client, nowMillis = 1L) }
@@ -127,12 +127,12 @@ class PendingFlowControlEmitTest {
     fun pendingMaxStreamData_perStreamDrain() =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamData[3L] = 1_024L
                 client.pendingMaxStreamData[7L] = 2_048L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             val sizeBefore = client.application.sentPackets.size
             runCatching { drainOutbound(client, nowMillis = 1L) }
@@ -159,13 +159,13 @@ class PendingFlowControlEmitTest {
     fun multiplePending_drainEmitsAllInOnePacket(): Unit =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamsUni = 150L
                 client.pendingMaxStreamsBidi = 200L
                 client.pendingMaxData = 1_000_000L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             val sizeBefore = client.application.sentPackets.size
             runCatching { drainOutbound(client, nowMillis = 1L) }
@@ -216,11 +216,11 @@ class PendingFlowControlEmitTest {
             // advertised cap. The writer drains it as-is — supersede check
             // is in step 6 (the setter side), not here.
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamsUni = 50L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             val sizeBefore = client.application.sentPackets.size
             runCatching { drainOutbound(client, nowMillis = 1L) }
@@ -242,11 +242,11 @@ class PendingFlowControlEmitTest {
     fun pendingClearedAcrossDrains() =
         runBlocking {
             val client = handshakedClient()
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.pendingMaxStreamsUni = 150L
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             // Drain once: pending consumed.
             runCatching { drainOutbound(client, nowMillis = 1L) }

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

@@ -93,12 +93,12 @@ class ResetStopSendingEmitTest {
                     .single()
 
             // Simulate loss.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensLost(listOf(token))
                 client.application.sentPackets.remove(firstEntry.key)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             // Per-stream emit-pending should be re-flagged.
             assertTrue(stream.resetEmitPending, "loss must re-flag resetEmitPending")
@@ -137,21 +137,21 @@ class ResetStopSendingEmitTest {
                     .single()
 
             // ACK first.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensAcked(listOf(token))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             assertEquals(true, stream.resetAcked)
             assertEquals(false, stream.resetEmitPending)
 
             // Now a stale loss notification arrives. Defensive: drop.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensLost(listOf(token))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             assertEquals(false, stream.resetEmitPending, "stale loss after ACK must not re-flag emit-pending")
         }
@@ -248,11 +248,11 @@ class ResetStopSendingEmitTest {
                     connectionId = byteArrayOf(1, 2, 3, 4),
                     statelessResetToken = ByteArray(16) { it.toByte() },
                 )
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensLost(listOf(token))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             assertEquals(token, client.pendingNewConnectionId[1L])
 

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

@@ -69,7 +69,7 @@ class RetransmitIntegrationTest {
             //    ACK'd by reordering — its PN < largestAckedPn -
             //    PACKET_THRESHOLD ⇒ declared lost.
             val futurePn = msuPn + 4L
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 // Inject a phantom SentPacket at futurePn so the loss
                 // detector has a credible "newly acked" reference, then
@@ -102,7 +102,7 @@ class RetransmitIntegrationTest {
                 // 5. Dispatch lost tokens — pendingMaxStreamsUni gets set.
                 client.onTokensLost(lostMsuPacket.tokens)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             assertEquals(
                 capAfterFirstDrain,
@@ -148,24 +148,24 @@ class RetransmitIntegrationTest {
 
             // Second bump: open more peer-uni streams to cross the
             // (already extended) threshold again.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 2))
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 3))
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 4))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             runCatching { drainOutbound(client, nowMillis = 2L) }
             val secondCap = client.advertisedMaxStreamsUni
             assertTrue(secondCap > firstCap, "second drain must advertise a still-higher cap; saw $firstCap → $secondCap")
 
             // Now declare the FIRST emit lost via direct dispatch.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = firstCap)))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
             // Supersede check: firstCap != advertisedMaxStreamsUni (now == secondCap),
             // so pending must remain null.
@@ -216,12 +216,12 @@ class RetransmitIntegrationTest {
 
     private fun crossPeerUniHalfWindow(client: QuicConnection) =
         runBlocking {
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0))
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
         }
 }

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

@@ -200,12 +200,12 @@ class SentPacketTrackingTest {
     /** Cross the half-window threshold (cap=4, two peer-uni streams ⇒ count >= cap-half=2). */
     private fun crossPeerUniHalfWindow(client: QuicConnection) =
         runBlocking {
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0))
                 client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1))
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
         }
 }

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

@@ -82,7 +82,7 @@ class StreamRetransmitTest {
             val firstPn = firstPacketEntry.key
 
             // Simulate loss via direct dispatch.
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 val streamToken =
                     firstPacketEntry.value.tokens
@@ -93,7 +93,7 @@ class StreamRetransmitTest {
                 // detector would have done this).
                 client.application.sentPackets.remove(firstPn)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
 
             // SendBuffer should have re-queued the bytes for retransmit.
@@ -133,11 +133,11 @@ class StreamRetransmitTest {
             val packet =
                 client.application.sentPackets.entries
                     .first { it.value.tokens.any { t -> t is RecoveryToken.Stream } }
-            client.lock.lock()
+            client.streamsLock.lock()
             try {
                 client.onTokensAcked(packet.value.tokens)
             } finally {
-                client.lock.unlock()
+                client.streamsLock.unlock()
             }
 
             // After ACK: enqueue more, observe that the buffer