engine: split packet loss by direction using the server's observations
"3 % loss" sends an engineer looking in both directions at once. The server records every packet it received per sequence number, so the two cases are distinguishable: sent-but-never-seen is upstream loss, seen-but-no-reply is downstream. The findings say which, and say what is not implicated. Downstream loss is measured against what reached the server, not against what was sent — the other denominator counts every upstream loss twice and overstates the return path. Per-direction jitter comes out of the same records without needing synchronised clocks: (server_rx - client_tx) carries a constant unknown offset, and differencing successive samples cancels it, so RFC 3393 variation is honestly attributable to a direction even though absolute latency is not. Correlation is by wire sequence number, not loop index — the counter is shared with every packet type on the session. ProbeSession exposes it even for a lost probe, since that is precisely the packet whose direction is in question. Live against fmr: 0.08 ms upstream jitter vs 0.85 ms downstream, an asymmetry a round-trip test cannot see. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
3e7e3b8d33
commit
4ffa6e4ae2
@@ -694,3 +694,37 @@ Two process notes from this round:
|
|||||||
token. Caught by deploying and *looking at the response*, not by trusting a green build.
|
token. Caught by deploying and *looking at the response*, not by trusting a green build.
|
||||||
- The live suite is now six tests (`LiveServerTest`, `LiveMeasurement`, `LiveGranted`,
|
- The live suite is now six tests (`LiveServerTest`, `LiveMeasurement`, `LiveGranted`,
|
||||||
`LiveUpload`, `LiveCompat`, `LiveEnrollment`), all green against fmr from the PC with no device.
|
`LiveUpload`, `LiveCompat`, `LiveEnrollment`), all green against fmr from the PC with no device.
|
||||||
|
|
||||||
|
### Directional loss: which way is the packet loss? (2026-08-01)
|
||||||
|
A round trip can only report that *something* was lost somewhere, which is the least useful form
|
||||||
|
of the answer — "3 % loss" sends an engineer looking in both directions at once. The server
|
||||||
|
already records every packet it received per sequence number (§6), so the two cases are actually
|
||||||
|
distinguishable, and `train.udp_updown` now reports them separately:
|
||||||
|
|
||||||
|
- sent, never seen by the server → **upstream** loss
|
||||||
|
- seen by the server, reply never arrived → **downstream** loss
|
||||||
|
|
||||||
|
Findings name the direction and say what is *not* implicated, which is half the value:
|
||||||
|
`connectivity.loss_upstream` ("the return path is not implicated: replies came back for everything
|
||||||
|
that arrived"), `connectivity.loss_downstream`, `nat.udp_unreachable_upstream`.
|
||||||
|
|
||||||
|
Two things the implementation gets deliberately right:
|
||||||
|
- **Downstream loss is measured against what reached the server**, not against what was sent.
|
||||||
|
Using "sent" as the denominator counts every upstream loss a second time and overstates the
|
||||||
|
return path. Pinned by a test with loss in both directions at once.
|
||||||
|
- **Per-direction jitter without synchronised clocks.** Absolute one-way delay would need clock
|
||||||
|
sync and we deliberately have none (the two-clock rule). But `server_rx − client_tx` carries a
|
||||||
|
constant unknown offset, and differencing successive samples cancels it — so RFC 3393 one-way
|
||||||
|
delay variation *is* honestly attributable to a direction even though latency is not. A test
|
||||||
|
pins that a 10-second clock offset changes nothing.
|
||||||
|
|
||||||
|
Correlation is by **wire sequence number**, which is not the loop index: the counter is shared
|
||||||
|
with every other packet type on the session, so "the nth echo" is not "sequence n". `ProbeSession`
|
||||||
|
now exposes `lastSeq`, including for a probe that was lost — a lost packet still has a sequence
|
||||||
|
number, and that number is exactly what tells you which way it was lost.
|
||||||
|
|
||||||
|
Live against fmr: 20/20 both ways, and jitter of **0.08 ms upstream vs 0.85 ms downstream** — a
|
||||||
|
tenfold asymmetry that a round-trip measurement cannot see at all.
|
||||||
|
|
||||||
|
10 unit tests on the arithmetic (a wrong denominator here does not crash, it produces a plausible
|
||||||
|
number pointing at the wrong half of the network) plus the live correlation check.
|
||||||
|
|||||||
@@ -0,0 +1,107 @@
|
|||||||
|
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||||
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
|
||||||
|
package app.echo_lot.engine
|
||||||
|
|
||||||
|
import kotlinx.serialization.SerialName
|
||||||
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Splits a round-trip train into its two directions using what the server witnessed.
|
||||||
|
*
|
||||||
|
* A round trip can only report that *something* was lost somewhere. That is the least useful form
|
||||||
|
* of the answer: "3 % loss" sends an engineer looking in both directions at once. The server
|
||||||
|
* records every packet it received, per sequence number (probe-protocol.md §6), so the two cases
|
||||||
|
* are actually distinguishable:
|
||||||
|
*
|
||||||
|
* - sent, never seen by the server → **upstream** loss
|
||||||
|
* - seen by the server, reply never arrived → **downstream** loss
|
||||||
|
*
|
||||||
|
* The same records give one-way delay *variation* per direction. Absolute one-way delay would
|
||||||
|
* need synchronised clocks and we deliberately have none (measurement-schema.md's two-clock rule),
|
||||||
|
* but the variation does not: (server_rx − client_tx) contains an unknown constant clock offset,
|
||||||
|
* and differencing successive samples cancels it. So jitter is honestly attributable to a
|
||||||
|
* direction even though latency is not.
|
||||||
|
*/
|
||||||
|
object Directional {
|
||||||
|
|
||||||
|
/** One probe as the client saw it. [tRxNs] null means no reply came back. */
|
||||||
|
data class Sample(val seq: Int, val tTxNs: Long, val tRxNs: Long?)
|
||||||
|
|
||||||
|
/** One probe as the server saw it: its own receive and transmit stamps, on its own clock. */
|
||||||
|
data class ServerSighting(val seq: Int, val tRxNs: Long, val tTxNs: Long)
|
||||||
|
|
||||||
|
fun analyse(sent: List<Sample>, seen: List<ServerSighting>): DirectionalMetrics {
|
||||||
|
val byServerSeq = seen.associateBy { it.seq }
|
||||||
|
// Only sequences we actually sent count. A server record for a sequence we have no note
|
||||||
|
// of is not evidence about this train — it is a bug or a stray, and silently folding it
|
||||||
|
// in would produce loss percentages above 100 or below zero.
|
||||||
|
val relevant = sent.filter { byServerSeq.containsKey(it.seq) }
|
||||||
|
|
||||||
|
val nSent = sent.size
|
||||||
|
val nSeen = relevant.size
|
||||||
|
val nReplied = sent.count { it.tRxNs != null }
|
||||||
|
|
||||||
|
// A reply can only exist if the request arrived, so downstream loss is measured against
|
||||||
|
// what the server saw, not against what we sent — otherwise upstream loss is counted twice.
|
||||||
|
val lostUp = nSent - nSeen
|
||||||
|
val lostDown = (nSeen - nReplied).coerceAtLeast(0)
|
||||||
|
|
||||||
|
val upDeltas = relevant.sortedBy { it.seq }
|
||||||
|
.map { byServerSeq.getValue(it.seq).tRxNs - it.tTxNs }
|
||||||
|
val downDeltas = sent.filter { it.tRxNs != null && byServerSeq.containsKey(it.seq) }
|
||||||
|
.sortedBy { it.seq }
|
||||||
|
.map { it.tRxNs!! - byServerSeq.getValue(it.seq).tTxNs }
|
||||||
|
|
||||||
|
return DirectionalMetrics(
|
||||||
|
sent = nSent,
|
||||||
|
seenByServer = nSeen,
|
||||||
|
repliesReceived = nReplied,
|
||||||
|
lostUpstream = lostUp,
|
||||||
|
lostDownstream = lostDown,
|
||||||
|
lossUpstreamPct = pct(lostUp, nSent),
|
||||||
|
// Denominator is what reached the server: of the packets that got there, how many
|
||||||
|
// replies came back.
|
||||||
|
lossDownstreamPct = pct(lostDown, nSeen),
|
||||||
|
jitterUpstreamMs = jitterMs(upDeltas),
|
||||||
|
jitterDownstreamMs = jitterMs(downDeltas),
|
||||||
|
/** True when the server saw nothing at all, which is a different fault from loss. */
|
||||||
|
noneReachedServer = nSent > 0 && nSeen == 0,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Mean absolute difference between consecutive one-way samples (RFC 3393 IPDV, averaged).
|
||||||
|
*
|
||||||
|
* Differencing is what makes this legitimate without synchronised clocks: each sample carries
|
||||||
|
* the same unknown offset between the two clocks, and the difference cancels it. Fewer than
|
||||||
|
* two samples yields null rather than zero — "no jitter" and "not enough data to say" are
|
||||||
|
* different claims and only one of them is true here.
|
||||||
|
*/
|
||||||
|
private fun jitterMs(oneWayNs: List<Long>): Double? {
|
||||||
|
if (oneWayNs.size < 2) return null
|
||||||
|
val deltas = oneWayNs.zipWithNext { a, b -> kotlin.math.abs(b - a) }
|
||||||
|
return round2(deltas.average() / 1_000_000.0)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun pct(part: Int, whole: Int): Double =
|
||||||
|
if (whole <= 0) 0.0 else round2(part * 100.0 / whole)
|
||||||
|
|
||||||
|
private fun round2(v: Double) = Math.round(v * 100.0) / 100.0
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Directional metrics for train.udp_updown; recomputable from the columnar evidence. */
|
||||||
|
@Serializable
|
||||||
|
data class DirectionalMetrics(
|
||||||
|
val sent: Int,
|
||||||
|
@SerialName("seen_by_server") val seenByServer: Int,
|
||||||
|
@SerialName("replies_received") val repliesReceived: Int,
|
||||||
|
@SerialName("lost_upstream") val lostUpstream: Int,
|
||||||
|
@SerialName("lost_downstream") val lostDownstream: Int,
|
||||||
|
@SerialName("loss_upstream_pct") val lossUpstreamPct: Double,
|
||||||
|
@SerialName("loss_downstream_pct") val lossDownstreamPct: Double,
|
||||||
|
/** One-way delay variation (RFC 3393), per direction. Null when there were too few samples. */
|
||||||
|
@SerialName("jitter_upstream_ms") val jitterUpstreamMs: Double? = null,
|
||||||
|
@SerialName("jitter_downstream_ms") val jitterDownstreamMs: Double? = null,
|
||||||
|
@SerialName("none_reached_server") val noneReachedServer: Boolean = false,
|
||||||
|
)
|
||||||
@@ -10,8 +10,13 @@ import app.echo_lot.measurement.*
|
|||||||
import app.echo_lot.protocol.ControlClient
|
import app.echo_lot.protocol.ControlClient
|
||||||
import app.echo_lot.protocol.ProbeSession
|
import app.echo_lot.protocol.ProbeSession
|
||||||
import kotlinx.serialization.json.Json
|
import kotlinx.serialization.json.Json
|
||||||
|
import kotlinx.serialization.json.JsonArray
|
||||||
import kotlinx.serialization.json.JsonObject
|
import kotlinx.serialization.json.JsonObject
|
||||||
import kotlinx.serialization.json.encodeToJsonElement
|
import kotlinx.serialization.json.encodeToJsonElement
|
||||||
|
import kotlinx.serialization.json.intOrNull
|
||||||
|
import kotlinx.serialization.json.jsonObject
|
||||||
|
import kotlinx.serialization.json.jsonPrimitive
|
||||||
|
import kotlinx.serialization.json.longOrNull
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Runs the server-facing measurements against one target and assembles a [MeasurementDocument]:
|
* Runs the server-facing measurements against one target and assembles a [MeasurementDocument]:
|
||||||
@@ -71,7 +76,7 @@ class ServerMeasurement(
|
|||||||
// re-primed source is never recorded and every granted send goes to the old, closed port.
|
// re-primed source is never recorded and every granted send goes to the old, closed port.
|
||||||
// Session identity lives on the server; the socket must live as long as it does.
|
// Session identity lives on the server; the socket must live as long as it does.
|
||||||
ProbeSession(cfg.credential, session, cfg.udpHost, cfg.udpPort).use { ps ->
|
ProbeSession(cfg.credential, session, cfg.udpHost, cfg.udpPort).use { ps ->
|
||||||
val (test, findings) = echoTrain(cfg, ps, startMono)
|
val (test, findings) = echoTrain(cfg, ps, startMono, control, session.sessionId)
|
||||||
tests.add(test)
|
tests.add(test)
|
||||||
allFindings.addAll(findings)
|
allFindings.addAll(findings)
|
||||||
|
|
||||||
@@ -105,6 +110,7 @@ class ServerMeasurement(
|
|||||||
|
|
||||||
private fun echoTrain(
|
private fun echoTrain(
|
||||||
cfg: Config, ps: ProbeSession, startMono: Long,
|
cfg: Config, ps: ProbeSession, startMono: Long,
|
||||||
|
control: ControlClient? = null, sessionId: String? = null,
|
||||||
): Pair<Test, List<Finding>> {
|
): Pair<Test, List<Finding>> {
|
||||||
val testId = ids.uuid()
|
val testId = ids.uuid()
|
||||||
val seqs = ArrayList<Int>()
|
val seqs = ArrayList<Int>()
|
||||||
@@ -114,9 +120,15 @@ class ServerMeasurement(
|
|||||||
val rtts = ArrayList<Double>()
|
val rtts = ArrayList<Double>()
|
||||||
val observedPorts = LinkedHashSet<Int>()
|
val observedPorts = LinkedHashSet<Int>()
|
||||||
|
|
||||||
|
// Wire sequence numbers, kept so the server's observations can be correlated packet by
|
||||||
|
// packet. They are not 0..n-1: the counter is shared with every other packet type on the
|
||||||
|
// session, so "the nth echo" is not "sequence n".
|
||||||
|
val wireSeqs = ArrayList<Int>()
|
||||||
|
|
||||||
for (i in 0 until cfg.echoCount) {
|
for (i in 0 until cfg.echoCount) {
|
||||||
val txMono = ids.monoNs() - startMono
|
val txMono = ids.monoNs() - startMono
|
||||||
val r = ps.echo(cfg.echoPaddingBytes)
|
val r = ps.echo(cfg.echoPaddingBytes)
|
||||||
|
wireSeqs.add(ps.lastSeq)
|
||||||
seqs.add(i)
|
seqs.add(i)
|
||||||
tTx.add(txMono)
|
tTx.add(txMono)
|
||||||
sizes.add(Wire_HEADER + cfg.echoPaddingBytes)
|
sizes.add(Wire_HEADER + cfg.echoPaddingBytes)
|
||||||
@@ -129,6 +141,20 @@ class ServerMeasurement(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Ask the server what it actually received. This is what turns "3 % loss somewhere" into
|
||||||
|
// "3 % loss upstream" - the least useful form of the answer into a usable one.
|
||||||
|
val directional: DirectionalMetrics? =
|
||||||
|
if (control != null && sessionId != null) {
|
||||||
|
runCatching {
|
||||||
|
val samples = wireSeqs.indices.map {
|
||||||
|
Directional.Sample(wireSeqs[it], tTx[it] ?: 0L, tRx[it])
|
||||||
|
}
|
||||||
|
Directional.analyse(samples, serverSightings(control, cfg, sessionId))
|
||||||
|
}.getOrNull() // an older server without the endpoint simply yields no split
|
||||||
|
} else {
|
||||||
|
null
|
||||||
|
}
|
||||||
|
|
||||||
val sent = cfg.echoCount
|
val sent = cfg.echoCount
|
||||||
val received = rtts.size
|
val received = rtts.size
|
||||||
val lossPct = if (sent == 0) 0.0 else (sent - received) * 100.0 / sent
|
val lossPct = if (sent == 0) 0.0 else (sent - received) * 100.0 / sent
|
||||||
@@ -138,6 +164,9 @@ class ServerMeasurement(
|
|||||||
epochMonoNs = startMono, seq = seqs, tTxNs = tTx, tRxNs = tRx, sizeBytes = sizes,
|
epochMonoNs = startMono, seq = seqs, tTxNs = tTx, tRxNs = tRx, sizeBytes = sizes,
|
||||||
).toEvidence()
|
).toEvidence()
|
||||||
|
|
||||||
|
val directionalJson = directional?.let {
|
||||||
|
json.encodeToJsonElement(DirectionalMetrics.serializer(), it) as JsonObject
|
||||||
|
}
|
||||||
val metrics: JsonObject = json.encodeToJsonElement(
|
val metrics: JsonObject = json.encodeToJsonElement(
|
||||||
EchoMetrics(
|
EchoMetrics(
|
||||||
sent = sent, received = received, lossPct = round1(lossPct),
|
sent = sent, received = received, lossPct = round1(lossPct),
|
||||||
@@ -147,7 +176,7 @@ class ServerMeasurement(
|
|||||||
observedPorts = observedPorts.toList(),
|
observedPorts = observedPorts.toList(),
|
||||||
natRebindingDetected = natRebinding,
|
natRebindingDetected = natRebinding,
|
||||||
)
|
)
|
||||||
) as JsonObject
|
).let { base -> JsonObject((base as JsonObject) + (directionalJson ?: JsonObject(emptyMap()))) }
|
||||||
|
|
||||||
val status = when {
|
val status = when {
|
||||||
received == 0 -> TestStatus.FAILED
|
received == 0 -> TestStatus.FAILED
|
||||||
@@ -170,6 +199,35 @@ class ServerMeasurement(
|
|||||||
"High UDP loss to the server (${round1(lossPct)}%)",
|
"High UDP loss to the server (${round1(lossPct)}%)",
|
||||||
"A large fraction of ECHO probes were lost, indicating an unreliable UDP path."))
|
"A large fraction of ECHO probes were lost, indicating an unreliable UDP path."))
|
||||||
}
|
}
|
||||||
|
// Naming the direction is the entire value of the split, so the findings do.
|
||||||
|
directional?.let { d ->
|
||||||
|
when {
|
||||||
|
d.noneReachedServer && received == 0 -> findings.add(
|
||||||
|
finding("nat.udp_unreachable_upstream", Category.CONNECTIVITY, Severity.HIGH, testId,
|
||||||
|
"Nothing reached the server",
|
||||||
|
"The server received none of the ${d.sent} probes, so the traffic is being " +
|
||||||
|
"dropped on the way out, not on the way back. A firewall or NAT on " +
|
||||||
|
"this side of the path is the place to look."),
|
||||||
|
)
|
||||||
|
d.lossUpstreamPct >= 2.0 -> findings.add(
|
||||||
|
finding("connectivity.loss_upstream", Category.CONNECTIVITY, Severity.MEDIUM, testId,
|
||||||
|
"${d.lossUpstreamPct} % of probes were lost on the way to the server",
|
||||||
|
"${d.lostUpstream} of ${d.sent} probes never reached the server. The " +
|
||||||
|
"return path is not implicated: replies came back for everything that " +
|
||||||
|
"arrived."),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if (d.lossDownstreamPct >= 2.0) {
|
||||||
|
findings.add(
|
||||||
|
finding("connectivity.loss_downstream", Category.CONNECTIVITY, Severity.MEDIUM, testId,
|
||||||
|
"${d.lossDownstreamPct} % of replies were lost on the way back",
|
||||||
|
"The server received ${d.seenByServer} probes and answered them, but " +
|
||||||
|
"${d.lostDownstream} of those replies never arrived. The outbound path " +
|
||||||
|
"is fine; the fault is on the return leg."),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (natRebinding) {
|
if (natRebinding) {
|
||||||
findings.add(finding("nat.udp_rebinding", Category.NAT, Severity.MEDIUM, testId,
|
findings.add(finding("nat.udp_rebinding", Category.NAT, Severity.MEDIUM, testId,
|
||||||
"NAT remapped the UDP source port mid-flow",
|
"NAT remapped the UDP source port mid-flow",
|
||||||
@@ -178,6 +236,29 @@ class ServerMeasurement(
|
|||||||
return test to findings
|
return test to findings
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The server's per-packet record of this session's echoes (spec section 6). Filtered to
|
||||||
|
* ECHO_REQ, because the observation list also holds MTU probes and anything else we sent -
|
||||||
|
* counting those as train packets would invent loss that is not there.
|
||||||
|
*/
|
||||||
|
private fun serverSightings(
|
||||||
|
control: ControlClient, cfg: Config, sessionId: String,
|
||||||
|
): List<Directional.ServerSighting> {
|
||||||
|
val body = control.observations(cfg.credential, sessionId)
|
||||||
|
val packets = Json.parseToJsonElement(body).jsonObject["udp"]
|
||||||
|
?.jsonObject?.get("packets") as? JsonArray ?: return emptyList()
|
||||||
|
return packets.mapNotNull { el ->
|
||||||
|
val o = el as? JsonObject ?: return@mapNotNull null
|
||||||
|
val type = o["type"]?.jsonPrimitive?.intOrNull ?: return@mapNotNull null
|
||||||
|
if (type != ECHO_REQ_TYPE) return@mapNotNull null
|
||||||
|
Directional.ServerSighting(
|
||||||
|
seq = o["seq"]?.jsonPrimitive?.intOrNull ?: return@mapNotNull null,
|
||||||
|
tRxNs = o["t_rx_ns"]?.jsonPrimitive?.longOrNull ?: return@mapNotNull null,
|
||||||
|
tTxNs = o["t_tx_ns"]?.jsonPrimitive?.longOrNull ?: return@mapNotNull null,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
|
private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
|
||||||
Finding(
|
Finding(
|
||||||
id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
|
id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
|
||||||
@@ -186,6 +267,7 @@ class ServerMeasurement(
|
|||||||
|
|
||||||
private companion object {
|
private companion object {
|
||||||
const val Wire_HEADER = 32
|
const val Wire_HEADER = 32
|
||||||
|
const val ECHO_REQ_TYPE = 0x01
|
||||||
fun round1(v: Double) = Math.round(v * 10.0) / 10.0
|
fun round1(v: Double) = Math.round(v * 10.0) / 10.0
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,155 @@
|
|||||||
|
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||||
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
|
||||||
|
package app.echo_lot.engine
|
||||||
|
|
||||||
|
import app.echo_lot.engine.Directional.Sample
|
||||||
|
import app.echo_lot.engine.Directional.ServerSighting
|
||||||
|
import kotlin.test.Test
|
||||||
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertFalse
|
||||||
|
import kotlin.test.assertNotNull
|
||||||
|
import kotlin.test.assertNull
|
||||||
|
import kotlin.test.assertTrue
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The arithmetic that turns "3 % loss somewhere" into "3 % loss upstream". Getting a denominator
|
||||||
|
* wrong here does not crash anything — it produces a plausible number pointing at the wrong half
|
||||||
|
* of the network, which is worse than no number at all. Hence a test per claim.
|
||||||
|
*/
|
||||||
|
class DirectionalTest {
|
||||||
|
|
||||||
|
/** A clean train: every packet sent, seen and answered. Server clock offset by a constant. */
|
||||||
|
private fun clean(n: Int, offsetNs: Long = 5_000_000_000L): Pair<List<Sample>, List<ServerSighting>> {
|
||||||
|
val sent = (1..n).map { Sample(it, tTxNs = it * 10_000_000L, tRxNs = it * 10_000_000L + 4_000_000L) }
|
||||||
|
val seen = (1..n).map {
|
||||||
|
ServerSighting(it, tRxNs = offsetNs + it * 10_000_000L + 2_000_000L,
|
||||||
|
tTxNs = offsetNs + it * 10_000_000L + 2_100_000L)
|
||||||
|
}
|
||||||
|
return sent to seen
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun aCleanTrainReportsNoLossInEitherDirection() {
|
||||||
|
val (sent, seen) = clean(10)
|
||||||
|
val m = Directional.analyse(sent, seen)
|
||||||
|
assertEquals(10, m.sent)
|
||||||
|
assertEquals(10, m.seenByServer)
|
||||||
|
assertEquals(10, m.repliesReceived)
|
||||||
|
assertEquals(0.0, m.lossUpstreamPct)
|
||||||
|
assertEquals(0.0, m.lossDownstreamPct)
|
||||||
|
assertFalse(m.noneReachedServer)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The whole point: a packet the server never saw was lost on the way there.
|
||||||
|
@Test
|
||||||
|
fun packetsTheServerNeverSawAreUpstreamLoss() {
|
||||||
|
val (sent, seen) = clean(10)
|
||||||
|
val m = Directional.analyse(sent, seen.filter { it.seq !in setOf(3, 7) })
|
||||||
|
assertEquals(2, m.lostUpstream)
|
||||||
|
assertEquals(0, m.lostDownstream)
|
||||||
|
assertEquals(20.0, m.lossUpstreamPct)
|
||||||
|
assertEquals(0.0, m.lossDownstreamPct, "a packet that never arrived cannot be lost coming back")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun repliesThatNeverArrivedAreDownstreamLoss() {
|
||||||
|
val (sent, seen) = clean(10)
|
||||||
|
val withHoles = sent.map { if (it.seq in setOf(2, 5)) it.copy(tRxNs = null) else it }
|
||||||
|
val m = Directional.analyse(withHoles, seen)
|
||||||
|
assertEquals(0, m.lostUpstream)
|
||||||
|
assertEquals(2, m.lostDownstream)
|
||||||
|
assertEquals(20.0, m.lossDownstreamPct)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Downstream loss is measured against what actually reached the server. Using "sent" as the
|
||||||
|
// denominator would count every upstream loss a second time and overstate the return path.
|
||||||
|
@Test
|
||||||
|
fun downstreamLossIsRelativeToWhatReachedTheServer() {
|
||||||
|
val (sent, seen) = clean(10)
|
||||||
|
// 5 lost on the way there; of the 5 that arrived, 1 reply is lost coming back.
|
||||||
|
val seenPartial = seen.filter { it.seq > 5 }
|
||||||
|
val withHole = sent.map {
|
||||||
|
when {
|
||||||
|
it.seq <= 5 -> it.copy(tRxNs = null) // never got there, so never came back
|
||||||
|
it.seq == 6 -> it.copy(tRxNs = null) // arrived, reply lost
|
||||||
|
else -> it
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val m = Directional.analyse(withHole, seenPartial)
|
||||||
|
assertEquals(5, m.lostUpstream)
|
||||||
|
assertEquals(50.0, m.lossUpstreamPct)
|
||||||
|
assertEquals(1, m.lostDownstream)
|
||||||
|
assertEquals(20.0, m.lossDownstreamPct, "1 of the 5 that arrived, not 1 of 10")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun aServerThatSawNothingIsCalledOutSeparately() {
|
||||||
|
val (sent, _) = clean(6)
|
||||||
|
val m = Directional.analyse(sent.map { it.copy(tRxNs = null) }, emptyList())
|
||||||
|
assertTrue(m.noneReachedServer)
|
||||||
|
assertEquals(100.0, m.lossUpstreamPct)
|
||||||
|
assertEquals(0.0, m.lossDownstreamPct, "with nothing arriving there is no return path to blame")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Jitter is legitimate without synchronised clocks because the offset cancels when successive
|
||||||
|
// one-way samples are differenced. This pins that: a huge constant offset must not show up.
|
||||||
|
@Test
|
||||||
|
fun jitterIsUnaffectedByTheClockOffsetBetweenTheTwoMachines() {
|
||||||
|
val (sent, near) = clean(10, offsetNs = 0)
|
||||||
|
val (_, far) = clean(10, offsetNs = 9_999_999_999L)
|
||||||
|
val a = Directional.analyse(sent, near)
|
||||||
|
val b = Directional.analyse(sent, far)
|
||||||
|
assertEquals(a.jitterUpstreamMs, b.jitterUpstreamMs,
|
||||||
|
"a constant clock offset must cancel when consecutive samples are differenced")
|
||||||
|
assertEquals(0.0, assertNotNull(a.jitterUpstreamMs), "an evenly spaced train has no jitter")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun jitterReflectsUnevenArrival() {
|
||||||
|
val sent = listOf(
|
||||||
|
Sample(1, 0, 10_000_000),
|
||||||
|
Sample(2, 10_000_000, 20_000_000),
|
||||||
|
Sample(3, 20_000_000, 30_000_000),
|
||||||
|
)
|
||||||
|
// Server receive times drift: +2ms, +7ms, +3ms relative to send.
|
||||||
|
val seen = listOf(
|
||||||
|
ServerSighting(1, 2_000_000, 2_100_000),
|
||||||
|
ServerSighting(2, 17_000_000, 17_100_000),
|
||||||
|
ServerSighting(3, 23_000_000, 23_100_000),
|
||||||
|
)
|
||||||
|
val m = Directional.analyse(sent, seen)
|
||||||
|
// one-way samples: 2ms, 7ms, 3ms → |7-2| and |3-7| → mean 4.5ms
|
||||||
|
assertEquals(4.5, assertNotNull(m.jitterUpstreamMs))
|
||||||
|
}
|
||||||
|
|
||||||
|
// "No jitter" and "not enough data to say" are different claims, and only one is true here.
|
||||||
|
@Test
|
||||||
|
fun tooFewSamplesReportsNoJitterRatherThanZero() {
|
||||||
|
val m = Directional.analyse(
|
||||||
|
listOf(Sample(1, 0, 10_000_000)),
|
||||||
|
listOf(ServerSighting(1, 2_000_000, 2_100_000)),
|
||||||
|
)
|
||||||
|
assertNull(m.jitterUpstreamMs)
|
||||||
|
assertNull(m.jitterDownstreamMs)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A server record for a sequence we never sent is not evidence about this train; folding it
|
||||||
|
// in would yield loss percentages outside 0–100.
|
||||||
|
@Test
|
||||||
|
fun strayServerRecordsAreIgnored() {
|
||||||
|
val (sent, seen) = clean(5)
|
||||||
|
val m = Directional.analyse(sent, seen + ServerSighting(99, 1, 2) + ServerSighting(100, 3, 4))
|
||||||
|
assertEquals(5, m.seenByServer)
|
||||||
|
assertEquals(0.0, m.lossUpstreamPct)
|
||||||
|
assertTrue(m.lossDownstreamPct in 0.0..100.0)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun anEmptyTrainDoesNotDivideByZero() {
|
||||||
|
val m = Directional.analyse(emptyList(), emptyList())
|
||||||
|
assertEquals(0.0, m.lossUpstreamPct)
|
||||||
|
assertEquals(0.0, m.lossDownstreamPct)
|
||||||
|
assertFalse(m.noneReachedServer, "nothing sent is not the same as nothing arriving")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -7,6 +7,7 @@ import app.echo_lot.measurement.*
|
|||||||
import kotlinx.serialization.json.Json
|
import kotlinx.serialization.json.Json
|
||||||
import kotlin.test.Test
|
import kotlin.test.Test
|
||||||
import kotlin.test.assertEquals
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertNotNull
|
||||||
import kotlin.test.assertTrue
|
import kotlin.test.assertTrue
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -59,6 +60,18 @@ class LiveMeasurementTest {
|
|||||||
println("metrics: $metrics")
|
println("metrics: $metrics")
|
||||||
assertTrue(metrics.toString().contains("rtt_ms_avg"))
|
assertTrue(metrics.toString().contains("rtt_ms_avg"))
|
||||||
|
|
||||||
|
// The directional split is the point of asking the server what it saw: without it a
|
||||||
|
// lossy path is reported as "loss" with no direction, which sends an engineer looking
|
||||||
|
// in both at once. Correlation is by wire sequence number, so a mismatch here means the
|
||||||
|
// two sides disagree about which packet is which.
|
||||||
|
val m = metrics.toString()
|
||||||
|
assertTrue(m.contains("seen_by_server"), "no directional split in the metrics: $m")
|
||||||
|
val seen = Regex(""""seen_by_server":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
|
||||||
|
assertNotNull(seen, "seen_by_server missing")
|
||||||
|
assertEquals(20, seen, "the server should have seen every probe on a healthy path")
|
||||||
|
assertTrue(m.contains("jitter_upstream_ms"), "no per-direction jitter: $m")
|
||||||
|
println("directional: $m")
|
||||||
|
|
||||||
assertTrue(doc.summary != null)
|
assertTrue(doc.summary != null)
|
||||||
// A healthy local->fmr path should be green (no loss, no rebinding) or yellow.
|
// A healthy local->fmr path should be green (no loss, no rebinding) or yellow.
|
||||||
println("summary: ${doc.summary}")
|
println("summary: ${doc.summary}")
|
||||||
|
|||||||
@@ -44,13 +44,26 @@ class ProbeSession(
|
|||||||
*/
|
*/
|
||||||
fun echo(paddingBytes: Int = 40): EchoResult? {
|
fun echo(paddingBytes: Int = 40): EchoResult? {
|
||||||
val t0 = System.nanoTime()
|
val t0 = System.nanoTime()
|
||||||
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, ++seq, nowNs(), key, ByteArray(paddingBytes))
|
val wireSeq = ++seq
|
||||||
|
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, wireSeq, nowNs(), key, ByteArray(paddingBytes))
|
||||||
socket.send(DatagramPacket(pkt, pkt.size, server))
|
socket.send(DatagramPacket(pkt, pkt.size, server))
|
||||||
|
// A lost probe still has a sequence number, and that number is what lets the server's
|
||||||
|
// observations say whether it was lost going out or coming back — so report it either way.
|
||||||
|
lastSeq = wireSeq
|
||||||
val resp = receive(Wire.TYPE_ECHO_RESP) ?: return null
|
val resp = receive(Wire.TYPE_ECHO_RESP) ?: return null
|
||||||
val rttMs = (System.nanoTime() - t0) / 1_000_000.0
|
val rttMs = (System.nanoTime() - t0) / 1_000_000.0
|
||||||
return EchoResult(rttMs, Observation.parse(resp.payload))
|
return EchoResult(rttMs, Observation.parse(resp.payload), wireSeq)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The wire sequence number of the most recent [echo], including one that was lost.
|
||||||
|
*
|
||||||
|
* Exposed because the caller cannot derive it: the counter is shared with every other packet
|
||||||
|
* type on this session, so "the nth echo" is not "sequence n".
|
||||||
|
*/
|
||||||
|
var lastSeq: Int = 0
|
||||||
|
private set
|
||||||
|
|
||||||
/** One MTU probe of [totalSize] bytes (DF is set by the OS on the socket where supported).
|
/** One MTU probe of [totalSize] bytes (DF is set by the OS on the socket where supported).
|
||||||
* Returns the size the server acknowledged receiving, or null if the probe was lost. */
|
* Returns the size the server acknowledged receiving, or null if the probe was lost. */
|
||||||
fun mtuProbe(totalSize: Int): Int? {
|
fun mtuProbe(totalSize: Int): Int? {
|
||||||
@@ -112,5 +125,5 @@ class ProbeSession(
|
|||||||
|
|
||||||
override fun close() = socket.close()
|
override fun close() = socket.close()
|
||||||
|
|
||||||
data class EchoResult(val rttMs: Double, val observation: Observation?)
|
data class EchoResult(val rttMs: Double, val observation: Observation?, val seq: Int = 0)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user