diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index aa41d8b..0a9a965 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -1,6 +1,6 @@ import db from '@adonisjs/lucid/services/db' import type { TransactionClientContract } from '@adonisjs/lucid/types/database' -import type { EdgeBatch, ProbeCandidate, ReleaseChunk } from '#lib/edge/index' +import { UPLOAD_GAP_MS, type EdgeBatch, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' import ChunkPartitions from '#services/chunk_partitions' export class StaleBatch extends Error {} @@ -95,7 +95,10 @@ class BatchStore { /** * Release aggregates in the catalog, one write-once chunk per release, and * probe candidates (a batch's, or files the fragment merger completed). - * Releases are sorted by key, so concurrent writers lock them in the same order. + * A chunk joins the release of its key whose articles the range's group + * owns and whose posting window is within UPLOAD_GAP_MS of its own, else + * starts one: re-uploads are releases of their own (decision d). Keys are + * locked in order, so concurrent writers don't deadlock or both start one. */ async writeReleases( trx: TransactionClientContract, @@ -104,64 +107,97 @@ class BatchStore { releases: ReleaseChunk[], candidates: ProbeCandidate[] ) { - const out = { newReleases: 0, candidates: 0, ids: new Map() } - const ids = out.ids - // Releases, sorted by key so concurrent batches lock them in the same order. - const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key)) + const out = { newReleases: 0, candidates: 0 } + const ids = new Map() + const slot = (key: string, upload: number) => `${key}\n${upload}` + const { group_id: groupId } = await trx.from('ranges').where('id', rangeId).select('group_id').firstOrFail() + const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key) || a.upload - b.upload) for (let i = 0; i < sorted.length; i += 1000) { - const chunk = sorted.slice(i, i + 1000) - const { rows } = await trx.rawQuery( - `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2, complete) - SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2, - CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END - FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) - ON CONFLICT (release_key) DO UPDATE SET - files = releases.files + excluded.files, - segments = releases.segments + excluded.segments, - bytes = releases.bytes + excluded.bytes, - chunks = releases.chunks + 1, - posted_at = least(releases.posted_at, excluded.posted_at), - file_total = coalesce(releases.file_total, excluded.file_total), - first_article = least(releases.first_article, excluded.first_article), - last_article = greatest(releases.last_article, excluded.last_article), - readable = releases.readable OR excluded.readable, - has_par2 = releases.has_par2 OR excluded.has_par2, - -- As it stands (the counter's files all found); settling makes it final. - complete = CASE WHEN coalesce(releases.file_total, excluded.file_total) IS NULL THEN NULL - ELSE releases.files + excluded.files >= coalesce(releases.file_total, excluded.file_total) END, - last_seen_at = now() - RETURNING id, release_key, (xmax = 0) AS inserted`, - [ - JSON.stringify( - chunk.map((r) => ({ - key: r.key, - name: r.name, - source: r.source, - poster: r.poster, - groups: r.groups, - posted_at: r.postedAt, - file_total: r.fileTotal, - files: r.files, - segments: r.segments, - bytes: r.bytes, - first_article: r.firstArticle, - last_article: r.lastArticle, - readable: r.readable, - has_par2: r.hasPar2, - })) - ), - ] + const part = sorted.slice(i, i + 1000) + const keys = [...new Set(part.map((r) => r.key))] + await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended(k, 0)) FROM unnest(?::text[]) AS k ORDER BY k`, [keys]) + const windows = part.map((r, idx) => ({ idx, key: r.key, first: r.postedAt, last: r.postedLast ?? r.postedAt })) + const { rows: matches } = await trx.rawQuery( + `SELECT DISTINCT ON (v.idx) v.idx, r.id + FROM jsonb_to_recordset(?::jsonb) AS v(idx int, key text, first bigint, last bigint) + JOIN releases r ON r.identity_key = v.key AND r.group_id = ? + WHERE v.first IS NULL OR r.posted_at IS NULL + OR (r.posted_at <= to_timestamp((v.last + ?) / 1000.0) AND coalesce(r.last_posted_at, r.posted_at) >= to_timestamp((v.first - ?) / 1000.0)) + ORDER BY v.idx, r.last_posted_at DESC NULLS LAST, r.id DESC`, + [JSON.stringify(windows), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS] ) - for (const row of rows) { - ids.set(row.release_key, Number(row.id)) - if (row.inserted) out.newReleases++ + const joined = new Map(matches.map((m: { idx: number; id: string }) => [m.idx, Number(m.id)])) + const row = (r: ReleaseChunk) => ({ + name: r.name, + source: r.source, + poster: r.poster, + groups: r.groups, + first: r.postedAt, + last: r.postedLast ?? r.postedAt, + file_total: r.fileTotal, + files: r.files, + segments: r.segments, + bytes: r.bytes, + first_article: r.firstArticle, + last_article: r.lastArticle, + readable: r.readable, + has_par2: r.hasPar2, + }) + const updates = part.flatMap((r, idx) => (joined.has(idx) ? [{ id: joined.get(idx)!, ...row(r) }] : [])) + if (updates.length > 0) { + await trx.rawQuery( + `WITH v AS (SELECT * FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)), + a AS (SELECT id, sum(files) AS files, sum(segments) AS segments, sum(bytes) AS bytes, count(*) AS chunks, min(first) AS first, max(last) AS last, + max(file_total) AS file_total, min(first_article) AS first_article, max(last_article) AS last_article, bool_or(readable) AS readable, bool_or(has_par2) AS has_par2 + FROM v GROUP BY id) + UPDATE releases r SET + files = r.files + a.files, + segments = r.segments + a.segments, + bytes = r.bytes + a.bytes, + chunks = r.chunks + a.chunks, + posted_at = least(r.posted_at, to_timestamp(a.first / 1000.0)), + last_posted_at = greatest(r.last_posted_at, to_timestamp(a.last / 1000.0)), + file_total = coalesce(r.file_total, a.file_total), + first_article = least(r.first_article, a.first_article), + last_article = greatest(r.last_article, a.last_article), + readable = r.readable OR a.readable, + has_par2 = r.has_par2 OR a.has_par2, + -- As it stands (the counter's files all found); settling makes it final. + complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN NULL ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END, + last_seen_at = now() + FROM a WHERE r.id = a.id`, + [JSON.stringify(updates)] + ) + part.forEach((r, idx) => joined.has(idx) && ids.set(slot(r.key, r.upload), joined.get(idx)!)) + } + const fresh = part.filter((_, idx) => !joined.has(idx)) + if (fresh.length > 0) { + // The first upload of a key keeps the key; later ones are key:2, key:3, … + const { rows: counts } = await trx.rawQuery(`SELECT identity_key, count(*) AS n FROM releases WHERE identity_key = ANY(?) GROUP BY 1`, [[...new Set(fresh.map((r) => r.key))]]) + const taken = new Map(counts.map((c: { identity_key: string; n: string }) => [c.identity_key, Number(c.n)])) + const inserts = fresh.map((r) => { + const n = (taken.get(r.key) ?? 0) + 1 + taken.set(r.key, n) + return { release_key: n === 1 ? r.key : `${r.key}:${n}`, key: r.key, upload: r.upload, ...row(r) } + }) + const { rows } = await trx.rawQuery( + `INSERT INTO releases (release_key, identity_key, group_id, name, source, poster, groups, posted_at, last_posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2, complete) + SELECT v.release_key, v.key, ?, v.name, v.source, v.poster, v.groups, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, + v.first_article, v.last_article, v.readable, v.has_par2, CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END + FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, key text, name text, source text, poster text, groups jsonb, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) + RETURNING id, release_key`, + [groupId, JSON.stringify(inserts)] + ) + const byReleaseKey = new Map(rows.map((x: { id: string; release_key: string }) => [x.release_key, Number(x.id)])) + for (const r of inserts) ids.set(slot(r.key, r.upload), byReleaseKey.get(r.release_key)!) + out.newReleases += rows.length } } // One write-once chunk per release. for (let i = 0; i < sorted.length; i += 500) { const rows = sorted.slice(i, i + 500).map((r) => ({ - release_id: ids.get(r.key)!, + release_id: ids.get(slot(r.key, r.upload))!, range_id: rangeId, batch_from: batchFrom, files: r.files, @@ -171,12 +207,12 @@ class BatchStore { if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows) } - const kept = candidates.filter((c) => ids.has(c.releaseKey)) + const kept = candidates.filter((c) => ids.has(slot(c.releaseKey, c.upload ?? 0))) for (let i = 0; i < kept.length; i += 2000) { const rows = kept .slice(i, i + 2000) .map((c) => ({ - release_id: ids.get(c.releaseKey)!, + release_id: ids.get(slot(c.releaseKey, c.upload ?? 0))!, kind: c.kind, message_id: c.messageId, file: c.file, diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts index e4596c9..a803792 100644 --- a/app/services/fragment_merger.ts +++ b/app/services/fragment_merger.ts @@ -3,7 +3,7 @@ import BatchStore from '#services/batch_store' import ChunkPartitions from '#services/chunk_partitions' import RuleSets from '#services/rule_sets' import { RuleSet } from '#lib/rules/index' -import { candidatesOf, chunkOf, mergeFragments, releaseKeyOf, type Fragment, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' +import { candidatesOf, chunkOf, mergeFragments, releaseKeyOf, uploadsOf, type Fragment, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' import type { AggregatedFile } from '#lib/files/index' import type { EdgeSegment } from '#lib/edge/index' @@ -85,10 +85,12 @@ class FragmentMerger { for (const [key, { files, rangeId }] of byRelease) { const verdict = rules.judge(key, files.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId }))) if (verdict.rule !== null) continue - releases.push({ rangeId, chunk: chunkOf(key, files), candidates: candidatesOf(key, files) }) + uploadsOf(files).forEach((list, upload) => releases.push({ rangeId, chunk: chunkOf(key, list, upload), candidates: candidatesOf(key, list, upload) })) } await ChunkPartitions.ensure() await db.transaction(async (trx) => { + // In key order, as batches lock keys (no deadlocks). + releases.sort((a, b) => a.chunk.key.localeCompare(b.chunk.key) || a.chunk.upload - b.chunk.upload) for (const { rangeId, chunk, candidates } of releases) await BatchStore.writeReleases(trx, rangeId, 0, [chunk], candidates) const done = [...completed.flatMap((c) => c.ids), ...incomplete.flatMap((c) => c.ids)] for (let i = 0; i < done.length; i += 5000) await trx.from('fragments').whereIn('id', done.slice(i, i + 5000)).delete() diff --git a/commands/releases_duplicates.ts b/commands/releases_duplicates.ts new file mode 100644 index 0000000..5fd028f --- /dev/null +++ b/commands/releases_duplicates.ts @@ -0,0 +1,82 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import { decodeChunk } from '#lib/chunks/index' + +/** + * How often a release holds a file name more than once (decoded from its + * chunks): copies of one upload posted again (multi-posting, reposts that + * share a key) or different files that share a name. Copies of the same + * size are counted as the same file posted again. + */ +export default class ReleasesDuplicates extends BaseCommand { + static commandName = 'releases:duplicates' + static description = 'Counts file names a release holds more than once, in a sample' + static options: CommandOptions = { startApp: true } + + @flags.number({ description: 'Releases sampled (with more than one file)', default: 20000 }) + declare sample: number + + async run() { + const ids: string[] = ( + await db.rawQuery(`SELECT id FROM releases TABLESAMPLE SYSTEM (5) WHERE files > 1 LIMIT ?`, [this.sample]) + ).rows.map((r: { id: string }) => r.id) + const kind = (name: string) => + /\.(jpe?g|png|gif|bmp|webp)$/i.test(name) ? 'image' + : /\.(mp3|flac|m4a|ogg|wav)\b/i.test(name) ? 'audio' + : /\.(par2)$/i.test(name) ? 'par2' + : /\.(rar|r\d\d|7z|zip|\d{3})$/i.test(name) ? 'archive' + : /\.(pdf|epub|mobi|cbr|cbz|djvu|chm)$/i.test(name) ? 'document' + : /\.(nfo|sfv|nzb|txt|diz)$/i.test(name) ? 'sidecar' + : 'other' + let releases = 0 + let withDuplicates = 0 + let files = 0 + let extra = 0 + let completeOnlyByCopies = 0 + const byKind = new Map() + const how = { sameFile: 0, differentFile: 0, sameDay: 0, laterRepost: 0 } + const examples: string[] = [] + for (const id of ids) { + const release = await db.from('releases').where('id', id).select('name', 'file_total', 'files').first() + const chunks = await db.from('release_chunks').where('release_id', id).select('payload') + const byName = new Map>() + for (const c of chunks) { + for (const f of decodeChunk(c.payload)) { + const copies = byName.get(f.name) ?? new Map() + copies.set(f.segments[0]?.[1] ?? '', { parts: f.segments.length, bytes: f.segments.reduce((n, [, , b]) => n + b, 0), at: f.postedAt }) + byName.set(f.name, copies) + } + } + releases++ + let dup = false + for (const [name, copies] of byName) { + files += copies.size + if (copies.size < 2) continue + dup = true + extra += copies.size - 1 + byKind.set(kind(name), (byKind.get(kind(name)) ?? 0) + copies.size - 1) + const list = [...copies.values()] + // The same file again: as many parts, within 3% in bytes (article sizes include headers). + const same = list.every((c) => c.parts === list[0].parts && Math.abs(c.bytes - list[0].bytes) <= 0.03 * list[0].bytes) + if (same) how.sameFile++ + else how.differentFile++ + const times = list.map((c) => c.at).filter((t): t is number => t !== null) + if (times.length > 1 && Math.max(...times) - Math.min(...times) > 86_400_000) how.laterRepost++ + else how.sameDay++ + if (!same && examples.length < 8) examples.push(`${id} ${release.name.slice(0, 60)} | ${name.slice(0, 60)} | ${list.map((c) => `${c.parts}p ${(c.bytes / 1e6).toFixed(1)}MB`).join(', ')}`) + } + if (!dup) continue + withDuplicates++ + const total = Number(release.file_total ?? 0) + if (total && Number(release.files) >= total && byName.size < total) completeOnlyByCopies++ + } + const pct = (n: number, of: number) => `${((100 * n) / Math.max(1, of)).toFixed(2)}%` + this.logger.info(`${releases} releases with 2+ files sampled`) + this.logger.info(` with a file name more than once: ${withDuplicates} (${pct(withDuplicates, releases)})`) + this.logger.info(` extra copies: ${extra} of ${files} files (${pct(extra, files)}); by type: ${[...byKind].sort((a, b) => b[1] - a[1]).map(([k, n]) => `${k} ${n}`).join(', ')}`) + this.logger.info(` duplicated names: same file again ${how.sameFile}, different (parts or size) ${how.differentFile}; copies the same day ${how.sameDay}, a day or more apart ${how.laterRepost}`) + this.logger.info(` 'complete' only by counting copies: ${completeOnlyByCopies}`) + for (const e of examples) this.logger.info(` different: ${e}`) + } +} diff --git a/commands/releases_upload_gaps.ts b/commands/releases_upload_gaps.ts new file mode 100644 index 0000000..e41a896 --- /dev/null +++ b/commands/releases_upload_gaps.ts @@ -0,0 +1,58 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import { decodeChunk } from '#lib/chunks/index' + +/** + * For telling uploads apart (re-uploads under one release key): how long + * one upload takes to post (the span of a release's files' dates, in + * releases without a file name twice), and how far apart the copies of a + * file name are in releases that have one twice. + */ +export default class ReleasesUploadGaps extends BaseCommand { + static commandName = 'releases:upload-gaps' + static description = 'Posting spans of uploads and gaps between re-uploads, in a sample' + static options: CommandOptions = { startApp: true } + + @flags.number({ description: 'Releases sampled (with more than one file)', default: 20000 }) + declare sample: number + + async run() { + const ids: string[] = (await db.rawQuery(`SELECT id FROM releases TABLESAMPLE SYSTEM (5) WHERE files > 1 LIMIT ?`, [this.sample])).rows.map((r: { id: string }) => r.id) + const bins = [10, 60, 180, 360, 720, 1440, 4320, Number.POSITIVE_INFINITY] + const label = (m: number) => (m === Number.POSITIVE_INFINITY ? '>3d' : m >= 1440 ? `≤${m / 1440}d` : m >= 60 ? `≤${m / 60}h` : `≤${m}min`) + const spans = new Array(bins.length).fill(0) + const gaps = new Array(bins.length).fill(0) + const bin = (minutes: number) => bins.findIndex((b) => minutes <= b) + for (const id of ids) { + const chunks = await db.from('release_chunks').where('release_id', id).select('payload') + const byName = new Map() + const times: number[] = [] + for (const c of chunks) { + for (const f of decodeChunk(c.payload)) { + if (f.postedAt === null) continue + times.push(f.postedAt) + byName.set(f.name, [...(byName.get(f.name) ?? []), f.postedAt]) + } + } + if (times.length < 2) continue + const twice = [...byName.values()].filter((t) => t.length > 1) + if (twice.length === 0) { + spans[bin((Math.max(...times) - Math.min(...times)) / 60_000)]++ + continue + } + // The gap between a name's copies (sorted; each step), so a release reposted 3 times counts 2 gaps per name. + for (const t of twice) { + t.sort((a, b) => a - b) + for (let i = 1; i < t.length; i++) gaps[bin((t[i] - t[i - 1]) / 60_000)]++ + } + } + const show = (counts: number[]) => { + const total = counts.reduce((a, b) => a + b, 0) + let cumulative = 0 + return bins.map((b, i) => `${label(b)} ${((100 * (cumulative += counts[i])) / Math.max(1, total)).toFixed(1)}%`).join(', ') + ` (n=${total})` + } + this.logger.info(`one upload's posting span (cumulative): ${show(spans)}`) + this.logger.info(`gap between copies of a name (cumulative): ${show(gaps)}`) + } +} diff --git a/database/migrations/1770000001400_release_uploads.ts b/database/migrations/1770000001400_release_uploads.ts new file mode 100644 index 0000000..f869de8 --- /dev/null +++ b/database/migrations/1770000001400_release_uploads.ts @@ -0,0 +1,26 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Re-uploads as releases of their own (REBUILD.md, decision d): releases + * keep the identity key their files gave (identity_key; release_key is that + * for the first upload, then key:2, key:3, …), the group that owns their + * articles, and their posting window (posted_at … last_posted_at). A batch's + * chunk joins a release with its key and group whose window is within + * UPLOAD_GAP_MS, else starts one. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(`ALTER TABLE releases ADD COLUMN IF NOT EXISTS identity_key varchar(40)`) + this.schema.raw(`ALTER TABLE releases ADD COLUMN IF NOT EXISTS group_id integer`) + this.schema.raw(`ALTER TABLE releases ADD COLUMN IF NOT EXISTS last_posted_at timestamptz`) + this.schema.raw(`UPDATE releases SET identity_key = release_key, last_posted_at = posted_at WHERE identity_key IS NULL`) + this.schema.raw(`ALTER TABLE releases ALTER COLUMN identity_key SET NOT NULL`) + this.schema.raw(`ALTER TABLE releases ALTER COLUMN release_key TYPE varchar(48)`) + this.schema.raw(`CREATE INDEX IF NOT EXISTS releases_identity_idx ON releases (identity_key, group_id)`) + } + + async down() { + this.schema.raw(`DROP INDEX IF EXISTS releases_identity_idx`) + this.schema.raw(`ALTER TABLE releases DROP COLUMN IF EXISTS identity_key, DROP COLUMN IF EXISTS group_id, DROP COLUMN IF EXISTS last_posted_at`) + } +} diff --git a/lib/edge/index.ts b/lib/edge/index.ts index b08668d..a308a76 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -1,4 +1,4 @@ -export { RangeIndexer, chunkOf, candidatesOf, releaseKeyOf, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' +export { RangeIndexer, chunkOf, candidatesOf, releaseKeyOf, uploadsOf, UPLOAD_GAP_MS, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js' export { toWire, fromWire, type WireBatch } from './wire.js' export type { Lease } from './lease.js' diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index 2a23a16..56cfe8a 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -8,12 +8,16 @@ import { encodeChunk, type ChunkFile } from '../chunks/index.js' /** A release's files from one batch (REBUILD.md, chunk store): what the catalog adds up, and the encoded chunk. */ export interface ReleaseChunk { key: string + /** Which upload of the key in this batch (0, 1, …; uploadsOf): the catalog keeps uploads apart. */ + upload: number name: string source: string fileTotal: number | null poster: string groups: string[] postedAt: number | null + /** The latest article's date: with postedAt, the upload's posting window. */ + postedLast: number | null files: number segments: number bytes: number @@ -35,6 +39,8 @@ export type CandidateKind = 'archive' | 'par2' | 'lone' | '7z-tail' /** An article phase 2 should read (REBUILD.md, probe queue). */ export interface ProbeCandidate { releaseKey: string + /** Its chunk's upload (ReleaseChunk.upload). */ + upload: number kind: CandidateKind messageId: string file: string @@ -106,7 +112,7 @@ const MAX_CANDIDATES = 4 * waiting for parts. */ export class RangeIndexer { - private aggregator = new FileAggregator() + private aggregator = new FileAggregator(UPLOAD_GAP_MS) private scanner = new SegmentScanner() private segments: EdgeSegment[] = [] private rules: RuleSet @@ -194,22 +200,24 @@ export class RangeIndexer { } } - /** Judges and encodes the releases completed in this batch. */ + /** Judges and encodes the releases completed in this batch: each upload of a key on its own. */ private judgeReleases() { const byRelease = this.completed this.completed = new Map() - for (const [key, list] of byRelease) { - const verdict = this.rules.judge( - key, - list.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId })) - ) - if (verdict.rule === null) { - this.batch.releases.push(chunkOf(key, list)) - this.batch.candidates.push(...candidatesOf(key, list)) - continue - } - this.count(verdict.rule, verdict.key, list.reduce((n, f) => n + f.segments.length, 0), list.length) - if (verdict.sample) this.batch.samples.push({ ...chunkOf(key, list), rule: verdict.rule, ruleKey: verdict.key }) + for (const [key, all] of byRelease) { + uploadsOf(all).forEach((list, upload) => { + const verdict = this.rules.judge( + key, + list.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId })) + ) + if (verdict.rule === null) { + this.batch.releases.push(chunkOf(key, list, upload)) + this.batch.candidates.push(...candidatesOf(key, list, upload)) + return + } + this.count(verdict.rule, verdict.key, list.reduce((n, f) => n + f.segments.length, 0), list.length) + if (verdict.sample) this.batch.samples.push({ ...chunkOf(key, list, upload), rule: verdict.rule, ruleKey: verdict.key }) + }) } } @@ -284,6 +292,51 @@ export function releaseKeyOf(s: EdgeSegment): string { return s.file.release?.key ?? hash('sha1', `file\n${s.fileKey}`) } +/** + * A pause of more than this in posting starts another upload of the same + * release key (a re-upload): 91% of uploads post within 3 h, 73% of + * re-uploaded copies come later (LOG.md 2026-10-09 (10)). + */ +export const UPLOAD_GAP_MS = 3 * 60 * 60 * 1000 + +/** + * A key's files split into uploads where posting pauses for more than + * UPLOAD_GAP_MS: files in order of their first article's date; a file + * starts another upload when it begins more than UPLOAD_GAP_MS after the + * latest article posted so far (so a file that takes hours to post doesn't + * split its upload). Earliest upload first, each in the files' own order; + * files without a date go with the first. + */ +export function uploadsOf>(files: F[]): F[][] { + if (files.length < 2) return [files] + const first = files.map((f) => postedOf(f.segments)) + const last = files.map((f) => latestOf(f.segments)) + const dated = files.map((_, i) => i).filter((i) => first[i] !== null).sort((a, b) => first[a]! - first[b]! || a - b) + if (dated.length < 2) return [files] + const upload = new Array(files.length).fill(0) + let current = 0 + let end = last[dated[0]]! + for (let k = 1; k < dated.length; k++) { + const i = dated[k] + if (first[i]! - end > UPLOAD_GAP_MS) { + current++ + end = last[i]! + } else if (last[i]! > end) { + end = last[i]! + } + upload[i] = current + } + if (current === 0) return [files] + return Array.from({ length: current + 1 }, (_, u) => files.filter((_, i) => upload[i] === u)) +} + +/** The latest article date of a file's segments, or null. */ +function latestOf(segments: EdgeSegment[]): number | null { + let max: number | null = null + for (const s of segments) if (Number.isFinite(s.postedAt) && (max === null || s.postedAt > max)) max = s.postedAt + return max +} + function postedOf(segments: EdgeSegment[]): number | null { let min: number | null = null for (const s of segments) { @@ -293,7 +346,7 @@ function postedOf(segments: EdgeSegment[]): number | null { return min } -export function chunkOf(key: string, files: AggregatedFile[]): ReleaseChunk { +export function chunkOf(key: string, files: AggregatedFile[], upload = 0): ReleaseChunk { const first = files[0].segments[0].file const chunkFiles: ChunkFile[] = files.map((f) => { const m = f.segments[0].file @@ -304,6 +357,7 @@ export function chunkOf(key: string, files: AggregatedFile[]): Rele let firstArticle = Number.POSITIVE_INFINITY let lastArticle = 0 let postedAt: number | null = null + let postedLast: number | null = null for (const f of files) { for (const s of f.segments) { segments++ @@ -311,18 +365,21 @@ export function chunkOf(key: string, files: AggregatedFile[]): Rele const n = s.articleNumber if (n < firstArticle) firstArticle = n if (n > lastArticle) lastArticle = n + if (Number.isFinite(s.postedAt) && (postedLast === null || s.postedAt > postedLast)) postedLast = s.postedAt } const t = postedOf(f.segments) if (t !== null && (postedAt === null || t < postedAt)) postedAt = t } return { key, + upload, name: first.release?.name ?? first.name, source: first.release?.source ?? 'file', fileTotal: files.find((f) => f.segments[0].file.release?.fileTotal)?.segments[0].file.release?.fileTotal ?? null, poster: first.poster, groups: first.groups, postedAt, + postedLast, files: files.length, segments, bytes, @@ -339,7 +396,7 @@ export function chunkOf(key: string, files: AggregatedFile[]): Rele * volume, of its par2 index, of a lone random-named file, and a 7z set's * last volume (its listing is in the tail). At most a few per release. */ -export function candidatesOf(key: string, files: AggregatedFile[]): ProbeCandidate[] { +export function candidatesOf(key: string, files: AggregatedFile[], upload = 0): ProbeCandidate[] { const out: ProbeCandidate[] = [] let tail: { volume: number; file: AggregatedFile } | null = null for (const f of files) { @@ -349,12 +406,12 @@ export function candidatesOf(key: string, files: AggregatedFile[]): if (!tail || volume > tail.volume) tail = { volume, file: f } continue } - if (isArchiveStart(name)) out.push({ releaseKey: key, kind: 'archive', messageId: f.segments[0].messageId, file: name }) - else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, kind: 'par2', messageId: f.segments[0].messageId, file: name }) + if (isArchiveStart(name)) out.push({ releaseKey: key, upload, kind: 'archive', messageId: f.segments[0].messageId, file: name }) + else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, upload, kind: 'par2', messageId: f.segments[0].messageId, file: name }) } - if (tail) out.push({ releaseKey: key, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name }) + if (tail) out.push({ releaseKey: key, upload, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name }) if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].file.name)) { - out.push({ releaseKey: key, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name }) + out.push({ releaseKey: key, upload, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name }) } return out.slice(0, MAX_CANDIDATES) } diff --git a/lib/files/aggregator.ts b/lib/files/aggregator.ts index e25b56b..83cb034 100644 --- a/lib/files/aggregator.ts +++ b/lib/files/aggregator.ts @@ -14,11 +14,13 @@ export interface AggregatableSegment { totalParts: number messageId: string bytes: number + /** ms since 1970 (NaN or absent: unknown): tells a re-upload's copy of a file from the first. */ + postedAt?: number } export interface AggregatedFile { fileKey: string - /** One segment per part number, in part order; the first article seen wins (a later one is a repost). */ + /** One segment per part number, in part order; within a copy the first article seen wins (a re-send). */ segments: S[] } @@ -27,14 +29,32 @@ interface OpenFile { totalParts: number /** Batch in which a segment last arrived. */ touched: number + /** Posting times seen (NaN: none). */ + earliest: number + latest: number } export class FileAggregator { - private open = new Map>() + /** Per file key its open copies, oldest first: a re-upload of a file is a copy of its own. */ + private open = new Map[]>() private batch = 0 /** Segments with a part number outside 1..total: never part of a complete file. */ private stray: S[] = [] + /** + * `uploadGapMs`: a part already in every open copy, posted further than + * this from all of them, starts another copy (a re-upload). A missing part + * always fills a copy, the nearest in time (a poster re-sending what was + * missing), and a part posted near a copy that has it is a re-send. + */ + constructor(private uploadGapMs = 3 * 60 * 60 * 1000) {} + + /** How far `t` is from a copy's posting times (0 when unknown or within them). */ + private distance(file: OpenFile, t: number | undefined): number { + if (t === undefined || !Number.isFinite(t) || !Number.isFinite(file.earliest)) return 0 + return Math.max(0, file.earliest - t, t - file.latest) + } + /** Adds one batch of segments; returns the files they completed. */ add(segments: S[]): AggregatedFile[] { this.batch++ @@ -44,22 +64,51 @@ export class FileAggregator { this.stray.push(segment) continue } - let file = this.open.get(segment.fileKey) + let copies = this.open.get(segment.fileKey) + if (!copies) { + copies = [] + this.open.set(segment.fileKey, copies) + } + const t = segment.postedAt + let file: OpenFile | undefined + let best = Number.POSITIVE_INFINITY + for (const copy of copies) { + if (copy.parts.has(segment.number)) continue + const d = this.distance(copy, t) + if (d < best) { + best = d + file = copy + } + } if (!file) { - file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch } - this.open.set(segment.fileKey, file) + const near = copies.find((copy) => this.distance(copy, t) <= this.uploadGapMs) + if (near) { + // A re-send of a part this copy has: the first article stays. + near.touched = this.batch + touched.add(segment.fileKey) + continue + } + file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch, earliest: Number.NaN, latest: Number.NaN } + copies.push(file) + } + file.parts.set(segment.number, segment) + if (t !== undefined && Number.isFinite(t)) { + if (!(t >= file.earliest)) file.earliest = t + if (!(t <= file.latest)) file.latest = t } - if (!file.parts.has(segment.number)) file.parts.set(segment.number, segment) file.touched = this.batch touched.add(segment.fileKey) } const complete: AggregatedFile[] = [] for (const key of touched) { - const file = this.open.get(key)! - if (file.parts.size < file.totalParts) continue - this.open.delete(key) - complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) }) + const copies = this.open.get(key)! + for (const file of [...copies]) { + if (file.parts.size < file.totalParts) continue + copies.splice(copies.indexOf(file), 1) + complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) }) + } + if (copies.length === 0) this.open.delete(key) } return complete } @@ -71,16 +120,21 @@ export class FileAggregator { evict(idleBatches = 0): S[] { const out = this.stray this.stray = [] - for (const [key, file] of this.open) { - if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue - this.open.delete(key) - out.push(...file.parts.values()) + for (const [key, copies] of this.open) { + for (const file of [...copies]) { + if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue + copies.splice(copies.indexOf(file), 1) + out.push(...file.parts.values()) + } + if (copies.length === 0) this.open.delete(key) } return out } /** Files still waiting for parts. */ get openFiles(): number { - return this.open.size + let n = 0 + for (const copies of this.open.values()) n += copies.length + return n } } diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index 797b711..c36abed 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -32,6 +32,32 @@ function indexer() { return new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } }, edgeArticles: 1000 }) } +/** One file's articles, posted `minutes` apart from `start` (an ISO time), in `group`. */ +function upload(at: number, name: string, parts: number, counter: string, start: string, group = GROUP, minutes = 1): Overview[] { + return Array.from({ length: parts }, (_, i) => { + id++ + const date = new Date(Date.parse(start) + i * minutes * 60_000).toUTCString().replace('GMT', '+0000') + return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p ', 'date': date, 'message-id': ``, ':bytes': '700000', 'xref': `news.example ${group}:${at + i}` }] as Overview + }) +} + +async function rangeIn(groupName: string, start: number, nodeId?: number) { + const g = (await db.from('groups').where('name', groupName).first()) ?? (await db.table('groups').insert({ name: groupName, priority: groupName === GROUP ? 1 : 2 }).returning('*'))[0] + const node = nodeId ?? Number((await db.table('nodes').insert({ name: `n${groupName}${start}` }).returning('id'))[0].id) + const [range] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 99_999, next_id: start, status: 'leased', node_id: node, rule_version: 1 }).returning('id') + return { rangeId: Number(range.id), nodeId: node } +} + +async function storeIn(groupName: string, start: number, overviews: Overview[]) { + const { rangeId, nodeId } = await rangeIn(groupName, start) + const ix = new RangeIndexer({ group: groupName, start, end: start + 99_999, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1, 'alt.binaries.other': 2 } }, edgeArticles: 1 }) + ix.add(overviews, start + 100_000) + ix.finish() + await BatchStore.store(rangeId, nodeId, ix.take()) +} + +const releasesOf = () => db.from('releases').orderBy('id').select('release_key', 'identity_key', 'files', 'posted_at', 'last_posted_at') + test.group('BatchStore', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) @@ -73,4 +99,32 @@ test.group('BatchStore', (group) => { ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000) await assert.rejects(() => BatchStore.store(rangeId, nodeId + 1, ix.take()), /not leased to this node/) }) + test('a re-upload is a release of its own: in one range, or a later range; a continuation joins', async ({ assert }) => { + // One range: the same file posted Monday and again Thursday. + await storeIn(GROUP, 1, [ + ...upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-05T10:00:00Z'), + ...upload(20, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-05T10:05:00Z'), + ...upload(500, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-08T10:00:00Z'), + ]) + let rows = await releasesOf() + assert.lengthOf(rows, 2) + assert.equal(rows[1].release_key, `${rows[0].release_key}:2`) + assert.deepEqual(rows.map((r) => [r.identity_key, r.files]), [[rows[0].release_key, 2], [rows[0].release_key, 1]]) + + // The next range: Thursday's par2 an hour later joins Thursday's release; a Saturday copy is a third. + await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-08T11:00:00Z')) + await storeIn(GROUP, 200_001, upload(200_010, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-10T10:00:00Z')) + rows = await releasesOf() + assert.deepEqual(rows.map((r) => [r.release_key.split(':')[1] ?? '1', r.files]), [['1', 2], ['2', 2], ['3', 1]]) + const chunks = await db.from('release_chunks').count('* as n').first() + assert.equal(Number(chunks!.n), 4) + }) + + test('the same upload in another group (multi-posted) is a release of its own', async ({ assert }) => { + await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z')) + await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other')) + const rows = await releasesOf() + assert.lengthOf(rows, 2) + assert.equal(rows[0].identity_key, rows[1].identity_key) + }) }) diff --git a/tests/functional/settler.spec.ts b/tests/functional/settler.spec.ts index 1a4adc5..51cb154 100644 --- a/tests/functional/settler.spec.ts +++ b/tests/functional/settler.spec.ts @@ -9,7 +9,7 @@ test.group('Settler', (group) => { test('settles quiet releases: complete by the [i/N] counter, unknown without one', async ({ assert }) => { const old = new Date(Date.now() - 7 * 3600 * 1000) const row = (key: string, files: number, fileTotal: number | null, lastSeen: Date) => ({ - release_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files, file_total: fileTotal, last_seen_at: lastSeen, + release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files, file_total: fileTotal, last_seen_at: lastSeen, }) await db.table('releases').multiInsert([row('a', 10, 10, old), row('b', 8, 10, old), row('c', 3, null, old), row('d', 10, 10, new Date())]) assert.equal(await Settler.run(), 3) diff --git a/tests/unit/edge/range_indexer.spec.ts b/tests/unit/edge/range_indexer.spec.ts index efab6e5..0ca8efb 100644 --- a/tests/unit/edge/range_indexer.spec.ts +++ b/tests/unit/edge/range_indexer.spec.ts @@ -81,4 +81,20 @@ test.group('edge / taking over a range', () => { indexer.add(post(500_001, 'A.Release.2020.mkv', 1), 600_000) assert.deepEqual([indexer.take().from, indexer.take().from], [500_001, 600_000]) }) + test('uploads split where posting pauses for more than 3 hours, counted from the latest article', ({ assert }) => { + const art = (n: number, subject: string, minutes: number): Overview => [n, { 'subject': subject, 'from': 'p ', 'date': new Date(Date.UTC(2026, 9, 5, 10) + minutes * 60_000).toUTCString(), 'message-id': ``, ':bytes': '1000', 'xref': `news.example ${GROUP}:${n}` }] + const ix = new RangeIndexer({ group: GROUP, start: 1, end: 10_000, rules }) + ix.add([ + // A big file posted over 4 hours, then its par2 right after: one upload. + ...Array.from({ length: 5 }, (_, i) => art(100 + i, `Big.Release [1/2] - "Big.Release.mkv" yEnc (${i + 1}/5)`, i * 60)), + art(200, 'Big.Release [2/2] - "Big.Release.par2" yEnc (1/1)', 245), + // The same file again a day later: another. + ...Array.from({ length: 5 }, (_, i) => art(300 + i, `Big.Release [1/2] - "Big.Release.mkv" yEnc (${i + 1}/5)`, 1440 + i)), + ], 10_001) + ix.finish() + const batch = ix.take() + assert.deepEqual(batch.releases.map((r) => [r.upload, r.files]), [[0, 2], [1, 1]]) + assert.equal(batch.releases[0].postedLast! - batch.releases[0].postedAt!, 245 * 60_000) + assert.equal(batch.releases[0].key, batch.releases[1].key) + }) }) diff --git a/tests/unit/files/aggregator.spec.ts b/tests/unit/files/aggregator.spec.ts index 0125170..8e2e500 100644 --- a/tests/unit/files/aggregator.spec.ts +++ b/tests/unit/files/aggregator.spec.ts @@ -36,4 +36,20 @@ test.group('aggregate / FileAggregator', () => { assert.deepEqual(aggregator.evict(0).map((s) => s.number), [1, 2]) assert.equal(aggregator.openFiles, 0) }) + test('a re-upload is a copy of its own; a missing part sent later fills the first', ({ assert }) => { + const hour = 3_600_000 + const at = (number: number, h: number, id: string) => ({ ...seg('a', number, 3, `${id}@test`), postedAt: h * hour }) + const aggregator = new FileAggregator() + // Parts 1 and 2 at 10:00; part 1 again right after (a re-send: ignored). + assert.lengthOf(aggregator.add([at(1, 10, 'a1'), at(2, 10, 'a2'), at(1, 10.1, 'resend')]), 0) + // A day later: parts 1 and 2 again (a re-upload, another copy), and part 3 (fills the nearest copy lacking it: the new one). + assert.lengthOf(aggregator.add([at(1, 34, 'b1'), at(2, 34, 'b2')]), 0) + assert.equal(aggregator.openFiles, 2) + const [second] = aggregator.add([at(3, 34.2, 'b3')]) + assert.deepEqual(second.segments.map((s) => s.messageId), ['b1@test', 'b2@test', 'b3@test']) + // The first copy's missing part, sent two days later: it completes the first. + const [first] = aggregator.add([at(3, 58, 'a3')]) + assert.deepEqual(first.segments.map((s) => s.messageId), ['a1@test', 'a2@test', 'a3@test']) + assert.equal(aggregator.openFiles, 0) + }) }) diff --git a/tools/edge-bench-rs/check.ts b/tools/edge-bench-rs/check.ts index f4d0187..c4a4a51 100644 --- a/tools/edge-bench-rs/check.ts +++ b/tools/edge-bench-rs/check.ts @@ -50,9 +50,9 @@ const release = (r: Item, b: number) => { const ts: Record = { b: [], r: [], s: [], c: [], n: [], f: [] } batches.forEach((batch, b) => { ts.b.push({ b, from: batch.from, next: batch.next, headers: batch.headers }) - for (const r of batch.releases) ts.r.push(release({ key: r.key, name: r.name, source: r.source, fileTotal: r.fileTotal, poster: r.poster, groups: r.groups, postedAt: r.postedAt, files: r.files, segments: r.segments, bytes: r.bytes, firstArticle: r.firstArticle, lastArticle: r.lastArticle, readable: r.readable, hasPar2: r.hasPar2, payload: r.payload }, b)) - for (const s of batch.samples) ts.s.push({ b, key: s.key, rule: s.rule, ruleKey: s.ruleKey, segments: s.segments }) - for (const c of batch.candidates) ts.c.push({ b, releaseKey: c.releaseKey, kind: c.kind, messageId: c.messageId, file: c.file }) + for (const r of batch.releases) ts.r.push(release({ upload: r.upload, key: r.key, name: r.name, source: r.source, fileTotal: r.fileTotal, poster: r.poster, groups: r.groups, postedAt: r.postedAt, postedLast: r.postedLast, files: r.files, segments: r.segments, bytes: r.bytes, firstArticle: r.firstArticle, lastArticle: r.lastArticle, readable: r.readable, hasPar2: r.hasPar2, payload: r.payload }, b)) + for (const s of batch.samples) ts.s.push({ b, key: s.key, upload: s.upload, rule: s.rule, ruleKey: s.ruleKey, segments: s.segments }) + for (const c of batch.candidates) ts.c.push({ b, upload: c.upload, releaseKey: c.releaseKey, kind: c.kind, messageId: c.messageId, file: c.file }) for (const n of batch.counts) ts.n.push({ b, rule: n.rule, key: n.key, articles: n.articles, files: n.files }) for (const f of batch.fragments) ts.f.push({ b, fileKey: f.fileKey, name: f.name, totalParts: f.totalParts, postedAt: f.postedAt, segments: f.segments }) }) @@ -65,15 +65,15 @@ for (const line of (await readFile(dumpPath, 'utf8')).split('\n')) { const { t, ...item } = JSON.parse(line) if (t === 'b') b++ if (t === 'r') rs.r.push(release({ ...item, payload: Buffer.from(item.payload, 'base64') }, b)) - else if (t === 's') rs.s.push({ b, key: item.key, rule: item.rule, ruleKey: item.ruleKey, segments: item.segments }) + else if (t === 's') rs.s.push({ b, key: item.key, upload: item.upload, rule: item.rule, ruleKey: item.ruleKey, segments: item.segments }) else rs[t].push({ b, ...item }) } const keyOf: Record string> = { b: (i) => String(i.b), - r: (i) => `${i.b} ${i.key}`, - s: (i) => `${i.b} ${i.key}`, - c: (i) => `${i.b} ${i.releaseKey} ${i.kind} ${i.messageId}`, + r: (i) => `${i.b} ${i.key} ${i.upload}`, + s: (i) => `${i.b} ${i.key} ${i.upload}`, + c: (i) => `${i.b} ${i.releaseKey} ${i.upload} ${i.kind} ${i.messageId}`, n: (i) => `${i.b} ${i.rule} ${i.key}`, f: (i) => `${i.b} ${i.fileKey}`, } diff --git a/tools/edge-bench-rs/src/indexer.rs b/tools/edge-bench-rs/src/indexer.rs index a2f7559..c170614 100644 --- a/tools/edge-bench-rs/src/indexer.rs +++ b/tools/edge-bench-rs/src/indexer.rs @@ -31,12 +31,14 @@ pub struct Seg { pub struct ReleaseChunk { pub key: String, + pub upload: usize, pub name: String, pub source: &'static str, pub file_total: Option, pub poster: String, pub groups: Vec, pub posted_at: Option, + pub posted_last: Option, pub files: usize, pub segments: usize, pub bytes: i64, @@ -49,6 +51,7 @@ pub struct ReleaseChunk { pub struct Candidate { pub release_key: String, + pub upload: usize, pub kind: &'static str, pub message_id: String, pub file: String, @@ -275,60 +278,167 @@ fn scan_overview_text(text: &str, mut line: impl FnMut(Option, &str, &str, } } +/// A re-upload's copy of a file is kept apart (FileAggregator, uploadGapMs). +pub const UPLOAD_GAP_MS: i64 = 3 * 60 * 60 * 1000; + struct OpenFile { parts: Vec, seen: FxHashSet, total: u32, touched: u64, + /// Posting times seen. + earliest: Option, + latest: Option, +} + +impl OpenFile { + fn distance(&self, t: Option) -> i64 { + match (t, self.earliest, self.latest) { + (Some(t), Some(e), Some(l)) => 0.max(e - t).max(t - l), + _ => 0, + } + } } #[derive(Default)] struct Aggregator { - open: FxHashMap, + /// Per file key its open copies, oldest first. + open: FxHashMap>, batch: u64, } impl Aggregator { fn add(&mut self, segments: Vec) -> Vec> { self.batch += 1; + let batch = self.batch; let mut touched: Vec = Vec::new(); let mut touched_set: FxHashSet = FxHashSet::default(); for s in segments { - let key = &s.file.file_key; - let file = match self.open.get_mut(key.as_str()) { - Some(f) => f, - None => self.open.entry(key.clone()).or_insert(OpenFile { parts: Vec::new(), seen: FxHashSet::default(), total: s.file.total_parts, touched: self.batch }), - }; + let key = s.file.file_key.clone(); if !touched_set.contains(key.as_str()) { touched_set.insert(key.clone()); touched.push(key.clone()); } - file.touched = self.batch; - if file.seen.insert(s.number) { - file.parts.push(s); + let copies = self.open.entry(key).or_default(); + let t = s.posted_at; + // A copy lacking the part, the nearest in time. + let mut chosen: Option = None; + let mut best = i64::MAX; + for (i, copy) in copies.iter().enumerate() { + if copy.seen.contains(&s.number) { + continue; + } + let d = copy.distance(t); + if d < best { + best = d; + chosen = Some(i); + } } + let index = match chosen { + Some(i) => i, + None => { + if let Some(near) = copies.iter_mut().find(|c| c.distance(t) <= UPLOAD_GAP_MS) { + near.touched = batch; + continue; + } + copies.push(OpenFile { parts: Vec::new(), seen: FxHashSet::default(), total: s.file.total_parts, touched: batch, earliest: None, latest: None }); + copies.len() - 1 + } + }; + let file = &mut copies[index]; + file.seen.insert(s.number); + if let Some(t) = t { + if file.earliest.is_none_or(|e| t < e) { + file.earliest = Some(t); + } + if file.latest.is_none_or(|l| t > l) { + file.latest = Some(t); + } + } + file.touched = batch; + file.parts.push(s); } let mut complete = Vec::new(); for key in touched { - let f = self.open.get(&key).unwrap(); - if (f.parts.len() as u32) < f.total { - continue; + let copies = self.open.get_mut(&key).unwrap(); + let mut i = 0; + while i < copies.len() { + if (copies[i].parts.len() as u32) < copies[i].total { + i += 1; + continue; + } + let mut f = copies.remove(i); + f.parts.sort_by_key(|s| s.number); + complete.push(f.parts); + } + if copies.is_empty() { + self.open.remove(&key); } - let mut f = self.open.remove(&key).unwrap(); - f.parts.sort_by_key(|s| s.number); - complete.push(f.parts); } complete } - /// Files idle for `idle` batches (all when 0): their segments, file by file in arrival order. + /// Files idle for `idle` batches (all when 0): their segments, per file key (its copies one after another). fn evict(&mut self, idle: u64) -> Vec> { let batch = self.batch; - let gone: Vec = self.open.iter().filter(|(_, f)| !(idle > 0 && batch - f.touched < idle)).map(|(k, _)| k.clone()).collect(); - gone.into_iter().map(|k| self.open.remove(&k).unwrap().parts).collect() + let mut out = Vec::new(); + let keys: Vec = self.open.keys().cloned().collect(); + for key in keys { + let copies = self.open.get_mut(&key).unwrap(); + let mut gone: Vec = Vec::new(); + let mut i = 0; + while i < copies.len() { + if idle > 0 && batch - copies[i].touched < idle { + i += 1; + continue; + } + gone.extend(copies.remove(i).parts); + } + if copies.is_empty() { + self.open.remove(&key); + } + if !gone.is_empty() { + out.push(gone); + } + } + out } } +/// uploadsOf: a key's files split where posting pauses for more than UPLOAD_GAP_MS after the latest article so far. +fn uploads_of(files: Vec>) -> Vec>> { + if files.len() < 2 { + return vec![files]; + } + let first: Vec> = files.iter().map(|f| posted_of(f)).collect(); + let last: Vec> = files.iter().map(|f| f.iter().filter_map(|s| s.posted_at).max()).collect(); + let mut dated: Vec = (0..files.len()).filter(|&i| first[i].is_some()).collect(); + if dated.len() < 2 { + return vec![files]; + } + dated.sort_by_key(|&i| (first[i].unwrap(), i)); + let mut upload = vec![0usize; files.len()]; + let mut current = 0; + let mut end = last[dated[0]].unwrap(); + for &i in &dated[1..] { + if first[i].unwrap() - end > UPLOAD_GAP_MS { + current += 1; + end = last[i].unwrap(); + } else if last[i].unwrap() > end { + end = last[i].unwrap(); + } + upload[i] = current; + } + if current == 0 { + return vec![files]; + } + let mut out: Vec>> = (0..=current).map(|_| Vec::new()).collect(); + for (i, f) in files.into_iter().enumerate() { + out[upload[i]].push(f); + } + out +} + pub struct RangeIndexer { group: String, start: i64, @@ -452,22 +562,23 @@ impl RangeIndexer { let order = std::mem::take(&mut self.completed_order); let mut by_release = std::mem::take(&mut self.completed); for key in order { - let list = by_release.remove(&key).unwrap(); - let judged: Vec = list - .iter() - .map(|f| JudgedFile { name: &f[0].file.name, total_parts: f[0].file.total_parts, groups: &f[0].file.groups, first_message_id: &f[0].message_id }) - .collect(); - match names::judge(&key, &judged) { - Verdict::Keep => { - let candidates = candidates_of(&key, &list); - self.batch.releases.push(chunk_of(&key, &list)); - self.batch.candidates.extend(candidates); - } - Verdict::Drop { rule, key: rule_key, sample } => { - let segments: usize = list.iter().map(|f| f.len()).sum(); - self.count(rule, rule_key.clone(), segments as u64, list.len() as u64); - if sample { - self.batch.samples.push((chunk_of(&key, &list), rule, rule_key)); + for (upload, list) in uploads_of(by_release.remove(&key).unwrap()).into_iter().enumerate() { + let judged: Vec = list + .iter() + .map(|f| JudgedFile { name: &f[0].file.name, total_parts: f[0].file.total_parts, groups: &f[0].file.groups, first_message_id: &f[0].message_id }) + .collect(); + match names::judge(&key, &judged) { + Verdict::Keep => { + let candidates = candidates_of(&key, &list, upload); + self.batch.releases.push(chunk_of(&key, &list, upload)); + self.batch.candidates.extend(candidates); + } + Verdict::Drop { rule, key: rule_key, sample } => { + let segments: usize = list.iter().map(|f| f.len()).sum(); + self.count(rule, rule_key.clone(), segments as u64, list.len() as u64); + if sample { + self.batch.samples.push((chunk_of(&key, &list, upload), rule, rule_key)); + } } } } @@ -528,7 +639,7 @@ fn posted_of(segments: &[Seg]) -> Option { segments.iter().filter_map(|s| s.posted_at).min() } -fn chunk_of(key: &str, files: &[Vec]) -> ReleaseChunk { +fn chunk_of(key: &str, files: &[Vec], upload: usize) -> ReleaseChunk { let first = &files[0][0].file; let chunk_files: Vec = files .iter() @@ -542,8 +653,14 @@ fn chunk_of(key: &str, files: &[Vec]) -> ReleaseChunk { let mut first_article = i64::MAX; let mut last_article = 0; let mut posted_at: Option = None; + let mut posted_last: Option = None; for f in files { for s in f { + if let Some(t) = s.posted_at { + if posted_last.is_none_or(|p| t > p) { + posted_last = Some(t); + } + } segments += 1; bytes += s.bytes; first_article = first_article.min(s.article); @@ -557,12 +674,14 @@ fn chunk_of(key: &str, files: &[Vec]) -> ReleaseChunk { } ReleaseChunk { key: key.to_string(), + upload, name: first.release.name.clone(), source: first.release.source, file_total: files.iter().find_map(|f| f[0].file.release.file_total_truthy()), poster: first.poster.clone(), groups: first.groups.clone(), posted_at, + posted_last, files: files.len(), segments, bytes, @@ -574,10 +693,10 @@ fn chunk_of(key: &str, files: &[Vec]) -> ReleaseChunk { } } -fn candidates_of(key: &str, files: &[Vec]) -> Vec { +fn candidates_of(key: &str, files: &[Vec], upload: usize) -> Vec { let mut out = Vec::new(); let mut tail: Option<(i64, &Vec)> = None; - let c = |kind, s: &Seg| Candidate { release_key: key.to_string(), kind, message_id: s.message_id.clone(), file: s.file.name.clone() }; + let c = |kind, s: &Seg| Candidate { release_key: key.to_string(), upload, kind, message_id: s.message_id.clone(), file: s.file.name.clone() }; for f in files { let name = &f[0].file.name; if let Some(volume) = names::seven_zip_volume(name).filter(|&v| v > 1) { @@ -593,7 +712,7 @@ fn candidates_of(key: &str, files: &[Vec]) -> Vec { } } if let Some((_, f)) = tail { - out.push(Candidate { release_key: key.to_string(), kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone() }); + out.push(Candidate { release_key: key.to_string(), upload, kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone() }); } if out.is_empty() && files.len() == 1 && is_random_file(&files[0][0].file.name) { out.push(c("lone", &files[0][0])); @@ -601,3 +720,56 @@ fn candidates_of(key: &str, files: &[Vec]) -> Vec { out.truncate(4); out } + +#[cfg(test)] +mod tests { + use super::*; + + const GROUP: &str = "alt.binaries.test"; + + fn line(n: i64, subject: &str, minutes: i64) -> String { + let date = chrono::DateTime::from_timestamp_millis(1_791_194_400_000 + minutes * 60_000).unwrap().format("%a, %d %b %Y %H:%M:%S +0000"); + format!("{n}\t{subject}\tp \t{date}\t\t\t1000\t10\tXref: news.example {GROUP}:{n}") + } + + #[test] + fn uploads_split_as_in_the_typescript() { + let mut priorities = FxHashMap::default(); + priorities.insert(GROUP.to_string(), 1); + let mut ix = RangeIndexer::new(GROUP, 1, 10_000, priorities, 5); + let mut lines: Vec = (0..5).map(|i| line(100 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), i * 60)).collect(); + lines.push(line(200, r#"Big.Release [2/2] - "Big.Release.par2" yEnc (1/1)"#, 245)); + lines.extend((0..5).map(|i| line(300 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), 1440 + i))); + ix.add_text(&lines.join("\r\n"), 10_001); + ix.finish(); + let batch = ix.take(); + let got: Vec<(usize, usize)> = batch.releases.iter().map(|r| (r.upload, r.files)).collect(); + assert_eq!(got, vec![(0, 2), (1, 1)]); + assert_eq!(batch.releases[0].posted_last.unwrap() - batch.releases[0].posted_at.unwrap(), 245 * 60_000); + assert_eq!(batch.releases[0].key, batch.releases[1].key); + } + + #[test] + fn copies_and_fills_as_in_the_typescript() { + let hour = 3_600_000i64; + let meta = Rc::new(FileMeta { + file_key: "a".into(), + name: "a.bin".into(), + poster: "p".into(), + subject: "s".into(), + total_parts: 3, + groups: vec![], + release: names::release_identity("p", "s", "a.bin"), + }); + let at = |number: u32, h: f64, id: &str| Seg { file: meta.clone(), number, message_id: id.into(), bytes: 100, article: 0, posted_at: Some((h * hour as f64) as i64) }; + let mut agg = Aggregator::default(); + assert!(agg.add(vec![at(1, 10.0, "a1"), at(2, 10.0, "a2"), at(1, 10.1, "resend")]).is_empty()); + assert!(agg.add(vec![at(1, 34.0, "b1"), at(2, 34.0, "b2")]).is_empty()); + assert_eq!(agg.open.get("a").unwrap().len(), 2); + let second = agg.add(vec![at(3, 34.2, "b3")]); + assert_eq!(second[0].iter().map(|s| s.message_id.as_str()).collect::>(), vec!["b1", "b2", "b3"]); + let first = agg.add(vec![at(3, 58.0, "a3")]); + assert_eq!(first[0].iter().map(|s| s.message_id.as_str()).collect::>(), vec!["a1", "a2", "a3"]); + assert!(agg.open.is_empty()); + } +} diff --git a/tools/edge-bench-rs/src/main.rs b/tools/edge-bench-rs/src/main.rs index efa8c0b..7f7241c 100644 --- a/tools/edge-bench-rs/src/main.rs +++ b/tools/edge-bench-rs/src/main.rs @@ -72,7 +72,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { line.push_str(&format!("{{\"t\":\"b\",\"from\":{},\"next\":{},\"headers\":{}}}", batch.from, batch.next, batch.headers)); emit(&mut line); let release = |r: &indexer::ReleaseChunk, kind: &str, line: &mut String| { - line.push_str(&format!("{{\"t\":\"{kind}\",\"key\":")); + line.push_str(&format!("{{\"t\":\"{kind}\",\"upload\":{},\"key\":", r.upload)); json_str(&r.key, line); line.push_str(",\"name\":"); json_str(&r.name, line); @@ -89,6 +89,8 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { } line.push_str("],\"postedAt\":"); opt_num(r.posted_at, line); + line.push_str(",\"postedLast\":"); + opt_num(r.posted_last, line); line.push_str(&format!( ",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"payload\":\"{}\"", r.files, @@ -114,7 +116,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { emit(&mut line); } for c in &batch.candidates { - line.push_str("{\"t\":\"c\",\"releaseKey\":"); + line.push_str(&format!("{{\"t\":\"c\",\"upload\":{},\"releaseKey\":", c.upload)); json_str(&c.release_key, &mut line); line.push_str(&format!(",\"kind\":\"{}\",\"messageId\":", c.kind)); json_str(&c.message_id, &mut line);