From 046b808792788c32f86cbbd71de16cbe77aac31a Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Sat, 24 Oct 2026 13:19:07 +0200 Subject: [PATCH] upwatch: hold a seeding torrent app for one line test, then put its limit back exactly (R-1250) When the agent waits for quiet and qBittorrent or Transmission runs: record the limit first, hold at 1 KiB/s, ask the agent for a test on request, put the old limit back (deferred; a household change during the test stays). Start-up puts back a limit a killed controller left; the 50 % rule skips while a hold record exists. Commit only. --- CHANGELOG.md | 24 ++ controller/cmd/controller/upwatch_wiring.go | 56 +++- .../cmd/controller/upwatch_wiring_test.go | 51 ++- controller/internal/agentapi/client.go | 14 +- controller/internal/agentapi/linetest.go | 31 ++ controller/internal/agentapi/linetest_test.go | 28 ++ controller/internal/upwatch/explain.go | 51 ++- controller/internal/upwatch/linehold.go | 249 +++++++++++++++ controller/internal/upwatch/linehold_test.go | 301 ++++++++++++++++++ controller/internal/upwatch/service.go | 17 +- 10 files changed, 789 insertions(+), 33 deletions(-) create mode 100644 controller/internal/upwatch/linehold.go create mode 100644 controller/internal/upwatch/linehold_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b41e24..1df0761 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,27 @@ +## Unreleased (next: v0.343.0) — the line test for a box that seeds all day (R-1250) (2026-10-24) — committed, not released + +**MinAgent: 0.158.1** (unchanged): an agent before 0.165.0 never reports `waiting_for_quiet`, so nothing is held. Attended +brief 2026-10-24, Part 4 (commit only; the release rides the next normal train). Design: felhom.eu `08` §9. + +- **`internal/upwatch/linehold.go`** — when the agent says a test is due and waits only for quiet, and qBittorrent (or + Transmission) runs: read the app's upload limit exactly as it stored it, write the record + `upwatch.line_test_hold` FIRST, hold the app at 1 KiB/s (Transmission: 1 kB/s, limit on), call the agent's + `POST /host/line-test/run` (about 55 s), then put the old limit back — the household's own included — and clear the + record. The release is deferred (a panic puts it back too); a hold whose write failed is undone before anything else; + a household change made during the test stays; an app that cannot be read keeps the record, and no new hold starts + until it is put back. `startUpwatch` puts back a limit a killed controller left (`RecoverHold`) before any rule runs, + and each 15-minute tick runs the hold before qBittorrent's 50 % rule, which skips while a record exists (it would + read 1 KiB/s as the household's change and stop for good). +- **`agentapi.LineTestRun`** + `LineTestState.WaitingForQuiet`; `postSlow` gives one call its own timeout (2 min) on the + pinned transport — the client's 15 s would cut the 55 s answer. +- **Transmission**: `rpc` (Basic auth + the session-id handshake) shared by `Ask` and the new `ReadLimit`/`WriteLimit` + (`session-get`/`session-set` of `speed-limit-up` + `speed-limit-up-enabled`); a `result` other than success is an error. +- Tests (each seen failing): `TestHold_PutsTheHouseholdsLimitBackExactly` (measured, failed, refused, agent unreachable), + `…PutsBackNoLimit`, `…StartupPutsBackALimitLeftHeld`, `…APanicInTheTestStillPutsTheLimitBack`, + `…AHouseholdChangeDuringTheTestStays`, `…AFailedHoldIsUndoneAndNoTestRuns`, `…AnUnreachableAppKeepsTheRecordUntilItIsBack`, + `…OnlyWhenTheAgentWaitsAndAnAppRuns`, `…QBLimitRuleSkipsAHeldLimit`, `…TheTwoAppsReadAndWriteTheirLimit`, + `TestLineTestRunOutlastsTheClientTimeout`, `TestStartUpwatchRecoversTheHoldFirst`, `TestUpwatchHoldGetsBothAgentCalls`. + ## v0.342.0 (2026-10-24) — RELEASED on the operator's yes (attended release run; image sha256:d002d3b7…, from `42cda07`, CI 2283) — the upload and CPU watch, the tested line and qBittorrent's limit **MinAgent: 0.158.1** (unchanged). **Hub dependency:** the four new event types need hub 0.169.0 (committed the same diff --git a/controller/cmd/controller/upwatch_wiring.go b/controller/cmd/controller/upwatch_wiring.go index 4071eb0..05cf2d4 100644 --- a/controller/cmd/controller/upwatch_wiring.go +++ b/controller/cmd/controller/upwatch_wiring.go @@ -25,16 +25,19 @@ const baselineMinMinutes = 1200 // newUpwatch builds the upload and CPU watch (internal/upwatch; felhom.eu `08` §9). It does not start it. func newUpwatch(sett *settings.Settings, stackMgr *stacks.Manager, backupMgr *backup.Manager, metricsStore *metrics.MetricsStore, - notifier *notify.Notifier, measured func(context.Context) (float64, time.Time, bool), logger *log.Logger) *upwatch.Service { + notifier *notify.Notifier, line agentLine, logger *log.Logger) *upwatch.Service { box := mediawire.DockerBox{} qb := &upwatch.QBittorrent{Box: box} + tr := &upwatch.Transmission{Box: box} svc := &upwatch.Service{Store: sett, QB: qb, QBRunning: func() bool { return box.Running("qbittorrent") }, QBShare: qbLimitShare, - Measured: measured, Log: logger} + Measured: line.Measured, Log: logger, + // R-1250: a box that seeds all day is line-tested by holding its torrent app low for one test on request. + Torrents: []upwatch.Holdable{qb, tr}, TorrentRunning: box.Running, LineWaiting: line.Waiting, LineRun: line.Run} w := &upwatch.Watcher{ Source: &upwatch.DockerAPI{}, Explainers: map[string]upwatch.Explainer{ "qbittorrent": qb, - "transmission": &upwatch.Transmission{Box: box}, + "transmission": tr, "plex": &upwatch.Plex{Token: func() string { return mediawire.PlexToken(box) }, URL: mediawire.PlexURL}, }, Rules: upwatch.DefaultRules(), @@ -58,26 +61,46 @@ func newUpwatch(sett *settings.Settings, stackMgr *stacks.Manager, backupMgr *ba return svc } +// agentLine is the host's line test as the watch uses it; every field nil without a local API. +type agentLine struct { + Measured func(context.Context) (float64, time.Time, bool) + Waiting func(context.Context) bool // waiting_for_quiet (agent ≥ 0.165.0, R-1250) + Run func(context.Context) (upwatch.LineRunResult, error) // POST /host/line-test/run +} + // agentLineTest reads the host's weekly line test through the agent's local API (agent ≥ 0.164.0; an older agent -// answers 404 → not measured). nil without a local API. -func agentLineTest(endpoint, token, fingerprint string, logger *log.Logger) func(context.Context) (float64, time.Time, bool) { +// answers 404 → not measured, and never waiting). Empty without a local API. +func agentLineTest(endpoint, token, fingerprint string, logger *log.Logger) agentLine { if endpoint == "" || token == "" { - return nil + return agentLine{} } ac, err := agentapi.New(endpoint, token, fingerprint) if err != nil { logger.Printf("[WARN] [upwatch] agent client init failed (%v) — no line test; qBittorrent gets no limit", err) - return nil + return agentLine{} } - return func(ctx context.Context) (float64, time.Time, bool) { + read := func(ctx context.Context) (agentapi.LineTestState, error) { cctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() - st, err := ac.LineTest(cctx) - if err != nil { - logger.Printf("[DEBUG] [upwatch] line test read: %v", err) - return 0, time.Time{}, false - } - return st.Mbps, st.MeasuredAtTime(), st.Present + return ac.LineTest(cctx) + } + return agentLine{ + Measured: func(ctx context.Context) (float64, time.Time, bool) { + st, err := read(ctx) + if err != nil { + logger.Printf("[DEBUG] [upwatch] line test read: %v", err) + return 0, time.Time{}, false + } + return st.Mbps, st.MeasuredAtTime(), st.Present + }, + Waiting: func(ctx context.Context) bool { + st, err := read(ctx) + return err == nil && st.WaitingForQuiet + }, + Run: func(ctx context.Context) (upwatch.LineRunResult, error) { + r, err := ac.LineTestRun(ctx) + return upwatch.LineRunResult{OK: r.OK, Mbps: r.Mbps, Refused: r.Refused, Error: r.Error}, err + }, } } @@ -103,8 +126,10 @@ func dailyBaseline(ms *metrics.MetricsStore) func(string, []string) (float64, bo } } -// startUpwatch runs the minute watcher and the 15-minute qBittorrent limit rule until ctx ends. +// startUpwatch runs the minute watcher and the 15-minute qBittorrent limit rule until ctx ends. A torrent limit a +// stopped controller left at the line-test value is put back first (R-1250), before any rule reads it. func startUpwatch(ctx context.Context, svc *upwatch.Service, logger *log.Logger) { + svc.RecoverHold(ctx) svc.RefreshMeasured(ctx) go svc.Watcher.Run(ctx, time.Minute) go func() { @@ -115,6 +140,7 @@ func startUpwatch(ctx context.Context, svc *upwatch.Service, logger *log.Logger) case <-ctx.Done(): return case <-t.C: + svc.SeedLineTestNow(ctx) // R-1250: holds the torrent app only while the agent waits for quiet svc.QBLimitNow(ctx) } } diff --git a/controller/cmd/controller/upwatch_wiring_test.go b/controller/cmd/controller/upwatch_wiring_test.go index be7130f..0fc0c43 100644 --- a/controller/cmd/controller/upwatch_wiring_test.go +++ b/controller/cmd/controller/upwatch_wiring_test.go @@ -3,7 +3,9 @@ package main import ( "io" "log" + "os" "path/filepath" + "strings" "testing" "gitea.dooplex.hu/admin/felhom-controller/internal/notify" @@ -35,7 +37,7 @@ func TestNewUpwatch_ExplainersAndEmit(t *testing.T) { if err != nil { t.Fatal(err) } - svc := newUpwatch(sett, nil, nil, nil, notify.New("", "", "", sett, lg, false), nil, lg) + svc := newUpwatch(sett, nil, nil, nil, notify.New("", "", "", sett, lg, false), agentLine{}, lg) for _, a := range []string{"qbittorrent", "transmission", "plex"} { if _, ok := svc.Watcher.Explainers[a]; !ok { t.Errorf("no explainer for %s", a) @@ -47,7 +49,54 @@ func TestNewUpwatch_ExplainersAndEmit(t *testing.T) { if _, ok := svc.Watcher.Source.(*upwatch.DockerAPI); !ok { t.Fatal("the watch must read the Engine API's exact counters, not docker stats' text") } + // R-1250: both torrent apps can be held, qBittorrent first; without a local API there is no hold at all. + if len(svc.Torrents) != 2 || svc.Torrents[0].Stack() != "qbittorrent" || svc.Torrents[1].Stack() != "transmission" || + svc.TorrentRunning == nil || svc.LineWaiting != nil || svc.LineRun != nil { + t.Fatalf("line-test hold not wired: %d torrents, waiting=%v run=%v", len(svc.Torrents), svc.LineWaiting != nil, svc.LineRun != nil) + } if svc.Watcher.Baseline != nil { t.Fatal("no metrics store, yet a baseline: an app with no explainer would be judged without history") } } + +// R-1250: the start-up check puts a held limit back BEFORE the first rule reads it, and every tick runs the hold +// before the 50 % rule. Asserted on startUpwatch's source order. +// RED-PROOF: move svc.RecoverHold below the ticker, or drop SeedLineTestNow → fails. +func TestStartUpwatchRecoversTheHoldFirst(t *testing.T) { + src, err := os.ReadFile("upwatch_wiring.go") + if err != nil { + t.Fatal(err) + } + body := string(src) + i := strings.Index(body, "func startUpwatch(") + if i < 0 { + t.Fatal("no startUpwatch") + } + body = body[i:] + rec, ticker := strings.Index(body, "svc.RecoverHold(ctx)"), strings.Index(body, "time.NewTicker(") + seed, rule := strings.Index(body, "svc.SeedLineTestNow(ctx)"), strings.Index(body, "svc.QBLimitNow(ctx)") + if rec < 0 || rec > ticker { + t.Fatalf("RecoverHold at %d, the ticker at %d: a held limit must go back before anything else runs", rec, ticker) + } + if seed < 0 || seed > rule { + t.Fatalf("SeedLineTestNow at %d, QBLimitNow at %d: the hold must run before the 50 %% rule", seed, rule) + } +} + +// With a local API, agentLineTest yields all three halves, and newUpwatch hands the hold both agent calls. +// RED-PROOF: drop LineRun from newUpwatch's Service, or Run from agentLineTest → fails. +func TestUpwatchHoldGetsBothAgentCalls(t *testing.T) { + lg := log.New(io.Discard, "", 0) + line := agentLineTest("127.0.0.1:1", "token-canary", strings.Repeat("ab", 32), lg) + if line.Measured == nil || line.Waiting == nil || line.Run == nil { + t.Fatalf("agentLineTest with a local API: measured=%v waiting=%v run=%v", line.Measured != nil, line.Waiting != nil, line.Run != nil) + } + sett, err := settings.Load(filepath.Join(t.TempDir(), "settings.json"), lg) + if err != nil { + t.Fatal(err) + } + svc := newUpwatch(sett, nil, nil, nil, notify.New("", "", "", sett, lg, false), line, lg) + if svc.Measured == nil || svc.LineWaiting == nil || svc.LineRun == nil { + t.Fatalf("newUpwatch dropped an agent call: measured=%v waiting=%v run=%v", svc.Measured != nil, svc.LineWaiting != nil, svc.LineRun != nil) + } +} diff --git a/controller/internal/agentapi/client.go b/controller/internal/agentapi/client.go index b01fbee..d34d03a 100644 --- a/controller/internal/agentapi/client.go +++ b/controller/internal/agentapi/client.go @@ -1291,6 +1291,18 @@ func (c *Client) get(ctx context.Context, path string) (json.RawMessage, error) // post issues an authenticated JSON POST and unwraps the {ok,data,error} envelope. The agent // returns 200 or 202 for accepted requests. func (c *Client) post(ctx context.Context, path string, body any) (json.RawMessage, error) { + return c.postWith(c.hc, ctx, path, body) +} + +// postSlow is post with its own overall timeout, for the few synchronous routes that outlast the client's 15 s +// (POST /host/line-test/run, about 55 s). It shares the pinned transport. +func (c *Client) postSlow(ctx context.Context, path string, body any, timeout time.Duration) (json.RawMessage, error) { + hc := *c.hc + hc.Timeout = timeout + return c.postWith(&hc, ctx, path, body) +} + +func (c *Client) postWith(hc *http.Client, ctx context.Context, path string, body any) (json.RawMessage, error) { buf, err := json.Marshal(body) if err != nil { return nil, err @@ -1302,7 +1314,7 @@ func (c *Client) post(ctx context.Context, path string, body any) (json.RawMessa req.Header.Set("Authorization", "Bearer "+c.token) req.Header.Set("Content-Type", "application/json") start := time.Now() - resp, err := c.hc.Do(req) + resp, err := hc.Do(req) if err != nil { logx.Debugf(c.logger, "[agentapi] POST %s failed after %dms: %v", path, time.Since(start).Milliseconds(), err) return nil, fmt.Errorf("agentapi: POST %s: %w", path, err) diff --git a/controller/internal/agentapi/linetest.go b/controller/internal/agentapi/linetest.go index b8fbb37..7411dfa 100644 --- a/controller/internal/agentapi/linetest.go +++ b/controller/internal/agentapi/linetest.go @@ -16,6 +16,37 @@ type LineTestState struct { Mbps float64 `json:"mbps,omitempty"` MeasuredAt string `json:"measured_at,omitempty"` // RFC3339 UTC Method string `json:"method,omitempty"` + // WaitingForQuiet (agent ≥ 0.165.0, R-1250): a test is due and could start now, but the host's uplink has not been + // quiet for 10 minutes. With a seeding torrent app the controller then holds the app's limit and calls LineTestRun. + WaitingForQuiet bool `json:"waiting_for_quiet,omitempty"` +} + +// LineTestRunResult mirrors POST /host/line-test/run. OK=false with Refused: nothing ran (not-due, outside-window, +// tries-used, no-tunnel, heavy-op, not-quiet, server-busy); OK=false without it: the runs failed. +type LineTestRunResult struct { + OK bool `json:"ok"` + Mbps float64 `json:"mbps,omitempty"` + RunsMbps []float64 `json:"runs_mbps,omitempty"` + Refused string `json:"refused,omitempty"` + Error string `json:"error,omitempty"` +} + +// LineTestRunTimeout bounds the synchronous request: about 55 s on the agent (5 s settle, 10 s quiet check, three +// 10-second runs 5 s apart), each run itself bounded at 25 s. +const LineTestRunTimeout = 2 * time.Minute + +// LineTestRun calls POST /host/line-test/run (R-1250): a test on request while the caller holds its torrent app's +// upload down. An agent before the route answers 404 (*StatusError). +func (c *Client) LineTestRun(ctx context.Context) (LineTestRunResult, error) { + var out LineTestRunResult + body, err := c.postSlow(ctx, "/host/line-test/run", struct{}{}, LineTestRunTimeout) + if err != nil { + return out, err + } + if err := json.Unmarshal(body, &out); err != nil { + return out, fmt.Errorf("agentapi: decode /host/line-test/run: %w", err) + } + return out, nil } // LineTest calls GET /host/line-test. diff --git a/controller/internal/agentapi/linetest_test.go b/controller/internal/agentapi/linetest_test.go index b3e44b4..e2171c8 100644 --- a/controller/internal/agentapi/linetest_test.go +++ b/controller/internal/agentapi/linetest_test.go @@ -34,3 +34,31 @@ func TestLineTestDecodesAndAnOlderAgentIs404(t *testing.T) { t.Fatalf("an older agent must answer a typed 404, got %v", err) } } + +// R-1250: the request decodes its outcome, waiting_for_quiet decodes, and the request outlasts the client's own +// timeout (the agent answers after about 55 s; the client's default is 15 s). +func TestLineTestRunOutlastsTheClientTimeout(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("GET /host/line-test", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"ok":true,"data":{"present":false,"waiting_for_quiet":true}}`)) + }) + mux.HandleFunc("POST /host/line-test/run", func(w http.ResponseWriter, r *http.Request) { + time.Sleep(300 * time.Millisecond) + _, _ = w.Write([]byte(`{"ok":true,"data":{"ok":true,"mbps":93.5,"runs_mbps":[92,93.5,93]}}`)) + }) + s := httptest.NewTLSServer(mux) + defer s.Close() + c := clientFor(t, s, strings.TrimPrefix(s.URL, "https://")) + st, err := c.LineTest(context.Background()) + if err != nil || st.Present || !st.WaitingForQuiet { + t.Fatalf("GET = %+v %v, want present:false waiting_for_quiet:true", st, err) + } + c.hc.Timeout = 100 * time.Millisecond // stands in for the 15 s default against a 55 s answer + if _, err := c.post(context.Background(), "/host/line-test/run", struct{}{}); err == nil { + t.Fatal("the plain POST should time out here — the seam does not test anything") + } + res, err := c.LineTestRun(context.Background()) + if err != nil || !res.OK || res.Mbps != 93.5 || len(res.RunsMbps) != 3 { + t.Fatalf("LineTestRun = %+v %v", res, err) + } +} diff --git a/controller/internal/upwatch/explain.go b/controller/internal/upwatch/explain.go index f10dd99..a9e8790 100644 --- a/controller/internal/upwatch/explain.go +++ b/controller/internal/upwatch/explain.go @@ -211,10 +211,32 @@ type Transmission struct { func (t *Transmission) Stack() string { return "transmission" } func (t *Transmission) Ask(ctx context.Context) (Report, error) { - pw, err := t.Box.Env("transmission", "PASS") + b, err := t.rpc(ctx, `{"method":"session-stats"}`) if err != nil { return Report{}, err } + var st struct { + Arguments struct { + Current struct { + UploadedBytes uint64 `json:"uploadedBytes"` + } `json:"current-stats"` + } `json:"arguments"` + } + if err := json.Unmarshal(b, &st); err != nil { + return Report{}, err + } + return Report{UploadCounter: st.Arguments.Current.UploadedBytes, HasUploadCounter: true}, nil +} + +var rpcMethodRe = regexp.MustCompile(`"method":"([a-z-]+)"`) + +// rpc is one Transmission RPC call: Basic auth with the stack's PASS, and the X-Transmission-Session-Id handshake +// (the first answer is 409 carrying it). It returns the 200 body; `result` other than "success" is an error. +func (t *Transmission) rpc(ctx context.Context, body string) ([]byte, error) { + pw, err := t.Box.Env("transmission", "PASS") + if err != nil { + return nil, err + } base := t.URL if base == "" { base = "http://transmission:9091" @@ -224,32 +246,31 @@ func (t *Transmission) Ask(ctx context.Context) (Report, error) { c = &http.Client{Timeout: 20 * time.Second} } h := map[string]string{"Authorization": "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+pw)), "Host": "transmission:9091"} - body := []byte(`{"method":"session-stats"}`) + method := body + if m := rpcMethodRe.FindStringSubmatch(body); m != nil { + method = m[1] + } for i := 0; i < 2; i++ { - code, b, hd, err := httpDo(ctx, c, "POST", base+"/transmission/rpc", h, body, "application/json") + code, b, hd, err := httpDo(ctx, c, "POST", base+"/transmission/rpc", h, []byte(body), "application/json") if err != nil { - return Report{}, err + return nil, err } if code == 409 { h["X-Transmission-Session-Id"] = hd.Get("X-Transmission-Session-Id") continue } if code != 200 { - return Report{}, fmt.Errorf("transmission session-stats: HTTP %d", code) + return nil, fmt.Errorf("transmission %s: HTTP %d", method, code) } - var st struct { - Arguments struct { - Current struct { - UploadedBytes uint64 `json:"uploadedBytes"` - } `json:"current-stats"` - } `json:"arguments"` + var r struct { + Result string `json:"result"` } - if err := json.Unmarshal(b, &st); err != nil { - return Report{}, err + if err := json.Unmarshal(b, &r); err == nil && r.Result != "" && r.Result != "success" { + return nil, fmt.Errorf("transmission %s: %s", method, r.Result) } - return Report{UploadCounter: st.Arguments.Current.UploadedBytes, HasUploadCounter: true}, nil + return b, nil } - return Report{}, fmt.Errorf("transmission session-stats: no session id") + return nil, fmt.Errorf("transmission %s: no session id", method) } // ── Plex ───────────────────────────────────────────────────────────────────────────────────────────────────────── diff --git a/controller/internal/upwatch/linehold.go b/controller/internal/upwatch/linehold.go new file mode 100644 index 0000000..ac071cc --- /dev/null +++ b/controller/internal/upwatch/linehold.go @@ -0,0 +1,249 @@ +package upwatch + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "gitea.dooplex.hu/admin/felhom-controller/internal/logx" +) + +// ── the line test for a box that seeds all day (R-1250, the brief of 2026-10-24) ───────────────────────────────── +// +// The host's weekly line test (the agent, R-1247) starts only after 10 minutes of quiet uplink. A torrent app that +// seeds all day never gives them, so that box never gets a tested line, and qBittorrent never gets its 50 % limit. +// When the agent says a test is due and waits only for quiet (GET /host/line-test `waiting_for_quiet`), the +// controller: +// +// 1. reads the app's global upload limit exactly as the app stored it, and writes it to settings.json +// (`upwatch.line_test_hold`) BEFORE it touches the app; +// 2. sets the app's upload to HoldValue (1 KiB/s); +// 3. asks the agent for a test on request (POST /host/line-test/run: 5 s settle, 10 s quiet check, three 10-second +// runs 5 s apart — about 55 s); +// 4. puts the old limit back — the household's own limit included, exactly as it was — and clears the record. +// +// Whatever fails after step 1, the limit is put back first. A controller that stops between 1 and 4 finds the record +// at start (RecoverHold) and puts it back then. If the app holds anything but the test value when the limit goes +// back, the household changed it during the test: theirs stays. qBittorrent's 50 % rule (QBLimitNow) does not run +// while a record exists — it would read the test value as the household's change and stop for good. +// +// Pinned by linehold_test.go. + +// HeldLimit is a torrent app's global upload limit as the app stored it. +type HeldLimit struct { + Bytes int64 `json:"bytes,omitempty"` // qBittorrent: B/s, 0 = none + Enabled bool `json:"enabled,omitempty"` // Transmission: speed-limit-up-enabled + KBps int64 `json:"kbps,omitempty"` // Transmission: speed-limit-up, kB/s +} + +// LineHold is the record of a held app (settings.json `upwatch.line_test_hold`). +type LineHold struct { + App string `json:"app"` + Prev HeldLimit `json:"prev"` + Held HeldLimit `json:"held"` // what the app stored for the test value + At string `json:"at"` +} + +// Holdable is one torrent app's global upload limit. +type Holdable interface { + Stack() string + ReadLimit(ctx context.Context) (HeldLimit, error) + // WriteLimit sets the limit and returns what the app stored. + WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit, error) + // HoldValue is the test value: very low, never zero (zero is „no limit"). + HoldValue() HeldLimit +} + +// LineRunResult is the agent's answer to a test on request. +type LineRunResult struct { + OK bool + Mbps float64 + Refused string // the agent's refusal code: nothing ran + Error string +} + +// holdTimeout bounds each app call made to put a limit back; it is not the caller's context, which may be ending. +const holdTimeout = 30 * time.Second + +func (s *Service) hold() *LineHold { + return decode(s.Store.GetUpwatch()).Hold +} + +func (s *Service) saveHold(h *LineHold) error { + return s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) { + st := decode(old) + st.Hold = h + return json.Marshal(st) + }) +} + +func (s *Service) torrentApp(stack string) Holdable { + for _, t := range s.Torrents { + if t.Stack() == stack { + return t + } + } + return nil +} + +func (s *Service) torrentRunning(stack string) bool { + return s.TorrentRunning != nil && s.TorrentRunning(stack) +} + +// RecoverHold puts back a limit a hold left behind (main calls it at start; SeedLineTestNow calls it first). It +// answers true when no record remains. +func (s *Service) RecoverHold(ctx context.Context) bool { + h := s.hold() + if h == nil { + return true + } + app := s.torrentApp(h.App) + if app == nil { + logx.Warnf(s.Log, "[upwatch] a line-test hold names %q, which this controller cannot reach — record dropped", h.App) + return s.saveHold(nil) == nil + } + if !s.torrentRunning(h.App) { + logx.Infof(s.Log, "[upwatch] %s is held at the line-test value but not running — put back when it runs", h.App) + return false + } + logx.Infof(s.Log, "[upwatch] %s was left at the line-test value (since %s) — putting its limit back", h.App, h.At) + return s.release(ctx, app, *h) +} + +// release puts the old limit back when the app still holds the test value, and clears the record. It keeps the +// record (answers false) only when the app cannot be read or written: the next tick tries again. +func (s *Service) release(ctx context.Context, app Holdable, h LineHold) bool { + rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), holdTimeout) + defer cancel() + cur, err := app.ReadLimit(rctx) + if err != nil { + logx.Warnf(s.Log, "[upwatch] %s: reading its limit to put it back: %v — tried again in 15 minutes", h.App, err) + return false + } + switch { + case cur == h.Prev: + logx.Infof(s.Log, "[upwatch] %s's upload limit is as it was (%+v)", h.App, cur) + case cur != h.Held: + logx.Infof(s.Log, "[upwatch] %s's limit changed during the line test (%+v) — the household's: left as it is", h.App, cur) + default: + got, err := app.WriteLimit(rctx, h.Prev) + if err != nil { + logx.Warnf(s.Log, "[upwatch] %s: putting its limit back to %+v: %v — tried again in 15 minutes", h.App, h.Prev, err) + return false + } + if got != h.Prev { + logx.Warnf(s.Log, "[upwatch] %s stored %+v when its limit %+v was put back", h.App, got, h.Prev) + } else { + logx.Infof(s.Log, "[upwatch] %s's upload limit is back as it was (%+v)", h.App, got) + } + } + if err := s.saveHold(nil); err != nil { + logx.Warnf(s.Log, "[upwatch] clearing the line-test hold record: %v", err) + return false + } + return true +} + +// SeedLineTestNow runs the hold once when the agent waits for quiet and a torrent app is running (main calls it every +// 15 minutes, before QBLimitNow). A no-op otherwise. +func (s *Service) SeedLineTestNow(ctx context.Context) { + if !s.RecoverHold(ctx) || s.LineWaiting == nil || s.LineRun == nil { + return + } + var app Holdable + for _, t := range s.Torrents { + if s.torrentRunning(t.Stack()) { + app = t + break + } + } + if app == nil || !s.LineWaiting(ctx) { + return + } + prev, err := app.ReadLimit(ctx) + if err != nil { + logx.Debugf(s.Log, "[upwatch] line test: reading %s's limit: %v — not held", app.Stack(), err) + return + } + h := LineHold{App: app.Stack(), Prev: prev, Held: app.HoldValue(), At: time.Now().UTC().Format(time.RFC3339)} + if err := s.saveHold(&h); err != nil { + logx.Warnf(s.Log, "[upwatch] line test: saving the hold record: %v — %s not held", err, app.Stack()) + return + } + stored, err := app.WriteLimit(ctx, h.Held) + if err != nil { + // The write may have taken although its read-back failed: release compares against the intended value. + logx.Warnf(s.Log, "[upwatch] line test: holding %s: %v — putting it back", app.Stack(), err) + s.release(ctx, app, h) + return + } + if stored != h.Held { + h.Held = stored + if err := s.saveHold(&h); err != nil { + logx.Warnf(s.Log, "[upwatch] line test: saving the hold record: %v — putting %s back", err, app.Stack()) + s.release(ctx, app, h) + return + } + } + logx.Infof(s.Log, "[upwatch] line test: %s held at %+v (was %+v) — asking the agent for a test", app.Stack(), stored, prev) + // Deferred: a panic in the call puts the limit back too. A killed process does not — that is RecoverHold's. + res, err := func() (LineRunResult, error) { + defer s.release(ctx, app, h) + return s.LineRun(ctx) + }() + switch { + case err != nil: + logx.Warnf(s.Log, "[upwatch] line test on request: %v", err) + case res.OK: + logx.Infof(s.Log, "[upwatch] line test on request: %.1f Mbit/s", res.Mbps) + s.RefreshMeasured(ctx) + case res.Refused != "": + logx.Infof(s.Log, "[upwatch] line test on request refused by the agent: %s (%s)", res.Refused, res.Error) + default: + logx.Warnf(s.Log, "[upwatch] line test on request failed: %s", res.Error) + } +} + +// ── the two torrent apps ───────────────────────────────────────────────────────────────────────────────────────── + +func (q *QBittorrent) ReadLimit(ctx context.Context) (HeldLimit, error) { + b, err := q.UploadLimit(ctx, nil) + return HeldLimit{Bytes: b}, err +} + +func (q *QBittorrent) WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit, error) { + b, err := q.SetUploadLimit(ctx, l.Bytes) + return HeldLimit{Bytes: b}, err +} + +// HoldValue: 1 KiB/s, qBittorrent's smallest stored step. +func (q *QBittorrent) HoldValue() HeldLimit { return HeldLimit{Bytes: 1024} } + +func (t *Transmission) ReadLimit(ctx context.Context) (HeldLimit, error) { + var st struct { + Arguments struct { + Up int64 `json:"speed-limit-up"` + Enabled bool `json:"speed-limit-up-enabled"` + } `json:"arguments"` + } + b, err := t.rpc(ctx, `{"method":"session-get","arguments":{"fields":["speed-limit-up","speed-limit-up-enabled"]}}`) + if err != nil { + return HeldLimit{}, err + } + if err := json.Unmarshal(b, &st); err != nil { + return HeldLimit{}, err + } + return HeldLimit{Enabled: st.Arguments.Enabled, KBps: st.Arguments.Up}, nil +} + +func (t *Transmission) WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit, error) { + body := fmt.Sprintf(`{"method":"session-set","arguments":{"speed-limit-up":%d,"speed-limit-up-enabled":%t}}`, l.KBps, l.Enabled) + if _, err := t.rpc(ctx, body); err != nil { + return HeldLimit{}, err + } + return t.ReadLimit(ctx) +} + +// HoldValue: 1 kB/s with the limit switched on. +func (t *Transmission) HoldValue() HeldLimit { return HeldLimit{Enabled: true, KBps: 1} } diff --git a/controller/internal/upwatch/linehold_test.go b/controller/internal/upwatch/linehold_test.go new file mode 100644 index 0000000..9a32bac --- /dev/null +++ b/controller/internal/upwatch/linehold_test.go @@ -0,0 +1,301 @@ +package upwatch + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// fakeTorrent is a torrent app's limit in memory. failWrite fails the next write AFTER storing it (a write that +// took while its read-back failed); failRead fails reads. +type fakeTorrent struct { + limit HeldLimit + writes []HeldLimit + failWrite bool + failRead bool +} + +func (f *fakeTorrent) Stack() string { return "qbittorrent" } +func (f *fakeTorrent) HoldValue() HeldLimit { return HeldLimit{Bytes: 1024} } +func (f *fakeTorrent) ReadLimit(context.Context) (HeldLimit, error) { + if f.failRead { + return HeldLimit{}, errors.New("connection refused") + } + return f.limit, nil +} +func (f *fakeTorrent) WriteLimit(_ context.Context, l HeldLimit) (HeldLimit, error) { + f.limit = l + f.writes = append(f.writes, l) + if f.failWrite { + f.failWrite = false + return HeldLimit{}, errors.New("read-back: EOF") + } + return l, nil +} + +type holdRig struct { + svc *Service + app *fakeTorrent + store *memStore + runs int + waiting bool + running bool + result LineRunResult + runErr error + during func() // runs inside the agent's test +} + +// newHoldRig: qBittorrent running with the household's own 2.5 MB/s limit, the agent waiting for quiet. +func newHoldRig(t *testing.T) *holdRig { + r := &holdRig{app: &fakeTorrent{limit: HeldLimit{Bytes: 2_499_584}}, store: &memStore{}, waiting: true, running: true, + result: LineRunResult{OK: true, Mbps: 93.5}} + r.svc = &Service{Store: r.store, Torrents: []Holdable{r.app}, + TorrentRunning: func(string) bool { return r.running }, + LineWaiting: func(context.Context) bool { return r.waiting }, + LineRun: func(context.Context) (LineRunResult, error) { + r.runs++ + if r.during != nil { + r.during() + } + return r.result, r.runErr + }} + return r +} + +func (r *holdRig) record() *LineHold { return decode(r.store.GetUpwatch()).Hold } + +// R-1250, the brief's three cases: the old limit — the household's own — comes back exactly after a measured test, +// after a failed one, and after a refused one; during the test the app holds the test value and the record exists. +// RED-PROOF: drop the release after LineRun → every case keeps 1024; write the record after the hold → the in-test +// check sees no record. +func TestHold_PutsTheHouseholdsLimitBackExactly(t *testing.T) { + for _, tc := range []struct { + name string + result LineRunResult + err error + }{ + {"measured", LineRunResult{OK: true, Mbps: 93.5}, nil}, + {"failed test", LineRunResult{Error: "dial tcp 10.77.0.1:5201: i/o timeout"}, nil}, + {"refused", LineRunResult{Refused: "not-quiet", Error: "the uplink sent 16.0 Mbit/s while held"}, nil}, + {"agent unreachable", LineRunResult{}, errors.New("agentapi: POST /host/line-test/run: context deadline exceeded")}, + } { + t.Run(tc.name, func(t *testing.T) { + r := newHoldRig(t) + r.result, r.runErr = tc.result, tc.err + r.during = func() { + if r.app.limit != (HeldLimit{Bytes: 1024}) { + t.Fatalf("during the test the app holds %+v, want 1 KiB/s", r.app.limit) + } + h := r.record() + if h == nil || h.Prev != (HeldLimit{Bytes: 2_499_584}) || h.Held != (HeldLimit{Bytes: 1024}) { + t.Fatalf("during the test the record is %+v — a restart now could not put the limit back", h) + } + } + r.svc.SeedLineTestNow(context.Background()) + if r.runs != 1 { + t.Fatalf("agent asked %d times, want 1", r.runs) + } + if r.app.limit != (HeldLimit{Bytes: 2_499_584}) { + t.Fatalf("after the test the app holds %+v, want the household's 2499584 back", r.app.limit) + } + if r.record() != nil { + t.Fatal("the hold record was not cleared") + } + }) + } +} + +// „no limit" (0) is a limit too: it comes back as 0, never as the test value or the 50 % default. +func TestHold_PutsBackNoLimit(t *testing.T) { + r := newHoldRig(t) + r.app.limit = HeldLimit{} + r.svc.SeedLineTestNow(context.Background()) + if r.app.limit != (HeldLimit{}) || len(r.app.writes) != 2 { + t.Fatalf("limit %+v after writes %+v, want 0 after hold+release", r.app.limit, r.app.writes) + } +} + +// A controller restart in the middle: the record is on disk and the app still holds the test value. The start-up +// check puts the old limit back. +// RED-PROOF: make RecoverHold return true without releasing → the app stays at 1024. +func TestHold_StartupPutsBackALimitLeftHeld(t *testing.T) { + r := newHoldRig(t) + // What a killed controller leaves: the record written, the app at the test value (no defer runs on SIGKILL). + _ = r.store.UpdateUpwatch(func(json.RawMessage) (json.RawMessage, error) { + return json.Marshal(State{Hold: &LineHold{App: "qbittorrent", Prev: HeldLimit{Bytes: 2_499_584}, Held: HeldLimit{Bytes: 1024}}}) + }) + r.app.limit = HeldLimit{Bytes: 1024} + restarted := &Service{Store: r.store, Torrents: []Holdable{r.app}, TorrentRunning: func(string) bool { return true }} + if !restarted.RecoverHold(context.Background()) { + t.Fatal("RecoverHold left a record") + } + if r.app.limit != (HeldLimit{Bytes: 2_499_584}) || r.record() != nil { + t.Fatalf("after the restart the app holds %+v (record %+v), want the household's limit back", r.app.limit, r.record()) + } +} + +// A panic inside the agent call still puts the limit back before it travels on. +func TestHold_APanicInTheTestStillPutsTheLimitBack(t *testing.T) { + r := newHoldRig(t) + r.during = func() { panic("agentapi bug") } + func() { + defer func() { _ = recover() }() + r.svc.SeedLineTestNow(context.Background()) + }() + if r.app.limit != (HeldLimit{Bytes: 2_499_584}) || r.record() != nil { + t.Fatalf("after a panic: limit %+v record %+v", r.app.limit, r.record()) + } +} + +// The household changed the limit while it was held: theirs stays. +func TestHold_AHouseholdChangeDuringTheTestStays(t *testing.T) { + r := newHoldRig(t) + r.during = func() { r.app.limit = HeldLimit{Bytes: 5_120_000} } + r.svc.SeedLineTestNow(context.Background()) + if r.app.limit != (HeldLimit{Bytes: 5_120_000}) || r.record() != nil { + t.Fatalf("limit %+v record %+v, want the household's new 5120000 kept and the record cleared", r.app.limit, r.record()) + } +} + +// If anything fails, the limit is put back FIRST: a hold write that took but failed its read-back is undone, and +// the agent is never asked. +func TestHold_AFailedHoldIsUndoneAndNoTestRuns(t *testing.T) { + r := newHoldRig(t) + r.app.failWrite = true + r.svc.SeedLineTestNow(context.Background()) + if r.runs != 0 { + t.Fatal("the agent was asked after the hold failed") + } + if r.app.limit != (HeldLimit{Bytes: 2_499_584}) || r.record() != nil { + t.Fatalf("limit %+v record %+v, want the old limit back and no record", r.app.limit, r.record()) + } +} + +// The app cannot be reached when the limit should go back: the record stays, and the next tick puts it back. While a +// record exists no new hold starts. +func TestHold_AnUnreachableAppKeepsTheRecordUntilItIsBack(t *testing.T) { + r := newHoldRig(t) + r.during = func() { r.app.failRead = true } + r.svc.SeedLineTestNow(context.Background()) + if r.record() == nil || r.app.limit != (HeldLimit{Bytes: 1024}) { + t.Fatalf("limit %+v record %+v, want the record kept while the app is unreachable", r.app.limit, r.record()) + } + r.svc.SeedLineTestNow(context.Background()) + if r.runs != 1 { + t.Fatal("a second hold started while the first was not put back") + } + r.app.failRead = false + r.running = false + if r.svc.RecoverHold(context.Background()) || r.app.limit != (HeldLimit{Bytes: 1024}) { + t.Fatal("a stopped app cannot be put back; the record must wait") + } + r.running, r.during = true, nil + r.svc.SeedLineTestNow(context.Background()) // recovers first; the agent is still waiting, so a new test follows + if r.app.limit != (HeldLimit{Bytes: 2_499_584}) || r.record() != nil || r.runs != 2 { + t.Fatalf("limit %+v record %+v runs %d", r.app.limit, r.record(), r.runs) + } +} + +// No hold without a reason: the agent not waiting, or no torrent app running. +func TestHold_OnlyWhenTheAgentWaitsAndAnAppRuns(t *testing.T) { + r := newHoldRig(t) + r.waiting = false + r.svc.SeedLineTestNow(context.Background()) + r.waiting, r.running = true, false + r.svc.SeedLineTestNow(context.Background()) + if r.runs != 0 || len(r.app.writes) != 0 { + t.Fatalf("runs %d writes %+v, want none", r.runs, r.app.writes) + } +} + +// qBittorrent's 50 % rule must not run on a held limit: it would read 1 KiB/s as the household's change and stop for +// good (QBLimitState.Household). +// RED-PROOF: remove the Hold check in QBLimitNow → Household turns true. +func TestHold_QBLimitRuleSkipsAHeldLimit(t *testing.T) { + logins, expire, limit := 0, false, int64(1024) + srv := fakeQBServer(t, "pw-canary", &logins, &expire, &limit) + defer srv.Close() + st := &memStore{} + _ = st.UpdateUpwatch(func(json.RawMessage) (json.RawMessage, error) { + return json.Marshal(State{QBLimit: &QBLimitState{BoxSetBytes: 5_654 * 1024, LineMbps: 92.6}, + Hold: &LineHold{App: "qbittorrent", Prev: HeldLimit{Bytes: 5_654 * 1024}, Held: HeldLimit{Bytes: 1024}}}) + }) + svc := &Service{Store: st, QB: &QBittorrent{Box: envBox{"qbittorrent/QBT_ADMIN_PASSWORD": "pw-canary"}, URL: srv.URL}, + QBShare: 0.5} + svc.QBLimitNow(context.Background()) + if q := decode(st.GetUpwatch()).QBLimit; q == nil || q.Household || limit != 1024 { + t.Fatalf("QBLimitNow on a held limit: record %+v, limit %d — the box's limit became the household's", q, limit) + } +} + +// The two apps' real APIs: qBittorrent's global limit, Transmission's session-get/session-set with the session-id +// handshake. The test value is never zero (zero is „no limit"). +func TestHold_TheTwoAppsReadAndWriteTheirLimit(t *testing.T) { + logins, expire, limit := 0, false, int64(2_499_584) + srv := fakeQBServer(t, "pw-canary", &logins, &expire, &limit) + defer srv.Close() + qb := &QBittorrent{Box: envBox{"qbittorrent/QBT_ADMIN_PASSWORD": "pw-canary"}, URL: srv.URL} + if l, err := qb.ReadLimit(context.Background()); err != nil || l.Bytes != 2_499_584 { + t.Fatalf("qBittorrent read %+v %v", l, err) + } + if l, err := qb.WriteLimit(context.Background(), qb.HoldValue()); err != nil || l != (HeldLimit{Bytes: 1024}) || limit != 1024 { + t.Fatalf("qBittorrent hold: %+v %v, stored %d", l, err, limit) + } + + tr := struct { + up int64 + enabled bool + }{up: 500, enabled: false} + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Transmission-Session-Id") != "sid-1" { + w.Header().Set("X-Transmission-Session-Id", "sid-1") + w.WriteHeader(409) + return + } + if u, p, ok := r.BasicAuth(); !ok || u != "admin" || p != "tr-canary" { + w.WriteHeader(401) + return + } + b, _ := io.ReadAll(r.Body) + var req struct { + Method string `json:"method"` + Arguments map[string]any `json:"arguments"` + } + _ = json.Unmarshal(b, &req) + switch req.Method { + case "session-get": + fmt.Fprintf(w, `{"arguments":{"speed-limit-up":%d,"speed-limit-up-enabled":%t},"result":"success"}`, tr.up, tr.enabled) + case "session-set": + if v, ok := req.Arguments["speed-limit-up"].(float64); ok { + tr.up = int64(v) + } + if v, ok := req.Arguments["speed-limit-up-enabled"].(bool); ok { + tr.enabled = v + } + fmt.Fprint(w, `{"arguments":{},"result":"success"}`) + default: + fmt.Fprint(w, `{"result":"method name not recognized"}`) + } + })) + defer ts.Close() + tm := &Transmission{Box: envBox{"transmission/PASS": "tr-canary"}, URL: ts.URL} + prev, err := tm.ReadLimit(context.Background()) + if err != nil || prev != (HeldLimit{KBps: 500}) { + t.Fatalf("Transmission read %+v %v, want 500 kB/s switched off", prev, err) + } + if l, err := tm.WriteLimit(context.Background(), tm.HoldValue()); err != nil || l != (HeldLimit{Enabled: true, KBps: 1}) { + t.Fatalf("Transmission hold: %+v %v", l, err) + } + if l, err := tm.WriteLimit(context.Background(), prev); err != nil || l != prev || tr.enabled || tr.up != 500 { + t.Fatalf("Transmission put back: %+v %v (server %d %v), want 500 switched off", l, err, tr.up, tr.enabled) + } + if _, err := tm.rpc(context.Background(), `{"method":"no-such"}`); err == nil || !strings.Contains(err.Error(), "not recognized") { + t.Fatalf("an RPC result other than success must be an error, got %v", err) + } +} diff --git a/controller/internal/upwatch/service.go b/controller/internal/upwatch/service.go index 66acc70..71957a9 100644 --- a/controller/internal/upwatch/service.go +++ b/controller/internal/upwatch/service.go @@ -14,6 +14,8 @@ import ( type State struct { Line []LineRecord `json:"line,omitempty"` QBLimit *QBLimitState `json:"qb_limit,omitempty"` + // Hold is a torrent app held low for a line test on request (R-1250, linehold.go); nil when none. + Hold *LineHold `json:"line_test_hold,omitempty"` } // Store is the settings half the service needs (settings.Settings satisfies it). @@ -44,7 +46,14 @@ type Service struct { // to ep0 inside the tunnel. nil or ok=false = not measured. Read every 15 minutes (RefreshMeasured), never per // minute. Measured func(ctx context.Context) (mbps float64, at time.Time, ok bool) - Log *log.Logger + // The line test for a box that seeds all day (R-1250, linehold.go). Torrents are the apps that can be held, in + // order of preference; TorrentRunning says whether one runs now; LineWaiting is the agent's waiting_for_quiet; + // LineRun is its test on request. Any nil = no hold. + Torrents []Holdable + TorrentRunning func(stack string) bool + LineWaiting func(ctx context.Context) bool + LineRun func(ctx context.Context) (LineRunResult, error) + Log *log.Logger mu sync.Mutex measured float64 @@ -137,6 +146,12 @@ func (s *Service) QBLimitNow(ctx context.Context) { s.RefreshMeasured(ctx) line, known := s.MeasuredLine() cur := decode(s.Store.GetUpwatch()) + if cur.Hold != nil { + // A line-test hold is in place or was left behind: qBittorrent holds the test value, which the rule would + // read as the household's change and then never touch the limit again (linehold.go). + logx.Debugf(s.Log, "[upwatch] qBittorrent limit: a line-test hold record exists — rule skipped until it is cleared") + return + } var st QBLimitState if cur.QBLimit != nil { st = *cur.QBLimit