1use 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#[derive(Debug, Clone)]
15pub struct ManagedExecutionStore {
16 path: PathBuf,
17}
18
19#[derive(Debug, Clone)]
21pub enum ManagedExecutionReservation {
22 Reserved(BoxRecord),
24 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#[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 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 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 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 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 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 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 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 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}