Rebuild: a lean overview scan for the edge worker
The worker reads the XOVER reply's text directly: fields by position, the owner from the Xref without arrays (crossposts leave before any subject parsing), a subject parsed once per file (cached without its segment counter), a file's details as one shared object, dates as numbers. Garbage collection had been ~22% of the time. Same results as parseOverviews + segmentFromHeader on 800k real lines. edge:bench (one core): boneless 137k → 193k headers/s (CPU 21 → 12.8 s), ath (2013) 143k → 303k; peak RSS 1.15 / 0.87 GB. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
981bc01810
commit
638cfa1dcd
5 files changed
+292
-59
No files matched your search
+66
-52
@@ -1,9 +1,8 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import type { Overview } from '../nntp/index.js'
|
||||
import { parseXrefGroups } from '../nntp/index.js'
|
||||
import { segmentFromHeader, type IdentityCache, type IngestSegment } from '../headers/index.js'
|
||||
import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/index.js'
|
||||
import { RuleSet, isRandomFile, ownerOf, type Rules } from '../rules/index.js'
|
||||
import { RuleSet, isRandomFile, type Rules } from '../rules/index.js'
|
||||
import { SegmentScanner, scanOverviewText, type EdgeSegment } from './scan.js'
|
||||
import { encodeChunk, type ChunkFile } from '../chunks/index.js'
|
||||
|
||||
/** A release's files from one batch (REBUILD.md, chunk store): what the catalog adds up, and the encoded chunk. */
|
||||
@@ -100,12 +99,12 @@ const MAX_CANDIDATES = 4
|
||||
* waiting for parts.
|
||||
*/
|
||||
export class RangeIndexer {
|
||||
private aggregator = new FileAggregator<IngestSegment>()
|
||||
private aggregator = new FileAggregator<EdgeSegment>()
|
||||
private scanner = new SegmentScanner()
|
||||
private segments: EdgeSegment[] = []
|
||||
private rules: RuleSet
|
||||
private batch: EdgeBatch
|
||||
private counts = new Map<string, RuleCount>()
|
||||
/** File identities seen in this batch (cleared with each take, so it stays small). */
|
||||
private identities: IdentityCache = new Map()
|
||||
private readonly idle: number
|
||||
private readonly edge: number
|
||||
|
||||
@@ -116,24 +115,38 @@ export class RangeIndexer {
|
||||
this.batch = this.empty(options.start)
|
||||
}
|
||||
|
||||
/** One chunk of overview lines; `next` is the first article after it. */
|
||||
/** One chunk of overview lines (parsed); `next` is the first article after it. */
|
||||
add(overviews: Overview[], next: number) {
|
||||
for (const [n, h] of overviews) this.line(n, h.subject ?? '', h.from ?? '', h.date ?? '', h['message-id'] ?? '', h[':bytes'] ?? '', h.xref ?? '')
|
||||
this.endChunk(next)
|
||||
}
|
||||
|
||||
/** One chunk as the XOVER reply's text (the worker's path: no object per line). */
|
||||
addText(text: string, next: number) {
|
||||
scanOverviewText(text, this.handleLine)
|
||||
this.endChunk(next)
|
||||
}
|
||||
|
||||
private handleLine = (n: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string) =>
|
||||
this.line(n, subject, from, date, messageId, bytes, xref)
|
||||
|
||||
private line(articleNumber: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string) {
|
||||
if (!Number.isFinite(articleNumber) || !messageId) return
|
||||
const { group, rules } = this.options
|
||||
const segments: IngestSegment[] = []
|
||||
for (const [articleNumber, header] of overviews) {
|
||||
if (!Number.isFinite(articleNumber) || !header['message-id']) continue
|
||||
this.batch.headers++
|
||||
const groups = parseXrefGroups(header.xref)
|
||||
const owner = ownerOf(groups.length > 0 ? groups : [group], rules.priorities) ?? group
|
||||
if (owner !== group) {
|
||||
this.count('crosspost', owner, 1, 0)
|
||||
continue
|
||||
}
|
||||
const segment = segmentFromHeader(header, group, articleNumber, this.identities)
|
||||
if (segment) segments.push(segment)
|
||||
else this.count('obfuscated', '', 1, 0)
|
||||
this.batch.headers++
|
||||
const owner = xref ? this.scanner.owner(xref, group, rules.priorities) : group
|
||||
if (owner !== group) {
|
||||
this.count('crosspost', owner, 1, 0)
|
||||
return
|
||||
}
|
||||
this.files(this.aggregator.add(segments))
|
||||
const segment = this.scanner.segment(articleNumber, subject, from, date, messageId, bytes, xref, group)
|
||||
if (segment) this.segments.push(segment)
|
||||
else this.count('obfuscated', '', 1, 0)
|
||||
}
|
||||
|
||||
private endChunk(next: number) {
|
||||
this.files(this.aggregator.add(this.segments))
|
||||
this.segments = []
|
||||
this.cutOff(this.aggregator.evict(this.idle))
|
||||
this.batch.next = next
|
||||
}
|
||||
@@ -147,7 +160,7 @@ export class RangeIndexer {
|
||||
/** What was found since the last take, and a fresh batch. */
|
||||
take(): EdgeBatch {
|
||||
this.judgeReleases()
|
||||
this.identities.clear()
|
||||
this.scanner.clear()
|
||||
const out = this.batch
|
||||
out.counts = [...this.counts.values()]
|
||||
this.counts.clear()
|
||||
@@ -160,9 +173,9 @@ export class RangeIndexer {
|
||||
}
|
||||
|
||||
/** Completed files, gathered per release until the batch is taken: one chunk per release per batch. */
|
||||
private completed = new Map<string, AggregatedFile<IngestSegment>[]>()
|
||||
private completed = new Map<string, AggregatedFile<EdgeSegment>[]>()
|
||||
|
||||
private files(files: AggregatedFile<IngestSegment>[]) {
|
||||
private files(files: AggregatedFile<EdgeSegment>[]) {
|
||||
for (const file of files) {
|
||||
const key = releaseKeyOf(file.segments[0])
|
||||
const list = this.completed.get(key)
|
||||
@@ -178,7 +191,7 @@ export class RangeIndexer {
|
||||
for (const [key, list] of byRelease) {
|
||||
const verdict = this.rules.judge(
|
||||
key,
|
||||
list.map((f) => ({ name: f.segments[0].name, totalParts: f.segments[0].totalParts, groups: f.segments[0].groups, firstMessageId: f.segments[0].messageId }))
|
||||
list.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId }))
|
||||
)
|
||||
if (verdict.rule === null) {
|
||||
this.batch.releases.push(chunkOf(key, list))
|
||||
@@ -191,7 +204,7 @@ export class RangeIndexer {
|
||||
}
|
||||
|
||||
/** Given-up segments: kept per file near the range's edges (the neighbour may complete them), counted otherwise. */
|
||||
private cutOff(segments: IngestSegment[]) {
|
||||
private cutOff(segments: EdgeSegment[]) {
|
||||
const { start, end } = this.options
|
||||
const byFile = new Map<string, Fragment>()
|
||||
for (const s of segments) {
|
||||
@@ -207,8 +220,8 @@ export class RangeIndexer {
|
||||
this.batch.fragments.push(fragment)
|
||||
}
|
||||
fragment.segments.push([s.number, s.messageId, s.bytes, n])
|
||||
const t = s.postedAt?.getTime()
|
||||
if (t !== undefined && Number.isFinite(t) && (fragment.postedAt === null || t < fragment.postedAt)) fragment.postedAt = t
|
||||
const t = s.postedAt
|
||||
if (Number.isFinite(t) && (fragment.postedAt === null || t < fragment.postedAt)) fragment.postedAt = t
|
||||
}
|
||||
}
|
||||
|
||||
@@ -229,24 +242,24 @@ export class RangeIndexer {
|
||||
}
|
||||
|
||||
/** The release key; a file without a recognisable release becomes a release of its own. */
|
||||
function releaseKeyOf(s: IngestSegment): string {
|
||||
return s.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex')
|
||||
function releaseKeyOf(s: EdgeSegment): string {
|
||||
return s.file.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex')
|
||||
}
|
||||
|
||||
function postedOf(segments: IngestSegment[]): number | null {
|
||||
function postedOf(segments: EdgeSegment[]): number | null {
|
||||
let min: number | null = null
|
||||
for (const s of segments) {
|
||||
const t = s.postedAt?.getTime()
|
||||
if (t !== undefined && Number.isFinite(t) && (min === null || t < min)) min = t
|
||||
const t = s.postedAt
|
||||
if (Number.isFinite(t) && (min === null || t < min)) min = t
|
||||
}
|
||||
return min
|
||||
}
|
||||
|
||||
function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseChunk {
|
||||
const first = files[0].segments[0]
|
||||
function chunkOf(key: string, files: AggregatedFile<EdgeSegment>[]): ReleaseChunk {
|
||||
const first = files[0].segments[0].file
|
||||
const chunkFiles: ChunkFile[] = files.map((f) => {
|
||||
const s = f.segments[0]
|
||||
return { name: s.name, poster: s.poster, subject: s.subject, postedAt: postedOf(f.segments), groups: s.groups, segments: f.segments.map((x) => [x.number, x.messageId, x.bytes]) }
|
||||
const m = f.segments[0].file
|
||||
return { name: m.name, poster: m.poster, subject: m.subject, postedAt: postedOf(f.segments), groups: m.groups, segments: f.segments.map((x) => [x.number, x.messageId, x.bytes]) }
|
||||
})
|
||||
let segments = 0
|
||||
let bytes = 0
|
||||
@@ -257,7 +270,7 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
|
||||
for (const s of f.segments) {
|
||||
segments++
|
||||
bytes += s.bytes
|
||||
const n = Number(s.articleNumber)
|
||||
const n = s.articleNumber
|
||||
if (n < firstArticle) firstArticle = n
|
||||
if (n > lastArticle) lastArticle = n
|
||||
}
|
||||
@@ -268,7 +281,7 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
|
||||
key,
|
||||
name: first.release?.name ?? first.name,
|
||||
source: first.release?.source ?? 'file',
|
||||
fileTotal: files.find((f) => f.segments[0].release?.fileTotal)?.segments[0].release?.fileTotal ?? null,
|
||||
fileTotal: files.find((f) => f.segments[0].file.release?.fileTotal)?.segments[0].file.release?.fileTotal ?? null,
|
||||
poster: first.poster,
|
||||
groups: first.groups,
|
||||
postedAt,
|
||||
@@ -288,11 +301,11 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
|
||||
* volume, of its par2 index, of a lone random-named file, and a 7z set's
|
||||
* last volume (its listing is in the tail). At most a few per release.
|
||||
*/
|
||||
function candidatesOf(key: string, files: AggregatedFile<IngestSegment>[]): ProbeCandidate[] {
|
||||
function candidatesOf(key: string, files: AggregatedFile<EdgeSegment>[]): ProbeCandidate[] {
|
||||
const out: ProbeCandidate[] = []
|
||||
let tail: { volume: number; file: AggregatedFile<IngestSegment> } | null = null
|
||||
let tail: { volume: number; file: AggregatedFile<EdgeSegment> } | null = null
|
||||
for (const f of files) {
|
||||
const name = f.segments[0].name
|
||||
const name = f.segments[0].file.name
|
||||
const volume = sevenZipVolume(name)
|
||||
if (volume !== null && volume > 1) {
|
||||
if (!tail || volume > tail.volume) tail = { volume, file: f }
|
||||
@@ -301,22 +314,23 @@ function candidatesOf(key: string, files: AggregatedFile<IngestSegment>[]): Prob
|
||||
if (isArchiveStart(name)) out.push({ releaseKey: key, kind: 'archive', messageId: f.segments[0].messageId, file: name })
|
||||
else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, kind: 'par2', messageId: f.segments[0].messageId, file: name })
|
||||
}
|
||||
if (tail) out.push({ releaseKey: key, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].name })
|
||||
if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].name)) {
|
||||
out.push({ releaseKey: key, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].name })
|
||||
if (tail) out.push({ releaseKey: key, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name })
|
||||
if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].file.name)) {
|
||||
out.push({ releaseKey: key, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name })
|
||||
}
|
||||
return out.slice(0, MAX_CANDIDATES)
|
||||
}
|
||||
|
||||
function fragmentOf(s: IngestSegment): Fragment {
|
||||
function fragmentOf(s: EdgeSegment): Fragment {
|
||||
const m = s.file
|
||||
return {
|
||||
fileKey: s.fileKey,
|
||||
name: s.name,
|
||||
poster: s.poster,
|
||||
subject: s.subject,
|
||||
totalParts: s.totalParts,
|
||||
groups: s.groups,
|
||||
release: s.release ? { key: s.release.key, name: s.release.name, source: s.release.source, ...(s.release.fileTotal ? { fileTotal: s.release.fileTotal } : {}) } : null,
|
||||
name: m.name,
|
||||
poster: m.poster,
|
||||
subject: m.subject,
|
||||
totalParts: m.totalParts,
|
||||
groups: m.groups,
|
||||
release: m.release ? { key: m.release.key, name: m.release.name, source: m.release.source, ...(m.release.fileTotal ? { fileTotal: m.release.fileTotal } : {}) } : null,
|
||||
postedAt: null,
|
||||
segments: [],
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user