nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(quic): kill dead levelLock + add bug-resistant batched-open API

Browse Source 
Audit follow-up. Two cleanups consolidated into one commit since they
share the goal of "make the lock contract obvious from the API".

(1) Remove LevelState.levelLock entirely.

The lock-split refactor introduced a per-level Mutex with a docstring
claiming the writer/parser would acquire it around encode + sentPackets
record + ACK observation. Neither actually does. SendBuffer's internal
synchronized(this) is what serializes cryptoSend mutations against
takeChunk and markAcked. handlePtoFired's levelLock acquisition was
the only production usage and it serialized only against itself.

Removed:
  - LevelState.levelLock
  - handlePtoFired's withLock wrapper (now a non-suspend fun)
  - All docstring references to a third lock domain

The pre-existing sentPackets HashMap race (writer mutates under
streamsLock, parser reads without sync) is unchanged — out of scope
for this PR. Acquisition order is now `lifecycleLock → streamsLock`,
flat.

(2) Add openBidiStreamsBatch + openUniStreamsBatch — the bug-resistant
high-level API for the prepareRequests / moq audio-rooms patterns.

The previous shape required callers to manually do
`streamsLock.withLock { repeat(N) { openBidiStreamLocked() ... } }`.
That contract regressed twice on this very branch: once held the wrong
lock (lifecycleLock alias), once skipped the wrap entirely. Both shapes
silently emitted one STREAM per packet under multiplex load.

The new API encapsulates the lock + the per-item init lambda:

    conn.openBidiStreamsBatch(items) { stream, item ->
        stream.send.enqueue(encode(item))
        stream.send.finish()
        Handle(stream)
    }

Callers physically cannot hold the wrong lock. Migrated:
  - Http3GetClient.prepareRequests
  - HqInteropGetClient.prepareRequests

Also added `openUniStreamLocked` + `openUniStreamsBatch` symmetric to
the bidi versions. moq audio-rooms eventually wants to open many uni
streams in burst; the same one-stream-per-packet bug lurks if every
open serializes through its own lock acquisition.

`openBidiStreamLocked` / `openUniStreamLocked` remain public (with
their `check(streamsLock.isLocked)` guards) for the rare custom-batch
callers that need to mix bidi+uni opens under a single hold. Most
callers should use the *Batch variants going forward.

Test coverage extended:
  - openBidiStreamsBatch happy path
  - openUniStreamLocked throws without streamsLock
  - openUniStreamsBatch happy path

Six tests in BatchedOpenLockContractTest now pin the contract.

https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT
This commit is contained in:
Claude
2026-05-07 03:31:30 +00:00
parent 07dd572423
commit a0a604b8e7
6 changed files with 178 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt
+12 -17
View File
@@ -23,25 +23,20 @@ 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]. */
/**
* Per-encryption-level state owned by [QuicConnection].
*
* Concurrency: [cryptoSend] / [cryptoReceive] use their internal
* [SendBuffer] / [ReceiveBuffer] `synchronized(this)` blocks for
* thread safety — the writer's `takeChunk`, the parser's `markAcked`,
* and PTO-driven `requeueAllInflight` are all serialized through
* those leaf locks. [sentPackets] is currently mutated by the writer
* (under [QuicConnection.streamsLock]) and read by the parser without
* synchronization; that race is pre-existing audit-tracked and not
* fixed by [LevelState] today.
*/
class LevelState {
/**
* 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 =
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+89 -31
View File
@@ -717,22 +717,24 @@ class QuicConnection(
/**
* Lock-split refactor (2026-05-08): split the previous single
* `lock` into three independent mutexes so the read loop, send
* `lock` into two 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.
* stream-id counters, connection-level flow-control bookkeeping,
* packet-number space + sentPackets retention + CRYPTO buffer
* mutations at every encryption level. The writer's drain and
* the parser's feed both take it.
* - [lifecycleLock] guards [status] / [closeReason] /
* [closeErrorCode] transitions.
*
* Per-stream and per-level buffer mutations serialize through
* `synchronized(this)` inside `SendBuffer` / `ReceiveBuffer` /
* `AckTracker` — those leaf locks are safe to take with or
* without an outer mutex held.
*
* 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.
* `lifecycleLock` → `streamsLock`. Never go the other way.
*
* The historical `lock` field is retained as an alias of
* [lifecycleLock] for source-compatibility with external callers
@@ -744,7 +746,7 @@ class QuicConnection(
val lifecycleLock: Mutex = Mutex()
@Deprecated(
"Use streamsLock / lifecycleLock / LevelState.levelLock as appropriate. Lock-split refactor 2026-05-08.",
"Use streamsLock or lifecycleLock as appropriate. Lock-split refactor 2026-05-08.",
replaceWith = ReplaceWith("streamsLock"),
)
val lock: Mutex
@@ -761,11 +763,38 @@ class QuicConnection(
suspend fun openBidiStream(): QuicStream = streamsLock.withLock { openBidiStreamLocked() }
/**
* The streamsLock-holding part of [openBidiStream]. Public so
* batched-multiplex callers (prepareRequests) can acquire
* [streamsLock] ONCE and bracket multiple stream opens — without
* yielding to the send loop between opens. Caller MUST hold
* [streamsLock].
* Atomically open one bidi stream per [items] entry under a single
* [streamsLock] hold and run [init] for each (stream, item) inside
* the lock. The send loop cannot interject between opens — when it
* next drains it sees ALL N streams' frames ready and packs them
* into coalesced packets instead of emitting one tiny packet per
* stream.
*
* This is the bug-resistant API for the prepareRequests pattern.
* The previous shape (caller manually wraps `streamsLock.withLock`
* around a loop of [openBidiStreamLocked]) regressed twice: once
* by holding the wrong lock, and once by skipping the wrapper
* entirely. Both shapes failed silently as "one STREAM per packet"
* under multiplex load, while the unit tests passed.
*
* Callers that just need a single stream should still use
* [openBidiStream]. [openBidiStreamLocked] remains public for the
* rare custom-batching scenarios that need finer control, but
* those callers should generally migrate to this API.
*/
suspend fun <I, R> openBidiStreamsBatch(
items: List<I>,
init: (QuicStream, I) -> R,
): List<R> =
streamsLock.withLock {
items.map { init(openBidiStreamLocked(), it) }
}
/**
* The streamsLock-holding primitive used by [openBidiStream] and
* [openBidiStreamsBatch]. Public so callers that need a custom
* batching shape (e.g. mixed bidi+uni opens) can compose it under
* a manual [streamsLock] hold. Caller MUST hold [streamsLock].
*/
fun openBidiStreamLocked(): QuicStream {
// Mutex.isLocked is the only check we have — kotlinx.coroutines
@@ -808,21 +837,50 @@ class QuicConnection(
* [QuicStream.bestEffort]). Used for moq-lite group streams
* carrying real-time Opus audio.
*/
suspend fun openUniStream(bestEffort: Boolean = false): QuicStream =
suspend fun openUniStream(bestEffort: Boolean = false): QuicStream = streamsLock.withLock { openUniStreamLocked(bestEffort) }
/**
* The streamsLock-holding primitive used by [openUniStream] and
* [openUniStreamsBatch]. Caller MUST hold [streamsLock].
*/
fun openUniStreamLocked(bestEffort: Boolean = false): QuicStream {
check(streamsLock.isLocked) {
"openUniStreamLocked requires streamsLock to be held"
}
if (nextLocalUniIndex >= peerMaxStreamsUni) {
throw QuicStreamLimitException(
"peer-granted uni stream cap reached " +
"(used=$nextLocalUniIndex limit=$peerMaxStreamsUni)",
)
}
val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++)
val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, bestEffort = bestEffort)
stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni
stream.receiveLimit = 0L // can't receive
streams[id] = stream
streamsList += stream
return stream
}
/**
* Bug-resistant counterpart to [openBidiStreamsBatch] for uni
* streams. Atomically open one client-uni stream per [items]
* entry under a single [streamsLock] hold and run [init] for
* each (stream, item).
*
* Use this for moq audio-rooms and any other path that opens many
* uni streams in burst — without batching, each open releases the
* lock and the send loop can interject, emitting one stream per
* packet (the same shape that broke bidi multiplexing on
* 2026-05-06).
*/
suspend fun <I, R> openUniStreamsBatch(
items: List<I>,
bestEffort: Boolean = false,
init: (QuicStream, I) -> R,
): List<R> =
streamsLock.withLock {
if (nextLocalUniIndex >= peerMaxStreamsUni) {
throw QuicStreamLimitException(
"peer-granted uni stream cap reached " +
"(used=$nextLocalUniIndex limit=$peerMaxStreamsUni)",
)
}
val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++)
val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, bestEffort = bestEffort)
stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni
stream.receiveLimit = 0L // can't receive
streams[id] = stream
streamsList += stream
stream
items.map { init(openUniStreamLocked(bestEffort), it) }
}
/** Snapshot of peer-granted bidi cap. Reads do not need the lock — long writes are atomic on every supported platform. */
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt
+15 -13
View File
@@ -37,10 +37,11 @@ import kotlinx.coroutines.withTimeoutOrNull
*
* 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.
* [feedDatagram] and [drainOutbound] internally acquire `streamsLock`
* for the precise critical sections they touch — leaving app
* coroutines (`openBidiStream`, etc.) free to run in parallel with the
* I/O loops. Per-stream and per-level buffers serialize through their
* leaf `synchronized(this)` blocks.
*
* The send loop is woken by a `Channel<Unit>(CONFLATED)` rather than a
* polling timer — no idle CPU. App writes ([QuicConnection.queueDatagram]
@@ -253,20 +254,21 @@ class QuicConnectionDriver(
* (commits c0d7b6031, then again in the lock-split refactor) without
* any test breaking.
*
* `pendingPing` and `consecutivePtoCount` are `@Volatile` so we set
* them outside any external lock. `requeueAllInflightCrypto` mutates
* the level's `cryptoSend` buffer; we take `levelLock` for the
* requeue so the writer's `takeChunk` can't observe a half-mutated
* inflight queue mid-build.
* Concurrency: `pendingPing` and `consecutivePtoCount` are `@Volatile`.
* [QuicConnection.requeueAllInflightCrypto] delegates to
* [com.vitorpamplona.quic.stream.SendBuffer.requeueAllInflight] which
* is `synchronized(this)` internally, so it's safe to call without
* an external lock — even concurrent with the writer's `takeChunk`.
* If the parser concurrently runs `discardKeys` on the same level,
* `requeueAllInflight` operates on the buffer reference we captured
* (or the fresh one — both are valid) and is at worst a no-op.
*/
internal suspend fun handlePtoFired(conn: QuicConnection) {
internal fun handlePtoFired(conn: QuicConnection) {
conn.pendingPing = true
if (conn.application.sendProtection == null) {
val level = highestPreApplicationLevel(conn)
if (level != null) {
conn.levelState(level).levelLock.withLock {
conn.requeueAllInflightCrypto(level)
}
conn.requeueAllInflightCrypto(level)
}
}
conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6)
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt
+45
View File
@@ -116,6 +116,51 @@ class BatchedOpenLockContractTest {
}
}
@Test
fun `openBidiStreamsBatch is the bug-resistant API for the prepareRequests pattern`() {
runBlocking {
val client = handshakedClient()
// The new high-level API — caller can't hold the wrong
// lock because it doesn't take any. Encapsulates the
// streamsLock acquisition + per-item init under the lock.
val payloads = (0 until 64).map { "req-$it".encodeToByteArray() }
val streams =
client.openBidiStreamsBatch(payloads) { stream, payload ->
stream.send.enqueue(payload)
stream.send.finish()
stream
}
assertEquals(64, streams.size)
assertEquals(64, streams.map { it.streamId }.toSet().size)
}
}
@Test
fun `openUniStreamLocked throws when streamsLock is not held`() {
runBlocking {
val client = handshakedClient()
assertFailsWith<IllegalStateException> {
client.openUniStreamLocked()
}
}
}
@Test
fun `openUniStreamsBatch holds streamsLock for the whole batch`() {
runBlocking {
val client = handshakedClient()
// moq audio-rooms shape: many uni streams in burst. Without
// the batched API, each open releases the lock and the
// send loop interjects (the same shape that broke bidi
// multiplexing on 2026-05-06).
val items = List(16) { it }
val streams =
client.openUniStreamsBatch(items) { stream, _ -> stream }
assertEquals(16, streams.size)
assertEquals(16, streams.map { it.streamId }.toSet().size)
}
}
private fun handshakedClient(): QuicConnection =
runBlocking {
val client =