Daan MeijerandClaude Opus 5.5
35d0b2da65
5.4M waiting inspection jobs took 3.3 GB of Redis and most of its single thread. Producers now add file ids to inspection_queue; index-scheduler feeds up to INSPECTION_FEED_MAX (50k) into BullMQ. inspection:from-redis moves the existing backlog over once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
57 lines
2.7 KiB
TypeScript
57 lines
2.7 KiB
TypeScript
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-<id>, content-<id>-<time>), so no job is loaded.
|
|
* index-scheduler then feeds a working set back in.
|
|
*/
|
|
export default class InspectionFromRedis extends BaseCommand {
|
|
public static commandName = 'inspection:from-redis'
|
|
public static description = 'Moves waiting inspection jobs from Redis into the inspection_queue table.'
|
|
|
|
public static options: CommandOptions = {
|
|
startApp: true,
|
|
}
|
|
|
|
@flags.boolean({ description: 'Only count, change nothing' })
|
|
declare dryRun: boolean
|
|
|
|
public async run() {
|
|
const queue = QueueService.collectionQueue
|
|
const client = await queue.client
|
|
const prefix = `${queue.opts.prefix ?? 'bull'}:${queue.name}`
|
|
try {
|
|
const idOf = (jobId: string) => {
|
|
const m = /^(?:file|content)-(\d+)/.exec(jobId)
|
|
return m ? Number(m[1]) : null
|
|
}
|
|
let moved = 0
|
|
let skipped = 0
|
|
const take = async (ids: string[], priority: number) => {
|
|
const fileIds = ids.map(idOf).filter((id): id is number => id !== null)
|
|
skipped += ids.length - fileIds.length
|
|
if (!this.dryRun) await InspectionQueue.add(fileIds, priority)
|
|
moved += fileIds.length
|
|
}
|
|
const waiting = await client.llen(`${prefix}:wait`)
|
|
for (let i = 0; i < waiting; i += 50_000) await take(await client.lrange(`${prefix}:wait`, i, i + 49_999), INSPECTION_PRIORITY.normal)
|
|
// Prioritized jobs were the low-priority niconico captures.
|
|
const prioritized = await client.zcard(`${prefix}:prioritized`)
|
|
for (let i = 0; i < prioritized; i += 50_000) await take(await client.zrange(`${prefix}:prioritized`, i, i + 49_999), INSPECTION_PRIORITY.low)
|
|
this.logger.info(`${this.dryRun ? 'Would move' : 'Moved'} ${moved} jobs into inspection_queue (${skipped} with other ids skipped).`)
|
|
if (this.dryRun) return
|
|
// Waiting and prioritized jobs only; active ones finish as they are.
|
|
await queue.drain()
|
|
const counts = await queue.getJobCounts('wait', 'prioritized', 'active')
|
|
this.logger.info(`BullMQ now: ${counts.wait} waiting, ${counts.prioritized} prioritized, ${counts.active} active; table: ${await InspectionQueue.size()}.`)
|
|
} finally {
|
|
await QueueService.closeAll()
|
|
}
|
|
}
|
|
}
|