felhom-controller/controller/internal/upwatch/linehold.go

package upwatch

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"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. waits until every held app REPORTS its own upload near zero (waitDrained) — qBittorrent 5.2.4 kept sending at
//     full speed for ~14 s after its limit fell to 1 KiB/s (measured on demo-hp 2026-10-27, felhom.eu
//     `audits/linetest-2026-10-27/`), so the agent's quiet check, 5–15 s after the hold, refused every test;
//  4. 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);
//  5. 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
	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 one held app (settings.json `upwatch.line_test_holds`).
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 reads the limit that applies NOW (qBittorrent: the slot of its current speed mode).
	ReadLimit(ctx context.Context) (HeldLimit, error)
	// 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 in prev's slot: very low, never zero (zero is „no limit").
	HoldValue(prev HeldLimit) HeldLimit
	// UploadRate is the app's own upload speed now, B/s — read while held, to know when it has really slowed.
	UploadRate(ctx context.Context) (int64, error)
}

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

// 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) 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 = 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 {
			return t
		}
	}
	return nil
}

func (s *Service) torrentRunning(stack string) bool {
	return s.TorrentRunning != nil && s.TorrentRunning(stack)
}

func (s *Service) torrentInstalled(stack string) bool {
	return s.TorrentInstalled == nil || s.TorrentInstalled(stack)
}

// 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.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
	}
	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.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])
	}
}

// The drain wait (step 3). The agent refuses a test when the uplink sends 2 Mbit/s or more 5–15 s after the call, so the
// controller asks only when every held app reports its own upload at or below drainQuietBps (0.13 Mbit/s) on
// drainQuietReadings readings in a row, drainPoll apart. It never waits longer than drainMax, and asks then anyway (the
// agent judges the line). An app whose speed cannot be read gets a plain drainFallback wait instead — the measured
// drain was ~14 s. Pinned by linehold_r1250_drain_test.go.
const (
	drainPoll          = 2 * time.Second
	drainMax           = 90 * time.Second
	drainFallback      = 20 * time.Second
	drainQuietBps      = 16 * 1024
	drainQuietReadings = 2
)

// drainSleep waits d; false when ctx ends first. A variable so the tests can count the waiting instead of doing it.
var drainSleep = func(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// waitDrained returns when every held app has slowed down to the test value, or when it cannot tell (see above).
func (s *Service) waitDrained(ctx context.Context, apps []Holdable) {
	var waited time.Duration
	quiet := 0
	for {
		var worst int64
		var readErr error
		for _, app := range apps {
			r, err := app.UploadRate(ctx)
			if err != nil {
				readErr = fmt.Errorf("%s: %v", app.Stack(), err)
				break
			}
			if r > worst {
				worst = r
			}
		}
		switch {
		case readErr != nil:
			logx.Infof(s.Log, "[upwatch] line test: cannot read the upload speed (%v) — waiting %s before asking the agent", readErr, drainFallback)
			if rest := drainFallback - waited; rest > 0 {
				drainSleep(ctx, rest)
			}
			return
		case worst <= drainQuietBps:
			quiet++
			if quiet >= drainQuietReadings {
				logx.Infof(s.Log, "[upwatch] line test: the held app(s) slowed down after %s — asking the agent", waited)
				return
			}
		default:
			quiet = 0
			logx.Debugf(s.Log, "[upwatch] line test: held, still sending %d B/s after %s", worst, waited)
		}
		if waited >= drainMax {
			logx.Infof(s.Log, "[upwatch] line test: the held app(s) still send %d B/s after %s — asking the agent anyway", worst, waited)
			return
		}
		if !drainSleep(ctx, drainPoll) {
			return
		}
		waited += drainPoll
	}
}

// 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). 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 apps []Holdable
	for _, t := range s.Torrents {
		if s.torrentRunning(t.Stack()) {
			apps = append(apps, t)
		}
	}
	if len(apps) == 0 || !s.LineWaiting(ctx) {
		return
	}
	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})
	}
	// 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.releaseAll(ctx, apps, hs)
		s.waitDrained(ctx, apps)
		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 ─────────────────────────────────────────────────────────────────────────────────────────

// 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) {
	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) {
	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)
}

// NormalLimit / SetNormalLimit are the 50 % rule's side (R-1292): the NORMAL slot (`up_limit`) whatever the mode.
func (q *QBittorrent) NormalLimit(ctx context.Context) (int64, error) {
	l, err := q.ReadLimitLike(ctx, HeldLimit{Alt: false})
	return l.Bytes, err
}

func (q *QBittorrent) SetNormalLimit(ctx context.Context, bytesPerSec int64) (int64, error) {
	l, err := q.WriteLimit(ctx, HeldLimit{Bytes: bytesPerSec, Alt: false})
	return l.Bytes, err
}

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

// UploadRate: transfer/info `up_info_speed` (B/s, payload and overhead together).
func (q *QBittorrent) UploadRate(ctx context.Context) (int64, error) {
	h, err := q.session(ctx, true)
	if err != nil {
		return 0, err
	}
	code, b, _, err := httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/transfer/info", h, nil, "")
	if err != nil || code != 200 {
		return 0, fmt.Errorf("qBittorrent transfer/info: HTTP %d %v", code, err)
	}
	var ti struct {
		UpSpeed *int64 `json:"up_info_speed"`
	}
	if err := json.Unmarshal(b, &ti); err != nil {
		return 0, fmt.Errorf("qBittorrent transfer/info: %v", err)
	}
	if ti.UpSpeed == nil {
		return 0, fmt.Errorf("qBittorrent transfer/info: no up_info_speed")
	}
	return *ti.UpSpeed, nil
}

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

// 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) HeldLimit { return HeldLimit{Enabled: true, KBps: 1} }

// UploadRate: session-stats `uploadSpeed` (B/s).
func (t *Transmission) UploadRate(ctx context.Context) (int64, error) {
	b, err := t.rpc(ctx, `{"method":"session-stats"}`)
	if err != nil {
		return 0, err
	}
	var st struct {
		Arguments struct {
			UploadSpeed *int64 `json:"uploadSpeed"`
		} `json:"arguments"`
	}
	if err := json.Unmarshal(b, &st); err != nil {
		return 0, err
	}
	if st.Arguments.UploadSpeed == nil {
		return 0, fmt.Errorf("transmission session-stats: no uploadSpeed")
	}
	return *st.Arguments.UploadSpeed, nil
}