98 lines
3.8 KiB
TypeScript
98 lines
3.8 KiB
TypeScript
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
|
import { CommandOptions } from '@adonisjs/core/types/ace'
|
|
import indexerConfig from '#config/indexer'
|
|
import QueueService from '#services/QueueService'
|
|
import RedisService from '#services/RedisService'
|
|
import DeferredService, { isDeferred } from '#services/DeferredService'
|
|
|
|
const BATCH = 2000
|
|
|
|
/** Rebuilds the wait list (KEYS[1]) without the ids in the set KEYS[2], in one atomic pass. */
|
|
const FILTER_LIST = `
|
|
local ids = redis.call('LRANGE', KEYS[1], 0, -1)
|
|
local keep = {}
|
|
for _, id in ipairs(ids) do
|
|
if redis.call('SISMEMBER', KEYS[2], id) == 0 then keep[#keep + 1] = id end
|
|
end
|
|
redis.call('DEL', KEYS[1])
|
|
for i = 1, #keep, 5000 do
|
|
redis.call('RPUSH', KEYS[1], unpack(keep, i, math.min(i + 4999, #keep)))
|
|
end
|
|
return #ids - #keep`
|
|
|
|
/**
|
|
* Removes waiting body jobs whose message-id is on the skip or defer list,
|
|
* e.g. after adding a pattern (CONCEPT.md D24, D28). Deferred ones are
|
|
* recorded in deferred_files first, so they can still be processed later.
|
|
*/
|
|
export default class BodyQueuePrune extends BaseCommand {
|
|
public static commandName = 'body-queue:prune'
|
|
public static description = 'Drops waiting body jobs that are now skipped or deferred.'
|
|
|
|
public static options: CommandOptions = {
|
|
startApp: true,
|
|
}
|
|
|
|
@flags.boolean({ description: 'Only count' })
|
|
declare dryRun: boolean
|
|
|
|
@flags.boolean({ description: 'Drop deferred jobs without recording them (already recorded by an earlier run)' })
|
|
declare skipRecord: boolean
|
|
|
|
public async run() {
|
|
const redis = RedisService.client
|
|
const queue = QueueService.bodyQueue
|
|
const waitKey = queue.toKey('wait')
|
|
let skipped = 0
|
|
let deferred = 0
|
|
let scanned = 0
|
|
const dropped: string[] = []
|
|
try {
|
|
const ids: string[] = await redis.lrange(waitKey, 0, -1)
|
|
for (let i = 0; i < ids.length; i += BATCH) {
|
|
const batch = ids.slice(i, i + BATCH)
|
|
const data: [Error | null, string | null][] = await batch
|
|
.reduce((pipeline, id) => pipeline.hget(queue.toKey(id), 'data'), redis.pipeline())
|
|
.exec()
|
|
const drop: string[] = []
|
|
const toDefer = new Map<string, { articleNumber: number; header: Record<string, string> }[]>()
|
|
batch.forEach((id, j) => {
|
|
const job = data[j][1] ? JSON.parse(data[j][1]!) : null
|
|
if (!job?.header) return
|
|
const messageId = job.header['message-id'] ?? ''
|
|
if (isDeferred(job.header)) {
|
|
const list = toDefer.get(job.group) ?? []
|
|
list.push({ articleNumber: Number(job.articleNumber), header: job.header })
|
|
toDefer.set(job.group, list)
|
|
drop.push(id)
|
|
deferred++
|
|
} else if (indexerConfig.skipBodyMessageIds.some((re) => re.test(messageId))) {
|
|
drop.push(id)
|
|
skipped++
|
|
}
|
|
})
|
|
scanned += batch.length
|
|
if (this.dryRun || drop.length === 0) continue
|
|
if (!this.skipRecord) for (const [group, articles] of toDefer) await DeferredService.record(group, articles)
|
|
dropped.push(...drop)
|
|
}
|
|
if (dropped.length > 0) {
|
|
const dropKey = queue.toKey('prune-ids')
|
|
await redis.del(dropKey)
|
|
for (let i = 0; i < dropped.length; i += BATCH) await redis.sadd(dropKey, ...dropped.slice(i, i + BATCH))
|
|
await redis.eval(FILTER_LIST, 2, waitKey, dropKey)
|
|
await redis.del(dropKey)
|
|
for (let i = 0; i < dropped.length; i += BATCH) {
|
|
await redis.unlink(...dropped.slice(i, i + BATCH).map((id) => queue.toKey(id)))
|
|
}
|
|
}
|
|
this.logger.info(
|
|
`Scanned ${scanned} waiting body jobs: ${this.dryRun ? 'would drop' : 'dropped'} ${deferred} deferred and ${skipped} skipped.`
|
|
)
|
|
} finally {
|
|
await QueueService.closeAll()
|
|
await RedisService.close()
|
|
}
|
|
}
|
|
}
|