mirror_worker: address streaming review nits

- Count only entry-contributing packages toward a chunk flush, so a run
  of already-persisted packages no longer triggers empty flushes; drop
  the now-redundant empty-chunk guard (commit_packages >= 1).
- Mark StreamBuffer eof on a stream error so is_eof() stays consistent
  and a recovering caller won't re-poll the failed stream.
- Explain the clean-vs-mid-package truncation branch in
  parse_next_package, and fix the stale parse_header doc that described
  a nested-Result return the code no longer uses.

Author: lukevalenta

crates/mirror_worker/src/add_entries.rs

@@ -393,17 +393,21 @@ where
         }
         packages_received += 1;
 
-        // Buffer only the not-yet-persisted tail of this package.
+        // Buffer only the not-yet-persisted tail of this package. Packages
+        // wholly below the request-start frontier are already persisted, so
+        // they contribute no entries and don't count toward a chunk flush;
+        // `chunk_pkgs` therefore tracks only packages that added buffered
+        // entries.
         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));
             chunk_end = pkg_end;
-        }
-        chunk_pkgs += 1;
+            chunk_pkgs += 1;
 
-        if chunk_pkgs == commit_packages {
-            if !chunk.is_empty() {
+            // `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 {
                 (frontier_size, frontier_hash) = flush_chunk(
                     bucket,
                     env,
@@ -414,8 +418,8 @@ where
                     &mut chunk,
                 )
                 .await?;
+                chunk_pkgs = 0;
             }
-            chunk_pkgs = 0;
         }
     }
 
@@ -721,8 +725,9 @@ enum ParseOutcome {
 
 /// Read the `add-entries` request header from `buf`, pulling more
 /// chunks from the underlying stream until the header parses or the
-/// stream errors. Returns `Ok(Ok(header))` on success or `Ok(Err(resp))`
-/// where `resp` is a fully-formed 400 response on malformed input.
+/// stream errors. Returns the parsed header, or
+/// [`AppError::BadRequest`] (400) if the input is malformed or truncated
+/// before the header is complete.
 ///
 /// The header has a bounded maximum size (u16 origin + u64s + u16
 /// ticket + hash + u8 proof-size + 63 hashes <= ~131 KB), so the
@@ -787,6 +792,11 @@ where
             }
             Err(ParseError::Io(ref e)) if e.kind() == ErrorKind::UnexpectedEof => {
                 if !buf.pull_one().await? {
+                    // Stream ended mid-parse. An empty buffer means the
+                    // previous package consumed exactly all buffered bytes
+                    // and this call started a fresh (never-arriving)
+                    // package: a clean between-package truncation. A
+                    // non-empty buffer holds a partial package.
                     if buf.len() == 0 {
                         return Ok(ParseOutcome::CleanEof);
                     }

crates/mirror_worker/src/stream_buffer.rs

@@ -35,9 +35,9 @@ pub(crate) struct StreamBuffer<S> {
     /// are reclaimed on the next [`Self::pull_one`]. The live buffer is
     /// `buf[start..]`.
     start: usize,
-    /// Set when the underlying stream has signalled end-of-stream.
-    /// Subsequent [`Self::pull_one`] calls return `Ok(false)`
-    /// without polling the (already-finished) stream.
+    /// Set when the underlying stream has signalled end-of-stream or
+    /// yielded an error. Subsequent [`Self::pull_one`] calls return
+    /// `Ok(false)` without polling the (already-finished) stream.
     eof: bool,
 }
 
@@ -60,11 +60,13 @@ where
     /// Pull one chunk from the underlying stream and append to the
     /// internal buffer. Returns `Ok(true)` if a chunk was appended,
     /// `Ok(false)` if the stream ended (clean EOF). Once the stream has
-    /// ended, all subsequent calls return `Ok(false)` without polling
-    /// the stream again.
+    /// ended or errored, all subsequent calls return `Ok(false)` without
+    /// polling the stream again.
     ///
     /// # Errors
-    /// Propagates any error from the underlying stream.
+    /// Propagates any error from the underlying stream. An error is
+    /// terminal: [`Self::is_eof`] is set so a caller that recovers from
+    /// the error does not re-poll the already-failed stream.
     pub async fn pull_one(&mut self) -> std::result::Result<bool, E> {
         if self.eof {
             return Ok(false);
@@ -80,7 +82,10 @@ where
                 self.buf.extend_from_slice(&chunk);
                 Ok(true)
             }
-            Some(Err(e)) => Err(e),
+            Some(Err(e)) => {
+                self.eof = true;
+                Err(e)
+            }
             None => {
                 self.eof = true;
                 Ok(false)
@@ -191,5 +196,9 @@ mod tests {
         assert!(buf.pull_one().await.unwrap());
         let err = buf.pull_one().await.unwrap_err();
         assert_eq!(err.to_string(), "boom");
+        // An error is terminal: eof is set and further pulls are no-ops
+        // rather than re-polling the already-failed stream.
+        assert!(buf.is_eof());
+        assert!(!buf.pull_one().await.unwrap());
     }
 }