Merge branch 'rebuild' into voyager-3

REBUILD.md

@@ -245,6 +245,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

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<number>()
+  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 }
+}

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. */

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<number>()
 
@@ -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<Buffer> {
     let age: number | undefined
-    const request = (start: (work: (client: XoverClient) => Promise<Buffer>) => Promise<Buffer> | null, first: boolean): Request<Buffer> | null => {
+    const request = (
+      start: (work: (client: XoverClient) => Promise<Buffer>) => Promise<Buffer> | null,
+      first: boolean
+    ): Request<Buffer> | 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<number>): Promise<Target[]> {
     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<string, number>()
-    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)
       }
     }

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.`
+    )
+  }
+}

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`
+      )
+  }
+}

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`)
+  }
+}

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))
+  })
+})