mirror_worker: cap buffered entries by bytes, not just package count

commit_packages bounds only the number of buffered packages, but a
single package can hold 256 entries of up to 65535 bytes (~16 MiB), so
a count-only cap could buffer far past the isolate's memory ceiling.
Add max_chunk_bytes (default 16 MiB) and flush early once buffered
entry bytes reach it, whichever cap trips first.

Author: lukevalenta

crates/mirror_worker/config.schema.json

@@ -37,6 +37,12 @@
             "default": 32,
             "description": "How many entry packages add-entries verifies before flushing them to storage and advancing the persisted-entry frontier. Bounds in-memory buffering and gives durable mid-request progress on large uploads. Defaults to 32 (the recommended per-request package budget) when omitted; capped at 1024 to bound worst-case buffering."
         },
+        "max_chunk_bytes": {
+            "type": "integer",
+            "minimum": 1,
+            "default": 16777216,
+            "description": "Byte ceiling on the entries buffered between flushes. add-entries flushes early once buffered entry bytes reach this many, even if fewer than commit_packages packages have accumulated. Bounds peak memory independent of package sizes, since a single package can hold up to ~16 MiB. Defaults to 16777216 (16 MiB) when omitted."
+        },
         "logs": {
             "type": "object",
             "description": "CAs this mirror mirrors, keyed by log_key_name: the CA cosigner's note-signature name (the CA ID) on the checkpoints it ingests. Used as a signed-note key name at runtime, so per c2sp.org/signed-note it MUST NOT contain '+', whitespace, or control characters.",

crates/mirror_worker/config/src/lib.rs

@@ -69,6 +69,16 @@ pub struct AppConfig {
     /// (see [`Self::commit_packages`]). Consumed by
     /// [`mirror_worker`](../mirror_worker/)'s `add_entries`.
     pub commit_packages: Option<u64>,
+    /// Byte ceiling on the entries buffered between flushes. `add-entries`
+    /// flushes early once the buffered entry bytes reach this many, even if
+    /// fewer than `commit_packages` packages have accumulated. This bounds
+    /// peak memory independent of package sizes: `commit_packages` alone
+    /// caps only the package *count*, and a single package can hold up to
+    /// 256 entries of 65535 bytes (~16 MiB), so a count-only bound can
+    /// exceed the isolate's memory ceiling. `None` falls back to a default
+    /// (see [`Self::max_chunk_bytes`]). Consumed by
+    /// [`mirror_worker`](../mirror_worker/)'s `add_entries`.
+    pub max_chunk_bytes: Option<u64>,
     /// CAs this mirror mirrors, keyed by `log_key_name`: the CA
     /// cosigner's note-signature name (the CA ID) carried by the
     /// checkpoints it ingests.
@@ -136,6 +146,18 @@ impl AppConfig {
         self.commit_packages.unwrap_or(32)
     }
 
+    /// Byte ceiling on buffered entries before `add-entries` flushes early,
+    /// falling back to 16 MiB when `max_chunk_bytes` is unset. This bounds
+    /// peak in-memory buffering regardless of how large individual packages
+    /// are, complementing the `commit_packages` count cap. 16 MiB matches
+    /// the worst-case size of a single spec-maximal package (256 entries *
+    /// 65535 bytes), so the default never flushes mid-package for a
+    /// compliant upload yet still caps a pathological one.
+    #[must_use]
+    pub fn max_chunk_bytes(&self) -> u64 {
+        self.max_chunk_bytes.unwrap_or(16 * 1024 * 1024)
+    }
+
     /// Validate the configuration beyond what `serde` and the JSON schema
     /// can express.
     ///
@@ -359,6 +381,7 @@ mod tests {
             monitoring_prefix: None,
             clean_interval_secs: None,
             commit_packages: None,
+            max_chunk_bytes: None,
             logs: HashMap::from([(
                 "example.com/log1".to_owned(),
                 LogParams {
@@ -491,6 +514,14 @@ mod tests {
         assert_eq!(cfg.commit_packages(), 8);
     }
 
+    #[test]
+    fn max_chunk_bytes_defaults_to_16_mib() {
+        let mut cfg = good_app_config();
+        assert_eq!(cfg.max_chunk_bytes(), 16 * 1024 * 1024);
+        cfg.max_chunk_bytes = Some(1024);
+        assert_eq!(cfg.max_chunk_bytes(), 1024);
+    }
+
     #[test]
     fn validate_rejects_inverted_window() {
         let cfg = with_log(|log| {
@@ -560,6 +591,7 @@ mod tests {
             monitoring_prefix: None,
             clean_interval_secs: None,
             commit_packages: None,
+            max_chunk_bytes: None,
             logs: HashMap::from([(
                 "a".repeat(250),
                 LogParams {

crates/mirror_worker/src/add_entries.rs

@@ -370,6 +370,11 @@ where
     // config.schema.json caps commit_packages (max 1024), enforced by the
     // build script, so this always fits usize; the fallback is unreachable.
     let commit_packages = usize::try_from(crate::CONFIG.commit_packages()).unwrap_or(usize::MAX);
+    // Byte ceiling on buffered entries; flush early when reached so peak
+    // memory is bounded regardless of package sizes. Saturating to
+    // usize::MAX on a 32-bit target just means "never trip the byte cap",
+    // leaving the package-count cap in force.
+    let max_chunk_bytes = usize::try_from(crate::CONFIG.max_chunk_bytes()).unwrap_or(usize::MAX);
 
     // Entries below the request-start frontier are already persisted; new
     // persistence begins at this fixed boundary.
@@ -384,6 +389,7 @@ where
     let mut chunk: Vec<Vec<u8>> = Vec::new();
     let mut chunk_end = frontier_size;
     let mut chunk_pkgs = 0usize;
+    let mut chunk_bytes = 0usize;
     let mut packages_received: u64 = 0;
     let mut truncated = false;
 
@@ -441,13 +447,21 @@ where
         if pkg_end > initial_next {
             let skip = usize::try_from(initial_next.saturating_sub(pkg_start))
                 .map_err(|_| Error::from("skip count overflows usize"))?;
-            chunk.extend(pkg.entries.into_iter().skip(skip));
+            let tail = pkg.entries.into_iter().skip(skip);
+            for entry in tail {
+                chunk_bytes = chunk_bytes.saturating_add(entry.len());
+                chunk.push(entry);
+            }
             chunk_end = pkg_end;
             chunk_pkgs += 1;
 
-            // `commit_packages >= 1` (config), so a full chunk always holds
-            // at least one package's entries: no empty-flush guard needed.
-            if chunk_pkgs == commit_packages {
+            // Flush when either cap trips: `commit_packages` bounds the
+            // package count, `max_chunk_bytes` bounds peak memory when
+            // individual packages are large. `commit_packages >= 1`
+            // (config), so a full chunk always holds at least one package's
+            // entries; the byte cap only fires after entries were buffered,
+            // so neither branch can flush an empty chunk.
+            if chunk_pkgs == commit_packages || chunk_bytes >= max_chunk_bytes {
                 (frontier_size, frontier_hash) = flush_chunk(
                     bucket,
                     env,
@@ -459,6 +473,7 @@ where
                 )
                 .await?;
                 chunk_pkgs = 0;
+                chunk_bytes = 0;
             }
         }
     }