From f7e1dcbb276806e6bd38fd33e210c7952e489300 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Sat, 24 Oct 2026 08:59:42 +0200 Subject: [PATCH] =?UTF-8?q?R-1247:=20the=20weekly=20upload=20line=20test?= =?UTF-8?q?=20=E2=80=94=20the=20agent's=20own=20iperf3=20client=20to=20ep0?= =?UTF-8?q?'s=20tunnel-only=20server,=20Tuesday=2010:00-15:59=20after=2010?= =?UTF-8?q?=20quiet=20minutes,=20under=20the=20heavy-op=20gate,=20best=20o?= =?UTF-8?q?f=203;=20line=5Ftest=20stanza=20+=20GET=20/host/line-test;=2015?= =?UTF-8?q?=20tests=20red-proved;=20unreleased=20(next=20v0.164.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 29 ++ CONTEXT.md | 6 + REUSE.md | 1 + cmd/felhom-agent/linetest_wiring_test.go | 64 ++++ cmd/felhom-agent/main.go | 19 + internal/config/config.go | 11 + internal/hub/collect.go | 17 + internal/hub/report.go | 18 + internal/linetest/iperf.go | 294 +++++++++++++++ internal/linetest/linetest.go | 439 ++++++++++++++++++++++ internal/linetest/linetest_test.go | 433 +++++++++++++++++++++ internal/localapi/linetest.go | 37 ++ internal/localapi/linetest_test.go | 51 +++ internal/localapi/server.go | 4 + internal/wgtunnel/linetest_target_test.go | 32 ++ internal/wgtunnel/manager.go | 20 +- 16 files changed, 1474 insertions(+), 1 deletion(-) create mode 100644 cmd/felhom-agent/linetest_wiring_test.go create mode 100644 internal/linetest/iperf.go create mode 100644 internal/linetest/linetest.go create mode 100644 internal/linetest/linetest_test.go create mode 100644 internal/localapi/linetest.go create mode 100644 internal/localapi/linetest_test.go create mode 100644 internal/wgtunnel/linetest_target_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a93ca5..8a3e516 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,32 @@ +## Unreleased (next: v0.164.0; 2026-10-24 attended release run) — the weekly upload line test (R-1247) + +Operator ruling 2026-10-24 07:26: once a week the box's HOST measures its upload line with a 10-second iperf3 upload to an +iperf3 server on ep0 that answers **only inside the WireGuard tunnel** (felhom.eu `scripts/ep0-iperf3/`; installed the same +day). The agent speaks the iperf3 protocol itself (operator's pick, 2026-10-24: nothing is installed on any host). +**No sudoers or config-bundle change.** **MinController: none** — the controller (≥ 0.342.0) reads the new route and +treats a 404 as „not measured". + +- **`internal/linetest`** — `Client` (the iperf3 3.x client protocol: cookie, params, N data streams, TEST_END, results + exchange; the value is the SERVER's received bytes) and `Tester` (the schedule): due weekly from Tuesday 10:00 local, + starts 10:00–15:59 only (never in the night chain), from a per-host slot spread over 10:00–13:59; only after 10 quiet + minutes (every one-minute transmit rate of the host's physical uplink below 2 Mbit/s); under the one-heavy-op gate + (`line-test`); 3 runs of 10 s with 4 streams, 30 s apart, the highest counts; each run bounded by 25 s; „server busy" → + tried again at the next tick the same day, at most 6 tries a day; no operational tunnel → no test, no value. The last + result is persisted (`/line-test.json`). +- **Report:** a `line_test` stanza (`schedule`, `last_ok_at`, `mbps`, `runs_mbps`, `server`, `last_attempt_at`, + `last_error`). **Local API:** `GET /host/line-test` → `{present, mbps, measured_at, method:"iperf3-tunnel"}`. +- `wgtunnel.Manager.LineTestTarget()` — the desired block's `pbs_tunnel_ip`, only while the tunnel is operational + (cached, no exec). Config `line_test.disable` (default on). +- **Proven live** (felhom.eu `audits/line-test-2026-10-24/`): a probe built from this package on demo-hp's host read + **90.7 Mbit/s** in 10.4 s (the push floor said „at least 10"); a second test 2 s later → „server busy" in 0.1 s while the + first went on undisturbed (90.6). +- Tests, each red-proved by a mutation (15): `TestClientSpeaksTheProtocolAndCountsTheServersBytes` (against a fake + iperf3 server), `TestClientBusyServerIsErrBusy`, `TestClientHonoursTheDeadlineAgainstAStalledServer`, + `TestEligibleWindowNeverInTheNightChain`, `TestNeverTestsInTheNightOrBeforeTuesday`, `TestBusyServerIsRetriedLaterTheSameDay`, + `TestTriesPerDayAreBounded`, `TestNoTunnelNoTest`, `TestBusyUplinkOrHeavyOpWaits`, `TestTickMeasuresTheHighestOfThreeRuns`, + `TestStateSurvivesARestartAndReportsIt`, `TestUplinkCountsTheBridgesPhysicalPortsOnly`, `TestLineTestRoute`, + `TestLineTestTargetOnlyWhileOperational`, `TestMainWiresTheLineTest`. + ## v0.163.0 — the tunnel heals itself after a clock that ran ahead (D30, R-1204); root SSH from the LAN by key only (D4, R-982) (2026-10-23) — RELEASED on the operator's yes (day brief); the vouch is the operator's Binary sha256 `561b5316cd511de9fa9d8b8be4553da7add684f715d3bea533ca330ff7f3763c`; config bundle sha256 `3afb88d9fbbc4adc16083fa52a98cd4974583c6378a77a1a6e0ed8ae6d1b80c8`. Built from `1109092` (CI 2246 green). Both sections below are in it. diff --git a/CONTEXT.md b/CONTEXT.md index df3dd5a..efee6c0 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -1,5 +1,11 @@ # CONTEXT — felhom-agent working state +> **2026-10-24 (attended release run) — v0.164.0: the weekly upload line test (R-1247; operator ruling 07:26).** +> `internal/linetest`: the agent's own iperf3 client (operator's pick — nothing installed on hosts) to ep0's tunnel-only +> server `10.77.0.1:5201`, Tuesday 10:00-15:59, after 10 quiet minutes, under the heavy-op gate, best of 3; `line_test` +> stanza + `GET /host/line-test` (the controller's qBittorrent limit, D41 = 50 %). demo-hp measured 90.7 Mbit/s. Release +> needs the operator's vouch. + > **2026-10-22/23 night — D30 on `main` (`ae69ad4`), unreleased; the D4 patches v2 ready.** Operator ruling D30 > (2026-10-22): a handshake stamped in the future for 15 min → a new WG key pair (`wgtunnel/rekey.go`; old key kept as > `wg/private.key.prev`), registered on `POST /hosts//wg` (the hub re-keys in place; its full-list push swaps the diff --git a/REUSE.md b/REUSE.md index b01147e..d33180a 100644 --- a/REUSE.md +++ b/REUSE.md @@ -171,6 +171,7 @@ | `localapi.GuestAPI` / `BackupService` / `BackupStore` / `TokenAuthority` | internal/localapi/server.go | `*proxmox.Client`, `*backup.BackupRunner`, `*backup.Store`, `*TokenStore` | `fakeGuests`/`fakeBackups`/`fakeStore` internal/localapi/server_test.go | | `backup.InFlight` | internal/backup/inflight.go | `TryAcquire(what) (release, busy, ok)` / `Busy()` | THE host-wide "one heavy guest operation at a time" gate — shared by the local-API backup path and the restore-test scheduler (R-85) | A **LINK** guard, not a lock one: the scratch VMID never touches the live guest's vzdump lock, but an offsite restore PULLS multi-GB over the tunnel a backup PUSHES one. Callers **DEFER, never cancel** — a deferred restore-test costs coverage, a cancelled backup costs the backup. A nil gate is ungated (pre-R-85 callers). | | `fstrim.Trimmer` (R-444) | internal/fstrim/fstrim.go | `New(runner, guests, gate, statePath, logger)` / `Pass(ctx)` / `GuestDiskTrimStatus(ctx)` / `ParseTrimmed(out)` | the weekly `pct fstrim ` of owned running guests (Wednesday from 10:00 local, starts 10:00-20:59 only), under `backup.InFlight`; last result per guest persisted and reported as `guest_disk_trim` | A busy gate DEFERS to the next hourly tick, never waits; a failed trim retries at most `MaxAttemptsPerWeek`; the report reads the persisted record, it never runs pct | +| `linetest.Tester` + `linetest.Client` (R-1247) | internal/linetest/linetest.go · internal/linetest/iperf.go | `New(runner, target, gate, uplink, hostID, statePath, logger)` / `Tick(ctx)` / `Value()` / `LineTestStatus(ctx)`; `Client{Streams, Time}.Run(ctx, addr)` | the weekly upload line test: a 10-second iperf3 upload (the agent speaks the iperf3 protocol — nothing installed on the host) to ep0's tunnel-only server (`wgtunnel.Manager.LineTestTarget`), Tuesday 10:00-15:59 local from a per-host slot, after 10 quiet minutes of the PHYSICAL uplink (`SysUplinkTx`: the bridge's ports with a `device` link — the bridge counter misses bridged guest traffic), under `backup.InFlight`; best of 3; reported as `line_test`, served as GET /host/line-test | **The value is the SERVER's received bytes**, never the client's written ones (the send buffer). `Run` refuses a context without a deadline. iperf3 numbers streams 1, 3, 4, 5 — the server matches results by those ids. „server busy" (ACCESS_DENIED) stops the measurement and is retried at the next tick, at most `MaxTriesPerDay` | | `capability` store-grant probe (`storeGrantStatuses` / `storeGrantVerdict` / `Client.Permissions`) | cmd/felhom-agent/main.go, internal/proxmox/query.go | *"may the agent READ this backup tier?"*, one `capability.Status` per configured tier | R-185. **Never infer permission from an empty content listing** — `{"data":[]}` is what a FORBIDDEN tier and a NEWBORN tier both return, and that ambiguity hid an unreadable host tier on both demo boxes. Ask `/access/permissions` **as the agent's own token** (root always says yes). **The ungranted answer is not empty and not a 403** — it carries the privileges inherited from the box-wide `/` grant, so test for **`Datastore.AllocateSpace`** specifically; path-presence or `Datastore.Audit` reports a blinded storage healthy. Probed set comes from `BackupTiers()`, never a fixed list. Critical except the `local` fallback. Composes AROUND the sudo prober (the `poolReadStatus` precedent); `Status`'s wire shape is untouched so the hub alert is free. Unreachable PVE ⇒ degraded, never ok. | | `backup.RestoreTestState` | internal/backup/restoretest_state.go | `RecordSuccess(target,archive,tier,verified,t)` / `ProvenArchive(target)` / `ProvenRestoreTests(ctx)` / `LastSuccess(target)` / `OldestFirst(targets)` | Per-tier restore-test PROOF state, persisted (atomic tmp+rename) — **which archive** was proven, and when (R-86) | **Credit ONLY on success** — a permanently failing tier must keep sorting first, or it looks freshly proven and stops being retried. Ties break on target id: without it, two tiers proven in the same second rotate by Go's randomised map order. **This one NEEDS persistence unlike R-84** — R-84 had ground truth to consult (the archive is still on the storage); a restore-test destroys its scratch and leaves no artifact. **R-86: the ARCHIVE is the state, the time is metadata** — a time alone cannot answer "have we proven THIS archive", which is the due-check's whole question. A pre-R-86 file (bare RFC3339 per target) keeps its time and yields NO proven archive, so each tier is due once after the upgrade; reading a legacy time as proof of the current archive would invent a guarantee. **R-189: it is also the REPORTABLE half of the restore-test signal.** The in-memory `backup.Store` holds only this process's latest run, and under per-archive due-ness the agent will not re-test a proven archive — so a proof lost to a restart is not repeated for a whole archive generation (observed live: a passing 14.5 GB offsite restore reached no host-report). `ProvenRestoreTests` renders the stored proofs as `hub.RestoreTest` entries and the collector merges them; a record missing the archive or the tier is NOT emitted, because an unproven tier reading as proven is worse than the defect. **Only successes are stored, deliberately:** a success suppresses future work, a failure causes it. | | `backup.BackupSuccessState` | internal/backup/backup_state.go | `RecordBackupSuccess(target, b)` / `LastKnownSuccess(target, vmid)` | Newest SUCCESSFUL backup per tier+guest, persisted (atomic tmp+rename) — the due-check's fallback when the tier's storage cannot be read after a restart (R-894) | **Read ONLY when the storage cannot be read** — a storage that answers is the ground truth (R-84), and an archive absent there must make the tier due even when this file remembers one. A saved copy older than the cadence still reads due. Only successes are written. | diff --git a/cmd/felhom-agent/linetest_wiring_test.go b/cmd/felhom-agent/linetest_wiring_test.go new file mode 100644 index 0000000..997e46f --- /dev/null +++ b/cmd/felhom-agent/linetest_wiring_test.go @@ -0,0 +1,64 @@ +package main + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" +) + +// R-1247: the weekly line test has the fstrim shape (component + reporter seam + goroutine) plus a local-API reader, +// so its wiring is asserted from the AST. It must share the ONE heavy-op gate (heavyOps) — or it could run beside a +// backup — and take its server from the tunnel manager, never a literal address. +func TestMainWiresTheLineTest(t *testing.T) { + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, "main.go", nil, 0) + if err != nil { + t.Fatalf("parse main.go: %v", err) + } + var withGate, withTunnelTarget, reporterWired, localAPIWired, started bool + ast.Inspect(f, func(n ast.Node) bool { + switch node := n.(type) { + case *ast.CallExpr: + if fn, ok := node.Fun.(*ast.SelectorExpr); ok { + switch fn.Sel.Name { + case "New": + if pkg, ok := fn.X.(*ast.Ident); ok && pkg.Name == "linetest" && len(node.Args) >= 3 { + if id, ok := node.Args[2].(*ast.Ident); ok && id.Name == "heavyOps" { + withGate = true + } + if sel, ok := node.Args[1].(*ast.SelectorExpr); ok && sel.Sel.Name == "LineTestTarget" { + withTunnelTarget = true + } + } + case "SetLineTestReporter": + reporterWired = true + case "SetLineTest": + localAPIWired = true + } + } + case *ast.GoStmt: + if sel, ok := node.Call.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" { + if id, ok := sel.X.(*ast.Ident); ok && id.Name == "lineTest" { + started = true + } + } + } + return true + }) + if !withGate { + t.Error("main.go never calls linetest.New(..., heavyOps, ...) — no line test, or one outside the heavy-op gate") + } + if !withTunnelTarget { + t.Error("linetest.New does not take its server from wgMgr.LineTestTarget") + } + if !reporterWired { + t.Error("main.go never calls collector.SetLineTestReporter — the line_test stanza never reaches the hub") + } + if !localAPIWired { + t.Error("main.go never calls localSrv.SetLineTest — the controller's GET /host/line-test always reads present:false") + } + if !started { + t.Error("main.go never starts the line test with `go lineTest.Run(ctx)`") + } +} diff --git a/cmd/felhom-agent/main.go b/cmd/felhom-agent/main.go index 12dbaa1..516a46f 100644 --- a/cmd/felhom-agent/main.go +++ b/cmd/felhom-agent/main.go @@ -43,6 +43,7 @@ import ( "gitea.dooplex.hu/admin/felhom-agent/internal/guestnet" "gitea.dooplex.hu/admin/felhom-agent/internal/hub" "gitea.dooplex.hu/admin/felhom-agent/internal/lanresolver" + "gitea.dooplex.hu/admin/felhom-agent/internal/linetest" "gitea.dooplex.hu/admin/felhom-agent/internal/localapi" applog "gitea.dooplex.hu/admin/felhom-agent/internal/log" "gitea.dooplex.hu/admin/felhom-agent/internal/mgmtplane" @@ -1641,6 +1642,24 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int } else { logger.Info("fstrim: weekly guest disk trim disabled by config (disk_trim.disable)") } + // R-1247 (operator ruling 2026-10-24 07:26): the weekly upload line test — a 10-second iperf3 upload from this host + // to ep0's tunnel-only server, Tuesday 10:00-15:59 local, after 10 quiet minutes, under the one-heavy-op gate. Only + // with the offsite tunnel (the server is reachable nowhere else). Not part of the errc fan-out. + switch { + case !cfg.LineTest.Enabled(): + logger.Info("linetest: weekly upload line test disabled by config (line_test.disable)") + case wgMgr == nil: + logger.Info("linetest: weekly upload line test off — no offsite tunnel on this host") + default: + lineTest := linetest.New(linetest.Client{Streams: linetest.Streams, Time: linetest.TestTime}, wgMgr.LineTestTarget, + heavyOps, linetest.SysUplinkTx("/"), cfg.Hub.HostID, + filepath.Join(cfg.OOB.WithDefaults().StateDir, "line-test.json"), logger) + collector.SetLineTestReporter(lineTest) + if localSrv != nil { + localSrv.SetLineTest(lineTest) + } + go lineTest.Run(ctx) + } if lanLoop != nil { lanServers = 1 go func() { errc <- lanLoop.Run(ctx) }() diff --git a/internal/config/config.go b/internal/config/config.go index 4afb776..171877e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -34,6 +34,7 @@ type Config struct { WGTunnel WGTunnelConfig `json:"wg_tunnel"` GuestNet GuestNetConfig `json:"guest_net"` DiskTrim DiskTrimConfig `json:"disk_trim"` + LineTest LineTestConfig `json:"line_test"` OOB OOBConfig `json:"oob"` SelfUpdate SelfUpdateConfig `json:"selfupdate"` LogLevel string `json:"log_level"` // debug|info|warn|error (default info) @@ -150,6 +151,16 @@ type DiskTrimConfig struct { // Enabled reports whether the weekly trim should run. func (d DiskTrimConfig) Enabled() bool { return !d.Disable } +// LineTestConfig configures the R-1247 weekly upload line test (internal/linetest). DEFAULT-ON: it reaches only ep0's +// tunnel-only iperf3 server, so it runs only on a box that already has an operational offsite tunnel (operator +// ruling 2026-10-24 07:26: every box with a tunnel measures its line weekly). +type LineTestConfig struct { + Disable bool `json:"disable"` +} + +// Enabled reports whether the weekly line test should run. +func (l LineTestConfig) Enabled() bool { return !l.Disable } + // GuestNetConfig configures the R-54 guest-network watchdog (internal/guestnet). // // **This is the repo's first DEFAULT-ON feature gate, and the inversion is deliberate.** Every other diff --git a/internal/hub/collect.go b/internal/hub/collect.go index 4a6ee69..325c2ba 100644 --- a/internal/hub/collect.go +++ b/internal/hub/collect.go @@ -104,6 +104,12 @@ type GuestDiskTrimReporter interface { GuestDiskTrimStatus(ctx context.Context) *GuestDiskTrimStatus } +// LineTestReporter is the R-1247 seam the weekly line test plugs into (hub does not import linetest). nil → no +// line_test stanza. +type LineTestReporter interface { + LineTestStatus(ctx context.Context) *LineTestStatus +} + // Collector builds a HostReport from read-only sources. All deps are behind narrow // interfaces for unit testing. type Collector struct { @@ -124,6 +130,7 @@ type Collector struct { ctrlSup ControllerSupervisorReporter // R-523: in-guest controller supervisor (nil → stanza omitted) guestNet GuestNetReporter // R-54: per-guest network watchdog (nil → stanza omitted) diskTrim GuestDiskTrimReporter // R-444: weekly guest disk trim (nil → stanza omitted) + lineTest LineTestReporter // R-1247: weekly upload line test (nil → stanza omitted) deliberateOff func() *DeliberateOff // R-998: household power-off record of this boot (nil → omitted) foreignKey ForeignKeyArchiveReporter // R-366 slice 2 (nil → omitted) selfUpdate SelfUpdateReporter // D1: agent self-update pending status (nil → false) @@ -259,6 +266,12 @@ func (c *Collector) SetGuestDiskTrimReporter(r GuestDiskTrimReporter) *Collector return c } +// SetLineTestReporter wires the R-1247 weekly line test as a report source (nil-safe → stanza omitted). +func (c *Collector) SetLineTestReporter(r LineTestReporter) *Collector { + c.lineTest = r + return c +} + // SetDeliberateOffSource wires the R-998 power-off record (poweroff.State.Current, adapted). func (c *Collector) SetDeliberateOffSource(f func() *DeliberateOff) *Collector { c.deliberateOff = f @@ -427,6 +440,10 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) { if c.diskTrim != nil { report.GuestDiskTrim = c.diskTrim.GuestDiskTrimStatus(ctx) } + // R-1247: the last weekly line test (nil reporter = not wired → stanza omitted). + if c.lineTest != nil { + report.LineTest = c.lineTest.LineTestStatus(ctx) + } // R-366 slice 2: archives the restore-test skipped as another key's (nil → not evaluated yet → omitted). if c.foreignKey != nil { report.ForeignKeyArchives = c.foreignKey.ForeignKeyArchives(ctx) diff --git a/internal/hub/report.go b/internal/hub/report.go index 60f36c8..4344ca0 100644 --- a/internal/hub/report.go +++ b/internal/hub/report.go @@ -129,6 +129,11 @@ type HostReport struct { // wired; an empty `guests` list means the job runs and no guest has been trimmed yet. No secret. GuestDiskTrim *GuestDiskTrimStatus `json:"guest_disk_trim,omitempty"` + // LineTest is the weekly upload line test stanza (R-1247; operator ruling 2026-10-24): the schedule, the last + // measured upload line (best of the runs, Mbit/s, as ep0's iperf3 server counted it) and the last attempt. Present + // only when the job is wired. No secret. + LineTest *LineTestStatus `json:"line_test,omitempty"` + // ForeignKeyArchives (R-366 slice 2, `09` §3 decision 168): per backup tier, the whole-guest archives the // restore-test SKIPPED because they were written with another key (an earlier install of this box). This box // cannot open them; the hub turns a CHANGE of this list into one operator event. Absent = not evaluated yet since @@ -232,6 +237,19 @@ type GuestDiskTrimStatus struct { Guests []GuestDiskTrim `json:"guests,omitempty"` } +// LineTestStatus is the R-1247 weekly upload line test stanza. `last_ok_at` + `mbps` are the last SUCCESSFUL +// measurement ("" / 0 = never); `last_attempt_at` + `last_error` the last attempt, whatever its verdict — never read +// the attempt time as a result. +type LineTestStatus struct { + Schedule string `json:"schedule"` + LastOKAt string `json:"last_ok_at,omitempty"` + Mbps float64 `json:"mbps,omitempty"` + RunsMbps []float64 `json:"runs_mbps,omitempty"` + Server string `json:"server,omitempty"` + LastAttemptAt string `json:"last_attempt_at,omitempty"` + LastError string `json:"last_error,omitempty"` +} + // GuestDiskTrim is one guest's LAST trim attempt. `ok` with `last_attempt_at` is the verdict of that attempt — never // read the time alone as success; `last_ok_at` is the last attempt that succeeded ("" = never). `bytes_trimmed` is // the sum of the "(N bytes) trimmed" lines `pct fstrim` printed, over `mounts` mount points. diff --git a/internal/linetest/iperf.go b/internal/linetest/iperf.go new file mode 100644 index 0000000..6faf6d0 --- /dev/null +++ b/internal/linetest/iperf.go @@ -0,0 +1,294 @@ +package linetest + +import ( + "context" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "strconv" + "sync" + "sync/atomic" + "time" +) + +// A minimal iperf3 CLIENT for one TCP upload test (client → server), speaking the iperf3 control protocol (3.x) to the +// stock `iperf3 --server` on ep0 (operator ruling 2026-10-24: the agent speaks iperf3 itself, nothing is installed on +// the host). The number that counts is the SERVER's: the bytes it received, from its results JSON — bytes the client +// handed to its socket include what still sat in the send buffer when the timer ended. +// +// The protocol, as the server drives it (one state byte at a time on the control connection): +// +// client → 37-byte cookie +// server → PARAM_EXCHANGE(9) client → params JSON (4-byte big-endian length + bytes) +// server → CREATE_STREAMS(10) client opens N data connections, each starts with the same cookie +// server → TEST_START(1), TEST_RUNNING(2) client sends for the test time, then → TEST_END(4) +// server → EXCHANGE_RESULTS(13) client → its results JSON; server → its results JSON +// server → DISPLAY_RESULTS(14) client → IPERF_DONE(16) +// +// A server already running a test answers ACCESS_DENIED(-1) and closes: that is ErrBusy (another box is testing; try +// again later the same day). SERVER_ERROR(-3) / SERVER_TERMINATE(-2) end the test with an error. + +const ( + stParamExchange = 9 + stCreateStreams = 10 + stTestStart = 1 + stTestRunning = 2 + stTestEnd = 4 + stExchangeResults = 13 + stDisplayResults = 14 + stIperfDone = 16 + stAccessDenied = -1 + stServerTerminate = -2 + stServerError = -3 + + cookieSize = 37 // 36 characters + NUL + blockSize = 128 * 1024 + maxJSONBytes = 1 << 20 +) + +// ErrBusy: the server is running another client's test. +var ErrBusy = errors.New("iperf3 server busy") + +// Result is one upload test. +type Result struct { + Mbps float64 // the server's received bits per second / 1e6 + ServerBytes int64 // bytes the server received, all streams + ClientBytes int64 // bytes the client wrote (diagnostics only) + ServerSeconds float64 // the server's measured duration +} + +// Dialer opens a TCP connection (net.Dialer in production; a seam for tests). +type Dialer interface { + DialContext(ctx context.Context, network, address string) (net.Conn, error) +} + +// Client runs one test. The whole run is bounded by ctx; Run sets every connection's deadline from it. +type Client struct { + Dialer Dialer + Streams int // parallel streams (-P) + Time time.Duration // test time (-t) +} + +func (c Client) dialer() Dialer { + if c.Dialer != nil { + return c.Dialer + } + return &net.Dialer{Timeout: 5 * time.Second} +} + +func newCookie() ([]byte, error) { + const alphabet = "abcdefghijklmnopqrstuvwxyz234567" + b := make([]byte, cookieSize) + if _, err := rand.Read(b[:cookieSize-1]); err != nil { + return nil, err + } + for i := 0; i < cookieSize-1; i++ { + b[i] = alphabet[int(b[i])%len(alphabet)] + } + b[cookieSize-1] = 0 + return b, nil +} + +func writeJSON(w io.Writer, v any) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(len(data))) + if _, err := w.Write(append(hdr[:], data...)); err != nil { + return err + } + return nil +} + +func readJSON(r io.Reader, v any) error { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return err + } + n := binary.BigEndian.Uint32(hdr[:]) + if n == 0 || n > maxJSONBytes { + return fmt.Errorf("iperf3: results JSON of %d bytes refused", n) + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return err + } + return json.Unmarshal(buf, v) +} + +type streamResult struct { + ID int `json:"id"` + Bytes int64 `json:"bytes"` + Retransmits int64 `json:"retransmits"` + Jitter float64 `json:"jitter"` + Errors int64 `json:"errors"` + OmittedErrors int64 `json:"omitted_errors"` + Packets int64 `json:"packets"` + OmittedPackets int64 `json:"omitted_packets"` + StartTime float64 `json:"start_time"` + EndTime float64 `json:"end_time"` +} + +type results struct { + CPUTotal float64 `json:"cpu_util_total"` + CPUUser float64 `json:"cpu_util_user"` + CPUSystem float64 `json:"cpu_util_system"` + SenderHasRetransmits int `json:"sender_has_retransmits"` + CongestionUsed string `json:"congestion_used,omitempty"` + Streams []streamResult `json:"streams"` +} + +// streamIDs are the ids iperf3 gives its streams: the first is 1, every later one is (streams so far) + 2 — so 1, 3, +// 4, 5 … (iperf_add_stream). The server matches our results by these ids. +func streamIDs(n int) []int { + ids := make([]int, n) + for i := range ids { + if i == 0 { + ids[i] = 1 + } else { + ids[i] = i + 2 + } + } + return ids +} + +// Run performs one upload test to addr ("10.77.0.1:5201"). +func (c Client) Run(ctx context.Context, addr string) (Result, error) { + var res Result + streams, dur := c.Streams, c.Time + if streams < 1 { + streams = 1 + } + if dur <= 0 { + dur = 10 * time.Second + } + deadline, ok := ctx.Deadline() + if !ok { + return res, errors.New("iperf3: Run needs a context with a deadline") + } + cookie, err := newCookie() + if err != nil { + return res, err + } + ctrl, err := c.dialer().DialContext(ctx, "tcp", addr) + if err != nil { + return res, fmt.Errorf("iperf3: connect: %w", err) + } + defer ctrl.Close() + ctrl.SetDeadline(deadline) + if _, err := ctrl.Write(cookie); err != nil { + return res, fmt.Errorf("iperf3: cookie: %w", err) + } + + var data []net.Conn + defer func() { + for _, d := range data { + d.Close() + } + }() + var sent atomic.Int64 + var server results + var wg sync.WaitGroup + for { + var st [1]byte + if _, err := io.ReadFull(ctrl, st[:]); err != nil { + return res, fmt.Errorf("iperf3: control read: %w", err) + } + switch int8(st[0]) { + case stParamExchange: + params := map[string]any{"tcp": true, "omit": 0, "time": int(dur / time.Second), "num": 0, "blockcount": 0, + "parallel": streams, "len": blockSize, "pacing_timer": 1000, "client_version": "3.18"} + if err := writeJSON(ctrl, params); err != nil { + return res, fmt.Errorf("iperf3: params: %w", err) + } + case stCreateStreams: + for i := 0; i < streams; i++ { + d, err := c.dialer().DialContext(ctx, "tcp", addr) + if err != nil { + return res, fmt.Errorf("iperf3: data stream %d: %w", i+1, err) + } + d.SetDeadline(deadline) + data = append(data, d) + if _, err := d.Write(cookie); err != nil { + return res, fmt.Errorf("iperf3: data stream %d cookie: %w", i+1, err) + } + } + case stTestStart: + case stTestRunning: + stopAt := time.Now().Add(dur) + if stopAt.After(deadline) { + return res, errors.New("iperf3: no time left for the test inside the deadline") + } + buf := make([]byte, blockSize) + for _, d := range data { + d.SetWriteDeadline(stopAt) + wg.Add(1) + go func(d net.Conn) { + defer wg.Done() + for { + n, err := d.Write(buf) + sent.Add(int64(n)) + if err != nil { + return + } + } + }(d) + } + wg.Wait() + if _, err := ctrl.Write([]byte{stTestEnd}); err != nil { + return res, fmt.Errorf("iperf3: test end: %w", err) + } + case stExchangeResults: + ids := streamIDs(len(data)) + mine := results{SenderHasRetransmits: -1, CongestionUsed: "cubic"} + per := sent.Load() / int64(max(len(data), 1)) + for _, id := range ids { + mine.Streams = append(mine.Streams, streamResult{ID: id, Bytes: per, Retransmits: -1, EndTime: dur.Seconds()}) + } + if err := writeJSON(ctrl, mine); err != nil { + return res, fmt.Errorf("iperf3: send results: %w", err) + } + if err := readJSON(ctrl, &server); err != nil { + return res, fmt.Errorf("iperf3: read server results: %w", err) + } + case stDisplayResults: + ctrl.Write([]byte{stIperfDone}) + return summarize(server, sent.Load(), dur) + case stAccessDenied: + return res, ErrBusy + case stServerTerminate, stServerError: + var codes [8]byte + io.ReadFull(ctrl, codes[:]) + return res, fmt.Errorf("iperf3: server error (state %d, iperf errno %d, errno %d)", int8(st[0]), + int32(binary.BigEndian.Uint32(codes[:4])), int32(binary.BigEndian.Uint32(codes[4:]))) + default: + return res, fmt.Errorf("iperf3: unexpected state %s", strconv.Itoa(int(int8(st[0])))) + } + } +} + +// summarize turns the server's results into a Result. No stream bytes is an error, never a zero line. +func summarize(server results, clientBytes int64, dur time.Duration) (Result, error) { + res := Result{ClientBytes: clientBytes} + for _, s := range server.Streams { + res.ServerBytes += s.Bytes + if s.EndTime > res.ServerSeconds { + res.ServerSeconds = s.EndTime + } + } + if res.ServerBytes <= 0 { + return res, errors.New("iperf3: the server reported no received bytes") + } + secs := res.ServerSeconds + if secs <= 0 { + secs = dur.Seconds() + } + res.Mbps = float64(res.ServerBytes) * 8 / secs / 1e6 + return res, nil +} diff --git a/internal/linetest/linetest.go b/internal/linetest/linetest.go new file mode 100644 index 0000000..464ca77 --- /dev/null +++ b/internal/linetest/linetest.go @@ -0,0 +1,439 @@ +// 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. +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" +) + +// 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 + } + } +} + +// Tick is one look. +func (t *Tester) Tick(ctx context.Context) { + lt := t.now().In(t.loc) + t.mu.Lock() + st := t.st + samples := append([]txSample(nil), t.samples...) + t.mu.Unlock() + if !Due(st, lt) { + return + } + anchor := weekAnchor(lt) + anchorDay := lt.YearDay() == anchor.YearDay() && lt.Year() == anchor.Year() + if !EligibleAt(lt, t.slot, anchorDay) { + return + } + day := lt.Format("2006-01-02") + if st.TriesDay == day && st.TriesToday >= MaxTriesPerDay { + return + } + ip, ok := t.target() + if !ok { + t.logger.Debug("linetest: no operational tunnel — no test") + return + } + 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, netip.AddrPortFrom(ip, ServerPort).String(), day) +} + +func (t *Tester) measure(ctx context.Context, addr, day string) { + start := t.now() + var runs []float64 + var lastErr error + busy := false + for i := 0; i < RunsPerMeasurement; i++ { + if i > 0 && !t.sleep(ctx, RunGap) { + 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) + } +} + +// 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") +} diff --git a/internal/linetest/linetest_test.go b/internal/linetest/linetest_test.go new file mode 100644 index 0000000..8eea692 --- /dev/null +++ b/internal/linetest/linetest_test.go @@ -0,0 +1,433 @@ +package linetest + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "io" + "net" + "net/netip" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" +) + +// ── a fake iperf3 server (the server half of the protocol, as `iperf3 --server` drives it) ───────────────────────── + +type fakeServer struct { + ln net.Listener + busy bool // answer ACCESS_DENIED + stall bool // accept the cookie, then never answer + recvSeconds float64 // what the server reports as its duration + gotIDs []int // the stream ids the client reported + gotParams map[string]any + mu sync.Mutex + received int64 +} + +func newFakeServer(t *testing.T) *fakeServer { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + s := &fakeServer{ln: ln, recvSeconds: 1} + t.Cleanup(func() { ln.Close() }) + go s.serve() + return s +} + +func (s *fakeServer) addr() string { return s.ln.Addr().String() } + +func wjson(w io.Writer, v any) { + b, _ := json.Marshal(v) + var h [4]byte + binary.BigEndian.PutUint32(h[:], uint32(len(b))) + w.Write(append(h[:], b...)) +} + +func (s *fakeServer) serve() { + ctrl, err := s.ln.Accept() + if err != nil { + return + } + defer ctrl.Close() + cookie := make([]byte, cookieSize) + if _, err := io.ReadFull(ctrl, cookie); err != nil { + return + } + if s.busy { + ctrl.Write([]byte{0xff}) + return + } + if s.stall { + time.Sleep(5 * time.Second) + return + } + ctrl.Write([]byte{stParamExchange}) + var params map[string]any + if readJSON(ctrl, ¶ms) != nil { + return + } + s.gotParams = params + n := int(params["parallel"].(float64)) + ctrl.Write([]byte{stCreateStreams}) + var wg sync.WaitGroup + var conns []net.Conn + for i := 0; i < n; i++ { + d, err := s.ln.Accept() + if err != nil { + return + } + c := make([]byte, cookieSize) + io.ReadFull(d, c) + conns = append(conns, d) + wg.Add(1) + go func(d net.Conn) { + defer wg.Done() + k, _ := io.Copy(io.Discard, d) + s.mu.Lock() + s.received += k + s.mu.Unlock() + }(d) + } + ctrl.Write([]byte{stTestStart, stTestRunning}) + var st [1]byte + if _, err := io.ReadFull(ctrl, st[:]); err != nil || st[0] != stTestEnd { + return + } + ctrl.Write([]byte{stExchangeResults}) + var theirs results + if readJSON(ctrl, &theirs) != nil { + return + } + for _, x := range theirs.Streams { + s.gotIDs = append(s.gotIDs, x.ID) + } + // The server counts what it RECEIVED; report a fixed figure so the test can tell it from the client's own count. + wjson(ctrl, results{Streams: []streamResult{{ID: 1, Bytes: 12_500_000, EndTime: s.recvSeconds}}}) + ctrl.Write([]byte{stDisplayResults}) + io.ReadFull(ctrl, st[:]) + for _, d := range conns { + d.Close() + } + wg.Wait() +} + +func TestClientSpeaksTheProtocolAndCountsTheServersBytes(t *testing.T) { + s := newFakeServer(t) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + r, err := Client{Streams: 4, Time: 300 * time.Millisecond}.Run(ctx, s.addr()) + if err != nil { + t.Fatalf("Run: %v", err) + } + if r.ServerBytes != 12_500_000 || r.Mbps != 100 { + t.Fatalf("result = %+v; want the SERVER's 12.5 MB in 1 s = 100 Mbit/s", r) + } + if want := []int{1, 3, 4, 5}; len(s.gotIDs) != 4 || s.gotIDs[0] != want[0] || s.gotIDs[1] != want[1] || s.gotIDs[3] != want[3] { + t.Fatalf("stream ids sent = %v, want %v (iperf3 numbers its streams 1, 3, 4, 5)", s.gotIDs, want) + } + if s.gotParams["tcp"] != true || s.gotParams["parallel"].(float64) != 4 { + t.Fatalf("params = %v", s.gotParams) + } +} + +func TestClientBusyServerIsErrBusy(t *testing.T) { + s := newFakeServer(t) + s.busy = true + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := (Client{Streams: 4, Time: time.Second}).Run(ctx, s.addr()); !errors.Is(err, ErrBusy) { + t.Fatalf("err = %v, want ErrBusy", err) + } +} + +func TestClientHonoursTheDeadlineAgainstAStalledServer(t *testing.T) { + s := newFakeServer(t) + s.stall = true + ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond) + defer cancel() + start := time.Now() + _, err := Client{Streams: 4, Time: time.Second}.Run(ctx, s.addr()) + if err == nil { + t.Fatal("a stalled server gave no error") + } + if took := time.Since(start); took > 1500*time.Millisecond { + t.Fatalf("Run took %v against a stalled server; the deadline is 400 ms", took) + } +} + +func TestClientRefusesAContextWithoutDeadline(t *testing.T) { + if _, err := (Client{}).Run(context.Background(), "127.0.0.1:1"); err == nil { + t.Fatal("Run without a deadline must refuse (an unbounded test is the thing RunTimeout exists to prevent)") + } +} + +// ── the schedule ──────────────────────────────────────────────────────────────────────────────────────────────── + +func TestEligibleWindowNeverInTheNightChain(t *testing.T) { + loc := time.FixedZone("CET", 3600) + start := time.Date(2026, 10, 19, 0, 0, 0, 0, loc) // a Monday + for m := 0; m < 7*24*60; m++ { + lt := start.Add(time.Duration(m) * time.Minute) + for _, anchorDay := range []bool{true, false} { + if !EligibleAt(lt, 0, anchorDay) { + continue + } + if h := lt.Hour(); h < StartHour || h >= EndHour || (h >= 1 && h <= 6) { + t.Fatalf("eligible at %s — outside 10:00-15:59 or inside the night chain (01:00-06:59)", lt.Format("Mon 15:04")) + } + } + } +} + +func TestSlotDelaysTheFirstStartOnTheDueDayOnly(t *testing.T) { + loc := time.UTC + tue := time.Date(2026, 10, 20, 10, 30, 0, 0, loc) + if EligibleAt(tue, 45, true) { + t.Fatal("10:30 with slot 45 must wait until 10:45 on the due day") + } + if !EligibleAt(tue.Add(15*time.Minute), 45, true) { + t.Fatal("10:45 with slot 45 must be eligible") + } + if !EligibleAt(tue.Add(24*time.Hour), 45, false) { + t.Fatal("a catch-up day does not wait for the slot again") + } + if a, b := SlotMinute("demo-hp-bb76ea"), SlotMinute("demo-felhom-8363b5"); a < 0 || a >= SlotSpreadMinutes || b < 0 || b >= SlotSpreadMinutes { + t.Fatalf("slots out of range: %d %d", a, b) + } +} + +func TestDueWeeklyFromTuesday(t *testing.T) { + loc := time.UTC + mon := time.Date(2026, 10, 19, 12, 0, 0, 0, loc) + tue := time.Date(2026, 10, 20, 12, 0, 0, 0, loc) + if !Due(State{}, mon) { + t.Fatal("never measured is due") + } + st := State{LastOKAt: time.Date(2026, 10, 14, 11, 0, 0, 0, loc)} // last Wednesday + if Due(st, mon) { + t.Fatal("measured after last Tuesday 10:00 is not due on Monday") + } + if !Due(st, tue) { + t.Fatal("from Tuesday 10:00 a new measurement is due") + } +} + +// ── the job ───────────────────────────────────────────────────────────────────────────────────────────────────── + +type fakeRunner struct { + mu sync.Mutex + calls int + out []runOut +} +type runOut struct { + mbps float64 + err error +} + +func (f *fakeRunner) Run(ctx context.Context, addr string) (Result, error) { + f.mu.Lock() + defer f.mu.Unlock() + i := f.calls + f.calls++ + if i >= len(f.out) { + return Result{}, errors.New("no more scripted runs") + } + return Result{Mbps: f.out[i].mbps, ServerBytes: 1}, f.out[i].err +} + +type fakeGate struct{ busy string } + +func (g *fakeGate) TryAcquire(what string) (func(), string, bool) { + if g.busy != "" { + return nil, g.busy, false + } + return func() {}, "", true +} + +type rig struct { + t *Tester + run *fakeRunner + gate *fakeGate + now time.Time + tx uint64 + tunnel bool +} + +// newRig: Tuesday 12:00 UTC (eligible, any slot), tunnel up, 11 quiet samples taken. +func newRig(t *testing.T, outs ...runOut) *rig { + r := &rig{run: &fakeRunner{out: outs}, gate: &fakeGate{}, now: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), tunnel: true} + r.t = New(r.run, func() (netip.Addr, bool) { return netip.MustParseAddr("10.77.0.1"), r.tunnel }, r.gate, + func() (uint64, error) { return r.tx, nil }, "host-x", filepath.Join(t.TempDir(), "line-test.json"), nil) + r.t.loc = time.UTC + r.t.now = func() time.Time { return r.now } + r.t.sleep = func(context.Context, time.Duration) bool { return true } + r.quietMinutes(QuietMinutes + 1) + return r +} + +func (r *rig) minute(bytes uint64) { + r.now = r.now.Add(time.Minute) + r.tx += bytes + r.t.sample() +} + +func (r *rig) quietMinutes(n int) { + for i := 0; i < n; i++ { + r.minute(1_000_000) // 1 MB/min = 0.13 Mbit/s + } +} + +func TestTickMeasuresTheHighestOfThreeRuns(t *testing.T) { + r := newRig(t, runOut{mbps: 88}, runOut{mbps: 91.5}, runOut{mbps: 90}) + r.t.Tick(context.Background()) + mbps, _, ok := r.t.Value() + if !ok || mbps != 91.5 || r.run.calls != RunsPerMeasurement { + t.Fatalf("value %.1f ok=%v calls=%d; want the highest of 3 runs = 91.5", mbps, ok, r.run.calls) + } + r.t.Tick(context.Background()) + if r.run.calls != RunsPerMeasurement { + t.Fatal("a measured week must not test again") + } +} + +func TestBusyServerIsRetriedLaterTheSameDay(t *testing.T) { + r := newRig(t, runOut{err: ErrBusy}, runOut{mbps: 90}, runOut{mbps: 89}, runOut{mbps: 92}) + r.t.Tick(context.Background()) + if _, _, ok := r.t.Value(); ok || r.run.calls != 1 { + t.Fatalf("busy: ok=%v calls=%d; want no value and the measurement stopped after the busy answer", ok, r.run.calls) + } + if r.t.st.TriesToday != 1 || r.t.st.LastError == "" { + t.Fatalf("busy attempt not recorded: %+v", r.t.st) + } + r.quietMinutes(1) + r.t.Tick(context.Background()) + if mbps, _, ok := r.t.Value(); !ok || mbps != 92 { + t.Fatalf("retry later the same day: value %.1f ok=%v", mbps, ok) + } +} + +func TestTriesPerDayAreBounded(t *testing.T) { + outs := make([]runOut, 20) + for i := range outs { + outs[i] = runOut{err: ErrBusy} + } + r := newRig(t, outs...) + for i := 0; i < 10; i++ { + r.quietMinutes(1) + r.t.Tick(context.Background()) + } + if r.run.calls != MaxTriesPerDay { + t.Fatalf("calls = %d in one day, want at most %d", r.run.calls, MaxTriesPerDay) + } +} + +func TestNoTunnelNoTest(t *testing.T) { + r := newRig(t, runOut{mbps: 90}) + r.tunnel = false + r.t.Tick(context.Background()) + if r.run.calls != 0 { + t.Fatal("tested without an operational tunnel") + } + if _, _, ok := r.t.Value(); ok { + t.Fatal("a value without a test") + } +} + +func TestBusyUplinkOrHeavyOpWaits(t *testing.T) { + r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90}) + r.minute(60_000_000) // one minute at 8 Mbit/s: the family is uploading + r.t.Tick(context.Background()) + if r.run.calls != 0 { + t.Fatal("tested while the uplink was busy in the last 10 minutes") + } + r.quietMinutes(QuietMinutes) + r.gate.busy = "backup:felhom-pbs" + r.t.Tick(context.Background()) + if r.run.calls != 0 { + t.Fatal("tested beside a heavy operation") + } + r.gate.busy = "" + r.t.Tick(context.Background()) + if r.run.calls == 0 { + t.Fatal("never tested once the uplink was quiet and the gate free") + } +} + +func TestNeverTestsInTheNightOrBeforeTuesday(t *testing.T) { + r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90}) + r.now = time.Date(2026, 10, 20, 3, 0, 0, 0, time.UTC) // Tuesday 03:00 — the night chain + r.quietMinutes(QuietMinutes + 1) + r.t.Tick(context.Background()) + r.t.st.LastOKAt = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + r.now = time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) // Monday noon, measured last week + r.quietMinutes(QuietMinutes + 1) + r.t.Tick(context.Background()) + if r.run.calls != 0 { + t.Fatalf("tested %d times at night or when not due", r.run.calls) + } +} + +func TestStateSurvivesARestartAndReportsIt(t *testing.T) { + r := newRig(t, runOut{mbps: 90.7}, runOut{err: errors.New("reset")}, runOut{mbps: 80}) + r.t.Tick(context.Background()) + again := New(r.run, nil, r.gate, nil, "host-x", r.t.statePath, nil) + mbps, at, ok := again.Value() + if !ok || mbps != 90.7 || at.IsZero() { + t.Fatalf("after a restart: %.1f %v %v", mbps, at, ok) + } + b, _ := json.Marshal(again.LineTestStatus(context.Background())) + var back hub.LineTestStatus + json.Unmarshal(b, &back) + if back.Mbps != 90.7 || back.LastOKAt == "" || len(back.RunsMbps) != 2 || back.Schedule == "" { + t.Fatalf("report stanza = %s", b) + } +} + +// ── the uplink counter ────────────────────────────────────────────────────────────────────────────────────────── + +func TestUplinkCountsTheBridgesPhysicalPortsOnly(t *testing.T) { + root := t.TempDir() + w := func(p, s string) { + os.MkdirAll(filepath.Dir(filepath.Join(root, p)), 0o755) + os.WriteFile(filepath.Join(root, p), []byte(s), 0o644) + } + w("proc/net/route", "Iface\tDestination\tGateway\tFlags\tRefCnt\tUse\tMetric\tMask\n"+ + "wg-felhom\t0000004D\t00000000\t0001\t0\t0\t0\tFFFFFFFF\n"+ + "vmbr0\t00000000\t0100A8C0\t0003\t0\t0\t0\t00000000\n") + w("sys/class/net/vmbr0/statistics/tx_bytes", "5\n") + w("sys/class/net/vmbr0/brif/enp1s0/.keep", "") + w("sys/class/net/vmbr0/brif/veth9201i0/.keep", "") + w("sys/class/net/enp1s0/device/.keep", "") + w("sys/class/net/enp1s0/statistics/tx_bytes", "1000\n") + w("sys/class/net/veth9201i0/statistics/tx_bytes", "777777\n") + got, err := SysUplinkTx(root)() + if err != nil || got != 1000 { + t.Fatalf("uplink tx = %d, %v; want 1000 (the physical port only — not the bridge, not the guest's veth)", got, err) + } +} + +func TestQuietNeedsTenFullMinutes(t *testing.T) { + base := time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC) + var s []txSample + for i := 0; i <= QuietMinutes; i++ { + s = append(s, txSample{at: base.Add(time.Duration(i) * time.Minute), tx: uint64(i) * 1_000_000}) + } + if q, _ := quiet(s); !q { + t.Fatal("ten quiet minutes must read quiet") + } + if q, _ := quiet(s[1:]); q { + t.Fatal("nine minutes of samples must not read quiet") + } + gap := append([]txSample(nil), s...) + gap[5].at = gap[5].at.Add(-10 * time.Minute) + if q, _ := quiet(gap); q { + t.Fatal("a gap in the samples must not read quiet") + } +} diff --git a/internal/localapi/linetest.go b/internal/localapi/linetest.go new file mode 100644 index 0000000..f057cc4 --- /dev/null +++ b/internal/localapi/linetest.go @@ -0,0 +1,37 @@ +package localapi + +import ( + "net/http" + "time" +) + +// LineTestReader is the R-1247 weekly upload line test (internal/linetest.Tester) as the local API reads it. +type LineTestReader interface { + Value() (mbps float64, at time.Time, ok bool) +} + +// LineTestResponse is GET /host/line-test. Present=false: never measured, or the job is not wired (no tunnel, disabled, +// or an agent before the route answers 404 — the controller reads all three as „unknown"). +type LineTestResponse struct { + Present bool `json:"present"` + Mbps float64 `json:"mbps,omitempty"` + MeasuredAt string `json:"measured_at,omitempty"` // RFC3339 UTC + Method string `json:"method,omitempty"` // "iperf3-tunnel": a 10-second upload to ep0 inside the tunnel +} + +// SetLineTest wires the weekly line test (main.go builds it after the local API server). +func (s *Server) SetLineTest(r LineTestReader) { s.lineTest = r } + +func (s *Server) handleLineTest(w http.ResponseWriter, r *http.Request, vmid int) { + if s.lineTest == nil { + writeOK(w, LineTestResponse{Present: false}) + return + } + mbps, at, ok := s.lineTest.Value() + if !ok { + writeOK(w, LineTestResponse{Present: false}) + return + } + s.logger.Debug("local-api: /host/line-test served", "vmid", vmid, "mbps", mbps, "measured_at", at.UTC().Format(time.RFC3339)) + writeOK(w, LineTestResponse{Present: true, Mbps: mbps, MeasuredAt: at.UTC().Format(time.RFC3339), Method: "iperf3-tunnel"}) +} diff --git a/internal/localapi/linetest_test.go b/internal/localapi/linetest_test.go new file mode 100644 index 0000000..c3fd237 --- /dev/null +++ b/internal/localapi/linetest_test.go @@ -0,0 +1,51 @@ +package localapi + +import ( + "encoding/json" + "log/slog" + "net/http/httptest" + "strings" + "testing" + "time" +) + +type fixedLine struct { + mbps float64 + ok bool +} + +func (f fixedLine) Value() (float64, time.Time, bool) { + return f.mbps, time.Date(2026, 10, 24, 6, 31, 0, 0, time.UTC), f.ok +} + +func lineTestBody(t *testing.T, s *Server) LineTestResponse { + t.Helper() + rec := httptest.NewRecorder() + s.handleLineTest(rec, httptest.NewRequest("GET", "/host/line-test", nil), 9201) + var env struct { + OK bool `json:"ok"` + Data LineTestResponse `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { + t.Fatalf("decode %q: %v", rec.Body.String(), err) + } + return env.Data +} + +// R-1247: the controller sets qBittorrent's limit only from a PRESENT value; never measured and not wired must both +// read present:false, never a zero line. +func TestLineTestRoute(t *testing.T) { + s := &Server{logger: slog.Default()} + if got := lineTestBody(t, s); got.Present { + t.Fatalf("not wired: %+v, want present:false", got) + } + s.SetLineTest(fixedLine{ok: false}) + if got := lineTestBody(t, s); got.Present || got.Mbps != 0 { + t.Fatalf("never measured: %+v, want present:false", got) + } + s.SetLineTest(fixedLine{mbps: 90.7, ok: true}) + got := lineTestBody(t, s) + if !got.Present || got.Mbps != 90.7 || got.Method != "iperf3-tunnel" || !strings.HasPrefix(got.MeasuredAt, "2026-10-24T06:31") { + t.Fatalf("measured: %+v", got) + } +} diff --git a/internal/localapi/server.go b/internal/localapi/server.go index da55318..3f974fa 100644 --- a/internal/localapi/server.go +++ b/internal/localapi/server.go @@ -336,6 +336,7 @@ type Server struct { formatJobs *FormatJobStore // F20-BUG3 detached-format job record (optional) staleLock StaleLockController // F2-b startup stale-lock recovery (optional) partialSweep func(ctx context.Context, vmid int) // R-1006 start-up sweep of dead vzdump leftovers (optional) + lineTest LineTestReader // R-1247 weekly line test (optional; nil → present:false) // guestPower (F-REBOOT) is per-guest start-attempt state for the guest-power watchdog. // Guarded by guestPowerMu in guestpower.go; in-memory on purpose (see guestPowerState). guestPower map[int]guestPowerState @@ -573,6 +574,9 @@ func (s *Server) Handler() http.Handler { // R-856 (`09` §3 decision 143): the host crash guard's record of the last HOST boot — the controller // waits ~15 min with app mails after an unclean one. Read-only; a missing/garbled file = present:false. mux.HandleFunc("GET /host/crash-guard", s.withGuest(s.handleCrashGuard)) + // R-1247: the weekly upload line test's last measurement — the controller's qBittorrent limit and upload watch + // read it. Read-only; never measured (or the job not wired) = present:false. + mux.HandleFunc("GET /host/line-test", s.withGuest(s.handleLineTest)) // Disk management (slice 8C) — self-scoped; format routes through the data-bearing classifier+gate. mux.HandleFunc("GET /disks", s.withGuest(s.handleDisks)) mux.HandleFunc("GET /disks/candidates", s.withGuest(s.handleDiskCandidates)) diff --git a/internal/wgtunnel/linetest_target_test.go b/internal/wgtunnel/linetest_target_test.go new file mode 100644 index 0000000..76fb42e --- /dev/null +++ b/internal/wgtunnel/linetest_target_test.go @@ -0,0 +1,32 @@ +package wgtunnel + +import ( + "context" + "testing" +) + +// R-1247: the line test's server is the desired block's pbs_tunnel_ip, and ONLY while the tunnel is operational +// (registered + unit active) — „no tunnel → no test" rests on this answer. +func TestLineTestTargetOnlyWhileOperational(t *testing.T) { + m, _, active, _ := testManager(t, &fakeHub{}) + ctx := context.Background() + if _, ok := m.LineTestTarget(); ok { + t.Fatal("a target before any Apply") + } + m.Apply(ctx, false, nil) // registers + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) // installs; the unit is not active yet + if _, ok := m.LineTestTarget(); ok { + t.Fatal("a target while the unit is not active") + } + *active = true + m.Apply(ctx, true, testBlock(pub)) + ip, ok := m.LineTestTarget() + if !ok || ip.String() != "10.77.0.1" { + t.Fatalf("operational tunnel: target %v ok=%v, want 10.77.0.1", ip, ok) + } + m.Apply(ctx, true, nil) // revoked + if _, ok := m.LineTestTarget(); ok { + t.Fatal("a target after the hub revoked the tunnel") + } +} diff --git a/internal/wgtunnel/manager.go b/internal/wgtunnel/manager.go index 74d9287..0d8a7ae 100644 --- a/internal/wgtunnel/manager.go +++ b/internal/wgtunnel/manager.go @@ -128,6 +128,7 @@ type Manager struct { type convSnapshot struct { wgBlockDesired bool operational bool + serverIP netip.Addr // the desired block's pbs_tunnel_ip (R-1247 line test target); invalid when none } // TunnelConvergence returns the cached (desired, operational) snapshot without any exec (fast-tick @@ -139,6 +140,17 @@ func (m *Manager) TunnelConvergence() (desired, operational bool) { return m.conv.wgBlockDesired, m.conv.operational } +// LineTestTarget is the R-1247 line test's server: ep0's tunnel address, ok only while the tunnel is operational +// (registered + unit active, as of the last Apply). Cached — no exec. +func (m *Manager) LineTestTarget() (netip.Addr, bool) { + m.mu.Lock() + defer m.mu.Unlock() + if !m.conv.operational || !m.conv.serverIP.IsValid() { + return netip.Addr{}, false + } + return m.conv.serverIP, true +} + // NewManager builds a Manager. stateDir is the agent state dir (default /var/lib/felhom-agent — // note the FELHOM_WG sudoers install entry hard-codes the staged path under it). func NewManager(runner proxmox.Runner, registrar Registrar, stateDir string, logger *slog.Logger) *Manager { @@ -348,7 +360,13 @@ func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WireWiregu operational = m.isActive(ctx) } } - m.conv = convSnapshot{wgBlockDesired: desired, operational: operational} + var server netip.Addr + if desired { + if a, err := netip.ParseAddr(block.Endpoint.PBSTunnelIP); err == nil && a.Is4() { + server = a + } + } + m.conv = convSnapshot{wgBlockDesired: desired, operational: operational, serverIP: server} }() keyExists := false