feat(quic-interop): wire qlog observer into the production endpoint

Agent B's QlogWriter previously lived only in :quic's jvmTest scope, hooked into the standalone InteropRunner. Brings it into :quic-interop proper: - QlogWriter.kt + QlogWriterTest.kt copied into :quic-interop with package com.vitorpamplona.quic.interop.runner. - Jackson dep added to :quic-interop's build.gradle.kts. - InteropClient reads $QLOGDIR (the runner sets it inside the container per docker-compose.yml). When set, opens a QlogWriter at <QLOGDIR>/client.sqlog, passes as QuicConnection.qlogObserver, closes on every exit path (handshake_failed / udp_failed / transfer_timeout / ok). - QUIC_INTEROP_DEBUG=1 now prints qlogdir alongside the other env fields. Net effect: any future runner-driven test failure dumps a structured qlog file the user can drag straight into qvis (qvis.quictools.info) to see frame-by-frame what the client decided to do. The retry test failure on aioquic + picoquic that we still need to debug now produces a usable artifact. NOTE: the QlogWriter copy in :quic's jvmTest stays in place as the helper for the standalone InteropRunner main(). Slight duplication; acceptable while :quic-interop is its own module. https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT
2026-05-07 00:35:11 +00:00
parent 7d5fcd710f
commit 5a582ec3f5
3 changed files with 496 additions and 0 deletions
@@ -71,6 +71,7 @@ fun main() {
    // does NOT export a DOWNLOADS env var. Hard-code the mount path.
    val downloadsDir = File("/downloads")
    val keyLogPath = System.getenv("SSLKEYLOGFILE")?.takeIf { it.isNotBlank() }
+    val qlogDir = System.getenv("QLOGDIR")?.takeIf { it.isNotBlank() }?.let { File(it) }

    // One-line context header. Verbose per-field dump deferred to debug
    // mode (env var QUIC_INTEROP_DEBUG=1) so the runner's aggregated output
@@ -81,6 +82,7 @@ fun main() {
        System.err.println("requests:       $requests")
        System.err.println("downloads dir:  ${downloadsDir.absolutePath} (exists=${downloadsDir.isDirectory})")
        System.err.println("sslkeylogfile:  ${keyLogPath ?: "(unset)"}")
+        System.err.println("qlogdir:        ${qlogDir?.absolutePath ?: "(unset)"}")
    }

    val cipherSuites =
@@ -155,6 +157,7 @@ fun main() {
                    offeredAlpns = offeredAlpns,
                    initialVersion = initialVersion,
                    keyLogPath = keyLogPath,
+                    qlogDir = qlogDir,
                    parallel = (testcase == "multiplexing"),
                )
            }
@@ -186,6 +189,7 @@ private fun runTransferTest(
    offeredAlpns: List<Alpn>,
    initialVersion: Int,
    keyLogPath: String?,
+    qlogDir: File?,
    parallel: Boolean,
 ): Int {
    val urls =
@@ -215,6 +219,15 @@ private fun runTransferTest(
                    return@runBlocking "udp_failed: ${t.message ?: t::class.simpleName}"
                }
            val keyLogger = keyLogPath?.let { SslKeyLogger(File(it)) }
+            val qlogWriter =
+                qlogDir?.let { dir ->
+                    dir.mkdirs()
+                    // ODCID is unknown until the connection generates one in
+                    // its init block; we'd need to plumb through, but for the
+                    // header it's fine to start with a placeholder and the
+                    // packet-sent events will carry SCID/DCID anyway.
+                    QlogWriter(file = File(dir, "client.sqlog"), odcidHex = "client")
+                }
            val conn =
                QuicConnection(
                    serverName = host,
@@ -229,6 +242,7 @@ private fun runTransferTest(
                                TlsConstants.CIPHER_TLS_CHACHA20_POLY1305_SHA256,
                            ),
                    extraSecretsListener = keyLogger?.listener,
+                    qlogObserver = qlogWriter ?: com.vitorpamplona.quic.observability.QlogObserver.NoOp,
                )
            val driver = QuicConnectionDriver(conn, socket, scope)
            driver.start()
@@ -240,6 +254,7 @@ private fun runTransferTest(
            if (handshake == null || handshake.isFailure) {
                runCatching { driver.close() }
                conn.tls.clientRandom?.let { keyLogger?.flush(it) }
+                runCatching { qlogWriter?.close() }
                return@runBlocking "handshake_failed"
            }

@@ -297,6 +312,7 @@ private fun runTransferTest(

            runCatching { driver.close() }
            conn.tls.clientRandom?.let { keyLogger?.flush(it) }
+            runCatching { qlogWriter?.close() }
            delay(50)
            outcome
        }
@@ -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.runner
+
+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<String>,
+    ) {
+        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<String>,
+    ) {
+        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<Long>,
+    ) {
+        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<String, String>,
+    ) {
+        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<String>,
+    ) {
+        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<String, Any?>,
+    ) {
+        val event =
+            linkedMapOf<String, Any?>(
+                "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()
+        }
+    }
+}
@@ -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.runner
+
+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 '<category>:<event>' 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())
+        }
+}