diff --git a/echolot-app/core-engine/src/test/kotlin/app/echo_lot/engine/LiveThroughputTest.kt b/echolot-app/core-engine/src/test/kotlin/app/echo_lot/engine/LiveThroughputTest.kt new file mode 100644 index 0000000..1d95479 --- /dev/null +++ b/echolot-app/core-engine/src/test/kotlin/app/echo_lot/engine/LiveThroughputTest.kt @@ -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") + } +} diff --git a/server/internal/dataplane/throughput.go b/server/internal/dataplane/throughput.go index a8ce169..c1f1f26 100644 --- a/server/internal/dataplane/throughput.go +++ b/server/internal/dataplane/throughput.go @@ -138,19 +138,24 @@ func (s *Server) DownThroughput( 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 { + 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 { diff --git a/server/internal/session/grant.go b/server/internal/session/grant.go index 7aebda6..12ae9b0 100644 --- a/server/internal/session/grant.go +++ b/server/internal/session/grant.go @@ -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 // than sleeping, so callers stay in control of pacing). 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() defer g.mu.Unlock() if time.Now().After(g.ExpiresAt) { - return false + return false, RefusalExpired } 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() - allowed := float64(g.MaxKbps) * 125 * elapsed // kbps -> bytes/s is kbps*1000/8 = kbps*125 - if elapsed > 0.05 && float64(g.sentBytes+int64(n)) > allowed { - return false + allowed := burstBytes(bytesPerSec) + bytesPerSec*elapsed + if float64(g.sentBytes+int64(n)) > allowed { + return false, RefusalRate } 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. diff --git a/server/internal/session/grant_test.go b/server/internal/session/grant_test.go index e472283..2e0d98f 100644 --- a/server/internal/session/grant_test.go +++ b/server/internal/session/grant_test.go @@ -92,3 +92,67 @@ func TestGrantEnforcesRate(t *testing.T) { 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") + } +}