diff --git a/REBUILD.md b/REBUILD.md index 4c135e4..48eb408 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -62,7 +62,12 @@ the cap is shared by the node's processes). Also built (2026-10-09): the live tail (new articles planned every 2 minutes from one LIST ACTIVE per name prefix, at priority −1: leased and -fetched before history); the catalog API, Newznab (`/api`) and the web +fetched before history; since 2026-10-27 a round grows the group's open +tail range, a planned one at the tail's priority nothing spooled or +took, up to RANGE_ARTICLES before it adds rows, as a tail nothing leased +had become a sliver a group a round, 127k rows of a median 14k articles; +`ranges:merge-tail` packs what such rounds left); the catalog API, +Newznab (`/api`) and the web interface, from voyager 1 (the current build), with categories from the name; voyager 1's 29 learned garbage signatures (`rules:signatures`, `database/data/voyager1_signatures.tsv`). diff --git a/app/services/planner.ts b/app/services/planner.ts index 71012b8..eb4d4d4 100644 --- a/app/services/planner.ts +++ b/app/services/planner.ts @@ -75,32 +75,99 @@ class Planner { * 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 }> { + async planTail(now = new Date()): Promise { const groups = await db.from('groups').where('active', true).where('tail', 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)])) + if (groups.length === 0) return { groups: 0, ranges: 0, extended: 0 } + const { rows: ends } = await db.rawQuery( + `SELECT DISTINCT ON (group_id) group_id, id, start_id, end_id, + (status = 'planned' AND priority = ? AND NOT capture AND spooled_chunks = 0 AND attempts = 0 AND node_id IS NULL) AS open + FROM ranges ORDER BY group_id, start_id DESC`, + [coordinatorConfig.tail.priority] + ) + const last = new Map( + ends.map((r: any) => [Number(r.group_id), { id: Number(r.id), start: Number(r.start_id), end: Number(r.end_id), open: r.open === true }]) + ) 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 })), + const plan = tailRanges( + groups.map((g) => { + const l = last.get(Number(g.id)) + return { id: Number(g.id), plannedTo: l?.end ?? null, plannedAt: new Date(g.planned_at), high: highs.get(g.name) ?? null, open: l?.open ? { id: l.id, start: l.start } : 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 } + return this.applyTail(plan, now) } + + /** The tail's plan written: the open ranges extended (one taken meanwhile gets the slice as a row instead), the rows added. */ + async applyTail(plan: TailPlan, now = new Date()): Promise { + const rows = [...plan.rows] + let extended = 0 + if (rows.length === 0 && plan.extended.length === 0) return { groups: 0, ranges: 0, extended: 0 } + await db.transaction(async (trx) => { + for (const e of plan.extended) { + const { rows: done } = await trx.rawQuery( + `UPDATE ranges SET end_id = ? WHERE id = ? AND status = 'planned' AND spooled_chunks = 0 AND attempts = 0 AND node_id IS NULL AND end_id = ? RETURNING id`, + [e.end_id, e.id, e.from - 1] + ) + if (done.length > 0) extended++ + else rows.push({ group_id: e.group_id, start_id: e.from, end_id: e.end_id, next_id: e.from, priority: coordinatorConfig.tail.priority }) + } + if (rows.length > 0) await trx.table('ranges').multiInsert(rows) + const highs = new Map() + for (const r of [...rows, ...plan.extended]) highs.set(r.group_id, Math.max(highs.get(r.group_id) ?? 0, r.end_id)) + for (const [id, high] of highs) await trx.from('groups').where('id', id).update({ high, planned_at: now }) + }) + return { groups: new Set([...rows, ...plan.extended].map((r) => r.group_id)).size, ranges: rows.length, extended } + } + + /** + * Adjacent planned tail ranges of a group and priority, never spooled or taken, merged into ranges of at most + * RANGE_ARTICLES (ranges:merge-tail): the tail planned a sliver a group every round while nothing leased the tail + * (127k rows of a median 14k articles on 2026-10-27; LOG.md (118)). The first row of a run grows, the others go. + */ + async mergeTail(dryRun = false): Promise<{ candidates: number; merged: number; dropped: number }> { + const { rows: candidates } = await db.rawQuery( + `SELECT id, group_id, start_id, end_id, priority FROM ranges + WHERE status = 'planned' AND priority < 0 AND NOT capture AND spooled_chunks = 0 AND attempts = 0 AND node_id IS NULL + ORDER BY group_id, start_id` + ) + const merges = tailMerges( + candidates.map((r: any) => ({ id: Number(r.id), group_id: Number(r.group_id), start_id: Number(r.start_id), end_id: Number(r.end_id), priority: Number(r.priority) })), + coordinatorConfig.rangeArticles + ) + const dropped = merges.reduce((n, m) => n + m.drop.length, 0) + if (dryRun || merges.length === 0) return { candidates: candidates.length, merged: merges.length, dropped } + await db.transaction(async (trx) => { + // The lease picker reads several tables under this lock; the rows' guards catch a range taken meanwhile. + await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('group-lease', 0))`) + const { rows: gone } = await trx.rawQuery( + `DELETE FROM ranges WHERE id = ANY(?::bigint[]) AND status = 'planned' AND spooled_chunks = 0 AND attempts = 0 AND node_id IS NULL RETURNING id`, + [merges.flatMap((m) => m.drop)] + ) + if (gone.length !== dropped) throw new Error(`${dropped - gone.length} of the ranges to drop were taken meanwhile; nothing merged`) + const { rows: grown } = await trx.rawQuery( + `UPDATE ranges r SET end_id = v.end_id FROM (SELECT unnest(?::bigint[]) AS id, unnest(?::bigint[]) AS end_id) v + WHERE r.id = v.id AND r.status = 'planned' AND r.spooled_chunks = 0 AND r.attempts = 0 AND r.node_id IS NULL RETURNING r.id`, + [merges.map((m) => m.id), merges.map((m) => m.end_id)] + ) + if (grown.length !== merges.length) throw new Error(`${merges.length - grown.length} of the ranges to grow were taken meanwhile; nothing merged`) + }) + return { candidates: candidates.length, merged: merges.length, dropped } + } +} + +export interface TailPlanned { + groups: number + /** Rows added. */ + ranges: number + /** Open ranges grown. */ + extended: number } /** A LIST ACTIVE pattern for a group: its first two name parts (`alt.binaries.*`), or the name itself. */ @@ -117,29 +184,90 @@ export interface TailGroup { plannedAt: Date /** The server's newest article (null: not listed). */ high: number | null + /** + * Its last range when the tail may grow it instead of adding a row: a planned range at the tail's priority that + * nothing spooled, took or tried (its start: the cut at rangeArticles counts from there). + */ + open?: { id: number; start: number } | null +} + +export interface TailRow { + group_id: number + start_id: number + end_id: number + next_id: number + priority: number +} + +export interface TailPlan { + /** New ranges. */ + rows: TailRow[] + /** Open ranges grown to end_id (from: the first new article, the old end + 1). */ + extended: { id: number; group_id: number; from: number; end_id: number }[] } /** - * 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. + * The tail's new ranges: per group, the articles after its last range, once there are `minArticles` of them or + * the group's last planning is `maxDelayMinutes` ago. An open last range grows up to `rangeArticles` first, so a + * tail nothing leases stays one range a group instead of a sliver a round; the rest is cut at `rangeArticles`. */ 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 }[] = [] +): TailPlan { + const rows: TailRow[] = [] + const extended: TailPlan['extended'] = [] 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) { + let start = g.plannedTo + 1 + if (g.open && g.open.start + options.rangeArticles - 1 >= start) { + const end = Math.min(g.open.start + options.rangeArticles - 1, g.high) + extended.push({ id: g.open.id, group_id: g.id, from: start, end_id: end }) + start = end + 1 + } + for (; 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 + return { rows, extended } +} + +export interface TailMerge { + /** The run's first range, grown to end_id. */ + id: number + end_id: number + /** The ranges it absorbed. */ + drop: number[] +} + +/** + * Runs of adjacent ranges of one group and priority (sorted by group and start) packed into ranges of at most + * `rangeArticles`: the first range of each packed run grows, the others are dropped. Only runs that merge anything. + */ +export function tailMerges( + ranges: { id: number; group_id: number; start_id: number; end_id: number; priority: number }[], + rangeArticles: number +): TailMerge[] { + const merges: TailMerge[] = [] + let cur: (TailMerge & { group_id: number; start_id: number; priority: number }) | null = null + const flush = () => { + if (cur && cur.drop.length > 0) merges.push({ id: cur.id, end_id: cur.end_id, drop: cur.drop }) + } + for (const r of ranges) { + if (cur && r.group_id === cur.group_id && r.priority === cur.priority && r.start_id === cur.end_id + 1 && r.end_id - cur.start_id + 1 <= rangeArticles) { + cur.end_id = r.end_id + cur.drop.push(r.id) + continue + } + flush() + cur = { id: r.id, group_id: r.group_id, start_id: r.start_id, end_id: r.end_id, priority: r.priority, drop: [] } + } + flush() + return merges } export default new Planner() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index cce4e78..4a4eea9 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -117,8 +117,11 @@ export default class CoordinatorRun extends BaseCommand { } 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.`) + const { groups, ranges, extended } = await Planner.planTail() + if (ranges + extended > 0) + this.logger.info( + `Tail: ${ranges} ranges added, ${extended} grown, for ${groups} groups.` + ) } } catch (error: any) { this.logger.error(`Coordinator round failed: ${error.message}`) diff --git a/commands/ranges_merge_tail.ts b/commands/ranges_merge_tail.ts new file mode 100644 index 0000000..de1a4ad --- /dev/null +++ b/commands/ranges_merge_tail.ts @@ -0,0 +1,23 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import Planner from '#services/planner' + +/** + * Merges adjacent planned tail ranges (Planner.mergeTail): the tail planner added a sliver a group every round + * while nothing leased the tail; now it grows the group's open range instead, and this packs what is there. + */ +export default class RangesMergeTail extends BaseCommand { + static commandName = 'ranges:merge-tail' + static description = 'Merges adjacent planned tail ranges into ranges of RANGE_ARTICLES' + static options: CommandOptions = { startApp: true } + + @flags.boolean({ description: 'Only count what would be merged' }) + declare dryRun: boolean + + async run() { + const { candidates, merged, dropped } = await Planner.mergeTail(this.dryRun) + this.logger.info( + `${candidates} planned tail ranges: ${dropped} ${this.dryRun ? 'would merge' : 'merged'} into ${merged} grown ranges (${candidates - dropped} left).` + ) + } +} diff --git a/commands/ranges_plan.ts b/commands/ranges_plan.ts index 8534ea9..8e1b6bb 100644 --- a/commands/ranges_plan.ts +++ b/commands/ranges_plan.ts @@ -14,8 +14,13 @@ export default class RangesPlan extends BaseCommand { async run() { try { - const { groups, ranges } = this.tail ? await Planner.planTail() : await Planner.planAll() - this.logger.info(`${ranges} ${this.tail ? 'tail ' : ''}ranges planned over ${groups} groups.`) + if (this.tail) { + const { groups, ranges, extended } = await Planner.planTail() + this.logger.info(`Tail: ${ranges} ranges added, ${extended} grown, over ${groups} groups.`) + } else { + const { groups, ranges } = await Planner.planAll() + this.logger.info(`${ranges} ranges planned over ${groups} groups.`) + } } finally { await NntpPool.close() } diff --git a/tests/functional/planner_tail.spec.ts b/tests/functional/planner_tail.spec.ts new file mode 100644 index 0000000..65cfa43 --- /dev/null +++ b/tests/functional/planner_tail.spec.ts @@ -0,0 +1,142 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import Planner from '#services/planner' +import coordinatorConfig from '#config/coordinator' + +const TAIL = coordinatorConfig.tail.priority + +test.group('Planner / tail ranges', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('applyTail grows an open range, or adds the slice when the range was taken meanwhile', async ({ + assert, + }) => { + const [g] = await db + .table('groups') + .insert({ + name: 'alt.binaries.tail', + priority: 1, + tail: true, + planned_at: new Date('2026-10-27T08:00:00Z'), + }) + .returning('id') + const [open] = await db + .table('ranges') + .insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, priority: TAIL }) + .returning('id') + const [taken] = await db + .table('ranges') + .insert({ + group_id: g.id, + start_id: 301, + end_id: 400, + next_id: 301, + priority: TAIL, + status: 'leased', + node_id: 1, + }) + .returning('id') + const now = new Date('2026-10-27T09:00:00Z') + const planned = await Planner.applyTail( + { + rows: [ + { group_id: Number(g.id), start_id: 2001, end_id: 2500, next_id: 2001, priority: TAIL }, + ], + extended: [ + { id: Number(open.id), group_id: Number(g.id), from: 301, end_id: 2000 }, + { id: Number(taken.id), group_id: Number(g.id), from: 401, end_id: 600 }, + ], + }, + now + ) + assert.deepEqual(planned, { groups: 1, ranges: 2, extended: 1 }) + const ranges = await db + .from('ranges') + .where('group_id', g.id) + .orderBy('start_id') + .select('start_id', 'end_id', 'status', 'priority') + assert.deepEqual( + ranges.map((r) => [Number(r.start_id), Number(r.end_id), r.status, Number(r.priority)]), + [ + [1, 2000, 'planned', TAIL], + [301, 400, 'leased', TAIL], + [401, 600, 'planned', TAIL], + [2001, 2500, 'planned', TAIL], + ] + ) + const updated = await db.from('groups').where('id', g.id).firstOrFail() + assert.equal(Number(updated.high), 2500) + assert.equal(new Date(updated.planned_at).toISOString(), now.toISOString()) + }) + + test('mergeTail packs adjacent planned tail ranges and leaves taken, spooled, capture and history ranges alone', async ({ + assert, + }) => { + const saved = coordinatorConfig.rangeArticles + coordinatorConfig.rangeArticles = 1000 + try { + const [g] = await db + .table('groups') + .insert({ name: 'alt.binaries.tail', priority: 1, tail: true }) + .returning('id') + const gid = Number(g.id) + const range = (start_id: number, end_id: number, extra: Record = {}) => ({ + group_id: gid, + start_id, + end_id, + next_id: start_id, + priority: TAIL, + ...extra, + }) + await db.table('ranges').multiInsert([ + // History, then a run of five slivers (the fifth is spooled: a break), two more, a capture range, the tail's rest. + range(1, 1000, { priority: 500, status: 'done' }), + range(1001, 1100), + range(1101, 1200), + range(1201, 1300), + range(1301, 1400), + range(1401, 1500, { spooled_chunks: 1 }), + range(1501, 1600), + range(1601, 1700), + range(1701, 1800, { capture: true }), + range(1801, 2600), + range(2601, 2700), + range(2701, 2800, { status: 'leased', node_id: 1 }), + ]) + assert.deepEqual(await Planner.mergeTail(true), { candidates: 8, merged: 3, dropped: 5 }) + assert.equal( + (await db.from('ranges').where('group_id', gid).count('* as n').first())!.n, + '12' + ) + assert.deepEqual(await Planner.mergeTail(), { candidates: 8, merged: 3, dropped: 5 }) + const ranges = await db + .from('ranges') + .where('group_id', gid) + .orderBy('start_id') + .select('start_id', 'end_id', 'next_id', 'status', 'priority', 'capture') + assert.deepEqual( + ranges.map((r) => [ + Number(r.start_id), + Number(r.end_id), + Number(r.next_id), + r.status, + Number(r.priority), + r.capture, + ]), + [ + [1, 1000, 1, 'done', 500, false], + [1001, 1400, 1001, 'planned', TAIL, false], + [1401, 1500, 1401, 'planned', TAIL, false], + [1501, 1700, 1501, 'planned', TAIL, false], + [1701, 1800, 1701, 'planned', TAIL, true], + [1801, 2700, 1801, 'planned', TAIL, false], + [2701, 2800, 2701, 'leased', TAIL, false], + ] + ) + assert.deepEqual(await Planner.mergeTail(), { candidates: 3, merged: 0, dropped: 0 }) + } finally { + coordinatorConfig.rangeArticles = saved + } + }) +}) diff --git a/tests/unit/coordinator/tail.spec.ts b/tests/unit/coordinator/tail.spec.ts index f2e0b8c..a654648 100644 --- a/tests/unit/coordinator/tail.spec.ts +++ b/tests/unit/coordinator/tail.spec.ts @@ -1,5 +1,5 @@ import { test } from '@japa/runner' -import { tailRanges, wildmatOf } from '#services/planner' +import { tailMerges, tailRanges, wildmatOf } from '#services/planner' const options = { rangeArticles: 1000, minArticles: 100, maxDelayMinutes: 15, priority: -1 } const now = new Date('2026-10-09T10:00:00Z') @@ -7,7 +7,7 @@ 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( + const { rows, extended } = tailRanges( [ // Enough new articles: planned at once, cut at rangeArticles. { id: 1, plannedTo: 5000, plannedAt: minutesAgo(2), high: 7500 }, @@ -32,6 +32,94 @@ test.group('coordinator / live tail', () => { [3, 101, 103, 101, -1], ] ) + assert.deepEqual(extended, []) + }) + + test('an open last range grows up to rangeArticles before rows are added', ({ assert }) => { + const { rows, extended } = tailRanges( + [ + // Room in the open range for everything new. + { + id: 1, + plannedTo: 5300, + plannedAt: minutesAgo(20), + high: 5350, + open: { id: 11, start: 5001 }, + }, + // The open range fills up, the rest is cut as rows. + { + id: 2, + plannedTo: 5300, + plannedAt: minutesAgo(2), + high: 7500, + open: { id: 12, start: 5001 }, + }, + // An open range already full: rows only. + { + id: 3, + plannedTo: 6000, + plannedAt: minutesAgo(2), + high: 6500, + open: { id: 13, start: 5001 }, + }, + // Not open (taken, spooled, another priority): rows. + { id: 4, plannedTo: 5300, plannedAt: minutesAgo(20), high: 5350, open: null }, + ], + now, + options + ) + assert.deepEqual( + extended.map((e) => [e.id, e.group_id, e.from, e.end_id]), + [ + [11, 1, 5301, 5350], + [12, 2, 5301, 6000], + ] + ) + assert.deepEqual( + rows.map((r) => [r.group_id, r.start_id, r.end_id]), + [ + [2, 6001, 7000], + [2, 7001, 7500], + [3, 6001, 6500], + [4, 5301, 5350], + ] + ) + }) + + test('adjacent ranges of a group and priority pack into rangeArticles; gaps and other priorities break a run', ({ + assert, + }) => { + const r = (id: number, group_id: number, start_id: number, end_id: number, priority = -3) => ({ + id, + group_id, + start_id, + end_id, + priority, + }) + const merges = tailMerges( + [ + // 1–300 + 301–600 + 601–900 fill 1000 with 901–1000; 1001–1100 starts the next run alone. + r(1, 1, 1, 300), + r(2, 1, 301, 600), + r(3, 1, 601, 900), + r(4, 1, 901, 1000), + r(5, 1, 1001, 1100), + // A gap, then two adjacent at another priority. + r(6, 1, 1200, 1250), + r(7, 1, 1251, 1300, -1), + r(8, 1, 1301, 1400, -1), + // Another group; one that would overflow stays alone. + r(9, 2, 1, 900), + r(10, 2, 901, 1100), + r(11, 2, 1101, 1150), + ], + 1000 + ) + assert.deepEqual(merges, [ + { id: 1, end_id: 1000, drop: [2, 3, 4] }, + { id: 7, end_id: 1400, drop: [8] }, + { id: 10, end_id: 1150, drop: [11] }, + ]) }) test('LIST ACTIVE patterns by the first two name parts', ({ assert }) => {