soma_infra/
container_exec.rs

1use std::path::{Component, Path, PathBuf};
2use std::sync::Arc;
3
4use async_trait::async_trait;
5use serde::{Deserialize, Serialize};
6use soma_fleet::{HostId, HostRecord, TopologyRevision};
7use soma_ops::{MutationSendState, OperationId, OperationName, Timestamp};
8use tokio_util::sync::CancellationToken;
9
10use crate::{InfraError, InfraResult, MutationResult};
11
12const MAX_COMMAND_ARGUMENTS: usize = 256;
13const MAX_ARGUMENT_CHARS: usize = 4096;
14const MAX_OUTPUT_BYTES: usize = 96 * 1024;
15
16/// One non-interactive bounded Docker exec request.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct ContainerExecRequest {
19    operation_id: OperationId,
20    operation: OperationName,
21    container: String,
22    command: Vec<String>,
23    user: Option<String>,
24    working_dir: Option<PathBuf>,
25    deadline: Timestamp,
26}
27
28impl ContainerExecRequest {
29    /// Creates a one-shot non-TTY exec request.
30    pub fn new(
31        operation_id: OperationId,
32        operation: OperationName,
33        container: impl Into<String>,
34        command: Vec<String>,
35        user: Option<String>,
36        working_dir: Option<PathBuf>,
37        deadline: Timestamp,
38    ) -> InfraResult<Self> {
39        let container = container.into();
40        validate_text("container", &container, 256)?;
41        if command.is_empty() || command.len() > MAX_COMMAND_ARGUMENTS {
42            return Err(invalid(format!(
43                "container exec requires 1-{MAX_COMMAND_ARGUMENTS} command arguments"
44            )));
45        }
46        for argument in &command {
47            validate_text("command argument", argument, MAX_ARGUMENT_CHARS)?;
48        }
49        if let Some(user) = &user {
50            validate_text("exec user", user, 256)?;
51        }
52        let working_dir = working_dir.map(validate_working_dir).transpose()?;
53        if deadline <= Timestamp::now() {
54            return Err(invalid("container exec deadline must be in the future"));
55        }
56        Ok(Self {
57            operation_id,
58            operation,
59            container,
60            command,
61            user,
62            working_dir,
63            deadline,
64        })
65    }
66
67    /// Returns the operation identity.
68    #[must_use]
69    pub fn operation_id(&self) -> &OperationId {
70        &self.operation_id
71    }
72    /// Returns the canonical operation name.
73    #[must_use]
74    pub fn operation(&self) -> &OperationName {
75        &self.operation
76    }
77    /// Returns the target container identifier.
78    #[must_use]
79    pub fn container(&self) -> &str {
80        &self.container
81    }
82    /// Returns direct exec argv.
83    #[must_use]
84    pub fn command(&self) -> &[String] {
85        &self.command
86    }
87    /// Returns the optional Docker exec user.
88    #[must_use]
89    pub fn user(&self) -> Option<&str> {
90        self.user.as_deref()
91    }
92    /// Returns the optional absolute container working directory.
93    #[must_use]
94    pub fn working_dir(&self) -> Option<&Path> {
95        self.working_dir.as_deref()
96    }
97    /// Returns the absolute deadline.
98    #[must_use]
99    pub const fn deadline(&self) -> Timestamp {
100        self.deadline
101    }
102    /// Returns the stdout byte ceiling.
103    #[must_use]
104    pub const fn max_stdout_bytes(&self) -> usize {
105        MAX_OUTPUT_BYTES
106    }
107    /// Returns the stderr byte ceiling.
108    #[must_use]
109    pub const fn max_stderr_bytes(&self) -> usize {
110        MAX_OUTPUT_BYTES
111    }
112}
113
114/// Completed non-interactive Docker exec.
115#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116pub struct ContainerExecReceipt {
117    /// Target host.
118    pub host: HostId,
119    /// Exact topology revision.
120    pub topology_revision: TopologyRevision,
121    /// Target container.
122    pub container: String,
123    /// Direct command argv.
124    pub command: Vec<String>,
125    /// Optional exec user.
126    pub user: Option<String>,
127    /// Optional container working directory.
128    pub working_dir: Option<PathBuf>,
129    /// Bounded stdout.
130    pub stdout: String,
131    /// Bounded stderr.
132    pub stderr: String,
133    /// Docker exec exit code when available.
134    pub exit_code: Option<i64>,
135    /// Whether either output stream exceeded its ceiling.
136    pub truncated: bool,
137    /// Whether UTF-8 replacement was required.
138    pub encoding_lossy: bool,
139    /// Backend send state.
140    pub send_state: MutationSendState,
141}
142
143/// Product-neutral non-interactive Docker exec driver.
144#[async_trait]
145pub trait ContainerExecMutator: Send + Sync {
146    /// Executes one direct argv command without a shell or TTY.
147    async fn exec_container(
148        &self,
149        host: &HostRecord,
150        request: &ContainerExecRequest,
151        cancellation: &CancellationToken,
152    ) -> MutationResult<ContainerExecReceipt>;
153}
154
155/// Supplies one host-bound Docker exec client.
156#[async_trait]
157pub trait ContainerExecClientProvider: Send + Sync {
158    /// Creates an exec client bound to the exact host revision.
159    async fn exec_client(
160        &self,
161        host: &HostRecord,
162        cancellation: &CancellationToken,
163    ) -> InfraResult<Arc<dyn ContainerExecMutator>>;
164}
165
166fn validate_text(field: &'static str, value: &str, max: usize) -> InfraResult<()> {
167    let count = value.chars().count();
168    if count == 0 || count > max || value.as_bytes().contains(&0) {
169        Err(invalid(format!("invalid {field}")))
170    } else {
171        Ok(())
172    }
173}
174
175fn validate_working_dir(path: PathBuf) -> InfraResult<PathBuf> {
176    if !path.is_absolute()
177        || path
178            .components()
179            .any(|part| matches!(part, Component::ParentDir | Component::CurDir))
180        || path.to_string_lossy().as_bytes().contains(&0)
181    {
182        Err(invalid(format!(
183            "container working directory must be absolute and normalized: {}",
184            path.display()
185        )))
186    } else {
187        Ok(path)
188    }
189}
190
191fn invalid(message: impl Into<String>) -> InfraError {
192    InfraError::InvalidRequest {
193        domain: "container-exec",
194        message: message.into(),
195    }
196}
197
198#[cfg(test)]
199#[path = "container_exec_tests.rs"]
200mod tests;
soma_infra/container_exec.rs

soma_infra/
container_exec.rs
