minstrel/internal/lifecycle/lifecycle_test.go

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)
		}
	}
}