soma_infra/
docker_cleanup_engine.rs

1use soma_fleet::HostRecord;
2use soma_ops::MutationSendState;
3use tokio_util::sync::CancellationToken;
4
5use crate::{
6    ContainerListOptions, ContainerState, DockerCleanupClient, DockerPruneFingerprint,
7    DockerPruneOutcome, DockerPruneRequest, DockerPruneTarget, ImageIdentity, ImageListOptions,
8    ImageRemovalFingerprint, ImageRemovalOutcome, ImageRemovalRequest, ImageSummary, InfraError,
9    InfraResult, MutationFailure, MutationResult,
10};
11
12/// Verified Docker image-removal and prune coordinator.
13#[derive(Debug, Clone, Copy, Default)]
14pub struct DockerCleanupEngine;
15
16impl DockerCleanupEngine {
17    /// Resolves one image reference into a stable local identity.
18    pub async fn inspect_image(
19        &self,
20        client: &dyn DockerCleanupClient,
21        host: &HostRecord,
22        reference: &str,
23        cancellation: &CancellationToken,
24    ) -> InfraResult<ImageRemovalFingerprint> {
25        let images = client
26            .list_images(
27                host,
28                &ImageListOptions {
29                    all: true,
30                    dangling_only: false,
31                },
32                cancellation,
33            )
34            .await?;
35        let identity =
36            find_image(&images, reference).ok_or_else(|| InfraError::InvalidRequest {
37                domain: "docker-cleanup",
38                message: format!("image not found: {reference}"),
39            })?;
40        ImageRemovalFingerprint::new(reference, identity)
41    }
42
43    /// Captures the deterministic inventory relevant to one prune target.
44    pub async fn inspect_prune(
45        &self,
46        client: &dyn DockerCleanupClient,
47        host: &HostRecord,
48        target: DockerPruneTarget,
49        cancellation: &CancellationToken,
50    ) -> InfraResult<DockerPruneFingerprint> {
51        let containers = if target_includes(target, DockerPruneTarget::Containers) {
52            client
53                .list_containers(host, &ContainerListOptions::default(), cancellation)
54                .await?
55                .into_iter()
56                .filter(|container| {
57                    !matches!(
58                        container.state,
59                        ContainerState::Running
60                            | ContainerState::Paused
61                            | ContainerState::Restarting
62                            | ContainerState::Removing
63                    )
64                })
65                .filter_map(|container| container.id)
66                .collect()
67        } else {
68            Vec::new()
69        };
70        let images = if target_includes(target, DockerPruneTarget::Images) {
71            client
72                .list_images(
73                    host,
74                    &ImageListOptions {
75                        all: true,
76                        dangling_only: true,
77                    },
78                    cancellation,
79                )
80                .await?
81                .into_iter()
82                .map(|image| image.id)
83                .collect()
84        } else {
85            Vec::new()
86        };
87        let volumes = if target_includes(target, DockerPruneTarget::Volumes) {
88            client
89                .list_volumes(host, cancellation)
90                .await?
91                .into_iter()
92                .map(|volume| volume.name)
93                .collect()
94        } else {
95            Vec::new()
96        };
97        let networks = if target_includes(target, DockerPruneTarget::Networks) {
98            client
99                .list_networks(host, cancellation)
100                .await?
101                .into_iter()
102                .filter_map(|network| network.id.or(network.name))
103                .collect()
104        } else {
105            Vec::new()
106        };
107        let build_cache_bytes = if target_includes(target, DockerPruneTarget::BuildCache) {
108            client
109                .disk_usage(host, cancellation)
110                .await?
111                .build_cache
112                .size_bytes
113        } else {
114            0
115        };
116        DockerPruneFingerprint {
117            host: host.id().clone(),
118            topology_revision: host.revision().clone(),
119            target,
120            containers,
121            images,
122            volumes,
123            networks,
124            build_cache_bytes,
125            sha256: String::new(),
126        }
127        .finalize()
128    }
129
130    /// Removes one exact planned image and verifies it is absent afterward.
131    pub async fn remove_image(
132        &self,
133        client: &dyn DockerCleanupClient,
134        host: &HostRecord,
135        request: &ImageRemovalRequest,
136        cancellation: &CancellationToken,
137    ) -> MutationResult<ImageRemovalOutcome> {
138        admit(request.force, request.deadline, cancellation)?;
139        let current = self
140            .inspect_image(client, host, &request.fingerprint.reference, cancellation)
141            .await
142            .map_err(not_sent)?;
143        if current != request.fingerprint {
144            return Err(not_sent(InfraError::InvalidRequest {
145                domain: "docker-cleanup",
146                message: "image identity changed after planning".into(),
147            }));
148        }
149        let receipt = client.remove_image(host, request, cancellation).await?;
150        let images = client
151            .list_images(
152                host,
153                &ImageListOptions {
154                    all: true,
155                    dangling_only: false,
156                },
157                cancellation,
158            )
159            .await
160            .map_err(|error| MutationFailure::new(receipt.send_state, error))?;
161        let removed = find_image(&images, &request.fingerprint.reference).is_none()
162            && images
163                .iter()
164                .all(|image| image.id != request.fingerprint.identity.id);
165        if !removed {
166            return Err(MutationFailure::new(
167                receipt.send_state,
168                InfraError::Docker("removed image remains visible after mutation".into()),
169            ));
170        }
171        Ok(ImageRemovalOutcome {
172            before: request.fingerprint.clone(),
173            removed,
174            receipt,
175        })
176    }
177
178    /// Prunes one exact planned inventory and verifies reported identities are gone.
179    pub async fn prune(
180        &self,
181        client: &dyn DockerCleanupClient,
182        host: &HostRecord,
183        request: &DockerPruneRequest,
184        cancellation: &CancellationToken,
185    ) -> MutationResult<DockerPruneOutcome> {
186        admit(request.force, request.deadline, cancellation)?;
187        let current = self
188            .inspect_prune(client, host, request.fingerprint.target, cancellation)
189            .await
190            .map_err(not_sent)?;
191        if current != request.fingerprint {
192            return Err(not_sent(InfraError::InvalidRequest {
193                domain: "docker-cleanup",
194                message: "prune inventory changed after planning".into(),
195            }));
196        }
197        let receipt = client.prune(host, request, cancellation).await?;
198        let after = self
199            .inspect_prune(client, host, request.fingerprint.target, cancellation)
200            .await
201            .map_err(|error| MutationFailure::new(receipt.send_state, error))?;
202        verify_prune(&receipt, &request.fingerprint, &after)
203            .map_err(|error| MutationFailure::new(receipt.send_state, error))?;
204        let changed = receipt
205            .scopes
206            .iter()
207            .any(|scope| !scope.deleted.is_empty() || scope.space_reclaimed > 0);
208        Ok(DockerPruneOutcome {
209            before: request.fingerprint.clone(),
210            after,
211            receipt,
212            changed,
213        })
214    }
215}
216
217fn find_image(images: &[ImageSummary], reference: &str) -> Option<ImageIdentity> {
218    images.iter().find_map(|image| {
219        let matches = image.id == reference
220            || image.repo_tags.iter().any(|tag| tag == reference)
221            || image.repo_digests.iter().any(|digest| digest == reference);
222        matches.then(|| ImageIdentity {
223            id: image.id.clone(),
224            repo_tags: image.repo_tags.clone(),
225            repo_digests: image.repo_digests.clone(),
226        })
227    })
228}
229
230fn target_includes(target: DockerPruneTarget, candidate: DockerPruneTarget) -> bool {
231    target == DockerPruneTarget::All || target == candidate
232}
233
234fn verify_prune(
235    receipt: &crate::DockerPruneReceipt,
236    before: &DockerPruneFingerprint,
237    after: &DockerPruneFingerprint,
238) -> InfraResult<()> {
239    for scope in &receipt.scopes {
240        if scope.target == DockerPruneTarget::BuildCache {
241            if scope.space_reclaimed > 0
242                && after.build_cache_bytes
243                    > before
244                        .build_cache_bytes
245                        .saturating_sub(scope.space_reclaimed)
246            {
247                return Err(InfraError::Docker(
248                    "build-cache usage did not reflect reported reclaimed bytes".into(),
249                ));
250            }
251            continue;
252        }
253        let remaining = match scope.target {
254            DockerPruneTarget::Containers => &after.containers,
255            DockerPruneTarget::Images => &after.images,
256            DockerPruneTarget::Volumes => &after.volumes,
257            DockerPruneTarget::Networks => &after.networks,
258            DockerPruneTarget::BuildCache | DockerPruneTarget::All => continue,
259        };
260        if scope
261            .deleted
262            .iter()
263            .any(|deleted| remaining.contains(deleted))
264        {
265            return Err(InfraError::Docker(format!(
266                "{} prune verification still sees a deleted identity",
267                scope.target.as_str()
268            )));
269        }
270    }
271    Ok(())
272}
273
274fn admit(
275    force: bool,
276    deadline: soma_ops::Timestamp,
277    cancellation: &CancellationToken,
278) -> MutationResult<()> {
279    if !force {
280        return Err(not_sent(InfraError::InvalidRequest {
281            domain: "docker-cleanup",
282            message: "force=true is required".into(),
283        }));
284    }
285    if cancellation.is_cancelled() {
286        return Err(not_sent(soma_fleet::FleetError::Cancelled.into()));
287    }
288    if deadline <= soma_ops::Timestamp::now() {
289        return Err(not_sent(soma_fleet::FleetError::DeadlineExceeded.into()));
290    }
291    Ok(())
292}
293
294fn not_sent(error: InfraError) -> MutationFailure {
295    MutationFailure::new(MutationSendState::NotSent, error)
296}
297
298#[cfg(test)]
299#[path = "docker_cleanup_engine_tests.rs"]
300mod tests;
soma_infra/docker_cleanup_engine.rs

soma_infra/
docker_cleanup_engine.rs
