Владение целевым путём при повторной раскладке (state-reconciliation)

Завершённая загрузка ложно «воскресала» из deleted в orphaned, когда её
целевой путь переиспользовала другая загрузка (повторная закачка того же
фильма в другом качестве): сверка проверяла лишь существование пути, не
проверяя, что файл по нему — наша раскладка.

Вводим инвариант «один целевой путь — один владелец»:

- при успешной раскладке на освободившийся чужой путь владение переходит
  к новой загрузке — прежние file_link на этот путь помечаются статусом
  superseded и перестают считаться целью при сверке;
- deleted исключён из desyncStates — терминальное состояние больше не
  переоценивается (источник к нему не вернётся из-за идемпотентности,
  цель отбирается переходом владения);
- Undo снимает только реально свои разложенные ссылки (superseded
  пропускает — файл по пути теперь чужой хардлинк);
- ошибку перехода владения трактуем как некритичную (WARN-and-continue):
  файлы уже разложены, рассинхрон чужих задач исправит следующий тик.

Без миграции схемы (status — TEXT). Дельта влита в основную спеку,
обновлены workflow.md и jellyfin-layout.md.

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

internal/layout/layout.go

@@ -213,10 +213,11 @@ func (l *Layouter) seriesDst(root, folder, base string, f *PlanFile) (string, Ki
 type LinkStatus string
 
 const (
-	StatusLinked    LinkStatus = "linked"    // хардлинк создан
-	StatusCopied    LinkStatus = "copied"    // хардлинк невозможен — файл скопирован (фолбэк)
-	StatusExists    LinkStatus = "exists"    // уже была (тот же inode) — идемпотентно
-	StatusCollision LinkStatus = "collision" // цель занята другим файлом
+	StatusLinked     LinkStatus = "linked"     // хардлинк создан
+	StatusCopied     LinkStatus = "copied"     // хардлинк невозможен — файл скопирован (фолбэк)
+	StatusExists     LinkStatus = "exists"     // уже была (тот же inode) — идемпотентно
+	StatusCollision  LinkStatus = "collision"  // цель занята другим файлом
+	StatusSuperseded LinkStatus = "superseded" // путь перехватила другая загрузка (см. state-reconciliation, владение путём)
 )
 
 // Result — итог по одной ссылке.

internal/store/recognition.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 )
 
 // Recognition — строка таблицы recognition (попытка распознавания).
@@ -195,6 +196,34 @@ VALUES (?, ?, ?, ?, ?, ?)`
 	return nil
 }
 
+// SupersedeForeignLinks помечает статусом superseded ссылки ДРУГИХ загрузок,
+// разложенные по тем же dst_path, что и переданные. Реализует инвариант
+// «один целевой путь — один владелец» (см. state-reconciliation): свежая
+// раскладка забирает владение освободившимся путём, прежние записи перестают
+// считаться целью при сверке. Затрагивает только активные статусы раскладки
+// (linked/copied/exists) и не трогает саму загрузку (download_id != ?).
+func (s *Store) SupersedeForeignLinks(ctx context.Context, downloadID int64, dstPaths []string) error {
+	if len(dstPaths) == 0 {
+		return nil
+	}
+	ph := make([]string, len(dstPaths))
+	args := make([]any, 0, len(dstPaths)+1)
+	args = append(args, downloadID)
+	for i, p := range dstPaths {
+		ph[i] = "?"
+		args = append(args, p)
+	}
+	q := fmt.Sprintf(`
+UPDATE file_link SET status = 'superseded'
+WHERE download_id != ?
+  AND status IN ('linked', 'copied', 'exists')
+  AND dst_path IN (%s)`, strings.Join(ph, ", "))
+	if _, err := s.DB.ExecContext(ctx, q, args...); err != nil {
+		return fmt.Errorf("supersede foreign links: %w", err)
+	}
+	return nil
+}
+
 // LatestBatchID возвращает apply_batch_id последнего применённого батча
 // загрузки (для undo) либо пустую строку, если ссылок нет.
 func (s *Store) LatestBatchID(ctx context.Context, downloadID int64) (string, error) {

internal/store/recognition_test.go

@@ -159,6 +159,56 @@ func TestFileLinks_BatchLifecycle(t *testing.T) {
 	}
 }
 
+func TestSupersedeForeignLinks(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	owner := seedDownload(t, st)
+	foreign, err := st.CreateDownload(ctx,
+		newDownloading("bbccddeeff00112233445566778899aabbccddee"))
+	if err != nil {
+		t.Fatalf("seed foreign: %v", err)
+	}
+
+	shared := "/m/Movie (2024).mkv"
+	// foreign разложена по shared (linked) и по своему пути (exists);
+	// owner разложен по shared и по третьему пути.
+	if err := st.CreateFileLinks(ctx, []FileLink{
+		{DownloadID: foreign, ApplyBatchID: "f", SrcPath: "/d/f.mkv", DstPath: shared, Kind: "video", Status: "linked"},
+		{DownloadID: foreign, ApplyBatchID: "f", SrcPath: "/d/g.mkv", DstPath: "/m/Other (2024).mkv", Kind: "video", Status: "exists"},
+		{DownloadID: owner, ApplyBatchID: "o", SrcPath: "/d/o.mkv", DstPath: shared, Kind: "video", Status: "linked"},
+	}); err != nil {
+		t.Fatalf("create links: %v", err)
+	}
+
+	if err := st.SupersedeForeignLinks(ctx, owner, []string{shared}); err != nil {
+		t.Fatalf("supersede: %v", err)
+	}
+
+	links, _ := st.ListFileLinksByBatch(ctx, "f")
+	for _, l := range links {
+		switch l.DstPath {
+		case shared:
+			if l.Status != "superseded" {
+				t.Errorf("чужая ссылка на %q = %q, want superseded", shared, l.Status)
+			}
+		default: // /m/Other — другой путь, не трогаем
+			if l.Status != "exists" {
+				t.Errorf("ссылка на %q = %q, want exists (не тронута)", l.DstPath, l.Status)
+			}
+		}
+	}
+	// Свою ссылку owner не суперсидит (download_id != self).
+	own, _ := st.ListFileLinksByBatch(ctx, "o")
+	if len(own) != 1 || own[0].Status != "linked" {
+		t.Errorf("своя ссылка = %+v, want linked", own)
+	}
+
+	// Пустой список путей — no-op, без ошибки.
+	if err := st.SupersedeForeignLinks(ctx, owner, nil); err != nil {
+		t.Errorf("пустой dstPaths должен быть no-op: %v", err)
+	}
+}
+
 func TestCandidates_Lifecycle(t *testing.T) {
 	st := newTestStore(t)
 	ctx := context.Background()

internal/worker/reconcile.go

@@ -14,14 +14,16 @@ import (
 )
 
 // desyncStates — состояния, которые ведёт сверка с реальностью: уже
-// разложенные (done) и сами состояния рассинхрона. Активные и
+// разложенные (done) и восстановимые состояния рассинхрона. deleted сюда не
+// входит — оно терминально и сверкой не переоценивается (источник к
+// терминальной задаче не вернётся из-за идемпотентности, а цель отбирается
+// переходом владения путём; см. state-reconciliation). Активные и
 // пользовательски-терминальные (reverted/cancelled/failed/stuck) сверка не
-// трогает (см. state-reconciliation).
+// трогает.
 var desyncStates = []store.State{
 	store.StateDone,
 	store.StateTargetMissing,
 	store.StateOrphaned,
-	store.StateDeleted,
 }
 
 // deriveState выводит состояние задачи из двумерной матрицы «источник × цель»

internal/worker/reconcile_test.go

@@ -99,6 +99,18 @@ func TestReconcilePartialTargetLoss(t *testing.T) {
 	}
 }
 
+func TestReconcileSkipsDeleted(t *testing.T) {
+	// deleted терминально: сверка его не переоценивает, даже если по бывшему
+	// пути снова появился файл (его забрала другая загрузка) и источника нет.
+	f := newReconcileFixture(t, store.StateDeleted, false, true)
+	if err := f.w.Poll(context.Background()); err != nil {
+		t.Fatalf("Poll: %v", err)
+	}
+	if got := f.st.downloads[1].State; got != store.StateDeleted {
+		t.Errorf("state = %q, want deleted (сверка не трогает терминальное)", got)
+	}
+}
+
 func TestReconcileDebounce(t *testing.T) {
 	// Порог 3: первые два промаха не помечают, третий — помечает orphaned;
 	// возврат источника лечит обратно и сбрасывает счётчик.

internal/worker/review.go

@@ -280,6 +280,22 @@ func (w *Worker) linkPlan(ctx context.Context, d *store.Download, plan recognize
 		if err := w.store.CreateFileLinks(ctx, fl); err != nil {
 			return fmt.Errorf("persist links: %w", err)
 		}
+		// Инвариант «один целевой путь — один владелец»: забираем владение
+		// фактически разложенными путями у прежних загрузок (см.
+		// state-reconciliation). Сверка тех загрузок перестанет считать эти
+		// пути своей целью и не «воскресит» их. Это учётная операция, не
+		// безопасность данных: файлы уже разложены, поэтому её сбой не
+		// стрэндит задачу в linking — логируем WARN и доводим до done, а
+		// рассинхрон чужих задач исправит следующий тик сверки.
+		owned := make([]string, 0, len(fl))
+		for _, l := range fl {
+			if isLaidOut(l.Status) {
+				owned = append(owned, l.DstPath)
+			}
+		}
+		if err := w.store.SupersedeForeignLinks(ctx, d.ID, owned); err != nil {
+			logctx.From(ctx).Warn("supersede foreign links failed", "error", err)
+		}
 	}
 
 	if applyErr != nil {
@@ -494,9 +510,15 @@ func (w *Worker) Undo(ctx context.Context, id int64) error {
 	if err != nil {
 		return fmt.Errorf("undo: %w", err)
 	}
-	links := make([]layout.Link, len(rows))
-	for i, r := range rows {
-		links[i] = layout.Link{Src: r.SrcPath, Dst: r.DstPath, Kind: layout.Kind(r.Kind)}
+	// Снимаем только реально разложенные нами ссылки. superseded-строки —
+	// путь забрала другая загрузка (см. state-reconciliation, владение
+	// путём); файл по нему теперь её хардлинк, трогать его нельзя.
+	links := make([]layout.Link, 0, len(rows))
+	for _, r := range rows {
+		if !isLaidOut(r.Status) {
+			continue
+		}
+		links = append(links, layout.Link{Src: r.SrcPath, Dst: r.DstPath, Kind: layout.Kind(r.Kind)})
 	}
 	n, err := w.layouter.Undo(ctx, links)
 	if err != nil {

internal/worker/review_test.go

@@ -353,6 +353,25 @@ func (m *memStore) CreateFileLinks(_ context.Context, links []store.FileLink) er
 	m.links = append(m.links, links...)
 	return nil
 }
+func (m *memStore) SupersedeForeignLinks(_ context.Context, downloadID int64, dstPaths []string) error {
+	if len(dstPaths) == 0 {
+		return nil
+	}
+	want := make(map[string]bool, len(dstPaths))
+	for _, p := range dstPaths {
+		want[p] = true
+	}
+	for i := range m.links {
+		l := &m.links[i]
+		if l.DownloadID == downloadID || !want[l.DstPath] {
+			continue
+		}
+		if isLaidOut(l.Status) {
+			l.Status = string(layout.StatusSuperseded)
+		}
+	}
+	return nil
+}
 func (m *memStore) LatestBatchID(_ context.Context, id int64) (string, error) {
 	for i := len(m.links) - 1; i >= 0; i-- {
 		if m.links[i].DownloadID == id {
@@ -769,6 +788,72 @@ func TestApply_CollisionStaysReview(t *testing.T) {
 	}
 }
 
+func TestApply_SupersedesForeignOwnerOfPath(t *testing.T) {
+	// Прежняя загрузка (id=2) была разложена по тому же пути, её файл удалён.
+	// Новая раскладка (id=1) ложится на освободившийся путь и забирает владение.
+	plan := seriesResult().Plan
+	f := newApplyFixture(t, plan)
+	e01 := filepath.Join(f.series, "Show (2006)", "Season 02", "Show (2006) S02E01.mkv")
+	f.st.put(completedDownload(2))
+	f.st.links = append(f.st.links, store.FileLink{
+		DownloadID: 2, ApplyBatchID: "old", SrcPath: "/old/e1.mkv", DstPath: e01,
+		Kind: "video", Status: "linked",
+	})
+
+	if err := f.w.Apply(context.Background(), 1); err != nil {
+		t.Fatalf("Apply: %v", err)
+	}
+
+	// Чужая ссылка на перехваченный путь — superseded.
+	var foreign *store.FileLink
+	for i := range f.st.links {
+		if f.st.links[i].DownloadID == 2 {
+			foreign = &f.st.links[i]
+		}
+	}
+	if foreign == nil || foreign.Status != string(layout.StatusSuperseded) {
+		t.Errorf("чужая ссылка status = %v, want superseded", foreign)
+	}
+	// Свои ссылки (id=1) не тронуты — download_id != self.
+	for _, l := range f.st.links {
+		if l.DownloadID == 1 && !isLaidOut(l.Status) {
+			t.Errorf("своя ссылка %q стала %q, ожидали разложенную", l.DstPath, l.Status)
+		}
+	}
+	// Прежняя загрузка больше не владеет путём → цель отсутствует.
+	present, err := f.w.targetPresent(context.Background(), 2)
+	if err != nil {
+		t.Fatalf("targetPresent: %v", err)
+	}
+	if present {
+		t.Error("targetPresent(2) = true, want false (путь забран)")
+	}
+}
+
+func TestApply_CollisionKeepsForeignOwner(t *testing.T) {
+	// Если файл прежней загрузки ВСЁ ЕЩЁ на месте — коллизия → review, владение
+	// не отбирается (supersede не срабатывает для не-разложенного пути).
+	plan := seriesResult().Plan
+	f := newApplyFixture(t, plan)
+	e01 := filepath.Join(f.series, "Show (2006)", "Season 02", "Show (2006) S02E01.mkv")
+	_ = os.MkdirAll(filepath.Dir(e01), 0o755)
+	_ = os.WriteFile(e01, []byte("foreign"), 0o644)
+	f.st.put(completedDownload(2))
+	f.st.links = append(f.st.links, store.FileLink{
+		DownloadID: 2, ApplyBatchID: "old", SrcPath: "/old/e1.mkv", DstPath: e01,
+		Kind: "video", Status: "linked",
+	})
+
+	if err := f.w.Apply(context.Background(), 1); err == nil {
+		t.Fatal("want collision error")
+	}
+	for _, l := range f.st.links {
+		if l.DownloadID == 2 && l.Status != "linked" {
+			t.Errorf("чужая ссылка стала %q при коллизии, владение не должно отбираться", l.Status)
+		}
+	}
+}
+
 func TestUndo_RevertsLinks(t *testing.T) {
 	plan := seriesResult().Plan
 	f := newApplyFixture(t, plan)

internal/worker/worker.go

@@ -55,6 +55,7 @@ type Store interface {
 	SetOverride(ctx context.Context, downloadID int64, field, value string) error
 	ListOverrides(ctx context.Context, downloadID int64) (map[string]string, error)
 	CreateFileLinks(ctx context.Context, links []store.FileLink) error
+	SupersedeForeignLinks(ctx context.Context, downloadID int64, dstPaths []string) error
 	LatestBatchID(ctx context.Context, downloadID int64) (string, error)
 	ListFileLinksByBatch(ctx context.Context, batchID string) ([]store.FileLink, error)
 	DeleteFileLinksByBatch(ctx context.Context, batchID string) error

internal/worker/worker_test.go

@@ -114,7 +114,10 @@ func (f *fakeStore) ListOverrides(_ context.Context, _ int64) (map[string]string
 	return nil, nil
 }
 func (f *fakeStore) CreateFileLinks(_ context.Context, _ []store.FileLink) error { return nil }
-func (f *fakeStore) LatestBatchID(_ context.Context, _ int64) (string, error)    { return "", nil }
+func (f *fakeStore) SupersedeForeignLinks(_ context.Context, _ int64, _ []string) error {
+	return nil
+}
+func (f *fakeStore) LatestBatchID(_ context.Context, _ int64) (string, error) { return "", nil }
 func (f *fakeStore) ListFileLinksByBatch(_ context.Context, _ string) ([]store.FileLink, error) {
 	return nil, nil
 }