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 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 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 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 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}