soma_infra/
process_compose_build.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
5use soma_ops::{MutationSendState, ProgressEvent, Timestamp};
6use tokio_util::sync::CancellationToken;
7
8use crate::{
9    ComposeBuildMutator, ComposeBuildReceipt, ComposeBuildRequest, InfraError, MutationFailure,
10    MutationProgressReporter, MutationResult,
11};
12
13const OUTPUT_LIMIT: usize = 16 * 1024 * 1024;
14const MAX_PROGRESS_ERRORS: usize = 16;
15
16/// Process-backed Compose build driver.
17pub struct CommandComposeBuildMutator<E> {
18    executor: Arc<E>,
19}
20impl<E> CommandComposeBuildMutator<E> {
21    /// Creates a driver from a fleet command executor.
22    #[must_use]
23    pub fn new(executor: Arc<E>) -> Self {
24        Self { executor }
25    }
26}
27
28#[async_trait]
29impl<E> ComposeBuildMutator for CommandComposeBuildMutator<E>
30where
31    E: CommandExecutor,
32{
33    async fn build_compose(
34        &self,
35        host: &HostRecord,
36        request: &ComposeBuildRequest,
37        progress: &dyn MutationProgressReporter,
38        cancellation: &CancellationToken,
39    ) -> MutationResult<ComposeBuildReceipt> {
40        ensure_admitted(request.deadline(), cancellation)?;
41        let mut errors = Vec::new();
42        report(
43            progress,
44            request,
45            1,
46            "build",
47            "starting Compose image build",
48            &mut errors,
49        );
50        let mut args = vec![
51            "compose".into(),
52            "--progress".into(),
53            "plain".into(),
54            "-f".into(),
55            request
56                .project()
57                .config_file()
58                .to_string_lossy()
59                .into_owned(),
60            "build".into(),
61        ];
62        if let Some(service) = request.service() {
63            args.extend(["--".into(), service.into()]);
64        }
65        let command = CommandRequest::new("docker", args, request.deadline())
66            .map_err(soma_fleet::FleetError::from)
67            .and_then(|request| {
68                request
69                    .with_output_limits(OUTPUT_LIMIT, OUTPUT_LIMIT)
70                    .map_err(soma_fleet::FleetError::from)
71            })
72            .map_err(|error| {
73                MutationFailure::new(MutationSendState::NotSent, InfraError::from(error))
74            })?;
75        let output = self
76            .executor
77            .execute(host, &command, cancellation)
78            .await
79            .map_err(|error| {
80                MutationFailure::new(MutationSendState::Unknown, InfraError::from(error))
81            })?;
82        report(
83            progress,
84            request,
85            2,
86            "build",
87            "Compose image build process completed",
88            &mut errors,
89        );
90        if output.exit_code() != Some(0) {
91            return Err(MutationFailure::new(
92                MutationSendState::Sent,
93                InfraError::CommandFailed {
94                    domain: "compose-build",
95                    host: host.id().clone(),
96                    exit_code: output.exit_code(),
97                    stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
98                },
99            ));
100        }
101        Ok(ComposeBuildReceipt {
102            host: host.id().clone(),
103            topology_revision: host.revision().clone(),
104            project: request.project().name().into(),
105            service: request.service().map(str::to_owned),
106            send_state: MutationSendState::Sent,
107            stdout: String::from_utf8_lossy(output.stdout()).into_owned(),
108            stderr: String::from_utf8_lossy(output.stderr()).into_owned(),
109            output_truncated: output.truncated(),
110            progress_delivery_errors: errors,
111        })
112    }
113}
114fn report(
115    sink: &dyn MutationProgressReporter,
116    request: &ComposeBuildRequest,
117    sequence: u64,
118    phase: &str,
119    message: &str,
120    errors: &mut Vec<String>,
121) {
122    let event = ProgressEvent::new(
123        request.operation_id().clone(),
124        request.operation().clone(),
125        sequence,
126        Timestamp::now(),
127        phase,
128    )
129    .and_then(|event| event.with_message(message));
130    if let Ok(event) = event
131        && let Err(error) = sink.report(&event)
132        && errors.len() < MAX_PROGRESS_ERRORS
133    {
134        errors.push(error);
135    }
136}
137fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
138    if cancellation.is_cancelled() {
139        return Err(MutationFailure::new(
140            MutationSendState::NotSent,
141            soma_fleet::FleetError::Cancelled.into(),
142        ));
143    }
144    if Timestamp::now() >= deadline {
145        return Err(MutationFailure::new(
146            MutationSendState::NotSent,
147            soma_fleet::FleetError::DeadlineExceeded.into(),
148        ));
149    }
150    Ok(())
151}
152
153#[cfg(test)]
154#[path = "process_compose_build_tests.rs"]
155mod tests;
soma_infra/process_compose_build.rs

soma_infra/
process_compose_build.rs
