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 <noreply@anthropic.com>
This commit is contained in:
1 parent
35d0b2da65
commit
211820067a
13 files changed
+683
-6
No files matched your search
@@ -25,6 +25,7 @@ import SevenZipService from '#services/SevenZipService'
|
|||||||
import SfvService from '#services/SfvService'
|
import SfvService from '#services/SfvService'
|
||||||
import ArchiveWalkService from '#services/ArchiveWalkService'
|
import ArchiveWalkService from '#services/ArchiveWalkService'
|
||||||
import { segmentsOf } from '#services/FileStore'
|
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
|
* 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
|
nicovideo?: string
|
||||||
/** The real name from the yEnc header, for a file posted under a random subject (D36). */
|
/** The real name from the yEnc header, for a file posted under a random subject (D36). */
|
||||||
yencName?: string
|
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<InspectTask | null> {
|
async prepare(fileId: number): Promise<InspectTask | null> {
|
||||||
const file = await db.from('files').where('id', fileId).first()
|
const file = await db.from('files').where('id', fileId).first()
|
||||||
if (!file || !this.isCandidate(file.filename)) return null
|
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) {
|
if ((sevenZipVolume(String(file.real_name ?? file.filename)) ?? 0) > 1) {
|
||||||
const volumes = await SevenZipService.volumes(file)
|
const volumes = await SevenZipService.volumes(file)
|
||||||
if (!volumes || volumes[0].archive) return null
|
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<ContentResult> {
|
async apply(task: InspectTask, probe: InspectProbe): Promise<ContentResult> {
|
||||||
|
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<ContentResult> {
|
||||||
const result: ContentResult = { named: 0, contentType: probe.contentType }
|
const result: ContentResult = { named: 0, contentType: probe.contentType }
|
||||||
const file = await db.from('files').where('id', task.fileId).first()
|
const file = await db.from('files').where('id', task.fileId).first()
|
||||||
if (!file) return result
|
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 === 'nzbe') lines.push(`"${filename}" is an encrypted NZB container.`)
|
||||||
if (content?.contentType === 'rclone') lines.push(`"${filename}" is rclone-encrypted.`)
|
if (content?.contentType === 'rclone') lines.push(`"${filename}" is rclone-encrypted.`)
|
||||||
if (content?.yencName) lines.push(`"${filename}" is "${content.yencName}" (yEnc name).`)
|
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?.nicovideo) lines.push(`"${filename}" → "${content.nicovideo}".`)
|
||||||
if (content?.sfvMoved) lines.push(`SFV "${filename}" moved ${content.sfvMoved} files into its release.`)
|
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.`)
|
if (content?.contentType === 'yencbin') lines.push(`"${filename}" is a yEncBin encrypted container.`)
|
||||||
|
|||||||
@@ -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<string> | null = null
|
||||||
|
private loadedAt = 0
|
||||||
|
|
||||||
|
/** The dropped signatures, cached for CACHE_MS. */
|
||||||
|
async droppedSignatures(): Promise<Set<string>> {
|
||||||
|
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<string>) {
|
||||||
|
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<T>(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<string, 'keep' | 'drop' | 'check'>()
|
||||||
|
const seen = new Set<string>()
|
||||||
|
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<string | null> {
|
||||||
|
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<string | null> {
|
||||||
|
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<number> {
|
||||||
|
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()
|
||||||
@@ -4,6 +4,7 @@ import type { QueryClientContract } from '@adonisjs/lucid/types/database'
|
|||||||
import type { Knex } from 'knex'
|
import type { Knex } from 'knex'
|
||||||
import QueueService from '#services/QueueService'
|
import QueueService from '#services/QueueService'
|
||||||
import NameService from '#services/NameService'
|
import NameService from '#services/NameService'
|
||||||
|
import GarbageFilter from '#services/GarbageFilter'
|
||||||
import type { IngestSegment } from '#lib/ingest/segment'
|
import type { IngestSegment } from '#lib/ingest/segment'
|
||||||
|
|
||||||
// Header parsing is pure and lives in lib, so seeding nodes need no database (CONCEPT.md D37).
|
// 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).
|
* no-op. See CONCEPT.md (D12).
|
||||||
*/
|
*/
|
||||||
class IngestService {
|
class IngestService {
|
||||||
async ingest(segments: IngestSegment[]): Promise<CompletedBinary[]> {
|
async ingest(batch: IngestSegment[]): Promise<CompletedBinary[]> {
|
||||||
|
// 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 []
|
if (segments.length === 0) return []
|
||||||
|
|
||||||
// Short statements, no enclosing transaction: every step is idempotent, and
|
// Short statements, no enclosing transaction: every step is idempotent, and
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import SfvService from '#services/SfvService'
|
|||||||
import ContentService from '#services/ContentService'
|
import ContentService from '#services/ContentService'
|
||||||
import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire'
|
import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire'
|
||||||
import InspectionQueue from '#services/InspectionQueue'
|
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. */
|
/** Articles from a range's edge within which a file may continue in the neighbouring range. */
|
||||||
const EDGE_ARTICLES = 50_000
|
const EDGE_ARTICLES = 50_000
|
||||||
@@ -15,6 +16,8 @@ export interface SeedStoreResult {
|
|||||||
existing: number
|
existing: number
|
||||||
releases: number
|
releases: number
|
||||||
inspections: 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 {
|
class SeedService {
|
||||||
async store(batch: SeedFile[]): Promise<SeedStoreResult> {
|
async store(batch: SeedFile[]): Promise<SeedStoreResult> {
|
||||||
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.
|
// 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
|
if (files.length === 0) return result
|
||||||
const knex = db.connection().getWriteClient()
|
const knex = db.connection().getWriteClient()
|
||||||
const now = new Date()
|
const now = new Date()
|
||||||
|
|||||||
@@ -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.`)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -11,6 +11,8 @@ import StagingService from '#services/StagingService'
|
|||||||
import FilePartitions from '#services/FilePartitions'
|
import FilePartitions from '#services/FilePartitions'
|
||||||
import ReleaseCompleteness from '#services/ReleaseCompleteness'
|
import ReleaseCompleteness from '#services/ReleaseCompleteness'
|
||||||
import FileKeys from '#services/FileKeys'
|
import FileKeys from '#services/FileKeys'
|
||||||
|
import GarbageFilter from '#services/GarbageFilter'
|
||||||
|
import indexerConfig from '#config/indexer'
|
||||||
import Shutdown from '#services/Shutdown'
|
import Shutdown from '#services/Shutdown'
|
||||||
|
|
||||||
const CLEAN_BATCH = 1000
|
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). */
|
/** When release completeness was last recomputed (it runs less often than the sweep). */
|
||||||
private completenessAt = 0
|
private completenessAt = 0
|
||||||
|
/** When the garbage signatures were last learned (D42). */
|
||||||
|
private learnedAt = 0
|
||||||
/** When split uploads were last merged. */
|
/** When split uploads were last merged. */
|
||||||
private mergedAt = 0
|
private mergedAt = 0
|
||||||
|
|
||||||
@@ -59,6 +63,11 @@ export default class GcScheduler extends BaseCommand {
|
|||||||
} catch (err: any) {
|
} catch (err: any) {
|
||||||
this.logger.error(`file_keys fill failed: ${err.message}`)
|
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)
|
if (!this.once) await Shutdown.sleep(gcConfig.intervalMs)
|
||||||
} while (!this.once && !Shutdown.requested)
|
} while (!this.once && !Shutdown.requested)
|
||||||
} finally {
|
} 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.`)
|
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() {
|
private async rebuildIndexes() {
|
||||||
const key = 'gc:reindex:last'
|
const key = 'gc:reindex:last'
|
||||||
const last = Number((await RedisService.client.get(key)) ?? 0)
|
const last = Number((await RedisService.client.get(key)) ?? 0)
|
||||||
|
|||||||
@@ -51,7 +51,7 @@ export default class SeedRange extends BaseCommand {
|
|||||||
const first = Number(this.start)
|
const first = Number(this.start)
|
||||||
if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${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 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
|
let storeMs = 0
|
||||||
const cpuStart = process.cpuUsage()
|
const cpuStart = process.cpuUsage()
|
||||||
const started = Date.now()
|
const started = Date.now()
|
||||||
@@ -90,6 +90,7 @@ export default class SeedRange extends BaseCommand {
|
|||||||
storeMs += Date.now() - t
|
storeMs += Date.now() - t
|
||||||
stats.files += stored.files
|
stats.files += stored.files
|
||||||
stats.existing += stored.existing
|
stats.existing += stored.existing
|
||||||
|
stats.dropped += stored.dropped
|
||||||
stats.inspections += stored.inspections
|
stats.inspections += stored.inspections
|
||||||
}
|
}
|
||||||
this.progress(stats, started)
|
this.progress(stats, started)
|
||||||
@@ -106,7 +107,7 @@ export default class SeedRange extends BaseCommand {
|
|||||||
this.logger.info(
|
this.logger.info(
|
||||||
`${this.group} ${first}-${last}: ${n(stats.headers)} headers in ${seconds.toFixed(0)} s (${n(Math.round(stats.headers / seconds))}/s). ` +
|
`${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.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.`
|
`${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.`)
|
this.logger.info(`Store ${(storeMs / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`)
|
||||||
|
|||||||
@@ -95,6 +95,25 @@ export default {
|
|||||||
maxHddGb: env.get('BACKFILL_MAX_HDD_GB', 150),
|
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. */
|
/** 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),
|
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). */
|
/** Files of at most this many parts are downloaded completely (small archives, NZBs, par2 index files). */
|
||||||
|
|||||||
@@ -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`)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -151,6 +151,17 @@ export class DeferredRangeSchema extends BaseModel {
|
|||||||
declare lastArticle: bigint | number
|
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 {
|
export class FetchRangeSchema extends BaseModel {
|
||||||
static $columns = ['backfill', 'createdAt', 'endId', 'error', 'groupId', 'id', 'requeues', 'startId', 'status', 'updatedAt'] as const
|
static $columns = ['backfill', 'createdAt', 'endId', 'error', 'groupId', 'id', 'requeues', 'startId', 'status', 'updatedAt'] as const
|
||||||
$columns = FetchRangeSchema.$columns
|
$columns = FetchRangeSchema.$columns
|
||||||
@@ -576,6 +587,19 @@ export class SfvEntrySchema extends BaseModel {
|
|||||||
declare releaseId: bigint | number
|
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 {
|
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
|
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
|
$columns = SpotSchema.$columns
|
||||||
@@ -638,6 +662,23 @@ export class StagedSegmentSchema extends BaseModel {
|
|||||||
declare stagedAt: DateTime
|
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 {
|
export class UserSchema extends BaseModel {
|
||||||
static $columns = ['createdAt', 'email', 'fullName', 'id', 'password', 'updatedAt'] as const
|
static $columns = ['createdAt', 'email', 'fullName', 'id', 'password', 'updatedAt'] as const
|
||||||
$columns = UserSchema.$columns
|
$columns = UserSchema.$columns
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -108,6 +108,9 @@ export default await Env.create(new URL('../', import.meta.url), {
|
|||||||
GC_AWAIT_INSPECTION_SECONDS: Env.schema.number.optional(),
|
GC_AWAIT_INSPECTION_SECONDS: Env.schema.number.optional(),
|
||||||
GC_MERGE_INTERVAL_MS: Env.schema.number.optional(),
|
GC_MERGE_INTERVAL_MS: Env.schema.number.optional(),
|
||||||
INSPECTION_FEED_MAX: 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(),
|
ARCHIVE_WALK_MAX_SEGMENTS: Env.schema.number.optional(),
|
||||||
DEFERRED_SETTLE_MS: Env.schema.number.optional(),
|
DEFERRED_SETTLE_MS: Env.schema.number.optional(),
|
||||||
DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(),
|
DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(),
|
||||||
|
|||||||
@@ -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 <poster@example.invalid>',
|
||||||
|
'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<string, unknown> = {}, file: Record<string, unknown> = {}) {
|
||||||
|
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<IngestSegment>()
|
||||||
|
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<IngestSegment>()
|
||||||
|
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)
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user