R-1261: the line hold holds every running seeder, drops the record of an app removed while held, and puts back the qBittorrent slot it held (alternative speed). Red-proved. Commit only.
gates / gates (push) Successful in 1m29s
gates / gates (push) Successful in 1m29s
This commit is contained in:
1 parent
fe9837708a
commit
a79a292d72
7 files changed
+475
-88
No files matched your search
@@ -4,6 +4,8 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
|
||||
@@ -33,11 +35,12 @@ import (
|
||||
// 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
|
||||
Alt bool `json:"alt,omitempty"` // qBittorrent: the ALTERNATIVE speed limit's slot (`alt_up_limit`), else `up_limit`
|
||||
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`).
|
||||
// LineHold is the record of one held app (settings.json `upwatch.line_test_holds`).
|
||||
type LineHold struct {
|
||||
App string `json:"app"`
|
||||
Prev HeldLimit `json:"prev"`
|
||||
@@ -48,11 +51,15 @@ type LineHold struct {
|
||||
// Holdable is one torrent app's global upload limit.
|
||||
type Holdable interface {
|
||||
Stack() string
|
||||
// ReadLimit reads the limit that applies NOW (qBittorrent: the slot of its current speed mode).
|
||||
ReadLimit(ctx context.Context) (HeldLimit, error)
|
||||
// WriteLimit sets the limit and returns what the app stored.
|
||||
// ReadLimitLike reads the same slot as like, whatever applies now — the release compares and writes the slot it
|
||||
// held, so a household's mode switch during the test never leaves the test value behind (R-1261).
|
||||
ReadLimitLike(ctx context.Context, like HeldLimit) (HeldLimit, error)
|
||||
// WriteLimit sets the limit in l's slot and returns what the app stored there.
|
||||
WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit, error)
|
||||
// HoldValue is the test value: very low, never zero (zero is „no limit").
|
||||
HoldValue() HeldLimit
|
||||
// HoldValue is the test value in prev's slot: very low, never zero (zero is „no limit").
|
||||
HoldValue(prev HeldLimit) HeldLimit
|
||||
}
|
||||
|
||||
// LineRunResult is the agent's answer to a test on request.
|
||||
@@ -66,18 +73,49 @@ type LineRunResult struct {
|
||||
// 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
|
||||
// heldList is every hold record, the 0.343.0 single one included (R-1261: a record that version left is honoured).
|
||||
func heldList(st State) []LineHold {
|
||||
out := append([]LineHold(nil), st.Holds...)
|
||||
if st.Hold != nil {
|
||||
dup := false
|
||||
for _, h := range out {
|
||||
dup = dup || h.App == st.Hold.App
|
||||
}
|
||||
if !dup {
|
||||
out = append(out, *st.Hold)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *Service) saveHold(h *LineHold) error {
|
||||
func (s *Service) holds() []LineHold {
|
||||
return heldList(decode(s.Store.GetUpwatch()))
|
||||
}
|
||||
|
||||
// saveHolds stores the full list (nil clears it) and drops the 0.343.0 single record.
|
||||
func (s *Service) saveHolds(hs []LineHold) error {
|
||||
return s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) {
|
||||
st := decode(old)
|
||||
st.Hold = h
|
||||
st.Hold = nil
|
||||
st.Holds = hs
|
||||
if len(hs) == 0 {
|
||||
st.Holds = nil
|
||||
}
|
||||
return json.Marshal(st)
|
||||
})
|
||||
}
|
||||
|
||||
// dropHold removes one app's record, keeping the others.
|
||||
func (s *Service) dropHold(app string) error {
|
||||
var keep []LineHold
|
||||
for _, h := range s.holds() {
|
||||
if h.App != app {
|
||||
keep = append(keep, h)
|
||||
}
|
||||
}
|
||||
return s.saveHolds(keep)
|
||||
}
|
||||
|
||||
func (s *Service) torrentApp(stack string) Holdable {
|
||||
for _, t := range s.Torrents {
|
||||
if t.Stack() == stack {
|
||||
@@ -91,32 +129,43 @@ 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)
|
||||
func (s *Service) torrentInstalled(stack string) bool {
|
||||
return s.TorrentInstalled == nil || s.TorrentInstalled(stack)
|
||||
}
|
||||
|
||||
// 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.
|
||||
// RecoverHold puts back every 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 {
|
||||
for _, h := range s.holds() {
|
||||
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)
|
||||
_ = s.dropHold(h.App)
|
||||
continue
|
||||
}
|
||||
if !s.torrentRunning(h.App) {
|
||||
if !s.torrentInstalled(h.App) {
|
||||
// R-1261: removed while held — there is no limit to put back, and a record kept for it would stop
|
||||
// qBittorrent's 50 % rule for good.
|
||||
logx.Infof(s.Log, "[upwatch] %s was removed while held for the line test — its record is dropped (nothing to put back)", h.App)
|
||||
_ = s.dropHold(h.App)
|
||||
continue
|
||||
}
|
||||
logx.Infof(s.Log, "[upwatch] %s is held at the line-test value but not running — put back when it runs", h.App)
|
||||
continue
|
||||
}
|
||||
logx.Infof(s.Log, "[upwatch] %s was left at the line-test value (since %s) — putting its limit back", h.App, h.At)
|
||||
s.release(ctx, app, h)
|
||||
}
|
||||
return len(s.holds()) == 0
|
||||
}
|
||||
|
||||
// release puts the old limit back when the app still holds the test value (in the slot it held), and drops this app's
|
||||
// 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)
|
||||
cur, err := app.ReadLimitLike(rctx, h.Held)
|
||||
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
|
||||
@@ -138,58 +187,74 @@ func (s *Service) release(ctx context.Context, app Holdable, h LineHold) bool {
|
||||
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)
|
||||
if err := s.dropHold(h.App); err != nil {
|
||||
logx.Warnf(s.Log, "[upwatch] clearing the line-test hold record of %s: %v", h.App, err)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// releaseAll puts back every app of this hold.
|
||||
func (s *Service) releaseAll(ctx context.Context, held []Holdable, hs []LineHold) {
|
||||
for i := range hs {
|
||||
s.release(ctx, held[i], hs[i])
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
// 15 minutes, before QBLimitNow). EVERY running seeder is held — one left seeding keeps the uplink busy and the agent
|
||||
// refuses the test „not quiet" every time (R-1261). 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
|
||||
var apps []Holdable
|
||||
for _, t := range s.Torrents {
|
||||
if s.torrentRunning(t.Stack()) {
|
||||
app = t
|
||||
break
|
||||
apps = append(apps, t)
|
||||
}
|
||||
}
|
||||
if app == nil || !s.LineWaiting(ctx) {
|
||||
if len(apps) == 0 || !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)
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
hs := make([]LineHold, 0, len(apps))
|
||||
for _, app := range apps {
|
||||
prev, err := app.ReadLimit(ctx)
|
||||
if err != nil {
|
||||
// One seeder that cannot be read cannot be held, and the test would be refused anyway: hold none.
|
||||
logx.Debugf(s.Log, "[upwatch] line test: reading %s's limit: %v — nothing held", app.Stack(), err)
|
||||
return
|
||||
}
|
||||
hs = append(hs, LineHold{App: app.Stack(), Prev: prev, Held: app.HoldValue(prev), At: now})
|
||||
}
|
||||
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.
|
||||
// Every record FIRST, then the apps: a stop between the two finds the record and puts the limit back.
|
||||
if err := s.saveHolds(hs); err != nil {
|
||||
logx.Warnf(s.Log, "[upwatch] line test: saving the hold records: %v — nothing held", err)
|
||||
return
|
||||
}
|
||||
for i, app := range apps {
|
||||
stored, err := app.WriteLimit(ctx, hs[i].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 every held app back", app.Stack(), err)
|
||||
s.releaseAll(ctx, apps, hs)
|
||||
return
|
||||
}
|
||||
if stored != hs[i].Held {
|
||||
hs[i].Held = stored
|
||||
if err := s.saveHolds(hs); err != nil {
|
||||
logx.Warnf(s.Log, "[upwatch] line test: saving the hold records: %v — putting every held app back", err)
|
||||
s.releaseAll(ctx, apps, hs)
|
||||
return
|
||||
}
|
||||
}
|
||||
logx.Infof(s.Log, "[upwatch] line test: %s held at %+v (was %+v)", app.Stack(), stored, hs[i].Prev)
|
||||
}
|
||||
logx.Infof(s.Log, "[upwatch] line test: %d app(s) held — asking the agent for a test", len(apps))
|
||||
// Deferred: a panic in the call puts the limits back too. A killed process does not — that is RecoverHold's.
|
||||
res, err := func() (LineRunResult, error) {
|
||||
defer s.release(ctx, app, h)
|
||||
defer s.releaseAll(ctx, apps, hs)
|
||||
return s.LineRun(ctx)
|
||||
}()
|
||||
switch {
|
||||
@@ -207,18 +272,76 @@ func (s *Service) SeedLineTestNow(ctx context.Context) {
|
||||
|
||||
// ── the two torrent apps ─────────────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// qBittorrent's limit calls (transfer/uploadLimit, setUploadLimit) act on the slot of its CURRENT speed mode — measured
|
||||
// on 5.2.4 (felhom.eu `audits/night-2026-10-25/r1261-qbittorrent-api-measured.txt`): in alternative mode they read and
|
||||
// write `alt_up_limit`. The hold therefore names its slot and reads/writes it through app/preferences (`up_limit`,
|
||||
// `alt_up_limit`, both B/s, either writable in either mode), so a mode switch during the test cannot leave 1 KiB/s.
|
||||
|
||||
func (q *QBittorrent) ReadLimit(ctx context.Context) (HeldLimit, error) {
|
||||
b, err := q.UploadLimit(ctx, nil)
|
||||
return HeldLimit{Bytes: b}, err
|
||||
h, err := q.session(ctx, true)
|
||||
if err != nil {
|
||||
return HeldLimit{}, err
|
||||
}
|
||||
code, b, _, err := httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/transfer/speedLimitsMode", h, nil, "")
|
||||
if err != nil || code != 200 {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent speedLimitsMode: HTTP %d %v", code, err)
|
||||
}
|
||||
return q.readSlot(ctx, h, strings.TrimSpace(string(b)) == "1")
|
||||
}
|
||||
|
||||
func (q *QBittorrent) ReadLimitLike(ctx context.Context, like HeldLimit) (HeldLimit, error) {
|
||||
h, err := q.session(ctx, true)
|
||||
if err != nil {
|
||||
return HeldLimit{}, err
|
||||
}
|
||||
return q.readSlot(ctx, h, like.Alt)
|
||||
}
|
||||
|
||||
func qbSlotKey(alt bool) string {
|
||||
if alt {
|
||||
return "alt_up_limit"
|
||||
}
|
||||
return "up_limit"
|
||||
}
|
||||
|
||||
func (q *QBittorrent) readSlot(ctx context.Context, h map[string]string, alt bool) (HeldLimit, error) {
|
||||
code, b, _, err := httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/app/preferences", h, nil, "")
|
||||
if err != nil || code != 200 {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent preferences: HTTP %d %v", code, err)
|
||||
}
|
||||
var prefs map[string]json.RawMessage
|
||||
if err := json.Unmarshal(b, &prefs); err != nil {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent preferences: %v", err)
|
||||
}
|
||||
raw, ok := prefs[qbSlotKey(alt)]
|
||||
if !ok {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent preferences: no %s", qbSlotKey(alt))
|
||||
}
|
||||
var v int64
|
||||
if err := json.Unmarshal(raw, &v); err != nil {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent preferences %s: %q", qbSlotKey(alt), raw)
|
||||
}
|
||||
return HeldLimit{Bytes: v, Alt: alt}, nil
|
||||
}
|
||||
|
||||
func (q *QBittorrent) WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit, error) {
|
||||
b, err := q.SetUploadLimit(ctx, l.Bytes)
|
||||
return HeldLimit{Bytes: b}, err
|
||||
h, err := q.session(ctx, true)
|
||||
if err != nil {
|
||||
return HeldLimit{}, err
|
||||
}
|
||||
js, _ := json.Marshal(map[string]int64{qbSlotKey(l.Alt): l.Bytes})
|
||||
form := url.Values{"json": {string(js)}}.Encode()
|
||||
code, _, _, err := httpDo(ctx, q.client(), "POST", q.base()+"/api/v2/app/setPreferences", h, []byte(form), "application/x-www-form-urlencoded")
|
||||
if err != nil || code != 200 {
|
||||
return HeldLimit{}, fmt.Errorf("qBittorrent setPreferences %s: HTTP %d %v", qbSlotKey(l.Alt), code, err)
|
||||
}
|
||||
return q.readSlot(ctx, h, l.Alt)
|
||||
}
|
||||
|
||||
// HoldValue: 1 KiB/s, qBittorrent's smallest stored step.
|
||||
func (q *QBittorrent) HoldValue() HeldLimit { return HeldLimit{Bytes: 1024} }
|
||||
// HoldValue: 1 KiB/s in the slot that applied when the hold began.
|
||||
func (q *QBittorrent) HoldValue(prev HeldLimit) HeldLimit {
|
||||
return HeldLimit{Bytes: 1024, Alt: prev.Alt}
|
||||
}
|
||||
|
||||
func (t *Transmission) ReadLimit(ctx context.Context) (HeldLimit, error) {
|
||||
var st struct {
|
||||
@@ -245,5 +368,10 @@ func (t *Transmission) WriteLimit(ctx context.Context, l HeldLimit) (HeldLimit,
|
||||
return t.ReadLimit(ctx)
|
||||
}
|
||||
|
||||
// ReadLimitLike: Transmission's speed-limit-up is one slot (its alternative speed has its own fields, never touched).
|
||||
func (t *Transmission) ReadLimitLike(ctx context.Context, _ HeldLimit) (HeldLimit, error) {
|
||||
return t.ReadLimit(ctx)
|
||||
}
|
||||
|
||||
// HoldValue: 1 kB/s with the limit switched on.
|
||||
func (t *Transmission) HoldValue() HeldLimit { return HeldLimit{Enabled: true, KBps: 1} }
|
||||
func (t *Transmission) HoldValue(HeldLimit) HeldLimit { return HeldLimit{Enabled: true, KBps: 1} }
|
||||
Reference in new issue
Block a user