Rebuild: fresh AdonisJS 7 API app on Postgres
The rebuild branch starts over (REBUILD.md): the API starter kit, with Postgres instead of SQLite (its own database, DB_SEARCH_PATH for a shared one in development) and the #lib alias. Phase 1's code is imported from the current app module by module. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed
1 parent
699395c51f
commit
96f0ade10e
264 files changed
+777
-24132
No files matched your search
@@ -1,70 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import QueueService from '#services/QueueService'
|
||||
import NntpService from '#services/NntpService'
|
||||
import ArchiveWalkService from '#services/ArchiveWalkService'
|
||||
|
||||
const BATCH = 5000
|
||||
|
||||
/**
|
||||
* Complete archive listings for releases inspected before archive walks
|
||||
* existed (CONCEPT.md D40): queues those whose listing leaves most of the
|
||||
* release unaccounted for. worker:collection walks them, one at a time and
|
||||
* behind inspection. With release ids, walks those right here instead.
|
||||
*/
|
||||
export default class ArchiveWalk extends BaseCommand {
|
||||
public static commandName = 'content:walk-archives'
|
||||
public static description = 'Queues releases with partial archive listings for a (low priority) header walk.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.spread({ description: 'Release ids to walk now (instead of queueing)', required: false })
|
||||
declare releaseIds?: string[]
|
||||
|
||||
@flags.number({ description: 'Queue at most this many releases' })
|
||||
declare limit?: number
|
||||
|
||||
@flags.boolean({ description: 'Also releases that were walked before' })
|
||||
declare again: boolean
|
||||
|
||||
public async run() {
|
||||
try {
|
||||
if (this.releaseIds?.length) {
|
||||
for (const id of this.releaseIds) {
|
||||
this.logger.info(JSON.stringify(await ArchiveWalkService.walk(Number(id))))
|
||||
}
|
||||
return
|
||||
}
|
||||
let lastId = 0
|
||||
let queued = 0
|
||||
while (!this.limit || queued < this.limit) {
|
||||
const query = db
|
||||
.from('releases as r')
|
||||
.where('r.id', '>', lastId)
|
||||
.whereNotNull('r.contents')
|
||||
.orderBy('r.id')
|
||||
.limit(BATCH)
|
||||
.select('r.id', 'r.size', 'r.contents_walked_at', db.raw(`(
|
||||
SELECT count(*) FROM jsonb_array_elements(r.contents)) AS n`), db.raw(`(
|
||||
SELECT coalesce(sum((e->>'size')::bigint), 0) FROM jsonb_array_elements(r.contents) e) AS listed`))
|
||||
const rows = await query
|
||||
if (rows.length === 0) break
|
||||
lastId = Number(rows.at(-1).id)
|
||||
const jobs = rows
|
||||
.filter((r) => this.again || !r.contents_walked_at)
|
||||
.filter((r) => ArchiveWalkService.worthWalking(Number(r.n), Number(r.listed), Number(r.size)))
|
||||
.slice(0, this.limit ? this.limit - queued : undefined)
|
||||
.map((r) => ({ name: 'walk', data: { releaseId: Number(r.id) }, opts: { jobId: `walk-${r.id}-${Date.now()}` } }))
|
||||
if (jobs.length > 0) await QueueService.archiveWalkQueue.addBulk(jobs)
|
||||
queued += jobs.length
|
||||
}
|
||||
this.logger.info(`Queued ${queued} releases for an archive walk.`)
|
||||
} finally {
|
||||
await QueueService.closeAll()
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,91 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
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
|
||||
|
||||
/**
|
||||
* Queues completed files for content inspection (names and groupings from
|
||||
* par2 and RAR headers, CONCEPT.md D21) that were completed before it
|
||||
* existed. Costs one article download per file, done by worker:collection.
|
||||
*/
|
||||
export default class ContentBackfill extends BaseCommand {
|
||||
public static commandName = 'content:backfill'
|
||||
public static description = 'Queues existing files with random names for content inspection.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'Queue at most this many files' })
|
||||
declare limit?: number
|
||||
|
||||
@flags.boolean({ description: 'Also files that were inspected before (after release:rebuild --reset-content)' })
|
||||
declare all: boolean
|
||||
|
||||
@flags.boolean({ description: 'Only archives whose listing contains NZBs, to import those (CONCEPT.md D28)' })
|
||||
declare archiveNzbs: boolean
|
||||
|
||||
@flags.boolean({ description: 'Only unlisted 7z archives (first volumes: .7z, .7z.001)' })
|
||||
declare sevenZip: boolean
|
||||
|
||||
@flags.boolean({ description: 'Archive starts never inspected (RAR first volumes, 7z, .bin), also files from NZBs/spots' })
|
||||
declare archives: boolean
|
||||
|
||||
@flags.boolean({ description: 'SFV files not read yet (their listings group the files they name, D34)' })
|
||||
declare sfv: boolean
|
||||
|
||||
@flags.boolean({ description: 'niconico captures not read yet (their releases get the video title, D34)' })
|
||||
declare nicovideo: boolean
|
||||
|
||||
public async run() {
|
||||
let lastId = 0
|
||||
let queued = 0
|
||||
try {
|
||||
while (!this.limit || queued < this.limit) {
|
||||
// Files never inspected: no fingerprint on their binary and no archive listing.
|
||||
const query = db.from('files as f').leftJoin('file_details as d', 'd.file_id', 'f.id').where('f.id', '>', lastId)
|
||||
// Files from NZBs (spots) have no binary.
|
||||
if (this.sevenZip || this.archives || this.sfv || this.nicovideo) query.leftJoin('binaries as b', 'b.file_id', 'f.id')
|
||||
else query.join('binaries as b', 'b.file_id', 'f.id')
|
||||
if (this.nicovideo) {
|
||||
query.whereNull('d.content_type').whereRaw(`f.filename ~* '^\\d{9,11}_nicovideo_jp_watch_sm\\d+\\.tar\\.zst$'`)
|
||||
} else if (this.sfv) {
|
||||
query.whereNull('d.content_type').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.sfv$'`)
|
||||
} else if (this.archives) {
|
||||
query.whereNull('d.content_type').whereNull('d.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.(rar|7z|7z\\.0*1|bin)$'`)
|
||||
} else if (this.sevenZip) {
|
||||
query.whereNull('d.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.7z(\\.0*1){0,1}$'`)
|
||||
} else if (this.archiveNzbs) query.whereRaw(`d.archive::text ~* '\\.nzb"'`)
|
||||
else if (!this.all) query.whereNull('b.md5_16k').whereNull('d.archive')
|
||||
const files = await query
|
||||
.orderBy('f.id')
|
||||
.limit(BATCH)
|
||||
.select('f.id', 'f.filename')
|
||||
if (files.length === 0) break
|
||||
lastId = Number(files.at(-1).id)
|
||||
|
||||
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) => 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.`)
|
||||
} finally {
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,22 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import IndexMaintenance from '#services/IndexMaintenance'
|
||||
|
||||
/** Rebuilds bloated indexes online (gc:scheduler also does this weekly). */
|
||||
export default class DbReindex extends BaseCommand {
|
||||
public static commandName = 'db:reindex'
|
||||
public static description = 'Rebuilds B-tree indexes that are mostly empty space (online, REINDEX CONCURRENTLY).'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({ description: 'Only show which indexes would be rebuilt' })
|
||||
declare dryRun: boolean
|
||||
|
||||
public async run() {
|
||||
const rebuilt = await IndexMaintenance.rebuildBloated({ dryRun: this.dryRun })
|
||||
const saved = rebuilt.reduce((sum, r) => sum + r.before - r.after, 0)
|
||||
this.logger.info(`${rebuilt.length} indexes ${this.dryRun ? 'would be rebuilt' : `rebuilt, ${Math.round(saved / 1024 / 1024)} MB freed`}.`)
|
||||
}
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import NntpService from '#services/NntpService'
|
||||
import QueueService from '#services/QueueService'
|
||||
import DeferredService from '#services/DeferredService'
|
||||
|
||||
/**
|
||||
* Sends recorded deferred ranges (fully obfuscated ngPost articles, CONCEPT.md
|
||||
* D28) to the body queue, oldest first. Every article in them is downloaded:
|
||||
* their headers can't be grouped into files.
|
||||
*/
|
||||
export default class DeferredExpandRanges extends BaseCommand {
|
||||
public static commandName = 'deferred:expand-ranges'
|
||||
public static description = 'Queues the bodies of deferred article ranges (costly: every article is downloaded).'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'Ranges to expand', default: 10 })
|
||||
declare limit: number
|
||||
|
||||
public async run() {
|
||||
try {
|
||||
const { ranges, articles } = await DeferredService.expandRanges(this.limit)
|
||||
this.logger.info(`Expanded ${ranges} ranges: ${articles} articles sent to the body queue.`)
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import { unquotedFileName } from '#services/NzbImportService'
|
||||
|
||||
/**
|
||||
* Files named after a whole NZB subject without quotes (spot NZBs), e.g.
|
||||
* `Adventure.part0055.rar - [056/175] (1/1)`: renamed to the filename in it.
|
||||
*/
|
||||
export default class FilesCleanNames extends BaseCommand {
|
||||
public static commandName = 'files:clean-names'
|
||||
public static description = 'Strips subject leftovers (counters, "yEnc") from file names that came from NZBs.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({ description: 'Only show what would change' })
|
||||
declare dryRun: boolean
|
||||
|
||||
public async run() {
|
||||
const files = await db
|
||||
.from('files')
|
||||
.whereRaw(`filename ~* '(\\(\\d+/\\d+\\)|yEnc|\\[\\d+/\\d+\\])\\s*$'`)
|
||||
.select('id', 'filename')
|
||||
const changes = files
|
||||
.map((f) => ({ id: Number(f.id), from: f.filename as string, to: unquotedFileName(f.filename) }))
|
||||
.filter((c) => c.to !== c.from)
|
||||
for (const c of changes.slice(0, 5)) this.logger.info(`"${c.from}" → "${c.to}"`)
|
||||
if (!this.dryRun) {
|
||||
for (const c of changes) await db.from('files').where('id', c.id).update({ filename: c.to })
|
||||
}
|
||||
this.logger.info(`${changes.length} of ${files.length} file names ${this.dryRun ? 'would change' : 'cleaned'}.`)
|
||||
}
|
||||
}
|
||||
@@ -1,36 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
|
||||
/**
|
||||
* One-off (LOG.md 2026-10-08 (14)): copies the content types and archive
|
||||
* listings still on `files` into file_details, a partition at a time.
|
||||
* Values written since the deploy (by the new code, into file_details) win:
|
||||
* a file that has details already is skipped.
|
||||
*/
|
||||
export default class FilesMoveDetails extends BaseCommand {
|
||||
public static commandName = 'files:move-details'
|
||||
public static description = "Copies files' content types and archive listings into file_details."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
const { rows: partitions } = await db.rawQuery(
|
||||
`SELECT c.relname FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid WHERE i.inhparent = 'files'::regclass ORDER BY c.relname`
|
||||
)
|
||||
let total = 0
|
||||
for (const { relname } of partitions) {
|
||||
const started = Date.now()
|
||||
const result = await db.rawQuery(
|
||||
`INSERT INTO file_details (file_id, content_type, archive)
|
||||
SELECT id, content_type, archive FROM ${relname} WHERE content_type IS NOT NULL OR archive IS NOT NULL
|
||||
ON CONFLICT (file_id) DO NOTHING`
|
||||
)
|
||||
total += result.rowCount ?? 0
|
||||
this.logger.info(`${relname}: ${result.rowCount} copied in ${Math.round((Date.now() - started) / 1000)} s.`)
|
||||
}
|
||||
this.logger.info(`${total} files' details copied.`)
|
||||
}
|
||||
}
|
||||
@@ -1,38 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import GarbageFilter from '#services/GarbageFilter'
|
||||
|
||||
/**
|
||||
* Learns the garbage upload signatures now (gc:scheduler does it every
|
||||
* GARBAGE_LEARN_INTERVAL_MS) and lists them (CONCEPT.md D42). --seed first
|
||||
* fills signature_probes from a sample of the files probed before it
|
||||
* existed, so there is something to learn from right away.
|
||||
*/
|
||||
export default class GarbageLearn extends BaseCommand {
|
||||
public static commandName = 'garbage:learn'
|
||||
public static description = 'Learns which upload signatures are garbage and lists them.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'First add probed releases from TABLESAMPLE SYSTEM (percent) of files' })
|
||||
declare seed?: number
|
||||
|
||||
public async run() {
|
||||
if (this.seed) {
|
||||
const started = Date.now()
|
||||
const added = await GarbageFilter.seed(this.seed)
|
||||
this.logger.info(`Seeded ${added} probed releases from ${this.seed}% of files in ${Math.round((Date.now() - started) / 1000)} s.`)
|
||||
}
|
||||
const { signatures, dropped, changed } = await GarbageFilter.learn()
|
||||
for (const line of changed) this.logger.info(`Changed: ${line}`)
|
||||
const rows = await db.from('upload_signatures').orderBy('probed', 'desc').limit(40)
|
||||
for (const row of rows) {
|
||||
const rate = row.probed ? ((100 * row.named) / row.probed).toFixed(1) : '-'
|
||||
this.logger.info(`${row.dropped ? 'DROP' : 'keep'} ${row.tool.padEnd(10)} ${(row.group_name || '(no group)').padEnd(36)} ${String(row.named).padStart(6)}/${String(row.probed).padEnd(6)} named (${rate}%)`)
|
||||
}
|
||||
this.logger.info(`${dropped} of ${signatures} signatures dropped.`)
|
||||
}
|
||||
}
|
||||
@@ -1,424 +0,0 @@
|
||||
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 GarbageFilter from '#services/GarbageFilter'
|
||||
import indexerConfig from '#config/indexer'
|
||||
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
|
||||
/** When the garbage signatures were last learned (D42). */
|
||||
private learnedAt = 0
|
||||
/** When split uploads were last merged. */
|
||||
private mergedAt = 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}`)
|
||||
}
|
||||
try {
|
||||
await this.learnGarbage()
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Garbage signatures 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). Four are kept ahead, so a failure here
|
||||
// can wait for the next sweep; it used to end the sweep, deletes and all (LOG.md 2026-10-07 (4)).
|
||||
if (!this.dryRun) {
|
||||
try {
|
||||
await FilePartitions.ensure()
|
||||
} catch (err: any) {
|
||||
this.logger.error(`files partitions: ${err.message}; trying again next sweep.`)
|
||||
}
|
||||
}
|
||||
|
||||
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. It reads
|
||||
// every release and its files: hourly, not every sweep (LOG.md 2026-10-06 (12)).
|
||||
if (this.once || Date.now() - this.mergedAt >= gcConfig.mergeIntervalMs) {
|
||||
this.mergedAt = Date.now()
|
||||
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.`)
|
||||
}
|
||||
|
||||
/** Learns which upload signatures are garbage from the probed releases (CONCEPT.md D42). */
|
||||
private async learnGarbage() {
|
||||
if (this.dryRun || (!this.once && Date.now() - this.learnedAt < indexerConfig.garbage.learnIntervalMs)) return
|
||||
this.learnedAt = Date.now()
|
||||
const { signatures, dropped, changed } = await GarbageFilter.learn()
|
||||
this.logger.info(`Garbage signatures: ${dropped} of ${signatures} dropped.`)
|
||||
for (const line of changed) this.logger.info(`Garbage signature ${line}`)
|
||||
}
|
||||
|
||||
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.
|
||||
*/
|
||||
/**
|
||||
* Binaries without new segments for staleAfterSeconds: incomplete ones are given up on,
|
||||
* finished ones have their file. A finished one whose file waits for content inspection
|
||||
* stays until it's inspected (its fingerprint names it from par2 sets, D16), at most
|
||||
* awaitInspectionSeconds (LOG.md 2026-10-06 (11)). Three deletes, one per partial index,
|
||||
* so none reads past millions of waiting rows.
|
||||
*/
|
||||
private async deleteStaleBinaries(): Promise<number> {
|
||||
const stale = new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000)
|
||||
const waited = new Date(Date.now() - gcConfig.binaries.awaitInspectionSeconds * 1000)
|
||||
const deletes = [
|
||||
db.from('binaries').where('last_seen_at', '<', stale).whereNotNull('file_id').whereRaw('awaiting_inspection IS NOT TRUE'),
|
||||
db.from('binaries').where('last_seen_at', '<', stale).where((q) => q.whereNull('completed_at').orWhereNotNull('conflicted_at')),
|
||||
db.from('binaries').where('last_seen_at', '<', waited).whereRaw('awaiting_inspection IS TRUE'),
|
||||
]
|
||||
let total = 0
|
||||
for (const query of deletes) {
|
||||
if (this.dryRun) total += Number((await query.count('* as count'))[0].count)
|
||||
else total += Number(await query.delete())
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
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
|
||||
AND awaiting_inspection IS NOT TRUE
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -1,79 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import Group from '#models/group'
|
||||
import NntpService from '#services/NntpService'
|
||||
import { parseGroupResponse } from '#lib/nntp/group'
|
||||
|
||||
/**
|
||||
* Adds newsgroups to index (or reactivates them). The scheduler picks them
|
||||
* up on its next tick, starting INDEX_INITIAL_BACKFILL articles back.
|
||||
*/
|
||||
export default class GroupAdd extends BaseCommand {
|
||||
public static commandName = 'group:add'
|
||||
public static description = 'Adds newsgroups to index.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.spread({ description: 'Group names, e.g. alt.binaries.hdtv' })
|
||||
declare names: string[]
|
||||
|
||||
@flags.boolean({ description: 'Deactivate the groups instead' })
|
||||
declare disable: boolean
|
||||
|
||||
@flags.boolean({
|
||||
description: 'Also download bodies to identify obfuscated posts (one body per article; bandwidth-heavy)',
|
||||
})
|
||||
declare bodies: boolean
|
||||
|
||||
@flags.number({ description: 'Start this many articles back instead of INDEX_INITIAL_BACKFILL (new groups only)' })
|
||||
declare backfill?: number
|
||||
|
||||
@flags.boolean({ description: 'With --backfill: also for groups indexed before, skipping what was missed since' })
|
||||
declare reset: boolean
|
||||
|
||||
@flags.boolean({ description: "Start at the server's oldest article: the group's full history (also for groups indexed before)" })
|
||||
declare fromStart: boolean
|
||||
|
||||
public async run() {
|
||||
for (const name of this.names) {
|
||||
if (!/^[a-z0-9][a-z0-9.+_-]*$/i.test(name)) {
|
||||
this.logger.error(`Not a valid group name: ${name}`)
|
||||
continue
|
||||
}
|
||||
const group = await Group.firstOrCreate({ name }, { name, active: !this.disable, fetchBodies: this.bodies })
|
||||
group.active = !this.disable
|
||||
if (this.bodies) group.fetchBodies = true
|
||||
if (this.fromStart) {
|
||||
// The scheduler continues after last_queued_id; 0 would mean "new group" to it.
|
||||
const { first } = await this.groupInfo(name)
|
||||
group.lastQueuedId = first > 1n ? first - 1n : 1n
|
||||
group.lastIndexedId = group.lastQueuedId
|
||||
} else if (this.backfill !== undefined && (this.reset || BigInt(group.lastQueuedId ?? 0) === 0n)) {
|
||||
const { last } = await this.groupInfo(name)
|
||||
group.lastQueuedId = (last > BigInt(this.backfill) ? last - BigInt(this.backfill) : 0n)
|
||||
group.lastIndexedId = group.lastQueuedId
|
||||
}
|
||||
await group.save()
|
||||
this.logger.info(
|
||||
`${name}: ${group.active ? 'active' : 'inactive'}, bodies ${group.fetchBodies ? 'on' : 'off'}` +
|
||||
(this.backfill !== undefined || this.fromStart ? `, starting after article ${group.lastQueuedId}` : '')
|
||||
)
|
||||
}
|
||||
if (this.backfill !== undefined || this.fromStart) await NntpService.shutdown()
|
||||
}
|
||||
|
||||
private async groupInfo(name: string) {
|
||||
const conn = await NntpService.acquire()
|
||||
let failed = false
|
||||
try {
|
||||
return parseGroupResponse((await conn.group(name)).resp)
|
||||
} catch (error) {
|
||||
failed = true
|
||||
throw error
|
||||
} finally {
|
||||
NntpService.release(conn, failed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import NntpService from '#services/NntpService'
|
||||
import BackfillService from '#services/BackfillService'
|
||||
import { isSpotGroup } from '#services/SpotService'
|
||||
|
||||
/**
|
||||
* Indexes a group's older headers (CONCEPT.md D32): from where live
|
||||
* indexing began back to N days ago, header-only and behind live indexing.
|
||||
* index:scheduler queues the ranges; this only sets where to go.
|
||||
*/
|
||||
export default class GroupBackfill extends BaseCommand {
|
||||
public static commandName = 'group:backfill'
|
||||
public static description = "Indexes groups' older headers (header-only, behind live indexing)."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.spread({ description: 'Groups (default: every active binary group)', required: false })
|
||||
declare names?: string[]
|
||||
|
||||
@flags.number({ description: 'How many days back', default: 21 })
|
||||
declare days: number
|
||||
|
||||
@flags.boolean({ description: 'Stop backfilling these groups (ranges already queued still run)' })
|
||||
declare stop: boolean
|
||||
|
||||
public async run() {
|
||||
const names = this.names?.length
|
||||
? this.names
|
||||
: (await db.from('groups').where('active', true).orderBy('name').select('name')).map((g) => g.name as string).filter((n) => !isSpotGroup(n))
|
||||
try {
|
||||
for (const name of names) {
|
||||
if (this.stop) {
|
||||
await BackfillService.stop(name)
|
||||
this.logger.info(`${name}: backfill stopped.`)
|
||||
continue
|
||||
}
|
||||
const range = await BackfillService.start(name, this.days)
|
||||
this.logger.info(
|
||||
range
|
||||
? `${name}: backfilling articles ${range.from}–${range.to} (${(range.to - range.from).toLocaleString('en')} articles, ${this.days} days).`
|
||||
: `${name}: nothing to backfill.`
|
||||
)
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,176 +0,0 @@
|
||||
import { writeFile } from 'node:fs/promises'
|
||||
import { setTimeout as sleep } from 'node:timers/promises'
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import NntpService from '#services/NntpService'
|
||||
import indexerConfig from '#config/indexer'
|
||||
import { parseGroupResponse } from '#lib/nntp/group'
|
||||
import { parseSubject } from '#lib/subject/index'
|
||||
|
||||
interface GroupRate {
|
||||
name: string
|
||||
perMinute: number
|
||||
sampled?: number
|
||||
normal?: number
|
||||
obfuscated?: number
|
||||
skipped?: number
|
||||
avgArticleBytes?: number
|
||||
avgHeaderBytes?: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists the busiest newsgroups (by articles posted per minute, from two
|
||||
* LIST ACTIVE snapshots) and optionally samples their newest headers to
|
||||
* estimate what indexing them would cost. See LOG.md (2026-09-28 (14)).
|
||||
*
|
||||
* node ace groups:top --minutes 5 --limit 1000 --sample 300 --out /data/top-groups.txt
|
||||
*/
|
||||
export default class GroupsTop extends BaseCommand {
|
||||
public static commandName = 'groups:top'
|
||||
public static description = 'Lists the busiest binary groups and estimates the bandwidth to index them.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.string({ description: 'Group pattern for LIST ACTIVE', default: 'alt.binaries.*' })
|
||||
declare pattern: string
|
||||
|
||||
@flags.number({ description: 'Minutes between the two snapshots', default: 5 })
|
||||
declare minutes: number
|
||||
|
||||
@flags.number({ description: 'Number of groups to list', default: 1000 })
|
||||
declare limit: number
|
||||
|
||||
@flags.number({ description: 'Headers to sample per listed group (0 = no sampling)', default: 0 })
|
||||
declare sample: number
|
||||
|
||||
@flags.string({ description: 'Write the list to this file (default: print only)' })
|
||||
declare out?: string
|
||||
|
||||
public async run() {
|
||||
const conn = await NntpService.acquire()
|
||||
let failed = false
|
||||
try {
|
||||
this.logger.info(`Snapshot 1 of ${this.pattern}…`)
|
||||
const first = await this.snapshot(conn)
|
||||
this.logger.info(`${first.size} groups. Waiting ${this.minutes} min for snapshot 2…`)
|
||||
const started = Date.now()
|
||||
await sleep(this.minutes * 60_000)
|
||||
const second = await this.snapshot(conn)
|
||||
const minutes = (Date.now() - started) / 60_000
|
||||
|
||||
const groups: GroupRate[] = []
|
||||
for (const [name, high] of second) {
|
||||
const before = first.get(name)
|
||||
if (before === undefined || high <= before) continue
|
||||
groups.push({ name, perMinute: Number(high - before) / minutes })
|
||||
}
|
||||
groups.sort((a, b) => b.perMinute - a.perMinute)
|
||||
const top = groups.slice(0, this.limit)
|
||||
this.logger.info(`${groups.length} active groups; listing ${top.length}.`)
|
||||
|
||||
if (this.sample > 0) {
|
||||
for (const [i, group] of top.entries()) {
|
||||
try {
|
||||
Object.assign(group, await this.sampleGroup(conn, group.name))
|
||||
} catch (error: any) {
|
||||
this.logger.warning(`${group.name}: ${error.message}`)
|
||||
}
|
||||
if ((i + 1) % 50 === 0) this.logger.info(`Sampled ${i + 1}/${top.length} groups…`)
|
||||
}
|
||||
}
|
||||
|
||||
const text = this.format(top, minutes)
|
||||
if (this.out) {
|
||||
await writeFile(this.out, text)
|
||||
this.logger.info(`Written to ${this.out}`)
|
||||
} else {
|
||||
console.log(text)
|
||||
}
|
||||
} catch (error) {
|
||||
failed = true
|
||||
throw error
|
||||
} finally {
|
||||
NntpService.release(conn, failed)
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
|
||||
/** Group name → highest article number, from LIST ACTIVE. */
|
||||
private async snapshot(conn: any): Promise<Map<string, bigint>> {
|
||||
const { data } = await conn.list(this.pattern)
|
||||
const result = new Map<string, bigint>()
|
||||
for (const line of Buffer.from(data).toString('latin1').split('\r\n')) {
|
||||
const [name, high] = line.split(' ')
|
||||
if (name && /^\d+$/.test(high ?? '')) result.set(name, BigInt(high))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/** Classifies the newest headers the way worker:fetch does. */
|
||||
private async sampleGroup(conn: any, name: string) {
|
||||
const { last } = parseGroupResponse((await conn.group(name)).resp)
|
||||
const start = last - BigInt(this.sample) + 1n
|
||||
const { overviews } = await conn.xover((start > 0n ? start : 1n).toString(), last.toString())
|
||||
let normal = 0
|
||||
let skipped = 0
|
||||
let obfuscated = 0
|
||||
let obfuscatedBytes = 0
|
||||
let headerBytes = 0
|
||||
for (const [, header] of overviews as [number, Record<string, string>][]) {
|
||||
if (!header) continue
|
||||
headerBytes += Object.values(header).reduce((sum, v) => sum + String(v).length + 1, 0)
|
||||
if (header.subject && parseSubject(header.subject)) normal++
|
||||
else if (indexerConfig.skipBodyMessageIds.some((re) => re.test(header['message-id'] ?? ''))) skipped++
|
||||
else {
|
||||
obfuscated++
|
||||
obfuscatedBytes += Number(header[':bytes']) || 0
|
||||
}
|
||||
}
|
||||
const sampled = normal + skipped + obfuscated
|
||||
return {
|
||||
sampled,
|
||||
normal,
|
||||
skipped,
|
||||
obfuscated,
|
||||
avgArticleBytes: obfuscated ? obfuscatedBytes / obfuscated : 0,
|
||||
avgHeaderBytes: sampled ? headerBytes / sampled : 0,
|
||||
}
|
||||
}
|
||||
|
||||
private format(groups: GroupRate[], minutes: number): string {
|
||||
const sampled = groups.some((g) => g.sampled)
|
||||
const lines = [
|
||||
`# Busiest ${this.pattern} groups by articles/min, measured over ${minutes.toFixed(1)} min on ${new Date().toISOString()}`,
|
||||
sampled
|
||||
? '# rank\tgroup\tarticles/min\tobfuscated%\tskipped%\tbody Mbit/s\theader Mbit/s'
|
||||
: '# rank\tgroup\tarticles/min',
|
||||
]
|
||||
let bodyTotal = 0
|
||||
let headerTotal = 0
|
||||
let articlesTotal = 0
|
||||
for (const [i, g] of groups.entries()) {
|
||||
articlesTotal += g.perMinute
|
||||
if (!g.sampled) {
|
||||
lines.push(`${i + 1}\t${g.name}\t${Math.round(g.perMinute)}`)
|
||||
continue
|
||||
}
|
||||
const obfuscatedShare = g.obfuscated! / g.sampled
|
||||
// Bodies: every obfuscated article is downloaded once. Headers: every article.
|
||||
const bodyMbit = (g.perMinute * obfuscatedShare * g.avgArticleBytes! * 8) / 60 / 1e6
|
||||
const headerMbit = (g.perMinute * g.avgHeaderBytes! * 8) / 60 / 1e6
|
||||
bodyTotal += bodyMbit
|
||||
headerTotal += headerMbit
|
||||
lines.push(
|
||||
[i + 1, g.name, Math.round(g.perMinute), (100 * obfuscatedShare).toFixed(1), ((100 * g.skipped!) / g.sampled).toFixed(1), bodyMbit.toFixed(1), headerMbit.toFixed(2)].join('\t')
|
||||
)
|
||||
}
|
||||
lines.push(`# total: ${Math.round(articlesTotal)} articles/min`)
|
||||
if (sampled) {
|
||||
lines.push(`# bandwidth to index like alt.binaries.test (headers + obfuscated bodies): ${(bodyTotal + headerTotal).toFixed(0)} Mbit/s`)
|
||||
lines.push(`# of which bodies ${bodyTotal.toFixed(0)} Mbit/s, headers ${headerTotal.toFixed(1)} Mbit/s (skip list applied)`)
|
||||
}
|
||||
return lines.join('\n') + '\n'
|
||||
}
|
||||
}
|
||||
@@ -1,144 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import Group from '#models/group'
|
||||
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'
|
||||
|
||||
export default class IndexScheduler extends BaseCommand {
|
||||
public static commandName = 'index:scheduler'
|
||||
public static description = 'Periodically checks for new articles and schedules them for fetching.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
private pool = NntpService
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Scheduler started. Awaiting tasks...')
|
||||
|
||||
const schedule = async () => {
|
||||
this.logger.info('Checking for new headers...')
|
||||
const groups = await Group.query().where('active', true)
|
||||
|
||||
if (groups.length === 0) {
|
||||
this.logger.info('No active groups to index. Add some via `node ace db:seed` or manually.')
|
||||
return
|
||||
}
|
||||
|
||||
let conn
|
||||
let failed = false
|
||||
try {
|
||||
conn = await this.pool.acquire()
|
||||
|
||||
for (const group of groups) {
|
||||
try {
|
||||
const { first: firstArticle, last: lastArticle } = await this.pool.selectGroup(
|
||||
conn,
|
||||
group.name
|
||||
)
|
||||
|
||||
await FetchRangeService.requeue(group)
|
||||
|
||||
// Postgres returns bigint columns as strings.
|
||||
const lastQueued = BigInt(group.lastQueuedId ?? 0)
|
||||
const { batchSize, maxBatchesPerTick, initialBackfill } = indexerConfig
|
||||
let startId =
|
||||
lastQueued === 0n ? lastArticle - initialBackfill + 1n : lastQueued + 1n
|
||||
if (startId < firstArticle) startId = firstArticle
|
||||
|
||||
if (startId > lastArticle) {
|
||||
await FetchRangeService.advanceWatermark(group)
|
||||
this.logger.info(`No new headers for group ${group.name}.`)
|
||||
continue
|
||||
}
|
||||
|
||||
let batches = 0
|
||||
let endId = startId - 1n
|
||||
while (endId < lastArticle && batches < maxBatchesPerTick) {
|
||||
const batchStart = endId + 1n
|
||||
endId = batchStart + batchSize - 1n < lastArticle ? batchStart + batchSize - 1n : lastArticle
|
||||
await FetchRangeService.queue(group, batchStart, endId)
|
||||
batches++
|
||||
}
|
||||
|
||||
this.logger.info(
|
||||
`Queued ${batches} fetch job(s) for ${group.name}: articles ${startId} to ${endId}` +
|
||||
(endId < lastArticle ? ` (${lastArticle - endId} left for next tick)` : '')
|
||||
)
|
||||
|
||||
group.lastQueuedId = endId
|
||||
await group.save()
|
||||
const watermark = await FetchRangeService.advanceWatermark(group)
|
||||
if (endId - watermark > batchSize * BigInt(maxBatchesPerTick) * 3n) {
|
||||
this.logger.warning(
|
||||
`${group.name}: indexed up to ${watermark}, ${endId - watermark} articles behind the queue.`
|
||||
)
|
||||
}
|
||||
} catch (err: any) {
|
||||
failed = true
|
||||
this.logger.error(`Error processing group ${group.name}: ${err.message}`)
|
||||
}
|
||||
}
|
||||
} catch (err: any) {
|
||||
failed = true
|
||||
this.logger.error(`Error in scheduler main loop: ${err.message}`)
|
||||
} finally {
|
||||
if (conn) {
|
||||
this.pool.release(conn, failed)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const split = async () => {
|
||||
try {
|
||||
const result = await SplitService.splitPending()
|
||||
if (result.parents > 0) {
|
||||
this.logger.info(
|
||||
`Split ${result.parents} conflicted binaries: ${result.volumes} verified volumes, ${result.completed} completed.`
|
||||
)
|
||||
}
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Error splitting conflicted binaries: ${err.message}`)
|
||||
}
|
||||
}
|
||||
|
||||
// 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()
|
||||
if (queued > 0) this.logger.info(`Queued ${queued} backfill range(s).`)
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Error queuing backfill: ${err.message}`)
|
||||
}
|
||||
}
|
||||
|
||||
// Run immediately, then wait a full interval after each run so runs never overlap. A failed
|
||||
// round (e.g. Postgres restarting) is logged and the next one tries again: an error that escaped
|
||||
// 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, feedInspection]) {
|
||||
try {
|
||||
await step()
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Scheduler round failed, retrying next round: ${err.message}`)
|
||||
}
|
||||
}
|
||||
await Shutdown.sleep(indexerConfig.intervalMs)
|
||||
}
|
||||
await this.pool.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
@@ -1,56 +0,0 @@
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,21 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import ReleaseMergeService from '#services/ReleaseMergeService'
|
||||
|
||||
/** Merges uploads split per poster (gc:scheduler also does this every sweep). */
|
||||
export default class ReleaseMergeSplit extends BaseCommand {
|
||||
public static commandName = 'release:merge-split'
|
||||
public static description = 'Merges releases of one upload that was posted under a different poster per file.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({ description: 'Only count them' })
|
||||
declare dryRun: boolean
|
||||
|
||||
public async run() {
|
||||
const { names, releases } = await ReleaseMergeService.mergeSplitPosts({ dryRun: this.dryRun })
|
||||
this.logger.info(`${names} split uploads; ${releases} releases ${this.dryRun ? 'would be' : ''} merged into their first.`)
|
||||
}
|
||||
}
|
||||
@@ -1,165 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import IngestService from '#services/IngestService'
|
||||
import NameService from '#services/NameService'
|
||||
import QueueService from '#services/QueueService'
|
||||
import { rarReleaseIdentity, releaseIdentity } from '#lib/release/index'
|
||||
|
||||
const BATCH = 5000
|
||||
|
||||
/**
|
||||
* Recomputes the release of every binary (and its file) with the current
|
||||
* rules in lib/release. Used after changing those rules.
|
||||
*/
|
||||
export default class ReleaseRebuild extends BaseCommand {
|
||||
public static commandName = 'release:rebuild'
|
||||
public static description = 'Reassigns all binaries and files to releases using the current grouping rules.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({
|
||||
description:
|
||||
'Also drop everything derived from file contents (par2 lists, par2/RAR groupings, real names), ' +
|
||||
'to rebuild it with content:backfill --all',
|
||||
})
|
||||
declare resetContent: boolean
|
||||
|
||||
public async run() {
|
||||
if (this.resetContent) await this.dropContentDerived()
|
||||
|
||||
let lastId = 0
|
||||
let assigned = 0
|
||||
let unassigned = 0
|
||||
try {
|
||||
while (true) {
|
||||
const binaries = await db
|
||||
.from('binaries')
|
||||
.where('id', '>', lastId)
|
||||
.orderBy('id')
|
||||
.limit(BATCH)
|
||||
.select('id', 'poster', 'subject', 'name', 'groups', 'posted_at')
|
||||
if (binaries.length === 0) break
|
||||
lastId = Number(binaries.at(-1).id)
|
||||
|
||||
const segments = binaries.map((binary) => {
|
||||
// Body-derived binaries store a synthetic subject, see segmentFromYenc.
|
||||
const weak = /^"[^"]*" yEnc$/.test(binary.subject)
|
||||
return {
|
||||
id: Number(binary.id),
|
||||
poster: binary.poster,
|
||||
groups: binary.groups,
|
||||
postedAt: binary.posted_at ? new Date(binary.posted_at) : undefined,
|
||||
weakIdentity: weak,
|
||||
release: releaseIdentity({ poster: binary.poster, subject: binary.subject, name: binary.name, weak }),
|
||||
}
|
||||
})
|
||||
|
||||
await db.transaction(async (trx) => {
|
||||
const releaseIds = await IngestService.upsertReleases(trx.knexClient, segments)
|
||||
const ids = segments.map((s) => s.id)
|
||||
const releaseIdsPerBinary = segments.map((s) =>
|
||||
s.release ? releaseIds.get(s.release.key)! : null
|
||||
)
|
||||
await trx.rawQuery(
|
||||
`UPDATE binaries b SET release_id = v.release_id
|
||||
FROM unnest(?::bigint[], ?::bigint[]) AS v(id, release_id)
|
||||
WHERE b.id = v.id AND b.release_id IS DISTINCT FROM v.release_id
|
||||
AND NOT (v.release_id IS NULL
|
||||
AND b.release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))
|
||||
-- A posted NZB decided this grouping (NzbImportService); name rules don't override it.
|
||||
AND NOT EXISTS (SELECT 1 FROM files n JOIN file_details nd ON nd.file_id = n.id WHERE n.release_id = b.release_id AND nd.content_type = 'nzb')`,
|
||||
[ids, releaseIdsPerBinary] as any
|
||||
)
|
||||
})
|
||||
assigned += segments.filter((s) => s.release).length
|
||||
unassigned += segments.filter((s) => !s.release).length
|
||||
}
|
||||
|
||||
// Files grouped by their RAR headers: regroup from the stored listing (no downloads).
|
||||
let rarGrouped = 0
|
||||
let rarCursor = 0
|
||||
while (true) {
|
||||
const rows = await db
|
||||
.from('files as f')
|
||||
.join('binaries as b', 'b.file_id', 'f.id')
|
||||
.join('file_details as d', 'd.file_id', 'f.id')
|
||||
.whereNotNull('d.archive')
|
||||
.whereNull('b.release_id')
|
||||
.where('f.id', '>', rarCursor)
|
||||
.orderBy('f.id')
|
||||
.limit(BATCH)
|
||||
.select('f.id as file_id', 'd.archive', 'b.id as binary_id', 'b.poster', 'b.groups', 'b.posted_at')
|
||||
if (rows.length === 0) break
|
||||
rarCursor = Number(rows.at(-1).file_id)
|
||||
const segments = rows.map((r) => ({
|
||||
fileId: Number(r.file_id),
|
||||
binaryId: Number(r.binary_id),
|
||||
poster: r.poster,
|
||||
groups: r.groups,
|
||||
postedAt: r.posted_at ? new Date(r.posted_at) : undefined,
|
||||
weakIdentity: true,
|
||||
release: rarReleaseIdentity(r.archive?.entries ?? []),
|
||||
}))
|
||||
const ids = await IngestService.upsertReleases(db.connection().getWriteClient(), segments)
|
||||
for (const s of segments.filter((x) => x.release)) {
|
||||
const releaseId = ids.get(s.release!.key)
|
||||
await db.from('binaries').where('id', s.binaryId).update({ release_id: releaseId })
|
||||
await db.from('files').where('id', s.fileId).update({ release_id: releaseId })
|
||||
rarGrouped++
|
||||
}
|
||||
}
|
||||
|
||||
// Real names from par2 for binaries with a known size (no downloads).
|
||||
let named = 0
|
||||
let nameCursor = 0
|
||||
while (true) {
|
||||
const rows = await db.from('binaries').whereNotNull('size').where('id', '>', nameCursor).orderBy('id').limit(BATCH).select('id')
|
||||
if (rows.length === 0) break
|
||||
nameCursor = Number(rows.at(-1).id)
|
||||
named += await NameService.resolve(rows.map((r) => Number(r.id)))
|
||||
}
|
||||
|
||||
const files = await db.rawQuery(
|
||||
`UPDATE files f SET release_id = b.release_id FROM binaries b
|
||||
WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`
|
||||
)
|
||||
const orphans = await db.rawQuery(
|
||||
`DELETE FROM releases r WHERE 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)`
|
||||
)
|
||||
if (this.resetContent) {
|
||||
this.logger.info('Content-derived data dropped; run `node ace content:backfill --all` to rebuild it.')
|
||||
}
|
||||
this.logger.info(
|
||||
`Binaries: ${assigned} in a release, ${unassigned} without, ${rarGrouped} regrouped from RAR headers, ` +
|
||||
`${named} named from par2. ` +
|
||||
`Files updated: ${files.rowCount}. Empty releases removed: ${orphans.rowCount}.`
|
||||
)
|
||||
} finally {
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
|
||||
/** par2 lists, par2/RAR release assignments and recovered names. */
|
||||
private async dropContentDerived() {
|
||||
await db.transaction(async (trx) => {
|
||||
await trx.rawQuery('DELETE FROM release_files')
|
||||
await trx.rawQuery(`UPDATE binaries SET real_name = NULL WHERE real_name IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE files SET real_name = NULL WHERE real_name IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE file_details SET archive = NULL WHERE archive IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE releases SET contents = NULL, encrypted = NULL WHERE contents IS NOT NULL OR encrypted IS NOT NULL`)
|
||||
// Binaries whose name rules find no release were grouped from contents (or wrongly by size).
|
||||
await trx.rawQuery(`
|
||||
UPDATE binaries SET release_id = NULL
|
||||
WHERE subject ~ '^"[^"]*" yEnc$' AND release_id IS NOT NULL
|
||||
AND (name !~ '\\.[A-Za-z0-9]{1,5}$' OR release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))`)
|
||||
await trx.rawQuery(`
|
||||
UPDATE files f SET release_id = b.release_id FROM binaries b
|
||||
WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`)
|
||||
})
|
||||
this.logger.info('Dropped par2 lists, content-derived groupings and recovered names.')
|
||||
}
|
||||
}
|
||||
@@ -1,27 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { renameFromStoredContents, renameHiddenNamed } from '#services/ArchiveService'
|
||||
import { regroupAndRenameFromPar2 } from '#services/NameService'
|
||||
|
||||
/** Renames random-named releases from their stored archive listings (no downloads). */
|
||||
export default class ReleaseRename extends BaseCommand {
|
||||
public static commandName = 'release:rename'
|
||||
public static description = 'Renames random-named releases from their par2 listings and archive contents.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({ description: 'Only releases named after a hidden archive entry (e.g. a decoy ".x.mkv")' })
|
||||
declare hidden: boolean
|
||||
|
||||
public async run() {
|
||||
if (this.hidden) {
|
||||
this.logger.info(`Renamed ${await renameHiddenNamed()} releases named after a hidden archive entry.`)
|
||||
return
|
||||
}
|
||||
const par2 = await regroupAndRenameFromPar2()
|
||||
this.logger.info(`par2: ${par2.regrouped} files moved into the set that names them, ${par2.renamed} releases renamed.`)
|
||||
this.logger.info(`Renamed ${await renameFromStoredContents()} releases from their archive contents.`)
|
||||
}
|
||||
}
|
||||
@@ -1,57 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
|
||||
/** Prints how well files are being grouped into releases. */
|
||||
export default class ReleaseStats extends BaseCommand {
|
||||
public static commandName = 'release:stats'
|
||||
public static description = 'Shows release recognition statistics.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
const { rows } = await db.rawQuery(`
|
||||
WITH r AS (
|
||||
SELECT r.id, r.file_total, r.poster IS NULL AS weak, r.source,
|
||||
count(b.id) AS binaries,
|
||||
count(b.id) FILTER (WHERE b.completed_at IS NOT NULL) AS completed
|
||||
FROM releases r LEFT JOIN binaries b ON b.release_id = r.id
|
||||
GROUP BY r.id
|
||||
)
|
||||
SELECT
|
||||
count(*) AS releases,
|
||||
count(*) FILTER (WHERE completed > 0) AS with_files,
|
||||
count(*) FILTER (WHERE completed >= 2) AS multi_file,
|
||||
count(*) FILTER (WHERE file_total IS NOT NULL AND completed >= file_total) AS complete_by_counter,
|
||||
count(*) FILTER (WHERE weak) AS from_obfuscated,
|
||||
count(*) FILTER (WHERE weak AND completed > 0) AS from_obfuscated_with_files
|
||||
FROM r`)
|
||||
const { rows: coverage } = await db.rawQuery(`
|
||||
SELECT count(*) AS completed_files,
|
||||
count(*) FILTER (WHERE release_id IS NOT NULL) AS in_release
|
||||
FROM binaries WHERE completed_at IS NOT NULL`)
|
||||
const { rows: bySource } = await db.rawQuery(`
|
||||
SELECT coalesce(r.source, 'unknown') AS source, count(DISTINCT r.id) AS releases,
|
||||
count(b.id) FILTER (WHERE b.completed_at IS NOT NULL) AS completed_files
|
||||
FROM releases r JOIN binaries b ON b.release_id = r.id GROUP BY 1 ORDER BY 2 DESC`)
|
||||
// Readable names: descriptive as posted, or recovered from contents.
|
||||
const { rows: naming } = await db.rawQuery(`
|
||||
WITH f AS (
|
||||
SELECT real_name IS NOT NULL AS recovered,
|
||||
is_readable_name(coalesce(real_name, filename)) AS readable_name,
|
||||
regexp_replace(coalesce(real_name, filename),
|
||||
'(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z(\\.\\d{3})?|zip|\\d{3}|nfo|sfv|nzb|tar|zst|mkv|mp4|avi|flac|mp3|jpg))+$', '', 'i') AS stem,
|
||||
coalesce(real_name, filename) AS name
|
||||
FROM files
|
||||
)
|
||||
SELECT count(*) AS files,
|
||||
count(*) FILTER (WHERE recovered) AS names_recovered,
|
||||
count(*) FILTER (WHERE name ~ '\\.[A-Za-z0-9]{1,5}$' AND readable_name) AS readable
|
||||
FROM f`)
|
||||
this.logger.info(JSON.stringify({ ...rows[0], ...coverage[0] }))
|
||||
this.logger.info(`naming: ${JSON.stringify(naming[0])}`)
|
||||
this.logger.info(`by source: ${JSON.stringify(bySource)}`)
|
||||
}
|
||||
}
|
||||
@@ -1,461 +0,0 @@
|
||||
import { gzipSync, unzipSync } from 'node:zlib'
|
||||
import { spawn, type ChildProcess } from 'node:child_process'
|
||||
import { availableParallelism } from 'node:os'
|
||||
import { setTimeout as sleep } from 'node:timers/promises'
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import nntpConfig from '#config/nntp'
|
||||
import indexerConfig from '#config/indexer'
|
||||
import { XoverClient, parseOverviews, type Overview } from '#lib/nntp/xover_client'
|
||||
import { runRange, type OverviewSource } from '#lib/seed/runner'
|
||||
import type { SeedBatch } from '#lib/seed/wire'
|
||||
import type { OverviewHeader } from '#lib/ingest/segment'
|
||||
import { probeInspection, probeWalk, type ContentTask } from '#lib/content/probe'
|
||||
|
||||
class LeaseLost extends Error {}
|
||||
class Stopping extends Error {}
|
||||
|
||||
interface InspectJob {
|
||||
queue: string
|
||||
jobId: string
|
||||
token: string
|
||||
task: ContentTask
|
||||
}
|
||||
|
||||
interface Lease {
|
||||
id: number
|
||||
group: string
|
||||
start: number
|
||||
end: number
|
||||
}
|
||||
|
||||
/**
|
||||
* A seeding node (CONCEPT.md D38): leases article ranges from the seed API
|
||||
* over HTTPS, fetches and aggregates their headers over its own NNTP
|
||||
* connection(s) (D37), and posts the results back in gzip-compressed
|
||||
* batches. Outgoing requests only: no ports, database or Redis needed.
|
||||
*
|
||||
* SEED_URL=https://seed.example SEED_TOKEN=sn_… node ace seed:node --ranges 4 --inspectors 2
|
||||
*/
|
||||
export default class SeedNode extends BaseCommand {
|
||||
public static commandName = 'seed:node'
|
||||
public static description =
|
||||
'Runs a seeding node: leases ranges from the seed API and submits what it aggregated.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: false,
|
||||
staysAlive: true,
|
||||
}
|
||||
|
||||
@flags.string({ description: 'Seed API base URL (default: SEED_URL)' })
|
||||
declare url?: string
|
||||
|
||||
@flags.string({ description: 'Node token (default: SEED_TOKEN)' })
|
||||
declare token?: string
|
||||
|
||||
@flags.number({
|
||||
description:
|
||||
'Ranges seeded at once (with SEED_FETCH=direct, one NNTP connection each); 0 for an inspection-only node (default: SEED_RANGES or 1)',
|
||||
default: process.env.SEED_RANGES ? Number(process.env.SEED_RANGES) : 1,
|
||||
})
|
||||
declare ranges: number
|
||||
|
||||
@flags.number({
|
||||
description:
|
||||
'Content inspections and archive walks worked on at once, articles through the seed API (default: SEED_INSPECTORS or 0)',
|
||||
default: Number(process.env.SEED_INSPECTORS) || 0,
|
||||
})
|
||||
declare inspectors: number
|
||||
|
||||
@flags.number({ description: 'Articles per XOVER', default: 20_000 })
|
||||
declare chunk: number
|
||||
|
||||
@flags.number({ description: 'XOVER chunks per batch posted to the API', default: 5 })
|
||||
declare submitEvery: number
|
||||
|
||||
@flags.string({
|
||||
description:
|
||||
'relay: headers come through the seed API (no NNTP account needed); direct: own NNTP connections (default: SEED_FETCH or relay)',
|
||||
default: process.env.SEED_FETCH || 'relay',
|
||||
})
|
||||
declare fetch: string
|
||||
|
||||
@flags.number({
|
||||
description:
|
||||
'Processes the ranges and inspections are spread over; one process uses one core (default: SEED_PROCESSES or the number of cores)',
|
||||
default: process.env.SEED_PROCESSES ? Number(process.env.SEED_PROCESSES) : availableParallelism(),
|
||||
})
|
||||
declare processes: number
|
||||
|
||||
@flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' })
|
||||
declare once: boolean
|
||||
|
||||
private stopping = false
|
||||
/** In a worker process: its number, in front of its log lines. */
|
||||
private readonly child = process.env.SEED_PROCESS ?? ''
|
||||
|
||||
public async run() {
|
||||
const base = (this.url ?? process.env.SEED_URL ?? '').replace(/\/+$/, '')
|
||||
const token = this.token ?? process.env.SEED_TOKEN ?? ''
|
||||
if (!base || !token) {
|
||||
this.logger.error('Set --url and --token (or SEED_URL and SEED_TOKEN).')
|
||||
this.exitCode = 1
|
||||
return
|
||||
}
|
||||
// JavaScript runs on one core per process: several slots are spread over worker processes.
|
||||
const processes = Math.max(1, Math.min(this.processes || 1, this.ranges + this.inspectors))
|
||||
if (!this.child && processes > 1) return this.supervise(base, token, processes)
|
||||
|
||||
const stop = () => {
|
||||
if (!this.stopping && !this.child) this.logger.info('Stopping: giving back the ranges in progress…')
|
||||
this.stopping = true
|
||||
}
|
||||
process.on('SIGTERM', stop)
|
||||
process.on('SIGINT', stop)
|
||||
|
||||
const api = new SeedApi(base, token, (message) => this.logger.warning(message))
|
||||
if (!this.child) this.logger.info(
|
||||
`Seeding node for ${base}: ${this.ranges} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}` +
|
||||
(this.inspectors > 0 ? `; ${this.inspectors} inspection(s) at once, articles through the seed API.` : '.')
|
||||
)
|
||||
await Promise.all([
|
||||
...Array.from({ length: this.ranges }, (_, i) => this.work(api, i + 1)),
|
||||
...Array.from({ length: this.inspectors }, (_, i) => this.inspect(api, i + 1)),
|
||||
])
|
||||
if (!this.child) this.logger.info('Seeding node stopped.')
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs the slots in `processes` worker processes (this command again, with
|
||||
* SEED_PROCESS set), each with its share of the ranges and inspections. A
|
||||
* worker that exits while the node runs is restarted; SIGTERM/SIGINT are
|
||||
* passed on, and the node stops when all workers have given their work back.
|
||||
*/
|
||||
private async supervise(base: string, token: string, processes: number) {
|
||||
// Slots dealt out in turn (ranges first, then inspections), so every process has work.
|
||||
const share = (from: number, to: number, i: number) =>
|
||||
Array.from({ length: to - from }, (_, k) => from + k).filter((k) => k % processes === i).length
|
||||
this.logger.info(
|
||||
`Seeding node for ${base}: ${this.ranges} range(s) and ${this.inspectors} inspection(s) at once, over ${processes} processes, ` +
|
||||
`${this.fetch === 'direct' ? 'own NNTP connections' : 'articles and headers through the seed API'}.`
|
||||
)
|
||||
const children = new Set<ChildProcess>()
|
||||
const start = (n: number): Promise<void> =>
|
||||
new Promise((resolve) => {
|
||||
const args = [
|
||||
...process.execArgv,
|
||||
process.argv[1],
|
||||
'seed:node',
|
||||
`--ranges=${share(0, this.ranges, n - 1)}`,
|
||||
`--inspectors=${share(this.ranges, this.ranges + this.inspectors, n - 1)}`,
|
||||
`--fetch=${this.fetch}`,
|
||||
`--chunk=${this.chunk}`,
|
||||
`--submit-every=${this.submitEvery}`,
|
||||
...(this.once ? ['--once'] : []),
|
||||
]
|
||||
const child = spawn(process.execPath, args, {
|
||||
env: { ...process.env, SEED_URL: base, SEED_TOKEN: token, SEED_PROCESS: String(n) },
|
||||
stdio: 'inherit',
|
||||
})
|
||||
children.add(child)
|
||||
child.on('exit', (code, signal) => {
|
||||
children.delete(child)
|
||||
if (this.stopping || this.once || code === 0) return resolve()
|
||||
this.logger.warning(`Process ${n} exited (${signal ?? code}); restarting in 10 s`)
|
||||
setTimeout(() => start(n).then(resolve), 10_000)
|
||||
})
|
||||
})
|
||||
const stop = (signal: NodeJS.Signals) => {
|
||||
if (!this.stopping) this.logger.info('Stopping: giving back the ranges and inspections in progress…')
|
||||
this.stopping = true
|
||||
for (const child of children) child.kill(signal)
|
||||
}
|
||||
process.on('SIGTERM', () => stop('SIGTERM'))
|
||||
process.on('SIGINT', () => stop('SIGINT'))
|
||||
await Promise.all(Array.from({ length: processes }, (_, i) => start(i + 1)))
|
||||
this.logger.info('Seeding node stopped.')
|
||||
}
|
||||
|
||||
private async work(api: SeedApi, worker: number) {
|
||||
const log = (message: string) => this.logger.info(`[${this.child && `${this.child}/`}${worker}] ${message}`)
|
||||
// The NNTP connection is opened for a lease and kept while leases keep coming; with
|
||||
// nothing to lease it's closed, so an idle node holds no connections.
|
||||
let client: XoverClient | undefined
|
||||
try {
|
||||
while (!this.stopping) {
|
||||
const lease = await this.lease(api, log)
|
||||
if (!lease) {
|
||||
client?.close()
|
||||
client = undefined
|
||||
if (this.once) return
|
||||
await this.pause(60_000)
|
||||
continue
|
||||
}
|
||||
if (this.fetch !== 'direct') {
|
||||
await this.workRange(api, new RelaySource(api, lease.id), lease, log)
|
||||
continue
|
||||
}
|
||||
if (!client) {
|
||||
try {
|
||||
client = await this.connect()
|
||||
} catch (error: any) {
|
||||
// Never started: the lease doesn't count as one of the range's attempts.
|
||||
const ipLimit = /simultaneous IP/i.test(error.message)
|
||||
log(`NNTP login failed: ${error.message}; trying again in ${ipLimit ? 5 : 1} min`)
|
||||
await api
|
||||
.post(`/api/v1/seed/ranges/${lease.id}/release`, {
|
||||
error: `NNTP login: ${error.message}`,
|
||||
uncounted: true,
|
||||
})
|
||||
.catch(() => {})
|
||||
await this.pause(ipLimit ? 300_000 : 60_000)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if (!(await this.workRange(api, client, lease, log))) {
|
||||
client.close()
|
||||
client = undefined
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
client?.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Content inspection and archive walks (CONCEPT.md D41): lease a few jobs,
|
||||
* probe each with articles through the seed API, post the results. The
|
||||
* indexer applies them; a job left unanswered goes back after 10 minutes.
|
||||
*/
|
||||
private async inspect(api: SeedApi, slot: number) {
|
||||
const log = (message: string) => this.logger.info(`[inspect ${this.child && `${this.child}/`}${slot}] ${message}`)
|
||||
const fetchArticle = (id: string) => api.get(`/api/v1/seed/articles?id=${encodeURIComponent(id)}`, 4)
|
||||
while (!this.stopping) {
|
||||
let jobs: InspectJob[] = []
|
||||
try {
|
||||
jobs = ((await api.post('/api/v1/seed/inspect/lease', { max: 2 })) as { jobs: InspectJob[] } | null)?.jobs ?? []
|
||||
} catch (error: any) {
|
||||
if (error instanceof Unauthorized) throw error
|
||||
log(`Lease failed: ${error.message}`)
|
||||
}
|
||||
if (jobs.length === 0) {
|
||||
if (this.once) return
|
||||
await this.pause(30_000)
|
||||
continue
|
||||
}
|
||||
for (const job of jobs) {
|
||||
const ref = { queue: job.queue, jobId: job.jobId, token: job.token }
|
||||
if (this.stopping) {
|
||||
await api.post('/api/v1/seed/inspect/release', ref).catch(() => {})
|
||||
continue
|
||||
}
|
||||
try {
|
||||
const result =
|
||||
job.task.kind === 'inspect'
|
||||
? await probeInspection(job.task, fetchArticle, (message) => log(message))
|
||||
: await probeWalk(job.task, fetchArticle)
|
||||
await api.post('/api/v1/seed/inspect/done', { ...ref, result: JSON.stringify(result) })
|
||||
} catch (error: any) {
|
||||
if (error instanceof Unauthorized) throw error
|
||||
// The job's lock expired and it went to another node: nothing to report.
|
||||
if (error instanceof LeaseLost) continue
|
||||
log(`${job.task.kind === 'inspect' ? `"${job.task.name}"` : `walk of release ${job.task.releaseId}`} failed: ${error.message}`)
|
||||
await api.post('/api/v1/seed/inspect/failed', { ...ref, error: error.message }).catch(() => {})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private connect() {
|
||||
return XoverClient.connect({
|
||||
host: nntpConfig.host,
|
||||
port: nntpConfig.port,
|
||||
secure: nntpConfig.secure,
|
||||
servername: nntpConfig.tlsServername,
|
||||
user: nntpConfig.user,
|
||||
password: nntpConfig.password?.release(),
|
||||
timeoutMs: nntpConfig.commandTimeoutMs,
|
||||
compress: true,
|
||||
})
|
||||
}
|
||||
|
||||
/** The next range, or null when there is none (or the API can't be reached; logged). */
|
||||
private async lease(api: SeedApi, log: (message: string) => void): Promise<Lease | null> {
|
||||
try {
|
||||
return ((await api.post('/api/v1/seed/lease')) as { range: Lease } | null)?.range ?? null
|
||||
} catch (error: any) {
|
||||
if (error instanceof Unauthorized) throw error
|
||||
log(`Lease failed: ${error.message}`)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/** Works one leased range; false when the NNTP connection should be replaced. */
|
||||
private async workRange(
|
||||
api: SeedApi,
|
||||
client: OverviewSource,
|
||||
lease: Lease,
|
||||
log: (message: string) => void
|
||||
): Promise<boolean> {
|
||||
log(`${lease.group} ${lease.start}–${lease.end}`)
|
||||
const started = Date.now()
|
||||
let headers = 0
|
||||
let chain: Promise<void> = Promise.resolve()
|
||||
// Where the range resumes after what was stored: a batch split in two reports it with its first half.
|
||||
let reported = lease.start
|
||||
const posting = new Set<Promise<void>>()
|
||||
// Throws once a post has failed, so the range stops fetching.
|
||||
const failedSoFar = () => Promise.race([chain, Promise.resolve()])
|
||||
try {
|
||||
await runRange({
|
||||
client,
|
||||
group: lease.group,
|
||||
start: lease.start,
|
||||
end: lease.end,
|
||||
chunk: this.chunk,
|
||||
submitEvery: this.submitEvery,
|
||||
isDeferred: (header: OverviewHeader) =>
|
||||
indexerConfig.deferBodyMessageIds.some((re) => re.test(header['message-id'] ?? '')),
|
||||
// Batches are posted in order, one after the other, while fetching goes on; with two
|
||||
// waiting, fetching waits too. A failed post stops the chain, so no batch is stored
|
||||
// past a missing one (the range would resume beyond it).
|
||||
sink: async (batch: SeedBatch) => {
|
||||
if (this.stopping) throw new Stopping()
|
||||
while (posting.size >= 2) await Promise.race(posting)
|
||||
const post = chain.then(async () => {
|
||||
await this.postBatch(api, lease.id, batch, reported)
|
||||
reported = batch.next
|
||||
headers += batch.headers
|
||||
})
|
||||
chain = post
|
||||
posting.add(post)
|
||||
post.catch(() => {}).finally(() => posting.delete(post))
|
||||
await failedSoFar()
|
||||
},
|
||||
})
|
||||
await chain
|
||||
await api.post(`/api/v1/seed/ranges/${lease.id}/done`)
|
||||
const seconds = (Date.now() - started) / 1000
|
||||
log(
|
||||
`done: ${headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s (${Math.round(headers / seconds).toLocaleString('en')}/s)`
|
||||
)
|
||||
return true
|
||||
} catch (error: any) {
|
||||
if (error instanceof Unauthorized) throw error
|
||||
// An XOVER may still be in flight: the connection is replaced either way.
|
||||
if (error instanceof LeaseLost) {
|
||||
log(`Range ${lease.id} passed to another node; dropping it.`)
|
||||
return false
|
||||
}
|
||||
const reason = error instanceof Stopping ? 'node stopping' : error.message
|
||||
if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`)
|
||||
// Let posts in flight finish (or fail) first, so the range resumes after what was stored.
|
||||
await chain.catch(() => {})
|
||||
await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {})
|
||||
if (!this.stopping) await this.pause(30_000)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Posts a batch; one the seed API finds too large (413: dense recent ranges) is
|
||||
* posted in two halves, at most three splits deep. Only the second half moves
|
||||
* the range's resume point, so a failure in between loses nothing.
|
||||
*/
|
||||
private async postBatch(api: SeedApi, rangeId: number, batch: SeedBatch, reported: number, depth = 0): Promise<void> {
|
||||
try {
|
||||
await api.post(`/api/v1/seed/ranges/${rangeId}/batches`, batch)
|
||||
} catch (error: any) {
|
||||
const parts = batch.files.length + batch.fragments.length
|
||||
if (!/: 413 /.test(String(error.message)) || depth >= 3 || parts < 2) throw error
|
||||
const files = Math.ceil(batch.files.length / 2)
|
||||
const fragments = Math.ceil(batch.fragments.length / 2)
|
||||
const first: SeedBatch = { ...batch, next: reported, headers: 0, obfuscated: 0, deferred: 0, files: batch.files.slice(0, files), fragments: batch.fragments.slice(0, fragments) }
|
||||
const second: SeedBatch = { ...batch, files: batch.files.slice(files), fragments: batch.fragments.slice(fragments) }
|
||||
await this.postBatch(api, rangeId, first, reported, depth + 1)
|
||||
await this.postBatch(api, rangeId, second, reported, depth + 1)
|
||||
}
|
||||
}
|
||||
|
||||
/** Sleeps, but wakes up early when the node is stopping. */
|
||||
private async pause(ms: number) {
|
||||
for (let waited = 0; waited < ms && !this.stopping; waited += 1000) await sleep(1000)
|
||||
}
|
||||
}
|
||||
|
||||
class Unauthorized extends Error {}
|
||||
|
||||
/** The seed API client: JSON over HTTPS, gzip request bodies, retries for network errors and 5xx. */
|
||||
class SeedApi {
|
||||
constructor(
|
||||
private base: string,
|
||||
private token: string,
|
||||
private warn: (message: string) => void
|
||||
) {}
|
||||
|
||||
async post(path: string, body?: unknown): Promise<unknown> {
|
||||
const payload = body === undefined ? undefined : gzipSync(JSON.stringify(body))
|
||||
const result = await this.request('POST', path, payload)
|
||||
if (result === null) return null
|
||||
const text = result.toString('utf8')
|
||||
return text ? JSON.parse(text) : null
|
||||
}
|
||||
|
||||
/** A binary GET (a relay chunk, an article). */
|
||||
async get(path: string, attempts = 14): Promise<Buffer> {
|
||||
return (await this.request('GET', path, undefined, attempts)) ?? Buffer.alloc(0)
|
||||
}
|
||||
|
||||
private async request(method: string, path: string, payload?: Buffer, attempts = 14): Promise<Buffer | null> {
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
let status = 0
|
||||
let data = Buffer.alloc(0)
|
||||
try {
|
||||
const response = await fetch(this.base + path, {
|
||||
method,
|
||||
headers: {
|
||||
authorization: `Bearer ${this.token}`,
|
||||
...(payload ? { 'content-type': 'application/json', 'content-encoding': 'gzip' } : {}),
|
||||
},
|
||||
body: payload,
|
||||
signal: AbortSignal.timeout(10 * 60 * 1000),
|
||||
})
|
||||
status = response.status
|
||||
data = Buffer.from(await response.arrayBuffer())
|
||||
} catch (error: any) {
|
||||
data = Buffer.from(error.message)
|
||||
}
|
||||
const text = () => data.toString('utf8').slice(0, 300)
|
||||
if (status === 204) return null
|
||||
if (status >= 200 && status < 300) return data
|
||||
if (status === 401) throw new Unauthorized('The seed API rejected this node token')
|
||||
if (status === 409) throw new LeaseLost(text())
|
||||
// 404 comes from the proxy while the seed API restarts (a deploy): this API has no 404s of its own.
|
||||
if (status >= 400 && status < 500 && status !== 404)
|
||||
throw new Error(`Seed API ${path}: ${status} ${text()}`)
|
||||
// Network errors, 404 and 5xx: storing a batch twice is harmless, so try again.
|
||||
// ~20 minutes in all, longer than a lease lasts: a restart or maintenance of the API is waited out.
|
||||
if (attempt >= attempts) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`)
|
||||
const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1))
|
||||
this.warn(`Seed API ${path}: ${status || text()}; retrying in ${wait / 1000} s`)
|
||||
await sleep(wait)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** A range's overviews through the seed API's relay (CONCEPT.md D38): no news server account on the node. */
|
||||
class RelaySource implements OverviewSource {
|
||||
constructor(
|
||||
private api: SeedApi,
|
||||
private rangeId: number
|
||||
) {}
|
||||
|
||||
async group() {}
|
||||
|
||||
async xover(from: number, to: number): Promise<Overview[]> {
|
||||
const raw = await this.api.get(
|
||||
`/api/v1/seed/ranges/${this.rangeId}/chunk?from=${from}&to=${to}`
|
||||
)
|
||||
return raw.length > 0 ? parseOverviews(unzipSync(raw).toString('utf8')) : []
|
||||
}
|
||||
}
|
||||
@@ -1,47 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import NntpService from '#services/NntpService'
|
||||
import SeedOrchestrator from '#services/SeedOrchestrator'
|
||||
|
||||
/**
|
||||
* Plans seed ranges (CONCEPT.md D38) for groups: every article from below
|
||||
* what live indexing and the backfill cover down to `--days` back (or the
|
||||
* server's oldest article) that no earlier range covers. Running it again
|
||||
* plans only what's missing. A group's backfill stops where its plan
|
||||
* starts. Ranges are interleaved across groups, newest first.
|
||||
*
|
||||
* node ace seed:plan alt.binaries.ath alt.binaries.mom --days 365
|
||||
*/
|
||||
export default class SeedPlan extends BaseCommand {
|
||||
public static commandName = 'seed:plan'
|
||||
public static description = 'Plans article ranges of groups for seeding nodes to lease.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.spread({ description: 'Groups' })
|
||||
declare groups: string[]
|
||||
|
||||
@flags.number({ description: 'How many days back (default: everything the server has)' })
|
||||
declare days?: number
|
||||
|
||||
@flags.number({ description: 'Articles per range', default: 2_000_000 })
|
||||
declare size: number
|
||||
|
||||
@flags.number({ description: 'Leased before ranges of a lower priority (default 0)', default: 0 })
|
||||
declare priority: number
|
||||
|
||||
public async run() {
|
||||
try {
|
||||
const summary = await SeedOrchestrator.plan(this.groups, { days: this.days, size: this.size, priority: this.priority })
|
||||
for (const s of summary) {
|
||||
this.logger.info(
|
||||
`${s.group}: ${s.ranges.toLocaleString('en')} new ranges in ${s.gaps} gap(s) not yet planned, articles ${s.from.toLocaleString('en')}–${s.to.toLocaleString('en')}.`
|
||||
)
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,126 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import nntpConfig from '#config/nntp'
|
||||
import SeedService from '#services/SeedService'
|
||||
import QueueService from '#services/QueueService'
|
||||
import { isDeferred } from '#services/DeferredService'
|
||||
import { XoverClient } from '#lib/nntp/xover_client'
|
||||
import { runRange } from '#lib/seed/runner'
|
||||
import type { SeedBatch } from '#lib/seed/wire'
|
||||
|
||||
/**
|
||||
* Edge aggregation (CONCEPT.md D37) on this machine: indexes a range of a
|
||||
* group's history over one NNTP connection (gzip-compressed overviews when
|
||||
* the server has them), grouping headers into files in memory and storing
|
||||
* one row per file directly. seed:node does the same for a remote node,
|
||||
* through the seed API. Header-only, like the backfill.
|
||||
*
|
||||
* node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000
|
||||
* node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run
|
||||
*/
|
||||
export default class SeedRange extends BaseCommand {
|
||||
public static commandName = 'seed:range'
|
||||
public static description = "Indexes a range of a group's headers by aggregating them into files on this node."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.string({ description: 'Group' })
|
||||
declare group: string
|
||||
|
||||
@flags.string({ description: 'First article number' })
|
||||
declare start: string
|
||||
|
||||
@flags.number({ description: 'Articles to index', default: 1_000_000 })
|
||||
declare count: number
|
||||
|
||||
@flags.number({ description: 'Articles per XOVER', default: 20_000 })
|
||||
declare chunk: number
|
||||
|
||||
@flags.number({ description: 'Batches without a new part before an open file is given up as fragments', default: 5 })
|
||||
declare idle: number
|
||||
|
||||
@flags.boolean({ description: 'Fetch and aggregate only; store nothing' })
|
||||
declare dryRun: boolean
|
||||
|
||||
@flags.boolean({ description: 'Ask the server for gzip-compressed overviews', default: true })
|
||||
declare compress: boolean
|
||||
|
||||
public async run() {
|
||||
const first = Number(this.start)
|
||||
if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`)
|
||||
const last = first + this.count - 1
|
||||
const stats = { headers: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, dropped: 0, fragments: 0, inspections: 0 }
|
||||
let storeMs = 0
|
||||
const cpuStart = process.cpuUsage()
|
||||
const started = Date.now()
|
||||
|
||||
const client = await XoverClient.connect({
|
||||
host: nntpConfig.host,
|
||||
port: nntpConfig.port,
|
||||
secure: nntpConfig.secure,
|
||||
servername: nntpConfig.tlsServername,
|
||||
user: nntpConfig.user,
|
||||
password: nntpConfig.password?.release(),
|
||||
timeoutMs: nntpConfig.commandTimeoutMs,
|
||||
compress: this.compress,
|
||||
})
|
||||
this.logger.info(`Connected${client.compressed ? ', overviews gzip-compressed' : ''}.`)
|
||||
try {
|
||||
await runRange({
|
||||
client,
|
||||
group: this.group,
|
||||
start: first,
|
||||
end: last,
|
||||
chunk: this.chunk,
|
||||
idle: this.idle,
|
||||
submitEvery: 1,
|
||||
isDeferred,
|
||||
sink: async (batch: SeedBatch) => {
|
||||
stats.headers += batch.headers
|
||||
stats.obfuscated += batch.obfuscated
|
||||
stats.deferred += batch.deferred
|
||||
stats.complete += batch.files.length
|
||||
stats.fragments += batch.fragments.length
|
||||
if (!this.dryRun) {
|
||||
const t = Date.now()
|
||||
const stored = await SeedService.store(batch.files)
|
||||
await SeedService.storeFragments(batch.fragments, { start: first, end: last })
|
||||
storeMs += Date.now() - t
|
||||
stats.files += stored.files
|
||||
stats.existing += stored.existing
|
||||
stats.dropped += stored.dropped
|
||||
stats.inspections += stored.inspections
|
||||
}
|
||||
this.progress(stats, started)
|
||||
},
|
||||
})
|
||||
} finally {
|
||||
client.close()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
|
||||
const seconds = (Date.now() - started) / 1000
|
||||
const cpu = process.cpuUsage(cpuStart)
|
||||
const n = (x: number) => x.toLocaleString('en')
|
||||
this.logger.info(
|
||||
`${this.group} ${first}-${last}: ${n(stats.headers)} headers in ${seconds.toFixed(0)} s (${n(Math.round(stats.headers / seconds))}/s). ` +
|
||||
`${n(stats.obfuscated)} obfuscated, ${n(stats.deferred)} deferred. ` +
|
||||
`${n(stats.complete)} complete files (${n(stats.files)} new, ${n(stats.existing)} known, ${n(stats.dropped)} garbage), ` +
|
||||
`${n(stats.fragments)} fragment segments, ${n(stats.inspections)} queued for inspection.`
|
||||
)
|
||||
this.logger.info(`Store ${(storeMs / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`)
|
||||
}
|
||||
|
||||
private lastReport = 0
|
||||
private progress(stats: Record<string, number>, started: number) {
|
||||
if (Date.now() - this.lastReport < 30_000) return
|
||||
this.lastReport = Date.now()
|
||||
const seconds = (Date.now() - started) / 1000
|
||||
this.logger.info(
|
||||
`${stats.headers.toLocaleString('en')} headers (${Math.round(stats.headers / seconds).toLocaleString('en')}/s), ` +
|
||||
`${stats.complete.toLocaleString('en')} files, ${stats.fragments.toLocaleString('en')} fragment segments.`
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -1,26 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import SeedOrchestrator from '#services/SeedOrchestrator'
|
||||
|
||||
/** Seeding progress per range status and per node (CONCEPT.md D38). */
|
||||
export default class SeedStatus extends BaseCommand {
|
||||
public static commandName = 'seed:status'
|
||||
public static description = 'Shows seeding progress per range status and per node.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
const status = await SeedOrchestrator.status()
|
||||
const n = (x: unknown) => Number(x ?? 0).toLocaleString('en')
|
||||
for (const r of status.ranges) {
|
||||
this.logger.info(`${r.status}: ${n(r.ranges)} ranges, ${n(r.reached)} of ${n(r.articles)} articles reached, ${n(r.headers)} headers, ${n(r.files)} files`)
|
||||
}
|
||||
for (const node of status.nodes) {
|
||||
this.logger.info(`node ${node.name}${node.disabled ? ' (disabled)' : ''}: ${node.leases} leases, ${n(node.headers)} headers, last seen ${node.last_seen_at ?? 'never'}`)
|
||||
}
|
||||
this.logger.info(`Headers in ranges finished in the last hour: ${n(status.headersDoneLastHour)}`)
|
||||
if (status.paused) this.logger.warning(`No new leases: ${status.paused} (BACKFILL_MAX_NVME_GB / BACKFILL_MAX_HDD_GB).`)
|
||||
}
|
||||
}
|
||||
@@ -1,37 +0,0 @@
|
||||
import { BaseCommand, args, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import SeedOrchestrator from '#services/SeedOrchestrator'
|
||||
|
||||
/**
|
||||
* Creates a seeding node, or gives it a new token (the old one stops
|
||||
* working), and prints the token once (CONCEPT.md D38).
|
||||
*
|
||||
* node ace seed:token hetzner-1
|
||||
* node ace seed:token hetzner-1 --disable
|
||||
*/
|
||||
export default class SeedToken extends BaseCommand {
|
||||
public static commandName = 'seed:token'
|
||||
public static description = 'Creates a seeding node or replaces its token; prints the token once.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@args.string({ description: 'Node name' })
|
||||
declare name: string
|
||||
|
||||
@flags.boolean({ description: 'Disable the node instead (its leases pass to others after they expire)' })
|
||||
declare disable: boolean
|
||||
|
||||
public async run() {
|
||||
if (this.disable) {
|
||||
const n = await db.from('seed_nodes').where('name', this.name).update({ disabled: true })
|
||||
this.logger.info(n ? `${this.name}: disabled.` : `${this.name}: no such node.`)
|
||||
return
|
||||
}
|
||||
const token = await SeedOrchestrator.issueToken(this.name)
|
||||
this.logger.info(`${this.name}: token (shown once, keep it on the node only):`)
|
||||
console.log(token)
|
||||
}
|
||||
}
|
||||
@@ -1,25 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import SpotService from '#services/SpotService'
|
||||
|
||||
/** Passwords from the titles and descriptions of spots read so far, onto their releases. */
|
||||
export default class SpotPasswords extends BaseCommand {
|
||||
public static commandName = 'spot:passwords'
|
||||
public static description = "Takes archive passwords from spots' titles and descriptions onto their releases."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
const spots = await db
|
||||
.from('spots')
|
||||
.whereNotNull('release_id')
|
||||
.where((q) => q.whereNotNull('description').orWhere('title', 'like', '%{{%}}%'))
|
||||
.select('message_id')
|
||||
let found = 0
|
||||
for (const spot of spots) if (await SpotService.applyPassword(spot.message_id)) found++
|
||||
this.logger.info(`${found} of ${spots.length} spots name a password in their title or description.`)
|
||||
}
|
||||
}
|
||||
@@ -1,55 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import env from '#start/env'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import NntpService from '#services/NntpService'
|
||||
import SpotService from '#services/SpotService'
|
||||
|
||||
const BATCH = 500
|
||||
|
||||
/**
|
||||
* Reads the article and NZB of listed spots that were listed from their
|
||||
* overview only (LOG.md 2026-10-08 (10)): description, the NZB's password or
|
||||
* one in the spot's text, placeholder NZBs. Newest first; can be stopped and
|
||||
* started again (worker:spots does the same when idle, a batch at a time).
|
||||
* Connections: NNTP_POOL_SIZE of this process, --concurrency at a time.
|
||||
*/
|
||||
export default class SpotPostprocess extends BaseCommand {
|
||||
public static commandName = 'spot:postprocess'
|
||||
public static description = "Reads listed spots' articles and NZBs for their passwords and details."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'Spots read at a time (default: NNTP_POOL_SIZE)' })
|
||||
declare concurrency?: number
|
||||
|
||||
@flags.number({ description: 'Stop after this many spots' })
|
||||
declare limit?: number
|
||||
|
||||
public async run() {
|
||||
const concurrency = this.concurrency ?? env.get('NNTP_POOL_SIZE', 4)
|
||||
const [{ count }] = await db.from('spots').where('status', 'listed').whereNull('nzb_read_at').count('* as count')
|
||||
this.logger.info(`${count} listed spots to read, ${concurrency} at a time.`)
|
||||
const totals: Record<string, number> = {}
|
||||
const started = Date.now()
|
||||
let done = 0
|
||||
try {
|
||||
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
||||
const counts = await SpotService.readListings(Math.min(BATCH, this.limit ? this.limit - done : BATCH), concurrency)
|
||||
const read = Object.values(counts).reduce((sum, n) => sum + n, 0)
|
||||
if (read === 0) break
|
||||
for (const [k, v] of Object.entries(counts)) totals[k] = (totals[k] ?? 0) + v
|
||||
done += read
|
||||
const rate = done / ((Date.now() - started) / 1000)
|
||||
this.logger.info(`${done} read (${rate.toFixed(1)}/s): ${Object.entries(totals).map(([k, v]) => `${v} ${k}`).join(', ')}.`)
|
||||
// Usenet unreachable: everything was given back; don't spin on it.
|
||||
if (counts.retry === read) await Shutdown.sleep(30_000)
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import NntpService from '#services/NntpService'
|
||||
import SpotService from '#services/SpotService'
|
||||
|
||||
/**
|
||||
* Imported spots whose files sit in a release that isn't the spot's own
|
||||
* (another source, or another name): their NZB is read again, which moves
|
||||
* the files into the spot's release under its title (CONCEPT.md D29).
|
||||
*/
|
||||
export default class SpotReimport extends BaseCommand {
|
||||
public static commandName = 'spot:reimport'
|
||||
public static description = "Moves imported spots' files into their own release, named after the spot."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.boolean({ description: 'Only count them' })
|
||||
declare dryRun: boolean
|
||||
|
||||
public async run() {
|
||||
const spots = await db
|
||||
.from('spots as s')
|
||||
.join('releases as r', 'r.id', 's.release_id')
|
||||
.where('s.status', 'imported')
|
||||
.where((q) => q.whereNot('r.source', 'spot').orWhereRaw('r.name <> s.title'))
|
||||
.orderBy('s.posted_at', 'desc')
|
||||
.select('s.*', 'r.name as release_name')
|
||||
this.logger.info(`${spots.length} imported spots are in a release that isn't their own.`)
|
||||
if (this.dryRun) return
|
||||
let moved = 0
|
||||
let failed = 0
|
||||
try {
|
||||
for (const spot of spots) {
|
||||
try {
|
||||
const releaseId = await SpotService.reimport(spot)
|
||||
if (releaseId) moved++
|
||||
else failed++
|
||||
} catch (error: any) {
|
||||
failed++
|
||||
this.logger.warning(`${spot.title}: ${error.message}`)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
this.logger.info(`${moved} spots re-imported, ${failed} without a usable NZB (dummy, gone) or failed.`)
|
||||
}
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import fs from 'node:fs/promises'
|
||||
import path from 'node:path'
|
||||
import queueConfig from '#config/queue'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import QueueService from '#services/QueueService'
|
||||
import env from '#start/env'
|
||||
import NntpService from '#services/NntpService'
|
||||
import IngestService, { segmentFromYenc } from '#services/IngestService'
|
||||
import { parseYencMeta, YencError } from '#lib/yenc/index'
|
||||
import { firstChunkMd5 } from '#services/ContentService'
|
||||
|
||||
export default class BodyWorker extends BaseCommand {
|
||||
public static commandName = 'worker:body'
|
||||
public static description = 'Starts a worker to process article bodies for yEnc metadata.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting body worker...')
|
||||
|
||||
const pool = NntpService
|
||||
|
||||
const worker = new Worker('body-queue', async (job) => {
|
||||
const { header, group, articleNumber } = job.data
|
||||
const messageId = header['message-id']
|
||||
this.logger.debug(`Processing header with unparsable subject: ${header.subject}`)
|
||||
|
||||
let conn
|
||||
let failed = false
|
||||
try {
|
||||
conn = await pool.acquire()
|
||||
const bodyBuffer = await pool.fetchBody(conn, messageId)
|
||||
|
||||
let meta
|
||||
try {
|
||||
meta = parseYencMeta(bodyBuffer)
|
||||
} catch (parseError: any) {
|
||||
if (parseError instanceof YencError && bodyBuffer.indexOf('=ybegin ') === -1) {
|
||||
this.logger.debug(`Not a yEnc post, skipping: ${messageId}`)
|
||||
return
|
||||
}
|
||||
this.logger.error(`Failed to parse yEnc data for message ID ${messageId}. Dumping buffer.`)
|
||||
const debugDir = path.join(this.app.appRoot.pathname, 'debug')
|
||||
await fs.mkdir(debugDir, { recursive: true })
|
||||
const timestamp = new Date().toISOString().replace(/:/g, '-')
|
||||
const dumpFile = path.join(debugDir, `body-error-${timestamp}-${messageId.replace(/[<>]/g, '')}.bin`)
|
||||
await fs.writeFile(dumpFile, bodyBuffer)
|
||||
this.logger.error(`Problematic body buffer saved to: ${dumpFile}`)
|
||||
throw parseError
|
||||
}
|
||||
|
||||
pool.release(conn)
|
||||
conn = undefined
|
||||
|
||||
const segment = segmentFromYenc(header, group, meta, articleNumber)
|
||||
if (segment?.number === 1) segment.md5_16k = firstChunkMd5(bodyBuffer, meta.begin.size)
|
||||
if (!segment) {
|
||||
this.logger.warning(`Inconsistent yEnc part numbering in ${messageId}, skipping.`)
|
||||
return
|
||||
}
|
||||
const completed = await IngestService.ingest([segment])
|
||||
this.logger.debug(
|
||||
`Stored part ${segment.number}/${segment.totalParts} of "${segment.name}"` +
|
||||
(completed.length > 0 ? ' (file complete)' : '')
|
||||
)
|
||||
} catch (error: any) {
|
||||
failed = true
|
||||
this.logger.error(`Error in body worker for message ID ${messageId}: ${error.message}`)
|
||||
throw error
|
||||
} finally {
|
||||
if (conn) {
|
||||
pool.release(conn, failed)
|
||||
}
|
||||
}
|
||||
}, {
|
||||
connection: queueConfig.connection,
|
||||
// Each job holds one NNTP connection while downloading, so keep this <= the pool size.
|
||||
concurrency: env.get('BODY_CONCURRENCY', 1),
|
||||
})
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`Body job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('Body worker started and listening for jobs.')
|
||||
await Shutdown.wait()
|
||||
this.logger.info('Shutting down: finishing active jobs...')
|
||||
await worker.close()
|
||||
await NntpService.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
@@ -1,75 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import env from '#start/env'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import QueueService from '#services/QueueService'
|
||||
import NntpService from '#services/NntpService'
|
||||
import ContentService, { describeContent } from '#services/ContentService'
|
||||
import ArchiveWalkService from '#services/ArchiveWalkService'
|
||||
import File from '#models/file'
|
||||
|
||||
export default class CollectionWorker extends BaseCommand {
|
||||
public static commandName = 'worker:collection'
|
||||
public static description = 'Starts a worker to process file collections (e.g., RAR archives).'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting collection worker...')
|
||||
|
||||
const worker = new Worker('collection-queue', async (job) => {
|
||||
const { fileId } = job.data
|
||||
this.logger.debug(`Processing file ID ${fileId} for collection.`)
|
||||
|
||||
const file = await File.find(fileId)
|
||||
if (!file) {
|
||||
this.logger.error(`File with ID ${fileId} not found in the database.`)
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const content = await ContentService.inspect(fileId)
|
||||
for (const line of describeContent(file.filename, content)) this.logger.info(line)
|
||||
} catch (error: any) {
|
||||
this.logger.warning(`Content inspection of "${file.filename}" failed: ${error.message}`)
|
||||
}
|
||||
}, {
|
||||
connection: queueConfig.connection,
|
||||
// Each job holds one NNTP connection while downloading; keep <= the pool size.
|
||||
concurrency: env.get('COLLECTION_CONCURRENCY', 1),
|
||||
})
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`Collection job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
// Complete archive listings (D40): background work beside inspection, on the same NNTP pool.
|
||||
const walkConcurrency = env.get('ARCHIVE_WALK_CONCURRENCY', 1)
|
||||
const walker = walkConcurrency > 0
|
||||
? new Worker('archive-walk-queue', async (job) => {
|
||||
const o = await ArchiveWalkService.walk(Number(job.data.releaseId))
|
||||
if (o.skipped) return o
|
||||
this.logger.info(
|
||||
`Archive walk of release ${o.releaseId}: ${o.before} → ${o.entries} entries from ${o.segments} segments` +
|
||||
(o.complete ? '' : ` (stopped: ${o.stopped})`) +
|
||||
(o.renamed ? `, renamed to "${o.renamed}"` : '')
|
||||
)
|
||||
return o
|
||||
}, { connection: queueConfig.connection, concurrency: walkConcurrency })
|
||||
: null
|
||||
walker?.on('failed', (job, err) => {
|
||||
this.logger.error(`Archive walk job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('Collection worker started and listening for jobs.')
|
||||
await Shutdown.wait()
|
||||
this.logger.info('Shutting down: finishing active jobs...')
|
||||
await Promise.all([worker.close(), walker?.close()])
|
||||
await NntpService.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import env from '#start/env'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import NntpService from '#services/NntpService'
|
||||
import QueueService from '#services/QueueService'
|
||||
import DeferredService from '#services/DeferredService'
|
||||
|
||||
const IDLE_MS = 60_000
|
||||
|
||||
/**
|
||||
* Works through `deferred_files` (CONCEPT.md D28): one body per file, and
|
||||
* small files completely. Not started by default (compose profile
|
||||
* `deferred`); run it when there is bandwidth to spare.
|
||||
*/
|
||||
export default class DeferredWorker extends BaseCommand {
|
||||
public static commandName = 'worker:deferred'
|
||||
public static description = 'Samples deferred files (e.g. ngPost uploads): one article body per file.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'Stop after this many files (default: keep running)' })
|
||||
declare limit?: number
|
||||
|
||||
public async run() {
|
||||
const concurrency = env.get('DEFERRED_CONCURRENCY', 4)
|
||||
this.logger.info(`Deferred worker started (concurrency ${concurrency}).`)
|
||||
const counts: Record<string, number> = {}
|
||||
let done = 0
|
||||
try {
|
||||
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
||||
try {
|
||||
const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity))
|
||||
if (batch.length === 0) {
|
||||
if (this.limit) break
|
||||
await Shutdown.sleep(IDLE_MS)
|
||||
continue
|
||||
}
|
||||
let next = 0
|
||||
await Promise.all(
|
||||
Array.from({ length: concurrency }, async () => {
|
||||
while (next < batch.length) {
|
||||
const status = await DeferredService.process(batch[next++])
|
||||
counts[status] = (counts[status] ?? 0) + 1
|
||||
}
|
||||
})
|
||||
)
|
||||
done += batch.length
|
||||
this.logger.info(`${done} deferred files processed: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`)
|
||||
} catch (err: any) {
|
||||
// E.g. Postgres restarting: wait and claim again, instead of ending the loop (LOG.md 2026-10-07 (3)).
|
||||
this.logger.error(`Round failed, retrying in 30 s: ${err.message}`)
|
||||
await Shutdown.sleep(30_000)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,118 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import env from '#start/env'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import indexerConfig from '#config/indexer'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import QueueService from '#services/QueueService'
|
||||
import NntpService from '#services/NntpService'
|
||||
import FetchRangeService from '#services/FetchRangeService'
|
||||
import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService'
|
||||
import DeferredService, { bodyJob, isDeferred, type DeferredArticle } from '#services/DeferredService'
|
||||
import SpotService, { isSpotGroup } from '#services/SpotService'
|
||||
import StagingService from '#services/StagingService'
|
||||
|
||||
export default class FetchWorker extends BaseCommand {
|
||||
public static commandName = 'worker:fetch'
|
||||
public static description = 'Starts a worker that fetches headers (XOVER) and stores them as binaries/parts.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting fetch worker...')
|
||||
|
||||
const pool = NntpService
|
||||
const bodyQueue = QueueService.bodyQueue
|
||||
|
||||
const worker = new Worker('nntp-fetch-queue', async (job) => {
|
||||
const { groupName, startId, endId } = job.data
|
||||
this.logger.info(`Processing fetch job for ${groupName}, articles ${startId}-${endId}`)
|
||||
|
||||
let overviews: [number, Record<string, string>][]
|
||||
let conn
|
||||
let failed = false
|
||||
try {
|
||||
conn = await pool.acquire()
|
||||
await pool.timed(conn.group(groupName), `GROUP ${groupName}`)
|
||||
const reply = await pool.timed<any>(conn.xover(startId, endId), `XOVER ${groupName} ${startId}-${endId}`)
|
||||
// An empty reply parses as one line without fields ([null, {}]).
|
||||
overviews = reply.overviews.filter(([number, header]: [number | null, Record<string, string>]) => number !== null && header?.['message-id'])
|
||||
} catch (error: any) {
|
||||
failed = true
|
||||
this.logger.error(`Error fetching headers for ${groupName}: ${error.message}`)
|
||||
throw error
|
||||
} finally {
|
||||
if (conn) {
|
||||
pool.release(conn, failed)
|
||||
}
|
||||
}
|
||||
|
||||
if (isSpotGroup(groupName)) {
|
||||
const spots = await SpotService.recordHeaders(groupName, overviews)
|
||||
if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)
|
||||
this.logger.info(`${groupName} ${startId}-${endId}: ${overviews.length} headers, ${spots} spots queued.`)
|
||||
return
|
||||
}
|
||||
|
||||
const group = await db.from('groups').where('name', groupName).first()
|
||||
// Backfill is header-only: old bodies would be far more than the link carries (CONCEPT.md D32).
|
||||
const fetchBodies = Boolean(group?.fetch_bodies) && !job.data.backfill
|
||||
|
||||
const segments: IngestSegment[] = []
|
||||
const obfuscated: { articleNumber: number; header: Record<string, string> }[] = []
|
||||
const deferred: DeferredArticle[] = []
|
||||
let skipped = 0
|
||||
for (const [articleNumber, header] of overviews) {
|
||||
if (!header) continue
|
||||
const segment = segmentFromHeader(header, groupName, articleNumber)
|
||||
if (segment) segments.push(segment)
|
||||
// Recorded per file (or as article ranges) in every group, bodies or not; processed later if at all.
|
||||
else if (isDeferred(header)) deferred.push({ articleNumber, header })
|
||||
else if (!fetchBodies || indexerConfig.skipBodyMessageIds.some((re) => re.test(header['message-id'] ?? ''))) skipped++
|
||||
else obfuscated.push({ articleNumber, header })
|
||||
}
|
||||
|
||||
// Lone segments of new files wait in staging until their file proves real (D35).
|
||||
const { ingest, staged } = StagingService.enabled(groupName)
|
||||
? await StagingService.filter(segments, { counters: StagingService.countersEnabled(groupName) })
|
||||
: { ingest: segments, staged: 0 }
|
||||
const completed = await IngestService.ingest(ingest)
|
||||
const deferredFiles = await DeferredService.record(groupName, deferred)
|
||||
|
||||
if (obfuscated.length > 0) {
|
||||
await bodyQueue.addBulk(obfuscated.map(({ articleNumber, header }) => bodyJob(header, groupName, articleNumber)))
|
||||
}
|
||||
|
||||
if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)
|
||||
|
||||
this.logger.info(
|
||||
`${groupName} ${startId}-${endId}: ${overviews.length} headers, ${ingest.length} stored${staged ? `, ${staged} staged` : ''}, ` +
|
||||
`${obfuscated.length} sent to body-queue${skipped ? ` (${skipped} skipped)` : ''}` +
|
||||
`${deferred.length ? `, ${deferred.length} deferred (${deferredFiles} files)` : ''}, ${completed.length} files completed.`
|
||||
)
|
||||
}, {
|
||||
connection: queueConfig.connection,
|
||||
// XOVER is quick; most of a job is storing. Each concurrent job uses one DB connection (DB_POOL_MAX).
|
||||
concurrency: env.get('FETCH_CONCURRENCY', 5),
|
||||
})
|
||||
|
||||
worker.on('failed', async (job, err) => {
|
||||
this.logger.error(`Fetch job ${job?.id} failed for group ${job?.data.groupName}: ${err.message}`)
|
||||
// Only after the last attempt; the scheduler re-queues the range later.
|
||||
if (job?.data.rangeId && job.attemptsMade >= (job.opts.attempts ?? 1)) {
|
||||
await FetchRangeService.markFailed(job.data.rangeId, err.message).catch(() => {})
|
||||
}
|
||||
})
|
||||
|
||||
this.logger.info('Fetch worker started and listening for jobs.')
|
||||
await Shutdown.wait()
|
||||
this.logger.info('Shutting down: finishing active jobs...')
|
||||
await worker.close()
|
||||
await NntpService.shutdown()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
@@ -1,54 +0,0 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import env from '#start/env'
|
||||
import BinaryFileService from '#services/BinaryFileService'
|
||||
import queueConfig from '#config/queue'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import QueueService from '#services/QueueService'
|
||||
|
||||
export default class FileWorker extends BaseCommand {
|
||||
public static commandName = 'worker:file'
|
||||
public static description = 'Starts a worker that turns completed binaries into files.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting file worker...')
|
||||
|
||||
const worker = new Worker('file-queue', async (job) => {
|
||||
// A batch of completed binaries ('process-files'), or one from before batching ('process-file').
|
||||
const ids: number[] = (job.data.binaryIds ?? [job.data.binaryId]).map(Number)
|
||||
const result = await BinaryFileService.store(ids)
|
||||
if (result.saved || result.existing) {
|
||||
this.logger.info(
|
||||
`Files: ${result.saved} saved, ${result.existing} existed already` +
|
||||
(result.skipped ? `, ${result.skipped} skipped` : '') +
|
||||
(result.inspections ? `, ${result.inspections} queued for inspection` : '') +
|
||||
` (${ids.length} binaries).`
|
||||
)
|
||||
}
|
||||
// The rest is stored; the job's retry picks these up (the others are linked now and skipped).
|
||||
if (result.incomplete.length > 0) {
|
||||
throw new Error(`${result.incomplete.length} of ${ids.length} binaries don't have all their parts yet.`)
|
||||
}
|
||||
}, {
|
||||
connection: queueConfig.connection,
|
||||
// One job at a time could not keep up with the backfill (130k waiting); jobs are per binary and
|
||||
// the file insert ignores duplicates, so they can run side by side.
|
||||
concurrency: env.get('FILE_CONCURRENCY', 8),
|
||||
})
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`File job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('File worker started and listening for jobs.')
|
||||
await Shutdown.wait()
|
||||
this.logger.info('Shutting down: finishing active jobs...')
|
||||
await worker.close()
|
||||
await QueueService.closeAll()
|
||||
}
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import env from '#start/env'
|
||||
import Shutdown from '#services/Shutdown'
|
||||
import NntpService from '#services/NntpService'
|
||||
import SpotService from '#services/SpotService'
|
||||
|
||||
const IDLE_MS = 30_000
|
||||
/** Spots claimed at a time; most of a history backfill is listed in bulk. */
|
||||
const BATCH = 500
|
||||
/** Imported spots whose details are read per idle round. */
|
||||
const DETAILS_BATCH = 50
|
||||
/** Listed spots whose article and NZB are read per idle round (passwords, dummies). */
|
||||
const LISTINGS_BATCH = 50
|
||||
|
||||
/**
|
||||
* Works through queued Spotnet spots (CONCEPT.md D29): verifies each one,
|
||||
* downloads its compressed NZB (a few KB to a few MB) and imports it. When
|
||||
* idle, reads what listed spots were listed without: details, passwords.
|
||||
*/
|
||||
export default class SpotsWorker extends BaseCommand {
|
||||
public static commandName = 'worker:spots'
|
||||
public static description = 'Reads Spotnet spots: verifies them and imports their NZBs.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
@flags.number({ description: 'Stop after this many spots (default: keep running)' })
|
||||
declare limit?: number
|
||||
|
||||
public async run() {
|
||||
const concurrency = env.get('SPOT_CONCURRENCY', 4)
|
||||
this.logger.info(`Spot worker started (concurrency ${concurrency}).`)
|
||||
const counts: Record<string, number> = {}
|
||||
let done = 0
|
||||
try {
|
||||
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
||||
try {
|
||||
const batch = await SpotService.claim(Math.min(BATCH, this.limit ? this.limit - done : Infinity))
|
||||
if (batch.length === 0) {
|
||||
if (this.limit) break
|
||||
// Idle: read the details of imported spots that don't have them yet (newest first), then
|
||||
// the article and NZB of listed spots (their passwords; LOG.md 2026-10-08 (10)).
|
||||
if ((await SpotService.fillDetails(DETAILS_BATCH)) > 0) continue
|
||||
const listings = await SpotService.readListings(LISTINGS_BATCH, concurrency)
|
||||
const read = Object.values(listings).reduce((sum, n) => sum + n, 0)
|
||||
if (read === 0 || listings.retry === read) await Shutdown.sleep(IDLE_MS)
|
||||
continue
|
||||
}
|
||||
// Old, verified spots are listed in bulk; the rest need their header (and maybe NZB) read.
|
||||
const listable = batch.filter((spot) => SpotService.isListable(spot))
|
||||
const single = batch.filter((spot) => !SpotService.isListable(spot))
|
||||
counts.listed = (counts.listed ?? 0) + (await SpotService.list(listable))
|
||||
let next = 0
|
||||
await Promise.all(
|
||||
Array.from({ length: concurrency }, async () => {
|
||||
while (next < single.length) {
|
||||
const status = await SpotService.process(single[next++])
|
||||
counts[status] = (counts[status] ?? 0) + 1
|
||||
}
|
||||
})
|
||||
)
|
||||
done += batch.length
|
||||
this.logger.info(`${done} spots: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`)
|
||||
} catch (err: any) {
|
||||
// E.g. Postgres restarting: wait and claim again, instead of ending the loop (LOG.md 2026-10-07 (3)).
|
||||
this.logger.error(`Round failed, retrying in 30 s: ${err.message}`)
|
||||
await Shutdown.sleep(30_000)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await NntpService.shutdown()
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user