soma_infra/
compose_down_engine.rs

1use soma_fleet::HostRecord;
2use soma_ops::{MutationSendState, VerificationStatus};
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    ComposeDownClient, ComposeDownOutcome, ComposeDownRequest, ComposeRecreateFingerprint,
7    InfraError, MutationFailure, MutationResult, MutationVerification,
8    compose_recreate_fingerprint,
9};
10
11/// Verified Docker Compose teardown coordinator.
12#[derive(Debug, Clone, Copy, Default)]
13pub struct ComposeDownEngine;
14
15impl ComposeDownEngine {
16    /// Captures the current normalized Compose config and status fingerprint.
17    pub async fn inspect(
18        &self,
19        client: &dyn ComposeDownClient,
20        host: &HostRecord,
21        project: &crate::ComposeProjectRef,
22        deadline: soma_ops::Timestamp,
23        cancellation: &CancellationToken,
24    ) -> crate::InfraResult<(ComposeRecreateFingerprint, crate::ComposeStatus)> {
25        let config = client.config(host, project, deadline, cancellation).await?;
26        let status = client
27            .status(host, project, None, deadline, cancellation)
28            .await?;
29        let fingerprint = compose_recreate_fingerprint(&config, &status)?;
30        Ok((fingerprint, status))
31    }
32
33    /// Executes and independently verifies Compose teardown.
34    pub async fn execute(
35        &self,
36        client: &dyn ComposeDownClient,
37        host: &HostRecord,
38        request: &ComposeDownRequest,
39        cancellation: &CancellationToken,
40    ) -> MutationResult<ComposeDownOutcome> {
41        admit(request, cancellation)?;
42        let (current, before) = self
43            .inspect(
44                client,
45                host,
46                request.project(),
47                request.deadline(),
48                cancellation,
49            )
50            .await
51            .map_err(not_sent)?;
52        if current != *request.expected() {
53            return Err(not_sent(InfraError::InvalidRequest {
54                domain: "compose-down",
55                message: "Compose config or service state changed after planning".into(),
56            }));
57        }
58        let receipt = client.down_compose(host, request, cancellation).await?;
59        let after = client
60            .status(
61                host,
62                request.project(),
63                None,
64                request.deadline(),
65                cancellation,
66            )
67            .await
68            .map_err(|error| MutationFailure::new(receipt.send_state, error))?;
69        if !after.services.is_empty() {
70            return Err(MutationFailure::new(
71                receipt.send_state,
72                InfraError::InvalidRequest {
73                    domain: "compose-down",
74                    message: "Compose services remain after down".into(),
75                },
76            ));
77        }
78        Ok(ComposeDownOutcome {
79            host: host.id().clone(),
80            topology_revision: host.revision().clone(),
81            project: request.project().name().to_owned(),
82            changed: !before.services.is_empty() || request.remove_volumes(),
83            before,
84            after,
85            receipt,
86            verification_status: VerificationStatus::Verified,
87            verification: MutationVerification {
88                status: "verified".into(),
89                summary: "Compose status reports no remaining services".into(),
90            },
91        })
92    }
93}
94
95fn admit(request: &ComposeDownRequest, cancellation: &CancellationToken) -> MutationResult<()> {
96    if request.remove_volumes() && !request.force() {
97        return Err(not_sent(InfraError::InvalidRequest {
98            domain: "compose-down",
99            message: "remove_volumes=true requires force=true".into(),
100        }));
101    }
102    if cancellation.is_cancelled() {
103        return Err(not_sent(soma_fleet::FleetError::Cancelled.into()));
104    }
105    if request.deadline() <= soma_ops::Timestamp::now() {
106        return Err(not_sent(soma_fleet::FleetError::DeadlineExceeded.into()));
107    }
108    Ok(())
109}
110
111fn not_sent(error: InfraError) -> MutationFailure {
112    MutationFailure::new(MutationSendState::NotSent, error)
113}
114
115#[cfg(test)]
116#[path = "compose_down_engine_tests.rs"]
117mod tests;
soma_infra/compose_down_engine.rs

soma_infra/
compose_down_engine.rs
