oci: Retry transient registry failures when pulling

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" }

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]);

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()
                 };
 

crates/composefs-integration-tests/src/main.rs

@@ -57,6 +57,63 @@ pub(crate) fn cfsctl() -> Result<PathBuf> {
     )
 }
 
+/// 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<SkopeoVersion> {
+    let version = output
+        .strip_prefix("skopeo version ")?
+        .split_whitespace()
+        .next()?;
+    // Drop suffixes such as "-dev"
+    let mut parts = version
+        .split(['.', '-', '+'])
+        .map(|p| p.parse::<u32>().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<SkopeoVersion> {
+    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

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;

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)