soma_infra/
compose_pull_engine.rs

1use soma_fleet::HostRecord;
2use soma_ops::{MutationSendState, Timestamp, VerificationStatus};
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    ComposePullClient, ComposePullOutcome, ComposePullRequest, ComposePulledImage,
7    ImageListOptions, ImageReader, InfraError, MutationFailure, MutationProgressReporter,
8    MutationResult, MutationVerification,
9};
10
11/// Coordinates Compose image pulls and verifies every configured local image identity.
12#[derive(Debug, Clone, Copy, Default)]
13pub struct ComposePullEngine;
14
15impl ComposePullEngine {
16    /// Pulls and verifies one Compose project or service.
17    pub async fn execute(
18        &self,
19        compose: &dyn ComposePullClient,
20        images: &dyn ImageReader,
21        host: &HostRecord,
22        request: &ComposePullRequest,
23        progress: &dyn MutationProgressReporter,
24        cancellation: &CancellationToken,
25    ) -> MutationResult<ComposePullOutcome> {
26        ensure_admitted(request, cancellation)?;
27        let config = compose
28            .config(host, request.project(), request.deadline(), cancellation)
29            .await
30            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
31        let expected = expected_images(&config.services, request.service())?;
32        let before_rows = images
33            .list_images(host, &ImageListOptions::default(), cancellation)
34            .await
35            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
36        let before = expected
37            .iter()
38            .map(|(service, reference)| {
39                (
40                    service.clone(),
41                    reference.clone(),
42                    crate::image_pull_engine::find_image(&before_rows, reference),
43                )
44            })
45            .collect::<Vec<_>>();
46        let receipt = compose
47            .pull_compose_images(host, request, progress, cancellation)
48            .await?;
49        let after_read = images
50            .list_images(host, &ImageListOptions::default(), cancellation)
51            .await;
52        let (rows, verification_status, summary) = match after_read {
53            Ok(after_rows) => verified_rows(before, &after_rows),
54            Err(error) => {
55                let rows = before
56                    .into_iter()
57                    .map(|(service, reference, before)| ComposePulledImage {
58                        service,
59                        reference,
60                        before,
61                        after: None,
62                        changed: false,
63                        verified: false,
64                    })
65                    .collect();
66                (
67                    rows,
68                    VerificationStatus::Inconclusive,
69                    format!("Compose pull completed but image verification failed: {error}"),
70                )
71            }
72        };
73        let changed = rows.iter().any(|row| row.changed);
74        Ok(ComposePullOutcome {
75            host: host.id().clone(),
76            topology_revision: host.revision().clone(),
77            project: request.project().name().to_owned(),
78            service: request.service().map(str::to_owned),
79            send_state: receipt.send_state,
80            images: rows,
81            changed,
82            progress_delivery_errors: receipt.progress_delivery_errors,
83            output_truncated: receipt.output_truncated,
84            verification_status,
85            verification: MutationVerification {
86                status: format!("{verification_status:?}").to_ascii_lowercase(),
87                summary,
88            },
89        })
90    }
91}
92
93type BeforeImage = (String, String, Option<crate::ImageIdentity>);
94
95fn verified_rows(
96    before: Vec<BeforeImage>,
97    after_rows: &[crate::ImageSummary],
98) -> (Vec<ComposePulledImage>, VerificationStatus, String) {
99    let rows = before
100        .into_iter()
101        .map(|(service, reference, before)| {
102            let after = crate::image_pull_engine::find_image(after_rows, &reference);
103            let changed = match (&before, &after) {
104                (Some(before), Some(after)) => before.id != after.id,
105                (None, Some(_)) => true,
106                _ => false,
107            };
108            ComposePulledImage {
109                service,
110                reference,
111                before,
112                verified: after.is_some(),
113                after,
114                changed,
115            }
116        })
117        .collect::<Vec<_>>();
118    if rows.iter().all(|row| row.verified) {
119        (
120            rows,
121            VerificationStatus::Verified,
122            "all configured Compose image references resolve locally".into(),
123        )
124    } else {
125        (
126            rows,
127            VerificationStatus::Failed,
128            "one or more configured Compose image references were not found locally".into(),
129        )
130    }
131}
132
133fn expected_images(
134    services: &std::collections::BTreeMap<String, crate::ComposeServiceConfig>,
135    selected: Option<&str>,
136) -> MutationResult<Vec<(String, String)>> {
137    if let Some(selected) = selected {
138        let service = services
139            .get(selected)
140            .ok_or_else(|| invalid_request(format!("Compose service {selected} was not found")))?;
141        let image = service
142            .image
143            .clone()
144            .filter(|image| !image.is_empty())
145            .ok_or_else(|| {
146                invalid_request(format!("Compose service {selected} has no image reference"))
147            })?;
148        return Ok(vec![(selected.to_owned(), image)]);
149    }
150    let images = services
151        .iter()
152        .filter_map(|(name, service)| service.image.clone().map(|image| (name.clone(), image)))
153        .filter(|(_, image)| !image.is_empty())
154        .collect::<Vec<_>>();
155    if images.is_empty() {
156        Err(invalid_request(
157            "Compose project has no pullable image references".into(),
158        ))
159    } else {
160        Ok(images)
161    }
162}
163
164fn invalid_request(message: String) -> MutationFailure {
165    MutationFailure::new(
166        MutationSendState::NotSent,
167        InfraError::InvalidRequest {
168            domain: "compose-pull",
169            message,
170        },
171    )
172}
173
174fn ensure_admitted(
175    request: &ComposePullRequest,
176    cancellation: &CancellationToken,
177) -> MutationResult<()> {
178    if cancellation.is_cancelled() {
179        return Err(MutationFailure::new(
180            MutationSendState::NotSent,
181            soma_fleet::FleetError::Cancelled.into(),
182        ));
183    }
184    if Timestamp::now() >= request.deadline() {
185        return Err(MutationFailure::new(
186            MutationSendState::NotSent,
187            soma_fleet::FleetError::DeadlineExceeded.into(),
188        ));
189    }
190    Ok(())
191}
192
193#[cfg(test)]
194#[path = "compose_pull_engine_tests.rs"]
195mod tests;
soma_infra/compose_pull_engine.rs

soma_infra/
compose_pull_engine.rs
