From 211820067aee715c96cf36ee7ac70af36297b41e Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Thu, 8 Oct 2026 11:21:01 +0200 Subject: [PATCH] Drop garbage uploads early (D42) Random-named uploads that never get a real name or password were ~90% of what seeding stores. Two filters: - before storing (seeding, ingest): random-named uploads of a dropped signature (posting tool from the Message-ID, first group) or of a release dropped after its probe aren't stored; 2% of a dropped signature's releases are still kept and probed. Stored files of dropped signatures aren't probed. - after a probe: a random-named release that is still unnamed, without password, names from files or spot, with all its probes done, and encrypted, a container or missing its first article, is deleted and tombstoned in dropped_releases. Signatures are learned by gc:scheduler from a sample of probed releases (signature_probes); garbage:learn --seed fills it from existing files. Co-Authored-By: Claude Opus 5.5 --- app/services/ContentService.ts | 17 +- app/services/GarbageFilter.ts | 217 ++++++++++++++++++ app/services/IngestService.ts | 10 +- app/services/SeedService.ts | 15 +- commands/GarbageLearn.ts | 38 +++ commands/GcScheduler.ts | 18 ++ commands/SeedRange.ts | 5 +- config/indexer.ts | 19 ++ .../1768620765270_garbage_filter.ts | 64 ++++++ database/schema.ts | 41 ++++ lib/garbage/index.ts | 52 +++++ start/env.ts | 3 + tests/functional/garbage.spec.ts | 190 +++++++++++++++ 13 files changed, 683 insertions(+), 6 deletions(-) create mode 100644 app/services/GarbageFilter.ts create mode 100644 commands/GarbageLearn.ts create mode 100644 database/migrations/1768620765270_garbage_filter.ts create mode 100644 lib/garbage/index.ts create mode 100644 tests/functional/garbage.spec.ts diff --git a/app/services/ContentService.ts b/app/services/ContentService.ts index 21c05a2..8aeed7d 100644 --- a/app/services/ContentService.ts +++ b/app/services/ContentService.ts @@ -25,6 +25,7 @@ import SevenZipService from '#services/SevenZipService' import SfvService from '#services/SfvService' import ArchiveWalkService from '#services/ArchiveWalkService' import { segmentsOf } from '#services/FileStore' +import GarbageFilter from '#services/GarbageFilter' /** * MD5 of the first 16 KiB of a file, as par2 records it, from the body of @@ -49,6 +50,8 @@ export interface ContentResult { nicovideo?: string /** The real name from the yEnc header, for a file posted under a random subject (D36). */ yencName?: string + /** Why the file's release was deleted as garbage after this probe (D42). */ + garbage?: string } /** @@ -129,6 +132,8 @@ class ContentService { async prepare(fileId: number): Promise { const file = await db.from('files').where('id', fileId).first() if (!file || !this.isCandidate(file.filename)) return null + // Stored before its signature was found to be garbage (D42): not worth an article. + if (await GarbageFilter.skipProbe(file)) return null if ((sevenZipVolume(String(file.real_name ?? file.filename)) ?? 0) > 1) { const volumes = await SevenZipService.volumes(file) if (!volumes || volumes[0].archive) return null @@ -148,8 +153,17 @@ class ContentService { } } - /** The database half after a probe: names, groupings, listings and flags from what it found. */ + /** + * The database half after a probe: names, groupings, listings and flags from + * what it found. Then the release is deleted if it turned out to be garbage (D42). + */ async apply(task: InspectTask, probe: InspectProbe): Promise { + const result = await this.applyProbe(task, probe) + result.garbage = (await GarbageFilter.afterProbe(task.fileId, probe.contentType)) ?? undefined + return result + } + + private async applyProbe(task: InspectTask, probe: InspectProbe): Promise { const result: ContentResult = { named: 0, contentType: probe.contentType } const file = await db.from('files').where('id', task.fileId).first() if (!file) return result @@ -277,6 +291,7 @@ export function describeContent(filename: string, content: ContentResult | null) if (content?.contentType === 'nzbe') lines.push(`"${filename}" is an encrypted NZB container.`) if (content?.contentType === 'rclone') lines.push(`"${filename}" is rclone-encrypted.`) if (content?.yencName) lines.push(`"${filename}" is "${content.yencName}" (yEnc name).`) + if (content?.garbage) lines.push(`"${filename}": release deleted as garbage (${content.garbage}, no name or password).`) if (content?.nicovideo) lines.push(`"${filename}" → "${content.nicovideo}".`) if (content?.sfvMoved) lines.push(`SFV "${filename}" moved ${content.sfvMoved} files into its release.`) if (content?.contentType === 'yencbin') lines.push(`"${filename}" is a yEncBin encrypted container.`) diff --git a/app/services/GarbageFilter.ts b/app/services/GarbageFilter.ts new file mode 100644 index 0000000..218fe20 --- /dev/null +++ b/app/services/GarbageFilter.ts @@ -0,0 +1,217 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' +import indexerConfig from '#config/indexer' +import { looksRandom } from '#services/ArchiveService' +import { fileStem } from '#lib/release/index' +import { explored, signatureGroup, signatureKey, uploadTool } from '#lib/garbage/index' + +/** What the filter needs to know about an upload before it is stored. */ +export interface Upload { + /** The posted file name. */ + name: string + /** Any of its articles' Message-IDs (one tool posts all of a file). */ + messageId: string | null | undefined + groups: unknown + /** Its release's key (the file key when it has none): kept or dropped as a whole. */ + releaseKey: string +} + +/** Contents that can't name a random-named release without a key (LOG.md 2026-10-08 (5)). */ +const CONTAINERS = ['rclone', 'nzbe', 'yencbin'] + +/** How long the dropped signatures are cached per process. */ +const CACHE_MS = 5 * 60 * 1000 + +export function isRandomName(name: string) { + return looksRandom(fileStem(name)) +} + +/** + * Garbage: random-named uploads that never get a real name and have no + * known password (CONCEPT.md D42). Two filters: + * - before storing (seeding, ingest): random-named uploads whose signature + * (posting tool, first group) is dropped, or whose release was dropped + * after its probe, aren't stored; + * - after a probe: a random-named release whose probes are all done, that + * is still without a real name, password, names from its files or spot, + * and is encrypted, a container or missing its first article, is deleted. + * The dropped signatures are learned from a sample of probed releases + * (learn(), gc:scheduler). + */ +class GarbageFilter { + private dropped: Set | null = null + private loadedAt = 0 + + /** The dropped signatures, cached for CACHE_MS. */ + async droppedSignatures(): Promise> { + if (this.dropped && Date.now() - this.loadedAt < CACHE_MS) return this.dropped + const rows = await db.from('upload_signatures').where('dropped', true).select('tool', 'group_name') + this.dropped = new Set(rows.map((row) => signatureKey(row.tool, row.group_name))) + this.loadedAt = Date.now() + return this.dropped + } + + /** Forgets the cached signatures (after learn(), and in tests). */ + reset() { + this.dropped = null + } + + /** Whether a random-named upload's signature is dropped and it isn't in the explored share. */ + private droppedBySignature(upload: Upload, signatures: Set) { + if (signatures.size === 0) return false + if (!signatures.has(signatureKey(uploadTool(upload.messageId), signatureGroup(upload.groups)))) return false + return !explored(upload.releaseKey, indexerConfig.garbage.explorePercent) + } + + /** + * The items worth storing: drops random-named uploads of dropped signatures + * (but explorePercent of their releases) and of dropped releases. + */ + async keep(items: T[], upload: (item: T) => Upload): Promise<{ kept: T[]; dropped: number }> { + if (!indexerConfig.garbage.enabled || items.length === 0) return { kept: items, dropped: 0 } + const signatures = await this.droppedSignatures() + // Per release, kept as a whole when any of its files has a readable name. A file is + // looked at once: an ingest batch has many articles of each. + const verdicts = new Map() + const seen = new Set() + for (const item of items) { + const u = upload(item) + if (verdicts.get(u.releaseKey) === 'keep' || seen.has(`${u.releaseKey}\n${u.name}`)) continue + seen.add(`${u.releaseKey}\n${u.name}`) + if (!isRandomName(u.name)) verdicts.set(u.releaseKey, 'keep') + else if (!verdicts.has(u.releaseKey)) verdicts.set(u.releaseKey, this.droppedBySignature(u, signatures) ? 'drop' : 'check') + } + const check = [...verdicts].filter(([, v]) => v === 'check').map(([key]) => key) + for (let i = 0; i < check.length; i += 5000) { + const rows = await db.from('dropped_releases').whereIn('release_key', check.slice(i, i + 5000)).select('release_key') + for (const row of rows) verdicts.set(row.release_key, 'drop') + } + const kept = items.filter((item) => verdicts.get(upload(item).releaseKey) !== 'drop') + return { kept, dropped: items.length - kept.length } + } + + /** + * Whether a stored file's probe can be skipped: a random-named file of a + * dropped signature, outside the explored share (files stored before its + * signature was dropped). Its binary stops waiting for inspection. + */ + async skipProbe(file: { id: number | string; filename: string; real_name?: string | null; first_message_id?: string | null; groups: unknown; release_id?: number | string | null }) { + if (!indexerConfig.garbage.enabled || file.real_name || !isRandomName(file.filename)) return false + const signatures = await this.droppedSignatures() + if (!signatures.has(signatureKey(uploadTool(file.first_message_id), signatureGroup(file.groups)))) return false + const release = file.release_id ? await db.from('releases').where('id', file.release_id).select('release_key').first() : null + if (explored(release?.release_key ?? String(file.id), indexerConfig.garbage.explorePercent)) return false + await db.from('binaries').where('file_id', file.id).where('awaiting_inspection', true).update({ awaiting_inspection: null }) + return true + } + + /** + * After a file's probe has been applied: notes its release for learning + * (a sample) and deletes the release when it is garbage. Never throws: + * the probe's result is stored already. Returns why the release was + * deleted, or null. + */ + async afterProbe(fileId: number, contentType: string | null): Promise { + try { + const file = await db.from('files').where('id', fileId).select('id', 'filename', 'real_name', 'first_message_id', 'groups', 'release_id').first() + if (!file?.release_id || !isRandomName(String(file.filename))) return null + const releaseId = Number(file.release_id) + if (releaseId % indexerConfig.garbage.probeSample === 0) { + await db.rawQuery(`INSERT INTO signature_probes (release_id, tool, group_name) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [ + releaseId, + uploadTool(file.first_message_id), + signatureGroup(file.groups), + ]) + } + if (!indexerConfig.garbage.enabled) return null + return await this.dropIfGarbage(releaseId, contentType) + } catch (error: any) { + logger.warn(`Garbage check after probing file ${fileId}: ${error.message}`) + return null + } + } + + private async dropIfGarbage(releaseId: number, contentType: string | null): Promise { + const { rows } = await db.rawQuery( + `SELECT r.release_key, r.container, r.encrypted, + is_real_release_name(r.name) OR r.password IS NOT NULL OR coalesce(r.search_files, '') <> '' + OR EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id) AS named, + EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id AND b.awaiting_inspection) AS waiting + FROM releases r WHERE r.id = ?`, + [releaseId] + ) + const release = rows[0] + if (!release || release.named || release.waiting) return null + const reason = CONTAINERS.includes(release.container) ? release.container : release.encrypted === true ? 'encrypted' : contentType === 'missing' ? 'missing' : null + if (!reason) return null + await db.transaction(async (trx) => { + await trx.rawQuery(`INSERT INTO dropped_releases (release_key, reason) VALUES (?, ?) ON CONFLICT DO NOTHING`, [release.release_key, reason]) + await trx.rawQuery(`DELETE FROM binaries WHERE release_id = ?`, [releaseId]) + await trx.rawQuery(`DELETE FROM files WHERE release_id = ?`, [releaseId]) + await trx.rawQuery(`DELETE FROM releases WHERE id = ?`, [releaseId]) + }) + return reason + } + + /** + * Learns the signatures from the probed releases of the last windowDays: + * dropped when at least minProbes are known and at most maxNamedRate got a + * real name, password, names from their files or a spot. A signature with + * fewer probes keeps its verdict (a dropped one only gets explored probes). + * Releases deleted as garbage count as unnamed. Returns the signatures + * whose verdict changed. + */ + async learn(): Promise<{ signatures: number; dropped: number; changed: string[] }> { + const { minProbes, maxNamedRate, windowDays } = indexerConfig.garbage + await db.rawQuery(`DELETE FROM signature_probes WHERE probed_at < now() - make_interval(days => ?)`, [windowDays]) + const before = new Map((await db.from('upload_signatures').select('tool', 'group_name', 'dropped')).map((row) => [signatureKey(row.tool, row.group_name), row.dropped])) + await db.rawQuery( + `INSERT INTO upload_signatures (tool, group_name, probed, named, dropped, updated_at) + SELECT tool, group_name, probed, named, probed >= ?::int AND named <= probed * ?::float8, now() + FROM ( + SELECT p.tool, p.group_name, count(*)::int AS probed, + (count(*) FILTER (WHERE is_real_release_name(r.name) OR r.password IS NOT NULL OR coalesce(r.search_files, '') <> '' + OR EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id)))::int AS named + FROM signature_probes p LEFT JOIN releases r ON r.id = p.release_id + GROUP BY p.tool, p.group_name + ) s + ON CONFLICT (tool, group_name) DO UPDATE SET + probed = excluded.probed, named = excluded.named, updated_at = now(), + dropped = CASE WHEN excluded.probed >= ?::int THEN excluded.dropped ELSE upload_signatures.dropped END`, + [minProbes, maxNamedRate, minProbes] + ) + const after = await db.from('upload_signatures').select('tool', 'group_name', 'dropped', 'probed', 'named') + const changed = after + .filter((row) => (before.get(signatureKey(row.tool, row.group_name)) ?? false) !== row.dropped) + .map((row) => `${row.tool} ${row.group_name || '(no group)'}: ${row.dropped ? 'dropped' : 'kept'} (${row.named}/${row.probed} named)`) + this.reset() + return { signatures: after.length, dropped: after.filter((row) => row.dropped).length, changed } + } + + /** + * Fills signature_probes from a sample of files probed before it existed + * (garbage:learn --seed): random-named files with a content type, from + * TABLESAMPLE SYSTEM (percent) of files. Returns the releases added. + */ + async seed(percent: number): Promise { + const { rows } = await db.rawQuery( + `SELECT DISTINCT ON (f.release_id) f.release_id, f.first_message_id, f.groups + FROM files f TABLESAMPLE SYSTEM (?) + WHERE f.content_type IS NOT NULL AND f.release_id IS NOT NULL AND NOT is_readable_name(f.filename)`, + [percent] + ) + let added = 0 + for (let i = 0; i < rows.length; i += 2000) { + const chunk = rows.slice(i, i + 2000) + const result = await db.rawQuery( + `INSERT INTO signature_probes (release_id, tool, group_name) + SELECT * FROM unnest(?::bigint[], ?::text[], ?::text[]) ON CONFLICT DO NOTHING`, + [chunk.map((row: any) => row.release_id), chunk.map((row: any) => uploadTool(row.first_message_id)), chunk.map((row: any) => signatureGroup(row.groups))] + ) + added += result.rowCount ?? 0 + } + return added + } +} + +export default new GarbageFilter() diff --git a/app/services/IngestService.ts b/app/services/IngestService.ts index a76d4c4..051841d 100644 --- a/app/services/IngestService.ts +++ b/app/services/IngestService.ts @@ -4,6 +4,7 @@ import type { QueryClientContract } from '@adonisjs/lucid/types/database' import type { Knex } from 'knex' import QueueService from '#services/QueueService' import NameService from '#services/NameService' +import GarbageFilter from '#services/GarbageFilter' import type { IngestSegment } from '#lib/ingest/segment' // Header parsing is pure and lives in lib, so seeding nodes need no database (CONCEPT.md D37). @@ -110,7 +111,14 @@ function validDate(date?: Date) { * no-op. See CONCEPT.md (D12). */ class IngestService { - async ingest(segments: IngestSegment[]): Promise { + async ingest(batch: IngestSegment[]): Promise { + // Garbage isn't stored at all: no release, binary or parts (CONCEPT.md D42). + const { kept: segments } = await GarbageFilter.keep(batch, (s) => ({ + name: s.name, + messageId: s.messageId, + groups: s.groups, + releaseKey: s.release?.key ?? s.fileKey, + })) if (segments.length === 0) return [] // Short statements, no enclosing transaction: every step is idempotent, and diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index f8f49e0..efe3dd6 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -5,6 +5,7 @@ import SfvService from '#services/SfvService' import ContentService from '#services/ContentService' import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire' import InspectionQueue from '#services/InspectionQueue' +import GarbageFilter from '#services/GarbageFilter' /** Articles from a range's edge within which a file may continue in the neighbouring range. */ const EDGE_ARTICLES = 50_000 @@ -15,6 +16,8 @@ export interface SeedStoreResult { existing: number releases: number inspections: number + /** Random-named files of garbage signatures or dropped releases, not stored (D42). */ + dropped: number } /** @@ -26,9 +29,17 @@ export interface SeedStoreResult { */ class SeedService { async store(batch: SeedFile[]): Promise { - const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0 } + const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0, dropped: 0 } // A repost can complete the same file twice in one batch; ON CONFLICT DO UPDATE takes each row once. - const files = [...new Map(batch.map((file) => [file.fileKey, file])).values()] + const unique = [...new Map(batch.map((file) => [file.fileKey, file])).values()] + // Garbage isn't stored at all: no release, file or binary (CONCEPT.md D42). + const { kept: files, dropped } = await GarbageFilter.keep(unique, (file) => ({ + name: file.name, + messageId: file.segments[0]?.[1], + groups: file.groups, + releaseKey: file.release?.key ?? file.fileKey, + })) + result.dropped = dropped if (files.length === 0) return result const knex = db.connection().getWriteClient() const now = new Date() diff --git a/commands/GarbageLearn.ts b/commands/GarbageLearn.ts new file mode 100644 index 0000000..3c4ffdd --- /dev/null +++ b/commands/GarbageLearn.ts @@ -0,0 +1,38 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import GarbageFilter from '#services/GarbageFilter' + +/** + * Learns the garbage upload signatures now (gc:scheduler does it every + * GARBAGE_LEARN_INTERVAL_MS) and lists them (CONCEPT.md D42). --seed first + * fills signature_probes from a sample of the files probed before it + * existed, so there is something to learn from right away. + */ +export default class GarbageLearn extends BaseCommand { + public static commandName = 'garbage:learn' + public static description = 'Learns which upload signatures are garbage and lists them.' + + public static options: CommandOptions = { + startApp: true, + } + + @flags.number({ description: 'First add probed releases from TABLESAMPLE SYSTEM (percent) of files' }) + declare seed?: number + + public async run() { + if (this.seed) { + const started = Date.now() + const added = await GarbageFilter.seed(this.seed) + this.logger.info(`Seeded ${added} probed releases from ${this.seed}% of files in ${Math.round((Date.now() - started) / 1000)} s.`) + } + const { signatures, dropped, changed } = await GarbageFilter.learn() + for (const line of changed) this.logger.info(`Changed: ${line}`) + const rows = await db.from('upload_signatures').orderBy('probed', 'desc').limit(40) + for (const row of rows) { + const rate = row.probed ? ((100 * row.named) / row.probed).toFixed(1) : '-' + this.logger.info(`${row.dropped ? 'DROP' : 'keep'} ${row.tool.padEnd(10)} ${(row.group_name || '(no group)').padEnd(36)} ${String(row.named).padStart(6)}/${String(row.probed).padEnd(6)} named (${rate}%)`) + } + this.logger.info(`${dropped} of ${signatures} signatures dropped.`) + } +} diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index 5b45166..15e2c57 100644 --- a/commands/GcScheduler.ts +++ b/commands/GcScheduler.ts @@ -11,6 +11,8 @@ import StagingService from '#services/StagingService' import FilePartitions from '#services/FilePartitions' import ReleaseCompleteness from '#services/ReleaseCompleteness' import FileKeys from '#services/FileKeys' +import GarbageFilter from '#services/GarbageFilter' +import indexerConfig from '#config/indexer' import Shutdown from '#services/Shutdown' const CLEAN_BATCH = 1000 @@ -34,6 +36,8 @@ export default class GcScheduler extends BaseCommand { /** When release completeness was last recomputed (it runs less often than the sweep). */ private completenessAt = 0 + /** When the garbage signatures were last learned (D42). */ + private learnedAt = 0 /** When split uploads were last merged. */ private mergedAt = 0 @@ -59,6 +63,11 @@ export default class GcScheduler extends BaseCommand { } catch (err: any) { this.logger.error(`file_keys fill failed: ${err.message}`) } + try { + await this.learnGarbage() + } catch (err: any) { + this.logger.error(`Garbage signatures failed: ${err.message}`) + } if (!this.once) await Shutdown.sleep(gcConfig.intervalMs) } while (!this.once && !Shutdown.requested) } finally { @@ -155,6 +164,15 @@ export default class GcScheduler extends BaseCommand { if (checked > 0) this.logger.info(`Release completeness: ${checked} changed releases checked in ${Math.round((Date.now() - started) / 1000)} s.`) } + /** Learns which upload signatures are garbage from the probed releases (CONCEPT.md D42). */ + private async learnGarbage() { + if (this.dryRun || (!this.once && Date.now() - this.learnedAt < indexerConfig.garbage.learnIntervalMs)) return + this.learnedAt = Date.now() + const { signatures, dropped, changed } = await GarbageFilter.learn() + this.logger.info(`Garbage signatures: ${dropped} of ${signatures} dropped.`) + for (const line of changed) this.logger.info(`Garbage signature ${line}`) + } + private async rebuildIndexes() { const key = 'gc:reindex:last' const last = Number((await RedisService.client.get(key)) ?? 0) diff --git a/commands/SeedRange.ts b/commands/SeedRange.ts index b38777c..913c6a0 100644 --- a/commands/SeedRange.ts +++ b/commands/SeedRange.ts @@ -51,7 +51,7 @@ export default class SeedRange extends BaseCommand { const first = Number(this.start) if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`) const last = first + this.count - 1 - const stats = { headers: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 } + const stats = { headers: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, dropped: 0, fragments: 0, inspections: 0 } let storeMs = 0 const cpuStart = process.cpuUsage() const started = Date.now() @@ -90,6 +90,7 @@ export default class SeedRange extends BaseCommand { storeMs += Date.now() - t stats.files += stored.files stats.existing += stored.existing + stats.dropped += stored.dropped stats.inspections += stored.inspections } this.progress(stats, started) @@ -106,7 +107,7 @@ export default class SeedRange extends BaseCommand { this.logger.info( `${this.group} ${first}-${last}: ${n(stats.headers)} headers in ${seconds.toFixed(0)} s (${n(Math.round(stats.headers / seconds))}/s). ` + `${n(stats.obfuscated)} obfuscated, ${n(stats.deferred)} deferred. ` + - `${n(stats.complete)} complete files (${n(stats.files)} new, ${n(stats.existing)} known), ` + + `${n(stats.complete)} complete files (${n(stats.files)} new, ${n(stats.existing)} known, ${n(stats.dropped)} garbage), ` + `${n(stats.fragments)} fragment segments, ${n(stats.inspections)} queued for inspection.` ) this.logger.info(`Store ${(storeMs / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`) diff --git a/config/indexer.ts b/config/indexer.ts index e169eaa..7b63476 100644 --- a/config/indexer.ts +++ b/config/indexer.ts @@ -95,6 +95,25 @@ export default { maxHddGb: env.get('BACKFILL_MAX_HDD_GB', 150), }, + /** + * Dropping garbage uploads early (CONCEPT.md D42). A signature (posting tool, first + * group) is dropped once at least minProbes of its probed random-named releases are + * known and at most maxNamedRate of them got a real name or password; its + * random-named files are then not stored, except explorePercent of its releases. + */ + garbage: { + /** Off: nothing is filtered or deleted (the learning still runs). */ + enabled: env.get('GARBAGE_FILTER', true), + explorePercent: env.get('GARBAGE_EXPLORE_PERCENT', 2), + minProbes: 100, + maxNamedRate: 0.005, + /** Probed releases learned from: 1 in this many (by release id), for the last windowDays. */ + probeSample: 4, + windowDays: 30, + /** How often gc:scheduler learns the signatures again. */ + learnIntervalMs: env.get('GARBAGE_LEARN_INTERVAL_MS', 3 * 60 * 60 * 1000), + }, + /** worker:deferred leaves a file alone until it has had no new articles for this long. */ deferredSettleMs: env.get('DEFERRED_SETTLE_MS', 15 * 60 * 1000), /** Files of at most this many parts are downloaded completely (small archives, NZBs, par2 index files). */ diff --git a/database/migrations/1768620765270_garbage_filter.ts b/database/migrations/1768620765270_garbage_filter.ts new file mode 100644 index 0000000..d16de2b --- /dev/null +++ b/database/migrations/1768620765270_garbage_filter.ts @@ -0,0 +1,64 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Dropping garbage uploads early (CONCEPT.md D42): random-named uploads that + * never get a real name or password. + * - upload_signatures: per posting tool (from the Message-ID) and first group, + * how many probed random-named releases got a real name; `dropped` ones + * aren't stored any more (GarbageFilter). + * - signature_probes: a sample of probed random-named releases, what + * upload_signatures is learned from (the outcome is read from the release + * at learning time, so names that come later count too). + * - dropped_releases: releases deleted after their probe, by release_key, so + * seeding and ingest don't add them again. + * - is_real_release_name(): a name someone would search for. is_readable_name + * lets spam through (usenetszene net, VTS_01_1.VOB, Nieuwe map, 100 MiB): + * this also wants a year, episode, resolution, codec or language tag, or + * three words (LOG.md 2026-10-08 (5)). No `?` in its regexes: knex takes it + * for a binding (hence {0,1}). + */ +export default class extends BaseSchema { + async up() { + await this.db.rawQuery(` + CREATE TABLE IF NOT EXISTS upload_signatures ( + tool varchar(32) NOT NULL, + group_name text NOT NULL, + probed integer NOT NULL DEFAULT 0, + named integer NOT NULL DEFAULT 0, + dropped boolean NOT NULL DEFAULT false, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (tool, group_name))`) + await this.db.rawQuery(` + CREATE TABLE IF NOT EXISTS signature_probes ( + release_id bigint PRIMARY KEY, + tool varchar(32) NOT NULL, + group_name text NOT NULL, + probed_at timestamptz NOT NULL DEFAULT now())`) + await this.db.rawQuery(`CREATE INDEX IF NOT EXISTS signature_probes_probed_at_idx ON signature_probes (probed_at)`) + await this.db.rawQuery(` + CREATE TABLE IF NOT EXISTS dropped_releases ( + release_key varchar(40) PRIMARY KEY, + reason varchar(16) NOT NULL, + dropped_at timestamptz NOT NULL DEFAULT now())`) + await this.db.rawQuery(` + CREATE OR REPLACE FUNCTION is_real_release_name(name text) RETURNS boolean IMMUTABLE LANGUAGE sql AS $$ + SELECT name IS NOT NULL AND is_readable_name(name) + AND NOT ( + name ~* '(usenet|video_ts|nieuwe map|new folder|(^|[^a-z0-9])mib([^a-z0-9]|$))' + OR name ~* '^(vts_|sample|cd[0-9])' + OR name ~ '^\\(.*\\)$' + OR name ~ '^[A-Za-z0-9]{8,}$') + AND ( + name ~* '(^|[^a-z0-9])((19|20)[0-9]{2}|(480|576|720|1080|2160)[pi]|dl|nl)([^a-z0-9]|$)' + OR name ~* '(s[0-9]{2}e[0-9]{2}|x26[45]|h\\.{0,1}26[45]|hevc|web[- .]{0,1}(dl|rip)|bluray|remux|german|dutch|multi|flac|mp3|epub|pdf|ac3|dts|xxx)' + OR (SELECT count(*) FROM regexp_matches(name, '[A-Za-z]{3,}', 'g')) >= 3) + $$`) + } + + async down() { + await this.db.rawQuery(`DROP FUNCTION IF EXISTS is_real_release_name(text)`) + await this.db.rawQuery(`DROP TABLE IF EXISTS dropped_releases`) + await this.db.rawQuery(`DROP TABLE IF EXISTS signature_probes`) + await this.db.rawQuery(`DROP TABLE IF EXISTS upload_signatures`) + } +} diff --git a/database/schema.ts b/database/schema.ts index b6060dd..ec5c874 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -151,6 +151,17 @@ export class DeferredRangeSchema extends BaseModel { declare lastArticle: bigint | number } +export class DroppedReleaseSchema extends BaseModel { + static $columns = ['droppedAt', 'reason', 'releaseKey'] as const + $columns = DroppedReleaseSchema.$columns + @column.dateTime() + declare droppedAt: DateTime + @column() + declare reason: string + @column({ isPrimary: true }) + declare releaseKey: string +} + export class FetchRangeSchema extends BaseModel { static $columns = ['backfill', 'createdAt', 'endId', 'error', 'groupId', 'id', 'requeues', 'startId', 'status', 'updatedAt'] as const $columns = FetchRangeSchema.$columns @@ -576,6 +587,19 @@ export class SfvEntrySchema extends BaseModel { declare releaseId: bigint | number } +export class SignatureProbeSchema extends BaseModel { + static $columns = ['groupName', 'probedAt', 'releaseId', 'tool'] as const + $columns = SignatureProbeSchema.$columns + @column() + declare groupName: string + @column.dateTime() + declare probedAt: DateTime + @column({ isPrimary: true }) + declare releaseId: bigint | number + @column() + declare tool: string +} + export class SpotSchema extends BaseModel { static $columns = ['articleNumber', 'category', 'createdAt', 'description', 'detailsAt', 'error', 'groupName', 'image', 'keyId', 'messageId', 'postedAt', 'poster', 'processedAt', 'releaseId', 'size', 'spotterId', 'status', 'subcats', 'tag', 'title', 'verified', 'website'] as const $columns = SpotSchema.$columns @@ -638,6 +662,23 @@ export class StagedSegmentSchema extends BaseModel { declare stagedAt: DateTime } +export class UploadSignatureSchema extends BaseModel { + static $columns = ['dropped', 'groupName', 'named', 'probed', 'tool', 'updatedAt'] as const + $columns = UploadSignatureSchema.$columns + @column() + declare dropped: boolean + @column() + declare groupName: string + @column() + declare named: number + @column() + declare probed: number + @column({ isPrimary: true }) + declare tool: string + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime +} + export class UserSchema extends BaseModel { static $columns = ['createdAt', 'email', 'fullName', 'id', 'password', 'updatedAt'] as const $columns = UserSchema.$columns diff --git a/lib/garbage/index.ts b/lib/garbage/index.ts new file mode 100644 index 0000000..f99590f --- /dev/null +++ b/lib/garbage/index.ts @@ -0,0 +1,52 @@ +import { createHash } from 'node:crypto' + +/** + * Upload signatures (CONCEPT.md D42): which tool posted an article, read from + * its Message-ID, and the group it was posted to first. Together they predict + * whether a random-named upload will ever get a name: ngPost uploads in + * a.b.ath almost never do, nyuu uploads in a.b.friends always do (LOG.md + * 2026-10-08 (5)). + */ +const TOOLS: [string, RegExp][] = [ + // FmFwIaTpGgUiSfYcMwFlMeRo-1790637260478@nyuu: 12 random capital+small letter pairs, a ms timestamp + ['nyuu', /^(?:[A-Z][a-z]){12}-\d{13}@nyuu$/], + ['nyuu-host', /^(?:[A-Z][a-z]){12}-\d{13}@.+$/], + // b94268bc702040f7b811d5318f727e4f@ngPost: a UUID's hex + ['ngPost', /^[0-9a-f]{32}@ngPost$/], + ['yuun', /^[0-9a-f]{32}@[0-9a-f]{32}\.yuun$/], + ['hex32', /^[0-9a-f]{32}@[A-Za-z0-9.-]+$/], + ['JBinUp', /@JBin(?:Up|Down)\.local$/], + // part1of69.lOlWzvSHabJCBTVrs0ts@AdultPoster.local + ['partNofM', /^part\d+of\d+\./i], + // 1790923856877$1729905757859$1@localhost + ['ts-dollar', /^\d{13}\$\d+\$\d+@localhost$/], + // t3Y4mV92YCp0McxDc5@zLZlsw.RYqes: random local part and random host + ['random', /^[A-Za-z0-9]{10,40}@[A-Za-z0-9]{3,12}\.[A-Za-z0-9]{2,6}$/], +] + +/** The posting tool of a Message-ID (without angle brackets), or 'other'. */ +export function uploadTool(messageId: string | null | undefined): string { + if (!messageId) return 'other' + for (const [tool, pattern] of TOOLS) if (pattern.test(messageId)) return tool + return 'other' +} + +/** The group of a signature: the first the upload was posted to, as stored in `groups`. */ +export function signatureGroup(groups: unknown): string { + return Array.isArray(groups) && typeof groups[0] === 'string' ? groups[0] : '' +} + +/** The key of a signature in a Set. */ +export function signatureKey(tool: string, group: string) { + return `${tool}\t${group}` +} + +/** + * Whether an upload is in the share of dropped signatures that is still kept + * (and probed), so their rates keep being measured. Decided per release key, + * so a release is kept or dropped as a whole. + */ +export function explored(key: string, percent: number): boolean { + if (percent <= 0) return false + return createHash('md5').update(key).digest().readUInt32BE(0) % 10_000 < percent * 100 +} diff --git a/start/env.ts b/start/env.ts index ae28b92..4de92f3 100644 --- a/start/env.ts +++ b/start/env.ts @@ -108,6 +108,9 @@ export default await Env.create(new URL('../', import.meta.url), { GC_AWAIT_INSPECTION_SECONDS: Env.schema.number.optional(), GC_MERGE_INTERVAL_MS: Env.schema.number.optional(), INSPECTION_FEED_MAX: Env.schema.number.optional(), + GARBAGE_FILTER: Env.schema.boolean.optional(), + GARBAGE_EXPLORE_PERCENT: Env.schema.number.optional(), + GARBAGE_LEARN_INTERVAL_MS: Env.schema.number.optional(), ARCHIVE_WALK_MAX_SEGMENTS: Env.schema.number.optional(), DEFERRED_SETTLE_MS: Env.schema.number.optional(), DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(), diff --git a/tests/functional/garbage.spec.ts b/tests/functional/garbage.spec.ts new file mode 100644 index 0000000..5ae8fcb --- /dev/null +++ b/tests/functional/garbage.spec.ts @@ -0,0 +1,190 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import indexerConfig from '#config/indexer' +import FileStore from '#services/FileStore' +import SeedService from '#services/SeedService' +import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService' +import ContentService from '#services/ContentService' +import GarbageFilter from '#services/GarbageFilter' +import QueueService from '#services/QueueService' +import { FileAggregator } from '#lib/aggregate/index' +import { toSeedFile } from '#lib/seed/wire' +import { explored, uploadTool } from '#lib/garbage/index' + +const GROUP = 'alt.binaries.test' +let n = 0 + +/** Headers of a file posted by ngPost (or nyuu), all its parts. */ +function upload(name: string, parts: number, tool: 'ngPost' | 'nyuu' = 'ngPost') { + const id = () => { + n++ + return tool === 'ngPost' ? `${n.toString(16).padStart(32, '0')}@ngPost` : `${'FmFwIaTpGgUiSfYcMwFl'}${String.fromCharCode(65 + (n % 26))}e${String.fromCharCode(65 + ((n >> 5) % 26))}o-${String(1790637260000 + n)}@nyuu` + } + return Array.from({ length: parts }, (_, i) => + segmentFromHeader( + { + 'subject': `[1/1] - "${name}" yEnc (${i + 1}/${parts})`, + 'from': 'poster ', + 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': `<${id()}>`, + ':bytes': '700000', + }, + GROUP, + n + )! + ) +} + +async function dropSignature(tool: string, group = GROUP) { + await db.table('upload_signatures').insert({ tool, group_name: group, probed: 500, named: 0, dropped: true }) + GarbageFilter.reset() +} + +/** A random-named release with one stored file, as seeding leaves it. */ +async function storedRelease(release: Record = {}, file: Record = {}) { + const now = new Date() + n++ + const [r] = await db + .table('releases') + .insert({ release_key: `garbage${n}`.padEnd(40, 'g'), name: 'Hs7dKq2LmZ9xWv3RtYp1Qa', source: 'stem', groups: JSON.stringify([GROUP]), first_seen_at: now, last_seen_at: now, ...release }) + .returning(['id', 'release_key']) + const [{ id }] = await FileStore.insert([ + { filename: 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', poster: 'p@ngPost.com', date: now.getTime(), parts: 1, size: 10, release_id: r.id, groups: JSON.stringify([GROUP]), + segments: [{ number: 1, id: `${n.toString(16).padStart(32, 'a')}@ngPost`, size: 10 }], created_at: now, updated_at: now, ...file }, + ]) + return { releaseId: Number(r.id), releaseKey: r.release_key as string, fileId: id } +} + +test.group('Garbage filter (D42)', (group) => { + const explore = indexerConfig.garbage.explorePercent + group.each.setup(() => { + // Nothing explored, so a test's release is dropped whatever its key hashes to. + indexerConfig.garbage.explorePercent = 0 + GarbageFilter.reset() + return testUtils.db().withGlobalTransaction() + }) + group.each.teardown(() => { + indexerConfig.garbage.explorePercent = explore + GarbageFilter.reset() + }) + group.teardown(() => QueueService.closeAll()) + + test('posting tools are told apart by their Message-IDs', ({ assert }) => { + assert.equal(uploadTool('FmFwIaTpGgUiSfYcMwFlMeRo-1790637260478@nyuu'), 'nyuu') + assert.equal(uploadTool('b94268bc702040f7b811d5318f727e4f@ngPost'), 'ngPost') + assert.equal(uploadTool('part1of69.lOlWzvSHabJCBTVrs0ts@AdultPoster.local'), 'partNofM') + assert.equal(uploadTool('DLugnv3ZrVBv8Z9I69nU_19o112@JBinUp.local'), 'JBinUp') + assert.equal(uploadTool('t3Y4mV92YCp0McxDc5@zLZlsw.RYqes'), 'random') + assert.equal(uploadTool('seed-1-12@test'), 'other') + assert.equal(uploadTool(null), 'other') + }) + + test('the explored share is stable per key and about the percentage', ({ assert }) => { + assert.equal(explored('some-release', 2), explored('some-release', 2)) + assert.isFalse(explored('some-release', 0)) + const kept = Array.from({ length: 10_000 }, (_, i) => explored(`release-${i}`, 2)).filter(Boolean).length + assert.isAbove(kept, 150) + assert.isBelow(kept, 250) + }) + + test('seeding skips random-named files of a dropped signature, not readable or other ones', async ({ assert }) => { + await dropSignature('ngPost') + const aggregator = new FileAggregator() + const files = aggregator + .add([...upload('Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 2), ...upload('Some.Show.S01E01.1080p.WEB.part01.rar', 2), ...upload('Pq4ZmXc8VbN2LkJh6GfDs3.part01.rar', 2, 'nyuu')]) + .map(toSeedFile) + assert.lengthOf(files, 3) + const result = await SeedService.store(files) + assert.include(result, { files: 2, dropped: 1 }) + assert.isNull(await db.from('files').where('filename', 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar').first()) + assert.isNull(await db.from('releases').where('name', 'Hs7dKq2LmZ9xWv3RtYp1Qa').first()) + assert.isNotNull(await db.from('files').where('filename', 'Some.Show.S01E01.1080p.WEB.part01.rar').first()) + assert.isNotNull(await db.from('files').where('filename', 'Pq4ZmXc8VbN2LkJh6GfDs3.part01.rar').first()) + }) + + test('ingest skips them too, and a release with a readable file is kept whole', async ({ assert }) => { + await dropSignature('ngPost') + await IngestService.ingest(upload('Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 2)) + assert.isNull(await db.from('binaries').where('name', 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar').first()) + + // Off: everything is stored. + indexerConfig.garbage.enabled = false + try { + await IngestService.ingest(upload('Hs7dKq2LmZ9xWv3RtYp1Qa.part02.rar', 2)) + } finally { + indexerConfig.garbage.enabled = true + } + assert.isNotNull(await db.from('binaries').where('name', 'Hs7dKq2LmZ9xWv3RtYp1Qa.part02.rar').first()) + }) + + test('a release dropped after its probe is not stored again', async ({ assert }) => { + const aggregator = new FileAggregator() + const files = aggregator.add(upload('Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 2)).map(toSeedFile) + await db.table('dropped_releases').insert({ release_key: files[0].release!.key, reason: 'encrypted' }) + assert.include(await SeedService.store(files), { files: 0, dropped: 1 }) + }) + + test('after its probe, an encrypted random-named release without name or password is deleted', async ({ assert }) => { + const garbage = await storedRelease({ encrypted: true }) + assert.equal(await GarbageFilter.afterProbe(garbage.fileId, 'rar5'), 'encrypted') + assert.isNull(await db.from('files').where('id', garbage.fileId).first()) + assert.isNull(await db.from('releases').where('id', garbage.releaseId).first()) + assert.equal((await db.from('dropped_releases').where('release_key', garbage.releaseKey).first()).reason, 'encrypted') + + const container = await storedRelease({ container: 'rclone', encrypted: true }) + assert.equal(await GarbageFilter.afterProbe(container.fileId, 'rclone'), 'rclone') + const missing = await storedRelease() + assert.equal(await GarbageFilter.afterProbe(missing.fileId, 'missing'), 'missing') + }) + + test('kept after the probe: a password, a real name, other probes waiting, nothing unreadable found', async ({ assert }) => { + const password = await storedRelease({ encrypted: true, password: 'secret', password_source: 'spot' }) + assert.isNull(await GarbageFilter.afterProbe(password.fileId, 'rar5')) + const named = await storedRelease({ encrypted: true, name: 'Some.Movie.2024.1080p.WEB-DL.x264-GRP' }) + assert.isNull(await GarbageFilter.afterProbe(named.fileId, 'rar5')) + + const waiting = await storedRelease({ encrypted: true }) + const now = new Date() + await db.table('binaries').insert({ file_key: `waiting${n}`, poster: 'p', subject: 's', name: 'Hs7dKq2LmZ9xWv3RtYp1Qa.par2', total_parts: 1, groups: '[]', + first_seen_at: now, last_seen_at: now, release_id: waiting.releaseId, awaiting_inspection: true }) + assert.isNull(await GarbageFilter.afterProbe(waiting.fileId, 'rar5')) + + const readable = await storedRelease() + assert.isNull(await GarbageFilter.afterProbe(readable.fileId, 'rar5')) + assert.isNotNull(await db.from('releases').where('id', readable.releaseId).first()) + }) + + test('learns dropped signatures from probed releases, and keeps a verdict without enough probes', async ({ assert }) => { + const { minProbes } = indexerConfig.garbage + const now = new Date() + const releases = await db + .table('releases') + .insert(Array.from({ length: minProbes }, (_, i) => ({ release_key: `learn${i}`.padEnd(40, 'l'), name: `Hs7dKq2LmZ9xWv3RtYp${i}`, groups: '[]', first_seen_at: now, last_seen_at: now }))) + .returning('id') + await db.table('signature_probes').insert(releases.map((r) => ({ release_id: r.id, tool: 'ngPost', group_name: GROUP }))) + // Five probes of another signature: too few to decide. + await db.table('signature_probes').insert(releases.slice(0, 5).map((r, i) => ({ release_id: Number(r.id) + 1_000_000_000 + i, tool: 'nyuu', group_name: GROUP }))) + + const learned = await GarbageFilter.learn() + assert.include(learned.changed.join('\n'), `ngPost ${GROUP}: dropped`) + const rows = await db.from('upload_signatures').where('group_name', GROUP).orderBy('tool') + assert.deepEqual(rows.map((r) => [r.tool, r.dropped, r.probed]), [['ngPost', true, minProbes], ['nyuu', false, 5]]) + + // One real name in a hundred is above the 0.5% limit: kept again. + await db.from('releases').where('id', releases[0].id).update({ name: 'Some.Movie.2024.1080p.WEB-DL.x264-GRP' }) + await GarbageFilter.learn() + assert.isFalse((await db.from('upload_signatures').where({ tool: 'ngPost', group_name: GROUP }).first()).dropped) + }) + + test('a stored file of a dropped signature is not probed, and stops waiting for it', async ({ assert }) => { + const stored = await storedRelease() + const now = new Date() + await db.table('binaries').insert({ file_key: `skip${n}`, poster: 'p', subject: 's', name: 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', total_parts: 1, groups: '[]', + first_seen_at: now, last_seen_at: now, release_id: stored.releaseId, file_id: stored.fileId, awaiting_inspection: true }) + assert.isNotNull(await ContentService.prepare(stored.fileId)) + await dropSignature('ngPost') + assert.isNull(await ContentService.prepare(stored.fileId)) + assert.isNull((await db.from('binaries').where('file_id', stored.fileId).first()).awaiting_inspection) + }) +})