1use std::collections::BTreeMap;
4use std::num::NonZeroU16;
5use std::pin::Pin;
6use std::time::Duration;
7
8use async_trait::async_trait;
9use chrono::{DateTime, Utc};
10use serde::{Deserialize, Serialize};
11use thiserror::Error;
12use tokio::io::{AsyncRead, AsyncWrite};
13
14use crate::config::{BoxConfig, ResourceConfig};
15use crate::execution::ResolvedExecutionPlan;
16use crate::log::{LogConfig, LogEntry};
17
18#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
20#[serde(try_from = "String", into = "String")]
21pub struct ExecutionId(String);
22
23impl ExecutionId {
24 pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
25 let value = value.into();
26 if value.trim().is_empty() {
27 return Err(ExecutionManagerError::InvalidRequest(
28 "execution ID cannot be empty".to_string(),
29 ));
30 }
31 Ok(Self(value))
32 }
33
34 pub fn as_str(&self) -> &str {
35 &self.0
36 }
37}
38
39impl std::fmt::Display for ExecutionId {
40 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41 formatter.write_str(&self.0)
42 }
43}
44
45impl TryFrom<String> for ExecutionId {
46 type Error = ExecutionManagerError;
47
48 fn try_from(value: String) -> Result<Self, Self::Error> {
49 Self::new(value)
50 }
51}
52
53impl From<ExecutionId> for String {
54 fn from(value: ExecutionId) -> Self {
55 value.0
56 }
57}
58
59#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
65#[serde(try_from = "String", into = "String")]
66pub struct ExecutionSnapshotId(String);
67
68impl ExecutionSnapshotId {
69 pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
70 let value = value.into();
71 if value.is_empty()
72 || value.len() > 128
73 || !value
74 .bytes()
75 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
76 {
77 return Err(ExecutionManagerError::InvalidRequest(
78 "execution snapshot ID must match [A-Za-z0-9_-]{1,128}".to_string(),
79 ));
80 }
81 Ok(Self(value))
82 }
83
84 pub fn as_str(&self) -> &str {
85 &self.0
86 }
87}
88
89impl std::fmt::Display for ExecutionSnapshotId {
90 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91 formatter.write_str(&self.0)
92 }
93}
94
95impl TryFrom<String> for ExecutionSnapshotId {
96 type Error = ExecutionManagerError;
97
98 fn try_from(value: String) -> Result<Self, Self::Error> {
99 Self::new(value)
100 }
101}
102
103impl From<ExecutionSnapshotId> for String {
104 fn from(value: ExecutionSnapshotId) -> Self {
105 value.0
106 }
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
111#[serde(try_from = "String", into = "String")]
112pub struct OperationId(String);
113
114impl OperationId {
115 pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
116 let value = value.into();
117 if value.trim().is_empty() {
118 return Err(ExecutionManagerError::InvalidRequest(
119 "operation ID cannot be empty".to_string(),
120 ));
121 }
122 Ok(Self(value))
123 }
124
125 pub fn as_str(&self) -> &str {
126 &self.0
127 }
128}
129
130impl std::fmt::Display for OperationId {
131 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132 formatter.write_str(&self.0)
133 }
134}
135
136impl TryFrom<String> for OperationId {
137 type Error = ExecutionManagerError;
138
139 fn try_from(value: String) -> Result<Self, Self::Error> {
140 Self::new(value)
141 }
142}
143
144impl From<OperationId> for String {
145 fn from(value: OperationId) -> Self {
146 value.0
147 }
148}
149
150#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
152#[serde(try_from = "u64", into = "u64")]
153pub struct ExecutionGeneration(u64);
154
155impl ExecutionGeneration {
156 pub const INITIAL: Self = Self(1);
157
158 pub fn new(value: u64) -> ExecutionManagerResult<Self> {
159 if value == 0 {
160 return Err(ExecutionManagerError::InvalidRequest(
161 "execution generation must be greater than zero".to_string(),
162 ));
163 }
164 Ok(Self(value))
165 }
166
167 pub const fn get(self) -> u64 {
168 self.0
169 }
170}
171
172impl TryFrom<u64> for ExecutionGeneration {
173 type Error = ExecutionManagerError;
174
175 fn try_from(value: u64) -> Result<Self, Self::Error> {
176 Self::new(value)
177 }
178}
179
180impl From<ExecutionGeneration> for u64 {
181 fn from(value: ExecutionGeneration) -> Self {
182 value.0
183 }
184}
185
186#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
188#[serde(rename_all = "kebab-case")]
189pub enum ExecutionRestartPolicy {
190 #[default]
192 No,
193 Always,
195 OnFailure,
197 UnlessStopped,
199}
200
201impl ExecutionRestartPolicy {
202 pub const fn as_str(self) -> &'static str {
204 match self {
205 Self::No => "no",
206 Self::Always => "always",
207 Self::OnFailure => "on-failure",
208 Self::UnlessStopped => "unless-stopped",
209 }
210 }
211}
212
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215pub struct ExecutionHealthCheck {
216 pub cmd: Vec<String>,
218 #[serde(default = "default_health_interval")]
220 pub interval_secs: u64,
221 #[serde(default = "default_health_timeout")]
223 pub timeout_secs: u64,
224 #[serde(default = "default_health_retries")]
226 pub retries: u32,
227 #[serde(default)]
229 pub start_period_secs: u64,
230}
231
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
238pub struct ExecutionRecordPolicy {
239 #[serde(default)]
241 pub name: Option<String>,
242 #[serde(default)]
244 pub auto_remove: bool,
245 #[serde(default)]
247 pub restart_policy: ExecutionRestartPolicy,
248 #[serde(default)]
250 pub max_restart_count: u32,
251 #[serde(default)]
253 pub health_check: Option<ExecutionHealthCheck>,
254 #[serde(default)]
256 pub healthcheck_disabled: bool,
257 #[serde(default)]
259 pub log_config: LogConfig,
260 #[serde(default)]
262 pub volume_names: Vec<String>,
263 #[serde(default)]
265 pub platform: Option<String>,
266 #[serde(default)]
268 pub init: bool,
269 #[serde(default)]
271 pub devices: Vec<String>,
272 #[serde(default)]
274 pub gpus: Option<String>,
275 #[serde(default)]
277 pub shm_size: Option<u64>,
278 #[serde(default)]
280 pub stop_signal: Option<String>,
281 #[serde(default)]
283 pub stop_timeout: Option<u64>,
284 #[serde(default)]
286 pub oom_kill_disable: bool,
287 #[serde(default)]
289 pub oom_score_adj: Option<i32>,
290}
291
292impl Default for ExecutionRecordPolicy {
293 fn default() -> Self {
294 Self {
295 name: None,
296 auto_remove: false,
297 restart_policy: ExecutionRestartPolicy::No,
298 max_restart_count: 0,
299 health_check: None,
300 healthcheck_disabled: false,
301 log_config: LogConfig::default(),
302 volume_names: Vec::new(),
303 platform: None,
304 init: false,
305 devices: Vec::new(),
306 gpus: None,
307 shm_size: None,
308 stop_signal: None,
309 stop_timeout: None,
310 oom_kill_disable: false,
311 oom_score_adj: None,
312 }
313 }
314}
315
316fn default_health_interval() -> u64 {
317 30
318}
319
320fn default_health_timeout() -> u64 {
321 5
322}
323
324fn default_health_retries() -> u32 {
325 3
326}
327
328#[derive(Debug, Clone, Serialize, Deserialize)]
330pub struct CreateExecutionRequest {
331 pub external_sandbox_id: String,
333 pub config: BoxConfig,
335 pub labels: BTreeMap<String, String>,
337 #[serde(default)]
339 pub policy: ExecutionRecordPolicy,
340 #[serde(default)]
344 pub rootfs_snapshot_id: Option<ExecutionSnapshotId>,
345}
346
347#[derive(Debug, Clone, Serialize, Deserialize)]
349pub struct ExecutionReservation {
350 pub execution_id: ExecutionId,
351 pub generation: ExecutionGeneration,
352 pub plan: ResolvedExecutionPlan,
353 pub resources: ResourceConfig,
354 pub created_at: DateTime<Utc>,
355}
356
357#[derive(Debug, Clone, Serialize, Deserialize)]
359pub struct ExecutionLease {
360 pub execution_id: ExecutionId,
361 pub generation: ExecutionGeneration,
362 pub plan: ResolvedExecutionPlan,
363 pub resources: ResourceConfig,
364 pub started_at: DateTime<Utc>,
365}
366
367#[derive(Debug, Clone, Serialize, Deserialize)]
369pub struct ExecutionSnapshot {
370 pub snapshot_id: ExecutionSnapshotId,
371 pub size_bytes: u64,
372 pub state: ExecutionState,
374 pub lease: ExecutionLease,
376}
377
378#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
380#[serde(rename_all = "snake_case")]
381pub enum ExecutionState {
382 Created,
383 Creating,
384 Running,
385 Paused,
386 Stopped,
387 Failed,
388}
389
390#[derive(Debug, Clone, Serialize, Deserialize)]
392pub struct ExecutionStatus {
393 pub execution_id: ExecutionId,
394 pub generation: ExecutionGeneration,
395 pub state: ExecutionState,
396 pub plan: ResolvedExecutionPlan,
397}
398
399#[derive(Debug, Clone, Copy, PartialEq, Eq)]
401pub enum KillOutcome {
402 Killed,
403 AlreadyStopped,
404}
405
406#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
408pub struct KillExecutionOptions {
409 #[serde(default)]
412 pub signal: Option<i32>,
413 #[serde(default)]
416 pub timeout_secs: Option<u64>,
417}
418
419#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
421pub struct RestartExecutionOptions {
422 #[serde(default)]
425 pub stop_timeout_secs: Option<u64>,
426}
427
428#[derive(Debug, Clone)]
430pub enum ReconcileOutcome {
431 Absent,
432 Created(ExecutionReservation),
433 Creating,
434 Ready(ExecutionLease),
435 Failed,
436}
437
438#[derive(Debug, Error)]
440pub enum ExecutionManagerError {
441 #[error("invalid execution request: {0}")]
442 InvalidRequest(String),
443 #[error("execution not found: {0}")]
444 NotFound(ExecutionId),
445 #[error("execution conflict for {execution_id}: {message}")]
446 Conflict {
447 execution_id: ExecutionId,
448 message: String,
449 },
450 #[error("execution backend unavailable: {0}")]
451 Unavailable(String),
452 #[error("execution lifecycle failed: {0}")]
453 Internal(String),
454}
455
456pub type ExecutionManagerResult<T> = std::result::Result<T, ExecutionManagerError>;
457
458pub trait ExecutionPortIo: AsyncRead + AsyncWrite + Send + Unpin {}
460
461impl<T> ExecutionPortIo for T where T: AsyncRead + AsyncWrite + Send + Unpin {}
462
463pub type ExecutionPortStream = Pin<Box<dyn ExecutionPortIo>>;
464
465#[async_trait]
471pub trait ExecutionPortConnector: Send + Sync {
472 async fn connect_port(
473 &self,
474 execution_id: &ExecutionId,
475 generation: ExecutionGeneration,
476 port: NonZeroU16,
477 timeout: Duration,
478 ) -> ExecutionManagerResult<ExecutionPortStream>;
479}
480
481#[async_trait]
483pub trait ExecutionManager: Send + Sync {
484 async fn create(
486 &self,
487 _request: CreateExecutionRequest,
488 _operation_id: &OperationId,
489 ) -> ExecutionManagerResult<ExecutionReservation> {
490 Err(ExecutionManagerError::Unavailable(
491 "this execution manager does not support staged create".to_string(),
492 ))
493 }
494
495 async fn start(
497 &self,
498 _execution_id: &ExecutionId,
499 _generation: ExecutionGeneration,
500 ) -> ExecutionManagerResult<ExecutionLease> {
501 Err(ExecutionManagerError::Unavailable(
502 "this execution manager does not support staged start".to_string(),
503 ))
504 }
505
506 async fn create_and_start(
511 &self,
512 request: CreateExecutionRequest,
513 operation_id: &OperationId,
514 ) -> ExecutionManagerResult<ExecutionLease> {
515 let reservation = self.create(request, operation_id).await?;
516 self.start(&reservation.execution_id, reservation.generation)
517 .await
518 }
519
520 async fn inspect(&self, execution_id: &ExecutionId) -> ExecutionManagerResult<ExecutionStatus>;
521
522 async fn read_logs(
524 &self,
525 _execution_id: &ExecutionId,
526 _generation: ExecutionGeneration,
527 ) -> ExecutionManagerResult<Vec<LogEntry>> {
528 Err(ExecutionManagerError::Unavailable(
529 "this execution manager does not expose structured logs".to_string(),
530 ))
531 }
532
533 async fn create_filesystem_snapshot(
536 &self,
537 _execution_id: &ExecutionId,
538 _generation: ExecutionGeneration,
539 _snapshot_id: &ExecutionSnapshotId,
540 ) -> ExecutionManagerResult<ExecutionSnapshot> {
541 Err(ExecutionManagerError::Unavailable(
542 "this execution manager does not support filesystem snapshots".to_string(),
543 ))
544 }
545
546 async fn filesystem_snapshot_size(
549 &self,
550 _snapshot_id: &ExecutionSnapshotId,
551 ) -> ExecutionManagerResult<Option<u64>> {
552 Err(ExecutionManagerError::Unavailable(
553 "this execution manager does not expose filesystem snapshots".to_string(),
554 ))
555 }
556
557 async fn delete_filesystem_snapshot(
560 &self,
561 _snapshot_id: &ExecutionSnapshotId,
562 ) -> ExecutionManagerResult<bool> {
563 Err(ExecutionManagerError::Unavailable(
564 "this execution manager does not support filesystem snapshot deletion".to_string(),
565 ))
566 }
567
568 async fn pause(
570 &self,
571 execution_id: &ExecutionId,
572 generation: ExecutionGeneration,
573 keep_memory: bool,
574 ) -> ExecutionManagerResult<ExecutionLease>;
575
576 async fn resume(
577 &self,
578 execution_id: &ExecutionId,
579 generation: ExecutionGeneration,
580 ) -> ExecutionManagerResult<ExecutionLease>;
581
582 async fn restart(
585 &self,
586 execution_id: &ExecutionId,
587 generation: ExecutionGeneration,
588 operation_id: &OperationId,
589 ) -> ExecutionManagerResult<ExecutionLease> {
590 self.restart_with_options(
591 execution_id,
592 generation,
593 operation_id,
594 RestartExecutionOptions::default(),
595 )
596 .await
597 }
598
599 async fn restart_with_options(
601 &self,
602 _execution_id: &ExecutionId,
603 _generation: ExecutionGeneration,
604 _operation_id: &OperationId,
605 _options: RestartExecutionOptions,
606 ) -> ExecutionManagerResult<ExecutionLease> {
607 Err(ExecutionManagerError::Unavailable(
608 "this execution manager does not support restart".to_string(),
609 ))
610 }
611
612 async fn kill(
613 &self,
614 execution_id: &ExecutionId,
615 generation: ExecutionGeneration,
616 ) -> ExecutionManagerResult<KillOutcome>;
617
618 async fn kill_with_options(
622 &self,
623 execution_id: &ExecutionId,
624 generation: ExecutionGeneration,
625 _options: KillExecutionOptions,
626 ) -> ExecutionManagerResult<KillOutcome> {
627 self.kill(execution_id, generation).await
628 }
629
630 async fn remove(
636 &self,
637 _execution_id: &ExecutionId,
638 _generation: ExecutionGeneration,
639 ) -> ExecutionManagerResult<bool> {
640 Err(ExecutionManagerError::Unavailable(
641 "this execution manager does not support execution removal".to_string(),
642 ))
643 }
644
645 async fn reconcile(
646 &self,
647 operation_id: &OperationId,
648 ) -> ExecutionManagerResult<ReconcileOutcome>;
649}
650
651#[cfg(test)]
652mod tests {
653 use super::*;
654
655 #[test]
656 fn identifiers_reject_empty_values() {
657 assert!(matches!(
658 ExecutionId::new(" "),
659 Err(ExecutionManagerError::InvalidRequest(_))
660 ));
661 assert!(matches!(
662 OperationId::new(""),
663 Err(ExecutionManagerError::InvalidRequest(_))
664 ));
665 }
666
667 #[test]
668 fn generation_rejects_zero() {
669 assert!(matches!(
670 ExecutionGeneration::new(0),
671 Err(ExecutionManagerError::InvalidRequest(_))
672 ));
673 assert_eq!(ExecutionGeneration::INITIAL.get(), 1);
674 assert!(serde_json::from_str::<ExecutionGeneration>("0").is_err());
675 }
676
677 #[test]
678 fn identifier_deserialization_preserves_invariants() {
679 assert!(serde_json::from_str::<ExecutionId>("\"\"").is_err());
680 assert!(serde_json::from_str::<OperationId>("\" \"").is_err());
681 }
682
683 #[test]
684 fn snapshot_identifiers_are_safe_managed_directory_names() {
685 for valid in ["snapshot-1", "SNAPSHOT_2", "a"] {
686 assert_eq!(ExecutionSnapshotId::new(valid).unwrap().as_str(), valid);
687 }
688 for invalid in [
689 "",
690 ".",
691 "..",
692 "../snapshot",
693 "snapshot/path",
694 "snapshot:tag",
695 "snapshot id",
696 ] {
697 assert!(matches!(
698 ExecutionSnapshotId::new(invalid),
699 Err(ExecutionManagerError::InvalidRequest(_))
700 ));
701 }
702 assert!(ExecutionSnapshotId::new("x".repeat(129)).is_err());
703 assert!(serde_json::from_str::<ExecutionSnapshotId>("\"../snapshot\"").is_err());
704 }
705
706 #[test]
707 fn legacy_creation_requests_default_record_policy() {
708 let request: CreateExecutionRequest = serde_json::from_value(serde_json::json!({
709 "external_sandbox_id": "sandbox-1",
710 "config": BoxConfig::default(),
711 "labels": {"purpose": "compatibility"}
712 }))
713 .unwrap();
714
715 assert_eq!(request.policy, ExecutionRecordPolicy::default());
716 assert_eq!(request.policy.restart_policy, ExecutionRestartPolicy::No);
717 assert!(request.rootfs_snapshot_id.is_none());
718 }
719
720 #[test]
721 fn restart_policy_has_stable_record_values() {
722 assert_eq!(ExecutionRestartPolicy::No.as_str(), "no");
723 assert_eq!(ExecutionRestartPolicy::Always.as_str(), "always");
724 assert_eq!(ExecutionRestartPolicy::OnFailure.as_str(), "on-failure");
725 assert_eq!(
726 ExecutionRestartPolicy::UnlessStopped.as_str(),
727 "unless-stopped"
728 );
729 assert_eq!(
730 serde_json::to_value(ExecutionRestartPolicy::OnFailure).unwrap(),
731 "on-failure"
732 );
733 }
734}