diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt index 7f3fd956c9..1ba9c1ad32 100644 --- a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt @@ -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, 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 } diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/QlogWriter.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/QlogWriter.kt new file mode 100644 index 0000000000..7f495f965c --- /dev/null +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/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.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, + ) { + 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/interop/src/test/kotlin/com/vitorpamplona/quic/interop/runner/QlogWriterTest.kt b/quic/interop/src/test/kotlin/com/vitorpamplona/quic/interop/runner/QlogWriterTest.kt new file mode 100644 index 0000000000..340be74cca --- /dev/null +++ b/quic/interop/src/test/kotlin/com/vitorpamplona/quic/interop/runner/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.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 ':' 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()) + } +}