app: speak train (0x03-0x05) and mark downtrains with DSCP

ProbeSession sends paced upstream trains and fetches the server's columnar received view; UpstreamTrainMeasurement lines both ledgers up per sequence number into TrainEvidence - the directional loss attribution a round trip cannot make. Loss is computed against the server's total count, not its row list, so a buffer-capped report can never invent loss on big trains. A missing report stays ambiguous by name (report lost, or server predates trains) instead of being blamed on the train. downtrain actions can now request a DSCP marking, recording dscp_applied so a survival comparison never blames the path for a marking the sender skipped. Also logged: fmr runs v0.11.2 built from commits this repo's remote never saw, so --self-update on fmr is OFF-LIMITS until that lineage is repaired - the fresh server-v0.9.2 release is newest-created but semantically older, and the updater compares strings, not SemVer. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-02 13:34:52 +02:00
parent d574c76630
commit 515a6aef04
7 changed files with 424 additions and 3 deletions
@@ -168,6 +168,10 @@ class DownstreamMeasurement(private val ids: IdSource) {
        trainCount: Int = 100,
        trainSizeBytes: Int = 300,
        trainIntervalUs: Int = 3_000,
+        // DSCP to mark the train with (0-63), or -1 to leave packets unmarked. Pairing a marked
+        // downtrain with the server-observed DSCP of an upstream train is the two-direction
+        // sec.dscp_ecn_survival measurement.
+        trainDscp: Int = -1,
    ): Pair<List<Test>, List<Finding>> {
        val tests = ArrayList<Test>()
        val findings = ArrayList<Finding>()
@@ -175,7 +179,8 @@ class DownstreamMeasurement(private val ids: IdSource) {
        val df = bigSend(credential, sessionId, control, probe, sessionRef, sizes, df = true)
        val frag = bigSend(credential, sessionId, control, probe, sessionRef, sizes, df = false)
        val train = downTrain(
-            credential, sessionId, control, probe, sessionRef, trainCount, trainSizeBytes, trainIntervalUs,
+            credential, sessionId, control, probe, sessionRef, trainCount, trainSizeBytes,
+            trainIntervalUs, trainDscp,
        )

        tests.add(df.test); tests.add(frag.test); tests.add(train.test)
@@ -338,15 +343,19 @@ class DownstreamMeasurement(private val ids: IdSource) {

    private fun downTrain(
        credential: String, sessionId: String, control: ControlClient, probe: ProbeSession,
-        sessionRef: String, count: Int, sizeBytes: Int, intervalUs: Int,
+        sessionRef: String, count: Int, sizeBytes: Int, intervalUs: Int, dscp: Int = -1,
    ): TrainResult {
        val testId = ids.uuid()
        val started = ids.monoNs()

+        // dscp is only sent when requested: an older server rejects unknown-value problems
+        // louder than absent keys, and unmarked is the correct default for a plain loss train.
+        val dscpField = if (dscp in 0..63) ""","dscp":$dscp""" else ""
        val reply = runCatching {
            control.action(
                credential, sessionId,
-                """{"action":"downtrain","count":$count,"size_bytes":$sizeBytes,"interval_us":$intervalUs}""",
+                """{"action":"downtrain","count":$count,"size_bytes":$sizeBytes,""" +
+                    """"interval_us":$intervalUs$dscpField}""",
            )
        }
        if (reply.isFailure) {
@@ -394,6 +403,12 @@ class DownstreamMeasurement(private val ids: IdSource) {
                interArrivalMsAvg = interArrival.average().takeIf { interArrival.isNotEmpty() }?.let(::round1),
                interArrivalMsMax = interArrival.maxOrNull()?.let(::round1),
                sendIntervalUs = intervalUs,
+                dscpRequested = dscp.takeIf { it in 0..63 },
+                // The server says whether it could actually mark (dscp_applied); recorded so a
+                // survival comparison never blames the path for a marking the sender skipped.
+                dscpApplied = reply.getOrNull()?.let {
+                    Regex("\"dscp_applied\"\\s*:\\s*(true|false)").find(it)?.groupValues?.get(1)?.toBoolean()
+                },
            ),
        ) as JsonObject

@@ -490,4 +505,6 @@ data class DownTrainMetrics(
    @SerialName("inter_arrival_ms_avg") val interArrivalMsAvg: Double? = null,
    @SerialName("inter_arrival_ms_max") val interArrivalMsMax: Double? = null,
    @SerialName("send_interval_us") val sendIntervalUs: Int,
+    @SerialName("dscp_requested") val dscpRequested: Int? = null,
+    @SerialName("dscp_applied") val dscpApplied: Boolean? = null,
 )
@@ -88,6 +88,14 @@ class ServerMeasurement(
            tests.add(test)
            allFindings.addAll(findings)

+            // The upstream train needs no grant and no capability beyond udp-probe itself; a
+            // server that predates trains simply never answers the report request, which the
+            // measurement reports as exactly that ambiguity rather than as network loss.
+            val (utTest, utFindings) = UpstreamTrainMeasurement(ids)
+                .run(ps, sessionRef = "sess-1")
+            tests.add(utTest)
+            allFindings.addAll(utFindings)
+
            // Downstream needs a session the server has already seen traffic from — the echo
            // train just provided that — and a server that advertises the grants. Skipped
            // quietly against an older server rather than reported as a failure of the network.
@@ -0,0 +1,159 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.engine
+
+import app.echo_lot.measurement.*
+import app.echo_lot.protocol.ProbeSession
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.encodeToJsonElement
+
+/**
+ * train.udp_updown — the client sends a paced train (types 0x03), then asks the server what
+ * arrived (0x04 → 0x05) and lines both views up per sequence number.
+ *
+ * This is the measurement a round trip cannot make: an echo run only says "lost somewhere", the
+ * train's two ledgers say lost on the way OUT, specifically, because the server's report names
+ * exactly which sequence numbers reached it. The downstream direction has its own test
+ * (train.udp_downstream) under a grant; this one needs none, since the client generates all the
+ * traffic itself.
+ *
+ * The evidence is the schema's columnar TrainEvidence: one index per sent packet, with the
+ * server-side columns null where a packet never arrived. Server timestamps are on the server's
+ * own clock — only differences within that clock mean anything unless time.server_offset maps
+ * them (two-clock rule).
+ */
+class UpstreamTrainMeasurement(private val ids: IdSource) {
+
+    private val json = Json { encodeDefaults = true; explicitNulls = true }
+
+    fun run(
+        probe: ProbeSession,
+        sessionRef: String,
+        count: Int = 200,
+        sizeBytes: Int = 200,
+        interPacketMs: Long = 5,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+        // The id only needs to be unique within this session; a clash across sessions is
+        // meaningless because trains are buffered per session on the server.
+        val trainId = (System.nanoTime() and 0x7FFFFFFF).toInt()
+
+        val sent = probe.sendTrain(trainId, count, sizeBytes, interPacketMs)
+        // Let the tail arrive before asking for the ledger; packets still in flight when the
+        // report is cut would read as upstream loss.
+        Thread.sleep(300)
+        val report = probe.trainReport(trainId)
+
+        if (report == null) {
+            return Test(
+                id = testId, type = TestType.TRAIN_UDP_UPDOWN, sessionRef = sessionRef,
+                tier = Tier.APP, startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.FAILED,
+                // Honest ambiguity: an old server drops 0x04 silently, and a lost report looks
+                // identical from here. Neither says anything about the train itself.
+                error = TestError(
+                    "no_report",
+                    "no train report arrived — the report was lost, or the server predates trains",
+                ),
+            ) to emptyList()
+        }
+
+        val bySeq = report.rows.associateBy { it.seq }
+        fun col255(v: Int): Int? = v.takeIf { it != 255 } // 255 = "not observed" on the wire
+
+        val evidence = TrainEvidence(
+            epochMonoNs = started,
+            seq = sent.map { it.seq },
+            tTxNs = sent.map { it.tTxNs },
+            tSrvRxNs = sent.map { bySeq[it.seq]?.tRxNs },
+            tRxNs = sent.map { null }, // upstream only: nothing comes back per packet
+            sizeBytes = sent.map { it.sizeBytes },
+            ttlSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.ttl) } },
+            dscpSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.dscp) } },
+            ecnSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.ecn) } },
+            evidenceTruncated = report.truncated,
+        ).toEvidence()
+
+        // Loss against the server's total count, not its row list: rows past the server's buffer
+        // cap are counted but not kept, and treating them as lost would invent loss exactly on
+        // the biggest trains.
+        val lossPct = if (sent.isEmpty()) 0.0 else {
+            (sent.size - report.received).coerceAtLeast(0) * 100.0 / sent.size
+        }
+        val metrics = json.encodeToJsonElement(
+            UpstreamTrainMetrics(
+                sent = sent.size,
+                receivedByServer = report.received,
+                lossPct = round1(lossPct),
+                reportPartsExpected = report.partsExpected,
+                reportPartsReceived = report.partsReceived,
+                truncated = report.truncated,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        if (sent.isNotEmpty() && report.received == 0) {
+            findings.add(
+                Finding(
+                    id = ids.uuid(),
+                    code = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.code,
+                    category = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.category,
+                    severity = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.severity,
+                    confidence = Confidence.HIGH,
+                    title = "The server received none of ${sent.size} upstream packets",
+                    description = "Every train packet vanished on the way out, while the " +
+                        "report request's reply made it back — the outbound path drops this " +
+                        "traffic, the return path works.",
+                    evidenceRefs = listOf(EvidenceRef(testId)),
+                ),
+            )
+        } else if (lossPct >= 2.0) {
+            findings.add(
+                Finding(
+                    id = ids.uuid(),
+                    code = FindingRegistry.LOSS_UPSTREAM.code,
+                    category = FindingRegistry.LOSS_UPSTREAM.category,
+                    severity = FindingRegistry.LOSS_UPSTREAM.severity,
+                    confidence = Confidence.HIGH,
+                    title = "Upstream loss of ${round1(lossPct)} %",
+                    description = "The server received ${report.received} of the ${sent.size} " +
+                        "packets this device sent, and its per-sequence ledger names the " +
+                        "missing ones. This is outbound loss specifically; the return path " +
+                        "delivered the report.",
+                    evidenceRefs = listOf(EvidenceRef(testId)),
+                ),
+            )
+        }
+
+        val status = when {
+            report.received == 0 && sent.isNotEmpty() -> TestStatus.FAILED
+            report.partsReceived < report.partsExpected -> TestStatus.PARTIAL
+            else -> TestStatus.OK
+        }
+        return Test(
+            id = testId, type = TestType.TRAIN_UDP_UPDOWN, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = status, evidence = evidence, metrics = metrics,
+        ) to findings
+    }
+
+    private fun round1(v: Double) = Math.round(v * 10.0) / 10.0
+}
+
+/** Metrics for train.udp_updown. */
+@Serializable
+data class UpstreamTrainMetrics(
+    val sent: Int,
+    /** The server's total count — includes packets past its row buffer (counted, not listed). */
+    @SerialName("received_by_server") val receivedByServer: Int,
+    @SerialName("loss_pct") val lossPct: Double,
+    @SerialName("report_parts_expected") val reportPartsExpected: Int,
+    @SerialName("report_parts_received") val reportPartsReceived: Int,
+    /** The server's row buffer overflowed: rows are a sample, the count is still complete. */
+    val truncated: Boolean,
+)
@@ -0,0 +1,65 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.engine
+
+import app.echo_lot.measurement.TestStatus
+import app.echo_lot.protocol.ControlClient
+import app.echo_lot.protocol.ProbeSession
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Upstream train (types 0x03-0x05) against a LIVE server. Self-skips without ECHOLOT_LIVE_*.
+ *
+ * What is asserted is the ledger property: the server's report must account for what was sent,
+ * per sequence number, because directional loss attribution is the entire reason trains exist —
+ * a test that only checked "a report came back" would pass against a server that counts nothing.
+ */
+class LiveUpstreamTrainTest {
+
+    private val url = System.getenv("ECHOLOT_LIVE_URL")
+    private val pin = System.getenv("ECHOLOT_LIVE_PIN")
+    private val cred = System.getenv("ECHOLOT_LIVE_CRED")
+    private val udp = System.getenv("ECHOLOT_LIVE_UDP")
+    private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
+
+    @Test
+    fun serverLedgerAccountsForTheTrain() {
+        if (url == null || pin == null || cred == null || udp == null) {
+            println("LiveUpstreamTrainTest skipped (no ECHOLOT_LIVE_* env)"); return
+        }
+        val control = ControlClient(url, setOf(pin), "0.2.0")
+        val session = control.createSession(cred, target)
+        val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
+
+        val (test, findings) = ProbeSession(cred, session, host, port).use { ps ->
+            ps.echo() // prime the session so its source is known
+            UpstreamTrainMeasurement(SystemIdSource()).run(
+                ps, sessionRef = "sess-1", count = 120, sizeBytes = 200, interPacketMs = 3,
+            )
+        }
+        control.deleteSession(cred, session.sessionId)
+
+        val m = assertNotNull(test.metrics).toString()
+        println("updown: ${test.status} $m")
+        for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
+
+        assertEquals(TestStatus.OK, test.status, "train report incomplete or absent: $m")
+
+        val sent = Regex(""""sent":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        val received = Regex(""""received_by_server":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        assertNotNull(sent); assertNotNull(received)
+        assertTrue(sent > 0, "nothing was sent: $m")
+        // Over a working path the ledger must be near-complete; a lossy wifi may drop a few, but
+        // a server that fails to count would show up as massive phantom loss here.
+        assertTrue(received >= sent * 9 / 10, "server counted $received of $sent: $m")
+
+        // The columnar evidence must carry a server timestamp for arrived packets — that column
+        // is what one-way delay math consumes after timesync.
+        val ev = assertNotNull(test.evidence).toString()
+        assertTrue(ev.contains("t_srv_rx_ns"), "no server rx column in evidence")
+    }
+}