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" } 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]); diff --git a/crates/composefs-ctl/src/lib.rs b/crates/composefs-ctl/src/lib.rs index 62dba2a0..bccceeca 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, @@ -144,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) { @@ -162,7 +166,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. @@ -406,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. @@ -1826,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. @@ -1851,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 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,