diff --git a/.gitea/workflows/deploy.yaml b/.gitea/workflows/deploy.yaml index 627f112..da0265b 100644 --- a/.gitea/workflows/deploy.yaml +++ b/.gitea/workflows/deploy.yaml @@ -83,8 +83,8 @@ jobs: echo "$HOME/.cargo/bin" >> "$GITHUB_PATH" - name: cargo test run: cargo test --release --manifest-path edge/Cargo.toml - - name: Rust and TypeScript identical on the fixtures - run: edge/check-fixtures.sh + - name: The Rust indexer over the fixtures + run: edge/bench-fixtures.sh deploy: needs: [test, edge] diff --git a/REBUILD.md b/REBUILD.md index c2f869d..dd8d9f0 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -401,21 +401,26 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md 2. **The window.** The worker reads a block in article order and keeps its open files and releases in memory (counters, names, sizes; the segment lists spooled to a file per block on the worker's disk). The *frontier* - is the median Date of the last 4,096 articles with a plausible date - (inside the block's span ± 7 days); Date headers alone aren't trusted - (clock skew, bogus years). A release is **finished** when the frontier - is 24 h past its last part (`WINDOW_HOURS`), or at the block's end: - written once, with its completeness decided then (every announced file - there; par2 decides later for those without a counter), never updated by - the header pass again. Incomplete releases are kept (user, 2026-10-24). - A fill posted after the window is a release of its own (the judge's - `repeated` rule already knows reposts). Releases still open at a block's - end are written flagged `edge`; the block merger (replaces the fragment - merger) joins the edge releases of adjacent blocks of a group by - identity key once both are done, and clears the flag. - The settler goes: `settled_at` is the finish. The **tail** keeps its - latency: a tail lease emits a release provisionally as today and the - window finalizes it once (two writes, not one per range). + is the median Date of the last 1,024 articles with a plausible date + (within 7 days of the frontier; 1,024 such dates in a row that agree + among themselves move it, the group having paused); Date headers alone + aren't trusted (clock skew, bogus years). 1,024, not more: the median + trails the newest article by half the ring, a day in a group of 500 + articles a day. A release is **finished** when the frontier is 24 h past + its last part (`WINDOW_MS`): written once, with its completeness decided + then (every announced file there and no file short of parts; par2 + decides later for those without a counter), never updated by the header + pass again. Incomplete releases are kept (user, 2026-10-24). A fill + posted after the window is a release of its own (the judge's `repeated` + rule already knows reposts). At the block's end the releases the window + hasn't closed on are written **provisionally** (`settled_at` null) and + flagged `edge`, as are the finished releases posted within 24 h of the + block's first frontier; the block merger (replaces the fragment merger) + joins the edge releases of adjacent blocks of a group by identity key + once both are done, finishes them and clears the flag. The settler goes: + `settled_at` is the finish. The **tail** needs no path of its own: a + tail lease's open releases at its end are such provisional edges, the + next lease's continuation joins them. 3. **Albums.** Files of one poster in one owner group whose subjects differ only by a number sequence, or share a directory-like prefix, posted @@ -571,11 +576,38 @@ service compare the source code. Tests insert fixtures through `tests/functional/helpers/releases.ts` (`insertRelease`: text keys, poster and groups by name, source by name). The unit and functional suites pass. -Next: the window and albums in the Rust edge (`finished`, `edge`, the -provisional tail and its finish), block leases and the planner (then the -store keys by the leased block and the lease columns leave `ranges`), the -block merger, the prober writing `release_texts` kind 2, the corpus replay -harness. +### The window in the Rust edge (built 2026-10-25) + +`edge/src/indexer.rs`: the scanner and the file aggregator as before; then +the window instead of the range path. `Frontier` keeps a ring of the last +`FRONTIER_ARTICLES` (1,024) plausible dates (within `PLAUSIBLE_MS`, 7 +days, of the frontier; 1,024 implausible dates in a row that agree move +it), recomputed once per chunk; established after 64 dated articles +(`settle` at the block's end for smaller blocks); its first value is the +block's start in posting time. Per chunk: completed files, then the files +whose last part is before frontier − `WINDOW_MS` (24 h; undated ones idle +for `idle` chunks), go into the open release of their key whose posting +span is within `UPLOAD_GAP_MS` (3 h) of theirs, else start one (the +upload number per key, distinct within a batch); releases whose last part +is before frontier − `WINDOW_MS` are judged and written `finished`, as an +`edge` when posted within `WINDOW_MS` of the block's start. At the end +(`finish`) everything left is written: finished when the window closed on +it, else `finished: false, edge: true`. A file short of parts stays in +its release and sets `partial` (the store: `complete = false`). A lone +article of a multi-part random file is still counted `per-article`, not +written. Gone: fragments and the cut-off, `uploads_of` (online now), +`check.ts` and the fixture comparison (the parity with the TypeScript +indexer; CI runs `edge/bench-fixtures.sh` instead), the `incomplete` count +(incomplete releases are kept). The wire carries `finished`, `edge` and +`partial`; the store reads them (`partial` into `complete`). Memory: the +window holds a day of a group's segments in memory (the busiest group's +~50 M); the spool file per block of the design is not built yet. + +Next: albums in the edge (design 3), block leases and the planner (then +the store keys by the leased block and the lease columns leave `ranges`; +until then every range end leaves provisional edges), the block merger +(joins and finishes the edges), the prober writing `release_texts` kind 2, +the corpus replay harness. ## Decisions diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index a916cb5..c7f9b6d 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -198,6 +198,7 @@ class BatchStore { has_par2: r.hasPar2, finished: r.finished !== false, edge: r.edge === true, + partial: r.partial === true, sidecar_only: sidecarOnly(r), })) // Inserted, or a provisional release of the key finished (the chunk's files added up); a finished release @@ -207,11 +208,12 @@ class BatchStore { first_article, last_article, readable, has_par2, complete, edge, sidecar_only, settled_at) SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.group_set_id, v.poster_id, v.name, v.source, 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, + v.first_article, v.last_article, v.readable, v.has_par2, + CASE WHEN v.partial THEN false WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END, v.edge, v.sidecar_only, CASE WHEN v.finished THEN now() END FROM jsonb_to_recordset(?::jsonb) AS v(idx int, release_key text, identity_key text, name text, source int, poster_id int, group_set_id int, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean, - finished boolean, edge boolean, sidecar_only boolean) + finished boolean, edge boolean, partial boolean, sidecar_only 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 + CASE WHEN excluded.files > 0 THEN 1 ELSE 0 END, @@ -219,7 +221,8 @@ class BatchStore { file_total = coalesce(excluded.file_total, releases.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, - complete = CASE WHEN coalesce(excluded.file_total, releases.file_total) IS NULL THEN releases.complete + complete = CASE WHEN releases.complete IS FALSE AND releases.file_total IS NULL THEN false + WHEN coalesce(excluded.file_total, releases.file_total) IS NULL THEN releases.complete ELSE releases.files + excluded.files >= coalesce(excluded.file_total, releases.file_total) END, edge = excluded.edge, sidecar_only = coalesce(releases.sidecar_only, excluded.sidecar_only), settled_at = excluded.settled_at WHERE releases.settled_at IS NULL diff --git a/edge/Cargo.toml b/edge/Cargo.toml index 54c3e53..849cd6d 100644 --- a/edge/Cargo.toml +++ b/edge/Cargo.toml @@ -2,7 +2,7 @@ name = "edge" version = "0.1.0" edition = "2021" -description = "Voyager 2's edge worker (edge-node) and its benchmark (edge-bench): the RangeIndexer in Rust, kept identical to lib/edge (check.ts)" +description = "The edge worker (edge-node) and its benchmark (edge-bench): voyager 3's window indexing in Rust" [[bin]] name = "edge-bench" diff --git a/edge/README.md b/edge/README.md index 06e4de2..303e7b5 100644 --- a/edge/README.md +++ b/edge/README.md @@ -6,18 +6,21 @@ from the node API, reads their prefetched XOVER chunks, finds releases and posts batches. `edge-bench` (`src/main.rs`) replays a captured range through the same indexing (`src/lib.rs`) for timing and for comparing. -The TypeScript `RangeIndexer` (`lib/edge`, `node ace edge:node`) is the -reference: `check.ts` runs both on a capture and compares every output field -by field. An identity or rule change goes into both and has to come out -identical. CI (`.gitea/workflows/deploy.yaml`, job `edge`) runs `cargo test` -and `check-fixtures.sh`, which compares both on the committed captures in -`fixtures/` (25k overview lines each of a.b.models, a.b.ath and -a.b.boneless, in 5 chunks); a push to `rebuild` deploys only when they -match. +Since voyager 3 (2026-10-25) the Rust indexer is the one implementation of +the window (REBUILD.md, "Voyager 3: window indexing"; `src/indexer.rs`): +files and releases are finished when the frontier (the median Date of the +last 1,024 plausibly dated articles) is 24 h past their last part, or at +the block's end (provisionally, as edges), and written once. The +TypeScript `RangeIndexer` (`lib/edge`) keeps the range path for the corpus +scanner and the tests; the field-by-field comparison (`check.ts`) went +with the parity. CI (`.gitea/workflows/deploy.yaml`, job `edge`) runs +`cargo test` and `bench-fixtures.sh`, which replays the committed captures +in `fixtures/` (25k overview lines each of a.b.ath, a.b.boneless, a.b.fta +and a.b.models, in 5 chunks). ```sh cargo test --release --manifest-path edge/Cargo.toml -edge/check-fixtures.sh # from the repository root +edge/bench-fixtures.sh # from the repository root ``` In production since 2026-10-09: 122M headers in 10 minutes with 0.37 cores diff --git a/edge/bench-fixtures.sh b/edge/bench-fixtures.sh new file mode 100755 index 0000000..3fc035c --- /dev/null +++ b/edge/bench-fixtures.sh @@ -0,0 +1,11 @@ +#!/bin/sh +# The Rust indexer over the committed captures (edge/fixtures: 25k overview lines each of a.b.ath, a.b.boneless, +# a.b.fta and a.b.models, in 5 chunks): a smoke test of the window on real headers, exiting non-zero when the bench +# fails on one. CI runs it after cargo test; run it from the repository root after a change to the indexer. +set -eu +cd "$(dirname "$0")/.." +cargo build --release --quiet --manifest-path edge/Cargo.toml --bin edge-bench +for capture in edge/fixtures/*/; do + echo "== $(basename "$capture")" + edge/target/release/edge-bench "$capture" --batch-chunks 2 --idle 2 | tail -1 +done diff --git a/edge/check-fixtures.sh b/edge/check-fixtures.sh deleted file mode 100755 index 5bf4193..0000000 --- a/edge/check-fixtures.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/bin/sh -# The Rust and TypeScript RangeIndexers on the committed captures (edge/fixtures: 25k -# overview lines each of a.b.models, a.b.ath and a.b.boneless, in 5 chunks), compared -# by check.ts field by field: exits non-zero on any difference. CI runs it; run it from -# the repository root after any change to either side. -set -eu -cd "$(dirname "$0")/.." -cargo build --release --quiet --manifest-path edge/Cargo.toml --bin edge-bench -out=$(mktemp -d) -trap 'rm -rf "$out"' EXIT -status=0 -for capture in edge/fixtures/*/; do - name=$(basename "$capture") - edge/target/release/edge-bench "$capture" --batch-chunks 2 --idle 2 --dump "$out/$name.jsonl" --wire "$out/$name.wire.jsonl" >/dev/null 2>&1 - echo "== $name" - node --import=@poppinss/ts-exec edge/check.ts "$capture" "$out/$name.jsonl" --wire "$out/$name.wire.jsonl" --batch-chunks 2 --idle 2 || status=1 -done -exit $status diff --git a/edge/check.ts b/edge/check.ts deleted file mode 100644 index c7a3bf3..0000000 --- a/edge/check.ts +++ /dev/null @@ -1,139 +0,0 @@ -/** - * Compares edge-bench's output (--dump) with the TypeScript RangeIndexer - * on the same captured range, item by item: releases (every field, and the - * payload decoded with lib/chunks), samples, probe candidates, rule counts, - * fragments and batch boundaries. - * - * node --import=@poppinss/ts-exec edge/check.ts [--wire batches.jsonl] [--batch-chunks 10] [--idle 5] - * - * With --wire (edge-bench --wire): also each batch as the Rust worker - * posts it, against toWire of the TypeScript batch, field by field. - */ -import { readFile, readdir } from 'node:fs/promises' -import { join } from 'node:path' -import { unzipSync } from 'node:zlib' -import { isDeepStrictEqual } from 'node:util' -import { RangeIndexer, toWire, type EdgeBatch } from '#lib/edge/index' -import { DEFAULT_RULES, type Rules } from '#lib/rules/index' -import { decodeChunk } from '#lib/chunks/index' - -const [dir, dumpPath] = process.argv.slice(2) -const flag = (name: string, fallback: number) => { - const i = process.argv.indexOf(name) - return i === -1 ? fallback : Number(process.argv[i + 1]) -} -const batchChunks = flag('--batch-chunks', 10) -const idle = flag('--idle', 5) - -const { group, start, end } = JSON.parse(await readFile(join(dir, 'range.json'), 'utf8')) -const rules: Rules = { ...DEFAULT_RULES, priorities: { [group]: 1 }, droppedSignatures: [] } -const indexer = new RangeIndexer({ group, start, end, rules, idleChunks: idle }) -const files = (await readdir(dir)).filter((f) => f.endsWith('.gz')).sort() -const batches: EdgeBatch[] = [] -for (const [i, file] of files.entries()) { - const next = Number(file.split('-')[2].replace('.gz', '')) + 1 - indexer.addText(unzipSync(await readFile(join(dir, file))).toString('utf8'), next) - const last = i === files.length - 1 - if (last) indexer.finish() - if (last || (i + 1) % batchChunks === 0) batches.push(indexer.take()) -} - -/** Canonical lines per kind, in a stable order, from both sides. */ -type Item = Record -const sortFiles = (decoded: ReturnType) => [...decoded].sort((a, b) => (a.name + a.subject < b.name + b.subject ? -1 : 1)) -const payloadBytes = { ts: 0, rs: 0 } -let side: 'ts' | 'rs' = 'ts' -const release = (r: Item, b: number) => { - const { payload, ...rest } = r as Item & { payload: Buffer } - // The compressed size isn't compared (node's zstd and the zstd crate frame alike data a little - // differently); what the payload decodes to is. - payloadBytes[side] += payload.length - return { b, ...rest, decoded: sortFiles(decodeChunk(payload)) } -} -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, postedLow: batch.postedLow, postedHigh: batch.postedHigh }) - 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 }) -}) - -const rs: Record = { b: [], r: [], s: [], c: [], n: [], f: [] } -side = 'rs' -let b = -1 -for (const line of (await readFile(dumpPath, 'utf8')).split('\n')) { - if (!line) continue - 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, 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} ${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}`, -} -const names: Record = { b: 'batches', r: 'releases', s: 'samples', c: 'candidates', n: 'rule counts', f: 'fragments' } -let failed = false -for (const kind of Object.keys(keyOf)) { - const a = new Map(ts[kind].map((i) => [keyOf[kind](i), i])) - const z = new Map(rs[kind].map((i) => [keyOf[kind](i), i])) - const missing = [...a.keys()].filter((k) => !z.has(k)) - const extra = [...z.keys()].filter((k) => !a.has(k)) - const differ = [...a.keys()].filter((k) => z.has(k) && !isDeepStrictEqual(a.get(k), z.get(k))) - const ok = missing.length === 0 && extra.length === 0 && differ.length === 0 - failed ||= !ok - console.log(`${ok ? 'same' : 'DIFF'} ${names[kind]}: ${a.size} TypeScript, ${z.size} Rust${ok ? '' : `; ${missing.length} only in TypeScript, ${extra.length} only in Rust, ${differ.length} differ`}`) - for (const k of [...missing.slice(0, 2), ...extra.slice(0, 2)]) console.log(` only one side: ${k}`) - for (const k of differ.slice(0, 2)) { - const x = a.get(k)! - const y = z.get(k)! - const fields = Object.keys(x).filter((f) => !isDeepStrictEqual(x[f], y[f])) - console.log(` ${k}: ${fields.map((f) => `${f}: ${JSON.stringify(x[f]).slice(0, 160)} vs ${JSON.stringify(y[f]).slice(0, 160)}`).join('; ')}`) - } -} -console.log(`payload bytes: ${payloadBytes.ts} TypeScript, ${payloadBytes.rs} Rust`) - -const wireIndex = process.argv.indexOf('--wire') -if (wireIndex !== -1) { - // Order within a batch doesn't count; payloads by what they decode to. - const by = (f: (x: any) => string) => (a: any, b: any) => (f(a) < f(b) ? -1 : f(a) > f(b) ? 1 : 0) - const normal = (w: any) => ({ - ...w, - releases: w.releases.map((r: any) => ({ ...r, payload: sortFiles(decodeChunk(Buffer.from(r.payload, 'base64'))) })).sort(by((r) => `${r.key} ${r.upload}`)), - samples: w.samples.map((r: any) => ({ ...r, payload: sortFiles(decodeChunk(Buffer.from(r.payload, 'base64'))) })).sort(by((r) => `${r.key} ${r.upload}`)), - candidates: [...w.candidates].sort(by((c) => JSON.stringify(c))), - counts: [...w.counts].sort(by((c) => `${c.rule} ${c.key}`)), - fragments: [...w.fragments].sort(by((f) => `${f.fileKey} ${f.segments[0]?.[1]}`)), - }) - const rust = (await readFile(process.argv[wireIndex + 1], 'utf8')).split('\n').filter(Boolean).map((l) => normal(JSON.parse(l))) - const typescript = batches.map((b) => normal(JSON.parse(JSON.stringify(toWire(b))))) - let wireSame = rust.length === typescript.length - const firstDiff = (x: any, y: any, path: string): string | null => { - if (isDeepStrictEqual(x, y)) return null - if (typeof x !== 'object' || typeof y !== 'object' || x === null || y === null) return `${path}: ${JSON.stringify(x)?.slice(0, 120)} vs ${JSON.stringify(y)?.slice(0, 120)}` - for (const k of new Set([...Object.keys(x), ...Object.keys(y)])) { - const d = firstDiff(x[k], y[k], `${path}.${k}`) - if (d) return d - } - return path - } - for (let i = 0; i < Math.max(rust.length, typescript.length); i++) { - const d = firstDiff(typescript[i], rust[i], `batch ${i}`) - if (d) { - wireSame = false - console.log(`DIFF wire: ${d}`) - break - } - } - if (wireSame) console.log(`same wire batches: ${typescript.length} TypeScript, ${rust.length} Rust`) - failed ||= !wireSame -} -process.exit(failed ? 1 : 0) diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index f067bd4..50a9091 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -1,6 +1,7 @@ -//! lib/edge/scan.ts (SegmentScanner, scanOverviewText), lib/files/aggregator.ts -//! and lib/edge/range_indexer.ts (RangeIndexer, chunkOf, candidatesOf), in the -//! same order of operations, so the outputs can be compared one to one. +//! The edge's indexing: lib/edge/scan.ts (SegmentScanner, scanOverviewText) and lib/files/aggregator.ts as in the +//! TypeScript, then voyager 3's window (REBUILD.md, "Voyager 3: window indexing", 2): files and releases are +//! finished when the frontier (the median Date of the last FRONTIER_ARTICLES plausible ones) is WINDOW_MS past +//! their last part, or at the block's end (provisionally, as edges), and written once. use crate::codec::{encode_chunk, ChunkFile}; use crate::js::{self, sha1_hex}; @@ -46,6 +47,12 @@ pub struct ReleaseChunk { pub last_article: i64, pub readable: bool, pub has_par2: bool, + /// The window closed on it (false: still open at the block's end, written provisionally). + pub finished: bool, + /// Open at the block's end, or within the window of its start: the block merger's. + pub edge: bool, + /// A file misses parts. + pub partial: bool, pub payload: Vec, /// The first files' subjects (at most SUBJECTS): edge-bench's --releases dump for the grouping sweeps. pub subjects: Vec, @@ -61,18 +68,6 @@ pub struct Candidate { pub file: String, } -pub struct Fragment { - pub file_key: String, - pub name: String, - pub poster: String, - pub subject: String, - pub total_parts: u32, - pub groups: Vec, - pub release: Release, - pub posted_at: Option, - pub segments: Vec<(u32, String, i64, i64)>, -} - #[derive(Default)] pub struct Batch { pub from: i64, @@ -86,7 +81,6 @@ pub struct Batch { pub candidates: Vec, /// (rule, key) → (articles, files) pub counts: Vec<(&'static str, String, u64, u64)>, - pub fragments: Vec, } struct Counter { @@ -470,8 +464,108 @@ 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). +/// A re-upload's copy of a file is kept apart (FileAggregator, uploadGapMs); a file joins the open release of its +/// key posted within this gap of it, else starts one (uploadsOf, online). pub const UPLOAD_GAP_MS: i64 = 3 * 60 * 60 * 1000; +/// The window (REBUILD.md, "Voyager 3: window indexing", 2): a file still missing parts, and a release, are finished +/// once the frontier is this far past their last part. +pub const WINDOW_MS: i64 = 24 * 60 * 60 * 1000; +/// The frontier is the median Date of the last this many plausibly dated articles: enough that a minority of skewed +/// dates can't move it, few enough that it trails the newest article by little even in a quiet group (half the +/// ring: a day at 500 articles a day). +pub const FRONTIER_ARTICLES: usize = 1024; +/// A Date further than this from the frontier isn't believed (clock skew, bogus years); FRONTIER_ARTICLES of them +/// in a row that agree among themselves move it (the group paused for longer). +pub const PLAUSIBLE_MS: i64 = 7 * 24 * 60 * 60 * 1000; +/// The frontier is established once this many dated articles were seen (a lone bogus date doesn't set it). +const FRONTIER_MIN: usize = 64; + +fn median(dates: &mut [i64]) -> i64 { + let mid = dates.len() / 2; + *dates.select_nth_unstable(mid).1 +} + +/// Where the block is in posting time: the median Date of the last FRONTIER_ARTICLES plausible ones. Date headers +/// alone aren't trusted: a date more than PLAUSIBLE_MS from the frontier is set aside, and only FRONTIER_ARTICLES +/// such dates in a row, agreeing among themselves, move the frontier there. +#[derive(Default)] +pub struct Frontier { + ring: Vec, + at: usize, + value: Option, + /// The frontier when first established: the block's start in posting time (releases within WINDOW_MS of it are edges). + start: Option, + /// Dates set aside since the last plausible one. + aside: Vec, + pending: usize, +} + +impl Frontier { + pub fn add(&mut self, t: i64) { + if let Some(f) = self.value { + if (t - f).abs() > PLAUSIBLE_MS { + self.aside.push(t); + if self.aside.len() >= FRONTIER_ARTICLES { + let m = median(&mut self.aside.clone()); + if self.aside.iter().filter(|&&d| (d - m).abs() <= PLAUSIBLE_MS).count() * 2 >= self.aside.len() { + self.ring = self.aside.iter().copied().filter(|&d| (d - m).abs() <= PLAUSIBLE_MS).collect(); + self.ring.truncate(FRONTIER_ARTICLES); + self.at = self.ring.len() % FRONTIER_ARTICLES; + self.value = Some(m); + } + self.aside.clear(); + } + return; + } + self.aside.clear(); + } + if self.ring.len() < FRONTIER_ARTICLES { + self.ring.push(t); + } else { + self.ring[self.at] = t; + } + self.at = (self.at + 1) % FRONTIER_ARTICLES; + self.pending += 1; + } + + /// Recomputes the frontier (once per chunk: a median of FRONTIER_ARTICLES isn't for every article). + pub fn update(&mut self) { + if self.pending == 0 || (self.value.is_none() && self.ring.len() < FRONTIER_MIN) { + return; + } + self.pending = 0; + let m = median(&mut self.ring.clone()); + self.value = Some(m); + if self.start.is_none() { + self.start = Some(m); + } + } + + /// At the block's end: whatever dates there are set the frontier (a small block never reached FRONTIER_MIN). + pub fn settle(&mut self) { + if self.value.is_none() && !self.ring.is_empty() { + self.pending = 1; + let m = median(&mut self.ring.clone()); + self.value = Some(m); + self.start = Some(m); + } else { + self.update(); + } + } + + pub fn value(&self) -> Option { + self.value + } + + pub fn start(&self) -> Option { + self.start + } + + /// The posting time before which a last part means the window closed. + fn closed_before(&self) -> Option { + self.value.map(|f| f - WINDOW_MS) + } +} struct OpenFile { parts: Vec, @@ -570,105 +664,97 @@ impl Aggregator { complete } - /// 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> { + /// Files the window closed on, still missing parts: those whose last part is before `closed_before`, undated ones + /// idle for `idle` batches (all when `all`). Their segments, part order. + fn evict(&mut self, closed_before: Option, idle: u64, all: bool) -> Vec> { let batch = self.batch; 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 { + let c = &copies[i]; + let closed = match c.latest { + Some(l) => closed_before.is_some_and(|before| l < before), + None => idle > 0 && batch - c.touched >= idle, + }; + if !all && !closed { i += 1; continue; } - gone.extend(copies.remove(i).parts); + let mut f = copies.remove(i); + f.parts.sort_by_key(|s| s.number); + out.push(f.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(); +/// A release still open in the window: the files of one upload of its key (one after another, as they finished). +struct OpenRelease { + upload: usize, + files: Vec>, + first: Option, + last: Option, + touched: u64, +} + +impl OpenRelease { + /// How far a file's posting span is from the release's (0 when either is undated). + fn distance(&self, first: Option, last: Option) -> i64 { + match (first, last, self.first, self.last) { + (Some(ff), Some(fl), Some(rf), Some(rl)) => 0.max(rf - fl).max(ff - rl), + _ => 0, } - 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, rules: Rules, - start: i64, end: i64, priorities: FxHashMap, idle: u64, - chunk_articles: i64, scanner: Scanner, aggregator: Aggregator, + frontier: Frontier, + /// Per release key its open uploads, oldest first. + open: FxHashMap>, + /// The next upload number per key (distinct within a batch: the store's slot is the key and the upload). + uploads: FxHashMap, batch: Batch, counts: FxHashMap<(&'static str, String), (u64, u64)>, count_order: Vec<(&'static str, String)>, - completed: FxHashMap>>, - completed_order: Vec, /// Posting times of this batch's segments (posted_span at take). dates: Vec, + pending_files: usize, } impl RangeIndexer { - /// `resume_at`: where this worker starts (a range taken over midway); its first batch starts there. - pub fn new(group: &str, start: i64, end: i64, resume_at: i64, rules: Rules, idle: u64) -> Self { + /// `resume_at`: where this worker starts (a range taken over midway); its first batch starts there. `idle`: batches + /// (chunks) after which an undated file or release is given up. + pub fn new(group: &str, _start: i64, end: i64, resume_at: i64, rules: Rules, idle: u64) -> Self { RangeIndexer { group: group.to_string(), - start, end, priorities: rules.priorities.clone(), rules, idle, - chunk_articles: 0, scanner: Scanner::default(), aggregator: Aggregator::default(), + frontier: Frontier::default(), + open: FxHashMap::default(), + uploads: FxHashMap::default(), batch: Batch { from: resume_at, next: resume_at, ..Default::default() }, counts: FxHashMap::default(), count_order: Vec::new(), - completed: FxHashMap::default(), - completed_order: Vec::new(), dates: Vec::new(), + pending_files: 0, } } @@ -699,7 +785,10 @@ impl RangeIndexer { }); } self.batch.headers += headers; - self.dates.extend(segments.iter().filter_map(|s| s.posted_at)); + for t in segments.iter().filter_map(|s| s.posted_at) { + self.dates.push(t); + self.frontier.add(t); + } for owner in crossposts { self.count("crosspost", owner, 1, 0); } @@ -714,31 +803,36 @@ impl RangeIndexer { let done = self.aggregator.add(segments); unsafe { PROF[3] += t.elapsed().as_nanos() as u64; } self.files(done); - if self.chunk_articles == 0 { - self.chunk_articles = next - self.batch.from; - } + self.frontier.update(); let t = std::time::Instant::now(); - let gone = self.aggregator.evict(self.idle); - self.cut_off(gone, false); + let gone = self.aggregator.evict(self.frontier.closed_before(), self.idle, false); + self.files(gone); + self.close(false); unsafe { PROF[4] += t.elapsed().as_nanos() as u64; } self.batch.next = next; } + /// The block's end: every file and release left is written, finished when the window closed on it, else + /// provisionally and as an edge (the block merger finishes it with the neighbour block's part). pub fn finish(&mut self) { - let gone = self.aggregator.evict(0); - self.cut_off(gone, true); + self.frontier.settle(); + let gone = self.aggregator.evict(None, 0, true); + self.files(gone); + self.close(true); self.batch.next = self.end + 1; } - /// Files completed and fragments cut since the last take (RangeIndexer.pending). + /// Files in the releases finished since the last take (RangeIndexer.pending). pub fn pending(&self) -> usize { - self.batch.fragments.len() + self.completed.values().map(|l| l.len()).sum::() + self.pending_files + } + + /// Where the block is in posting time (None until FRONTIER_MIN dated articles were read). + pub fn frontier(&self) -> Option { + self.frontier.value() } pub fn take(&mut self) -> Batch { - let t = std::time::Instant::now(); - self.judge_releases(); - unsafe { PROF[5] += t.elapsed().as_nanos() as u64; } self.scanner.files.clear(); let next = self.batch.next; let mut out = std::mem::replace(&mut self.batch, Batch { from: next, next, ..Default::default() }); @@ -747,87 +841,115 @@ impl RangeIndexer { let (articles, files) = self.counts.remove(&id).unwrap(); out.counts.push((id.0, id.1, articles, files)); } + self.pending_files = 0; + // Upload numbers only have to be distinct within a batch: those of keys with nothing open go. + let open = &self.open; + self.uploads.retain(|key, _| open.contains_key(key)); out } + /// Finished files (complete, or the window closed on them) into the open release of their key and upload. fn files(&mut self, files: Vec>) { + let batch = self.aggregator.batch; for file in files { - let key = release_key_of(&file[0]); - match self.completed.get_mut(&key) { - Some(list) => list.push(file), - None => { - self.completed_order.push(key.clone()); - self.completed.insert(key, vec![file]); - } - } - } - } - - fn judge_releases(&mut self) { - let order = std::mem::take(&mut self.completed_order); - let mut by_release = std::mem::take(&mut self.completed); - for key in order { - 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, poster: &f[0].file.poster }) - .collect(); - match names::judge(&key, &judged, &self.rules) { - 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)); - } - } - } - } - } - } - - fn cut_off(&mut self, files: Vec>, at_end: bool) { - let window = self.idle as i64 * self.chunk_articles.max(1); - for list in files { - if list.len() == 1 && is_random_file(&list[0].file.name) { - let tool = names::upload_tool(&list[0].message_id); + if file.len() == 1 && (file[0].number != 1 || file[0].file.total_parts > 1) && is_random_file(&file[0].file.name) { + // One article of a multi-part random file (its upload per article: ngPost's, nyuu's): not a release. + let tool = names::upload_tool(&file[0].message_id); self.count("per-article", tool.to_string(), 1, 0); continue; } - let min_part = list.iter().map(|s| s.number).min().unwrap(); - let max_part = list.iter().map(|s| s.number).max().unwrap(); - let first_article = list.iter().map(|s| s.article).min().unwrap(); - let cut_at_start = min_part > 1 && first_article < self.start + window; - let cut_at_end = at_end && max_part < list[0].file.total_parts; - if !cut_at_start && !cut_at_end { - self.count("incomplete", String::new(), list.len() as u64, 1); - continue; - } - let f = &list[0].file; - let mut fragment = Fragment { - file_key: f.file_key.clone(), - name: f.name.clone(), - poster: f.poster.clone(), - subject: f.subject.clone(), - total_parts: f.total_parts, - groups: f.groups.clone(), - release: f.release.clone(), - posted_at: None, - segments: Vec::with_capacity(list.len()), + let key = release_key_of(&file[0]); + let first = posted_of(&file); + let last = file.iter().filter_map(|s| s.posted_at).max(); + let uploads = self.open.entry(key.clone()).or_default(); + let near = uploads.iter_mut().filter(|u| u.distance(first, last) <= UPLOAD_GAP_MS).min_by_key(|u| u.distance(first, last)); + let release = match near { + Some(u) => u, + None => { + let n = self.uploads.entry(key).or_insert(0); + let upload = *n; + *n += 1; + uploads.push(OpenRelease { upload, files: Vec::new(), first: None, last: None, touched: batch }); + uploads.last_mut().unwrap() + } }; - for s in &list { - fragment.segments.push((s.number, s.message_id.clone(), s.bytes, s.article)); - if let Some(t) = s.posted_at { - if fragment.posted_at.is_none_or(|p| t < p) { - fragment.posted_at = Some(t); - } + if let Some(f) = first { + if release.first.is_none_or(|r| f < r) { + release.first = Some(f); + } + } + if let Some(l) = last { + if release.last.is_none_or(|r| l > r) { + release.last = Some(l); + } + } + release.touched = batch; + release.files.push(file); + } + } + + /// Releases the window closed on (their last part before the frontier less WINDOW_MS; undated ones idle for + /// `idle` batches): judged and written. At the end (`all`), the rest too: provisionally, as edges. + fn close(&mut self, all: bool) { + let before = self.frontier.closed_before(); + let start = self.frontier.start(); + let batch = self.aggregator.batch; + let idle = self.idle; + let mut done: Vec<(String, OpenRelease, bool)> = Vec::new(); + let keys: Vec = self.open.keys().cloned().collect(); + for key in keys { + let uploads = self.open.get_mut(&key).unwrap(); + let mut i = 0; + while i < uploads.len() { + let u = &uploads[i]; + let closed = match u.last { + Some(l) => before.is_some_and(|b| l < b), + None => idle > 0 && batch - u.touched >= idle, + }; + if !all && !closed { + i += 1; + continue; + } + done.push((key.clone(), uploads.remove(i), closed)); + } + if uploads.is_empty() { + self.open.remove(&key); + } + } + for (key, u, closed) in done { + // An edge: still open at the block's end, or posted within the window of the block's start (its other + // part may be in the neighbour block); undated ones can't be placed, so they are edges too. + let near_start = match (start, u.first) { + (Some(s), Some(f)) => f < s + WINDOW_MS, + _ => true, + }; + self.emit(&key, u, closed, !closed || near_start); + } + } + + fn emit(&mut self, key: &str, release: OpenRelease, finished: bool, edge: bool) { + let list = release.files; + 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, poster: &f[0].file.poster }) + .collect(); + match names::judge(key, &judged, &self.rules) { + Verdict::Keep => { + let candidates = candidates_of(key, &list, release.upload); + let mut chunk = chunk_of(key, &list, release.upload); + chunk.finished = finished; + chunk.edge = edge; + self.pending_files += chunk.files; + self.batch.releases.push(chunk); + 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, release.upload), rule, rule_key)); } } - self.batch.fragments.push(fragment); } } @@ -904,6 +1026,9 @@ fn chunk_of(key: &str, files: &[Vec], upload: usize) -> ReleaseChunk { last_article, readable: chunk_files.iter().any(|f| !is_random_file(f.name)), has_par2: chunk_files.iter().any(|f| names::is_par2(f.name)), + finished: true, + edge: false, + partial: files.iter().any(|f| (f.len() as u32) < f[0].file.total_parts), payload: encode_chunk(&chunk_files), subjects: chunk_files.iter().take(SUBJECTS).map(|f| f.subject.to_string()).collect(), } @@ -968,6 +1093,83 @@ mod tests { 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); + // A day of posting in all: the window never closed on them, so at the end they are provisional edges. + assert!(batch.releases.iter().all(|r| !r.finished && r.edge && !r.partial)); + } + + /// `n` single-part filler files posted `minutes` apart from `start`: dated articles that move the frontier. + fn fillers(from: i64, n: i64, start: i64, minutes: i64) -> Vec { + (0..n).map(|i| line(from + i, &format!(r#"Filler.{} [1/1] - "filler.{}.jpg" yEnc (1/1)"#, from + i, from + i), start + i * minutes)).collect() + } + + #[test] + fn the_window_finishes_releases_and_keeps_incomplete_files() { + let mut priorities = FxHashMap::default(); + priorities.insert(GROUP.to_string(), 1); + let mut ix = RangeIndexer::new(GROUP, 1, 1_000_000, 1, Rules::defaults(priorities), 5); + let day = 24 * 60; + // Chunk 1: a release at the block's start (two of three parts of its file), and 100 dated articles after it. + let mut lines = vec![ + line(10, r#"Early.Show [1/1] - "Early.Show.mkv" yEnc (1/3)"#, 0), + line(11, r#"Early.Show [1/1] - "Early.Show.mkv" yEnc (2/3)"#, 1), + ]; + lines.extend(fillers(100, 100, 2, 1)); + ix.add_text(&lines.join(" +"), 1000); + assert!(ix.frontier().is_some()); + let b1 = ix.take(); + assert!(b1.releases.iter().all(|r| r.name.starts_with("Filler"))); + // Chunk 2, three days on (more articles than before, so the median moves there): the window closed on the + // early show (its file still short a part) and on the first fillers. + let mut lines = vec![line(2000, r#"Later.Show [1/1] - "Later.Show.mkv" yEnc (1/1)"#, 3 * day)]; + lines.extend(fillers(2100, 300, 3 * day + 1, 1)); + ix.add_text(&lines.join(" +"), 3000); + let b2 = ix.take(); + let early = b2.releases.iter().find(|r| r.name == "Early Show").expect("the early show is written"); + assert!(early.finished && early.edge && early.partial, "finished {} edge {} partial {}", early.finished, early.edge, early.partial); + assert_eq!((early.files, early.segments), (1, 2)); + assert!(!b2.releases.iter().any(|r| r.name == "Later Show")); + // Chunk 3, three more days: the later show is finished, and not an edge (a day past the block's start). + ix.add_text(&fillers(4000, 500, 6 * day, 1).join(" +"), 5000); + let b3 = ix.take(); + let later = b3.releases.iter().find(|r| r.name == "Later Show").expect("the later show is written"); + assert!(later.finished && !later.edge && !later.partial); + // The end: the last fillers are still inside the window, so they are provisional edges. + ix.finish(); + let b4 = ix.take(); + assert_eq!(b4.releases.len(), 500); + assert!(b4.releases.iter().all(|r| !r.finished && r.edge)); + assert_eq!(b4.next, 1_000_001); + } + + #[test] + fn the_frontier_ignores_implausible_dates_and_follows_a_pause() { + let hour = 3_600_000i64; + let mut f = Frontier::default(); + for i in 0..200 { + f.add(i * hour / 10); + } + f.update(); + let settled = f.value().unwrap(); + assert!((settled - 10 * hour).abs() <= hour); + assert_eq!(f.start(), Some(settled)); + // A bogus year, and one a year back: set aside. + for _ in 0..100 { + f.add(4_000_000_000_000); + f.add(-400 * 24 * hour); + } + f.update(); + assert_eq!(f.value(), Some(settled)); + // The group paused for a month: FRONTIER_ARTICLES agreeing dates move the frontier there. + let later = 30 * 24 * hour; + for i in 0..FRONTIER_ARTICLES as i64 { + f.add(later + i * 1000); + } + f.update(); + assert!(f.value().unwrap() >= later); + assert_eq!(f.start(), Some(settled)); } #[test] diff --git a/edge/src/main.rs b/edge/src/main.rs index 2e56643..b025212 100644 --- a/edge/src/main.rs +++ b/edge/src/main.rs @@ -123,7 +123,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { line.push_str(",\"postedLast\":"); opt_num(r.posted_last, line); line.push_str(&format!( - ",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"payload\":\"{}\"", + ",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"finished\":{},\"edge\":{},\"partial\":{},\"payload\":\"{}\"", r.files, r.segments, r.bytes, @@ -131,6 +131,9 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { r.last_article, r.readable, r.has_par2, + r.finished, + r.edge, + r.partial, b64.encode(&r.payload) )); }; @@ -162,25 +165,6 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { line.push_str(&format!(",\"articles\":{articles},\"files\":{files}}}")); emit(&mut line); } - for f in &batch.fragments { - line.push_str("{\"t\":\"f\",\"fileKey\":"); - json_str(&f.file_key, &mut line); - line.push_str(",\"name\":"); - json_str(&f.name, &mut line); - line.push_str(&format!(",\"totalParts\":{},\"postedAt\":", f.total_parts)); - opt_num(f.posted_at, &mut line); - line.push_str(",\"segments\":["); - for (i, (n, id, bytes, article)) in f.segments.iter().enumerate() { - if i > 0 { - line.push(','); - } - line.push_str(&format!("[{n},")); - json_str(id, &mut line); - line.push_str(&format!(",{bytes},{article}]")); - } - line.push_str("]}"); - emit(&mut line); - } } fn main() { @@ -208,7 +192,7 @@ fn main() { let (mut inflate_s, mut index_s) = (0.0, 0.0); let (mut compressed, mut text_bytes) = (0usize, 0usize); - let mut out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, fragments, samples + let mut out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, samples let mut keys = std::collections::HashSet::new(); let mut decompressor = libdeflater::Decompressor::new(); let wall = Instant::now(); @@ -242,8 +226,7 @@ fn main() { out.4 += r.payload.len(); } out.5 += batch.candidates.len(); - out.6 += batch.fragments.len(); - out.7 += batch.samples.len(); + out.6 += batch.samples.len(); if let Some(d) = dump.as_mut() { dump_batch(&batch, d); } @@ -269,7 +252,7 @@ fn main() { ); println!(" inflate {inflate_s:.2} s, index {index_s:.2} s; peak RSS {} MB", peak_rss_mb()); println!( - " kept: {} release chunks ({} releases), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} fragments, {} samples", + " kept: {} release chunks ({} releases), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} samples", out.1, keys.len(), out.2, @@ -277,7 +260,6 @@ fn main() { out.4 as f64 / 1e6, out.4 as f64 / out.3.max(1) as f64, out.5, - out.6, - out.7 + out.6 ); } diff --git a/edge/src/wire.rs b/edge/src/wire.rs index 4508c3f..7e1b940 100644 --- a/edge/src/wire.rs +++ b/edge/src/wire.rs @@ -1,7 +1,7 @@ //! A batch as the coordinator takes it (lib/edge/wire.ts toWire): JSON with //! the TypeScript names, chunk payloads in base64. -use crate::indexer::{Batch, Candidate, Fragment, ReleaseChunk}; +use crate::indexer::{Batch, Candidate, ReleaseChunk}; use base64::Engine; use serde::Serialize; @@ -24,6 +24,9 @@ struct WireRelease<'a> { last_article: i64, readable: bool, has_par2: bool, + finished: bool, + edge: bool, + partial: bool, payload: String, #[serde(skip_serializing_if = "Option::is_none")] rule: Option<&'a str>, @@ -49,30 +52,6 @@ struct WireCount<'a> { files: u64, } -#[derive(Serialize)] -#[serde(rename_all = "camelCase")] -struct WireFragmentRelease<'a> { - key: &'a str, - name: &'a str, - source: &'a str, - #[serde(skip_serializing_if = "Option::is_none")] - file_total: Option, -} - -#[derive(Serialize)] -#[serde(rename_all = "camelCase")] -struct WireFragment<'a> { - file_key: &'a str, - name: &'a str, - poster: &'a str, - subject: &'a str, - total_parts: u32, - groups: &'a [String], - release: Option>, - posted_at: Option, - segments: Vec<(u32, &'a str, i64, i64)>, -} - #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct WireBatch<'a> { @@ -85,7 +64,6 @@ struct WireBatch<'a> { samples: Vec>, candidates: Vec>, counts: Vec>, - fragments: Vec>, } fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRelease<'a> { @@ -106,6 +84,9 @@ fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRel last_article: r.last_article, readable: r.readable, has_par2: r.has_par2, + finished: r.finished, + edge: r.edge, + partial: r.partial, payload: base64::engine::general_purpose::STANDARD.encode(&r.payload), rule: rule.map(|(rule, _)| rule), rule_key: rule.map(|(_, key)| key), @@ -116,21 +97,6 @@ fn candidate(c: &Candidate) -> WireCandidate<'_> { WireCandidate { release_key: &c.release_key, upload: c.upload, kind: c.kind, message_id: &c.message_id, file: &c.file } } -fn fragment(f: &Fragment) -> WireFragment<'_> { - WireFragment { - file_key: &f.file_key, - name: &f.name, - poster: &f.poster, - subject: &f.subject, - total_parts: f.total_parts, - groups: &f.groups, - // A named file always has a release; fileTotal only when truthy, as fragmentOf. - release: Some(WireFragmentRelease { key: &f.release.key, name: &f.release.name, source: f.release.source, file_total: f.release.file_total_truthy() }), - posted_at: f.posted_at, - segments: f.segments.iter().map(|(n, id, bytes, article)| (*n, id.as_str(), *bytes, *article)).collect(), - } -} - /// toWire(batch) as JSON bytes. pub fn to_json(batch: &Batch) -> Vec { let wire = WireBatch { @@ -143,7 +109,6 @@ pub fn to_json(batch: &Batch) -> Vec { samples: batch.samples.iter().map(|(r, rule, key)| release(r, Some((rule, key.as_str())))).collect(), candidates: batch.candidates.iter().map(candidate).collect(), counts: batch.counts.iter().map(|(rule, key, articles, files)| WireCount { rule, key, articles: *articles, files: *files }).collect(), - fragments: batch.fragments.iter().map(fragment).collect(), }; serde_json::to_vec(&wire).unwrap() } diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index 68f1938..63e10fe 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -31,8 +31,10 @@ export interface ReleaseChunk { * later chunk of the same release key (the totals then are the release's; the payload only the files not sent yet). */ finished?: boolean - /** Still open at its block's end: the block merger joins it with the neighbour block's part. */ + /** Still open at its block's end, or posted within the window of its start: the block merger joins it with the neighbour block's part. */ edge?: boolean + /** A file of the release misses parts (the window closed on it): not complete, whatever the counter says. */ + partial?: boolean payload: Buffer } @@ -95,7 +97,8 @@ export interface EdgeBatch { samples: DroppedChunk[] candidates: ProbeCandidate[] counts: RuleCount[] - fragments: Fragment[] + /** The range path's cut-off files (the TypeScript indexer); the Rust edge's window has none, the store ignores them. */ + fragments?: Fragment[] } export interface RangeIndexerOptions { @@ -210,7 +213,7 @@ export class RangeIndexer { /** Files completed and fragments cut since the last take: how big the next batch is. */ get pending(): number { - let files = this.batch.fragments.length + let files = this.batch.fragments!.length for (const list of this.completed.values()) files += list.length return files } @@ -294,7 +297,7 @@ export class RangeIndexer { fragment.segments.push([s.number, s.messageId, s.bytes, s.articleNumber]) if (Number.isFinite(s.postedAt) && (fragment.postedAt === null || s.postedAt < fragment.postedAt)) fragment.postedAt = s.postedAt } - this.batch.fragments.push(fragment) + this.batch.fragments!.push(fragment) } } diff --git a/tests/unit/edge/range_indexer.spec.ts b/tests/unit/edge/range_indexer.spec.ts index e85489c..655779e 100644 --- a/tests/unit/edge/range_indexer.spec.ts +++ b/tests/unit/edge/range_indexer.spec.ts @@ -83,7 +83,7 @@ test.group('edge / RangeIndexer', () => { indexer.add(post(999_990, 'End.Cut.2020.part1.rar', 3).slice(0, 2), 1_000_001) indexer.finish() const batch = indexer.take() - assert.deepEqual(batch.fragments.map((f) => [f.name, f.segments.map((s) => s[0])]).sort(), [['End.Cut.2020.part1.rar', [1, 2]], ['Start.Cut.2020.part1.rar', [2, 3]]]) + assert.deepEqual(batch.fragments!.map((f) => [f.name, f.segments.map((s) => s[0])]).sort(), [['End.Cut.2020.part1.rar', [1, 2]], ['Start.Cut.2020.part1.rar', [2, 3]]]) assert.deepEqual(batch.counts.map((c) => [c.rule, c.articles, c.files]), [['incomplete', 2, 1]]) assert.lengthOf(batch.releases, 0) })