diff --git a/.env.test b/.env.test index 28d2da0..32d335d 100644 --- a/.env.test +++ b/.env.test @@ -1 +1,2 @@ SESSION_DRIVER=memory +DB_DATABASE=usenet_rebuild_test diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 1598886..7ceeac1 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -69,7 +69,7 @@ class BatchStore { const rows = releases.slice(i, i + 500).map((r) => ({ release_id: ids.get(r.key)!, range_id: rangeId, - batch_seq: batch.from, + batch_from: batch.from, files: r.files, segments: r.segments, payload: r.payload, @@ -100,7 +100,13 @@ class BatchStore { } for (let i = 0; i < batch.fragments.length; i += 2000) { - const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ range_id: rangeId, file_key: f.fileKey, article_number: f.articleNumber, segment: JSON.stringify(f) })) + const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ + range_id: rangeId, + file_key: f.fileKey, + article_number: Math.min(...f.segments.map((s) => s[3])), + segments: f.segments.length, + fragment: JSON.stringify(f), + })) await trx.insertQuery().table('fragments').multiInsert(rows) } diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index ccf6c43..31f021b 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -2,39 +2,48 @@ import db from '@adonisjs/lucid/services/db' import logger from '@adonisjs/core/services/logger' import coordinatorConfig from '#config/coordinator' import NntpPool from '#services/nntp_pool' -import Spool from '#services/spool' +import Spool, { type ChunkSpan } from '#services/spool' interface Target { id: number group: string - start: number - end: number - next: number - leased: boolean + /** Chunks not spooled yet, in order. */ + missing: ChunkSpan[] + /** All of the range's chunks from its resume point. */ + total: number } +/** How often the targets are read again from the database. */ +const REFRESH_MS = 2000 + /** - * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch): - * the missing chunks of leased ranges first, then the next planned ranges, - * about SPOOL_AHEAD_PER_LEASE per active lease (at least SPOOL_MIN_AHEAD), - * within SPOOL_MAX_GB. One range per connection at a time, in order. + * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch). + * Every connection takes the next missing chunk: the leased ranges' first + * (a worker waits for them; all connections work on the same range, in + * order), then the next planned ranges, about SPOOL_AHEAD_PER_LEASE per + * active lease (at least SPOOL_MIN_AHEAD), within SPOOL_MAX_GB. One + * connection fills ~26 k overview lines/s, so a range on one connection + * kept its worker waiting (LOG.md 2026-10-09). */ export class Prefetcher { - private busy = new Set() + private targets: Target[] = [] + private claimed = new Set() + private running = 0 + private refreshedAt = 0 private stopped = false - /** One round: starts work for free connections; returns how many ranges it started. */ - async tick(): Promise { - const free = NntpPool.size - this.busy.size - if (free <= 0) return 0 - const targets = await this.targets(free) - for (const target of targets) { - this.busy.add(target.id) - this.spoolRange(target) - .catch((error) => logger.warn(`Spooling range ${target.id}: ${error.message}`)) - .finally(() => this.busy.delete(target.id)) + /** One round: refreshes the targets now and then, and puts idle connections to work. */ + async tick() { + if (Date.now() - this.refreshedAt >= REFRESH_MS) { + this.refreshedAt = Date.now() + this.targets = await this.readTargets() + } + while (!this.stopped && this.running < NntpPool.size && this.peek()) { + this.running++ + this.work() + .catch((error) => logger.warn(`Prefetching: ${error.message}`)) + .finally(() => this.running--) } - return targets.length } stop() { @@ -42,49 +51,64 @@ export class Prefetcher { } get active() { - return this.busy.size + return this.running } - private async targets(limit: number): Promise { - const busy = [...this.busy] - const select = (q: any) => - q - .from('ranges as r') - .join('groups as g', 'g.id', 'r.group_id') - .whereNull('r.spooled_at') - .whereNotIn('r.id', busy.length > 0 ? busy : [0]) - .select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.status') - // Leased ranges whose chunks aren't all there: a worker is waiting for them. - const leased = await select(db.query()).where('r.status', 'leased').orderBy('r.leased_at').limit(limit) - const out: Target[] = leased.map(toTarget) - if (out.length >= limit) return out - - const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases') - const [{ ahead }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ahead') - const wanted = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases)) - Number(ahead) - if (wanted <= 0 || (await Spool.size()) >= coordinatorConfig.spool.maxBytes) return out - const planned = await select(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(Math.min(limit - out.length, wanted)) - return [...out, ...planned.map(toTarget)] - } - - private async spoolRange(target: Target) { - let count = 0 - for (const span of Spool.chunksOf({ start: target.start, end: target.end }, target.next)) { - if (this.stopped) return - if (!(await Spool.has(target.id, span))) { + /** One connection's loop: chunks until there is nothing left to fetch. */ + private async work() { + while (!this.stopped) { + const next = this.take() + if (!next) return + const { target, span } = next + try { const data = await NntpPool.use(async (client) => { await client.group(target.group) return client.xoverRaw(span.from, span.to) }) await Spool.write(target.id, span, data) + const { rows } = await db.rawQuery( + `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`, + [target.id] + ) + if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + } finally { + this.claimed.delete(`${target.id}:${span.from}`) } - count++ - await db.from('ranges').where('id', target.id).update({ spooled_chunks: count }) } - await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + } + + private peek(): boolean { + return this.targets.some((t) => t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))) + } + + /** The first unclaimed missing chunk, in target order. */ + private take(): { target: Target; span: ChunkSpan } | null { + for (const target of this.targets) { + const index = target.missing.findIndex((s) => !this.claimed.has(`${target.id}:${s.from}`)) + if (index === -1) continue + const [span] = target.missing.splice(index, 1) + this.claimed.add(`${target.id}:${span.from}`) + return { target, span } + } + return null + } + + private async readTargets(): Promise { + const rows = (q: any) => + q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').whereNull('r.spooled_at').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id') + // Leased ranges whose chunks aren't all there: a worker is waiting for them. + const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at') + const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases') + const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases)) + const room = (await Spool.size()) < coordinatorConfig.spool.maxBytes + const planned = room ? await rows(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead) : [] + const targets: Target[] = [] + for (const row of [...leased, ...planned]) { + const spans = Spool.chunksOf({ start: Number(row.start_id), end: Number(row.end_id) }, Number(row.next_id)) + const missing: ChunkSpan[] = [] + for (const span of spans) if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span) + targets.push({ id: Number(row.id), group: row.group, missing, total: spans.length }) + } + return targets } } - -function toTarget(row: any): Target { - return { id: Number(row.id), group: row.group, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id), leased: row.status === 'leased' } -} diff --git a/app/services/rule_sets.ts b/app/services/rule_sets.ts index 731b088..8435674 100644 --- a/app/services/rule_sets.ts +++ b/app/services/rule_sets.ts @@ -12,12 +12,15 @@ class RuleSets { async current(): Promise { let row = await db.from('rule_sets').orderBy('version', 'desc').first() if (!row) { - const { version: _v, ...rules } = DEFAULT_RULES + const { version: _version, ...rules } = DEFAULT_RULES await db.table('rule_sets').insert({ version: 1, rules: JSON.stringify({ ...rules, droppedSignatures: [] }), note: 'default' }).onConflict('version').ignore() row = await db.from('rule_sets').orderBy('version', 'desc').first() } const groups = await db.from('groups').where('active', true).select('name', 'priority') + const { version: _v, ...defaults } = DEFAULT_RULES return { + // A field newer than the stored set takes today's default. + ...defaults, ...(row.rules as StoredRules), version: Number(row.version), priorities: Object.fromEntries(groups.map((g) => [g.name, Number(g.priority)])), diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index 3422c25..9d5901a 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -7,6 +7,8 @@ import env from '#start/env' import { XoverClient, parseOverviews } from '#lib/nntp/index' import { RangeIndexer } from '#lib/edge/index' import { DEFAULT_RULES, type Rules } from '#lib/rules/index' +import db from '@adonisjs/lucid/services/db' +import BatchStore from '#services/batch_store' /** * Measures the edge worker on a real range (REBUILD.md: TypeScript or Rust): @@ -43,6 +45,9 @@ export default class EdgeBench extends BaseCommand { @flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 }) declare idle: number + @flags.boolean({ description: 'Also store the batches (BatchStore, the configured database), timed' }) + declare store: boolean + @flags.string({ description: 'Indexed groups in priority order, comma-separated (owner rule)' }) declare priorities?: string @@ -92,6 +97,16 @@ export default class EdgeBench extends BaseCommand { const counts = new Map() const keys = new Set() let batches = 0 + let storeMs = 0 + let rangeId = 0 + let nodeId = 0 + if (this.store) { + const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000) }).returning('id') + const [n] = await db.table('nodes').insert({ name: `bench-${Date.now()}` }).returning('id') + nodeId = Number(n.id) + const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id') + rangeId = Number(r.id) + } const cpuStart = process.cpuUsage() const wall = Date.now() for (const file of files) { @@ -114,6 +129,11 @@ export default class EdgeBench extends BaseCommand { at = performance.now() const batch = indexer.take() t.index += performance.now() - at + if (this.store) { + const storedAt = performance.now() + await BatchStore.store(rangeId, nodeId, batch) + storeMs += performance.now() - storedAt + } out.headers += batch.headers out.releases += batch.releases.length batches++ @@ -138,5 +158,6 @@ export default class EdgeBench extends BaseCommand { this.logger.info(` ${batches} batches of ${this.batchChunks} chunks: ${out.releases} release chunks for ${keys.size} releases (${(out.releases / Math.max(1, keys.size)).toFixed(2)} chunks per release)`) this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`) for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(` dropped ${k}: ${v.toLocaleString('en')} articles`) + if (this.store) this.logger.info(` stored ${batches} batches in ${(storeMs / 1000).toFixed(1)} s (${(storeMs / Math.max(1, out.releases)).toFixed(2)} ms per release chunk)`) } } diff --git a/commands/edge_node.ts b/commands/edge_node.ts index 5d96fcf..d1e7f6e 100644 --- a/commands/edge_node.ts +++ b/commands/edge_node.ts @@ -1,7 +1,7 @@ import { BaseCommand, flags } from '@adonisjs/core/ace' import type { CommandOptions } from '@adonisjs/core/types/ace' import { setTimeout as sleep } from 'node:timers/promises' -import { 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' @@ -68,10 +68,13 @@ export default class EdgeNode extends BaseCommand { const { range, rules } = lease const started = Date.now() const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, rules }) - const totals = { headers: 0, releases: 0, candidates: 0 } + const totals = { headers: 0, releases: 0, candidates: 0, bytes: 0, postMs: 0 } let chunks = 0 const post = async (batch: EdgeBatch) => { + const at = Date.now() const stored = await api.batch(range.id, batch) + totals.postMs += Date.now() - at + totals.bytes += stored.bytes totals.headers += batch.headers totals.releases += stored.releases totals.candidates += stored.candidates @@ -94,7 +97,8 @@ export default class EdgeNode extends BaseCommand { const seconds = (Date.now() - started) / 1000 this.logger.info( `${range.group} ${range.start}–${range.end}: ${totals.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` + - `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} release chunks, ${totals.candidates} probe candidates.` + `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} release chunks, ${totals.candidates} probe candidates; ` + + `posted ${(totals.bytes / 1e6).toFixed(1)} MB in ${(totals.postMs / 1000).toFixed(1)} s.` ) } catch (error: any) { if (error instanceof LeaseLost) { @@ -144,10 +148,12 @@ class Api { return response.body as unknown as AsyncIterable } - async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number }> { - const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body: JSON.stringify(toWire(batch)) }) + async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number; bytes: number }> { + // Gzipped: chunk payloads are compressed already, the rest (names, fragments, candidates) compresses well. + const body = gzipSync(JSON.stringify(toWire(batch)), { level: 1 }) + const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body, headers: { 'Content-Encoding': 'gzip' } }) if (!response.ok) throw new Error(`batch: HTTP ${response.status}`) - return response.json() as Promise<{ releases: number; candidates: number }> + return { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length } } async call(rangeId: number, action: 'finish' | 'release' | 'heartbeat') { diff --git a/config/bodyparser.ts b/config/bodyparser.ts index b774fcf..7886690 100644 --- a/config/bodyparser.ts +++ b/config/bodyparser.ts @@ -27,10 +27,11 @@ const bodyParserConfig = defineConfig({ * Config for the JSON parser. */ json: { - /** - * Normalize empty string values to null. - */ - convertEmptyStringsToNull: true, + // Edge worker batches carry their releases' chunks (base64) and edge fragments; the limit is on the inflated body. + limit: '256mb', + + // Workers send '' on purpose (e.g. a rule count's key); null would break NOT NULL columns. + convertEmptyStringsToNull: false, /** * Content types handled by the JSON parser. @@ -66,8 +67,7 @@ const bodyParserConfig = defineConfig({ /** * Maximum accepted payload size for multipart requests. */ - // Edge worker batches carry their releases' chunks (a few MB, base64). - limit: '64mb', + limit: '20mb', /** * Content types handled by the multipart parser. diff --git a/database/migrations/1770000000500_create_release_chunks.ts b/database/migrations/1770000000500_create_release_chunks.ts index 9c0a387..afef892 100644 --- a/database/migrations/1770000000500_create_release_chunks.ts +++ b/database/migrations/1770000000500_create_release_chunks.ts @@ -17,7 +17,7 @@ export default class extends BaseSchema { id bigserial, release_id bigint NOT NULL, range_id bigint NOT NULL, - batch_seq integer NOT NULL, + batch_from bigint NOT NULL, files integer NOT NULL, segments integer NOT NULL, payload bytea NOT NULL, diff --git a/database/migrations/1770000000600_create_fragments.ts b/database/migrations/1770000000600_create_fragments.ts index a020974..b80e130 100644 --- a/database/migrations/1770000000600_create_fragments.ts +++ b/database/migrations/1770000000600_create_fragments.ts @@ -1,9 +1,9 @@ import { BaseSchema } from '@adonisjs/lucid/schema' /** - * Segments of files cut off by a range's edge (REBUILD.md, fragments): kept - * by range, merged when the neighbouring range is done, then written as a - * chunk. + * Files cut off by a range's edge (REBUILD.md, fragments): one row per file + * and range, with that range's segments; merged when the neighbouring range + * is done, then written as a chunk. */ export default class extends BaseSchema { async up() { @@ -11,8 +11,10 @@ export default class extends BaseSchema { table.bigIncrements('id') table.bigInteger('range_id').notNullable() table.string('file_key', 40).notNullable() + // The file's first article in this range. table.bigInteger('article_number').notNullable() - table.jsonb('segment').notNullable() + table.integer('segments').notNullable() + table.jsonb('fragment').notNullable() table.index(['file_key']) table.index(['range_id']) }) diff --git a/database/schema.ts b/database/schema.ts index 3eb293a..ddc7964 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -33,18 +33,20 @@ export class AuthAccessTokenSchema extends BaseModel { } export class FragmentSchema extends BaseModel { - static $columns = ['articleNumber', 'fileKey', 'id', 'rangeId', 'segment'] as const + static $columns = ['articleNumber', 'fileKey', 'fragment', 'id', 'rangeId', 'segments'] as const $columns = FragmentSchema.$columns @column() declare articleNumber: bigint | number @column() declare fileKey: string + @column() + declare fragment: any @column({ isPrimary: true }) declare id: bigint | number @column() declare rangeId: bigint | number @column() - declare segment: any + declare segments: number } export class GroupSchema extends BaseModel { @@ -165,10 +167,10 @@ export class RangeSchema extends BaseModel { } export class ReleaseChunkSchema extends BaseModel { - static $columns = ['batchSeq', 'createdAt', 'files', 'id', 'payload', 'rangeId', 'releaseId', 'segments'] as const + static $columns = ['batchFrom', 'createdAt', 'files', 'id', 'payload', 'rangeId', 'releaseId', 'segments'] as const $columns = ReleaseChunkSchema.$columns @column() - declare batchSeq: number + declare batchFrom: bigint | number @column.dateTime({ autoCreate: true }) declare createdAt: DateTime @column() diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index e34adfb..5c5c438 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -43,13 +43,17 @@ export interface ProbeCandidate { /** What a rule dropped in a batch (REBUILD.md, rules and recovery). */ export interface RuleCount { - rule: 'crosspost' | 'obfuscated' | 'garbage' | 'per-article' | 'incomplete' + rule: 'crosspost' | 'obfuscated' | 'executable' | 'garbage' | 'per-article' | 'incomplete' key: string articles: number files: number } -/** A segment of a file cut off by the range's edges, for the neighbouring range. */ +/** + * A file cut off by the range's edges, for the neighbouring range: its + * details once, and the segments this range had, as + * [part, message-id, bytes, article number]. + */ export interface Fragment { fileKey: string name: string @@ -58,11 +62,8 @@ export interface Fragment { totalParts: number groups: string[] release: { key: string; name: string; source: string; fileTotal?: number } | null - number: number - messageId: string - bytes: number - articleNumber: number postedAt: number | null + segments: [number, string, number, number][] } export interface EdgeBatch { @@ -186,13 +187,25 @@ export class RangeIndexer { } } - /** Given-up segments: kept 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[]) { const { start, end } = this.options + const byFile = new Map() for (const s of segments) { const n = Number(s.articleNumber) - if (Number.isFinite(n) && (n < start + this.edge || n > end - this.edge)) this.batch.fragments.push(fragmentOf(s)) - else this.count('incomplete', '', 1, 0) + if (!Number.isFinite(n) || (n >= start + this.edge && n <= end - this.edge)) { + this.count('incomplete', '', 1, 0) + continue + } + let fragment = byFile.get(s.fileKey) + if (!fragment) { + fragment = fragmentOf(s) + byFile.set(s.fileKey, fragment) + 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 } } @@ -301,10 +314,7 @@ function fragmentOf(s: IngestSegment): Fragment { 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, - number: s.number, - messageId: s.messageId, - bytes: s.bytes, - articleNumber: Number(s.articleNumber), - postedAt: s.postedAt && Number.isFinite(s.postedAt.getTime()) ? s.postedAt.getTime() : null, + postedAt: null, + segments: [], } } diff --git a/lib/rules/rule_set.ts b/lib/rules/rule_set.ts index e46b5f3..ed8edf0 100644 --- a/lib/rules/rule_set.ts +++ b/lib/rules/rule_set.ts @@ -18,6 +18,12 @@ export interface Rules { perArticleMaxParts: number /** Share of each rule's drops kept in full as a sample (by a hash of the release key). */ samplePercent: number + /** + * Drop releases of bare executables (malware floods posted under film and + * software names; 92% of the catalog rows of the first test ranges, LOG.md + * 2026-10-09). Genuine software comes in archives. + */ + dropExecutables: boolean } export const DEFAULT_RULES: Omit = { @@ -25,9 +31,14 @@ export const DEFAULT_RULES: Omit = { droppedTools: ['ngPost', 'hex32', 'yuun', 'JBinUp', 'random'], perArticleMaxParts: 3, samplePercent: 1, + dropExecutables: true, } -export type Verdict = { rule: null } | { rule: 'garbage' | 'per-article'; key: string; sample: boolean } +export type Verdict = { rule: null } | { rule: 'executable' | 'garbage' | 'per-article'; key: string; sample: boolean } + +/** Files that come along with a release's content: they don't make it one of executables or not. */ +const SIDECAR = /\.(par2|nfo|sfv|txt|jpe?g|png|nzb)$/i +const EXECUTABLE = /\.(exe|scr|com|pif|bat|cmd|vbs|lnk)\s*$/i export function isRandomFile(name: string) { return looksRandom(fileStem(name)) @@ -52,15 +63,21 @@ export class RuleSet { } /** - * A release's verdict from its files: kept as a whole when any file has a - * readable name; else dropped as garbage when its signature or tool is + * A release's verdict from its files: dropped when its content is bare + * executables; else kept as a whole when any file has a readable name; else dropped as garbage when its signature or tool is * dropped, as per-article random subjects when every file has few parts. */ judge(releaseKey: string, files: { name: string; totalParts: number; groups: string[]; firstMessageId: string }[]): Verdict { + const sample = explored(releaseKey, this.rules.samplePercent) + if (this.rules.dropExecutables) { + const content = files.filter((f) => !SIDECAR.test(f.name)) + if (content.length > 0 && content.every((f) => EXECUTABLE.test(f.name))) { + return { rule: 'executable', key: EXECUTABLE.exec(content[0].name)![1].toLowerCase(), sample } + } + } if (files.some((f) => !isRandomFile(f.name))) return { rule: null } const first = files[0] const tool = uploadTool(first.firstMessageId) - const sample = explored(releaseKey, this.rules.samplePercent) // Garbage first: a dropped tool or signature isn't worth reading bodies for either. const group = first.groups[0] ?? '' if (this.tools.has(tool) || this.signatures.has(signatureKey(tool, group))) return { rule: 'garbage', key: `${tool} ${group}`, sample } diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index d4c5d14..78a31e4 100644 --- a/tests/functional/node_api.spec.ts +++ b/tests/functional/node_api.spec.ts @@ -19,6 +19,8 @@ function chunk(at: number): Buffer { const lines = [1, 2].map( (n) => `${at + n - 1}\tSome.Show.S01E0${at % 7} [1/1] - "Some.Show.S01E0${at % 7}.1080p.mkv" yEnc (${n}/2)\tp \tMon, 05 Oct 2026 10:00:00 +0000\t\t\t700000\t5000\tXref: news.example ${GROUP}:${at + n - 1}` ) + // An obfuscated subject: counted under the key '' (which must survive the body parser). + lines.push(`${at + 2}\tobfuscated\tp \tMon, 05 Oct 2026 10:00:00 +0000\t\t\t700000\t5000\tXref: news.example ${GROUP}:${at + 2}`) return deflateSync(Buffer.from(lines.join('\r\n') + '\r\n')) } @@ -75,6 +77,8 @@ test.group('Node API', (group) => { // The same batch again: refused. ;(await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)).assertStatus(409) + assert.deepEqual((await db.from('rule_counts').where('range_id', rangeId)).map((c) => [c.rule, c.key, Number(c.articles)]), [['obfuscated', '', 3]]) + ;(await client.post(`/api/v1/node/ranges/${rangeId}/finish`).bearerToken(token)).assertStatus(200) const range = await db.from('ranges').where('id', rangeId).first() assert.equal(range.status, 'done') diff --git a/tests/unit/edge/range_indexer.spec.ts b/tests/unit/edge/range_indexer.spec.ts index bc2281c..7381b53 100644 --- a/tests/unit/edge/range_indexer.spec.ts +++ b/tests/unit/edge/range_indexer.spec.ts @@ -60,7 +60,7 @@ test.group('edge / RangeIndexer', () => { indexer.add([...post(5, 'Edge.File.2020.part1.rar', 3).slice(0, 2), ...post(500_000, 'Middle.File.2020.part1.rar', 3).slice(0, 2)], 600_000) indexer.finish() const batch = indexer.take() - assert.deepEqual(batch.fragments.map((f) => [f.name, f.number]), [['Edge.File.2020.part1.rar', 1], ['Edge.File.2020.part1.rar', 2]]) + assert.deepEqual(batch.fragments.map((f) => [f.name, f.segments.map((s) => [s[0], s[3]])]), [['Edge.File.2020.part1.rar', [[1, 5], [2, 6]]]]) assert.deepEqual(batch.counts.map((c) => [c.rule, c.articles]), [['incomplete', 2]]) assert.lengthOf(batch.releases, 0) }) diff --git a/tests/unit/rules/rule_set.spec.ts b/tests/unit/rules/rule_set.spec.ts index 66caf0b..20c6fd8 100644 --- a/tests/unit/rules/rule_set.spec.ts +++ b/tests/unit/rules/rule_set.spec.ts @@ -21,6 +21,12 @@ test.group('rules / verdicts', () => { assert.deepEqual(rules.judge('k', [file('Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 50, NGPOST), file('Some.Show.S01E01.1080p.nfo', 1, NGPOST)]), { rule: null }) }) + test('bare executables are dropped, sidecars aside; an executable among other content is kept', ({ assert }) => { + assert.include(rules.judge('k', [file('How to Make Love to a Woman 2010.scr', 10, NYUU), file('How to Make Love to a Woman 2010.par2', 1, NYUU)]), { rule: 'executable', key: 'scr' }) + assert.include(rules.judge('k', [file('Adobe Illustrator CS2 + ssg keygen.zip .exe', 3, NYUU)]), { rule: 'executable', key: 'exe' }) + assert.deepEqual(rules.judge('k', [file('Some.App.v1.0.part1.rar', 50, NYUU), file('setup.exe', 2, NYUU)]), { rule: null }) + }) + test('random names: per-article with few parts, garbage by tool or signature, else kept', ({ assert }) => { assert.include(rules.judge('k', [file('Hs7dKq2LmZ9xWv3RtYp1Qa', 2, NYUU)]), { rule: 'per-article', key: 'nyuu' }) assert.include(rules.judge('k', [file('Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 50, NGPOST)]), { rule: 'garbage', key: 'ngPost alt.binaries.ath' })