From c3312023dc8f90edaec75dec601e8c54129c8190 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 21:28:37 +0000 Subject: [PATCH 1/2] Docker proxies and imports: an upstream lost mid-blob answers 503, not 500 Staging an upstream blob tags the body's read errors with an UpstreamBodyError inside the I/O error, as push.rs tags request bodies, and From for Failure maps it to the upstream-unreachable failure (503, logged as a warning). An import's item names the source registry as unreachable. Staging's own I/O errors stay 500. Fixes #689 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012e6DJZDiJztZTi98zcU8Mj --- crates/server/src/images/import.rs | 16 +++++-- crates/server/src/images/mod.rs | 67 ++++++++++++++++++++++++++++++ crates/server/src/images/proxy.rs | 5 +-- crates/server/tests/registry.rs | 54 ++++++++++++++++++++++++ docs/changes/2.2.0/689.fixed.md | 6 +++ 5 files changed, 141 insertions(+), 7 deletions(-) create mode 100644 docs/changes/2.2.0/689.fixed.md diff --git a/crates/server/src/images/import.rs b/crates/server/src/images/import.rs index 2e7055e2..a6e1aa68 100644 --- a/crates/server/src/images/import.rs +++ b/crates/server/src/images/import.rs @@ -49,7 +49,7 @@ 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 +609,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}" diff --git a/crates/server/src/images/mod.rs b/crates/server/src/images/mod.rs index 81a9b424..7ee15f0f 100644 --- a/crates/server/src/images/mod.rs +++ b/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::())) +} + impl From for Failure { fn from(e: forge_storage::CasError) -> Self { match e { @@ -352,6 +386,7 @@ impl From 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); + } } diff --git a/crates/server/src/images/proxy.rs b/crates/server/src/images/proxy.rs index 5a5c6d47..35449ee4 100644 --- a/crates/server/src/images/proxy.rs +++ b/crates/server/src/images/proxy.rs @@ -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) { diff --git a/crates/server/tests/registry.rs b/crates/server/tests/registry.rs index 2456fe82..195839b1 100644 --- a/crates/server/tests/registry.rs +++ b/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>>, + /// 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). diff --git a/docs/changes/2.2.0/689.fixed.md b/docs/changes/2.2.0/689.fixed.md new file mode 100644 index 00000000..ad2da1a3 --- /dev/null +++ b/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`. From 06f5a5539c05c1a262920517269c2b6da8a19693 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 21:38:01 +0000 Subject: [PATCH 2/2] Drop the imports the upstream body reader left unused Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012e6DJZDiJztZTi98zcU8Mj --- crates/server/src/images/import.rs | 1 - crates/server/src/images/proxy.rs | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/server/src/images/import.rs b/crates/server/src/images/import.rs index a6e1aa68..96df4ead 100644 --- a/crates/server/src/images/import.rs +++ b/crates/server/src/images/import.rs @@ -42,7 +42,6 @@ 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; diff --git a/crates/server/src/images/proxy.rs b/crates/server/src/images/proxy.rs index 35449ee4..ef07dd26 100644 --- a/crates/server/src/images/proxy.rs +++ b/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;