From cfc305feb30cad69bdfdb23c3cb2df60b075f0c3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 00:21:42 +0000 Subject: [PATCH] feat(quic): add qlog observer infrastructure for interop diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hooks every QUIC protocol decision (packets sent / received / dropped, TLS key updates, transport params, ALPN, loss detection, PTO, close) into a [QlogObserver] interface. Production callers default to [QlogObserver.NoOp] (zero allocation, single virtual call); the :quic interop runner wires a [QlogWriter] writing JSON-NDJSON (qlog 0.3 / JSON-SEQ format) to `/client.sqlog`, consumable by qvis (https://qvis.quictools.info/) and Wireshark. Goal: every interop-runner test failure produces a qlog file the operator can drop into qvis to see exactly what we did differently from the spec. Hooked call sites: - QuicConnection.start → connection_started, parameters_set(local), version_information - QuicConnection.close → connection_closed(local) - QuicConnection.markClosedExternally → connection_closed(remote) - QuicConnection.applyPeerTransportParameters → parameters_set(remote) - QuicConnection.tlsListener (handshake/app keys) → security:key_updated - QuicConnection.tlsListener (handshake done) → alpn_information - QuicConnectionWriter.buildLongHeaderFromFrames → packet_sent (initial / handshake) - QuicConnectionWriter.buildApplicationPacket → packet_sent (1-rtt) - QuicConnectionWriter.buildBestLevelPacket → packet_sent (close-path) - QuicConnectionParser.feedLongHeaderPacket → packet_received / packet_dropped - QuicConnectionParser.feedShortHeaderPacket → packet_received / packet_dropped - QuicConnectionParser AckFrame loss-detect → recovery:packet_lost - QuicConnectionDriver.sendLoop PTO branch → recovery:loss_timer_updated (pto) --- .../quic/connection/QuicConnection.kt | 67 ++++ .../quic/connection/QuicConnectionDriver.kt | 14 +- .../quic/connection/QuicConnectionParser.kt | 92 ++++- .../quic/connection/QuicConnectionWriter.kt | 91 +++-- .../quic/observability/QlogObserver.kt | 255 ++++++++++++++ .../quic/observability/QlogObserverTest.kt | 286 ++++++++++++++++ .../quic/interop/InteropRunner.kt | 21 ++ .../vitorpamplona/quic/interop/QlogWriter.kt | 319 ++++++++++++++++++ .../quic/interop/QlogWriterTest.kt | 161 +++++++++ 9 files changed, 1267 insertions(+), 39 deletions(-) create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/observability/QlogObserver.kt create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/observability/QlogObserverTest.kt create mode 100644 quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriter.kt create mode 100644 quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriterTest.kt diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index cbdbe2e9d..668579670 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection import com.vitorpamplona.quic.crypto.InitialSecrets import com.vitorpamplona.quic.crypto.PlatformAesOneBlock import com.vitorpamplona.quic.crypto.bestAes128GcmAead +import com.vitorpamplona.quic.observability.QlogObserver import com.vitorpamplona.quic.stream.QuicStream import com.vitorpamplona.quic.stream.StreamId import com.vitorpamplona.quic.tls.TlsClient @@ -73,6 +74,13 @@ class QuicConnection( .toEpochMilliseconds() }, val alpnList: List = listOf(TlsConstants.ALPN_H3), + /** + * Optional qlog observer (draft-marx-qlog). Production callers + * leave this at [QlogObserver.NoOp] (zero overhead). Interop / + * test runners attach a JSON-NDJSON writer so a failed run + * produces a `client.sqlog` consumable by qvis. + */ + val qlogObserver: QlogObserver = QlogObserver.NoOp, ) { val sourceConnectionId: ConnectionId = ConnectionId.random(8) var destinationConnectionId: ConnectionId = ConnectionId.random(8) @@ -317,6 +325,8 @@ class QuicConnection( ) { handshake.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret) handshake.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret) + qlogObserver.onKeyUpdated("client", EncryptionLevel.HANDSHAKE) + qlogObserver.onKeyUpdated("server", EncryptionLevel.HANDSHAKE) } override fun onApplicationKeysReady( @@ -326,12 +336,15 @@ class QuicConnection( ) { application.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret) application.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret) + qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION) + qlogObserver.onKeyUpdated("server", EncryptionLevel.APPLICATION) } override fun onHandshakeComplete() { handshakeComplete = true if (status == Status.HANDSHAKING) status = Status.CONNECTED applyPeerTransportParameters() + tls.negotiatedAlpn?.let { qlogObserver.onAlpnNegotiated(it.decodeToString()) } handshakeDoneSignal.complete(Unit) } } @@ -375,11 +388,52 @@ class QuicConnection( /** Begin the handshake — emits ClientHello into Initial CRYPTO. */ fun start() { + // qlog: emit connection_started + initial transport_parameters_set + // before any wire traffic so the trace makes chronological sense + // when handed to qvis. + qlogObserver.onConnectionStarted( + serverName = serverName, + dcid = destinationConnectionId.bytes, + scid = sourceConnectionId.bytes, + ) + qlogObserver.onTransportParametersSet("local", localTransportParametersSummary()) + // RFC 9000 §6: we're not doing version negotiation, so the + // chosen version is unconditional. + qlogObserver.onVersionInformation("v1", emptyList()) tls.start() // Drain ClientHello bytes into the Initial-level CRYPTO send buffer. tls.pollOutbound(TlsClient.Level.INITIAL)?.let { initial.cryptoSend.enqueue(it) } } + private fun localTransportParametersSummary(): Map { + val out = LinkedHashMap(8) + out["initial_max_data"] = config.initialMaxData.toString() + out["initial_max_stream_data_bidi_local"] = config.initialMaxStreamDataBidiLocal.toString() + out["initial_max_stream_data_bidi_remote"] = config.initialMaxStreamDataBidiRemote.toString() + out["initial_max_stream_data_uni"] = config.initialMaxStreamDataUni.toString() + out["initial_max_streams_bidi"] = config.initialMaxStreamsBidi.toString() + out["initial_max_streams_uni"] = config.initialMaxStreamsUni.toString() + out["max_idle_timeout"] = config.maxIdleTimeoutMillis.toString() + out["max_udp_payload_size"] = config.maxUdpPayloadSize.toString() + out["max_datagram_frame_size"] = config.maxDatagramFrameSize.toString() + return out + } + + private fun peerTransportParametersSummary(tp: TransportParameters): Map { + val out = LinkedHashMap(10) + tp.initialMaxData?.let { out["initial_max_data"] = it.toString() } + tp.initialMaxStreamDataBidiLocal?.let { out["initial_max_stream_data_bidi_local"] = it.toString() } + tp.initialMaxStreamDataBidiRemote?.let { out["initial_max_stream_data_bidi_remote"] = it.toString() } + tp.initialMaxStreamDataUni?.let { out["initial_max_stream_data_uni"] = it.toString() } + tp.initialMaxStreamsBidi?.let { out["initial_max_streams_bidi"] = it.toString() } + tp.initialMaxStreamsUni?.let { out["initial_max_streams_uni"] = it.toString() } + tp.maxIdleTimeoutMillis?.let { out["max_idle_timeout"] = it.toString() } + tp.maxUdpPayloadSize?.let { out["max_udp_payload_size"] = it.toString() } + tp.maxDatagramFrameSize?.let { out["max_datagram_frame_size"] = it.toString() } + tp.maxAckDelay?.let { out["max_ack_delay"] = it.toString() } + return out + } + private fun buildLocalTransportParameters(): TransportParameters = TransportParameters( initialMaxData = config.initialMaxData, @@ -425,6 +479,7 @@ class QuicConnection( return } peerTransportParameters = tp + qlogObserver.onTransportParametersSet("remote", peerTransportParametersSummary(tp)) sendConnectionFlowCredit = tp.initialMaxData ?: 0L peerMaxStreamsBidi = tp.initialMaxStreamsBidi ?: 0L peerMaxStreamsUni = tp.initialMaxStreamsUni ?: 0L @@ -631,12 +686,15 @@ class QuicConnection( errorCode: Long, reason: String, ) { + var firedQlog = false lock.withLock { if (status == Status.CLOSED || status == Status.CLOSING) return@withLock closeErrorCode = errorCode closeReason = reason status = Status.CLOSING + firedQlog = true } + if (firedQlog) qlogObserver.onConnectionClosed("local", errorCode, reason) // If a caller is suspended on awaitHandshake() and we're tearing down // before completion, fail the deferred so the caller throws instead // of hanging forever. @@ -648,7 +706,16 @@ class QuicConnection( /** Called by the parser on inbound CONNECTION_CLOSE or by the driver on read-loop death. */ internal fun markClosedExternally(reason: String) { + val wasClosed = status == Status.CLOSED if (status != Status.CLOSED) status = Status.CLOSED + if (!wasClosed) { + // "remote" covers both peer-initiated CONNECTION_CLOSE and + // local invariant violations (CID mismatch, frame decode + // failure) that the parser surfaces as markClosedExternally. + // The reason string is the discriminator the trace consumer + // reads. + qlogObserver.onConnectionClosed("remote", closeErrorCode, reason) + } if (!handshakeComplete) { signalHandshakeFailed(QuicConnectionClosedException("connection closed externally: $reason")) } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index 48f198607..052e8ea9a 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -149,11 +149,15 @@ class QuicConnectionDriver( // PING on the next drain (RFC 9002 §6.2.4 probe // packet). The peer's ACK feeds loss detection + // retransmit (steps 5–6). - connection.lock.withLock { - connection.pendingPing = true - connection.consecutivePtoCount = - (connection.consecutivePtoCount + 1).coerceAtMost(6) - } + val newPtoCount = + connection.lock + .withLock { + connection.pendingPing = true + connection.consecutivePtoCount = + (connection.consecutivePtoCount + 1).coerceAtMost(6) + connection.consecutivePtoCount + } + connection.qlogObserver.onPtoFired(newPtoCount, ptoMillis) } } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index 284b05f4f..d1f643835 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -37,6 +37,7 @@ import com.vitorpamplona.quic.frame.ResetStreamFrame import com.vitorpamplona.quic.frame.StopSendingFrame import com.vitorpamplona.quic.frame.StreamFrame import com.vitorpamplona.quic.frame.decodeFrames +import com.vitorpamplona.quic.observability.qlogFrameName import com.vitorpamplona.quic.packet.LongHeaderPacket import com.vitorpamplona.quic.packet.LongHeaderType import com.vitorpamplona.quic.packet.ShortHeaderPacket @@ -87,12 +88,33 @@ private fun feedLongHeaderPacket( val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null val level = when (peeked.type) { - LongHeaderType.INITIAL -> EncryptionLevel.INITIAL - LongHeaderType.HANDSHAKE -> EncryptionLevel.HANDSHAKE - LongHeaderType.ZERO_RTT, LongHeaderType.RETRY -> return null // unsupported in client + LongHeaderType.INITIAL -> { + EncryptionLevel.INITIAL + } + + LongHeaderType.HANDSHAKE -> { + EncryptionLevel.HANDSHAKE + } + + LongHeaderType.ZERO_RTT, LongHeaderType.RETRY -> { + // Not supported by client; surface as a drop so qvis can + // see we ignored a packet rather than silently moving on. + conn.qlogObserver.onPacketDropped( + "unsupported long-header type ${peeked.type}", + peeked.totalLength, + ) + return null + } } val state = conn.levelState(level) - val proto = state.receiveProtection ?: return null + val proto = state.receiveProtection + if (proto == null) { + conn.qlogObserver.onPacketDropped( + "no receive keys at level $level", + peeked.totalLength, + ) + return null + } val parsed = LongHeaderPacket.parseAndDecrypt( bytes = datagram, @@ -103,7 +125,14 @@ private fun feedLongHeaderPacket( hp = proto.hp, hpKey = proto.hpKey, largestReceivedInSpace = state.pnSpace.largestReceived, - ) ?: return null + ) + if (parsed == null) { + conn.qlogObserver.onPacketDropped( + "AEAD auth failed or header parse failed at level $level", + peeked.totalLength, + ) + return null + } state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) @@ -112,6 +141,14 @@ private fun feedLongHeaderPacket( conn.destinationConnectionId = parsed.packet.scid } + if (conn.qlogObserver !== com.vitorpamplona.quic.observability.QlogObserver.NoOp) { + conn.qlogObserver.onPacketReceived( + level = level, + packetNumber = parsed.packet.packetNumber, + sizeBytes = parsed.consumed, + frames = peekFrameNames(parsed.packet.payload), + ) + } dispatchFrames(conn, level, parsed.packet.payload, parsed.packet.packetNumber, nowMillis) return parsed.consumed } @@ -123,7 +160,14 @@ private fun feedShortHeaderPacket( nowMillis: Long, ) { val state = conn.levelState(EncryptionLevel.APPLICATION) - val proto = state.receiveProtection ?: return + val proto = state.receiveProtection + if (proto == null) { + conn.qlogObserver.onPacketDropped( + "no application receive keys", + datagram.size - offset, + ) + return + } val parsed = ShortHeaderPacket.parseAndDecrypt( bytes = datagram, @@ -135,11 +179,42 @@ private fun feedShortHeaderPacket( hp = proto.hp, hpKey = proto.hpKey, largestReceivedInSpace = state.pnSpace.largestReceived, - ) ?: return + ) + if (parsed == null) { + conn.qlogObserver.onPacketDropped( + "AEAD auth failed or header parse failed at level APPLICATION", + datagram.size - offset, + ) + return + } state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) + if (conn.qlogObserver !== com.vitorpamplona.quic.observability.QlogObserver.NoOp) { + conn.qlogObserver.onPacketReceived( + level = EncryptionLevel.APPLICATION, + packetNumber = parsed.packet.packetNumber, + sizeBytes = datagram.size - offset, + frames = peekFrameNames(parsed.packet.payload), + ) + } dispatchFrames(conn, EncryptionLevel.APPLICATION, parsed.packet.payload, parsed.packet.packetNumber, nowMillis) } +/** + * Decode the payload's frames just to surface their qlog names. Reuses + * the same [com.vitorpamplona.quic.frame.decodeFrames] path as + * [dispatchFrames]; if it throws (malformed peer payload), we return + * an empty list — the dispatch path will catch the same exception + * and surface the close via `markClosedExternally`. + */ +private fun peekFrameNames(payload: ByteArray): List = + try { + com.vitorpamplona.quic.frame + .decodeFrames(payload) + .map { qlogFrameName(it::class.simpleName ?: "frame") } + } catch (_: QuicCodecException) { + emptyList() + } + private fun dispatchFrames( conn: QuicConnection, level: EncryptionLevel, @@ -225,6 +300,9 @@ private fun dispatchFrames( for (lostPacket in lost) { conn.onTokensLost(lostPacket.tokens) } + if (lost.isNotEmpty()) { + conn.qlogObserver.onLossDetected(level, lost.map { it.packetNumber }) + } } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 7b5598738..c2463c3a4 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -37,6 +37,8 @@ import com.vitorpamplona.quic.frame.ResetStreamFrame import com.vitorpamplona.quic.frame.StopSendingFrame import com.vitorpamplona.quic.frame.StreamFrame import com.vitorpamplona.quic.frame.encodeFrames +import com.vitorpamplona.quic.observability.QlogObserver +import com.vitorpamplona.quic.observability.qlogFrameName import com.vitorpamplona.quic.packet.LongHeaderPacket import com.vitorpamplona.quic.packet.LongHeaderPlaintextPacket import com.vitorpamplona.quic.packet.LongHeaderType @@ -176,23 +178,26 @@ private fun buildBestLevelPacket( if (app.sendProtection != null) { val proto = app.sendProtection!! val pn = app.pnSpace.allocateOutbound() - return ShortHeaderPacket.build( - ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload), - proto.aead, - proto.key, - proto.iv, - proto.hp, - proto.hpKey, - largestAckedInSpace = -1L, - ) + val built = + ShortHeaderPacket.build( + ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) + emitQlogSent(conn, EncryptionLevel.APPLICATION, pn, built.size, frames) + return built } val hs = conn.handshake if (hs.sendProtection != null) { - return buildLongHeaderPacket(conn, EncryptionLevel.HANDSHAKE, payload) + return buildLongHeaderPacket(conn, EncryptionLevel.HANDSHAKE, payload, frames) } val init = conn.initial if (init.sendProtection != null) { - return buildLongHeaderPacket(conn, EncryptionLevel.INITIAL, payload) + return buildLongHeaderPacket(conn, EncryptionLevel.INITIAL, payload, frames) } return null } @@ -201,6 +206,7 @@ private fun buildLongHeaderPacket( conn: QuicConnection, level: EncryptionLevel, payload: ByteArray, + frames: List, ): ByteArray { val state = conn.levelState(level) val proto = state.sendProtection!! @@ -211,22 +217,25 @@ private fun buildLongHeaderPacket( EncryptionLevel.HANDSHAKE -> LongHeaderType.HANDSHAKE EncryptionLevel.APPLICATION -> error("APPLICATION uses short-header packets") } - return LongHeaderPacket.build( - LongHeaderPlaintextPacket( - type = type, - version = QuicVersion.V1, - dcid = conn.destinationConnectionId, - scid = conn.sourceConnectionId, - packetNumber = pn, - payload = payload, - ), - proto.aead, - proto.key, - proto.iv, - proto.hp, - proto.hpKey, - largestAckedInSpace = -1L, - ) + val built = + LongHeaderPacket.build( + LongHeaderPlaintextPacket( + type = type, + version = QuicVersion.V1, + dcid = conn.destinationConnectionId, + scid = conn.sourceConnectionId, + packetNumber = pn, + payload = payload, + ), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) + emitQlogSent(conn, level, pn, built.size, frames) + return built } /** @@ -345,9 +354,34 @@ private fun buildLongHeaderFromFrames( sizeBytes = packet.size, tokens = tokens, ) + emitQlogSent(conn, level, pn, packet.size, frames) return packet } +/** + * Fire [QlogObserver.onPacketSent] for one outbound packet. Skipped + * fast for the [QlogObserver.NoOp] default so production callers pay + * only one identity comparison + one virtual call. + */ +private fun emitQlogSent( + conn: QuicConnection, + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, +) { + val observer = conn.qlogObserver + if (observer === QlogObserver.NoOp) return + val frameNames = ArrayList(frames.size) + for (f in frames) frameNames += qlogFrameName(f::class.simpleName ?: "frame") + observer.onPacketSent( + level = level, + packetNumber = packetNumber, + sizeBytes = sizeBytes, + frames = frameNames, + ) +} + private fun buildApplicationPacket( conn: QuicConnection, nowMillis: Long, @@ -510,6 +544,9 @@ private fun buildApplicationPacket( sizeBytes = sizeBytes.getOrNull()?.size ?: 0, tokens = tokens.toList(), ) + sizeBytes.getOrNull()?.let { built -> + emitQlogSent(conn, EncryptionLevel.APPLICATION, pn, built.size, frames) + } // Re-throw on encrypt failure so callers (driver loop) see the // same exception they did before this change. The bookkeeping // entry is still in place; if the throw was transient, retransmit diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/observability/QlogObserver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/observability/QlogObserver.kt new file mode 100644 index 000000000..e977c4f11 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/observability/QlogObserver.kt @@ -0,0 +1,255 @@ +/* + * 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.observability + +import com.vitorpamplona.quic.connection.EncryptionLevel + +/** + * qlog (draft-marx-qlog) observer interface for QUIC connection events. + * + * Tools like qvis (https://qvis.quictools.info/) and Wireshark consume + * qlog files to render sequence diagrams + RTT graphs + recovery + * timelines. The on-the-wire format is JSON-NDJSON (one JSON object + * per line); this interface decouples event emission from format — + * production code can pass [NoOp] (zero overhead), and the + * `:quic` interop runner attaches a JSON-writing implementation that + * produces a `client.sqlog` consumable by qvis. + * + * Goal: every interop-runner test failure produces a qlog file the + * caller can drop into qvis to see exactly what we did differently + * from the spec. + * + * Performance: hooks are on the hot packet-send/receive path, so + * [NoOp] must be a JIT-able single virtual call with no allocation. + * Default-method bodies in Kotlin compile to single bytecode `RETURN`, + * which HotSpot inlines reliably. + */ +interface QlogObserver { + /** Called once after [com.vitorpamplona.quic.connection.QuicConnection.start] runs. */ + fun onConnectionStarted( + serverName: String, + dcid: ByteArray, + scid: ByteArray, + ) + + /** + * Called when the connection transitions to CLOSED, either locally + * (close()) or due to an inbound CONNECTION_CLOSE / read-loop + * termination ([com.vitorpamplona.quic.connection.QuicConnection.markClosedExternally]). + * + * @param initiator `"local"` if we initiated, `"remote"` otherwise. + */ + fun onConnectionClosed( + initiator: String, + errorCode: Long, + reason: String, + ) + + /** + * One outbound packet at [level] just hit the wire. Called once per + * coalesced packet inside a UDP datagram, so a single + * datagram carrying Initial + Handshake fires twice. + */ + fun onPacketSent( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) + + /** One inbound packet at [level] was successfully decrypted + dispatched. */ + fun onPacketReceived( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) + + /** + * An inbound packet (or whole datagram) was dropped on the floor — + * AEAD AUTH FAIL, unknown DCID, missing receive keys, version + * mismatch, frame decode failure, etc. + */ + fun onPacketDropped( + reason: String, + sizeBytes: Int, + ) + + /** + * TLS produced new keys at the given encryption level. + * + * @param keyType `"server"` or `"client"` (which direction the + * key applies to). + */ + fun onKeyUpdated( + keyType: String, + level: EncryptionLevel, + ) + + /** + * RFC 9002 §6.1 loss detection declared one or more outbound + * packets at [level] lost. + */ + fun onLossDetected( + level: EncryptionLevel, + lostPacketNumbers: List, + ) + + /** + * RFC 9002 §6.2 PTO timer expired; the writer will emit a PING on + * the next drain to elicit an ACK from the peer. + * + * @param consecutivePtoCount the new PTO count (post-increment). + * @param ptoMillis the PTO duration that just expired. + */ + fun onPtoFired( + consecutivePtoCount: Int, + ptoMillis: Long, + ) + + /** + * Congestion-controller state transition (slow-start ↔ + * recovery ↔ congestion-avoidance). qvis renders this as + * background bands on the RTT timeline. + */ + fun onCongestionStateUpdated(newState: String) + + /** + * QUIC transport parameters were set by [initiator] (`"local"` at + * connection-open, `"remote"` once the peer's params arrive in + * EncryptedExtensions). [params] is a flat label→value map of + * the parameters that the implementation chose to surface; the + * exact key set is not part of the qlog 0.3 contract. + */ + fun onTransportParametersSet( + initiator: String, + params: Map, + ) + + /** + * The TLS handshake completed and an ALPN was selected (or `null` + * was chosen — [alpn] is the human-readable name, e.g. `"h3"`). + */ + fun onAlpnNegotiated(alpn: String) + + /** + * RFC 9000 §6 Version Information. Single-fire — emitted once + * after Version Negotiation resolves. + */ + fun onVersionInformation( + chosenVersion: String, + otherVersionsOffered: List, + ) + + /** + * No-op observer. Default for production callers — every method + * is an empty body that the JIT inlines. No allocation, no I/O. + */ + object NoOp : QlogObserver { + override fun onConnectionStarted( + serverName: String, + dcid: ByteArray, + scid: ByteArray, + ) = Unit + + override fun onConnectionClosed( + initiator: String, + errorCode: Long, + reason: String, + ) = Unit + + override fun onPacketSent( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) = Unit + + override fun onPacketReceived( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) = Unit + + override fun onPacketDropped( + reason: String, + sizeBytes: Int, + ) = Unit + + override fun onKeyUpdated( + keyType: String, + level: EncryptionLevel, + ) = Unit + + override fun onLossDetected( + level: EncryptionLevel, + lostPacketNumbers: List, + ) = Unit + + override fun onPtoFired( + consecutivePtoCount: Int, + ptoMillis: Long, + ) = Unit + + override fun onCongestionStateUpdated(newState: String) = Unit + + override fun onTransportParametersSet( + initiator: String, + params: Map, + ) = Unit + + override fun onAlpnNegotiated(alpn: String) = Unit + + override fun onVersionInformation( + chosenVersion: String, + otherVersionsOffered: List, + ) = Unit + } +} + +/** + * Map a [com.vitorpamplona.quic.frame.Frame] subclass simple-name to + * the qlog frame_type label. qlog 0.3 specifies snake_case with + * `_frame` suffix stripped — e.g. `MaxStreamsFrame` → `max_streams`. + * + * Lives next to [QlogObserver] so writer + parser hot paths share the + * same conversion (used to fill [QlogObserver.onPacketSent.frames] and + * [QlogObserver.onPacketReceived.frames] without round-tripping through + * Jackson). + */ +fun qlogFrameName(simpleClassName: String): String { + // Strip trailing "Frame" then convert CamelCase to snake_case. + val noSuffix = + if (simpleClassName.endsWith("Frame")) { + simpleClassName.dropLast("Frame".length) + } else { + simpleClassName + } + if (noSuffix.isEmpty()) return simpleClassName.lowercase() + val sb = StringBuilder(noSuffix.length + 4) + for (i in noSuffix.indices) { + val c = noSuffix[i] + if (i > 0 && c.isUpperCase()) sb.append('_') + sb.append(c.lowercaseChar()) + } + return sb.toString() +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/observability/QlogObserverTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/observability/QlogObserverTest.kt new file mode 100644 index 000000000..eaec42808 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/observability/QlogObserverTest.kt @@ -0,0 +1,286 @@ +/* + * 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.observability + +import com.vitorpamplona.quic.connection.ConnectionId +import com.vitorpamplona.quic.connection.EncryptionLevel +import com.vitorpamplona.quic.connection.InMemoryQuicPipe +import com.vitorpamplona.quic.connection.QuicConnection +import com.vitorpamplona.quic.connection.QuicConnectionConfig +import com.vitorpamplona.quic.connection.TransportParameters +import com.vitorpamplona.quic.connection.feedDatagram +import com.vitorpamplona.quic.tls.InProcessTlsServer +import com.vitorpamplona.quic.tls.PermissiveCertificateValidator +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.withLock +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * End-to-end smoke for [QlogObserver]: + * + * 1. NoOp safety — a connection driven through start → handshake → + * close with [QlogObserver.NoOp] must complete without throwing. + * 2. RecordingQlogObserver captures the expected events (start, + * local transport_params, packet_sent for ClientHello, close). + * 3. Malformed inbound datagram surfaces a `packet_dropped` event. + * + * The recorded-event list documented in the implementation report + * comes from the second test below. + */ +class QlogObserverTest { + @Test + fun noOpObserver_handshakeAndCloseDoNotThrow(): Unit = + runBlocking { + val client = handshakedClient(QlogObserver.NoOp) + // Drive a tiny operation post-handshake. + client.lock.withLock { + // No-op — just confirm we can still take the lock. + assertEquals(QuicConnection.Status.CONNECTED, client.status) + } + client.close(0L, "done") + // close() flips status to CLOSING; the writer's next + // drainOutbound emits CONNECTION_CLOSE and transitions to + // CLOSED. We don't run the driver here, so stop at CLOSING + // and assert that — the point of this test is just that + // the no-op observer doesn't throw on any call site. + assertTrue( + client.status == QuicConnection.Status.CLOSING || + client.status == QuicConnection.Status.CLOSED, + "expected CLOSING or CLOSED, was ${client.status}", + ) + } + + @Test + fun recordingObserver_capturesStartParametersSentAndClose(): Unit = + runBlocking { + val recorder = RecordingQlogObserver() + val client = handshakedClient(recorder) + client.close(0L, "shutdown") + + val names = recorder.events.map { it.name } + assertTrue( + names.contains("connectionStarted"), + "expected connectionStarted in $names", + ) + // Local transport parameters set at start. + val localTps = + recorder.events + .filter { it.name == "transportParametersSet" } + .map { it.payload["initiator"] } + assertTrue( + localTps.contains("local"), + "expected local transportParametersSet in $localTps", + ) + // At least one packet_sent for the ClientHello at INITIAL level. + val initialSends = + recorder.events.filter { + it.name == "packetSent" && it.payload["level"] == EncryptionLevel.INITIAL + } + assertTrue( + initialSends.isNotEmpty(), + "expected at least one INITIAL-level packetSent (ClientHello) " + + "but saw ${recorder.events.map { it.name to it.payload["level"] }}", + ) + assertTrue( + names.contains("connectionClosed"), + "expected connectionClosed in $names", + ) + } + + @Test + fun malformedDatagram_recordsPacketDropped(): Unit = + runBlocking { + val recorder = RecordingQlogObserver() + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = PermissiveCertificateValidator(), + qlogObserver = recorder, + ) + client.start() + // Long-header packet bytes that look parseable enough to peek + // but won't decrypt — the receive keys for HANDSHAKE/APPLICATION + // aren't installed yet, so feedDatagram drops with "no receive + // keys at level …". 0xC0 = long header, INITIAL with the + // smallest valid layout: 0xC0 | version(0x00000001) | dcil=0 + // | scil=0 | token_len=0 | length=2 | pn=0x00 | one byte of + // garbage. AEAD-decrypt will fail on this Initial too, since + // the keys are derived from a different DCID. + val garbage = + byteArrayOf( + 0xC0.toByte(), // long header initial + 0x00, + 0x00, + 0x00, + 0x01, // version v1 + 0x00, // dcil = 0 + 0x00, // scil = 0 + 0x00, // token length varint = 0 + 0x02, // length = 2 + 0x00, // pn byte + 0x00, // payload byte (will fail AEAD) + ) + client.lock.withLock { + feedDatagram(client, garbage, nowMillis = 1L) + } + val drops = recorder.events.filter { it.name == "packetDropped" } + assertTrue( + drops.isNotEmpty(), + "expected at least one packetDropped event but saw ${recorder.events.map { it.name }}", + ) + } + + private fun handshakedClient(observer: QlogObserver): QuicConnection = + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = PermissiveCertificateValidator(), + qlogObserver = observer, + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 1_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + initialMaxStreamsBidi = 100, + initialMaxStreamsUni = 100, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + client + } +} + +/** + * Test-only [QlogObserver] that appends every callback into a list, + * with a tiny serialized-payload shape so assertions read like + * structured pattern-matches rather than a soup of positional args. + */ +internal class RecordingQlogObserver : QlogObserver { + data class Event( + val name: String, + val payload: Map, + ) + + val events: MutableList = mutableListOf() + + private fun add( + name: String, + payload: Map, + ) { + events += Event(name, payload) + } + + override fun onConnectionStarted( + serverName: String, + dcid: ByteArray, + scid: ByteArray, + ) = add( + "connectionStarted", + mapOf("serverName" to serverName, "dcid_size" to dcid.size, "scid_size" to scid.size), + ) + + override fun onConnectionClosed( + initiator: String, + errorCode: Long, + reason: String, + ) = add( + "connectionClosed", + mapOf("initiator" to initiator, "errorCode" to errorCode, "reason" to reason), + ) + + override fun onPacketSent( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) = add( + "packetSent", + mapOf("level" to level, "pn" to packetNumber, "size" to sizeBytes, "frames" to frames), + ) + + override fun onPacketReceived( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) = add( + "packetReceived", + mapOf("level" to level, "pn" to packetNumber, "size" to sizeBytes, "frames" to frames), + ) + + override fun onPacketDropped( + reason: String, + sizeBytes: Int, + ) = add("packetDropped", mapOf("reason" to reason, "size" to sizeBytes)) + + override fun onKeyUpdated( + keyType: String, + level: EncryptionLevel, + ) = add("keyUpdated", mapOf("keyType" to keyType, "level" to level)) + + override fun onLossDetected( + level: EncryptionLevel, + lostPacketNumbers: List, + ) = add("lossDetected", mapOf("level" to level, "lost" to lostPacketNumbers)) + + override fun onPtoFired( + consecutivePtoCount: Int, + ptoMillis: Long, + ) = add("ptoFired", mapOf("count" to consecutivePtoCount, "ptoMillis" to ptoMillis)) + + override fun onCongestionStateUpdated(newState: String) = add("congestionStateUpdated", mapOf("newState" to newState)) + + override fun onTransportParametersSet( + initiator: String, + params: Map, + ) = add("transportParametersSet", mapOf("initiator" to initiator, "params" to params)) + + override fun onAlpnNegotiated(alpn: String) = add("alpnNegotiated", mapOf("alpn" to alpn)) + + override fun onVersionInformation( + chosenVersion: String, + otherVersionsOffered: List, + ) = add( + "versionInformation", + mapOf("chosen" to chosenVersion, "offered" to otherVersionsOffered), + ) +} diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/InteropRunner.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/InteropRunner.kt index 514df4d5d..b638c3e5b 100644 --- a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/InteropRunner.kt +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/InteropRunner.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quic.interop import com.vitorpamplona.quic.connection.QuicConnection import com.vitorpamplona.quic.connection.QuicConnectionConfig import com.vitorpamplona.quic.connection.QuicConnectionDriver +import com.vitorpamplona.quic.observability.QlogObserver import com.vitorpamplona.quic.tls.PermissiveCertificateValidator import com.vitorpamplona.quic.transport.UdpSocket import kotlinx.coroutines.CoroutineScope @@ -31,6 +32,7 @@ import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeoutOrNull +import java.io.File /** * Standalone interop runner. Drives [QuicConnection] against a real QUIC @@ -58,6 +60,20 @@ fun main(args: Array) { println("timeout: ${timeoutSec}s") println() + // qlog: if QLOGDIR is set, drop a `client.sqlog` file at + // /client.sqlog so the operator can hand the failed run + // to qvis. The interop-runner contract from the IETF QUIC team's + // Docker harness (quic-interop-runner) sets this env var on every + // test invocation. + val qlogDir = + (System.getenv("QLOGDIR") ?: System.getProperty("QLOGDIR"))?.takeIf { it.isNotBlank() } + val qlogWriter: QlogWriter? = + qlogDir?.let { dir -> + val parent = File(dir).also { it.mkdirs() } + QlogWriter(File(parent, "client.sqlog"), odcidHex = "00") + } + val qlogObserver: QlogObserver = qlogWriter ?: QlogObserver.NoOp + val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) val outcome = runBlocking { @@ -73,6 +89,7 @@ fun main(args: Array) { serverName = host, config = QuicConnectionConfig(), tlsCertificateValidator = PermissiveCertificateValidator(), + qlogObserver = qlogObserver, ) val driver = QuicConnectionDriver(conn, socket, scope) driver.start() @@ -110,6 +127,10 @@ fun main(args: Array) { // are torn down before main() exits. Without this the JVM hangs on stray // IO-dispatcher threads. scope.cancel() + // Flush + close the qlog file before exiting. Without this, an + // exitProcess() call below could leave the trailing events + // unflushed. + runCatching { qlogWriter?.close() } when (outcome) { is InteropOutcome.Connected -> { diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriter.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriter.kt new file mode 100644 index 000000000..585d17c3e --- /dev/null +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriter.kt @@ -0,0 +1,319 @@ +/* + * 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.interop + +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.quic.connection.EncryptionLevel +import com.vitorpamplona.quic.observability.QlogObserver +import java.io.BufferedWriter +import java.io.Closeable +import java.io.File +import java.io.FileWriter +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +/** + * JSON-NDJSON qlog writer (qlog 0.3 / JSON-SEQ format) used by the + * `:quic` interop runner. One JSON object per line; first line is the + * qlog header, subsequent lines are events. + * + * Tools like qvis (https://qvis.quictools.info/) consume the resulting + * `.sqlog` file to render sequence diagrams + RTT graphs + recovery + * timelines. + * + * **Goal: every interop-runner test failure produces a qlog file the + * caller can drop into qvis to see exactly what we did differently + * from the spec.** + * + * Threading: [java.io.BufferedWriter] is not safe for concurrent + * writers; we hold a [ReentrantLock] around each line emit so the + * read + send loops can fire events concurrently without interleaving + * partial JSON. + */ +class QlogWriter( + file: File, + private val odcidHex: String, + private val mapper: ObjectMapper = DEFAULT_MAPPER, + /** Wall-clock provider; tests inject a deterministic source. */ + private val nowMillis: () -> Long = { System.currentTimeMillis() }, +) : QlogObserver, + Closeable { + // append=false → truncate any prior trace at this path so a reused + // QLOGDIR doesn't accumulate stale events from a previous run. + private val writer: BufferedWriter = BufferedWriter(FileWriter(file, false)) + private val lock = ReentrantLock() + private val startMillis: Long = nowMillis() + + init { + // qlog 0.3 JSON-SEQ header. qvis tolerates both `qlog_format` + // values "JSON-SEQ" and "NDJSON"; we use JSON-SEQ to match the + // most-common production qlog files (Chromium, mvfst). + val header = + mapOf( + "qlog_version" to "0.3", + "qlog_format" to "JSON-SEQ", + "title" to "amethyst :quic client trace", + "trace" to + mapOf( + "vantage_point" to mapOf("type" to "client", "name" to "amethyst-quic"), + "common_fields" to + mapOf( + "ODCID" to odcidHex, + "reference_time" to startMillis, + "time_format" to "relative", + ), + ), + ) + writeLineLocked(mapper.writeValueAsString(header)) + } + + override fun onConnectionStarted( + serverName: String, + dcid: ByteArray, + scid: ByteArray, + ) { + emit( + "transport:connection_started", + mapOf( + "ip_version" to "v4_or_v6", + "server_name" to serverName, + "dst_cid" to hex(dcid), + "src_cid" to hex(scid), + ), + ) + } + + override fun onConnectionClosed( + initiator: String, + errorCode: Long, + reason: String, + ) { + emit( + "transport:connection_closed", + mapOf( + "owner" to initiator, + "application_code" to errorCode, + "reason" to reason, + ), + ) + } + + override fun onPacketSent( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) { + emit( + "transport:packet_sent", + mapOf( + "header" to + mapOf( + "packet_type" to packetTypeFor(level), + "packet_number" to packetNumber, + ), + "raw" to mapOf("length" to sizeBytes), + "frames" to frames.map { mapOf("frame_type" to it) }, + ), + ) + } + + override fun onPacketReceived( + level: EncryptionLevel, + packetNumber: Long, + sizeBytes: Int, + frames: List, + ) { + emit( + "transport:packet_received", + mapOf( + "header" to + mapOf( + "packet_type" to packetTypeFor(level), + "packet_number" to packetNumber, + ), + "raw" to mapOf("length" to sizeBytes), + "frames" to frames.map { mapOf("frame_type" to it) }, + ), + ) + } + + override fun onPacketDropped( + reason: String, + sizeBytes: Int, + ) { + emit( + "transport:packet_dropped", + mapOf( + "trigger" to reason, + "raw" to mapOf("length" to sizeBytes), + ), + ) + } + + override fun onKeyUpdated( + keyType: String, + level: EncryptionLevel, + ) { + emit( + "security:key_updated", + mapOf( + "key_type" to "${keyType}_${packetTypeFor(level)}_secret", + "trigger" to "tls", + ), + ) + } + + override fun onLossDetected( + level: EncryptionLevel, + lostPacketNumbers: List, + ) { + for (pn in lostPacketNumbers) { + emit( + "recovery:packet_lost", + mapOf( + "header" to + mapOf( + "packet_type" to packetTypeFor(level), + "packet_number" to pn, + ), + "trigger" to "reordering_threshold_or_time_threshold", + ), + ) + } + } + + override fun onPtoFired( + consecutivePtoCount: Int, + ptoMillis: Long, + ) { + emit( + "recovery:loss_timer_updated", + mapOf( + "event_type" to "expired", + "timer_type" to "pto", + "pto_count" to consecutivePtoCount, + "delta" to ptoMillis, + ), + ) + } + + override fun onCongestionStateUpdated(newState: String) { + emit( + "recovery:congestion_state_updated", + mapOf("new" to newState), + ) + } + + override fun onTransportParametersSet( + initiator: String, + params: Map, + ) { + emit( + "transport:parameters_set", + mapOf( + "owner" to initiator, + "params" to params, + ), + ) + } + + override fun onAlpnNegotiated(alpn: String) { + emit( + "transport:alpn_information", + mapOf("chosen_alpn" to alpn), + ) + } + + override fun onVersionInformation( + chosenVersion: String, + otherVersionsOffered: List, + ) { + emit( + "transport:version_information", + mapOf( + "chosen_version" to chosenVersion, + "client_versions" to otherVersionsOffered, + ), + ) + } + + override fun close() { + lock.withLock { + try { + writer.flush() + } finally { + writer.close() + } + } + } + + private fun emit( + name: String, + data: Map, + ) { + val event = + linkedMapOf( + "time" to (nowMillis() - startMillis), + "name" to name, + "data" to data, + ) + // Serialize OUTSIDE the lock so concurrent emitters don't + // serialize their JSON serially. The lock is only held while + // appending the line to the file. + val line = mapper.writeValueAsString(event) + writeLineLocked(line) + } + + private fun writeLineLocked(line: String) { + lock.withLock { + writer.write(line) + writer.write("\n") + // Flush every line so a hard-killed process still leaves a + // partial-but-parseable trace. + writer.flush() + } + } + + companion object { + private val DEFAULT_MAPPER: ObjectMapper = jacksonObjectMapper() + + private fun packetTypeFor(level: EncryptionLevel): String = + when (level) { + EncryptionLevel.INITIAL -> "initial" + EncryptionLevel.HANDSHAKE -> "handshake" + EncryptionLevel.APPLICATION -> "1RTT" + } + + private val HEX_CHARS = "0123456789abcdef".toCharArray() + + fun hex(bytes: ByteArray): String { + val sb = StringBuilder(bytes.size * 2) + for (b in bytes) { + val v = b.toInt() and 0xFF + sb.append(HEX_CHARS[v ushr 4]) + sb.append(HEX_CHARS[v and 0x0F]) + } + return sb.toString() + } + } +} diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriterTest.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriterTest.kt new file mode 100644 index 000000000..72f5d1384 --- /dev/null +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/interop/QlogWriterTest.kt @@ -0,0 +1,161 @@ +/* + * 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.interop + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.quic.connection.EncryptionLevel +import org.junit.Test +import java.io.File +import java.nio.file.Files +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertTrue + +/** + * Validates the JSON-NDJSON shape qvis (https://qvis.quictools.info/) + * expects: header line + one event per line, every line independently + * parseable as JSON. + * + * Drives [QlogWriter] through one event of each type to confirm we + * don't emit anything that breaks the format. + */ +class QlogWriterTest { + @Test + fun headerThenOneEventPerLine_allParseable() { + val tmp = Files.createTempFile("amethyst-qlog-test", ".sqlog").toFile() + tmp.deleteOnExit() + var clock = 0L + QlogWriter(tmp, odcidHex = "deadbeef", nowMillis = { clock }).use { w -> + clock = 5L + w.onConnectionStarted("example.test", byteArrayOf(1, 2, 3), byteArrayOf(4, 5, 6)) + clock = 7L + w.onTransportParametersSet("local", mapOf("initial_max_data" to "1000000")) + clock = 10L + w.onPacketSent(EncryptionLevel.INITIAL, packetNumber = 0, sizeBytes = 1200, frames = listOf("crypto")) + clock = 15L + w.onPacketReceived(EncryptionLevel.INITIAL, packetNumber = 0, sizeBytes = 800, frames = listOf("crypto", "ack")) + clock = 20L + w.onPacketDropped("AEAD auth failed", sizeBytes = 80) + clock = 25L + w.onKeyUpdated("server", EncryptionLevel.HANDSHAKE) + clock = 30L + w.onLossDetected(EncryptionLevel.APPLICATION, lostPacketNumbers = listOf(3L, 5L)) + clock = 35L + w.onPtoFired(consecutivePtoCount = 1, ptoMillis = 333) + clock = 40L + w.onCongestionStateUpdated("recovery") + clock = 45L + w.onAlpnNegotiated("h3") + clock = 50L + w.onVersionInformation("v1", emptyList()) + clock = 55L + w.onConnectionClosed("local", errorCode = 0, reason = "done") + } + + val lines = tmp.readLines().filter { it.isNotBlank() } + assertTrue(lines.size >= 12, "expected >= 12 lines (header + at least 11 events) but got ${lines.size}") + + val mapper = jacksonObjectMapper() + + // Line 1: qlog header. + val header = mapper.readTree(lines[0]) + assertEquals("0.3", header.get("qlog_version").asText(), "qlog_version must be 0.3") + assertEquals("JSON-SEQ", header.get("qlog_format").asText(), "qlog_format must be JSON-SEQ") + val vp = header.get("trace").get("vantage_point") + assertEquals("client", vp.get("type").asText()) + assertEquals( + "deadbeef", + header + .get("trace") + .get("common_fields") + .get("ODCID") + .asText(), + ) + + // Lines 2..N: event objects with `time`, `name`, `data`. + for (i in 1 until lines.size) { + val node = mapper.readTree(lines[i]) + assertNotNull(node.get("time"), "line $i missing 'time': ${lines[i]}") + val name = node.get("name") + assertNotNull(name, "line $i missing 'name': ${lines[i]}") + assertTrue( + name.asText().contains(":"), + "name '${name.asText()}' must be in ':' form", + ) + assertNotNull(node.get("data"), "line $i missing 'data': ${lines[i]}") + } + + // Spot-check specific events made it through. + val names = lines.drop(1).map { mapper.readTree(it).get("name").asText() } + assertTrue(names.contains("transport:connection_started"), names.toString()) + assertTrue(names.contains("transport:packet_sent"), names.toString()) + assertTrue(names.contains("transport:packet_received"), names.toString()) + assertTrue(names.contains("transport:packet_dropped"), names.toString()) + assertTrue(names.contains("security:key_updated"), names.toString()) + assertTrue(names.contains("recovery:packet_lost"), names.toString()) + assertTrue(names.contains("recovery:loss_timer_updated"), names.toString()) + assertTrue(names.contains("transport:parameters_set"), names.toString()) + assertTrue(names.contains("transport:alpn_information"), names.toString()) + assertTrue(names.contains("transport:version_information"), names.toString()) + assertTrue(names.contains("transport:connection_closed"), names.toString()) + } + + @Test + fun timesAreRelativeToConstructorTime() { + val tmp = Files.createTempFile("amethyst-qlog-rel", ".sqlog").toFile() + tmp.deleteOnExit() + var clock = 1_000L + QlogWriter(tmp, odcidHex = "00", nowMillis = { clock }).use { w -> + clock = 1_050L + w.onAlpnNegotiated("h3") + } + val lines = tmp.readLines().filter { it.isNotBlank() } + val mapper = jacksonObjectMapper() + val event = mapper.readTree(lines[1]) + assertEquals(50L, event.get("time").asLong(), "time must be relative to constructor (1050 - 1000)") + } + + @Test + fun fileEndsWithNewline_qvisCompatible() { + val tmp = Files.createTempFile("amethyst-qlog-nl", ".sqlog").toFile() + tmp.deleteOnExit() + QlogWriter(tmp, odcidHex = "00").use { w -> + w.onAlpnNegotiated("h3") + } + val bytes = tmp.readBytes() + assertTrue(bytes.isNotEmpty(), "file must not be empty") + assertEquals('\n'.code.toByte(), bytes.last(), "file must end with '\\n' so trailing event parses") + } + + @Test + fun handlesEmptyFramesList(): Unit = + File.createTempFile("amethyst-qlog-empty", ".sqlog").let { tmp -> + tmp.deleteOnExit() + QlogWriter(tmp, odcidHex = "00").use { w -> + w.onPacketSent(EncryptionLevel.INITIAL, 0, 1200, emptyList()) + } + val mapper = jacksonObjectMapper() + val lines = tmp.readLines().filter { it.isNotBlank() } + val frames = mapper.readTree(lines[1]).get("data").get("frames") + assertTrue(frames.isArray, "frames must be an array even when empty") + assertEquals(0, frames.size()) + } +}