soma_infra/
compose_build_engine.rs

1use soma_fleet::HostRecord;
2use soma_ops::{MutationSendState, VerificationStatus};
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    BuildContextInspector, ComposeBuildMutator, ComposeBuildOutcome, ComposeBuildRequest,
7    ComposeBuiltImage, ImageListOptions, ImageReader, MutationFailure, MutationProgressReporter,
8    MutationResult, MutationVerification,
9};
10
11/// Borrowed services required for one Compose build execution.
12#[derive(Clone, Copy)]
13pub struct ComposeBuildServices<'a> {
14    /// Descriptor-confined context inspector.
15    pub contexts: &'a dyn BuildContextInspector,
16    /// Compose build mutation driver.
17    pub mutator: &'a dyn ComposeBuildMutator,
18    /// Docker image-store reader used for verification.
19    pub images: &'a dyn ImageReader,
20}
21
22/// Coordinates context drift checks, Compose build, and image verification.
23#[derive(Debug, Clone, Copy, Default)]
24pub struct ComposeBuildEngine;
25
26impl ComposeBuildEngine {
27    /// Builds selected Compose services and verifies every output image.
28    pub async fn execute(
29        &self,
30        services: ComposeBuildServices<'_>,
31        host: &HostRecord,
32        request: &ComposeBuildRequest,
33        progress: &dyn MutationProgressReporter,
34        cancellation: &CancellationToken,
35    ) -> MutationResult<ComposeBuildOutcome> {
36        ensure_admitted(request, cancellation)?;
37        for artifact in request.artifacts() {
38            let actual = services
39                .contexts
40                .fingerprint(host, &artifact.context, request.deadline(), cancellation)
41                .await
42                .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
43            if actual != artifact.fingerprint {
44                return Err(MutationFailure::new(
45                    MutationSendState::NotSent,
46                    crate::InfraError::InvalidRequest {
47                        domain: "compose-build",
48                        message: format!(
49                            "build context changed after planning for service {}",
50                            artifact.service
51                        ),
52                    },
53                ));
54            }
55        }
56        let before = services
57            .images
58            .list_images(host, &ImageListOptions::default(), cancellation)
59            .await
60            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
61        let receipt = services
62            .mutator
63            .build_compose(host, request, progress, cancellation)
64            .await?;
65        let after_read = services
66            .images
67            .list_images(host, &ImageListOptions::default(), cancellation)
68            .await;
69        let (rows, status, summary) = match after_read {
70            Ok(after) => {
71                let rows = request
72                    .artifacts()
73                    .iter()
74                    .map(|artifact| {
75                        let old = crate::image_pull_engine::find_image(&before, &artifact.image);
76                        let new = crate::image_pull_engine::find_image(&after, &artifact.image);
77                        let changed = match (&old, &new) {
78                            (Some(a), Some(b)) => a.id != b.id,
79                            (None, Some(_)) => true,
80                            _ => false,
81                        };
82                        ComposeBuiltImage {
83                            service: artifact.service.clone(),
84                            image: artifact.image.clone(),
85                            context: artifact.fingerprint.clone(),
86                            before: old,
87                            verified: new.is_some(),
88                            after: new,
89                            changed,
90                        }
91                    })
92                    .collect::<Vec<_>>();
93                let verified = rows.iter().all(|row| row.verified);
94                (
95                    rows,
96                    if verified {
97                        VerificationStatus::Verified
98                    } else {
99                        VerificationStatus::Failed
100                    },
101                    if verified {
102                        "all Compose build output images resolve locally".into()
103                    } else {
104                        "one or more Compose build output images were not found locally".into()
105                    },
106                )
107            }
108            Err(error) => (
109                Vec::new(),
110                VerificationStatus::Inconclusive,
111                format!("Compose build completed but image verification failed: {error}"),
112            ),
113        };
114        let changed = rows.iter().any(|row| row.changed);
115        Ok(ComposeBuildOutcome {
116            host: host.id().clone(),
117            topology_revision: host.revision().clone(),
118            project: request.project().name().into(),
119            service: request.service().map(str::to_owned),
120            images: rows,
121            changed,
122            send_state: receipt.send_state,
123            stdout: receipt.stdout,
124            stderr: receipt.stderr,
125            output_truncated: receipt.output_truncated,
126            progress_delivery_errors: receipt.progress_delivery_errors,
127            verification_status: status,
128            verification: MutationVerification {
129                status: format!("{status:?}").to_ascii_lowercase(),
130                summary,
131            },
132        })
133    }
134}
135fn ensure_admitted(
136    request: &ComposeBuildRequest,
137    cancellation: &CancellationToken,
138) -> MutationResult<()> {
139    if cancellation.is_cancelled() {
140        return Err(MutationFailure::new(
141            MutationSendState::NotSent,
142            soma_fleet::FleetError::Cancelled.into(),
143        ));
144    }
145    if soma_ops::Timestamp::now() >= request.deadline() {
146        return Err(MutationFailure::new(
147            MutationSendState::NotSent,
148            soma_fleet::FleetError::DeadlineExceeded.into(),
149        ));
150    }
151    Ok(())
152}
153
154#[cfg(test)]
155#[path = "compose_build_engine_tests.rs"]
156mod tests;
soma_infra/compose_build_engine.rs

soma_infra/
compose_build_engine.rs
