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