Corpus: store chunks without the lines the worker only counts

History fetched ahead into the corpus (bulk HDD) is ~60 TB raw for the
1,180B missing articles, and the indexer keeps only ~38% of headers.
With CORPUS_FILTER (default on), Corpus.write leaves out the lines the
edge worker drops on their own, decided per line as RangeIndexer.line
does: a crosspost another indexed group owns, and a subject that names no
file part (~48% of the indexed ranges' headers). Release-level rules
(garbage, per-article, executable, poster) judge files across chunks and
aren't applied. The chunk is compressed again as zlib (the worker takes
it as before); beside it a .drops (gzip JSON) keeps the rule version,
the owners' priorities and per rule the article numbers (as differences,
crossposts by owner), and range.json says filtered. A test shows the
worker makes the same releases, candidates and fragments of a filtered
chunk. corpus:filter filters chunks stored raw (planned capture ranges;
--all for every range).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

REBUILD.md

@@ -230,6 +230,20 @@ range with the deployed build's edge-bench (`replay`, ~seconds a range) and
 reports suspects and the change since the previous build (`report`). A rule
 change is judged on the whole corpus before a reset.
 
+Since 2026-10-23 also history fetched ahead of the indexing: while the
+catalog is over its NVMe cap, ranges marked `capture` (priority >= 0, not
+leased) are fetched into the corpus by idle loops, at most perGroup /
+perRange of a group at a time, to be indexed from disk once there's room
+(LOG.md 2026-10-23 (100)). Stored filtered (`CORPUS_FILTER`, default on):
+without the lines the edge worker only counts, decided per line as it does
+(a crosspost another indexed group owns, a subject that names no file part;
+~48% of headers), the chunk compressed again as zlib; beside it a `.drops`
+(gzip JSON: rule version, the owners' priorities, per rule the article
+numbers as differences, crossposts by owner) and `filtered` in range.json.
+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.
+
 ## Phase 2
 
 Built: the prober (2026-10-11; `probe:run`, LOG.md 2026-10-11 (2)). It

app/services/corpus.ts

@@ -1,11 +1,22 @@
 import { access, mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises'
 import { join } from 'node:path'
-import { unzip } from 'node:zlib'
+import { deflate as deflateCb, gzip as gzipCb, unzip } from 'node:zlib'
 import { promisify } from 'node:util'
 import coordinatorConfig from '#config/coordinator'
+import { filterOverviews } from '#lib/edge/index'
 import type { ChunkSpan } from '#services/spool'
 
 const inflate = promisify(unzip)
+const deflate = promisify(deflateCb)
+const gzip = promisify(gzipCb)
+
+/** What a filtered chunk was filtered with (CORPUS_FILTER): the rule set and the group priorities of its lease. */
+export interface CorpusFilter {
+  ruleVersion: number
+  priorities: Record<string, number>
+}
+/** The rules a filtered chunk leaves out (lib/edge/corpus_filter.ts), in its range.json. */
+export const FILTERED = ['crosspost', 'obfuscated']
 
 export interface CorpusRange {
   group: string
@@ -19,7 +30,10 @@ export interface CorpusRange {
  * every chunk is there, the layout edge-bench reads (edge/fixtures). The
  * grouping sweeps replay them with each build's rules instead of waiting for
  * a re-index; the prefetcher reads a stored chunk instead of fetching it
- * again. Off without CORPUS_DIR.
+ * again. Off without CORPUS_DIR. Filtered (CORPUS_FILTER; LOG.md 2026-10-23
+ * (101)): the lines the edge worker only counts are left out, ~half of them,
+ * and each chunk's `.drops` beside it (gzip JSON, ChunkDrops) lists their
+ * article numbers by rule; its range.json says `filtered`.
  */
 class Corpus {
   get dir(): string | undefined {
@@ -61,28 +75,40 @@ class Corpus {
 
   /**
    * Stores a chunk that inflates (a reply cut short isn't kept), and the
-   * range.json once the range is complete (`total` chunks).
+   * range.json once the range is complete (`total` chunks). With `filter`,
+   * without the lines the edge worker drops on their own, compressed again
+   * as the news server's (zlib), with its `.drops` written first.
    */
-  async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number) {
+  async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number, filter?: CorpusFilter) {
     if (!this.dir) return
+    let raw: Buffer
     try {
-      await inflate(data)
+      raw = await inflate(data)
     } catch {
       return
     }
     const dir = this.rangeDir(range)
     const file = this.path(range, span)
     await mkdir(dir, { recursive: true })
-    await writeFile(`${file}.part`, data)
+    let stored = data
+    if (filter) {
+      // Latin-1 both ways: every byte back as it was.
+      const { kept, drops } = filterOverviews(raw.toString('latin1'), range.group, filter.priorities, filter.ruleVersion)
+      stored = await deflate(Buffer.from(kept, 'latin1'), { level: 9 })
+      await writeFile(`${file}.drops.part`, await gzip(JSON.stringify(drops)))
+      await rename(`${file}.drops.part`, `${file}.drops`)
+    }
+    await writeFile(`${file}.part`, stored)
     await rename(`${file}.part`, file)
     const chunks = (await readdir(dir)).filter((f) => f.endsWith('.gz')).length
-    if (chunks >= total) await writeFile(join(dir, 'range.json'), JSON.stringify(range))
+    if (chunks >= total) await writeFile(join(dir, 'range.json'), JSON.stringify(filter ? { ...range, filtered: FILTERED } : range))
   }
 
   /** A chunk a worker couldn't inflate after all: not served again, and the range not complete. */
   async removeChunk(range: CorpusRange, span: ChunkSpan) {
     if (!this.dir) return
     await rm(this.path(range, span), { force: true })
+    await rm(`${this.path(range, span)}.drops`, { force: true })
     await rm(join(this.rangeDir(range), 'range.json'), { force: true })
   }
 }

app/services/prefetcher.ts

@@ -3,7 +3,8 @@ import logger from '@adonisjs/core/services/logger'
 import coordinatorConfig from '#config/coordinator'
 import NntpPool from '#services/nntp_pool'
 import Spool, { type ChunkSpan } from '#services/spool'
-import Corpus from '#services/corpus'
+import Corpus, { type CorpusFilter } from '#services/corpus'
+import RuleSets from '#services/rule_sets'
 import CatalogSpace from '#services/catalog_space'
 import { hedge, type Request, type XoverClient } from '#lib/nntp/index'
 
@@ -175,7 +176,7 @@ 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)
+          if (target.capture) await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter())
         }
         if (target.corpusOnly) {
           this.stats.captured++
@@ -415,6 +416,18 @@ export class Prefetcher {
     return targets
   }
 
+  private filterRules: { at: number; filter: CorpusFilter } | null = null
+
+  /** The rules corpus chunks are filtered with (CORPUS_FILTER), read again once a minute; undefined when off. */
+  private async corpusFilter(): Promise<CorpusFilter | undefined> {
+    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 } }
+    }
+    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

commands/corpus_filter.ts

@@ -0,0 +1,54 @@
+import { BaseCommand, flags } from '@adonisjs/core/ace'
+import type { CommandOptions } from '@adonisjs/core/types/ace'
+import { access, readFile, readdir, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import db from '@adonisjs/lucid/services/db'
+import Corpus, { FILTERED } from '#services/corpus'
+import RuleSets from '#services/rule_sets'
+import coordinatorConfig from '#config/coordinator'
+
+/**
+ * Filters corpus chunks stored as the news server sent them (before CORPUS_FILTER, 2026-10-23): each chunk without a
+ * `.drops` of the planned capture ranges (--all: of every range in the corpus) written again as Corpus.write would.
+ */
+export default class CorpusFilterCommand extends BaseCommand {
+  static commandName = 'corpus:filter'
+  static description = 'Filters stored corpus chunks (planned capture ranges; --all: every range) as CORPUS_FILTER does'
+  static options: CommandOptions = { startApp: true }
+
+  @flags.boolean({ description: 'Every range in the corpus, also done ones (the grouping sweeps read them raw)' })
+  declare all: boolean
+
+  async run() {
+    const dir = coordinatorConfig.corpus.dir
+    if (!dir) return this.logger.error('CORPUS_DIR is not set')
+    const rules = await RuleSets.current()
+    const filter = { ruleVersion: rules.version, priorities: rules.priorities }
+    const query = db.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').where('r.capture', true).select('g.name as group', 'r.start_id', 'r.end_id')
+    const ranges = await (this.all ? query : query.where('r.status', 'planned'))
+    let chunks = 0
+    let before = 0
+    let after = 0
+    for (const r of ranges) {
+      const range = { group: r.group, start: Number(r.start_id), end: Number(r.end_id) }
+      const rangeDir = join(dir, range.group, `${range.start}-${range.end}`)
+      const files = await readdir(rangeDir).catch(() => [] as string[])
+      const gz = files.filter((f) => f.endsWith('.gz'))
+      for (const name of gz) {
+        if (files.includes(`${name}.drops`)) continue
+        const m = /^\d{4}-(\d+)-(\d+)\.gz$/.exec(name)
+        if (!m) continue
+        const data = await readFile(join(rangeDir, name))
+        await Corpus.write(range, { from: Number(m[1]), to: Number(m[2]) }, data, Number.POSITIVE_INFINITY, filter)
+        before += data.length
+        after += (await readFile(join(rangeDir, name))).length
+        chunks++
+      }
+      // A complete range's range.json says it's filtered now (Corpus.write only writes it at the last chunk).
+      const json = join(rangeDir, 'range.json')
+      if (gz.length > 0 && (await access(json).then(() => true, () => false))) await writeFile(json, JSON.stringify({ ...range, filtered: FILTERED }))
+      if (chunks > 0 && chunks % 1000 < gz.length) this.logger.info(`${chunks} chunks filtered: ${(before / 1e9).toFixed(1)} → ${(after / 1e9).toFixed(1)} GB.`)
+    }
+    this.logger.info(`Done: ${chunks} chunks of ${ranges.length} ranges filtered, ${(before / 1e9).toFixed(1)} → ${(after / 1e9).toFixed(1)} GB.`)
+  }
+}

config/coordinator.ts

@@ -131,6 +131,12 @@ export default {
      * ranges waited behind it for hours (1,400 marked, none stored, 34 of 40 connections idle; LOG.md 2026-10-12 (27)).
      */
     ahead: env.get('CORPUS_AHEAD', 16),
+    /**
+     * Stored without the lines the edge worker drops on their own (crossposts another indexed group owns, obfuscated
+     * subjects; ~48% of the indexed ranges' headers), what was left out in each chunk's `.drops`: twice the history
+     * in the same space (LOG.md 2026-10-23 (101)). Off: as the news server sent them.
+     */
+    filter: env.get('CORPUS_FILTER', true),
   },
   spool: {
     dir: env.get('SPOOL_DIR', app.tmpPath('spool')),

lib/edge/corpus_filter.ts

@@ -0,0 +1,78 @@
+import { SegmentScanner, scanOverviewText } from './scan.js'
+
+/**
+ * What a corpus chunk left out (REBUILD.md, corpus; LOG.md 2026-10-23 (101)): the overview lines the edge worker only
+ * counts and never indexes, decided per line as RangeIndexer.line does: a crosspost an indexed group with a lower
+ * priority number owns (by owner), and a subject that names no file part (obfuscated). Article numbers ascending,
+ * stored as differences. Release-level rules (garbage, per-article, executable, poster) judge files put together
+ * across chunks, so they aren't applied here.
+ */
+export interface ChunkDrops {
+  version: 1
+  /** The rule set (rule_sets.version) and group priorities the decision was made with. */
+  ruleVersion: number
+  priorities: Record<string, number>
+  /** Overview lines in the chunk, and kept. */
+  lines: number
+  kept: number
+  /** By owning group, article numbers as differences (the first from 0). */
+  crosspost: Record<string, number[]>
+  obfuscated: number[]
+}
+
+const deltas = (numbers: number[]) => numbers.map((n, i) => (i === 0 ? n : n - numbers[i - 1]))
+
+/** The article numbers back from ChunkDrops differences. */
+export function undelta(differences: number[]): number[] {
+  let at = 0
+  return differences.map((d) => (at += d))
+}
+
+/**
+ * A chunk's overview text without the lines the edge worker drops on their own, and what was dropped. Lines kept
+ * byte for byte (with their line ends); lines the worker skips uncounted (no number or message-id) are kept.
+ */
+export function filterOverviews(text: string, group: string, priorities: Record<string, number>, ruleVersion: number): { kept: string; drops: ChunkDrops } {
+  const scanner = new SegmentScanner()
+  const crosspost = new Map<string, number[]>()
+  const obfuscated: number[] = []
+  const dropped = new Set<number>()
+  let lines = 0
+  scanOverviewText(text, (n, subject, from, date, messageId, bytes, xref) => {
+    lines++
+    if (!Number.isFinite(n) || !messageId) return
+    const owner = xref ? scanner.owner(xref, group, priorities) : group
+    if (owner !== group) {
+      const list = crosspost.get(owner)
+      if (list) list.push(n)
+      else crosspost.set(owner, [n])
+      dropped.add(n)
+    } else if (!scanner.segment(n, subject, from, date, messageId, bytes, xref, group)) {
+      obfuscated.push(n)
+      dropped.add(n)
+    }
+  })
+  const out: string[] = []
+  let start = 0
+  while (start < text.length) {
+    let end = text.indexOf('\n', start)
+    end = end === -1 ? text.length : end + 1
+    const tab = text.indexOf('\t', start)
+    const n = tab !== -1 && tab < end ? Number.parseInt(text.slice(start, tab), 10) : Number.NaN
+    if (!dropped.has(n)) out.push(text.slice(start, end))
+    start = end
+  }
+  const sorted = (list: number[]) => [...list].sort((a, b) => a - b)
+  return {
+    kept: out.join(''),
+    drops: {
+      version: 1,
+      ruleVersion,
+      priorities: Object.fromEntries(Object.entries(priorities).filter(([name]) => crosspost.has(name) || name === group)),
+      lines,
+      kept: lines - dropped.size,
+      crosspost: Object.fromEntries([...crosspost].map(([owner, list]) => [owner, deltas(sorted(list))])),
+      obfuscated: deltas(sorted(obfuscated)),
+    },
+  }
+}

lib/edge/index.ts

@@ -5,3 +5,4 @@ export type { Lease } from './lease.js'
 export { mergeFragments } from './merge.js'
 export type { EdgeSegment, FileMeta } from './scan.js'
 export { throttle } from './throttle.js'
+export { filterOverviews, undelta, type ChunkDrops } from './corpus_filter.js'

start/env.ts

@@ -51,6 +51,7 @@ export default await Env.create(new URL('../', import.meta.url), {
   NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(),
   NNTP_CONNECTION_MAX_MINUTES: Env.schema.number.optional(),
   NNTP_IDLE_CLOSE_SECONDS: Env.schema.number.optional(),
+  CORPUS_FILTER: Env.schema.boolean.optional(),
   NNTP_STALL_KB_PER_MINUTE: Env.schema.number.optional(),
 
   // Newznab API key (empty: no key needed)

tests/unit/edge/corpus_filter.spec.ts

@@ -0,0 +1,83 @@
+import { test } from '@japa/runner'
+import { mkdtemp, readFile, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { deflateSync, gunzipSync, inflateSync } from 'node:zlib'
+import coordinatorConfig from '#config/coordinator'
+import Corpus from '#services/corpus'
+import { RangeIndexer, filterOverviews, undelta } from '#lib/edge/index'
+import { DEFAULT_RULES } from '#lib/rules/index'
+
+const GROUP = 'alt.binaries.test'
+const priorities = { 'alt.binaries.boneless': 1, [GROUP]: 2, 'alt.binaries.x': 3 }
+
+/** An overview line as XOVER sends it (number, subject, from, date, message-id, references, bytes, lines, Xref). */
+const line = (n: number, subject: string, xref = `news.example ${GROUP}:${n}`, messageId = `<a${n}@test>`) =>
+  [n, subject, 'poster <p@x>', 'Mon, 05 Oct 2026 10:00:00 +0000', messageId, '', '700000', '5000', `Xref: ${xref}`].join('\t')
+
+const LINES = [
+  line(100, 'Some.Show.S01E01 [1/2] - "Some.Show.S01E01.1080p.part1.rar" yEnc (1/2)'),
+  // Owned by boneless (priority 1): the worker counts it there.
+  line(101, 'Other [1/1] - "Other.mkv" yEnc (1/1)', `news.example ${GROUP}:101 alt.binaries.boneless:9`),
+  line(102, 'Some.Show.S01E01 [1/2] - "Some.Show.S01E01.1080p.part1.rar" yEnc (2/2)'),
+  // No file part in the subject.
+  line(103, 'obfuscated'),
+  // Also in a lower-priority group (3): ours.
+  line(104, 'Some.Show.S01E01 [2/2] - "Some.Show.S01E01.1080p.par2" yEnc (1/1)', `news.example ${GROUP}:104 alt.binaries.x:7`),
+  line(105, 'Pq4ZmXc8VbN2LkJh6GfDs3'),
+  // Without a message-id: the worker skips it uncounted, so it stays.
+  line(106, 'x', undefined, ''),
+]
+const TEXT = LINES.map((l) => `${l}\r\n`).join('')
+
+function index(text: string) {
+  const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities }, edgeArticles: 10 })
+  indexer.addText(text, 1001)
+  indexer.finish()
+  return indexer.take()
+}
+
+test.group('edge / corpus filter', () => {
+  test('leaves out crossposts another group owns and obfuscated subjects, keeps the rest byte for byte', ({ assert }) => {
+    const { kept, drops } = filterOverviews(TEXT, GROUP, priorities, 7)
+    assert.equal(kept, [LINES[0], LINES[2], LINES[4], LINES[6]].map((l) => `${l}\r\n`).join(''))
+    assert.equal(drops.ruleVersion, 7)
+    assert.equal(drops.lines, 7)
+    assert.equal(drops.kept, 4)
+    assert.deepEqual(Object.keys(drops.crosspost), ['alt.binaries.boneless'])
+    assert.deepEqual(undelta(drops.crosspost['alt.binaries.boneless']), [101])
+    assert.deepEqual(undelta(drops.obfuscated), [103, 105])
+    assert.deepEqual(drops.priorities, { 'alt.binaries.boneless': 1, [GROUP]: 2 })
+  })
+
+  test('the worker makes the same releases, candidates and fragments of the filtered chunk', ({ assert }) => {
+    const raw = index(TEXT)
+    const filtered = index(filterOverviews(TEXT, GROUP, priorities, 7).kept)
+    assert.isAbove(raw.releases.length, 0)
+    assert.deepEqual(filtered.releases, raw.releases)
+    assert.deepEqual(filtered.candidates, raw.candidates)
+    assert.deepEqual(filtered.fragments, raw.fragments)
+    // Only the counts of what was left out differ.
+    const without = (b: typeof raw) => b.counts.filter((c) => c.rule !== 'crosspost' && c.rule !== 'obfuscated')
+    assert.deepEqual(without(filtered), without(raw))
+  })
+
+  test('the corpus stores a filtered chunk as zlib with its .drops, and says so in range.json', async ({ assert, cleanup }) => {
+    const dir = await mkdtemp(join(tmpdir(), 'corpus-'))
+    const before = coordinatorConfig.corpus.dir
+    coordinatorConfig.corpus.dir = dir
+    cleanup(async () => {
+      coordinatorConfig.corpus.dir = before
+      await rm(dir, { recursive: true, force: true })
+    })
+    const range = { group: GROUP, start: 1, end: 100_000 }
+    const span = { from: 1, to: 100_000 }
+    await Corpus.write(range, span, deflateSync(Buffer.from(TEXT, 'latin1')), 1, { ruleVersion: 7, priorities })
+    const stored = (await Corpus.read(range, span))!
+    assert.equal(inflateSync(stored).toString('latin1'), filterOverviews(TEXT, GROUP, priorities, 7).kept)
+    const file = join(dir, GROUP, '1-100000', '0000-1-100000.gz')
+    const drops = JSON.parse(gunzipSync(await readFile(`${file}.drops`)).toString())
+    assert.deepEqual(undelta(drops.obfuscated), [103, 105])
+    assert.deepEqual(JSON.parse(await readFile(join(dir, GROUP, '1-100000', 'range.json'), 'utf8')), { ...range, filtered: ['crosspost', 'obfuscated'] })
+  })
+})