Приём: гард дедуп-дозаписи хешей (F1) и апгрейд catched-magnet до torrent (F6)

Два дефекта дедуп-веток приёма (ревью Fable 2026-07-08), оба про инвариант «≤1 активная загрузка на infohash» и сохранность источника. F1: дедуп-ветка CreateDownloadIfNoActive дописывала все хеши входящего источника в найденную активную задачу без пер-хеш гарда владения (в отличие от AddInfohashes). Гибрид {v1,v2}, дедупнувшись на задачу B (владелец v2), крал v1 у активной A → две активные владели v1. Теперь дозапись под тем же гардом: хеш, которым владеет другая активная задача, не дописывается. F6: при дедупе .torrent-байт на пойманную magnet-задачу (catched) байты выбрасывались, source_type оставался magnet → worker добавлял по magnet-URL → вечный metaDL → failed (magnet закрытого трекера без DHT метаданные не докачает). Новый guarded-метод UpgradeCatchedMagnetToTorrent атомарно сохраняет байты и меняет source_type magnet→torrent, но только пока задача в catched (worker источник ещё не отдал). Ingest зовёт апгрейд на обоих дедуп-путях. Это целевое исключение из правила спеки «при дедупе байты не сохраняем» — оформлено MODIFIED-дельтой ingest. Схема БД не меняется (download_torrent и source_type уже есть). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-08 17:18:28 +03:00
parent b1bca98738
commit 4475fbd548
11 changed files with 605 additions and 13 deletions
@@ -35,6 +35,11 @@ type Store interface {
 	// AddInfohashes доносит задаче недостающие хеши (guarded). Нужен на
 	// быстром дедуп-пути, который не доходит до CreateDownloadIfNoActive.
 	AddInfohashes(ctx context.Context, downloadID string, hashes []string) error
+	// UpgradeCatchedMagnetToTorrent при дедупе входящих байтов `.torrent` на
+	// пойманную (`catched`) magnet-задачу сохраняет байты и меняет source_type
+	// на torrent (атомарно). No-op, если задача уже добавлена/не magnet. Чинит
+	// magnet закрытого трекера, который иначе застрянет в metaDL.
+	UpgradeCatchedMagnetToTorrent(ctx context.Context, downloadID string, torrentBlob []byte) (bool, error)
 }

 // Service — реализация быстрого приёма.
@@ -109,14 +114,9 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
 	if existing != nil {
 		// Гонка с параллельным приёмом/discover: активная задача появилась
 		// после быстрого чека — присоединяемся к ней (хеши донёс сам
-		// CreateDownloadIfNoActive).
+		// CreateDownloadIfNoActive; existing уже с подгруженными хешами).
 		log.Info("download attached to active", "download_id", existing.ID, "state", existing.State)
-		return Result{
-			DownloadID:   existing.ID,
-			Infohashes:   src.infohashes,
-			State:        existing.State,
-			Deduplicated: true,
-		}, nil
+		return s.attached(ctx, src, existing), nil
 	}

 	log.Info("download catched", "download_id", d.ID)
@@ -208,6 +208,17 @@ func (s *Service) attached(ctx context.Context, src parsedSource, existing *stor
 			logctx.FromOr(ctx, s.log).Warn("ingest top-up infohashes failed", "error", err)
 		}
 	}
+	// F6: входящее — байты `.torrent`, а активная задача поймана как magnet и
+	// ещё не отдана в qBittorrent (catched) → сохраняем байты и переключаем
+	// источник на torrent, чтобы воркер добавил раздачу файлом (magnet
+	// закрытого трекера иначе застрянет в metaDL). Best-effort: не валит приём.
+	if src.sourceType == store.SourceTorrent && len(src.torrentBlob) > 0 {
+		if upgraded, err := s.store.UpgradeCatchedMagnetToTorrent(ctx, existing.ID, src.torrentBlob); err != nil {
+			logctx.FromOr(ctx, s.log).Warn("ingest torrent upgrade failed", "error", err)
+		} else if upgraded {
+			logctx.FromOr(ctx, s.log).Info("catched magnet upgraded to torrent", "download_id", existing.ID)
+		}
+	}
 	return Result{
 		DownloadID:   existing.ID,
 		Infohashes:   src.infohashes,
@@ -16,11 +16,14 @@ const sampleMagnet = "magnet:?xt=urn:btih:541ADCFF3B6DD5DBA7088EA83317D9D6FAC331
 const sampleInfohash = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"

 type fakeStore struct {
-	active   *store.Download
-	created  []store.Download
-	hashes   [][]string
-	blobs    [][]byte
-	toppedUp []string
+	active      *store.Download
+	created     []store.Download
+	hashes      [][]string
+	blobs       [][]byte
+	toppedUp    []string
+	upgradeID   string // downloadID последнего вызова UpgradeCatchedMagnetToTorrent
+	upgradeBlob []byte // байты, переданные в апгрейд
+	upgradeUp   bool   // что вернуть из UpgradeCatchedMagnetToTorrent
 }

 func (f *fakeStore) FindActiveByInfohash(_ context.Context, _ ...string) (*store.Download, error) {
@@ -44,6 +47,12 @@ func (f *fakeStore) AddInfohashes(_ context.Context, id string, hashes []string)
 	return nil
 }

+func (f *fakeStore) UpgradeCatchedMagnetToTorrent(_ context.Context, id string, blob []byte) (bool, error) {
+	f.upgradeID = id
+	f.upgradeBlob = blob
+	return f.upgradeUp, nil
+}
+
 func newService(st Store) *Service {
 	return New(st, slog.New(slog.NewTextHandler(io.Discard, nil)))
 }
@@ -83,6 +83,45 @@ func TestIngestTorrentDedupNoBlob(t *testing.T) {
 	}
 }

+// F6: дедуп .torrent на пойманную (catched) magnet-задачу вызывает апгрейд —
+// сохранение байтов и смену источника (magnet закрытого трекера иначе застрянет
+// в metaDL).
+func TestIngestTorrentUpgradesCatchedMagnet(t *testing.T) {
+	data, hash := buildTorrent(t, "Dune", "http://t/ann")
+	existing := &store.Download{
+		ID:         "cm",
+		State:      store.StateCatched,
+		SourceType: store.SourceMagnet,
+		Infohashes: []store.Infohash{{Infohash: hash, Kind: store.HashV1}},
+	}
+	fs := &fakeStore{active: existing, upgradeUp: true}
+	res, err := newService(fs).Ingest(context.Background(), Request{TorrentData: data})
+	if err != nil {
+		t.Fatalf("Ingest: %v", err)
+	}
+	if !res.Deduplicated || res.DownloadID != "cm" {
+		t.Errorf("ожидался дедуп на cm, res = %+v", res)
+	}
+	if fs.upgradeID != "cm" {
+		t.Errorf("апгрейд не вызван для существующей задачи (upgradeID=%q)", fs.upgradeID)
+	}
+	if !bytes.Equal(fs.upgradeBlob, data) {
+		t.Errorf("в апгрейд переданы не те байты")
+	}
+}
+
+// Дедуп magnet-ссылки (не .torrent) апгрейд не вызывает — нечего сохранять.
+func TestIngestMagnetDedupNoUpgrade(t *testing.T) {
+	existing := &store.Download{ID: "cm", State: store.StateCatched, SourceType: store.SourceMagnet}
+	fs := &fakeStore{active: existing}
+	if _, err := newService(fs).Ingest(context.Background(), Request{Source: sampleMagnet}); err != nil {
+		t.Fatalf("Ingest: %v", err)
+	}
+	if fs.upgradeID != "" {
+		t.Errorf("апгрейд не должен вызываться для magnet-дедупа")
+	}
+}
+
 func TestIngestTorrentTooLarge(t *testing.T) {
 	fs := &fakeStore{}
 	big := make([]byte, MaxTorrentSize+1)
@@ -272,8 +272,18 @@ func (s *Store) CreateDownloadIfNoActive(ctx context.Context, d *Download, hashe
 		// Дедуп нашёл активного владельца по одному из хешей — остальные хеши
 		// norm принадлежат тому же торренту (гибридный magnet): дописываем
 		// недостающие, иначе второй хеш молча теряется и последующий приём по
-		// нему создал бы вторую активную задачу.
+		// нему создал бы вторую активную задачу. Дозапись — под тем же пер-хеш
+		// гардом владения, что и AddInfohashes: хеш, которым владеет ДРУГАЯ
+		// активная задача (split-identity/крафт-магнет), не дописываем, иначе
+		// две активные владели бы одним инфохэшем (нарушение инварианта).
 		for _, h := range norm {
+			owner, err := findActiveByInfohash(ctx, tx, []string{h}, existing.ID)
+			if err != nil {
+				return nil, fmt.Errorf("create download: %w", err)
+			}
+			if owner != nil {
+				continue // чужой активный владелец — не крадём хеш
+			}
 			if _, err := tx.ExecContext(ctx,
 				`INSERT OR IGNORE INTO download_infohash (download_id, infohash, kind, created_at) VALUES (?, ?, ?, ?)`,
 				existing.ID, h, HashKind(h), now); err != nil {
@@ -335,6 +345,57 @@ func (s *Store) GetTorrentData(ctx context.Context, downloadID string) ([]byte,
 	return data, nil
 }

+// UpgradeCatchedMagnetToTorrent апгрейдит пойманную magnet-задачу до torrent
+// при дедупе входящих байтов `.torrent`: в одной write-транзакции сохраняет
+// байты и меняет source_type magnet→torrent, но ТОЛЬКО пока задача в `catched`
+// (воркер источник ещё не отдал в qBittorrent) и её source_type всё ещё
+// `magnet`. Это целевое исключение из правила «при дедупе байты не сохраняем»:
+// magnet на закрытом трекере без DHT метаданные не докачает и застрянет в
+// metaDL, а поданный пользователем `.torrent` их несёт. Возвращает true, если
+// апгрейд применён; (false, nil) — если задача не подходит (уже добавлена,
+// отменена или не magnet) либо байты пусты (защитный no-op). Гард через
+// RowsAffected UPDATE — та же ре-валидация состояния, что у PromoteCatched.
+func (s *Store) UpgradeCatchedMagnetToTorrent(ctx context.Context, downloadID string, torrentBlob []byte) (bool, error) {
+	if len(torrentBlob) == 0 {
+		return false, nil
+	}
+	tx, err := s.DB.BeginTxx(ctx, nil)
+	if err != nil {
+		return false, fmt.Errorf("upgrade %s to torrent: begin tx: %w", downloadID, err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	res, err := tx.ExecContext(ctx, `
+UPDATE download
+SET source_type = ?, updated_at = ?
+WHERE id = ? AND source_type = ? AND state = ?`,
+		string(SourceTorrent), FormatTime(Now()), downloadID, string(SourceMagnet), string(StateCatched))
+	if err != nil {
+		return false, fmt.Errorf("upgrade %s to torrent: %w", downloadID, err)
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("upgrade %s to torrent: %w", downloadID, err)
+	}
+	if n == 0 {
+		// Не catched-magnet (уже добавлена/отменена/torrent) — апгрейд не нужен.
+		if err := tx.Commit(); err != nil {
+			return false, fmt.Errorf("upgrade %s to torrent: commit: %w", downloadID, err)
+		}
+		return false, nil
+	}
+	// У magnet-задачи блоба не было; OR REPLACE — страховка идемпотентности.
+	if _, err := tx.ExecContext(ctx,
+		`INSERT OR REPLACE INTO download_torrent (download_id, data) VALUES (?, ?)`,
+		downloadID, torrentBlob); err != nil {
+		return false, fmt.Errorf("upgrade %s to torrent: store blob: %w", downloadID, err)
+	}
+	if err := tx.Commit(); err != nil {
+		return false, fmt.Errorf("upgrade %s to torrent: commit: %w", downloadID, err)
+	}
+	return true, nil
+}
+
 // ActivateIfNoOtherActive атомарно возвращает загрузку в активное состояние
 // (retry/восстановление сверкой/relink): в одной write-транзакции проверяет,
 // что никакая ДРУГАЯ активная загрузка не владеет любым из хешей этой, и
@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"errors"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -167,6 +168,50 @@ func TestFindActiveByInfohash(t *testing.T) {
 	}
 }

+// F1: дедуп-ветка CreateDownloadIfNoActive доносит недостающие хеши под пер-хеш
+// гардом владения — хеш, которым владеет ДРУГАЯ активная задача (split-identity
+// гибрида), красть не должна, иначе две активные владели бы одним инфохэшем.
+func TestDedupTopUpDoesNotStealForeignHash(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	v1 := hashN(1)
+	v2 := hashN(2)
+
+	// A владеет v1, B владеет v2 — две активные задачи одного гибрида (split).
+	idA := mustCreate(t, st, v1)
+	idB := mustCreate(t, st, v2)
+
+	// Приём гибрида {v1, v2} → дедуп на одну из активных; чужой хеш не крадём.
+	existing, err := st.CreateDownloadIfNoActive(ctx, newDownloading(), []string{v1, v2}, nil)
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	if existing == nil {
+		t.Fatal("ожидался дедуп на активную задачу")
+	}
+
+	assertHashes(t, st, idA, v1) // A по-прежнему владеет только v1
+	assertHashes(t, st, idB, v2) // B по-прежнему владеет только v2
+}
+
+// assertHashes проверяет точный набор инфохэшей загрузки.
+func assertHashes(t *testing.T, st *Store, id string, want ...string) {
+	t.Helper()
+	d, err := st.GetDownload(context.Background(), id)
+	if err != nil {
+		t.Fatalf("get %s: %v", id, err)
+	}
+	got := d.HashList()
+	if len(got) != len(want) {
+		t.Fatalf("%s hashes = %v, want %v", id, got, want)
+	}
+	for _, w := range want {
+		if !slices.Contains(got, w) {
+			t.Fatalf("%s hashes = %v, want содержит %s", id, got, w)
+		}
+	}
+}
+
 // Состояния рассинхрона (target_missing/orphaned/deleted) терминальны: задача
 // в них не должна считаться «активной» (иначе relink/ingest-дедуп решат, что
 // для infohash уже есть активная задача). Регрессия: FindActiveByInfohash и
@@ -65,3 +65,110 @@ func TestTorrentBlobMissing(t *testing.T) {
 		t.Errorf("ожидался ErrNotFound, got %v", err)
 	}
 }
+
+// F6: апгрейд пойманного magnet до torrent сохраняет байты и меняет source_type.
+func TestUpgradeCatchedMagnetToTorrent(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	d := &Download{SourceType: SourceMagnet, SourceRef: "magnet:?xt=urn:btih:" + hashN(1), State: StateCatched}
+	if _, err := st.CreateDownloadIfNoActive(ctx, d, []string{hashN(1)}, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+
+	blob := []byte("d8:announce…real torrent")
+	upgraded, err := st.UpgradeCatchedMagnetToTorrent(ctx, d.ID, blob)
+	if err != nil || !upgraded {
+		t.Fatalf("upgrade: upgraded=%v err=%v", upgraded, err)
+	}
+
+	got, err := st.GetDownload(ctx, d.ID)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got.SourceType != SourceTorrent {
+		t.Errorf("source_type = %q, want torrent", got.SourceType)
+	}
+	data, err := st.GetTorrentData(ctx, d.ID)
+	if err != nil {
+		t.Fatalf("get torrent data: %v", err)
+	}
+	if string(data) != string(blob) {
+		t.Errorf("blob = %q, want %q", data, blob)
+	}
+}
+
+// Апгрейд применим только в catched: уже добавленный (downloading) magnet не
+// трогаем — его судьба решается retry/сверкой, а не приёмом.
+func TestUpgradeSkippedWhenNotCatched(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	d := &Download{SourceType: SourceMagnet, SourceRef: "m", State: StateDownloading}
+	if _, err := st.CreateDownloadIfNoActive(ctx, d, []string{hashN(2)}, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+
+	upgraded, err := st.UpgradeCatchedMagnetToTorrent(ctx, d.ID, []byte("bytes"))
+	if err != nil {
+		t.Fatalf("upgrade: %v", err)
+	}
+	if upgraded {
+		t.Error("downloading-magnet апгрейдить не должны")
+	}
+	got, err := st.GetDownload(ctx, d.ID)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got.SourceType != SourceMagnet {
+		t.Errorf("source_type сменился на %q, а не должен", got.SourceType)
+	}
+	if _, err := st.GetTorrentData(ctx, d.ID); !errors.Is(err, ErrNotFound) {
+		t.Errorf("байты не должны сохраняться, got err=%v", err)
+	}
+}
+
+// Не-magnet источник апгрейд не трогает (гард source_type=magnet), байты не
+// перезаписывает.
+func TestUpgradeSkippedForTorrentSource(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	d := &Download{SourceType: SourceTorrent, SourceRef: "Rel", State: StateCatched}
+	if _, err := st.CreateDownloadIfNoActive(ctx, d, []string{hashN(3)}, []byte("orig")); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+
+	upgraded, err := st.UpgradeCatchedMagnetToTorrent(ctx, d.ID, []byte("new"))
+	if err != nil {
+		t.Fatalf("upgrade: %v", err)
+	}
+	if upgraded {
+		t.Error("torrent-источник апгрейдить не нужно")
+	}
+	data, err := st.GetTorrentData(ctx, d.ID)
+	if err != nil {
+		t.Fatalf("get torrent data: %v", err)
+	}
+	if string(data) != "orig" {
+		t.Errorf("байты перезаписаны: %q", data)
+	}
+}
+
+// Пустые байты — защитный no-op.
+func TestUpgradeEmptyBlobNoOp(t *testing.T) {
+	st := newTestStore(t)
+	ctx := context.Background()
+	d := &Download{SourceType: SourceMagnet, SourceRef: "m", State: StateCatched}
+	if _, err := st.CreateDownloadIfNoActive(ctx, d, []string{hashN(4)}, nil); err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	upgraded, err := st.UpgradeCatchedMagnetToTorrent(ctx, d.ID, nil)
+	if err != nil || upgraded {
+		t.Fatalf("пустой блоб: upgraded=%v err=%v", upgraded, err)
+	}
+	got, err := st.GetDownload(ctx, d.ID)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got.SourceType != SourceMagnet {
+		t.Errorf("source_type сменился на %q при пустом блобе", got.SourceType)
+	}
+}