a3s_box_runtime/local_execution/
api.rs

1use a3s_box_core::{
2    CreateExecutionRequest, ExecutionEventBatch, ExecutionEventsRequest, ExecutionGeneration,
3    ExecutionId, ExecutionLease, ExecutionManager, ExecutionManagerError, ExecutionManagerResult,
4    ExecutionProcessInventory, ExecutionReservation, ExecutionResourceUpdate, ExecutionSnapshot,
5    ExecutionSnapshotId, ExecutionState, ExecutionStats, ExecutionStatus, KillExecutionOptions,
6    KillOutcome, OperationId, ReconcileOutcome, RestartExecutionOptions,
7};
8use async_trait::async_trait;
9
10use super::support::{managed_state, outcome_from_record, require_generation, state_conflict};
11use super::{
12    build_managed_record, status_from_record, LocalExecutionManager, ManagedExecutionState,
13    RuntimeUpdate,
14};
15
16#[async_trait]
17impl ExecutionManager for LocalExecutionManager {
18    async fn create(
19        &self,
20        request: CreateExecutionRequest,
21        operation_id: &OperationId,
22    ) -> ExecutionManagerResult<ExecutionReservation> {
23        let execution_id = ExecutionId::new(uuid::Uuid::new_v4().to_string())?;
24        let mut record = build_managed_record(
25            &self.home_dir,
26            &execution_id,
27            operation_id.clone(),
28            request,
29            chrono::Utc::now(),
30        )?;
31        let route = self.backend.route_for_create(&record)?;
32        record
33            .managed_execution
34            .as_mut()
35            .ok_or_else(|| {
36                ExecutionManagerError::Internal(format!(
37                    "new execution {execution_id} has no managed lifecycle metadata"
38                ))
39            })?
40            .runtime_route = route;
41        self.backend.preflight(&record).await?;
42        self.backend
43            .plan_create_resources(&record)
44            .await?
45            .apply_to(&mut record)?;
46        let reservation = self.reserve(record).await?;
47        super::record::reservation_from_record(reservation.record())
48    }
49
50    async fn start(
51        &self,
52        execution_id: &ExecutionId,
53        expected_generation: ExecutionGeneration,
54    ) -> ExecutionManagerResult<ExecutionLease> {
55        let _lifecycle_lock =
56            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
57        let record = self
58            .get(execution_id)
59            .await?
60            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
61        super::record::validate_record_health(&record)?;
62        require_generation(&record, execution_id, expected_generation)?;
63        self.ensure_started(record).await
64    }
65
66    async fn inspect(&self, execution_id: &ExecutionId) -> ExecutionManagerResult<ExecutionStatus> {
67        let lifecycle_lock =
68            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
69        let manager = self.clone();
70        let execution_id = execution_id.clone();
71        let execution_id_label = execution_id.clone();
72        tokio::spawn(async move {
73            // Inspection can reconcile provider state, release resources, and
74            // persist a terminal observation. Once its lifecycle lock has been
75            // acquired, finish that projection even if the caller is cancelled;
76            // a replay must wait for this same authoritative operation.
77            let _lifecycle_lock = lifecycle_lock;
78            let record = manager
79                .get(&execution_id)
80                .await?
81                .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
82            let record = manager.stabilize_snapshot(record).await?;
83            let (record, state) = manager.observe_record(record).await?;
84            status_from_record(&record, state)
85        })
86        .await
87        .map_err(|error| {
88            ExecutionManagerError::Internal(format!(
89                "inspection task failed for {execution_id_label}: {error}"
90            ))
91        })?
92    }
93
94    async fn read_logs(
95        &self,
96        execution_id: &ExecutionId,
97        expected_generation: ExecutionGeneration,
98    ) -> ExecutionManagerResult<Vec<a3s_box_core::log::LogEntry>> {
99        self.read_structured_logs(execution_id, expected_generation)
100            .await
101    }
102
103    async fn list_processes(
104        &self,
105        execution_id: &ExecutionId,
106        expected_generation: ExecutionGeneration,
107    ) -> ExecutionManagerResult<ExecutionProcessInventory> {
108        let record = self
109            .require_observable_record(execution_id, expected_generation)
110            .await?;
111        self.require_same_observable_runtime(&record, execution_id, expected_generation)
112            .await?;
113        let inventory = self.backend.list_processes(&record).await?;
114        self.require_same_observable_runtime(&record, execution_id, expected_generation)
115            .await?;
116        if inventory.execution_id != *execution_id || inventory.generation != expected_generation {
117            return Err(ExecutionManagerError::Internal(format!(
118                "backend returned process inventory for a different execution generation than {execution_id} generation {}",
119                expected_generation.get()
120            )));
121        }
122        inventory.validate()?;
123        Ok(inventory)
124    }
125
126    async fn stats(
127        &self,
128        execution_id: &ExecutionId,
129        expected_generation: ExecutionGeneration,
130    ) -> ExecutionManagerResult<ExecutionStats> {
131        let record = self
132            .require_observable_record(execution_id, expected_generation)
133            .await?;
134        self.require_same_observable_runtime(&record, execution_id, expected_generation)
135            .await?;
136        let stats = self.backend.stats(&record).await?;
137        self.require_same_observable_runtime(&record, execution_id, expected_generation)
138            .await?;
139        if stats.execution_id != *execution_id || stats.generation != expected_generation {
140            return Err(ExecutionManagerError::Internal(format!(
141                "backend returned stats for a different execution generation than {execution_id} generation {}",
142                expected_generation.get()
143            )));
144        }
145        stats.validate()?;
146        Ok(stats)
147    }
148
149    async fn events(
150        &self,
151        execution_id: &ExecutionId,
152        expected_generation: ExecutionGeneration,
153        request: ExecutionEventsRequest,
154    ) -> ExecutionManagerResult<ExecutionEventBatch> {
155        request.validate()?;
156        let after_sequence = request.after_sequence;
157        let record = self
158            .require_observable_record(execution_id, expected_generation)
159            .await?;
160        self.require_same_observable_runtime(&record, execution_id, expected_generation)
161            .await?;
162        let batch = self.backend.events(&record, request).await?;
163        self.require_same_observable_runtime(&record, execution_id, expected_generation)
164            .await?;
165        if batch.execution_id != *execution_id || batch.generation != expected_generation {
166            return Err(ExecutionManagerError::Internal(format!(
167                "backend returned events for a different execution generation than {execution_id} generation {}",
168                expected_generation.get()
169            )));
170        }
171        batch.validate_after(after_sequence)?;
172        Ok(batch)
173    }
174
175    async fn update_resources(
176        &self,
177        execution_id: &ExecutionId,
178        expected_generation: ExecutionGeneration,
179        operation_id: &OperationId,
180        update: ExecutionResourceUpdate,
181    ) -> ExecutionManagerResult<ExecutionLease> {
182        update.validate()?;
183        let _lifecycle_lock =
184            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
185        let record = self
186            .get(execution_id)
187            .await?
188            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
189        let record = self.stabilize_snapshot(record).await?;
190        require_generation(&record, execution_id, expected_generation)?;
191        let state = managed_state(&record)?;
192        if let Some(completed) = record
193            .managed_execution
194            .as_ref()
195            .and_then(|metadata| metadata.last_resource_update.as_ref())
196            .filter(|completed| completed.operation_id == *operation_id)
197        {
198            if completed.generation != expected_generation || completed.update != update {
199                return Err(ExecutionManagerError::Conflict {
200                    execution_id: execution_id.clone(),
201                    message: format!(
202                        "resource update operation {operation_id} was already completed with different intent"
203                    ),
204                });
205            }
206            if state != ManagedExecutionState::Running {
207                return Err(state_conflict(
208                    &record,
209                    execution_id,
210                    "replay resource update",
211                ));
212            }
213            return super::record::lease_from_record(&record);
214        }
215        if state == ManagedExecutionState::UpdatingResources {
216            let pending = record
217                .managed_execution
218                .as_ref()
219                .and_then(|metadata| metadata.pending_operation.as_ref());
220            if !matches!(
221                pending,
222                Some(crate::ManagedExecutionOperation::UpdateResources {
223                    operation_id: pending_id,
224                    update: pending_update,
225                }) if pending_id == operation_id && pending_update == &update
226            ) {
227                return Err(ExecutionManagerError::Conflict {
228                    execution_id: execution_id.clone(),
229                    message: "another resource update is already in progress".to_string(),
230                });
231            }
232            return self.finish_resource_update(record).await;
233        }
234        if state != ManagedExecutionState::Running {
235            return Err(state_conflict(&record, execution_id, "update resources"));
236        }
237        self.backend
238            .preflight_resource_update(&record, &update)
239            .await?;
240        let claimed = self
241            .transition(
242                &record,
243                ManagedExecutionState::Running,
244                ManagedExecutionState::UpdatingResources,
245                RuntimeUpdate::ResourceUpdateClaim {
246                    operation_id: operation_id.clone(),
247                    update,
248                },
249            )
250            .await?;
251        self.finish_resource_update(claimed).await
252    }
253
254    async fn create_filesystem_snapshot(
255        &self,
256        execution_id: &ExecutionId,
257        expected_generation: ExecutionGeneration,
258        snapshot_id: &ExecutionSnapshotId,
259    ) -> ExecutionManagerResult<ExecutionSnapshot> {
260        let _lifecycle_lock =
261            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
262        self.create_snapshot(execution_id, expected_generation, snapshot_id)
263            .await
264    }
265
266    async fn filesystem_snapshot_size(
267        &self,
268        snapshot_id: &ExecutionSnapshotId,
269    ) -> ExecutionManagerResult<Option<u64>> {
270        self.snapshot_size(snapshot_id).await
271    }
272
273    async fn delete_filesystem_snapshot(
274        &self,
275        snapshot_id: &ExecutionSnapshotId,
276    ) -> ExecutionManagerResult<bool> {
277        self.delete_snapshot(snapshot_id).await
278    }
279
280    async fn pause(
281        &self,
282        execution_id: &ExecutionId,
283        expected_generation: ExecutionGeneration,
284        keep_memory: bool,
285    ) -> ExecutionManagerResult<ExecutionLease> {
286        let _lifecycle_lock =
287            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
288        let record = self
289            .get(execution_id)
290            .await?
291            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
292        let record = self.stabilize_snapshot(record).await?;
293        require_generation(&record, execution_id, expected_generation)?;
294        if managed_state(&record)? != ManagedExecutionState::Running {
295            return Err(state_conflict(&record, execution_id, "pause"));
296        }
297        let backend_operation_id =
298            OperationId::new(format!("managed-pause-{}", uuid::Uuid::new_v4().simple()))?;
299        let claimed = self
300            .transition(
301                &record,
302                ManagedExecutionState::Running,
303                ManagedExecutionState::Pausing,
304                RuntimeUpdate::PauseClaim {
305                    keep_memory,
306                    operation_id: backend_operation_id,
307                },
308            )
309            .await?;
310        self.finish_pause(claimed).await
311    }
312
313    async fn resume(
314        &self,
315        execution_id: &ExecutionId,
316        expected_generation: ExecutionGeneration,
317    ) -> ExecutionManagerResult<ExecutionLease> {
318        let _lifecycle_lock =
319            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
320        let record = self
321            .get(execution_id)
322            .await?
323            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
324        let record = self.stabilize_snapshot(record).await?;
325        require_generation(&record, execution_id, expected_generation)?;
326        if managed_state(&record)? != ManagedExecutionState::Paused {
327            return Err(state_conflict(&record, execution_id, "resume"));
328        }
329        let backend_operation_id =
330            OperationId::new(format!("managed-resume-{}", uuid::Uuid::new_v4().simple()))?;
331        let claimed = self
332            .transition(
333                &record,
334                ManagedExecutionState::Paused,
335                ManagedExecutionState::Resuming,
336                RuntimeUpdate::ResumeClaim(backend_operation_id),
337            )
338            .await?;
339        self.finish_resume(claimed).await
340    }
341
342    async fn restart_with_options(
343        &self,
344        execution_id: &ExecutionId,
345        expected_generation: ExecutionGeneration,
346        operation_id: &OperationId,
347        options: RestartExecutionOptions,
348    ) -> ExecutionManagerResult<ExecutionLease> {
349        let _lifecycle_lock =
350            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
351        let record = self
352            .get(execution_id)
353            .await?
354            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
355        super::record::validate_record_health(&record)?;
356        self.restart_record(record, expected_generation, operation_id, options)
357            .await
358    }
359
360    async fn kill(
361        &self,
362        execution_id: &ExecutionId,
363        expected_generation: ExecutionGeneration,
364    ) -> ExecutionManagerResult<KillOutcome> {
365        self.kill_with_options(
366            execution_id,
367            expected_generation,
368            KillExecutionOptions::default(),
369        )
370        .await
371    }
372
373    async fn kill_with_options(
374        &self,
375        execution_id: &ExecutionId,
376        expected_generation: ExecutionGeneration,
377        options: KillExecutionOptions,
378    ) -> ExecutionManagerResult<KillOutcome> {
379        validate_kill_options(options)?;
380        let _lifecycle_lock =
381            super::lifecycle_lock::acquire(&self.home_dir, execution_id.as_str()).await?;
382        let record = self
383            .get(execution_id)
384            .await?
385            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
386        let record = self.stabilize_snapshot(record).await?;
387        require_generation(&record, execution_id, expected_generation)?;
388        let state = managed_state(&record)?;
389        if state.is_terminal() {
390            return Ok(KillOutcome::AlreadyStopped);
391        }
392        if matches!(
393            state,
394            ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting
395        ) {
396            return Err(state_conflict(&record, execution_id, "kill"));
397        }
398        let claimed = if state == ManagedExecutionState::Killing {
399            record
400        } else {
401            self.transition(
402                &record,
403                state,
404                ManagedExecutionState::Killing,
405                RuntimeUpdate::KillClaim(options),
406            )
407            .await?
408        };
409        self.finish_kill(claimed).await
410    }
411
412    async fn remove(
413        &self,
414        execution_id: &ExecutionId,
415        expected_generation: ExecutionGeneration,
416    ) -> ExecutionManagerResult<bool> {
417        self.remove_execution(execution_id, expected_generation)
418            .await
419    }
420
421    async fn reconcile(
422        &self,
423        operation_id: &OperationId,
424    ) -> ExecutionManagerResult<ReconcileOutcome> {
425        let Some(initial_record) = self.get_by_operation(operation_id).await? else {
426            return Ok(ReconcileOutcome::Absent);
427        };
428        let _lifecycle_lock =
429            super::lifecycle_lock::acquire(&self.home_dir, &initial_record.id).await?;
430        let Some(record) = self.get_by_operation(operation_id).await? else {
431            return Ok(ReconcileOutcome::Absent);
432        };
433        if record.id != initial_record.id {
434            return Err(ExecutionManagerError::Unavailable(format!(
435                "operation {operation_id} changed execution identity while waiting for its lifecycle lock"
436            )));
437        }
438        super::record::validate_record_health(&record)?;
439        match managed_state(&record)? {
440            ManagedExecutionState::Creating | ManagedExecutionState::Created => Ok(
441                ReconcileOutcome::Created(super::record::reservation_from_record(&record)?),
442            ),
443            ManagedExecutionState::Starting => self.recover_start(record).await,
444            ManagedExecutionState::Pausing => {
445                let (record, state) = self.observe_record(record).await?;
446                if managed_state(&record)? == ManagedExecutionState::Pausing
447                    && state == ExecutionState::Running
448                {
449                    return self.finish_pause(record).await.map(ReconcileOutcome::Ready);
450                }
451                outcome_from_record(record, state)
452            }
453            ManagedExecutionState::Resuming => {
454                let (record, state) = self.observe_record(record).await?;
455                if managed_state(&record)? == ManagedExecutionState::Resuming
456                    && state == ExecutionState::Paused
457                {
458                    return self
459                        .finish_resume(record)
460                        .await
461                        .map(ReconcileOutcome::Ready);
462                }
463                outcome_from_record(record, state)
464            }
465            ManagedExecutionState::UpdatingResources => {
466                let execution_id = super::record::execution_id(&record)?;
467                match self.finish_resource_update(record).await {
468                    Ok(lease) => Ok(ReconcileOutcome::Ready(lease)),
469                    Err(error) => {
470                        let current = self
471                            .get(&execution_id)
472                            .await?
473                            .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
474                        match managed_state(&current)? {
475                            ManagedExecutionState::Stopped | ManagedExecutionState::Failed => {
476                                Ok(ReconcileOutcome::Failed)
477                            }
478                            _ => Err(error),
479                        }
480                    }
481                }
482            }
483            ManagedExecutionState::Snapshotting => self
484                .recover_snapshot(record)
485                .await
486                .map(ReconcileOutcome::Ready),
487            ManagedExecutionState::Killing => {
488                self.finish_kill(record).await?;
489                Ok(ReconcileOutcome::Failed)
490            }
491            ManagedExecutionState::Removing => {
492                self.finish_remove(record).await?;
493                Ok(ReconcileOutcome::Absent)
494            }
495            ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => self
496                .resume_restart(record)
497                .await
498                .map(ReconcileOutcome::Ready),
499            _ => {
500                let (record, state) = self.observe_record(record).await?;
501                outcome_from_record(record, state)
502            }
503        }
504    }
505}
506
507fn validate_kill_options(options: KillExecutionOptions) -> ExecutionManagerResult<()> {
508    if options
509        .signal
510        .is_some_and(|signal| signal <= 0 || 128_i32.checked_add(signal).is_none())
511    {
512        return Err(ExecutionManagerError::InvalidRequest(
513            "kill signal must be positive and representable as a Box exit code".to_string(),
514        ));
515    }
516    if options
517        .timeout_secs
518        .is_some_and(|timeout| timeout.checked_mul(1_000).is_none())
519    {
520        return Err(ExecutionManagerError::InvalidRequest(
521            "kill timeout is too large".to_string(),
522        ));
523    }
524    Ok(())
525}
a3s_box_runtime/local_execution/api.rs

a3s_box_runtime/local_execution/
api.rs
