Восстановление зависших загрузок и уведомления о падении (state-reconciliation)

Долгий metaDL больше не убивается агрессивным таймаутом: дефолт
magnet_timeout 30m → 24h (страховочный предохранитель), базис отсчёта —
added_on из qBittorrent, а не created_at (переживает retry/усыновление).

Авто-восстановление: фоновая сверка возвращает в поток задачи, упавшие по
нашей нетерпеливости (magnet_timeout/stalled), когда источник ожил и
продвинулся за условие падения (downloading/completed по статусу торрента);
qbit_error не воскрешается. Конфликт idempotency (infohash занят другой
активной задачей) — оставляем в failed.

Уведомления: любой переход в failed/stuck пингует автора (включая приёмный
qbit_add через ingest), с дебаунсом против спама при флаппинге stalled.
Ручной retry добавлен в веб-UI и Telegram; Retry перецепляется к живому
торренту вместо слепого Add.

Дельта state-reconciliation влита в живые спеки; обновлён workflow.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

internal/worker/worker.go

@@ -38,6 +38,7 @@ const (
 // 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 int64) (*store.Download, error)
 	SetDownloadState(ctx context.Context, id int64, state store.State, errCode, errMsg string) error
 	SetSourceMissCount(ctx context.Context, id int64, n int) error
@@ -94,6 +95,17 @@ const (
 	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). Вызывается неблокирующе.
@@ -136,8 +148,18 @@ type Worker struct {
 	newID    func() string    // генератор apply_batch_id (подменяется в тестах)
 	notifier Notifier         // опц. исходящие пинги
 	scanner  Scanner          // опц. пересканирование Jellyfin
+
+	// failNotified — дебаунс повторных EventFailed по задаче (download_id →
+	// время последнего пинга). Мерцающий stalled-торрент колеблется
+	// stuck↔downloading; без дебаунса каждый цикл слал бы уведомление. Память
+	// процесса: при рестарте дебаунс сбрасывается — допустимо. Доступ под w.mu.
+	failNotified map[int64]time.Time
 }
 
+// failNotifyDebounce — минимальный интервал между уведомлениями о падении
+// одной задачи (см. failNotified).
+const failNotifyDebounce = time.Hour
+
 // SetNotifier подключает исходящие пинги (до запуска Run).
 func (w *Worker) SetNotifier(n Notifier) { w.notifier = n }
 
@@ -148,14 +170,15 @@ func (w *Worker) SetScanner(s Scanner) { w.scanner = s }
 // распознавания и раскладки) — тогда 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,
+		store:        st,
+		qbt:          qb,
+		recognizer:   rec,
+		layouter:     lay,
+		cfg:          cfg,
+		log:          log,
+		now:          time.Now,
+		newID:        defaultBatchID,
+		failNotified: map[int64]time.Time{},
 	}
 }
 
@@ -246,6 +269,10 @@ func (w *Worker) Poll(ctx context.Context) error {
 	// Сверка разложенных задач с реальностью (источник в qBit + хардлинки на ФС)
 	// — отдельно от активных, по двумерной матрице (см. state-reconciliation).
 	w.reconcileDesync(ctx, byHash)
+
+	// Восстановление задач, упавших по нашей нетерпеливости (magnet_timeout/
+	// stalled), если их источник в qBittorrent ожил и продвинулся.
+	w.reconcileRecovery(ctx, byHash)
 	return nil
 }
 
@@ -257,7 +284,7 @@ func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent)
 	case classReady:
 		w.transition(ctx, d, store.StateCompleted, "", "")
 	case classErrored:
-		w.transition(ctx, d, store.StateFailed, "qbit_error", "qBittorrent state: "+t.State)
+		w.transition(ctx, d, store.StateFailed, errCodeQbitError, "qBittorrent state: "+t.State)
 	case classDownloading:
 		w.checkTimeouts(ctx, d, t)
 	case classBusy:
@@ -265,26 +292,43 @@ func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent)
 	}
 }
 
-// checkTimeouts помечает зависшие задачи. Возраст считаем от created_at:
-// для metaDL это время с момента добавления (огрублённо, но достаточно).
+// 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) {
-	created, err := d.CreatedTime()
-	if err != nil {
-		logctx.From(ctx).Warn("cannot parse created_at", "value", d.CreatedAt, "error", err)
-		return
-	}
-	age := w.now().Sub(created)
+	age := w.torrentAge(d, t)
 
 	switch {
 	case isMeta(t.State) && w.cfg.MagnetTimeout > 0 && age > w.cfg.MagnetTimeout:
-		w.transition(ctx, d, store.StateFailed, "magnet_timeout",
+		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, "stalled",
+		w.transition(ctx, d, store.StateStuck, errCodeStalled,
 			fmt.Sprintf("stalled for %s", age.Truncate(time.Second)))
 	}
 }
 
+// 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-логгер, падаем на
@@ -308,6 +352,10 @@ func (w *Worker) transition(ctx context.Context, d store.Download, state store.S
 			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)
+			}
 		}
 	}
 
@@ -323,6 +371,25 @@ func (w *Worker) transition(ctx context.Context, d store.Download, state store.S
 	}
 }
 
+// shouldNotifyFail дебаунсит повторные уведомления о падении одной задачи
+// (мерцающий stalled-торрент: stuck↔downloading), чтобы не спамить. Вызывается
+// под w.mu. НЕ сбрасываем запись при восстановлении — иначе дебаунс не гасил бы
+// флаппинг.
+func (w *Worker) shouldNotifyFail(id int64) 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 int64) error {
@@ -356,7 +423,18 @@ func (w *Worker) Retry(ctx context.Context, id int64) error {
 	if d.State != store.StateFailed && d.State != store.StateStuck {
 		return fmt.Errorf("retry: download %d is %s, only failed/stuck are retriable", id, d.State)
 	}
-	if d.SourceType == store.SourceMagnet {
+	// Если раздача уже жива в qBittorrent — перецепляемся к ней, повторный Add
+	// не нужен (и вреден: вслепую дублировал бы торрент). Add — только когда
+	// источника в qBittorrent нет. Базис таймаута берётся от added_on, поэтому
+	// возврат в downloading не роняет задачу снова на ближайшем тике.
+	alive := false
+	if d.Infohash.Valid {
+		_, alive, err = w.torrentByInfohash(ctx, d.Infohash.String)
+		if err != nil {
+			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,