feat(quic): add qlog observer infrastructure for interop diagnostics
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 `<QLOGDIR>/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)
This commit is contained in:
@@ -24,6 +24,7 @@ import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection
|
|||||||
import com.vitorpamplona.quic.crypto.InitialSecrets
|
import com.vitorpamplona.quic.crypto.InitialSecrets
|
||||||
import com.vitorpamplona.quic.crypto.PlatformAesOneBlock
|
import com.vitorpamplona.quic.crypto.PlatformAesOneBlock
|
||||||
import com.vitorpamplona.quic.crypto.bestAes128GcmAead
|
import com.vitorpamplona.quic.crypto.bestAes128GcmAead
|
||||||
|
import com.vitorpamplona.quic.observability.QlogObserver
|
||||||
import com.vitorpamplona.quic.stream.QuicStream
|
import com.vitorpamplona.quic.stream.QuicStream
|
||||||
import com.vitorpamplona.quic.stream.StreamId
|
import com.vitorpamplona.quic.stream.StreamId
|
||||||
import com.vitorpamplona.quic.tls.TlsClient
|
import com.vitorpamplona.quic.tls.TlsClient
|
||||||
@@ -73,6 +74,13 @@ class QuicConnection(
|
|||||||
.toEpochMilliseconds()
|
.toEpochMilliseconds()
|
||||||
},
|
},
|
||||||
val alpnList: List<ByteArray> = listOf(TlsConstants.ALPN_H3),
|
val alpnList: List<ByteArray> = 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)
|
val sourceConnectionId: ConnectionId = ConnectionId.random(8)
|
||||||
var destinationConnectionId: ConnectionId = ConnectionId.random(8)
|
var destinationConnectionId: ConnectionId = ConnectionId.random(8)
|
||||||
@@ -317,6 +325,8 @@ class QuicConnection(
|
|||||||
) {
|
) {
|
||||||
handshake.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret)
|
handshake.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret)
|
||||||
handshake.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret)
|
handshake.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret)
|
||||||
|
qlogObserver.onKeyUpdated("client", EncryptionLevel.HANDSHAKE)
|
||||||
|
qlogObserver.onKeyUpdated("server", EncryptionLevel.HANDSHAKE)
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun onApplicationKeysReady(
|
override fun onApplicationKeysReady(
|
||||||
@@ -326,12 +336,15 @@ class QuicConnection(
|
|||||||
) {
|
) {
|
||||||
application.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret)
|
application.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret)
|
||||||
application.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret)
|
application.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret)
|
||||||
|
qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION)
|
||||||
|
qlogObserver.onKeyUpdated("server", EncryptionLevel.APPLICATION)
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun onHandshakeComplete() {
|
override fun onHandshakeComplete() {
|
||||||
handshakeComplete = true
|
handshakeComplete = true
|
||||||
if (status == Status.HANDSHAKING) status = Status.CONNECTED
|
if (status == Status.HANDSHAKING) status = Status.CONNECTED
|
||||||
applyPeerTransportParameters()
|
applyPeerTransportParameters()
|
||||||
|
tls.negotiatedAlpn?.let { qlogObserver.onAlpnNegotiated(it.decodeToString()) }
|
||||||
handshakeDoneSignal.complete(Unit)
|
handshakeDoneSignal.complete(Unit)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -375,11 +388,52 @@ class QuicConnection(
|
|||||||
|
|
||||||
/** Begin the handshake — emits ClientHello into Initial CRYPTO. */
|
/** Begin the handshake — emits ClientHello into Initial CRYPTO. */
|
||||||
fun start() {
|
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()
|
tls.start()
|
||||||
// Drain ClientHello bytes into the Initial-level CRYPTO send buffer.
|
// Drain ClientHello bytes into the Initial-level CRYPTO send buffer.
|
||||||
tls.pollOutbound(TlsClient.Level.INITIAL)?.let { initial.cryptoSend.enqueue(it) }
|
tls.pollOutbound(TlsClient.Level.INITIAL)?.let { initial.cryptoSend.enqueue(it) }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun localTransportParametersSummary(): Map<String, String> {
|
||||||
|
val out = LinkedHashMap<String, String>(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<String, String> {
|
||||||
|
val out = LinkedHashMap<String, String>(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 =
|
private fun buildLocalTransportParameters(): TransportParameters =
|
||||||
TransportParameters(
|
TransportParameters(
|
||||||
initialMaxData = config.initialMaxData,
|
initialMaxData = config.initialMaxData,
|
||||||
@@ -425,6 +479,7 @@ class QuicConnection(
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
peerTransportParameters = tp
|
peerTransportParameters = tp
|
||||||
|
qlogObserver.onTransportParametersSet("remote", peerTransportParametersSummary(tp))
|
||||||
sendConnectionFlowCredit = tp.initialMaxData ?: 0L
|
sendConnectionFlowCredit = tp.initialMaxData ?: 0L
|
||||||
peerMaxStreamsBidi = tp.initialMaxStreamsBidi ?: 0L
|
peerMaxStreamsBidi = tp.initialMaxStreamsBidi ?: 0L
|
||||||
peerMaxStreamsUni = tp.initialMaxStreamsUni ?: 0L
|
peerMaxStreamsUni = tp.initialMaxStreamsUni ?: 0L
|
||||||
@@ -631,12 +686,15 @@ class QuicConnection(
|
|||||||
errorCode: Long,
|
errorCode: Long,
|
||||||
reason: String,
|
reason: String,
|
||||||
) {
|
) {
|
||||||
|
var firedQlog = false
|
||||||
lock.withLock {
|
lock.withLock {
|
||||||
if (status == Status.CLOSED || status == Status.CLOSING) return@withLock
|
if (status == Status.CLOSED || status == Status.CLOSING) return@withLock
|
||||||
closeErrorCode = errorCode
|
closeErrorCode = errorCode
|
||||||
closeReason = reason
|
closeReason = reason
|
||||||
status = Status.CLOSING
|
status = Status.CLOSING
|
||||||
|
firedQlog = true
|
||||||
}
|
}
|
||||||
|
if (firedQlog) qlogObserver.onConnectionClosed("local", errorCode, reason)
|
||||||
// If a caller is suspended on awaitHandshake() and we're tearing down
|
// If a caller is suspended on awaitHandshake() and we're tearing down
|
||||||
// before completion, fail the deferred so the caller throws instead
|
// before completion, fail the deferred so the caller throws instead
|
||||||
// of hanging forever.
|
// 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. */
|
/** Called by the parser on inbound CONNECTION_CLOSE or by the driver on read-loop death. */
|
||||||
internal fun markClosedExternally(reason: String) {
|
internal fun markClosedExternally(reason: String) {
|
||||||
|
val wasClosed = status == Status.CLOSED
|
||||||
if (status != Status.CLOSED) 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) {
|
if (!handshakeComplete) {
|
||||||
signalHandshakeFailed(QuicConnectionClosedException("connection closed externally: $reason"))
|
signalHandshakeFailed(QuicConnectionClosedException("connection closed externally: $reason"))
|
||||||
}
|
}
|
||||||
|
|||||||
+9
-5
@@ -149,11 +149,15 @@ class QuicConnectionDriver(
|
|||||||
// PING on the next drain (RFC 9002 §6.2.4 probe
|
// PING on the next drain (RFC 9002 §6.2.4 probe
|
||||||
// packet). The peer's ACK feeds loss detection +
|
// packet). The peer's ACK feeds loss detection +
|
||||||
// retransmit (steps 5–6).
|
// retransmit (steps 5–6).
|
||||||
connection.lock.withLock {
|
val newPtoCount =
|
||||||
connection.pendingPing = true
|
connection.lock
|
||||||
connection.consecutivePtoCount =
|
.withLock {
|
||||||
(connection.consecutivePtoCount + 1).coerceAtMost(6)
|
connection.pendingPing = true
|
||||||
}
|
connection.consecutivePtoCount =
|
||||||
|
(connection.consecutivePtoCount + 1).coerceAtMost(6)
|
||||||
|
connection.consecutivePtoCount
|
||||||
|
}
|
||||||
|
connection.qlogObserver.onPtoFired(newPtoCount, ptoMillis)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+85
-7
@@ -37,6 +37,7 @@ import com.vitorpamplona.quic.frame.ResetStreamFrame
|
|||||||
import com.vitorpamplona.quic.frame.StopSendingFrame
|
import com.vitorpamplona.quic.frame.StopSendingFrame
|
||||||
import com.vitorpamplona.quic.frame.StreamFrame
|
import com.vitorpamplona.quic.frame.StreamFrame
|
||||||
import com.vitorpamplona.quic.frame.decodeFrames
|
import com.vitorpamplona.quic.frame.decodeFrames
|
||||||
|
import com.vitorpamplona.quic.observability.qlogFrameName
|
||||||
import com.vitorpamplona.quic.packet.LongHeaderPacket
|
import com.vitorpamplona.quic.packet.LongHeaderPacket
|
||||||
import com.vitorpamplona.quic.packet.LongHeaderType
|
import com.vitorpamplona.quic.packet.LongHeaderType
|
||||||
import com.vitorpamplona.quic.packet.ShortHeaderPacket
|
import com.vitorpamplona.quic.packet.ShortHeaderPacket
|
||||||
@@ -87,12 +88,33 @@ private fun feedLongHeaderPacket(
|
|||||||
val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null
|
val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null
|
||||||
val level =
|
val level =
|
||||||
when (peeked.type) {
|
when (peeked.type) {
|
||||||
LongHeaderType.INITIAL -> EncryptionLevel.INITIAL
|
LongHeaderType.INITIAL -> {
|
||||||
LongHeaderType.HANDSHAKE -> EncryptionLevel.HANDSHAKE
|
EncryptionLevel.INITIAL
|
||||||
LongHeaderType.ZERO_RTT, LongHeaderType.RETRY -> return null // unsupported in client
|
}
|
||||||
|
|
||||||
|
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 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 =
|
val parsed =
|
||||||
LongHeaderPacket.parseAndDecrypt(
|
LongHeaderPacket.parseAndDecrypt(
|
||||||
bytes = datagram,
|
bytes = datagram,
|
||||||
@@ -103,7 +125,14 @@ private fun feedLongHeaderPacket(
|
|||||||
hp = proto.hp,
|
hp = proto.hp,
|
||||||
hpKey = proto.hpKey,
|
hpKey = proto.hpKey,
|
||||||
largestReceivedInSpace = state.pnSpace.largestReceived,
|
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)
|
state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis)
|
||||||
|
|
||||||
@@ -112,6 +141,14 @@ private fun feedLongHeaderPacket(
|
|||||||
conn.destinationConnectionId = parsed.packet.scid
|
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)
|
dispatchFrames(conn, level, parsed.packet.payload, parsed.packet.packetNumber, nowMillis)
|
||||||
return parsed.consumed
|
return parsed.consumed
|
||||||
}
|
}
|
||||||
@@ -123,7 +160,14 @@ private fun feedShortHeaderPacket(
|
|||||||
nowMillis: Long,
|
nowMillis: Long,
|
||||||
) {
|
) {
|
||||||
val state = conn.levelState(EncryptionLevel.APPLICATION)
|
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 =
|
val parsed =
|
||||||
ShortHeaderPacket.parseAndDecrypt(
|
ShortHeaderPacket.parseAndDecrypt(
|
||||||
bytes = datagram,
|
bytes = datagram,
|
||||||
@@ -135,11 +179,42 @@ private fun feedShortHeaderPacket(
|
|||||||
hp = proto.hp,
|
hp = proto.hp,
|
||||||
hpKey = proto.hpKey,
|
hpKey = proto.hpKey,
|
||||||
largestReceivedInSpace = state.pnSpace.largestReceived,
|
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)
|
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)
|
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<String> =
|
||||||
|
try {
|
||||||
|
com.vitorpamplona.quic.frame
|
||||||
|
.decodeFrames(payload)
|
||||||
|
.map { qlogFrameName(it::class.simpleName ?: "frame") }
|
||||||
|
} catch (_: QuicCodecException) {
|
||||||
|
emptyList()
|
||||||
|
}
|
||||||
|
|
||||||
private fun dispatchFrames(
|
private fun dispatchFrames(
|
||||||
conn: QuicConnection,
|
conn: QuicConnection,
|
||||||
level: EncryptionLevel,
|
level: EncryptionLevel,
|
||||||
@@ -225,6 +300,9 @@ private fun dispatchFrames(
|
|||||||
for (lostPacket in lost) {
|
for (lostPacket in lost) {
|
||||||
conn.onTokensLost(lostPacket.tokens)
|
conn.onTokensLost(lostPacket.tokens)
|
||||||
}
|
}
|
||||||
|
if (lost.isNotEmpty()) {
|
||||||
|
conn.qlogObserver.onLossDetected(level, lost.map { it.packetNumber })
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+64
-27
@@ -37,6 +37,8 @@ import com.vitorpamplona.quic.frame.ResetStreamFrame
|
|||||||
import com.vitorpamplona.quic.frame.StopSendingFrame
|
import com.vitorpamplona.quic.frame.StopSendingFrame
|
||||||
import com.vitorpamplona.quic.frame.StreamFrame
|
import com.vitorpamplona.quic.frame.StreamFrame
|
||||||
import com.vitorpamplona.quic.frame.encodeFrames
|
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.LongHeaderPacket
|
||||||
import com.vitorpamplona.quic.packet.LongHeaderPlaintextPacket
|
import com.vitorpamplona.quic.packet.LongHeaderPlaintextPacket
|
||||||
import com.vitorpamplona.quic.packet.LongHeaderType
|
import com.vitorpamplona.quic.packet.LongHeaderType
|
||||||
@@ -176,23 +178,26 @@ private fun buildBestLevelPacket(
|
|||||||
if (app.sendProtection != null) {
|
if (app.sendProtection != null) {
|
||||||
val proto = app.sendProtection!!
|
val proto = app.sendProtection!!
|
||||||
val pn = app.pnSpace.allocateOutbound()
|
val pn = app.pnSpace.allocateOutbound()
|
||||||
return ShortHeaderPacket.build(
|
val built =
|
||||||
ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload),
|
ShortHeaderPacket.build(
|
||||||
proto.aead,
|
ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload),
|
||||||
proto.key,
|
proto.aead,
|
||||||
proto.iv,
|
proto.key,
|
||||||
proto.hp,
|
proto.iv,
|
||||||
proto.hpKey,
|
proto.hp,
|
||||||
largestAckedInSpace = -1L,
|
proto.hpKey,
|
||||||
)
|
largestAckedInSpace = -1L,
|
||||||
|
)
|
||||||
|
emitQlogSent(conn, EncryptionLevel.APPLICATION, pn, built.size, frames)
|
||||||
|
return built
|
||||||
}
|
}
|
||||||
val hs = conn.handshake
|
val hs = conn.handshake
|
||||||
if (hs.sendProtection != null) {
|
if (hs.sendProtection != null) {
|
||||||
return buildLongHeaderPacket(conn, EncryptionLevel.HANDSHAKE, payload)
|
return buildLongHeaderPacket(conn, EncryptionLevel.HANDSHAKE, payload, frames)
|
||||||
}
|
}
|
||||||
val init = conn.initial
|
val init = conn.initial
|
||||||
if (init.sendProtection != null) {
|
if (init.sendProtection != null) {
|
||||||
return buildLongHeaderPacket(conn, EncryptionLevel.INITIAL, payload)
|
return buildLongHeaderPacket(conn, EncryptionLevel.INITIAL, payload, frames)
|
||||||
}
|
}
|
||||||
return null
|
return null
|
||||||
}
|
}
|
||||||
@@ -201,6 +206,7 @@ private fun buildLongHeaderPacket(
|
|||||||
conn: QuicConnection,
|
conn: QuicConnection,
|
||||||
level: EncryptionLevel,
|
level: EncryptionLevel,
|
||||||
payload: ByteArray,
|
payload: ByteArray,
|
||||||
|
frames: List<Frame>,
|
||||||
): ByteArray {
|
): ByteArray {
|
||||||
val state = conn.levelState(level)
|
val state = conn.levelState(level)
|
||||||
val proto = state.sendProtection!!
|
val proto = state.sendProtection!!
|
||||||
@@ -211,22 +217,25 @@ private fun buildLongHeaderPacket(
|
|||||||
EncryptionLevel.HANDSHAKE -> LongHeaderType.HANDSHAKE
|
EncryptionLevel.HANDSHAKE -> LongHeaderType.HANDSHAKE
|
||||||
EncryptionLevel.APPLICATION -> error("APPLICATION uses short-header packets")
|
EncryptionLevel.APPLICATION -> error("APPLICATION uses short-header packets")
|
||||||
}
|
}
|
||||||
return LongHeaderPacket.build(
|
val built =
|
||||||
LongHeaderPlaintextPacket(
|
LongHeaderPacket.build(
|
||||||
type = type,
|
LongHeaderPlaintextPacket(
|
||||||
version = QuicVersion.V1,
|
type = type,
|
||||||
dcid = conn.destinationConnectionId,
|
version = QuicVersion.V1,
|
||||||
scid = conn.sourceConnectionId,
|
dcid = conn.destinationConnectionId,
|
||||||
packetNumber = pn,
|
scid = conn.sourceConnectionId,
|
||||||
payload = payload,
|
packetNumber = pn,
|
||||||
),
|
payload = payload,
|
||||||
proto.aead,
|
),
|
||||||
proto.key,
|
proto.aead,
|
||||||
proto.iv,
|
proto.key,
|
||||||
proto.hp,
|
proto.iv,
|
||||||
proto.hpKey,
|
proto.hp,
|
||||||
largestAckedInSpace = -1L,
|
proto.hpKey,
|
||||||
)
|
largestAckedInSpace = -1L,
|
||||||
|
)
|
||||||
|
emitQlogSent(conn, level, pn, built.size, frames)
|
||||||
|
return built
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -345,9 +354,34 @@ private fun buildLongHeaderFromFrames(
|
|||||||
sizeBytes = packet.size,
|
sizeBytes = packet.size,
|
||||||
tokens = tokens,
|
tokens = tokens,
|
||||||
)
|
)
|
||||||
|
emitQlogSent(conn, level, pn, packet.size, frames)
|
||||||
return packet
|
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<com.vitorpamplona.quic.frame.Frame>,
|
||||||
|
) {
|
||||||
|
val observer = conn.qlogObserver
|
||||||
|
if (observer === QlogObserver.NoOp) return
|
||||||
|
val frameNames = ArrayList<String>(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(
|
private fun buildApplicationPacket(
|
||||||
conn: QuicConnection,
|
conn: QuicConnection,
|
||||||
nowMillis: Long,
|
nowMillis: Long,
|
||||||
@@ -510,6 +544,9 @@ private fun buildApplicationPacket(
|
|||||||
sizeBytes = sizeBytes.getOrNull()?.size ?: 0,
|
sizeBytes = sizeBytes.getOrNull()?.size ?: 0,
|
||||||
tokens = tokens.toList(),
|
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
|
// Re-throw on encrypt failure so callers (driver loop) see the
|
||||||
// same exception they did before this change. The bookkeeping
|
// same exception they did before this change. The bookkeeping
|
||||||
// entry is still in place; if the throw was transient, retransmit
|
// entry is still in place; if the throw was transient, retransmit
|
||||||
|
|||||||
@@ -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<String>,
|
||||||
|
)
|
||||||
|
|
||||||
|
/** One inbound packet at [level] was successfully decrypted + dispatched. */
|
||||||
|
fun onPacketReceived(
|
||||||
|
level: EncryptionLevel,
|
||||||
|
packetNumber: Long,
|
||||||
|
sizeBytes: Int,
|
||||||
|
frames: List<String>,
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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<Long>,
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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<String, String>,
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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<String>,
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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<String>,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onPacketReceived(
|
||||||
|
level: EncryptionLevel,
|
||||||
|
packetNumber: Long,
|
||||||
|
sizeBytes: Int,
|
||||||
|
frames: List<String>,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onPacketDropped(
|
||||||
|
reason: String,
|
||||||
|
sizeBytes: Int,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onKeyUpdated(
|
||||||
|
keyType: String,
|
||||||
|
level: EncryptionLevel,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onLossDetected(
|
||||||
|
level: EncryptionLevel,
|
||||||
|
lostPacketNumbers: List<Long>,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onPtoFired(
|
||||||
|
consecutivePtoCount: Int,
|
||||||
|
ptoMillis: Long,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onCongestionStateUpdated(newState: String) = Unit
|
||||||
|
|
||||||
|
override fun onTransportParametersSet(
|
||||||
|
initiator: String,
|
||||||
|
params: Map<String, String>,
|
||||||
|
) = Unit
|
||||||
|
|
||||||
|
override fun onAlpnNegotiated(alpn: String) = Unit
|
||||||
|
|
||||||
|
override fun onVersionInformation(
|
||||||
|
chosenVersion: String,
|
||||||
|
otherVersionsOffered: List<String>,
|
||||||
|
) = 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()
|
||||||
|
}
|
||||||
@@ -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<String, Any?>,
|
||||||
|
)
|
||||||
|
|
||||||
|
val events: MutableList<Event> = mutableListOf()
|
||||||
|
|
||||||
|
private fun add(
|
||||||
|
name: String,
|
||||||
|
payload: Map<String, Any?>,
|
||||||
|
) {
|
||||||
|
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<String>,
|
||||||
|
) = 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<String>,
|
||||||
|
) = 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<Long>,
|
||||||
|
) = 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<String, String>,
|
||||||
|
) = 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<String>,
|
||||||
|
) = add(
|
||||||
|
"versionInformation",
|
||||||
|
mapOf("chosen" to chosenVersion, "offered" to otherVersionsOffered),
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -23,6 +23,7 @@ package com.vitorpamplona.quic.interop
|
|||||||
import com.vitorpamplona.quic.connection.QuicConnection
|
import com.vitorpamplona.quic.connection.QuicConnection
|
||||||
import com.vitorpamplona.quic.connection.QuicConnectionConfig
|
import com.vitorpamplona.quic.connection.QuicConnectionConfig
|
||||||
import com.vitorpamplona.quic.connection.QuicConnectionDriver
|
import com.vitorpamplona.quic.connection.QuicConnectionDriver
|
||||||
|
import com.vitorpamplona.quic.observability.QlogObserver
|
||||||
import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
|
import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
|
||||||
import com.vitorpamplona.quic.transport.UdpSocket
|
import com.vitorpamplona.quic.transport.UdpSocket
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
@@ -31,6 +32,7 @@ import kotlinx.coroutines.SupervisorJob
|
|||||||
import kotlinx.coroutines.cancel
|
import kotlinx.coroutines.cancel
|
||||||
import kotlinx.coroutines.runBlocking
|
import kotlinx.coroutines.runBlocking
|
||||||
import kotlinx.coroutines.withTimeoutOrNull
|
import kotlinx.coroutines.withTimeoutOrNull
|
||||||
|
import java.io.File
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Standalone interop runner. Drives [QuicConnection] against a real QUIC
|
* Standalone interop runner. Drives [QuicConnection] against a real QUIC
|
||||||
@@ -58,6 +60,20 @@ fun main(args: Array<String>) {
|
|||||||
println("timeout: ${timeoutSec}s")
|
println("timeout: ${timeoutSec}s")
|
||||||
println()
|
println()
|
||||||
|
|
||||||
|
// qlog: if QLOGDIR is set, drop a `client.sqlog` file at
|
||||||
|
// <QLOGDIR>/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 scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
|
||||||
val outcome =
|
val outcome =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
@@ -73,6 +89,7 @@ fun main(args: Array<String>) {
|
|||||||
serverName = host,
|
serverName = host,
|
||||||
config = QuicConnectionConfig(),
|
config = QuicConnectionConfig(),
|
||||||
tlsCertificateValidator = PermissiveCertificateValidator(),
|
tlsCertificateValidator = PermissiveCertificateValidator(),
|
||||||
|
qlogObserver = qlogObserver,
|
||||||
)
|
)
|
||||||
val driver = QuicConnectionDriver(conn, socket, scope)
|
val driver = QuicConnectionDriver(conn, socket, scope)
|
||||||
driver.start()
|
driver.start()
|
||||||
@@ -110,6 +127,10 @@ fun main(args: Array<String>) {
|
|||||||
// are torn down before main() exits. Without this the JVM hangs on stray
|
// are torn down before main() exits. Without this the JVM hangs on stray
|
||||||
// IO-dispatcher threads.
|
// IO-dispatcher threads.
|
||||||
scope.cancel()
|
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) {
|
when (outcome) {
|
||||||
is InteropOutcome.Connected -> {
|
is InteropOutcome.Connected -> {
|
||||||
|
|||||||
@@ -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<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
|
||||||
|
|
||||||
|
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())
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user