felhom-agent/internal/linetest/linetest.go

// Package linetest is the weekly upload line-speed test (R-1247; operator ruling 2026-10-24 07:26).
//
// Why: the controller's upload watch and qBittorrent's default limit (50 % of the line, D41) need the household's
// upload line. The passive reading — the nightly off-site push — is only a low floor (2026-10-24 night: demo-hp „at
// least 10 Mbit/s"; this test read 90.7 on the same line). So once a week the HOST (the tunnel lives on the host)
// runs a 10-second iperf3 upload to ep0's tunnel-only iperf3 server (felhom.eu scripts/ep0-iperf3). No file is left
// anywhere and nothing counts against the household's storage.
//
// The rule, each part pinned by a test in linetest_test.go:
//   - Weekly: due from Tuesday 10:00 local until one measurement succeeded since then (a box that was off catches up
//     on a later day, inside the window).
//   - Daytime only: a test starts only 10:00–15:59 local — never in the night chain's window (01:00–06:59), and each
//     host starts from its own minute (SlotMinute, spread over 10:00–13:59) so the fleet does not test at once.
//   - Never beside a backup, a restore-test, an OS update or a trim: the test holds the host-wide one-heavy-op gate.
//   - Only after QuietMinutes of quiet uplink: every one-minute transmit rate of the host's physical uplink below
//     QuietMbps — so neither the family's upload nor a restore is measured against itself.
//   - Up to RunsPerMeasurement runs of TestTime each; the highest counts. A run is bounded by RunTimeout.
//   - „Server busy" (another box is testing) → tried again at the next tick the same day; at most MaxTriesPerDay.
//   - No tunnel (no operational wg-felhom with a server address) → no test and no value; the controller then keeps its
//     passive floor and sets no qBittorrent limit.
//   - The last result is persisted, so a restart neither loses it nor re-tests.
//   - A box whose torrent app seeds all day is never quiet for 10 minutes (R-1250). For it the controller holds the
//     app's upload limit very low and asks for a test on request (RunNow): every rule above except the 10 quiet
//     minutes still holds, the quiet is checked over RequestQuietWindow instead, and the runs are RequestRunGap
//     apart, so the whole request ends in about 55 s.
package linetest

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"log/slog"
	"net/netip"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
)

const (
	Weekday            = time.Tuesday
	StartHour          = 10 // first eligible local hour (inclusive)
	EndHour            = 16 // first NOT-eligible local hour: the last start is 15:59
	SlotSpreadMinutes  = 240
	QuietMinutes       = 10
	QuietMbps          = 2.0
	RunsPerMeasurement = 3
	TestTime           = 10 * time.Second
	Streams            = 4
	RunTimeout         = 25 * time.Second
	RunGap             = 30 * time.Second
	MaxTriesPerDay     = 6
	ServerPort         = 5201
	TickInterval       = time.Minute
	FirstTickDelay     = 5 * time.Minute
	GateName           = "line-test"

	// A test on request (R-1250): the controller holds the app's limit while it waits, so it is short.
	RequestSettle      = 5 * time.Second  // the held app's last packets leave before the quiet is read
	RequestQuietWindow = 10 * time.Second // the uplink must send below QuietMbps over this window
	RequestRunGap      = 5 * time.Second
)

// ScheduleText is carried on the host report.
const ScheduleText = "weekly, due Tuesday from 10:00 host-local time; starts 10:00-15:59 after 10 quiet minutes; never beside a backup, restore-test or update"

// Gate is the host-wide one-heavy-operation gate (*backup.InFlight).
type Gate interface {
	TryAcquire(what string) (release func(), busy string, ok bool)
}

// Target answers the iperf3 server's tunnel address, ok=false when the tunnel is not operational.
type Target func() (netip.Addr, bool)

// Runner runs one upload test (Client in production).
type Runner interface {
	Run(ctx context.Context, addr string) (Result, error)
}

// UplinkTx reads the host's physical-uplink transmit byte counter.
type UplinkTx func() (uint64, error)

// State is what is persisted (line-test.json).
type State struct {
	LastOKAt      time.Time `json:"last_ok_at,omitempty"`
	Mbps          float64   `json:"mbps,omitempty"`
	RunsMbps      []float64 `json:"runs_mbps,omitempty"`
	Server        string    `json:"server,omitempty"`
	LastAttemptAt time.Time `json:"last_attempt_at,omitempty"`
	LastError     string    `json:"last_error,omitempty"`
	TriesDay      string    `json:"tries_day,omitempty"` // local date of TriesToday
	TriesToday    int       `json:"tries_today,omitempty"`
}

type txSample struct {
	at time.Time
	tx uint64
}

// Tester is the weekly job.
type Tester struct {
	runner    Runner
	target    Target
	gate      Gate
	uplink    UplinkTx
	statePath string
	slot      int // minutes after StartHour:00 this host may first start
	logger    *slog.Logger
	loc       *time.Location
	now       func() time.Time
	sleep     func(context.Context, time.Duration) bool

	mu      sync.Mutex
	st      State
	samples []txSample
}

// New builds the job and loads the persisted state (missing = empty; corrupt = logged, empty — the cost is one extra
// test, never a lost value elsewhere).
func New(runner Runner, target Target, gate Gate, uplink UplinkTx, hostID, statePath string, logger *slog.Logger) *Tester {
	if logger == nil {
		logger = slog.Default()
	}
	t := &Tester{runner: runner, target: target, gate: gate, uplink: uplink, statePath: statePath,
		slot: SlotMinute(hostID), logger: logger, loc: time.Local, now: time.Now, sleep: sleepCtx}
	t.load()
	return t
}

func sleepCtx(ctx context.Context, d time.Duration) bool {
	tm := time.NewTimer(d)
	defer tm.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-tm.C:
		return true
	}
}

// SlotMinute spreads hosts over the first SlotSpreadMinutes of the window, stable per host id.
func SlotMinute(hostID string) int {
	h := fnv.New32a()
	h.Write([]byte(hostID))
	return int(h.Sum32() % SlotSpreadMinutes)
}

func (t *Tester) load() {
	data, err := os.ReadFile(t.statePath)
	if err != nil {
		if !os.IsNotExist(err) {
			t.logger.Warn("linetest: state read failed — starting empty", "path", t.statePath, "err", err)
		}
		return
	}
	if err := json.Unmarshal(data, &t.st); err != nil {
		t.logger.Warn("linetest: state file corrupt — starting empty", "path", t.statePath, "err", err)
		t.st = State{}
	}
}

func (t *Tester) saveLocked() error {
	data, err := json.MarshalIndent(t.st, "", "  ")
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(t.statePath), 0o755); err != nil {
		return err
	}
	tmp := t.statePath + ".tmp"
	if err := os.WriteFile(tmp, data, 0o600); err != nil {
		os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, t.statePath)
}

// EligibleAt reports whether a test may START at local time lt for a host whose slot is slot minutes after
// StartHour:00 on the due day. On a catch-up day (anchorDay=false) the slot is not waited for again.
func EligibleAt(lt time.Time, slot int, anchorDay bool) bool {
	h := lt.Hour()
	if h < StartHour || h >= EndHour {
		return false
	}
	if anchorDay {
		mins := (h-StartHour)*60 + lt.Minute()
		return mins >= slot
	}
	return true
}

// weekAnchor is the most recent Tuesday StartHour:00 at or before lt.
func weekAnchor(lt time.Time) time.Time {
	daysBack := (int(lt.Weekday()) - int(Weekday) + 7) % 7
	d := lt.AddDate(0, 0, -daysBack)
	a := time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
	if a.After(lt) {
		d = d.AddDate(0, 0, -7)
		a = time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
	}
	return a
}

// Due reports whether a measurement is due at lt given the state.
func Due(st State, lt time.Time) bool {
	return st.LastOKAt.IsZero() || st.LastOKAt.In(lt.Location()).Before(weekAnchor(lt))
}

// sample records the uplink counter (once a minute) and keeps QuietMinutes+1 samples.
func (t *Tester) sample() {
	if t.uplink == nil {
		return
	}
	tx, err := t.uplink()
	if err != nil {
		t.logger.Debug("linetest: uplink counter unreadable", "err", err)
		t.mu.Lock()
		t.samples = nil // a gap is not quiet
		t.mu.Unlock()
		return
	}
	t.mu.Lock()
	t.samples = append(t.samples, txSample{at: t.now(), tx: tx})
	if len(t.samples) > QuietMinutes+1 {
		t.samples = t.samples[len(t.samples)-(QuietMinutes+1):]
	}
	t.mu.Unlock()
}

// quiet reports whether the last QuietMinutes minutes each sent below QuietMbps, and the highest minute seen.
func quiet(samples []txSample) (bool, float64) {
	if len(samples) < QuietMinutes+1 {
		return false, 0
	}
	worst := 0.0
	for i := 1; i < len(samples); i++ {
		dt := samples[i].at.Sub(samples[i-1].at).Seconds()
		if dt <= 0 || dt > 3*TickInterval.Seconds() || samples[i].tx < samples[i-1].tx {
			return false, worst // a gap or a counter reset is not proof of quiet
		}
		mbps := float64(samples[i].tx-samples[i-1].tx) * 8 / dt / 1e6
		if mbps > worst {
			worst = mbps
		}
	}
	return worst < QuietMbps, worst
}

// Run ticks every TickInterval until ctx ends. It never returns an error: a failed test is a reported fact.
func (t *Tester) Run(ctx context.Context) {
	t.logger.Info("linetest: weekly upload line test starting", "schedule", ScheduleText, "slot_minute", t.slot)
	if !t.sleep(ctx, FirstTickDelay) {
		return
	}
	for {
		t.sample()
		t.Tick(ctx)
		if !t.sleep(ctx, TickInterval) {
			return
		}
	}
}

// Refusal is why a test on request did not run (RunNow). Code is stable for the controller.
type Refusal struct {
	Code string // not-due | outside-window | tries-used | no-tunnel | heavy-op | not-quiet
	Msg  string
}

func (r *Refusal) Error() string { return "linetest: " + r.Code + ": " + r.Msg }

// ready applies every rule except the quiet one: due this week, inside the window (the slot on the due day), tries
// left today, the tunnel up. It answers the server address and the local day.
func (t *Tester) ready() (addr, day string, refusal *Refusal) {
	lt := t.now().In(t.loc)
	t.mu.Lock()
	st := t.st
	t.mu.Unlock()
	if !Due(st, lt) {
		return "", "", &Refusal{"not-due", "measured this week"}
	}
	anchor := weekAnchor(lt)
	anchorDay := lt.YearDay() == anchor.YearDay() && lt.Year() == anchor.Year()
	if !EligibleAt(lt, t.slot, anchorDay) {
		return "", "", &Refusal{"outside-window", "a test starts 10:00-15:59 local, from this host's slot on the due day"}
	}
	day = lt.Format("2006-01-02")
	if st.TriesDay == day && st.TriesToday >= MaxTriesPerDay {
		return "", "", &Refusal{"tries-used", fmt.Sprintf("%d tries today", st.TriesToday)}
	}
	ip, ok := t.target()
	if !ok {
		return "", "", &Refusal{"no-tunnel", "no operational tunnel"}
	}
	return netip.AddrPortFrom(ip, ServerPort).String(), day, nil
}

// Tick is one look.
func (t *Tester) Tick(ctx context.Context) {
	addr, day, refusal := t.ready()
	if refusal != nil {
		if refusal.Code == "no-tunnel" {
			t.logger.Debug("linetest: no operational tunnel — no test")
		}
		return
	}
	t.mu.Lock()
	samples := append([]txSample(nil), t.samples...)
	t.mu.Unlock()
	if q, worst := quiet(samples); !q {
		t.logger.Debug("linetest: uplink not quiet — waiting", "worst_minute_mbps", fmt.Sprintf("%.1f", worst), "samples", len(samples))
		return
	}
	release, busy, ok := t.gate.TryAcquire(GateName)
	if !ok {
		t.logger.Info("linetest: deferred — a heavy operation is in flight", "busy", busy)
		return
	}
	defer release()
	t.measure(ctx, addr, day, RunGap)
}

// WaitingForQuiet reports that a measurement is due and could start now but the uplink has not been quiet for
// QuietMinutes — the case a seeding torrent app makes permanent (R-1250). The controller reads it from
// GET /host/line-test and only then holds its app's limit and calls RunNow.
func (t *Tester) WaitingForQuiet() bool {
	if _, _, refusal := t.ready(); refusal != nil {
		return false
	}
	t.mu.Lock()
	samples := append([]txSample(nil), t.samples...)
	t.mu.Unlock()
	if len(samples) < QuietMinutes+1 {
		return false // not yet watched for 10 minutes (a fresh start): not known to be busy, so no hold
	}
	q, _ := quiet(samples)
	return !q
}

// RunNow is a test on request (POST /host/line-test/run, R-1250): every rule of the weekly test except the 10 quiet
// minutes, which the caller replaces by holding its own upload down. The uplink must still be quiet over
// RequestQuietWindow (after RequestSettle), so a request never measures the family's upload against itself. A
// *Refusal says why nothing ran; any other error is a failed measurement (the same as a failed weekly attempt).
func (t *Tester) RunNow(ctx context.Context) (mbps float64, runs []float64, err error) {
	addr, day, refusal := t.ready()
	if refusal != nil {
		return 0, nil, refusal
	}
	release, busy, ok := t.gate.TryAcquire(GateName)
	if !ok {
		return 0, nil, &Refusal{"heavy-op", busy}
	}
	defer release()
	if t.uplink == nil {
		return 0, nil, &Refusal{"not-quiet", "no uplink counter"}
	}
	if !t.sleep(ctx, RequestSettle) {
		return 0, nil, ctx.Err()
	}
	tx0, err0 := t.uplink()
	at0 := t.now()
	if !t.sleep(ctx, RequestQuietWindow) {
		return 0, nil, ctx.Err()
	}
	tx1, err1 := t.uplink()
	if err0 != nil || err1 != nil || tx1 < tx0 {
		return 0, nil, &Refusal{"not-quiet", "uplink counter unreadable or reset"}
	}
	secs := t.now().Sub(at0).Seconds()
	if secs <= 0 {
		secs = RequestQuietWindow.Seconds()
	}
	if m := float64(tx1-tx0) * 8 / secs / 1e6; m >= QuietMbps {
		t.logger.Info("linetest: test on request refused — the uplink is not quiet", "mbps", fmt.Sprintf("%.1f", m))
		return 0, nil, &Refusal{"not-quiet", fmt.Sprintf("the uplink sent %.1f Mbit/s while held", m)}
	}
	t.logger.Info("linetest: test on request (a held torrent app, R-1250)", "server", addr)
	if err := t.measure(ctx, addr, day, RequestRunGap); err != nil {
		return 0, nil, err
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.st.Mbps, append([]float64(nil), t.st.RunsMbps...), nil
}

// measure runs the runs, gap apart, and stores the result. It returns nil only when a value was stored.
func (t *Tester) measure(ctx context.Context, addr, day string, gap time.Duration) error {
	start := t.now()
	var runs []float64
	var lastErr error
	busy := false
	for i := 0; i < RunsPerMeasurement; i++ {
		if i > 0 && !t.sleep(ctx, gap) {
			break
		}
		rctx, cancel := context.WithTimeout(ctx, RunTimeout)
		r, err := t.runner.Run(rctx, addr)
		cancel()
		if errors.Is(err, ErrBusy) {
			busy, lastErr = true, err
			break
		}
		if err != nil {
			lastErr = err
			t.logger.Warn("linetest: run failed", "run", i+1, "server", addr, "err", err)
			continue
		}
		runs = append(runs, r.Mbps)
		t.logger.Info(fmt.Sprintf("linetest: run %d: %.1f Mbit/s", i+1, r.Mbps), "server", addr,
			"server_bytes", r.ServerBytes, "server_seconds", r.ServerSeconds)
	}

	t.mu.Lock()
	defer t.mu.Unlock()
	if t.st.TriesDay != day {
		t.st.TriesDay, t.st.TriesToday = day, 0
	}
	t.st.TriesToday++
	t.st.LastAttemptAt = start.UTC()
	if len(runs) > 0 {
		best := runs[0]
		for _, r := range runs {
			if r > best {
				best = r
			}
		}
		t.st.LastOKAt, t.st.Mbps, t.st.RunsMbps, t.st.Server, t.st.LastError = start.UTC(), best, runs, addr, ""
		t.logger.Info(fmt.Sprintf("linetest: upload line measured: %.1f Mbit/s (best of %d)", best, len(runs)),
			"server", addr, "runs", len(runs), "duration_s", t.now().Sub(start).Seconds())
	} else {
		msg := "no run succeeded"
		if lastErr != nil {
			msg = lastErr.Error()
		}
		if len(msg) > 300 {
			msg = msg[:300]
		}
		t.st.LastError = msg
		if busy {
			t.logger.Info("linetest: the test server is busy — trying again later today", "server", addr, "tries_today", t.st.TriesToday)
		} else {
			t.logger.Warn("linetest: measurement failed", "server", addr, "tries_today", t.st.TriesToday, "err", msg)
		}
	}
	if err := t.saveLocked(); err != nil {
		t.logger.Warn("linetest: state save failed — the result will not survive a restart", "path", t.statePath, "err", err)
	}
	if len(runs) > 0 {
		return nil
	}
	if lastErr == nil {
		lastErr = errors.New("linetest: no run succeeded")
	}
	return lastErr
}

// Value is the measured line (ok=false: never measured).
func (t *Tester) Value() (mbps float64, at time.Time, ok bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.st.LastOKAt.IsZero() || t.st.Mbps <= 0 {
		return 0, time.Time{}, false
	}
	return t.st.Mbps, t.st.LastOKAt, true
}

// LineTestStatus implements hub.LineTestReporter: a pure read of the persisted state.
func (t *Tester) LineTestStatus(context.Context) *hub.LineTestStatus {
	t.mu.Lock()
	defer t.mu.Unlock()
	out := &hub.LineTestStatus{Schedule: ScheduleText, Mbps: t.st.Mbps, RunsMbps: t.st.RunsMbps, Server: t.st.Server,
		LastError: t.st.LastError}
	if !t.st.LastOKAt.IsZero() {
		out.LastOKAt = t.st.LastOKAt.UTC().Format(time.RFC3339)
	}
	if !t.st.LastAttemptAt.IsZero() {
		out.LastAttemptAt = t.st.LastAttemptAt.UTC().Format(time.RFC3339)
	}
	return out
}

// ── the host's physical uplink counter ──────────────────────────────────────────────────────────────────────────

// SysUplinkTx reads the transmit bytes of the host's physical uplink under root (normally "/"): the interface of the
// default route in /proc/net/route, and — when it is a bridge (vmbr0), as on every Proxmox host — the sum of its
// PHYSICAL member ports (those with a `device` link). The bridge's own counter does not see the guests' bridged
// traffic, which is the traffic this check exists for.
func SysUplinkTx(root string) UplinkTx {
	return func() (uint64, error) {
		ifc, err := defaultRouteIface(filepath.Join(root, "proc/net/route"))
		if err != nil {
			return 0, err
		}
		sys := filepath.Join(root, "sys/class/net")
		ports := []string{ifc}
		if ents, err := os.ReadDir(filepath.Join(sys, ifc, "brif")); err == nil {
			ports = ports[:0]
			for _, e := range ents {
				if _, err := os.Stat(filepath.Join(sys, e.Name(), "device")); err == nil {
					ports = append(ports, e.Name())
				}
			}
			if len(ports) == 0 {
				return 0, fmt.Errorf("linetest: bridge %s has no physical port", ifc)
			}
		}
		var sum uint64
		for _, p := range ports {
			b, err := os.ReadFile(filepath.Join(sys, p, "statistics/tx_bytes"))
			if err != nil {
				return 0, err
			}
			n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64)
			if err != nil {
				return 0, err
			}
			sum += n
		}
		return sum, nil
	}
}

func defaultRouteIface(path string) (string, error) {
	b, err := os.ReadFile(path)
	if err != nil {
		return "", err
	}
	for _, line := range strings.Split(string(b), "\n")[1:] {
		f := strings.Fields(line)
		if len(f) >= 8 && f[1] == "00000000" && f[7] == "00000000" {
			return f[0], nil
		}
	}
	return "", errors.New("linetest: no default route")
}