From 137e931e9e816e2ff6dff25da067e393e175ec60 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sun, 25 Oct 2026 09:47:28 +0100 Subject: [PATCH] Posting-time windows into the corpus: ranges:dates, ranges:capture ranges:dates samples each active group's posting dates from the news server (group_dates) and estimates every range's date (ranges.est_date). ranges:capture --from --to marks a window's ranges of every group for the corpus. The corpus fill takes capture ranges in priority order, done ones too, and remembers complete ones (ranges.stored_at). Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 14 ++ app/services/group_dates.ts | 122 ++++++++++++ app/services/leases.ts | 56 +++++- app/services/prefetcher.ts | 161 ++++++++++++--- commands/ranges_capture.ts | 80 ++++++++ commands/ranges_dates.ts | 188 ++++++++++++++++++ .../migrations/1770000004900_range_dates.ts | 27 +++ tests/unit/services/group_dates.spec.ts | 75 +++++++ 8 files changed, 683 insertions(+), 40 deletions(-) create mode 100644 app/services/group_dates.ts create mode 100644 commands/ranges_capture.ts create mode 100644 commands/ranges_dates.ts create mode 100644 database/migrations/1770000004900_range_dates.ts create mode 100644 tests/unit/services/group_dates.spec.ts diff --git a/REBUILD.md b/REBUILD.md index 8c44f10..0baa19f 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -244,6 +244,20 @@ The worker makes the same releases of a filtered chunk; rule_counts lack those two rules' counts. `corpus:filter` filters chunks stored raw. A corpus filtered this way no longer serves a sweep of those two rules. +Windows by posting time (2026-10-25; voyager 3's test data, LOG.md (104)): +`ranges:dates` samples when each active group's articles were posted (one +XOVER of 32 articles every 500M articles of the group, 16 … 1000 samples, +the median Date of each slice, kept monotone; `group_dates`) and estimates +every range's posting date from the curve (`ranges.est_date`; the edges of +a window are off by up to one sample interval, three weeks in boneless). +`ranges:capture --from 2025-01-01 --to 2026-01-01` marks every active +group's ranges posted in the window `capture`: planned ones not started at +`--priority` (0), done ones as they are, both fetched into the corpus by the +fill, which takes capture ranges in priority order, done ones too (they're +fetched again, corpus only), and skips those complete (`ranges.stored_at`, +set when it sees the range.json). A year of every group is the data to +replay voyager 3's 30-day blocks on. + ## Phase 2 Built: the prober (2026-10-11; `probe:run`, LOG.md 2026-10-11 (2)). It diff --git a/app/services/group_dates.ts b/app/services/group_dates.ts new file mode 100644 index 0000000..24c9093 --- /dev/null +++ b/app/services/group_dates.ts @@ -0,0 +1,122 @@ +import type { Overview, XoverClient } from '#lib/nntp/index' + +/** When an article of a group was posted: the median Date of a slice of articles at `article`. */ +export interface DateSample { + article: number + postedAt: Date +} + +/** Before this, a Date header is a broken clock (the oldest retention is 2008). */ +const OLDEST = Date.UTC(1995, 0, 1) +const DAY = 86_400_000 + +/** + * The median of the overviews' Date headers that parse and are plausible (1995 … a day from `now`), with at least + * `min` of them; null otherwise (an expired or empty slice). + */ +export function medianDate(overviews: Overview[], now = new Date(), min = 3): Date | null { + const latest = now.getTime() + DAY + const times = overviews + .map(([, fields]) => Date.parse(fields.date ?? '')) + .filter((t) => t >= OLDEST && t <= latest) + .sort((a, b) => a - b) + if (times.length < min) return null + return new Date(times[Math.floor(times.length / 2)]) +} + +/** Where a group is sampled: `count` article numbers spread over low … high, the first at low and the last `size` before high. */ +export function samplePoints(low: number, high: number, count: number, size: number): number[] { + const last = Math.max(low, high - size + 1) + if (count <= 1 || last <= low) return [low] + const points = new Set() + for (let i = 0; i < count; i++) points.add(Math.round(low + ((last - low) * i) / (count - 1))) + return [...points].sort((a, b) => a - b) +} + +/** + * The samples as a posting-time curve: in article order, a sample later than both of the next two by over a day + * (a slice of one poster's wrong clock) is dropped, then each date is at least its predecessor's (articles arrive + * in number order, bar stragglers and old crossposts). + */ +export function monotone(samples: DateSample[]): DateSample[] { + const sorted = [...samples].sort((a, b) => a.article - b.article) + const kept = sorted.filter((s, i) => { + const next = sorted.slice(i + 1, i + 3) + return next.length < 2 || next.some((n) => n.postedAt.getTime() + DAY >= s.postedAt.getTime()) + }) + let floor = Number.NEGATIVE_INFINITY + return kept.map((s) => { + floor = Math.max(floor, s.postedAt.getTime()) + return { article: s.article, postedAt: new Date(floor) } + }) +} + +/** + * When `article` was posted by the group's samples (monotone, in article order): interpolated between the two around + * it, the first or last sample's date beyond them; null without samples. + */ +export function estimateDate(samples: DateSample[], article: number): Date | null { + if (samples.length === 0) return null + if (article <= samples[0].article) return samples[0].postedAt + const last = samples[samples.length - 1] + if (article >= last.article) return last.postedAt + let lo = 0 + let hi = samples.length - 1 + while (hi - lo > 1) { + const mid = (lo + hi) >> 1 + if (samples[mid].article <= article) lo = mid + else hi = mid + } + const a = samples[lo] + const b = samples[hi] + const f = (article - a.article) / (b.article - a.article) + return new Date(a.postedAt.getTime() + f * (b.postedAt.getTime() - a.postedAt.getTime())) +} + +/** How many samples a group gets: one per `every` articles of its span, at least `min` and at most `max`. */ +export function sampleCount( + low: number, + high: number, + every: number, + min = 16, + max = 1000 +): number { + return Math.min(max, Math.max(min, Math.ceil((high - low) / every) + 1)) +} + +/** + * Samples a group over one connection: GROUP, then an XOVER of `size` articles at each of `count` points (a slice + * with too few articles that parse, expired or sparse, is widened to 32 × size once). The curve (monotone) and the + * server's span. + */ +export async function sampleGroup( + client: XoverClient, + name: string, + options: { every: number; size: number }, + now = new Date() +): Promise<{ samples: DateSample[]; low: number; high: number; commands: number }> { + const info = await client.group(name) + const samples: DateSample[] = [] + let commands = 1 + for (const point of samplePoints( + info.low, + info.high, + sampleCount(info.low, info.high, options.every), + options.size + )) { + commands++ + let date = medianDate( + await client.xover(point, Math.min(point + options.size - 1, info.high)), + now + ) + if (!date) { + commands++ + date = medianDate( + await client.xover(point, Math.min(point + options.size * 32 - 1, info.high)), + now + ) + } + if (date) samples.push({ article: point, postedAt: date }) + } + return { samples: monotone(samples), low: info.low, high: info.high, commands } +} diff --git a/app/services/leases.ts b/app/services/leases.ts index 8afc17e..8135ccc 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -41,7 +41,12 @@ class Leases { AND l.heartbeat_at >= now() - make_interval(secs => ?)) < ${MAX_LEASES_PER_GROUP} ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`, - [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds, coordinatorConfig.leaseTtlSeconds] + [ + nodeId, + rules.version, + coordinatorConfig.leaseTtlSeconds, + coordinatorConfig.leaseTtlSeconds, + ] ) let { rows } = await take(true) if (rows.length === 0) ({ rows } = await take(false)) @@ -49,7 +54,16 @@ class Leases { if (!row) return null await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() }) const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail() - return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules } + return { + range: { + id: Number(row.id), + group: group.name, + start: Number(row.start_id), + end: Number(row.end_id), + next: Number(row.next_id), + }, + rules, + } } /** @@ -70,7 +84,8 @@ class Leases { async leased(rangeId: number, nodeId: number) { const range = await db.from('ranges').where('id', rangeId).first() - if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`) + if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) + throw new LeaseLost(`Range ${rangeId} is not leased to this node`) return range } @@ -83,8 +98,14 @@ class Leases { /** The worker reached the range's end: done, and its spooled chunks go. */ async finish(rangeId: number, nodeId: number) { const range = await this.leased(rangeId, nodeId) - if (Number(range.next_id) <= Number(range.end_id)) throw new LeaseLost(`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`) - await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() }) + if (Number(range.next_id) <= Number(range.end_id)) + throw new LeaseLost( + `Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}` + ) + await db + .from('ranges') + .where('id', rangeId) + .update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() }) await Spool.remove(rangeId) } @@ -95,12 +116,19 @@ class Leases { */ async corrupt(rangeId: number, nodeId: number, from: number) { const range = await this.leased(rangeId, nodeId) - const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from) + const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find( + (s) => s.from === from + ) if (span) { await Spool.removeChunk(rangeId, span) if (range.capture) { const group = await db.from('groups').where('id', range.group_id).select('name').first() - await Corpus.removeChunk({ group: group.name, start: Number(range.start_id), end: Number(range.end_id) }, span) + await Corpus.removeChunk( + { group: group.name, start: Number(range.start_id), end: Number(range.end_id) }, + span + ) + // Not complete in the corpus any more: the corpus fill fetches the chunk again. + await db.from('ranges').where('id', rangeId).update({ stored_at: null }) } } const plain: number[] = (range.plain_chunks ?? []).map(Number) @@ -108,11 +136,21 @@ class Leases { // refreshed yet) and spooled again (range 161820, LOG.md 2026-10-11), so a second try plain first. if (plain.filter((p) => p === from).length >= 2) { // Fetched plain (deflated here) and still not inflating: not the server's compression; set aside. - await db.from('ranges').where('id', rangeId).update({ status: 'failed', node_id: null, error: `chunk ${from} doesn't inflate, also fetched plain` }) + await db + .from('ranges') + .where('id', rangeId) + .update({ + status: 'failed', + node_id: null, + error: `chunk ${from} doesn't inflate, also fetched plain`, + }) return } // The server's compressed reply is cut short (the same bytes every time): fetched plain next time. - await db.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`, [from, rangeId]) + await db.rawQuery( + `UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`, + [from, rangeId] + ) } /** A stopping worker gives its range back; the next one resumes it. */ diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 8cc6f2b..d2b0391 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -73,7 +73,13 @@ export class Prefetcher { private writingNow = 0 private sampler?: NodeJS.Timeout /** Requests in flight: what, since when, on which connection (the report names the oldest). */ - private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>() + private inFlight = new Set<{ + rangeId: number + from: number + to: number + since: number + client: XoverClient + }>() /** Capture ranges found complete in the corpus (not looked at again). */ private stored = new Set() @@ -163,10 +169,14 @@ export class Prefetcher { } const corpusRange = { group: target.group, start: target.start, end: target.end } // A range in the corpus is read from disk (a reset re-indexes it without the news server). - let data = target.capture && !target.corpusOnly ? await Corpus.read(corpusRange, span) : null + let data = + target.capture && !target.corpusOnly ? await Corpus.read(corpusRange, span) : null const stored = data !== null try { - if (!data) data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span) + if (!data) + data = target.plain.has(span.from) + ? await NntpPool.usePlain(fetch) + : await this.fetchHedged(target.id, target.group, span) } finally { this.fetchingNow-- } @@ -176,7 +186,8 @@ export class Prefetcher { this.stats.chunks++ this.stats.articles += span.to - span.from + 1 this.stats.fetchMs += Date.now() - fetchedAt - if (target.capture) await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter()) + if (target.capture) + await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter()) } if (target.corpusOnly) { this.stats.captured++ @@ -193,7 +204,8 @@ export class Prefetcher { `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`, [target.id] ) - if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + if (Number(rows[0]?.spooled_chunks) >= target.total) + await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) } finally { this.writingNow-- this.stats.writeMs += Date.now() - writtenAt @@ -215,7 +227,10 @@ export class Prefetcher { */ private async fetchHedged(rangeId: number, group: string, span: ChunkSpan): Promise { let age: number | undefined - const request = (start: (work: (client: XoverClient) => Promise) => Promise | null, first: boolean): Request | null => { + const request = ( + start: (work: (client: XoverClient) => Promise) => Promise | null, + first: boolean + ): Request | null => { let client: XoverClient | undefined let cancelled = false const done = start(async (c) => { @@ -247,7 +262,9 @@ export class Prefetcher { data = await c.xoverRaw(span.from, span.to) } catch (error) { if (!late) throw error - throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB in a minute, ${Math.round(c.buffered / 1000)} kB in); fetching it again`) + throw new Error( + `Chunk ${span.from}–${span.to} of range ${rangeId} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB in a minute, ${Math.round(c.buffered / 1000)} kB in); fetching it again` + ) } finally { clearInterval(watch) this.inFlight.delete(entry) @@ -290,7 +307,8 @@ export class Prefetcher { /** The longest-running request: range, chunk, how long, how much came in. */ private oldest() { - let oldest: { rangeId: number; from: number; to: number; since: number; client: XoverClient } | undefined + let oldest: + { rangeId: number; from: number; to: number; since: number; client: XoverClient } | undefined for (const r of this.inFlight) if (!oldest || r.since < oldest.since) oldest = r if (!oldest) return 'none in flight' return `oldest request ${Math.round((Date.now() - oldest.since) / 1000)} s (range ${oldest.rangeId}, ${oldest.from}–${oldest.to}, ${Math.round(oldest.client.buffered / 1000)} kB in)` @@ -315,7 +333,12 @@ export class Prefetcher { const { chunks, articles, fetchMs, hedged, hedgeWon, byAge, samples, requests } = this.stats if (chunks > 0) { const ages = byAge - .map((b, i) => `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace('– min', '+ min')) + .map((b, i) => + `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace( + '– min', + '+ min' + ) + ) .join(', ') logger.info( `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` + @@ -341,13 +364,18 @@ export class Prefetcher { } private peek(): boolean { - return this.targets.some((t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))) + return this.targets.some( + (t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)) + ) } /** Whether the target is under its range's and its group's limit of connections. */ private open(target: Target) { const { perRange, perGroup } = coordinatorConfig.spool - return (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup) + return ( + (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && + (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup) + ) } private count(target: Target, by: number) { @@ -378,26 +406,58 @@ export class Prefetcher { this.sizedAt = Date.now() this.spoolBytes = await Spool.size() } - const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q) - const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks', 'r.capture')) + const mine = (q: any) => + this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q + const rows = (q: any) => + mine( + q + .from('ranges as r') + .join('groups as g', 'g.id', 'r.group_id') + .select( + 'r.id', + 'g.name as group', + 'r.start_id', + 'r.end_id', + 'r.next_id', + 'r.plain_chunks', + 'r.capture' + ) + ) // Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them. const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at') // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. - const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length) - const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) - const [{ ready }] = await mine(db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)).count('* as ready') + const ahead = Math.max( + coordinatorConfig.spool.minAhead, + coordinatorConfig.spool.aheadPerLease * leased.length + ) + const started = await rows(db.query()) + .where('r.status', 'planned') + .where('r.spooled_chunks', '>', 0) + .whereNull('r.spooled_at') + .orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) + const [{ ready }] = await mine( + db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0) + ).count('* as ready') const room = this.spoolBytes < coordinatorConfig.spool.maxBytes // With at most perRange connections a range, enough ranges still being fetched for all loops: fully // spooled ones count as ready but need no connection (30 of 40 busy on small ranges; LOG.md 2026-10-10 (2)). // 8 more than that: ranges with a chunk or two left take fewer than perRange (with +2, 3 of 40 loops idle). const { perRange } = coordinatorConfig.spool - const short = perRange > 0 ? Math.ceil(this.loops / perRange) + 8 - (leased.length + started.length) : 0 + const short = + perRange > 0 ? Math.ceil(this.loops / perRange) + 8 - (leased.length + started.length) : 0 const want = Math.max(ahead - Number(ready), short) // The catalog at its cap on the NVMe: no history fetched ahead (CatalogSpace). const full = await CatalogSpace.full() - const fresh = room && want > 0 - ? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').where((q: any) => (full ? q.where('r.priority', '<', 0) : q)).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(want) - : [] + const fresh = + room && want > 0 + ? await rows(db.query()) + .where('r.status', 'planned') + .where('r.spooled_chunks', 0) + .whereRaw('r.next_id <= r.end_id') + .where((q: any) => (full ? q.where('r.priority', '<', 0) : q)) + .orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) + .limit(want) + : [] const targets: Target[] = [] for (const row of [...leased, ...(room ? started : []), ...fresh]) { const start = Number(row.start_id) @@ -405,14 +465,25 @@ export class Prefetcher { const total = Spool.chunksOf({ start, end }).length const missing: ChunkSpan[] = [] for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) { - if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span) + if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) + missing.push(span) } if (missing.length > 0) { - targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total, start, end, capture: row.capture === true }) + targets.push({ + id: Number(row.id), + group: row.group, + plain: new Set((row.plain_chunks ?? []).map(Number)), + missing, + total, + start, + end, + capture: row.capture === true, + }) } } // Last, for loops with nothing else to do: capture ranges not stored yet, into the corpus only. - for (const target of await this.corpusTargets(rows, new Set(targets.map((t) => t.id)))) targets.push(target) + for (const target of await this.corpusTargets(rows, new Set(targets.map((t) => t.id)))) + targets.push(target) return targets } @@ -423,23 +494,38 @@ export class Prefetcher { if (!coordinatorConfig.corpus.filter) return undefined if (!this.filterRules || Date.now() - this.filterRules.at > 60_000) { const rules = await RuleSets.current() - this.filterRules = { at: Date.now(), filter: { ruleVersion: rules.version, priorities: rules.priorities } } + this.filterRules = { + at: Date.now(), + filter: { ruleVersion: rules.version, priorities: rules.priorities }, + } } return this.filterRules.filter } /** - * Up to corpus.ahead planned capture ranges, not leased soon, whose chunks aren't all in the corpus; as many of a - * group as its connections allow (perGroup / perRange): in id order the 16 were one group's, which took 4 of the - * loops and left the rest idle (30k articles/s; LOG.md 2026-10-23 (100)). + * Up to corpus.ahead capture ranges whose chunks aren't all in the corpus, by priority: planned ones not started + * (not leased soon), and done ones (a window of every group, ranges:capture, takes those indexed already too); + * as many of a group as its connections allow (perGroup / perRange): in id order the 16 were one group's, which + * took 4 of the loops and left the rest idle (30k articles/s; LOG.md 2026-10-23 (100)). A range found complete + * gets stored_at and isn't read again. */ private async corpusTargets(rows: (q: any) => any, taken: Set): Promise { const { dir, ahead } = coordinatorConfig.corpus if (!dir || ahead <= 0) return [] const { perRange, perGroup } = coordinatorConfig.spool - const ofGroup = perGroup > 0 && perRange > 0 ? Math.max(1, Math.floor(perGroup / perRange)) : Infinity + const ofGroup = + perGroup > 0 && perRange > 0 ? Math.max(1, Math.floor(perGroup / perRange)) : Infinity const groups = new Map() - const candidates = await rows(db.query()).where('r.status', 'planned').where('r.capture', true).where('r.spooled_chunks', 0).orderBy('r.id') + const candidates = await rows(db.query()) + .where('r.capture', true) + .whereNull('r.stored_at') + .where((q: any) => + q + .where((p: any) => p.where('r.status', 'planned').where('r.spooled_chunks', 0)) + .orWhere('r.status', 'done') + ) + .orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) + .limit(ahead * 16) const targets: Target[] = [] for (const row of candidates) { if (targets.length >= ahead) break @@ -451,13 +537,26 @@ export class Prefetcher { const range = { group: row.group, start, end } if (await Corpus.complete(range)) { this.stored.add(id) + await db.from('ranges').where('id', id).update({ stored_at: new Date() }) continue } const chunks = Spool.chunksOf({ start, end }) const missing: ChunkSpan[] = [] - for (const span of chunks) if (!this.claimed.has(`${id}:${span.from}`) && !(await Corpus.has(range, span))) missing.push(span) + for (const span of chunks) + if (!this.claimed.has(`${id}:${span.from}`) && !(await Corpus.has(range, span))) + missing.push(span) if (missing.length > 0) { - targets.push({ id, group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total: chunks.length, start, end, capture: true, corpusOnly: true }) + targets.push({ + id, + group: row.group, + plain: new Set((row.plain_chunks ?? []).map(Number)), + missing, + total: chunks.length, + start, + end, + capture: true, + corpusOnly: true, + }) groups.set(row.group, (groups.get(row.group) ?? 0) + 1) } } diff --git a/commands/ranges_capture.ts b/commands/ranges_capture.ts new file mode 100644 index 0000000..6e4d908 --- /dev/null +++ b/commands/ranges_capture.ts @@ -0,0 +1,80 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' + +/** Bytes a filtered corpus article takes on disk (334 GB for 18.0B articles, 2026-10-25). */ +const BYTES_PER_ARTICLE = 18.5 + +/** + * Marks every active group's ranges posted in a window (by ranges.est_date, from ranges:dates) `capture`, so the + * corpus fill fetches them: voyager 3's test data, a year of all groups at a time (REBUILD.md, corpus). Planned + * ranges not started get --priority (the fill's order, then the indexing's once there's room); done ones are + * fetched again into the corpus only. Ranges leased or half spooled are left out (their first chunks would be + * missing) and reported. + */ +export default class RangesCapture extends BaseCommand { + static commandName = 'ranges:capture' + static description = + 'Marks the ranges posted in a window for the corpus: --from 2025-01-01 --to 2026-01-01' + static options: CommandOptions = { startApp: true } + + @flags.string({ description: 'First day of the window (ISO date)', required: true }) + declare from: string + + @flags.string({ description: 'Day after the window (ISO date)', required: true }) + declare to: string + + @flags.number({ + description: 'Priority for the planned ranges marked (default 0: before the history)', + }) + declare priority?: number + + @flags.boolean({ description: 'Only report what the window holds' }) + declare dryRun: boolean + + async run() { + const from = new Date(this.from) + const to = new Date(this.to) + if (Number.isNaN(from.getTime()) || Number.isNaN(to.getTime()) || to <= from) + return this.logger.error('--from and --to: ISO dates, from before to') + const priority = this.priority ?? 0 + // A range covers est_date up to the next range's; the last range of a group up to now. + const hits = ` + WITH spans AS ( + SELECT r.id, r.status, r.capture, r.spooled_chunks, r.end_id - r.start_id + 1 AS n, r.est_date AS d0, + lead(r.est_date) OVER (PARTITION BY r.group_id ORDER BY r.start_id) AS d1 + FROM ranges r JOIN groups g ON g.id = r.group_id WHERE g.active AND r.est_date IS NOT NULL + ), hit AS (SELECT * FROM spans WHERE d0 < ? AND coalesce(d1, now()) > ?)` + const { rows } = await db.rawQuery( + `${hits} + SELECT CASE WHEN capture THEN 'marked already' WHEN status = 'planned' AND spooled_chunks = 0 THEN 'planned' + WHEN status = 'done' THEN 'done' ELSE 'in flight' END AS kind, + count(*) AS ranges, round(sum(n) / 1e9, 2) AS articles + FROM hit GROUP BY 1 ORDER BY 1`, + [to, from] + ) + const fetch = rows + .filter((r: any) => r.kind === 'planned' || r.kind === 'done') + .reduce((sum: number, r: any) => sum + Number(r.articles), 0) + this.logger.info(`${this.from} to ${this.to}:`) + for (const r of rows) + this.logger.info(` ${r.kind}: ${r.ranges} ranges, ${r.articles}B articles`) + this.logger.info( + `To fetch: ${fetch.toFixed(2)}B articles, ~${((fetch * 1e9 * BYTES_PER_ARTICLE) / 1e12).toFixed(2)} TB in the corpus.` + ) + if (this.dryRun) return + const { rows: marked } = await db.rawQuery( + `${hits} + UPDATE ranges r SET capture = true, + priority = CASE WHEN r.status = 'planned' THEN ? ELSE r.priority END + FROM hit WHERE r.id = hit.id AND NOT hit.capture + AND (hit.status = 'done' OR (hit.status = 'planned' AND hit.spooled_chunks = 0)) + RETURNING r.status`, + [to, from, priority] + ) + const planned = marked.filter((r: any) => r.status === 'planned').length + this.logger.info( + `Marked ${planned} planned ranges at priority ${priority} and ${marked.length - planned} done ranges.` + ) + } +} diff --git a/commands/ranges_dates.ts b/commands/ranges_dates.ts new file mode 100644 index 0000000..b448c0a --- /dev/null +++ b/commands/ranges_dates.ts @@ -0,0 +1,188 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' +import NntpPool from '#services/nntp_pool' +import { NntpShare } from '#services/nntp_shares' +import { estimateDate, sampleGroup, type DateSample } from '#services/group_dates' + +/** + * When each active group's articles were posted (group_dates, sampled from the news server) and from that every + * range's estimated posting date (ranges.est_date), so ranges can be chosen by era: ranges:capture marks a + * posting-time window of every group for the corpus (REBUILD.md, corpus). A group gets a sample every --every + * articles (16 … 1000): boneless (161B articles) ~320, three weeks apart; ~15k XOVER commands of 32 articles in all. + */ +export default class RangesDates extends BaseCommand { + static commandName = 'ranges:dates' + static description = + "Samples when each active group's articles were posted and estimates every range's date" + static options: CommandOptions = { startApp: true } + + @flags.string({ description: 'Only groups whose name matches this SQL LIKE pattern' }) + declare group?: string + + @flags.number({ description: 'Articles between two samples of a group (default 500M)' }) + declare every?: number + + @flags.number({ + description: + 'Connections, a share of the account beside the coordinator and the prober (default 6)', + }) + declare connections?: number + + @flags.boolean({ description: 'Only groups without samples yet' }) + declare missing: boolean + + @flags.boolean({ + description: 'No sampling: estimate the ranges from the samples stored already', + }) + declare estimateOnly: boolean + + async run() { + try { + if (!this.estimateOnly) await this.sample() + await this.estimate() + await this.report() + } finally { + await NntpPool.close() + } + } + + private groups() { + const q = db.from('groups').where('active', true).select('id', 'name').orderBy('priority') + if (this.group) q.where('name', 'like', this.group) + if (this.missing) + q.whereNotExists((sub) => + sub.from('group_dates').whereRaw('group_dates.group_id = groups.id') + ) + return q + } + + private async sample() { + const every = this.every ?? 500_000_000 + const connections = this.connections ?? 6 + const groups = await this.groups() + const share = new NntpShare( + 'dates', + coordinatorConfig.nntpShares.coordinatorPriority + 1, + 1, + connections + ) + const first = await NntpPool.share(share) + const sharing = setInterval( + () => + NntpPool.share(share).catch((error) => this.logger.warning(`NNTP share: ${error.message}`)), + coordinatorConfig.nntpShares.everyMs + ) + this.logger.info( + `Sampling ${groups.length} groups over ${first.granted} connections, a sample every ${every / 1e6}M articles.` + ) + const startedAt = Date.now() + let done = 0 + let commands = 0 + let samples = 0 + const queue = [...groups] + const worker = async () => { + for (let g = queue.shift(); g; g = queue.shift()) { + try { + const result = await NntpPool.use((client) => + sampleGroup(client, g.name, { every, size: 32 }) + ) + await db.transaction(async (trx) => { + await trx.from('group_dates').where('group_id', g.id).delete() + if (result.samples.length > 0) + await trx.table('group_dates').multiInsert( + result.samples.map((s) => ({ + group_id: g.id, + article: s.article, + posted_at: s.postedAt, + })) + ) + }) + commands += result.commands + samples += result.samples.length + if (result.samples.length < 2) + this.logger.warning( + `${g.name}: ${result.samples.length} samples over ${result.low}–${result.high}.` + ) + } catch (error: any) { + this.logger.warning(`${g.name}: ${error.message}`) + } + done++ + if (done % 50 === 0) + this.logger.info( + `${done} of ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.` + ) + } + } + try { + await Promise.all( + Array.from({ length: Math.max(1, Math.min(connections, groups.length)) }, worker) + ) + } finally { + clearInterval(sharing) + } + this.logger.info( + `Sampled ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.` + ) + } + + /** Every range of the groups sampled gets est_date from the group's curve (1000 a statement). */ + private async estimate() { + const groups = await this.groups().whereExists((sub) => + sub.from('group_dates').whereRaw('group_dates.group_id = groups.id') + ) + let updated = 0 + for (const g of groups) { + const rows = await db + .from('group_dates') + .where('group_id', g.id) + .orderBy('article') + .select('article', 'posted_at') + const samples: DateSample[] = rows.map((r) => ({ + article: Number(r.article), + postedAt: new Date(r.posted_at), + })) + const ranges = await db + .from('ranges') + .where('group_id', g.id) + .orderBy('start_id') + .select('id', 'start_id') + for (let i = 0; i < ranges.length; i += 1000) { + const batch = ranges.slice(i, i + 1000) + const values = batch.map(() => '(?::bigint, ?::timestamptz)').join(', ') + const bindings = batch.flatMap((r) => [ + Number(r.id), + estimateDate(samples, Number(r.start_id))!.toISOString(), + ]) + await db.rawQuery( + `UPDATE ranges AS r SET est_date = v.d FROM (VALUES ${values}) AS v(id, d) WHERE r.id = v.id`, + bindings + ) + updated += batch.length + } + } + this.logger.info(`${updated} ranges of ${groups.length} groups dated.`) + } + + /** The planned history by posting year: what a window would take. */ + private async report() { + const { rows } = await db.rawQuery(` + SELECT extract(year FROM est_date)::int AS year, count(*) AS ranges, count(DISTINCT group_id) AS groups, + round(sum(end_id - start_id + 1) / 1e9, 1) AS articles, + round(sum(end_id - start_id + 1) FILTER (WHERE status = 'done') / 1e9, 1) AS done, + round(sum(end_id - start_id + 1) FILTER (WHERE capture) / 1e9, 1) AS capture + FROM ranges WHERE est_date IS NOT NULL GROUP BY 1 ORDER BY 1`) + const undatedRows = await db.rawQuery( + `SELECT count(*) AS undated FROM ranges WHERE est_date IS NULL` + ) + const undated = undatedRows.rows[0].undated + this.logger.info( + `By estimated posting year (B articles; done: indexed, capture: marked for the corpus); ${undated} ranges undated:` + ) + for (const r of rows) + this.logger.info( + ` ${r.year}: ${String(r.ranges).padStart(7)} ranges of ${String(r.groups).padStart(3)} groups, ${String(r.articles).padStart(6)}B articles, ${r.done}B done, ${r.capture}B capture` + ) + } +} diff --git a/database/migrations/1770000004900_range_dates.ts b/database/migrations/1770000004900_range_dates.ts new file mode 100644 index 0000000..aa4c4a6 --- /dev/null +++ b/database/migrations/1770000004900_range_dates.ts @@ -0,0 +1,27 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * When a group's articles were posted, sampled from the news server (ranges:dates; app/services/group_dates.ts), and + * from those each range's estimated posting date, so ranges can be chosen by era: ranges:capture --from --to marks a + * posting-time window of every group for the corpus (voyager 3's test data; REBUILD.md, corpus). `stored_at`: the + * range is complete in the corpus (the prefetcher saw its range.json), so the corpus fill skips it without looking. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE IF NOT EXISTS group_dates ( + group_id integer NOT NULL REFERENCES groups (id) ON DELETE CASCADE, + article bigint NOT NULL, + posted_at timestamptz NOT NULL, + PRIMARY KEY (group_id, article) + )`) + this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS est_date timestamptz`) + this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS stored_at timestamptz`) + } + + async down() { + this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS stored_at`) + this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS est_date`) + this.schema.raw(`DROP TABLE IF EXISTS group_dates`) + } +} diff --git a/tests/unit/services/group_dates.spec.ts b/tests/unit/services/group_dates.spec.ts new file mode 100644 index 0000000..be610a8 --- /dev/null +++ b/tests/unit/services/group_dates.spec.ts @@ -0,0 +1,75 @@ +import { test } from '@japa/runner' +import type { Overview } from '#lib/nntp/index' +import { + estimateDate, + medianDate, + monotone, + sampleCount, + samplePoints, +} from '#services/group_dates' + +const now = new Date('2026-10-25T10:00:00Z') +const overview = (n: number, date: string): Overview => [n, { subject: 's', from: 'f', date }] +const day = (d: string) => new Date(`${d}T00:00:00Z`) + +test.group('group dates', () => { + test('the median of the plausible Date headers, or null for too few', ({ assert }) => { + const slice = [ + overview(1, 'Mon, 02 Mar 2015 10:00:00 +0000'), + overview(2, 'Tue, 03 Mar 2015 10:00:00 GMT'), + overview(3, 'Wed, 04 Mar 2015 10:00:00 UTC'), + // A broken clock, a future one, and junk: left out. + overview(4, 'Sat, 01 Jan 1980 00:00:00 +0000'), + overview(5, 'Fri, 01 Jan 2027 00:00:00 +0000'), + overview(6, 'not a date'), + overview(7, 'Thu, 05 Mar 2015 10:00:00 +0100'), + ] + assert.equal(medianDate(slice, now)!.toISOString(), '2015-03-04T10:00:00.000Z') + assert.isNull(medianDate(slice.slice(0, 2), now)) + assert.isNull(medianDate([], now)) + }) + + test('sample points span low to high, the last slice ending at high', ({ assert }) => { + assert.deepEqual(samplePoints(100, 1099, 5, 100), [100, 325, 550, 775, 1000]) + assert.deepEqual(samplePoints(100, 150, 5, 100), [100]) + assert.equal(sampleCount(0, 161e9, 500e6), 323) + assert.equal(sampleCount(0, 1000, 500e6), 16) + assert.equal(sampleCount(0, 10e12, 500e6), 1000) + }) + + test('the curve drops a spike, raises a dip, and keeps the order', ({ assert }) => { + const curve = monotone([ + { article: 300, postedAt: day('2015-03-01') }, + { article: 100, postedAt: day('2015-01-01') }, + // One poster's clock in the future for a whole slice. + { article: 200, postedAt: day('2019-06-01') }, + // Old crossposts: earlier than the sample before. + { article: 400, postedAt: day('2014-12-01') }, + { article: 500, postedAt: day('2015-05-01') }, + ]) + assert.deepEqual( + curve.map((s) => [s.article, s.postedAt.toISOString().slice(0, 10)]), + [ + [100, '2015-01-01'], + [300, '2015-03-01'], + [400, '2015-03-01'], + [500, '2015-05-01'], + ] + ) + }) + + test('a date is interpolated between the samples around it and clamped beyond them', ({ + assert, + }) => { + const curve = [ + { article: 100, postedAt: day('2015-01-01') }, + { article: 300, postedAt: day('2015-03-02') }, + { article: 500, postedAt: day('2015-05-01') }, + ] + assert.equal(estimateDate(curve, 200)!.toISOString().slice(0, 10), '2015-01-31') + assert.equal(estimateDate(curve, 300)!.toISOString().slice(0, 10), '2015-03-02') + assert.equal(estimateDate(curve, 1)!.toISOString().slice(0, 10), '2015-01-01') + assert.equal(estimateDate(curve, 900)!.toISOString().slice(0, 10), '2015-05-01') + assert.isNull(estimateDate([], 200)) + }) +})