// Package lifecycle looks after a download from the grab to its removal // (M502 #5523), ported from Fabled Librarian's internal/health and its seed // release (System 104). // // - A download the client has failed is given up at once. // - One nobody has shared, or still fetching its file list, after the // stall setting is retried: the trackers asked again and the files // re-read, three times a pass apart. Then it is given up. // - Giving up removes the download with its files, holds the release back // for a week, ends the grab and searches for the album again. There is // no limit per album: what stops it churning is the hold on each release. // - A running download the client no longer lists, for half an hour, ends // and its album is searched again. The release is not held back: nothing // was wrong with it. // - An imported download seeds (its files were copied) until its share // ratio reaches the setting or its days are up, then is removed. // // Every condition and every act goes in the intervention ledger, one open // row per (client, download, condition), so the Activity tab can say what // is wrong with a download and what was done. package lifecycle import ( "context" "errors" "fmt" "log/slog" "strings" "sync" "time" "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/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/downloader" "git.fabledsword.com/bvandeusen/minstrel/internal/indexer" ) // Conditions, matching download_interventions_trigger_check. const ( NoPeers = "no_peers" NoMetadata = "no_metadata" ClientError = "client_error" ClientUnreachable = "client_unreachable" Gone = "gone" NotRemoved = "not_removed" ) // Acts, matching download_interventions_action_check. const ( actRetried = "retried" actGivenUp = "given_up" actRemoved = "removed" ) // End reasons a grab is given with, matching grabs_end_reason_check. const ( EndStalled = "stalled" EndFailed = "failed" EndAbandoned = "abandoned" EndGone = "gone" ) const ( // interval: three retries a pass apart is half an hour of asking before // a stalled download is given up. interval = 10 * time.Minute passTimeout = 5 * time.Minute // Retries is how many times one stuck download is nudged first. Retries = 3 // StallBlock is how long a release given up on is held back. StallBlock = 7 * 24 * time.Hour // goneAfter: a client restarting can answer without a download for a // while; half an hour missing is gone. goneAfter = 30 * time.Minute // goneWindow: a release gone twice within it is held back (#6176). goneWindow = 7 * 24 * time.Hour ) // running are the import states of a grab still downloading: the lifecycle // acts only on these, so an import under way is never pulled from under. var running = []string{"", "downloading", "waiting"} // Settings are the limits, from acquisition_settings. type Settings struct { // SeedRatio is the share ratio an imported download seeds to; 0 removes // it once imported. SeedRatio float64 // SeedTime is the longest it seeds; 0 is no limit. SeedTime time.Duration // Patience is how long a download may sit with nobody sharing it before // it is retried, then given up; 0 never gives up for that. Patience time.Duration } // Defaults are FL's rulings. var Defaults = Settings{SeedRatio: 2, SeedTime: 14 * 24 * time.Hour, Patience: 6 * time.Hour} // Service runs the lifecycle passes. type Service struct { pool *pgxpool.Pool // ClientFor builds a download client by its code. Defaults to // acquire.DownloadClientFor. ClientFor func(ctx context.Context, code string) (downloader.Driver, error) // Now is the clock; nil is time.Now. Now func() time.Time logger *slog.Logger mu sync.Mutex // one pass at a time } // New returns a Service. func New(pool *pgxpool.Pool, logger *slog.Logger) *Service { if logger == nil { logger = slog.Default() } return &Service{pool: pool, logger: logger, ClientFor: func(ctx context.Context, code string) (downloader.Driver, error) { return acquire.DownloadClientFor(ctx, pool, code) }} } func (s *Service) now() time.Time { if s.Now != nil { return s.Now() } return time.Now() } // Run passes a minute after start, then every ten minutes, until ctx ends. // Each pass is bounded, so a client that never answers costs one pass. func (s *Service) Run(ctx context.Context) { first := time.NewTimer(time.Minute) defer first.Stop() select { case <-ctx.Done(): return case <-first.C: } t := time.NewTicker(interval) defer t.Stop() for { rep := s.Pass(ctx) if rep.Retried+rep.GivenUp+rep.Gone+rep.Released+rep.Removed > 0 || rep.Error != "" { s.logger.Info("lifecycle: pass", "retried", rep.Retried, "given_up", rep.GivenUp, "gone", rep.Gone, "released", rep.Released, "removed", rep.Removed, "raised", rep.Raised, "settled", rep.Settled, "error", rep.Error) } select { case <-ctx.Done(): return case <-t.C: } } } // Report is one pass. type Report struct { // Raised and Settled count ledger writes; the rest count acts. Raised, Settled int Retried, GivenUp, Gone, Released, Removed int Error string } // Pass looks at every running download, every seeding one and every one // left behind, once. func (s *Service) Pass(ctx context.Context) Report { s.mu.Lock() defer s.mu.Unlock() ctx, cancel := context.WithTimeout(ctx, passTimeout) defer cancel() var rep Report q := dbq.New(s.pool) set := s.settings(ctx, q) if _, err := q.DeleteExpiredBlocks(ctx); err != nil { s.logger.Warn("lifecycle: lift expired holds", "err", err) } if n, err := q.SettleEndedGrabs(ctx); err == nil { rep.Settled += int(n) } live, err := q.ListLiveGrabs(ctx) if err != nil { rep.Error = err.Error() return rep } seeds, err := q.ListSeedingGrabs(ctx) if err != nil { rep.Error = err.Error() return rep } left, err := q.ListLeftovers(ctx) if err != nil { rep.Error = err.Error() return rep } work := map[string]*clientWork{} var order []string of := func(code string) *clientWork { if w, ok := work[code]; ok { return w } order = append(order, code) work[code] = &clientWork{} return work[code] } for _, g := range live { if g.InfoHash != "" { of(g.Client).live = append(of(g.Client).live, g) } } for _, g := range seeds { if g.InfoHash != "" { of(g.Client).seeds = append(of(g.Client).seeds, g) } } for _, l := range left { of(l.Client).left = append(of(l.Client).left, l) } for _, code := range order { if ctx.Err() != nil { break } s.passClient(ctx, q, code, work[code], set, &rep) } return rep } type clientWork struct { live []dbq.ListLiveGrabsRow seeds []dbq.ListSeedingGrabsRow left []dbq.DownloadIntervention } func (s *Service) settings(ctx context.Context, q *dbq.Queries) Settings { row, err := q.GetAcquisitionSettings(ctx) if err != nil { // A blip keeps the defaults rather than stopping the pass. return Defaults } return Settings{ SeedRatio: float64(row.SeedRatio), SeedTime: time.Duration(row.SeedDays) * 24 * time.Hour, Patience: time.Duration(row.StallHours) * time.Hour, } } func (s *Service) passClient(ctx context.Context, q *dbq.Queries, code string, w *clientWork, set Settings, rep *Report) { d, err := s.ClientFor(ctx, code) if err != nil { s.raise(ctx, q, code, "", pgtype.UUID{}, "", ClientUnreachable, firstLine(err.Error()), rep) return } var hashes []string for _, g := range w.live { hashes = append(hashes, g.InfoHash) } for _, g := range w.seeds { hashes = append(hashes, g.InfoHash) } for _, l := range w.left { hashes = append(hashes, l.InfoHash) } ts, err := d.Torrents(ctx, hashes...) if err != nil { // Nothing about any one download is observed through a client that // cannot answer. s.raise(ctx, q, code, "", pgtype.UUID{}, "", ClientUnreachable, firstLine(err.Error()), rep) return } for _, t := range ts { if t.EngineDown != "" { s.raise(ctx, q, code, "", pgtype.UUID{}, "", ClientUnreachable, t.EngineDown, rep) return } } s.settle(ctx, q, code, "", ClientUnreachable, rep) byHash := make(map[string]downloader.Torrent, len(ts)) for _, t := range ts { byHash[indexer.NormalizeHash(t.Hash)] = t } // ABSENCE IS ONLY A MEASUREMENT FROM A CLIENT THAT LISTS SOMETHING. An // engine switched off, or a store that would not read, answers with an // empty list; then nothing is concluded from what is missing. listed := len(ts) > 0 if !listed { all, err := d.Torrents(ctx) listed = err == nil && len(all) > 0 } for _, l := range w.left { t, ok := byHash[indexer.NormalizeHash(l.InfoHash)] s.removeAgain(ctx, q, d, l, t, ok, listed, rep) } for _, g := range w.live { t, ok := byHash[indexer.NormalizeHash(g.InfoHash)] s.judge(ctx, q, d, g, t, ok, listed, set, rep) } for _, g := range w.seeds { t, ok := byHash[indexer.NormalizeHash(g.InfoHash)] s.seed(ctx, q, d, g, t, ok, set, rep) } } // judge decides what, if anything, is wrong with one running download, and // answers each condition either way: raised when true, settled when not. func (s *Service) judge(ctx context.Context, q *dbq.Queries, d downloader.Driver, g dbq.ListLiveGrabsRow, t downloader.Torrent, ok, listed bool, set Settings, rep *Report) { if !isRunning(g.ImportState) { return // the importer's, or a person's } if !ok { if !listed { return } row, raised := s.raise(ctx, q, g.Client, g.InfoHash, g.ID, g.Name, Gone, "The download client no longer lists it.", rep) if raised && s.now().Sub(row.FirstSeen.Time) >= goneAfter { s.endGone(ctx, q, g, rep) } return } s.settle(ctx, q, g.Client, g.InfoHash, Gone, rep) if t.Done() { // Finished: whatever was wrong while it downloaded is over, and the // import is the importer's. for _, c := range []string{ClientError, NoMetadata, NoPeers} { s.settle(ctx, q, g.Client, g.InfoHash, c, rep) } return } // The client's own verdict, given up on at once: it has already made // the decision a retry would ask about (FL #4749). if t.Failed() { why := t.FailReason if why == "" { why = fmt.Sprintf("the download client reports it %q", t.State) } if _, raised := s.raise(ctx, q, g.Client, g.InfoHash, g.ID, g.Name, ClientError, why, rep); raised { s.giveUp(ctx, q, d, g, ClientError, EndFailed, why, rep) } return } s.settle(ctx, q, g.Client, g.InfoHash, ClientError, rep) // Below here the stall setting applies, measured from the grab. With it // off nothing is raised and what was raised clears. age := s.now().Sub(g.GrabbedAt.Time) patient := set.Patience > 0 && age > set.Patience if patient && t.Fetching() { row, raised := s.raise(ctx, q, g.Client, g.InfoHash, g.ID, g.Name, NoMetadata, fmt.Sprintf("Still fetching its file list after %s.", span(age)), rep) if raised { s.ladder(ctx, q, d, g, NoMetadata, row.Remediations, age, rep) } } else { s.settle(ctx, q, g.Client, g.InfoHash, NoMetadata, rep) } // Not while fetching: a torrent with no file list has no swarm to be // alone in. if patient && !t.Fetching() && t.Alone() { row, raised := s.raise(ctx, q, g.Client, g.InfoHash, g.ID, g.Name, NoPeers, fmt.Sprintf("Nobody has shared it in %s; %.0f%% downloaded.", span(age), t.Progress*100), rep) if raised { s.ladder(ctx, q, d, g, NoPeers, row.Remediations, age, rep) } } else { s.settle(ctx, q, g.Client, g.InfoHash, NoPeers, rep) } } // ladder retries a stuck download until the retries are spent, then gives // it up. The rung is the ledger row's, so a restart hands out no more. func (s *Service) ladder(ctx context.Context, q *dbq.Queries, d downloader.Driver, g dbq.ListLiveGrabsRow, trigger string, done int32, age time.Duration, rep *Report) { if int(done) < Retries { // Counted before it happens: a retry that happened and was not // counted is a bound that never binds. n, err := q.CountRemediation(ctx, dbq.CountRemediationParams{ Client: g.Client, InfoHash: g.InfoHash, Trigger: trigger, Action: actRetried}) if err != nil { return } // Exchange schedules its own announces and has no recheck: for it a // retry is the time between passes, and the ladder still climbs. if err := d.Reannounce(ctx, g.InfoHash); err != nil && !errors.Is(err, downloader.ErrUnsupported) { s.logger.Warn("lifecycle: reannounce", "grab", g.Name, "err", err) } if err := d.Recheck(ctx, g.InfoHash); err != nil && !errors.Is(err, downloader.ErrUnsupported) { s.logger.Warn("lifecycle: recheck", "grab", g.Name, "err", err) } s.logger.Info("lifecycle: retried a stalled download", "grab", g.Name, "trigger", trigger, "attempt", n) rep.Retried++ return } why := fmt.Sprintf("taken %s ago and never finished: %s", span(age), reasonFor(trigger)) s.giveUp(ctx, q, d, g, trigger, EndStalled, why, rep) } // giveUp ends the grab, holds the release back for StallBlock, removes the // download with its files and searches for the album again. func (s *Service) giveUp(ctx context.Context, q *dbq.Queries, d downloader.Driver, g dbq.ListLiveGrabsRow, trigger, end, why string, rep *Report) { until := s.now().Add(StallBlock) detail := capital(why) + fmt.Sprintf(". Held back until %s; the album is searched for again.", until.Format("2 Jan")) // The update is the lock: an import that claimed the download meanwhile // keeps it. if _, err := q.GiveUpGrab(ctx, dbq.GiveUpGrabParams{ID: g.ID, EndReason: end, EndDetail: detail, States: running}); err != nil { return } // Held back before it is removed: if the pass dies between the two, the // release is still not taken again. if err := acquire.NewMemory(s.pool).Block(ctx, dbq.BlockReleaseParams{ InfoHash: g.InfoHash, Name: g.Name, Reason: acquire.BlockStalled, Detail: capital(why) + ".", WantID: g.WantID, ExpiresAt: pgtype.Timestamptz{Time: until, Valid: true}, }); err != nil { s.logger.Warn("lifecycle: hold the release back", "grab", g.Name, "err", err) } removed := s.remove(ctx, q, d, g.Client, g.InfoHash, g.ID, g.Name, rep) if g.WantID.Valid { if err := q.SearchWantAgain(ctx, g.WantID); err != nil { s.logger.Warn("lifecycle: search again", "grab", g.Name, "err", err) } } if _, err := q.CountRemediation(ctx, dbq.CountRemediationParams{ Client: g.Client, InfoHash: g.InfoHash, Trigger: trigger, Action: actGivenUp}); err != nil { s.logger.Warn("lifecycle: stamp the ledger", "grab", g.Name, "err", err) } if removed { if n, err := q.SettleInterventionsFor(ctx, dbq.SettleInterventionsForParams{ Client: g.Client, InfoHash: g.InfoHash}); err == nil { rep.Settled += int(n) } } s.logger.Info("lifecycle: gave up on a download", "grab", g.Name, "trigger", trigger, "why", why) rep.GivenUp++ } // endGone ends a grab whose download left the client, and searches again. // The first time, the release is not held back: nothing was found wrong with // it. A release that leaves its client again within goneWindow is held back for // acquire.GoneBlock (#6176): whatever keeps the client from listing it, taking // it again on every pass found 38 grabs of one release in a day. func (s *Service) endGone(ctx context.Context, q *dbq.Queries, g dbq.ListLiveGrabsRow, rep *Report) { detail := "The download client no longer lists it: removed outside Minstrel. The album is searched for again." before, err := q.CountGoneGrabs(ctx, dbq.CountGoneGrabsParams{ InfoHash: g.InfoHash, Since: pgtype.Timestamptz{Time: s.now().Add(-goneWindow), Valid: true}}) if err != nil { s.logger.Warn("lifecycle: count earlier gone grabs", "grab", g.Name, "err", err) } again := before > 0 until := s.now().Add(acquire.GoneBlock) if again { detail = fmt.Sprintf("The download client no longer lists it, again: held back until %s; the album is searched for again.", until.Format("2 Jan 15:04")) } if _, err := q.GiveUpGrab(ctx, dbq.GiveUpGrabParams{ID: g.ID, EndReason: EndGone, EndDetail: detail, States: running}); err != nil { return } if again { if err := acquire.NewMemory(s.pool).Block(ctx, dbq.BlockReleaseParams{ InfoHash: g.InfoHash, Name: g.Name, Reason: acquire.BlockGone, Detail: "It left its download client twice without finishing.", WantID: g.WantID, ExpiresAt: pgtype.Timestamptz{Time: until, Valid: true}, }); err != nil { s.logger.Warn("lifecycle: hold the release back", "grab", g.Name, "err", err) } } if g.WantID.Valid { if err := q.SearchWantAgain(ctx, g.WantID); err != nil { s.logger.Warn("lifecycle: search again", "grab", g.Name, "err", err) } } if _, err := q.CountRemediation(ctx, dbq.CountRemediationParams{ Client: g.Client, InfoHash: g.InfoHash, Trigger: Gone, Action: actGivenUp}); err != nil { s.logger.Warn("lifecycle: stamp the ledger", "grab", g.Name, "err", err) } if n, err := q.SettleInterventionsFor(ctx, dbq.SettleInterventionsForParams{ Client: g.Client, InfoHash: g.InfoHash}); err == nil { rep.Settled += int(n) } s.logger.Info("lifecycle: a download left its client", "grab", g.Name) rep.Gone++ } // seed removes an imported download once it has shared enough. // // Absence is not a verdict: a restarting client answers with a partial list. // A seeding download missing from it is left until its time is up, then // recorded as released; there is nothing to delete. func (s *Service) seed(ctx context.Context, q *dbq.Queries, d downloader.Driver, g dbq.ListSeedingGrabsRow, t downloader.Torrent, ok bool, set Settings, rep *Report) { overdue := set.SeedTime > 0 && g.ImportedAt.Valid && s.now().Sub(g.ImportedAt.Time) >= set.SeedTime if !ok { if overdue { s.release(ctx, q, g.ID, rep) } return } if err := q.SetGrabRatio(ctx, dbq.SetGrabRatioParams{ID: g.ID, Ratio: float32(t.Ratio)}); err != nil { s.logger.Warn("lifecycle: record the ratio", "grab", g.Name, "err", err) } if t.Ratio < set.SeedRatio && !overdue { return } s.remove(ctx, q, d, g.Client, g.InfoHash, g.ID, g.Name, rep) s.logger.Info("lifecycle: seeded enough, removed the download", "grab", g.Name, "ratio", t.Ratio, "overdue", overdue) s.release(ctx, q, g.ID, rep) } func (s *Service) release(ctx context.Context, q *dbq.Queries, id pgtype.UUID, rep *Report) { if err := q.ReleaseSeed(ctx, id); err != nil { s.logger.Warn("lifecycle: record the seeding ended", "err", err) return } rep.Released++ } // remove deletes a download and its files. A client that refuses leaves an // open not_removed row, which every pass tries again. func (s *Service) remove(ctx context.Context, q *dbq.Queries, d downloader.Driver, code, hash string, grab pgtype.UUID, name string, rep *Report) bool { if err := d.Delete(ctx, true, hash); err != nil { s.logger.Warn("lifecycle: remove the download", "grab", name, "err", err) s.raise(ctx, q, code, hash, grab, name, NotRemoved, notRemoved(err), rep) return false } return true } // removeAgain asks the client once more to remove a download left behind, // unless it has been taken again since. func (s *Service) removeAgain(ctx context.Context, q *dbq.Queries, d downloader.Driver, l dbq.DownloadIntervention, t downloader.Torrent, ok, listed bool, rep *Report) { if !ok { if listed { s.settle(ctx, q, l.Client, l.InfoHash, NotRemoved, rep) } return } live, err := q.LiveGrabOnHash(ctx, l.InfoHash) if err != nil { return } if live { s.settle(ctx, q, l.Client, l.InfoHash, NotRemoved, rep) return } if err := d.Delete(ctx, true, l.InfoHash); err != nil { s.raise(ctx, q, l.Client, l.InfoHash, l.GrabID, t.Name, NotRemoved, notRemoved(err), rep) return } if _, err := q.CountRemediation(ctx, dbq.CountRemediationParams{ Client: l.Client, InfoHash: l.InfoHash, Trigger: NotRemoved, Action: actRemoved}); err != nil { s.logger.Warn("lifecycle: stamp the ledger", "hash", l.InfoHash, "err", err) } s.logger.Info("lifecycle: removed a download left behind", "name", l.Name) rep.Removed++ } // raise records a condition or agrees with the open row that says it. The // bool is false when the write failed: then nothing is acted on, since // without the row's count there is no bound. func (s *Service) raise(ctx context.Context, q *dbq.Queries, code, hash string, grab pgtype.UUID, name, trigger, detail string, rep *Report) (dbq.DownloadIntervention, bool) { row, err := q.RaiseIntervention(ctx, dbq.RaiseInterventionParams{ Client: code, InfoHash: hash, GrabID: grab, Name: name, Trigger: trigger, Detail: detail}) if err != nil { s.logger.Warn("lifecycle: record", "client", code, "hash", hash, "trigger", trigger, "err", err) return row, false } rep.Raised++ return row, true } func (s *Service) settle(ctx context.Context, q *dbq.Queries, code, hash, trigger string, rep *Report) { n, err := q.SettleIntervention(ctx, dbq.SettleInterventionParams{Client: code, InfoHash: hash, Trigger: trigger}) if err != nil { s.logger.Warn("lifecycle: settle", "client", code, "hash", hash, "trigger", trigger, "err", err) return } rep.Settled += int(n) } // ErrNotAbandonable: the download is not running; an import, or a person's // decision on one, has it. var ErrNotAbandonable = errors.New("lifecycle: that download is not running") // Abandon is a person giving up on a running download: the release is set // aside for good (their word about it), the download removed with its files, // and the album searched for again. func (s *Service) Abandon(ctx context.Context, grabID, by pgtype.UUID) (dbq.Grab, error) { q := dbq.New(s.pool) g, err := q.GetGrab(ctx, grabID) if err != nil { return dbq.Grab{}, err } if g.EndedAt.Valid || !isRunning(g.ImportState) { return dbq.Grab{}, ErrNotAbandonable } const why = "Given up by you before it finished. The album is searched for again." ended, err := q.GiveUpGrab(ctx, dbq.GiveUpGrabParams{ID: g.ID, EndReason: EndAbandoned, EndDetail: why, States: running}) if err != nil { return dbq.Grab{}, ErrNotAbandonable } if g.InfoHash != "" { if err := acquire.NewMemory(s.pool).Block(ctx, dbq.BlockReleaseParams{ InfoHash: g.InfoHash, Name: g.Name, Reason: acquire.BlockOperator, Detail: "Given up by you before it finished.", WantID: g.WantID, BlockedBy: by, }); err != nil { s.logger.Warn("lifecycle: set the release aside", "grab", g.Name, "err", err) } var rep Report if d, err := s.ClientFor(ctx, g.Client); err != nil { s.raise(ctx, q, g.Client, g.InfoHash, g.ID, g.Name, NotRemoved, notRemoved(err), &rep) } else if s.remove(ctx, q, d, g.Client, g.InfoHash, g.ID, g.Name, &rep) { _, _ = q.SettleInterventionsFor(ctx, dbq.SettleInterventionsForParams{Client: g.Client, InfoHash: g.InfoHash}) } } if g.WantID.Valid { if err := q.SearchWantAgain(ctx, g.WantID); err != nil { s.logger.Warn("lifecycle: search again", "grab", g.Name, "err", err) } } return ended, nil } func isRunning(state string) bool { for _, r := range running { if state == r { return true } } return false } // reasonFor is a condition in a person's words. func reasonFor(trigger string) string { switch trigger { case NoPeers: return "nobody was sharing it" case NoMetadata: return "its file list never arrived" case ClientError: return "the download client gave up on it" } return "it stalled" } func notRemoved(err error) string { return "Given up on, but the download client would not remove it: " + firstLine(err.Error()) + ". Minstrel asks again every ten minutes." } // span is a duration as a person says it. func span(d time.Duration) string { switch { case d < 2*time.Hour: return fmt.Sprintf("%d minutes", int(d.Minutes())) case d < 48*time.Hour: return fmt.Sprintf("%d hours", int(d.Hours())) } return fmt.Sprintf("%d days", int(d.Hours()/24)) } func capital(s string) string { if s == "" { return s } return strings.ToUpper(s[:1]) + s[1:] } func firstLine(s string) string { if i := strings.IndexByte(s, '\n'); i >= 0 { return strings.TrimSpace(s[:i]) } return strings.TrimSpace(s) }