fix(sabnzbd): keep failed jobs on the first history page

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": { ... } }

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

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

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

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

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() {