Два дефекта дедуп-веток приёма (ревью 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>
169 lines
6.4 KiB
Go
169 lines
6.4 KiB
Go
package ingest
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha1"
|
|
"encoding/hex"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/anacrolix/torrent/bencode"
|
|
"github.com/anacrolix/torrent/metainfo"
|
|
|
|
"git.vakhrushev.me/av/jellybit/internal/store"
|
|
)
|
|
|
|
// buildTorrent собирает валидные байты .torrent и ожидаемый v1-инфохэш.
|
|
func buildTorrent(t *testing.T, name, announce string) (data []byte, infohash string) {
|
|
t.Helper()
|
|
info := metainfo.Info{Name: name, Length: 1024, PieceLength: 512, Pieces: make([]byte, 40)}
|
|
infoBytes, err := bencode.Marshal(info)
|
|
if err != nil {
|
|
t.Fatalf("marshal info: %v", err)
|
|
}
|
|
sum := sha1.Sum(infoBytes)
|
|
mi := metainfo.MetaInfo{InfoBytes: infoBytes, Announce: announce}
|
|
var buf bytes.Buffer
|
|
if err := mi.Write(&buf); err != nil {
|
|
t.Fatalf("write metainfo: %v", err)
|
|
}
|
|
return buf.Bytes(), hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
func TestIngestTorrentFile(t *testing.T) {
|
|
data, wantHash := buildTorrent(t, "Dune.Part.Two.2024.mkv", "http://bt.rutracker.org/ann")
|
|
fs := &fakeStore{}
|
|
res, err := newService(fs).Ingest(context.Background(), Request{TorrentData: data, Context: "мой текст"})
|
|
if err != nil {
|
|
t.Fatalf("Ingest: %v", err)
|
|
}
|
|
if res.State != store.StateCatched || res.Deduplicated {
|
|
t.Errorf("res = %+v", res)
|
|
}
|
|
if len(res.Infohashes) != 1 || res.Infohashes[0] != wantHash {
|
|
t.Errorf("infohashes = %v, want [%s]", res.Infohashes, wantHash)
|
|
}
|
|
if len(fs.created) != 1 {
|
|
t.Fatalf("создано задач: %d", len(fs.created))
|
|
}
|
|
d := fs.created[0]
|
|
if d.SourceType != store.SourceTorrent {
|
|
t.Errorf("source_type = %q, want torrent", d.SourceType)
|
|
}
|
|
if d.SourceRef != "Dune.Part.Two.2024.mkv" {
|
|
t.Errorf("source_ref = %q (имя раздачи, не URL)", d.SourceRef)
|
|
}
|
|
// Контекст: текст пользователя первым, затем синтез из полей файла.
|
|
if !strings.HasPrefix(d.Context, "мой текст") {
|
|
t.Errorf("context не с текста пользователя: %q", d.Context)
|
|
}
|
|
if !strings.Contains(d.Context, "Dune.Part.Two") || !strings.Contains(d.Context, "rutracker.org") {
|
|
t.Errorf("context без синтеза из файла: %q", d.Context)
|
|
}
|
|
// Байты сохранены в транзакции создания.
|
|
if len(fs.blobs) != 1 || !bytes.Equal(fs.blobs[0], data) {
|
|
t.Errorf("байты .torrent не переданы в CreateDownloadIfNoActive")
|
|
}
|
|
}
|
|
|
|
func TestIngestTorrentDedupNoBlob(t *testing.T) {
|
|
data, _ := buildTorrent(t, "X", "http://t/ann")
|
|
// Активная задача уже есть → дедуп; байты писаться не должны.
|
|
fs := &fakeStore{active: &store.Download{ID: "existing", State: store.StateDownloading}}
|
|
res, err := newService(fs).Ingest(context.Background(), Request{TorrentData: data})
|
|
if err != nil {
|
|
t.Fatalf("Ingest: %v", err)
|
|
}
|
|
if !res.Deduplicated || res.DownloadID != "existing" {
|
|
t.Errorf("ожидался дедуп, res = %+v", res)
|
|
}
|
|
if len(fs.created) != 0 || len(fs.blobs) != 0 {
|
|
t.Errorf("при дедупе не должно быть создания/записи байтов")
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
_, err := newService(fs).Ingest(context.Background(), Request{TorrentData: big})
|
|
if err == nil {
|
|
t.Fatal("ожидалась ошибка превышения размера")
|
|
}
|
|
if len(fs.created) != 0 {
|
|
t.Errorf("при превышении размера задача не создаётся")
|
|
}
|
|
}
|
|
|
|
// У раздачи без имени source_ref берётся из имени файла (фолбек).
|
|
func TestIngestTorrentNameFallback(t *testing.T) {
|
|
// Info без name → BestName() == "" → фолбек на TorrentName.
|
|
info := metainfo.Info{Name: "", Length: 1024, PieceLength: 512, Pieces: make([]byte, 40)}
|
|
infoBytes, err := bencode.Marshal(info)
|
|
if err != nil {
|
|
t.Fatalf("marshal: %v", err)
|
|
}
|
|
mi := metainfo.MetaInfo{InfoBytes: infoBytes, Announce: "http://t/ann"}
|
|
var buf bytes.Buffer
|
|
if err := mi.Write(&buf); err != nil {
|
|
t.Fatalf("write: %v", err)
|
|
}
|
|
|
|
fs := &fakeStore{}
|
|
_, err = newService(fs).Ingest(context.Background(),
|
|
Request{TorrentData: buf.Bytes(), TorrentName: "Fallback.Name.torrent"})
|
|
if err != nil {
|
|
t.Fatalf("Ingest: %v", err)
|
|
}
|
|
if len(fs.created) != 1 || fs.created[0].SourceRef != "Fallback.Name.torrent" {
|
|
t.Errorf("source_ref = %q, want фолбек на имя файла", fs.created[0].SourceRef)
|
|
}
|
|
}
|
|
|
|
func TestIngestTorrentInvalid(t *testing.T) {
|
|
fs := &fakeStore{}
|
|
_, err := newService(fs).Ingest(context.Background(), Request{TorrentData: []byte("not a torrent")})
|
|
if err == nil {
|
|
t.Fatal("ожидалась ошибка разбора .torrent")
|
|
}
|
|
}
|