Files
jellybit/internal/worker/worker.go
T
avandClaude Fable 5 37f2f6481a Идентичность на ULID: download_infohash, guarded-дедуп, миграция (ulid-identity)
Все сущности переехали с INTEGER AUTOINCREMENT на TEXT ULID (lowercase,
internal/ident — единая точка генерации и разбора; oklog/ulid). Инфохэши
загрузки — множество (download_infohash, v1/v2 гибридных торрентов): дедуп
и сопоставление в поллинге по любому из хешей, magnet-парсер отдаёт оба
хеша гибридной ссылки, усечённый v2-хеш v2-only раздач не хранится.

Инвариант «не более одной активной загрузки на infohash» вместо снятого
unique-индекса держат guarded-методы store в одной write-транзакции
(_txlock=immediate): CreateDownloadIfNoActive (приём/adopt, с доносом
недостающих хешей), ActivateIfNoOtherActive (retry/recovery/relink, отказ
до побочных эффектов), guarded AddInfohashes; SetDownloadState отклоняет
терминал→активное как механический бэкстоп.

Миграция 0006 — первая Go-миграция goose: пересоздание таблиц при
включённых FK, backfill ULID с timestamp из created_at (хронология id
сохранена), разнос infohash, удаление idempotency_key. BREAKING: формат id
в URL/логах/Telegram, REST-поля id (string) и infohashes (список).

Новая конвенция docs/conventions/database.md (без числовых PK), корреляция
в логах grep'ом по голому ULID, ER-схема обновлена. Спеки: новая capability
identity, MODIFIED в state-reconciliation; change заархивирован. Пройдены
ревью дизайна и кода (по 8 углов), все находки исправлены с
регрессионными тестами.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 21:25:00 +03:00

626 lines
29 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// Package worker — владелец машины состояний. Поллит qBittorrent по
// категории, переводит задачи между состояниями и сериализует команды
// транспортов (cancel/retry), чтобы два транспорта не гонялись за одно
// состояние.
//
// Ф1 ведёт задачу downloading → completed, плюс stuck/failed по таймаутам и
// ошибкам qBittorrent. Ф3 продолжает: completed → recognizing (вызов
// recognize) → review; команды ревью (apply/refine/reject/defer/undo,
// переключение типа, пометка «игнор») раскладывают файлы хардлинками через
// layout. Распознавание зовётся в поллинг-цикле, команды — из транспортов;
// всё под per-download блокировкой w.mu.
package worker
import (
"context"
"errors"
"fmt"
"log/slog"
"slices"
"strings"
"sync"
"time"
"git.vakhrushev.me/av/jellybit/internal/ident"
"git.vakhrushev.me/av/jellybit/internal/layout"
"git.vakhrushev.me/av/jellybit/internal/logctx"
"git.vakhrushev.me/av/jellybit/internal/qbt"
"git.vakhrushev.me/av/jellybit/internal/recognize"
"git.vakhrushev.me/av/jellybit/internal/store"
)
// Стадии (capability) — адресуют запись к подсистеме при корреляции по
// download_id. Уровень логов от стадии не зависит.
const (
capIngest = "ingest" // приём/скачивание (поллинг, reconcile, переходы)
capRecognize = "recognition" // распознавание фильма/сериала
capFileLayout = "file-layout" // раскладка хардлинками
capReview = "review" // ручные команды ревью
)
// Store — нужная worker часть хранилища.
type Store interface {
ListDownloadsByState(ctx context.Context, states ...store.State) ([]store.Download, error)
ListRecoverable(ctx context.Context, codes ...string) ([]store.Download, error)
GetDownload(ctx context.Context, id string) (*store.Download, error)
SetDownloadState(ctx context.Context, id string, state store.State, errCode, errMsg string) error
SetSourceMissCount(ctx context.Context, id string, n int) error
SetSourceAddedAt(ctx context.Context, id string, t time.Time) error
// Идентичность/инвариант «одна активная загрузка на infohash».
ExistsByInfohash(ctx context.Context, hashes ...string) (bool, error)
CreateDownloadIfNoActive(ctx context.Context, d *store.Download, hashes []string) (*store.Download, error)
ActivateIfNoOtherActive(ctx context.Context, id string, state store.State, errCode, errMsg string) error
AddInfohashes(ctx context.Context, downloadID string, hashes []string) error
// Ф3: распознавание, ревью, раскладка.
CreateRecognition(ctx context.Context, r *store.Recognition, reasons []string) (string, error)
GetCurrentRecognition(ctx context.Context, downloadID string) (*store.Recognition, error)
AddHint(ctx context.Context, downloadID string, text string) error
ListHints(ctx context.Context, downloadID string) ([]string, error)
SetOverride(ctx context.Context, downloadID string, field, value string) error
ListOverrides(ctx context.Context, downloadID string) (map[string]string, error)
CreateFileLinks(ctx context.Context, links []store.FileLink) error
SupersedeForeignLinks(ctx context.Context, downloadID string, dstPaths []string) error
LatestBatchID(ctx context.Context, downloadID string) (string, error)
ListFileLinksByBatch(ctx context.Context, batchID string) ([]store.FileLink, error)
DeleteFileLinksByBatch(ctx context.Context, batchID string) error
// Кандидаты базы метаданных (ручной выбор в review).
CreateCandidates(ctx context.Context, cands []store.MetadataCandidate) error
ListCandidatesByRecognition(ctx context.Context, recognitionID string) ([]store.MetadataCandidate, error)
GetCandidate(ctx context.Context, id string) (*store.MetadataCandidate, error)
SetCandidateChosen(ctx context.Context, recognitionID, candidateID string) error
}
// QBittorrent — нужная worker часть клиента qBittorrent.
type QBittorrent interface {
Torrents(ctx context.Context, category string) ([]qbt.Torrent, error)
Add(ctx context.Context, ar qbt.AddRequest) error
Files(ctx context.Context, hash string) ([]qbt.File, error)
}
// Recognizer — распознаватель (recognize.Recognizer).
type Recognizer interface {
Recognize(ctx context.Context, in recognize.Input) (recognize.Result, error)
}
// Layouter — раскладчик хардлинками (layout.Layouter).
type Layouter interface {
BuildLinks(p layout.Plan) ([]layout.Link, error)
Apply(ctx context.Context, links []layout.Link) ([]layout.Result, error)
Undo(ctx context.Context, links []layout.Link) (int, error)
}
// NotifyEvent — повод позвать пользователя.
type NotifyEvent string
const (
EventReview NotifyEvent = "review" // задача ждёт подтверждения
EventDone NotifyEvent = "done" // раскладка завершена
EventOrphaned NotifyEvent = "orphaned" // источник пропал, цель — последняя копия
EventTargetMissing NotifyEvent = "target_missing" // цель удалена, доступен relink
EventFailed NotifyEvent = "failed" // задача упала/зависла (failed/stuck)
)
// Коды ошибок (error_code) при переходе в failed/stuck. Восстановимые
// (magnet_timeout/stalled) — следствие нашей нетерпеливости: сверка воскрешает
// такие задачи при оживлении источника (см. reconcileRecovery). qbit_error —
// реальная ошибка qBittorrent, восстановлению не подлежит.
const (
errCodeMagnetTimeout = "magnet_timeout"
errCodeStalled = "stalled"
errCodeQbitError = "qbit_error"
)
// Notifier — исходящие пинги (Telegram). Вызывается неблокирующе.
type Notifier interface {
Notify(ctx context.Context, downloadID string, event NotifyEvent)
}
// Scanner — триггер пересканирования медиатеки Jellyfin. Вызывается
// неблокирующе после успешной раскладки, чтобы новые файлы быстрее появились
// в проигрывателе.
type Scanner interface {
RefreshLibraries(ctx context.Context) error
}
// Config — параметры воркера.
type Config struct {
Category string
Tag string // метка для усыновления существующих раздач (discovery)
SavePath string
PathMap map[string]string // трансляция save_path qBit → хост-путь (обычно пусто)
PollInterval time.Duration
StuckAfter time.Duration // stalledDL дольше → stuck
MagnetTimeout time.Duration // metaDL дольше → failed
// SourceMissingThreshold — порог дебаунса пропажи источника (тиков сверки).
// <1 трактуется как 1 (помечаем при первой же устойчивой пропаже).
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
qbt QBittorrent
recognizer Recognizer
layouter Layouter
cfg Config
log *slog.Logger
mu sync.Mutex // сериализует переходы (поллинг + команды)
now func() time.Time // подменяется в тестах
newID func() string // генератор apply_batch_id (подменяется в тестах)
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; без дебаунса каждый цикл слал бы уведомление. Память
// процесса: при рестарте дебаунс сбрасывается — допустимо. Доступ под w.mu.
failNotified map[string]time.Time
}
// failNotifyDebounce — минимальный интервал между уведомлениями о падении
// одной задачи (см. failNotified).
const failNotifyDebounce = time.Hour
// SetNotifier подключает исходящие пинги (до запуска Run).
func (w *Worker) SetNotifier(n Notifier) { w.notifier = n }
// SetScanner подключает пересканирование Jellyfin (до запуска Run).
func (w *Worker) SetScanner(s Scanner) { w.scanner = s }
// New собирает воркер. recognizer/layouter могут быть nil (Ф1 без Ф3-ступеней
// распознавания и раскладки) — тогда completed-задачи не двигаются дальше.
func New(st Store, qb QBittorrent, rec Recognizer, lay Layouter, cfg Config, log *slog.Logger) *Worker {
return &Worker{
store: st,
qbt: qb,
recognizer: rec,
layouter: lay,
cfg: cfg,
log: log,
now: time.Now,
newID: defaultBatchID,
failNotified: map[string]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 — идентификатор батча раскладки (ULID, единая точка
// генерации id — internal/ident; сортируем по времени, удобен в логах).
func defaultBatchID() string {
return ident.NewID()
}
// scoped кладёт в ctx scoped-логгер загрузки (capability + download_id
// [+ infohash]); стадии и внешние клиенты достают его из ctx и дописывают эти
// ключи на каждую запись сами — без ручного доклеивания download_id.
func (w *Worker) scoped(ctx context.Context, capability string, id string, infohash string) context.Context {
log := w.log.With("capability", capability, "download_id", id)
if infohash != "" {
log = log.With("infohash", infohash)
}
return logctx.With(ctx, log)
}
// Run крутит цикл поллинга до отмены ctx.
func (w *Worker) Run(ctx context.Context) {
w.log.Info("worker started", "poll_interval", w.cfg.PollInterval, "category", w.cfg.Category)
t := time.NewTicker(w.cfg.PollInterval)
defer t.Stop()
w.pollOnce(ctx)
for {
select {
case <-ctx.Done():
w.log.Info("worker stopped")
return
case <-t.C:
w.pollOnce(ctx)
}
}
}
func (w *Worker) pollOnce(ctx context.Context) {
if err := w.Poll(ctx); err != nil {
w.log.Warn("poll failed", "error", err)
}
// Ф3: распознаём завершённые загрузки (и перезапускаем по подсказке).
if w.recognizer != nil {
w.recognizePending(ctx)
}
}
// Poll сверяет активные задачи с состоянием qBittorrent и двигает их.
// Листаем все торренты (а не только свою категорию), чтобы reconcile нашёл и
// усыновлённые по тегу раздачи, а discovery — увидел новые.
func (w *Worker) Poll(ctx context.Context) error {
torrents, err := w.qbt.Torrents(ctx, "")
if err != nil {
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 != "" {
key := strings.ToLower(h)
byHash[key] = t
live[key] = l
}
}
}
// Снимок зависит только от torrents — свопаем сразу, до store-операций
// ниже (их ранний return по ошибке не должен лишать UI свежей телеметрии).
w.setLive(live)
w.mu.Lock()
defer w.mu.Unlock()
// Усыновляем новые раздачи с нашей категорией/тегом до reconcile.
w.discover(ctx, torrents)
active, err := w.store.ListDownloadsByState(ctx, store.StateDownloading)
if err != nil {
return fmt.Errorf("poll: list active: %w", err)
}
for _, d := range active {
if len(d.Infohashes) == 0 {
continue // нечем сопоставить (в Ф1 не случается: magnet всегда с infohash)
}
t, ok := torrentFor(d, byHash)
if !ok {
w.log.Warn("active download not found in qbittorrent",
"capability", capIngest, "download_id", d.ID, "infohash", d.PrimaryInfohash())
continue
}
w.captureInfohashes(ctx, d, t)
w.captureSourceAddedAt(ctx, d, t)
w.reconcile(ctx, d, t)
}
// Сверка разложенных задач с реальностью (источник в qBit + хардлинки на ФС)
// — отдельно от активных, по двумерной матрице (см. state-reconciliation).
w.reconcileDesync(ctx, byHash)
// Восстановление задач, упавших по нашей нетерпеливости (magnet_timeout/
// stalled), если их источник в qBittorrent ожил и продвинулся.
w.reconcileRecovery(ctx, byHash)
return nil
}
// reconcile двигает одну задачу по состоянию её торрента. Вызывается под
// w.mu.
func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent) {
ctx = w.scoped(ctx, capIngest, d.ID, d.PrimaryInfohash())
switch classify(t.State) {
case classReady:
w.transition(ctx, d, store.StateCompleted, "", "")
case classErrored:
w.transition(ctx, d, store.StateFailed, errCodeQbitError, "qBittorrent state: "+t.State)
case classDownloading:
w.checkTimeouts(ctx, d, t)
case classBusy:
// moving/checking — ждём, файлы ещё не на финальном месте.
}
}
// checkTimeouts помечает зависшие задачи. Возраст считаем от факта в
// qBittorrent (added_on), а не от created_at: базис переживает retry и
// усыновление раздачи (см. design download-failure-recovery). magnet_timeout —
// редкий страховочный предохранитель (дефолт 24h); настоящие провалы ловит
// classErrored, а ожившие задачи воскрешает reconcileRecovery.
func (w *Worker) checkTimeouts(ctx context.Context, d store.Download, t qbt.Torrent) {
age := w.torrentAge(d, t)
switch {
case isMeta(t.State) && w.cfg.MagnetTimeout > 0 && age > w.cfg.MagnetTimeout:
w.transition(ctx, d, store.StateFailed, errCodeMagnetTimeout,
fmt.Sprintf("no metadata after %s", age.Truncate(time.Second)))
case isStalledDL(t.State) && w.cfg.StuckAfter > 0 && age > w.cfg.StuckAfter:
w.transition(ctx, d, store.StateStuck, errCodeStalled,
fmt.Sprintf("stalled for %s", age.Truncate(time.Second)))
}
}
// captureSourceAddedAt однократно сохраняет время добавления торрента в
// qBittorrent (added_on) у задачи — базис сортировки списка. Пишем только при
// первом наблюдении (в БД source_added_at ещё пуст, SQL-гард в store); значение
// неизменно, поэтому повторные тики его не трогают. Учётная операция: её сбой не
// двигает задачу, лишь логируем WARN. Вызывается под w.mu.
func (w *Worker) captureSourceAddedAt(ctx context.Context, d store.Download, t qbt.Torrent) {
if d.SourceAddedAt.Valid || t.AddedOn <= 0 {
return
}
if err := w.store.SetSourceAddedAt(ctx, d.ID, time.Unix(t.AddedOn, 0)); err != nil {
w.log.Warn("capture source_added_at failed",
"capability", capIngest, "download_id", d.ID, "error", err)
}
}
// torrentFor ищет торрент загрузки в карте byHash по любому из её хешей.
func torrentFor(d store.Download, byHash map[string]qbt.Torrent) (qbt.Torrent, bool) {
for _, h := range d.HashList() {
if t, ok := byHash[h]; ok {
return t, true
}
}
return qbt.Torrent{}, false
}
// captureInfohashes дописывает загрузке хеши, которые qBittorrent знает, а мы
// ещё нет (гибридный торрент раскрывает v1+v2 после получения метаданных).
// Хеши собирает torrentHashes (усечённый t.Hash v2-only раздач отсеян).
// AddInfohashes под гардом: хеш, которым владеет другая активная задача,
// дописан не будет (ErrInfohashTaken). Учётная операция: сбой не двигает
// задачу, лишь логируем WARN. Под w.mu.
func (w *Worker) captureInfohashes(ctx context.Context, d store.Download, t qbt.Torrent) {
known := d.HashList()
var missing []string
for _, h := range torrentHashes(t) {
if !slices.Contains(known, h) {
missing = append(missing, h)
}
}
if len(missing) == 0 {
return
}
if err := w.store.AddInfohashes(ctx, d.ID, missing); err != nil {
w.log.Warn("capture infohashes failed",
"capability", capIngest, "download_id", d.ID, "error", err)
}
}
// torrentAge — возраст торрента: от added_on в qBittorrent (надёжный базис,
// переживает retry/усыновление), с фолбэком на created_at задачи, если qBit не
// отдал added_on.
func (w *Worker) torrentAge(d store.Download, t qbt.Torrent) time.Duration {
if t.AddedOn > 0 {
return w.now().Sub(time.Unix(t.AddedOn, 0).UTC())
}
created, err := d.CreatedTime()
if err != nil {
// Ни added_on от qBit, ни разбираемого created_at — возраст неизвестен,
// таймауты не сработают; фиксируем диагностикой.
w.log.Warn("cannot determine torrent age",
"capability", capIngest, "download_id", d.ID,
"created_at", d.CreatedAt, "error", err)
return 0
}
return w.now().Sub(created)
}
// transition пишет новое состояние и логирует переход.
func (w *Worker) transition(ctx context.Context, d store.Download, state store.State, code, msg string) {
// FromOr, а не From: если вызывающий не завёл scoped-логгер, падаем на
// w.log (настроенный), а не на slog.Default().
log := logctx.FromOr(ctx, w.log)
if err := w.store.SetDownloadState(ctx, d.ID, state, code, msg); err != nil {
log.Error("state transition failed", "from", d.State, "to", state, "error", err)
return
}
log.Info("state transition", "from", d.State, "to", state, "code", code)
// Пинги — неблокирующе и в отдельном контексте: вызов уходит в сеть, а
// мы под w.mu (Notify читает состояние уже после освобождения замка).
if w.notifier != nil {
switch state {
case store.StateReview:
go w.notifier.Notify(context.Background(), d.ID, EventReview)
case store.StateDone:
go w.notifier.Notify(context.Background(), d.ID, EventDone)
case store.StateOrphaned:
go w.notifier.Notify(context.Background(), d.ID, EventOrphaned)
case store.StateTargetMissing:
go w.notifier.Notify(context.Background(), d.ID, EventTargetMissing)
case store.StateFailed, store.StateStuck:
if w.shouldNotifyFail(d.ID) {
go w.notifier.Notify(context.Background(), d.ID, EventFailed)
}
}
}
// Раскладка завершена — просим Jellyfin пересканировать библиотеку, чтобы
// новые файлы быстрее появились в проигрывателе. Тоже неблокирующе и вне
// w.mu; недоступность Jellyfin не влияет на состояние задачи.
if w.scanner != nil && state == store.StateDone {
// Скан Jellyfin — неблокирующе и вне w.mu, в фоновом ctx со scoped-логгером
// (download_id для корреляции ext.*-записи клиента). Недоступность Jellyfin
// на задачу не влияет; ошибку вызова логирует сам клиент (ext.*), здесь гасим.
gctx := w.scoped(context.Background(), capFileLayout, d.ID, d.PrimaryInfohash())
go func() { _ = w.scanner.RefreshLibraries(gctx) }()
}
}
// shouldNotifyFail дебаунсит повторные уведомления о падении одной задачи
// (мерцающий stalled-торрент: stuck↔downloading), чтобы не спамить. Вызывается
// под w.mu. НЕ сбрасываем запись при восстановлении — иначе дебаунс не гасил бы
// флаппинг.
func (w *Worker) shouldNotifyFail(id string) bool {
now := w.now()
if last, ok := w.failNotified[id]; ok && now.Sub(last) < failNotifyDebounce {
return false
}
w.failNotified[id] = now
// Лёгкая чистка устаревших записей, чтобы карта не росла без предела.
for k, t := range w.failNotified {
if now.Sub(t) >= failNotifyDebounce {
delete(w.failNotified, k)
}
}
return true
}
// Cancel отклоняет задачу. Торрент в qBittorrent не трогаем — он продолжает
// раздачу (источник неприкосновенен).
func (w *Worker) Cancel(ctx context.Context, id string) error {
w.mu.Lock()
defer w.mu.Unlock()
d, err := w.store.GetDownload(ctx, id)
if err != nil {
return fmt.Errorf("cancel: %w", err)
}
if d.State.IsTerminal() {
return fmt.Errorf("cancel: download %s is already terminal (%s)", id, d.State)
}
if err := w.store.SetDownloadState(ctx, id, store.StateCancelled, "", ""); err != nil {
return fmt.Errorf("cancel: %w", err)
}
logctx.From(w.scoped(ctx, capReview, id, d.PrimaryInfohash())).Info("download cancelled", "from", d.State)
return nil
}
// Retry повторяет застрявшую/упавшую задачу: заново отдаёт источник в
// qBittorrent и возвращает в downloading.
func (w *Worker) Retry(ctx context.Context, id string) error {
w.mu.Lock()
defer w.mu.Unlock()
d, err := w.store.GetDownload(ctx, id)
if err != nil {
return fmt.Errorf("retry: %w", err)
}
if d.State != store.StateFailed && d.State != store.StateStuck {
return fmt.Errorf("retry: download %s is %s, only failed/stuck are retriable", id, d.State)
}
// Если раздача уже жива в qBittorrent — перецепляемся к ней, повторный Add
// не нужен (и вреден: вслепую дублировал бы торрент). Add — только когда
// источника в qBittorrent нет. Базис таймаута берётся от added_on, поэтому
// возврат в downloading не роняет задачу снова на ближайшем тике.
alive := false
if hashes := d.HashList(); len(hashes) > 0 {
_, alive, err = w.torrentByInfohash(ctx, hashes)
if err != nil {
return fmt.Errorf("retry: %w", err)
}
}
// Гард инварианта — ДО побочного эффекта в qBittorrent: пока задача лежала
// в failed, тем же infohash могла завладеть другая активная задача — тогда
// отказываем, не добавив торрент повторно (см. design ulid-identity, D4).
if err := w.store.ActivateIfNoOtherActive(ctx, id, store.StateDownloading, "", ""); err != nil {
if errors.Is(err, store.ErrInfohashTaken) {
return fmt.Errorf("retry: для этого торрента уже есть другая активная задача: %w", ErrConflict)
}
return fmt.Errorf("retry: %w", err)
}
if !alive && d.SourceType == store.SourceMagnet {
if err := w.qbt.Add(ctx, qbt.AddRequest{
URLs: []string{d.SourceRef},
Category: w.cfg.Category,
SavePath: w.cfg.SavePath,
}); err != nil {
// Активация уже прошла — откатываем задачу в прежнее состояние,
// чтобы не оставить «качающуюся» задачу без раздачи в qBittorrent.
if rbErr := w.store.SetDownloadState(ctx, id, d.State, d.ErrorCode.String, d.ErrorMsg.String); rbErr != nil {
w.log.Error("retry rollback failed",
"capability", capReview, "download_id", id, "error", rbErr)
}
return fmt.Errorf("retry: add to qbittorrent: %w", err)
}
}
logctx.From(w.scoped(ctx, capReview, id, d.PrimaryInfohash())).Info("download retried", "from", d.State)
return nil
}
// class — класс состояния торрента qBittorrent.
type class int
const (
classDownloading class = iota // ещё качается
classReady // готов к раскладке
classErrored // ошибка
classBusy // moving/checking — переходный момент, ждём
)
// classify относит состояние qBittorrent к классу (см. architecture.md,
// «Завершение в qBittorrent»). Учитываем и v5-имена (stopped* вместо
// paused*).
func classify(state string) class {
switch state {
case "uploading", "stalledUP", "pausedUP", "stoppedUP", "queuedUP", "forcedUP":
return classReady
case "error", "missingFiles":
return classErrored
case "moving", "checkingUP", "checkingResumeData", "allocating":
return classBusy
default:
// downloading, stalledDL, metaDL, forcedMetaDL, queuedDL, checkingDL,
// forcedDL, pausedDL, stoppedDL, unknown — считаем «ещё качается».
return classDownloading
}
}
func isMeta(state string) bool {
return state == "metaDL" || state == "forcedMetaDL"
}
func isStalledDL(state string) bool {
return state == "stalledDL"
}