soma_infra/
process_compose_recreate.rs

1use async_trait::async_trait;
2use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
3use soma_ops::{MutationSendState, Timestamp};
4use tokio_util::sync::CancellationToken;
5
6use crate::{
7    CommandComposeInspector, ComposeRecreateMutator, ComposeRecreateReceipt,
8    ComposeRecreateRequest, InfraError, MutationFailure, MutationResult,
9};
10
11const OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
12
13#[async_trait]
14impl<E> ComposeRecreateMutator for CommandComposeInspector<E>
15where
16    E: CommandExecutor,
17{
18    async fn recreate_compose(
19        &self,
20        host: &HostRecord,
21        request: &ComposeRecreateRequest,
22        cancellation: &CancellationToken,
23    ) -> MutationResult<ComposeRecreateReceipt> {
24        ensure_admitted(request.deadline(), cancellation)?;
25        let args = vec![
26            "compose".into(),
27            "-f".into(),
28            request
29                .project()
30                .config_file()
31                .to_string_lossy()
32                .into_owned(),
33            "up".into(),
34            "-d".into(),
35            "--force-recreate".into(),
36        ];
37        let command = CommandRequest::new("docker", args, request.deadline())
38            .map_err(soma_fleet::FleetError::from)
39            .and_then(|request| {
40                request
41                    .with_output_limits(OUTPUT_LIMIT, OUTPUT_LIMIT)
42                    .map_err(soma_fleet::FleetError::from)
43            })
44            .map_err(|error| {
45                MutationFailure::new(MutationSendState::NotSent, InfraError::from(error))
46            })?;
47        let output = self
48            .executor
49            .execute(host, &command, cancellation)
50            .await
51            .map_err(|error| {
52                MutationFailure::new(MutationSendState::Unknown, InfraError::from(error))
53            })?;
54        if output.exit_code() != Some(0) {
55            return Err(MutationFailure::new(
56                MutationSendState::Sent,
57                InfraError::CommandFailed {
58                    domain: "compose-recreate",
59                    host: host.id().clone(),
60                    exit_code: output.exit_code(),
61                    stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
62                },
63            ));
64        }
65        Ok(ComposeRecreateReceipt {
66            host: host.id().clone(),
67            topology_revision: host.revision().clone(),
68            project: request.project().name().to_owned(),
69            send_state: MutationSendState::Sent,
70            stdout: String::from_utf8_lossy(output.stdout()).trim().to_owned(),
71            stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
72            output_truncated: output.truncated(),
73        })
74    }
75}
76
77fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
78    if cancellation.is_cancelled() {
79        return Err(MutationFailure::new(
80            MutationSendState::NotSent,
81            soma_fleet::FleetError::Cancelled.into(),
82        ));
83    }
84    if Timestamp::now() >= deadline {
85        return Err(MutationFailure::new(
86            MutationSendState::NotSent,
87            soma_fleet::FleetError::DeadlineExceeded.into(),
88        ));
89    }
90    Ok(())
91}
92
93#[cfg(test)]
94#[path = "process_compose_recreate_tests.rs"]
95mod tests;
soma_infra/process_compose_recreate.rs

soma_infra/
process_compose_recreate.rs
