diff --git a/docs/docs/4. API/endpoints.md b/docs/docs/4. API/endpoints.md index f0f3fb2ab..9d58723d5 100644 --- a/docs/docs/4. API/endpoints.md +++ b/docs/docs/4. API/endpoints.md @@ -56,7 +56,7 @@ The per-user **API key** (visible in **System → Settings**) is only honoured b | Endpoint | Accepts | |----------|---------| -| `/api/sabnzbd/*` — SABnzbd-compatible API | `?apikey=` or `ma_username` + `ma_password` | +| `/sabnzbd` — SABnzbd-compatible API | `?apikey=` or `ma_username` + `ma_password` | | `POST /api/arrs/webhook` — Sonarr/Radarr webhook | `?apikey=` (required) | | `POST /api/import/file` — manual NZB file import | `?apikey=` (required) | @@ -83,9 +83,28 @@ The Stremio addon is **not** authenticated with the API key. It uses a separate | **Auth** | `/api/auth` | Login, registration, auth config | | **User** | `/api/user` | Current user info, token refresh, API key management | +## SABnzbd History + +`GET /sabnzbd?mode=history&output=json&apikey=YOUR_API_KEY` returns completed and failed jobs in the SABnzbd response format (`history.slots` and `history.noofslots`). + +The normal history page is ordered newest first. `start` is the offset (default `0`), and `limit` is the page size (default `100`, capped at `1000`). Use `category` to filter jobs by category. + +For Sonarr/Radarr compatibility, the first page (`start=0`, including when omitted) also appends up to **1,000 retained failed jobs** that are absent from the normal page. Sonarr/Radarr poll the first page, so this keeps older failures visible during bursts of newer completed jobs. The extra jobs use the same category filter and exclude jobs marked `skip_arr_notification`; their availability follows the failed-job retention policy. + +This behavior applies to **all clients** calling this endpoint, without client detection: + +- The first page can contain more jobs than the requested `limit`. Extra failures are appended after the normal page, so the complete response may not be in chronological order. +- A failed job appended to the first page can appear again at its normal position on a later page. Clients traversing pages should deduplicate by `nzo_id`. +- `history.noofslots` remains the total number of unique jobs in the filtered history, rather than the number of slots returned on the first page. +- Later pages (`start>0`) retain normal pagination. Requests with `nzo_ids` use direct job lookup and do not append unrelated failures. + +For example, `start=0&limit=5` can return the five newest completed jobs plus an older failed job, for six slots in total. + +The frontend queue (`/api/queue`) and import history (`/api/import/history`) use separate endpoints; this first-page behavior does not change their ordering or pagination. + ## Response Format -All endpoints return a consistent JSON envelope: +REST endpoints under `/api` return a consistent JSON envelope: ```json { "success": true, "data": { ... } } diff --git a/internal/api/sabnzbd_handlers.go b/internal/api/sabnzbd_handlers.go index ff3b1363b..e061d5b1e 100644 --- a/internal/api/sabnzbd_handlers.go +++ b/internal/api/sabnzbd_handlers.go @@ -835,6 +835,33 @@ func (s *Server) handleSABnzbdHistory(c *fiber.Ctx) error { return s.writeSABnzbdErrorFiber(c, "Failed to get history") } + // #498: Sonarr and Radarr only ever read this first page (start=0, + // limit=DownloadClientHistoryLimit, default 60). A tracked download that is + // in neither the queue nor that page is dropped without a downloadFailed + // event, so no blocklist and no retry. During a burst of grabs, newer + // completed rows can push a failed row off the page before the *arr polls. + // Append every failed row still kept in import_queue that the page does not + // already hold. The page itself, noofslots and later pages are unchanged; a + // failed row can therefore appear both here and at its own position on a + // later page. + if start == 0 { + failedRows, err := s.queueRepo.ListSABnzbdFailedHistory(ctx, categoryFilter, maxSABnzbdHistoryLimit) + if err != nil { + return s.writeSABnzbdErrorFiber(c, "Failed to get history") + } + onPage := make(map[int64]bool, len(historyRows)) + for _, row := range historyRows { + if row.Source == "failed_queue" { + onPage[row.ID] = true + } + } + for _, row := range failedRows { + if !onPage[row.ID] { + historyRows = append(historyRows, row) + } + } + } + slots := make([]SABnzbdHistorySlot, 0, len(historyRows)) var totalBytes int64 itemBasePath := s.calculateItemBasePath() diff --git a/internal/api/sabnzbd_history_failed_first_test.go b/internal/api/sabnzbd_history_failed_first_test.go new file mode 100644 index 000000000..781cd0b20 --- /dev/null +++ b/internal/api/sabnzbd_history_failed_first_test.go @@ -0,0 +1,168 @@ +package api + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "github.com/gofiber/fiber/v2" + "github.com/kipsilabs/altmount/internal/database" + "github.com/kipsilabs/altmount/internal/importer" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Sonarr and Radarr poll SABnzbd history with start=0 and a fixed limit +// (DownloadClientHistoryLimit, default 60) and never read a later page. A +// tracked download that is in neither the queue nor that first page is made +// untrackable without a downloadFailed event, so a failed job must not be +// pushed off the first page by newer completed jobs (#498). + +func newSABnzbdHistoryTestServer(t *testing.T) (*Server, *database.Repository, *sql.DB) { + t.Helper() + + dbPath := filepath.Join(t.TempDir(), "test.db") + db, err := database.NewDB(database.Config{Type: "sqlite", DatabasePath: dbPath}) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + repo := database.NewRepository(db.Connection(), db.Dialect()) + server := &Server{ + queueRepo: repo, + importerService: &importer.Service{}, // non-nil sentinel, not used by the history handler + } + return server, repo, db.Connection() +} + +// addHistoryQueueRow inserts an import_queue row with the given status, +// category, download id and completion time. +func addHistoryQueueRow(t *testing.T, repo *database.Repository, conn *sql.DB, name string, status database.QueueStatus, category, downloadID string, at time.Time) int64 { + t.Helper() + + cat := category + item := &database.ImportQueueItem{ + NzbPath: "/nzbs/" + name + ".nzb", + Status: database.QueueStatusPending, + Priority: database.QueuePriorityNormal, + Category: &cat, + } + require.NoError(t, repo.AddToQueue(context.Background(), item)) + + ts := at.UTC().Format("2006-01-02 15:04:05") + _, err := conn.Exec(`UPDATE import_queue + SET status = ?, category = ?, download_id = ?, created_at = ?, updated_at = ?, completed_at = ? + WHERE id = ?`, string(status), category, downloadID, ts, ts, ts, item.ID) + require.NoError(t, err) + return item.ID +} + +func getSABnzbdHistory(t *testing.T, server *Server, query string) SABnzbdHistoryObject { + t.Helper() + + app := fiber.New() + app.Get("/sabnzbd", server.handleSABnzbdHistory) + + resp, err := app.Test(httptest.NewRequest("GET", "/sabnzbd?mode=history&"+query, nil)) + require.NoError(t, err) + require.Equal(t, 200, resp.StatusCode) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + + var out SABnzbdCompleteHistoryResponse + require.NoError(t, json.Unmarshal(body, &out), "body: %s", body) + return out.History +} + +func slotIDs(slots []SABnzbdHistorySlot) []string { + ids := make([]string, 0, len(slots)) + for _, s := range slots { + ids = append(ids, s.NzoID) + } + return ids +} + +// seedBurst creates one failed job in "tv", then `completed` newer completed +// jobs in "tv", plus one failed job in "movies" that a "tv" request must not see. +func seedBurst(t *testing.T, completed int) (*Server, []string) { + t.Helper() + + server, repo, conn := newSABnzbdHistoryTestServer(t) + base := time.Now().UTC().Add(-2 * time.Hour) + + addHistoryQueueRow(t, repo, conn, "failed-tv", database.QueueStatusFailed, "tv", "nzo_failed_tv", base) + addHistoryQueueRow(t, repo, conn, "failed-movies", database.QueueStatusFailed, "movies", "nzo_failed_movies", base) + + // Newest first, the order the history view returns them in. + newestFirst := make([]string, completed) + for i := 0; i < completed; i++ { + id := fmt.Sprintf("nzo_done_%02d", i) + addHistoryQueueRow(t, repo, conn, fmt.Sprintf("done-%02d", i), database.QueueStatusCompleted, "tv", id, + base.Add(time.Duration(i+1)*time.Minute)) + newestFirst[completed-1-i] = id + } + return server, newestFirst +} + +func TestSABnzbdHistory_FailedRowOnFirstPageDuringBurst(t *testing.T) { + const limit = 5 + server, newestCompleted := seedBurst(t, 8) + + page := getSABnzbdHistory(t, server, fmt.Sprintf("start=0&limit=%d&category=tv", limit)) + ids := slotIDs(page.Slots) + + // The failed tv job is older than every completed job, so by time alone it + // is on page 2. It must still be on page 0. + require.Contains(t, ids, "nzo_failed_tv", "failed job fell off the first history page") + for _, s := range page.Slots { + if s.NzoID == "nzo_failed_tv" { + assert.Equal(t, "Failed", s.Status) + } + } + + // The normal page is untouched: the `limit` newest completed jobs, in order. + require.GreaterOrEqual(t, len(ids), limit) + assert.Equal(t, newestCompleted[:limit], ids[:limit]) + + // The category filter applies to the extra rows as well. + assert.NotContains(t, ids, "nzo_failed_movies") + + // noofslots stays the real total of the tv view: 8 completed + 1 failed. + assert.Equal(t, 9, page.Noofslots) + + // Slot indices stay unique and continue after the normal page. + for i, s := range page.Slots { + assert.Equal(t, i, s.Index) + } +} + +func TestSABnzbdHistory_LaterPagesUnchanged(t *testing.T) { + server, newestCompleted := seedBurst(t, 8) + + page := getSABnzbdHistory(t, server, "start=5&limit=5&category=tv") + // Natural order: completed 5..7, then the failed job. No extra rows here. + assert.Equal(t, append(append([]string{}, newestCompleted[5:]...), "nzo_failed_tv"), slotIDs(page.Slots)) + assert.Equal(t, 9, page.Noofslots) + assert.Equal(t, 5, page.Slots[0].Index) +} + +func TestSABnzbdHistory_FailedAlreadyOnPageNotDuplicated(t *testing.T) { + server, _ := seedBurst(t, 2) + + page := getSABnzbdHistory(t, server, "start=0&limit=10&category=tv") + assert.Equal(t, []string{"nzo_done_01", "nzo_done_00", "nzo_failed_tv"}, slotIDs(page.Slots)) + assert.Equal(t, 3, page.Noofslots) +} + +func TestSABnzbdHistory_ByNzoIDsUnchanged(t *testing.T) { + server, _ := seedBurst(t, 8) + + page := getSABnzbdHistory(t, server, "start=0&limit=5&category=tv&nzo_ids=nzo_done_03") + assert.Equal(t, []string{"nzo_done_03"}, slotIDs(page.Slots)) +} diff --git a/internal/database/repository.go b/internal/database/repository.go index 78d16a03c..0003dd383 100644 --- a/internal/database/repository.go +++ b/internal/database/repository.go @@ -1404,6 +1404,39 @@ func (r *Repository) ListSABnzbdHistory(ctx context.Context, category string, li return out, rows.Err() } +// ListSABnzbdFailedHistory returns the failed-queue rows of the SABnzbd history +// view, newest first, at most max rows. It reads the same union as +// ListSABnzbdHistory, so the category and skip_arr_notification filters match +// the main page exactly. Failed rows stay in import_queue only until the +// FailedItemRetentionHours cleanup deletes them; when that cleanup is off they +// are kept forever, which is why max is required. +func (r *Repository) ListSABnzbdFailedHistory(ctx context.Context, category string, max int) ([]*SABnzbdHistoryRow, error) { + body, args := sabnzbdHistoryUnion(category) + query := `SELECT source, id, download_id, name, status, file_size, completed_at, category, storage_path, metadata, error_message + FROM (` + body + `) t + WHERE source = 'failed_queue' + ORDER BY sort_time DESC, id DESC + LIMIT ?` + args = append(args, max) + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list failed sabnzbd history: %w", err) + } + defer rows.Close() + + var out []*SABnzbdHistoryRow + for rows.Next() { + var row SABnzbdHistoryRow + if err := rows.Scan(&row.Source, &row.ID, &row.DownloadID, &row.Name, &row.Status, + &row.FileSize, &row.CompletedAt, &row.Category, &row.StoragePath, &row.Metadata, &row.ErrorMessage); err != nil { + return nil, fmt.Errorf("failed to scan failed sabnzbd history row: %w", err) + } + out = append(out, &row) + } + return out, rows.Err() +} + // CountSABnzbdHistory returns the exact pre-pagination total for the SABnzbd // history view (for noofslots). func (r *Repository) CountSABnzbdHistory(ctx context.Context, category string) (int, error) { diff --git a/internal/database/sabnzbd_history_test.go b/internal/database/sabnzbd_history_test.go index 7ffd911e9..722a92c28 100644 --- a/internal/database/sabnzbd_history_test.go +++ b/internal/database/sabnzbd_history_test.go @@ -149,6 +149,37 @@ func TestSABnzbdHistory_PaginationAndCategory(t *testing.T) { assert.Empty(t, empty) } +func TestSABnzbdFailedHistory_FiltersAndCap(t *testing.T) { + repo, db := newSABHistoryRepo(t) + ctx := context.Background() + base := time.Now().UTC().Add(-time.Hour) + + insertQ(t, db, 1, "failed", "tv", base, false) + insertQ(t, db, 2, "failed", "tv", base.Add(time.Minute), false) + insertQ(t, db, 3, "failed", "tv", base.Add(2*time.Minute), true) // skip_arr_notification: never shown + insertQ(t, db, 4, "failed", "movies", base.Add(3*time.Minute), false) + insertQ(t, db, 5, "completed", "tv", base.Add(4*time.Minute), false) + insertH(t, db, 6, 0, "tv", base.Add(5*time.Minute)) + + rows, err := repo.ListSABnzbdFailedHistory(ctx, "tv", 10) + require.NoError(t, err) + var ids []int64 + for _, r := range rows { + assert.Equal(t, "failed_queue", r.Source) + ids = append(ids, r.ID) + } + assert.Equal(t, []int64{2, 1}, ids, "newest first, tv only, no skip_arr rows") + + rows, err = repo.ListSABnzbdFailedHistory(ctx, "tv", 1) + require.NoError(t, err) + require.Len(t, rows, 1) + assert.Equal(t, int64(2), rows[0].ID) + + rows, err = repo.ListSABnzbdFailedHistory(ctx, "", 10) + require.NoError(t, err) + assert.Len(t, rows, 3, "empty category means all categories") +} + func TestSABnzbdHistory_LiveQueueOwnsProvisionalHistory(t *testing.T) { for _, status := range []string{"failed", "pending", "processing", "paused", "completed"} { for _, skip := range []bool{false, true} { diff --git a/internal/nzbfilesystem/readat_failure_test.go b/internal/nzbfilesystem/readat_failure_test.go index 727b3e9bd..ee094c4d1 100644 --- a/internal/nzbfilesystem/readat_failure_test.go +++ b/internal/nzbfilesystem/readat_failure_test.go @@ -5,12 +5,12 @@ import ( "testing" "time" + "github.com/javi11/nntppool/v5" "github.com/kipsilabs/altmount/internal/config" "github.com/kipsilabs/altmount/internal/database" metapb "github.com/kipsilabs/altmount/internal/metadata/proto" "github.com/kipsilabs/altmount/internal/testsupport/fakepool" "github.com/kipsilabs/altmount/internal/testsupport/segments" - "github.com/javi11/nntppool/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -130,12 +130,20 @@ func TestReadAtContext_StalledFetch_TimesOut(t *testing.T) { ctx := context.Background() fp := fakepool.New() configurePoolForFile(fp, segCount, segSize, fakepool.SegmentBehavior{}) - fp.BlockUntil(make(chan struct{})) // never released, never cancelled + gate := make(chan struct{}) + fp.BlockUntil(gate) + t.Cleanup(func() { close(gate) }) cfg := config.DefaultConfig() cfg.Streaming.ReadTimeoutSeconds = 1 mvf := withConfig(newTestMVF(t, ctx, fp, segCount, segSize, 2), cfg) + // Other readers share in-flight articles by message ID. Keep the stalled + // articles separate so a successful fetch cannot bypass this pool. + for i, seg := range mvf.meta.SegmentData { + seg.Id = t.Name() + "-" + seg.Id + fp.SetBehavior(seg.Id, fakepool.SegmentBehavior{Bytes: segments.Payload(i, segSize)}) + } type result struct { n int @@ -170,12 +178,20 @@ func TestReadAtContext_StalledEphemeralFetch_TimesOut(t *testing.T) { ctx := context.Background() fp := fakepool.New() configurePoolForFile(fp, segCount, segSize, fakepool.SegmentBehavior{}) - fp.BlockUntil(make(chan struct{})) // never released, never cancelled + gate := make(chan struct{}) + fp.BlockUntil(gate) + t.Cleanup(func() { close(gate) }) cfg := config.DefaultConfig() cfg.Streaming.ReadTimeoutSeconds = 1 mvf := withConfig(newTestMVF(t, ctx, fp, segCount, segSize, 2), cfg) + // Other readers share in-flight articles by message ID. Keep the stalled + // articles separate so a successful fetch cannot bypass this pool. + for i, seg := range mvf.meta.SegmentData { + seg.Id = t.Name() + "-" + seg.Id + fp.SetBehavior(seg.Id, fakepool.SegmentBehavior{Bytes: segments.Payload(i, segSize)}) + } done := make(chan error, 1) go func() {