Живые обновления прогресса и раздел «Раздача» (live-status)
Воркер ведёт in-memory снимок телеметрии раздач (прогресс, скорость, ETA, рейтинг, сиды/пиры, отдано) под отдельным RWMutex, обновляя его на каждом тике поллинга сразу после построения byHash — без лишних вызовов qBittorrent и без хранения в БД (волатильно). qbt.Torrent дополнен полями телеметрии. Веб-UI читает снимок через узкий контракт LiveStatus: карточки активных загрузок показывают живой прогресс-бар (htmx-поллинг фрагмента every 3s, точечно — без сброса фильтров), на странице загрузки появилась секция «Раздача» для сидирующих задач. Начальный кадр рендерится сразу со значениями; при отсутствии данных UI деградирует штатно. Капабилити live-status (OpenSpec), web-ui дополнен. Change заархивирован. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,62 @@
|
||||
package worker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.vakhrushev.me/av/jellybit/internal/qbt"
|
||||
"git.vakhrushev.me/av/jellybit/internal/store"
|
||||
)
|
||||
|
||||
// TestPollBuildsLiveSnapshot: после Poll снимок несёт телеметрию качающейся и
|
||||
// сидирующей задач (с верным Seeding), доступную по любому из трёх хэшей;
|
||||
// неизвестный/пустой infohash → ok=false.
|
||||
func TestPollBuildsLiveSnapshot(t *testing.T) {
|
||||
qb := &fakeQbt{torrents: []qbt.Torrent{
|
||||
// Качается (Category пуст → discover не усыновляет, store не мешает).
|
||||
{Hash: "aaa", State: "downloading", Progress: 0.5, Dlspeed: 1000, Eta: 120},
|
||||
// Сидирует, торрент v2 (три ключа).
|
||||
{
|
||||
Hash: "bbb", InfohashV1: "bbb1", InfohashV2: "BBB2",
|
||||
State: "uploading", Progress: 1.0,
|
||||
Ratio: 2.5, NumSeeds: 3, NumLeechs: 1, Uploaded: 999, Upspeed: 50,
|
||||
},
|
||||
}}
|
||||
w := newTestWorker(&fakeStore{downloads: map[int64]*store.Download{}}, qb)
|
||||
|
||||
if err := w.Poll(context.Background()); err != nil {
|
||||
t.Fatalf("Poll: %v", err)
|
||||
}
|
||||
|
||||
dl, ok := w.Live("aaa")
|
||||
if !ok {
|
||||
t.Fatal("нет телеметрии качающейся задачи")
|
||||
}
|
||||
if dl.Seeding {
|
||||
t.Error("качающаяся задача помечена Seeding")
|
||||
}
|
||||
if dl.Progress != 0.5 || dl.DlSpeed != 1000 || dl.ETA != 120 {
|
||||
t.Errorf("телеметрия качания неверна: %+v", dl)
|
||||
}
|
||||
|
||||
// Сидирующая задача находится по любому из трёх хэшей (lowercase).
|
||||
for _, h := range []string{"bbb", "bbb1", "BBB2", "bbb2"} {
|
||||
sd, ok := w.Live(h)
|
||||
if !ok {
|
||||
t.Fatalf("нет телеметрии раздачи по ключу %q", h)
|
||||
}
|
||||
if !sd.Seeding {
|
||||
t.Errorf("ключ %q: Seeding=false для uploading", h)
|
||||
}
|
||||
if sd.Ratio != 2.5 || sd.Seeds != 3 || sd.Peers != 1 || sd.Uploaded != 999 || sd.UpSpeed != 50 {
|
||||
t.Errorf("ключ %q: статистика раздачи неверна: %+v", h, sd)
|
||||
}
|
||||
}
|
||||
|
||||
if _, ok := w.Live("unknownhash"); ok {
|
||||
t.Error("неизвестный infohash вернул ok=true")
|
||||
}
|
||||
if _, ok := w.Live(""); ok {
|
||||
t.Error("пустой infohash вернул ok=true")
|
||||
}
|
||||
}
|
||||
@@ -134,6 +134,40 @@ type Config struct {
|
||||
SourceMissingThreshold int
|
||||
}
|
||||
|
||||
// Live — живая телеметрия одной раздачи из снимка воркера. Курированный срез
|
||||
// qbt.Torrent: читатели (httpapi) не зависят от пакета qbt, контракт чтения
|
||||
// узкий. Seeding вычисляется воркером (classify), чтобы трактовка завершённости
|
||||
// не дублировалась в транспорте.
|
||||
type Live struct {
|
||||
Progress float64 // доля 0..1
|
||||
DlSpeed int64 // скорость загрузки, байт/с
|
||||
ETA int64 // оценка до завершения, с (8640000 ≈ ∞)
|
||||
State string // сырое состояние qBittorrent
|
||||
Seeding bool // торрент завершён и раздаётся
|
||||
|
||||
Ratio float64 // рейтинг отдачи (может быть <0 = ∞/н/д)
|
||||
Seeds int // подключённые сиды
|
||||
Peers int // подключённые личи
|
||||
Uploaded int64 // отдано всего, байт
|
||||
UpSpeed int64 // скорость отдачи, байт/с
|
||||
}
|
||||
|
||||
// liveFrom собирает Live из торрента qBittorrent (Seeding — через classify).
|
||||
func liveFrom(t qbt.Torrent) Live {
|
||||
return Live{
|
||||
Progress: t.Progress,
|
||||
DlSpeed: t.Dlspeed,
|
||||
ETA: t.Eta,
|
||||
State: t.State,
|
||||
Seeding: classify(t.State) == classReady,
|
||||
Ratio: t.Ratio,
|
||||
Seeds: t.NumSeeds,
|
||||
Peers: t.NumLeechs,
|
||||
Uploaded: t.Uploaded,
|
||||
UpSpeed: t.Upspeed,
|
||||
}
|
||||
}
|
||||
|
||||
// Worker — поллер и владелец переходов.
|
||||
type Worker struct {
|
||||
store Store
|
||||
@@ -149,6 +183,14 @@ type Worker struct {
|
||||
notifier Notifier // опц. исходящие пинги
|
||||
scanner Scanner // опц. пересканирование Jellyfin
|
||||
|
||||
// live — снимок живой телеметрии раздач (ключ — lowercase infohash, по три
|
||||
// ключа на торрент, как byHash). Обновляется атомарным свопом карты на
|
||||
// каждом тике Poll. Отдельный RWMutex (не w.mu): UI читает телеметрию часто,
|
||||
// смешивать частые чтения с замком переходов — лишняя конкуренция. Снимок
|
||||
// волатилен, в БД не хранится.
|
||||
liveMu sync.RWMutex
|
||||
live map[string]Live
|
||||
|
||||
// failNotified — дебаунс повторных EventFailed по задаче (download_id →
|
||||
// время последнего пинга). Мерцающий stalled-торрент колеблется
|
||||
// stuck↔downloading; без дебаунса каждый цикл слал бы уведомление. Память
|
||||
@@ -179,9 +221,30 @@ func New(st Store, qb QBittorrent, rec Recognizer, lay Layouter, cfg Config, log
|
||||
now: time.Now,
|
||||
newID: defaultBatchID,
|
||||
failNotified: map[int64]time.Time{},
|
||||
live: map[string]Live{},
|
||||
}
|
||||
}
|
||||
|
||||
// Live возвращает живую телеметрию раздачи по infohash (любому из v1/v2/hash).
|
||||
// ok=false, если infohash пуст или раздачи не было в последнем тике поллинга —
|
||||
// тогда читатель деградирует без живых значений. Чтение под RLock.
|
||||
func (w *Worker) Live(infohash string) (Live, bool) {
|
||||
if infohash == "" {
|
||||
return Live{}, false
|
||||
}
|
||||
w.liveMu.RLock()
|
||||
defer w.liveMu.RUnlock()
|
||||
l, ok := w.live[strings.ToLower(infohash)]
|
||||
return l, ok
|
||||
}
|
||||
|
||||
// setLive атомарно подменяет снимок телеметрии готовой картой.
|
||||
func (w *Worker) setLive(snap map[string]Live) {
|
||||
w.liveMu.Lock()
|
||||
w.live = snap
|
||||
w.liveMu.Unlock()
|
||||
}
|
||||
|
||||
// defaultBatchID — уникальный идентификатор батча раскладки.
|
||||
func defaultBatchID() string {
|
||||
return fmt.Sprintf("b-%d", time.Now().UnixNano())
|
||||
@@ -235,13 +298,20 @@ func (w *Worker) Poll(ctx context.Context) error {
|
||||
return fmt.Errorf("poll: list torrents: %w", err)
|
||||
}
|
||||
byHash := make(map[string]qbt.Torrent, len(torrents)*2)
|
||||
live := make(map[string]Live, len(torrents)*2)
|
||||
for _, t := range torrents {
|
||||
l := liveFrom(t)
|
||||
for _, h := range []string{t.Hash, t.InfohashV1, t.InfohashV2} {
|
||||
if h != "" {
|
||||
byHash[strings.ToLower(h)] = t
|
||||
key := strings.ToLower(h)
|
||||
byHash[key] = t
|
||||
live[key] = l
|
||||
}
|
||||
}
|
||||
}
|
||||
// Снимок зависит только от torrents — свопаем сразу, до store-операций
|
||||
// ниже (их ранний return по ошибке не должен лишать UI свежей телеметрии).
|
||||
w.setLive(live)
|
||||
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
Reference in New Issue
Block a user