soma_infra/
compose_recreate_engine.rs

1use soma_fleet::HostRecord;
2use soma_ops::{MutationSendState, Timestamp, VerificationStatus};
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    ComposeRecreateClient, ComposeRecreateOutcome, ComposeRecreateRequest, MutationFailure,
7    MutationResult, MutationVerification, compose_recreate_fingerprint,
8};
9
10/// Coordinates Compose drift checks, force-recreate, and post-state verification.
11#[derive(Debug, Clone, Copy, Default)]
12pub struct ComposeRecreateEngine;
13
14impl ComposeRecreateEngine {
15    /// Recreates a Compose project and verifies every configured service is running.
16    pub async fn execute(
17        &self,
18        client: &dyn ComposeRecreateClient,
19        host: &HostRecord,
20        request: &ComposeRecreateRequest,
21        cancellation: &CancellationToken,
22    ) -> MutationResult<ComposeRecreateOutcome> {
23        ensure_admitted(request, cancellation)?;
24        let config = client
25            .config(host, request.project(), request.deadline(), cancellation)
26            .await
27            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
28        let before = client
29            .status(
30                host,
31                request.project(),
32                None,
33                request.deadline(),
34                cancellation,
35            )
36            .await
37            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
38        let current = compose_recreate_fingerprint(&config, &before)
39            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
40        if current != *request.expected() {
41            return Err(MutationFailure::new(
42                MutationSendState::NotSent,
43                crate::InfraError::InvalidRequest {
44                    domain: "compose-recreate",
45                    message: "Compose configuration or service pre-state changed after planning"
46                        .into(),
47                },
48            ));
49        }
50        let receipt = client.recreate_compose(host, request, cancellation).await?;
51        let after_read = client
52            .status(
53                host,
54                request.project(),
55                None,
56                request.deadline(),
57                cancellation,
58            )
59            .await;
60        let (after, verification_status, summary) = match after_read {
61            Ok(after) => {
62                let mut observed = after
63                    .services
64                    .iter()
65                    .map(|service| service.service.clone())
66                    .collect::<Vec<_>>();
67                observed.sort();
68                observed.dedup();
69                let healthy = after.services.iter().all(|service| {
70                    service
71                        .state
72                        .as_deref()
73                        .is_some_and(|state| state.eq_ignore_ascii_case("running"))
74                        && service.health.as_deref().is_none_or(|health| {
75                            health.eq_ignore_ascii_case("healthy") || health.is_empty()
76                        })
77                        && service.exit_code.unwrap_or(0) == 0
78                });
79                if observed == request.expected().services && healthy {
80                    (
81                        Some(after),
82                        VerificationStatus::Verified,
83                        "all configured Compose services are running after force-recreate".into(),
84                    )
85                } else {
86                    (
87                        Some(after),
88                        VerificationStatus::Failed,
89                        "Compose force-recreate completed without the expected healthy service set"
90                            .into(),
91                    )
92                }
93            }
94            Err(error) => (
95                None,
96                VerificationStatus::Inconclusive,
97                format!("Compose force-recreate completed but status verification failed: {error}"),
98            ),
99        };
100        Ok(ComposeRecreateOutcome {
101            host: host.id().clone(),
102            topology_revision: host.revision().clone(),
103            project: request.project().name().to_owned(),
104            before,
105            after,
106            changed: true,
107            send_state: receipt.send_state,
108            stdout: receipt.stdout,
109            stderr: receipt.stderr,
110            output_truncated: receipt.output_truncated,
111            verification_status,
112            verification: MutationVerification {
113                status: format!("{verification_status:?}").to_ascii_lowercase(),
114                summary,
115            },
116        })
117    }
118}
119
120fn ensure_admitted(
121    request: &ComposeRecreateRequest,
122    cancellation: &CancellationToken,
123) -> MutationResult<()> {
124    if cancellation.is_cancelled() {
125        return Err(MutationFailure::new(
126            MutationSendState::NotSent,
127            soma_fleet::FleetError::Cancelled.into(),
128        ));
129    }
130    if Timestamp::now() >= request.deadline() {
131        return Err(MutationFailure::new(
132            MutationSendState::NotSent,
133            soma_fleet::FleetError::DeadlineExceeded.into(),
134        ));
135    }
136    Ok(())
137}
138
139#[cfg(test)]
140#[path = "compose_recreate_engine_tests.rs"]
141mod tests;
soma_infra/compose_recreate_engine.rs

soma_infra/
compose_recreate_engine.rs
