🐛 fix(pypi): tell a stalled upload from a malformed one

Every multipart read error answered 400. That is the right family, since the
bytes come from the client either way, but it merges two conditions whose
correct next move differs. A malformed form repeated unchanged fails again. A
stall says nothing about the form at all, so repeating the upload is exactly
what the client should do, and 400 told it the opposite.

A stalled body now answers 408 through the same classification the registry
uses, so the two ecosystems cannot drift on what a stalled request body means.

Author: gaborbernat

crates/peryx-ecosystem-pypi/src/serving/upload_form.rs

@@ -1,4 +1,5 @@
-use axum::extract::Multipart;
+use axum::extract::{Multipart, multipart};
+use peryx_driver::body::BodyFailure;
 use axum::http::{StatusCode, header};
 use axum::response::{IntoResponse, Response};
 use blake2::Blake2bVar;
@@ -38,7 +39,7 @@ pub(super) async fn collect_form(
     let mut form = UploadForm::default();
     let mut staged = None;
     let mut budget = FormBudget::default();
-    while let Some(field) = multipart.next_field().await.map_err(reject)? {
+    while let Some(field) = multipart.next_field().await.map_err(|error| body_reject(&error))? {
         budget.add_part()?;
         let field_name = field.name().unwrap_or_default().to_owned();
         if field_name == "content" {
@@ -214,7 +215,7 @@ async fn read_text_field(
     budget: &mut FormBudget,
 ) -> HttpResult<String> {
     let mut bytes = Vec::new();
-    while let Some(chunk) = field.chunk().await.map_err(reject)? {
+    while let Some(chunk) = field.chunk().await.map_err(|error| body_reject(&error))? {
         if bytes.len().saturating_add(chunk.len()) > limit {
             return Err((
                 StatusCode::BAD_REQUEST,
@@ -230,7 +231,7 @@ async fn read_text_field(
 }
 
 async fn drain_field(mut field: axum::extract::multipart::Field<'_>, budget: &mut FormBudget) -> HttpResult<()> {
-    while let Some(chunk) = field.chunk().await.map_err(reject)? {
+    while let Some(chunk) = field.chunk().await.map_err(|error| body_reject(&error))? {
         budget.add_text(chunk.len())?;
     }
     Ok(())
@@ -312,6 +313,24 @@ fn reject(err: impl std::fmt::Display) -> Response {
     (StatusCode::BAD_REQUEST, format!("bad upload: {err}")).into_response()
 }
 
+/// A read of the multipart body that failed.
+///
+/// The bytes come from the client either way, so the only question is whether it stopped sending or
+/// sent something the server could not read. `BodyFailure` answers that once for every ecosystem
+/// instead of each guessing from the message, and the two answers differ for the client: a stall says
+/// nothing about the form, so repeating the upload is the right move, while a malformed form repeated
+/// unchanged fails again.
+fn body_reject(error: &multipart::MultipartError) -> Response {
+    match BodyFailure::of(error) {
+        BodyFailure::Stalled(after) => (
+            StatusCode::REQUEST_TIMEOUT,
+            format!("upload stopped: the request body sent nothing for {after:?}"),
+        )
+            .into_response(),
+        BodyFailure::Interrupted => reject(error),
+    }
+}
+
 fn storage_reject(err: impl std::fmt::Display) -> Response {
     tracing::error!(error = %err, "upload staging failed");
     (

crates/peryx-ecosystem-pypi/tests/unit/tests/http/upload.rs

@@ -2190,3 +2190,30 @@ fn multipart_body_with_content_length(declared: u64, content: &[u8]) -> (String,
     head.extend_from_slice(&body[header_end + 4..]);
     (content_type, head)
 }
+
+/// A client that stops mid-form is told the server gave up waiting on a message that never arrived.
+/// It is not told its form was malformed, which would be wrong about the form and would suggest
+/// repeating the upload is pointless, and it is not told an upstream failed, since none serves a
+/// request body.
+#[tokio::test(start_paused = true)]
+async fn test_a_stalled_upload_body_reports_a_timeout() {
+    let h = harness().await;
+    let (content_type, body) = multipart_body(&[], Some(("peryxpkg-1.0-py3-none-any.whl", b"ab")));
+    let opening = Bytes::copy_from_slice(&body[..body.len() / 2]);
+    let stalling = futures_util::StreamExt::chain(
+        futures_util::stream::once(async move { Ok::<_, Infallible>(opening) }),
+        futures_util::stream::once(std::future::pending::<Result<Bytes, Infallible>>()),
+    );
+
+    let (status, _) = post_upload_body_with_headers_response(
+        &h.state,
+        "/root/pypi/",
+        Some(&upload_auth()),
+        &content_type,
+        &[],
+        Body::from_stream(stalling),
+    )
+    .await;
+
+    assert_eq!(status, StatusCode::REQUEST_TIMEOUT);
+}