a3s_box_runtime/local_execution/
api.rs1use a3s_box_core::{
2 CreateExecutionRequest, ExecutionGeneration, ExecutionId, ExecutionLease, ExecutionManager,
3 ExecutionManagerError, ExecutionManagerResult, ExecutionReservation, ExecutionSnapshot,
4 ExecutionSnapshotId, ExecutionState, ExecutionStatus, KillOutcome, OperationId,
5 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 record = self
40 .get(execution_id)
41 .await?
42 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
43 require_generation(&record, execution_id, expected_generation)?;
44 self.ensure_started(record).await
45 }
46
47 async fn inspect(&self, execution_id: &ExecutionId) -> ExecutionManagerResult<ExecutionStatus> {
48 let record = self
49 .get(execution_id)
50 .await?
51 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
52 let record = self.stabilize_snapshot(record).await?;
53 let (record, state) = self.observe_record(record).await?;
54 status_from_record(&record, state)
55 }
56
57 async fn read_logs(
58 &self,
59 execution_id: &ExecutionId,
60 expected_generation: ExecutionGeneration,
61 ) -> ExecutionManagerResult<Vec<a3s_box_core::log::LogEntry>> {
62 self.read_structured_logs(execution_id, expected_generation)
63 .await
64 }
65
66 async fn create_filesystem_snapshot(
67 &self,
68 execution_id: &ExecutionId,
69 expected_generation: ExecutionGeneration,
70 snapshot_id: &ExecutionSnapshotId,
71 ) -> ExecutionManagerResult<ExecutionSnapshot> {
72 self.create_snapshot(execution_id, expected_generation, snapshot_id)
73 .await
74 }
75
76 async fn filesystem_snapshot_size(
77 &self,
78 snapshot_id: &ExecutionSnapshotId,
79 ) -> ExecutionManagerResult<Option<u64>> {
80 self.snapshot_size(snapshot_id).await
81 }
82
83 async fn delete_filesystem_snapshot(
84 &self,
85 snapshot_id: &ExecutionSnapshotId,
86 ) -> ExecutionManagerResult<bool> {
87 self.delete_snapshot(snapshot_id).await
88 }
89
90 async fn pause(
91 &self,
92 execution_id: &ExecutionId,
93 expected_generation: ExecutionGeneration,
94 keep_memory: bool,
95 ) -> ExecutionManagerResult<ExecutionLease> {
96 let record = self
97 .get(execution_id)
98 .await?
99 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
100 let record = self.stabilize_snapshot(record).await?;
101 require_generation(&record, execution_id, expected_generation)?;
102 if managed_state(&record)? != ManagedExecutionState::Running {
103 return Err(state_conflict(&record, execution_id, "pause"));
104 }
105 let claimed = self
106 .transition(
107 &record,
108 ManagedExecutionState::Running,
109 ManagedExecutionState::Pausing,
110 RuntimeUpdate::PauseClaim(keep_memory),
111 )
112 .await?;
113 self.finish_pause(claimed).await
114 }
115
116 async fn resume(
117 &self,
118 execution_id: &ExecutionId,
119 expected_generation: ExecutionGeneration,
120 ) -> ExecutionManagerResult<ExecutionLease> {
121 let record = self
122 .get(execution_id)
123 .await?
124 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
125 let record = self.stabilize_snapshot(record).await?;
126 require_generation(&record, execution_id, expected_generation)?;
127 if managed_state(&record)? != ManagedExecutionState::Paused {
128 return Err(state_conflict(&record, execution_id, "resume"));
129 }
130 let claimed = self
131 .transition(
132 &record,
133 ManagedExecutionState::Paused,
134 ManagedExecutionState::Resuming,
135 RuntimeUpdate::None,
136 )
137 .await?;
138 self.finish_resume(claimed).await
139 }
140
141 async fn restart_with_options(
142 &self,
143 execution_id: &ExecutionId,
144 expected_generation: ExecutionGeneration,
145 operation_id: &OperationId,
146 options: RestartExecutionOptions,
147 ) -> ExecutionManagerResult<ExecutionLease> {
148 let record = self
149 .get(execution_id)
150 .await?
151 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
152 self.restart_record(record, expected_generation, operation_id, options)
153 .await
154 }
155
156 async fn kill(
157 &self,
158 execution_id: &ExecutionId,
159 expected_generation: ExecutionGeneration,
160 ) -> ExecutionManagerResult<KillOutcome> {
161 let record = self
162 .get(execution_id)
163 .await?
164 .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
165 let record = self.stabilize_snapshot(record).await?;
166 require_generation(&record, execution_id, expected_generation)?;
167 let state = managed_state(&record)?;
168 if state.is_terminal() {
169 return Ok(KillOutcome::AlreadyStopped);
170 }
171 if matches!(
172 state,
173 ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting
174 ) {
175 return Err(state_conflict(&record, execution_id, "kill"));
176 }
177 let claimed = if state == ManagedExecutionState::Killing {
178 record
179 } else {
180 self.transition(
181 &record,
182 state,
183 ManagedExecutionState::Killing,
184 RuntimeUpdate::None,
185 )
186 .await?
187 };
188 self.finish_kill(claimed).await
189 }
190
191 async fn reconcile(
192 &self,
193 operation_id: &OperationId,
194 ) -> ExecutionManagerResult<ReconcileOutcome> {
195 let Some(record) = self.get_by_operation(operation_id).await? else {
196 return Ok(ReconcileOutcome::Absent);
197 };
198 match managed_state(&record)? {
199 ManagedExecutionState::Creating | ManagedExecutionState::Created => Ok(
200 ReconcileOutcome::Created(super::record::reservation_from_record(&record)?),
201 ),
202 ManagedExecutionState::Starting => self.recover_start(record).await,
203 ManagedExecutionState::Pausing => {
204 let (record, state) = self.observe_record(record).await?;
205 if managed_state(&record)? == ManagedExecutionState::Pausing
206 && state == ExecutionState::Running
207 {
208 return self.finish_pause(record).await.map(ReconcileOutcome::Ready);
209 }
210 outcome_from_record(record, state)
211 }
212 ManagedExecutionState::Resuming => {
213 let (record, state) = self.observe_record(record).await?;
214 if managed_state(&record)? == ManagedExecutionState::Resuming
215 && state == ExecutionState::Paused
216 {
217 return self
218 .finish_resume(record)
219 .await
220 .map(ReconcileOutcome::Ready);
221 }
222 outcome_from_record(record, state)
223 }
224 ManagedExecutionState::Snapshotting => self
225 .recover_snapshot(record)
226 .await
227 .map(ReconcileOutcome::Ready),
228 ManagedExecutionState::Killing => {
229 self.finish_kill(record).await?;
230 Ok(ReconcileOutcome::Failed)
231 }
232 ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => self
233 .resume_restart(record)
234 .await
235 .map(ReconcileOutcome::Ready),
236 _ => {
237 let (record, state) = self.observe_record(record).await?;
238 outcome_from_record(record, state)
239 }
240 }
241 }
242}