Rebuild: the batch store (catalog upserts, write-once chunks, candidates, rule counts, fragments, resume point)

A batch is stored in one transaction and only at its range's resume
point, so a retried batch can't count twice. Garbage is judged before
per-article: a dropped tool isn't worth reading bodies for either.

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

REBUILD.md

@@ -65,8 +65,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
 
 3. **Chunk store** (bulk, append-only). One record per release per batch:
    its files' names, sizes and segment lists, compactly encoded (template
-   codec for message-ids, zstd). Written once, never updated; a re-run of a
-   range replaces that range's chunks (no per-file dedup). ~3 TB.
+   codec for message-ids, zstd). Written once, never updated. A batch is
+   stored in one transaction and only when it starts at its range's resume
+   point, so a batch sent twice or late is refused and the catalog's sums
+   stay right (no per-file dedup); a recovery run for a retracted rule only
+   adds what that rule had dropped. ~3 TB.
 
 4. **Release catalog** (hot, Postgres on NVMe). One row per release: key,
    name, poster, groups, posted_at, file count, size, segments, expected

app/services/batch_store.ts

@@ -0,0 +1,116 @@
+import db from '@adonisjs/lucid/services/db'
+import type { EdgeBatch } from '#lib/edge/index'
+import ChunkPartitions from '#services/chunk_partitions'
+
+export class StaleBatch extends Error {}
+
+export interface StoredBatch {
+  releases: number
+  newReleases: number
+  candidates: number
+  fragments: number
+}
+
+/**
+ * Stores one batch of a leased range (REBUILD.md, parts 3–7), in one
+ * transaction: release aggregates in the catalog, one chunk per release,
+ * probe candidates, rule counts and samples, edge fragments, and the
+ * range's resume point. A batch that doesn't start at the resume point was
+ * sent twice or too late, and is refused (StaleBatch), so the sums stay right.
+ */
+class BatchStore {
+  async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise<StoredBatch> {
+    await ChunkPartitions.ensure()
+    return db.transaction(async (trx) => {
+      const range = await trx.from('ranges').where('id', rangeId).forUpdate().first()
+      if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
+      if (Number(range.next_id) !== batch.from) throw new StaleBatch(`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`)
+      const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, candidates: 0, fragments: batch.fragments.length }
+
+      // Releases, sorted by key so concurrent batches lock them in the same order.
+      const releases = [...batch.releases].sort((a, b) => a.key.localeCompare(b.key))
+      const ids = new Map<string, number>()
+      for (let i = 0; i < releases.length; i += 1000) {
+        const chunk = releases.slice(i, i + 1000)
+        const { rows } = await trx.rawQuery(
+          `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2)
+           SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2
+           FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)
+           ON CONFLICT (release_key) DO UPDATE SET
+             files = releases.files + excluded.files,
+             segments = releases.segments + excluded.segments,
+             bytes = releases.bytes + excluded.bytes,
+             chunks = releases.chunks + 1,
+             posted_at = least(releases.posted_at, excluded.posted_at),
+             file_total = coalesce(releases.file_total, excluded.file_total),
+             first_article = least(releases.first_article, excluded.first_article),
+             last_article = greatest(releases.last_article, excluded.last_article),
+             readable = releases.readable OR excluded.readable,
+             has_par2 = releases.has_par2 OR excluded.has_par2,
+             last_seen_at = now()
+           RETURNING id, release_key, (xmax = 0) AS inserted`,
+          [
+            JSON.stringify(
+              chunk.map((r) => ({
+                key: r.key, name: r.name, source: r.source, poster: r.poster, groups: r.groups, posted_at: r.postedAt, file_total: r.fileTotal,
+                files: r.files, segments: r.segments, bytes: r.bytes, first_article: r.firstArticle, last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2,
+              }))
+            ),
+          ]
+        )
+        for (const row of rows) {
+          ids.set(row.release_key, Number(row.id))
+          if (row.inserted) result.newReleases++
+        }
+      }
+
+      // One write-once chunk per release.
+      for (let i = 0; i < releases.length; i += 500) {
+        const rows = releases.slice(i, i + 500).map((r) => ({
+          release_id: ids.get(r.key)!,
+          range_id: rangeId,
+          batch_seq: batch.from,
+          files: r.files,
+          segments: r.segments,
+          payload: r.payload,
+        }))
+        if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows)
+      }
+
+      const candidates = batch.candidates.filter((c) => ids.has(c.releaseKey))
+      for (let i = 0; i < candidates.length; i += 2000) {
+        const rows = candidates.slice(i, i + 2000).map((c) => ({ release_id: ids.get(c.releaseKey)!, kind: c.kind, message_id: c.messageId, file: c.file }))
+        const added = await trx.knexQuery().table('probe_candidates').insert(rows).onConflict(['release_id', 'kind', 'message_id']).ignore().returning('id')
+        result.candidates += added.length
+      }
+
+      if (batch.counts.length > 0) {
+        await trx.rawQuery(
+          `INSERT INTO rule_counts (range_id, rule, key, articles, files)
+           SELECT ?, v.rule, v.key, v.articles, v.files FROM jsonb_to_recordset(?::jsonb) AS v(rule text, key text, articles bigint, files bigint)
+           ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`,
+          [rangeId, JSON.stringify(batch.counts)]
+        )
+      }
+
+      for (const s of batch.samples) {
+        await trx.insertQuery().table('rule_samples').insert({
+          range_id: rangeId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, name: s.name, files: s.files, segments: s.segments, payload: s.payload,
+        })
+      }
+
+      for (let i = 0; i < batch.fragments.length; i += 2000) {
+        const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ range_id: rangeId, file_key: f.fileKey, article_number: f.articleNumber, segment: JSON.stringify(f) }))
+        await trx.insertQuery().table('fragments').multiInsert(rows)
+      }
+
+      await trx
+        .from('ranges')
+        .where('id', rangeId)
+        .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date() })
+      return result
+    })
+  }
+}
+
+export default new BatchStore()

app/services/chunk_partitions.ts

@@ -0,0 +1,32 @@
+import db from '@adonisjs/lucid/services/db'
+import env from '#start/env'
+
+/** Months ahead that always have a release_chunks partition. */
+const AHEAD = 2
+
+/**
+ * release_chunks is partitioned by month of writing (REBUILD.md, decision a).
+ * Inserts fail without a partition for now(), so the current and the next
+ * months' partitions are created before they are needed; checked once per
+ * process and month.
+ */
+class ChunkPartitions {
+  private ensuredFor: string | null = null
+
+  async ensure(now = new Date()) {
+    const month = now.toISOString().slice(0, 7)
+    if (this.ensuredFor === month) return
+    const tablespace = env.get('CHUNK_TABLESPACE')
+    for (let i = 0; i <= AHEAD; i++) {
+      const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1))
+      const to = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i + 1, 1))
+      const name = `release_chunks_${from.toISOString().slice(0, 7).replace('-', '')}`
+      await db.rawQuery(
+        `CREATE TABLE IF NOT EXISTS ${name} PARTITION OF release_chunks FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')${tablespace ? ` TABLESPACE ${tablespace}` : ''}`
+      )
+    }
+    this.ensuredFor = month
+  }
+}
+
+export default new ChunkPartitions()

config/database.ts

@@ -18,8 +18,6 @@ const dbConfig = defineConfig({
         password: env.get('DB_PASSWORD'),
         database: env.get('DB_DATABASE'),
       },
-      // Local development may share a database, in its own schema (DB_SEARCH_PATH=rebuild).
-      ...(env.get('DB_SEARCH_PATH') ? { searchPath: env.get('DB_SEARCH_PATH')!.split(',') } : {}),
       pool: { min: 0, max: env.get('DB_POOL_MAX', 10) },
       migrations: {
         naturalSort: true,

database/migrations/1770000000100_create_groups.ts

@@ -0,0 +1,27 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The indexed groups (REBUILD.md, group list). `priority` decides a
+ * crosspost's owner: the indexed group in its Xref with the lowest number.
+ * New groups get numbers after the existing ones, so they never take
+ * articles from groups already done.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('groups', (table) => {
+      table.increments('id')
+      table.string('name', 255).notNullable().unique()
+      table.integer('priority').notNullable().unique()
+      table.boolean('active').notNullable().defaultTo(true)
+      // From GROUP when planned: the server's oldest and newest article.
+      table.bigInteger('low').nullable()
+      table.bigInteger('high').nullable()
+      table.timestamp('planned_at', { useTz: true }).nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('groups')
+  }
+}

database/migrations/1770000000200_create_rule_sets.ts

@@ -0,0 +1,21 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Versions of what workers drop (REBUILD.md, rules and recovery): dropped
+ * signatures and tools, the per-article limit, the sample share. Every range
+ * records the version it ran with.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_sets', (table) => {
+      table.integer('version').primary()
+      table.jsonb('rules').notNullable()
+      table.text('note').nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_sets')
+  }
+}

database/migrations/1770000000300_create_ranges.ts

@@ -0,0 +1,38 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * A group's history in leasable pieces (REBUILD.md, coordinator):
+ * planned → (spooled by the prefetcher) → leased → done. `next_id` is the
+ * resume point: everything before it has been stored.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('ranges', (table) => {
+      table.bigIncrements('id')
+      table.integer('group_id').notNullable().references('id').inTable('groups').onDelete('CASCADE')
+      table.bigInteger('start_id').notNullable()
+      table.bigInteger('end_id').notNullable()
+      table.bigInteger('next_id').notNullable()
+      table.string('status', 8).notNullable().defaultTo('planned')
+      // Lower first; the coordinator plans the oldest era first.
+      table.integer('priority').notNullable().defaultTo(0)
+      table.integer('rule_version').nullable()
+      table.integer('node_id').nullable()
+      table.integer('attempts').notNullable().defaultTo(0)
+      // Chunks of this range in the spool (prefetch); all of them when spooled_at is set.
+      table.integer('spooled_chunks').notNullable().defaultTo(0)
+      table.timestamp('spooled_at', { useTz: true }).nullable()
+      table.timestamp('leased_at', { useTz: true }).nullable()
+      table.timestamp('heartbeat_at', { useTz: true }).nullable()
+      table.timestamp('done_at', { useTz: true }).nullable()
+      table.bigInteger('headers').notNullable().defaultTo(0)
+      table.text('error').nullable()
+      table.unique(['group_id', 'start_id'])
+    })
+    this.schema.raw(`CREATE INDEX ranges_todo_idx ON ranges (priority, id) WHERE status IN ('planned', 'leased')`)
+  }
+
+  async down() {
+    this.schema.dropTable('ranges')
+  }
+}

database/migrations/1770000000400_create_releases.ts

@@ -0,0 +1,41 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The release catalog (REBUILD.md, part 4): one row per release, the hot
+ * part on fast disk. Updated with aggregates once per batch; the files
+ * themselves are in release_chunks.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('releases', (table) => {
+      table.bigIncrements('id')
+      table.string('release_key', 40).notNullable().unique()
+      table.text('name').notNullable()
+      table.string('source', 16).notNullable()
+      table.text('poster').nullable()
+      table.jsonb('groups').notNullable()
+      table.timestamp('posted_at', { useTz: true }).nullable()
+      // Files announced by the [i/N] counter, when present.
+      table.integer('file_total').nullable()
+      table.integer('files').notNullable().defaultTo(0)
+      table.bigInteger('segments').notNullable().defaultTo(0)
+      table.bigInteger('bytes').notNullable().defaultTo(0)
+      table.integer('chunks').notNullable().defaultTo(0)
+      table.bigInteger('first_article').nullable()
+      table.bigInteger('last_article').nullable()
+      table.boolean('readable').notNullable().defaultTo(false)
+      table.boolean('has_par2').notNullable().defaultTo(false)
+      table.boolean('complete').nullable()
+      table.timestamp('settled_at', { useTz: true }).nullable()
+      table.timestamp('first_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.timestamp('last_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+    this.schema.raw(`CREATE EXTENSION IF NOT EXISTS pg_trgm`)
+    this.schema.raw(`CREATE INDEX releases_name_trgm ON releases USING gin (name gin_trgm_ops)`)
+    this.schema.raw(`CREATE INDEX releases_posted_at_idx ON releases (posted_at DESC)`)
+  }
+
+  async down() {
+    this.schema.dropTable('releases')
+  }
+}

database/migrations/1770000000500_create_release_chunks.ts

@@ -0,0 +1,34 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+import env from '#start/env'
+
+/**
+ * The chunk store (REBUILD.md, part 3; decision a): one write-once row per
+ * release per batch, its files encoded by lib/chunks. Partitioned by month
+ * of writing, in CHUNK_TABLESPACE (the HDD) when set; ChunkPartitions keeps
+ * the next months' partitions ready. A re-run of a range replaces its chunks
+ * (by range_id).
+ */
+export default class extends BaseSchema {
+  async up() {
+    const tablespace = env.get('CHUNK_TABLESPACE')
+    const where = tablespace ? ` TABLESPACE ${tablespace}` : ''
+    this.schema.raw(`
+      CREATE TABLE release_chunks (
+        id bigserial,
+        release_id bigint NOT NULL,
+        range_id bigint NOT NULL,
+        batch_seq integer NOT NULL,
+        files integer NOT NULL,
+        segments integer NOT NULL,
+        payload bytea NOT NULL,
+        created_at timestamptz NOT NULL DEFAULT now(),
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)${where}`)
+    this.schema.raw(`CREATE INDEX release_chunks_release_idx ON release_chunks (release_id)`)
+    this.schema.raw(`CREATE INDEX release_chunks_range_idx ON release_chunks (range_id)`)
+  }
+
+  async down() {
+    this.schema.raw(`DROP TABLE IF EXISTS release_chunks CASCADE`)
+  }
+}

database/migrations/1770000000600_create_fragments.ts

@@ -0,0 +1,24 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Segments of files cut off by a range's edge (REBUILD.md, fragments): kept
+ * by range, merged when the neighbouring range is done, then written as a
+ * chunk.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('fragments', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('file_key', 40).notNullable()
+      table.bigInteger('article_number').notNullable()
+      table.jsonb('segment').notNullable()
+      table.index(['file_key'])
+      table.index(['range_id'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('fragments')
+  }
+}

database/migrations/1770000000700_create_probe_candidates.ts

@@ -0,0 +1,25 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The probe queue (REBUILD.md, part 6): per release the few articles phase 2
+ * should read. Phase 1 only fills it.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('probe_candidates', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('release_id').notNullable()
+      table.string('kind', 8).notNullable()
+      table.text('message_id').notNullable()
+      table.text('file').notNullable()
+      table.string('status', 8).notNullable().defaultTo('waiting')
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.unique(['release_id', 'kind', 'message_id'])
+    })
+    this.schema.raw(`CREATE INDEX probe_candidates_waiting_idx ON probe_candidates (id) WHERE status = 'waiting'`)
+  }
+
+  async down() {
+    this.schema.dropTable('probe_candidates')
+  }
+}

database/migrations/1770000000800_create_rule_bookkeeping.ts

@@ -0,0 +1,35 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * What the rules dropped, per range (REBUILD.md, rules and recovery), and
+ * the sample of it kept in full.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_counts', (table) => {
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('key').notNullable()
+      table.bigInteger('articles').notNullable().defaultTo(0)
+      table.bigInteger('files').notNullable().defaultTo(0)
+      table.primary(['range_id', 'rule', 'key'])
+    })
+    this.schema.createTable('rule_samples', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('rule_key').notNullable()
+      table.string('release_key', 40).notNullable()
+      table.text('name').notNullable()
+      table.integer('files').notNullable()
+      table.integer('segments').notNullable()
+      table.binary('payload').notNullable()
+      table.index(['rule', 'rule_key'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_samples')
+    this.schema.dropTable('rule_counts')
+  }
+}

database/migrations/1770000000900_create_nodes.ts

@@ -0,0 +1,30 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/** Edge workers and their API tokens (the `node` guard). */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('nodes', (table) => {
+      table.increments('id')
+      table.string('name', 64).notNullable().unique()
+      table.timestamp('last_seen_at', { useTz: true }).nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+    this.schema.createTable('node_access_tokens', (table) => {
+      table.increments('id')
+      table.integer('tokenable_id').notNullable().unsigned().references('id').inTable('nodes').onDelete('CASCADE')
+      table.string('type').notNullable()
+      table.string('name').nullable()
+      table.string('hash').notNullable()
+      table.text('abilities').notNullable()
+      table.timestamp('created_at')
+      table.timestamp('updated_at')
+      table.timestamp('last_used_at').nullable()
+      table.timestamp('expires_at').nullable()
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('node_access_tokens')
+    this.schema.dropTable('nodes')
+  }
+}

database/schema.ts

@@ -32,6 +32,255 @@ export class AuthAccessTokenSchema extends BaseModel {
   declare updatedAt: DateTime | null
 }
 
+export class FragmentSchema extends BaseModel {
+  static $columns = ['articleNumber', 'fileKey', 'id', 'rangeId', 'segment'] as const
+  $columns = FragmentSchema.$columns
+  @column()
+  declare articleNumber: bigint | number
+  @column()
+  declare fileKey: string
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column()
+  declare rangeId: bigint | number
+  @column()
+  declare segment: any
+}
+
+export class GroupSchema extends BaseModel {
+  static $columns = ['active', 'createdAt', 'high', 'id', 'low', 'name', 'plannedAt', 'priority'] as const
+  $columns = GroupSchema.$columns
+  @column()
+  declare active: boolean
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime
+  @column()
+  declare high: bigint | number | null
+  @column({ isPrimary: true })
+  declare id: number
+  @column()
+  declare low: bigint | number | null
+  @column()
+  declare name: string
+  @column.dateTime()
+  declare plannedAt: DateTime | null
+  @column()
+  declare priority: number
+}
+
+export class NodeAccessTokenSchema extends BaseModel {
+  static $columns = ['abilities', 'createdAt', 'expiresAt', 'hash', 'id', 'lastUsedAt', 'name', 'tokenableId', 'type', 'updatedAt'] as const
+  $columns = NodeAccessTokenSchema.$columns
+  @column()
+  declare abilities: string
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime | null
+  @column.dateTime()
+  declare expiresAt: DateTime | null
+  @column()
+  declare hash: string
+  @column({ isPrimary: true })
+  declare id: number
+  @column.dateTime()
+  declare lastUsedAt: DateTime | null
+  @column()
+  declare name: string | null
+  @column()
+  declare tokenableId: number
+  @column()
+  declare type: string
+  @column.dateTime({ autoCreate: true, autoUpdate: true })
+  declare updatedAt: DateTime | null
+}
+
+export class NodeSchema extends BaseModel {
+  static $columns = ['createdAt', 'id', 'lastSeenAt', 'name'] as const
+  $columns = NodeSchema.$columns
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime
+  @column({ isPrimary: true })
+  declare id: number
+  @column.dateTime()
+  declare lastSeenAt: DateTime | null
+  @column()
+  declare name: string
+}
+
+export class ProbeCandidateSchema extends BaseModel {
+  static $columns = ['createdAt', 'file', 'id', 'kind', 'messageId', 'releaseId', 'status'] as const
+  $columns = ProbeCandidateSchema.$columns
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime
+  @column()
+  declare file: string
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column()
+  declare kind: string
+  @column()
+  declare messageId: string
+  @column()
+  declare releaseId: bigint | number
+  @column()
+  declare status: string
+}
+
+export class RangeSchema extends BaseModel {
+  static $columns = ['attempts', 'doneAt', 'endId', 'error', 'groupId', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'priority', 'ruleVersion', 'spooledAt', 'spooledChunks', 'startId', 'status'] as const
+  $columns = RangeSchema.$columns
+  @column()
+  declare attempts: number
+  @column.dateTime()
+  declare doneAt: DateTime | null
+  @column()
+  declare endId: bigint | number
+  @column()
+  declare error: string | null
+  @column()
+  declare groupId: number
+  @column()
+  declare headers: bigint | number
+  @column.dateTime()
+  declare heartbeatAt: DateTime | null
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column.dateTime()
+  declare leasedAt: DateTime | null
+  @column()
+  declare nextId: bigint | number
+  @column()
+  declare nodeId: number | null
+  @column()
+  declare priority: number
+  @column()
+  declare ruleVersion: number | null
+  @column.dateTime()
+  declare spooledAt: DateTime | null
+  @column()
+  declare spooledChunks: number
+  @column()
+  declare startId: bigint | number
+  @column()
+  declare status: string
+}
+
+export class ReleaseChunkSchema extends BaseModel {
+  static $columns = ['batchSeq', 'createdAt', 'files', 'id', 'payload', 'rangeId', 'releaseId', 'segments'] as const
+  $columns = ReleaseChunkSchema.$columns
+  @column()
+  declare batchSeq: number
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime
+  @column()
+  declare files: number
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column()
+  declare payload: Buffer
+  @column()
+  declare rangeId: bigint | number
+  @column()
+  declare releaseId: bigint | number
+  @column()
+  declare segments: number
+}
+
+export class ReleaseSchema extends BaseModel {
+  static $columns = ['bytes', 'chunks', 'complete', 'fileTotal', 'files', 'firstArticle', 'firstSeenAt', 'groups', 'hasPar2', 'id', 'lastArticle', 'lastSeenAt', 'name', 'postedAt', 'poster', 'readable', 'releaseKey', 'segments', 'settledAt', 'source'] as const
+  $columns = ReleaseSchema.$columns
+  @column()
+  declare bytes: bigint | number
+  @column()
+  declare chunks: number
+  @column()
+  declare complete: boolean | null
+  @column()
+  declare fileTotal: number | null
+  @column()
+  declare files: number
+  @column()
+  declare firstArticle: bigint | number | null
+  @column.dateTime()
+  declare firstSeenAt: DateTime
+  @column()
+  declare groups: any
+  @column()
+  declare hasPar2: boolean
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column()
+  declare lastArticle: bigint | number | null
+  @column.dateTime()
+  declare lastSeenAt: DateTime
+  @column()
+  declare name: string
+  @column.dateTime()
+  declare postedAt: DateTime | null
+  @column()
+  declare poster: string | null
+  @column()
+  declare readable: boolean
+  @column()
+  declare releaseKey: string
+  @column()
+  declare segments: bigint | number
+  @column.dateTime()
+  declare settledAt: DateTime | null
+  @column()
+  declare source: string
+}
+
+export class RuleCountSchema extends BaseModel {
+  static $columns = ['articles', 'files', 'key', 'rangeId', 'rule'] as const
+  $columns = RuleCountSchema.$columns
+  @column()
+  declare articles: bigint | number
+  @column()
+  declare files: bigint | number
+  @column()
+  declare key: string
+  @column({ isPrimary: true })
+  declare rangeId: bigint | number
+  @column()
+  declare rule: string
+}
+
+export class RuleSampleSchema extends BaseModel {
+  static $columns = ['files', 'id', 'name', 'payload', 'rangeId', 'releaseKey', 'rule', 'ruleKey', 'segments'] as const
+  $columns = RuleSampleSchema.$columns
+  @column()
+  declare files: number
+  @column({ isPrimary: true })
+  declare id: bigint | number
+  @column()
+  declare name: string
+  @column()
+  declare payload: Buffer
+  @column()
+  declare rangeId: bigint | number
+  @column()
+  declare releaseKey: string
+  @column()
+  declare rule: string
+  @column()
+  declare ruleKey: string
+  @column()
+  declare segments: number
+}
+
+export class RuleSetSchema extends BaseModel {
+  static $columns = ['createdAt', 'note', 'rules', 'version'] as const
+  $columns = RuleSetSchema.$columns
+  @column.dateTime({ autoCreate: true })
+  declare createdAt: DateTime
+  @column()
+  declare note: string | null
+  @column()
+  declare rules: any
+  @column({ isPrimary: true })
+  declare version: number
+}
+
 export class UserSchema extends BaseModel {
   static $columns = ['createdAt', 'email', 'fullName', 'id', 'password', 'updatedAt'] as const
   $columns = UserSchema.$columns

lib/edge/range_indexer.ts

@@ -20,6 +20,9 @@ export interface ReleaseChunk {
   bytes: number
   firstArticle: number
   lastArticle: number
+  /** A file with a readable name; a par2 file. */
+  readable: boolean
+  hasPar2: boolean
   payload: Buffer
 }
 
@@ -63,6 +66,8 @@ export interface Fragment {
 }
 
 export interface EdgeBatch {
+  /** The range's resume point this batch starts at. */
+  from: number
   /** First article number not covered yet. */
   next: number
   headers: number
@@ -193,8 +198,8 @@ export class RangeIndexer {
     }
   }
 
-  private empty(next: number): EdgeBatch {
-    return { next, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] }
+  private empty(from: number): EdgeBatch {
+    return { from, next: from, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] }
   }
 }
 
@@ -247,6 +252,8 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
     bytes,
     firstArticle,
     lastArticle,
+    readable: chunkFiles.some((f) => !isRandomFile(f.name)),
+    hasPar2: chunkFiles.some((f) => /\.par2$/i.test(f.name)),
     payload: encodeChunk(chunkFiles),
   }
 }

lib/rules/rule_set.ts

@@ -53,17 +53,18 @@ export class RuleSet {
 
   /**
    * A release's verdict from its files: kept as a whole when any file has a
-   * readable name; else dropped as per-article random subjects when every
-   * file has few parts, as garbage when its signature or tool is dropped.
+   * readable name; else dropped as garbage when its signature or tool is
+   * dropped, as per-article random subjects when every file has few parts.
    */
   judge(releaseKey: string, files: { name: string; totalParts: number; groups: string[]; firstMessageId: string }[]): Verdict {
     if (files.some((f) => !isRandomFile(f.name))) return { rule: null }
     const first = files[0]
     const tool = uploadTool(first.firstMessageId)
     const sample = explored(releaseKey, this.rules.samplePercent)
-    if (files.every((f) => f.totalParts <= this.rules.perArticleMaxParts)) return { rule: 'per-article', key: tool, sample }
+    // Garbage first: a dropped tool or signature isn't worth reading bodies for either.
     const group = first.groups[0] ?? ''
     if (this.tools.has(tool) || this.signatures.has(signatureKey(tool, group))) return { rule: 'garbage', key: `${tool} ${group}`, sample }
+    if (files.every((f) => f.totalParts <= this.rules.perArticleMaxParts)) return { rule: 'per-article', key: tool, sample }
     return { rule: null }
   }
 }

start/env.ts

@@ -32,7 +32,8 @@ export default await Env.create(new URL('../', import.meta.url), {
   DB_PASSWORD: Env.schema.string.optional(),
   DB_DATABASE: Env.schema.string(),
   DB_POOL_MAX: Env.schema.number.optional(),
-  DB_SEARCH_PATH: Env.schema.string.optional(),
+  // Tablespace for the chunk store (the HDD on homelab).
+  CHUNK_TABLESPACE: Env.schema.string.optional(),
   DB_DEBUG: Env.schema.boolean.optional(),
 
   // NNTP (the coordinator's relay; edge:bench --capture)

tests/functional/batch_store.spec.ts

@@ -0,0 +1,76 @@
+import { test } from '@japa/runner'
+import db from '@adonisjs/lucid/services/db'
+import testUtils from '@adonisjs/core/services/test_utils'
+import BatchStore, { StaleBatch } from '#services/batch_store'
+import { RangeIndexer } from '#lib/edge/index'
+import { decodeChunk } from '#lib/chunks/index'
+import { DEFAULT_RULES } from '#lib/rules/index'
+import type { Overview } from '#lib/nntp/index'
+
+const GROUP = 'alt.binaries.test'
+let id = 0
+
+function post(at: number, name: string, parts: number, counter: string, tool: 'ngPost' | 'plain' = 'plain'): Overview[] {
+  return Array.from({ length: parts }, (_, i) => {
+    id++
+    const mid = tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test`
+    return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p <p@x>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<${mid}>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}` }] as Overview
+  })
+}
+
+async function leasedRange() {
+  const [group] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
+  const [node] = await db.table('nodes').insert({ name: 'test-node' }).returning('id')
+  const [range] = await db
+    .table('ranges')
+    .insert({ group_id: group.id, start_id: 1, end_id: 1_000_000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 })
+    .returning('id')
+  return { rangeId: Number(range.id), nodeId: Number(node.id) }
+}
+
+function indexer() {
+  return new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } }, edgeArticles: 1000 })
+}
+
+test.group('BatchStore', (group) => {
+  group.each.setup(() => testUtils.db().withGlobalTransaction())
+
+  test('stores releases, their chunks, probe candidates, rule counts and fragments, and moves the resume point', async ({ assert }) => {
+    const { rangeId, nodeId } = await leasedRange()
+    const ix = indexer()
+    ix.add([
+      ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'),
+      ...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'),
+      ...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'),
+      ...post(10, 'Edge.Cut.2020.part1.rar', 3, 'Edge.Cut [1/1]').slice(0, 1),
+    ], 500_000)
+    const first = ix.take()
+    const stored = await BatchStore.store(rangeId, nodeId, first)
+    assert.include(stored, { releases: 1, newReleases: 1, candidates: 2, fragments: 0 })
+
+    const release = await db.from('releases').first()
+    assert.include(release, { files: 2, segments: '3', chunks: 1, readable: true, has_par2: true, file_total: 2 })
+    const chunk = await db.from('release_chunks').where('release_id', release.id).first()
+    assert.deepEqual(decodeChunk(chunk.payload).map((f) => f.name).sort(), ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar'])
+    assert.deepEqual((await db.from('probe_candidates').orderBy('kind')).map((c) => c.kind), ['archive', 'par2'])
+    assert.deepEqual((await db.from('rule_counts')).map((c) => [c.rule, c.key, Number(c.articles)]), [['garbage', 'ngPost alt.binaries.test', 3]])
+    assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 500_000)
+
+    // The same batch again (a retry after a lost reply) is refused, and changes nothing.
+    await assert.rejects(() => BatchStore.store(rangeId, nodeId, first), StaleBatch)
+    assert.equal((await db.from('releases').first()).files, 2)
+
+    // The end of the range: the cut-off file near the start edge is kept as a fragment.
+    ix.finish()
+    const last = ix.take()
+    assert.include(await BatchStore.store(rangeId, nodeId, last), { fragments: 1 })
+    assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 1_000_001)
+  })
+
+  test('a range leased to another node is refused', async ({ assert }) => {
+    const { rangeId, nodeId } = await leasedRange()
+    const ix = indexer()
+    ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000)
+    await assert.rejects(() => BatchStore.store(rangeId, nodeId + 1, ix.take()), /not leased to this node/)
+  })
+})