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

refactor(nestsclient,quic): simplify-pass cleanups on moq-lite Lite-03/04 audit

Browse Source 
Surface a few code-quality wins surfaced by the simplify pass on the
moq-lite Lite-03/04 audit. All semantic-preserving; tests untouched
in behavior.

  - **Derive `MoqLiteSession.version` from `transport.negotiatedSubProtocol`**
    instead of carrying it as a separate constructor parameter. The
    version was already redundant with the transport's negotiated
    ALPN; deriving it eliminates the parameter on
    `MoqLiteSession.client(...)` and the `resolveMoqLiteVersion()` /
    `moqVersion` plumbing in `connectNestsListener` /
    `connectNestsSpeaker`. Tests drive the version via
    `FakeWebTransport.pair(negotiatedSubProtocol = …)`, which already
    plumbs through to the session's derivation.

  - **Extract `MoqLiteSession.rejectSubscribe(bidi, errorCode, reason)`**
    helper that writes a `SubscribeDrop` body and `RESET_STREAM`s the
    bidi. Both Subscribe-rejection arms (`BROADCAST_DOES_NOT_EXIST`
    and `TRACK_DOES_NOT_EXIST`) now share this helper instead of
    duplicating the runCatching / write / reset shape. Future drop
    codes plug in without growing the duplication.

  - **Make `StrippedWtStream.stopSending` non-nullable.** The demux
    always wires it (every surfaced stream has a read side), so the
    `?:` "defensive" fallbacks in `StrippedWtReadStreamAdapter` and
    `StrippedWtBidiStreamAdapter` were dead code. Tightens the
    `StrippedWtStream` contract and shrinks the adapters.

  - **Extract `SEQ_BITS=23` / `SEQ_MASK=0x007F_FFFF` / `SEQ_MAX`
    constants** in `MoqLiteSession.openGroupStream`. The bit-pack
    formula now reads as `(trackPriority and 0xFF) shl SEQ_BITS) or
    (sequence and SEQ_MASK)` — layout is named, not derived from the
    hex literals scattered through the body.

Items deferred (noted but not done):

  - `handleInboundBidi` arm extraction — the per-arm variable capture
    (`dispatched`, `inboundSub`, `inboundSubPublisher`,
    `inboundAnnouncePublisher`) makes a clean refactor harder than
    the readability win. Defer until a future feature changes the
    dispatch shape and the refactor falls out naturally.
  - `Fake{Bidi,Read}Stream` / `ChannelWriteStream` constructor
    parameter explosion — the cells-as-nullable-named-params shape
    is already readable; the candidate `FakeStreamSignals` struct
    would force a left/right-direction flag on the bidi side, which
    is uglier than the explicit `myReset` / `peerReset` naming.
  - `MoqLiteCodec` `object` → versioned `class` — the
    `version: MoqLiteVersion = LITE_03` parameter on each method is
    the idiomatic Kotlin shape for an optional-with-default
    discriminator; converting to a class would bind state to
    instances and complicate the test path that exercises both
    versions per-call.
  - `announce()` / `probe()` pump abstraction — borderline; the
    embedded logging + tracing in `announce()` would have to be
    parameterized into the abstraction. Worth revisiting if a third
    similar pump appears.
  - `MoqLiteSubscribeDropCode` / `MoqLiteStreamCancelCode` sealed-
    class refactor — `Long` constants match the wire shape; sealed
    types would force conversion at every API boundary
    (codec accepts `Long`, transport accepts `Long`).

https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
This commit is contained in:
Claude
2026-05-09 15:37:42 +00:00
parent 064654512d
commit b2362f6504
5 changed files with 94 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt
+9 -25
View File
@@ -109,20 +109,19 @@ suspend fun connectNestsListener(
// moq-lite has NO setup message — the WebTransport handshake itself
// is the handshake. Version (Lite-03 vs Lite-04) is selected by the
// ALPN exchange (`wt-available-protocols` / `wt-protocol`); fall
// back to Lite-03 if the server didn't echo `wt-protocol` (older
// single-protocol deployments).
val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)
// ALPN exchange (`wt-available-protocols` / `wt-protocol`) and
// surfaced via [MoqLiteSession.version]; the session itself derives
// it from `transport.negotiatedSubProtocol`.
val moq =
try {
MoqLiteSession.client(webTransport, scope, moqVersion)
MoqLiteSession.client(webTransport, scope)
} catch (t: Throwable) {
runCatching { webTransport.close(0, "moq-lite session init failed") }
state.value = NestsListenerState.Failed("moq-lite session init failed: ${t.message}", t)
return failedListener(state)
}
state.value = NestsListenerState.Connected(room, versionCode(moqVersion))
state.value = NestsListenerState.Connected(room, versionCode(moq.version))
return MoqLiteNestsListener(
session = moq,
mutableState = state,
@@ -145,21 +144,6 @@ const val MOQ_LITE_03_VERSION: Long = 0x6D71_6C03L
*/
const val MOQ_LITE_04_VERSION: Long = 0x6D71_6C04L
/**
* Resolve the [MoqLiteVersion] to use for a session given the ALPN
* the WebTransport server selected (or `null` if the server didn't
* echo `wt-protocol`, e.g. older single-protocol deployments).
*
* Falls back to [MoqLiteVersion.LITE_03] when the negotiated value
* is unknown or absent — every nostrnests deployment supports
* Lite-03, so this stays safe under any of:
* - server is older than draft-13 (no `wt-protocol` echo)
* - server picked something we don't recognise (e.g. a future
* `moq-lite-05`)
* - client offered only one protocol
*/
internal fun resolveMoqLiteVersion(negotiatedSubProtocol: String?): MoqLiteVersion = MoqLiteVersion.fromAlpn(negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03
/** Map a [MoqLiteVersion] to its synthetic [MOQ_LITE_03_VERSION]-family code. */
internal fun versionCode(version: MoqLiteVersion): Long =
when (version) {
@@ -273,18 +257,18 @@ suspend fun connectNestsSpeaker(
state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake)
// moq-lite has NO setup message. Same logic as the listener path —
// version is selected by the ALPN exchange.
val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)
// version is selected by the ALPN exchange and derived inside
// [MoqLiteSession.version].
val moq =
try {
MoqLiteSession.client(webTransport, scope, moqVersion)
MoqLiteSession.client(webTransport, scope)
} catch (t: Throwable) {
runCatching { webTransport.close(0, "moq-lite session init failed") }
state.value = NestsSpeakerState.Failed("moq-lite session init failed: ${t.message}", t)
return failedSpeaker(state)
}
state.value = NestsSpeakerState.Connected(room, versionCode(moqVersion))
state.value = NestsSpeakerState.Connected(room, versionCode(moq.version))
return MoqLiteNestsSpeaker(
session = moq,
speakerPubkeyHex = speakerPubkeyHex,
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+72 -71
View File
@@ -77,24 +77,24 @@ class MoqLiteSession internal constructor(
*/
internal val transport: WebTransportSession,
private val scope: CoroutineScope,
/**
* Wire-format version this session speaks. Selected at
* WebTransport CONNECT time via the
* `wt-available-protocols` / `wt-protocol` exchange (see
* [com.vitorpamplona.nestsclient.transport.WebTransportSession.negotiatedSubProtocol]).
* Defaults to [MoqLiteVersion.LITE_03] for backward compat with
* call sites that don't pass an explicit version (most tests +
* the legacy single-version factory). Production
* `connectNestsListener` / `connectNestsSpeaker` pass the
* resolved version from the negotiated ALPN.
*
* The version selects between the version-conditional codec
* branches in [MoqLiteCodec] for `Announce.hops`,
* `AnnouncePlease.excludeHop`, and `Probe.rtt`. All other
* messages are wire-identical between the two versions.
*/
val version: MoqLiteVersion = MoqLiteVersion.LITE_03,
) {
/**
* Wire-format version this session speaks. Derived from the
* WebTransport ALPN the server selected via
* `wt-available-protocols` / `wt-protocol`
* ([WebTransportSession.negotiatedSubProtocol]); falls back to
* [MoqLiteVersion.LITE_03] when the server didn't echo
* `wt-protocol` or echoed something we don't recognise (older
* single-protocol deployments / forward-compat).
*
* Selects between the version-conditional codec branches in
* [MoqLiteCodec] for `Announce.hops`, `AnnouncePlease.excludeHop`,
* and `Probe.rtt`. All other messages are wire-identical between
* the two versions. Tests drive this via
* `FakeWebTransport.pair(negotiatedSubProtocol = …)`.
*/
val version: MoqLiteVersion =
MoqLiteVersion.fromAlpn(transport.negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03
private val state = Mutex()
private val subscriptionsBySubscribeId: MutableMap<Long, ListenerSubscription> = HashMap()
private var nextSubscribeId: Long = 0L
@@ -883,6 +883,31 @@ class MoqLiteSession internal constructor(
}
}
/**
* Reject an inbound Subscribe with a typed error code. Writes the
* `SubscribeDrop` body so a watcher that decodes the application
* message sees the typed reason, then `RESET_STREAM(errorCode)`s
* the bidi so a watcher inspecting only the QUIC layer also sees
* the same code (audit M3 — Lite-03 conveys errors via
* RESET_STREAM, distinguishing "publisher rejected" from
* "publisher gracefully shut down" / FIN). Errors during the
* write/reset are swallowed: the bidi is already on its way out.
*/
private suspend fun rejectSubscribe(
bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
errorCode: Long,
reason: String,
) {
runCatching {
bidi.write(
MoqLiteCodec.encodeSubscribeDrop(
MoqLiteSubscribeDrop(errorCode = errorCode, reasonPhrase = reason),
),
)
bidi.reset(errorCode)
}
}
private suspend fun handleInboundBidi(bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream) {
// Single long-running collector for the bidi's full lifetime.
// Pre-fix this dispatch was split into a `firstOrNull()` to
@@ -1031,28 +1056,13 @@ class MoqLiteSession internal constructor(
"SUBSCRIBE inbound id=${sub.id} broadcast='${sub.broadcast}' does not match " +
"publisher.suffix='$ourSuffix' — replying SubscribeDrop+RESET"
}
runCatching {
bidi.write(
MoqLiteCodec.encodeSubscribeDrop(
MoqLiteSubscribeDrop(
errorCode = MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST,
reasonPhrase =
"broadcast '${sub.broadcast}' is not published on this session " +
"(we publish '$ourSuffix')",
),
),
)
// Lite-03 conveys errors on any stream via
// `RESET_STREAM(application_error_code)` (audit
// M3). The Drop body is the application-level
// signal; the reset is the QUIC-level signal
// that carries the same code, distinguishing
// "publisher rejected this subscribe" from
// "publisher gracefully shut down" (which
// would be a plain FIN). Pre-fix we FINed,
// overlapping the two semantics.
bidi.reset(MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST)
}
rejectSubscribe(
bidi = bidi,
errorCode = MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST,
reason =
"broadcast '${sub.broadcast}' is not published on this session " +
"(we publish '$ourSuffix')",
)
dispatched = true
return@collect
}
@@ -1060,38 +1070,17 @@ class MoqLiteSession internal constructor(
// single-track-per-publisher model, only one match is possible.
val targetPublisher = publishersSnapshot.firstOrNull { it.track == sub.track }
if (targetPublisher == null) {
// Reply SubscribeDrop with a TRACK_DOES_NOT_EXIST
// error code BEFORE we RESET — without this the
// peer's response wait resolves only on
// bidi tear-down with no indication WHY (looks
// identical to "publisher disappeared mid-
// subscribe"). Drop carries the error code +
// reason phrase the watcher can log /
// surface, and matches what kixelated's
// `rs/moq-lite/src/lite/subscribe.rs`
// expects for an unrecognised track on a
// live broadcast. RESET (audit M3) replaces
// the prior FIN: Lite-03 conveys errors via
// RESET_STREAM, distinguishing "publisher
// rejected" from "publisher gracefully shut
// down."
Log.w("NestTx") {
"SUBSCRIBE inbound id=${sub.id} track='${sub.track}' has no matching publisher " +
"on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop+RESET"
}
runCatching {
bidi.write(
MoqLiteCodec.encodeSubscribeDrop(
MoqLiteSubscribeDrop(
errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST,
reasonPhrase =
"track '${sub.track}' is not published on this broadcast " +
"(available: ${publishersSnapshot.joinToString(",") { it.track }})",
),
),
)
bidi.reset(MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST)
}
rejectSubscribe(
bidi = bidi,
errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST,
reason =
"track '${sub.track}' is not published on this broadcast " +
"(available: ${publishersSnapshot.joinToString(",") { it.track }})",
)
dispatched = true
return@collect
}
@@ -1286,8 +1275,8 @@ class MoqLiteSession internal constructor(
// groups of any LOWER-priority track via the high byte.
// Production sessions cycle on JWT refresh every 9 min, so the
// wrap is defensive only.
val seq23 = sequence.coerceAtMost(0x7F_FFFFL).toInt() and 0x007F_FFFF
val packedPriority = ((trackPriority and 0xFF) shl 23) or seq23
val seqLow = sequence.coerceAtMost(SEQ_MAX.toLong()).toInt() and SEQ_MASK
val packedPriority = ((trackPriority and 0xFF) shl SEQ_BITS) or seqLow
uni.setPriority(packedPriority)
uni.write(Varint.encode(MoqLiteDataType.Group.code))
uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence)))
@@ -1607,6 +1596,19 @@ class MoqLiteSession internal constructor(
*/
const val DEFAULT_TRACK_PRIORITY: Int = 0x80
/**
* Bit layout for the QUIC stream priority value packed by
* [openGroupStream]. Bit 31 is reserved as the sign bit (so
* the writer's signed `sortedByDescending` orders correctly);
* bits 30..23 carry [DEFAULT_TRACK_PRIORITY]-family
* `trackPriority u8`; bits 22..0 carry the low [SEQ_BITS] of
* the group sequence. See the kdoc on [openGroupStream] for
* the full rationale.
*/
private const val SEQ_BITS: Int = 23
private const val SEQ_MASK: Int = 0x007F_FFFF
private const val SEQ_MAX: Int = SEQ_MASK
/**
* Bitrate hint (bits/sec) we report on inbound moq-lite Probe
* bidis as a publisher. Mirrors the upper-bound of an Opus
@@ -1638,7 +1640,6 @@ class MoqLiteSession internal constructor(
fun client(
transport: WebTransportSession,
pumpScope: CoroutineScope,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): MoqLiteSession = MoqLiteSession(transport, pumpScope, version)
): MoqLiteSession = MoqLiteSession(transport, pumpScope)
}
}
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt
+7 -2
View File
@@ -352,8 +352,13 @@ class MoqLiteSessionTest {
// client-side announce-watch flow surfaces them
// unchanged. Pre-fix the codec was Lite-03-only and
// would have read the count and discarded the IDs.
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope, MoqLiteVersion.LITE_04)
// Drive the negotiated sub-protocol via the fake so the
// session derives `version = LITE_04` itself (matching the
// production path, where `MoqLiteSession.version` reads
// through `transport.negotiatedSubProtocol`).
val (clientSide, serverSide) =
FakeWebTransport.pair(negotiatedSubProtocol = MoqLiteAlpn.LITE_04)
val session = MoqLiteSession.client(clientSide, pumpScope)
assertEquals(MoqLiteVersion.LITE_04, session.version)
val peer =
nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt
+2 -10
View File
@@ -470,14 +470,7 @@ private class StrippedWtReadStreamAdapter(
override fun incoming(): Flow<ByteArray> = stripped.data
override suspend fun stopSending(errorCode: Long) {
// Demux unconditionally wires `stopSending` for every surfaced
// stream — the contract in `StrippedWtStream` is "non-null on
// streams with a read side", which is true here. The `?:`
// fallback exists only to defend against a future demux bug
// that forgets to wire it; Unit is a safe no-op for tests
// that mock a stripped stream without a real receiver.
val ss = stripped.stopSending ?: return
ss(errorCode)
stripped.stopSending(errorCode)
}
}
@@ -522,8 +515,7 @@ private class StrippedWtBidiStreamAdapter(
}
override suspend fun stopSending(errorCode: Long) {
val ss = stripped.stopSending ?: return
ss(errorCode)
stripped.stopSending(errorCode)
}
/**
quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt
+4 -3
View File
@@ -78,10 +78,11 @@ class StrippedWtStream(
/**
* Send `STOP_SENDING(applicationErrorCode)` on the receive half
* (RFC 9000 §3.5) — asks the peer to RESET its corresponding
* send side. Always non-null on streams surfaced here: every
* stripped stream has a read side. First call wins.
* send side. Non-null because every stripped stream surfaced
* here has a read side; the demux always wires it. First call
* wins.
*/
val stopSending: (suspend (errorCode: Long) -> Unit)? = null,
val stopSending: suspend (errorCode: Long) -> Unit,
)
/**