116 lines
4.3 KiB
TypeScript
116 lines
4.3 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
|
import logger from '@adonisjs/core/services/logger'
|
|
import QueueService from '#services/QueueService'
|
|
|
|
/** A failed range is retried after this long... */
|
|
const REQUEUE_AFTER_MS = 5 * 60 * 1000
|
|
/** ...at most this many times; after that it stays visible as a gap. */
|
|
const MAX_REQUEUES = 5
|
|
/** A queued range without a job in Redis (e.g. Redis was flushed) is re-queued after this long. */
|
|
const LOST_AFTER_MS = 60 * 60 * 1000
|
|
|
|
interface Group {
|
|
id: number
|
|
name: string
|
|
}
|
|
|
|
/**
|
|
* Tracks the article ranges given to worker:fetch. A range only counts as
|
|
* indexed once the worker reports it done, so failed or lost fetch jobs are
|
|
* retried instead of silently skipped. See CONCEPT.md (D18).
|
|
*/
|
|
class FetchRangeService {
|
|
async queue(group: Group, startId: bigint, endId: bigint) {
|
|
const now = new Date()
|
|
const [row] = await db
|
|
.table('fetch_ranges')
|
|
.insert({ group_id: group.id, start_id: startId.toString(), end_id: endId.toString(), created_at: now, updated_at: now })
|
|
.returning(['id', 'requeues'])
|
|
await this.addJob(group.name, Number(row.id), 0, startId, endId)
|
|
}
|
|
|
|
private async addJob(groupName: string, rangeId: number, requeues: number, startId: bigint, endId: bigint) {
|
|
await QueueService.nntpFetchQueue.add(
|
|
'fetch-headers',
|
|
{ groupName, startId: startId.toString(), endId: endId.toString(), rangeId },
|
|
// The job id changes per attempt: BullMQ ignores an add() for an id it still knows.
|
|
{ jobId: this.jobId(rangeId, requeues) }
|
|
)
|
|
}
|
|
|
|
private jobId(rangeId: number, requeues: number) {
|
|
return `range-${rangeId}-${requeues}`
|
|
}
|
|
|
|
async markDone(rangeId: number) {
|
|
await db.from('fetch_ranges').where('id', rangeId).update({ status: 'done', error: null, updated_at: new Date() })
|
|
}
|
|
|
|
async markFailed(rangeId: number, error: string) {
|
|
await db.from('fetch_ranges').where('id', rangeId).update({ status: 'failed', error, updated_at: new Date() })
|
|
}
|
|
|
|
/** Re-queues failed ranges, and queued ranges whose job no longer exists. */
|
|
async requeue(group: Group): Promise<number> {
|
|
const now = Date.now()
|
|
const candidates = await db
|
|
.from('fetch_ranges')
|
|
.where('group_id', group.id)
|
|
.where('requeues', '<', MAX_REQUEUES)
|
|
.where((q) =>
|
|
q
|
|
.where((f) => f.where('status', 'failed').where('updated_at', '<', new Date(now - REQUEUE_AFTER_MS)))
|
|
.orWhere((l) => l.where('status', 'queued').where('updated_at', '<', new Date(now - LOST_AFTER_MS)))
|
|
)
|
|
.orderBy('start_id')
|
|
|
|
let requeued = 0
|
|
for (const range of candidates) {
|
|
if (range.status === 'queued') {
|
|
const job = await QueueService.nntpFetchQueue.getJob(this.jobId(Number(range.id), range.requeues))
|
|
const state = job ? await job.getState() : 'missing'
|
|
if (state !== 'missing' && state !== 'failed' && state !== 'unknown') continue
|
|
}
|
|
await db
|
|
.from('fetch_ranges')
|
|
.where('id', range.id)
|
|
.update({ status: 'queued', requeues: range.requeues + 1, updated_at: new Date() })
|
|
await this.addJob(group.name, Number(range.id), range.requeues + 1, BigInt(range.start_id), BigInt(range.end_id))
|
|
requeued++
|
|
logger.warn(
|
|
`Re-queued ${group.name} ${range.start_id}-${range.end_id} (attempt ${range.requeues + 2}): ${range.error ?? 'job lost'}`
|
|
)
|
|
}
|
|
return requeued
|
|
}
|
|
|
|
/**
|
|
* Moves `groups.last_indexed_id` up to just below the first range that is
|
|
* not done yet, and forgets done ranges. Returns the new watermark.
|
|
*/
|
|
async advanceWatermark(group: Group): Promise<bigint> {
|
|
const { rows } = await db.rawQuery(
|
|
`UPDATE groups g SET last_indexed_id = greatest(g.last_indexed_id, coalesce(
|
|
(SELECT min(start_id) - 1 FROM fetch_ranges WHERE group_id = g.id AND status <> 'done'),
|
|
g.last_queued_id))
|
|
WHERE g.id = ? RETURNING last_indexed_id`,
|
|
[group.id]
|
|
)
|
|
await db.from('fetch_ranges').where('group_id', group.id).where('status', 'done').delete()
|
|
return BigInt(rows[0].last_indexed_id)
|
|
}
|
|
|
|
/** Ranges that failed for good (for monitoring). */
|
|
async abandoned(group: Group) {
|
|
return db
|
|
.from('fetch_ranges')
|
|
.where('group_id', group.id)
|
|
.where('status', 'failed')
|
|
.where('requeues', '>=', MAX_REQUEUES)
|
|
.count('* as count')
|
|
.first()
|
|
}
|
|
}
|
|
|
|
export default new FetchRangeService()
|