diff --git a/app/services/BinaryFileService.ts b/app/services/BinaryFileService.ts index 13d4f15..e7700d6 100644 --- a/app/services/BinaryFileService.ts +++ b/app/services/BinaryFileService.ts @@ -1,8 +1,8 @@ import db from '@adonisjs/lucid/services/db' import FileStore from '#services/FileStore' import SfvService from '#services/SfvService' -import QueueService from '#services/QueueService' import ContentService from '#services/ContentService' +import InspectionQueue from '#services/InspectionQueue' export interface BinaryFilesResult { /** New files. */ @@ -134,8 +134,8 @@ class BinaryFileService { // By the posted name, as before: inspection skips later volumes and readable media names anyway. const jobs = added .filter((file) => ContentService.isCandidate(file.posted)) - .map((file) => ({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } })) - if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs) + .map((file) => file.id) + if (jobs.length > 0) await InspectionQueue.add(jobs) result.inspections = jobs.length return result } diff --git a/app/services/InspectionQueue.ts b/app/services/InspectionQueue.ts new file mode 100644 index 0000000..53f6203 --- /dev/null +++ b/app/services/InspectionQueue.ts @@ -0,0 +1,66 @@ +import db from '@adonisjs/lucid/services/db' +import env from '#start/env' +import QueueService from '#services/QueueService' + +/** Files per transaction when moving from the table to BullMQ. */ +const FEED_BATCH = 5000 + +/** Priorities: lower goes first. */ +export const INSPECTION_PRIORITY = { targeted: -1, normal: 0, low: 10 } as const + +/** + * The content inspection backlog (CONCEPT.md D21, D41; LOG.md 2026-10-07 (8)). + * Producers add file ids to the inspection_queue table; feed() moves them + * into BullMQ's collection-queue while fewer than INSPECTION_FEED_MAX jobs + * wait there, where worker:collection and inspecting nodes take them as + * before. Redis holds a working set instead of a backlog of days. + */ +class InspectionQueue { + readonly feedMax = env.get('INSPECTION_FEED_MAX', 50_000) + + /** Queues files for inspection; a file already queued keeps the higher (lower number) priority. */ + async add(fileIds: number[], priority: number = INSPECTION_PRIORITY.normal) { + const ids = [...new Set(fileIds)] + for (let i = 0; i < ids.length; i += 10_000) { + await db.rawQuery( + `INSERT INTO inspection_queue (file_id, priority) + SELECT id, ?::smallint FROM unnest(?::bigint[]) AS id + ON CONFLICT (file_id) DO UPDATE SET priority = least(inspection_queue.priority, excluded.priority)`, + [priority, ids.slice(i, i + 10_000)] + ) + } + } + + /** Tops BullMQ up to feedMax waiting jobs; returns how many were moved. */ + async feed(): Promise { + const queue = QueueService.collectionQueue + const counts = await queue.getJobCounts('wait', 'prioritized') + let room = this.feedMax - (counts.wait ?? 0) - (counts.prioritized ?? 0) + let moved = 0 + while (room > 0) { + const n = await db.transaction(async (trx) => { + const { rows } = await trx.rawQuery( + `SELECT file_id FROM inspection_queue ORDER BY priority, queued_at LIMIT ? FOR UPDATE SKIP LOCKED`, + [Math.min(room, FEED_BATCH)] + ) + if (rows.length === 0) return 0 + const ids = rows.map((row: { file_id: string }) => Number(row.file_id)) + // Added before the rows go: a failure leaves them queued (a job that went in anyway is deduplicated by its id). + await queue.addBulk(ids.map((id: number) => ({ name: 'process-collection', data: { fileId: id }, opts: { jobId: `file-${id}` } }))) + await trx.rawQuery(`DELETE FROM inspection_queue WHERE file_id = ANY(?::bigint[])`, [ids]) + return ids.length + }) + if (n === 0) break + moved += n + room -= n + } + return moved + } + + async size(): Promise { + const { rows } = await db.rawQuery(`SELECT count(*) AS n FROM inspection_queue`) + return Number(rows[0].n) + } +} + +export default new InspectionQueue() diff --git a/app/services/NzbImportService.ts b/app/services/NzbImportService.ts index d48bf6d..b81798d 100644 --- a/app/services/NzbImportService.ts +++ b/app/services/NzbImportService.ts @@ -7,9 +7,9 @@ import { parseNzb, type ParsedNzb } from '#lib/nzb/index' import { parseSubject } from '#lib/subject/index' import { fileStem } from '#lib/release/index' import { looksRandom } from '#services/ArchiveService' -import QueueService from '#services/QueueService' import FileStore from '#services/FileStore' import { isArchiveStart } from '#lib/content/index' +import InspectionQueue from '#services/InspectionQueue' export interface NzbImportResult { releaseId: number @@ -207,9 +207,7 @@ class NzbImportService { // Their first segment tells what the archives hold (contents, password); nothing else reads it for files from an NZB. const archives = added.filter((f) => isArchiveStart(f.filename)) if (archives.length) { - await QueueService.collectionQueue.addBulk( - archives.map((f) => ({ name: 'process-collection', data: { fileId: f.id }, opts: { jobId: `file-${f.id}` } })) - ) + await InspectionQueue.add(archives.map((f) => Number(f.id))) } return result } diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index 546e9fe..f8f49e0 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -3,8 +3,8 @@ import IngestService from '#services/IngestService' import FileStore, { insertLimits } from '#services/FileStore' import SfvService from '#services/SfvService' import ContentService from '#services/ContentService' -import QueueService from '#services/QueueService' import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire' +import InspectionQueue from '#services/InspectionQueue' /** Articles from a range's edge within which a file may continue in the neighbouring range. */ const EDGE_ARTICLES = 50_000 @@ -112,8 +112,8 @@ class SeedService { await SfvService.claimAll(added) const candidates = added .filter((file) => ContentService.isCandidate(file.name)) - .map((file) => ({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } })) - if (candidates.length > 0) await QueueService.collectionQueue.addBulk(candidates) + .map((file) => file.id) + if (candidates.length > 0) await InspectionQueue.add(candidates) result.inspections = candidates.length return result } diff --git a/commands/ContentBackfill.ts b/commands/ContentBackfill.ts index bc2df18..a904e5a 100644 --- a/commands/ContentBackfill.ts +++ b/commands/ContentBackfill.ts @@ -4,6 +4,7 @@ import db from '@adonisjs/lucid/services/db' import QueueService from '#services/QueueService' import ContentService from '#services/ContentService' import { isArchiveStart } from '#lib/content/index' +import InspectionQueue, { INSPECTION_PRIORITY } from '#services/InspectionQueue' const BATCH = 1000 @@ -71,19 +72,15 @@ export default class ContentBackfill extends BaseCommand { const jobs = files .filter((f) => (this.archives ? isArchiveStart(f.filename) : this.sfv || this.nicovideo || ContentService.isCandidate(f.filename))) .slice(0, this.limit ? this.limit - queued : undefined) - .map((f) => ({ - name: 'process-collection', - data: { fileId: Number(f.id) }, - // A new id per run: BullMQ ignores an add() for an id it still knows. - // Targeted backfills go ahead of whatever is waiting. - opts: { - jobId: `content-${f.id}-${Date.now()}`, - lifo: this.archiveNzbs || this.sevenZip || this.archives || this.sfv, - // niconico captures are many (~190k) and only cosmetic: behind everything else. - ...(this.nicovideo ? { priority: 10 } : {}), - }, - })) - if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs) + .map((f) => Number(f.id)) + // Into the inspection backlog (InspectionQueue): targeted backfills go ahead of what's waiting; + // niconico captures are many (~190k) and only cosmetic, so behind everything else. + const priority = this.nicovideo + ? INSPECTION_PRIORITY.low + : this.archiveNzbs || this.sevenZip || this.archives || this.sfv + ? INSPECTION_PRIORITY.targeted + : INSPECTION_PRIORITY.normal + if (jobs.length > 0) await InspectionQueue.add(jobs, priority) queued += jobs.length } this.logger.info(`Queued ${queued} files for content inspection.`) diff --git a/commands/IndexScheduler.ts b/commands/IndexScheduler.ts index a7d46a0..ea23fd0 100644 --- a/commands/IndexScheduler.ts +++ b/commands/IndexScheduler.ts @@ -5,6 +5,7 @@ import NntpService from '#services/NntpService' import QueueService from '#services/QueueService' import indexerConfig from '#config/indexer' import Shutdown from '#services/Shutdown' +import InspectionQueue from '#services/InspectionQueue' import FetchRangeService from '#services/FetchRangeService' import SplitService from '#services/SplitService' import BackfillService from '#services/BackfillService' @@ -108,6 +109,12 @@ export default class IndexScheduler extends BaseCommand { } } + // The inspection backlog lives in Postgres; BullMQ gets a working set (LOG.md 2026-10-07 (8)). + const feedInspection = async () => { + const moved = await InspectionQueue.feed() + if (moved > 0) this.logger.info(`Inspection: ${moved} files moved from the backlog to the queue.`) + } + const backfill = async () => { try { const queued = await BackfillService.tick() @@ -122,7 +129,7 @@ export default class IndexScheduler extends BaseCommand { // here once ended the loop while the process stayed up, and live indexing stopped for ~3 hours // (LOG.md 2026-10-07 (3)). while (!Shutdown.requested) { - for (const step of [schedule, backfill, split]) { + for (const step of [schedule, backfill, split, feedInspection]) { try { await step() } catch (err: any) { diff --git a/commands/InspectionFromRedis.ts b/commands/InspectionFromRedis.ts new file mode 100644 index 0000000..437eef6 --- /dev/null +++ b/commands/InspectionFromRedis.ts @@ -0,0 +1,56 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import QueueService from '#services/QueueService' +import InspectionQueue, { INSPECTION_PRIORITY } from '#services/InspectionQueue' + +/** + * One-off (LOG.md 2026-10-07 (8)): moves the inspection jobs waiting in + * BullMQ into the inspection_queue table, then empties BullMQ's waiting and + * prioritized jobs so Redis gives the memory back. The file id is read from + * the job id (file-, content--