soma_infra/
image_pull_engine.rs

1use soma_fleet::{HostRecord, TopologyRevision};
2use soma_ops::{MutationSendState, VerificationStatus};
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    DockerArtifactClient, ImageIdentity, ImageListOptions, ImagePullOutcome, ImagePullRequest,
7    ImageSummary, MutationFailure, MutationProgressReporter, MutationResult, MutationVerification,
8};
9
10/// Coordinates one image pull and independent image-store verification.
11#[derive(Debug, Clone, Copy, Default)]
12pub struct ImagePullEngine;
13
14impl ImagePullEngine {
15    /// Pulls an image and verifies its local content identity.
16    pub async fn execute(
17        &self,
18        client: &dyn DockerArtifactClient,
19        host: &HostRecord,
20        request: &ImagePullRequest,
21        progress: &dyn MutationProgressReporter,
22        cancellation: &CancellationToken,
23    ) -> MutationResult<ImagePullOutcome> {
24        ensure_admitted(request, cancellation)?;
25        let before = client
26            .list_images(host, &ImageListOptions::default(), cancellation)
27            .await
28            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
29        let before = find_image(&before, request.image());
30        let receipt = client
31            .pull_image(host, request, progress, cancellation)
32            .await?;
33        let after_read = client
34            .list_images(host, &ImageListOptions::default(), cancellation)
35            .await;
36        let (after, verification_status, summary) = match after_read {
37            Ok(images) => match find_image(&images, request.image()) {
38                Some(image) => (
39                    Some(image),
40                    VerificationStatus::Verified,
41                    "the requested image reference resolves to a local content identity".into(),
42                ),
43                None => (
44                    None,
45                    VerificationStatus::Failed,
46                    "the pull stream completed but the requested image reference was not found locally"
47                        .into(),
48                ),
49            },
50            Err(error) => (
51                None,
52                VerificationStatus::Inconclusive,
53                format!("the pull stream completed but image verification failed: {error}"),
54            ),
55        };
56        let changed = match (&before, &after) {
57            (Some(before), Some(after)) => before.id != after.id,
58            (None, Some(_)) => true,
59            _ => false,
60        };
61        Ok(ImagePullOutcome {
62            host: host.id().clone(),
63            topology_revision: TopologyRevision::clone(host.revision()),
64            image: request.image().to_owned(),
65            changed,
66            send_state: receipt.send_state,
67            before,
68            after,
69            total_events: receipt.total_events,
70            progress: receipt.progress,
71            progress_truncated: receipt.progress_truncated,
72            progress_delivery_errors: receipt.progress_delivery_errors,
73            verification_status,
74            verification: MutationVerification {
75                status: verification_status_text(verification_status),
76                summary,
77            },
78        })
79    }
80}
81
82fn ensure_admitted(
83    request: &ImagePullRequest,
84    cancellation: &CancellationToken,
85) -> MutationResult<()> {
86    if cancellation.is_cancelled() {
87        return Err(MutationFailure::new(
88            MutationSendState::NotSent,
89            soma_fleet::FleetError::Cancelled.into(),
90        ));
91    }
92    if soma_ops::Timestamp::now() >= request.deadline() {
93        return Err(MutationFailure::new(
94            MutationSendState::NotSent,
95            soma_fleet::FleetError::DeadlineExceeded.into(),
96        ));
97    }
98    Ok(())
99}
100
101pub(crate) fn find_image(images: &[ImageSummary], reference: &str) -> Option<ImageIdentity> {
102    let canonical = crate::canonical_image_reference(reference);
103    images
104        .iter()
105        .find(|image| {
106            image.id == reference
107                || image.repo_tags.iter().any(|tag| tag == &canonical)
108                || image.repo_digests.iter().any(|digest| digest == reference)
109        })
110        .map(|image| ImageIdentity {
111            id: image.id.clone(),
112            repo_tags: image.repo_tags.clone(),
113            repo_digests: image.repo_digests.clone(),
114        })
115}
116
117fn verification_status_text(status: VerificationStatus) -> String {
118    format!("{status:?}").to_ascii_lowercase()
119}
120
121#[cfg(test)]
122#[path = "image_pull_engine_tests.rs"]
123mod tests;
soma_infra/image_pull_engine.rs

soma_infra/
image_pull_engine.rs
