From 52f97106bdc9a1a45eb4dd47c81f5cf06d31a00c Mon Sep 17 00:00:00 2001 From: Piotr Komborski <163394+combor@users.noreply.github.com> Date: Wed, 7 Oct 2026 09:25:34 +0100 Subject: [PATCH] Preserve UI history after Sonarr and Radarr cleanup --- README.md | 2 +- cmd/magnetowid/arrs_test.go | 92 ++++++++++- docs/usage.md | 14 +- internal/downloader/archive_test.go | 237 ++++++++++++++++++++++++++++ internal/downloader/queue.go | 110 ++++++++++--- internal/downloader/store.go | 42 +++-- internal/sabnzbd/sabnzbd.go | 2 +- internal/sabnzbd/sabnzbd_test.go | 68 +++++++- internal/web/overrides.go | 2 +- internal/web/templates/history.html | 9 +- internal/web/view.go | 2 + internal/web/web.go | 2 +- internal/web/web_test.go | 48 +++++- 13 files changed, 583 insertions(+), 47 deletions(-) create mode 100644 internal/downloader/archive_test.go diff --git a/README.md b/README.md index 4acedea..9349138 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,7 @@ Open `http://:8484/` and sign in with the API key. | Page | What it is for | |---|---| | **Queue** | The running download's progress and time left, and what is up next. Pause, resume or remove downloads. | -| **History** | Finished and failed downloads from the last 30 days, with the error of any that failed. | +| **History** | Completed and failed downloads from the last 90 days, kept after Sonarr or Radarr removes them. | | **Overrides** | Forms to [correct a match](docs/usage.md#correcting-matches) for a series or a film. | | **Setup** | The values to enter in Sonarr and Radarr, and whether each site is reachable. | diff --git a/cmd/magnetowid/arrs_test.go b/cmd/magnetowid/arrs_test.go index 11fad8b..ffedfb3 100644 --- a/cmd/magnetowid/arrs_test.go +++ b/cmd/magnetowid/arrs_test.go @@ -106,9 +106,12 @@ func testSonarr(ctx context.Context, t *testing.T, a *arr, mw *magnetowid) { // TVDB search must register the series for RSS. a.command(ctx, t, map[string]any{"name": "EpisodeSearch", "episodeIds": []int{e2}}) - a.checkGrab(ctx, t, "episodeId", e2, "Days.of.Honor.S01E02.POLISH.1080p.WEB-DL.AAC.H.264-TVP", "seriesMatchType", "") + downloadID := a.checkGrab(ctx, t, "episodeId", e2, "Days.of.Honor.S01E02.POLISH.1080p.WEB-DL.AAC.H.264-TVP", "seriesMatchType", "") checkFile(t, a.waitImport(ctx, t, episodeFile(e2))) a.checkSubtitles(t, 1) + mw.checkArchived(ctx, t, a, downloadID) + mw.restart(ctx, t) + mw.checkArchived(ctx, t, a, downloadID) // Episode 3 aired yesterday in the fixture. mw.waitFeed(ctx, t, url.Values{"t": {"tvsearch"}, "cat": {"5000,5040"}}, "Days.of.Honor.S01E03.") @@ -142,9 +145,10 @@ func testRadarr(ctx context.Context, t *testing.T, a *arr, mw *magnetowid, sites // Cube is dated 1998 in Radarr, 1997 on TVP. cubeID := a.addMovie(ctx, t, 431) a.command(ctx, t, map[string]any{"name": "MoviesSearch", "movieIds": []int{cubeID}}) - a.checkGrab(ctx, t, "movieId", cubeID, "Cube.1998.POLISH.1080p.WEB-DL.AAC.H.264-TVP", "movieMatchType", "") + downloadID := a.checkGrab(ctx, t, "movieId", cubeID, "Cube.1998.POLISH.1080p.WEB-DL.AAC.H.264-TVP", "movieMatchType", "") checkFile(t, a.waitImport(ctx, t, movieFile(cubeID))) a.checkSubtitles(t, 1) + mw.checkArchived(ctx, t, a, downloadID) // Search before TVP lists the film, then restart to bypass the 10-minute // feed cache and verify the watch list survives. @@ -152,6 +156,7 @@ func testRadarr(ctx context.Context, t *testing.T, a *arr, mw *magnetowid, sites a.command(ctx, t, map[string]any{"name": "MoviesSearch", "movieIds": []int{sexmissionID}}) sites.release(seksmisja) mw.restart(ctx, t) + mw.checkArchived(ctx, t, a, downloadID) mw.waitFeed(ctx, t, url.Values{"t": {"movie"}, "cat": {"2000,2040"}}, "Seksmisja.1984.") a.command(ctx, t, map[string]any{"name": "RssSync"}) a.checkGrab(ctx, t, "movieId", sexmissionID, "Seksmisja.1984.POLISH.1080p.WEB-DL.AAC.H.264-TVP", "movieMatchType", "Rss") @@ -267,6 +272,76 @@ func (mw *magnetowid) restart(ctx context.Context, t *testing.T) { mw.start(ctx, t) } +// Confirm real client cleanup hides the job from the API but retains its UI row. +func (mw *magnetowid) checkArchived(ctx context.Context, t *testing.T, a *arr, id string) { + t.Helper() + wait, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + req, err := http.NewRequestWithContext(wait, http.MethodPost, "http://"+mw.addr+"/ui/login", + strings.NewReader(url.Values{"apikey": {mw.apiKey}}.Encode())) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusSeeOther || len(resp.Cookies()) == 0 { + t.Fatalf("UI sign-in returned %s", resp.Status) + } + cookies := resp.Cookies() + params := url.Values{"apikey": {mw.apiKey}, "mode": {"history"}, "output": {"json"}} + for { + a.command(wait, t, map[string]any{"name": "RefreshMonitoredDownloads"}) + var history struct { + History struct { + Slots []struct { + ID string `json:"nzo_id"` + } + } + } + if err := getJSON(wait, "http://"+mw.addr+"/api?"+params.Encode(), &history); err != nil { + t.Fatal(err) + } + visible := false + for _, slot := range history.History.Slots { + visible = visible || slot.ID == id + } + if !visible { + req, err := http.NewRequestWithContext(wait, http.MethodGet, "http://"+mw.addr+"/ui/history", nil) + if err != nil { + t.Fatal(err) + } + for _, c := range cookies { + req.AddCookie(c) + } + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil || resp.StatusCode != http.StatusOK { + t.Fatalf("history page: %s, %v", resp.Status, err) + } + _, row, found := strings.Cut(string(body), `id="job-`+id+`"`) + row, _, _ = strings.Cut(row, "") + if !found || !strings.Contains(row, "Archived") || !strings.Contains(row, "Completed") || strings.Contains(row, `name="files"`) { + t.Fatalf("imported download %s lacks its archived UI record", id) + } + return + } + select { + case <-wait.Done(): + t.Fatalf("%s never cleared imported download %s", a.name, id) + case <-time.After(500 * time.Millisecond): + } + } +} + // The first RSS request starts an asynchronous rebuild. func (mw *magnetowid) waitFeed(ctx context.Context, t *testing.T, params url.Values, prefix string) { t.Helper() @@ -438,6 +513,8 @@ func (a *arr) connect(ctx context.Context, t *testing.T, mw *magnetowid, categor client := a.schema(ctx, t, "/downloadclient/schema", "Sabnzbd") client["name"] = "magnetowid" client["enable"] = true + client["removeCompletedDownloads"] = true + client["removeFailedDownloads"] = true setFields(t, client, map[string]any{"host": host, "port": port, "apiKey": mw.apiKey, categoryField: category}) var added struct{ ID int } a.call(ctx, t, "POST", "/downloadclient", client, &added) @@ -458,6 +535,8 @@ func (a *arr) connect(ctx context.Context, t *testing.T, mw *magnetowid, categor a.call(ctx, t, "GET", "/config/mediamanagement", nil, &media) media["importExtraFiles"] = true media["extraFileExtensions"] = "srt" + // Releases advertise full runtimes, but the fixture downloads tiny videos. + media["skipFreeSpaceCheckWhenImporting"] = true a.call(ctx, t, "PUT", fmt.Sprintf("/config/mediamanagement/%v", media["id"]), media, nil) } @@ -517,7 +596,7 @@ func (a *arr) addMovie(ctx context.Context, t *testing.T, tmdbID int) int { return added.ID } -func (a *arr) checkGrab(ctx context.Context, t *testing.T, idField string, id int, release, matchField, source string) { +func (a *arr) checkGrab(ctx context.Context, t *testing.T, idField string, id int, release, matchField, source string) string { t.Helper() var history struct{ Records []map[string]any } a.call(ctx, t, "GET", "/history?pageSize=100&sortKey=date&sortDirection=descending", nil, &history) @@ -544,7 +623,11 @@ func (a *arr) checkGrab(ctx context.Context, t *testing.T, idField string, id in if source != "" && grab.Data["releaseSource"] != source { t.Errorf("%s grabbed %s from %v, want %s", a.name, grab.SourceTitle, grab.Data["releaseSource"], source) } - return + downloadID, _ := r["downloadId"].(string) + if downloadID == "" { + t.Fatal("grab history lacks download ID") + } + return downloadID } // Report search rejection reasons on failure. var releases []struct { @@ -561,6 +644,7 @@ func (a *arr) checkGrab(ctx context.Context, t *testing.T, idField string, id in } } t.FailNow() + return "" } func (a *arr) waitImport(ctx context.Context, t *testing.T, file func() map[string]any) map[string]any { diff --git a/docs/usage.md b/docs/usage.md index 913384d..4d408e8 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -182,10 +182,14 @@ every second. Pause or resume the queue or a job, or remove a job; removing an unfinished download deletes its partial files. Resumed downloads continue where they stopped. -**History** lists finished and failed downloads, newest first, and refreshes -every five seconds. Downloads stay for 30 days unless Sonarr or Radarr removes -them first. Removing one can also delete its folder; Sonarr and Radarr cannot -import a download removed from history. +**History** lists completed and failed downloads, newest first, and refreshes +every five seconds. Records stay for 90 days after the download finishes, +unless you remove them. When Sonarr or Radarr removes a download, its record +stays here as **Archived**, with its original result. + +Removing an archived record keeps the files. If an unarchived download has a +folder, you can delete it too. Removing a record before Sonarr or Radarr imports +the download prevents that import. **Overrides** lists each site's [overrides](#correcting-matches), with forms to add, change and remove them for series and films. A form that can't be saved @@ -328,7 +332,7 @@ Include the steps to reproduce and any relevant error message, with API keys rem known audio language. Probes run during search and are cached for a day. Downloads keep one audio track; subtitle conversion supports TTML only. - **Restarts:** `.magnetowid-jobs.db` stores the queue, history, and watch lists. - Completed history expires after 30 days. The download filesystem must support + Download history expires after 90 days. The download filesystem must support file locks; only one magnetowid instance can use it at a time. - **Partial downloads:** magnetowid fetches HLS segments, four at a time, into `.incomplete/` in the download folder, then remuxes them with ffmpeg. diff --git a/internal/downloader/archive_test.go b/internal/downloader/archive_test.go new file mode 100644 index 0000000..e783d77 --- /dev/null +++ b/internal/downloader/archive_test.go @@ -0,0 +1,237 @@ +package downloader + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "reflect" + "slices" + "testing" + "time" + + bolt "go.etcd.io/bbolt" + + "github.com/combor/magnetowid/internal/nzb" + "github.com/combor/magnetowid/internal/provider" +) + +func TestClientRemovalArchivesFinishedJobs(t *testing.T) { + for _, failed := range []bool{false, true} { + for _, deleteFiles := range []bool{false, true} { + t.Run(fmt.Sprintf("failed=%t/files=%t", failed, deleteFiles), func(t *testing.T) { + dir := t.TempDir() + p := &fakeProvider{} + if failed { + p.err = provider.ErrUnavailable + } + q := newQueue(t, dir, p, &fakeEngine{}) + stop := run(t, q) + want := waitFinished(t, q, add(t, q, "Movie.2020", "movies", 0, nzb.Ref{Provider: "fake", ID: "1"})) + stop() + if ok, err := q.RemoveForClient(want.ID, deleteFiles); !ok || err != nil { + t.Fatalf("remove = %v, %v", ok, err) + } + check := func() { + t.Helper() + if len(q.Jobs()) != 0 { + t.Fatalf("client still sees archived jobs: %+v", q.Jobs()) + } + all := q.AllJobs() + if len(all) != 1 || !all[0].Archived { + t.Fatalf("UI history = %+v", all) + } + gotJSON, _ := json.Marshal(all[0]) + wantJSON, _ := json.Marshal(want) + if string(gotJSON) != string(wantJSON) { + t.Errorf("archive = %s, want %s", gotJSON, wantJSON) + } + } + check() + q.db.Close() + q = newQueue(t, dir, p, &fakeEngine{}) + check() + if !failed { + _, err := os.Stat(want.Storage) + if (deleteFiles && !os.IsNotExist(err)) || (!deleteFiles && err != nil) { + t.Errorf("files after removal: %v", err) + } + } + for _, id := range []string{want.ID, "unknown"} { + if ok, err := q.RemoveForClient(id, true); ok || err != nil { + t.Errorf("repeated/unknown removal = %v, %v", ok, err) + } + } + check() + if ok, err := q.Cancel(want.ID); ok || err != nil { + t.Errorf("cancel archive = %v, %v", ok, err) + } + if ok, err := q.Delete(want.ID, true); !ok || err != nil { + t.Fatalf("forget archive = %v, %v", ok, err) + } + if !failed && !deleteFiles { + if _, err := os.Stat(want.Storage); err != nil { + t.Errorf("forgetting archive deleted retained files: %v", err) + } + } + q.db.Close() + q = newQueue(t, dir, p, &fakeEngine{}) + if len(q.AllJobs()) != 0 { + t.Fatal("forgotten archive returned after restart") + } + }) + } + } +} + +func TestArchivedStorageCanBeReusedSafely(t *testing.T) { + q := startQueue(t, &fakeProvider{}, &fakeEngine{}) + ref := nzb.Ref{Provider: "fake", ID: "1"} + first := waitFinished(t, q, add(t, q, "Movie.2020", "movies", 0, ref)) + if ok, err := q.RemoveForClient(first.ID, true); !ok || err != nil { + t.Fatalf("archive = %v, %v", ok, err) + } + second := waitFinished(t, q, add(t, q, "Movie.2020", "movies", 0, ref)) + if second.Storage != first.Storage { + t.Fatalf("archive reserved an unused folder: %q != %q", second.Storage, first.Storage) + } + if ok, err := q.RemoveForClient(first.ID, true); ok || err != nil { + t.Fatalf("repeat removal = %v, %v", ok, err) + } + if ok, err := q.Delete(first.ID, true); !ok || err != nil { + t.Fatalf("forget archive = %v, %v", ok, err) + } + if _, err := os.Stat(filepath.Join(second.Storage, "Movie.2020.mp4")); err != nil { + t.Errorf("new download lost its files: %v", err) + } +} + +func TestClientRemovalCancelsUnfinishedJobs(t *testing.T) { + for _, running := range []bool{false, true} { + t.Run(fmt.Sprint(running), func(t *testing.T) { + e := &blockingEngine{started: make(chan string, 1)} + q := newQueue(t, t.TempDir(), &fakeProvider{}, e) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "1"}) + stop := func() {} + if running { + stop = run(t, q) + <-e.started + } + if ok, err := q.RemoveForClient(id, true); !ok || err != nil { + t.Fatalf("remove = %v, %v", ok, err) + } + stop() + if len(q.AllJobs()) != 0 { + t.Fatal("cancelled job was archived") + } + if _, err := os.Stat(q.workDir(id)); !os.IsNotExist(err) { + t.Errorf("unfinished files remain: %v", err) + } + }) + } +} + +func TestArchiveWriteFailuresLeaveRecordsAndFiles(t *testing.T) { + q := newQueue(t, t.TempDir(), &fakeProvider{}, &fakeEngine{}) + stop := run(t, q) + ref := nzb.Ref{Provider: "fake", ID: "1"} + a := waitFinished(t, q, add(t, q, "archived", "tv", 0, ref)) + b := waitFinished(t, q, add(t, q, "completed", "tv", 0, ref)) + stop() + if _, err := q.RemoveForClient(a.ID, false); err != nil { + t.Fatal(err) + } + before := q.AllJobs() + q.db.Close() + if ok, err := q.RemoveForClient(b.ID, true); ok || err == nil { + t.Errorf("archive without saving = %v, %v", ok, err) + } + if ok, err := q.Delete(a.ID, true); ok || err == nil { + t.Errorf("delete archive without saving = %v, %v", ok, err) + } + q.mu.Lock() + q.prune(time.Now().Add(HistoryRetention + time.Hour)) + q.mu.Unlock() + if !reflect.DeepEqual(before, q.AllJobs()) { + t.Fatal("failed writes changed history") + } + for _, j := range []Job{a, b} { + if _, err := os.Stat(j.Storage); err != nil { + t.Errorf("failed write removed files: %v", err) + } + } +} + +func TestPruneArchivesAndHistory(t *testing.T) { + dir := t.TempDir() + q := newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) + // Keep the exact boundary stable when reopening the database below. + now := time.Now().Add(time.Hour) + const retention = 90 * 24 * time.Hour + var want, folders []string + for _, archived := range []bool{false, true} { + for _, status := range []Status{StatusCompleted, StatusFailed} { + for _, age := range []time.Duration{retention + time.Nanosecond, retention, 45 * 24 * time.Hour} { + id := fmt.Sprintf("%t-%s-%s", archived, status, age) + j := &Job{ID: id, Status: status, Finished: now.Add(-age), Storage: filepath.Join(dir, id)} + if err := os.Mkdir(j.Storage, 0o755); err != nil { + t.Fatal(err) + } + folders = append(folders, j.Storage) + if err := q.put(j); err != nil { + t.Fatal(err) + } + q.jobs[id], q.order = j, append(q.order, id) + if archived { + if _, err := q.RemoveForClient(id, false); err != nil { + t.Fatal(err) + } + } + if age <= retention { + want = append(want, id) + } + } + } + } + // A retrying job has a finish time but must never be pruned. + id := add(t, q, "retrying", "tv", 0, nzb.Ref{Provider: "fake", ID: "1"}) + q.jobs[id].Finished = now.Add(-2 * HistoryRetention) + if err := q.put(q.jobs[id]); err != nil { + t.Fatal(err) + } + want = append(want, id) + q.mu.Lock() + q.prune(now) + q.mu.Unlock() + q.db.Close() + q = newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) + var got []string + for _, j := range q.AllJobs() { + got = append(got, j.ID) + } + slices.Sort(got) + slices.Sort(want) + if !slices.Equal(got, want) { + t.Errorf("retained %v, want %v", got, want) + } + for _, folder := range folders { + if _, err := os.Stat(folder); err != nil { + t.Errorf("pruning removed a download folder: %v", err) + } + } +} + +func TestExistingDatabaseGainsArchive(t *testing.T) { + dir := t.TempDir() + q := newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) + id := add(t, q, "a", "tv", 0, nzb.Ref{Provider: "fake", ID: "1"}) + // Reproduce the old layout, with no archive bucket. + if err := q.db.Update(func(tx *bolt.Tx) error { return tx.DeleteBucket(archivedBucket) }); err != nil { + t.Fatal(err) + } + q.db.Close() + q = newQueue(t, dir, &fakeProvider{}, &fakeEngine{}) + if jobs := q.AllJobs(); len(jobs) != 1 || jobs[0].ID != id || jobs[0].Archived { + t.Fatalf("existing jobs changed: %+v", jobs) + } +} diff --git a/internal/downloader/queue.go b/internal/downloader/queue.go index 709a4ac..6a07ca1 100644 --- a/internal/downloader/queue.go +++ b/internal/downloader/queue.go @@ -2,6 +2,7 @@ package downloader import ( + "cmp" "context" "crypto/rand" "encoding/hex" @@ -83,6 +84,8 @@ type Job struct { Priority int `json:"priority,omitzero"` // -1 low, 0 normal, 1 high, 2 force // Separate from Status so older versions can still load paused jobs. Paused bool `json:"paused,omitzero"` + // Derived from archive bucket membership, for UI snapshots only. + Archived bool `json:"-"` startFraction float64 // when this attempt's progress began; not saved } @@ -112,7 +115,7 @@ func (j Job) TimeLeft(now time.Time) time.Duration { func (j Job) Unreachable() bool { return strings.HasPrefix(j.Error, errUnreachable.Error()+": ") } // HistoryRetention is how long finished jobs are kept. -const HistoryRetention = 30 * 24 * time.Hour +const HistoryRetention = 90 * 24 * time.Hour const incompleteDir = ".incomplete" @@ -126,13 +129,14 @@ type Queue struct { retryDelay func(n int) time.Duration - mu sync.Mutex - jobs map[string]*Job - order []string - cancels map[string]context.CancelCauseFunc - outages map[string]*outage // by provider name - paused bool - wake chan struct{} + mu sync.Mutex + jobs map[string]*Job + archived map[string]*Job + order []string + cancels map[string]context.CancelCauseFunc + outages map[string]*outage // by provider name + paused bool + wake chan struct{} } type outage struct { @@ -142,7 +146,11 @@ type outage struct { // New requires an absolute download path. Close db only after Run returns. func New(dir string, db *bolt.DB, providers *provider.Registry, engine Engine, log *slog.Logger) (*Queue, error) { - jobs, err := loadJobs(db) + jobs, err := loadJobs(db, jobsBucket) + if err != nil { + return nil, err + } + archived, err := loadJobs(db, archivedBucket) if err != nil { return nil, err } @@ -158,6 +166,7 @@ func New(dir string, db *bolt.DB, providers *provider.Registry, engine Engine, l db: db, retryDelay: backoff, jobs: make(map[string]*Job), + archived: make(map[string]*Job), cancels: make(map[string]context.CancelCauseFunc), outages: make(map[string]*outage), paused: paused, @@ -167,6 +176,9 @@ func New(dir string, db *bolt.DB, providers *provider.Registry, engine Engine, l q.jobs[j.ID] = j q.order = append(q.order, j.ID) } + for _, j := range archived { + q.archived[j.ID] = j + } q.prune(time.Now()) q.cleanWork() return q, nil @@ -322,7 +334,7 @@ func (q *Queue) setJobsPaused(paused bool, ids []string) ([]string, error) { return done, nil } -// Jobs returns snapshots in insertion order. +// Jobs returns unarchived snapshots in insertion order for download clients. func (q *Queue) Jobs() []Job { q.mu.Lock() defer q.mu.Unlock() @@ -333,29 +345,81 @@ func (q *Queue) Jobs() []Job { return out } -// Delete cancels the job and optionally removes its files. It returns false for -// unknown IDs; a save failure leaves the job unchanged. +// AllJobs returns current and archived snapshots in insertion order for the UI. +func (q *Queue) AllJobs() []Job { + q.mu.Lock() + out := make([]Job, 0, len(q.jobs)+len(q.archived)) + for _, j := range q.jobs { + out = append(out, *j) + } + for _, j := range q.archived { + snapshot := *j + snapshot.Archived = true + out = append(out, snapshot) + } + q.mu.Unlock() + slices.SortFunc(out, func(a, b Job) int { + return cmp.Or(a.Added.Compare(b.Added), strings.Compare(a.ID, b.ID)) + }) + return out +} + +type removalMode int + +const ( + removePermanently removalMode = iota + removeUnfinished + removeForClient +) + +// Delete permanently removes a job, optionally deleting its files. Archived +// records never delete files. Unknown IDs return false; save failures change nothing. func (q *Queue) Delete(id string, deleteFiles bool) (bool, error) { - return q.discard(id, deleteFiles, false) + return q.discard(id, deleteFiles, removePermanently) +} + +// RemoveForClient archives finished jobs and cancels unfinished jobs. Already +// archived and unknown IDs return false without touching their records or files. +func (q *Queue) RemoveForClient(id string, deleteFiles bool) (bool, error) { + return q.discard(id, deleteFiles, removeForClient) } // Cancel deletes an unfinished job. It returns false for unknown and finished // jobs, which stay in history for import. func (q *Queue) Cancel(id string) (bool, error) { - return q.discard(id, false, true) + return q.discard(id, false, removeUnfinished) } -func (q *Queue) discard(id string, deleteFiles, unfinishedOnly bool) (bool, error) { +func (q *Queue) discard(id string, deleteFiles bool, mode removalMode) (bool, error) { q.mu.Lock() + if mode == removePermanently && q.archived[id] != nil { + defer q.mu.Unlock() + if err := q.remove(id); err != nil { + return false, err + } + delete(q.archived, id) + q.log.Info("archived job deleted", "id", id) + return true, nil + } job, ok := q.jobs[id] - if !ok || (unfinishedOnly && job.Status != StatusQueued && job.Status != StatusDownloading) { + if !ok || (mode == removeUnfinished && job.Status != StatusQueued && job.Status != StatusDownloading) { q.mu.Unlock() return false, nil } - if err := q.remove(id); err != nil { + archive := mode == removeForClient && (job.Status == StatusCompleted || job.Status == StatusFailed) + var err error + if archive { + err = q.archive(job) + } else { + err = q.remove(id) + } + if err != nil { q.mu.Unlock() return false, err } + if archive { + q.archived[id] = job + } // A running job's worker removes its files once the engine stops. running := job.Status == StatusDownloading delete(q.jobs, id) @@ -380,7 +444,11 @@ func (q *Queue) discard(id string, deleteFiles, unfinishedOnly bool) (bool, erro if !running { q.removeWork(id) } - q.log.Info("job deleted", "id", id, "delete_files", deleteFiles) + message := "job deleted" + if archive { + message = "job archived" + } + q.log.Info(message, "id", id, "delete_files", deleteFiles) return true, nil } @@ -560,6 +628,11 @@ func (q *Queue) prune(now time.Time) { old = append(old, id) } } + for id, j := range q.archived { + if now.Sub(j.Finished) > HistoryRetention { + old = append(old, id) + } + } if len(old) == 0 { return } @@ -570,6 +643,7 @@ func (q *Queue) prune(now time.Time) { } for _, id := range old { delete(q.jobs, id) + delete(q.archived, id) } q.order = slices.DeleteFunc(q.order, func(id string) bool { return q.jobs[id] == nil }) q.log.Info("forgot old finished jobs", "count", len(old)) diff --git a/internal/downloader/store.go b/internal/downloader/store.go index 0db15ea..0fef413 100644 --- a/internal/downloader/store.go +++ b/internal/downloader/store.go @@ -11,15 +11,16 @@ import ( ) var ( - jobsBucket = []byte("jobs") - queueBucket = []byte("queue") - pausedKey = []byte("paused") + jobsBucket = []byte("jobs") + archivedBucket = []byte("archived_jobs") + queueBucket = []byte("queue") + pausedKey = []byte("paused") ) -func loadJobs(db *bolt.DB) ([]*Job, error) { +func loadJobs(db *bolt.DB, bucket []byte) ([]*Job, error) { var jobs []*Job err := db.Update(func(tx *bolt.Tx) error { - b, err := tx.CreateBucketIfNotExists(jobsBucket) + b, err := tx.CreateBucketIfNotExists(bucket) if err != nil { return err } @@ -33,7 +34,7 @@ func loadJobs(db *bolt.DB) ([]*Job, error) { }) }) if err != nil { - return nil, fmt.Errorf("loading jobs from %s: %w", db.Path(), err) + return nil, fmt.Errorf("loading %s from %s: %w", bucket, db.Path(), err) } // Random job IDs do not preserve insertion order. slices.SortFunc(jobs, func(a, b *Job) int { @@ -88,13 +89,32 @@ func (q *Queue) put(job *Job) error { return nil } -// Requires q.mu. +// archive moves a finished job out of client history atomically. Requires q.mu. +func (q *Queue) archive(job *Job) error { + v, err := json.Marshal(job) + if err == nil { + err = q.db.Update(func(tx *bolt.Tx) error { + if err := tx.Bucket(archivedBucket).Put([]byte(job.ID), v); err != nil { + return err + } + return tx.Bucket(jobsBucket).Delete([]byte(job.ID)) + }) + } + if err != nil { + return fmt.Errorf("archiving job %s: %w", job.ID, err) + } + return nil +} + +// remove permanently forgets jobs in either bucket. Requires q.mu. func (q *Queue) remove(ids ...string) error { err := q.db.Update(func(tx *bolt.Tx) error { - b := tx.Bucket(jobsBucket) - for _, id := range ids { - if err := b.Delete([]byte(id)); err != nil { - return err + for _, bucket := range [][]byte{jobsBucket, archivedBucket} { + b := tx.Bucket(bucket) + for _, id := range ids { + if err := b.Delete([]byte(id)); err != nil { + return err + } } } return nil diff --git a/internal/sabnzbd/sabnzbd.go b/internal/sabnzbd/sabnzbd.go index 14715fc..ce887f5 100644 --- a/internal/sabnzbd/sabnzbd.go +++ b/internal/sabnzbd/sabnzbd.go @@ -167,7 +167,7 @@ func (h *Handler) delete(w http.ResponseWriter, r *http.Request) { deleteFiles := r.FormValue("del_files") == "1" var ids []string for _, id := range jobIDs(r) { - ok, err := h.Queue.Delete(id, deleteFiles) + ok, err := h.Queue.RemoveForClient(id, deleteFiles) if err != nil { h.Log.Error("deleting job", "id", id, "err", err) writeJSON(w, errorResponse(err.Error())) diff --git a/internal/sabnzbd/sabnzbd_test.go b/internal/sabnzbd/sabnzbd_test.go index 36d460a..400214e 100644 --- a/internal/sabnzbd/sabnzbd_test.go +++ b/internal/sabnzbd/sabnzbd_test.go @@ -29,7 +29,10 @@ func (fakeProvider) Name() string { return "fake" } func (fakeProvider) Search(context.Context, provider.Query) ([]provider.Item, error) { return nil, nil } -func (fakeProvider) Resolve(context.Context, string) (provider.Stream, error) { +func (fakeProvider) Resolve(_ context.Context, id string) (provider.Stream, error) { + if id == "gone" { + return provider.Stream{}, provider.ErrUnavailable + } return provider.Stream{URL: "http://example.invalid/master.m3u8"}, nil } @@ -303,11 +306,74 @@ func TestHistoryAndDelete(t *testing.T) { if len(q.Jobs()) != 0 { t.Errorf("jobs after delete = %v", q.Jobs()) } + if all := q.AllJobs(); len(all) != 1 || all[0].ID != id || !all[0].Archived { + t.Errorf("UI history after delete = %+v", all) + } if _, err := os.Stat(storage); !os.IsNotExist(err) { t.Errorf("storage not removed: %v", err) } } +func TestArchivedHistoryIsHidden(t *testing.T) { + srv, q := newServer(t, true) + var ids []string + for _, tc := range []struct{ name, category, ref string }{ + {"A", "tv", "1"}, {"B", "tv", "gone"}, {"C", "tv", "2"}, {"D", "movies", "3"}, + } { + id, err := q.Add(tc.name, tc.name+".nzb", tc.category, 0, false, nzb.Ref{Provider: "fake", ID: tc.ref}) + if err != nil { + t.Fatal(err) + } + ids = append(ids, id) + } + for deadline := time.Now().Add(5 * time.Second); ; { + h := call(t, srv, url.Values{"mode": {"history"}})["history"].(map[string]any) + if h["noofslots"] == float64(4) { + break + } + if time.Now().After(deadline) { + t.Fatal("jobs never finished") + } + time.Sleep(10 * time.Millisecond) + } + // Both deletion modes must archive a job that has finished. + for i, mode := range []string{"queue", "history"} { + out := call(t, srv, url.Values{"mode": {mode}, "name": {"delete"}, "value": {ids[i]}}) + if out["status"] != true || len(out["nzo_ids"].([]any)) != 1 { + t.Fatalf("%s delete = %v", mode, out) + } + } + if len(q.Jobs()) != 2 || len(q.AllJobs()) != 4 { + t.Fatalf("client/UI counts = %d/%d", len(q.Jobs()), len(q.AllJobs())) + } + for _, tc := range []struct { + params url.Values + name string + total float64 + }{ + {url.Values{"category": {"tv"}}, "C", 1}, + {url.Values{"category": {"movies"}}, "D", 1}, + {url.Values{"start": {"1"}, "limit": {"1"}}, "C", 2}, + } { + tc.params.Set("mode", "history") + h := call(t, srv, tc.params)["history"].(map[string]any) + slots := h["slots"].([]any) + if h["noofslots"] != tc.total || len(slots) != 1 || slots[0].(map[string]any)["name"] != tc.name { + t.Errorf("filtered history = %v", h) + } + } + for _, mode := range []string{"queue", "history"} { + out := call(t, srv, url.Values{"mode": {mode}, "name": {"delete"}, "value": {ids[0] + "," + ids[1] + ",unknown"}, "del_files": {"1"}}) + removed, _ := out["nzo_ids"].([]any) + if out["status"] != true || len(removed) != 0 || len(q.AllJobs()) != 4 { + t.Errorf("repeated delete = %v; UI history = %+v", out, q.AllJobs()) + } + } + if _, err := os.Stat(q.AllJobs()[0].Storage); err != nil { + t.Errorf("repeated deletion removed retained files: %v", err) + } +} + func TestUnsavedChangesFail(t *testing.T) { db := openDB(t) srv, q := newServerOn(t, db, false) diff --git a/internal/web/overrides.go b/internal/web/overrides.go index 1f009e6..e0b5c96 100644 --- a/internal/web/overrides.go +++ b/internal/web/overrides.go @@ -515,7 +515,7 @@ func (h *Handler) topbar(w http.ResponseWriter, _ *http.Request) { } func (h *Handler) formChrome() chrome { - c := newChrome(h.Queue.Jobs(), h.Queue.Paused(), h.Version, "overrides") + c := newChrome(h.Queue.AllJobs(), h.Queue.Paused(), h.Version, "overrides") c.form = true return c } diff --git a/internal/web/templates/history.html b/internal/web/templates/history.html index 7efee13..c6f803f 100644 --- a/internal/web/templates/history.html +++ b/internal/web/templates/history.html @@ -13,6 +13,7 @@ {{- with .Took}}took {{.}}{{end -}} {{- with .Attempt}}{{.}} attempts{{end -}} finished {{.Finished}} + {{- if .Archived}}Archived{{end}}

@@ -21,6 +22,9 @@

Remove it from history?

{{.Label}}

+ {{- if .Archived}} +

This removes the record and keeps the files.

+ {{- else}} {{- with .Storage}}