Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
638cfa1dcd | ||
|
|
981bc01810 |
No files matched your search
@@ -4,7 +4,7 @@ import { mkdir, readFile, readdir, writeFile } from 'node:fs/promises'
|
|||||||
import { join } from 'node:path'
|
import { join } from 'node:path'
|
||||||
import { unzipSync } from 'node:zlib'
|
import { unzipSync } from 'node:zlib'
|
||||||
import env from '#start/env'
|
import env from '#start/env'
|
||||||
import { XoverClient, parseOverviews } from '#lib/nntp/index'
|
import { XoverClient } from '#lib/nntp/index'
|
||||||
import { RangeIndexer } from '#lib/edge/index'
|
import { RangeIndexer } from '#lib/edge/index'
|
||||||
import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
|
import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
|
||||||
import db from '@adonisjs/lucid/services/db'
|
import db from '@adonisjs/lucid/services/db'
|
||||||
@@ -117,11 +117,9 @@ export default class EdgeBench extends BaseCommand {
|
|||||||
t.inflate += performance.now() - at
|
t.inflate += performance.now() - at
|
||||||
text += data.length
|
text += data.length
|
||||||
at = performance.now()
|
at = performance.now()
|
||||||
const overviews = parseOverviews(data)
|
|
||||||
t.parse += performance.now() - at
|
|
||||||
at = performance.now()
|
|
||||||
const next = Number(file.split('-')[2].replace('.gz', '')) + 1
|
const next = Number(file.split('-')[2].replace('.gz', '')) + 1
|
||||||
indexer.add(overviews, next)
|
// The worker's path: the reply's text straight into the indexer (parsing is part of "index").
|
||||||
|
indexer.addText(data, next)
|
||||||
const last = file === files.at(-1)
|
const last = file === files.at(-1)
|
||||||
if (last) indexer.finish()
|
if (last) indexer.finish()
|
||||||
t.index += performance.now() - at
|
t.index += performance.now() - at
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ import { BaseCommand, flags } from '@adonisjs/core/ace'
|
|||||||
import type { CommandOptions } from '@adonisjs/core/types/ace'
|
import type { CommandOptions } from '@adonisjs/core/types/ace'
|
||||||
import { setTimeout as sleep } from 'node:timers/promises'
|
import { setTimeout as sleep } from 'node:timers/promises'
|
||||||
import { gzipSync, unzipSync } from 'node:zlib'
|
import { gzipSync, unzipSync } from 'node:zlib'
|
||||||
import { parseOverviews } from '#lib/nntp/index'
|
|
||||||
import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
|
import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
|
||||||
|
|
||||||
const IDLE_MS = 30_000
|
const IDLE_MS = 30_000
|
||||||
@@ -81,7 +80,7 @@ export default class EdgeNode extends BaseCommand {
|
|||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
for await (const frame of decodeFrames(await api.stream(range.id))) {
|
for await (const frame of decodeFrames(await api.stream(range.id))) {
|
||||||
indexer.add(parseOverviews(unzipSync(frame.data).toString('utf8')), frame.to + 1)
|
indexer.addText(unzipSync(frame.data).toString('utf8'), frame.to + 1)
|
||||||
chunks++
|
chunks++
|
||||||
const last = frame.to >= range.end
|
const last = frame.to >= range.end
|
||||||
if (last) indexer.finish()
|
if (last) indexer.finish()
|
||||||
|
|||||||
+66
-49
@@ -1,9 +1,8 @@
|
|||||||
import { createHash } from 'node:crypto'
|
import { createHash } from 'node:crypto'
|
||||||
import type { Overview } from '../nntp/index.js'
|
import type { Overview } from '../nntp/index.js'
|
||||||
import { parseXrefGroups } from '../nntp/index.js'
|
|
||||||
import { segmentFromHeader, type IngestSegment } from '../headers/index.js'
|
|
||||||
import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/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'
|
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. */
|
/** A release's files from one batch (REBUILD.md, chunk store): what the catalog adds up, and the encoded chunk. */
|
||||||
@@ -100,7 +99,9 @@ const MAX_CANDIDATES = 4
|
|||||||
* waiting for parts.
|
* waiting for parts.
|
||||||
*/
|
*/
|
||||||
export class RangeIndexer {
|
export class RangeIndexer {
|
||||||
private aggregator = new FileAggregator<IngestSegment>()
|
private aggregator = new FileAggregator<EdgeSegment>()
|
||||||
|
private scanner = new SegmentScanner()
|
||||||
|
private segments: EdgeSegment[] = []
|
||||||
private rules: RuleSet
|
private rules: RuleSet
|
||||||
private batch: EdgeBatch
|
private batch: EdgeBatch
|
||||||
private counts = new Map<string, RuleCount>()
|
private counts = new Map<string, RuleCount>()
|
||||||
@@ -114,24 +115,38 @@ export class RangeIndexer {
|
|||||||
this.batch = this.empty(options.start)
|
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) {
|
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 { group, rules } = this.options
|
||||||
const segments: IngestSegment[] = []
|
this.batch.headers++
|
||||||
for (const [articleNumber, header] of overviews) {
|
const owner = xref ? this.scanner.owner(xref, group, rules.priorities) : group
|
||||||
if (!Number.isFinite(articleNumber) || !header['message-id']) continue
|
if (owner !== group) {
|
||||||
this.batch.headers++
|
this.count('crosspost', owner, 1, 0)
|
||||||
const groups = parseXrefGroups(header.xref)
|
return
|
||||||
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)
|
|
||||||
if (segment) segments.push(segment)
|
|
||||||
else this.count('obfuscated', '', 1, 0)
|
|
||||||
}
|
}
|
||||||
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.cutOff(this.aggregator.evict(this.idle))
|
||||||
this.batch.next = next
|
this.batch.next = next
|
||||||
}
|
}
|
||||||
@@ -145,6 +160,7 @@ export class RangeIndexer {
|
|||||||
/** What was found since the last take, and a fresh batch. */
|
/** What was found since the last take, and a fresh batch. */
|
||||||
take(): EdgeBatch {
|
take(): EdgeBatch {
|
||||||
this.judgeReleases()
|
this.judgeReleases()
|
||||||
|
this.scanner.clear()
|
||||||
const out = this.batch
|
const out = this.batch
|
||||||
out.counts = [...this.counts.values()]
|
out.counts = [...this.counts.values()]
|
||||||
this.counts.clear()
|
this.counts.clear()
|
||||||
@@ -157,9 +173,9 @@ export class RangeIndexer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Completed files, gathered per release until the batch is taken: one chunk per release per batch. */
|
/** 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) {
|
for (const file of files) {
|
||||||
const key = releaseKeyOf(file.segments[0])
|
const key = releaseKeyOf(file.segments[0])
|
||||||
const list = this.completed.get(key)
|
const list = this.completed.get(key)
|
||||||
@@ -175,7 +191,7 @@ export class RangeIndexer {
|
|||||||
for (const [key, list] of byRelease) {
|
for (const [key, list] of byRelease) {
|
||||||
const verdict = this.rules.judge(
|
const verdict = this.rules.judge(
|
||||||
key,
|
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) {
|
if (verdict.rule === null) {
|
||||||
this.batch.releases.push(chunkOf(key, list))
|
this.batch.releases.push(chunkOf(key, list))
|
||||||
@@ -188,7 +204,7 @@ export class RangeIndexer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Given-up segments: kept per file near the range's edges (the neighbour may complete them), counted otherwise. */
|
/** 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 { start, end } = this.options
|
||||||
const byFile = new Map<string, Fragment>()
|
const byFile = new Map<string, Fragment>()
|
||||||
for (const s of segments) {
|
for (const s of segments) {
|
||||||
@@ -204,8 +220,8 @@ export class RangeIndexer {
|
|||||||
this.batch.fragments.push(fragment)
|
this.batch.fragments.push(fragment)
|
||||||
}
|
}
|
||||||
fragment.segments.push([s.number, s.messageId, s.bytes, n])
|
fragment.segments.push([s.number, s.messageId, s.bytes, n])
|
||||||
const t = s.postedAt?.getTime()
|
const t = s.postedAt
|
||||||
if (t !== undefined && Number.isFinite(t) && (fragment.postedAt === null || t < fragment.postedAt)) fragment.postedAt = t
|
if (Number.isFinite(t) && (fragment.postedAt === null || t < fragment.postedAt)) fragment.postedAt = t
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -226,24 +242,24 @@ export class RangeIndexer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** The release key; a file without a recognisable release becomes a release of its own. */
|
/** The release key; a file without a recognisable release becomes a release of its own. */
|
||||||
function releaseKeyOf(s: IngestSegment): string {
|
function releaseKeyOf(s: EdgeSegment): string {
|
||||||
return s.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex')
|
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
|
let min: number | null = null
|
||||||
for (const s of segments) {
|
for (const s of segments) {
|
||||||
const t = s.postedAt?.getTime()
|
const t = s.postedAt
|
||||||
if (t !== undefined && Number.isFinite(t) && (min === null || t < min)) min = t
|
if (Number.isFinite(t) && (min === null || t < min)) min = t
|
||||||
}
|
}
|
||||||
return min
|
return min
|
||||||
}
|
}
|
||||||
|
|
||||||
function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseChunk {
|
function chunkOf(key: string, files: AggregatedFile<EdgeSegment>[]): ReleaseChunk {
|
||||||
const first = files[0].segments[0]
|
const first = files[0].segments[0].file
|
||||||
const chunkFiles: ChunkFile[] = files.map((f) => {
|
const chunkFiles: ChunkFile[] = files.map((f) => {
|
||||||
const s = f.segments[0]
|
const m = f.segments[0].file
|
||||||
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]) }
|
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 segments = 0
|
||||||
let bytes = 0
|
let bytes = 0
|
||||||
@@ -254,7 +270,7 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
|
|||||||
for (const s of f.segments) {
|
for (const s of f.segments) {
|
||||||
segments++
|
segments++
|
||||||
bytes += s.bytes
|
bytes += s.bytes
|
||||||
const n = Number(s.articleNumber)
|
const n = s.articleNumber
|
||||||
if (n < firstArticle) firstArticle = n
|
if (n < firstArticle) firstArticle = n
|
||||||
if (n > lastArticle) lastArticle = n
|
if (n > lastArticle) lastArticle = n
|
||||||
}
|
}
|
||||||
@@ -265,7 +281,7 @@ function chunkOf(key: string, files: AggregatedFile<IngestSegment>[]): ReleaseCh
|
|||||||
key,
|
key,
|
||||||
name: first.release?.name ?? first.name,
|
name: first.release?.name ?? first.name,
|
||||||
source: first.release?.source ?? 'file',
|
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,
|
poster: first.poster,
|
||||||
groups: first.groups,
|
groups: first.groups,
|
||||||
postedAt,
|
postedAt,
|
||||||
@@ -285,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
|
* 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.
|
* 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[] = []
|
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) {
|
for (const f of files) {
|
||||||
const name = f.segments[0].name
|
const name = f.segments[0].file.name
|
||||||
const volume = sevenZipVolume(name)
|
const volume = sevenZipVolume(name)
|
||||||
if (volume !== null && volume > 1) {
|
if (volume !== null && volume > 1) {
|
||||||
if (!tail || volume > tail.volume) tail = { volume, file: f }
|
if (!tail || volume > tail.volume) tail = { volume, file: f }
|
||||||
@@ -298,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 })
|
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 })
|
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 (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].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].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)
|
return out.slice(0, MAX_CANDIDATES)
|
||||||
}
|
}
|
||||||
|
|
||||||
function fragmentOf(s: IngestSegment): Fragment {
|
function fragmentOf(s: EdgeSegment): Fragment {
|
||||||
|
const m = s.file
|
||||||
return {
|
return {
|
||||||
fileKey: s.fileKey,
|
fileKey: s.fileKey,
|
||||||
name: s.name,
|
name: m.name,
|
||||||
poster: s.poster,
|
poster: m.poster,
|
||||||
subject: s.subject,
|
subject: m.subject,
|
||||||
totalParts: s.totalParts,
|
totalParts: m.totalParts,
|
||||||
groups: s.groups,
|
groups: m.groups,
|
||||||
release: s.release ? { key: s.release.key, name: s.release.name, source: s.release.source, ...(s.release.fileTotal ? { fileTotal: s.release.fileTotal } : {}) } : null,
|
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,
|
postedAt: null,
|
||||||
segments: [],
|
segments: [],
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,181 @@
|
|||||||
|
import { fileKey } from '../headers/index.js'
|
||||||
|
import { releaseIdentity, type ReleaseIdentity } from '../release/index.js'
|
||||||
|
|
||||||
|
/** A file's details, shared by all of its segments (one object per file, not per article). */
|
||||||
|
export interface FileMeta {
|
||||||
|
fileKey: string
|
||||||
|
name: string
|
||||||
|
poster: string
|
||||||
|
/** The subject without its segment counter. */
|
||||||
|
subject: string
|
||||||
|
totalParts: number
|
||||||
|
groups: string[]
|
||||||
|
release: ReleaseIdentity | null
|
||||||
|
}
|
||||||
|
|
||||||
|
/** One article of a file, as lean as the edge worker can keep it. */
|
||||||
|
export interface EdgeSegment {
|
||||||
|
fileKey: string
|
||||||
|
number: number
|
||||||
|
totalParts: number
|
||||||
|
messageId: string
|
||||||
|
bytes: number
|
||||||
|
articleNumber: number
|
||||||
|
/** ms since 1970, NaN when the Date header doesn't parse. */
|
||||||
|
postedAt: number
|
||||||
|
file: FileMeta
|
||||||
|
}
|
||||||
|
|
||||||
|
interface ParsedSubjectBase {
|
||||||
|
name: string
|
||||||
|
totalSegments: number
|
||||||
|
base: string
|
||||||
|
}
|
||||||
|
|
||||||
|
const SEGMENT_COUNTER = /\((\d+)\/(\d+)\)/g
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Builds segments from overview fields (REBUILD.md, edge worker), the same
|
||||||
|
* as lib/headers' segmentFromHeader but allocating as little as it can: a
|
||||||
|
* subject is parsed once per file (cached without its segment counter), a
|
||||||
|
* file's details are one shared object, and the Xref is read without
|
||||||
|
* building arrays. Garbage collection was ~20% of the worker's time
|
||||||
|
* (LOG.md 2026-10-09). The caches live until `clear`, once per batch.
|
||||||
|
*/
|
||||||
|
export class SegmentScanner {
|
||||||
|
private subjects = new Map<string, ParsedSubjectBase | null>()
|
||||||
|
private files = new Map<string, FileMeta>()
|
||||||
|
|
||||||
|
clear() {
|
||||||
|
this.subjects.clear()
|
||||||
|
this.files.clear()
|
||||||
|
}
|
||||||
|
|
||||||
|
/** The owner of an article by its Xref field: the indexed group with the lowest priority, else `group`. */
|
||||||
|
owner(xref: string, group: string, priorities: Record<string, number>): string {
|
||||||
|
// "news.example a.b.test:123 a.b.x:456": the first token is the server.
|
||||||
|
let at = xref.indexOf(' ')
|
||||||
|
if (at === -1) return group
|
||||||
|
let owner: string | null = null
|
||||||
|
let best = Number.POSITIVE_INFINITY
|
||||||
|
while (at !== -1) {
|
||||||
|
while (xref.charCodeAt(at) === 32) at++
|
||||||
|
if (at >= xref.length) break
|
||||||
|
const end = xref.indexOf(' ', at)
|
||||||
|
const colon = xref.lastIndexOf(':', end === -1 ? xref.length : end)
|
||||||
|
if (colon > at) {
|
||||||
|
const name = xref.slice(at, colon)
|
||||||
|
const p = priorities[name]
|
||||||
|
if (p !== undefined && p < best) {
|
||||||
|
best = p
|
||||||
|
owner = name
|
||||||
|
}
|
||||||
|
}
|
||||||
|
at = end
|
||||||
|
}
|
||||||
|
return owner ?? group
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A segment from an article's fields, or null when its subject is obfuscated. */
|
||||||
|
segment(articleNumber: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string, group: string): EdgeSegment | null {
|
||||||
|
if (!subject || !messageId) return null
|
||||||
|
const nameEnd = subject.lastIndexOf('"')
|
||||||
|
if (nameEnd <= 0) return null
|
||||||
|
// The segment counter: the last "(k/N)" after the filename (a file size may follow it).
|
||||||
|
SEGMENT_COUNTER.lastIndex = nameEnd
|
||||||
|
let counter: RegExpExecArray | null = null
|
||||||
|
for (let m = SEGMENT_COUNTER.exec(subject); m; m = SEGMENT_COUNTER.exec(subject)) counter = m
|
||||||
|
if (!counter) return null
|
||||||
|
const key = subject.slice(0, counter.index) + '\u0000' + subject.slice(counter.index + counter[0].length)
|
||||||
|
let parsed = this.subjects.get(key)
|
||||||
|
if (parsed === undefined) {
|
||||||
|
parsed = parseBase(subject, nameEnd, counter)
|
||||||
|
this.subjects.set(key, parsed)
|
||||||
|
}
|
||||||
|
if (!parsed) return null
|
||||||
|
const number = Number.parseInt(counter[1], 10)
|
||||||
|
if (number < 1 || number > parsed.totalSegments) return null
|
||||||
|
const id = `${from}\n${parsed.base}\n${parsed.totalSegments}`
|
||||||
|
let file = this.files.get(id)
|
||||||
|
if (!file) {
|
||||||
|
file = {
|
||||||
|
fileKey: fileKey(from, parsed.base, parsed.totalSegments),
|
||||||
|
name: parsed.name,
|
||||||
|
poster: from,
|
||||||
|
subject: parsed.base,
|
||||||
|
totalParts: parsed.totalSegments,
|
||||||
|
groups: xrefGroups(xref, group),
|
||||||
|
release: releaseIdentity({ poster: from, subject: parsed.base, name: parsed.name, weak: false }),
|
||||||
|
}
|
||||||
|
this.files.set(id, file)
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
fileKey: file.fileKey,
|
||||||
|
number,
|
||||||
|
totalParts: file.totalParts,
|
||||||
|
messageId: messageId.charCodeAt(0) === 60 ? messageId.slice(1, -1) : messageId,
|
||||||
|
bytes: Number.parseInt(bytes, 10) || 0,
|
||||||
|
articleNumber,
|
||||||
|
postedAt: Date.parse(date),
|
||||||
|
file,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** As parseSubject: the name, the segment count and the base, from a subject whose counter was found. */
|
||||||
|
function parseBase(subject: string, nameEnd: number, counter: RegExpExecArray): ParsedSubjectBase | null {
|
||||||
|
const nameStart = subject.lastIndexOf('"', nameEnd - 1)
|
||||||
|
if (nameStart === -1) return null
|
||||||
|
const name = subject.slice(nameStart + 1, nameEnd)
|
||||||
|
if (!name) return null
|
||||||
|
const totalSegments = Number.parseInt(counter[2], 10)
|
||||||
|
if (totalSegments < 1) return null
|
||||||
|
const base = (subject.slice(0, counter.index) + subject.slice(counter.index + counter[0].length)).replace(/\s+/g, ' ').trim()
|
||||||
|
return { name, totalSegments, base }
|
||||||
|
}
|
||||||
|
|
||||||
|
function xrefGroups(xref: string, group: string): string[] {
|
||||||
|
const groups: string[] = []
|
||||||
|
const tokens = xref.trim().split(/\s+/)
|
||||||
|
for (let i = 1; i < tokens.length; i++) {
|
||||||
|
const colon = tokens[i].lastIndexOf(':')
|
||||||
|
if (colon > 0) groups.push(tokens[i].slice(0, colon))
|
||||||
|
}
|
||||||
|
return groups.length > 0 ? groups : [group]
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Calls `line` for each overview line of an XOVER reply's text, with its
|
||||||
|
* fields as strings (no object per line): number, subject, from, date,
|
||||||
|
* message-id, :bytes, and the Xref without its "Xref: " name.
|
||||||
|
*/
|
||||||
|
export function scanOverviewText(
|
||||||
|
text: string,
|
||||||
|
line: (number: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string) => void
|
||||||
|
) {
|
||||||
|
let start = 0
|
||||||
|
const length = text.length
|
||||||
|
while (start < length) {
|
||||||
|
let end = text.indexOf('\n', start)
|
||||||
|
if (end === -1) end = length
|
||||||
|
const stop = end > start && text.charCodeAt(end - 1) === 13 ? end - 1 : end
|
||||||
|
if (stop > start) {
|
||||||
|
// number, subject, from, date, message-id, references, bytes, lines, xref
|
||||||
|
const t1 = text.indexOf('\t', start)
|
||||||
|
const t2 = t1 === -1 ? -1 : text.indexOf('\t', t1 + 1)
|
||||||
|
const t3 = t2 === -1 ? -1 : text.indexOf('\t', t2 + 1)
|
||||||
|
const t4 = t3 === -1 ? -1 : text.indexOf('\t', t3 + 1)
|
||||||
|
const t5 = t4 === -1 ? -1 : text.indexOf('\t', t4 + 1)
|
||||||
|
const t6 = t5 === -1 ? -1 : text.indexOf('\t', t5 + 1)
|
||||||
|
const t7 = t6 === -1 ? -1 : text.indexOf('\t', t6 + 1)
|
||||||
|
const t8 = t7 === -1 ? -1 : text.indexOf('\t', t7 + 1)
|
||||||
|
if (t5 !== -1 && t5 < stop) {
|
||||||
|
const field = (a: number, b: number) => (a === -1 || a >= stop ? '' : text.slice(a + 1, b === -1 || b > stop ? stop : b))
|
||||||
|
let xref = t8 === -1 || t8 >= stop ? '' : field(t8, text.indexOf('\t', t8 + 1))
|
||||||
|
if (xref.startsWith('Xref:')) xref = xref.slice(xref.charCodeAt(5) === 32 ? 6 : 5)
|
||||||
|
line(Number.parseInt(text.slice(start, t1), 10), text.slice(t1 + 1, t2), field(t2, t3), field(t3, t4), field(t4, t5), field(t6, t7), xref)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
start = end + 1
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,3 +1,3 @@
|
|||||||
export { parseSubject, type ParsedSubject } from './subject.js'
|
export { parseSubject, type ParsedSubject } from './subject.js'
|
||||||
export { fileKey, weakFileKey, counterFileKey, looksLikeFilename } from './file_key.js'
|
export { fileKey, weakFileKey, counterFileKey, looksLikeFilename } from './file_key.js'
|
||||||
export { segmentFromHeader, type IngestSegment, type OverviewHeader } from './segment.js'
|
export { segmentFromHeader, type IngestSegment, type OverviewHeader, type IdentityCache } from './segment.js'
|
||||||
+20
-3
@@ -51,22 +51,39 @@ function common(header: OverviewHeader, groupName: string, articleNumber?: bigin
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A file's identity, the same for all of its articles: worth computing once
|
||||||
|
* per file (two SHA-1 hashes and the release rules were ~25% of an edge
|
||||||
|
* worker's time; LOG.md 2026-10-09).
|
||||||
|
*/
|
||||||
|
export type IdentityCache = Map<string, { fileKey: string; release: ReleaseIdentity | null }>
|
||||||
|
|
||||||
/** Builds a segment from the subject line; null when the subject is obfuscated. */
|
/** Builds a segment from the subject line; null when the subject is obfuscated. */
|
||||||
export function segmentFromHeader(
|
export function segmentFromHeader(
|
||||||
header: OverviewHeader,
|
header: OverviewHeader,
|
||||||
groupName: string,
|
groupName: string,
|
||||||
articleNumber?: bigint | number
|
articleNumber?: bigint | number,
|
||||||
|
cache?: IdentityCache
|
||||||
): IngestSegment | null {
|
): IngestSegment | null {
|
||||||
const parsed = header.subject ? parseSubject(header.subject) : null
|
const parsed = header.subject ? parseSubject(header.subject) : null
|
||||||
if (!parsed || !header['message-id']) return null
|
if (!parsed || !header['message-id']) return null
|
||||||
const base = common(header, groupName, articleNumber)
|
const base = common(header, groupName, articleNumber)
|
||||||
|
const id = `${base.poster}\n${parsed.base}\n${parsed.totalSegments}`
|
||||||
|
let identity = cache?.get(id)
|
||||||
|
if (!identity) {
|
||||||
|
identity = {
|
||||||
|
fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments),
|
||||||
|
release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }),
|
||||||
|
}
|
||||||
|
cache?.set(id, identity)
|
||||||
|
}
|
||||||
return {
|
return {
|
||||||
...base,
|
...base,
|
||||||
fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments),
|
fileKey: identity.fileKey,
|
||||||
subject: parsed.base,
|
subject: parsed.base,
|
||||||
name: parsed.name,
|
name: parsed.name,
|
||||||
totalParts: parsed.totalSegments,
|
totalParts: parsed.totalSegments,
|
||||||
number: parsed.segment,
|
number: parsed.segment,
|
||||||
release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }),
|
release: identity.release,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
import { test } from '@japa/runner'
|
||||||
|
import { SegmentScanner, scanOverviewText } from '#lib/edge/scan'
|
||||||
|
import { segmentFromHeader } from '#lib/headers/index'
|
||||||
|
|
||||||
|
const priorities = { 'alt.binaries.boneless': 1, 'alt.binaries.test': 2 }
|
||||||
|
|
||||||
|
test.group('edge / scanOverviewText and SegmentScanner', () => {
|
||||||
|
test('reads the fields of overview lines, Xref without its name', ({ assert }) => {
|
||||||
|
const text =
|
||||||
|
'101\tShow - "a.mkv" yEnc (1/2)\tposter <p@x>\tMon, 05 Oct 2026 10:00:00 +0000\t<one@test>\t\t700000\t5000\tXref: news.example alt.binaries.test:101\r\n' +
|
||||||
|
'102\tShow - "a.mkv" yEnc (2/2) 1234\tposter <p@x>\tMon, 05 Oct 2026 10:00:01 +0000\t<two@test>\t<ref@x>\t1234\t10\tXref: news.example alt.binaries.test:102 alt.binaries.boneless:9\r\n'
|
||||||
|
const lines: string[][] = []
|
||||||
|
scanOverviewText(text, (...fields) => lines.push(fields.map(String)))
|
||||||
|
assert.deepEqual(lines, [
|
||||||
|
['101', 'Show - "a.mkv" yEnc (1/2)', 'poster <p@x>', 'Mon, 05 Oct 2026 10:00:00 +0000', '<one@test>', '700000', 'news.example alt.binaries.test:101'],
|
||||||
|
['102', 'Show - "a.mkv" yEnc (2/2) 1234', 'poster <p@x>', 'Mon, 05 Oct 2026 10:00:01 +0000', '<two@test>', '1234', 'news.example alt.binaries.test:102 alt.binaries.boneless:9'],
|
||||||
|
])
|
||||||
|
})
|
||||||
|
|
||||||
|
test('the owner is the indexed group with the lowest priority in the Xref', ({ assert }) => {
|
||||||
|
const scanner = new SegmentScanner()
|
||||||
|
assert.equal(scanner.owner('news.example alt.binaries.test:5 alt.binaries.boneless:9', 'alt.binaries.test', priorities), 'alt.binaries.boneless')
|
||||||
|
assert.equal(scanner.owner('news.example alt.binaries.test:5 alt.binaries.other:9', 'alt.binaries.test', priorities), 'alt.binaries.test')
|
||||||
|
assert.equal(scanner.owner('news.example', 'alt.binaries.test', priorities), 'alt.binaries.test')
|
||||||
|
})
|
||||||
|
|
||||||
|
test('segments match segmentFromHeader, and share their file details', ({ assert }) => {
|
||||||
|
const scanner = new SegmentScanner()
|
||||||
|
const subject = (n: number) => `[03/10] - "Some.Release.part02.rar" yEnc (${n}/137) 51315352`
|
||||||
|
const xref = 'news.example alt.binaries.test:7'
|
||||||
|
const a = scanner.segment(7, subject(12), 'poster <p@x>', 'Mon, 05 Oct 2026 10:00:00 +0000', '<m12@x>', '700000', xref, 'alt.binaries.test')!
|
||||||
|
const b = scanner.segment(8, subject(13), 'poster <p@x>', 'Mon, 05 Oct 2026 10:00:00 +0000', '<m13@x>', '700000', xref, 'alt.binaries.test')!
|
||||||
|
const old = segmentFromHeader({ 'subject': subject(12), 'from': 'poster <p@x>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': '<m12@x>', ':bytes': '700000', 'xref': xref }, 'alt.binaries.test', 7)!
|
||||||
|
assert.deepEqual([a.fileKey, a.number, a.totalParts, a.messageId, a.bytes, a.file.name, a.file.groups, a.file.release?.key], [old.fileKey, old.number, old.totalParts, old.messageId, old.bytes, old.name, old.groups, old.release?.key])
|
||||||
|
assert.strictEqual(a.file, b.file)
|
||||||
|
assert.equal(b.number, 13)
|
||||||
|
// Obfuscated, or a counter only before the name: no segment.
|
||||||
|
assert.isNull(scanner.segment(9, 'Xq8r2KdLm3Zw91PqA7zz', 'p', '', '<o@x>', '1', xref, 'alt.binaries.test'))
|
||||||
|
assert.isNull(scanner.segment(9, '(1/2) "x.bin" yEnc', 'p', '', '<o@x>', '1', xref, 'alt.binaries.test'))
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user