bootc_lib/

deploy.rs

//! Pull dispatch and deployment staging for the ostree backend.
//!
//! ## Planned Pull paths
//!
//! The top-level entry point for upgrade/switch will eventually select
//! among three paths based on the `unified` flag and filesystem capability:
//!
//! - **Unified + reflinks** (`unified = true`, XFS/btrfs): `pull_via_composefs_unified`
//!   — the planned three-store pipeline. Pulls into containers-storage first, then
//!   ZeroCopy into composefs, then synthesizes the ostree commit via FICLONE.
//!   See [`crate::store`] for the architecture diagram.
//!
//! - **Non-unified + reflinks** (`unified = false`, XFS/btrfs): `pull_via_composefs`
//!   — fetches from registry directly into composefs (no containers-storage),
//!   then synthesizes the ostree commit via FICLONE.
//!
//! - **No reflinks** (ext4): `pull` — the legacy ostree-native tar importer
//!   (`ostree_container::store::ImageImporter`).
//!
//! ## Planned composefs → ostree synthesis
//!
//! The synthesis plan relies on `import_from_composefs_repo` from
//! `ostree_ext::container::composefs_import` to walk the composefs
//! filesystem tree and for each regular file:
//!
//! 1. Reads uid/gid/mode/xattrs from composefs metadata. SELinux labels are
//!    computed in bulk before the walk via `selabel()` and looked up per-file;
//!    a NUL terminator is appended because composefs-rs omits it but the kernel
//!    stores it.
//! 2. Computes the ostree content checksum in-memory (SHA-256 of
//!    `uid:gid:mode:xattrs:file-content`).
//! 3. Issues `ioctl(FICLONE)` from the composefs object fd into a new `O_TMPFILE`
//!    in the ostree object directory.
//! 4. Applies metadata (`fchown`, `fchmod`, `fsetxattr`) and links the tmpfile
//!    into the ostree content-addressed path.
//!
//! `/etc` is remapped to `usr/etc`; virtual toplevel paths (`proc`, `sys`,
//! `dev`, etc.) are excluded — matching the ostree-container tar importer.
//!
//! ## Auto-detection
//!
//! `image_exists_in_unified_storage` checks whether the target image is already
//! present in bootc-owned containers-storage. Call sites use this to select
//! `unified = true` automatically without requiring an explicit flag from the
//! user once `bootc image set-unified` has been run.

use std::collections::HashSet;
use std::io::{BufRead, Write};
use std::os::fd::{AsFd, AsRawFd};
use std::process::Command;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result, anyhow};
use bootc_utils::skopeo_bin;
use cap_std::fs::{Dir, MetadataExt};
use cap_std_ext::cap_std;
use cap_std_ext::dirext::CapStdExtDirExt;
use fn_error_context::context;
use linux_kernel_cmdline::utf8::CmdlineOwned;
use ostree::{gio, glib};
use ostree_container::OstreeImageReference;
use ostree_ext::container as ostree_container;
use ostree_ext::container::store::{ImageImporter, ImportProgress, PrepareResult, PreparedImport};
use ostree_ext::oci_spec::image::{Descriptor, Digest};
use ostree_ext::ostree::Deployment;
use ostree_ext::ostree::{self, Sysroot};
use ostree_ext::sysroot::SysrootLock;
use ostree_ext::tokio_util::spawn_blocking_cancellable_flatten;

use crate::progress_jsonl::{Event, ProgressWriter, SubTaskBytes, SubTaskStep};
use crate::spec::ImageReference;
use crate::spec::{BootOrder, HostSpec};
use crate::status::labels_of_config;
use crate::store::Storage;
use crate::utils::async_task_with_spinner;

// TODO use https://github.com/ostreedev/ostree-rs-ext/pull/493/commits/afc1837ff383681b947de30c0cefc70080a4f87a
const BASE_IMAGE_PREFIX: &str = "ostree/container/baseimage/bootc";

// Match podman's default registry retry policy. A failed attempt has to rebuild
// the importer, so retries are made at the whole-pull boundary instead of for
// individual layers.
// TODO: Read this policy from the proxy once
// https://github.com/podman-container-tools/container-libs/pull/951 is available.
const PULL_MAX_RETRIES: u32 = 3;
const PULL_RETRY_INITIAL_DELAY: Duration = Duration::from_secs(2);

/// Progress output shared by all attempts to pull an image.
#[derive(Clone)]
pub(crate) struct PullProgress {
    bars: indicatif::MultiProgress,
    json: ProgressWriter,
    quiet: bool,
}

impl PullProgress {
    pub(crate) fn new(quiet: bool, json: ProgressWriter) -> Self {
        let bars = indicatif::MultiProgress::new();
        if quiet {
            bars.set_draw_target(indicatif::ProgressDrawTarget::hidden());
        }
        Self { bars, json, quiet }
    }

    fn println(&self, message: impl AsRef<str>) {
        if self.quiet {
            return;
        }
        if let Err(error) = self.bars.println(message) {
            tracing::debug!(%error, "Writing pull progress message");
        }
    }

    async fn send(&self, event: Event<'_>) {
        self.json.send(event.clone()).await;
        if let Event::ProgressSteps { description, .. } = &event {
            self.println(description);
        }
    }

    fn clear(&self) {
        if let Err(error) = self.bars.clear() {
            tracing::debug!(%error, "Clearing pull progress");
        }
    }
}

/// Create an ImageProxyConfig with bootc's user agent prefix set.
///
/// This allows registries to distinguish "image pulls for bootc client runs"
/// from other skopeo/containers-image users.
pub(crate) fn new_proxy_config() -> ostree_ext::containers_image_proxy::ImageProxyConfig {
    let mut c = ostree_ext::containers_image_proxy::ImageProxyConfig::default();
    c.user_agent_prefix = Some(format!("bootc/{}", env!("CARGO_PKG_VERSION")));
    c
}

/// Set on an ostree commit if this is a derived commit
const BOOTC_DERIVED_KEY: &str = "bootc.derived";

/// Variant of HostSpec but required to be filled out
pub(crate) struct RequiredHostSpec<'a> {
    pub(crate) image: &'a ImageReference,
}

/// State of a locally fetched image
pub(crate) struct ImageState {
    pub(crate) manifest_digest: Digest,
    pub(crate) version: Option<String>,
    pub(crate) ostree_commit: String,
}

impl<'a> RequiredHostSpec<'a> {
    /// Given a (borrowed) host specification, "unwrap" its internal
    /// options, giving a spec that is required to have a base container image.
    pub(crate) fn from_spec(spec: &'a HostSpec) -> Result<Self> {
        let image = spec
            .image
            .as_ref()
            .ok_or_else(|| anyhow::anyhow!("Missing image in specification"))?;
        Ok(Self { image })
    }
}

impl From<ostree_container::store::LayeredImageState> for ImageState {
    fn from(value: ostree_container::store::LayeredImageState) -> Self {
        let version = value.version().map(|v| v.to_owned());
        let ostree_commit = value.get_commit().to_owned();
        Self {
            manifest_digest: value.manifest_digest,
            version,
            ostree_commit,
        }
    }
}

impl ImageState {
    /// Fetch the manifest corresponding to this image.  May not be available in all backends.
    pub(crate) fn get_manifest(
        &self,
        repo: &ostree::Repo,
    ) -> Result<Option<ostree_ext::oci_spec::image::ImageManifest>> {
        ostree_container::store::query_image_commit(repo, &self.ostree_commit)
            .map(|v| Some(v.manifest))
    }
}

/// The importer settings every bootc pull uses, however the layers arrive.
fn configure_importer(
    imp: &mut ostree_container::store::ImageImporter,
    booted_deployment: Option<&ostree::Deployment>,
) {
    imp.require_bootable();
    // We do our own GC/prune in deploy::prune(), so skip the importer's internal one.
    imp.disable_gc();
    if let Some(deployment) = booted_deployment {
        imp.set_sepolicy_commit(deployment.csum().to_string());
    }
}

/// Wrapper for pulling a container image, wiring up status output.
pub(crate) async fn new_importer(
    repo: &ostree::Repo,
    imgref: &ostree_container::OstreeImageReference,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<ostree_container::store::ImageImporter> {
    new_importer_with_config(repo, imgref, new_proxy_config(), booted_deployment).await
}

/// Wrapper for pulling a container image with a custom proxy config (e.g. for unified storage).
pub(crate) async fn new_importer_with_config(
    repo: &ostree::Repo,
    imgref: &ostree_container::OstreeImageReference,
    config: ostree_ext::containers_image_proxy::ImageProxyConfig,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<ostree_container::store::ImageImporter> {
    let mut imp = ostree_container::store::ImageImporter::new(repo, imgref, config).await?;
    configure_importer(&mut imp, booted_deployment);
    Ok(imp)
}

pub(crate) fn check_bootc_label(config: &ostree_ext::oci_spec::image::ImageConfiguration) {
    if let Some(label) =
        labels_of_config(config).and_then(|labels| labels.get(crate::metadata::BOOTC_COMPAT_LABEL))
    {
        match label.as_str() {
            crate::metadata::COMPAT_LABEL_V1 => {}
            o => crate::journal::journal_print(
                libsystemd::logging::Priority::Warning,
                &format!(
                    "notice: Unknown {} value {}",
                    crate::metadata::BOOTC_COMPAT_LABEL,
                    o
                ),
            ),
        }
    } else {
        crate::journal::journal_print(
            libsystemd::logging::Priority::Warning,
            &format!(
                "notice: Image is missing label: {}",
                crate::metadata::BOOTC_COMPAT_LABEL
            ),
        )
    }
}

fn descriptor_of_progress(p: &ImportProgress) -> &Descriptor {
    match p {
        ImportProgress::OstreeChunkStarted(l) => l,
        ImportProgress::OstreeChunkCompleted(l) => l,
        ImportProgress::DerivedLayerStarted(l) => l,
        ImportProgress::DerivedLayerCompleted(l) => l,
    }
}

fn prefix_of_progress(p: &ImportProgress) -> &'static str {
    match p {
        ImportProgress::OstreeChunkStarted(_) | ImportProgress::OstreeChunkCompleted(_) => {
            "ostree chunk"
        }
        ImportProgress::DerivedLayerStarted(_) | ImportProgress::DerivedLayerCompleted(_) => {
            "layer"
        }
    }
}

/// Configuration for layer progress printing
struct LayerProgressConfig {
    layers: tokio::sync::mpsc::Receiver<ostree_container::store::ImportProgress>,
    layer_bytes: tokio::sync::watch::Receiver<Option<ostree_container::store::LayerProgress>>,
    digest: Box<str>,
    n_layers_to_fetch: usize,
    layers_total: usize,
    bytes_to_download: u64,
    bytes_total: u64,
    progress: PullProgress,
}

/// Write container fetch progress to standard output.
async fn handle_layer_progress_print(mut config: LayerProgressConfig) {
    let start = std::time::Instant::now();
    let mut total_read = 0u64;
    let bar = &config.progress.bars;
    let layers_bar = bar.add(indicatif::ProgressBar::new(
        config.n_layers_to_fetch.try_into().unwrap(),
    ));
    let byte_bar = bar.add(indicatif::ProgressBar::new(0));
    // let byte_bar = indicatif::ProgressBar::new(0);
    // byte_bar.set_draw_target(indicatif::ProgressDrawTarget::hidden());
    layers_bar.set_style(
        indicatif::ProgressStyle::default_bar()
            .template("{prefix} {bar} {pos}/{len} {wide_msg}")
            .unwrap(),
    );
    let taskname = "Fetching layers";
    layers_bar.set_prefix(taskname);
    layers_bar.set_message("");
    byte_bar.set_prefix("Fetching");
    byte_bar.set_style(
        indicatif::ProgressStyle::default_bar()
                .template(
                    " └ {prefix} {bar} {binary_bytes}/{binary_total_bytes} ({binary_bytes_per_sec}) {wide_msg}",
                )
                .unwrap()
        );

    let mut subtasks = vec![];
    let mut subtask: SubTaskBytes = Default::default();
    loop {
        tokio::select! {
            // Always handle layer changes first.
            biased;
            layer = config.layers.recv() => {
                if let Some(l) = layer {
                    let layer = descriptor_of_progress(&l);
                    let layer_type = prefix_of_progress(&l);
                    let short_digest = &layer.digest().digest()[0..21];
                    let layer_size = layer.size();
                    if l.is_starting() {
                        // Reset the progress bar
                        byte_bar.reset_elapsed();
                        byte_bar.reset_eta();
                        byte_bar.set_length(layer_size);
                        byte_bar.set_message(format!("{layer_type} {short_digest}"));

                        subtask = SubTaskBytes {
                            subtask: layer_type.into(),
                            description: format!("{layer_type}: {short_digest}").clone().into(),
                            id: short_digest.to_string().clone().into(),
                            bytes_cached: 0,
                            bytes: 0,
                            bytes_total: layer_size,
                        };
                    } else {
                        // Use the bar's length (actual blob size) rather than
                        // the manifest descriptor size for completion accounting.
                        let actual_size = byte_bar.length().unwrap_or(layer_size);
                        byte_bar.set_position(actual_size);
                        layers_bar.inc(1);
                        total_read = total_read.saturating_add(actual_size);
                        // Emit an event where bytes == total to signal completion.
                        subtask.bytes_total = actual_size;
                        subtask.bytes = actual_size;
                        subtasks.push(subtask.clone());
                        config.progress.json.send(Event::ProgressBytes {
                            task: "pulling".into(),
                            description: format!("Pulling Image: {}", config.digest).into(),
                            id: (*config.digest).into(),
                            bytes_cached: config.bytes_total - config.bytes_to_download,
                            bytes: total_read,
                            bytes_total: config.bytes_to_download,
                            steps_cached: (config.layers_total - config.n_layers_to_fetch) as u64,
                            steps: layers_bar.position(),
                            steps_total: config.n_layers_to_fetch as u64,
                            subtasks: subtasks.clone(),
                        }).await;
                    }
                } else {
                    // If the receiver is disconnected, then we're done
                    break
                };
            },
            r = config.layer_bytes.changed() => {
                if r.is_err() {
                    // If the receiver is disconnected, then we're done
                    break
                }
                let bytes = {
                    let bytes = config.layer_bytes.borrow_and_update();
                    bytes.as_ref().cloned()
                };
                if let Some(bytes) = bytes {
                    // Update the bar length from the actual blob size, which
                    // may differ from the manifest descriptor size (e.g.
                    // containers-storage stores layers uncompressed).
                    byte_bar.set_length(bytes.total);
                    byte_bar.set_position(bytes.fetched);
                    subtask.bytes_total = bytes.total;
                    subtask.bytes = byte_bar.position();
                    config.progress.json.send_lossy(Event::ProgressBytes {
                        task: "pulling".into(),
                        description: format!("Pulling Image: {}", config.digest).into(),
                        id: (*config.digest).into(),
                        bytes_cached: config.bytes_total - config.bytes_to_download,
                        bytes: total_read + byte_bar.position(),
                        bytes_total: config.bytes_to_download,
                        steps_cached: (config.layers_total - config.n_layers_to_fetch) as u64,
                        steps: layers_bar.position(),
                        steps_total: config.n_layers_to_fetch as u64,
                        subtasks: subtasks.clone().into_iter().chain([subtask.clone()]).collect(),
                    }).await;
                }
            }
        }
    }
    byte_bar.finish_and_clear();
    layers_bar.finish_and_clear();
    bar.remove(&byte_bar);
    bar.remove(&layers_bar);
    config.progress.clear();
    let end = std::time::Instant::now();
    let elapsed = end.duration_since(start);
    let persec = total_read as f64 / elapsed.as_secs_f64();
    let persec = indicatif::HumanBytes(persec as u64);
    config.progress.println(format!(
        "Fetched layers: {} in {} ({}/s)",
        indicatif::HumanBytes(total_read),
        indicatif::HumanDuration(elapsed),
        persec,
    ));

    // Since the progress notifier closed, we know import has started
    // use as a heuristic to begin import progress
    // Cannot be lossy or it is dropped
    config
        .progress
        .json
        .send(Event::ProgressSteps {
            task: "importing".into(),
            description: "Importing Image".into(),
            id: (*config.digest).into(),
            steps_cached: 0,
            steps: 0,
            steps_total: 1,
            subtasks: [SubTaskStep {
                subtask: "importing".into(),
                description: "Importing Image".into(),
                id: "importing".into(),
                completed: false,
            }]
            .into(),
        })
        .await;
}

/// Gather all bound images in all deployments, then prune the image store,
/// using the gathered images as the roots (that will not be GC'd).
pub(crate) async fn prune_container_store(sysroot: &Storage) -> Result<()> {
    let ostree = sysroot.get_ostree()?;
    let deployments = ostree.deployments();
    let mut all_bound_images = Vec::new();
    for deployment in deployments {
        let bound = crate::boundimage::query_bound_images_for_deployment(ostree, &deployment)?;
        all_bound_images.extend(bound.into_iter());
        // Also include the host image itself
        // Note: Use just the image name (not the full transport:image format) because
        // podman's image names don't include the transport prefix.
        if let Some(host_image) = crate::status::boot_entry_from_deployment(ostree, &deployment)?
            .image
            .map(|i| i.image)
        {
            all_bound_images.push(crate::boundimage::BoundImage {
                image: host_image.image.clone(),
                auth_file: None,
            });
        }
    }
    // Convert to a hashset of just the image names
    let image_names = HashSet::from_iter(all_bound_images.iter().map(|img| img.image.as_str()));
    let pruned = sysroot
        .get_ensure_imgstore()?
        .prune_except_roots(&image_names)
        .await?;
    tracing::debug!("Pruned images: {}", pruned.len());
    Ok(())
}

/// Core disk space check: verify that `bytes_to_fetch` fits within available space,
/// leaving at least `min_free` bytes reserved.
fn check_disk_space_inner(
    fd: impl AsFd,
    bytes_to_fetch: u64,
    min_free: u64,
    imgref: &ImageReference,
) -> Result<()> {
    let stat = rustix::fs::fstatvfs(fd)?;
    let bytes_avail = stat.f_bsize.checked_mul(stat.f_bavail).unwrap_or(u64::MAX);
    let usable = bytes_avail.saturating_sub(min_free);
    tracing::trace!("bytes_avail: {bytes_avail} min_free: {min_free} usable: {usable}");

    if bytes_to_fetch > usable {
        anyhow::bail!(
            "Insufficient free space for {image} (available: {available} required: {required})",
            available = ostree_ext::glib::format_size(usable),
            required = ostree_ext::glib::format_size(bytes_to_fetch),
            image = imgref.image,
        );
    }
    Ok(())
}

/// Verify there is sufficient disk space to pull an image into the ostree repo.
/// Respects the repository's configured min-free-space threshold.
pub(crate) fn check_disk_space_ostree(
    repo: &ostree::Repo,
    image_meta: &PreparedImportMeta,
    imgref: &ImageReference,
) -> Result<()> {
    let min_free = repo.min_free_space_bytes().unwrap_or(0);
    check_disk_space_inner(
        repo.dfd_borrow(),
        image_meta.bytes_to_fetch,
        min_free,
        imgref,
    )
}

/// Verify there is sufficient disk space to pull an image into the composefs store
/// via the ostree unified-storage path (uses `PreparedImportMeta`).
pub(crate) fn check_disk_space_unified(
    cfs: &crate::store::ComposefsRepository,
    image_meta: &PreparedImportMeta,
    imgref: &ImageReference,
) -> Result<()> {
    check_disk_space_inner(cfs.objects_dir()?, image_meta.bytes_to_fetch, 0, imgref)
}

/// Verify there is sufficient disk space to pull an image into the composefs store
/// for the native composefs backend (uses a raw `ImageManifest`).
pub(crate) fn check_disk_space_composefs(
    cfs: &crate::store::ComposefsRepository,
    manifest: &ostree_ext::oci_spec::image::ImageManifest,
    imgref: &ImageReference,
) -> Result<()> {
    let bytes_to_fetch: u64 = manifest
        .layers()
        .iter()
        .map(|l: &ostree_ext::oci_spec::image::Descriptor| l.size())
        .sum();
    check_disk_space_inner(cfs.objects_dir()?, bytes_to_fetch, 0, imgref)
}

pub(crate) struct PreparedImportMeta {
    pub imp: ImageImporter,
    pub prep: Box<PreparedImport>,
    pub digest: Digest,
    pub n_layers_to_fetch: usize,
    pub layers_total: usize,
    pub bytes_to_fetch: u64,
    pub bytes_total: u64,
}

pub(crate) enum PreparedPullResult {
    Ready(Box<PreparedImportMeta>),
    AlreadyPresent(Box<ImageState>),
}

pub(crate) async fn prepare_for_pull(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    target_imgref: Option<&OstreeImageReference>,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<PreparedPullResult> {
    let imgref_canonicalized = imgref.clone().canonicalize()?;
    tracing::debug!("Canonicalized image reference: {imgref_canonicalized:#}");
    let ostree_imgref = &OstreeImageReference::from(imgref_canonicalized);
    let mut imp = new_importer(repo, ostree_imgref, booted_deployment).await?;
    if let Some(target) = target_imgref {
        imp.set_target(target);
    }
    let prep = match imp.prepare().await? {
        PrepareResult::AlreadyPresent(c) => {
            println!("No changes in {imgref:#} => {}", c.manifest_digest);
            return Ok(PreparedPullResult::AlreadyPresent(Box::new((*c).into())));
        }
        PrepareResult::Ready(p) => p,
    };
    Ok(PreparedPullResult::Ready(Box::new(
        prepared_import_meta(imp, prep).await?,
    )))
}

/// Report on what a prepared import is going to do, and tally it up for the
/// progress display.
async fn prepared_import_meta(
    imp: ImageImporter,
    prep: Box<PreparedImport>,
) -> Result<PreparedImportMeta> {
    check_bootc_label(&prep.config);
    if let Some(warning) = prep.deprecated_warning() {
        ostree_ext::cli::print_deprecated_warning(warning).await;
    }
    ostree_ext::cli::print_layer_status(&prep);
    let layers_to_fetch = prep.layers_to_fetch().collect::<Result<Vec<_>>>()?;

    Ok(PreparedImportMeta {
        imp,
        n_layers_to_fetch: layers_to_fetch.len(),
        layers_total: prep.all_layers().count(),
        bytes_to_fetch: layers_to_fetch.iter().map(|(l, _)| l.layer.size()).sum(),
        bytes_total: prep.all_layers().map(|l| l.layer.size()).sum(),
        digest: prep.manifest_digest.clone(),
        prep,
    })
}

/// Prepare an import whose metadata and layer content both come from a delta
/// file plus the source image already in `repo`; nothing is fetched.
pub(crate) async fn prepare_for_pull_delta(
    repo: &ostree::Repo,
    progress: &PullProgress,
    imgref: &ImageReference,
    delta: Arc<crate::delta::Delta>,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<PreparedPullResult> {
    delta.validate_image_reference(imgref)?;
    let imgref_canonicalized = imgref.clone().canonicalize()?;
    tracing::debug!("Canonicalized image reference: {imgref_canonicalized:#}");
    let ostree_imgref = &OstreeImageReference::from(imgref_canonicalized);

    let mut imp = ImageImporter::new_without_proxy(repo, ostree_imgref)?;
    configure_importer(&mut imp, booted_deployment);

    let layer_source = Box::new(crate::delta_ostree::DeltaLayerSource::new(
        Arc::clone(&delta),
        repo,
    ));
    let prep = match imp
        .prepare_from_manifest(
            delta.target_manifest_digest().clone(),
            delta.target_manifest().clone(),
            delta.target_config().clone(),
            layer_source,
        )
        .await?
    {
        PrepareResult::AlreadyPresent(c) => {
            progress.println(format!("No changes in {imgref:#} => {}", c.manifest_digest));
            return Ok(PreparedPullResult::AlreadyPresent(Box::new((*c).into())));
        }
        PrepareResult::Ready(p) => p,
    };

    Ok(PreparedPullResult::Ready(Box::new(
        prepared_import_meta(imp, prep).await?,
    )))
}

/// Wrapper for applying a delta, wiring up status output.
pub(crate) async fn pull_delta(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    delta: Arc<crate::delta::Delta>,
    quiet: bool,
    prog: ProgressWriter,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<Box<ImageState>> {
    let progress = PullProgress::new(quiet, prog);
    progress.println(format!("Applying delta {}", delta.describe()));
    match prepare_for_pull_delta(repo, &progress, imgref, delta, booted_deployment).await? {
        PreparedPullResult::AlreadyPresent(existing) => Ok(existing),
        PreparedPullResult::Ready(prepared_image_meta) => {
            check_disk_space_ostree(repo, &prepared_image_meta, imgref)?;
            let r = pull_from_prepared(imgref, progress, *prepared_image_meta).await;
            if r.is_err() {
                // The importer publishes a layer's ref before the reconstruction
                // that feeds it has finished being verified, so a failed apply
                // can leave a ref to content that never passed its diff_id
                // check. Drop the unreferenced layers so a retry cannot pick one
                // up; failing to do so is not worth masking the real error.
                if let Err(e) = ostree_container::store::gc_image_layers(repo) {
                    tracing::warn!("Pruning layers after failed delta apply: {e:#}");
                }
            }
            r
        }
    }
}

/// Check whether the image exists in bootc's unified container storage.
///
/// This is used for auto-detection: if the image already exists in bootc storage
/// (e.g., from a previous `bootc image set-unified` or LBI pull), we can use
/// the unified storage path for faster imports.
///
/// Returns true if the image exists in bootc storage.
pub(crate) async fn image_exists_in_unified_storage(
    store: &Storage,
    imgref: &ImageReference,
) -> Result<bool> {
    let imgstore = store.get_ensure_imgstore()?;
    let image_ref_str = imgref.to_transport_image()?;
    imgstore.exists(&image_ref_str).await
}

/// Unified approach: Use bootc's CStorage to pull the image, then prepare from containers-storage.
/// This reuses the same infrastructure as LBIs.
pub(crate) async fn prepare_for_pull_unified(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    target_imgref: Option<&OstreeImageReference>,
    store: &Storage,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<PreparedPullResult> {
    // Get or initialize the bootc container storage (same as used for LBIs)
    let imgstore = store.get_ensure_imgstore()?;

    let image_ref_str = imgref.to_transport_image()?;

    // Always pull to ensure we have the latest image, whether from a remote
    // registry or a locally rebuilt image
    tracing::info!(
        "Unified pull: pulling from transport '{}' to bootc storage",
        &imgref.transport
    );

    // Pull the image into bootc containers-storage with per-layer
    // download progress via the podman native API.
    imgstore
        .pull_with_progress(&image_ref_str)
        .await
        .context("Pulling image into bootc containers-storage")?;

    // Now create a containers-storage reference to read from bootc storage
    tracing::info!("Unified pull: now importing from containers-storage transport");
    let containers_storage_imgref = ImageReference {
        transport: "containers-storage".to_string(),
        image: imgref.image.clone(),
        signature: imgref.signature.clone(),
    };
    let ostree_imgref = OstreeImageReference::from(containers_storage_imgref);

    // Configure the importer to use bootc storage as an additional image store
    let mut config = new_proxy_config();
    let mut cmd = Command::new(skopeo_bin());
    // Use the physical path to bootc storage from the Storage struct
    let storage_path = format!(
        "{}/{}",
        store.physical_root_path,
        crate::podstorage::CStorage::subpath()
    );
    crate::podstorage::set_additional_image_store(&mut cmd, &storage_path);
    config.skopeo_cmd = Some(cmd);

    // Use the preparation flow with the custom config
    let mut imp = new_importer_with_config(repo, &ostree_imgref, config, booted_deployment).await?;
    if let Some(target) = target_imgref {
        imp.set_target(target);
    }
    let prep = match imp.prepare().await? {
        PrepareResult::AlreadyPresent(c) => {
            println!("No changes in {imgref:#} => {}", c.manifest_digest);
            return Ok(PreparedPullResult::AlreadyPresent(Box::new((*c).into())));
        }
        PrepareResult::Ready(p) => p,
    };
    check_bootc_label(&prep.config);
    if let Some(warning) = prep.deprecated_warning() {
        ostree_ext::cli::print_deprecated_warning(warning).await;
    }
    ostree_ext::cli::print_layer_status(&prep);
    let layers_to_fetch = prep.layers_to_fetch().collect::<Result<Vec<_>>>()?;

    // Log that we're importing a new image from containers-storage
    const PULLING_NEW_IMAGE_ID: &str = "6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f1a0";
    tracing::info!(
        message_id = PULLING_NEW_IMAGE_ID,
        bootc.image.reference = &imgref.image,
        bootc.image.transport = "containers-storage",
        bootc.original_transport = &imgref.transport,
        bootc.status = "importing_from_storage",
        "Importing image from bootc storage: {}",
        ostree_imgref
    );

    let prepared_image = PreparedImportMeta {
        imp,
        n_layers_to_fetch: layers_to_fetch.len(),
        layers_total: prep.all_layers().count(),
        bytes_to_fetch: layers_to_fetch.iter().map(|(l, _)| l.layer.size()).sum(),
        bytes_total: prep.all_layers().map(|l| l.layer.size()).sum(),
        digest: prep.manifest_digest.clone(),
        prep,
    };

    Ok(PreparedPullResult::Ready(Box::new(prepared_image)))
}

/// Unified pull: Use podman to pull to containers-storage, then read from there
pub(crate) async fn pull_unified(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    target_imgref: Option<&OstreeImageReference>,
    quiet: bool,
    prog: ProgressWriter,
    store: &Storage,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<Box<ImageState>> {
    let progress = PullProgress::new(quiet, prog);
    match prepare_for_pull_unified(repo, imgref, target_imgref, store, booted_deployment).await? {
        PreparedPullResult::AlreadyPresent(existing) => {
            // Log that the image was already present (Debug level since it's not actionable)
            const IMAGE_ALREADY_PRESENT_ID: &str = "5c4d3e2f1a0b9c8d7e6f5a4b3c2d1e0f9";
            tracing::debug!(
                message_id = IMAGE_ALREADY_PRESENT_ID,
                bootc.image.reference = &imgref.image,
                bootc.image.transport = &imgref.transport,
                bootc.status = "already_present",
                "Image already present: {}",
                imgref
            );
            Ok(existing)
        }
        PreparedPullResult::Ready(prepared_image_meta) => {
            check_disk_space_unified(
                store.get_ensure_composefs()?.as_ref(),
                &prepared_image_meta,
                imgref,
            )?;
            // To avoid duplicate success logs, pass a containers-storage imgref to the importer
            let cs_imgref = ImageReference {
                transport: "containers-storage".to_string(),
                image: imgref.image.clone(),
                signature: imgref.signature.clone(),
            };
            pull_from_prepared(&cs_imgref, progress, *prepared_image_meta).await
        }
    }
}

#[context("Pulling")]
pub(crate) async fn pull_from_prepared(
    imgref: &ImageReference,
    progress: PullProgress,
    mut prepared_image: PreparedImportMeta,
) -> Result<Box<ImageState>> {
    let layer_progress = prepared_image.imp.request_progress();
    let layer_byte_progress = prepared_image.imp.request_layer_progress();
    let digest = prepared_image.digest.clone();
    let digest_imp = prepared_image.digest.clone();

    let printer_progress = progress.clone();
    let printer = tokio::task::spawn(async move {
        handle_layer_progress_print(LayerProgressConfig {
            layers: layer_progress,
            layer_bytes: layer_byte_progress,
            digest: digest.as_ref().into(),
            n_layers_to_fetch: prepared_image.n_layers_to_fetch,
            layers_total: prepared_image.layers_total,
            bytes_to_download: prepared_image.bytes_to_fetch,
            bytes_total: prepared_image.bytes_total,
            progress: printer_progress,
        })
        .await
    });
    let import = prepared_image.imp.import(prepared_image.prep).await;
    printer.await?;
    // Both the progress and the import are done, so import is done as well
    progress
        .json
        .send(Event::ProgressSteps {
            task: "importing".into(),
            description: "Importing Image".into(),
            id: digest_imp.clone().as_ref().into(),
            steps_cached: 0,
            steps: 1,
            steps_total: 1,
            subtasks: [SubTaskStep {
                subtask: "importing".into(),
                description: "Importing Image".into(),
                id: "importing".into(),
                completed: true,
            }]
            .into(),
        })
        .await;
    let import = import?;
    let imgref_canonicalized = imgref.clone().canonicalize()?;
    tracing::debug!("Canonicalized image reference: {imgref_canonicalized:#}");

    // Log successful import completion (skip if using unified storage to avoid double logging)
    let is_unified_path = imgref.transport == "containers-storage";
    if !is_unified_path {
        const IMPORT_COMPLETE_JOURNAL_ID: &str = "4d3e2f1a0b9c8d7e6f5a4b3c2d1e0f9a8";

        tracing::info!(
            message_id = IMPORT_COMPLETE_JOURNAL_ID,
            bootc.image.reference = &imgref.image,
            bootc.image.transport = &imgref.transport,
            bootc.manifest_digest = import.manifest_digest.as_ref(),
            bootc.ostree_commit = &import.merge_commit,
            "Successfully imported image: {}",
            imgref
        );
    }

    if let Some(msg) =
        ostree_container::store::image_filtered_content_warning(&import.filtered_files)
            .context("Image content warning")?
    {
        tracing::info!("{}", msg);
    }
    Ok(Box::new((*import).into()))
}

fn is_retryable_pull_error(error: &anyhow::Error) -> bool {
    // TODO: Also classify transient prepare/OpenImage failures once
    // containers-image-proxy exposes a typed error for them. Its current
    // RequestInitiationFailure cannot safely distinguish DNS/TCP failures from
    // permanent errors such as a missing image or a signature-policy rejection.
    // https://github.com/bootc-dev/bootc/issues/2466
    error.chain().any(|source| {
        matches!(
            source.downcast_ref::<ostree_ext::containers_image_proxy::Error>(),
            Some(ostree_ext::containers_image_proxy::Error::BlobError(
                ostree_ext::containers_image_proxy::GetBlobError::Retryable(_)
            ))
        )
    })
}

fn pull_retry_delay(retry: u32) -> Duration {
    PULL_RETRY_INITIAL_DELAY.saturating_mul(2_u32.saturating_pow(retry))
}

pub(crate) async fn retry_pull_operation<F, Fut, T>(
    progress: &PullProgress,
    mut operation: F,
) -> Result<T>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T>>,
{
    let mut retries = 0;
    loop {
        match operation().await {
            Ok(value) => return Ok(value),
            Err(error) if retries < PULL_MAX_RETRIES && is_retryable_pull_error(&error) => {
                let retry_delay = pull_retry_delay(retries);
                let attempt = retries + 1;
                let description = format!(
                    "Container image pull failed; retrying in {} seconds ({attempt}/{PULL_MAX_RETRIES}): {error}",
                    retry_delay.as_secs()
                );
                progress
                    .send(Event::ProgressSteps {
                        task: "pulling".into(),
                        description: description.into(),
                        id: "pull-retry".into(),
                        steps_cached: 0,
                        steps: attempt.into(),
                        steps_total: PULL_MAX_RETRIES.into(),
                        subtasks: Vec::new(),
                    })
                    .await;
                retries += 1;
                tokio::time::sleep(retry_delay).await;
            }
            Err(error) => return Err(error),
        }
    }
}

async fn pull_once(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    target_imgref: Option<&OstreeImageReference>,
    progress: PullProgress,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<Box<ImageState>> {
    match prepare_for_pull(repo, imgref, target_imgref, booted_deployment).await? {
        PreparedPullResult::AlreadyPresent(existing) => {
            // Log that the image was already present (Debug level since it's not actionable)
            const IMAGE_ALREADY_PRESENT_ID: &str = "5c4d3e2f1a0b9c8d7e6f5a4b3c2d1e0f9";
            tracing::debug!(
                message_id = IMAGE_ALREADY_PRESENT_ID,
                bootc.image.reference = &imgref.image,
                bootc.image.transport = &imgref.transport,
                bootc.status = "already_present",
                "Image already present: {}",
                imgref
            );
            Ok(existing)
        }
        PreparedPullResult::Ready(prepared_image_meta) => {
            // Check disk space before attempting to pull
            check_disk_space_ostree(repo, &prepared_image_meta, imgref)?;
            // Log that we're pulling a new image
            const PULLING_NEW_IMAGE_ID: &str = "6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f1a0";
            tracing::info!(
                message_id = PULLING_NEW_IMAGE_ID,
                bootc.image.reference = &imgref.image,
                bootc.image.transport = &imgref.transport,
                bootc.status = "pulling_new",
                "Pulling new image: {}",
                imgref
            );
            Ok(pull_from_prepared(imgref, progress, *prepared_image_meta).await?)
        }
    }
}

/// Wrapper for pulling a container image, wiring up status output.
pub(crate) async fn pull(
    repo: &ostree::Repo,
    imgref: &ImageReference,
    target_imgref: Option<&OstreeImageReference>,
    quiet: bool,
    prog: ProgressWriter,
    booted_deployment: Option<&ostree::Deployment>,
) -> Result<Box<ImageState>> {
    let progress = PullProgress::new(quiet, prog);
    let operation = || {
        pull_once(
            repo,
            imgref,
            target_imgref,
            progress.clone(),
            booted_deployment,
        )
    };
    if imgref.transport != "registry" {
        return operation().await;
    }
    retry_pull_operation(&progress, operation).await
}

pub(crate) async fn wipe_ostree(sysroot: Sysroot) -> Result<()> {
    tokio::task::spawn_blocking(move || {
        sysroot
            .write_deployments(&[], gio::Cancellable::NONE)
            .context("removing deployments")
    })
    .await??;

    Ok(())
}

pub(crate) async fn cleanup(sysroot: &Storage) -> Result<()> {
    // Log the cleanup operation to systemd journal
    const CLEANUP_JOURNAL_ID: &str = "2f1a0b9c8d7e6f5a4b3c2d1e0f9a8b7c6";

    tracing::info!(
        message_id = CLEANUP_JOURNAL_ID,
        "Starting cleanup of old images and deployments"
    );

    let bound_prune = prune_container_store(sysroot);

    // We create clones (just atomic reference bumps) here to move to the thread.
    let ostree = sysroot.get_ostree_cloned()?;
    let repo = ostree.repo();
    let repo_prune =
        ostree_ext::tokio_util::spawn_blocking_cancellable_flatten(move |cancellable| {
            let locked_sysroot = &SysrootLock::from_assumed_locked(&ostree);
            let cancellable = Some(cancellable);
            let repo = &repo;
            let txn = repo.auto_transaction(cancellable)?;
            let repo = txn.repo();

            // Regenerate our base references.  First, we delete the ones that exist
            for ref_entry in repo
                .list_refs_ext(
                    Some(BASE_IMAGE_PREFIX),
                    ostree::RepoListRefsExtFlags::NONE,
                    cancellable,
                )
                .context("Listing refs")?
                .keys()
            {
                repo.transaction_set_refspec(ref_entry, None);
            }

            // Then, for each deployment which is derived (e.g. has configmaps) we synthesize
            // a base ref to ensure that it's not GC'd.
            for (i, deployment) in ostree.deployments().into_iter().enumerate() {
                let commit = deployment.csum();
                if let Some(base) = get_base_commit(repo, &commit)? {
                    repo.transaction_set_refspec(&format!("{BASE_IMAGE_PREFIX}/{i}"), Some(&base));
                }
            }

            let pruned =
                ostree_container::deploy::prune(locked_sysroot).context("Pruning images")?;
            if !pruned.is_empty() {
                let size = glib::format_size(pruned.objsize);
                println!(
                    "Pruned images: {} (layers: {}, objsize: {})",
                    pruned.n_images, pruned.n_layers, size
                );
            } else {
                tracing::debug!("Nothing to prune");
            }

            Ok(())
        });

    // We run these in parallel mostly because we can.
    tokio::try_join!(repo_prune, bound_prune)?;
    Ok(())
}

/// If commit is a bootc-derived commit (e.g. has configmaps), return its base.
#[context("Finding base commit")]
pub(crate) fn get_base_commit(repo: &ostree::Repo, commit: &str) -> Result<Option<String>> {
    let commitv = repo.load_commit(commit)?.0;
    let commitmeta = commitv.child_value(0);
    let commitmeta = &glib::VariantDict::new(Some(&commitmeta));
    let r = commitmeta.lookup::<String>(BOOTC_DERIVED_KEY)?;
    Ok(r)
}

#[context("Writing deployment")]
async fn deploy(
    sysroot: &Storage,
    from: MergeState,
    image: &ImageState,
    origin: &glib::KeyFile,
    lock_finalization: bool,
) -> Result<Deployment> {
    // Compute the kernel argument overrides. In practice today this API is always expecting
    // a merge deployment. The kargs code also always looks at the booted root (which
    // is a distinct minor issue, but not super important as right now the install path
    // doesn't use this API).
    let (stateroot, override_kargs) = match &from {
        MergeState::MergeDeployment(deployment) => {
            let kargs = crate::bootc_kargs::get_kargs(sysroot, &deployment, image)?;
            (deployment.stateroot().into(), Some(kargs))
        }
        MergeState::Reset { stateroot, kargs } => (stateroot.clone(), Some(kargs.clone())),
    };
    // Clone all the things to move to worker thread
    let ostree = sysroot.get_ostree_cloned()?;
    // ostree::Deployment is incorrectly !Send 😢 so convert it to an integer
    let merge_deployment = from.as_merge_deployment();
    let merge_deployment = merge_deployment.map(|d| d.index() as usize);
    let ostree_commit = image.ostree_commit.to_string();
    // GKeyFile also isn't Send! So we serialize that as a string...
    let origin_data = origin.to_data();
    let r = async_task_with_spinner(
        "Deploying",
        spawn_blocking_cancellable_flatten(move |cancellable| -> Result<_> {
            let ostree = ostree;
            let stateroot = Some(stateroot);
            let mut opts = ostree::SysrootDeployTreeOpts::default();

            // Set finalization lock if requested
            opts.locked = lock_finalization;

            // Because the C API expects a Vec<&str>, convert the Cmdline to string slices.
            // The references borrow from the Cmdline, which outlives this usage.
            let override_kargs_refs = override_kargs
                .as_ref()
                .map(|kargs| kargs.iter_str().collect::<Vec<_>>());
            if let Some(kargs) = override_kargs_refs.as_ref() {
                opts.override_kernel_argv = Some(kargs);
            }

            let deployments = ostree.deployments();
            let merge_deployment = merge_deployment.map(|m| &deployments[m]);
            let origin = glib::KeyFile::new();
            origin.load_from_data(&origin_data, glib::KeyFileFlags::NONE)?;
            let d = ostree.stage_tree_with_options(
                stateroot.as_deref(),
                &ostree_commit,
                Some(&origin),
                merge_deployment,
                &opts,
                Some(cancellable),
            )?;
            Ok(d.index())
        }),
    )
    .await?;
    // SAFETY: We must have a staged deployment
    let ostree = sysroot.get_ostree()?;
    let staged = ostree.staged_deployment().unwrap();
    assert_eq!(staged.index(), r);
    Ok(staged)
}

#[context("Generating origin")]
fn origin_from_imageref(imgref: &ImageReference) -> Result<glib::KeyFile> {
    let origin = glib::KeyFile::new();
    let imgref = OstreeImageReference::from(imgref.clone());
    origin.set_string(
        "origin",
        ostree_container::deploy::ORIGIN_CONTAINER,
        imgref.to_string().as_str(),
    );
    Ok(origin)
}

/// The source of data for staging a new deployment
#[derive(Debug)]
pub(crate) enum MergeState {
    /// Use the provided merge deployment
    MergeDeployment(Deployment),
    /// Don't use a merge deployment, but only this
    /// provided initial state.
    Reset {
        stateroot: String,
        kargs: CmdlineOwned,
    },
}
impl MergeState {
    /// Initialize using the default merge deployment for the given stateroot.
    pub(crate) fn from_stateroot(sysroot: &Storage, stateroot: &str) -> Result<Self> {
        let ostree = sysroot.get_ostree()?;
        let merge_deployment = ostree.merge_deployment(Some(stateroot)).ok_or_else(|| {
            anyhow::anyhow!("No merge deployment found for stateroot {stateroot}")
        })?;
        Ok(Self::MergeDeployment(merge_deployment))
    }

    /// Cast this to a merge deployment case.
    pub(crate) fn as_merge_deployment(&self) -> Option<&Deployment> {
        match self {
            Self::MergeDeployment(d) => Some(d),
            Self::Reset { .. } => None,
        }
    }
}

/// Pull the bound images referenced by an imported commit before staging it.
#[context("Pulling bound images for ostree commit {commit}")]
async fn pull_bound_images_for_commit(sysroot: &Storage, commit: &str) -> Result<()> {
    let repo = sysroot.get_ostree()?.repo();
    let repo_dir = Dir::reopen_dir(&repo.dfd_borrow())?;
    let repo_tmp = repo_dir
        .open_dir("tmp")
        .context("Opening ostree repo tmp/")?;
    let td = cap_std_ext::cap_tempfile::TempDir::new_in(&repo_tmp)?;
    let checkout_mode = if repo.mode() == ostree::RepoMode::Bare {
        ostree::RepoCheckoutMode::None
    } else {
        ostree::RepoCheckoutMode::User
    };
    let checkout_opts = ostree::RepoCheckoutAtOptions {
        mode: checkout_mode,
        ..Default::default()
    };
    let root_name = "root";
    repo.checkout_at(
        Some(&checkout_opts),
        td.as_raw_fd(),
        root_name,
        commit,
        gio::Cancellable::NONE,
    )
    .context("Checking out imported commit")?;
    let root = td.open_dir(root_name)?;
    let bound_images = crate::boundimage::query_bound_images(&root)?;
    crate::boundimage::pull_images(sysroot, bound_images).await
}

/// Stage (queue deployment of) a fetched container image.
#[context("Staging")]
pub(crate) async fn stage(
    sysroot: &Storage,
    from: MergeState,
    image: &ImageState,
    spec: &RequiredHostSpec<'_>,
    prog: ProgressWriter,
    lock_finalization: bool,
) -> Result<()> {
    // Log the staging operation to systemd journal with comprehensive upgrade information
    const STAGE_JOURNAL_ID: &str = "8f7a2b1c3d4e5f6a7b8c9d0e1f2a3b4c";

    tracing::info!(
        message_id = STAGE_JOURNAL_ID,
        bootc.image.reference = &spec.image.image,
        bootc.image.transport = &spec.image.transport,
        bootc.manifest_digest = image.manifest_digest.as_ref(),
        "Staging image for deployment: {} (digest: {})",
        spec.image,
        image.manifest_digest
    );

    let mut subtask = SubTaskStep {
        subtask: "merging".into(),
        description: "Merging Image".into(),
        id: "fetching".into(),
        completed: false,
    };
    let mut subtasks = vec![];
    prog.send(Event::ProgressSteps {
        task: "staging".into(),
        description: "Deploying Image".into(),
        id: image.manifest_digest.clone().as_ref().into(),
        steps_cached: 0,
        steps: 0,
        steps_total: 3,
        subtasks: subtasks
            .clone()
            .into_iter()
            .chain([subtask.clone()])
            .collect(),
    })
    .await;

    subtask.completed = true;
    subtasks.push(subtask.clone());
    subtask.subtask = "bound_images".into();
    subtask.id = "bound_images".into();
    subtask.description = "Pulling Bound Images".into();
    subtask.completed = false;
    prog.send(Event::ProgressSteps {
        task: "staging".into(),
        description: "Deploying Image".into(),
        id: image.manifest_digest.clone().as_ref().into(),
        steps_cached: 0,
        steps: 1,
        steps_total: 3,
        subtasks: subtasks
            .clone()
            .into_iter()
            .chain([subtask.clone()])
            .collect(),
    })
    .await;
    pull_bound_images_for_commit(sysroot, &image.ostree_commit).await?;

    subtask.completed = true;
    subtasks.push(subtask.clone());
    subtask.subtask = "deploying".into();
    subtask.id = "deploying".into();
    subtask.description = "Deploying Image".into();
    subtask.completed = false;
    prog.send(Event::ProgressSteps {
        task: "staging".into(),
        description: "Deploying Image".into(),
        id: image.manifest_digest.clone().as_ref().into(),
        steps_cached: 0,
        steps: 1,
        steps_total: 3,
        subtasks: subtasks
            .clone()
            .into_iter()
            .chain([subtask.clone()])
            .collect(),
    })
    .await;
    let origin = origin_from_imageref(spec.image)?;
    crate::deploy::deploy(sysroot, from, image, &origin, lock_finalization).await?;

    subtask.completed = true;
    subtasks.push(subtask.clone());
    subtask.subtask = "cleanup".into();
    subtask.id = "cleanup".into();
    subtask.description = "Removing old images".into();
    subtask.completed = false;
    prog.send(Event::ProgressSteps {
        task: "staging".into(),
        description: "Deploying Image".into(),
        id: image.manifest_digest.clone().as_ref().into(),
        steps_cached: 0,
        steps: 2,
        steps_total: 3,
        subtasks: subtasks
            .clone()
            .into_iter()
            .chain([subtask.clone()])
            .collect(),
    })
    .await;
    crate::deploy::cleanup(sysroot).await?;

    if !lock_finalization {
        println!("Queued for next boot: {:#}", spec.image);
    } else {
        println!("Staged but not queued for next boot: {:#}", spec.image);
    }

    if let Some(version) = image.version.as_deref() {
        println!("  Version: {version}");
    }
    println!("  Digest: {}", image.manifest_digest);

    subtask.completed = true;
    subtasks.push(subtask.clone());
    prog.send(Event::ProgressSteps {
        task: "staging".into(),
        description: "Deploying Image".into(),
        id: image.manifest_digest.clone().as_ref().into(),
        steps_cached: 0,
        steps: 3,
        steps_total: 3,
        subtasks: subtasks
            .clone()
            .into_iter()
            .chain([subtask.clone()])
            .collect(),
    })
    .await;

    // Unconditionally create or update /run/reboot-required to signal a reboot is needed.
    // This is monitored by kured (Kubernetes Reboot Daemon).
    write_reboot_required(&image.manifest_digest.as_ref())?;

    Ok(())
}

/// Update the /run/reboot-required file with the image that will be active after a reboot.
fn write_reboot_required(image: &str) -> Result<()> {
    let reboot_message = format!("bootc: Reboot required for image: {}", image);
    let run_dir = Dir::open_ambient_dir("/run", cap_std::ambient_authority())?;
    run_dir
        .atomic_write("reboot-required", reboot_message.as_bytes())
        .context("Creating /run/reboot-required")?;

    Ok(())
}

pub(crate) const ROLLBACK_JOURNAL_ID: &str = "26f3b1eb24464d12aa5e7b544a6b5468";

/// Implementation of rollback functionality
pub(crate) async fn rollback(sysroot: &Storage) -> Result<()> {
    let ostree = sysroot.get_ostree()?;
    let (booted_ostree, deployments, host) = crate::status::get_status_require_booted(ostree)?;

    let new_spec = {
        let mut new_spec = host.spec.clone();
        new_spec.boot_order = new_spec.boot_order.swap();
        new_spec
    };

    let repo = &booted_ostree.repo();

    // Just to be sure
    host.spec.verify_transition(&new_spec)?;

    let reverting = new_spec.boot_order == BootOrder::Default;
    if reverting {
        println!("notice: Reverting queued rollback state");
    }
    let rollback_status = host
        .status
        .rollback
        .ok_or_else(|| anyhow!("No rollback available"))?;
    let rollback_image = rollback_status
        .query_image(repo)?
        .ok_or_else(|| anyhow!("Rollback is not container image based"))?;

    // Get current booted image for comparison
    let current_image = host
        .status
        .booted
        .as_ref()
        .and_then(|b| b.query_image(repo).ok()?);

    tracing::info!(
        message_id = ROLLBACK_JOURNAL_ID,
        bootc.manifest_digest = rollback_image.manifest_digest.as_ref(),
        bootc.ostree_commit = &rollback_image.merge_commit,
        bootc.rollback_type = if reverting { "revert" } else { "rollback" },
        bootc.current_manifest_digest = current_image
            .as_ref()
            .map(|i| i.manifest_digest.as_ref())
            .unwrap_or("none"),
        "Rolling back to image: {}",
        rollback_image.manifest_digest
    );
    // SAFETY: If there's a rollback status, then there's a deployment
    let rollback_deployment = deployments.rollback.expect("rollback deployment");
    let new_deployments = if reverting {
        [booted_ostree.deployment, rollback_deployment]
    } else {
        [rollback_deployment, booted_ostree.deployment]
    };
    let new_deployments = new_deployments
        .into_iter()
        .chain(deployments.other)
        .collect::<Vec<_>>();
    tracing::debug!("Writing new deployments: {new_deployments:?}");
    booted_ostree
        .sysroot
        .write_deployments(&new_deployments, gio::Cancellable::NONE)?;
    if reverting {
        println!("Next boot: current deployment");
    } else {
        println!("Next boot: rollback deployment");
    }

    write_reboot_required(rollback_image.manifest_digest.as_ref())?;

    sysroot.update_mtime()?;

    Ok(())
}

fn find_newest_deployment_name(deploysdir: &Dir) -> Result<String> {
    let mut dirs = Vec::new();
    for ent in deploysdir.entries()? {
        let ent = ent?;
        if !ent.file_type()?.is_dir() {
            continue;
        }
        let name = ent.file_name();
        let Some(name) = name.to_str() else {
            continue;
        };
        dirs.push((name.to_owned(), ent.metadata()?.mtime()));
    }
    dirs.sort_unstable_by(|a, b| a.1.cmp(&b.1));
    if let Some((name, _ts)) = dirs.pop() {
        Ok(name)
    } else {
        anyhow::bail!("No deployment directory found")
    }
}

// Implementation of `bootc switch --in-place`
pub(crate) fn switch_origin_inplace(root: &Dir, imgref: &ImageReference) -> Result<String> {
    // Log the in-place switch operation to systemd journal
    const SWITCH_INPLACE_JOURNAL_ID: &str = "3e2f1a0b9c8d7e6f5a4b3c2d1e0f9a8b7";

    tracing::info!(
        message_id = SWITCH_INPLACE_JOURNAL_ID,
        bootc.image.reference = &imgref.image,
        bootc.image.transport = &imgref.transport,
        bootc.switch_type = "in_place",
        "Performing in-place switch to image: {}",
        imgref
    );

    // First, just create the new origin file
    let origin = origin_from_imageref(imgref)?;
    let serialized_origin = origin.to_data();

    // Now, we can't rely on being officially booted (e.g. with the `ostree=` karg)
    // in a scenario like running in the anaconda %post.
    // Eventually, we should support a setup here where ostree-prepare-root
    // can officially be run to "enter" an ostree root in a supportable way.
    // Anyways for now, the brutal hack is to just scrape through the deployments
    // and find the newest one, which we will mutate.  If there's more than one,
    // ultimately the calling tooling should be fixed to set things up correctly.

    let mut ostree_deploys = root.open_dir("sysroot/ostree/deploy")?.entries()?;
    let deploydir = loop {
        if let Some(ent) = ostree_deploys.next() {
            let ent = ent?;
            if !ent.file_type()?.is_dir() {
                continue;
            }
            tracing::debug!("Checking {:?}", ent.file_name());
            let child_dir = ent
                .open_dir()
                .with_context(|| format!("Opening dir {:?}", ent.file_name()))?;
            if let Some(d) = child_dir.open_dir_optional("deploy")? {
                break d;
            }
        } else {
            anyhow::bail!("Failed to find a deployment");
        }
    };
    let newest_deployment = find_newest_deployment_name(&deploydir)?;
    let origin_path = format!("{newest_deployment}.origin");
    if !deploydir.try_exists(&origin_path)? {
        tracing::warn!("No extant origin for {newest_deployment}");
    }
    deploydir
        .atomic_write(&origin_path, serialized_origin.as_bytes())
        .context("Writing origin")?;
    Ok(newest_deployment)
}

/// A workaround for <https://github.com/ostreedev/ostree/issues/3193>
/// as generated by anaconda.
#[context("Updating /etc/fstab for anaconda+composefs")]
pub(crate) fn fixup_etc_fstab(root: &Dir) -> Result<()> {
    let fstab_path = "etc/fstab";
    // Read the old file
    let fd = root
        .open(fstab_path)
        .with_context(|| format!("Opening {fstab_path}"))
        .map(std::io::BufReader::new)?;

    // Helper function to possibly change a line from /etc/fstab.
    // Returns Ok(true) if we made a change (and we wrote the modified line)
    // otherwise returns Ok(false) and the caller should write the original line.
    fn edit_fstab_line(line: &str, mut w: impl Write) -> Result<bool> {
        if line.starts_with('#') {
            return Ok(false);
        }
        let parts = line.split_ascii_whitespace().collect::<Vec<_>>();

        let path_idx = 1;
        let options_idx = 3;
        let (&path, &options) = match (parts.get(path_idx), parts.get(options_idx)) {
            (None, _) => {
                tracing::debug!("No path in entry: {line}");
                return Ok(false);
            }
            (_, None) => {
                tracing::debug!("No options in entry: {line}");
                return Ok(false);
            }
            (Some(p), Some(o)) => (p, o),
        };
        // If this is not the root, we're not matching on it
        if path != "/" {
            return Ok(false);
        }
        // If options already contains `ro`, nothing to do
        if options.split(',').any(|s| s == "ro") {
            return Ok(false);
        }

        writeln!(w, "# {}", crate::generator::BOOTC_EDITED_STAMP)?;

        // SAFETY: we unpacked the options before.
        // This adds `ro` to the option list
        assert!(!options.is_empty()); // Split wouldn't have turned this up if it was empty
        let options = format!("{options},ro");
        for (i, part) in parts.into_iter().enumerate() {
            // TODO: would obviously be nicer to preserve whitespace...but...eh.
            if i > 0 {
                write!(w, " ")?;
            }
            if i == options_idx {
                write!(w, "{options}")?;
            } else {
                write!(w, "{part}")?
            }
        }
        // And add the trailing newline
        writeln!(w)?;
        Ok(true)
    }

    // Read the input, and atomically write a modified version
    root.atomic_replace_with(fstab_path, move |mut w| -> Result<()> {
        for line in fd.lines() {
            let line = line?;
            if !edit_fstab_line(&line, &mut w)? {
                writeln!(w, "{line}")?;
            }
        }
        Ok(())
    })
    .context("Replacing /etc/fstab")?;

    println!("Updated /etc/fstab to add `ro` for `/`");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{AsyncBufReadExt, BufReader};

    fn get_blob_failure(error: ostree_ext::containers_image_proxy::GetBlobError) -> anyhow::Error {
        let error = ostree_ext::containers_image_proxy::Error::BlobError(error);
        anyhow::Error::from(error).context("Unencapsulating base")
    }

    fn retryable_blob_failure(message: &str) -> anyhow::Error {
        get_blob_failure(ostree_ext::containers_image_proxy::GetBlobError::Retryable(
            message.into(),
        ))
    }

    fn permanent_blob_failure(message: &str) -> anyhow::Error {
        get_blob_failure(ostree_ext::containers_image_proxy::GetBlobError::Other(
            message.into(),
        ))
    }

    fn open_image_failure(message: &str) -> anyhow::Error {
        let error = ostree_ext::containers_image_proxy::Error::RequestInitiationFailure {
            method: "OpenImage".into(),
            error: message.into(),
        };
        anyhow::Error::from(error).context("Creating importer")
    }

    #[test]
    fn test_retryable_pull_error_classification() {
        let cases = [
            (retryable_blob_failure("502 Bad Gateway"), true),
            (permanent_blob_failure("blob unknown"), false),
            (open_image_failure("image unknown"), false),
        ];

        for (error, expected) in cases {
            assert_eq!(is_retryable_pull_error(&error), expected, "{error:#}");
        }
    }

    #[tokio::test(start_paused = true)]
    async fn test_retry_pull_operation_succeeds() -> Result<()> {
        let attempts = std::cell::Cell::new(0);
        let start = tokio::time::Instant::now();
        let progress = PullProgress::new(true, ProgressWriter::default());
        let value = retry_pull_operation(&progress, || {
            let attempt = attempts.get() + 1;
            attempts.set(attempt);
            async move {
                if attempt == 1 {
                    Err(retryable_blob_failure("502 Bad Gateway"))
                } else {
                    Ok(42)
                }
            }
        })
        .await?;

        assert_eq!(value, 42);
        assert_eq!(attempts.get(), 2);
        assert_eq!(start.elapsed(), pull_retry_delay(0));
        Ok(())
    }

    #[tokio::test(start_paused = true)]
    async fn test_retry_pull_operation_reports_progress() -> Result<()> {
        let attempts = std::cell::Cell::new(0);
        let (send, recv) = tokio::net::unix::pipe::pipe()?;
        let progress = PullProgress::new(true, ProgressWriter::from(send));

        retry_pull_operation(&progress, || {
            let attempt = attempts.get() + 1;
            attempts.set(attempt);
            async move {
                if attempt == 1 {
                    Err(retryable_blob_failure("502 Bad Gateway"))
                } else {
                    Ok(())
                }
            }
        })
        .await?;

        let mut lines = BufReader::new(recv).lines();
        let start_line = lines.next_line().await?.unwrap();
        let start: Event = serde_json::from_str(&start_line)?;
        assert!(matches!(start, Event::Start { version } if version == "0.1.0"));

        let retry_line = lines.next_line().await?.unwrap();
        let retry: Event = serde_json::from_str(&retry_line)?;
        assert_eq!(
            retry,
            Event::ProgressSteps {
                task: "pulling".into(),
                description:
                    "Container image pull failed; retrying in 2 seconds (1/3): Unencapsulating base"
                        .into(),
                id: "pull-retry".into(),
                steps_cached: 0,
                steps: 1,
                steps_total: 3,
                subtasks: Vec::new(),
            }
        );
        drop(progress);
        assert!(lines.next_line().await?.is_none());
        Ok(())
    }

    #[tokio::test(start_paused = true)]
    async fn test_retry_pull_operation_stops_after_max_attempts() {
        let attempts = std::cell::Cell::new(0);
        let start = tokio::time::Instant::now();
        let progress = PullProgress::new(true, ProgressWriter::default());
        let error = retry_pull_operation(&progress, || {
            attempts.set(attempts.get() + 1);
            async { Err::<(), _>(retryable_blob_failure("registry unavailable")) }
        })
        .await
        .unwrap_err();

        let expected_delay = (0..PULL_MAX_RETRIES)
            .map(pull_retry_delay)
            .fold(Duration::ZERO, Duration::saturating_add);
        assert_eq!(attempts.get(), PULL_MAX_RETRIES + 1);
        assert_eq!(start.elapsed(), expected_delay);
        assert_eq!(
            error.root_cause().to_string(),
            "retryable error: registry unavailable"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn test_retry_pull_operation_does_not_retry_permanent_blob_error() {
        let attempts = std::cell::Cell::new(0);
        let progress = PullProgress::new(true, ProgressWriter::default());
        let error = retry_pull_operation(&progress, || {
            attempts.set(attempts.get() + 1);
            async { Err::<(), _>(permanent_blob_failure("blob unknown")) }
        })
        .await
        .unwrap_err();

        assert_eq!(attempts.get(), 1);
        assert_eq!(error.root_cause().to_string(), "other error: blob unknown");
    }

    #[tokio::test(start_paused = true)]
    async fn test_retry_pull_operation_does_not_retry_opaque_prepare_error() {
        let attempts = std::cell::Cell::new(0);
        let progress = PullProgress::new(true, ProgressWriter::default());
        let error = retry_pull_operation(&progress, || {
            attempts.set(attempts.get() + 1);
            async { Err::<(), _>(open_image_failure("image unknown")) }
        })
        .await
        .unwrap_err();

        assert_eq!(attempts.get(), 1);
        assert_eq!(
            error.root_cause().to_string(),
            "failed to invoke method OpenImage: image unknown"
        );
    }

    #[test]
    fn test_new_proxy_config_user_agent() {
        let config = new_proxy_config();
        let prefix = config
            .user_agent_prefix
            .expect("user_agent_prefix should be set");
        assert!(
            prefix.starts_with("bootc/"),
            "User agent should start with bootc/"
        );
        // Verify the version is present (not just "bootc/")
        assert!(
            prefix.len() > "bootc/".len(),
            "Version should be present after bootc/"
        );
    }

    #[test]
    fn test_switch_inplace() -> Result<()> {
        use cap_std::fs::DirBuilderExt;

        let td = cap_std_ext::cap_tempfile::TempDir::new(cap_std::ambient_authority())?;
        let mut builder = cap_std::fs::DirBuilder::new();
        let builder = builder.recursive(true).mode(0o755);
        let deploydir = "sysroot/ostree/deploy/default/deploy";
        let target_deployment =
            "af36eb0086bb55ac601600478c6168f834288013d60f8870b7851f44bf86c3c5.0";
        td.ensure_dir_with(
            format!("sysroot/ostree/deploy/default/deploy/{target_deployment}"),
            builder,
        )?;
        let deploydir = &td.open_dir(deploydir)?;
        let orig_imgref = ImageReference {
            image: "quay.io/exampleos/original:sometag".into(),
            transport: "registry".into(),
            signature: None,
        };
        {
            let origin = origin_from_imageref(&orig_imgref)?;
            deploydir.atomic_write(
                format!("{target_deployment}.origin"),
                origin.to_data().as_bytes(),
            )?;
        }

        let target_imgref = ImageReference {
            image: "quay.io/someother/otherimage:latest".into(),
            transport: "registry".into(),
            signature: None,
        };

        let replaced = switch_origin_inplace(&td, &target_imgref).unwrap();
        assert_eq!(replaced, target_deployment);
        Ok(())
    }

    #[test]
    fn test_fixup_etc_fstab_default() -> Result<()> {
        let tempdir = cap_std_ext::cap_tempfile::tempdir(cap_std::ambient_authority())?;
        let default = "UUID=f7436547-20ac-43cb-aa2f-eac9632183f6 /boot auto ro 0 0\n";
        tempdir.create_dir_all("etc")?;
        tempdir.atomic_write("etc/fstab", default)?;
        fixup_etc_fstab(&tempdir).unwrap();
        assert_eq!(tempdir.read_to_string("etc/fstab")?, default);
        Ok(())
    }

    #[test]
    fn test_fixup_etc_fstab_multi() -> Result<()> {
        let tempdir = cap_std_ext::cap_tempfile::tempdir(cap_std::ambient_authority())?;
        let default = "UUID=f7436547-20ac-43cb-aa2f-eac9632183f6 /boot auto ro 0 0\n\
UUID=6907-17CA          /boot/efi               vfat    umask=0077,shortname=winnt 0 2\n";
        tempdir.create_dir_all("etc")?;
        tempdir.atomic_write("etc/fstab", default)?;
        fixup_etc_fstab(&tempdir).unwrap();
        assert_eq!(tempdir.read_to_string("etc/fstab")?, default);
        Ok(())
    }

    #[test]
    fn test_fixup_etc_fstab_ro() -> Result<()> {
        let tempdir = cap_std_ext::cap_tempfile::tempdir(cap_std::ambient_authority())?;
        let default = "UUID=f7436547-20ac-43cb-aa2f-eac9632183f6 /boot auto ro 0 0\n\
UUID=1eef9f42-40e3-4bd8-ae20-e9f2325f8b52 /                     xfs   ro 0 0\n\
UUID=6907-17CA          /boot/efi               vfat    umask=0077,shortname=winnt 0 2\n";
        tempdir.create_dir_all("etc")?;
        tempdir.atomic_write("etc/fstab", default)?;
        fixup_etc_fstab(&tempdir).unwrap();
        assert_eq!(tempdir.read_to_string("etc/fstab")?, default);
        Ok(())
    }

    #[test]
    fn test_fixup_etc_fstab_rw() -> Result<()> {
        let tempdir = cap_std_ext::cap_tempfile::tempdir(cap_std::ambient_authority())?;
        // This case uses `defaults`
        let default = "UUID=f7436547-20ac-43cb-aa2f-eac9632183f6 /boot auto ro 0 0\n\
UUID=1eef9f42-40e3-4bd8-ae20-e9f2325f8b52 /                     xfs   defaults 0 0\n\
UUID=6907-17CA          /boot/efi               vfat    umask=0077,shortname=winnt 0 2\n";
        let modified = "UUID=f7436547-20ac-43cb-aa2f-eac9632183f6 /boot auto ro 0 0\n\
# Updated by bootc-fstab-edit.service\n\
UUID=1eef9f42-40e3-4bd8-ae20-e9f2325f8b52 / xfs defaults,ro 0 0\n\
UUID=6907-17CA          /boot/efi               vfat    umask=0077,shortname=winnt 0 2\n";
        tempdir.create_dir_all("etc")?;
        tempdir.atomic_write("etc/fstab", default)?;
        fixup_etc_fstab(&tempdir).unwrap();
        assert_eq!(tempdir.read_to_string("etc/fstab")?, modified);
        Ok(())
    }
    #[test]
    fn test_check_disk_space_inner() -> Result<()> {
        let td = cap_std_ext::cap_tempfile::TempDir::new(cap_std::ambient_authority())?;
        let imgref = ImageReference {
            image: "quay.io/exampleos/exampleos:latest".into(),
            transport: "registry".into(),
            signature: None,
        };

        // 0 bytes needed always passes
        check_disk_space_inner(&*td, 0, 0, &imgref)?;

        // u64::MAX bytes needed always fails
        assert!(check_disk_space_inner(&*td, u64::MAX, 0, &imgref).is_err());

        // With min_free consuming all usable space, even a tiny fetch fails
        assert!(check_disk_space_inner(&*td, 1, u64::MAX, &imgref).is_err());

        Ok(())
    }
}