Обработка рассинхрона состояния с реальностью (state-reconciliation)
Распознаём ручное удаление источника (раздача в qBittorrent) и/или цели (разложенные хардлинки) и отражаем его в состоянии задачи, без автодействий. - Новая capability state-reconciliation (OpenSpec): фоновая сверка по матрице «источник × цель» → состояния target_missing/orphaned/deleted, переходы и самовосстановление (healing). - worker: reconcileDesync в Poll (только разложенные/desync-задачи), дебаунс пропажи источника (порог [worker].source_missing_threshold) и синхронный preflight перед действиями (relink/recognize/apply/undo) — не доверяем state в БД. - layout.Undo: отказ снять последнюю копию (nlink<=1 или нет источника), отказ всего батча без частичного отката (ErrLastCopy). - store: единый список terminalStates для IsTerminal и FindActiveByInfohash (иначе семантика «активности» разъезжается), столбец source_miss_count, миграция 0003. - httpapi/web и Telegram: показ новых состояний и уведомления о рассинхроне. - Доки: workflow.md, jellyfin-layout.md, database.md (+0003), config. Change заархивирован в openspec/changes/archive, дельта влита в openspec/specs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,196 @@
|
||||
package worker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"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/store"
|
||||
)
|
||||
|
||||
// desyncStates — состояния, которые ведёт сверка с реальностью: уже
|
||||
// разложенные (done) и сами состояния рассинхрона. Активные и
|
||||
// пользовательски-терминальные (reverted/cancelled/failed/stuck) сверка не
|
||||
// трогает (см. state-reconciliation).
|
||||
var desyncStates = []store.State{
|
||||
store.StateDone,
|
||||
store.StateTargetMissing,
|
||||
store.StateOrphaned,
|
||||
store.StateDeleted,
|
||||
}
|
||||
|
||||
// deriveState выводит состояние задачи из двумерной матрицы «источник × цель»
|
||||
// (см. state-reconciliation, D1).
|
||||
func deriveState(sourcePresent, targetPresent bool) store.State {
|
||||
switch {
|
||||
case sourcePresent && targetPresent:
|
||||
return store.StateDone
|
||||
case sourcePresent && !targetPresent:
|
||||
return store.StateTargetMissing
|
||||
case !sourcePresent && targetPresent:
|
||||
return store.StateOrphaned
|
||||
default:
|
||||
return store.StateDeleted
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileDesync сверяет разложенные/desync-задачи с реальностью. Вызывается
|
||||
// из Poll под w.mu. byHash — карта присутствующих в qBittorrent раздач.
|
||||
func (w *Worker) reconcileDesync(ctx context.Context, byHash map[string]qbt.Torrent) {
|
||||
cands, err := w.store.ListDownloadsByState(ctx, desyncStates...)
|
||||
if err != nil {
|
||||
w.log.Warn("reconcile list desync failed", "capability", capIngest, "error", err)
|
||||
return
|
||||
}
|
||||
for _, d := range cands {
|
||||
w.reconcileOneDesync(ctx, d, byHash)
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileOneDesync сверяет одну задачу: вычисляет присутствие источника (с
|
||||
// дебаунсом) и цели, выводит состояние и переходит при изменении.
|
||||
func (w *Worker) reconcileOneDesync(ctx context.Context, d store.Download, byHash map[string]qbt.Torrent) {
|
||||
if !d.Infohash.Valid {
|
||||
return // нечем сопоставить источник
|
||||
}
|
||||
ctx = w.scoped(ctx, capIngest, d.ID, d.Infohash.String)
|
||||
_, sourceSeen := byHash[strings.ToLower(d.Infohash.String)]
|
||||
|
||||
// Дебаунс пропажи источника: считаем удалённым только после порога подряд
|
||||
// идущих промахов; любое появление сбрасывает счётчик.
|
||||
sourcePresent := w.debounceSource(ctx, d, sourceSeen)
|
||||
|
||||
targetPresent, err := w.targetPresent(ctx, d.ID)
|
||||
if err != nil {
|
||||
logctx.From(ctx).Warn("reconcile target probe failed", "error", err)
|
||||
return // не можем проверить цель — не дёргаем состояние
|
||||
}
|
||||
|
||||
want := deriveState(sourcePresent, targetPresent)
|
||||
if want == d.State {
|
||||
return
|
||||
}
|
||||
w.transition(ctx, d, want, "reconcile", reconcileReason(sourcePresent, targetPresent))
|
||||
}
|
||||
|
||||
// debounceSource обновляет счётчик промахов источника и возвращает, считать ли
|
||||
// источник присутствующим с учётом порога. Запись счётчика — только при его
|
||||
// изменении (без лишних UPDATE на каждом тике).
|
||||
func (w *Worker) debounceSource(ctx context.Context, d store.Download, sourceSeen bool) bool {
|
||||
threshold := max(w.cfg.SourceMissingThreshold, 1)
|
||||
if sourceSeen {
|
||||
if d.SourceMissCount != 0 {
|
||||
if err := w.store.SetSourceMissCount(ctx, d.ID, 0); err != nil {
|
||||
logctx.From(ctx).Warn("reconcile reset miss count failed", "error", err)
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
miss := d.SourceMissCount + 1
|
||||
if err := w.store.SetSourceMissCount(ctx, d.ID, miss); err != nil {
|
||||
logctx.From(ctx).Warn("reconcile bump miss count failed", "error", err)
|
||||
}
|
||||
// До порога источник трактуется как присутствующий — задача не дёргается.
|
||||
return miss < threshold
|
||||
}
|
||||
|
||||
// targetPresent сообщает, существуют ли разложенные хардлинки задачи. Цель
|
||||
// считается присутствующей, только если существуют ВСЕ ссылки последнего
|
||||
// батча; частичная пропажа — это отсутствие цели (библиотека сломана → relink).
|
||||
func (w *Worker) targetPresent(ctx context.Context, id int64) (bool, error) {
|
||||
batch, err := w.store.LatestBatchID(ctx, id)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("latest batch: %w", err)
|
||||
}
|
||||
if batch == "" {
|
||||
return false, nil // ничего не разложено — цели нет
|
||||
}
|
||||
rows, err := w.store.ListFileLinksByBatch(ctx, batch)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("list links: %w", err)
|
||||
}
|
||||
laidOut := 0
|
||||
for _, r := range rows {
|
||||
// Все статусы, означающие реальный файл на диске: только что слинкован,
|
||||
// скопирован (фолбэк) или уже существовал тем же inode (идемпотентный
|
||||
// повтор apply). Прочие (collision и т.п.) — не наша разложенная цель.
|
||||
if !isLaidOut(r.Status) {
|
||||
continue
|
||||
}
|
||||
laidOut++
|
||||
if _, err := os.Lstat(r.DstPath); err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return false, nil
|
||||
}
|
||||
return false, fmt.Errorf("stat target %q: %w", r.DstPath, err)
|
||||
}
|
||||
}
|
||||
return laidOut > 0, nil
|
||||
}
|
||||
|
||||
// isLaidOut сообщает, означает ли статус file_link реально лежащий на ФС файл
|
||||
// нашей раскладки (а не коллизию или иной не-разложенный исход).
|
||||
func isLaidOut(status string) bool {
|
||||
switch layout.LinkStatus(status) {
|
||||
case layout.StatusLinked, layout.StatusCopied, layout.StatusExists:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileReason — человекочитаемая причина перехода для лога/UI.
|
||||
func reconcileReason(sourcePresent, targetPresent bool) string {
|
||||
switch {
|
||||
case sourcePresent && targetPresent:
|
||||
return "источник и цель на месте"
|
||||
case sourcePresent && !targetPresent:
|
||||
return "цель удалена, источник на месте"
|
||||
case !sourcePresent && targetPresent:
|
||||
return "источник удалён, цель (последняя копия) на месте"
|
||||
default:
|
||||
return "удалены и источник, и цель"
|
||||
}
|
||||
}
|
||||
|
||||
// --- Синхронный preflight перед действием (не доверяем state в БД) ---
|
||||
|
||||
// ensureSourcePresent синхронно (без дебаунса) проверяет, что раздача есть в
|
||||
// qBittorrent прямо сейчас. При отсутствии приводит состояние к реальности и
|
||||
// возвращает ErrConflict. Недоступность qBittorrent — честный отказ операции.
|
||||
func (w *Worker) ensureSourcePresent(ctx context.Context, d *store.Download, op string) error {
|
||||
if !d.Infohash.Valid {
|
||||
return fmt.Errorf("%s: download %d has no infohash", op, d.ID)
|
||||
}
|
||||
_, ok, err := w.torrentByInfohash(ctx, d.Infohash.String)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: %w", op, err)
|
||||
}
|
||||
if ok {
|
||||
return nil
|
||||
}
|
||||
// Источник пропал — немедленно приводим состояние к реальности.
|
||||
w.reconcileToReality(ctx, *d, false)
|
||||
return fmt.Errorf("%s: источник удалён из qBittorrent: %w", op, ErrConflict)
|
||||
}
|
||||
|
||||
// reconcileToReality выводит и проставляет состояние по уже известному факту об
|
||||
// источнике (sourcePresent) и фактически проверенной цели. Используется
|
||||
// preflight-проверками: немедленно, без дебаунса.
|
||||
func (w *Worker) reconcileToReality(ctx context.Context, d store.Download, sourcePresent bool) {
|
||||
targetPresent, err := w.targetPresent(ctx, d.ID)
|
||||
if err != nil {
|
||||
logctx.FromOr(ctx, w.log).Warn("preflight target probe failed", "error", err)
|
||||
return
|
||||
}
|
||||
want := deriveState(sourcePresent, targetPresent)
|
||||
if want == d.State {
|
||||
return
|
||||
}
|
||||
w.transition(ctx, d, want, "reconcile", reconcileReason(sourcePresent, targetPresent))
|
||||
}
|
||||
Reference in New Issue
Block a user