mirror_worker: stream and incrementally commit add-entries uploads

Replace the whole-body buffer with a streaming decode: gunzip and parse
the body as it arrives through a small stream-buffer adapter (retry the
parse as chunks land, consume on success), so a large upload no longer
holds the entire body in memory.

Persist incrementally every `commit_packages` packages (default 32, config
capped at 1024) so a long or interrupted upload advances the frontier as
it streams rather than only at the end. Each flush commits from the last
persisted frontier and is resumable; the running frontier is threaded
locally across flushes so the DO is not re-queried between chunks. Tiles
are immutable and content-addressed and the DO advance is a monotone
compare-and-swap, so a repeated or concurrent flush of the same range is
harmless.

Author: lukevalenta

crates/mirror_worker/config.dev.json

@@ -5,6 +5,7 @@
     "submission_prefix": "http://localhost:8787/",
     "monitoring_prefix": "http://localhost:8787/",
     "clean_interval_secs": 5,
+    "commit_packages": 2,
     "logs": {
         "oid/1.3.6.1.4.1.32473.2": {
             "description": "Dev-only MTC CA cosigner. Key name is the CA ID; the mirror serves log numbers 1-6 as origins oid/1.3.6.1.4.1.32473.2.0.<N>. log_public_keys holds a dev-only ML-DSA-44 SPKI.",

crates/mirror_worker/config.schema.json

@@ -30,6 +30,13 @@
             "default": 3600,
             "description": "How often (in seconds) the per-origin partial-tile cleaner wakes to clean orphaned partial tiles from object storage. Defaults to 3600 (one hour) when omitted."
         },
+        "commit_packages": {
+            "type": "integer",
+            "minimum": 1,
+            "maximum": 1024,
+            "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."
+        },
         "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

@@ -62,6 +62,13 @@ pub struct AppConfig {
     /// back to a one-hour default (see [`Self::clean_interval_secs`]).
     /// Consumed by [`mirror_worker`](../mirror_worker/)'s `cleaner_do`.
     pub clean_interval_secs: Option<u64>,
+    /// How many entry packages the `add-entries` handler 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. `None` falls back to a default of 32
+    /// (see [`Self::commit_packages`]). Consumed by
+    /// [`mirror_worker`](../mirror_worker/)'s `add_entries`.
+    pub commit_packages: 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.
@@ -118,6 +125,17 @@ impl AppConfig {
         self.clean_interval_secs.unwrap_or(3600)
     }
 
+    /// How many entry packages `add-entries` commits per flush, falling
+    /// back to 32 when `commit_packages` is unset. 32 matches the
+    /// per-request package budget clients are recommended to stay within
+    /// (tlog-mirror "Implementation Considerations"), so a compliant
+    /// single-request upload still commits once, while larger uploads
+    /// flush every 32 packages instead of buffering the whole body.
+    #[must_use]
+    pub fn commit_packages(&self) -> u64 {
+        self.commit_packages.unwrap_or(32)
+    }
+
     /// Validate the configuration beyond what `serde` and the JSON schema
     /// can express.
     ///
@@ -141,9 +159,9 @@ impl AppConfig {
     ///    signed-note key name length cap, since each origin is itself
     ///    used as a checkpoint origin.
     ///
-    /// Simple single-field bounds (e.g. the log-number ranges) are
-    /// expressed in `config.schema.json` and enforced by the build
-    /// script, so they are not re-checked here.
+    /// Simple single-field bounds (e.g. `commit_packages` and the
+    /// log-number ranges) are expressed in `config.schema.json` and
+    /// enforced by the build script, so they are not re-checked here.
     ///
     /// `log_key_name` uniqueness across log entries is not checked here;
     /// it is enforced earlier, during deserialization (see
@@ -340,6 +358,7 @@ mod tests {
             submission_prefix: "https://mirror.example/".to_owned(),
             monitoring_prefix: None,
             clean_interval_secs: None,
+            commit_packages: None,
             logs: HashMap::from([(
                 "example.com/log1".to_owned(),
                 LogParams {
@@ -464,6 +483,14 @@ mod tests {
             .expect("a valid log-number window is accepted");
     }
 
+    #[test]
+    fn commit_packages_defaults_to_32() {
+        let mut cfg = good_app_config();
+        assert_eq!(cfg.commit_packages(), 32);
+        cfg.commit_packages = Some(8);
+        assert_eq!(cfg.commit_packages(), 8);
+    }
+
     #[test]
     fn validate_rejects_inverted_window() {
         let cfg = with_log(|log| {
@@ -532,6 +559,7 @@ mod tests {
             submission_prefix: "https://mirror.example/".to_owned(),
             monitoring_prefix: None,
             clean_interval_secs: None,
+            commit_packages: None,
             logs: HashMap::from([(
                 "a".repeat(250),
                 LogParams {