Skip to main content

a3s_box_runtime/
managed_execution_store.rs

1//! Durable generation-fenced transitions for managed local executions.
2
3use std::path::{Path, PathBuf};
4
5use a3s_box_core::{ExecutionGeneration, ExecutionId, OperationId};
6use thiserror::Error;
7
8use crate::{
9    BoxRecord, BoxStateStore, ManagedExecutionOperation, ManagedExecutionState,
10    ManagedRestartCompletion, ManagedRestartOutcome,
11};
12
13/// Strict durable repository used by the local `ExecutionManager`.
14#[derive(Debug, Clone)]
15pub struct ManagedExecutionStore {
16    path: PathBuf,
17}
18
19/// Result of reserving an idempotent create operation.
20#[derive(Debug, Clone)]
21pub enum ManagedExecutionReservation {
22    /// The creation intent was inserted by this call.
23    Reserved(BoxRecord),
24    /// The operation already existed with the same creation intent.
25    Existing(BoxRecord),
26}
27
28impl ManagedExecutionReservation {
29    pub const fn is_new(&self) -> bool {
30        matches!(self, Self::Reserved(_))
31    }
32
33    pub fn record(&self) -> &BoxRecord {
34        match self {
35            Self::Reserved(record) | Self::Existing(record) => record,
36        }
37    }
38
39    pub fn into_record(self) -> BoxRecord {
40        match self {
41            Self::Reserved(record) | Self::Existing(record) => record,
42        }
43    }
44}
45
46/// Fail-closed errors from managed lifecycle persistence.
47#[derive(Debug, Error)]
48pub enum ManagedExecutionStoreError {
49    #[error("managed execution state I/O failed: {0}")]
50    Io(#[from] std::io::Error),
51    #[error("managed execution not found: {0}")]
52    NotFound(ExecutionId),
53    #[error("execution record is not managed: {0}")]
54    Unmanaged(ExecutionId),
55    #[error("managed execution conflict for {execution_id}: {message}")]
56    Conflict {
57        execution_id: ExecutionId,
58        message: String,
59    },
60    #[error("invalid managed execution record: {0}")]
61    InvalidRecord(String),
62    #[error("invalid managed execution transition for {execution_id}: {from} -> {to}")]
63    InvalidTransition {
64        execution_id: ExecutionId,
65        from: ManagedExecutionState,
66        to: ManagedExecutionState,
67    },
68}
69
70pub type ManagedExecutionStoreResult<T> = std::result::Result<T, ManagedExecutionStoreError>;
71
72impl ManagedExecutionStore {
73    pub fn new(path: impl Into<PathBuf>) -> Self {
74        Self { path: path.into() }
75    }
76
77    pub fn path(&self) -> &Path {
78        &self.path
79    }
80
81    /// Return one managed record without mutating or reconciling state.
82    pub fn get(
83        &self,
84        execution_id: &ExecutionId,
85    ) -> ManagedExecutionStoreResult<Option<BoxRecord>> {
86        let store = BoxStateStore::load_readonly(&self.path)?;
87        let Some(record) = store.find_by_id(execution_id.as_str()).cloned() else {
88            return Ok(None);
89        };
90        if record.managed_execution.is_none() {
91            return Err(ManagedExecutionStoreError::Unmanaged(execution_id.clone()));
92        }
93        Ok(Some(record))
94    }
95
96    /// Return every managed record without reconciling provider state.
97    ///
98    /// Legacy CLI records share the same state file and are deliberately
99    /// excluded. Loading remains strict: one malformed managed record fails the
100    /// complete snapshot instead of letting provider discovery skip corrupt
101    /// ownership metadata.
102    pub fn list(&self) -> ManagedExecutionStoreResult<Vec<BoxRecord>> {
103        let store = BoxStateStore::load_readonly(&self.path)?;
104        Ok(store
105            .records()
106            .iter()
107            .filter(|record| record.managed_execution.is_some())
108            .cloned()
109            .collect())
110    }
111
112    /// Return the record reserved by an idempotent creation operation.
113    pub fn get_by_operation_id(
114        &self,
115        operation_id: &OperationId,
116    ) -> ManagedExecutionStoreResult<Option<BoxRecord>> {
117        let store = BoxStateStore::load_readonly(&self.path)?;
118        Ok(store.find_by_operation_id(operation_id).cloned())
119    }
120
121    /// Atomically reserve one creation operation before backend side effects.
122    ///
123    /// Retrying the same operation with the same full request returns the
124    /// existing record. Reusing an operation ID for different creation intent
125    /// fails without changing durable state.
126    pub fn reserve(
127        &self,
128        mut record: BoxRecord,
129    ) -> ManagedExecutionStoreResult<ManagedExecutionReservation> {
130        let execution_id = validate_new_record(&record)?;
131        record.status = ManagedExecutionState::Created.as_status().to_string();
132        let incoming_metadata = record
133            .managed_execution
134            .as_ref()
135            .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?
136            .clone();
137
138        BoxStateStore::transact(&self.path, move |store| {
139            if let Some(existing) = store
140                .find_by_operation_id(&incoming_metadata.operation_id)
141                .cloned()
142            {
143                let existing_id = ExecutionId::new(existing.id.clone()).map_err(|error| {
144                    ManagedExecutionStoreError::InvalidRecord(error.to_string())
145                })?;
146                let existing_metadata = existing
147                    .managed_execution
148                    .as_ref()
149                    .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(existing_id.clone()))?;
150                if !same_creation_intent(existing_metadata, &incoming_metadata)? {
151                    return Err(ManagedExecutionStoreError::Conflict {
152                        execution_id: existing_id,
153                        message: format!(
154                            "operation {} was already reserved with different creation intent",
155                            incoming_metadata.operation_id
156                        ),
157                    });
158                }
159                return Ok(ManagedExecutionReservation::Existing(existing));
160            }
161
162            if store.find_by_id(execution_id.as_str()).is_some() {
163                return Err(ManagedExecutionStoreError::Conflict {
164                    execution_id,
165                    message: "execution ID is already present".to_string(),
166                });
167            }
168
169            store.records_mut().push(record.clone());
170            Ok(ManagedExecutionReservation::Reserved(record))
171        })
172    }
173
174    /// Claim terminal-record removal before deleting host resources.
175    ///
176    /// The durable `removing` state prevents another lifecycle operation from
177    /// reviving the execution while cleanup is in progress. A retry observes
178    /// the same claim and can resume cleanup after a process crash.
179    pub fn begin_remove(
180        &self,
181        execution_id: &ExecutionId,
182        expected_generation: ExecutionGeneration,
183    ) -> ManagedExecutionStoreResult<Option<BoxRecord>> {
184        let execution_id = execution_id.clone();
185        BoxStateStore::transact(&self.path, move |store| {
186            let Some(record) = store.find_by_id_mut(execution_id.as_str()) else {
187                return Ok(None);
188            };
189            let state = record
190                .managed_state()
191                .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
192                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
193            let metadata = record
194                .managed_execution
195                .as_mut()
196                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
197            if metadata.generation != expected_generation {
198                return Err(ManagedExecutionStoreError::Conflict {
199                    execution_id: execution_id.clone(),
200                    message: format!(
201                        "expected generation {}, found {}",
202                        expected_generation.get(),
203                        metadata.generation.get()
204                    ),
205                });
206            }
207            match state {
208                ManagedExecutionState::Removing => Ok(Some(record.clone())),
209                ManagedExecutionState::Created
210                | ManagedExecutionState::Stopped
211                | ManagedExecutionState::Failed => {
212                    record.status = ManagedExecutionState::Removing.as_status().to_string();
213                    metadata.pending_operation = Some(ManagedExecutionOperation::Remove);
214                    Ok(Some(record.clone()))
215                }
216                _ => Err(ManagedExecutionStoreError::Conflict {
217                    execution_id: execution_id.clone(),
218                    message: format!("cannot remove execution in state {state}"),
219                }),
220            }
221        })
222    }
223
224    /// Forget a generation only after its durable removal claim has completed.
225    pub fn finish_remove(
226        &self,
227        execution_id: &ExecutionId,
228        expected_generation: ExecutionGeneration,
229    ) -> ManagedExecutionStoreResult<bool> {
230        let execution_id = execution_id.clone();
231        BoxStateStore::transact(&self.path, move |store| {
232            let Some(record) = store.find_by_id(execution_id.as_str()) else {
233                return Ok(false);
234            };
235            let state = record
236                .managed_state()
237                .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
238                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
239            let metadata = record
240                .managed_execution
241                .as_ref()
242                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
243            if metadata.generation != expected_generation
244                || state != ManagedExecutionState::Removing
245            {
246                return Err(ManagedExecutionStoreError::Conflict {
247                    execution_id: execution_id.clone(),
248                    message: format!(
249                        "expected removing generation {}, found {state} generation {}",
250                        expected_generation.get(),
251                        metadata.generation.get()
252                    ),
253                });
254            }
255            Ok(store.remove_by_id(execution_id.as_str()))
256        })
257    }
258
259    /// Atomically compare generation and state, then persist one legal edge.
260    ///
261    /// Completing pause and resume, and advancing a restart from teardown to
262    /// startup, increments the runtime generation exactly once. Other edges
263    /// retain the current generation.
264    pub fn transition(
265        &self,
266        execution_id: &ExecutionId,
267        expected_generation: ExecutionGeneration,
268        expected_state: ManagedExecutionState,
269        next_state: ManagedExecutionState,
270    ) -> ManagedExecutionStoreResult<BoxRecord> {
271        self.transition_with(
272            execution_id,
273            expected_generation,
274            expected_state,
275            next_state,
276            |_| {},
277        )
278    }
279
280    /// Persist one legal transition and update runtime evidence in the same
281    /// transaction.
282    pub fn transition_with(
283        &self,
284        execution_id: &ExecutionId,
285        expected_generation: ExecutionGeneration,
286        expected_state: ManagedExecutionState,
287        next_state: ManagedExecutionState,
288        update: impl FnOnce(&mut BoxRecord),
289    ) -> ManagedExecutionStoreResult<BoxRecord> {
290        let execution_id = execution_id.clone();
291        BoxStateStore::transact(&self.path, move |store| {
292            let record = store
293                .find_by_id_mut(execution_id.as_str())
294                .ok_or_else(|| ManagedExecutionStoreError::NotFound(execution_id.clone()))?;
295            let actual_state = record
296                .managed_state()
297                .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
298                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
299            let metadata = record
300                .managed_execution
301                .as_ref()
302                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
303
304            if metadata.generation != expected_generation || actual_state != expected_state {
305                return Err(ManagedExecutionStoreError::Conflict {
306                    execution_id: execution_id.clone(),
307                    message: format!(
308                        "expected {expected_state} generation {}, found {actual_state} generation {}",
309                        expected_generation.get(),
310                        metadata.generation.get()
311                    ),
312                });
313            }
314
315            let next_generation = transition_generation(
316                &execution_id,
317                expected_state,
318                next_state,
319                expected_generation,
320            )?;
321            let original_metadata = record
322                .managed_execution
323                .as_ref()
324                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?
325                .clone();
326            update(record);
327            if record.id != execution_id.as_str() {
328                return Err(ManagedExecutionStoreError::InvalidRecord(format!(
329                    "transition changed execution ID {execution_id}"
330                )));
331            }
332            let updated_metadata = record
333                .managed_execution
334                .as_ref()
335                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
336            if updated_metadata.operation_id != original_metadata.operation_id
337                || !same_creation_intent(updated_metadata, &original_metadata)?
338            {
339                return Err(ManagedExecutionStoreError::InvalidRecord(format!(
340                    "transition changed creation identity for {execution_id}"
341                )));
342            }
343            record.status = next_state.as_status().to_string();
344            let metadata = record
345                .managed_execution
346                .as_mut()
347                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
348            metadata.generation = next_generation;
349            if expected_state == ManagedExecutionState::RestartStarting
350                && matches!(
351                    next_state,
352                    ManagedExecutionState::Running | ManagedExecutionState::Failed
353                )
354            {
355                let Some(ManagedExecutionOperation::Restart {
356                    operation_id,
357                    source_generation,
358                    stop_timeout_secs,
359                    ..
360                }) = metadata.pending_operation.as_ref()
361                else {
362                    return Err(ManagedExecutionStoreError::InvalidRecord(format!(
363                        "restart completion for {execution_id} has no persisted restart intent"
364                    )));
365                };
366                metadata.last_restart = Some(ManagedRestartCompletion {
367                    operation_id: operation_id.clone(),
368                    source_generation: *source_generation,
369                    target_generation: next_generation,
370                    outcome: if next_state == ManagedExecutionState::Running {
371                        ManagedRestartOutcome::Running
372                    } else {
373                        ManagedRestartOutcome::Failed
374                    },
375                    stop_timeout_secs: *stop_timeout_secs,
376                });
377            }
378            metadata.pending_operation = match next_state {
379                ManagedExecutionState::Starting => Some(ManagedExecutionOperation::Start),
380                ManagedExecutionState::Pausing => match metadata.pending_operation.take() {
381                    Some(operation @ ManagedExecutionOperation::Pause { .. }) => Some(operation),
382                    _ => Some(ManagedExecutionOperation::Pause { keep_memory: false }),
383                },
384                ManagedExecutionState::Resuming => Some(ManagedExecutionOperation::Resume),
385                ManagedExecutionState::Snapshotting => match metadata.pending_operation.take() {
386                    Some(operation @ ManagedExecutionOperation::Snapshot { .. }) => Some(operation),
387                    _ => {
388                        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
389                        "snapshot transition for {execution_id} has no persisted snapshot intent"
390                    )))
391                    }
392                },
393                ManagedExecutionState::Killing => match metadata.pending_operation.take() {
394                    Some(operation @ ManagedExecutionOperation::Kill { .. }) => Some(operation),
395                    _ => Some(ManagedExecutionOperation::Kill {
396                        signal: None,
397                        timeout_secs: None,
398                    }),
399                },
400                ManagedExecutionState::Removing => Some(ManagedExecutionOperation::Remove),
401                ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => {
402                    match metadata.pending_operation.take() {
403                        Some(operation @ ManagedExecutionOperation::Restart { .. }) => {
404                            Some(operation)
405                        }
406                        _ => {
407                            return Err(ManagedExecutionStoreError::InvalidRecord(format!(
408                            "restart transition for {execution_id} has no persisted restart intent"
409                        )))
410                        }
411                    }
412                }
413                ManagedExecutionState::Creating
414                | ManagedExecutionState::Created
415                | ManagedExecutionState::Running
416                | ManagedExecutionState::Paused
417                | ManagedExecutionState::Stopped
418                | ManagedExecutionState::Failed => None,
419            };
420            Ok(record.clone())
421        })
422    }
423}
424
425fn validate_new_record(record: &BoxRecord) -> ManagedExecutionStoreResult<ExecutionId> {
426    let execution_id = ExecutionId::new(record.id.clone())
427        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
428    let metadata = record
429        .managed_execution
430        .as_ref()
431        .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
432    metadata
433        .validate()
434        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
435    let state = record
436        .managed_state()
437        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
438        .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
439    if state != ManagedExecutionState::Created {
440        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
441            "new execution {execution_id} must be created, found {state}"
442        )));
443    }
444    if metadata.generation != ExecutionGeneration::INITIAL {
445        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
446            "new execution {execution_id} must start at generation {}",
447            ExecutionGeneration::INITIAL.get()
448        )));
449    }
450    Ok(execution_id)
451}
452
453fn same_creation_intent(
454    left: &crate::ManagedExecutionMetadata,
455    right: &crate::ManagedExecutionMetadata,
456) -> ManagedExecutionStoreResult<bool> {
457    let left_request = serde_json::to_value(&left.request)
458        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
459    let right_request = serde_json::to_value(&right.request)
460        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
461    Ok(left_request == right_request && left.plan == right.plan)
462}
463
464fn transition_generation(
465    execution_id: &ExecutionId,
466    from: ManagedExecutionState,
467    to: ManagedExecutionState,
468    current: ExecutionGeneration,
469) -> ManagedExecutionStoreResult<ExecutionGeneration> {
470    use ManagedExecutionState::{
471        Created, Creating, Failed, Killing, Paused, Pausing, RestartStarting, RestartStopping,
472        Resuming, Running, Snapshotting, Starting, Stopped,
473    };
474
475    let legal = matches!(
476        (from, to),
477        (Creating, Created | Starting | Killing | Stopped | Failed)
478            | (
479                Created,
480                Starting | Killing | RestartStopping | Stopped | Failed
481            )
482            | (
483                Starting,
484                Created | Creating | Running | Killing | Stopped | Failed
485            )
486            | (
487                Running,
488                Pausing | Snapshotting | Killing | RestartStopping | Stopped | Failed
489            )
490            | (Pausing, Paused | Running | Killing | Stopped | Failed)
491            | (
492                Paused,
493                Resuming | Snapshotting | Killing | RestartStopping | Stopped | Failed
494            )
495            | (Resuming, Running | Paused | Killing | Stopped | Failed)
496            | (Snapshotting, Running | Paused | Stopped | Failed)
497            | (Killing, Stopped | Failed)
498            | (Stopped | Failed, RestartStopping)
499            | (RestartStopping, RestartStarting)
500            | (RestartStarting, Running | Failed)
501    );
502    if !legal {
503        return Err(ManagedExecutionStoreError::InvalidTransition {
504            execution_id: execution_id.clone(),
505            from,
506            to,
507        });
508    }
509
510    if matches!(
511        (from, to),
512        (Pausing, Paused) | (Resuming, Running) | (RestartStopping, RestartStarting)
513    ) {
514        let value = current.get().checked_add(1).ok_or_else(|| {
515            ManagedExecutionStoreError::InvalidRecord(format!(
516                "execution {execution_id} generation is exhausted"
517            ))
518        })?;
519        return ExecutionGeneration::new(value)
520            .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()));
521    }
522    Ok(current)
523}
524
525#[cfg(test)]
526mod tests {
527    use std::sync::{Arc, Barrier};
528
529    use a3s_box_core::{CreateExecutionRequest, ExecutionIsolation, ExecutionSnapshotId};
530
531    use super::*;
532    use crate::ManagedExecutionMetadata;
533
534    fn managed_record(id: &str, operation: &str) -> BoxRecord {
535        let mut record: BoxRecord = serde_json::from_value(serde_json::json!({
536            "id": id,
537            "short_id": BoxRecord::make_short_id(id),
538            "name": format!("box-{id}"),
539            "image": "alpine:latest",
540            "isolation": "sandbox",
541            "status": "created",
542            "pid": null,
543            "cpus": 1,
544            "memory_mb": 128,
545            "volumes": [],
546            "env": {},
547            "cmd": ["sh"],
548            "box_dir": format!("/tmp/{id}"),
549            "console_log": format!("/tmp/{id}/console.log"),
550            "created_at": "2026-07-14T12:00:00Z",
551            "started_at": null,
552            "auto_remove": false
553        }))
554        .unwrap();
555        let config = a3s_box_core::BoxConfig {
556            image: "alpine:latest".to_string(),
557            isolation: ExecutionIsolation::Sandbox,
558            ..Default::default()
559        };
560        record.managed_execution = Some(
561            ManagedExecutionMetadata::new(
562                OperationId::new(operation).unwrap(),
563                ExecutionGeneration::INITIAL,
564                CreateExecutionRequest {
565                    external_sandbox_id: "sandbox-1".to_string(),
566                    config,
567                    labels: Default::default(),
568                    policy: Default::default(),
569                    rootfs_snapshot_id: None,
570                },
571            )
572            .unwrap(),
573        );
574        record
575    }
576
577    #[test]
578    fn reservation_is_idempotent_for_the_same_full_request() {
579        let directory = tempfile::tempdir().unwrap();
580        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
581
582        let first = store.reserve(managed_record("execution-1", "operation-1"));
583        let retry = store.reserve(managed_record("execution-2", "operation-1"));
584
585        assert!(first.unwrap().is_new());
586        let retry = retry.unwrap();
587        assert!(!retry.is_new());
588        assert_eq!(retry.record().id, "execution-1");
589        assert_eq!(
590            BoxStateStore::load(store.path()).unwrap().records().len(),
591            1
592        );
593    }
594
595    #[test]
596    fn reservation_rejects_operation_reuse_with_different_intent() {
597        let directory = tempfile::tempdir().unwrap();
598        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
599        store
600            .reserve(managed_record("execution-1", "operation-1"))
601            .unwrap();
602        let mut conflicting = managed_record("execution-2", "operation-1");
603        conflicting
604            .managed_execution
605            .as_mut()
606            .unwrap()
607            .request
608            .external_sandbox_id = "sandbox-2".to_string();
609
610        let error = store.reserve(conflicting).unwrap_err();
611
612        assert!(matches!(error, ManagedExecutionStoreError::Conflict { .. }));
613        assert_eq!(
614            BoxStateStore::load(store.path()).unwrap().records().len(),
615            1
616        );
617    }
618
619    #[test]
620    fn removal_claim_is_generation_fenced_durable_and_idempotent() {
621        let directory = tempfile::tempdir().unwrap();
622        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
623        let id = ExecutionId::new("execution-1").unwrap();
624        store
625            .reserve(managed_record(id.as_str(), "operation-1"))
626            .unwrap();
627
628        let claimed = store
629            .begin_remove(&id, ExecutionGeneration::INITIAL)
630            .unwrap()
631            .unwrap();
632        assert_eq!(
633            claimed.managed_state().unwrap(),
634            Some(ManagedExecutionState::Removing)
635        );
636        assert!(matches!(
637            claimed
638                .managed_execution
639                .as_ref()
640                .unwrap()
641                .pending_operation,
642            Some(ManagedExecutionOperation::Remove)
643        ));
644
645        let reopened = ManagedExecutionStore::new(store.path().to_path_buf());
646        assert_eq!(
647            reopened
648                .begin_remove(&id, ExecutionGeneration::INITIAL)
649                .unwrap()
650                .unwrap()
651                .managed_state()
652                .unwrap(),
653            Some(ManagedExecutionState::Removing)
654        );
655        assert!(matches!(
656            reopened.begin_remove(&id, ExecutionGeneration::new(2).unwrap()),
657            Err(ManagedExecutionStoreError::Conflict { .. })
658        ));
659
660        assert!(reopened
661            .finish_remove(&id, ExecutionGeneration::INITIAL)
662            .unwrap());
663        assert!(!reopened
664            .finish_remove(&id, ExecutionGeneration::INITIAL)
665            .unwrap());
666        assert!(reopened.get(&id).unwrap().is_none());
667    }
668
669    #[test]
670    fn pause_and_resume_completion_advance_generation_once() {
671        let directory = tempfile::tempdir().unwrap();
672        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
673        let id = ExecutionId::new("execution-1").unwrap();
674        store
675            .reserve(managed_record(id.as_str(), "operation-1"))
676            .unwrap();
677
678        store
679            .transition(
680                &id,
681                ExecutionGeneration::INITIAL,
682                ManagedExecutionState::Created,
683                ManagedExecutionState::Starting,
684            )
685            .unwrap();
686        let running = store
687            .transition(
688                &id,
689                ExecutionGeneration::INITIAL,
690                ManagedExecutionState::Starting,
691                ManagedExecutionState::Running,
692            )
693            .unwrap();
694        assert_eq!(
695            running.managed_execution.unwrap().generation,
696            ExecutionGeneration::INITIAL
697        );
698        store
699            .transition(
700                &id,
701                ExecutionGeneration::INITIAL,
702                ManagedExecutionState::Running,
703                ManagedExecutionState::Pausing,
704            )
705            .unwrap();
706        let paused = store
707            .transition(
708                &id,
709                ExecutionGeneration::INITIAL,
710                ManagedExecutionState::Pausing,
711                ManagedExecutionState::Paused,
712            )
713            .unwrap();
714        let generation_two = ExecutionGeneration::new(2).unwrap();
715        assert_eq!(paused.managed_execution.unwrap().generation, generation_two);
716        store
717            .transition(
718                &id,
719                generation_two,
720                ManagedExecutionState::Paused,
721                ManagedExecutionState::Resuming,
722            )
723            .unwrap();
724        let resumed = store
725            .transition(
726                &id,
727                generation_two,
728                ManagedExecutionState::Resuming,
729                ManagedExecutionState::Running,
730            )
731            .unwrap();
732        assert_eq!(
733            resumed.managed_execution.unwrap().generation,
734            ExecutionGeneration::new(3).unwrap()
735        );
736    }
737
738    #[test]
739    fn snapshot_intent_is_durable_and_preserves_runtime_generation() {
740        let directory = tempfile::tempdir().unwrap();
741        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
742        let id = ExecutionId::new("execution-1").unwrap();
743        store
744            .reserve(managed_record(id.as_str(), "operation-1"))
745            .unwrap();
746        store
747            .transition(
748                &id,
749                ExecutionGeneration::INITIAL,
750                ManagedExecutionState::Created,
751                ManagedExecutionState::Starting,
752            )
753            .unwrap();
754        store
755            .transition(
756                &id,
757                ExecutionGeneration::INITIAL,
758                ManagedExecutionState::Starting,
759                ManagedExecutionState::Running,
760            )
761            .unwrap();
762        let snapshot_id = ExecutionSnapshotId::new("snapshot-1").unwrap();
763        let claimed = store
764            .transition_with(
765                &id,
766                ExecutionGeneration::INITIAL,
767                ManagedExecutionState::Running,
768                ManagedExecutionState::Snapshotting,
769                |record| {
770                    record.managed_execution.as_mut().unwrap().pending_operation =
771                        Some(ManagedExecutionOperation::Snapshot {
772                            snapshot_id: snapshot_id.clone(),
773                            source_state: ManagedExecutionState::Running,
774                        });
775                },
776            )
777            .unwrap();
778        assert_eq!(
779            claimed.managed_execution.as_ref().unwrap().generation,
780            ExecutionGeneration::INITIAL
781        );
782        assert!(matches!(
783            claimed
784                .managed_execution
785                .as_ref()
786                .unwrap()
787                .pending_operation
788                .as_ref(),
789            Some(ManagedExecutionOperation::Snapshot {
790                snapshot_id,
791                source_state: ManagedExecutionState::Running,
792            }) if snapshot_id.as_str() == "snapshot-1"
793        ));
794
795        let reopened = ManagedExecutionStore::new(store.path().to_path_buf());
796        let persisted = reopened.get(&id).unwrap().unwrap();
797        assert_eq!(
798            persisted.managed_state().unwrap(),
799            Some(ManagedExecutionState::Snapshotting)
800        );
801        let completed = reopened
802            .transition(
803                &id,
804                ExecutionGeneration::INITIAL,
805                ManagedExecutionState::Snapshotting,
806                ManagedExecutionState::Running,
807            )
808            .unwrap();
809        let metadata = completed.managed_execution.unwrap();
810        assert_eq!(metadata.generation, ExecutionGeneration::INITIAL);
811        assert!(metadata.pending_operation.is_none());
812        assert!(matches!(
813            reopened.transition(
814                &id,
815                ExecutionGeneration::INITIAL,
816                ManagedExecutionState::Running,
817                ManagedExecutionState::Snapshotting,
818            ),
819            Err(ManagedExecutionStoreError::InvalidRecord(_))
820        ));
821    }
822
823    #[test]
824    fn restart_advances_generation_between_durable_teardown_and_startup() {
825        let directory = tempfile::tempdir().unwrap();
826        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
827        let id = ExecutionId::new("execution-1").unwrap();
828        store
829            .reserve(managed_record(id.as_str(), "operation-create"))
830            .unwrap();
831        store
832            .transition(
833                &id,
834                ExecutionGeneration::INITIAL,
835                ManagedExecutionState::Created,
836                ManagedExecutionState::Starting,
837            )
838            .unwrap();
839        store
840            .transition(
841                &id,
842                ExecutionGeneration::INITIAL,
843                ManagedExecutionState::Starting,
844                ManagedExecutionState::Running,
845            )
846            .unwrap();
847        let restart_operation = OperationId::new("operation-restart").unwrap();
848        let stopping = store
849            .transition_with(
850                &id,
851                ExecutionGeneration::INITIAL,
852                ManagedExecutionState::Running,
853                ManagedExecutionState::RestartStopping,
854                |record| {
855                    record.managed_execution.as_mut().unwrap().pending_operation =
856                        Some(ManagedExecutionOperation::Restart {
857                            operation_id: restart_operation.clone(),
858                            source_generation: ExecutionGeneration::INITIAL,
859                            source_state: ManagedExecutionState::Running,
860                            stop_timeout_secs: Some(10),
861                        });
862                },
863            )
864            .unwrap();
865        assert_eq!(
866            stopping.managed_execution.as_ref().unwrap().generation,
867            ExecutionGeneration::INITIAL
868        );
869
870        let starting = store
871            .transition(
872                &id,
873                ExecutionGeneration::INITIAL,
874                ManagedExecutionState::RestartStopping,
875                ManagedExecutionState::RestartStarting,
876            )
877            .unwrap();
878        let generation_two = ExecutionGeneration::new(2).unwrap();
879        assert_eq!(
880            starting.managed_execution.as_ref().unwrap().generation,
881            generation_two
882        );
883        let running = store
884            .transition(
885                &id,
886                generation_two,
887                ManagedExecutionState::RestartStarting,
888                ManagedExecutionState::Running,
889            )
890            .unwrap();
891        let metadata = running.managed_execution.unwrap();
892        assert_eq!(metadata.generation, generation_two);
893        assert!(metadata.pending_operation.is_none());
894        let completed = metadata.last_restart.unwrap();
895        assert_eq!(completed.operation_id, restart_operation);
896        assert_eq!(completed.source_generation, ExecutionGeneration::INITIAL);
897        assert_eq!(completed.target_generation, generation_two);
898        assert_eq!(completed.outcome, ManagedRestartOutcome::Running);
899        assert_eq!(completed.stop_timeout_secs, Some(10));
900    }
901
902    #[test]
903    fn stale_generation_and_invalid_edges_do_not_change_disk() {
904        let directory = tempfile::tempdir().unwrap();
905        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
906        let id = ExecutionId::new("execution-1").unwrap();
907        store
908            .reserve(managed_record(id.as_str(), "operation-1"))
909            .unwrap();
910        store
911            .transition(
912                &id,
913                ExecutionGeneration::INITIAL,
914                ManagedExecutionState::Created,
915                ManagedExecutionState::Starting,
916            )
917            .unwrap();
918        store
919            .transition(
920                &id,
921                ExecutionGeneration::INITIAL,
922                ManagedExecutionState::Starting,
923                ManagedExecutionState::Running,
924            )
925            .unwrap();
926
927        let stale = store.transition(
928            &id,
929            ExecutionGeneration::new(2).unwrap(),
930            ManagedExecutionState::Running,
931            ManagedExecutionState::Pausing,
932        );
933        let invalid = store.transition(
934            &id,
935            ExecutionGeneration::INITIAL,
936            ManagedExecutionState::Running,
937            ManagedExecutionState::Paused,
938        );
939
940        assert!(matches!(
941            stale,
942            Err(ManagedExecutionStoreError::Conflict { .. })
943        ));
944        assert!(matches!(
945            invalid,
946            Err(ManagedExecutionStoreError::InvalidTransition { .. })
947        ));
948        let persisted = store.get(&id).unwrap().unwrap();
949        assert_eq!(
950            persisted.managed_state().unwrap(),
951            Some(ManagedExecutionState::Running)
952        );
953        assert_eq!(
954            persisted.managed_execution.unwrap().generation,
955            ExecutionGeneration::INITIAL
956        );
957    }
958
959    #[cfg(unix)]
960    #[test]
961    fn concurrent_claims_have_one_winner() {
962        let directory = tempfile::tempdir().unwrap();
963        let store = Arc::new(ManagedExecutionStore::new(
964            directory.path().join("boxes.json"),
965        ));
966        let id = ExecutionId::new("execution-1").unwrap();
967        store
968            .reserve(managed_record(id.as_str(), "operation-1"))
969            .unwrap();
970        store
971            .transition(
972                &id,
973                ExecutionGeneration::INITIAL,
974                ManagedExecutionState::Created,
975                ManagedExecutionState::Starting,
976            )
977            .unwrap();
978        store
979            .transition(
980                &id,
981                ExecutionGeneration::INITIAL,
982                ManagedExecutionState::Starting,
983                ManagedExecutionState::Running,
984            )
985            .unwrap();
986        let barrier = Arc::new(Barrier::new(3));
987        let handles: Vec<_> = (0..2)
988            .map(|_| {
989                let store = Arc::clone(&store);
990                let id = id.clone();
991                let barrier = Arc::clone(&barrier);
992                std::thread::spawn(move || {
993                    barrier.wait();
994                    store.transition(
995                        &id,
996                        ExecutionGeneration::INITIAL,
997                        ManagedExecutionState::Running,
998                        ManagedExecutionState::Pausing,
999                    )
1000                })
1001            })
1002            .collect();
1003        barrier.wait();
1004        let results: Vec<_> = handles
1005            .into_iter()
1006            .map(|handle| handle.join().unwrap())
1007            .collect();
1008
1009        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
1010        assert_eq!(
1011            results
1012                .iter()
1013                .filter(|result| matches!(result, Err(ManagedExecutionStoreError::Conflict { .. })))
1014                .count(),
1015            1
1016        );
1017        assert_eq!(
1018            store.get(&id).unwrap().unwrap().managed_state().unwrap(),
1019            Some(ManagedExecutionState::Pausing)
1020        );
1021    }
1022}