diff --git a/REBUILD.md b/REBUILD.md index 7ae8905..e104bfe 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -550,6 +550,13 @@ releases; the inspector matches them again), `nntp_shares`, `nodes`, `release_texts`. The category trigger reads the groups from `group_sets`; the judge's dirty trigger watches `poster_id` instead of `poster`. +- `pending_releases` (migration 6300, the HDD): the unnamed releases + outside the catalog (decision f): `release_key` (primary), `identity_key`, + `group_id`, `name` (the random stem), `poster` text, the posting span, + `first_article`, `last_article`, counts, `partial`, `finished`, `tool`, + `candidates` jsonb (the probe targets by effort), `status` (0 waiting, 1 + named, 2 discarded), `tries`, `created_at`, `probed_at`; indexed by + (group, tool, status, created_at) for the ladder. - `release_texts(release_id, kind smallint, text varchar(255))` (6200), primary key on all three, one trigram index (fastupdate): every string search matches against, appended by whoever learns one and never an @@ -682,10 +689,12 @@ until the window is done (LOG.md 2026-10-26 (111) has the first local numbers, (112) the homelab run). Next: the replay on homelab over the 2016 window (the NVMe bytes per -header for real), the ordered range leases with the carried window in -the edge-node and its snapshot (design 1 and 5), the prober writing -`release_texts` kind 2, random-named singles without a chunk (decision f: -2.3 M of the first replay's 4.0 M singles are "Bob"'s 32-hex posts). Gone from the list: block +header for real), the pending ladder and the signature learner (decision +f, amended 2026-10-26: the prober working `pending_releases` cheapest +first, verdicts per tool and group, a named release moved into the +catalog with its segments read again), the ordered range leases with the +carried window in the edge-node and its snapshot (design 1 and 5), the +prober writing `release_texts` kind 2. Gone from the list: block leases, the planner and the block merger (design 1, amended). ## Decisions @@ -697,11 +706,33 @@ c. Group list: editable, owner by priority number (above). **Accepted.** d. Reposts: separate releases in phase 1, merged in phase 2. **Accepted.** e. Headers through the coordinator's relay, with prefetch (part 1). **Accepted.** -f. Random-named releases (no file with a readable name) keep no chunk in - the header pass: their file keys and probe candidates only, until - phase 2 names them (bodies) and fetches their segments again. The full - history is ~3 TB with them, ~1.2–1.4 TB without (725 M of the 1.25 B - kept files; LOG.md 2026-10-09). **Accepted** 2026-10-09, not built yet. +f. **Nothing enters the catalog without a usable name** (amended + 2026-10-26, the user's direction; before: random-named releases keep + no chunk but their rows). A release the window closed on whose files + all have random names is no release yet: it goes to `pending_releases` + (migration 6300, the HDD) with its keys, counts, article span, upload + tool and probe targets, no row in `releases`, `release_texts` or + `posters`, no chunk. The pending ladder works them **from least to most + effort**: (0) no fetch: a signature verdict (tool and group, poster) + learned from earlier probes, a spot or NZB that names the message-ids; + (1) a par2 main file's first article (it names the files); (2) the first + article of the first file (the yEnc name, an archive's listing in its + first 16 KiB); (3) a 7z's tail; then nothing more. A probe that yields + a usable name moves the release into the catalog, its segments read + again from the corpus (the range stored) or the server; one that + yields none counts against its signature, and a signature with + ≥100 probes and ≤0.5% named (voyager 2's garbage learner, LOG.md + 2026-10-08 (5)) is **discarded** from then on, a sample of it still + probed to notice a change. The reason, in the 2016 replay: one poster's + hash-named pieces ("Bob", 11 aliases, 3.2 M releases in five small + groups) were 76% of the rows and three quarters of the NVMe bytes for + 11% of the segments and nothing anyone can find; the full history has + 725 M random-named files (LOG.md 2026-10-09). Built 2026-10-26: the + edge sends an unnamed release with an empty payload and its tool, the + batch store writes `pending_releases` (LOG.md (114)). Not built: the + ladder itself (the prober working `pending_releases` cheapest first, + the signature verdicts, the move into the catalog with the segments + read again), and the spot/NZB match on pending keys. g. The account's NNTP connections shared by priority (`nntp_shares`): the coordinator (headers) before the prober, each between its minimum and maximum, 5 kept for the spotter, the spot inspector, the plain diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index c8df57c..28c94aa 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -8,6 +8,7 @@ import { decodeChunk } from '#lib/chunks/index' import { SIDECAR } from '#services/sidecar_folder' import { identityKey, releaseKey } from '#lib/release/keys' import { sourceCode } from '#lib/release/source' +import { uploadTool } from '#lib/rules/index' import { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind' /** @@ -30,6 +31,8 @@ export interface StoredBatch { /** Provisional (tail) releases the batch finished. */ finalized: number candidates: number + /** Unnamed releases written to the pending store instead of the catalog (decision f). */ + pending: number } /** A chunk's slot in its batch: the key and the upload the edge numbered it with (ProbeCandidate points at it). */ @@ -73,18 +76,29 @@ class BatchStore { newReleases: 0, finalized: 0, candidates: 0, + pending: 0, } + // Nothing enters the catalog without a usable name (decision f): an unnamed release (no readable file name) + // goes to the pending store with its probe targets, the named ones to the catalog. + const named = batch.releases.filter((r) => r.readable) + const unnamed = batch.releases.filter((r) => !r.readable) const written = await this.writeReleases( trx, { groupId: Number(range.group_id), blockId }, batch.from, - batch.releases, + named, batch.candidates ) result.newReleases = written.newReleases result.finalized = written.finalized result.candidates = written.candidates + result.pending = await this.writePending( + trx, + Number(range.group_id), + unnamed, + batch.candidates + ) if (batch.counts.length > 0) { await trx.rawQuery( @@ -153,6 +167,68 @@ class BatchStore { return Number(block.id) } + /** + * The unnamed releases of a batch into pending_releases (migration 6300, the HDD): keys, counts, the upload + * tool and the probe targets, no chunk. The pending ladder (REBUILD.md decision f) probes them cheapest first, + * and a name moves one into the catalog; a provisional tail release sent again adds its counts up. + */ + async writePending( + trx: TransactionClientContract, + groupId: number, + releases: ReleaseChunk[], + candidates: ProbeCandidate[] + ): Promise { + if (releases.length === 0) return 0 + const targets = new Map() + for (const c of candidates) { + const key = slot(c.releaseKey, c.upload ?? 0) + const list = targets.get(key) ?? [] + list.push({ kind: c.kind, messageId: c.messageId, file: c.file }) + targets.set(key, list) + } + let written = 0 + for (let i = 0; i < releases.length; i += 1000) { + const rows = releases.slice(i, i + 1000).map((r) => { + const probes = targets.get(slot(r.key, r.upload)) ?? [] + return { + release_key: releaseKey(r.key, groupId, r.firstArticle).toString('hex'), + identity_key: identityKey(r.key).toString('hex'), + name: r.name.slice(0, TEXT_MAX), + poster: r.poster || null, + first: r.postedAt, + last: r.postedLast ?? r.postedAt, + first_article: r.firstArticle, + last_article: r.lastArticle, + files: r.files, + segments: r.segments, + bytes: r.bytes, + partial: r.partial === true, + finished: r.finished !== false, + tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'), + candidates: probes, + } + }) + const { rows: done } = await trx.rawQuery( + `INSERT INTO pending_releases (release_key, identity_key, group_id, name, poster, posted_at, last_posted_at, first_article, last_article, + files, segments, bytes, partial, finished, tool, candidates) + SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.name, v.poster, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), + v.first_article, v.last_article, v.files, v.segments, v.bytes, v.partial, v.finished, v.tool, v.candidates + FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, identity_key text, name text, poster text, first bigint, last bigint, first_article bigint, last_article bigint, + files int, segments bigint, bytes bigint, partial boolean, finished boolean, tool text, candidates jsonb) + ON CONFLICT (release_key) DO UPDATE SET + files = pending_releases.files + excluded.files, segments = pending_releases.segments + excluded.segments, bytes = pending_releases.bytes + excluded.bytes, + last_posted_at = greatest(pending_releases.last_posted_at, excluded.last_posted_at), last_article = greatest(pending_releases.last_article, excluded.last_article), + partial = excluded.partial, finished = excluded.finished, + candidates = CASE WHEN jsonb_array_length(pending_releases.candidates) = 0 THEN excluded.candidates ELSE pending_releases.candidates END + WHERE NOT pending_releases.finished + RETURNING release_key`, + [groupId, JSON.stringify(rows)] + ) + written += done.length + } + return written + } + /** * The releases of a batch into the catalog, one chunk each, and their probe candidates. Every chunk is one * release: inserted, or (a provisional tail release sent again) finished. diff --git a/commands/replay_corpus.ts b/commands/replay_corpus.ts index 723708b..50521aa 100644 --- a/commands/replay_corpus.ts +++ b/commands/replay_corpus.ts @@ -333,6 +333,11 @@ export default class ReplayCorpus extends BaseCommand { `SELECT source, count(*)::bigint AS n FROM releases WHERE group_id = ? GROUP BY source ORDER BY source`, [groupId] ) + const { rows: pending } = await db.rawQuery( + `SELECT count(*)::bigint AS n, coalesce(sum(segments), 0)::bigint AS segments, count(*) FILTER (WHERE jsonb_array_length(candidates) = 0)::bigint AS unprobable + FROM pending_releases WHERE group_id = ?`, + [groupId] + ) const { rows: headerRows } = await db.rawQuery( `SELECT coalesce(sum(headers), 0)::bigint AS headers FROM ranges WHERE id = ANY(?::bigint[])`, [ids] @@ -343,12 +348,17 @@ export default class ReplayCorpus extends BaseCommand { sum(pg_total_relation_size(c.oid))::bigint AS bytes FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'public' AND c.relkind = 'r' - AND (c.relname IN ('releases', 'release_texts', 'posters', 'group_sets', 'probe_candidates', 'rule_counts', 'rule_samples', 'blocks') + AND (c.relname IN ('releases', 'release_texts', 'posters', 'group_sets', 'probe_candidates', 'rule_counts', 'rule_samples', 'blocks', 'pending_releases') OR c.relname LIKE 'release_chunks%') GROUP BY 1 ORDER BY 2 DESC` ) // On CHUNK_TABLESPACE (the HDD) in the catalog: the chunks (migration 3200), inspections and candidates (6000). - const hdd = new Set(['release_chunks', 'probe_candidates', 'release_inspections']) + const hdd = new Set([ + 'release_chunks', + 'probe_candidates', + 'release_inspections', + 'pending_releases', + ]) let nvme = 0 let other = 0 const c = counts[0] @@ -364,6 +374,9 @@ export default class ReplayCorpus extends BaseCommand { this.logger.info( `By source: ${sources.map((s: any) => `${s.source}: ${fmt(s.n)}`).join(', ')}.` ) + this.logger.info( + `Pending (unnamed, outside the catalog): ${fmt(pending[0].n)} releases, ${fmt(pending[0].segments)} segments, ${fmt(pending[0].unprobable)} without a probe target.` + ) for (const s of sizes) { const bytes = Number(s.bytes) if (hdd.has(s.name)) other += bytes diff --git a/database/migrations/1770000006300_pending_releases.ts b/database/migrations/1770000006300_pending_releases.ts new file mode 100644 index 0000000..dbcc7b4 --- /dev/null +++ b/database/migrations/1770000006300_pending_releases.ts @@ -0,0 +1,60 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' +import env from '#start/env' + +/** + * pending_releases (REBUILD.md decision f, 2026-10-26): nothing enters the catalog without a usable name. A release + * the window closed on whose files all have random names is written here instead of releases: its keys (as the + * catalog would store them), its counts and article span, the upload tool from its message-id (with the group its + * signature), and its probe targets by effort (the edge's candidates: a par2, an archive start, a lone file, a 7z + * tail), no chunk. The pending ladder probes them cheapest first and moves a named one into the catalog, reading its + * segments again from the corpus or the server; a signature that names nothing is discarded. On CHUNK_TABLESPACE + * (the HDD): written once, read by the ladder, never by search. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE pending_releases ( + -- lib/release/keys.ts: the identity, the group and the first article; the identity alone. + release_key bytea PRIMARY KEY, + identity_key bytea NOT NULL, + group_id integer NOT NULL, + -- The stem or post the edge keyed it by (a random name: no release name, but the inner-stem merge's handle). + name varchar(255) NOT NULL, + poster text, + posted_at timestamptz, + last_posted_at timestamptz, + first_article bigint NOT NULL, + last_article bigint NOT NULL, + files integer NOT NULL DEFAULT 0, + segments bigint NOT NULL DEFAULT 0, + bytes bigint NOT NULL DEFAULT 0, + -- A file misses parts. + partial boolean NOT NULL DEFAULT false, + -- False: a tail release still open; a later chunk of its key adds its counts up. + finished boolean NOT NULL DEFAULT true, + -- lib/rules uploadTool: ngPost, nyuu, hex32, random, other, … + tool varchar(16) NOT NULL, + -- [{kind, messageId, file}] by effort (lib/edge ProbeCandidate); empty: nothing to read, only a signature. + candidates jsonb NOT NULL DEFAULT '[]', + -- 0 waiting, 1 named (in the catalog), 2 discarded (its signature's verdict, or every probe gave nothing). + status smallint NOT NULL DEFAULT 0, + tries smallint NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + probed_at timestamptz + )`) + // The ladder's order: a group's and tool's waiting releases, oldest first. + this.schema.raw( + `CREATE INDEX pending_releases_ladder_idx ON pending_releases (group_id, tool, status, created_at)` + ) + const hdd = env.get('CHUNK_TABLESPACE') + if (hdd) { + this.schema.raw(`ALTER TABLE pending_releases SET TABLESPACE ${hdd}`) + this.schema.raw(`ALTER INDEX pending_releases_pkey SET TABLESPACE ${hdd}`) + this.schema.raw(`ALTER INDEX pending_releases_ladder_idx SET TABLESPACE ${hdd}`) + } + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS pending_releases`) + } +} diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index af8b4c5..c4ba7bd 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -53,6 +53,11 @@ pub struct ReleaseChunk { pub edge: bool, /// A file misses parts. pub partial: bool, + /// The upload tool from the first article's message-id (names::upload_tool), for an unnamed release: with the + /// group its signature, what the pending ladder learns verdicts by. None for a named one. + pub tool: Option<&'static str>, + /// The files' names and segments (codec); empty for an unnamed release (decision f: nothing in the catalog + /// without a usable name; the pending store keeps its keys and probe targets until a probe names it). pub payload: Vec, /// The first files' subjects (at most SUBJECTS): edge-bench's --releases dump for the grouping sweeps. pub subjects: Vec, @@ -1154,17 +1159,26 @@ impl RangeIndexer { let mut chunk = chunk_of(key, &list, upload); chunk.finished = finished; chunk.edge = edge; + let album_none = album.is_none(); if let Some(name) = album { chunk.name = name; chunk.source = "album"; chunk.file_total = None; - } else if let Some(total) = chunk.file_total.filter(|&t| t >= BATCH_COUNTER_MIN_TOTAL && (chunk.files as i64) < t) { + } + if !chunk.readable { + // Unnamed: no chunk (its segments are read again once a probe names it), the tool for its signature. + chunk.tool = Some(names::upload_tool(&list[0][0].message_id)); + chunk.payload = encode_chunk(&[]); + } + if chunk.readable { + self.pending_files += chunk.files; + } + if let Some(total) = chunk.file_total.filter(|&t| album_none && t >= BATCH_COUNTER_MIN_TOTAL && (chunk.files as i64) < t) { // A batch counter: no file total (the release is as complete as its files are). if self.batch_totals.get(&(chunk.poster.clone(), total)).is_some_and(|(n, _)| *n >= BATCH_COUNTER_RELEASES) { chunk.file_total = None; } } - self.pending_files += chunk.files; self.batch.releases.push(chunk); self.batch.candidates.extend(candidates); } @@ -1267,6 +1281,7 @@ fn chunk_of(key: &str, files: &[Vec], upload: usize) -> ReleaseChunk { finished: true, edge: false, partial: files.iter().any(|f| (f.len() as u32) < f[0].file.total_parts), + tool: None, payload: encode_chunk(&chunk_files), subjects: chunk_files.iter().take(SUBJECTS).map(|f| f.subject.to_string()).collect(), } @@ -1411,6 +1426,23 @@ mod tests { assert!(ix.albums.is_empty()); } + #[test] + fn an_unnamed_release_keeps_no_chunk_but_its_tool_and_probe_target() { + let mut priorities = FxHashMap::default(); + priorities.insert(GROUP.to_string(), 1); + let mut ix = RangeIndexer::new(GROUP, 1, 10_000, 1, Rules::defaults(priorities), 5); + let lines: Vec = (0..4).map(|i| line(100 + i, &format!(r#"[1/1] - "q3B9OrpKiCxUnzgtUGnp.part01.rar" yEnc ({}/4)"#, i + 1), i)).collect(); + ix.add_text(&lines.join("\n"), 10_001); + ix.finish(); + let batch = ix.take(); + assert_eq!(batch.releases.len(), 1); + let r = &batch.releases[0]; + assert!(!r.readable && r.tool == Some("other") && r.files == 1 && r.segments == 4); + assert_eq!(r.payload, encode_chunk(&[])); + assert_eq!(batch.candidates.iter().map(|c| c.kind).collect::>(), vec!["archive"]); + assert_eq!(ix.pending(), 0); + } + #[test] fn album_templates_and_names() { assert_eq!(album_template("1481439079025_014.jpg.data").as_deref(), Some("#_#.jpg.data")); diff --git a/edge/src/wire.rs b/edge/src/wire.rs index 7e1b940..b2a7e9c 100644 --- a/edge/src/wire.rs +++ b/edge/src/wire.rs @@ -27,6 +27,8 @@ struct WireRelease<'a> { finished: bool, edge: bool, partial: bool, + #[serde(skip_serializing_if = "Option::is_none")] + tool: Option<&'a str>, payload: String, #[serde(skip_serializing_if = "Option::is_none")] rule: Option<&'a str>, @@ -87,6 +89,7 @@ fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRel finished: r.finished, edge: r.edge, partial: r.partial, + tool: r.tool, payload: base64::engine::general_purpose::STANDARD.encode(&r.payload), rule: rule.map(|(rule, _)| rule), rule_key: rule.map(|(_, key)| key), diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index 63e10fe..c8ee31c 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -35,6 +35,11 @@ export interface ReleaseChunk { edge?: boolean /** A file of the release misses parts (the window closed on it): not complete, whatever the counter says. */ partial?: boolean + /** + * Voyager 3, an unnamed release (readable false): the upload tool from its first article's message-id, its + * signature with the group; the payload is empty (no chunk until a probe names it: pending_releases). + */ + tool?: string payload: Buffer } diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index c67201c..0d7313c 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -181,7 +181,13 @@ test.group('BatchStore', (group) => { ) const first = ix.take() const stored = await BatchStore.store(rangeId, nodeId, first) - assert.deepEqual(stored, { releases: 1, newReleases: 1, finalized: 0, candidates: 2 }) + assert.deepEqual(stored, { + releases: 1, + newReleases: 1, + finalized: 0, + candidates: 2, + pending: 0, + }) const release = await db.from('releases').first() assert.include(release, { @@ -487,6 +493,51 @@ test.group('BatchStore', (group) => { assert.notEqual(rows[0].release_key, rows[1].release_key) }) + test('an unnamed release goes to the pending store with its probe targets, not into the catalog', async ({ + assert, + }) => { + const { rangeId, nodeId } = await rangeIn(GROUP, 1) + const batch = batchOf(GROUP, 1, [ + ...upload(10, 'q3B9OrpKiCxUnzgtUGnp.part01.rar', 4, '[1/2]', '2026-10-05T10:00:00Z'), + ...upload(20, 'q3B9OrpKiCxUnzgtUGnp.par2', 1, '[2/2]', '2026-10-05T10:01:00Z'), + ]) + assert.lengthOf(batch.releases, 1) + assert.isFalse(batch.releases[0].readable) + const stored = await BatchStore.store(rangeId, nodeId, batch) + assert.include(stored, { releases: 1, newReleases: 0, candidates: 0, pending: 1 }) + assert.equal(Number((await db.from('releases').count('* as n').first())!.n), 0) + assert.equal(Number((await db.from('probe_candidates').count('* as n').first())!.n), 0) + assert.equal(Number((await db.from('release_chunks').count('* as n').first())!.n), 0) + const pending = await db.from('pending_releases').firstOrFail() + assert.include(pending, { + files: 2, + segments: '5', + partial: false, + finished: true, + tool: 'other', + status: 0, + poster: 'p ', + }) + assert.lengthOf(pending.release_key, 20) + assert.deepEqual( + pending.candidates.map((c: { kind: string; file: string }) => [c.kind, c.file]).sort(), + [ + ['archive', 'q3B9OrpKiCxUnzgtUGnp.part01.rar'], + ['par2', 'q3B9OrpKiCxUnzgtUGnp.par2'], + ] + ) + // Sent again as a later chunk of a tail release: its counts add up while it is open, not once finished. + const again = await BatchStore.store(rangeId, nodeId, { + ...batch, + from: batch.next, + next: batch.next + 1000, + releases: [{ ...batch.releases[0], files: 1, segments: 5, bytes: 100 }], + candidates: [], + }) + assert.include(again, { pending: 0 }) + assert.include(await db.from('pending_releases').firstOrFail(), { files: 2, segments: '5' }) + }) + test("an album's members stay findable by their own names (release_texts kind member)", async ({ assert, }) => {