From 084d4b1a9efd395380e24a05942a994608ed7f6e Mon Sep 17 00:00:00 2001 From: kikootwo Date: Tue, 18 Aug 2026 15:04:43 -0400 Subject: [PATCH] fix: prevent retry backlog starvation --- documentation/backend/services/scheduler.md | 6 +- .../processors/monitor-rss-feeds.processor.ts | 29 +++++++-- .../retry-failed-imports.processor.ts | 12 ++++ .../retry-missing-torrents.processor.ts | 25 ++++++- .../monitor-rss-feeds.processor.test.ts | 65 +++++++++++++++++++ .../retry-failed-imports.processor.test.ts | 16 +++++ .../retry-missing-torrents.processor.test.ts | 23 +++++++ 7 files changed, 167 insertions(+), 9 deletions(-) diff --git a/documentation/backend/services/scheduler.md b/documentation/backend/services/scheduler.md index c55ca17..5999d44 100644 --- a/documentation/backend/services/scheduler.md +++ b/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 diff --git a/src/lib/processors/monitor-rss-feeds.processor.ts b/src/lib/processors/monitor-rss-feeds.processor.ts index 1ae1e3c..53c4cb2 100644 --- a/src/lib/processors/monitor-rss-feeds.processor.ts +++ b/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 { 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`); diff --git a/src/lib/processors/retry-failed-imports.processor.ts b/src/lib/processors/retry-failed-imports.processor.ts index 3f03383..361e708 100644 --- a/src/lib/processors/retry-failed-imports.processor.ts +++ b/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]; diff --git a/src/lib/processors/retry-missing-torrents.processor.ts b/src/lib/processors/retry-missing-torrents.processor.ts index 3d5bcf2..b44b40c 100644 --- a/src/lib/processors/retry-missing-torrents.processor.ts +++ b/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, }); diff --git a/tests/processors/monitor-rss-feeds.processor.test.ts b/tests/processors/monitor-rss-feeds.processor.test.ts index 89b7c2b..bc41488 100644 --- a/tests/processors/monitor-rss-feeds.processor.test.ts +++ b/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 () => { diff --git a/tests/processors/retry-failed-imports.processor.test.ts b/tests/processors/retry-failed-imports.processor.test.ts index b203410..e0b93a8 100644 --- a/tests/processors/retry-failed-imports.processor.test.ts +++ b/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 () => { diff --git a/tests/processors/retry-missing-torrents.processor.test.ts b/tests/processors/retry-missing-torrents.processor.test.ts index b1b78c3..0353c42 100644 --- a/tests/processors/retry-missing-torrents.processor.test.ts +++ b/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'] }, + }, + })); }); });