edge-node: a range whose batch is too large goes on in smaller batches

Range 154218 (a.b.hdtv.tv-episodes) looped 13 times on HTTP 413: a
500k-article batch over the server's 256 MB body limit, though under
100k files. On a 413 the worker gives the range back and posts that
range's batches at half the articles and files from then on (again on
another 413, down to 1/64); stored batches stay. The message says how
large the batch was.

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

tools/edge-bench-rs/src/bin/edge-node.rs

@@ -97,13 +97,15 @@ struct StoredJson {
 enum ApiError {
     /// 409: the range is no longer this node's (or the batch was stale): stop it, don't give it back.
     LeaseLost(String),
+    /// 413: the batch was over the server's body limit; the range is tried again in smaller batches.
+    TooLarge(String),
     Other(String),
 }
 
 impl std::fmt::Display for ApiError {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         match self {
-            ApiError::LeaseLost(m) | ApiError::Other(m) => f.write_str(m),
+            ApiError::LeaseLost(m) | ApiError::TooLarge(m) | ApiError::Other(m) => f.write_str(m),
         }
     }
 }
@@ -194,6 +196,9 @@ impl Api {
         gz.write_all(&json).unwrap();
         let body = gz.finish().unwrap();
         let mut response = self.request("POST", &format!("ranges/{range}/batches"), Some(&body), true, "application/json")?;
+        if response.status().as_u16() == 413 {
+            return Err(ApiError::TooLarge(format!("batch: HTTP {} ({} MB gzipped, {} MB of JSON)", response.status(), body.len() / 1_000_000, json.len() / 1_000_000)));
+        }
         if response.status().as_u16() != 200 {
             return Err(ApiError::Other(format!("batch: HTTP {}", response.status())));
         }
@@ -322,6 +327,16 @@ fn inflate(raw: &[u8], d: &mut libdeflater::Decompressor) -> Result<Vec<u8>, Str
 struct Config {
     batch_articles: i64,
     bytes_per_second: Option<f64>,
+    /// Ranges whose batch was too large (413), and how much smaller their batches are now (halved each time).
+    smaller: Mutex<HashMap<i64, i64>>,
+}
+
+impl Config {
+    /// Articles and pending files a batch of this range is posted at.
+    fn limits(&self, range: i64) -> (i64, usize) {
+        let divisor = self.smaller.lock().unwrap().get(&range).copied().unwrap_or(1);
+        (self.batch_articles / divisor, MAX_PENDING_FILES / divisor as usize)
+    }
 }
 
 #[derive(Default)]
@@ -375,6 +390,7 @@ fn run_range(api: &Arc<Api>, lease: LeaseJson, config: &Config, stopping: &Atomi
             return;
         }
     };
+    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;
@@ -408,7 +424,7 @@ fn run_range(api: &Arc<Api>, lease: LeaseJson, config: &Config, stopping: &Atomi
                 if reached_end {
                     indexer.finish();
                 }
-                if reached_end || to + 1 - batch_from >= config.batch_articles || indexer.pending() >= MAX_PENDING_FILES {
+                if reached_end || to + 1 - batch_from >= batch_articles || indexer.pending() >= max_pending {
                     if post_tx.send(indexer.take()).is_err() {
                         break;
                     }
@@ -434,6 +450,18 @@ fn run_range(api: &Arc<Api>, lease: LeaseJson, config: &Config, stopping: &Atomi
             log("warn", &format!("Range {}: {m}", range.id));
             return;
         }
+        Some(ApiError::TooLarge(m)) => {
+            // Batches already stored stay; from the resume point on, half the size (a batch is at least a chunk).
+            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!("Range {}: {m}; giving it back, its batches 1/{divisor} the size from now on.", range.id));
+            let _ = api.call(range.id, "release");
+            return;
+        }
         Some(ApiError::Other(m)) => {
             log("error", &format!("Range {} failed: {m}; giving it back.", range.id));
             let _ = api.call(range.id, "release");
@@ -492,6 +520,7 @@ fn main() {
         batch_articles: flag("--batch-articles").map(|v| v.parse().unwrap()).unwrap_or(500_000),
         // 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()),
     });
     let stopping = Arc::new(AtomicBool::new(false));
     signal_hook::flag::register(signal_hook::consts::SIGTERM, stopping.clone()).unwrap();