Compare commits

..
Author SHA1 Message Date
mrambossekandClaude Fable 5 3c9af04e6f grant: replace the rate check with a token bucket
server-release / image (push) Successful in 15s
server-test / test (push) Successful in 32s
server-release / release (push) Successful in 33s
The live throughput test found it: a 3-second run delivered 104 packets and
stopped after 50 milliseconds.

The rate check exempted the first 50 ms entirely, meaning to be lenient at
startup. The effect was the opposite. A sender could dump an unbounded burst
into that free window, and the instant the check switched on it compared those
bytes against 50 ms worth of allowance and refused everything until real time
caught up. Every short test passed — downtrain sends 50 packets, big_send seven
— and every sustained send died about fifty milliseconds in.

A token bucket (allowance = burst + rate x elapsed) has no such cliff; it is
smooth from t=0. The burst is 100 ms of the allowed rate, floored at one
ordinary datagram so a single packet is never refused outright. The floor is
deliberately one datagram: at 8 kbps a 64 KB floor would be sixty-four seconds'
worth, which is precisely the instant dump the ceiling exists to prevent. The
existing rate test caught that when I first tried it, and it was right.

Second half of the same bug: callers treated any refusal as terminal. TryAllow
now says why, so a sender can pace through a transient "too fast just now" and
still stop dead on a spent budget or an expired grant.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-01 14:10:30 +02:00
mrambossekandClaude Fable 5 3333788d9e throughput: paced downstream rate, with the qualifier that makes it honest
server-release / image (push) Successful in 15s
server-test / test (push) Successful in 32s
server-release / release (push) Successful in 32s
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>
2026-08-01 14:05:23 +02:00
mrambossekandClaude Fable 5 35744c609e docs: record frag_send and the current testing state
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-01 13:49:43 +02:00
13 changed files with 827 additions and 10 deletions
+46
View File
@@ -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 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. 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.
@@ -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,
)
@@ -44,7 +44,9 @@ class LiveDownstreamTest {
for (t in tests) println("${t.type} status=${t.status} metrics=${t.metrics}") 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}") 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 byType = tests.associateBy { it.type }
val pmtud = assertNotNull(byType[TestType.MTU_PMTUD_DOWN], "no mtu.pmtud_down test") 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") val frag = assertNotNull(byType[TestType.MTU_FRAG_DELIVERY], "no mtu.frag_delivery test")
assertNotNull(frag.metrics?.get("largest_delivered_bytes")) 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") val train = assertNotNull(byType[TestType.TRAIN_UDP_DOWNSTREAM], "no downstream train")
assertNotNull(train.evidence, "a train without columnar evidence is not recomputable") assertNotNull(train.evidence, "a train without columnar evidence is not recomputable")
val received = train.metrics?.get("received")?.toString()?.toIntOrNull() ?: 0 val received = train.metrics?.get("received")?.toString()?.toIntOrNull() ?: 0
@@ -0,0 +1,68 @@
// 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
/**
* Downstream throughput against a LIVE server. Self-skips without ECHOLOT_LIVE_*.
*
* The assertions are about *honesty* rather than speed: a rate is only a measurement if the run
* was ended by the clock and the sender's own count backs it up. A test that just asserted "some
* Mbps arrived" would pass equally well against a broken implementation.
*/
class LiveThroughputTest {
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 measuresDownstreamRateAndSaysWhatLimitedIt() {
if (url == null || pin == null || cred == null || udp == null) {
println("LiveThroughputTest 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 grant binds to the observed source
ThroughputMeasurement(SystemIdSource()).run(
cred, session.sessionId, control, ps, sessionRef = "sess-1",
durationS = 3, kbps = 20_000,
)
}
control.deleteSession(cred, session.sessionId)
val m = assertNotNull(test.metrics).toString()
println("throughput: ${test.status} $m")
for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
assertEquals(TestStatus.OK, test.status, "no throughput traffic arrived: $m")
// The sender's own count must be present — without it, loss cannot be attributed and the
// number is not a measurement.
assertTrue(m.contains("sender_packets"), "no sender report to compare against: $m")
assertTrue(m.contains("limited_by"), "the result must say what ended the run: $m")
val received = Regex(""""received_kbps":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
assertNotNull(received)
assertTrue(received > 0, "measured 0 kbps: $m")
println("received ${received / 1000} Mbit/s")
// A run this short and this far below the ceiling should end on the clock. Anything else
// means the grant was the constraint, and then the rate says nothing about the path.
assertTrue(m.contains(""""limited_by":"duration""""),
"the run did not end on the clock, so the rate measures the server, not the path: $m")
}
}
@@ -38,6 +38,9 @@ object Wire {
*/ */
const val TYPE_FRAG_DATA: Int = 0x0D 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. */ /** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
fun wirePrefix(sessionId: String): ByteArray { fun wirePrefix(sessionId: String): ByteArray {
require(sessionId.length >= 16) { "session id too short" } require(sessionId.length >= 16) { "session id too short" }
+2 -1
View File
@@ -111,7 +111,7 @@ func serve(cfg *config.Config) error {
TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}, 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 — // 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. // a capability we cannot deliver turns a missing feature into a failed measurement.
rawFrag := dataplane.RawFragSupported() rawFrag := dataplane.RawFragSupported()
@@ -167,6 +167,7 @@ func serve(cfg *config.Config) error {
if rawFrag { if rawFrag {
ctl.FragSend = dp.FragSend ctl.FragSend = dp.FragSend
} }
ctl.DownThroughput = dp.DownThroughput
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop() defer stop()
+58 -1
View File
@@ -62,6 +62,8 @@ type Server struct {
// FragSend emits one datagram as hand-built IP fragments in a chosen order (may be nil: // 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). // 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) 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 // 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 // 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. // front and reported as such — the client must not read that as a downstream path limit.
@@ -221,7 +223,9 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
"udp": map[string]any{"packets_seen": packetsSeen, "packets": udp}, "udp": map[string]any{"packets_seen": packetsSeen, "packets": udp},
"tcp": tcp, "tcp": tcp,
"connect_back": cb, "connect_back": cb,
"dns_canary": dnsCanary, // 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 // TODO(spec §6): http echo records
}) })
} }
@@ -246,6 +250,10 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
DF *bool `json:"df"` DF *bool `json:"df"`
Mode string `json:"mode"` Mode string `json:"mode"`
FragBytes int `json:"frag_bytes"` 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 { if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"}) 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}, "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: default:
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"}) writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
} }
+187
View File
@@ -0,0 +1,187 @@
// 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
var seq uint32
for time.Now().Before(deadline) {
ok, why := g.TryAllow(sizeBytes)
if !ok {
if why == session.RefusalRate {
// Transient: the bucket is momentarily empty. Wait for the next slot and carry
// on. Ending the run here would report a rate measured over a fraction of a
// second, which is worse than reporting no rate at all.
res.LimitedBy = "rate"
time.Sleep(time.Duration(perPacketNs))
continue
}
// Terminal: the budget is spent, or the grant expired.
res.LimitedBy = why
break
}
// Reaching here means the run is progressing normally; the clock will end it.
res.LimitedBy = "duration"
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
}
@@ -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)
}
}
+2
View File
@@ -37,6 +37,8 @@ const (
TypeBigSend = 0x0C TypeBigSend = 0x0C
// TypeFragData is delivered only after IP reassembly, so its arrival IS the measurement. // TypeFragData is delivered only after IP reassembly, so its arrival IS the measurement.
TypeFragData = 0x0D TypeFragData = 0x0D
// TypeThroughputData is one packet of a sustained-rate downstream run.
TypeThroughputData = 0x0E
) )
type Server struct { type Server struct {
+49 -7
View File
@@ -74,22 +74,64 @@ func (s *Session) NewGrant(actionID string, wantBytes int64, wantKbps int, lim G
// enforces the byte ceiling, the expiry, and the average rate (by refusing early sends rather // enforces the byte ceiling, the expiry, and the average rate (by refusing early sends rather
// than sleeping, so callers stay in control of pacing). // than sleeping, so callers stay in control of pacing).
func (g *Grant) Allow(n int) bool { func (g *Grant) Allow(n int) bool {
ok, _ := g.TryAllow(n)
return ok
}
// Refusal reasons from TryAllow. The distinction is not cosmetic: "too fast just now" is
// transient and a caller should pace and carry on, while "budget" and "expired" are terminal and
// a caller that keeps trying is only wasting its own run.
const (
RefusalNone = ""
RefusalBudget = "budget"
RefusalExpired = "expired"
RefusalRate = "rate"
)
// TryAllow reports whether n more bytes may be sent now, consuming the budget when they may, and
// says why not when they may not.
//
// The rate limit is a token bucket: allowance = burst + rate x elapsed. An earlier version
// exempted the first 50 ms from the check entirely, meaning to be lenient at startup. The effect
// was the opposite - a sender could dump an unbounded burst into that window, and the moment the
// check switched on it compared those bytes against 50 ms worth of allowance and refused
// everything until real time caught up. A sustained send died about fifty milliseconds in, having
// looked perfectly fine in every short test. A bucket has no such cliff: it is smooth from t=0.
func (g *Grant) TryAllow(n int) (bool, string) {
g.mu.Lock() g.mu.Lock()
defer g.mu.Unlock() defer g.mu.Unlock()
if time.Now().After(g.ExpiresAt) { if time.Now().After(g.ExpiresAt) {
return false return false, RefusalExpired
} }
if g.sentBytes+int64(n) > g.MaxBytes { if g.sentBytes+int64(n) > g.MaxBytes {
return false return false, RefusalBudget
} }
// Average-rate check: bytes allowed so far = kbps/8 * elapsed_seconds. // kbps -> bytes/s is kbps*1000/8 = kbps*125.
bytesPerSec := float64(g.MaxKbps) * 125
elapsed := time.Since(g.started).Seconds() elapsed := time.Since(g.started).Seconds()
allowed := float64(g.MaxKbps) * 125 * elapsed // kbps -> bytes/s is kbps*1000/8 = kbps*125 allowed := burstBytes(bytesPerSec) + bytesPerSec*elapsed
if elapsed > 0.05 && float64(g.sentBytes+int64(n)) > allowed { if float64(g.sentBytes+int64(n)) > allowed {
return false return false, RefusalRate
} }
g.sentBytes += int64(n) g.sentBytes += int64(n)
return true return true, RefusalNone
}
// burstBytes is the bucket's depth: 100 ms of the allowed rate, floored at a single ordinary
// datagram.
//
// The floor exists only so that one packet is never refused outright by a very slow grant — it is
// deliberately one datagram and not more. A generous floor would undo the rate ceiling at low
// rates: at 8 kbps a 64 KB burst is sixty-four seconds' worth, which is exactly the instant dump
// the ceiling is there to prevent. One datagram is 1.5 seconds' worth at that rate and nothing at
// any realistic one.
func burstBytes(bytesPerSec float64) float64 {
const oneDatagram = 1500
b := bytesPerSec * 0.1
if b < oneDatagram {
b = oneDatagram
}
return b
} }
// Sent returns how many bytes this grant has consumed. // Sent returns how many bytes this grant has consumed.
+64
View File
@@ -92,3 +92,67 @@ func TestGrantEnforcesRate(t *testing.T) {
t.Fatalf("rate limit let %d bytes through in ~60ms at 8kbps", sent) t.Fatalf("rate limit let %d bytes through in ~60ms at 8kbps", sent)
} }
} }
// The bug this pins: the rate check used to exempt the first 50 ms entirely, so a sender could
// dump an unbounded burst into that window and then be refused for as long as it took real time
// to catch up. Every short test passed; a sustained send died about fifty milliseconds in. A
// token bucket has no such cliff, and the property that matters is that a sender pacing *at* the
// allowed rate is never refused for long.
func TestSustainedSendAtTheAllowedRateIsNotCutOff(t *testing.T) {
s := sessionWithSource(t)
const kbps = 8000 // 1 MB/s
const packet = 1200 // bytes
g := s.NewGrant("a1", 8<<20, kbps, DefaultGrantLimits)
// Pace at the allowed rate for a short run and count how much got through. A correct
// limiter passes essentially all of it; the old one stopped almost immediately.
perPacket := time.Duration(float64(packet) / (float64(kbps) * 125) * float64(time.Second))
deadline := time.Now().Add(300 * time.Millisecond)
sent, refusals := 0, 0
for time.Now().Before(deadline) {
if ok, why := g.TryAllow(packet); ok {
sent += packet
} else if why == RefusalRate {
refusals++
} else {
t.Fatalf("unexpected terminal refusal %q after %d bytes", why, sent)
}
time.Sleep(perPacket)
}
// 300 ms at 1 MB/s is ~300 KB. Allow generous slack for scheduler granularity, but a run
// that delivered only a few packets means the limiter cut it off.
if sent < 100_000 {
t.Fatalf("a sender pacing at the allowed rate got only %d bytes through in 300ms "+
"(%d rate refusals) — the limiter is cutting off sustained sends", sent, refusals)
}
}
// The other half: a rate refusal must be distinguishable from a spent budget, because one is
// transient and one is terminal, and a caller that cannot tell them apart either gives up early
// or spins forever.
func TestRefusalReasonsAreDistinguishable(t *testing.T) {
s := sessionWithSource(t)
// Budget: tiny ceiling, plenty of rate.
g := s.NewGrant("a1", 1000, 100_000, DefaultGrantLimits)
for i := 0; i < 20; i++ {
g.TryAllow(100)
}
if ok, why := g.TryAllow(100); ok || why != RefusalBudget {
t.Errorf("spent budget reported as ok=%v why=%q, want %q", ok, why, RefusalBudget)
}
// Rate: huge ceiling, minimal rate, so only the bucket can refuse.
g2 := s.NewGrant("a2", 1<<20, 8, DefaultGrantLimits)
sawRate := false
for i := 0; i < 100; i++ {
if ok, why := g2.TryAllow(1000); !ok && why == RefusalRate {
sawRate = true
break
}
}
if !sawRate {
t.Error("a sender far above the rate ceiling never got a rate refusal")
}
}
+34
View File
@@ -40,6 +40,7 @@ type Session struct {
packetsSeen uint64 packetsSeen uint64
udpObs []UDPObservation // ring, newest last, cap obsCap udpObs []UDPObservation // ring, newest last, cap obsCap
connectBack []ConnectBackResult connectBack []ConnectBackResult
throughput []ThroughputReport
} }
const obsCap = 4096 const obsCap = 4096
@@ -54,6 +55,18 @@ type UDPObservation struct {
Type uint8 `json:"type"` 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. // ConnectBackResult records one connect-back action outcome.
type ConnectBackResult struct { type ConnectBackResult struct {
ActionID string `json:"action_id"` ActionID string `json:"action_id"`
@@ -87,6 +100,27 @@ func (s *Session) Observations() (packetsSeen uint64, udp []UDPObservation, cb [
append([]ConnectBackResult(nil), s.connectBack...) 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 // DataSource returns the last verified data-plane source (invalid when the
// session has not sent data-plane traffic yet). // session has not sent data-plane traffic yet).
func (s *Session) DataSource() netip.AddrPort { func (s *Session) DataSource() netip.AddrPort {