fix: prevent retry backlog starvation

documentation/backend/services/scheduler.md

@@ -19,11 +19,11 @@ Manages recurring/scheduled jobs providing automated tasks (Plex scans, Audible
 1. **plex_library_scan** - Default: every 6 hours, full library scan, disabled by default (enable after setup)
 2. **plex_recently_added_check** - Default: every 5 minutes, lightweight polling of top 10 recently added items, enabled by default
 3. **audible_refresh** - Default: daily midnight, fetches 200 popular + 200 new releases, stores with rankings, disabled by default
-4. **retry_missing_torrents** - Default: daily midnight, processes union of `awaiting_search` ∪ `awaiting_release` (limit 50), handles both audiobook and ebook requests. Bidirectional transitions: `awaiting_search` → `awaiting_release` when release date is future + `indexer.skip_unreleased` ON; `awaiting_release` → `awaiting_search` + run search when release date has passed or setting OFF. Sole owner of these transitions. Enabled by default.
-5. **retry_failed_imports** - Default: every 6 hours, re-attempts 'awaiting_import' status (limit 50), enabled by default
+4. **retry_missing_torrents** - Default: daily midnight, processes union of `awaiting_search` ∪ eligible `awaiting_release` (limit 50), handles both audiobook and ebook requests. Fair ordering: never-searched first, then least-recently searched; future `awaiting_release` rows do not consume retry slots. Bidirectional transitions: `awaiting_search` → `awaiting_release` when release date is future + `indexer.skip_unreleased` ON; `awaiting_release` → `awaiting_search` + run search when release date has passed or setting OFF. Sole owner of these transitions. Enabled by default.
+5. **retry_failed_imports** - Default: every 6 hours, re-attempts `awaiting_import` status (limit 50), ordered by oldest `lastImportAt` so the backlog rotates fairly; enabled by default
 6. **find_missing_ebooks** - Default: daily midnight, scans `downloaded` ∪ `available` audiobook requests (limit 50) for missing ebook companions and triggers the existing ebook fetch flow (`addSearchEbookJob`). Gated by `ebook_auto_grab_enabled` AND at least one ebook source enabled (`ebook_annas_archive_enabled` or `ebook_indexer_search_enabled`; legacy `ebook_sidecar_enabled` accepted as Anna's fallback). Skips ebook children in-flight (`pending`, `awaiting_approval`, `searching`, `downloading`, `processing`, `awaiting_search`, `awaiting_release`) or `cancelled`. Retries `failed`/`warn` children up to **5 lifetime auto-retries** per audiobook, tracked in `Request.ebookAutoRetryCount` (nullable; processor-private — manual "Fetch Ebook" never reads/writes it). Per-candidate writes are wrapped in `prisma.$transaction` for race-safety with concurrent auto-grab; counter rolls back if `addSearchEbookJob` throws. Enabled by default. Returns `{ scanned, gapsFound, triggered, created, retried, skippedInFlight, skippedCancelled, skippedCapHit }`.
 7. **cleanup_seeded_torrents** - Default: every 30 mins, deletes torrents after seeding requirements met. Respects per-indexer `seedingTimeMinutes` AND `ratioLimit` (BOTH required when set; `0` disables that criterion; both `0` = never cleaned up). Undefined ratio with `ratioLimit > 0` = not met (safe-deny). Enabled by default.
-8. **monitor_rss_feeds** - Default: every 15 mins, checks RSS feeds from enabled indexers, matches against `awaiting_search` requests (audiobook and ebook, limit 100). Query is unchanged — release-date gate is applied AFTER a match is found: if matched book is unreleased + `indexer.skip_unreleased` ON, the match is skipped and request status is NOT mutated (retry job owns transitions). Enabled by default.
+8. **monitor_rss_feeds** - Default: every 15 mins, checks RSS feeds from enabled indexers, matches against all `awaiting_search` requests (audiobook and ebook) in deterministic pages of 100. Release-date gate is applied AFTER a match is found: if matched book is unreleased + `indexer.skip_unreleased` ON, the match is skipped and request status is NOT mutated (retry job owns transitions). Enabled by default.
 
 ## Architecture: Bull + Cron
 

src/lib/processors/monitor-rss-feeds.processor.ts

@@ -17,6 +17,8 @@ export interface MonitorRssFeedsPayload {
   scheduledJobId?: string;
 }
 
+const REQUEST_PAGE_SIZE = 100;
+
 export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): Promise<any> {
   const { jobId, scheduledJobId } = payload;
   const logger = RMABLogger.forJob(jobId, 'MonitorRssFeeds');
@@ -63,16 +65,35 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
     return { success: true, message: 'No RSS results', matched: 0 };
   }
 
-  // Get all active requests awaiting search (audiobooks and ebooks)
-  // Both types can be matched against RSS torrent feeds
-  const missingRequests = await prisma.request.findMany({
+  // Get every active request awaiting search (audiobooks and ebooks) in
+  // deterministic pages. A single unordered take(100) repeatedly inspected
+  // the same rows and made requests beyond that batch invisible to RSS.
+  const firstPage = await prisma.request.findMany({
     where: {
       status: 'awaiting_search',
       deletedAt: null,
     },
     include: { audiobook: true },
-    take: 100,
+    orderBy: { id: 'asc' },
+    take: REQUEST_PAGE_SIZE,
   });
+  const missingRequests = [...firstPage];
+  let page = firstPage;
+
+  while (page.length === REQUEST_PAGE_SIZE) {
+    page = await prisma.request.findMany({
+      where: {
+        status: 'awaiting_search',
+        deletedAt: null,
+      },
+      include: { audiobook: true },
+      orderBy: { id: 'asc' },
+      cursor: { id: page[page.length - 1].id },
+      skip: 1,
+      take: REQUEST_PAGE_SIZE,
+    });
+    missingRequests.push(...page);
+  }
 
   logger.info(`Found ${missingRequests.length} requests awaiting search`);
 

src/lib/processors/retry-failed-imports.processor.ts

@@ -61,6 +61,11 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
           take: 1,
         },
       },
+      orderBy: [
+        { lastImportAt: { sort: 'asc', nulls: 'first' } },
+        { createdAt: 'asc' },
+        { id: 'asc' },
+      ],
       take: 50, // Limit to 50 requests per run
     });
 
@@ -81,6 +86,13 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
 
     for (const request of requests) {
       try {
+        // Claim this retry slot immediately so every selected row rotates to
+        // the back of the next batch, including malformed rows we must skip.
+        await prisma.request.update({
+          where: { id: request.id },
+          data: { lastImportAt: new Date() },
+        });
+
         // Get the download path from the most recent download history
         const downloadHistory = request.downloadHistory[0];
 

src/lib/processors/retry-missing-torrents.processor.ts

@@ -30,15 +30,36 @@ export async function processRetryMissingTorrents(payload: RetryMissingTorrentsP
     const configService = getConfigService();
     const skipUnreleasedSetting = (await configService.get('indexer.skip_unreleased')) !== 'false';
 
-    // Find all active requests in awaiting_search OR awaiting_release status
+    // Release dates are stored as date-only values. When the release gate is
+    // enabled, do not let still-future awaiting_release rows consume one of
+    // the limited retry slots; they become eligible automatically on release day.
+    const todayUtc = new Date();
+    todayUtc.setUTCHours(0, 0, 0, 0);
+
+    // Process never-searched requests first, then rotate through the least
+    // recently searched. Without an explicit order, PostgreSQL repeatedly
+    // returned the same physical first 50 rows and starved the rest forever.
     const requests = await prisma.request.findMany({
       where: {
-        status: { in: ['awaiting_search', 'awaiting_release'] },
         deletedAt: null,
+        ...(skipUnreleasedSetting
+          ? {
+              OR: [
+                { status: 'awaiting_search' },
+                { status: 'awaiting_release', releaseDate: null },
+                { status: 'awaiting_release', releaseDate: { lte: todayUtc } },
+              ],
+            }
+          : { status: { in: ['awaiting_search', 'awaiting_release'] } }),
       },
       include: {
         audiobook: true,
       },
+      orderBy: [
+        { lastSearchAt: { sort: 'asc', nulls: 'first' } },
+        { createdAt: 'asc' },
+        { id: 'asc' },
+      ],
       take: 50,
     });
 

tests/processors/monitor-rss-feeds.processor.test.ts

@@ -34,6 +34,7 @@ function futureDate(days = 30): Date {
 
 describe('processMonitorRssFeeds', () => {
   beforeEach(() => {
+    vi.useRealTimers();
     vi.clearAllMocks();
     // Default to empty blocklist so the filter is a no-op unless a test overrides.
     prismaMock.blockedRelease.findMany.mockResolvedValue([]);
@@ -73,6 +74,70 @@ describe('processMonitorRssFeeds', () => {
       'req-1',
       expect.objectContaining({ title: 'Great Book', author: 'Author Name' })
     );
+    expect(prismaMock.request.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      orderBy: { id: 'asc' },
+      take: 100,
+    }));
+  });
+
+  it('pages beyond the first 100 awaiting requests', async () => {
+    vi.useFakeTimers();
+    configMock.get.mockImplementation(async (key: string) => {
+      if (key === 'prowlarr_indexers') {
+        return JSON.stringify([{ id: 1, name: 'Indexer', rssEnabled: true }]);
+      }
+      return null;
+    });
+    prowlarrMock.getAllRssFeeds.mockResolvedValue([
+      { title: 'Target Story - Target Author' },
+    ]);
+
+    const firstPage = Array.from({ length: 100 }, (_, index) => ({
+      id: `req-${String(index).padStart(3, '0')}`,
+      type: 'audiobook',
+      status: 'awaiting_search',
+      releaseDate: null,
+      audiobook: {
+        id: `a-${index}`,
+        title: `Unrelated Volume ${index}`,
+        author: 'Someone Else',
+        audibleAsin: `ASIN-${index}`,
+      },
+    }));
+    const secondPage = [{
+      id: 'req-target',
+      type: 'audiobook',
+      status: 'awaiting_search',
+      releaseDate: null,
+      audiobook: {
+        id: 'a-target',
+        title: 'Target Story',
+        author: 'Target Author',
+        audibleAsin: 'ASIN-TARGET',
+      },
+    }];
+    prismaMock.request.findMany
+      .mockResolvedValueOnce(firstPage)
+      .mockResolvedValueOnce(secondPage);
+
+    const { processMonitorRssFeeds } = await import('@/lib/processors/monitor-rss-feeds.processor');
+    const processing = processMonitorRssFeeds({ jobId: 'job-paged' });
+    await vi.runAllTimersAsync();
+    const result = await processing;
+
+    expect(result.totalMissing).toBe(101);
+    expect(result.matched).toBe(1);
+    expect(prismaMock.request.findMany).toHaveBeenNthCalledWith(2, expect.objectContaining({
+      orderBy: { id: 'asc' },
+      cursor: { id: 'req-099' },
+      skip: 1,
+      take: 100,
+    }));
+    expect(jobQueueMock.addSearchJob).toHaveBeenCalledWith(
+      'req-target',
+      expect.objectContaining({ title: 'Target Story', author: 'Target Author' })
+    );
+    vi.useRealTimers();
   });
 
   it('skips RSS auto-search when matched book is unreleased and setting ON', async () => {

tests/processors/retry-failed-imports.processor.test.ts

@@ -77,6 +77,18 @@ describe('processRetryFailedImports', () => {
     const result = await processRetryFailedImports({ jobId: 'job-1' });
 
     expect(result.success).toBe(true);
+    expect(prismaMock.request.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      orderBy: [
+        { lastImportAt: { sort: 'asc', nulls: 'first' } },
+        { createdAt: 'asc' },
+        { id: 'asc' },
+      ],
+      take: 50,
+    }));
+    expect(prismaMock.request.update).toHaveBeenCalledWith({
+      where: { id: 'req-1' },
+      data: { lastImportAt: expect.any(Date) },
+    });
     expect(jobQueueMock.addOrganizeJob).toHaveBeenCalledWith(
       'req-1',
       'a1',
@@ -109,6 +121,10 @@ describe('processRetryFailedImports', () => {
 
     expect(result.skipped).toBe(1);
     expect(result.triggered).toBe(0);
+    expect(prismaMock.request.update).toHaveBeenCalledWith({
+      where: { id: 'req-2' },
+      data: { lastImportAt: expect.any(Date) },
+    });
   });
 
   it('falls back to configured download dir when qBittorrent lookup fails', async () => {

tests/processors/retry-missing-torrents.processor.test.ts

@@ -53,6 +53,23 @@ describe('processRetryMissingTorrents', () => {
     const result = await processRetryMissingTorrents({ jobId: 'job-1' });
 
     expect(result.success).toBe(true);
+    expect(prismaMock.request.findMany).toHaveBeenCalledWith({
+      where: {
+        deletedAt: null,
+        OR: [
+          { status: 'awaiting_search' },
+          { status: 'awaiting_release', releaseDate: null },
+          { status: 'awaiting_release', releaseDate: { lte: expect.any(Date) } },
+        ],
+      },
+      include: { audiobook: true },
+      orderBy: [
+        { lastSearchAt: { sort: 'asc', nulls: 'first' } },
+        { createdAt: 'asc' },
+        { id: 'asc' },
+      ],
+      take: 50,
+    });
     expect(jobQueueMock.addSearchJob).toHaveBeenCalledWith(
       'req-1',
       expect.objectContaining({ id: 'a1', title: 'Book', author: 'Author' })
@@ -155,5 +172,11 @@ describe('processRetryMissingTorrents', () => {
     expect(prismaMock.request.update).not.toHaveBeenCalled();
     expect(jobQueueMock.addSearchJob).toHaveBeenCalled();
     expect(result.triggered).toBe(1);
+    expect(prismaMock.request.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      where: {
+        deletedAt: null,
+        status: { in: ['awaiting_search', 'awaiting_release'] },
+      },
+    }));
   });
 });