From fd0f2bde75d4d64bcaf53c66ecce5a43eab9d741 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 10:41:16 +0200 Subject: [PATCH] Rebuild: the live tail coordinator:run plans the articles after each group's last range every TAIL_EVERY_SECONDS (120): one LIST ACTIVE per name prefix gives every group's newest article, and new articles become ranges at priority -1, leased and prefetched before any history. A group gets a tail range once it has TAIL_MIN_ARTICLES (100k, a chunk) new or its last planning is TAIL_MAX_DELAY_MINUTES (15) ago, so quiet groups don't make a range every round. History (planAll, hourly) is now only for groups not planned yet. ranges:plan --tail plans one round. Found on the way: with XFEATURE COMPRESS GZIP on, Eweka compresses LIST ACTIVE's reply too ("215 NewsGroups Follow [COMPRESS=GZIP]"), and listActive waited for a plain end that never came (groups:add --top too). It reads the compressed reply now; the fake server in the test does both. Co-Authored-By: Claude Opus 5.5 --- app/services/planner.ts | 75 +++++++++++++++++++++++++++- commands/coordinator_run.ts | 8 +++ commands/ranges_plan.ts | 13 +++-- config/coordinator.ts | 14 ++++++ lib/nntp/xover_client.ts | 4 +- start/env.ts | 3 ++ tests/unit/coordinator/tail.spec.ts | 42 ++++++++++++++++ tests/unit/nntp/xover_client.spec.ts | 14 ++++++ 8 files changed, 166 insertions(+), 7 deletions(-) create mode 100644 tests/unit/coordinator/tail.spec.ts diff --git a/app/services/planner.ts b/app/services/planner.ts index 8af9888..694f448 100644 --- a/app/services/planner.ts +++ b/app/services/planner.ts @@ -54,8 +54,9 @@ class Planner { return rows.length } + /** History for groups not planned yet (new ones); what comes after is the tail's (planTail). */ async planAll(): Promise<{ groups: number; ranges: number }> { - const groups = await db.from('groups').where('active', true).orderBy('priority').select('id', 'name') + const groups = await db.from('groups').where('active', true).whereNull('planned_at').orderBy('priority').select('id', 'name') let ranges = 0 for (const group of groups) { // One group failing (e.g. gone from the server) doesn't stop the others. @@ -67,6 +68,78 @@ class Planner { } return { groups: groups.length, ranges } } + + /** + * The live tail: ranges for the articles after each planned group's last + * range, at the tail's priority (before any history). The newest article + * numbers come from LIST ACTIVE, one command per name prefix (GROUP per + * group took minutes for 1000 groups). + */ + async planTail(now = new Date()): Promise<{ groups: number; ranges: number }> { + const groups = await db.from('groups').where('active', true).whereNotNull('planned_at').select('id', 'name', 'planned_at') + if (groups.length === 0) return { groups: 0, ranges: 0 } + const { rows: ends } = await db.rawQuery(`SELECT DISTINCT ON (group_id) group_id, end_id FROM ranges ORDER BY group_id, start_id DESC`) + const plannedTo = new Map(ends.map((r: any) => [Number(r.group_id), Number(r.end_id)])) + const highs = new Map() + await NntpPool.use(async (client) => { + for (const pattern of new Set(groups.map((g) => wildmatOf(g.name)))) { + for (const g of await client.listActive(pattern)) highs.set(g.name, g.high) + } + }) + const rows = tailRanges( + groups.map((g) => ({ id: Number(g.id), plannedTo: plannedTo.get(Number(g.id)) ?? null, plannedAt: new Date(g.planned_at), high: highs.get(g.name) ?? null })), + now, + { rangeArticles: coordinatorConfig.rangeArticles, ...coordinatorConfig.tail } + ) + if (rows.length === 0) return { groups: 0, ranges: 0 } + await db.transaction(async (trx) => { + await trx.table('ranges').multiInsert(rows) + for (const id of new Set(rows.map((r) => r.group_id))) { + const high = Math.max(...rows.filter((r) => r.group_id === id).map((r) => r.end_id)) + await trx.from('groups').where('id', id).update({ high, planned_at: now }) + } + }) + return { groups: new Set(rows.map((r) => r.group_id)).size, ranges: rows.length } + } +} + +/** A LIST ACTIVE pattern for a group: its first two name parts (`alt.binaries.*`), or the name itself. */ +export function wildmatOf(name: string): string { + const parts = name.split('.') + return parts.length > 2 ? `${parts[0]}.${parts[1]}.*` : name +} + +export interface TailGroup { + id: number + /** End of its last range (null: none, then it has no tail yet). */ + plannedTo: number | null + /** When its last range was planned. */ + plannedAt: Date + /** The server's newest article (null: not listed). */ + high: number | null +} + +/** + * The tail's new ranges: per group, the articles after its last range, in + * ranges of at most `rangeArticles`, once there are `minArticles` of them or + * the group's last planning is `maxDelayMinutes` ago. + */ +export function tailRanges( + groups: TailGroup[], + now: Date, + options: { rangeArticles: number; minArticles: number; maxDelayMinutes: number; priority: number } +) { + const rows: { group_id: number; start_id: number; end_id: number; next_id: number; priority: number }[] = [] + for (const g of groups) { + if (g.plannedTo === null || g.high === null || g.high <= g.plannedTo) continue + const waiting = g.high - g.plannedTo + const waited = (now.getTime() - g.plannedAt.getTime()) / 60_000 + if (waiting < options.minArticles && waited < options.maxDelayMinutes) continue + for (let start = g.plannedTo + 1; start <= g.high; start += options.rangeArticles) { + rows.push({ group_id: g.id, start_id: start, end_id: Math.min(start + options.rangeArticles - 1, g.high), next_id: start, priority: options.priority }) + } + } + return rows } export default new Planner() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index f1c9a0d..c00d41b 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -6,6 +6,7 @@ import Planner from '#services/planner' import FragmentMerger from '#services/fragment_merger' import Settler from '#services/settler' import NntpPool from '#services/nntp_pool' +import coordinatorConfig from '#config/coordinator' const TICK_MS = 2000 const PLAN_EVERY_MS = 60 * 60 * 1000 @@ -55,7 +56,9 @@ export default class CoordinatorRun extends BaseCommand { await sleep(TICK_MS) } })() + // History for new groups hourly; the tail every few minutes. let plannedAt = 0 + let tailedAt = 0 let mergedAt = 0 while (!stopping) { try { @@ -66,6 +69,11 @@ export default class CoordinatorRun extends BaseCommand { const settled = await Settler.run() if (settled > 0) this.logger.info(`Settled ${settled} releases.`) } + if (main && Date.now() - tailedAt >= coordinatorConfig.tail.everySeconds * 1000) { + tailedAt = Date.now() + const { groups, ranges } = await Planner.planTail() + if (ranges > 0) this.logger.info(`Tail: ${ranges} ranges for ${groups} groups.`) + } if (main && Date.now() - mergedAt >= MERGE_EVERY_MS) { mergedAt = Date.now() // Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS. diff --git a/commands/ranges_plan.ts b/commands/ranges_plan.ts index 273b8a3..8534ea9 100644 --- a/commands/ranges_plan.ts +++ b/commands/ranges_plan.ts @@ -1,18 +1,21 @@ -import { BaseCommand } from '@adonisjs/core/ace' +import { BaseCommand, flags } from '@adonisjs/core/ace' import type { CommandOptions } from '@adonisjs/core/types/ace' import Planner from '#services/planner' import NntpPool from '#services/nntp_pool' -/** Cuts every active group's history (and new articles since) into ranges. */ +/** Cuts new groups' history into ranges, or (--tail) plans one round of the live tail. */ export default class RangesPlan extends BaseCommand { static commandName = 'ranges:plan' - static description = "Plans ranges for the indexed groups' articles" + static description = "Plans ranges for new groups' history, or one round of the live tail" static options: CommandOptions = { startApp: true } + @flags.boolean({ description: 'Plan the live tail once (what coordinator:run does every few minutes)' }) + declare tail: boolean + async run() { try { - const { groups, ranges } = await Planner.planAll() - this.logger.info(`${ranges} ranges planned over ${groups} groups.`) + const { groups, ranges } = this.tail ? await Planner.planTail() : await Planner.planAll() + this.logger.info(`${ranges} ${this.tail ? 'tail ' : ''}ranges planned over ${groups} groups.`) } finally { await NntpPool.close() } diff --git a/config/coordinator.ts b/config/coordinator.ts index 1030252..0928e4c 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -12,6 +12,20 @@ export default { * 2026-10-09). */ chunkArticles: env.get('CHUNK_ARTICLES', 100_000), + /** + * The live tail (REBUILD.md, coordinator): new articles become ranges + * every `everySeconds` (one LIST ACTIVE per name prefix), leased and + * fetched before any history. A group gets a tail range once it has + * `minArticles` new (a chunk) or its oldest unplanned article waited + * `maxDelayMinutes`, so a quiet group doesn't make a range every round. + */ + tail: { + everySeconds: env.get('TAIL_EVERY_SECONDS', 120), + minArticles: env.get('TAIL_MIN_ARTICLES', 100_000), + maxDelayMinutes: env.get('TAIL_MAX_DELAY_MINUTES', 15), + /** Before the history's 0 (oldest) … 1000 (newest). */ + priority: -1, + }, /** A leased range without a heartbeat for this long goes back to the queue. */ leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60), /** NNTP connections the coordinator fetches headers with. */ diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index eab8c69..de61169 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -105,7 +105,9 @@ export class XoverClient { return this.timed(async () => { const reply = await this.command(`LIST ACTIVE ${pattern}`) await this.expect(reply, ['215']) - return (await this.plainBody()) + // With XFEATURE COMPRESS GZIP on, Eweka compresses this list too ("215 NewsGroups Follow [COMPRESS=GZIP]"). + const body = reply.includes('COMPRESS=GZIP') ? (await this.compressedRaw(true)).text : await this.plainBody() + return body .split('\r\n') .filter(Boolean) .map((line) => { diff --git a/start/env.ts b/start/env.ts index a4246af..d953d64 100644 --- a/start/env.ts +++ b/start/env.ts @@ -47,6 +47,9 @@ export default await Env.create(new URL('../', import.meta.url), { // Coordinator RANGE_ARTICLES: Env.schema.number.optional(), + TAIL_EVERY_SECONDS: Env.schema.number.optional(), + TAIL_MIN_ARTICLES: Env.schema.number.optional(), + TAIL_MAX_DELAY_MINUTES: Env.schema.number.optional(), CHUNK_ARTICLES: Env.schema.number.optional(), LEASE_TTL_SECONDS: Env.schema.number.optional(), SPOOL_DIR: Env.schema.string.optional(), diff --git a/tests/unit/coordinator/tail.spec.ts b/tests/unit/coordinator/tail.spec.ts new file mode 100644 index 0000000..f2e0b8c --- /dev/null +++ b/tests/unit/coordinator/tail.spec.ts @@ -0,0 +1,42 @@ +import { test } from '@japa/runner' +import { tailRanges, wildmatOf } from '#services/planner' + +const options = { rangeArticles: 1000, minArticles: 100, maxDelayMinutes: 15, priority: -1 } +const now = new Date('2026-10-09T10:00:00Z') +const minutesAgo = (m: number) => new Date(now.getTime() - m * 60_000) + +test.group('coordinator / live tail', () => { + test('a chunk of new articles, or a long enough wait, makes tail ranges', ({ assert }) => { + const rows = tailRanges( + [ + // Enough new articles: planned at once, cut at rangeArticles. + { id: 1, plannedTo: 5000, plannedAt: minutesAgo(2), high: 7500 }, + // A few new articles, recently planned: waits. + { id: 2, plannedTo: 100, plannedAt: minutesAgo(2), high: 150 }, + // A few new articles, waited long enough. + { id: 3, plannedTo: 100, plannedAt: minutesAgo(20), high: 103 }, + // Nothing new, never planned, not listed: nothing. + { id: 4, plannedTo: 100, plannedAt: minutesAgo(60), high: 100 }, + { id: 5, plannedTo: null, plannedAt: minutesAgo(60), high: 500 }, + { id: 6, plannedTo: 100, plannedAt: minutesAgo(60), high: null }, + ], + now, + options + ) + assert.deepEqual( + rows.map((r) => [r.group_id, r.start_id, r.end_id, r.next_id, r.priority]), + [ + [1, 5001, 6000, 5001, -1], + [1, 6001, 7000, 6001, -1], + [1, 7001, 7500, 7001, -1], + [3, 101, 103, 101, -1], + ] + ) + }) + + test('LIST ACTIVE patterns by the first two name parts', ({ assert }) => { + assert.equal(wildmatOf('alt.binaries.boneless'), 'alt.binaries.*') + assert.equal(wildmatOf('alt.binaries.multimedia.anime'), 'alt.binaries.*') + assert.equal(wildmatOf('alt.binaries'), 'alt.binaries') + }) +}) diff --git a/tests/unit/nntp/xover_client.spec.ts b/tests/unit/nntp/xover_client.spec.ts index 0b770ba..a8fa821 100644 --- a/tests/unit/nntp/xover_client.spec.ts +++ b/tests/unit/nntp/xover_client.spec.ts @@ -39,6 +39,16 @@ function server(): Promise<{ port: number; close: () => void }> { sock.write('224 Overview Information Follows\r\n') send(Buffer.concat([body, Buffer.from('.\r\n')])) } + } else if (line.startsWith('LIST ACTIVE')) { + // As Eweka: compressed too once XFEATURE COMPRESS GZIP is on. + const body = Buffer.from('a.b.test 102 101 y\r\na.b.other 9000000000 2 y\r\n') + if (gzip) { + sock.write('215 NewsGroups Follow [COMPRESS=GZIP]\r\n') + send(Buffer.concat([deflateSync(body), Buffer.from('.\r\n')])) + } else { + sock.write('215 NewsGroups Follow\r\n') + send(Buffer.concat([body, Buffer.from('.\r\n')])) + } } else if (line === 'QUIT') sock.end('205 bye\r\n') } }) @@ -68,6 +78,10 @@ test.group('nntp / XoverClient', () => { assert.equal(overviews[1][1].subject, 'Show.S01E01 - "a.mkv" yEnc (2/2)') assert.deepEqual(await client.xover(1, 1), []) assert.lengthOf(await client.xover(101, 102), 2) + assert.deepEqual(await client.listActive('a.b.*'), [ + { name: 'a.b.test', high: 102, low: 101 }, + { name: 'a.b.other', high: 9_000_000_000, low: 2 }, + ]) } finally { client.close() close()