diff --git a/REBUILD.md b/REBUILD.md index e7889ca..8603279 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -60,6 +60,13 @@ most workers on batcave, a few on poweredge-kantoor with `edge:node --max-mbit` (e.g. 50 of its 160 Mbit/s ≈ 100–125 k articles/s; the cap is shared by the node's processes). +Every group is tiled from article 1 (2026-10-28): one "ancient" range, +status `ancient`, covers 1 … low watermark − 1, what had expired on the +server before the group was planned. Never leased, fetched or filled — +nothing selects it — and the status panel reports it apart ("Ancient: … +articles below the low watermark"); migration 6700 added it for the +groups planned before, `Planner.historyRows` for groups planned since. + 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; since 2026-10-27 a round grows the group's open diff --git a/app/services/planner.ts b/app/services/planner.ts index eb4d4d4..870ffa0 100644 --- a/app/services/planner.ts +++ b/app/services/planner.ts @@ -35,20 +35,8 @@ class Planner { async planGroup(group: { id: number; name: string }): Promise { const info = await NntpPool.use((client) => client.group(group.name)) - const size = coordinatorConfig.rangeArticles const last = await db.from('ranges').where('group_id', group.id).max('end_id as end').first() - const from = Math.max(info.low, Number(last?.end ?? 0) + 1) - const span = Math.max(1, info.high - info.low) - const rows = [] - for (let start = from; start <= info.high; start += size) { - rows.push({ - group_id: group.id, - start_id: start, - end_id: Math.min(start + size - 1, info.high), - next_id: start, - priority: Math.floor((1000 * (start - info.low)) / span), - }) - } + const rows = historyRows(group.id, info, Number(last?.end ?? 0), coordinatorConfig.rangeArticles) for (let i = 0; i < rows.length; i += 2000) await db.table('ranges').multiInsert(rows.slice(i, i + 2000)) await db.from('groups').where('id', group.id).update({ low: info.low, high: info.high, planned_at: new Date() }) return rows.length @@ -270,4 +258,34 @@ export function tailMerges( return merges } +/** + * The rows a group's history is cut into, from `lastEnd + 1` (0: the group has no ranges yet) up to the high + * watermark: RANGE_ARTICLES slices at the position priority. A group planned for the first time also gets one + * "ancient" range, articles 1 … low − 1: what had expired on the server before it was planned. Status + * 'ancient', never leased or fetched; it keeps the tiling complete from article 1 (migration 6700 added it for + * the groups planned before 2026-10-28). + */ +export function historyRows( + groupId: number, + info: { low: number; high: number }, + lastEnd: number, + size: number +): Record[] { + const rows: Record[] = [] + if (lastEnd === 0 && info.low > 1) + rows.push({ group_id: groupId, start_id: 1, end_id: info.low - 1, next_id: 1, priority: 0, status: 'ancient' }) + const from = Math.max(info.low, lastEnd + 1) + const span = Math.max(1, info.high - info.low) + for (let start = from; start <= info.high; start += size) { + rows.push({ + group_id: groupId, + start_id: start, + end_id: Math.min(start + size - 1, info.high), + next_id: start, + priority: Math.floor((1000 * (start - info.low)) / span), + }) + } + return rows +} + export default new Planner() diff --git a/app/services/search_service.ts b/app/services/search_service.ts index 9ebace1..c5e26c9 100644 --- a/app/services/search_service.ts +++ b/app/services/search_service.ts @@ -304,9 +304,9 @@ class SearchService { // History by articles; the pass: the history ranges before those deferred behind them (priority // 2000 apart; e.g. groups added later that go first). const { rows: ranges } = await db.rawQuery( - `WITH h AS (SELECT status, priority, end_id - start_id + 1 AS n FROM ranges WHERE priority >= 0), + `WITH h AS (SELECT status, priority, end_id - start_id + 1 AS n FROM ranges WHERE priority >= 0 AND status <> 'ancient'), pass AS (SELECT (floor(coalesce(min(priority) FILTER (WHERE status <> 'done'), 0) / 2000) + 1) * 2000 AS below FROM h) - SELECT (SELECT count(*) FILTER (WHERE status = 'done') FROM ranges) AS done, (SELECT count(*) FROM ranges) AS total, + SELECT (SELECT count(*) FILTER (WHERE status = 'done') FROM ranges) AS done, (SELECT count(*) FROM ranges WHERE status <> 'ancient') AS total, coalesce(sum(n) FILTER (WHERE status = 'done'), 0) AS history_done, coalesce(sum(n), 0) AS history, coalesce(sum(n) FILTER (WHERE status = 'done' AND priority < (SELECT below FROM pass)), 0) AS pass_done, coalesce(sum(n) FILTER (WHERE priority < (SELECT below FROM pass)), 0) AS pass, diff --git a/app/services/status.ts b/app/services/status.ts index 0a4c165..e1641ee 100644 --- a/app/services/status.ts +++ b/app/services/status.ts @@ -54,6 +54,8 @@ export interface Status { corpus: { pendingRanges: number; pendingArticles: number } spool: { ranges: number; chunks: number } tail: { groups: number; ranges: number; articles: number } + /** The 'ancient' ranges: articles below the low watermark the groups were planned at (expired, never fetched). */ + ancient: { groups: number; articles: number } held: HeldGroup[] snapshots: { count: number; gb: number; oldestHours: number | null; largest: SnapshotFile[] } releases: { estimated: number; provisional: number } @@ -100,8 +102,10 @@ class StatusService { count(*) FILTER (WHERE status = 'done') AS done, coalesce(sum(end_id - start_id + 1) FILTER (WHERE status = 'done'), 0) AS done_articles, count(*) FILTER (WHERE status = 'failed') AS failed - FROM ranges GROUP BY 1 ORDER BY min(priority)`, - [tail] + FROM ranges GROUP BY 1 + ORDER BY min(CASE WHEN priority <= ? THEN 0 WHEN priority < 0 THEN 1 WHEN capture AND priority <= 2 THEN 2 + priority + WHEN priority < 1000 THEN 10 ELSE 11 + priority / 1000 END)`, + [tail, tail] ) const { rows: hour } = await db.rawQuery( `SELECT coalesce(sum(end_id - start_id + 1) FILTER (WHERE done_at > now() - interval '1 hour'), 0) AS indexed, @@ -114,7 +118,9 @@ class StatusService { coalesce(sum(spooled_chunks) FILTER (WHERE status IN ('planned', 'leased')), 0) AS spooled_chunks, count(DISTINCT group_id) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_groups, count(*) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_ranges, - coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned' AND priority <= ?), 0) AS tail_articles + coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned' AND priority <= ?), 0) AS tail_articles, + count(DISTINCT group_id) FILTER (WHERE status = 'ancient') AS ancient_groups, + coalesce(sum(end_id - start_id + 1) FILTER (WHERE status = 'ancient'), 0) AS ancient_articles FROM ranges`, [tail, tail, tail] ) @@ -194,6 +200,7 @@ class StatusService { pendingArticles: Number(h.corpus_pending_articles), }, spool: { ranges: Number(h.spooled_ranges), chunks: Number(h.spooled_chunks) }, + ancient: { groups: Number(h.ancient_groups), articles: Number(h.ancient_articles) }, tail: { groups: Number(h.tail_groups), ranges: Number(h.tail_ranges), @@ -264,7 +271,7 @@ class StatusService { `Last hour: ${b(s.lastHour.indexedArticles)} articles indexed in ${n(s.lastHour.doneRanges)} ranges (${n(s.lastHour.indexedArticles / 3600)}/s); corpus fill ${b(s.lastHour.storedArticles)} in ${n(s.lastHour.storedRanges)} ranges, ${n(s.corpus.pendingRanges)} capture ranges (${b(s.corpus.pendingArticles)}) still to store` ) lines.push( - `Spool: ${n(s.spool.chunks)} chunks in ${n(s.spool.ranges)} ranges. Tail: ${b(s.tail.articles)} planned in ${n(s.tail.ranges)} ranges of ${n(s.tail.groups)} groups` + `Spool: ${n(s.spool.chunks)} chunks in ${n(s.spool.ranges)} ranges. Tail: ${b(s.tail.articles)} planned in ${n(s.tail.ranges)} ranges of ${n(s.tail.groups)} groups. Ancient: ${b(s.ancient.articles)} articles below the low watermark of ${n(s.ancient.groups)} groups` ) lines.push('') const table = (header: string[], rows: string[][]) => { diff --git a/database/migrations/1770000006700_ancient_ranges.ts b/database/migrations/1770000006700_ancient_ranges.ts new file mode 100644 index 0000000..355fdf5 --- /dev/null +++ b/database/migrations/1770000006700_ancient_ranges.ts @@ -0,0 +1,20 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * One "ancient" range a group, articles 1 … (its first range − 1): what had expired on the server before the + * group was planned (the planner cuts from the low watermark). Status 'ancient': nothing to fetch, never leased, + * so the tiling is complete from article 1 and the span below the low watermark is on record (user, 2026-10-28). + * Planner.historyRows adds it for groups planned from now on. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw( + `INSERT INTO ranges (group_id, start_id, end_id, next_id, priority, status) + SELECT group_id, 1, min(start_id) - 1, 1, 0, 'ancient' FROM ranges GROUP BY group_id HAVING min(start_id) > 1` + ) + } + + async down() { + this.schema.raw(`DELETE FROM ranges WHERE status = 'ancient'`) + } +} diff --git a/tests/functional/status.spec.ts b/tests/functional/status.spec.ts index 888d550..1a89576 100644 --- a/tests/functional/status.spec.ts +++ b/tests/functional/status.spec.ts @@ -94,6 +94,15 @@ test.group('Status', (group) => { ]) await writeFile(join(dir, `${g.id}.gz`), Buffer.alloc(2_000_000)) await writeFile(join(dir, `${g.id}-tail.gz`), Buffer.alloc(1000)) + // A group planned from its low watermark 1000: its ancient range counts in no band and is never leased. + const [old] = await db + .table('groups') + .insert({ name: 'alt.binaries.ancient', priority: 2 }) + .returning('id') + await db.table('ranges').multiInsert([ + { group_id: old.id, start_id: 1, end_id: 999, next_id: 1, priority: 0, status: 'ancient' }, + { group_id: old.id, start_id: 1000, end_id: 1099, next_id: 1000, priority: 0 }, + ]) const s = await StatusService.collect() assert.deepEqual( @@ -111,10 +120,11 @@ test.group('Status', (group) => { ['raised', 1, 100, 0, 0, 0, 0], ['window 0', 0, 0, 1, 1, 100, 0], ['window 2', 1, 100, 0, 0, 0, 0], - ['history 0-999', 1, 100, 0, 0, 0, 0], + ['history 0-999', 2, 200, 0, 0, 0, 0], ['history 4000+', 0, 0, 0, 0, 0, 1], ] ) + assert.deepEqual(s.ancient, { groups: 1, articles: 999 }) assert.deepEqual(s.lastHour, { indexedArticles: 100, doneRanges: 1, @@ -153,5 +163,6 @@ test.group('Status', (group) => { assert.include(text, 'window 0') assert.include(text, 'alt.binaries.status history status-node 301-400 351') assert.include(text, 'Snapshots: 2 files') + assert.include(text, 'Ancient: 999 articles below the low watermark of 1 groups') }) }) diff --git a/tests/unit/coordinator/planner_rows.spec.ts b/tests/unit/coordinator/planner_rows.spec.ts new file mode 100644 index 0000000..955b08c --- /dev/null +++ b/tests/unit/coordinator/planner_rows.spec.ts @@ -0,0 +1,36 @@ +import { test } from '@japa/runner' +import { historyRows } from '#services/planner' + +test.group('Planner / history rows', () => { + test('a group planned for the first time: the ancient range below the low watermark, then the slices', ({ + assert, + }) => { + const rows = historyRows(7, { low: 1001, high: 3500 }, 0, 1000) + assert.deepEqual( + rows.map((r) => [r.start_id, r.end_id, r.next_id, r.priority, r.status ?? 'planned']), + [ + [1, 1000, 1, 0, 'ancient'], + [1001, 2000, 1001, 0, 'planned'], + [2001, 3000, 2001, 400, 'planned'], + [3001, 3500, 3001, 800, 'planned'], + ] + ) + }) + + test('no ancient range when the group starts at article 1, nor when planning again', ({ + assert, + }) => { + assert.deepEqual( + historyRows(7, { low: 1, high: 1500 }, 0, 1000).map((r) => [r.start_id, r.end_id, r.status]), + [ + [1, 1000, undefined], + [1001, 1500, undefined], + ] + ) + // Planned before up to 3500: only the new articles, no second ancient range. + assert.deepEqual( + historyRows(7, { low: 1200, high: 4100 }, 3500, 1000).map((r) => [r.start_id, r.end_id]), + [[3501, 4100]] + ) + }) +})