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 the record reserved by an idempotent creation operation.
97    pub fn get_by_operation_id(
98        &self,
99        operation_id: &OperationId,
100    ) -> ManagedExecutionStoreResult<Option<BoxRecord>> {
101        let store = BoxStateStore::load_readonly(&self.path)?;
102        Ok(store.find_by_operation_id(operation_id).cloned())
103    }
104
105    /// Atomically reserve one creation operation before backend side effects.
106    ///
107    /// Retrying the same operation with the same full request returns the
108    /// existing record. Reusing an operation ID for different creation intent
109    /// fails without changing durable state.
110    pub fn reserve(
111        &self,
112        mut record: BoxRecord,
113    ) -> ManagedExecutionStoreResult<ManagedExecutionReservation> {
114        let execution_id = validate_new_record(&record)?;
115        record.status = ManagedExecutionState::Created.as_status().to_string();
116        let incoming_metadata = record
117            .managed_execution
118            .as_ref()
119            .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?
120            .clone();
121
122        BoxStateStore::transact(&self.path, move |store| {
123            if let Some(existing) = store
124                .find_by_operation_id(&incoming_metadata.operation_id)
125                .cloned()
126            {
127                let existing_id = ExecutionId::new(existing.id.clone()).map_err(|error| {
128                    ManagedExecutionStoreError::InvalidRecord(error.to_string())
129                })?;
130                let existing_metadata = existing
131                    .managed_execution
132                    .as_ref()
133                    .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(existing_id.clone()))?;
134                if !same_creation_intent(existing_metadata, &incoming_metadata)? {
135                    return Err(ManagedExecutionStoreError::Conflict {
136                        execution_id: existing_id,
137                        message: format!(
138                            "operation {} was already reserved with different creation intent",
139                            incoming_metadata.operation_id
140                        ),
141                    });
142                }
143                return Ok(ManagedExecutionReservation::Existing(existing));
144            }
145
146            if store.find_by_id(execution_id.as_str()).is_some() {
147                return Err(ManagedExecutionStoreError::Conflict {
148                    execution_id,
149                    message: "execution ID is already present".to_string(),
150                });
151            }
152
153            store.records_mut().push(record.clone());
154            Ok(ManagedExecutionReservation::Reserved(record))
155        })
156    }
157
158    /// Atomically compare generation and state, then persist one legal edge.
159    ///
160    /// Completing pause and resume, and advancing a restart from teardown to
161    /// startup, increments the runtime generation exactly once. Other edges
162    /// retain the current generation.
163    pub fn transition(
164        &self,
165        execution_id: &ExecutionId,
166        expected_generation: ExecutionGeneration,
167        expected_state: ManagedExecutionState,
168        next_state: ManagedExecutionState,
169    ) -> ManagedExecutionStoreResult<BoxRecord> {
170        self.transition_with(
171            execution_id,
172            expected_generation,
173            expected_state,
174            next_state,
175            |_| {},
176        )
177    }
178
179    /// Persist one legal transition and update runtime evidence in the same
180    /// transaction.
181    pub fn transition_with(
182        &self,
183        execution_id: &ExecutionId,
184        expected_generation: ExecutionGeneration,
185        expected_state: ManagedExecutionState,
186        next_state: ManagedExecutionState,
187        update: impl FnOnce(&mut BoxRecord),
188    ) -> ManagedExecutionStoreResult<BoxRecord> {
189        let execution_id = execution_id.clone();
190        BoxStateStore::transact(&self.path, move |store| {
191            let record = store
192                .find_by_id_mut(execution_id.as_str())
193                .ok_or_else(|| ManagedExecutionStoreError::NotFound(execution_id.clone()))?;
194            let actual_state = record
195                .managed_state()
196                .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
197                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
198            let metadata = record
199                .managed_execution
200                .as_ref()
201                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
202
203            if metadata.generation != expected_generation || actual_state != expected_state {
204                return Err(ManagedExecutionStoreError::Conflict {
205                    execution_id: execution_id.clone(),
206                    message: format!(
207                        "expected {expected_state} generation {}, found {actual_state} generation {}",
208                        expected_generation.get(),
209                        metadata.generation.get()
210                    ),
211                });
212            }
213
214            let next_generation = transition_generation(
215                &execution_id,
216                expected_state,
217                next_state,
218                expected_generation,
219            )?;
220            let original_metadata = record
221                .managed_execution
222                .as_ref()
223                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?
224                .clone();
225            update(record);
226            if record.id != execution_id.as_str() {
227                return Err(ManagedExecutionStoreError::InvalidRecord(format!(
228                    "transition changed execution ID {execution_id}"
229                )));
230            }
231            let updated_metadata = record
232                .managed_execution
233                .as_ref()
234                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
235            if updated_metadata.operation_id != original_metadata.operation_id
236                || !same_creation_intent(updated_metadata, &original_metadata)?
237            {
238                return Err(ManagedExecutionStoreError::InvalidRecord(format!(
239                    "transition changed creation identity for {execution_id}"
240                )));
241            }
242            record.status = next_state.as_status().to_string();
243            let metadata = record
244                .managed_execution
245                .as_mut()
246                .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
247            metadata.generation = next_generation;
248            if expected_state == ManagedExecutionState::RestartStarting
249                && matches!(
250                    next_state,
251                    ManagedExecutionState::Running | ManagedExecutionState::Failed
252                )
253            {
254                let Some(ManagedExecutionOperation::Restart {
255                    operation_id,
256                    source_generation,
257                    stop_timeout_secs,
258                    ..
259                }) = metadata.pending_operation.as_ref()
260                else {
261                    return Err(ManagedExecutionStoreError::InvalidRecord(format!(
262                        "restart completion for {execution_id} has no persisted restart intent"
263                    )));
264                };
265                metadata.last_restart = Some(ManagedRestartCompletion {
266                    operation_id: operation_id.clone(),
267                    source_generation: *source_generation,
268                    target_generation: next_generation,
269                    outcome: if next_state == ManagedExecutionState::Running {
270                        ManagedRestartOutcome::Running
271                    } else {
272                        ManagedRestartOutcome::Failed
273                    },
274                    stop_timeout_secs: *stop_timeout_secs,
275                });
276            }
277            metadata.pending_operation = match next_state {
278                ManagedExecutionState::Starting => Some(ManagedExecutionOperation::Start),
279                ManagedExecutionState::Pausing => match metadata.pending_operation.take() {
280                    Some(operation @ ManagedExecutionOperation::Pause { .. }) => Some(operation),
281                    _ => Some(ManagedExecutionOperation::Pause { keep_memory: false }),
282                },
283                ManagedExecutionState::Resuming => Some(ManagedExecutionOperation::Resume),
284                ManagedExecutionState::Snapshotting => match metadata.pending_operation.take() {
285                    Some(operation @ ManagedExecutionOperation::Snapshot { .. }) => Some(operation),
286                    _ => {
287                        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
288                        "snapshot transition for {execution_id} has no persisted snapshot intent"
289                    )))
290                    }
291                },
292                ManagedExecutionState::Killing => Some(ManagedExecutionOperation::Kill),
293                ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => {
294                    match metadata.pending_operation.take() {
295                        Some(operation @ ManagedExecutionOperation::Restart { .. }) => {
296                            Some(operation)
297                        }
298                        _ => {
299                            return Err(ManagedExecutionStoreError::InvalidRecord(format!(
300                            "restart transition for {execution_id} has no persisted restart intent"
301                        )))
302                        }
303                    }
304                }
305                ManagedExecutionState::Creating
306                | ManagedExecutionState::Created
307                | ManagedExecutionState::Running
308                | ManagedExecutionState::Paused
309                | ManagedExecutionState::Stopped
310                | ManagedExecutionState::Failed => None,
311            };
312            Ok(record.clone())
313        })
314    }
315}
316
317fn validate_new_record(record: &BoxRecord) -> ManagedExecutionStoreResult<ExecutionId> {
318    let execution_id = ExecutionId::new(record.id.clone())
319        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
320    let metadata = record
321        .managed_execution
322        .as_ref()
323        .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
324    metadata
325        .validate()
326        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
327    let state = record
328        .managed_state()
329        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?
330        .ok_or_else(|| ManagedExecutionStoreError::Unmanaged(execution_id.clone()))?;
331    if state != ManagedExecutionState::Created {
332        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
333            "new execution {execution_id} must be created, found {state}"
334        )));
335    }
336    if metadata.generation != ExecutionGeneration::INITIAL {
337        return Err(ManagedExecutionStoreError::InvalidRecord(format!(
338            "new execution {execution_id} must start at generation {}",
339            ExecutionGeneration::INITIAL.get()
340        )));
341    }
342    Ok(execution_id)
343}
344
345fn same_creation_intent(
346    left: &crate::ManagedExecutionMetadata,
347    right: &crate::ManagedExecutionMetadata,
348) -> ManagedExecutionStoreResult<bool> {
349    let left_request = serde_json::to_value(&left.request)
350        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
351    let right_request = serde_json::to_value(&right.request)
352        .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()))?;
353    Ok(left_request == right_request && left.plan == right.plan)
354}
355
356fn transition_generation(
357    execution_id: &ExecutionId,
358    from: ManagedExecutionState,
359    to: ManagedExecutionState,
360    current: ExecutionGeneration,
361) -> ManagedExecutionStoreResult<ExecutionGeneration> {
362    use ManagedExecutionState::{
363        Created, Creating, Failed, Killing, Paused, Pausing, RestartStarting, RestartStopping,
364        Resuming, Running, Snapshotting, Starting, Stopped,
365    };
366
367    let legal = matches!(
368        (from, to),
369        (Creating, Created | Starting | Killing | Stopped | Failed)
370            | (
371                Created,
372                Starting | Killing | RestartStopping | Stopped | Failed
373            )
374            | (
375                Starting,
376                Created | Creating | Running | Killing | Stopped | Failed
377            )
378            | (
379                Running,
380                Pausing | Snapshotting | Killing | RestartStopping | Stopped | Failed
381            )
382            | (Pausing, Paused | Running | Killing | Stopped | Failed)
383            | (
384                Paused,
385                Resuming | Snapshotting | Killing | RestartStopping | Stopped | Failed
386            )
387            | (Resuming, Running | Paused | Killing | Stopped | Failed)
388            | (Snapshotting, Running | Paused | Stopped | Failed)
389            | (Killing, Stopped | Failed)
390            | (Stopped | Failed, RestartStopping)
391            | (RestartStopping, RestartStarting)
392            | (RestartStarting, Running | Failed)
393    );
394    if !legal {
395        return Err(ManagedExecutionStoreError::InvalidTransition {
396            execution_id: execution_id.clone(),
397            from,
398            to,
399        });
400    }
401
402    if matches!(
403        (from, to),
404        (Pausing, Paused) | (Resuming, Running) | (RestartStopping, RestartStarting)
405    ) {
406        let value = current.get().checked_add(1).ok_or_else(|| {
407            ManagedExecutionStoreError::InvalidRecord(format!(
408                "execution {execution_id} generation is exhausted"
409            ))
410        })?;
411        return ExecutionGeneration::new(value)
412            .map_err(|error| ManagedExecutionStoreError::InvalidRecord(error.to_string()));
413    }
414    Ok(current)
415}
416
417#[cfg(test)]
418mod tests {
419    use std::sync::{Arc, Barrier};
420
421    use a3s_box_core::{CreateExecutionRequest, ExecutionIsolation, ExecutionSnapshotId};
422
423    use super::*;
424    use crate::ManagedExecutionMetadata;
425
426    fn managed_record(id: &str, operation: &str) -> BoxRecord {
427        let mut record: BoxRecord = serde_json::from_value(serde_json::json!({
428            "id": id,
429            "short_id": BoxRecord::make_short_id(id),
430            "name": format!("box-{id}"),
431            "image": "alpine:latest",
432            "isolation": "sandbox",
433            "status": "created",
434            "pid": null,
435            "cpus": 1,
436            "memory_mb": 128,
437            "volumes": [],
438            "env": {},
439            "cmd": ["sh"],
440            "box_dir": format!("/tmp/{id}"),
441            "console_log": format!("/tmp/{id}/console.log"),
442            "created_at": "2026-07-14T12:00:00Z",
443            "started_at": null,
444            "auto_remove": false
445        }))
446        .unwrap();
447        let config = a3s_box_core::BoxConfig {
448            image: "alpine:latest".to_string(),
449            isolation: ExecutionIsolation::Sandbox,
450            ..Default::default()
451        };
452        record.managed_execution = Some(
453            ManagedExecutionMetadata::new(
454                OperationId::new(operation).unwrap(),
455                ExecutionGeneration::INITIAL,
456                CreateExecutionRequest {
457                    external_sandbox_id: "sandbox-1".to_string(),
458                    config,
459                    labels: Default::default(),
460                    policy: Default::default(),
461                    rootfs_snapshot_id: None,
462                },
463            )
464            .unwrap(),
465        );
466        record
467    }
468
469    #[test]
470    fn reservation_is_idempotent_for_the_same_full_request() {
471        let directory = tempfile::tempdir().unwrap();
472        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
473
474        let first = store.reserve(managed_record("execution-1", "operation-1"));
475        let retry = store.reserve(managed_record("execution-2", "operation-1"));
476
477        assert!(first.unwrap().is_new());
478        let retry = retry.unwrap();
479        assert!(!retry.is_new());
480        assert_eq!(retry.record().id, "execution-1");
481        assert_eq!(
482            BoxStateStore::load(store.path()).unwrap().records().len(),
483            1
484        );
485    }
486
487    #[test]
488    fn reservation_rejects_operation_reuse_with_different_intent() {
489        let directory = tempfile::tempdir().unwrap();
490        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
491        store
492            .reserve(managed_record("execution-1", "operation-1"))
493            .unwrap();
494        let mut conflicting = managed_record("execution-2", "operation-1");
495        conflicting
496            .managed_execution
497            .as_mut()
498            .unwrap()
499            .request
500            .external_sandbox_id = "sandbox-2".to_string();
501
502        let error = store.reserve(conflicting).unwrap_err();
503
504        assert!(matches!(error, ManagedExecutionStoreError::Conflict { .. }));
505        assert_eq!(
506            BoxStateStore::load(store.path()).unwrap().records().len(),
507            1
508        );
509    }
510
511    #[test]
512    fn pause_and_resume_completion_advance_generation_once() {
513        let directory = tempfile::tempdir().unwrap();
514        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
515        let id = ExecutionId::new("execution-1").unwrap();
516        store
517            .reserve(managed_record(id.as_str(), "operation-1"))
518            .unwrap();
519
520        store
521            .transition(
522                &id,
523                ExecutionGeneration::INITIAL,
524                ManagedExecutionState::Created,
525                ManagedExecutionState::Starting,
526            )
527            .unwrap();
528        let running = store
529            .transition(
530                &id,
531                ExecutionGeneration::INITIAL,
532                ManagedExecutionState::Starting,
533                ManagedExecutionState::Running,
534            )
535            .unwrap();
536        assert_eq!(
537            running.managed_execution.unwrap().generation,
538            ExecutionGeneration::INITIAL
539        );
540        store
541            .transition(
542                &id,
543                ExecutionGeneration::INITIAL,
544                ManagedExecutionState::Running,
545                ManagedExecutionState::Pausing,
546            )
547            .unwrap();
548        let paused = store
549            .transition(
550                &id,
551                ExecutionGeneration::INITIAL,
552                ManagedExecutionState::Pausing,
553                ManagedExecutionState::Paused,
554            )
555            .unwrap();
556        let generation_two = ExecutionGeneration::new(2).unwrap();
557        assert_eq!(paused.managed_execution.unwrap().generation, generation_two);
558        store
559            .transition(
560                &id,
561                generation_two,
562                ManagedExecutionState::Paused,
563                ManagedExecutionState::Resuming,
564            )
565            .unwrap();
566        let resumed = store
567            .transition(
568                &id,
569                generation_two,
570                ManagedExecutionState::Resuming,
571                ManagedExecutionState::Running,
572            )
573            .unwrap();
574        assert_eq!(
575            resumed.managed_execution.unwrap().generation,
576            ExecutionGeneration::new(3).unwrap()
577        );
578    }
579
580    #[test]
581    fn snapshot_intent_is_durable_and_preserves_runtime_generation() {
582        let directory = tempfile::tempdir().unwrap();
583        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
584        let id = ExecutionId::new("execution-1").unwrap();
585        store
586            .reserve(managed_record(id.as_str(), "operation-1"))
587            .unwrap();
588        store
589            .transition(
590                &id,
591                ExecutionGeneration::INITIAL,
592                ManagedExecutionState::Created,
593                ManagedExecutionState::Starting,
594            )
595            .unwrap();
596        store
597            .transition(
598                &id,
599                ExecutionGeneration::INITIAL,
600                ManagedExecutionState::Starting,
601                ManagedExecutionState::Running,
602            )
603            .unwrap();
604        let snapshot_id = ExecutionSnapshotId::new("snapshot-1").unwrap();
605        let claimed = store
606            .transition_with(
607                &id,
608                ExecutionGeneration::INITIAL,
609                ManagedExecutionState::Running,
610                ManagedExecutionState::Snapshotting,
611                |record| {
612                    record.managed_execution.as_mut().unwrap().pending_operation =
613                        Some(ManagedExecutionOperation::Snapshot {
614                            snapshot_id: snapshot_id.clone(),
615                            source_state: ManagedExecutionState::Running,
616                        });
617                },
618            )
619            .unwrap();
620        assert_eq!(
621            claimed.managed_execution.as_ref().unwrap().generation,
622            ExecutionGeneration::INITIAL
623        );
624        assert!(matches!(
625            claimed
626                .managed_execution
627                .as_ref()
628                .unwrap()
629                .pending_operation
630                .as_ref(),
631            Some(ManagedExecutionOperation::Snapshot {
632                snapshot_id,
633                source_state: ManagedExecutionState::Running,
634            }) if snapshot_id.as_str() == "snapshot-1"
635        ));
636
637        let reopened = ManagedExecutionStore::new(store.path().to_path_buf());
638        let persisted = reopened.get(&id).unwrap().unwrap();
639        assert_eq!(
640            persisted.managed_state().unwrap(),
641            Some(ManagedExecutionState::Snapshotting)
642        );
643        let completed = reopened
644            .transition(
645                &id,
646                ExecutionGeneration::INITIAL,
647                ManagedExecutionState::Snapshotting,
648                ManagedExecutionState::Running,
649            )
650            .unwrap();
651        let metadata = completed.managed_execution.unwrap();
652        assert_eq!(metadata.generation, ExecutionGeneration::INITIAL);
653        assert!(metadata.pending_operation.is_none());
654        assert!(matches!(
655            reopened.transition(
656                &id,
657                ExecutionGeneration::INITIAL,
658                ManagedExecutionState::Running,
659                ManagedExecutionState::Snapshotting,
660            ),
661            Err(ManagedExecutionStoreError::InvalidRecord(_))
662        ));
663    }
664
665    #[test]
666    fn restart_advances_generation_between_durable_teardown_and_startup() {
667        let directory = tempfile::tempdir().unwrap();
668        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
669        let id = ExecutionId::new("execution-1").unwrap();
670        store
671            .reserve(managed_record(id.as_str(), "operation-create"))
672            .unwrap();
673        store
674            .transition(
675                &id,
676                ExecutionGeneration::INITIAL,
677                ManagedExecutionState::Created,
678                ManagedExecutionState::Starting,
679            )
680            .unwrap();
681        store
682            .transition(
683                &id,
684                ExecutionGeneration::INITIAL,
685                ManagedExecutionState::Starting,
686                ManagedExecutionState::Running,
687            )
688            .unwrap();
689        let restart_operation = OperationId::new("operation-restart").unwrap();
690        let stopping = store
691            .transition_with(
692                &id,
693                ExecutionGeneration::INITIAL,
694                ManagedExecutionState::Running,
695                ManagedExecutionState::RestartStopping,
696                |record| {
697                    record.managed_execution.as_mut().unwrap().pending_operation =
698                        Some(ManagedExecutionOperation::Restart {
699                            operation_id: restart_operation.clone(),
700                            source_generation: ExecutionGeneration::INITIAL,
701                            source_state: ManagedExecutionState::Running,
702                            stop_timeout_secs: Some(10),
703                        });
704                },
705            )
706            .unwrap();
707        assert_eq!(
708            stopping.managed_execution.as_ref().unwrap().generation,
709            ExecutionGeneration::INITIAL
710        );
711
712        let starting = store
713            .transition(
714                &id,
715                ExecutionGeneration::INITIAL,
716                ManagedExecutionState::RestartStopping,
717                ManagedExecutionState::RestartStarting,
718            )
719            .unwrap();
720        let generation_two = ExecutionGeneration::new(2).unwrap();
721        assert_eq!(
722            starting.managed_execution.as_ref().unwrap().generation,
723            generation_two
724        );
725        let running = store
726            .transition(
727                &id,
728                generation_two,
729                ManagedExecutionState::RestartStarting,
730                ManagedExecutionState::Running,
731            )
732            .unwrap();
733        let metadata = running.managed_execution.unwrap();
734        assert_eq!(metadata.generation, generation_two);
735        assert!(metadata.pending_operation.is_none());
736        let completed = metadata.last_restart.unwrap();
737        assert_eq!(completed.operation_id, restart_operation);
738        assert_eq!(completed.source_generation, ExecutionGeneration::INITIAL);
739        assert_eq!(completed.target_generation, generation_two);
740        assert_eq!(completed.outcome, ManagedRestartOutcome::Running);
741        assert_eq!(completed.stop_timeout_secs, Some(10));
742    }
743
744    #[test]
745    fn stale_generation_and_invalid_edges_do_not_change_disk() {
746        let directory = tempfile::tempdir().unwrap();
747        let store = ManagedExecutionStore::new(directory.path().join("boxes.json"));
748        let id = ExecutionId::new("execution-1").unwrap();
749        store
750            .reserve(managed_record(id.as_str(), "operation-1"))
751            .unwrap();
752        store
753            .transition(
754                &id,
755                ExecutionGeneration::INITIAL,
756                ManagedExecutionState::Created,
757                ManagedExecutionState::Starting,
758            )
759            .unwrap();
760        store
761            .transition(
762                &id,
763                ExecutionGeneration::INITIAL,
764                ManagedExecutionState::Starting,
765                ManagedExecutionState::Running,
766            )
767            .unwrap();
768
769        let stale = store.transition(
770            &id,
771            ExecutionGeneration::new(2).unwrap(),
772            ManagedExecutionState::Running,
773            ManagedExecutionState::Pausing,
774        );
775        let invalid = store.transition(
776            &id,
777            ExecutionGeneration::INITIAL,
778            ManagedExecutionState::Running,
779            ManagedExecutionState::Paused,
780        );
781
782        assert!(matches!(
783            stale,
784            Err(ManagedExecutionStoreError::Conflict { .. })
785        ));
786        assert!(matches!(
787            invalid,
788            Err(ManagedExecutionStoreError::InvalidTransition { .. })
789        ));
790        let persisted = store.get(&id).unwrap().unwrap();
791        assert_eq!(
792            persisted.managed_state().unwrap(),
793            Some(ManagedExecutionState::Running)
794        );
795        assert_eq!(
796            persisted.managed_execution.unwrap().generation,
797            ExecutionGeneration::INITIAL
798        );
799    }
800
801    #[cfg(unix)]
802    #[test]
803    fn concurrent_claims_have_one_winner() {
804        let directory = tempfile::tempdir().unwrap();
805        let store = Arc::new(ManagedExecutionStore::new(
806            directory.path().join("boxes.json"),
807        ));
808        let id = ExecutionId::new("execution-1").unwrap();
809        store
810            .reserve(managed_record(id.as_str(), "operation-1"))
811            .unwrap();
812        store
813            .transition(
814                &id,
815                ExecutionGeneration::INITIAL,
816                ManagedExecutionState::Created,
817                ManagedExecutionState::Starting,
818            )
819            .unwrap();
820        store
821            .transition(
822                &id,
823                ExecutionGeneration::INITIAL,
824                ManagedExecutionState::Starting,
825                ManagedExecutionState::Running,
826            )
827            .unwrap();
828        let barrier = Arc::new(Barrier::new(3));
829        let handles: Vec<_> = (0..2)
830            .map(|_| {
831                let store = Arc::clone(&store);
832                let id = id.clone();
833                let barrier = Arc::clone(&barrier);
834                std::thread::spawn(move || {
835                    barrier.wait();
836                    store.transition(
837                        &id,
838                        ExecutionGeneration::INITIAL,
839                        ManagedExecutionState::Running,
840                        ManagedExecutionState::Pausing,
841                    )
842                })
843            })
844            .collect();
845        barrier.wait();
846        let results: Vec<_> = handles
847            .into_iter()
848            .map(|handle| handle.join().unwrap())
849            .collect();
850
851        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
852        assert_eq!(
853            results
854                .iter()
855                .filter(|result| matches!(result, Err(ManagedExecutionStoreError::Conflict { .. })))
856                .count(),
857            1
858        );
859        assert_eq!(
860            store.get(&id).unwrap().unwrap().managed_state().unwrap(),
861            Some(ManagedExecutionState::Pausing)
862        );
863    }
864}