From 32fb6689585475e42709c35d61a5a1ce18c89ce0 Mon Sep 17 00:00:00 2001 From: Alexander Larsson Date: Tue, 29 Sep 2026 11:24:23 +0200 Subject: [PATCH 1/5] aboot parsing: Minor cleanup Avoid indexing arrays when parsing the split cmdline, instead use CStr::from_bytes_until_nul() and iterators. Signed-off-by: Alexander Larsson --- crates/composefs-boot/src/android_boot.rs | 40 ++++++++++++++++------- 1 file changed, 28 insertions(+), 12 deletions(-) diff --git a/crates/composefs-boot/src/android_boot.rs b/crates/composefs-boot/src/android_boot.rs index 7b73a7b8..ef7bf93a 100644 --- a/crates/composefs-boot/src/android_boot.rs +++ b/crates/composefs-boot/src/android_boot.rs @@ -3,6 +3,7 @@ //! This module provides functionality to parse Android boot image format version 2 files //! and extract embedded components like kernel, initrd, commandline and dtb. +use std::ffi::CStr; use std::io::{Read, Seek, SeekFrom}; use thiserror::Error; use zerocopy::{ @@ -115,12 +116,12 @@ impl AndroidBootImage { } // mkbootimg splits long command lines (with a null terminator for each) - let primary_len = nul_terminated_len(&header.cmdline); - let extra_len = nul_terminated_len(&header.extra_cmdline); + let primary = nul_terminated_bytes(&header.cmdline); + let extra = nul_terminated_bytes(&header.extra_cmdline); let mut cmdline = [0; TOTAL_CMDLINE_SIZE]; - cmdline[..primary_len].copy_from_slice(&header.cmdline[..primary_len]); - cmdline[primary_len..primary_len + extra_len] - .copy_from_slice(&header.extra_cmdline[..extra_len]); + for (dst, src) in cmdline.iter_mut().zip(primary.iter().chain(extra)) { + *dst = *src; + } Ok(Self { page_size, @@ -179,16 +180,12 @@ impl AndroidBootImage { /// Return the kernel command line stored in the image header. pub fn cmdline(&self) -> Result<&str, AndroidBootError> { - let end = nul_terminated_len(&self.cmdline); - Ok(std::str::from_utf8(&self.cmdline[..end])?) + Ok(std::str::from_utf8(nul_terminated_bytes(&self.cmdline))?) } } -fn nul_terminated_len(bytes: &[u8]) -> usize { - bytes - .iter() - .position(|byte| *byte == 0) - .unwrap_or(bytes.len()) +fn nul_terminated_bytes(bytes: &[u8]) -> &[u8] { + CStr::from_bytes_until_nul(bytes).map_or(bytes, CStr::to_bytes) } fn add_aligned( @@ -305,6 +302,25 @@ pub(crate) mod tests { Ok(()) } + #[test] + fn parses_cmdline_without_nul_terminators() -> Result<(), AndroidBootError> { + let mut bytes = image(b"kernel", b"ramdisk", b""); + let header = BootImageHeaderV2::mut_from_bytes(&mut bytes[..HEADER_SIZE]) + .map_err(|_| AndroidBootError::InvalidHeader)?; + header.cmdline.fill(b'x'); + header.extra_cmdline.fill(b'y'); + + let image = AndroidBootImage::parse(&mut Cursor::new(bytes))?; + let expected = format!( + "{}{}", + "x".repeat(CMDLINE_SIZE), + "y".repeat(EXTRA_CMDLINE_SIZE) + ); + assert_eq!(image.cmdline()?, expected); + assert!(CStr::from_bytes_until_nul(&image.cmdline).is_err()); + Ok(()) + } + #[test] fn rejects_non_utf8_cmdline() -> Result<(), AndroidBootError> { let bytes = image_with_cmdline(b"kernel", b"ramdisk", b"", &[0xff]); From 2c11ff02b0499883681a540b8a0f9390ae320f86 Mon Sep 17 00:00:00 2001 From: Colin Walters Date: Tue, 22 Sep 2026 23:23:04 -0400 Subject: [PATCH 2/5] cfsctl: Print progress messages when stderr is not a terminal indicatif hides progress bars when stderr is not a terminal, and MultiProgress::println() then silently discards the message. So status messages from a pull never show up in CI logs, which is where the upcoming retry warnings for flaky registries matter most. Note this means scripted (non-tty) callers now also see messages like "Fetching config ..." on stderr. Generated-by: AI Signed-off-by: Colin Walters --- crates/composefs-ctl/src/lib.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/crates/composefs-ctl/src/lib.rs b/crates/composefs-ctl/src/lib.rs index 62dba2a0..e7182d13 100644 --- a/crates/composefs-ctl/src/lib.rs +++ b/crates/composefs-ctl/src/lib.rs @@ -87,7 +87,8 @@ use composefs::{ /// /// Renders per-component progress bars via [`MultiProgress`]. When a component /// completes or is skipped the bar is removed; human-readable messages are -/// printed above the bar group via [`MultiProgress::println`]. +/// printed above the bar group via [`MultiProgress::println`], or directly to +/// stderr when it is not a terminal. #[cfg(any(feature = "oci", feature = "http", feature = "ostree"))] struct IndicatifReporter { multi: MultiProgress, @@ -162,7 +163,14 @@ impl ProgressReporter for IndicatifReporter { } } ProgressEvent::Message(msg) => { - let _ = self.multi.println(msg); + // Progress bars are hidden when stderr is not a terminal (e.g. in + // CI logs), and `println` then discards the message; print it + // directly instead so that e.g. retry warnings stay visible. + if self.multi.is_hidden() { + eprintln!("{msg}"); + } else { + let _ = self.multi.println(msg); + } } // `ProgressEvent` is #[non_exhaustive]: new variants added to the library // will be silently ignored here until cfsctl is updated to handle them. From 3ab9bd034ebc4ac6379f13c32bc29a993df89e30 Mon Sep 17 00:00:00 2001 From: Colin Walters Date: Mon, 28 Sep 2026 23:18:48 -0400 Subject: [PATCH 3/5] build: Use containers-image-proxy from git for Error::is_retryable() Prep for retrying pulls, which needs the proxy's classification of retryable errors. This is only until a containers-image-proxy release includes it; then it becomes a plain version bump. Generated-by: AI Signed-off-by: Colin Walters --- Cargo.toml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/Cargo.toml b/Cargo.toml index 2726a77e..6fd24c99 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,3 +56,8 @@ debug = true platforms = ["*-unknown-linux-gnu"] tier = "2" all-features = true + +# Temporary, for Error::is_retryable(): replace with a version bump once +# https://github.com/bootc-dev/containers-image-proxy-rs is released with it. +[patch.crates-io] +containers-image-proxy = { git = "https://github.com/cgwalters-forge/containers-image-proxy-rs", rev = "a35d1d65e6724d2e8d7cf69e2c8770c28802b760" } From fc1dab84c666a4e943d8e8426a64a2669f2f5de3 Mon Sep 17 00:00:00 2001 From: Colin Walters Date: Tue, 22 Sep 2026 23:23:04 -0400 Subject: [PATCH 4/5] oci: Retry transient registry failures when pulling Pulls through skopeo have no retries at all, so a single 503 or dropped connection from e.g. quay.io fails the whole pull; this is a regular source of CI flakes for bootc. The proxy doesn't retry for us: in podman the retry loop lives in the caller (c/common pkg/retry), so do the same here. Which errors are worth retrying is the proxy's call: since skopeo 1.19 it classifies each failure with c/common's IsErrorRetryable(), the heuristic podman's retry loop uses, and containers-image-proxy exposes that as Error::is_retryable(). With an older skopeo nothing is retried. Only errors from the proxy are considered; a local failure to import verified data is never retried. Retries happen per proxy operation (opening the image, fetching the manifest, the config, and each layer or delta blob), so a failure in one layer doesn't refetch the others. The schedule is podman's default: 3 retries after 1s, 2s and 4s, or a fixed delay like its --retry-delay; PullOptions::retry tunes or disables it. Only registry (docker://) pulls retry. Note pull_image() now retries with the default policy too. A layer is still only registered once the proxy has verified the whole blob, and each attempt starts a fresh request. We now also wait for the proxy's verdict when the import itself failed, since a broken transfer is then usually the root cause and decides whether to retry; the import error is attached to it as context. Since each attempt restarts the transfer, a layer's Started event is sent again per attempt, and cfsctl replaces that layer's progress bar. Retries are reported as progress messages, which is how cfsctl shows them, and only logged at debug level so they don't appear twice. Generated-by: AI Closes: #348 Signed-off-by: Colin Walters --- crates/composefs-ctl/src/lib.rs | 5 +- crates/composefs-oci/Cargo.toml | 2 +- crates/composefs-oci/src/lib.rs | 11 +- crates/composefs-oci/src/retry.rs | 350 +++++++++++++++ crates/composefs-oci/src/skopeo.rs | 679 +++++++++++++++++++++++------ crates/composefs/src/progress.rs | 3 + 6 files changed, 918 insertions(+), 132 deletions(-) create mode 100644 crates/composefs-oci/src/retry.rs diff --git a/crates/composefs-ctl/src/lib.rs b/crates/composefs-ctl/src/lib.rs index e7182d13..3b0f7f7e 100644 --- a/crates/composefs-ctl/src/lib.rs +++ b/crates/composefs-ctl/src/lib.rs @@ -145,7 +145,10 @@ impl ProgressReporter for IndicatifReporter { .progress_chars("##-"), ); bar.set_message(id.to_string()); - self.bars.lock().unwrap().insert(id, bar); + // A retried component is started again; replace its bar. + if let Some(old) = self.bars.lock().unwrap().insert(id, bar) { + old.finish_and_clear(); + } } ProgressEvent::Progress { id, fetched, .. } => { if let Some(bar) = self.bars.lock().unwrap().get(&id) { diff --git a/crates/composefs-oci/Cargo.toml b/crates/composefs-oci/Cargo.toml index 8057de01..bb29edee 100644 --- a/crates/composefs-oci/Cargo.toml +++ b/crates/composefs-oci/Cargo.toml @@ -41,7 +41,7 @@ serde_json = { version = "1.0", default-features = false, features = ["std"] } rand = { version = "0.10.0", default-features = false, optional = true } tar = { version = "0.4.38", default-features = false, optional = true } tar-core = "0.1.0" -tokio = { version = "1.24.2", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1.24.2", features = ["macros", "rt-multi-thread", "time"] } tokio-util = { version = "0.7", default-features = false, features = ["io", "io-util"] } tracing = { version = "0.1", default-features = false } zstd = { version = "0.14.0", default-features = false } diff --git a/crates/composefs-oci/src/lib.rs b/crates/composefs-oci/src/lib.rs index df8a6fc4..567e8a01 100644 --- a/crates/composefs-oci/src/lib.rs +++ b/crates/composefs-oci/src/lib.rs @@ -28,6 +28,7 @@ pub mod oci_image; pub mod oci_layout; /// Re-exported from [`composefs::progress`]; use that path directly in new code. pub mod progress; +pub mod retry; pub mod skopeo; pub mod tar; /// Shared wire types and client proxy for the `org.composefs.Oci` interface. @@ -149,6 +150,7 @@ pub use oci_image::{ untag_image, }; pub use progress::{ComponentId, NullReporter, ProgressEvent, ProgressReporter, SharedReporter}; +pub use retry::RetryPolicy; pub use skopeo::pull_image; /// Statistics from an image import operation. @@ -344,6 +346,11 @@ pub struct PullOptions<'a> { /// Supported uniformly across transports, including `containers-storage:` /// imports (see [`LocalFetchOpt`]). pub bootable: bool, + + /// How to retry transient registry failures, as classified by the image + /// proxy. Only applies to registry (`docker://`) pulls; use + /// [`RetryPolicy::none()`] to disable retrying. + pub retry: RetryPolicy, } impl<'a> std::fmt::Debug for PullOptions<'a> { @@ -362,6 +369,7 @@ impl<'a> std::fmt::Debug for PullOptions<'a> { }, ) .field("bootable", &self.bootable) + .field("retry", &self.retry) .finish() } } @@ -512,13 +520,14 @@ pub async fn pull( }); } - let (result, stats) = skopeo::pull_image( + let (result, stats) = skopeo::pull_image_with_retry( repo, imgref, reference, opts.img_proxy_config, reporter, boot_options.as_ref(), + &opts.retry, ) .await?; Ok(crate::PullResult { diff --git a/crates/composefs-oci/src/retry.rs b/crates/composefs-oci/src/retry.rs new file mode 100644 index 00000000..8d0b428f --- /dev/null +++ b/crates/composefs-oci/src/retry.rs @@ -0,0 +1,350 @@ +//! Retrying transient failures when fetching from a registry. +//! +//! The containers-image-proxy (skopeo) does not retry failed requests +//! itself; in podman the retry loop lives in the caller (`c/common`), and +//! the same holds for us. Registries such as quay.io intermittently return +//! 5xx errors or drop connections, so without retries a single blip fails +//! an entire pull. +//! +//! Which errors are transient is decided by the proxy, with the same +//! `IsErrorRetryable()` heuristic podman uses; see +//! [`containers_image_proxy::Error::is_retryable()`]. Proxies older than +//! skopeo 1.19 don't report that, so with them nothing is retried. +//! +//! Retries are done at the granularity of one proxy operation (opening the +//! image, fetching the manifest, or fetching and importing one blob), so a +//! failed layer does not force refetching the others. Every blob attempt +//! starts from a fresh proxy request: data from a failed attempt is never +//! reused, and a layer is only registered once the proxy has verified the +//! size and digest of the complete blob. + +use std::future::Future; +use std::time::Duration; + +use anyhow::{Context, Result}; +use containers_image_proxy::Error as ProxyError; + +use crate::progress::{ProgressEvent, ProgressReporter}; + +/// Default number of retries, matching podman's default in `containers.conf`. +const DEFAULT_MAX_RETRIES: u32 = 3; +/// Delay before the first retry when [`RetryPolicy::delay`] is unset; it +/// doubles for each further retry, as in podman. +const DEFAULT_INITIAL_DELAY: Duration = Duration::from_secs(1); + +/// How to retry transient failures while fetching an image from a registry. +/// +/// This mirrors podman's `--retry` and `--retry-delay` options. Only +/// errors that the image proxy classifies as transient (network failures, +/// HTTP 502-504 responses and the like) are retried; others such as +/// authentication failures or a missing image fail immediately. +/// +/// Start from [`RetryPolicy::default()`] or [`RetryPolicy::none()`] and +/// adjust the fields as needed. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct RetryPolicy { + /// Maximum number of retries after the initial attempt; zero disables + /// retrying. + pub max_retries: u32, + /// Fixed delay between attempts. If unset, the delay starts at one + /// second and doubles for each further retry. + pub delay: Option, +} + +impl Default for RetryPolicy { + fn default() -> Self { + Self::with_max_retries(DEFAULT_MAX_RETRIES) + } +} + +impl RetryPolicy { + /// A policy that never retries. + pub const fn none() -> Self { + Self::with_max_retries(0) + } + + /// The default policy with a different number of retries. + pub const fn with_max_retries(max_retries: u32) -> Self { + Self { + max_retries, + delay: None, + } + } + + /// The delay before retry number `retry`, starting at zero. + fn backoff_delay(&self, retry: u32) -> Duration { + self.delay + .unwrap_or_else(|| DEFAULT_INITIAL_DELAY.saturating_mul(2u32.saturating_pow(retry))) + } +} + +/// Whether `err` looks like a transient failure that is worth retrying. +/// +/// Only errors reported by the image proxy are considered; local failures, +/// such as a malformed layer that the proxy verified as matching its digest, +/// are never retried. +pub(crate) fn is_transient(err: &anyhow::Error) -> bool { + err.chain() + .filter_map(|cause| cause.downcast_ref::()) + .any(ProxyError::is_retryable) +} + +/// Run `op` until it succeeds, fails with a non-transient error, or the +/// retries in `policy` are exhausted. +/// +/// Each retry is reported as a [`ProgressEvent::Message`], which is how +/// callers such as `cfsctl` show it; it is only logged at debug level, so +/// it does not show up twice. +/// `what` describes the operation in those messages. +/// +/// `op` must start from scratch every time it is called; nothing from a +/// failed attempt may leak into the next one. +pub(crate) async fn with_retry( + policy: &RetryPolicy, + what: &str, + reporter: &dyn ProgressReporter, + mut op: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + let mut retry = 0; + loop { + let err = match op().await { + Ok(v) => return Ok(v), + Err(err) => err, + }; + if !is_transient(&err) { + return Err(err); + } + if retry >= policy.max_retries { + return if retry > 0 { + let retries = if retry == 1 { "retry" } else { "retries" }; + Err(err).with_context(|| format!("Giving up after {retry} {retries}")) + } else { + Err(err) + }; + } + let delay = policy.backoff_delay(retry); + retry += 1; + let msg = format!( + "{what}: transient error, retrying in {:.1}s ({retry}/{}): {err:#}", + delay.as_secs_f64(), + policy.max_retries + ); + tracing::debug!("{msg}"); + reporter.report(ProgressEvent::Message(msg)); + tokio::time::sleep(delay).await; + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use containers_image_proxy::GetBlobError; + + use super::*; + use crate::progress::NullReporter; + + /// A progress reporter that records messages. + #[derive(Debug, Default)] + struct MessageLog(Mutex>); + + impl ProgressReporter for MessageLog { + fn report(&self, event: ProgressEvent) { + if let ProgressEvent::Message(m) = event { + self.0.lock().unwrap().push(m); + } + } + } + + const FAST_RETRIES: RetryPolicy = RetryPolicy { + max_retries: 3, + delay: Some(Duration::ZERO), + }; + const TRANSIENT: &str = "received unexpected HTTP status: 503 Service Unavailable"; + const PERMANENT: &str = "unauthorized: authentication required"; + + /// A failed proxy request, as classified by the proxy. + fn proxy_failure(msg: &str, retryable: bool) -> ProxyError { + let (method, error) = ("GetBlob".into(), msg.into()); + if retryable { + ProxyError::RetryableRequestFailure { method, error } + } else { + ProxyError::RequestInitiationFailure { method, error } + } + } + + #[test] + fn test_is_transient() { + let proxy = |e: ProxyError| anyhow::Error::from(e); + let cases = [ + (proxy(proxy_failure(TRANSIENT, true)), true), + (proxy(proxy_failure(PERMANENT, false)), false), + // Only the proxy's classification counts, not the message + (proxy(proxy_failure(TRANSIENT, false)), false), + ( + proxy(GetBlobError::Retryable("connection reset".into()).into()), + true, + ), + ( + proxy(GetBlobError::Other("connection reset".into()).into()), + false, + ), + // Transient cause wrapped in context + ( + proxy(proxy_failure(TRANSIENT, true)).context("Failed to import layer sha256:abcd"), + true, + ), + // Both the proxy and the import failed: the proxy decides + ( + proxy(proxy_failure("unexpected EOF", true)) + .context("Import failed: unexpected EOF in tar stream") + .context("Fetching layer sha256:abcd"), + true, + ), + ( + proxy(proxy_failure("manifest unknown", false)) + .context("Import failed: unexpected EOF in tar stream"), + false, + ), + // Local errors are never retried + (anyhow::anyhow!("unexpected EOF in tar stream"), false), + ]; + for (err, expected) in cases { + assert_eq!(is_transient(&err), expected, "{err:#}"); + } + } + + #[test] + fn test_backoff_delay() { + let default = RetryPolicy::default(); + let fixed = RetryPolicy { + delay: Some(Duration::from_millis(500)), + ..RetryPolicy::default() + }; + // (policy, retry, expected) + let cases = [ + (&default, 0, Duration::from_secs(1)), + (&default, 1, Duration::from_secs(2)), + (&default, 2, Duration::from_secs(4)), + // Must not overflow + (&default, u32::MAX, Duration::from_secs(u32::MAX.into())), + (&fixed, 0, Duration::from_millis(500)), + (&fixed, 5, Duration::from_millis(500)), + ]; + for (policy, retry, expected) in cases { + assert_eq!( + policy.backoff_delay(retry), + expected, + "{policy:?} retry={retry}" + ); + } + } + + /// Drive [`with_retry`] with a fetcher that fails with the given errors + /// in order and then succeeds. + #[tokio::test] + async fn test_with_retry() { + const ONE_RETRY: RetryPolicy = RetryPolicy { + max_retries: 1, + ..FAST_RETRIES + }; + const T: (&str, bool) = (TRANSIENT, true); + const P: (&str, bool) = (PERMANENT, false); + // (policy, errors before success, expected attempts, expected + // success, expected context of the final error) + type Case<'a> = ( + RetryPolicy, + &'a [(&'a str, bool)], + usize, + bool, + Option<&'a str>, + ); + let cases: &[Case] = &[ + (FAST_RETRIES, &[], 1, true, None), + (FAST_RETRIES, &[T], 2, true, None), + (FAST_RETRIES, &[T; 3], 4, true, None), + // Out of retries + ( + FAST_RETRIES, + &[T; 4], + 4, + false, + Some("Giving up after 3 retries"), + ), + ( + ONE_RETRY, + &[T; 2], + 2, + false, + Some("Giving up after 1 retry"), + ), + // Permanent errors are not retried, even after transient ones + (FAST_RETRIES, &[P], 1, false, None), + (FAST_RETRIES, &[T, P], 2, false, None), + // Retrying disabled + (RetryPolicy::none(), &[T], 1, false, None), + ]; + for (policy, errors, expected_attempts, expected_ok, expected_context) in cases { + let log = MessageLog::default(); + let mut attempts = 0; + let r = with_retry(policy, "Fetching thing", &log, || { + let result = match errors.get(attempts) { + Some(&(msg, retryable)) => Err(proxy_failure(msg, retryable).into()), + None => Ok(attempts), + }; + attempts += 1; + std::future::ready(result) + }) + .await; + let ctx = format!("policy={policy:?} errors={errors:?}"); + assert_eq!(attempts, *expected_attempts, "{ctx}"); + assert_eq!(r.is_ok(), *expected_ok, "{ctx}: {r:?}"); + let messages = log.0.into_inner().unwrap(); + assert_eq!(messages.len(), expected_attempts - 1, "{ctx}"); + for (i, m) in messages.iter().enumerate() { + let n = i + 1; + let max = policy.max_retries; + assert!( + m.starts_with("Fetching thing: transient error, retrying in ") + && m.contains(&format!("({n}/{max}): ")) + && m.ends_with(TRANSIENT), + "{ctx}: {m}" + ); + } + if let Err(e) = r { + let outer = e.to_string(); + match expected_context { + Some(c) => assert_eq!(outer, *c, "{ctx}"), + None => assert!(outer.starts_with("failed to invoke method"), "{ctx}: {e:#}"), + } + } + } + } + + #[tokio::test] + async fn test_with_retry_sleeps() { + let policy = RetryPolicy { + max_retries: 1, + delay: Some(Duration::from_millis(50)), + }; + let start = std::time::Instant::now(); + let mut failed = false; + with_retry(&policy, "x", &NullReporter, || { + let r = if failed { + Ok(()) + } else { + Err(proxy_failure("connection reset by peer", true).into()) + }; + failed = true; + std::future::ready(r) + }) + .await + .unwrap(); + assert!(start.elapsed() >= Duration::from_millis(50)); + } +} diff --git a/crates/composefs-oci/src/skopeo.rs b/crates/composefs-oci/src/skopeo.rs index d8fa2d05..b1855cd9 100644 --- a/crates/composefs-oci/src/skopeo.rs +++ b/crates/composefs-oci/src/skopeo.rs @@ -8,7 +8,7 @@ //! asynchronously with parallelism control, and stores them in the composefs repository with proper //! fs-verity integration. It supports various image formats and compression types. -use std::{cmp::Reverse, process::Command, thread::available_parallelism}; +use std::{cmp::Reverse, future::Future, process::Command, thread::available_parallelism}; use std::{iter::zip, sync::Arc}; @@ -22,7 +22,11 @@ use fn_error_context::context; use crate::oci_layout::OciLayoutKind; use rustix::process::geteuid; -use tokio::{io::AsyncReadExt, sync::Semaphore, task::JoinSet}; +use tokio::{ + io::{AsyncBufRead, AsyncReadExt}, + sync::Semaphore, + task::JoinSet, +}; use composefs::{ fsverity::FsVerityHashValue, @@ -35,6 +39,7 @@ use crate::{ layer_identifier, oci_image::{manifest_identifier, tag_image}, progress::{ComponentId, ProgressEvent, ProgressRead, ProgressUnit, SharedReporter}, + retry::{RetryPolicy, with_retry}, }; /// Result of pulling an OCI image. @@ -79,6 +84,7 @@ struct ImageOp { img: OpenedImage, reporter: SharedReporter, transport: Transport, + retry: RetryPolicy, } impl ImageOp { @@ -87,12 +93,20 @@ impl ImageOp { image_ref: &ImageReference, img_proxy_config: Option, reporter: SharedReporter, + retry: RetryPolicy, ) -> Result { // Fail fast if the repository is not writable, before starting // the image proxy or doing any network I/O. repo.ensure_writable()?; let transport = image_ref.transport; + // Only a registry fails transiently; for local transports such as + // containers-storage: a failure would just repeat. + let retry = if transport == Transport::Registry { + retry + } else { + RetryPolicy::none() + }; // See https://github.com/containers/skopeo/issues/2563 let skopeo_cmd = if transport == Transport::ContainerStorage && !geteuid().is_root() { @@ -141,16 +155,19 @@ impl ImageOp { let proxy = containers_image_proxy::ImageProxy::new_with_config(config) .await .context("Creating ImageProxy")?; - let img = proxy - .open_image_ref(image_ref) - .await - .context("Opening image")?; + // Opening the image fetches its manifest from the registry. + let img = with_retry(&retry, "Opening image", &*reporter, || async { + Ok(proxy.open_image_ref(image_ref).await?) + }) + .await + .context("Opening image")?; Ok(ImageOp { repo: Arc::clone(repo), proxy, img, reporter, transport, + retry, }) } @@ -192,95 +209,18 @@ impl ImageOp { _ => descriptor, }; - let (blob_reader, driver) = self - .proxy - .get_blob(&self.img, descriptor.digest(), descriptor.size()) - .await?; - - // See https://github.com/containers/containers-image-proxy-rs/issues/71 - let blob_reader = blob_reader.take(descriptor.size()); - - let id = ComponentId::from(diff_id.to_string()); - self.reporter.report(ProgressEvent::Started { - id: id.clone(), - total: Some(descriptor.size()), - unit: ProgressUnit::Bytes, - }); - - // Wrap the blob reader to emit Progress events as compressed bytes are read. - // This sits before decompression so `fetched` tracks bytes-over-the-wire, - // matching the `total` from the descriptor size above. - // - // The watch channel provides backpressure: if the renderer is slow, intermediate - // byte counts are coalesced rather than queued, keeping the I/O path non-blocking. - let (blob_reader, progress_driver) = ProgressRead::new( - blob_reader, - Arc::clone(&self.reporter), - id.clone(), - Some(descriptor.size()), - ); - - let media_type = descriptor.media_type(); - let (object_id, layer_stats) = if is_tar_media_type(media_type) { - // Tar layers: decompress and split into a splitstream. - // Run the progress driver concurrently with the import. - let reader = decompress_async(blob_reader, media_type)?; - let (result, ()) = - tokio::join!(import_tar_async(self.repo.clone(), reader), progress_driver); - result? - } else { - // Non-tar layers (OCI artifacts): stream raw bytes to object store. - // Run the progress driver concurrently with the blob store. - let (store_result, ()) = - tokio::join!(store_blob_async(&self.repo, blob_reader), progress_driver); - let (object_id, size, method) = store_result?; - driver.await?; - - let mut stats = ImportStats::default(); - match method { - ObjectStoreMethod::Copied => { - stats.objects_copied += 1; - stats.bytes_copied += size; - } - ObjectStoreMethod::Reflinked => { - stats.objects_reflinked += 1; - stats.bytes_reflinked += size; - } - ObjectStoreMethod::Hardlinked => { - stats.objects_hardlinked += 1; - stats.bytes_hardlinked += size; - } - ObjectStoreMethod::AlreadyPresent => { - stats.objects_already_present += 1; - } - } - - let mut stream = self.repo.create_stream(OCI_BLOB_CONTENT_TYPE)?; - stream.add_external_size(size); - stream.write_reference(object_id)?; - let stream_id = self.repo.write_stream(stream, &content_id, None)?; - self.reporter.report(ProgressEvent::Done { - id, - transferred: size, - }); - return Ok((stream_id, stats)); - }; - - // skopeo is doing data checksums for us to make sure the content we received is equal - // to the claimed diff_id. We trust it, but we need to check it by awaiting the driver. - driver.await?; - - // Sync and register the stream with its content identifier - self.repo - .register_stream(&object_id, &content_id, None) - .await?; - - self.reporter.report(ProgressEvent::Done { - id, - transferred: descriptor.size(), - }); - - Ok((object_id, layer_stats)) + fetch_layer( + &self.repo, + &self.reporter, + &self.retry, + diff_id, + descriptor, + || { + self.proxy + .get_blob(&self.img, descriptor.digest(), descriptor.size()) + }, + ) + .await } } @@ -344,11 +284,10 @@ impl ImageOp { )) } else { // We need to add the config to the repo - self.reporter.report(ProgressEvent::Message(format!( - "Fetching config {config_digest}" - ))); + let what = format!("Fetching config {config_digest}"); + self.reporter.report(ProgressEvent::Message(what.clone())); - let raw_config = async { + let raw_config = with_retry(&self.retry, &what, &*self.reporter, || async { let (mut config, driver) = self.proxy.get_descriptor(&self.img, descriptor).await?; let config = async move { let mut s = Vec::new(); @@ -358,7 +297,7 @@ impl ImageOp { let (config, driver) = tokio::join!(config, driver); let _: () = driver?; config - } + }) .await .with_context(|| format!("Failed to fetch config {config_digest}"))?; @@ -455,11 +394,14 @@ impl ImageOp { /// Pull the image, storing manifest, config, and all layers. pub async fn pull(self: &Arc) -> Result<(PullResult, ImportStats)> { - let (manifest_digest_str, raw_manifest) = self - .proxy - .fetch_manifest_raw_oci(&self.img) - .await - .context("Fetching manifest")?; + let (manifest_digest_str, raw_manifest) = with_retry( + &self.retry, + "Fetching manifest", + &*self.reporter, + || async { Ok(self.proxy.fetch_manifest_raw_oci(&self.img).await?) }, + ) + .await + .context("Fetching manifest")?; let manifest_digest: OciDigest = manifest_digest_str .try_into() .context("Invalid manifest digest from image proxy")?; @@ -547,33 +489,191 @@ impl oci_delta::DeltaBlobReader for ProxyBlobReader fn open_blob(&self, desc: &Descriptor) -> oci_delta::BlobStreamFuture<'_> { let desc = desc.clone(); Box::pin(async move { - let (reader, driver) = self - .image_op - .proxy - .get_blob(&self.image_op.img, desc.digest(), desc.size()) - .await?; - - let tmpfile = self - .image_op - .repo - .create_object_tmpfile() - .context("Creating temp file for delta blob")?; - let copy_fut = async { - let mut async_dst = tokio::fs::File::from(std::fs::File::from(tmpfile)); - tokio::io::copy(&mut reader.take(desc.size()), &mut async_dst).await?; - tokio::io::AsyncWriteExt::flush(&mut async_dst).await?; - let mut std_file = async_dst.into_std().await; - use std::io::Seek; - std_file.seek(std::io::SeekFrom::Start(0))?; - anyhow::Ok(Box::new(std_file) as Box) - }; - let (file_result, driver_result) = tokio::join!(copy_fut, driver); - let _: () = driver_result?; - file_result + let op = &self.image_op; + let desc = &desc; + let what = format!("Fetching blob {}", desc.digest()); + // Each attempt downloads into a fresh anonymous tmpfile, so a + // failed attempt leaves nothing behind. + with_retry(&op.retry, &what, &*op.reporter, || async move { + let (reader, driver) = op + .proxy + .get_blob(&op.img, desc.digest(), desc.size()) + .await?; + + let tmpfile = op + .repo + .create_object_tmpfile() + .context("Creating temp file for delta blob")?; + let copy_fut = async { + let mut async_dst = tokio::fs::File::from(std::fs::File::from(tmpfile)); + tokio::io::copy(&mut reader.take(desc.size()), &mut async_dst).await?; + tokio::io::AsyncWriteExt::flush(&mut async_dst).await?; + let mut std_file = async_dst.into_std().await; + use std::io::Seek; + std_file.seek(std::io::SeekFrom::Start(0))?; + anyhow::Ok(Box::new(std_file) as Box) + }; + let (file_result, driver_result) = tokio::join!(copy_fut, driver); + let _: () = driver_result?; + file_result + }) + .await }) } } +/// Fetch a layer blob and import it into the repository, retrying transient +/// failures according to `policy`. +/// +/// `fetch` starts a new download of the blob, returning the data stream and +/// the proxy's "driver" future which reports whether the proxy verified the +/// size and digest of what it sent. It is called once per attempt. +async fn fetch_layer( + repo: &Arc>, + reporter: &SharedReporter, + policy: &RetryPolicy, + diff_id: &OciDigest, + descriptor: &Descriptor, + mut fetch: F, +) -> Result<(ObjectID, ImportStats)> +where + ObjectID: FsVerityHashValue, + F: FnMut() -> Fut, + Fut: Future>, + R: AsyncBufRead + Send + Unpin, + D: Future>, +{ + let id = ComponentId::from(diff_id.to_string()); + let content_id = layer_identifier(diff_id); + let what = format!("Fetching layer {}", descriptor.digest()); + let (object_id, stats, transferred) = with_retry(policy, &what, &**reporter, || { + // Reported for every attempt: each one refetches the blob from the + // start, so its progress restarts from zero. + reporter.report(ProgressEvent::Started { + id: id.clone(), + total: Some(descriptor.size()), + unit: ProgressUnit::Bytes, + }); + let fetched = fetch(); + let (id, content_id) = (&id, content_id.as_str()); + async move { + let (reader, driver) = fetched.await?; + import_layer_blob(repo, reporter, id, content_id, descriptor, reader, driver).await + } + }) + .await?; + + reporter.report(ProgressEvent::Done { id, transferred }); + Ok((object_id, stats)) +} + +/// A single attempt at importing a layer blob streamed from the image proxy. +/// +/// The layer is only registered under `content_id` once the proxy has +/// confirmed (via `driver`) that the complete blob with the expected size +/// and digest was sent. If the transfer fails part way, anything already +/// written to the repository is either a content-addressed object (correct +/// by construction, and reused by the next attempt) or unreferenced (and +/// removed by garbage collection). +/// +/// Returns the stream ID, import statistics and the number of bytes +/// transferred. +async fn import_layer_blob( + repo: &Arc>, + reporter: &SharedReporter, + id: &ComponentId, + content_id: &str, + descriptor: &Descriptor, + reader: impl AsyncBufRead + Send + Unpin, + driver: impl Future>, +) -> Result<(ObjectID, ImportStats, u64)> { + enum Imported { + Tar(ObjectID, ImportStats), + Blob(ObjectID, u64, ObjectStoreMethod), + } + + // See https://github.com/containers/containers-image-proxy-rs/issues/71 + let reader = reader.take(descriptor.size()); + + // Wrap the blob reader to emit Progress events as compressed bytes are read. + // This sits before decompression so `fetched` tracks bytes-over-the-wire, + // matching the `total` from the descriptor size. + // + // The watch channel provides backpressure: if the renderer is slow, intermediate + // byte counts are coalesced rather than queued, keeping the I/O path non-blocking. + let (reader, progress_driver) = ProgressRead::new( + reader, + Arc::clone(reporter), + id.clone(), + Some(descriptor.size()), + ); + + let media_type = descriptor.media_type(); + let import = async { + if is_tar_media_type(media_type) { + // Tar layers: decompress and split into a splitstream. + let reader = decompress_async(reader, media_type)?; + let (object_id, stats) = import_tar_async(repo.clone(), reader).await?; + anyhow::Ok(Imported::Tar(object_id, stats)) + } else { + // Non-tar layers (OCI artifacts): stream raw bytes to object store. + let (object_id, size, method) = store_blob_async(repo, reader).await?; + Ok(Imported::Blob(object_id, size, method)) + } + }; + let (imported, ()) = tokio::join!(import, progress_driver); + + // The reader has been dropped, so the proxy is done writing. It checks + // the size and digest of the blob for us, and reports the outcome via the + // driver. Check that even if the import failed: a failed transfer is + // then usually the root cause, and it decides whether to retry. + let imported = match (imported, driver.await) { + (Ok(imported), Ok(())) => imported, + (Err(import_err), Ok(())) => return Err(import_err), + (Ok(_), Err(proxy_err)) => return Err(proxy_err.into()), + // Keep the proxy's error as the source, so that retrying looks at it. + (Err(import_err), Err(proxy_err)) => { + return Err( + anyhow::Error::from(proxy_err).context(format!("Import failed: {import_err:#}")) + ); + } + }; + + match imported { + Imported::Tar(object_id, stats) => { + // Sync and register the stream with its content identifier + repo.register_stream(&object_id, content_id, None).await?; + Ok((object_id, stats, descriptor.size())) + } + Imported::Blob(object_id, size, method) => { + let mut stats = ImportStats::default(); + match method { + ObjectStoreMethod::Copied => { + stats.objects_copied += 1; + stats.bytes_copied += size; + } + ObjectStoreMethod::Reflinked => { + stats.objects_reflinked += 1; + stats.bytes_reflinked += size; + } + ObjectStoreMethod::Hardlinked => { + stats.objects_hardlinked += 1; + stats.bytes_hardlinked += size; + } + ObjectStoreMethod::AlreadyPresent => { + stats.objects_already_present += 1; + } + } + + let mut stream = repo.create_stream(OCI_BLOB_CONTENT_TYPE)?; + stream.add_external_size(size); + stream.write_reference(object_id)?; + let stream_id = repo.write_stream(stream, content_id, None)?; + Ok((stream_id, stats, size)) + } + } +} + /// Pull the target image, storing manifest, config, and layers. /// /// Returns `PullResult` containing both manifest and config digests/verities. @@ -589,6 +689,9 @@ impl oci_delta::DeltaBlobReader for ProxyBlobReader /// generated and linked in the same pass over the OCI layers, avoiding the /// extra tar walk a separate `boot::generate_boot_image()` call would /// otherwise require. +/// +/// Transient registry failures are retried with the default [`RetryPolicy`]; +/// use [`crate::pull`] with [`crate::PullOptions::retry`] to change that. pub async fn pull_image( repo: &Arc>, imgref: &str, @@ -596,6 +699,28 @@ pub async fn pull_image( img_proxy_config: Option, reporter: SharedReporter, boot_options: Option<&composefs::generic_tree::OciTransformOptions>, +) -> Result<(PullResult, ImportStats)> { + pull_image_with_retry( + repo, + imgref, + reference, + img_proxy_config, + reporter, + boot_options, + &RetryPolicy::default(), + ) + .await +} + +/// Like [`pull_image`], but retrying transient failures according to `retry`. +pub(crate) async fn pull_image_with_retry( + repo: &Arc>, + imgref: &str, + reference: Option<&str>, + img_proxy_config: Option, + reporter: SharedReporter, + boot_options: Option<&composefs::generic_tree::OciTransformOptions>, + retry: &RetryPolicy, ) -> Result<(PullResult, ImportStats)> { // Fail fast if the repository is not writable, before doing any I/O. repo.ensure_writable()?; @@ -626,7 +751,9 @@ pub async fn pull_image( crate::oci_layout::import_oci_layout(repo, kind, layout_path, layout_tag, reporter).await? } else { // Standard path: use skopeo proxy for other transports - let op = Arc::new(ImageOp::new(repo, &image_ref, img_proxy_config, reporter).await?); + let op = Arc::new( + ImageOp::new(repo, &image_ref, img_proxy_config, reporter, retry.clone()).await?, + ); op.pull() .await .with_context(|| format!("Unable to pull container image {imgref}"))? @@ -670,3 +797,297 @@ pub async fn pull( let (config_digest, config_verity) = result.into_config(); Ok((config_digest, config_verity, stats)) } + +#[cfg(test)] +mod tests { + use std::io::Cursor; + use std::sync::Mutex; + use std::time::Duration; + + use composefs::fsverity::Sha256HashValue; + use composefs::test::TestRepo; + use containers_image_proxy::Error as ProxyError; + use containers_image_proxy::oci_spec::image::MediaType; + + use super::*; + use crate::progress::ProgressReporter; + + const FAST_RETRIES: RetryPolicy = RetryPolicy { + max_retries: 2, + delay: Some(Duration::ZERO), + }; + + /// Size of a tar header block, which precedes each file's content. + const TAR_BLOCK_SIZE: usize = 512; + + /// What one call of the injected fetcher does. + #[derive(Debug, Clone, Copy)] + enum Attempt { + /// The request itself fails with this message, which the proxy + /// classifies as retryable or not. + RequestFails(&'static str, bool), + /// The stream ends half way and the proxy reports the (retryable) + /// network error that interrupted it. + Interrupted, + /// Full-length data with a flipped byte; the proxy reports the + /// digest mismatch, which is not retryable, only after it has all + /// been read. + Corrupted, + /// The stream ends half way, but the proxy claims success, so the + /// failure is local (as for a malformed layer) and must not be retried. + TruncatedUnnoticed, + /// The complete, correct blob. + Good, + } + + const HTTP_503: &str = "reading blob: received unexpected HTTP status: 503 Service Unavailable"; + const UNAUTHORIZED: &str = "reading blob: unauthorized: authentication required"; + const CONNECTION_RESET: &str = + "read tcp 10.0.0.1:1234->10.0.0.2:443: read: connection reset by peer"; + const ARTIFACT_MEDIA_TYPE: &str = "application/vnd.example.artifact"; + + /// An uncompressed tar layer with a single file. + fn test_layer() -> Vec { + let mut builder = ::tar::Builder::new(Vec::new()); + let data = b"hello from a flaky registry\n".repeat(100); + let mut header = ::tar::Header::new_gnu(); + header.set_size(data.len() as u64); + header.set_mode(0o644); + header.set_mtime(1234567890); + header.set_cksum(); + builder + .append_data(&mut header, "usr/hello.txt", data.as_slice()) + .unwrap(); + builder.into_inner().unwrap() + } + + /// The proxy reports failures, including the verdict of `FinishPipe`, + /// like this. + fn proxy_failure(method: &str, msg: String, retryable: bool) -> ProxyError { + let (method, error) = (method.into(), msg.into()); + if retryable { + ProxyError::RetryableRequestFailure { method, error } + } else { + ProxyError::RequestInitiationFailure { method, error } + } + } + + type FakeDriver = std::future::Ready>; + type FakeFetch = containers_image_proxy::Result<(Cursor>, FakeDriver)>; + + fn run_attempt(attempt: Attempt, blob: &[u8], digest: &OciDigest) -> FakeFetch { + let size = blob.len(); + let half = blob[..size / 2].to_vec(); + let (data, verdict) = match attempt { + Attempt::RequestFails(msg, retryable) => { + return Err(proxy_failure("GetBlob", msg.into(), retryable)); + } + Attempt::Interrupted => (half, Err((CONNECTION_RESET.to_string(), true))), + Attempt::Corrupted => { + let mut data = blob.to_vec(); + // Inside the file content, so the tar still parses + data[TAR_BLOCK_SIZE + 10] ^= 0xff; + ( + data, + Err((format!("corrupted blob, expecting {digest}"), false)), + ) + } + Attempt::TruncatedUnnoticed => (half, Ok(())), + Attempt::Good => (blob.to_vec(), Ok(())), + }; + let driver = std::future::ready( + verdict.map_err(|(msg, retryable)| proxy_failure("FinishPipe", msg, retryable)), + ); + Ok((Cursor::new(data), driver)) + } + + /// Records `Started` and `Done` events. + #[derive(Debug, Default)] + struct EventLog(Mutex>); + + impl ProgressReporter for EventLog { + fn report(&self, event: ProgressEvent) { + let name = match event { + ProgressEvent::Started { .. } => "started", + ProgressEvent::Done { .. } => "done", + _ => return, + }; + self.0.lock().unwrap().push(name); + } + } + + /// Run [`fetch_layer`] with a fetcher following `script`, returning + /// the result, the number of fetches and the progress events. + async fn fetch_scripted( + repo: &Arc>, + policy: &RetryPolicy, + descriptor: &Descriptor, + blob: &[u8], + script: &[Attempt], + ) -> (Result, usize, Vec<&'static str>) { + let events = Arc::new(EventLog::default()); + let reporter: SharedReporter = events.clone(); + let digest = descriptor.digest(); + let mut calls = 0; + let result = fetch_layer(repo, &reporter, policy, digest, descriptor, || { + let attempt = script[calls]; + calls += 1; + std::future::ready(run_attempt(attempt, blob, digest)) + }) + .await + .map(|(id, _stats)| id); + let events = events.0.lock().unwrap().clone(); + (result, calls, events) + } + + /// Drive [`fetch_layer`] with an injected fetcher that fails according + /// to a script, and check that only verified data is ever registered. + #[tokio::test] + async fn test_fetch_layer_retries() { + use Attempt::*; + + let layer = test_layer(); + let artifact = b"some artifact blob contents\n".repeat(50); + + // (media type, blob, policy, script, expected success) + type Case<'a> = (MediaType, &'a [u8], RetryPolicy, &'a [Attempt], bool); + let cases: &[Case] = &[ + (MediaType::ImageLayer, &layer, FAST_RETRIES, &[Good], true), + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[RequestFails(HTTP_503, true), Good], + true, + ), + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[Interrupted, Good], + true, + ), + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[Interrupted, RequestFails(HTTP_503, true), Good], + true, + ), + // Out of retries + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[Interrupted; 3], + false, + ), + // Errors the proxy doesn't consider transient are not retried + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[RequestFails(UNAUTHORIZED, false)], + false, + ), + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[Corrupted], + false, + ), + // Nor are local failures to import verified data + ( + MediaType::ImageLayer, + &layer, + FAST_RETRIES, + &[TruncatedUnnoticed], + false, + ), + // Retrying disabled + ( + MediaType::ImageLayer, + &layer, + RetryPolicy::none(), + &[Interrupted], + false, + ), + // Non-tar blobs take a different import path + ( + MediaType::Other(ARTIFACT_MEDIA_TYPE.into()), + &artifact, + FAST_RETRIES, + &[Interrupted, RequestFails(HTTP_503, true), Good], + true, + ), + ( + MediaType::Other(ARTIFACT_MEDIA_TYPE.into()), + &artifact, + FAST_RETRIES, + &[Corrupted], + false, + ), + ]; + + for (media_type, blob, policy, script, expected_ok) in cases { + let ctx = format!("media_type={media_type} policy={policy:?} script={script:?}"); + let digest = crate::sha256_content_digest(blob); + let descriptor = Descriptor::new(media_type.clone(), blob.len() as u64, digest); + let content_id = layer_identifier(descriptor.digest()); + + // The stream ID that a single clean fetch produces + let expected_id = { + let reference = TestRepo::::new(); + let (r, ..) = + fetch_scripted(&reference.repo, policy, &descriptor, blob, &[Good]).await; + r.unwrap() + }; + + let test_repo = TestRepo::::new(); + let (result, calls, events) = + fetch_scripted(&test_repo.repo, policy, &descriptor, blob, script).await; + + assert_eq!(calls, script.len(), "{ctx}"); + let registered = test_repo.repo.has_stream(&content_id).unwrap(); + if *expected_ok { + let id = result.unwrap_or_else(|e| panic!("{ctx}: {e:#}")); + assert_eq!(id, expected_id, "{ctx}"); + assert_eq!(registered, Some(expected_id), "{ctx}"); + let mut expected_events = vec!["started"; script.len()]; + expected_events.push("done"); + assert_eq!(events, expected_events, "{ctx}"); + } else { + assert!(result.is_err(), "{ctx}"); + // Data from failed attempts must never be registered + assert_eq!(registered, None, "{ctx}"); + // One `Started` per attempt, and no `Done` + assert_eq!(events, vec!["started"; script.len()], "{ctx}"); + } + } + } + + /// For a tar layer, the stream ID matches a direct import of the layer. + #[tokio::test] + async fn test_fetch_layer_matches_import() { + let layer = test_layer(); + let diff_id = crate::sha256_content_digest(&layer); + let descriptor = + Descriptor::new(MediaType::ImageLayer, layer.len() as u64, diff_id.clone()); + let reference = TestRepo::::new(); + let (imported, _) = + crate::import_layer(&reference.repo, &diff_id, None, Cursor::new(layer.clone())) + .await + .unwrap(); + let test_repo = TestRepo::::new(); + let (fetched, ..) = fetch_scripted( + &test_repo.repo, + &FAST_RETRIES, + &descriptor, + &layer, + &[Attempt::Interrupted, Attempt::Good], + ) + .await; + assert_eq!(fetched.unwrap(), imported); + } +} diff --git a/crates/composefs/src/progress.rs b/crates/composefs/src/progress.rs index 22379e85..a45f2c6f 100644 --- a/crates/composefs/src/progress.rs +++ b/crates/composefs/src/progress.rs @@ -62,6 +62,9 @@ pub enum ProgressUnit { #[non_exhaustive] pub enum ProgressEvent { /// A new component (layer/object) has started being fetched. + /// + /// This is sent again for the same component when a retry restarts its + /// transfer from the beginning; progress for it then starts over. Started { /// Identifier for this component. id: ComponentId, From a594bfe1f9c2ffa0f1aba07a4ba4622fb1ef4241 Mon Sep 17 00:00:00 2001 From: Colin Walters Date: Tue, 22 Sep 2026 23:23:04 -0400 Subject: [PATCH 5/5] cfsctl: Add oci pull --retry Expose the retry count for registry pulls, so it can be raised for flaky environments or set to 0 to fail fast. The varlink API keeps the default policy for now. The integration test serves a local OCI layout from a minimal in-process registry that answers the first manifest request with a 503 and hangs up on the first request for each blob (skopeo retries neither itself), and checks that the pull recovers and yields the pinned image ID, while --retry 0 fails. This goes through the real skopeo proxy, so it also checks that skopeo classifies both failures as retryable. That classification needs skopeo 1.19, so the test is skipped with older ones (e.g. Ubuntu 24.04 has 1.13); the skopeo version check is now shared by all integration tests. Generated-by: AI Signed-off-by: Colin Walters --- crates/composefs-ctl/src/lib.rs | 7 + .../composefs-integration-tests/src/main.rs | 57 +++ .../src/tests/mod.rs | 1 + .../src/tests/old_format.rs | 13 +- .../src/tests/registry_retry.rs | 350 ++++++++++++++++++ .../src/tests/zstd_chunked.rs | 13 +- 6 files changed, 417 insertions(+), 24 deletions(-) create mode 100644 crates/composefs-integration-tests/src/tests/registry_retry.rs diff --git a/crates/composefs-ctl/src/lib.rs b/crates/composefs-ctl/src/lib.rs index 3b0f7f7e..bccceeca 100644 --- a/crates/composefs-ctl/src/lib.rs +++ b/crates/composefs-ctl/src/lib.rs @@ -417,6 +417,11 @@ enum OciCommand { /// import path with zero-copy reflink/hardlink support. #[arg(long, value_enum, default_value_t = LocalFetchCli::Disabled)] local_fetch: LocalFetchCli, + /// Number of times to retry transient registry failures (as + /// classified by skopeo, e.g. network errors and HTTP 502-504), with + /// exponential backoff as in podman; 0 disables retrying. + #[arg(long, value_name = "N", default_value_t = composefs_oci::RetryPolicy::default().max_retries)] + retry: u32, }, /// Copy an OCI image (and its layers) from another composefs repository /// into this repository. @@ -1837,6 +1842,7 @@ where bootable, expected_digest, local_fetch, + retry, } => { // Parse before pulling so a malformed digest fails fast, // rather than after a potentially long-running fetch. @@ -1862,6 +1868,7 @@ where local_fetch: local_fetch.into(), progress: Some(reporter), bootable: use_bootable_opt, + retry: composefs_oci::RetryPolicy::with_max_retries(retry), ..Default::default() }; diff --git a/crates/composefs-integration-tests/src/main.rs b/crates/composefs-integration-tests/src/main.rs index ad74a036..2ff8c56f 100644 --- a/crates/composefs-integration-tests/src/main.rs +++ b/crates/composefs-integration-tests/src/main.rs @@ -57,6 +57,63 @@ pub(crate) fn cfsctl() -> Result { ) } +/// A skopeo version as `(major, minor, patch)`. +pub(crate) type SkopeoVersion = (u32, u32, u32); + +/// Parse the output of `skopeo --version`, e.g. +/// `skopeo version 1.22.2 commit: 02c8e50e...`. +fn parse_skopeo_version(output: &str) -> Option { + let version = output + .strip_prefix("skopeo version ")? + .split_whitespace() + .next()?; + // Drop suffixes such as "-dev" + let mut parts = version + .split(['.', '-', '+']) + .map(|p| p.parse::().ok()); + Some(( + parts.next()??, + parts.next()??, + parts.next().flatten().unwrap_or(0), + )) +} + +/// The version of the installed skopeo, or `None` if there is none (or its +/// version can't be parsed). +pub(crate) fn skopeo_version() -> Option { + let output = std::process::Command::new("skopeo") + .arg("--version") + .stderr(std::process::Stdio::null()) + .output() + .ok() + .filter(|o| o.status.success())?; + parse_skopeo_version(&String::from_utf8_lossy(&output.stdout)) +} + +/// Returns true if skopeo is available on the system. +pub(crate) fn have_skopeo() -> bool { + skopeo_version().is_some() +} + +fn test_parse_skopeo_version() -> Result<()> { + let cases = [ + ( + "skopeo version 1.22.2 commit: 02c8e50e431f9617", + Some((1, 22, 2)), + ), + ("skopeo version 1.13.3\n", Some((1, 13, 3))), + ("skopeo version 1.19.0-dev", Some((1, 19, 0))), + ("skopeo version 1.20", Some((1, 20, 0))), + ("skopeo version banana", None), + ("", None), + ]; + for (output, expected) in cases { + assert_eq!(parse_skopeo_version(output), expected, "{output:?}"); + } + Ok(()) +} +integration_test!(test_parse_skopeo_version); + /// Bind a listening Unix socket at a fresh tempdir path and spawn `cfsctl` /// against it via the systemd socket-activation protocol (`LISTEN_FDS=1`, the /// listening socket on fd 3, `LISTEN_PID` set in the child). The socket is diff --git a/crates/composefs-integration-tests/src/tests/mod.rs b/crates/composefs-integration-tests/src/tests/mod.rs index 7bef8d8a..22db564d 100644 --- a/crates/composefs-integration-tests/src/tests/mod.rs +++ b/crates/composefs-integration-tests/src/tests/mod.rs @@ -9,6 +9,7 @@ pub mod oci_compat; pub mod old_format; pub mod ostree; pub mod privileged; +pub mod registry_retry; pub mod upgrade; pub mod varlink; pub mod zstd_chunked; diff --git a/crates/composefs-integration-tests/src/tests/old_format.rs b/crates/composefs-integration-tests/src/tests/old_format.rs index ea4ab993..343dbd49 100644 --- a/crates/composefs-integration-tests/src/tests/old_format.rs +++ b/crates/composefs-integration-tests/src/tests/old_format.rs @@ -17,22 +17,11 @@ use anyhow::{Context, Result}; use std::path::PathBuf; use xshell::{Shell, cmd}; -use crate::{cfsctl, integration_test}; +use crate::{cfsctl, have_skopeo, integration_test}; /// Environment variable naming the old cfsctl binary. const CFSCTL_PATH_OLD: &str = "CFSCTL_PATH_OLD"; -/// Returns true if skopeo is available on the system. -fn have_skopeo() -> bool { - std::process::Command::new("skopeo") - .arg("--version") - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()) - .status() - .map(|s| s.success()) - .unwrap_or(false) -} - fn test_read_old_format_repo() -> Result<()> { let old_cfsctl = PathBuf::from( std::env::var_os(CFSCTL_PATH_OLD) diff --git a/crates/composefs-integration-tests/src/tests/registry_retry.rs b/crates/composefs-integration-tests/src/tests/registry_retry.rs new file mode 100644 index 00000000..70a051b7 --- /dev/null +++ b/crates/composefs-integration-tests/src/tests/registry_retry.rs @@ -0,0 +1,350 @@ +//! Integration test for retrying transient registry failures. +//! +//! Serves the deterministic OCI layout from [`create_oci_layout`] from a +//! minimal in-process registry that answers the first manifest request with +//! `503 Service Unavailable` and hangs up on the first request for each +//! blob, and pulls it with `cfsctl oci pull docker://` through skopeo. +//! skopeo (i.e. containers/image) does not retry either itself, so any +//! recovery is ours. Whether an error is worth retrying is decided by the +//! proxy, so this also checks that skopeo classifies both as retryable, +//! which needs skopeo 1.19 or newer. + +use std::collections::HashSet; +use std::io::{BufRead, BufReader, Write}; +use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use xshell::{Shell, cmd}; + +use crate::tests::cli::{OCI_LAYOUT_COMPOSEFS_ID, create_oci_layout, init_insecure_repo}; +use crate::{SkopeoVersion, cfsctl, integration_test, skopeo_version}; + +/// The first skopeo whose image proxy reports which errors are retryable +/// (protocol 0.2.8); with older ones nothing is retried. +const MIN_SKOPEO_VERSION: SkopeoVersion = (1, 19, 0); +/// Repository name the image is served under. +const IMAGE_NAME: &str = "composefs/retry-test"; +/// Requests that fail once: the manifest, plus the blobs of the image from +/// [`create_oci_layout`] (config and one layer). +const FAILURE_COUNT: usize = 3; +/// Key in [`FailedRequests`] for the manifest, which is fetched by tag or +/// digest. +const MANIFEST_KEY: &str = "manifest"; +/// How long to wait for a request before giving up on a connection, so a +/// client that never sends one can't hang a handler thread. +const READ_TIMEOUT: Duration = Duration::from_secs(5); +/// First byte of a TLS handshake record, as in a ClientHello. +const TLS_HANDSHAKE: u8 = 0x16; + +/// The manifest ([`MANIFEST_KEY`]) and blob digests that have already +/// failed once. +type FailedRequests = Mutex>; + +/// A read-only registry serving a single-image OCI layout over plain HTTP, +/// failing the first request for the manifest and for each blob. Every +/// response closes the connection, so each request is seen separately. +struct FlakyRegistry { + addr: SocketAddr, + failed: Arc, + shutdown: Arc, +} + +impl FlakyRegistry { + fn start(layout: &Path) -> Result { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?; + let addr = listener.local_addr()?; + let failed = Arc::new(FailedRequests::default()); + let shutdown = Arc::new(AtomicBool::new(false)); + let layout = layout.to_owned(); + let (failures, stop) = (Arc::clone(&failed), Arc::clone(&shutdown)); + std::thread::spawn(move || { + for conn in listener.incoming() { + if stop.load(Ordering::SeqCst) { + break; + } + let conn = match conn { + Ok(conn) => conn, + Err(e) => { + eprintln!("flaky registry: accept: {e}"); + continue; + } + }; + let (layout, failures) = (layout.clone(), Arc::clone(&failures)); + std::thread::spawn(move || { + if let Err(e) = handle_request(conn, &layout, &failures) { + eprintln!("flaky registry: {e:#}"); + } + }); + } + }); + Ok(Self { + addr, + failed, + shutdown, + }) + } + + fn failure_count(&self) -> usize { + self.failed.lock().unwrap().len() + } + + fn image_ref(&self) -> String { + format!("docker://{}/{IMAGE_NAME}:latest", self.addr) + } + + /// Create a home directory whose `registries.conf` allows plain HTTP to + /// this registry. skopeo only reads the per-user file from `$HOME` + /// (it ignores `CONTAINERS_REGISTRIES_CONF`), and cfsctl has no flag + /// for its `--tls-verify`. + fn create_home(&self, parent: &Path) -> Result { + let home = parent.join("home"); + let config_dir = home.join(".config/containers"); + std::fs::create_dir_all(&config_dir)?; + let conf = format!( + "[[registry]]\nlocation = \"{}\"\ninsecure = true\n", + self.addr + ); + std::fs::write(config_dir.join("registries.conf"), conf)?; + Ok(home) + } +} + +impl Drop for FlakyRegistry { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::SeqCst); + // Wake up the accept loop so it sees the flag. + let _ = TcpStream::connect(self.addr); + } +} + +fn respond( + conn: &mut TcpStream, + status: &str, + headers: &[(&str, &str)], + body: &[u8], + send_body: bool, +) -> Result<()> { + let mut head = format!( + "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n", + body.len() + ); + for (k, v) in headers { + head.push_str(&format!("{k}: {v}\r\n")); + } + head.push_str("\r\n"); + conn.write_all(head.as_bytes())?; + if send_body { + conn.write_all(body)?; + } + Ok(()) +} + +fn read_blob(layout: &Path, digest: &str) -> Result> { + let (alg, hex) = digest + .split_once(':') + .with_context(|| format!("invalid digest {digest}"))?; + Ok(std::fs::read(layout.join("blobs").join(alg).join(hex))?) +} + +/// How [`fail_once`] fails a request. +#[derive(Debug, Clone, Copy)] +enum Fault { + /// Respond with `503 Service Unavailable`. + Unavailable, + /// Close the connection without responding. + HangUp, +} + +/// Fail the first request for `key` as `fault` says; returns whether it did. +fn fail_once( + conn: &mut TcpStream, + failed: &FailedRequests, + key: &str, + fault: Fault, + send_body: bool, +) -> Result { + let first_request = failed.lock().unwrap().insert(key.to_owned()); + if first_request { + match fault { + Fault::Unavailable => respond(conn, "503 Service Unavailable", &[], b"", send_body)?, + // The caller drops the connection. + Fault::HangUp => {} + } + } + Ok(first_request) +} + +fn handle_request(mut conn: TcpStream, layout: &Path, failed: &FailedRequests) -> Result<()> { + conn.set_read_timeout(Some(READ_TIMEOUT))?; + // skopeo tries TLS before falling back to plain HTTP; just hang up on it, + // rather than waiting for a request line that never comes. + let mut first = [0u8]; + if conn.peek(&mut first)? == 0 || first[0] == TLS_HANDSHAKE { + return Ok(()); + } + let mut reader = BufReader::new(conn.try_clone()?); + let mut request_line = String::new(); + reader + .read_line(&mut request_line) + .context("Reading request line")?; + // Skip the headers; nothing in them matters here. + loop { + let mut line = Vec::new(); + if reader.read_until(b'\n', &mut line)? == 0 || line == b"\r\n" { + break; + } + } + let mut parts = request_line.split_whitespace(); + let (Some(method), Some(path)) = (parts.next(), parts.next()) else { + return Ok(()); + }; + let send_body = method != "HEAD"; + let not_found = |conn: &mut TcpStream| respond(conn, "404 Not Found", &[], b"", send_body); + + if path == "/v2/" { + return respond( + &mut conn, + "200 OK", + &[("Content-Type", "application/json")], + b"{}", + send_body, + ); + } + let Some(rest) = path.strip_prefix(&format!("/v2/{IMAGE_NAME}/")) else { + return not_found(&mut conn); + }; + if let Some(reference) = rest.strip_prefix("manifests/") { + if fail_once( + &mut conn, + failed, + MANIFEST_KEY, + Fault::Unavailable, + send_body, + )? { + return Ok(()); + } + let index: serde_json::Value = + serde_json::from_slice(&std::fs::read(layout.join("index.json"))?)?; + let desc = &index["manifests"][0]; + let (Some(digest), Some(media_type)) = + (desc["digest"].as_str(), desc["mediaType"].as_str()) + else { + anyhow::bail!("index.json has no manifest descriptor"); + }; + if reference.contains(':') && reference != digest { + return not_found(&mut conn); + } + let manifest = read_blob(layout, digest)?; + respond( + &mut conn, + "200 OK", + &[ + ("Content-Type", media_type), + ("Docker-Content-Digest", digest), + ], + &manifest, + send_body, + ) + } else if let Some(digest) = rest.strip_prefix("blobs/") { + if fail_once(&mut conn, failed, digest, Fault::HangUp, send_body)? { + return Ok(()); + } + match read_blob(layout, digest) { + Ok(blob) => respond( + &mut conn, + "200 OK", + &[("Content-Type", "application/octet-stream")], + &blob, + send_body, + ), + Err(e) => { + eprintln!("flaky registry: {e:#}"); + not_found(&mut conn) + } + } + } else { + not_found(&mut conn) + } +} + +/// A pull through skopeo survives a transient 503 on the manifest and +/// dropped connections on the config and layer fetches, and imports the +/// same image as a direct pull of the layout. With retries disabled the +/// same faults fail the pull, which shows they were injected. +fn test_pull_retries_transient_registry_errors() -> Result<()> { + let Some(version) = skopeo_version() else { + eprintln!("skopeo not found, skipping registry retry test"); + return Ok(()); + }; + if version < MIN_SKOPEO_VERSION { + eprintln!( + "skopeo {version:?} does not classify retryable errors (needs {MIN_SKOPEO_VERSION:?}), \ + skipping registry retry test" + ); + return Ok(()); + } + + let sh = Shell::new()?; + let cfsctl = cfsctl()?; + let fixture_dir = tempfile::tempdir()?; + let layout = create_oci_layout(fixture_dir.path())?; + + // (extra cfsctl args, expect success) + let cases: &[(&[&str], bool)] = &[(&[], true), (&["--retry", "0"], false)]; + for &(args, expect_ok) in cases { + let registry = FlakyRegistry::start(&layout)?; + let home = registry.create_home(fixture_dir.path())?; + let image = registry.image_ref(); + let repo_dir = init_insecure_repo(&sh, &cfsctl)?; + let repo = repo_dir.path(); + + let output = cmd!( + sh, + "{cfsctl} --insecure --repo {repo} oci pull {args...} {image} retry-image" + ) + .env("HOME", &home) + .env("XDG_CONFIG_HOME", home.join(".config")) + .ignore_status() + .output()?; + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + let ctx = format!("args={args:?}\nstdout:\n{stdout}\nstderr:\n{stderr}"); + + let failures = registry.failure_count(); + if !expect_ok { + assert_eq!(failures, 1, "should give up at the first failure; {ctx}"); + assert!(!output.status.success(), "pull should fail; {ctx}"); + assert!( + stderr.contains("503 Service Unavailable"), + "error should report the 503; {ctx}" + ); + continue; + } + + assert!(output.status.success(), "pull should succeed; {ctx}"); + assert_eq!(failures, FAILURE_COUNT, "{ctx}"); + let combined = format!("{stdout}{stderr}"); + let retries = combined.matches("transient error, retrying").count(); + assert_eq!( + retries, FAILURE_COUNT, + "each injected failure should be retried once; {ctx}" + ); + let config_digest = stdout + .lines() + .find_map(|l| l.strip_prefix("config").map(|s| s.trim().to_string())) + .with_context(|| format!("config digest in pull output; {ctx}"))?; + let at_config_digest = format!("@{config_digest}"); + let image_id = cmd!( + sh, + "{cfsctl} --insecure --repo {repo} oci compute-id {at_config_digest}" + ) + .read()?; + assert_eq!(image_id.trim(), OCI_LAYOUT_COMPOSEFS_ID); + } + Ok(()) +} +integration_test!(test_pull_retries_transient_registry_errors); diff --git a/crates/composefs-integration-tests/src/tests/zstd_chunked.rs b/crates/composefs-integration-tests/src/tests/zstd_chunked.rs index 536a43ce..85c8c126 100644 --- a/crates/composefs-integration-tests/src/tests/zstd_chunked.rs +++ b/crates/composefs-integration-tests/src/tests/zstd_chunked.rs @@ -8,18 +8,7 @@ use anyhow::Result; use xshell::{Shell, cmd}; use crate::tests::cli::{OCI_LAYOUT_COMPOSEFS_ID, create_oci_layout, init_insecure_repo}; -use crate::{cfsctl, integration_test}; - -/// Returns true if skopeo is available on the system. -fn have_skopeo() -> bool { - std::process::Command::new("skopeo") - .arg("--version") - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()) - .status() - .map(|s| s.success()) - .unwrap_or(false) -} +use crate::{cfsctl, have_skopeo, integration_test}; /// Regression test for multi-frame zstd decoding: before the decode fix, /// every `zstd:chunked` pull silently imported a truncated tar because the