210 lines
6.3 KiB
Go
210 lines
6.3 KiB
Go
package upwatch
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
|
|
)
|
|
|
|
// State is what the watch keeps in settings.json (`upwatch`).
|
|
type State struct {
|
|
Line []LineRecord `json:"line,omitempty"`
|
|
QBLimit *QBLimitState `json:"qb_limit,omitempty"`
|
|
}
|
|
|
|
// Store is the settings half the service needs (settings.Settings satisfies it).
|
|
type Store interface {
|
|
GetUpwatch() json.RawMessage
|
|
UpdateUpwatch(fn func(old json.RawMessage) (json.RawMessage, error)) error
|
|
}
|
|
|
|
func decode(raw json.RawMessage) State {
|
|
var st State
|
|
if len(raw) > 0 {
|
|
_ = json.Unmarshal(raw, &st) // a damaged block reads as empty: the line is re-learnt, the limit record too
|
|
}
|
|
return st
|
|
}
|
|
|
|
// Service ties the watch together for main: the minute watcher, the push sampler, the qBittorrent limit rule and
|
|
// what the pages show.
|
|
type Service struct {
|
|
Store Store
|
|
Watcher *Watcher
|
|
QB *QBittorrent // nil when the box has no qBittorrent wiring
|
|
// QBRunning reports whether qBittorrent is installed and running now.
|
|
QBRunning func() bool
|
|
// QBShare is the share of the measured line qBittorrent may upload at (operator decide row; CC's pick 0.5).
|
|
QBShare float64
|
|
// Measured reads the host's weekly line test (the agent's GET /host/line-test, R-1247): a 10-second iperf3 upload
|
|
// 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
|
|
|
|
mu sync.Mutex
|
|
measured float64
|
|
measuredAt time.Time
|
|
measuredOK bool
|
|
}
|
|
|
|
// RefreshMeasured reads the agent's line test once. A failed read keeps the last value (the agent persists it; an
|
|
// agent restart must not drop qBittorrent's limit to „unknown").
|
|
func (s *Service) RefreshMeasured(ctx context.Context) {
|
|
if s.Measured == nil {
|
|
return
|
|
}
|
|
mbps, at, ok := s.Measured(ctx)
|
|
if !ok || mbps <= 0 {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
changed := !s.measuredOK || s.measured != mbps
|
|
s.measured, s.measuredAt, s.measuredOK = mbps, at, true
|
|
s.mu.Unlock()
|
|
if changed {
|
|
logx.Infof(s.Log, "[upwatch] the host's weekly line test: %.1f Mbit/s (measured %s)", mbps, at.UTC().Format(time.RFC3339))
|
|
}
|
|
}
|
|
|
|
// MeasuredLine is the host's tested upload line (ok=false: never tested).
|
|
func (s *Service) MeasuredLine() (float64, bool) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.measured, s.measuredOK
|
|
}
|
|
|
|
// Line is the upload line the watch judges against, in Mbit/s: the higher of the weekly test and the off-site push's
|
|
// floor (both are lower bounds of the real line; the test is the close one). ok=false while neither exists.
|
|
func (s *Service) Line() (float64, bool) {
|
|
l, _, ok := s.line()
|
|
return l, ok
|
|
}
|
|
|
|
func (s *Service) line() (mbps float64, tested, ok bool) {
|
|
passive, pok := LineFromRecords(decode(s.Store.GetUpwatch()).Line)
|
|
m, mok := s.MeasuredLine()
|
|
switch {
|
|
case mok && m >= passive:
|
|
return m, true, true
|
|
case pok:
|
|
return passive, false, true
|
|
}
|
|
return 0, false, false
|
|
}
|
|
|
|
// RecordPush stores one off-site push's measurement.
|
|
func (s *Service) RecordPush(r LineRecord) {
|
|
err := s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) {
|
|
st := decode(old)
|
|
st.Line = AddRecord(st.Line, r)
|
|
return json.Marshal(st)
|
|
})
|
|
if err != nil {
|
|
logx.Warnf(s.Log, "[upwatch] saving the push measurement: %v", err)
|
|
return
|
|
}
|
|
l, ok := s.Line()
|
|
logx.Infof(s.Log, "[upwatch] off-site push measured: %d bytes in %d s, best 10 s %.1f Mbit/s (counts=%v); line now %s",
|
|
r.Bytes, r.Seconds, r.Peak10Mbps, r.Counts, lineWords(l, ok))
|
|
}
|
|
|
|
func lineWords(l float64, ok bool) string {
|
|
if !ok {
|
|
return "unknown (learning)"
|
|
}
|
|
return "at least " + formatMbps(l) + " Mbit/s"
|
|
}
|
|
|
|
// PushObserver is handed to the off-site backup: it is called at the push's start and returns the function that
|
|
// ends the measurement.
|
|
func (s *Service) PushObserver() func() {
|
|
stop := (&PushSampler{}).Start()
|
|
return func() { s.RecordPush(stop()) }
|
|
}
|
|
|
|
// QBLimitNow runs qBittorrent's limit rule once (main calls it every 15 minutes; a no-op without qBittorrent).
|
|
func (s *Service) QBLimitNow(ctx context.Context) {
|
|
if s.QB == nil || (s.QBRunning != nil && !s.QBRunning()) {
|
|
return
|
|
}
|
|
// D41 (operator 2026-10-24): 50 % of the TESTED line. The push floor is not used here — on demo-hp it read 10 Mbit/s
|
|
// on a 90 Mbit/s line, and half of that would have held a seeding qBittorrent at 5. No test yet → no limit.
|
|
s.RefreshMeasured(ctx)
|
|
line, known := s.MeasuredLine()
|
|
cur := decode(s.Store.GetUpwatch())
|
|
var st QBLimitState
|
|
if cur.QBLimit != nil {
|
|
st = *cur.QBLimit
|
|
}
|
|
ns, act, err := ApplyQBLimit(ctx, s.QB, st, line, known, s.QBShare, time.Now())
|
|
if err != nil {
|
|
logx.Debugf(s.Log, "[upwatch] qBittorrent limit: %v — tried again in 15 minutes", err)
|
|
return
|
|
}
|
|
if ns != st {
|
|
if err := s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) {
|
|
o := decode(old)
|
|
o.QBLimit = &ns
|
|
return json.Marshal(o)
|
|
}); err != nil {
|
|
logx.Warnf(s.Log, "[upwatch] saving the qBittorrent limit record: %v", err)
|
|
return
|
|
}
|
|
}
|
|
switch act {
|
|
case "set":
|
|
logx.Infof(s.Log, "[upwatch] qBittorrent upload limit set to %d B/s (%.0f %% of the tested line, %.1f Mbit/s)", ns.BoxSetBytes, s.QBShare*100, line)
|
|
case "household":
|
|
if !st.Household {
|
|
logx.Infof(s.Log, "[upwatch] qBittorrent's upload limit is the household's — the box leaves it alone from now on")
|
|
}
|
|
}
|
|
}
|
|
|
|
// View is what the pages show.
|
|
type View struct {
|
|
LineMbps float64
|
|
LineKnown bool
|
|
LineTested bool // LineMbps is the weekly test's (else the off-site push's floor, „at least")
|
|
QBLimitBps int64 // the limit the box set and still owns; 0 = none shown
|
|
}
|
|
|
|
// View reads the state for the pages.
|
|
func (s *Service) View() View {
|
|
st := decode(s.Store.GetUpwatch())
|
|
v := View{}
|
|
v.LineMbps, v.LineTested, v.LineKnown = s.line()
|
|
if st.QBLimit != nil && !st.QBLimit.Household {
|
|
v.QBLimitBps = st.QBLimit.BoxSetBytes
|
|
}
|
|
return v
|
|
}
|
|
|
|
// MbpsText renders bytes/s as Mbit/s for a page line (one decimal under 10).
|
|
func MbpsText(bps int64) string { return formatMbps(float64(bps) * 8 / 1e6) }
|
|
|
|
func formatMbps(v float64) string {
|
|
if v < 10 {
|
|
return trimZero(v)
|
|
}
|
|
return itoa(int64(v + 0.5))
|
|
}
|
|
|
|
func trimZero(v float64) string {
|
|
x := int64(v*10 + 0.5)
|
|
if x%10 == 0 {
|
|
return itoa(x / 10)
|
|
}
|
|
return itoa(x/10) + "." + itoa(x%10)
|
|
}
|
|
|
|
func itoa(v int64) string {
|
|
b, _ := json.Marshal(v)
|
|
return string(b)
|
|
}
|