Lease lanes for the tail, a mixed lease order, coordinator:status, the edge end to end

A group now has two lease lanes (migration 6600): `history` works the ranges the bounds hand out, `tail` the tail
planner's, each with its own holder, window, snapshot and takeover redo, so a thread on the tail never wades
through a group's history and a held group's tail isn't stuck behind it. The edge puts `--tail-threads`
(EDGE_TAIL_THREADS, 1 of 2+ by default) of its threads on the tail lane; the group routes take `?lane=`. The
lease order within a lane alternates the group with the most planned articles and the longest unheld one. The
takeover redo puts back only the lane's ranges the window did after the snapshot, and a group whose last range
is held by a holder gone quiet can be taken over.

`coordinator:status` prints the catalog's room, the ranges by band, the last hour's rates, the held groups with
their snapshots and the phase-2 backlogs. `tests/e2e` runs the real edge-node binary against the node API: the
corpus-fill race, a failed stream and the restart that restores the snapshot, a restart without the snapshot
file, a takeover (STREAM_CHUNK_WAIT_MINUTES is a setting now). CI builds the binary before the tests.

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

edge/src/bin/edge-node.rs

@@ -5,7 +5,11 @@
 //! --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] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
+//!   COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--tail-threads T] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
+//!
+//! `--tail-threads T` (EDGE_TAIL_THREADS; 1 of 2 or more threads by default): that many of the threads lease the
+//! groups' *tail* lane (`?lane=tail`: the tail planner's ranges, with a window and snapshot of their own), the rest
+//! the history lane (the windows and the history the coordinator's bounds hand out).
 //!
 //! 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
@@ -167,10 +171,10 @@ impl Api {
         }
     }
 
-    // Group leases (REBUILD.md design 1 and 5).
+    // Group leases (REBUILD.md design 1 and 5), per lane (`history` or `tail`).
 
-    fn lease_group(&self) -> Result<Option<GroupLeaseJson>, ApiError> {
-        let mut response = self.request("POST", "groups/lease", None, false, "application/json")?;
+    fn lease_group(&self, lane: &str) -> Result<Option<GroupLeaseJson>, ApiError> {
+        let mut response = self.request("POST", &format!("groups/lease?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             204 => Ok(None),
             200 => {
@@ -181,8 +185,8 @@ impl Api {
         }
     }
 
-    fn next_range(&self, group: i64) -> Result<Option<RangeSpanJson>, ApiError> {
-        let mut response = self.request("POST", &format!("groups/{group}/next"), None, false, "application/json")?;
+    fn next_range(&self, group: i64, lane: &str) -> Result<Option<RangeSpanJson>, ApiError> {
+        let mut response = self.request("POST", &format!("groups/{group}/next?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             204 => Ok(None),
             200 => {
@@ -193,8 +197,8 @@ impl Api {
         }
     }
 
-    fn group_call(&self, group: i64, action: &str) -> Result<(), ApiError> {
-        let response = self.request("POST", &format!("groups/{group}/{action}"), None, false, "application/json")?;
+    fn group_call(&self, group: i64, lane: &str, action: &str) -> Result<(), ApiError> {
+        let response = self.request("POST", &format!("groups/{group}/{action}?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             200..=299 => Ok(()),
             status => Err(ApiError::Other(format!("{action}: HTTP {status}"))),
@@ -202,11 +206,11 @@ impl Api {
     }
 
     /// The window's snapshot (bincode), gzipped, standing at `next`.
-    fn put_snapshot(&self, group: i64, next: i64, snapshot: &[u8]) -> Result<usize, ApiError> {
+    fn put_snapshot(&self, group: i64, lane: &str, 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 url = format!("{}/api/v1/node/groups/{group}/snapshot?next={next}&lane={lane}", 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[..]);
@@ -227,8 +231,8 @@ impl Api {
     }
 
     /// 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")?;
+    fn get_snapshot(&self, group: i64, lane: &str) -> Result<Option<(i64, Vec<u8>)>, ApiError> {
+        let response = self.request("GET", &format!("groups/{group}/snapshot?lane={lane}"), None, false, "application/octet-stream")?;
         match response.status().as_u16() {
             404 => Ok(None),
             200 => {
@@ -544,9 +548,10 @@ enum RangeOutcome {
 }
 
 /// 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.
+/// least, the range finished. The indexer keeps the window for the next range. `group_name` is the label for the
+/// log (the tail lane's says so).
 #[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 {
+fn run_group_range(api: &Arc<Api>, group_id: i64, lane: &str, 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));
@@ -597,7 +602,7 @@ fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: Range
         }
         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");
+            let _ = api.group_call(group_id, lane, "heartbeat");
         }
         match frame {
             Frame::KeepAlive => continue,
@@ -706,9 +711,9 @@ fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: Range
 }
 
 /// 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) {
+fn snapshot(api: &Api, group_id: i64, lane: &str, group_name: &str, indexer: &RangeIndexer) {
     match indexer.snapshot() {
-        Ok(bytes) => match api.put_snapshot(group_id, indexer.next(), &bytes) {
+        Ok(bytes) => match api.put_snapshot(group_id, lane, 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}")),
         },
@@ -716,14 +721,16 @@ fn snapshot(api: &Api, group_id: i64, group_name: &str, indexer: &RangeIndexer)
     }
 }
 
-/// 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) {
+/// One held group (one lane of it): 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, lane: &str, config: &Config, stopping: &AtomicBool) {
     let group_id = lease.group.id;
-    let group_name = lease.group.name.clone();
+    let name = lease.group.name.clone();
+    // The log's label; the indexer gets the group's name itself (keys, xrefs).
+    let group_name = if lane == "tail" { format!("{name} (tail)") } else { name.clone() };
     let rules = lease.rules.rules();
     let mut indexer = match lease.snapshot {
-        Some(s) => match api.get_snapshot(group_id) {
+        Some(s) => match api.get_snapshot(group_id, lane) {
             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));
@@ -731,20 +738,20 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
                 }
                 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)
+                    RangeIndexer::for_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)
+                RangeIndexer::for_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");
+                let _ = api.group_call(group_id, lane, "release");
                 return;
             }
         },
-        None => RangeIndexer::for_group(&group_name, rules.clone(), 5),
+        None => RangeIndexer::for_group(&name, rules.clone(), 5),
     };
     let totals = Arc::new(Mutex::new(Totals::default()));
     let started = Instant::now();
@@ -754,7 +761,7 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
         if stopping.load(Ordering::Relaxed) {
             break;
         }
-        let range = match api.next_range(group_id) {
+        let range = match api.next_range(group_id, lane) {
             Ok(Some(range)) => range,
             Ok(None) => {
                 log("info", &format!("{group_name}: caught up after {ranges} ranges; the group goes back."));
@@ -768,19 +775,19 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
         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) {
+        match run_group_range(api, group_id, lane, &group_name, range, &mut indexer, config, stopping, &totals) {
             RangeOutcome::Done => ranges += 1,
             // The same range again from the reported chunk: the stream waits for its new copy.
             RangeOutcome::Again => {}
             RangeOutcome::Stop => break,
         }
         if snapshotted.elapsed() >= config.snapshot_every {
-            snapshot(api, group_id, &group_name, &indexer);
+            snapshot(api, group_id, lane, &group_name, &indexer);
             snapshotted = Instant::now();
         }
     }
-    snapshot(api, group_id, &group_name, &indexer);
-    let _ = api.group_call(group_id, "release");
+    snapshot(api, group_id, lane, &group_name, &indexer);
+    let _ = api.group_call(group_id, lane, "release");
     let t = totals.lock().unwrap();
     log(
         "info",
@@ -807,6 +814,12 @@ fn main() {
         .filter(|&n| n > 0)
         .unwrap_or_else(|| std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
     let ranges: Option<usize> = flag("--ranges").map(|v| v.parse().unwrap());
+    // Of the threads, this many on the groups' tail lane: one by default when there are two or more.
+    let tail_threads: usize = flag("--tail-threads")
+        .or_else(|| std::env::var("EDGE_TAIL_THREADS").ok())
+        .map(|v| v.parse().unwrap())
+        .unwrap_or(if threads >= 2 { 1 } else { 0 });
+    assert!(tail_threads <= threads, "--tail-threads {tail_threads} is more than the {threads} threads");
     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),
@@ -819,12 +832,13 @@ fn main() {
     signal_hook::flag::register(signal_hook::consts::SIGTERM, stopping.clone()).unwrap();
     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()));
+    log("info", &format!("Starting {threads} worker threads{}{}.", if range_leases_wanted(&ranges) { String::new() } else { format!(", {tail_threads} on the tail") }, 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(|_| {
+        .map(|i| {
             let (api, config, stopping) = (api.clone(), config.clone(), stopping.clone());
+            let lane = if i < tail_threads { "tail" } else { "history" };
             std::thread::spawn(move || {
                 let mut done = 0;
                 if range_leases {
@@ -844,9 +858,9 @@ fn main() {
                     return;
                 }
                 while !stopping.load(Ordering::Relaxed) && groups.is_none_or(|n| done < n) {
-                    match api.lease_group() {
+                    match api.lease_group(lane) {
                         Ok(Some(lease)) => {
-                            run_group(&api, lease, &config, &stopping);
+                            run_group(&api, lease, lane, &config, &stopping);
                             done += 1;
                         }
                         Ok(None) if groups.is_some() => break,
@@ -865,6 +879,10 @@ fn main() {
     }
 }
 
+fn range_leases_wanted(ranges: &Option<usize>) -> bool {
+    ranges.is_some()
+}
+
 fn idle(stopping: &AtomicBool) {
     let until = Instant::now() + IDLE;
     while Instant::now() < until && !stopping.load(Ordering::Relaxed) {