throughput: paced downstream rate, with the qualifier that makes it honest

A throughput number reports the smallest limit on the path, and the sender's own
ceiling is one of the candidates. If the server was asked for 50 Mbps and 50
Mbps arrived, the network was never the constraint and "50 Mbps" says nothing
about it. So the result always carries limited_by and measures_network, and a
finding is raised only when the path is actually implicated.

Loss is computed against the *sender's* count, not the requested rate: the
server reports what it put on the wire, and the gap is the loss. A receiver
alone cannot tell "the network dropped it" from "the sender never sent it", and
guessing turns a healthy server-side limit into a phantom network fault. The
count is stored per action, not per packet — half a million packets of structs
would turn a measurement into memory exhaustion.

Sending is paced rather than flat out. An unpaced burst measures the server's
NIC and the first queue it meets, then collapses into loss that reads as a
network fault. The schedule is absolute rather than sleep-per-packet, which
would accumulate scheduler error and drift the rate down over a ten-second run.

Throughput gets its own grant budget sized from the request, so every other
action stays bounded at 8 MiB. When the byte cap binds before the clock does,
the *duration* is shortened and reported, rather than the run being truncated
halfway: promising thirty seconds and delivering twenty-one is the same
information with a surprise attached, and it keeps "the clock ended the run" as
the normal case — the only case where the rate is a clean property of the path.

That last behaviour came out of a test that failed honestly: 30 s at 100 Mbps
needs 375 MB against a 256 MB cap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/build-status.md

@@ -728,3 +728,49 @@ 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.
+
+### frag_send: crafted IP fragments, so *ordering* is testable (server-v0.6.0, 2026-08-01)
+`big_send` with `df=false` answers one question — do fragments get through. It cannot answer the
+more interesting one, because the kernel always emits fragments in order, first one first.
+
+The classic middlebox fault is exactly about that ordering. Only the **first** fragment carries the
+UDP header, and therefore the ports; a stateful firewall or NAT that has not seen it has no flow to
+match the rest against, and many simply drop them. That is invisible to every in-order test, and in
+the field it looks like "large DNS answers fail on this network" or "the tunnel breaks when the MTU
+drops" — it works until the network reorders, then fails intermittently, which is the hardest kind
+of fault to chase.
+
+So the server builds the fragments itself (raw socket, `IP_HDRINCL`) and controls their order:
+`in_order` (baseline), `reversed` (last fragment first), `first_last` (first fragment held back
+250 ms). The datagram is assembled and **signed whole** before being cut up, so what the client
+reassembles is indistinguishable from an ordinary packet — otherwise the test would be measuring
+our sender rather than the path. New test type `mtu.frag_ordering`; findings
+`mtu.fragments_blocked` and `mtu.fragment_reorder_sensitive`.
+
+Two details that would otherwise produce confidently wrong answers:
+- **The UDP checksum is computed, not left zero.** Zero is legal in IPv4 and would be less code,
+  but zero-checksum datagrams are dropped by some middleboxes — and that drop would be recorded as
+  a fragmentation failure, which is the wrong conclusion entirely.
+- **Fragment offsets are in 8-byte units**, so non-final fragments are rounded down to a multiple
+  of 8. A 100-byte fragment is not an error; it is a datagram no host will ever reassemble.
+
+`frag-send` is advertised only when a raw socket can actually be opened — checked by opening one,
+because a permission model has more ways to say no (userns, seccomp, LSM) than a capability bit has
+to say yes. fmr runs as root with `cap_net_raw` in its bounding set, so it is available there.
+
+Fragment ordering runs only after `mtu.frag_delivery` shows fragments arrive at all; otherwise the
+three orderings would each report "not delivered" and read as three faults instead of one.
+
+The header arithmetic is unit-tested (reassembly coverage with no gaps or double-delivery, MF
+flags, shared IP ID, 8-byte offsets, checksum verification over odd and even lengths). Because the
+code is `//go:build linux`, the tests are **cross-compiled and run on fmr** — there is no Go
+toolchain there, so `go test -c` plus scp is the loop.
+
+Live against fmr: 4 fragments per burst, and all three orderings reassembled — a healthy path, and
+the baseline against which a mobile network will be interesting.
+
+### Testing state (2026-08-01)
+Six live tests against fmr, all green, no device involved: `LiveServerTest`, `LiveMeasurement`,
+`LiveGranted`, `LiveDownstream`, `LiveUpload`, `LiveCompat`, `LiveEnrollment`. Plus 74 client unit
+tests and the full Go suite. Everything in the last several entries is verified from the PC; the
+app's UI (settings, history, deep-link enrollment) and `mtu.pmtud_up` remain device-only.

echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt

@@ -0,0 +1,201 @@
+// 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.ControlClient
+import app.echo_lot.protocol.ProbeSession
+import app.echo_lot.protocol.Wire
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.encodeToJsonElement
+import kotlinx.serialization.json.jsonArray
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+
+/**
+ * Downstream throughput: the server sends at a paced rate for a bounded time and the client
+ * measures what arrives (`perf.throughput_udp`).
+ *
+ * The number this produces is only meaningful with a qualifier attached, and getting that
+ * qualifier right is most of the work here. A throughput test reports the *smallest* limit on the
+ * path, and the sender's own ceiling is one of the candidates: if the server was asked for 50 Mbps
+ * and 50 Mbps arrived, the network was never the constraint and "50 Mbps" says nothing about it.
+ * Reporting that as a capacity measurement would be a confident lie, so the result always carries
+ * [ThroughputMetrics.limitedBy] and a finding is only raised when the network is actually
+ * implicated.
+ *
+ * Comparing against the *sender's* count rather than the requested rate is the other half: the
+ * server reports how much it actually put on the wire, and the gap between that and what arrived
+ * is the loss. A receiver alone cannot tell "the network dropped it" from "the sender never sent
+ * it", and guessing turns a healthy server-side limit into a phantom network fault.
+ */
+class ThroughputMeasurement(private val ids: IdSource) {
+
+    private val json = Json { encodeDefaults = true; explicitNulls = true }
+
+    fun run(
+        credential: String,
+        sessionId: String,
+        control: ControlClient,
+        probe: ProbeSession,
+        sessionRef: String,
+        durationS: Int = 5,
+        kbps: Int = 50_000,
+        sizeBytes: Int = 1200,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+
+        val reply = runCatching {
+            control.action(
+                credential, sessionId,
+                """{"action":"throughput","direction":"down","duration_s":$durationS,""" +
+                    """"kbps":$kbps,"size_bytes":$sizeBytes}""",
+            )
+        }
+        if (reply.isFailure) {
+            return Test(
+                id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+                startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.UNSUPPORTED,
+                error = TestError("action_refused", reply.exceptionOrNull()?.message ?: "throughput refused"),
+            ) to emptyList()
+        }
+
+        // The server may have shortened the run to fit its own byte budget; listen for what it
+        // actually promised, not for what we asked.
+        val plannedMs = parseInt(reply.getOrNull(), "duration_ms") ?: (durationS * 1000)
+
+        // A margin past the planned end so the tail of the run is not counted as loss: packets
+        // still in flight when we stop listening were not dropped, they were merely late.
+        val received = probe.collectGranted(plannedMs + 1_500L)
+            .filter { it.type == Wire.TYPE_THROUGHPUT_DATA }
+
+        val bytes = received.sumOf { it.sizeBytes.toLong() }
+        val spanNs = if (received.size >= 2) {
+            received.maxOf { it.tRxNs } - received.minOf { it.tRxNs }
+        } else {
+            0L
+        }
+        // Measured over the arrival span rather than our listening window, which includes the
+        // request round trip and the trailing margin and would understate the rate.
+        val receivedKbps = if (spanNs > 0) (bytes * 8 * 1_000_000 / spanNs).toInt() else 0
+
+        val sender = senderReport(control, credential, sessionId)
+        val sentPackets = sender?.packets ?: 0
+        val lossPct = if (sentPackets > 0) {
+            round2((sentPackets - received.size).coerceAtLeast(0) * 100.0 / sentPackets)
+        } else {
+            null
+        }
+
+        // Only a run the *clock* ended measured the network. One stopped by our own byte budget
+        // or rate ceiling measured this server.
+        val limitedBy = sender?.limitedBy ?: "unknown"
+        val networkLimited = limitedBy == "duration" &&
+            sender != null && receivedKbps > 0 && receivedKbps < sender.kbps * 9 / 10
+
+        val metrics = json.encodeToJsonElement(
+            ThroughputMetrics(
+                requestedKbps = kbps,
+                plannedDurationMs = plannedMs,
+                packetsReceived = received.size,
+                bytesReceived = bytes,
+                receivedKbps = receivedKbps,
+                senderPackets = sender?.packets,
+                senderBytes = sender?.bytes,
+                senderKbps = sender?.kbps,
+                lossPct = lossPct,
+                limitedBy = limitedBy,
+                measuresNetwork = networkLimited,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        when {
+            sender == null -> Unit // no sender report: nothing can be concluded, so nothing is
+            received.isEmpty() -> findings.add(
+                finding(
+                    "perf.throughput_no_delivery", Category.PERFORMANCE, Severity.HIGH, testId,
+                    "No throughput traffic arrived",
+                    "The server sent ${sender.packets} packets and none arrived. This is a " +
+                        "connectivity fault rather than a slow link.",
+                ),
+            )
+            networkLimited -> findings.add(
+                finding(
+                    "perf.throughput_below_offered", Category.PERFORMANCE, Severity.LOW, testId,
+                    "Downstream throughput ${receivedKbps / 1000} Mbit/s, below the " +
+                        "${sender.kbps / 1000} Mbit/s offered",
+                    "The server sent at ${sender.kbps / 1000} Mbit/s for the full run and " +
+                        "${receivedKbps / 1000} Mbit/s arrived" +
+                        (lossPct?.let { ", losing $it % of packets" } ?: "") +
+                        ". The path could not carry what was offered.",
+                ),
+            )
+        }
+
+        return Test(
+            id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = if (received.isEmpty()) TestStatus.FAILED else TestStatus.OK,
+            metrics = metrics,
+        ) to findings
+    }
+
+    private data class SenderReport(
+        val packets: Int, val bytes: Long, val kbps: Int, val limitedBy: String,
+    )
+
+    /** The server's own account of the run, from the observations API. */
+    private fun senderReport(
+        control: ControlClient, credential: String, sessionId: String,
+    ): SenderReport? = runCatching {
+        val arr = Json.parseToJsonElement(control.observations(credential, sessionId))
+            .jsonObject["throughput"]?.jsonArray ?: return null
+        val last = arr.lastOrNull()?.jsonObject ?: return null
+        SenderReport(
+            packets = last["packets"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+            bytes = last["bytes"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0,
+            kbps = last["kbps"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+            limitedBy = last["limited_by"]?.jsonPrimitive?.content ?: "unknown",
+        )
+    }.getOrNull()
+
+    private fun parseInt(body: String?, key: String): Int? =
+        body?.let { Regex("\"$key\"\\s*:\\s*(-?\\d+)").find(it)?.groupValues?.get(1)?.toIntOrNull() }
+
+    private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
+        Finding(
+            id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
+            title = title, description = desc, evidenceRefs = listOf(EvidenceRef(testId)),
+        )
+
+    private fun round2(v: Double) = Math.round(v * 100.0) / 100.0
+}
+
+/** Metrics for perf.throughput_udp. */
+@Serializable
+data class ThroughputMetrics(
+    @SerialName("requested_kbps") val requestedKbps: Int,
+    @SerialName("planned_duration_ms") val plannedDurationMs: Int,
+    @SerialName("packets_received") val packetsReceived: Int,
+    @SerialName("bytes_received") val bytesReceived: Long,
+    @SerialName("received_kbps") val receivedKbps: Int,
+    @SerialName("sender_packets") val senderPackets: Int? = null,
+    @SerialName("sender_bytes") val senderBytes: Long? = null,
+    @SerialName("sender_kbps") val senderKbps: Int? = null,
+    /** Against the sender's count, so a server-side limit is never counted as network loss. */
+    @SerialName("loss_pct") val lossPct: Double? = null,
+    /** What ended the run: duration | budget | rate | send_error | unknown. */
+    @SerialName("limited_by") val limitedBy: String,
+    /**
+     * Whether this number says anything about the network. False when the sender's own ceiling
+     * was the binding constraint — in which case the rate is a property of the test, not the path.
+     */
+    @SerialName("measures_network") val measuresNetwork: Boolean,
+)

echolot-app/core-engine/src/test/kotlin/app/echo_lot/engine/LiveDownstreamTest.kt

@@ -44,7 +44,9 @@ class LiveDownstreamTest {
         for (t in tests) println("${t.type}  status=${t.status}  metrics=${t.metrics}")
         for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
 
-        assertEquals(3, tests.size, "expected pmtud_down, frag_delivery and a downstream train")
+        // Assert on what is present, not on how many: adding a measurement should not be a
+        // test edit. (It was, once — hence the note.)
+        assertTrue(tests.size >= 3, "expected at least the three downstream tests, got ${tests.size}")
         val byType = tests.associateBy { it.type }
 
         val pmtud = assertNotNull(byType[TestType.MTU_PMTUD_DOWN], "no mtu.pmtud_down test")
@@ -58,6 +60,17 @@ class LiveDownstreamTest {
         val frag = assertNotNull(byType[TestType.MTU_FRAG_DELIVERY], "no mtu.frag_delivery test")
         assertNotNull(frag.metrics?.get("largest_delivered_bytes"))
 
+        // Fragment ordering runs only when fragments arrive at all, and only against a server
+        // that can craft them — so it is checked when present rather than required.
+        byType[TestType.MTU_FRAG_ORDERING]?.let { fo ->
+            val m = fo.metrics?.toString() ?: ""
+            println("fragment ordering: ${fo.status} $m")
+            if (fo.status != TestStatus.UNSUPPORTED) {
+                assertTrue(m.contains("in_order"), "no per-ordering result: $m")
+                assertTrue(m.contains("reversed"), "reversed ordering was never attempted: $m")
+            }
+        }
+
         val train = assertNotNull(byType[TestType.TRAIN_UDP_DOWNSTREAM], "no downstream train")
         assertNotNull(train.evidence, "a train without columnar evidence is not recomputable")
         val received = train.metrics?.get("received")?.toString()?.toIntOrNull() ?: 0

echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt

@@ -38,6 +38,9 @@ object Wire {
      */
     const val TYPE_FRAG_DATA: Int = 0x0D
 
+    /** One packet of a sustained-rate downstream run. */
+    const val TYPE_THROUGHPUT_DATA: Int = 0x0E
+
     /** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
     fun wirePrefix(sessionId: String): ByteArray {
         require(sessionId.length >= 16) { "session id too short" }

server/cmd/echolot-server/main.go

@@ -111,7 +111,7 @@ func serve(cfg *config.Config) error {
 		TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12},
 	}
 
-	caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo", "downtrain", "big-send"}
+	caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo", "downtrain", "big-send", "throughput"}
 	// Crafted fragments need a raw socket. Advertised only when one can actually be opened —
 	// a capability we cannot deliver turns a missing feature into a failed measurement.
 	rawFrag := dataplane.RawFragSupported()
@@ -167,6 +167,7 @@ func serve(cfg *config.Config) error {
 	if rawFrag {
 		ctl.FragSend = dp.FragSend
 	}
+	ctl.DownThroughput = dp.DownThroughput
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()

server/internal/control/control.go

@@ -62,6 +62,8 @@ type Server struct {
 	// FragSend emits one datagram as hand-built IP fragments in a chosen order (may be nil:
 	// needs a raw socket, so it is unavailable to an unprivileged server).
 	FragSend func(sess *session.Session, g *session.Grant, sizeBytes int, mode dataplane.FragMode, fragSize int) (dataplane.FragResult, error)
+	// DownThroughput sends paced traffic toward the client for a bounded time (may be nil).
+	DownThroughput func(sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int) (dataplane.ThroughputResult, error)
 	// EgressMTU reports the server's own measured egress path MTU (0 = unknown). With DF set
 	// we cannot emit a datagram larger than this, so requested sizes above it are refused up
 	// front and reported as such — the client must not read that as a downstream path limit.
@@ -221,6 +223,8 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
 		"udp":          map[string]any{"packets_seen": packetsSeen, "packets": udp},
 		"tcp":          tcp,
 		"connect_back": cb,
+		// The sender's own count, which is what makes the receiver's count mean something.
+		"throughput": sess.ThroughputReports(),
 		"dns_canary": dnsCanary,
 		// TODO(spec §6): http echo records
 	})
@@ -246,6 +250,10 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 		DF         *bool  `json:"df"`
 		Mode       string `json:"mode"`
 		FragBytes  int    `json:"frag_bytes"`
+		Direction  string `json:"direction"`
+		DurationS  int    `json:"duration_s"`
+		Kbps       int    `json:"kbps"`
+		Streams    int    `json:"streams"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
@@ -417,6 +425,55 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
 		})
 
+	case "throughput":
+		if s.DownThroughput == nil {
+			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "throughput not wired"})
+			return
+		}
+		// Only the downstream direction needs the server to send. Upstream is the client
+		// sending and the server counting, which needs no action at all — so asking for it here
+		// is a client bug worth naming rather than silently doing the other thing.
+		if req.Direction != "" && req.Direction != "down" {
+			writeJSON(w, http.StatusBadRequest, map[string]string{
+				"error": "only direction=down is an action; upstream throughput is measured by sending and reading observations",
+			})
+			return
+		}
+		// Planned once, here, so the response promises exactly what the run will do. A request
+		// that would outlast the server's byte cap comes back with a shorter duration rather
+		// than being truncated halfway.
+		durationMs, kbps := dataplane.ThroughputPlan(
+			clamp(req.DurationS, 1, 30)*1000, clamp(req.Kbps, 100, 200_000))
+		size := clamp(req.SizeBytes, dataMinPacket, 1472)
+		if req.SizeBytes == 0 {
+			size = 1200
+		}
+		g := sess.NewGrant(actionID, 0, kbps, dataplane.ThroughputLimits(durationMs, kbps))
+		if g == nil {
+			writeJSON(w, http.StatusConflict, noDataPlaneYet)
+			return
+		}
+		// Answered before the run so the client can start listening, then reported through the
+		// observations API. Doing it the other way round would have the client miss the first
+		// second of a ten-second test.
+		writeJSON(w, http.StatusAccepted, map[string]any{
+			"action_id": actionID, "direction": "down",
+			"duration_s": durationMs / 1000, "duration_ms": durationMs,
+			"requested_duration_s": clamp(req.DurationS, 1, 30),
+			"kbps":                 kbps, "size_bytes": size,
+			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
+		})
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		go func() {
+			result, err := s.DownThroughput(sess, g, durationMs, kbps, size)
+			slog.Info("throughput finished", "action", actionID, "packets", result.Packets,
+				"bytes", result.Bytes, "kbps", result.Kbps, "limited_by", result.LimitedBy, "err", err)
+			sess.RecordThroughput(actionID, result.Packets, result.Bytes, result.DurationMs,
+				result.Kbps, result.LimitedBy)
+		}()
+
 	default:
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
 	}

server/internal/dataplane/throughput.go

@@ -0,0 +1,182 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"echo-lot.app/server/internal/session"
+)
+
+// Sustained-rate sending (spec §5 throughput).
+//
+// This is the most expensive thing the server will do on a client's say-so, so it is also the
+// action where the §3.4 anti-amplification rules matter most. Three bounds apply, and all three
+// are enforced here rather than trusted to the caller:
+//
+//   - the destination is the session's *observed* data-plane source, verified by an HMAC-signed
+//     ECHO that arrived from that address, so this cannot be aimed at a third party;
+//   - the grant carries a byte budget and an average-rate ceiling, and the send stops the moment
+//     either is reached;
+//   - the duration is hard-capped, so a client that vanishes mid-test costs a bounded amount of
+//     traffic rather than an open-ended one.
+//
+// The measurement this produces is honest only if the client is told which limit it hit. A run
+// that saturates the grant ceiling has measured *us*, not the network, and reporting that as
+// throughput would be worse than not measuring at all — see ThroughputResult.LimitedBy.
+
+// ThroughputResult is what the server actually managed to send.
+type ThroughputResult struct {
+	Packets    int   `json:"packets"`
+	Bytes      int64 `json:"bytes"`
+	DurationMs int64 `json:"duration_ms"`
+	Kbps       int   `json:"kbps"`
+	// LimitedBy says what stopped it: "duration" (ran the full time, so the rate is the path's
+	// or ours to give), "budget" (hit the grant's byte ceiling), or "rate" (the pacing ceiling
+	// held it back). Only "duration" makes the number a property of the network.
+	LimitedBy string `json:"limited_by"`
+}
+
+// ThroughputLimits derives a grant sized for one throughput run.
+//
+// The default 8 MiB action budget is deliberately far too small for this — ten seconds at
+// 50 Mbps is 62 MB — so throughput gets its own budget computed from what it asked for, still
+// clamped to a ceiling. Sizing the budget to the request (rather than raising the global default)
+// keeps every *other* action bounded at 8 MiB.
+func ThroughputLimits(durationMs, kbps int) session.GrantLimits {
+	durationMs, kbps = ThroughputPlan(durationMs, kbps)
+	// bytes = kbps * 1000 / 8 * seconds, with a little headroom so the byte budget is not what
+	// stops a run that was meant to be stopped by the clock.
+	budget := int64(kbps) * 1000 / 8 * int64(durationMs) / 1000
+	budget = budget * 11 / 10
+	if budget > maxThroughputBytes {
+		budget = maxThroughputBytes
+	}
+	return session.GrantLimits{
+		MaxBytes: budget,
+		// A little above the pacing target on purpose: the pacer should be what controls the
+		// rate, and the grant should be the safety net. If they are equal, ordinary scheduling
+		// jitter trips the grant and the run is cut short for no real reason.
+		MaxKbps: kbps * 12 / 10,
+		MaxHold: time.Duration(durationMs)*time.Millisecond + 5*time.Second,
+	}
+}
+
+// ThroughputPlan reduces a request to what this server will actually run, and is the single
+// place that decides it.
+//
+// When the byte cap binds before the clock does, the *duration* is shortened rather than the run
+// being cut off partway. Truncating mid-run is not wrong exactly — the rate is still computed
+// over the elapsed time and limited_by says "budget" — but it means promising a client thirty
+// seconds and giving it twenty-one. Saying "twenty-one seconds" up front is the same information
+// without the surprise, and it keeps "the clock ended the run" as the normal case, which is the
+// only case where the number is a clean property of the network.
+func ThroughputPlan(durationMs, kbps int) (effectiveMs, effectiveKbps int) {
+	if durationMs <= 0 {
+		durationMs = 10_000
+	}
+	if durationMs > maxThroughputMs {
+		durationMs = maxThroughputMs
+	}
+	if kbps <= 0 || kbps > maxThroughputKbps {
+		kbps = maxThroughputKbps
+	}
+	bytesPerMs := int64(kbps) * 1000 / 8 / 1000
+	if bytesPerMs > 0 {
+		if maxMs := maxThroughputBytes / bytesPerMs; int64(durationMs) > maxMs {
+			durationMs = int(maxMs)
+		}
+	}
+	return durationMs, kbps
+}
+
+const (
+	maxThroughputMs    = 30_000
+	maxThroughputKbps  = 200_000
+	maxThroughputBytes = 256 << 20
+)
+
+// DownThroughput sends paced traffic toward the client for up to durationMs.
+//
+// Pacing is deliberate rather than "send as fast as possible": an unpaced burst measures the
+// server's NIC and the first queue it meets, then collapses into loss that looks like a network
+// fault. Spacing packets at the target rate makes loss mean what a reader will assume it means.
+func (s *Server) DownThroughput(
+	sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int,
+) (ThroughputResult, error) {
+	res := ThroughputResult{}
+
+	target := sess.DataSource()
+	if !target.IsValid() {
+		return res, fmt.Errorf("no observed data-plane source")
+	}
+	conn := s.connFor(target, sess.DataLocal())
+	if conn == nil {
+		return res, fmt.Errorf("no data-plane socket matches target family")
+	}
+
+	// Same plan the grant was sized from, so the two cannot disagree.
+	durationMs, kbps = ThroughputPlan(durationMs, kbps)
+	if sizeBytes < HeaderSize+16 {
+		sizeBytes = 1200 // a size that survives every common path unfragmented
+	}
+	if sizeBytes > 1472 {
+		sizeBytes = 1472
+	}
+
+	// Nanoseconds between packets to hit the target rate.
+	perPacketNs := int64(sizeBytes) * 8 * 1_000_000 / int64(kbps)
+	if perPacketNs < 1_000 {
+		perPacketNs = 1_000
+	}
+
+	payload := make([]byte, sizeBytes-HeaderSize)
+	deadline := time.Now().Add(time.Duration(durationMs) * time.Millisecond)
+	start := time.Now()
+	next := start
+
+	res.LimitedBy = "duration"
+	var seq uint32
+	for time.Now().Before(deadline) {
+		if !g.Allow(sizeBytes) {
+			// Distinguishing these two matters: a run stopped by the byte budget has not been
+			// running long enough for its rate to mean anything.
+			if g.Sent() >= g.MaxBytes {
+				res.LimitedBy = "budget"
+			} else {
+				res.LimitedBy = "rate"
+			}
+			break
+		}
+		binary.BigEndian.PutUint32(payload[0:4], seq)
+		binary.BigEndian.PutUint64(payload[4:12], uint64(time.Since(s.start).Nanoseconds()))
+		if err := s.sendErr(conn, target, sess, TypeThroughputData, seq, payload); err != nil {
+			// A send error mid-run is a local condition (buffer full, route gone). Stop and
+			// report what got out rather than pretending the rest was lost on the path.
+			res.LimitedBy = "send_error"
+			break
+		}
+		res.Packets++
+		res.Bytes += int64(sizeBytes)
+		seq++
+
+		// Absolute schedule, not sleep-per-packet: sleeping a fixed interval accumulates the
+		// scheduler's error and drifts the achieved rate below the target over a 10-second run.
+		next = next.Add(time.Duration(perPacketNs))
+		if d := time.Until(next); d > 0 {
+			time.Sleep(d)
+		}
+	}
+
+	elapsed := time.Since(start)
+	res.DurationMs = elapsed.Milliseconds()
+	// bits per millisecond is kilobits per second, so no scaling constant is needed - and none
+	// can be got wrong. Guarded because a run that ends inside a millisecond has no rate.
+	if res.DurationMs > 0 {
+		res.Kbps = int(res.Bytes * 8 / res.DurationMs)
+	}
+	return res, nil
+}

server/internal/dataplane/throughput_test.go

@@ -0,0 +1,99 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+import (
+	"testing"
+	"time"
+)
+
+// The grant has to be big enough that the *clock* ends a throughput run, not the byte budget. Get
+// this wrong and the test still "works": it stops early, reports a rate computed over a truncated
+// window, and nothing anywhere says the number is meaningless. So the sizing is pinned.
+func TestThroughputBudgetOutlastsTheRequestedRun(t *testing.T) {
+	cases := []struct{ durationMs, kbps int }{
+		{1_000, 1_000},
+		{10_000, 50_000},
+		{10_000, 200_000},
+		{30_000, 100_000},
+	}
+	for _, c := range cases {
+		// Against the *planned* duration, which is what will actually be run: a request the
+		// server shortens is answered with the shorter number, not truncated halfway.
+		planMs, planKbps := ThroughputPlan(c.durationMs, c.kbps)
+		lim := ThroughputLimits(c.durationMs, c.kbps)
+		needed := int64(planKbps) * 1000 / 8 * int64(planMs) / 1000
+		if lim.MaxBytes < needed {
+			t.Errorf("%d ms at %d kbps (planned %d ms) needs %d bytes, budget is %d - the run "+
+				"would stop early and report a rate over a truncated window",
+				c.durationMs, c.kbps, planMs, needed, lim.MaxBytes)
+		}
+	}
+}
+
+// The pacer should control the rate and the grant should be the safety net. If the grant's
+// ceiling equals the pacing target, ordinary scheduling jitter trips it and cuts the run short
+// for no real reason.
+func TestGrantRateCeilingSitsAboveThePacingTarget(t *testing.T) {
+	lim := ThroughputLimits(10_000, 50_000)
+	if lim.MaxKbps <= 50_000 {
+		t.Fatalf("grant ceiling %d kbps is not above the 50000 kbps pacing target", lim.MaxKbps)
+	}
+}
+
+// A client asking for more than the server will do must get the server's number, not its own.
+func TestThroughputRequestsAreClamped(t *testing.T) {
+	lim := ThroughputLimits(10*60*1000, 10_000_000) // ten minutes at 10 Gbps
+	if lim.MaxBytes > maxThroughputBytes {
+		t.Errorf("byte budget %d exceeds the hard cap %d", lim.MaxBytes, maxThroughputBytes)
+	}
+	if lim.MaxKbps > maxThroughputKbps*12/10 {
+		t.Errorf("rate ceiling %d exceeds the hard cap", lim.MaxKbps)
+	}
+	// The hold has to outlast the planned run, or the grant expires mid-send and the run is
+	// reported as rate-limited when it was really time-limited.
+	planMs, _ := ThroughputPlan(10*60*1000, 10_000_000)
+	if lim.MaxHold < time.Duration(planMs)*time.Millisecond {
+		t.Errorf("hold %v is shorter than the planned run of %d ms", lim.MaxHold, planMs)
+	}
+}
+
+// When the byte cap binds before the clock does, the server shortens the run and says so, rather
+// than accepting thirty seconds and delivering twenty-one. Same information, no surprise - and it
+// keeps "the clock ended the run" as the normal case, which is the only case where the resulting
+// rate is a clean property of the network.
+func TestAnOversizedRequestComesBackShorterRatherThanTruncated(t *testing.T) {
+	const kbps = 200_000
+	askedMs := 30_000
+	planMs, planKbps := ThroughputPlan(askedMs, kbps)
+
+	if planKbps != kbps {
+		t.Errorf("rate was reduced to %d; the duration should absorb the cap, not the rate", planKbps)
+	}
+	if planMs >= askedMs {
+		t.Fatalf("plan kept the full %d ms at %d kbps, which exceeds the %d byte cap",
+			askedMs, kbps, maxThroughputBytes)
+	}
+	// And what it does promise must fit.
+	if got := int64(planKbps) * 1000 / 8 * int64(planMs) / 1000; got > maxThroughputBytes {
+		t.Errorf("planned run needs %d bytes, over the %d cap", got, maxThroughputBytes)
+	}
+}
+
+// A short, ordinary request must come back untouched - the clamping only exists for the extremes.
+func TestAnOrdinaryRequestIsNotRewritten(t *testing.T) {
+	planMs, planKbps := ThroughputPlan(10_000, 50_000)
+	if planMs != 10_000 || planKbps != 50_000 {
+		t.Errorf("10 s at 50 Mbps was rewritten to %d ms at %d kbps", planMs, planKbps)
+	}
+}
+
+// Every action other than throughput stays on the small default budget. Throughput needs a big
+// one; raising the global default to suit it would quietly unbound everything else.
+func TestOnlyThroughputGetsTheLargeBudget(t *testing.T) {
+	big := ThroughputLimits(10_000, 50_000)
+	if big.MaxBytes <= 8<<20 {
+		t.Fatalf("throughput budget %d is no larger than the default action budget", big.MaxBytes)
+	}
+}

server/internal/dataplane/udp.go

@@ -37,6 +37,8 @@ const (
 	TypeBigSend       = 0x0C
 	// TypeFragData is delivered only after IP reassembly, so its arrival IS the measurement.
 	TypeFragData = 0x0D
+	// TypeThroughputData is one packet of a sustained-rate downstream run.
+	TypeThroughputData = 0x0E
 )
 
 type Server struct {

server/internal/session/session.go

@@ -40,6 +40,7 @@ type Session struct {
 	packetsSeen uint64
 	udpObs      []UDPObservation // ring, newest last, cap obsCap
 	connectBack []ConnectBackResult
+	throughput  []ThroughputReport
 }
 
 const obsCap = 4096
@@ -54,6 +55,18 @@ type UDPObservation struct {
 	Type  uint8  `json:"type"`
 }
 
+// ThroughputReport is the server's own account of a sustained send: what it managed to put on
+// the wire, and what stopped it. The client needs this to interpret its own count — the gap
+// between the two IS the loss, and without the sender's number a receiver can only guess.
+type ThroughputReport struct {
+	ActionID   string `json:"action_id"`
+	Packets    int    `json:"packets"`
+	Bytes      int64  `json:"bytes"`
+	DurationMs int64  `json:"duration_ms"`
+	Kbps       int    `json:"kbps"`
+	LimitedBy  string `json:"limited_by"`
+}
+
 // ConnectBackResult records one connect-back action outcome.
 type ConnectBackResult struct {
 	ActionID string  `json:"action_id"`
@@ -87,6 +100,27 @@ func (s *Session) Observations() (packetsSeen uint64, udp []UDPObservation, cb [
 		append([]ConnectBackResult(nil), s.connectBack...)
 }
 
+// RecordThroughput stores the server's account of one sustained send.
+//
+// Kept as a per-action summary rather than per-packet records: a ten-second run at 50 Mbps is
+// half a million packets, and holding one struct each would turn a measurement into a memory
+// exhaustion. The client has the per-packet view; the server only needs to say how many it sent.
+func (s *Session) RecordThroughput(actionID string, packets int, bytes, durationMs int64, kbps int, limitedBy string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.throughput = append(s.throughput, ThroughputReport{
+		ActionID: actionID, Packets: packets, Bytes: bytes,
+		DurationMs: durationMs, Kbps: kbps, LimitedBy: limitedBy,
+	})
+}
+
+// ThroughputReports returns the server's account of every sustained send in this session.
+func (s *Session) ThroughputReports() []ThroughputReport {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]ThroughputReport(nil), s.throughput...)
+}
+
 // DataSource returns the last verified data-plane source (invalid when the
 // session has not sent data-plane traffic yet).
 func (s *Session) DataSource() netip.AddrPort {