soma_infra/
host_exec_many.rs

1use std::path::PathBuf;
2use std::time::Duration;
3
4use serde::{Deserialize, Serialize};
5use soma_fleet::{FanoutPolicy, FanoutScheduler, HostId, HostRecord, TargetOutcomeKind};
6use soma_ops::MutationSendState;
7use tokio_util::sync::CancellationToken;
8
9use crate::{
10    HostExecMutator, HostExecReceipt, HostExecRequest, InfraError, MutationFailure, MutationResult,
11};
12
13/// Terminal classification for one host-exec fanout target.
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case")]
16pub enum HostExecTargetStatus {
17    /// Command completed with exit code zero.
18    Succeeded,
19    /// Command failed, returned nonzero, or lost backend certainty.
20    Failed,
21    /// Shared cancellation interrupted target accounting.
22    Cancelled,
23    /// Fanout target exceeded its per-target ceiling.
24    TimedOut,
25}
26
27/// Stable-order outcome for one fanout target.
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
29pub struct HostExecTargetResult {
30    /// Target host.
31    pub host: HostId,
32    /// Optional descriptor-bound working directory.
33    pub working_dir: Option<PathBuf>,
34    /// Terminal target status.
35    pub status: HostExecTargetStatus,
36    /// Completed command receipt when available.
37    pub receipt: Option<HostExecReceipt>,
38    /// Bounded error text when execution did not succeed.
39    pub error: Option<String>,
40    /// Conservative backend send state.
41    pub send_state: MutationSendState,
42}
43
44/// Complete stable-order host execution fanout outcome.
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46pub struct HostExecManyOutcome {
47    /// Per-target results in normalized request order.
48    pub results: Vec<HostExecTargetResult>,
49    /// Successful target count.
50    pub succeeded: usize,
51    /// Failed target count, including nonzero exits.
52    pub failed: usize,
53    /// Cancelled target count.
54    pub cancelled: usize,
55    /// Timed-out target count.
56    pub timed_out: usize,
57    /// Aggregate conservative send state.
58    pub send_state: MutationSendState,
59}
60
61impl HostExecManyOutcome {
62    /// Returns whether every target completed with exit code zero.
63    #[must_use]
64    pub fn all_succeeded(&self) -> bool {
65        self.succeeded == self.results.len()
66    }
67}
68
69/// Bounded stable-order host execution fanout coordinator.
70#[derive(Debug, Clone, Copy)]
71pub struct HostExecManyEngine {
72    scheduler: FanoutScheduler,
73}
74
75impl HostExecManyEngine {
76    /// Creates an engine with explicit concurrency and per-target timeout bounds.
77    pub fn new(max_concurrency: usize, per_target_timeout: Duration) -> Result<Self, InfraError> {
78        let policy = FanoutPolicy::new(max_concurrency, per_target_timeout).map_err(|error| {
79            InfraError::InvalidRequest {
80                domain: "host-exec-many",
81                message: error.to_string(),
82            }
83        })?;
84        Ok(Self {
85            scheduler: FanoutScheduler::new(policy),
86        })
87    }
88
89    /// Executes distinct host/request payloads and retains every terminal outcome.
90    pub async fn execute(
91        &self,
92        client: &dyn HostExecMutator,
93        targets: Vec<(HostRecord, HostExecRequest)>,
94        cancellation: CancellationToken,
95    ) -> MutationResult<HostExecManyOutcome> {
96        if targets.is_empty() {
97            return Err(MutationFailure::new(
98                MutationSendState::NotSent,
99                InfraError::InvalidRequest {
100                    domain: "host-exec-many",
101                    message: "at least one target is required".into(),
102                },
103            ));
104        }
105        if cancellation.is_cancelled() {
106            return Err(MutationFailure::new(
107                MutationSendState::NotSent,
108                soma_fleet::FleetError::Cancelled.into(),
109            ));
110        }
111        let descriptors = targets
112            .iter()
113            .map(|(host, request)| {
114                (
115                    host.id().clone(),
116                    request.working_dir().map(ToOwned::to_owned),
117                )
118            })
119            .collect::<Vec<_>>();
120        let report = self
121            .scheduler
122            .run_with_payload(targets, cancellation, |host, request, child| async move {
123                client.exec_host(&host, &request, &child).await
124            })
125            .await;
126        let mut results = Vec::with_capacity(descriptors.len());
127        for outcome in report.into_outcomes() {
128            let (index, host, kind) = outcome.into_parts();
129            let working_dir = descriptors[index].1.clone();
130            results.push(normalize_target(host, working_dir, kind));
131        }
132        let succeeded = count(&results, HostExecTargetStatus::Succeeded);
133        let failed = count(&results, HostExecTargetStatus::Failed);
134        let cancelled = count(&results, HostExecTargetStatus::Cancelled);
135        let timed_out = count(&results, HostExecTargetStatus::TimedOut);
136        let send_state = aggregate_send_state(&results);
137        Ok(HostExecManyOutcome {
138            results,
139            succeeded,
140            failed,
141            cancelled,
142            timed_out,
143            send_state,
144        })
145    }
146}
147
148fn normalize_target(
149    host: HostId,
150    working_dir: Option<PathBuf>,
151    kind: TargetOutcomeKind<HostExecReceipt, MutationFailure>,
152) -> HostExecTargetResult {
153    match kind {
154        TargetOutcomeKind::Succeeded(receipt) => {
155            let succeeded = receipt.exit_code == Some(0);
156            let send_state = receipt.send_state;
157            HostExecTargetResult {
158                host,
159                working_dir,
160                status: if succeeded {
161                    HostExecTargetStatus::Succeeded
162                } else {
163                    HostExecTargetStatus::Failed
164                },
165                error: if succeeded {
166                    None
167                } else {
168                    Some(format!(
169                        "command exited with status {:?}",
170                        receipt.exit_code
171                    ))
172                },
173                receipt: Some(receipt),
174                send_state,
175            }
176        }
177        TargetOutcomeKind::Failed(failure) => HostExecTargetResult {
178            host,
179            working_dir,
180            status: HostExecTargetStatus::Failed,
181            receipt: None,
182            error: Some(failure.error().to_string()),
183            send_state: failure.send_state(),
184        },
185        TargetOutcomeKind::Cancelled => HostExecTargetResult {
186            host,
187            working_dir,
188            status: HostExecTargetStatus::Cancelled,
189            receipt: None,
190            error: Some("target was cancelled after fanout admission".into()),
191            send_state: MutationSendState::Unknown,
192        },
193        TargetOutcomeKind::TimedOut => HostExecTargetResult {
194            host,
195            working_dir,
196            status: HostExecTargetStatus::TimedOut,
197            receipt: None,
198            error: Some("target exceeded its bounded execution timeout".into()),
199            send_state: MutationSendState::Unknown,
200        },
201    }
202}
203
204fn count(results: &[HostExecTargetResult], status: HostExecTargetStatus) -> usize {
205    results
206        .iter()
207        .filter(|result| result.status == status)
208        .count()
209}
210
211fn aggregate_send_state(results: &[HostExecTargetResult]) -> MutationSendState {
212    if results
213        .iter()
214        .any(|result| result.send_state == MutationSendState::Unknown)
215    {
216        MutationSendState::Unknown
217    } else if results
218        .iter()
219        .any(|result| result.send_state == MutationSendState::Sent)
220    {
221        MutationSendState::Sent
222    } else {
223        MutationSendState::NotSent
224    }
225}
226
227#[cfg(test)]
228#[path = "host_exec_many_tests.rs"]
229mod tests;
soma_infra/host_exec_many.rs

soma_infra/
host_exec_many.rs
