From 8d86f7d489aa14570f1b0e854d3bc64c0ed4d121 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 20:30:24 +0200 Subject: [PATCH] Seed fragments: keep only those near the range edges, unstaged About 5% of seeded headers are fragments. They went through staging, where they waited 30 minutes for parts that were never coming (the neighbouring range is worked on by another node, often hours later) and expired: the staging table grew to 22 GB and the NVMe part of the database reached its 60 GB limit, which stopped leasing and the backfill. Files cut by a range edge can still be completed by the neighbouring range, so fragments within 50k articles of an edge go straight into binaries and parts; the rest, files with articles missing for good, are dropped. Co-Authored-By: Claude Opus 5.5 --- app/services/SeedOrchestrator.ts | 2 +- app/services/SeedService.ts | 29 ++++++++++++++++++++--------- commands/SeedRange.ts | 2 +- tests/functional/seed.spec.ts | 12 +++++++++++- tests/functional/seed_api.spec.ts | 5 ++--- 5 files changed, 35 insertions(+), 15 deletions(-) diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index b6dc703..beafbf8 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -89,7 +89,7 @@ class SeedOrchestrator { async batch(nodeId: number, rangeId: number, batch: SeedBatch) { const range = await this.leased(nodeId, rangeId) const stored = await SeedService.store(batch.files.map(normalizeSeedFile)) - await SeedService.storeFragments(range.group_name, batch.fragments) + await SeedService.storeFragments(batch.fragments, { start: Number(range.start_id), end: Number(range.end_id) }) await db.rawQuery( `UPDATE seed_ranges SET next_id = greatest(next_id, least(?, end_id + 1)), heartbeat_at = now(), headers = headers + ?, files = files + ?, fragments = fragments + ? diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index e82350a..0a1668a 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -4,9 +4,11 @@ import FileStore from '#services/FileStore' import SfvService from '#services/SfvService' import ContentService from '#services/ContentService' import QueueService from '#services/QueueService' -import StagingService from '#services/StagingService' import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire' +/** Articles from a range's edge within which a file may continue in the neighbouring range. */ +const EDGE_ARTICLES = 50_000 + export interface SeedStoreResult { files: number /** Files that existed already (same file_key, e.g. from live indexing or a crosspost). */ @@ -96,14 +98,23 @@ class SeedService { return result } - /** Segments of files that didn't complete on a node: staging (D35) for lone ones, binaries and parts for the rest. */ - async storeFragments(group: string, fragments: SeedFragment[]): Promise { - if (fragments.length === 0) return - const segments = fragments.map(fromFragment) - const { ingest } = StagingService.enabled(group) - ? await StagingService.filter(segments, { counters: StagingService.countersEnabled(group) }) - : { ingest: segments } - await IngestService.ingest(ingest) + /** + * Segments of files that didn't complete on a node. Only files near the + * range's edges can continue in the neighbouring range (worked on by some + * node, maybe hours later): those go straight into binaries and parts + * (staging would drop them after 30 minutes). The rest are files with + * articles missing for good, and are dropped: staging them filled 22 GB + * of churn on the NVMe (LOG.md 2026-10-05 (10)). Returns the segments kept. + */ + async storeFragments(fragments: SeedFragment[], range: { start: number; end: number }): Promise { + const nearEdge = new Set() + for (const fragment of fragments) { + const n = Number(fragment.articleNumber) + if (Number.isFinite(n) && (n < range.start + EDGE_ARTICLES || n > range.end - EDGE_ARTICLES)) nearEdge.add(fragment.fileKey) + } + const kept = fragments.filter((fragment) => nearEdge.has(fragment.fileKey)) + if (kept.length > 0) await IngestService.ingest(kept.map(fromFragment)) + return kept.length } } diff --git a/commands/SeedRange.ts b/commands/SeedRange.ts index ade9884..b38777c 100644 --- a/commands/SeedRange.ts +++ b/commands/SeedRange.ts @@ -86,7 +86,7 @@ export default class SeedRange extends BaseCommand { if (!this.dryRun) { const t = Date.now() const stored = await SeedService.store(batch.files) - await SeedService.storeFragments(this.group, batch.fragments) + await SeedService.storeFragments(batch.fragments, { start: first, end: last }) storeMs += Date.now() - t stats.files += stored.files stats.existing += stored.existing diff --git a/tests/functional/seed.spec.ts b/tests/functional/seed.spec.ts index 2661938..25d7105 100644 --- a/tests/functional/seed.spec.ts +++ b/tests/functional/seed.spec.ts @@ -5,7 +5,7 @@ import { segmentFromHeader, type IngestSegment } from '#services/IngestService' import SeedService from '#services/SeedService' import QueueService from '#services/QueueService' import { FileAggregator } from '#lib/aggregate/index' -import { toSeedFile } from '#lib/seed/wire' +import { toFragment, toSeedFile } from '#lib/seed/wire' import { segmentsOf } from '#services/FileStore' function header(subject: string, n: number) { @@ -45,4 +45,14 @@ test.group('SeedService', (group) => { const again = await SeedService.store(files) assert.include(again, { files: 0, existing: 2 }) }) + + test('fragments are kept near the range edges only', async ({ assert }) => { + const subject = (name: string) => `Seed.Frag - "${name}" yEnc (1/3)` + const edge = segmentFromHeader(header(subject('edge.bin'), 1), 'alt.binaries.test', 1_000_100)! + const middle = segmentFromHeader(header(subject('middle.bin'), 2), 'alt.binaries.test', 1_500_000)! + const kept = await SeedService.storeFragments([toFragment(edge), toFragment(middle)], { start: 1_000_000, end: 2_999_999 }) + assert.equal(kept, 1) + assert.isOk(await db.from('binaries').where('name', 'edge.bin').first()) + assert.isNotOk(await db.from('binaries').where('name', 'middle.bin').first()) + }) }) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index 3769b91..8161d48 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -48,9 +48,8 @@ test.group('Seed API', (group) => { assert.include(((await posted.json()) as any).stored, { files: 1 }) const file = await db.from('files').where('filename', 'Api.Test.2026.mkv').first() assert.equal(file.parts, 2) - // The lone .nfo segment took the regular path (staged or stored as a binary). - const fragment = (await db.from('staged_segments').count('* as n').first()).n + (await db.from('binaries').where('name', 'Api.Test.2026.nfo').count('* as n').first()).n - assert.isAbove(Number(fragment), 0) + // The lone .nfo segment sits near the range's edge: kept as a binary (not staged) for the neighbouring range. + assert.isOk(await db.from('binaries').where('name', 'Api.Test.2026.nfo').first()) const row = await db.from('seed_ranges').where('id', range.id).first() assert.equal(Number(row.next_id), 51)