Daan MeijerandClaude Opus 5.5
68eb9a44fd
A seeding node fetches a range over one NNTP connection, parses headers with the live rules, groups them into files in memory (lib/aggregate) and stores per complete file its release, the file and a completed binary (SeedService), instead of one parts row per article. What doesn't complete within the window goes through the regular ingest path as fragments. `--dry-run` only fetches and aggregates, to measure a node. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
2.8 KiB
TypeScript
87 lines
2.8 KiB
TypeScript
/**
|
|
* Edge aggregation (CONCEPT.md D37): groups header segments into files in
|
|
* memory, so a seeding node writes one row per file instead of one per
|
|
* article. A file is done once every part number from 1 to its total has
|
|
* an article; what never completes within the node's window (an upload cut
|
|
* by the range edges, a random subject per article) leaves as fragments
|
|
* for the regular ingest path, where it can meet its other parts.
|
|
*/
|
|
|
|
/** What the aggregator needs of a segment (IngestSegment has all of it). */
|
|
export interface AggregatableSegment {
|
|
fileKey: string
|
|
number: number
|
|
totalParts: number
|
|
messageId: string
|
|
bytes: number
|
|
}
|
|
|
|
export interface AggregatedFile<S extends AggregatableSegment> {
|
|
fileKey: string
|
|
/** One segment per part number, in part order; the first article seen wins (a later one is a repost). */
|
|
segments: S[]
|
|
}
|
|
|
|
interface OpenFile<S> {
|
|
parts: Map<number, S>
|
|
totalParts: number
|
|
/** Batch in which a segment last arrived. */
|
|
touched: number
|
|
}
|
|
|
|
export class FileAggregator<S extends AggregatableSegment> {
|
|
private open = new Map<string, OpenFile<S>>()
|
|
private batch = 0
|
|
/** Segments with a part number outside 1..total: never part of a complete file. */
|
|
private stray: S[] = []
|
|
|
|
/** Adds one batch of segments; returns the files they completed. */
|
|
add(segments: S[]): AggregatedFile<S>[] {
|
|
this.batch++
|
|
const touched = new Set<string>()
|
|
for (const segment of segments) {
|
|
if (segment.number < 1 || segment.number > segment.totalParts) {
|
|
this.stray.push(segment)
|
|
continue
|
|
}
|
|
let file = this.open.get(segment.fileKey)
|
|
if (!file) {
|
|
file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch }
|
|
this.open.set(segment.fileKey, file)
|
|
}
|
|
if (!file.parts.has(segment.number)) file.parts.set(segment.number, segment)
|
|
file.touched = this.batch
|
|
touched.add(segment.fileKey)
|
|
}
|
|
|
|
const complete: AggregatedFile<S>[] = []
|
|
for (const key of touched) {
|
|
const file = this.open.get(key)!
|
|
if (file.parts.size < file.totalParts) continue
|
|
this.open.delete(key)
|
|
complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) })
|
|
}
|
|
return complete
|
|
}
|
|
|
|
/**
|
|
* Gives up on files that got no segment in the last `idleBatches` batches
|
|
* (all open files when 0) and returns their segments, plus stray ones.
|
|
*/
|
|
evict(idleBatches = 0): S[] {
|
|
const out = this.stray
|
|
this.stray = []
|
|
for (const [key, file] of this.open) {
|
|
if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue
|
|
this.open.delete(key)
|
|
out.push(...file.parts.values())
|
|
}
|
|
return out
|
|
}
|
|
|
|
/** Files still waiting for parts. */
|
|
get openFiles(): number {
|
|
return this.open.size
|
|
}
|
|
}
|