Daan MeijerandClaude Opus 5.5
f0ff8e200f
Copying all existing keys in the migration read the files rows from the HDD for over half an hour and held up the deploy. gc:scheduler now copies them in slices (FILE_KEYS_FILL_SECONDS per sweep) and FileStore checks every partition until the fill is complete. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
385 lines
15 KiB
TypeScript
385 lines
15 KiB
TypeScript
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
|
import { CommandOptions } from '@adonisjs/core/types/ace'
|
|
import type { Queue } from 'bullmq'
|
|
import db from '@adonisjs/lucid/services/db'
|
|
import gcConfig from '#config/gc'
|
|
import QueueService from '#services/QueueService'
|
|
import RedisService from '#services/RedisService'
|
|
import IndexMaintenance from '#services/IndexMaintenance'
|
|
import ReleaseMergeService from '#services/ReleaseMergeService'
|
|
import StagingService from '#services/StagingService'
|
|
import FilePartitions from '#services/FilePartitions'
|
|
import ReleaseCompleteness from '#services/ReleaseCompleteness'
|
|
import FileKeys from '#services/FileKeys'
|
|
import Shutdown from '#services/Shutdown'
|
|
|
|
const CLEAN_BATCH = 1000
|
|
const SCAN_COUNT = 1000
|
|
/** Binaries per statement when dropping copied parts (a few hundred parts each at most). */
|
|
const COPIED_PARTS_BATCH = 2000
|
|
|
|
export default class GcScheduler extends BaseCommand {
|
|
public static commandName = 'gc:scheduler'
|
|
public static description = 'Periodically removes old BullMQ jobs and stale binaries.'
|
|
|
|
public static options: CommandOptions = {
|
|
startApp: true,
|
|
}
|
|
|
|
@flags.boolean({ description: 'Run one sweep and exit' })
|
|
declare once: boolean
|
|
|
|
@flags.boolean({ description: 'Report what would be removed without changing anything' })
|
|
declare dryRun: boolean
|
|
|
|
/** When release completeness was last recomputed (it runs less often than the sweep). */
|
|
private completenessAt = 0
|
|
|
|
public async run() {
|
|
this.logger.info(
|
|
`GC started (interval ${gcConfig.intervalMs / 1000}s${this.dryRun ? ', dry run' : ''}).`
|
|
)
|
|
|
|
try {
|
|
do {
|
|
try {
|
|
await this.sweep()
|
|
} catch (err: any) {
|
|
this.logger.error(`GC sweep failed: ${err.message}`)
|
|
}
|
|
try {
|
|
await this.releaseCompleteness()
|
|
} catch (err: any) {
|
|
this.logger.error(`Release completeness failed: ${err.message}`)
|
|
}
|
|
try {
|
|
await this.fillFileKeys()
|
|
} catch (err: any) {
|
|
this.logger.error(`file_keys fill failed: ${err.message}`)
|
|
}
|
|
if (!this.once) await Shutdown.sleep(gcConfig.intervalMs)
|
|
} while (!this.once && !Shutdown.requested)
|
|
} finally {
|
|
await QueueService.closeAll()
|
|
await RedisService.close()
|
|
}
|
|
}
|
|
|
|
private async sweep() {
|
|
const { completedAgeSeconds, failedAgeSeconds } = gcConfig.jobs
|
|
for (const queue of [...QueueService.all, ...QueueService.legacy]) {
|
|
const completed = await this.cleanJobs(queue, 'completed', completedAgeSeconds)
|
|
const failed = await this.cleanJobs(queue, 'failed', failedAgeSeconds)
|
|
if (completed || failed) {
|
|
this.logger.info(
|
|
`${queue.name}: ${this.verb('remove', 'removed')} ${completed} completed and ${failed} failed jobs.`
|
|
)
|
|
}
|
|
}
|
|
|
|
// Inserts fail without a partition for their id (D39).
|
|
if (!this.dryRun) await FilePartitions.ensure()
|
|
|
|
const trimmed = await this.trimBodyQueue()
|
|
if (trimmed) this.logger.info(`Body queue: ${this.verb('drop', 'dropped')} ${trimmed} oldest waiting jobs.`)
|
|
|
|
const binaries = await this.deleteStaleBinaries()
|
|
this.logger.info(`Binaries: ${this.verb('delete', 'deleted')} ${binaries} stale.`)
|
|
|
|
if (!this.dryRun) {
|
|
// Each kind gets its own time budget; empty releases last, as deleting binaries empties them.
|
|
const budget = () => Date.now() + gcConfig.deleteBudgetMs
|
|
const parts = await this.dropCopiedParts(budget())
|
|
const single = await this.deleteSingleArticleBinaries(budget())
|
|
const unfingerprinted = await this.deleteUnfingerprintedBinaries(budget())
|
|
const releases = await this.deleteEmptyReleases(budget())
|
|
const staging = await StagingService.sweep(gcConfig.stagingTtlSeconds, budget())
|
|
if (staging.adopted || staging.expired) {
|
|
this.logger.info(`Staging: ${staging.adopted} segments adopted by their file, ${staging.expired} expired.`)
|
|
}
|
|
if (parts || single || unfingerprinted || releases) {
|
|
this.logger.info(
|
|
`Space: ${parts} copied parts, ${single} single-article binaries, ${unfingerprinted} unfingerprinted completed binaries, ` +
|
|
`${releases} empty releases deleted.`
|
|
)
|
|
}
|
|
}
|
|
|
|
const deferred = await this.deleteOldDeferredFiles()
|
|
if (deferred) this.logger.info(`Deferred files: ${this.verb('delete', 'deleted')} ${deferred} past retention.`)
|
|
|
|
const { expired, scanned } = await this.expireLegacyPendingFiles()
|
|
if (scanned > 0) {
|
|
this.logger.info(
|
|
`Legacy pending files: scanned ${scanned}, ${this.verb('set a TTL on', 'set a TTL on')} ${expired} without expiry.`
|
|
)
|
|
}
|
|
|
|
// Uploads split per poster (CONCEPT.md D35); the releases this empties go in a later sweep.
|
|
const merged = await ReleaseMergeService.mergeSplitPosts({ dryRun: this.dryRun })
|
|
if (merged.releases) this.logger.info(`Split uploads: ${this.verb('merge', 'merged')} ${merged.releases} releases into ${merged.names}.`)
|
|
|
|
await this.rebuildIndexes()
|
|
}
|
|
|
|
/** Rebuilds bloated indexes, at most every gcConfig.reindexIntervalMs (CONCEPT.md D33). */
|
|
/** The keys of files from before file_keys, a slice of time per sweep, until done. */
|
|
private async fillFileKeys() {
|
|
if (this.dryRun || (await FileKeys.complete())) return
|
|
const started = Date.now()
|
|
const { read, complete, nextId, untilId } = await FileKeys.fill(gcConfig.fileKeysFillMs)
|
|
this.logger.info(
|
|
`file_keys: ${read.toLocaleString('en')} keys copied in ${Math.round((Date.now() - started) / 1000)} s, ` +
|
|
(complete ? 'complete: file inserts use the filter now.' : `at file ${nextId.toLocaleString('en')} of ${untilId.toLocaleString('en')}.`)
|
|
)
|
|
}
|
|
|
|
private async releaseCompleteness() {
|
|
if (this.dryRun || (!this.once && Date.now() - this.completenessAt < gcConfig.releaseCompleteIntervalMs)) return
|
|
this.completenessAt = Date.now()
|
|
const started = Date.now()
|
|
const checked = await ReleaseCompleteness.sweep()
|
|
if (checked > 0) this.logger.info(`Release completeness: ${checked} changed releases checked in ${Math.round((Date.now() - started) / 1000)} s.`)
|
|
}
|
|
|
|
private async rebuildIndexes() {
|
|
const key = 'gc:reindex:last'
|
|
const last = Number((await RedisService.client.get(key)) ?? 0)
|
|
if (Date.now() - last < gcConfig.reindexIntervalMs) return
|
|
const rebuilt = await IndexMaintenance.rebuildBloated({ dryRun: this.dryRun })
|
|
if (!this.dryRun) await RedisService.client.set(key, String(Date.now()))
|
|
const saved = rebuilt.reduce((sum, r) => sum + r.before - r.after, 0)
|
|
this.logger.info(`Indexes: ${this.verb('rebuild', 'rebuilt')} ${rebuilt.length}, ${Math.round(saved / 1024 / 1024)} MB freed.`)
|
|
}
|
|
|
|
/**
|
|
* Drops the oldest waiting body jobs above gcConfig.bodyQueueMax. BullMQ
|
|
* workers take jobs from the right end of the wait list, so the oldest are
|
|
* there.
|
|
*/
|
|
private async trimBodyQueue(): Promise<number> {
|
|
const redis = RedisService.client
|
|
const key = QueueService.bodyQueue.toKey('wait')
|
|
const excess = (await redis.llen(key)) - gcConfig.bodyQueueMax
|
|
if (excess <= 0 || this.dryRun) return Math.max(excess, 0)
|
|
let dropped = 0
|
|
while (dropped < excess) {
|
|
const ids: string[] = (await redis.rpop(key, Math.min(5000, excess - dropped))) ?? []
|
|
if (ids.length === 0) break
|
|
await redis.unlink(...ids.map((id) => QueueService.bodyQueue.toKey(id)))
|
|
dropped += ids.length
|
|
}
|
|
return dropped
|
|
}
|
|
|
|
/**
|
|
* Deletes binaries (and, by cascade, their parts) that received no segment
|
|
* for `staleAfterSeconds`: incomplete ones are given up on; completed ones
|
|
* already live on in `files`; conflicted ones will never become a file.
|
|
* Completed ones still waiting for worker:file are kept.
|
|
*/
|
|
private async deleteStaleBinaries(): Promise<number> {
|
|
const query = db
|
|
.from('binaries')
|
|
.where('last_seen_at', '<', new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000))
|
|
.where((q) =>
|
|
q.whereNull('completed_at').orWhereNotNull('file_id').orWhereNotNull('conflicted_at')
|
|
)
|
|
|
|
if (this.dryRun) {
|
|
const [{ count }] = await query.count('* as count')
|
|
return Number(count)
|
|
}
|
|
return Number(await query.delete())
|
|
}
|
|
|
|
private async deleteOldDeferredFiles(): Promise<number> {
|
|
const cutoff = new Date(Date.now() - gcConfig.deferredRetentionDays * 24 * 60 * 60 * 1000)
|
|
let total = 0
|
|
for (const [table, column] of [['deferred_files', 'first_seen_at'], ['deferred_ranges', 'created_at']]) {
|
|
const query = db.from(table).where(column, '<', cutoff)
|
|
if (this.dryRun) {
|
|
const [{ count }] = await query.count('* as count')
|
|
total += Number(count)
|
|
} else {
|
|
total += Number(await query.delete())
|
|
}
|
|
}
|
|
return total
|
|
}
|
|
|
|
// Keyset cursors over the tables, kept between sweeps so one pass over
|
|
// millions of rows is spread over several sweeps.
|
|
private singleCursor = 0
|
|
private releaseCursor = 0
|
|
|
|
/**
|
|
* Header parts of binaries that are a file already. worker:file drops them
|
|
* when it creates the file, and ingest no longer adds them afterwards; this
|
|
* catches what slipped through. A cursor over binaries (as for single
|
|
* articles) visits each one once per pass, instead of searching all of
|
|
* them for parts on every batch (LOG.md 2026-10-04).
|
|
*/
|
|
private copiedCursor = 0
|
|
private async dropCopiedParts(budget: number): Promise<number> {
|
|
let total = 0
|
|
while (Date.now() < budget) {
|
|
const { rows } = await db.rawQuery(
|
|
`WITH batch AS (
|
|
SELECT id FROM binaries WHERE id > ? AND file_id IS NOT NULL ORDER BY id LIMIT ?
|
|
), gone AS (
|
|
DELETE FROM parts p USING batch WHERE p.binary_id = batch.id AND p.pcrc32 IS NULL
|
|
RETURNING 1)
|
|
SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
|
|
[this.copiedCursor, COPIED_PARTS_BATCH]
|
|
)
|
|
total += Number(rows[0].deleted)
|
|
if (rows[0].last === null) {
|
|
this.copiedCursor = 0
|
|
break
|
|
}
|
|
this.copiedCursor = Number(rows[0].last)
|
|
}
|
|
return total
|
|
}
|
|
|
|
/** Header-derived binaries that never got a second article (gcConfig.binaries.singleArticleAfterSeconds). */
|
|
private async deleteSingleArticleBinaries(budget: number): Promise<number> {
|
|
const cutoff = new Date(Date.now() - gcConfig.binaries.singleArticleAfterSeconds * 1000)
|
|
let total = 0
|
|
while (Date.now() < budget) {
|
|
const { rows } = await db.rawQuery(
|
|
`WITH batch AS (
|
|
SELECT id FROM binaries WHERE id > ? ORDER BY id LIMIT ?
|
|
), gone AS (
|
|
DELETE FROM binaries b USING batch
|
|
WHERE b.id = batch.id AND b.completed_at IS NULL AND b.size IS NULL AND b.total_parts > 1
|
|
AND b.last_seen_at < ? AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) <= 1
|
|
RETURNING 1)
|
|
SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
|
|
[this.singleCursor, gcConfig.deleteBatch, cutoff]
|
|
)
|
|
total += Number(rows[0].deleted)
|
|
if (rows[0].last === null) {
|
|
this.singleCursor = 0
|
|
break
|
|
}
|
|
this.singleCursor = Number(rows[0].last)
|
|
}
|
|
return total
|
|
}
|
|
|
|
/**
|
|
* Completed header-derived binaries without a fingerprint, some hours after
|
|
* completion: their file exists, their parts are gone, and par2 naming needs
|
|
* a size and MD5 they don't have. Oldest first, via the last_seen_at index.
|
|
*/
|
|
private async deleteUnfingerprintedBinaries(budget: number): Promise<number> {
|
|
const cutoff = new Date(Date.now() - gcConfig.binaries.unfingerprintedAfterSeconds * 1000)
|
|
let total = 0
|
|
while (Date.now() < budget) {
|
|
const result = await db.rawQuery(
|
|
`DELETE FROM binaries WHERE id IN (
|
|
SELECT id FROM binaries
|
|
WHERE last_seen_at < ? AND file_id IS NOT NULL AND completed_at IS NOT NULL
|
|
AND size IS NULL AND md5_16k IS NULL AND conflicted_at IS NULL
|
|
ORDER BY last_seen_at LIMIT ?)`,
|
|
[cutoff, gcConfig.deleteBatch]
|
|
)
|
|
total += result.rowCount
|
|
if (result.rowCount < gcConfig.deleteBatch) break
|
|
}
|
|
return total
|
|
}
|
|
|
|
/** Releases nothing refers to any more: no files, binaries, par2 listing or spot. */
|
|
private async deleteEmptyReleases(budget: number): Promise<number> {
|
|
const cutoff = new Date(Date.now() - gcConfig.emptyReleaseAfterSeconds * 1000)
|
|
let total = 0
|
|
while (Date.now() < budget) {
|
|
const { rows } = await db.rawQuery(
|
|
`WITH batch AS (
|
|
SELECT id FROM releases WHERE id > ? ORDER BY id LIMIT ?
|
|
), gone AS (
|
|
DELETE FROM releases r USING batch
|
|
WHERE r.id = batch.id AND r.file_count = 0 AND r.first_seen_at < ?
|
|
AND NOT EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id)
|
|
AND NOT EXISTS (SELECT 1 FROM files f WHERE f.release_id = r.id)
|
|
AND NOT EXISTS (SELECT 1 FROM release_files rf WHERE rf.release_id = r.id)
|
|
AND NOT EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id)
|
|
RETURNING 1)
|
|
SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
|
|
[this.releaseCursor, gcConfig.deleteBatch, cutoff]
|
|
)
|
|
total += Number(rows[0].deleted)
|
|
if (rows[0].last === null) {
|
|
this.releaseCursor = 0
|
|
break
|
|
}
|
|
this.releaseCursor = Number(rows[0].last)
|
|
}
|
|
return total
|
|
}
|
|
|
|
/** Removes jobs in `state` that finished more than `ageSeconds` ago. */
|
|
private async cleanJobs(queue: Queue, state: 'completed' | 'failed', ageSeconds: number) {
|
|
const graceMs = ageSeconds * 1000
|
|
if (this.dryRun) {
|
|
const cutoff = Date.now() - graceMs
|
|
// Finished-job sets are sorted by finishedOn, so a score range counts them cheaply.
|
|
return RedisService.client.zcount(queue.toKey(state), '-inf', cutoff)
|
|
}
|
|
|
|
let total = 0
|
|
let removed: string[]
|
|
do {
|
|
removed = await queue.clean(graceMs, CLEAN_BATCH, state)
|
|
total += removed.length
|
|
} while (removed.length === CLEAN_BATCH)
|
|
return total
|
|
}
|
|
|
|
/**
|
|
* Pending-file hashes of the old header worker (and the pre-AdonisJS app)
|
|
* never expire. Give them a TTL rather than deleting them outright.
|
|
*/
|
|
private async expireLegacyPendingFiles() {
|
|
const redis = RedisService.client
|
|
const { prefix, ttlSeconds } = gcConfig.legacyPendingFiles
|
|
let cursor = '0'
|
|
let scanned = 0
|
|
let expired = 0
|
|
|
|
do {
|
|
const [next, keys]: [string, string[]] = await redis.scan(
|
|
cursor,
|
|
'MATCH',
|
|
`${prefix}*`,
|
|
'COUNT',
|
|
SCAN_COUNT
|
|
)
|
|
cursor = next
|
|
if (keys.length === 0) continue
|
|
scanned += keys.length
|
|
|
|
const ttls: [Error | null, number][] = await keys
|
|
.reduce((pipeline, key) => pipeline.ttl(key), redis.pipeline())
|
|
.exec()
|
|
const orphaned = keys.filter((_, i) => ttls[i][1] === -1)
|
|
expired += orphaned.length
|
|
|
|
if (!this.dryRun && orphaned.length > 0) {
|
|
await orphaned
|
|
.reduce((pipeline, key) => pipeline.expire(key, ttlSeconds, 'NX'), redis.pipeline())
|
|
.exec()
|
|
}
|
|
} while (cursor !== '0')
|
|
|
|
return { scanned, expired }
|
|
}
|
|
|
|
private verb(present: string, past: string) {
|
|
return this.dryRun ? `would ${present}` : past
|
|
}
|
|
}
|