mirror_worker: map malformed-gzip bodies to 400, not 500

Gzip decode/trailer failures were surfaced as worker::Error, which the
handler's blanket From<worker::Error> mapped to 500. A malformed or
truncated gzip request body is a client fault. Introduce a typed
BodyError (Decode vs Transport) on the body stream so decode faults map
to 400 while genuine transport failures stay 500; parse_next_package now
returns ApiResult so the distinction survives to the handler.

Author: lukevalenta

crates/mirror_worker/src/add_entries.rs

@@ -48,7 +48,8 @@ use worker::*;
 use generic_log_worker::{ObjectBackend, util::now_millis};
 
 use crate::{
-    body, commit,
+    body::{self, BodyError},
+    commit,
     frontend_worker::{ApiResult, AppError},
     load_mirror_signer, load_ticket_sealer, log_verifiers,
     mirror_state_do::{
@@ -237,7 +238,7 @@ async fn stream_and_commit<S>(
     first_prefix: &[Vec<u8>],
 ) -> ApiResult<axum::response::Response>
 where
-    S: futures_util::Stream<Item = Result<Vec<u8>>> + Unpin,
+    S: futures_util::Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin,
 {
     let bucket = load_origin_bucket(env, &header.log_origin)?;
 
@@ -336,7 +337,7 @@ async fn persist_packages<S, O>(
     start: &NextEntry,
 ) -> ApiResult<StreamResult>
 where
-    S: futures_util::Stream<Item = Result<Vec<u8>>> + Unpin,
+    S: futures_util::Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin,
     O: ObjectBackend,
 {
     // config.schema.json caps commit_packages (max 1024), enforced by the
@@ -728,7 +729,7 @@ enum ParseOutcome {
 /// retry-on-`UnexpectedEof` loop terminates.
 async fn parse_header<S>(buf: &mut StreamBuffer<S>) -> ApiResult<AddEntriesRequestHeader>
 where
-    S: futures_util::Stream<Item = Result<Vec<u8>>> + Unpin,
+    S: futures_util::Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin,
 {
     loop {
         let mut cursor = Cursor::new(buf.buffered());
@@ -765,9 +766,12 @@ where
 /// Read the next entry package from `buf`, pulling more chunks from
 /// the underlying stream until the package parses or the stream ends.
 /// See [`ParseOutcome`] for the four cases.
-async fn parse_next_package<S>(buf: &mut StreamBuffer<S>, num_entries: u64) -> Result<ParseOutcome>
+async fn parse_next_package<S>(
+    buf: &mut StreamBuffer<S>,
+    num_entries: u64,
+) -> ApiResult<ParseOutcome>
 where
-    S: futures_util::Stream<Item = Result<Vec<u8>>> + Unpin,
+    S: futures_util::Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin,
 {
     // EOF with an empty buffer: clean truncation between packages.
     if buf.is_eof() && buf.len() == 0 {

crates/mirror_worker/src/body.rs

@@ -31,10 +31,31 @@ use worker::*;
 
 use crate::frontend_worker::{ApiResult, AppError};
 
+/// An error surfaced by a [`BodyStream`], distinguishing a client-side
+/// decode fault from a transport failure so the `add-entries` handler can
+/// map each to the right HTTP status (see `From<BodyError> for AppError`).
+#[derive(Debug)]
+pub(crate) enum BodyError {
+    /// Malformed or truncated gzip body: a client fault, mapped to 400.
+    Decode(String),
+    /// Transport failure reading the underlying request body: mapped to
+    /// 500.
+    Transport(Error),
+}
+
+impl std::fmt::Display for BodyError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            BodyError::Decode(e) => write!(f, "{e}"),
+            BodyError::Transport(e) => write!(f, "{e}"),
+        }
+    }
+}
+
 /// A boxed, `Unpin` body stream, the common type the `add-entries`
 /// handler feeds to [`crate::stream_buffer::StreamBuffer`] regardless of
 /// whether the request body was identity- or gzip-encoded.
-pub(crate) type BodyStream = Pin<Box<dyn Stream<Item = Result<Vec<u8>>>>>;
+pub(crate) type BodyStream = Pin<Box<dyn Stream<Item = std::result::Result<Vec<u8>, BodyError>>>>;
 
 /// Open the request body as a decoded chunk stream, honoring
 /// `Content-Encoding`.
@@ -62,7 +83,7 @@ pub(crate) fn decoded_stream(
     // `Result<Vec<u8>>` chunk contract the buffer/gunzip pipeline expects.
     let raw = body.into_data_stream().map(|r| {
         r.map(|b| b.to_vec())
-            .map_err(|e| Error::from(e.to_string()))
+            .map_err(|e| BodyError::Transport(Error::from(e.to_string())))
     });
     let stream: BodyStream = match encoding.as_str() {
         "" | "identity" => Box::pin(raw),
@@ -125,7 +146,7 @@ impl GzipInflater {
 /// terminal `Err` item, after which the stream ends.
 pub(crate) fn gunzip<S>(inner: S) -> BodyStream
 where
-    S: Stream<Item = Result<Vec<u8>>> + Unpin + 'static,
+    S: Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin + 'static,
 {
     struct DecodeState<S> {
         inner: S,
@@ -151,7 +172,7 @@ where
                             Ok(out) => out,
                             Err(e) => {
                                 st.done = true;
-                                return Some((Err(e), st));
+                                return Some((Err(BodyError::Decode(e.to_string())), st));
                             }
                         };
                         // A chunk may not yet yield any plaintext (partial
@@ -172,7 +193,7 @@ where
                         return match inflater.finish() {
                             Ok(tail) if !tail.is_empty() => Some((Ok(tail), st)),
                             Ok(_) => None,
-                            Err(e) => Some((Err(e), st)),
+                            Err(e) => Some((Err(BodyError::Decode(e.to_string())), st)),
                         };
                     }
                 }
@@ -199,13 +220,15 @@ mod tests {
     fn chunked_stream(
         bytes: &[u8],
         size: usize,
-    ) -> impl Stream<Item = Result<Vec<u8>>> + Unpin + 'static {
-        let chunks: Vec<Result<Vec<u8>>> =
+    ) -> impl Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin + 'static {
+        let chunks: Vec<std::result::Result<Vec<u8>, BodyError>> =
             bytes.chunks(size.max(1)).map(|c| Ok(c.to_vec())).collect();
         stream::iter(chunks)
     }
 
-    async fn collect(mut s: impl Stream<Item = Result<Vec<u8>>> + Unpin) -> Result<Vec<u8>> {
+    async fn collect(
+        mut s: impl Stream<Item = std::result::Result<Vec<u8>, BodyError>> + Unpin,
+    ) -> std::result::Result<Vec<u8>, BodyError> {
         let mut out = Vec::new();
         while let Some(item) = s.next().await {
             out.extend_from_slice(&item?);
@@ -249,7 +272,10 @@ mod tests {
         // ends mid-member; finish() must report the truncation.
         compressed.truncate(compressed.len() - 6);
         let err = collect(gunzip(chunked_stream(&compressed, 4))).await;
-        assert!(err.is_err(), "truncated gzip must surface an error");
+        assert!(
+            matches!(err, Err(BodyError::Decode(_))),
+            "truncated gzip must surface a client decode error"
+        );
     }
 
     #[tokio::test]
@@ -259,14 +285,18 @@ mod tests {
         let mid = compressed.len() / 2;
         compressed[mid] ^= 0xff;
         let err = collect(gunzip(chunked_stream(&compressed, 5))).await;
-        assert!(err.is_err(), "corrupt gzip must surface an error");
+        assert!(
+            matches!(err, Err(BodyError::Decode(_))),
+            "corrupt gzip must surface a client decode error"
+        );
     }
 
     #[tokio::test]
     async fn upstream_error_propagates() {
         let compressed = gzip(b"partial");
-        let mut chunks: Vec<Result<Vec<u8>>> = vec![Ok(compressed[..4].to_vec())];
-        chunks.push(Err(Error::from("boom")));
+        let mut chunks: Vec<std::result::Result<Vec<u8>, BodyError>> =
+            vec![Ok(compressed[..4].to_vec())];
+        chunks.push(Err(BodyError::Transport(Error::from("boom"))));
         let err = collect(gunzip(stream::iter(chunks))).await;
         assert!(err.is_err(), "upstream stream error must propagate");
     }

crates/mirror_worker/src/frontend_worker.rs

@@ -140,6 +140,16 @@ impl From<worker::Error> for AppError {
     }
 }
 
+impl From<crate::body::BodyError> for AppError {
+    fn from(err: crate::body::BodyError) -> Self {
+        match err {
+            // Malformed/truncated gzip is a client fault.
+            crate::body::BodyError::Decode(msg) => Self::BadRequest(msg),
+            crate::body::BodyError::Transport(e) => Self::InternalServerError(e.to_string()),
+        }
+    }
+}
+
 impl IntoResponse for AppError {
     fn into_response(self) -> axum::response::Response {
         match self {