Docker proxies and imports: an upstream lost mid-blob answers 503, not 500

crates/server/src/images/import.rs

@@ -42,14 +42,13 @@ use forge_plugin::api::images::{
     MAX_MANIFEST_BYTES,
 };
 use forge_plugin::api::{UpstreamAnswer, UpstreamOp};
-use futures_util::TryStreamExt;
 use reqwest::Method;
 
 use super::digest::ContentDigest;
 use super::dispatch::Target;
 use super::proxy::{content_type, read_limited, Upstream};
 use super::push::require_linked;
-use super::{Failure, ImgResult, Registry};
+use super::{is_upstream_read_error, upstream_body_reader, Failure, ImgResult, Registry};
 
 /// How deep indexes may nest below the copied manifest.
 const MAX_DEPTH: usize = 4;
@@ -609,12 +608,20 @@ impl Registry {
         if response.content_length().is_some_and(|l| l > max) {
             return Err(Failure::new(ImageError::TooLarge { limit: max }));
         }
-        let body = response.bytes_stream().map_err(std::io::Error::other);
         let staged = self
             .inner
             .cas
-            .stage(tokio_util::io::StreamReader::new(body), max)
-            .await?;
+            .stage(upstream_body_reader(response), max)
+            .await
+            .map_err(|e| {
+                if is_upstream_read_error(&e) {
+                    Failure::from(e).with_text(format!(
+                        "the source registry is unreachable: reading {digest} failed"
+                    ))
+                } else {
+                    Failure::from(e)
+                }
+            })?;
         if !digest.matches(&staged.sha256, &staged.sha512_base64) {
             return Err(Failure::unavailable(format!(
                 "the source served a blob that does not match its digest {digest}"

crates/server/src/images/mod.rs

@@ -332,6 +332,40 @@ impl std::fmt::Display for RequestBodyError {
 
 impl std::error::Error for RequestBodyError {}
 
+/// An upstream response body that could not be read: the connection was
+/// lost, or it stalled past the idle timeout or the download deadline.
+/// Travels inside the I/O error staging the body fails with, and is the
+/// upstream's fault, not storage's.
+#[derive(Debug)]
+pub(crate) struct UpstreamBodyError(pub reqwest::Error);
+
+impl std::fmt::Display for UpstreamBodyError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "reading the upstream response: {}", self.0)
+    }
+}
+
+impl std::error::Error for UpstreamBodyError {}
+
+/// An upstream response body as an `AsyncRead`. A body that cannot be read
+/// fails with an [`UpstreamBodyError`] inside the I/O error.
+pub(crate) fn upstream_body_reader(
+    response: reqwest::Response,
+) -> impl tokio::io::AsyncRead + Unpin + Send {
+    use futures_util::TryStreamExt;
+    tokio_util::io::StreamReader::new(
+        response
+            .bytes_stream()
+            .map_err(|e| std::io::Error::other(UpstreamBodyError(e))),
+    )
+}
+
+/// Whether staging failed because the upstream body could not be read.
+pub(crate) fn is_upstream_read_error(e: &forge_storage::CasError) -> bool {
+    matches!(e, forge_storage::CasError::Io(io)
+        if io.get_ref().is_some_and(|inner| inner.is::<UpstreamBodyError>()))
+}
+
 impl From<forge_storage::CasError> for Failure {
     fn from(e: forge_storage::CasError) -> Self {
         match e {
@@ -352,6 +386,7 @@ impl From<forge_storage::CasError> for Failure {
                 tracing::debug!(error = %e, "registry: the upload body could not be read");
                 Failure::new(ImageError::BodyUnreadable)
             }
+            e if is_upstream_read_error(&e) => proxy::upstream_unreachable(&e),
             other => Failure::internal("storage error", other),
         }
     }
@@ -399,4 +434,36 @@ mod tests {
         let e = Failure::from(CasError::Io(std::io::Error::other("disk full")));
         assert_eq!(e.error, ImageError::Internal);
     }
+
+    #[tokio::test]
+    async fn an_unreadable_upstream_body_is_the_upstreams_error() {
+        // A body cut short: its `Content-Length` promises more than it sends.
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            use tokio::io::AsyncWriteExt;
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let mut request = [0u8; 1024];
+            let _ = tokio::io::AsyncReadExt::read(&mut socket, &mut request).await;
+            let _ = socket
+                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 100\r\n\r\nshort")
+                .await;
+        });
+        let response = reqwest::get(format!("http://{addr}/")).await.unwrap();
+        let mut body = Vec::new();
+        let read =
+            tokio::io::AsyncReadExt::read_to_end(&mut upstream_body_reader(response), &mut body)
+                .await
+                .unwrap_err();
+        let e = CasError::Io(read);
+        assert!(is_upstream_read_error(&e));
+        assert_eq!(
+            Failure::from(e).error,
+            ImageError::Unavailable("the upstream registry is unreachable".into())
+        );
+        // Staging's own I/O errors stay server errors.
+        let e = CasError::Io(std::io::Error::other("disk full"));
+        assert!(!is_upstream_read_error(&e));
+        assert_eq!(Failure::from(e).error, ImageError::Internal);
+    }
 }

crates/server/src/images/proxy.rs

@@ -37,7 +37,7 @@ use forge_plugin::api::{
 };
 use forge_storage::{BlobRef, StagedBlob};
 use forge_telemetry::proxy::{Kind as LookupKind, Lookup, Outcome};
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::StreamExt;
 use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, AUTHORIZATION, WWW_AUTHENTICATE};
 use reqwest::Method;
 use tracing::Instrument;
@@ -47,7 +47,7 @@ use super::dispatch::Target;
 use super::download::Download;
 use super::pull::{BlobHit, ManifestHit};
 use super::push::{merge, require_linked};
-use super::{Failure, ImgResult, Registry};
+use super::{upstream_body_reader, Failure, ImgResult, Registry};
 
 /// Record a proxy `lookup` that answered `result`: what its path marked,
 /// else a hit for something found and "not found" for nothing. A failure is
@@ -993,11 +993,10 @@ impl Registry {
                 self.start_download(repo, &local, digest, size, response, upstream.http.clone())?;
             return Ok(Some(BlobHit::downloading(repo, &local, digest, download)));
         }
-        let body = response.bytes_stream().map_err(std::io::Error::other);
         let staged = self
             .inner
             .cas
-            .stage(tokio_util::io::StreamReader::new(body), max)
+            .stage(upstream_body_reader(response), max)
             .await?;
         upstream.http.count_bytes(staged.size);
         if !digest.matches(&staged.sha256, &staged.sha512_base64) {

crates/server/tests/registry.rs

@@ -2359,6 +2359,8 @@ struct MockRegistry {
     /// Send the first half of every blob `GET`, then wait for a permit of
     /// this before the rest.
     blob_hold: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
+    /// Send the first half of every blob `GET`, then drop the connection.
+    cut_blobs: AtomicBool,
 }
 
 impl MockRegistry {
@@ -2484,6 +2486,19 @@ impl MockRegistry {
                 if self.corrupt.load(Ordering::SeqCst) {
                     body.push(0);
                 }
+                if self.cut_blobs.load(Ordering::SeqCst) {
+                    let length = body.len();
+                    body.truncate(length / 2);
+                    let cut = futures_util::stream::iter([
+                        Ok::<_, std::io::Error>(body),
+                        Err(std::io::Error::other("connection lost")),
+                    ]);
+                    return (
+                        [(header::CONTENT_LENGTH, length.to_string())],
+                        Body::from_stream(cut),
+                    )
+                        .into_response();
+                }
                 let Some(hold) = self.blob_hold.lock().unwrap().clone() else {
                     return body.into_response();
                 };
@@ -2954,6 +2969,45 @@ async fn a_downloading_layer_that_does_not_match_its_digest_is_cut_short() {
     assert!(linked_soon(&srv, &proxy).await, "cached");
 }
 
+/// An upstream that loses the connection in the middle of a small blob (one
+/// Forge stages before it answers) is the upstream's failure: 503, not 500,
+/// and nothing is cached (#689).
+#[tokio::test]
+async fn an_upstream_cut_short_mid_blob_is_unreachable() {
+    let srv = common::start().await;
+    let mock = MockRegistry::start().await;
+    let config = br#"{"architecture":"amd64","os":"linux"}"#.repeat(64);
+    mock.put_blob(&config);
+    mock.cut_blobs.store(true, Ordering::SeqCst);
+    let proxy = repo_name("proxy");
+    create_repo(
+        &srv,
+        json!({
+            "name": proxy, "kind": "proxy", "format": "docker", "remote_url": mock.base(),
+            "upstream_username": "upstream", "upstream_password": "secret",
+        }),
+    )
+    .await;
+    let oci = admin_oci(&srv).await;
+    let name = format!("{proxy}/library/cut");
+    let path = format!("blobs/{}", sha(&config));
+
+    let resp = oci.get(&name, &path).await;
+    assert_eq!(resp.status(), 503);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert!(
+        body.to_string()
+            .contains("the upstream registry is unreachable"),
+        "{body}"
+    );
+    assert_eq!(links(&srv, &proxy).await, 0, "not cached");
+
+    mock.cut_blobs.store(false, Ordering::SeqCst);
+    let resp = oci.get(&name, &path).await;
+    assert_eq!(resp.status(), 200);
+    assert_eq!(resp.bytes().await.unwrap().as_ref(), config.as_slice());
+}
+
 /// A client that stops reading a layer a proxy serves while it downloads it
 /// holds none of the staging space once the layer is stored, and reads the
 /// rest from the store when it goes on (#763).

docs/changes/2.2.0/689.fixed.md

@@ -0,0 +1,6 @@
+A Docker proxy whose upstream lost the connection, or stalled past its
+timeouts, in the middle of a blob of at most 1 MiB (an image config) or of
+one sent without a `Content-Length` answered 500 and logged a storage
+error; it answers 503, `the upstream registry is unreachable`, and logs a
+warning. An image import's item that fails the same way names the source
+registry as unreachable instead of reading `internal error`.