The replay harness: a group's corpus through one carried window into a scratch catalog

replay:corpus runs edge-bench --group over a group's complete corpus ranges
in article order (one window, carried from range to range, finished after
the last range; edge/src/lease.rs holds the lease JSON the worker and the
bench share) and stores every batch through BatchStore as the worker
would, then reports releases finished, provisional, complete and the bytes
per header of the NVMe and HDD tables. Refuses a database not named
*_replay or *_scratch. The app image carries edge-bench; compose gets a
replay profile for homelab.

Design amended (REBUILD.md 1, 2, 5): the window is carried by one worker a
group, no block leases and no block merger; provisional writes stay for
the lease ends.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

edge/src/main.rs

@@ -1,10 +1,11 @@
-//! The edge worker's indexing in Rust, on a range captured by `edge:bench
-//! --capture` (REBUILD.md, part 2: TypeScript or Rust). Same work as
-//! edge:bench's replay: inflate each chunk, scan it, aggregate files, cut off
-//! idle ones, and every N chunks judge and encode the releases. `--dump`
-//! writes every output as JSON lines for edge/check.ts.
+//! The Rust indexer over stored chunks (REBUILD.md, part 2 and "Voyager 3"): a range captured by `edge:bench
+//! --capture` or stored in the corpus, or a group's corpus ranges in order through one window (`--group`, the
+//! replay harness behind `replay:corpus`). Same work as the edge-node worker: inflate each chunk, scan it,
+//! aggregate files, finish releases through the window, and every N chunks judge and encode them. `--dump`
+//! writes every output as JSON lines; `--wire` the batches as the worker posts them (`-`: stdout).
 //!
-//!   edge-bench <dir> [--batch-chunks 10] [--idle 5] [--dump out.jsonl] [--wire batches.jsonl]
+//!   edge-bench <range dir> | --group <group dir> [--from article] [--limit ranges] [--rules rules.json]
+//!              [--batch-chunks 10] [--idle 5] [--no-finish] [--dump out.jsonl] [--wire batches.jsonl|-] [--releases out.jsonl]
 
 use base64::Engine;
 use edge::indexer::{self, Batch, RangeIndexer};
@@ -167,99 +168,210 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
     }
 }
 
+/// A stored range: its directory and article span (range.json).
+struct Range {
+    dir: String,
+    start: i64,
+    end: i64,
+}
+
+/// The range in a directory with a range.json, and its group.
+fn range_in(dir: &str) -> Option<(String, Range)> {
+    let json = std::fs::read_to_string(format!("{dir}/range.json")).ok()?;
+    Some((json_field(&json, "group"), Range { dir: dir.to_string(), start: json_field(&json, "start").parse().ok()?, end: json_field(&json, "end").parse().ok()? }))
+}
+
+/// A group directory's complete ranges in article order (the corpus layout: `<group>/<start>-<end>/`).
+fn ranges_in(group_dir: &str) -> (String, Vec<Range>) {
+    let mut group = String::new();
+    let mut ranges: Vec<Range> = Vec::new();
+    for entry in std::fs::read_dir(group_dir).unwrap_or_else(|e| panic!("{group_dir}: {e}")) {
+        let path = entry.unwrap().path();
+        if !path.is_dir() {
+            continue;
+        }
+        if let Some((g, range)) = range_in(path.to_str().unwrap()) {
+            group = g;
+            ranges.push(range);
+        }
+    }
+    ranges.sort_by_key(|r| r.start);
+    (group, ranges)
+}
+
+fn date(ms: Option<i64>) -> String {
+    match ms.and_then(chrono::DateTime::from_timestamp_millis) {
+        Some(t) => t.format("%Y-%m-%d %H:%M").to_string(),
+        None => "-".to_string(),
+    }
+}
+
+/// Files waiting in finished releases after which a batch goes out whatever its chunks (edge-node's limit).
+const MAX_PENDING_FILES: usize = 100_000;
+
 fn main() {
     let args: Vec<String> = std::env::args().collect();
-    let dir = args.get(1).expect("usage: edge-bench <dir> [--batch-chunks N] [--idle N] [--dump file]").clone();
+    let usage = "usage: edge-bench <range dir> | --group <group dir> [--from article] [--limit ranges] [--rules rules.json] [--batch-chunks N] [--idle N] [--no-finish] [--dump file] [--wire file|-] [--releases file]";
     let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned();
+    let has = |name: &str| args.iter().any(|a| a == name);
     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()));
+    // Each batch as the worker posts it (wire.rs), one JSON per line: a file, or stdout (`-`) for replay:corpus.
+    let wire_path = flag("--wire");
+    let wire_stdout = wire_path.as_deref() == Some("-");
+    let mut wire_out: Option<Box<dyn Write>> = match wire_path.as_deref() {
+        Some("-") => Some(Box::new(std::io::BufWriter::with_capacity(1 << 20, std::io::stdout().lock()))),
+        Some(p) => Some(Box::new(std::io::BufWriter::new(std::fs::File::create(p).unwrap()))),
+        None => None,
+    };
     // One line per release chunk without its payload, with its first subjects (deploy/corpus_sweep.py).
     let mut releases_out = flag("--releases").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");
-    let start: i64 = json_field(&range, "start").parse().unwrap();
-    let end: i64 = json_field(&range, "end").parse().unwrap();
-    let mut files: Vec<String> = std::fs::read_dir(&dir).unwrap().map(|e| e.unwrap().file_name().into_string().unwrap()).filter(|f| f.ends_with(".gz")).collect();
-    files.sort();
+    // One range, or a group's ranges in order through one window (the window carries over from range to range;
+    // the end is only finished after the last one, unless --no-finish: a replay to be continued).
+    let (group, ranges) = match flag("--group") {
+        Some(dir) => {
+            let (group, mut ranges) = ranges_in(&dir);
+            // --from: ranges ending before it are skipped (a replay continued); --limit: this many at most.
+            if let Some(from) = flag("--from").map(|v| v.parse::<i64>().unwrap()) {
+                ranges.retain(|r| r.end >= from);
+            }
+            if let Some(limit) = flag("--limit").map(|v| v.parse::<usize>().unwrap()) {
+                ranges.truncate(limit);
+            }
+            (group, ranges)
+        }
+        None => {
+            let positional = (1..args.len()).find(|&i| !args[i].starts_with("--") && !args[i - 1].starts_with("--")).map(|i| args[i].clone());
+            let dir = positional.expect(usage);
+            let (group, range) = range_in(&dir).unwrap_or_else(|| panic!("{dir}: no range.json"));
+            (group, vec![range])
+        }
+    };
+    if ranges.is_empty() {
+        eprintln!("no complete range found");
+        std::process::exit(2);
+    }
+    let finish = !has("--no-finish");
+    let rules = match flag("--rules") {
+        Some(path) => {
+            let json = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("{path}: {e}"));
+            serde_json::from_str::<edge::lease::RulesJson>(&json).unwrap_or_else(|e| panic!("{path}: {e}")).rules()
+        }
+        None => {
+            let mut priorities = FxHashMap::default();
+            priorities.insert(group.clone(), 1);
+            names::Rules::defaults(priorities)
+        }
+    };
+    // The report goes to stderr when the wire takes stdout.
+    let say = |line: String| {
+        if wire_stdout {
+            eprintln!("{line}");
+        } else {
+            println!("{line}");
+        }
+    };
 
-    let mut priorities = FxHashMap::default();
-    priorities.insert(group.clone(), 1);
-    let mut indexer = RangeIndexer::new(&group, start, end, start, names::Rules::defaults(priorities), idle);
+    let (start, end) = (ranges[0].start, ranges[ranges.len() - 1].end);
+    let mut indexer = RangeIndexer::new(&group, start, end, start, rules, idle);
 
     let (mut inflate_s, mut index_s) = (0.0, 0.0);
-    let (mut compressed, mut text_bytes) = (0usize, 0usize);
+    let (mut compressed, mut text_bytes, mut chunks) = (0usize, 0usize, 0usize);
     let mut out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, samples
+    let mut finished = 0usize;
     let mut keys = std::collections::HashSet::new();
     let mut decompressor = libdeflater::Decompressor::new();
     let wall = Instant::now();
-    for (i, file) in files.iter().enumerate() {
-        let raw = std::fs::read(format!("{dir}/{file}")).unwrap();
-        compressed += raw.len();
-        let t = Instant::now();
-        let data = inflate(&raw, &mut decompressor);
-        inflate_s += t.elapsed().as_secs_f64();
-        text_bytes += data.len();
-        let t = Instant::now();
-        let text = String::from_utf8_lossy(&data);
-        let next: i64 = file.split('-').nth(2).unwrap().trim_end_matches(".gz").parse::<i64>().unwrap() + 1;
-        indexer.add_text(&text, next);
-        let last = i + 1 == files.len();
-        if last {
-            indexer.finish();
-        }
-        if !last && (i + 1) % batch_chunks != 0 {
+    let mut since_batch = 0usize;
+    for (ri, range) in ranges.iter().enumerate() {
+        let range_wall = Instant::now();
+        let headers_before = out.0;
+        let mut files: Vec<String> = std::fs::read_dir(&range.dir).unwrap().map(|e| e.unwrap().file_name().into_string().unwrap()).filter(|f| f.ends_with(".gz")).collect();
+        files.sort();
+        for (i, file) in files.iter().enumerate() {
+            let raw = std::fs::read(format!("{}/{file}", range.dir)).unwrap();
+            compressed += raw.len();
+            let t = Instant::now();
+            let data = inflate(&raw, &mut decompressor);
+            inflate_s += t.elapsed().as_secs_f64();
+            text_bytes += data.len();
+            chunks += 1;
+            let t = Instant::now();
+            let text = String::from_utf8_lossy(&data);
+            let next: i64 = file.split('-').nth(2).unwrap().trim_end_matches(".gz").parse::<i64>().unwrap() + 1;
+            indexer.add_text(&text, next);
+            since_batch += 1;
+            let range_end = i + 1 == files.len();
+            let last = range_end && ri + 1 == ranges.len();
+            if last && finish {
+                indexer.finish();
+            }
+            if !range_end && since_batch < batch_chunks && indexer.pending() < MAX_PENDING_FILES {
+                index_s += t.elapsed().as_secs_f64();
+                continue;
+            }
+            let batch = indexer.take();
+            since_batch = 0;
             index_s += t.elapsed().as_secs_f64();
-            continue;
+            out.0 += batch.headers;
+            out.1 += batch.releases.len();
+            for r in &batch.releases {
+                keys.insert(r.key.clone());
+                out.2 += r.files;
+                out.3 += r.segments;
+                out.4 += r.payload.len();
+                finished += r.finished as usize;
+            }
+            out.5 += batch.candidates.len();
+            out.6 += batch.samples.len();
+            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();
+                w.flush().unwrap();
+            }
+            if let Some(w) = releases_out.as_mut() {
+                dump_releases(&batch, &group, w);
+            }
         }
-        let batch = indexer.take();
-        index_s += t.elapsed().as_secs_f64();
-        out.0 += batch.headers;
-        out.1 += batch.releases.len();
-        for r in &batch.releases {
-            keys.insert(r.key.clone());
-            out.2 += r.files;
-            out.3 += r.segments;
-            out.4 += r.payload.len();
-        }
-        out.5 += batch.candidates.len();
-        out.6 += batch.samples.len();
-        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();
-        }
-        if let Some(w) = releases_out.as_mut() {
-            dump_releases(&batch, &group, w);
+        if ranges.len() > 1 {
+            eprintln!(
+                "{} {}–{}: {} headers in {:.1} s; frontier {}; {} releases out ({} finished), {} releases and {} files open; RSS {} MB",
+                group,
+                range.start,
+                range.end,
+                out.0 - headers_before,
+                range_wall.elapsed().as_secs_f64(),
+                date(indexer.frontier()),
+                out.1,
+                finished,
+                indexer.open_releases(),
+                indexer.open_files(),
+                peak_rss_mb()
+            );
         }
     }
     let total = inflate_s + index_s;
     #[allow(static_mut_refs)]
     let p = unsafe { indexer::PROF };
     eprintln!("  of which aggregate {:.2} s, evict+cutoff {:.2} s, judge+encode {:.2} s", p[3] as f64 / 1e9, p[4] as f64 / 1e9, p[5] as f64 / 1e9);
-    println!("{group} {start}–{end}: {} chunks, {} MB compressed, {} MB text", files.len(), compressed / 1_000_000, text_bytes / 1_000_000);
-    println!(
-        "{} headers in {:.2} s ({:.0}/s on one core; wall {:.2} s)",
-        out.0,
-        total,
-        out.0 as f64 / total,
-        wall.elapsed().as_secs_f64()
-    );
-    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, {} samples",
+    say(format!("{group} {start}–{end}: {} ranges, {chunks} chunks, {} MB compressed, {} MB text", ranges.len(), compressed / 1_000_000, text_bytes / 1_000_000));
+    say(format!("{} headers in {:.2} s ({:.0}/s on one core; wall {:.2} s)", out.0, total, out.0 as f64 / total, wall.elapsed().as_secs_f64()));
+    say(format!("  inflate {inflate_s:.2} s, index {index_s:.2} s; peak RSS {} MB", peak_rss_mb()));
+    say(format!(
+        "  kept: {} release chunks ({} releases, {} finished), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} samples",
         out.1,
         keys.len(),
+        finished,
         out.2,
         out.3,
         out.4 as f64 / 1e6,
         out.4 as f64 / out.3.max(1) as f64,
         out.5,
         out.6
-    );
+    ));
 }