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)