synapse_application/
mutation_runtime.rs

1use serde_json::Value;
2use soma_fleet::{HostId, HostRecord};
3use soma_infra::{
4    ComposeBuildEngine, ComposeDownEngine, ComposeMutationEngine, ComposePullEngine,
5    ComposeRecreateEngine, ContainerLifecycleAction, ContainerLifecycleEngine,
6    ContainerLifecycleRequest, ContainerRecreateEngine, DockerCleanupEngine, FileTransferEngine,
7    ImageBuildEngine, ImagePullEngine,
8};
9use soma_ops::{
10    AccessClass, AuthorizationEvidence, OperationContext, OperationName, OperationPlan, PlanStep,
11    PlannedChange, TargetKind, TargetRef, Timestamp, VerificationStrategy,
12};
13use tokio_util::sync::CancellationToken;
14
15use crate::mutation_ports::SynapseMutationPorts;
16use crate::runtime_params::required_str;
17use crate::{ExecutionError, SynapseCatalog};
18
19pub(crate) const DEFAULT_MUTATION_DEADLINE_MS: i64 = 30_000;
20
21/// Canonical Synapse mutation planner and executor.
22pub struct SynapseMutationRuntime {
23    pub(crate) catalog: &'static SynapseCatalog,
24    pub(crate) ports: SynapseMutationPorts,
25    lifecycle: ContainerLifecycleEngine,
26    pub(crate) compose: ComposeMutationEngine,
27    pub(crate) image_pull: ImagePullEngine,
28    pub(crate) compose_pull: ComposePullEngine,
29    pub(crate) image_build: ImageBuildEngine,
30    pub(crate) compose_build: ComposeBuildEngine,
31    pub(crate) container_recreate: ContainerRecreateEngine,
32    pub(crate) compose_recreate: ComposeRecreateEngine,
33    pub(crate) docker_cleanup: DockerCleanupEngine,
34    pub(crate) compose_down: ComposeDownEngine,
35    pub(crate) file_transfer: FileTransferEngine,
36}
37
38impl SynapseMutationRuntime {
39    /// Creates the mutation runtime from product-owned ports.
40    #[must_use]
41    pub fn new(ports: SynapseMutationPorts) -> Self {
42        Self {
43            catalog: SynapseCatalog::embedded(),
44            ports,
45            lifecycle: ContainerLifecycleEngine::default(),
46            compose: ComposeMutationEngine::default(),
47            image_pull: ImagePullEngine,
48            compose_pull: ComposePullEngine,
49            image_build: ImageBuildEngine,
50            compose_build: ComposeBuildEngine,
51            container_recreate: ContainerRecreateEngine,
52            compose_recreate: ComposeRecreateEngine,
53            docker_cleanup: DockerCleanupEngine,
54            compose_down: ComposeDownEngine,
55            file_transfer: FileTransferEngine,
56        }
57    }
58
59    /// Creates a runtime with an explicit container verification engine.
60    #[must_use]
61    pub fn with_lifecycle_engine(
62        ports: SynapseMutationPorts,
63        lifecycle: ContainerLifecycleEngine,
64    ) -> Self {
65        Self::with_engines(ports, lifecycle, ComposeMutationEngine::default())
66    }
67
68    /// Creates a runtime with explicit container and Compose verification engines.
69    #[must_use]
70    pub fn with_engines(
71        ports: SynapseMutationPorts,
72        lifecycle: ContainerLifecycleEngine,
73        compose: ComposeMutationEngine,
74    ) -> Self {
75        Self {
76            catalog: SynapseCatalog::embedded(),
77            ports,
78            lifecycle,
79            compose,
80            image_pull: ImagePullEngine,
81            compose_pull: ComposePullEngine,
82            image_build: ImageBuildEngine,
83            compose_build: ComposeBuildEngine,
84            container_recreate: ContainerRecreateEngine,
85            compose_recreate: ComposeRecreateEngine,
86            docker_cleanup: DockerCleanupEngine,
87            compose_down: ComposeDownEngine,
88            file_transfer: FileTransferEngine,
89        }
90    }
91
92    pub(crate) async fn plan_container(
93        &self,
94        operation: &OperationName,
95        parameters: &Value,
96        context: &OperationContext,
97    ) -> Result<OperationPlan, ExecutionError> {
98        let action = lifecycle_action(operation)?;
99        let spec = self.mutation_spec(operation)?;
100        self.catalog.validate_parameters(operation, parameters)?;
101        let host = self.resolve_host(required_str(parameters, "host")?).await?;
102        let container = required_str(parameters, "container_id")?;
103        let target = container_target(&host, container)?;
104        let summary = format!(
105            "{} container {container} on host {}",
106            action.action_label(),
107            host.id()
108        );
109        let change = PlannedChange::new(target.clone(), action.action_label(), summary.clone())?;
110        let step = PlanStep::new(1, operation.clone(), target.clone(), summary)?;
111        let verification = VerificationStrategy::new(
112            OperationName::new("container.inspect").expect("static operation name"),
113            format!(
114                "inspect container {container} until the {} post-state is observed",
115                action.action_label()
116            ),
117        )?;
118        OperationPlan::new(
119            context.operation_id().clone(),
120            operation.clone(),
121            target,
122            spec.risk(),
123            spec.reversibility(),
124        )?
125        .with_topology_revision(host.revision().to_string())?
126        .with_change(change)?
127        .with_prerequisite("the target Docker daemon is reachable")?
128        .with_step(step)?
129        .with_verification(verification)?
130        .with_rollback_guidance(rollback_guidance(action))
131        .map_err(ExecutionError::from)
132    }
133
134    pub(crate) async fn execute_container(
135        &self,
136        operation: &OperationName,
137        parameters: &Value,
138        context: &OperationContext,
139        plan: &OperationPlan,
140        authorization: &AuthorizationEvidence,
141        cancellation: &CancellationToken,
142    ) -> Result<soma_ops::OperationResult, ExecutionError> {
143        let started_at = Timestamp::now();
144        let action = lifecycle_action(operation)?;
145        let spec = self.mutation_spec(operation)?;
146        self.catalog.validate_parameters(operation, parameters)?;
147        let host = self.resolve_host(required_str(parameters, "host")?).await?;
148        let container = required_str(parameters, "container_id")?;
149        let target = container_target(&host, container)?;
150        self.validate_admission(
151            operation,
152            context,
153            plan,
154            authorization,
155            &target,
156            &host,
157            started_at,
158            spec.idempotent(),
159            "container.inspect",
160        )?;
161        let deadline = context.deadline().unwrap_or_else(|| {
162            Timestamp::from_unix_millis(
163                started_at
164                    .unix_millis()
165                    .saturating_add(DEFAULT_MUTATION_DEADLINE_MS),
166            )
167        });
168        let request = ContainerLifecycleRequest::new(container, action, deadline)?;
169        let client = match self.ports.docker.mutation_client(&host, cancellation).await {
170            Ok(client) => client,
171            Err(error) => {
172                return self.failure_result(
173                    operation,
174                    context,
175                    target,
176                    started_at,
177                    soma_ops::MutationSendState::NotSent,
178                    spec.retry(),
179                    error,
180                    None,
181                );
182            }
183        };
184        match self
185            .lifecycle
186            .execute(client.as_ref(), &host, &request, cancellation)
187            .await
188        {
189            Ok(outcome) => self.outcome_result(
190                operation,
191                context,
192                target,
193                started_at,
194                spec.retry(),
195                outcome,
196            ),
197            Err(failure) => self.failure_result(
198                operation,
199                context,
200                target,
201                started_at,
202                failure.send_state(),
203                spec.retry(),
204                failure.into_error(),
205                None,
206            ),
207        }
208    }
209
210    pub(crate) fn mutation_spec(
211        &self,
212        operation: &OperationName,
213    ) -> Result<&soma_ops::OperationSpec, ExecutionError> {
214        let spec = self
215            .catalog
216            .operation(operation)
217            .ok_or_else(|| crate::CompatibilityError::UnknownOperation(operation.clone()))?;
218        if spec.access() != AccessClass::Mutation
219            || (lifecycle_action(operation).is_err()
220                && crate::mutation_compose::compose_action(operation).is_err()
221                && !crate::mutation_pull::pull_operation(operation)
222                && !crate::mutation_build::build_operation(operation)
223                && !crate::mutation_recreate::recreate_operation(operation)
224                && !crate::mutation_exec::exec_operation(operation)
225                && !crate::mutation_final_contract::final_operation(operation))
226        {
227            return Err(ExecutionError::UnsupportedOperation(operation.clone()));
228        }
229        Ok(spec)
230    }
231
232    pub(crate) async fn resolve_host(&self, name: &str) -> Result<HostRecord, ExecutionError> {
233        let id = HostId::new(name).map_err(|error| ExecutionError::InvalidParameter {
234            field: "host".into(),
235            message: error.to_string(),
236        })?;
237        self.ports
238            .hosts
239            .snapshot()
240            .await?
241            .get(&id)
242            .cloned()
243            .ok_or_else(|| ExecutionError::HostNotFound(name.to_owned()))
244    }
245}
246
247pub(crate) fn lifecycle_action(
248    operation: &OperationName,
249) -> Result<ContainerLifecycleAction, ExecutionError> {
250    match operation.as_str() {
251        "container.start" => Ok(ContainerLifecycleAction::Start),
252        "container.stop" => Ok(ContainerLifecycleAction::Stop),
253        "container.restart" => Ok(ContainerLifecycleAction::Restart),
254        "container.pause" => Ok(ContainerLifecycleAction::Pause),
255        "container.resume" => Ok(ContainerLifecycleAction::Resume),
256        _ => Err(ExecutionError::UnsupportedOperation(operation.clone())),
257    }
258}
259
260pub(crate) fn container_target(
261    host: &HostRecord,
262    container: &str,
263) -> Result<TargetRef, ExecutionError> {
264    TargetRef::new(TargetKind::Container, container)?
265        .with_host(host.id().to_string())?
266        .with_revision(host.revision().to_string())
267        .map_err(ExecutionError::from)
268}
269
270fn rollback_guidance(action: ContainerLifecycleAction) -> &'static str {
271    match action {
272        ContainerLifecycleAction::Start => {
273            "stop the container to restore the previous stopped state"
274        }
275        ContainerLifecycleAction::Stop => "start the container to restore service availability",
276        ContainerLifecycleAction::Restart => {
277            "inspect container logs and restart again only after correcting the underlying fault"
278        }
279        ContainerLifecycleAction::Pause => "resume the container to restore process scheduling",
280        ContainerLifecycleAction::Resume => {
281            "pause the container to restore the previous paused state"
282        }
283    }
284}
285
286#[cfg(test)]
287#[path = "mutation_runtime_tests.rs"]
288mod tests;
synapse_application/mutation_runtime.rs

synapse_application/
mutation_runtime.rs
