diff --git a/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt b/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt new file mode 100644 index 0000000..e6b8b83 --- /dev/null +++ b/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> { + 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() + 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, +) diff --git a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt b/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt index 7969abe..5f9c46c 100644 --- a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt +++ b/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" } diff --git a/server/cmd/echolot-server/main.go b/server/cmd/echolot-server/main.go index b150506..5b23c4e 100644 --- a/server/cmd/echolot-server/main.go +++ b/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() diff --git a/server/internal/control/control.go b/server/internal/control/control.go index 6db09d0..9e95186 100644 --- a/server/internal/control/control.go +++ b/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,7 +223,9 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) { "udp": map[string]any{"packets_seen": packetsSeen, "packets": udp}, "tcp": tcp, "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 }) } @@ -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"}) } diff --git a/server/internal/dataplane/throughput.go b/server/internal/dataplane/throughput.go new file mode 100644 index 0000000..a8ce169 --- /dev/null +++ b/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 +} diff --git a/server/internal/dataplane/throughput_test.go b/server/internal/dataplane/throughput_test.go new file mode 100644 index 0000000..4ad0dde --- /dev/null +++ b/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) + } +} diff --git a/server/internal/dataplane/udp.go b/server/internal/dataplane/udp.go index 223ccb0..879ff0e 100644 --- a/server/internal/dataplane/udp.go +++ b/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 { diff --git a/server/internal/session/session.go b/server/internal/session/session.go index db1339b..cbf8046 100644 --- a/server/internal/session/session.go +++ b/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 {