echolot/server/internal/dataplane/throughput.go

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