🐛 fix(oci): answer a failed request body as a client error

An upload whose body failed answered 502 with "upstream transfer failed". No
upstream serves a request body: the bytes were coming from the client and peryx
is the server. The stall bound #2182 added made it reachable with the client
still connected, so a client that paused for thirty seconds was told something
upstream of peryx had gone wrong.

502 also carries a meaning to intermediaries, a bad response from an upstream
server, which invites a retry against a different backend for a condition no
backend change reaches.

A stall now answers 408, which says the server gave up waiting for a message
that never arrived, and names the offset a resumable session stands at so the
client knows where to continue. Anything else the body ends with answers 400,
since repeating it unchanged fails the same way. Both come from the shared
classification rather than from each call site reading an opaque error.

Four tests asserted the old status. Their intent survives unchanged: the stall
pair asserted a cut chunk must not read as accepted, which 408 satisfies, and
the monolithic case was named for the gateway status it happened to produce
rather than for a gateway being involved.

Author: gaborbernat

crates/peryx-driver/src/body.rs

@@ -1,4 +1,5 @@
 use std::io::{Read as _, Seek as _, SeekFrom};
+use std::time::Duration;
 
 use axum::body::Body;
 use bytes::Bytes;
@@ -91,3 +92,61 @@ pub fn pipelined_file(file: std::fs::File, offset: u64, length: u64) -> Body {
         rx.recv().await.map(|chunk| (chunk, rx))
     }))
 }
+
+/// The error a stalled body ends with, worded for the client that stopped sending rather than for the
+/// handler that was reading.
+#[derive(Debug)]
+pub struct Stalled(Duration);
+
+impl std::error::Error for Stalled {}
+
+impl std::fmt::Display for Stalled {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(formatter, "the request body sent nothing for {:?}", self.0)
+    }
+}
+
+impl Stalled {
+    #[must_use]
+    pub const fn new(after: Duration) -> Self {
+        Self(after)
+    }
+}
+
+/// Why reading a request body ended without the bytes the handler was waiting for.
+///
+/// The bytes of a request body come from the client, so no failure reading one is an upstream fault
+/// and none of them may answer `502`. What the client should do next still differs, so the edge that
+/// bounds the body is where the two are told apart: a handler holding an opaque body error cannot
+/// recover the distinction, and every handler that tried would derive it again.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BodyFailure {
+    /// The client sent no frame for the bound. The request never completed, so it may be repeated,
+    /// and a resumable session picks up at the offset its bytes reached.
+    Stalled(Duration),
+    /// The body stopped for some other reason: a dropped connection, or framing the server could not
+    /// read. Repeating it unchanged fails the same way.
+    Interrupted,
+}
+
+impl BodyFailure {
+    /// Classify the error a request-body stream ended with.
+    ///
+    /// The stall arrives wrapped by whatever read the body, so the whole source chain is searched
+    /// rather than the outermost error alone.
+    #[must_use]
+    pub fn of(error: &(dyn std::error::Error + 'static)) -> Self {
+        let mut current = Some(error);
+        while let Some(error) = current {
+            if let Some(stalled) = error.downcast_ref::<Stalled>() {
+                return Self::Stalled(stalled.0);
+            }
+            current = error.source();
+        }
+        Self::Interrupted
+    }
+}
+
+#[cfg(test)]
+#[path = "../tests/unit/body_failure_tests.rs"]
+mod body_failure_tests;

…x-http/tests/unit/request_stall_tests.rs …-driver/tests/unit/body_failure_tests.rscrates/peryx-http/tests/unit/request_stall_tests.rs renamed to crates/peryx-driver/tests/unit/body_failure_tests.rs crates/peryx-http/tests/unit/request_stall_tests.rs renamed to crates/peryx-driver/tests/unit/body_failure_tests.rs

@@ -19,17 +19,19 @@ impl std::fmt::Display for Wrapper {
     }
 }
 
+/// A handler never sees the stall itself: whatever read the body wraps it, so the chain is what the
+/// classification has to walk.
 #[test]
 fn test_a_stall_is_recognized_through_the_reader_that_wrapped_it() {
-    let stalled = Wrapper(Box::new(Stalled(Duration::from_secs(30))));
+    let stalled = Wrapper(Box::new(Stalled::new(Duration::from_secs(30))));
 
     assert_eq!(BodyFailure::of(&stalled), BodyFailure::Stalled(Duration::from_secs(30)));
 }
 
 #[test]
 fn test_a_bare_stall_is_recognized() {
     assert_eq!(
-        BodyFailure::of(&Stalled(Duration::from_secs(5))),
+        BodyFailure::of(&Stalled::new(Duration::from_secs(5))),
         BodyFailure::Stalled(Duration::from_secs(5))
     );
 }

crates/peryx-ecosystem-oci/src/registry/uploads.rs

@@ -3,7 +3,8 @@ use super::blobs::{
     publish_acknowledged, release_reservation, upload_epoch,
 };
 use super::*;
-use crate::error::{ErrorCode, error_response};
+use crate::error::{ErrorCode, error_response, error_response_with_status};
+use peryx_driver::body::BodyFailure;
 use crate::registry::acknowledge::BlobAck;
 use crate::registry::authority::{EpochCommit, commit_epoch};
 use crate::store::{self};
@@ -485,7 +486,7 @@ async fn append_to_stage(
     let mut stream = body.into_data_stream();
     let limit = index.policy.max_artifact_size();
     while let Some(chunk) = stream.next().await {
-        let chunk = chunk.map_err(|err| UploadBodyError::Fault(ServeError::Transport(err.to_string())))?;
+        let chunk = chunk.map_err(|err| UploadBodyError::ClientBody(BodyFailure::of(&err), Some(*offset)))?;
         let size = *offset + chunk.len() as u64;
         if limit.is_some_and(|limit| size > limit) {
             return Err(UploadBodyError::Denied(
@@ -516,16 +517,39 @@ async fn append_to_stage(
 enum UploadBodyError {
     Fault(ServeError),
     Denied(Response),
+    /// The client's bytes stopped arriving, carrying the offset a resumable session reached.
+    ClientBody(BodyFailure, Option<u64>),
     /// The session's row went while the chunk was being written, so the offset this append reached
     /// describes an upload the registry no longer has.
     Vanished,
 }
 
+/// Answer a request body that failed on the client's side.
+///
+/// Nothing upstream serves a request body, so `502` would name the wrong party and invite an
+/// intermediary to retry against a different backend for a condition no backend change reaches. A
+/// stall is the server giving up on a message that never arrived, which is what `408` says, and it
+/// leaves the client free to repeat the request or resume the session at the offset its bytes reached.
+fn client_body_response(failure: BodyFailure, resume: Option<u64>) -> Response {
+    match failure {
+        BodyFailure::Stalled(after) => {
+            let resumable = resume.map_or_else(String::new, |offset| format!(", resumable from byte {offset}"));
+            error_response_with_status(
+                StatusCode::REQUEST_TIMEOUT,
+                ErrorCode::BlobUploadInvalid,
+                &format!("the request body sent nothing for {after:?}{resumable}"),
+            )
+        }
+        BodyFailure::Interrupted => error_response(ErrorCode::BlobUploadInvalid, "the request body ended early"),
+    }
+}
+
 impl UploadBodyError {
     fn into_response(self) -> Result<Response, ServeError> {
         match self {
             Self::Fault(err) => Err(err),
             Self::Denied(response) => Ok(response),
+            Self::ClientBody(failure, resume) => Ok(client_body_response(failure, resume)),
             Self::Vanished => Ok(error_response(ErrorCode::BlobUploadUnknown, "upload unknown")),
         }
     }
@@ -556,7 +580,7 @@ async fn append_body(
     let mut stream = body.into_data_stream();
     let limit = index.policy.max_artifact_size();
     while let Some(chunk) = stream.next().await {
-        let chunk = chunk.map_err(|err| UploadBodyError::Fault(ServeError::Transport(err.to_string())))?;
+        let chunk = chunk.map_err(|err| UploadBodyError::ClientBody(BodyFailure::of(&err), None))?;
         let size = *offset + chunk.len() as u64;
         if limit.is_some_and(|limit| size > limit) {
             return Err(UploadBodyError::Denied(

crates/peryx-ecosystem-oci/tests/unit/tests/push_tests.rs

@@ -1478,7 +1478,7 @@ async fn test_session_finish_reports_an_admission_store_fault() {
 }
 
 #[tokio::test]
-async fn test_monolithic_upload_body_read_error_is_a_gateway_error() {
+async fn test_monolithic_upload_body_read_error_is_a_client_error() {
     let dir = tempfile::tempdir().unwrap();
     let (_state, app) = hosted_writable(&dir, TOKEN);
     let erroring = futures_util::stream::iter(vec![
@@ -1492,7 +1492,7 @@ async fn test_monolithic_upload_body_read_error_is_a_gateway_error() {
         .body(Body::from_stream(erroring))
         .unwrap();
     let response = app.clone().oneshot(request).await.unwrap();
-    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
     let _ = response.into_body().collect().await;
 }
 
@@ -2256,7 +2256,7 @@ async fn test_patch_body_read_error_keeps_session_resumable() {
         .body(Body::from_stream(chunks))
         .unwrap();
     let response = app.clone().oneshot(request).await.unwrap();
-    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
 
     let (status, headers, _) = send_with(&app, Method::GET, &location, &[("authorization", &auth(TOKEN))]).await;
     assert_eq!(status, StatusCode::NO_CONTENT);

crates/peryx-ecosystem-oci/tests/unit/tests/upload_stall_tests.rs

@@ -84,8 +84,8 @@ async fn test_a_stalled_chunk_leaves_a_session_the_client_can_resume() {
 
     assert_eq!(
         stalled,
-        StatusCode::BAD_GATEWAY,
-        "a cut chunk must not read as accepted"
+        StatusCode::REQUEST_TIMEOUT,
+        "a client that stopped sending is told the server gave up waiting, not that an upstream failed"
     );
     assert_eq!(
         landed.map(|record| record.offset),
@@ -183,7 +183,7 @@ async fn test_a_stalled_chunk_delays_the_reclaim_pass_by_one_bound() {
     let reclaimed = reclaim(&state).await;
     let waited = started.elapsed();
 
-    assert_eq!(appending.await.unwrap(), StatusCode::BAD_GATEWAY);
+    assert_eq!(appending.await.unwrap(), StatusCode::REQUEST_TIMEOUT);
     assert_eq!(reclaimed, 2, "the stalled session and the one queued behind it both go");
     assert!(
         (STALL_BOUND..STALL_BOUND * 2).contains(&waited),

crates/peryx-http/src/lib.rs

@@ -2,7 +2,6 @@
 
 pub mod handlers;
 mod request_stall;
-pub use request_stall::{BodyFailure, Stalled};
 mod response_framing;
 pub mod response_security;
 pub mod router;

crates/peryx-http/src/request_stall.rs

@@ -8,6 +8,7 @@ use std::time::Duration;
 
 use http_body::{Body, Frame, SizeHint};
 use pin_project_lite::pin_project;
+use peryx_driver::body::Stalled;
 use tokio::time::Sleep;
 
 pin_project! {
@@ -32,19 +33,6 @@ impl<B> StallBounded<B> {
     }
 }
 
-/// The error a stalled body ends with, worded for the client that stopped sending rather than for the
-/// handler that was reading.
-#[derive(Debug)]
-pub struct Stalled(Duration);
-
-impl std::error::Error for Stalled {}
-
-impl std::fmt::Display for Stalled {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(formatter, "the request body sent nothing for {:?}", self.0)
-    }
-}
-
 impl<B> Body for StallBounded<B>
 where
     B: Body,
@@ -60,7 +48,7 @@ where
         }
         let idle = this.idle.as_mut().as_pin_mut().expect("the wait was just armed");
         if idle.poll(cx).is_ready() {
-            return Poll::Ready(Some(Err(Box::new(Stalled(*this.stall)))));
+            return Poll::Ready(Some(Err(Box::new(Stalled::new(*this.stall)))));
         }
         let frame = ready!(this.body.poll_frame(cx));
         this.idle.set(None);
@@ -74,41 +62,3 @@ where
         self.body.size_hint()
     }
 }
-
-/// Why reading a request body ended without the bytes the handler was waiting for.
-///
-/// The bytes of a request body come from the client, so no failure reading one is an upstream fault
-/// and none of them may answer `502`. What the client should do next still differs, so the edge that
-/// bounds the body is where the two are told apart: a handler holding an opaque body error cannot
-/// recover the distinction, and every handler that tried would derive it again.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum BodyFailure {
-    /// The client sent no frame for the bound. The request never completed, so it may be repeated,
-    /// and a resumable session picks up at the offset its bytes reached.
-    Stalled(Duration),
-    /// The body stopped for some other reason: a dropped connection, or framing the server could not
-    /// read. Repeating it unchanged fails the same way.
-    Interrupted,
-}
-
-impl BodyFailure {
-    /// Classify the error a request-body stream ended with.
-    ///
-    /// The stall arrives wrapped by whatever read the body, so the whole source chain is searched
-    /// rather than the outermost error alone.
-    #[must_use]
-    pub fn of(error: &(dyn std::error::Error + 'static)) -> Self {
-        let mut current = Some(error);
-        while let Some(error) = current {
-            if let Some(stalled) = error.downcast_ref::<Stalled>() {
-                return Self::Stalled(stalled.0);
-            }
-            current = error.source();
-        }
-        Self::Interrupted
-    }
-}
-
-#[cfg(test)]
-#[path = "../tests/unit/request_stall_tests.rs"]
-mod request_stall_tests;