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

a3s_box_runtime/local_execution/
api.rs
