// 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 }