a3s_box_runtime/local_execution/
api.rs

1use a3s_box_core::{
2    CreateExecutionRequest, ExecutionGeneration, ExecutionId, ExecutionLease, ExecutionManager,
3    ExecutionManagerError, ExecutionManagerResult, ExecutionReservation, ExecutionSnapshot,
4    ExecutionSnapshotId, ExecutionState, ExecutionStatus, KillExecutionOptions, KillOutcome,
5    OperationId, ReconcileOutcome, RestartExecutionOptions,
6};
7use async_trait::async_trait;
8
9use super::support::{managed_state, outcome_from_record, require_generation, state_conflict};
10use super::{
11    build_managed_record, status_from_record, LocalExecutionManager, ManagedExecutionState,
12    RuntimeUpdate,
13};
14
15#[async_trait]
16impl ExecutionManager for LocalExecutionManager {
17    async fn create(
18        &self,
19        request: CreateExecutionRequest,
20        operation_id: &OperationId,
21    ) -> ExecutionManagerResult<ExecutionReservation> {
22        let execution_id = ExecutionId::new(uuid::Uuid::new_v4().to_string())?;
23        let record = build_managed_record(
24            &self.home_dir,
25            &execution_id,
26            operation_id.clone(),
27            request,
28            chrono::Utc::now(),
29        )?;
30        let reservation = self.reserve(record).await?;
31        super::record::reservation_from_record(reservation.record())
32    }
33
34    async fn start(
35        &self,
36        execution_id: &ExecutionId,
37        expected_generation: ExecutionGeneration,
38    ) -> ExecutionManagerResult<ExecutionLease> {
39        let _lifecycle_lock =
40            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
41        let record = self
42            .get(execution_id)
43            .await?
44            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
45        super::record::validate_record_health(&record)?;
46        require_generation(&record, execution_id, expected_generation)?;
47        self.ensure_started(record).await
48    }
49
50    async fn inspect(&self, execution_id: &ExecutionId) -> ExecutionManagerResult<ExecutionStatus> {
51        let _lifecycle_lock =
52            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
53        let record = self
54            .get(execution_id)
55            .await?
56            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
57        let record = self.stabilize_snapshot(record).await?;
58        let (record, state) = self.observe_record(record).await?;
59        status_from_record(&record, state)
60    }
61
62    async fn read_logs(
63        &self,
64        execution_id: &ExecutionId,
65        expected_generation: ExecutionGeneration,
66    ) -> ExecutionManagerResult<Vec<a3s_box_core::log::LogEntry>> {
67        self.read_structured_logs(execution_id, expected_generation)
68            .await
69    }
70
71    async fn create_filesystem_snapshot(
72        &self,
73        execution_id: &ExecutionId,
74        expected_generation: ExecutionGeneration,
75        snapshot_id: &ExecutionSnapshotId,
76    ) -> ExecutionManagerResult<ExecutionSnapshot> {
77        let _lifecycle_lock =
78            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
79        self.create_snapshot(execution_id, expected_generation, snapshot_id)
80            .await
81    }
82
83    async fn filesystem_snapshot_size(
84        &self,
85        snapshot_id: &ExecutionSnapshotId,
86    ) -> ExecutionManagerResult<Option<u64>> {
87        self.snapshot_size(snapshot_id).await
88    }
89
90    async fn delete_filesystem_snapshot(
91        &self,
92        snapshot_id: &ExecutionSnapshotId,
93    ) -> ExecutionManagerResult<bool> {
94        self.delete_snapshot(snapshot_id).await
95    }
96
97    async fn pause(
98        &self,
99        execution_id: &ExecutionId,
100        expected_generation: ExecutionGeneration,
101        keep_memory: bool,
102    ) -> ExecutionManagerResult<ExecutionLease> {
103        let _lifecycle_lock =
104            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
105        let record = self
106            .get(execution_id)
107            .await?
108            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
109        let record = self.stabilize_snapshot(record).await?;
110        require_generation(&record, execution_id, expected_generation)?;
111        if managed_state(&record)? != ManagedExecutionState::Running {
112            return Err(state_conflict(&record, execution_id, "pause"));
113        }
114        let claimed = self
115            .transition(
116                &record,
117                ManagedExecutionState::Running,
118                ManagedExecutionState::Pausing,
119                RuntimeUpdate::PauseClaim(keep_memory),
120            )
121            .await?;
122        self.finish_pause(claimed).await
123    }
124
125    async fn resume(
126        &self,
127        execution_id: &ExecutionId,
128        expected_generation: ExecutionGeneration,
129    ) -> ExecutionManagerResult<ExecutionLease> {
130        let _lifecycle_lock =
131            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
132        let record = self
133            .get(execution_id)
134            .await?
135            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
136        let record = self.stabilize_snapshot(record).await?;
137        require_generation(&record, execution_id, expected_generation)?;
138        if managed_state(&record)? != ManagedExecutionState::Paused {
139            return Err(state_conflict(&record, execution_id, "resume"));
140        }
141        let claimed = self
142            .transition(
143                &record,
144                ManagedExecutionState::Paused,
145                ManagedExecutionState::Resuming,
146                RuntimeUpdate::None,
147            )
148            .await?;
149        self.finish_resume(claimed).await
150    }
151
152    async fn restart_with_options(
153        &self,
154        execution_id: &ExecutionId,
155        expected_generation: ExecutionGeneration,
156        operation_id: &OperationId,
157        options: RestartExecutionOptions,
158    ) -> ExecutionManagerResult<ExecutionLease> {
159        let _lifecycle_lock =
160            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
161        let record = self
162            .get(execution_id)
163            .await?
164            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
165        super::record::validate_record_health(&record)?;
166        self.restart_record(record, expected_generation, operation_id, options)
167            .await
168    }
169
170    async fn kill(
171        &self,
172        execution_id: &ExecutionId,
173        expected_generation: ExecutionGeneration,
174    ) -> ExecutionManagerResult<KillOutcome> {
175        self.kill_with_options(
176            execution_id,
177            expected_generation,
178            KillExecutionOptions::default(),
179        )
180        .await
181    }
182
183    async fn kill_with_options(
184        &self,
185        execution_id: &ExecutionId,
186        expected_generation: ExecutionGeneration,
187        options: KillExecutionOptions,
188    ) -> ExecutionManagerResult<KillOutcome> {
189        validate_kill_options(options)?;
190        let _lifecycle_lock =
191            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
192        let record = self
193            .get(execution_id)
194            .await?
195            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
196        let record = self.stabilize_snapshot(record).await?;
197        require_generation(&record, execution_id, expected_generation)?;
198        let state = managed_state(&record)?;
199        if state.is_terminal() {
200            return Ok(KillOutcome::AlreadyStopped);
201        }
202        if matches!(
203            state,
204            ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting
205        ) {
206            return Err(state_conflict(&record, execution_id, "kill"));
207        }
208        let claimed = if state == ManagedExecutionState::Killing {
209            record
210        } else {
211            self.transition(
212                &record,
213                state,
214                ManagedExecutionState::Killing,
215                RuntimeUpdate::KillClaim(options),
216            )
217            .await?
218        };
219        self.finish_kill(claimed).await
220    }
221
222    async fn remove(
223        &self,
224        execution_id: &ExecutionId,
225        expected_generation: ExecutionGeneration,
226    ) -> ExecutionManagerResult<bool> {
227        self.remove_execution(execution_id, expected_generation)
228            .await
229    }
230
231    async fn reconcile(
232        &self,
233        operation_id: &OperationId,
234    ) -> ExecutionManagerResult<ReconcileOutcome> {
235        let Some(initial_record) = self.get_by_operation(operation_id).await? else {
236            return Ok(ReconcileOutcome::Absent);
237        };
238        let _lifecycle_lock =
239            super::lifecycle_lock::acquire(&self.home_dir, &initial_record.id).await?;
240        let Some(record) = self.get_by_operation(operation_id).await? else {
241            return Ok(ReconcileOutcome::Absent);
242        };
243        if record.id != initial_record.id {
244            return Err(ExecutionManagerError::Unavailable(format!(
245                "operation {operation_id} changed execution identity while waiting for its lifecycle lock"
246            )));
247        }
248        super::record::validate_record_health(&record)?;
249        match managed_state(&record)? {
250            ManagedExecutionState::Creating | ManagedExecutionState::Created => Ok(
251                ReconcileOutcome::Created(super::record::reservation_from_record(&record)?),
252            ),
253            ManagedExecutionState::Starting => self.recover_start(record).await,
254            ManagedExecutionState::Pausing => {
255                let (record, state) = self.observe_record(record).await?;
256                if managed_state(&record)? == ManagedExecutionState::Pausing
257                    && state == ExecutionState::Running
258                {
259                    return self.finish_pause(record).await.map(ReconcileOutcome::Ready);
260                }
261                outcome_from_record(record, state)
262            }
263            ManagedExecutionState::Resuming => {
264                let (record, state) = self.observe_record(record).await?;
265                if managed_state(&record)? == ManagedExecutionState::Resuming
266                    && state == ExecutionState::Paused
267                {
268                    return self
269                        .finish_resume(record)
270                        .await
271                        .map(ReconcileOutcome::Ready);
272                }
273                outcome_from_record(record, state)
274            }
275            ManagedExecutionState::Snapshotting => self
276                .recover_snapshot(record)
277                .await
278                .map(ReconcileOutcome::Ready),
279            ManagedExecutionState::Killing => {
280                self.finish_kill(record).await?;
281                Ok(ReconcileOutcome::Failed)
282            }
283            ManagedExecutionState::Removing => {
284                self.finish_remove(record).await?;
285                Ok(ReconcileOutcome::Absent)
286            }
287            ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => self
288                .resume_restart(record)
289                .await
290                .map(ReconcileOutcome::Ready),
291            _ => {
292                let (record, state) = self.observe_record(record).await?;
293                outcome_from_record(record, state)
294            }
295        }
296    }
297}
298
299fn validate_kill_options(options: KillExecutionOptions) -> ExecutionManagerResult<()> {
300    if options.signal.is_some_and(|signal| signal <= 0) {
301        return Err(ExecutionManagerError::InvalidRequest(
302            "kill signal must be positive".to_string(),
303        ));
304    }
305    if options
306        .timeout_secs
307        .is_some_and(|timeout| timeout.checked_mul(1_000).is_none())
308    {
309        return Err(ExecutionManagerError::InvalidRequest(
310            "kill timeout is too large".to_string(),
311        ));
312    }
313    Ok(())
314}
a3s_box_runtime/local_execution/api.rs

a3s_box_runtime/local_execution/
api.rs
