The window worker: group leases, the stream serving the corpus, the snapshot through the API (design 1 and 5)

A worker thread leases a group and takes its ranges in article order through
one carried window; the coordinator keeps one cursor a group (group_leases,
migration 6500) and hands out the next range, the rules and nothing about
where the chunks are: the range stream serves a stored range's chunks from the
corpus file and a live one's from the spool. The window's snapshot (bincode:
the frontier, the open files, releases and albums, the upload numbers, the
batch-counter table) goes to the coordinator every EDGE_SNAPSHOT_MINUTES and
when the group is given back, and the next holder restores it; a takeover
after LEASE_TTL_SECONDS plans the ranges from the snapshot's article again.
The prefetcher runs ahead of held groups' cursors and skips chunks the corpus
has. LEASE_MIN_PRIORITY..LEASE_MAX_PRIORITY bound what is leased. edge-node
holds groups by default; --ranges keeps the old per-range path.

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

edge/src/bin/edge-node.rs

@@ -1,16 +1,20 @@
-//! The edge worker in Rust (REBUILD.md, part 2; commands/edge_node.ts): leases
-//! ranges from the coordinator, reads their XOVER chunks from its stream,
-//! indexes them (the same RangeIndexer as edge-bench, kept identical to
-//! the TypeScript by check.ts) and posts the batches. One range per thread.
+//! The window worker in Rust (REBUILD.md, "Voyager 3: window indexing", 1 and 5): each thread leases a *group*
+//! from the coordinator, takes its ranges in article order, reads each range's XOVER chunks from the coordinator's
+//! stream (the corpus or the spool, the worker can't tell), indexes them through one carried window and posts the
+//! batches, a batch at least at every range end; the window's snapshot goes to the coordinator every
+//! --snapshot-minutes and when the group is given back, and the next holder restores it. A group with no range
+//! left (caught up: the tail) is given back and leased again later.
 //!
-//!   COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--batch-articles 500000] [--ranges N] [--max-mbit M]
+//!   COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
 //!
-//! SIGTERM/SIGINT: each thread gives its range back after the current chunk
-//! (the next worker resumes at the last stored batch).
+//! SIGTERM/SIGINT: each thread finishes the chunk, posts its batch, snapshots and gives its group back.
+//! `--ranges`: the old range leases (POST lease), one range per thread without a carried window (kept for a
+//! coordinator without group leases).
 
 use edge::indexer::{Batch, RangeIndexer};
-use edge::lease::LeaseJson;
+use edge::lease::{GroupLeaseJson, LeaseJson, NextRangeJson, RangeSpanJson};
 use edge::wire;
+use flate2::read::GzDecoder;
 use flate2::write::GzEncoder;
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -163,6 +167,81 @@ impl Api {
         }
     }
 
+    // Group leases (REBUILD.md design 1 and 5).
+
+    fn lease_group(&self) -> Result<Option<GroupLeaseJson>, ApiError> {
+        let mut response = self.request("POST", "groups/lease", None, false, "application/json")?;
+        match response.status().as_u16() {
+            204 => Ok(None),
+            200 => {
+                let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("groups/lease: {e}")))?;
+                serde_json::from_str(&text).map(Some).map_err(|e| ApiError::Other(format!("groups/lease: {e}")))
+            }
+            status => Err(ApiError::Other(format!("groups/lease: HTTP {status}"))),
+        }
+    }
+
+    fn next_range(&self, group: i64) -> Result<Option<RangeSpanJson>, ApiError> {
+        let mut response = self.request("POST", &format!("groups/{group}/next"), None, false, "application/json")?;
+        match response.status().as_u16() {
+            204 => Ok(None),
+            200 => {
+                let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("next: {e}")))?;
+                serde_json::from_str::<NextRangeJson>(&text).map(|n| Some(n.range)).map_err(|e| ApiError::Other(format!("next: {e}")))
+            }
+            status => Err(ApiError::Other(format!("next: HTTP {status}"))),
+        }
+    }
+
+    fn group_call(&self, group: i64, action: &str) -> Result<(), ApiError> {
+        let response = self.request("POST", &format!("groups/{group}/{action}"), None, false, "application/json")?;
+        match response.status().as_u16() {
+            200..=299 => Ok(()),
+            status => Err(ApiError::Other(format!("{action}: HTTP {status}"))),
+        }
+    }
+
+    /// The window's snapshot (bincode), gzipped, standing at `next`.
+    fn put_snapshot(&self, group: i64, next: i64, snapshot: &[u8]) -> Result<usize, ApiError> {
+        let mut gz = GzEncoder::new(Vec::with_capacity(snapshot.len() / 3), flate2::Compression::new(1));
+        gz.write_all(snapshot).unwrap();
+        let body = gz.finish().unwrap();
+        let url = format!("{}/api/v1/node/groups/{group}/snapshot?next={next}", self.base);
+        let auth = format!("Bearer {}", self.token);
+        for attempt in 1..=RETRIES {
+            let result = self.agent.put(&url).header("Authorization", &auth).header("Content-Type", "application/octet-stream").send(&body[..]);
+            match result {
+                Ok(response) => match response.status().as_u16() {
+                    200..=299 => return Ok(body.len()),
+                    409 => return Err(ApiError::LeaseLost("snapshot: the group is no longer this node's".into())),
+                    status if status < 500 => return Err(ApiError::Other(format!("snapshot: HTTP {status}"))),
+                    status if attempt == RETRIES => return Err(ApiError::Other(format!("snapshot: HTTP {status}"))),
+                    _ => {}
+                },
+                Err(error) if attempt == RETRIES => return Err(ApiError::Other(format!("snapshot: {error}"))),
+                Err(_) => {}
+            }
+            std::thread::sleep(Duration::from_millis(60_000.min(1000 * 2u64.pow(attempt))));
+        }
+        unreachable!()
+    }
+
+    /// The group's snapshot: its bytes (inflated) and the article it stands at; None without one.
+    fn get_snapshot(&self, group: i64) -> Result<Option<(i64, Vec<u8>)>, ApiError> {
+        let response = self.request("GET", &format!("groups/{group}/snapshot"), None, false, "application/octet-stream")?;
+        match response.status().as_u16() {
+            404 => Ok(None),
+            200 => {
+                let next: i64 = response.headers().get("x-snapshot-next").and_then(|v| v.to_str().ok()).and_then(|v| v.parse().ok()).ok_or_else(|| ApiError::Other("snapshot: no X-Snapshot-Next".into()))?;
+                let mut gz = GzDecoder::new(response.into_body().into_reader());
+                let mut bytes = Vec::new();
+                gz.read_to_end(&mut bytes).map_err(|e| ApiError::Other(format!("snapshot: {e}")))?;
+                Ok(Some((next, bytes)))
+            }
+            status => Err(ApiError::Other(format!("snapshot: HTTP {status}"))),
+        }
+    }
+
     fn corrupt(&self, range: i64, from: i64) -> Result<(), ApiError> {
         let body = format!("{{\"from\":{from}}}");
         let response = self.request("POST", &format!("ranges/{range}/corrupt"), Some(body.as_bytes()), false, "application/json")?;
@@ -275,6 +354,8 @@ fn inflate(raw: &[u8], d: &mut libdeflater::Decompressor) -> Result<Vec<u8>, Str
 struct Config {
     batch_articles: i64,
     bytes_per_second: Option<f64>,
+    /// Between two snapshots of a held group's window.
+    snapshot_every: Duration,
     /// Ranges whose batch was too large (413), and how much smaller their batches are now (halved each time).
     smaller: Mutex<HashMap<i64, i64>>,
 }
@@ -452,6 +533,255 @@ fn run_range(api: &Arc<Api>, lease: LeaseJson, config: &Config, stopping: &Atomi
     }
 }
 
+/// What one range of a held group came to.
+enum RangeOutcome {
+    Done,
+    /// The group is given back (a stop, a lost lease, a failure): no more ranges.
+    Stop,
+}
+
+/// One range of a held group through the carried window: the chunks streamed, batches posted at the range's end at
+/// least, the range finished. The indexer keeps the window for the next range.
+#[allow(clippy::too_many_arguments)]
+fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: RangeSpanJson, indexer: &mut RangeIndexer, config: &Config, stopping: &AtomicBool, totals: &Arc<Mutex<Totals>>) -> RangeOutcome {
+    let started = Instant::now();
+    indexer.begin_range(range.next, range.end);
+    let post_error: Arc<Mutex<Option<ApiError>>> = Arc::new(Mutex::new(None));
+    let last_contact = Arc::new(AtomicI64::new(0));
+    let since = Instant::now();
+    let (post_tx, post_rx) = sync_channel::<Batch>(1);
+    let poster = {
+        let (api, totals, post_error, last_contact) = (api.clone(), totals.clone(), post_error.clone(), last_contact.clone());
+        std::thread::spawn(move || {
+            for batch in post_rx {
+                if post_error.lock().unwrap().is_some() {
+                    continue;
+                }
+                let at = Instant::now();
+                match api.batch(range.id, &batch) {
+                    Ok((releases, candidates, bytes)) => {
+                        let mut t = totals.lock().unwrap();
+                        t.post_ms += at.elapsed().as_millis();
+                        t.bytes += bytes;
+                        t.headers += batch.headers;
+                        t.releases += releases;
+                        t.candidates += candidates;
+                        last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed);
+                    }
+                    Err(e) => *post_error.lock().unwrap() = Some(e),
+                }
+            }
+        })
+    };
+    let frames = match api.stream(range.id) {
+        Ok(stream) => read_frames(stream, config.bytes_per_second),
+        Err(e) => {
+            log("error", &format!("{group_name} {}–{}: stream failed: {e}; the group goes back.", range.start, range.end));
+            drop(post_tx);
+            let _ = poster.join();
+            return RangeOutcome::Stop;
+        }
+    };
+    let (batch_articles, max_pending) = config.limits(range.id);
+    let mut batch_from = range.next;
+    let mut reached_end = false;
+    let mut corrupt: Option<(i64, i64, String)> = None;
+    let mut failure: Option<String> = None;
+    let headers_before = totals.lock().unwrap().headers;
+    for frame in frames.iter() {
+        if stopping.load(Ordering::Relaxed) || post_error.lock().unwrap().is_some() {
+            break;
+        }
+        if since.elapsed().as_millis() as i64 - last_contact.load(Ordering::Relaxed) > HEARTBEAT.as_millis() as i64 {
+            last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed);
+            let _ = api.group_call(group_id, "heartbeat");
+        }
+        match frame {
+            Frame::KeepAlive => continue,
+            Frame::End => break,
+            Frame::Failed(e) => {
+                failure = Some(e);
+                break;
+            }
+            Frame::Chunk { from, to, text } => {
+                let data = match text {
+                    Ok(data) => data,
+                    Err(e) => {
+                        corrupt = Some((from, to, e));
+                        break;
+                    }
+                };
+                indexer.add_text(&String::from_utf8_lossy(&data), to + 1);
+                reached_end = to >= range.end;
+                // The window carries on past the range's end: no finish here; a batch at the end at least.
+                if reached_end || to + 1 - batch_from >= batch_articles || indexer.pending() >= max_pending {
+                    if post_tx.send(indexer.take()).is_err() {
+                        break;
+                    }
+                    batch_from = to + 1;
+                }
+                if reached_end {
+                    break;
+                }
+            }
+        }
+    }
+    // Stopped mid-range: the batch so far goes out, so the snapshot stands at a stored resume point.
+    if !reached_end && corrupt.is_none() && failure.is_none() && post_error.lock().unwrap().is_none() && indexer.next() > batch_from {
+        let _ = post_tx.send(indexer.take());
+    }
+    drop(post_tx);
+    let _ = poster.join();
+    let post_error = post_error.lock().unwrap().take();
+    if let Some((from, to, e)) = corrupt {
+        log("warn", &format!("{group_name} {}–{}: chunk {from}–{to} doesn't inflate ({e}); reported, the group goes back", range.start, range.end));
+        let _ = api.corrupt(range.id, from);
+        return RangeOutcome::Stop;
+    }
+    match post_error {
+        Some(ApiError::LeaseLost(m)) => {
+            log("warn", &format!("{group_name} {}–{}: {m}", range.start, range.end));
+            return RangeOutcome::Stop;
+        }
+        Some(ApiError::TooLarge(m)) => {
+            let divisor = {
+                let mut smaller = config.smaller.lock().unwrap();
+                let d = smaller.entry(range.id).or_insert(1);
+                *d = (*d * 2).min(64);
+                *d
+            };
+            log("warn", &format!("{group_name} {}–{}: {m}; the group goes back, its batches 1/{divisor} the size from now on.", range.start, range.end));
+            return RangeOutcome::Stop;
+        }
+        Some(ApiError::Other(m)) => {
+            log("error", &format!("{group_name} {}–{}: {m}; the group goes back.", range.start, range.end));
+            return RangeOutcome::Stop;
+        }
+        None => {}
+    }
+    if let Some(e) = failure {
+        log("error", &format!("{group_name} {}–{}: {e}; the group goes back.", range.start, range.end));
+        return RangeOutcome::Stop;
+    }
+    if !reached_end {
+        return RangeOutcome::Stop;
+    }
+    match api.call(range.id, "finish") {
+        Ok(()) => {
+            let t = totals.lock().unwrap();
+            let seconds = started.elapsed().as_secs_f64();
+            let headers = t.headers - headers_before;
+            log(
+                "info",
+                &format!(
+                    "{group_name} {}–{}: {} headers in {:.0} s ({:.0}/s); {} releases and {} files open in the window.",
+                    range.start,
+                    range.end,
+                    headers,
+                    seconds,
+                    headers as f64 / seconds.max(0.001),
+                    indexer.open_releases(),
+                    indexer.open_files()
+                ),
+            );
+            RangeOutcome::Done
+        }
+        Err(e) => {
+            log("warn", &format!("{group_name} {}–{}: finish: {e}", range.start, range.end));
+            RangeOutcome::Stop
+        }
+    }
+}
+
+/// The window's snapshot to the coordinator, standing at the indexer's next article.
+fn snapshot(api: &Api, group_id: i64, group_name: &str, indexer: &RangeIndexer) {
+    match indexer.snapshot() {
+        Ok(bytes) => match api.put_snapshot(group_id, indexer.next(), &bytes) {
+            Ok(gz) => log("info", &format!("{group_name}: snapshot at {} ({:.1} MB, {:.1} MB gzipped).", indexer.next(), bytes.len() as f64 / 1e6, gz as f64 / 1e6)),
+            Err(e) => log("warn", &format!("{group_name}: snapshot failed: {e}")),
+        },
+        Err(e) => log("warn", &format!("{group_name}: no snapshot: {e}")),
+    }
+}
+
+/// One held group: its window restored from the snapshot (or new), its ranges one after another until it is
+/// caught up or the worker stops; then a snapshot, and the group given back.
+fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &AtomicBool) {
+    let group_id = lease.group.id;
+    let group_name = lease.group.name.clone();
+    let rules = lease.rules.rules();
+    let mut indexer = match lease.snapshot {
+        Some(s) => match api.get_snapshot(group_id) {
+            Ok(Some((next, bytes))) => match RangeIndexer::restore(&bytes, rules.clone(), 5) {
+                Ok(ix) => {
+                    log("info", &format!("{group_name}: window restored from the snapshot at {next} ({:.1} MB).", bytes.len() as f64 / 1e6));
+                    ix
+                }
+                Err(e) => {
+                    log("warn", &format!("{group_name}: the snapshot at {} can't be restored ({e}); the window starts empty.", s.next));
+                    RangeIndexer::for_group(&group_name, rules.clone(), 5)
+                }
+            },
+            Ok(None) => {
+                log("warn", &format!("{group_name}: the lease names a snapshot at {} but there is none; the window starts empty.", s.next));
+                RangeIndexer::for_group(&group_name, rules.clone(), 5)
+            }
+            Err(e) => {
+                log("warn", &format!("{group_name}: snapshot: {e}; the group goes back."));
+                let _ = api.group_call(group_id, "release");
+                return;
+            }
+        },
+        None => RangeIndexer::for_group(&group_name, rules.clone(), 5),
+    };
+    let totals = Arc::new(Mutex::new(Totals::default()));
+    let started = Instant::now();
+    let mut snapshotted = Instant::now();
+    let mut ranges = 0;
+    loop {
+        if stopping.load(Ordering::Relaxed) {
+            break;
+        }
+        let range = match api.next_range(group_id) {
+            Ok(Some(range)) => range,
+            Ok(None) => {
+                log("info", &format!("{group_name}: caught up after {ranges} ranges; the group goes back."));
+                break;
+            }
+            Err(e) => {
+                log("warn", &format!("{group_name}: next range: {e}; the group goes back."));
+                break;
+            }
+        };
+        if range.next != indexer.next() && ranges > 0 {
+            log("warn", &format!("{group_name}: the window stands at {} but the next range resumes at {}: the window carries on from there.", indexer.next(), range.next));
+        }
+        match run_group_range(api, group_id, &group_name, range, &mut indexer, config, stopping, &totals) {
+            RangeOutcome::Done => ranges += 1,
+            RangeOutcome::Stop => break,
+        }
+        if snapshotted.elapsed() >= config.snapshot_every {
+            snapshot(api, group_id, &group_name, &indexer);
+            snapshotted = Instant::now();
+        }
+    }
+    snapshot(api, group_id, &group_name, &indexer);
+    let _ = api.group_call(group_id, "release");
+    let t = totals.lock().unwrap();
+    log(
+        "info",
+        &format!(
+            "{group_name}: {ranges} ranges, {} headers in {:.0} s, {} release chunks, {} probe candidates; posted {:.1} MB in {:.1} s.",
+            t.headers,
+            started.elapsed().as_secs_f64(),
+            t.releases,
+            t.candidates,
+            t.bytes as f64 / 1e6,
+            t.post_ms as f64 / 1000.0
+        ),
+    );
+}
+
 fn main() {
     let args: Vec<String> = std::env::args().collect();
     let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned();
@@ -466,6 +796,7 @@ fn main() {
     let max_mbit: Option<f64> = flag("--max-mbit").or_else(|| std::env::var("EDGE_MAX_MBIT").ok()).map(|v| v.parse().unwrap()).filter(|&m: &f64| m > 0.0);
     let config = Arc::new(Config {
         batch_articles: flag("--batch-articles").map(|v| v.parse().unwrap()).unwrap_or(500_000),
+        snapshot_every: Duration::from_secs(60 * flag("--snapshot-minutes").or_else(|| std::env::var("EDGE_SNAPSHOT_MINUTES").ok()).map(|v| v.parse::<u64>().unwrap()).unwrap_or(30)),
         // The node's cap, shared by its threads.
         bytes_per_second: max_mbit.map(|m| m * 1e6 / 8.0 / threads as f64),
         smaller: Mutex::new(HashMap::new()),
@@ -475,21 +806,39 @@ fn main() {
     signal_hook::flag::register(signal_hook::consts::SIGINT, stopping.clone()).unwrap();
     let api = Arc::new(Api::new(&url, &token));
     log("info", &format!("Starting {threads} worker threads{}.", max_mbit.map(|m| format!(" (download cap {m} Mbit/s)")).unwrap_or_default()));
+    let groups: Option<usize> = flag("--groups").map(|v| v.parse().unwrap());
+    let range_leases = ranges.is_some();
     let workers: Vec<_> = (0..threads)
         .map(|_| {
             let (api, config, stopping) = (api.clone(), config.clone(), stopping.clone());
             std::thread::spawn(move || {
                 let mut done = 0;
-                while !stopping.load(Ordering::Relaxed) && ranges.is_none_or(|n| done < n) {
-                    match api.lease() {
+                if range_leases {
+                    while !stopping.load(Ordering::Relaxed) && ranges.is_none_or(|n| done < n) {
+                        match api.lease() {
+                            Ok(Some(lease)) => {
+                                run_range(&api, lease, &config, &stopping);
+                                done += 1;
+                            }
+                            Ok(None) => break,
+                            Err(e) => {
+                                log("warn", &format!("Lease failed: {e}"));
+                                idle(&stopping);
+                            }
+                        }
+                    }
+                    return;
+                }
+                while !stopping.load(Ordering::Relaxed) && groups.is_none_or(|n| done < n) {
+                    match api.lease_group() {
                         Ok(Some(lease)) => {
-                            run_range(&api, lease, &config, &stopping);
+                            run_group(&api, lease, &config, &stopping);
                             done += 1;
                         }
-                        Ok(None) if ranges.is_some() => break,
+                        Ok(None) if groups.is_some() => break,
                         Ok(None) => idle(&stopping),
                         Err(e) => {
-                            log("warn", &format!("Lease failed: {e}"));
+                            log("warn", &format!("Group lease failed: {e}"));
                             idle(&stopping);
                         }
                     }