package lifecycle import ( "context" "errors" "io" "log/slog" "os" "strings" "testing" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/acquire" "git.fabledsword.com/bvandeusen/minstrel/internal/db" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/dbtest" "git.fabledsword.com/bvandeusen/minstrel/internal/downloader" "git.fabledsword.com/bvandeusen/minstrel/internal/wanted" ) func newPool(t *testing.T) *pgxpool.Pool { t.Helper() if testing.Short() { t.Skip("skipping integration test in -short mode") } dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") if dsn == "" { t.Skip("MINSTREL_TEST_DATABASE_URL not set") } if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatalf("migrate: %v", err) } pool, err := pgxpool.New(context.Background(), dsn) if err != nil { t.Fatalf("pool: %v", err) } t.Cleanup(pool.Close) dbtest.ResetDB(t, pool) return pool } const ( hash = "0123456789abcdef0123456789abcdef01234567" other = "fedcba9876543210fedcba9876543210fedcba98" ) // fakeClient lists the torrents it holds and records what it is asked. type fakeClient struct { downloader.Driver held map[string]downloader.Torrent err error deleteErr error deleted []string reannounce int recheck int } func (c *fakeClient) Torrents(_ context.Context, hashes ...string) ([]downloader.Torrent, error) { if c.err != nil { return nil, c.err } var out []downloader.Torrent if len(hashes) == 0 { for _, t := range c.held { out = append(out, t) } return out, nil } for _, h := range hashes { if t, ok := c.held[h]; ok { out = append(out, t) } } return out, nil } func (c *fakeClient) Delete(_ context.Context, _ bool, hashes ...string) error { if c.deleteErr != nil { return c.deleteErr } c.deleted = append(c.deleted, hashes...) for _, h := range hashes { delete(c.held, h) } return nil } func (c *fakeClient) Reannounce(context.Context, ...string) error { c.reannounce++; return nil } func (c *fakeClient) Recheck(context.Context, ...string) error { c.recheck++; return nil } type stack struct { q *dbq.Queries s *Service client *fakeClient want dbq.Want grab dbq.Grab clock time.Time } // newStack wants RAM and has a grab of it running in a fake client holding // t under hash. func newStack(t *testing.T, tor downloader.Torrent) *stack { t.Helper() ctx := context.Background() pool := newPool(t) q := dbq.New(pool) w, err := wanted.WantWith(ctx, q, wanted.WantParams{GroupMBID: "rg-ram", ArtistMBID: "mb-dp", ArtistName: "Daft Punk", Title: "Random Access Memories", PrimaryType: "Album", FirstReleaseDate: "2013-05-17", Reason: wanted.ReasonRequest}) if err != nil { t.Fatal(err) } g, err := acquire.NewMemory(pool).Grabbed(ctx, dbq.InsertGrabParams{WantID: w.ID, InfoHash: hash, Name: "Daft Punk - RAM [FLAC]", Key: hash, Client: "qbittorrent", SavePath: "/downloads", Origin: acquire.OriginSchedule}, pgtype.UUID{}, "") if err != nil { t.Fatal(err) } if err := q.MarkWantSearched(ctx, w.ID); err != nil { t.Fatal(err) } tor.Hash = hash st := &stack{q: q, client: &fakeClient{held: map[string]downloader.Torrent{hash: tor}}, want: w, grab: g, clock: time.Now()} st.s = New(pool, slog.New(slog.NewTextHandler(io.Discard, nil))) st.s.ClientFor = func(context.Context, string) (downloader.Driver, error) { return st.client, nil } st.s.Now = func() time.Time { return st.clock } return st } func (st *stack) grabNow(t *testing.T) dbq.Grab { t.Helper() g, err := st.q.GetGrab(context.Background(), st.grab.ID) if err != nil { t.Fatal(err) } return g } func (st *stack) searchedAgain(t *testing.T) bool { t.Helper() w, err := st.q.GetWantByReleaseGroup(context.Background(), "rg-ram") if err != nil { t.Fatal(err) } return !w.SearchedAt.Valid } func (st *stack) ledger(t *testing.T) []dbq.DownloadIntervention { t.Helper() rows, err := st.s.pool.Query(context.Background(), `SELECT * FROM download_interventions ORDER BY first_seen, id`) if err != nil { t.Fatal(err) } out, err := pgx.CollectRows(rows, pgx.RowToStructByPos[dbq.DownloadIntervention]) if err != nil { t.Fatal(err) } return out } func TestLifecycle_AFailedDownloadIsGivenUpAtOnce_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.4, State: "error", FailReason: "No space left on device"}) rep := st.s.Pass(context.Background()) if rep.GivenUp != 1 { t.Fatalf("pass = %+v", rep) } g := st.grabNow(t) if !g.EndedAt.Valid || g.EndReason != EndFailed { t.Fatalf("grab ended=%v reason=%q", g.EndedAt.Valid, g.EndReason) } if len(st.client.deleted) != 1 || !st.searchedAgain(t) { t.Errorf("deleted %v, searched again %v", st.client.deleted, st.searchedAgain(t)) } blocks, err := st.q.ListReleaseBlocks(context.Background()) if err != nil || len(blocks) != 1 || blocks[0].Reason != acquire.BlockStalled || !blocks[0].ExpiresAt.Valid { t.Fatalf("blocks = %+v %v", blocks, err) } if d := blocks[0].ExpiresAt.Time.Sub(st.clock); d < StallBlock-time.Minute || d > StallBlock+time.Minute { t.Errorf("held back for %s", d) } rows := st.ledger(t) if len(rows) != 1 || rows[0].Trigger != ClientError || rows[0].Action != actGivenUp || rows[0].Outcome != "resolved" || rows[0].Detail != "No space left on device" { t.Errorf("ledger = %+v", rows) } } func TestLifecycle_ADownloadNobodySharesIsRetriedThenGivenUp_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "stalledDL"}) ctx := context.Background() // Within the stall setting: nothing is wrong yet. if rep := st.s.Pass(ctx); rep.Retried+rep.GivenUp != 0 || len(st.ledger(t)) != 0 { t.Fatalf("a young download was judged: %+v", rep) } st.clock = st.clock.Add(7 * time.Hour) for i := 1; i <= Retries; i++ { if rep := st.s.Pass(ctx); rep.Retried != 1 || rep.GivenUp != 0 { t.Fatalf("pass %d = %+v", i, rep) } } if st.client.reannounce != Retries || st.client.recheck != Retries { t.Errorf("reannounced %d, rechecked %d", st.client.reannounce, st.client.recheck) } if st.grabNow(t).EndedAt.Valid { t.Fatal("given up before the retries ran out") } if rep := st.s.Pass(ctx); rep.GivenUp != 1 { t.Fatalf("last pass = %+v", rep) } g := st.grabNow(t) if g.EndReason != EndStalled || len(st.client.deleted) != 1 || !st.searchedAgain(t) { t.Errorf("grab %q: %s; deleted %v", g.EndReason, g.EndDetail, st.client.deleted) } rows := st.ledger(t) if len(rows) != 1 || rows[0].Trigger != NoPeers || rows[0].Remediations != Retries+1 || rows[0].Outcome != "resolved" { t.Errorf("ledger = %+v", rows) } } func TestLifecycle_AStalledDownloadThatRecoversIsSettled_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "stalledDL"}) ctx := context.Background() st.clock = st.clock.Add(7 * time.Hour) st.s.Pass(ctx) t2 := st.client.held[hash] t2.NumSeeds = 3 st.client.held[hash] = t2 st.s.Pass(ctx) rows := st.ledger(t) if len(rows) != 1 || rows[0].Outcome != "resolved" || rows[0].Action != actRetried { t.Fatalf("ledger = %+v", rows) } if st.grabNow(t).EndedAt.Valid { t.Error("a download that found peers was given up") } } func TestLifecycle_TheStallSettingOffNeverGivesUp_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "stalledDL"}) ctx := context.Background() if _, err := st.s.pool.Exec(ctx, `UPDATE acquisition_settings SET stall_hours = 0`); err != nil { t.Fatal(err) } st.clock = st.clock.Add(30 * 24 * time.Hour) for range Retries + 2 { if rep := st.s.Pass(ctx); rep.Retried+rep.GivenUp != 0 { t.Fatalf("pass = %+v", rep) } } } func TestLifecycle_ADownloadGoneFromTheClientEndsAfterHalfAnHour_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "downloading", NumSeeds: 4}) ctx := context.Background() delete(st.client.held, hash) st.client.held[other] = downloader.Torrent{Hash: other, Name: "Someone else's", Progress: 1, State: "uploading"} if rep := st.s.Pass(ctx); rep.Gone != 0 { t.Fatalf("ended at first sight: %+v", rep) } st.clock = st.clock.Add(31 * time.Minute) if rep := st.s.Pass(ctx); rep.Gone != 1 { t.Fatalf("pass = %+v", rep) } g := st.grabNow(t) if g.EndReason != EndGone || !st.searchedAgain(t) { t.Errorf("grab %q, searched again %v", g.EndReason, st.searchedAgain(t)) } if blocks, _ := st.q.ListReleaseBlocks(ctx); len(blocks) != 0 { t.Errorf("a release nothing was wrong with was held back: %+v", blocks) } } // A release that leaves its client a second time is held back, whatever the // cause (#6176): one release was taken 40 times in a day, each ending gone. func TestLifecycle_AReleaseGoneTwiceIsHeldBack_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "downloading", NumSeeds: 4}) ctx := context.Background() delete(st.client.held, hash) st.client.held[other] = downloader.Torrent{Hash: other, Name: "Someone else's", Progress: 1, State: "uploading"} st.s.Pass(ctx) st.clock = st.clock.Add(31 * time.Minute) if rep := st.s.Pass(ctx); rep.Gone != 1 { t.Fatalf("first gone: %+v", rep) } if blocks, _ := st.q.ListReleaseBlocks(ctx); len(blocks) != 0 { t.Fatalf("held back after one gone: %+v", blocks) } again, err := acquire.NewMemory(st.s.pool).Grabbed(ctx, dbq.InsertGrabParams{WantID: st.want.ID, InfoHash: hash, Name: "Daft Punk - RAM [FLAC]", Key: hash, Client: "qbittorrent", SavePath: "/downloads", Origin: acquire.OriginSchedule}, pgtype.UUID{}, "") if err != nil { t.Fatal(err) } // first_seen is the database's clock and the test's runs ahead of it, so // the second grab may end on either pass; count both. gone := st.s.Pass(ctx).Gone st.clock = st.clock.Add(31 * time.Minute) if gone += st.s.Pass(ctx).Gone; gone != 1 { t.Fatalf("second grab ended gone %d times, want 1", gone) } blocks, err := st.q.ListReleaseBlocks(ctx) if err != nil { t.Fatal(err) } if len(blocks) != 1 || blocks[0].Reason != acquire.BlockGone || !blocks[0].ExpiresAt.Valid { t.Fatalf("blocks = %+v, want one timed gone block", blocks) } g, err := st.q.GetGrab(ctx, again.ID) if err != nil { t.Fatal(err) } if g.EndReason != EndGone || !strings.Contains(g.EndDetail, "held back until") { t.Errorf("second grab %q: %s", g.EndReason, g.EndDetail) } } func TestLifecycle_AnEmptyListIsNotAMeasurement_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "downloading"}) ctx := context.Background() delete(st.client.held, hash) st.clock = st.clock.Add(24 * time.Hour) for range 3 { if rep := st.s.Pass(ctx); rep.Gone != 0 { t.Fatalf("pass = %+v", rep) } } if len(st.ledger(t)) != 0 || st.grabNow(t).EndedAt.Valid { t.Error("a client listing nothing was taken to have lost the download") } } func TestLifecycle_AnUnreachableClientIsRecordedAndSettled_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.3, State: "stalledDL"}) ctx := context.Background() st.clock = st.clock.Add(7 * time.Hour) st.client.err = errors.New("dial tcp: connection refused") st.s.Pass(ctx) down, err := st.q.ListClientsDown(ctx) if err != nil || len(down) != 1 || down[0].Client != "qbittorrent" { t.Fatalf("clients down = %+v %v", down, err) } if st.client.reannounce != 0 { t.Error("a download was judged through a client that did not answer") } st.client.err = nil st.s.Pass(ctx) if down, _ := st.q.ListClientsDown(ctx); len(down) != 0 { t.Errorf("still down: %+v", down) } } func TestLifecycle_AnImportedDownloadSeedsThenIsRemoved_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 1, State: "uploading", Ratio: 0.8}) ctx := context.Background() if _, err := st.q.RecordImport(ctx, dbq.RecordImportParams{ID: st.grab.ID, ImportState: "imported", ImportReason: "Imported.", ImportDetail: []byte("{}"), ImportedFiles: 3, EndReason: "imported"}); err != nil { t.Fatal(err) } if !st.grabNow(t).Seeding { t.Fatal("an import did not leave its download seeding") } if rep := st.s.Pass(ctx); rep.Released != 0 || len(st.client.deleted) != 0 { t.Fatalf("removed before the ratio: %+v", rep) } if g := st.grabNow(t); g.Ratio < 0.79 || g.Ratio > 0.81 { t.Errorf("ratio recorded %v", g.Ratio) } tor := st.client.held[hash] tor.Ratio = 2.1 st.client.held[hash] = tor if rep := st.s.Pass(ctx); rep.Released != 1 || len(st.client.deleted) != 1 { t.Fatalf("pass = %+v, deleted %v", rep, st.client.deleted) } if !st.grabNow(t).SeedReleasedAt.Valid { t.Error("the release was not recorded") } } func TestLifecycle_AnImportedDownloadIsRemovedWhenItsDaysAreUp_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 1, State: "uploading", Ratio: 0.1}) ctx := context.Background() if _, err := st.q.RecordImport(ctx, dbq.RecordImportParams{ID: st.grab.ID, ImportState: "imported", ImportReason: "Imported.", ImportDetail: []byte("{}"), ImportedFiles: 3, EndReason: "imported"}); err != nil { t.Fatal(err) } st.clock = st.clock.Add(15 * 24 * time.Hour) if rep := st.s.Pass(ctx); rep.Released != 1 || len(st.client.deleted) != 1 { t.Fatalf("pass = %+v", rep) } } func TestLifecycle_ADownloadTheClientWouldNotRemoveIsAskedAgain_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.4, State: "error"}) ctx := context.Background() st.client.deleteErr = errors.New("HTTP 500") if rep := st.s.Pass(ctx); rep.GivenUp != 1 { t.Fatalf("pass = %+v", rep) } if !st.grabNow(t).EndedAt.Valid { t.Fatal("a refused removal held the grab") } open := 0 for _, r := range st.ledger(t) { if r.Trigger == NotRemoved && r.Outcome == "open" { open++ } } if open != 1 { t.Fatalf("ledger = %+v", st.ledger(t)) } st.client.deleteErr = nil if rep := st.s.Pass(ctx); rep.Removed != 1 || len(st.client.deleted) != 1 { t.Fatalf("pass = %+v", rep) } for _, r := range st.ledger(t) { if r.Outcome == "open" { t.Errorf("still open: %+v", r) } } } func TestLifecycle_APersonGivingUpSetsTheReleaseAsideForGood_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "Daft Punk - RAM [FLAC]", Progress: 0.4, State: "downloading"}) ctx := context.Background() if _, err := st.s.Abandon(ctx, st.grab.ID, pgtype.UUID{}); err != nil { t.Fatal(err) } g := st.grabNow(t) if g.EndReason != EndAbandoned || len(st.client.deleted) != 1 || !st.searchedAgain(t) { t.Errorf("grab %q, deleted %v", g.EndReason, st.client.deleted) } blocks, _ := st.q.ListReleaseBlocks(ctx) if len(blocks) != 1 || blocks[0].Reason != acquire.BlockOperator || blocks[0].ExpiresAt.Valid { t.Errorf("blocks = %+v", blocks) } if _, err := st.s.Abandon(ctx, st.grab.ID, pgtype.UUID{}); !errors.Is(err, ErrNotAbandonable) { t.Errorf("abandoned twice: %v", err) } } func TestLifecycle_AHoldExpiresButABlockForGoodIsNeverShortened_Integration(t *testing.T) { st := newStack(t, downloader.Torrent{Name: "x", Progress: 0.4, State: "downloading"}) ctx := context.Background() mem := acquire.NewMemory(st.s.pool) past := pgtype.Timestamptz{Time: time.Now().Add(-time.Hour), Valid: true} if err := mem.Block(ctx, dbq.BlockReleaseParams{InfoHash: other, Name: "old", Reason: acquire.BlockStalled, ExpiresAt: past}); err != nil { t.Fatal(err) } if blocks, _ := st.q.ListReleaseBlocks(ctx); len(blocks) != 0 { t.Errorf("an expired hold still holds: %+v", blocks) } if n, err := st.q.DeleteExpiredBlocks(ctx); err != nil || n != 1 { t.Errorf("lifted %d %v", n, err) } if err := mem.Block(ctx, dbq.BlockReleaseParams{InfoHash: hash, Name: "x", Reason: acquire.BlockOperator}); err != nil { t.Fatal(err) } soon := pgtype.Timestamptz{Time: time.Now().Add(time.Hour), Valid: true} if err := mem.Block(ctx, dbq.BlockReleaseParams{InfoHash: hash, Name: "x", Reason: acquire.BlockStalled, ExpiresAt: soon}); err != nil { t.Fatal(err) } blocks, _ := st.q.ListReleaseBlocks(ctx) if len(blocks) != 1 || blocks[0].Reason != acquire.BlockOperator || blocks[0].ExpiresAt.Valid { t.Errorf("a block for good was shortened: %+v", blocks) } } func TestSpan(t *testing.T) { for _, c := range []struct { d time.Duration want string }{{45 * time.Minute, "45 minutes"}, {7 * time.Hour, "7 hours"}, {3 * 24 * time.Hour, "3 days"}} { if got := span(c.d); got != c.want { t.Errorf("span(%s) = %q, want %q", c.d, got, c.want) } } }