edge-bench-rs: a library for the worker; rules from the lease; the wire format

- src/lib.rs: the indexing modules as a library, for edge-bench-rs and
  the coming edge-node worker.
- Rules as a lease brings them (priorities, dropped signatures and
  tools, per-article parts, sample share, executables) instead of
  DEFAULT_RULES hard-coded; RangeIndexer takes them and a resume point.
- Fragments carry everything the TypeScript Fragment has (poster,
  subject, groups, release): the server stores them for the merger.
- wire.rs: a batch as toWire makes it (camelCase JSON, payloads in
  base64). edge-bench-rs --wire writes it; check.ts --wire compares it
  with toWire of the TypeScript batch: identical on models-a and
  ath-2013.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tools/edge-bench-rs/src/indexer.rs

@@ -4,7 +4,7 @@
 
 use crate::codec::{encode_chunk, ChunkFile};
 use crate::js::{self, sha1_hex};
-use crate::names::{self, is_random_file, JudgedFile, Release, Verdict};
+use crate::names::{self, is_random_file, JudgedFile, Release, Rules, Verdict};
 use rustc_hash::{FxHashMap, FxHashSet};
 use std::rc::Rc;
 
@@ -60,7 +60,11 @@ pub struct Candidate {
 pub struct Fragment {
     pub file_key: String,
     pub name: String,
+    pub poster: String,
+    pub subject: String,
     pub total_parts: u32,
+    pub groups: Vec<String>,
+    pub release: Release,
     pub posted_at: Option<i64>,
     pub segments: Vec<(u32, String, i64, i64)>,
 }
@@ -441,6 +445,7 @@ fn uploads_of(files: Vec<Vec<Seg>>) -> Vec<Vec<Vec<Seg>>> {
 
 pub struct RangeIndexer {
     group: String,
+    rules: Rules,
     start: i64,
     end: i64,
     priorities: FxHashMap<String, u32>,
@@ -456,17 +461,19 @@ pub struct RangeIndexer {
 }
 
 impl RangeIndexer {
-    pub fn new(group: &str, start: i64, end: i64, priorities: FxHashMap<String, u32>, idle: u64) -> Self {
+    /// `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 {
         RangeIndexer {
             group: group.to_string(),
             start,
             end,
-            priorities,
+            priorities: rules.priorities.clone(),
+            rules,
             idle,
             chunk_articles: 0,
             scanner: Scanner::default(),
             aggregator: Aggregator::default(),
-            batch: Batch { from: start, next: start, ..Default::default() },
+            batch: Batch { from: resume_at, next: resume_at, ..Default::default() },
             counts: FxHashMap::default(),
             count_order: Vec::new(),
             completed: FxHashMap::default(),
@@ -531,6 +538,11 @@ impl RangeIndexer {
         self.batch.next = self.end + 1;
     }
 
+    /// Files completed and fragments cut since the last take (RangeIndexer.pending).
+    pub fn pending(&self) -> usize {
+        self.batch.fragments.len() + self.completed.values().map(|l| l.len()).sum::<usize>()
+    }
+
     pub fn take(&mut self) -> Batch {
         let t = std::time::Instant::now();
         self.judge_releases();
@@ -567,7 +579,7 @@ impl RangeIndexer {
                     .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) {
+                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));
@@ -603,7 +615,17 @@ impl RangeIndexer {
                 continue;
             }
             let f = &list[0].file;
-            let mut fragment = Fragment { file_key: f.file_key.clone(), name: f.name.clone(), total_parts: f.total_parts, posted_at: None, segments: Vec::with_capacity(list.len()) };
+            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()),
+            };
             for s in &list {
                 fragment.segments.push((s.number, s.message_id.clone(), s.bytes, s.article));
                 if let Some(t) = s.posted_at {
@@ -736,7 +758,7 @@ mod tests {
     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 ix = RangeIndexer::new(GROUP, 1, 10_000, 1, Rules::defaults(priorities), 5);
         let mut lines: Vec<String> = (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)));

tools/edge-bench-rs/src/lib.rs

@@ -0,0 +1,9 @@
+//! The edge worker's indexing (lib/edge RangeIndexer and what it uses) in
+//! Rust, kept identical to the TypeScript (check.ts): the edge-bench-rs
+//! benchmark and the edge-node worker share it.
+
+pub mod codec;
+pub mod indexer;
+pub mod js;
+pub mod names;
+pub mod wire;

tools/edge-bench-rs/src/main.rs

@@ -4,16 +4,12 @@
 //! idle ones, and every N chunks judge and encode the releases. `--dump`
 //! writes every output as JSON lines for tools/edge-bench-rs/check.ts.
 //!
-//!   edge-bench-rs <dir> [--batch-chunks 10] [--idle 5] [--dump out.jsonl]
-
-mod codec;
-mod indexer;
-mod js;
-mod names;
+//!   edge-bench-rs <dir> [--batch-chunks 10] [--idle 5] [--dump out.jsonl] [--wire batches.jsonl]
 
 use base64::Engine;
-use indexer::{Batch, RangeIndexer};
-use js::json_str;
+use edge_bench_rs::indexer::{self, Batch, RangeIndexer};
+use edge_bench_rs::js::json_str;
+use edge_bench_rs::{names, wire};
 use rustc_hash::FxHashMap;
 use std::io::Write;
 use std::time::Instant;
@@ -159,6 +155,8 @@ fn main() {
     let batch_chunks: usize = flag("--batch-chunks").map(|v| v.parse().unwrap()).unwrap_or(10);
     let idle: u64 = flag("--idle").map(|v| v.parse().unwrap()).unwrap_or(5);
     let mut dump = flag("--dump").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap()));
+    // Each batch as the worker posts it (wire.rs), one JSON per line: check.ts compares it with toWire.
+    let mut wire_out = flag("--wire").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap()));
 
     let range = std::fs::read_to_string(format!("{dir}/range.json")).unwrap();
     let group = json_field(&range, "group");
@@ -169,7 +167,7 @@ fn main() {
 
     let mut priorities = FxHashMap::default();
     priorities.insert(group.clone(), 1);
-    let mut indexer = RangeIndexer::new(&group, start, end, priorities, idle);
+    let mut indexer = RangeIndexer::new(&group, start, end, start, names::Rules::defaults(priorities), idle);
 
     let (mut inflate_s, mut index_s) = (0.0, 0.0);
     let (mut compressed, mut text_bytes) = (0usize, 0usize);
@@ -212,6 +210,10 @@ fn main() {
         if let Some(d) = dump.as_mut() {
             dump_batch(&batch, d);
         }
+        if let Some(w) = wire_out.as_mut() {
+            w.write_all(&wire::to_json(&batch)).unwrap();
+            w.write_all(b"\n").unwrap();
+        }
     }
     let total = inflate_s + index_s;
     #[allow(static_mut_refs)]

tools/edge-bench-rs/src/names.rs

@@ -806,12 +806,40 @@ pub enum Verdict {
     Drop { rule: &'static str, key: String, sample: bool },
 }
 
-/// RuleSet.judge with DEFAULT_RULES (no dropped signatures).
-pub fn judge(release_key: &str, files: &[JudgedFile]) -> Verdict {
-    const DROPPED_TOOLS: [&str; 5] = ["ngPost", "hex32", "yuun", "JBinUp", "random"];
-    let sample = explored(release_key, 1.0);
+/// The rules a lease brings (lib/rules Rules): what the worker drops.
+#[derive(Clone, Debug)]
+pub struct Rules {
+    pub version: i64,
+    /// Indexed groups and their priority (owner rule).
+    pub priorities: rustc_hash::FxHashMap<String, u32>,
+    /// "tool\tgroup"
+    pub dropped_signatures: rustc_hash::FxHashSet<String>,
+    pub dropped_tools: rustc_hash::FxHashSet<String>,
+    pub per_article_max_parts: u32,
+    pub sample_percent: f64,
+    pub drop_executables: bool,
+}
+
+impl Rules {
+    /// DEFAULT_RULES with these priorities and no dropped signatures (edge:bench's).
+    pub fn defaults(priorities: rustc_hash::FxHashMap<String, u32>) -> Self {
+        Rules {
+            version: 1,
+            priorities,
+            dropped_signatures: Default::default(),
+            dropped_tools: ["ngPost", "hex32", "yuun", "JBinUp", "random"].iter().map(|t| t.to_string()).collect(),
+            per_article_max_parts: 3,
+            sample_percent: 1.0,
+            drop_executables: true,
+        }
+    }
+}
+
+/// RuleSet.judge.
+pub fn judge(release_key: &str, files: &[JudgedFile], rules: &Rules) -> Verdict {
+    let sample = explored(release_key, rules.sample_percent);
     let content: Vec<&JudgedFile> = files.iter().filter(|f| !test(&SIDECAR, f.name)).collect();
-    if !content.is_empty() && content.iter().all(|f| test(&EXECUTABLE, f.name)) {
+    if rules.drop_executables && !content.is_empty() && content.iter().all(|f| test(&EXECUTABLE, f.name)) {
         let c = EXECUTABLE.captures(content[0].name.as_bytes()).unwrap();
         let key = std::str::from_utf8(&c[1]).unwrap().to_lowercase();
         return Verdict::Drop { rule: "executable", key, sample };
@@ -822,10 +850,10 @@ pub fn judge(release_key: &str, files: &[JudgedFile]) -> Verdict {
     let first = &files[0];
     let tool = upload_tool(first.first_message_id);
     let group = first.groups.first().map(String::as_str).unwrap_or("");
-    if DROPPED_TOOLS.contains(&tool) {
+    if rules.dropped_tools.contains(tool) || rules.dropped_signatures.contains(&format!("{tool}\t{group}")) {
         return Verdict::Drop { rule: "garbage", key: format!("{tool} {group}"), sample };
     }
-    if files.iter().all(|f| f.total_parts <= 3) {
+    if files.iter().all(|f| f.total_parts <= rules.per_article_max_parts) {
         return Verdict::Drop { rule: "per-article", key: tool.to_string(), sample };
     }
     Verdict::Keep

tools/edge-bench-rs/src/wire.rs

@@ -0,0 +1,144 @@
+//! 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 base64::Engine;
+use serde::Serialize;
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct WireRelease<'a> {
+    key: &'a str,
+    upload: usize,
+    name: &'a str,
+    source: &'a str,
+    file_total: Option<i64>,
+    poster: &'a str,
+    groups: &'a [String],
+    posted_at: Option<i64>,
+    posted_last: Option<i64>,
+    files: usize,
+    segments: usize,
+    bytes: i64,
+    first_article: i64,
+    last_article: i64,
+    readable: bool,
+    has_par2: bool,
+    payload: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    rule: Option<&'a str>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    rule_key: Option<&'a str>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct WireCandidate<'a> {
+    release_key: &'a str,
+    upload: usize,
+    kind: &'a str,
+    message_id: &'a str,
+    file: &'a str,
+}
+
+#[derive(Serialize)]
+struct WireCount<'a> {
+    rule: &'a str,
+    key: &'a str,
+    articles: u64,
+    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<i64>,
+}
+
+#[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<WireFragmentRelease<'a>>,
+    posted_at: Option<i64>,
+    segments: Vec<(u32, &'a str, i64, i64)>,
+}
+
+#[derive(Serialize)]
+struct WireBatch<'a> {
+    from: i64,
+    next: i64,
+    headers: u64,
+    releases: Vec<WireRelease<'a>>,
+    samples: Vec<WireRelease<'a>>,
+    candidates: Vec<WireCandidate<'a>>,
+    counts: Vec<WireCount<'a>>,
+    fragments: Vec<WireFragment<'a>>,
+}
+
+fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRelease<'a> {
+    WireRelease {
+        key: &r.key,
+        upload: r.upload,
+        name: &r.name,
+        source: r.source,
+        file_total: r.file_total,
+        poster: &r.poster,
+        groups: &r.groups,
+        posted_at: r.posted_at,
+        posted_last: r.posted_last,
+        files: r.files,
+        segments: r.segments,
+        bytes: r.bytes,
+        first_article: r.first_article,
+        last_article: r.last_article,
+        readable: r.readable,
+        has_par2: r.has_par2,
+        payload: base64::engine::general_purpose::STANDARD.encode(&r.payload),
+        rule: rule.map(|(rule, _)| rule),
+        rule_key: rule.map(|(_, key)| key),
+    }
+}
+
+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<u8> {
+    let wire = WireBatch {
+        from: batch.from,
+        next: batch.next,
+        headers: batch.headers,
+        releases: batch.releases.iter().map(|r| release(r, None)).collect(),
+        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()
+}