1use crate::lino_value_json::{json_to_lino_value, lino_value_to_json};
4use chrono::Utc;
5use lino_objects_codec::{decode, encode, LinoValue};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::collections::HashMap;
9use std::env;
10use std::fs;
11use std::path::{Path, PathBuf};
12use std::process::{Command, Stdio};
13use uuid::Uuid;
14
15const DEFAULT_APP_FOLDER_NAME: &str = ".start-command";
17const LINO_DB_FILE: &str = "executions.lino";
19const LINKS_DB_FILE: &str = "executions.links";
21const LOCK_FILE: &str = "executions.lock";
23const LOCK_TIMEOUT_MS: u64 = 30000;
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "lowercase")]
29pub enum ExecutionStatus {
30 Executing,
31 Executed,
32}
33
34impl ExecutionStatus {
35 pub fn as_str(&self) -> &'static str {
36 match self {
37 ExecutionStatus::Executing => "executing",
38 ExecutionStatus::Executed => "executed",
39 }
40 }
41}
42
43impl std::fmt::Display for ExecutionStatus {
44 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45 write!(f, "{}", self.as_str())
46 }
47}
48
49#[derive(Debug, Clone, Serialize, Deserialize)]
51#[serde(rename_all = "camelCase")]
52pub struct ExecutionRecord {
53 pub uuid: String,
54 pub pid: Option<u32>,
55 pub status: ExecutionStatus,
56 pub exit_code: Option<i32>,
57 pub command: String,
58 pub log_path: String,
59 pub start_time: String,
60 pub end_time: Option<String>,
61 #[serde(default, skip_serializing_if = "Option::is_none")]
62 pub attempt: Option<crate::execution_attempt::ExecutionAttempt>,
63 #[serde(default, skip_serializing_if = "Vec::is_empty")]
64 pub attempt_history: Vec<Value>,
65 #[serde(default, skip_serializing_if = "Option::is_none")]
70 pub end_time_source: Option<String>,
71 #[serde(default, skip_serializing_if = "Option::is_none")]
74 pub observed_at: Option<String>,
75 #[serde(default, skip_serializing_if = "Option::is_none")]
78 pub stale_detected_at: Option<String>,
79 #[serde(default, skip_serializing_if = "Option::is_none")]
81 pub container_started_at: Option<String>,
82 #[serde(default, skip_serializing_if = "Option::is_none")]
83 pub oom_killed: Option<bool>,
84 #[serde(
88 default,
89 skip_serializing_if = "Option::is_none",
90 deserialize_with = "crate::cgroup_memory::deserialize_cgroup_memory"
91 )]
92 pub cgroup_memory: Option<crate::cgroup_memory::CgroupMemory>,
93 #[serde(default, skip_serializing_if = "Option::is_none")]
94 pub cpu_penalty: Option<Value>,
95 #[serde(default, skip_serializing_if = "Option::is_none")]
98 pub exit_reason: Option<String>,
99 #[serde(default, skip_serializing_if = "Option::is_none")]
102 pub memory_exhausted: Option<bool>,
103 #[serde(default, skip_serializing_if = "Option::is_none")]
105 pub memory_exhausted_reason: Option<String>,
106 pub working_directory: String,
107 pub shell: String,
108 pub platform: String,
109 #[serde(default)]
110 pub options: HashMap<String, Value>,
111}
112
113impl ExecutionRecord {
114 pub fn new(command: &str) -> Self {
116 let now = Utc::now();
117 ExecutionRecord {
118 uuid: Uuid::new_v4().to_string(),
119 pid: None,
120 status: ExecutionStatus::Executing,
121 exit_code: None,
122 command: command.to_string(),
123 log_path: String::new(),
124 start_time: now.to_rfc3339(),
125 end_time: None,
126 attempt: None,
127 attempt_history: Vec::new(),
128 end_time_source: None,
129 observed_at: None,
130 stale_detected_at: None,
131 container_started_at: None,
132 oom_killed: None,
133 cgroup_memory: None,
134 cpu_penalty: None,
135 exit_reason: None,
136 memory_exhausted: None,
137 memory_exhausted_reason: None,
138 working_directory: env::current_dir()
139 .map(|p| p.to_string_lossy().to_string())
140 .unwrap_or_default(),
141 shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
142 platform: std::env::consts::OS.to_string(),
143 options: HashMap::new(),
144 }
145 }
146
147 pub fn with_options(options: ExecutionRecordOptions) -> Self {
149 let mut record = Self::new(&options.command);
150 if let Some(uuid) = options.uuid {
151 record.uuid = uuid;
152 }
153 if let Some(pid) = options.pid {
154 record.pid = Some(pid);
155 }
156 if let Some(status) = options.status {
157 record.status = status;
158 }
159 if let Some(exit_code) = options.exit_code {
160 record.exit_code = Some(exit_code);
161 }
162 if let Some(log_path) = options.log_path {
163 record.log_path = log_path;
164 }
165 if let Some(start_time) = options.start_time {
166 record.start_time = start_time;
167 }
168 if let Some(end_time) = options.end_time {
169 record.end_time = Some(end_time);
170 }
171 if let Some(oom_killed) = options.oom_killed {
172 record.oom_killed = Some(oom_killed);
173 }
174 if let Some(working_directory) = options.working_directory {
175 record.working_directory = working_directory;
176 }
177 if let Some(shell) = options.shell {
178 record.shell = shell;
179 }
180 if let Some(platform) = options.platform {
181 record.platform = platform;
182 }
183 if let Some(opts) = options.options {
184 record.options = opts;
185 }
186 record
187 }
188
189 pub fn complete(&mut self, exit_code: i32) {
191 self.status = ExecutionStatus::Executed;
192 self.exit_code = Some(exit_code);
193 self.end_time = Some(Utc::now().to_rfc3339());
194 }
195
196 pub fn to_json(&self) -> Value {
198 serde_json::to_value(self).unwrap_or(Value::Null)
199 }
200
201 pub fn from_json(value: &Value) -> Option<Self> {
203 serde_json::from_value(value.clone()).ok()
204 }
205}
206
207#[derive(Debug, Default)]
209pub struct ExecutionRecordOptions {
210 pub uuid: Option<String>,
211 pub command: String,
212 pub pid: Option<u32>,
213 pub status: Option<ExecutionStatus>,
214 pub exit_code: Option<i32>,
215 pub log_path: Option<String>,
216 pub start_time: Option<String>,
217 pub end_time: Option<String>,
218 pub oom_killed: Option<bool>,
219 pub working_directory: Option<String>,
220 pub shell: Option<String>,
221 pub platform: Option<String>,
222 pub options: Option<HashMap<String, Value>>,
223}
224
225pub use crate::store_lock::LockManager;
226
227pub fn is_clink_installed() -> bool {
229 Command::new("clink")
230 .arg("--version")
231 .stdout(Stdio::null())
232 .stderr(Stdio::null())
233 .status()
234 .map(|s| s.success())
235 .unwrap_or(false)
236}
237
238pub fn get_default_app_folder() -> PathBuf {
240 if let Ok(custom) = env::var("START_APP_FOLDER") {
241 return PathBuf::from(custom);
242 }
243
244 let home = env::var("HOME")
245 .or_else(|_| env::var("USERPROFILE"))
246 .unwrap_or_else(|_| ".".to_string());
247 PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
248}
249
250#[derive(Clone)]
252pub struct ExecutionStore {
253 app_folder: PathBuf,
254 lino_db_path: PathBuf,
255 links_db_path: PathBuf,
256 lock_file_path: PathBuf,
257 use_links: bool,
258 verbose: bool,
259}
260
261#[derive(Debug, Default)]
263pub struct ExecutionStoreOptions {
264 pub app_folder: Option<PathBuf>,
265 pub use_links: Option<bool>,
266 pub verbose: bool,
267}
268
269impl ExecutionStore {
270 pub fn new() -> Self {
272 Self::with_options(ExecutionStoreOptions::default())
273 }
274
275 pub fn with_options(options: ExecutionStoreOptions) -> Self {
277 let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
278 let lino_db_path = app_folder.join(LINO_DB_FILE);
279 let links_db_path = app_folder.join(LINKS_DB_FILE);
280 let lock_file_path = app_folder.join(LOCK_FILE);
281 let use_links = options.use_links.unwrap_or_else(is_clink_installed);
282
283 let _ = fs::create_dir_all(&app_folder);
285
286 ExecutionStore {
287 app_folder,
288 lino_db_path,
289 links_db_path,
290 lock_file_path,
291 use_links,
292 verbose: options.verbose,
293 }
294 }
295
296 fn log(&self, message: &str) {
298 if self.verbose {
299 println!("[ExecutionStore] {}", message);
300 }
301 }
302
303 pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
305 if !self.lino_db_path.exists() {
306 return Vec::new();
307 }
308
309 match fs::read_to_string(&self.lino_db_path) {
310 Ok(content) => {
311 if content.trim().is_empty() {
312 return Vec::new();
313 }
314
315 match decode(&content) {
316 Ok(data) => {
317 if let LinoValue::Array(arr) = data {
318 arr.iter()
319 .map(lino_value_to_json)
320 .filter_map(|value| ExecutionRecord::from_json(&value))
321 .collect()
322 } else {
323 Vec::new()
324 }
325 }
326 Err(e) => {
327 self.log(&format!("Error decoding lino records: {}", e));
328 Vec::new()
329 }
330 }
331 }
332 Err(e) => {
333 self.log(&format!("Error reading lino records: {}", e));
334 Vec::new()
335 }
336 }
337 }
338
339 fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
341 let data: Vec<LinoValue> = records
342 .iter()
343 .map(|record| json_to_lino_value(&record.to_json()))
344 .collect();
345 let content = encode(&LinoValue::Array(data));
346 crate::atomic_write::atomic_write(&self.lino_db_path, content.as_bytes())?;
347 self.log(&format!("Wrote {} records to lino file", records.len()));
348 Ok(())
349 }
350
351 fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
353 let obj = record.to_json();
354 let mut links = Vec::new();
355
356 links.push(format!(
358 "({}: ExecutionRecord {})",
359 record.uuid, record.uuid
360 ));
361
362 if let Value::Object(map) = obj {
364 for (key, value) in map {
365 let escaped_value = match value {
366 Value::Object(_) | Value::Array(_) => {
367 serde_json::to_string(&value).unwrap_or_default()
368 }
369 Value::String(s) => s,
370 Value::Null => "null".to_string(),
371 other => other.to_string(),
372 };
373 links.push(format!(
374 "({}.{}: {} \"{}\")",
375 record.uuid,
376 key,
377 key,
378 escaped_value.replace('"', "\\\"")
379 ));
380 }
381 }
382
383 format!("() (({})))", links.join(") ("))
384 }
385
386 fn exec_clink(&self, query: &str) -> Result<String, String> {
388 match Command::new("clink")
389 .arg(query)
390 .arg("--db")
391 .arg(&self.links_db_path)
392 .output()
393 {
394 Ok(output) => {
395 if output.status.success() {
396 Ok(String::from_utf8_lossy(&output.stdout).to_string())
397 } else {
398 let stderr = String::from_utf8_lossy(&output.stderr).to_string();
399 self.log(&format!("Clink error: {}", stderr));
400 Err(stderr)
401 }
402 }
403 Err(e) => {
404 self.log(&format!("Clink execution error: {}", e));
405 Err(e.to_string())
406 }
407 }
408 }
409
410 fn write_links_record(&self, record: &ExecutionRecord) -> bool {
412 if !self.use_links {
413 return false;
414 }
415
416 let query = self.build_clink_create_query(record);
417 match self.exec_clink(&query) {
418 Ok(_) => {
419 self.log(&format!("Wrote record {} to links database", record.uuid));
420 true
421 }
422 Err(_) => false,
423 }
424 }
425
426 fn delete_links_record(&self, uuid: &str) -> bool {
428 if !self.use_links {
429 return false;
430 }
431
432 let query = format!("(($id: {} $any)) ()", uuid);
433 self.exec_clink(&query).is_ok()
434 }
435
436 pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
438 self.save_reserved(record, None)
439 }
440
441 pub fn reserve_launch(
442 &self,
443 record: &ExecutionRecord,
444 previous: &ExecutionRecord,
445 ) -> Result<(), String> {
446 self.save_reserved(record, Some(previous))
447 }
448
449 fn save_reserved(
450 &self,
451 record: &ExecutionRecord,
452 previous: Option<&ExecutionRecord>,
453 ) -> Result<(), String> {
454 let mut lock = LockManager::new(self.lock_file_path.clone());
455
456 if !lock.acquire(LOCK_TIMEOUT_MS) {
457 return Err("Failed to acquire lock for database write".to_string());
458 }
459
460 let mut records = self.read_lino_records();
462
463 let existing_index = records.iter().position(|r| r.uuid == record.uuid);
465 if let Some(previous) = previous {
466 let current = existing_index
467 .and_then(|i| records.get(i))
468 .ok_or("Execution disappeared before launch reservation")?;
469 if crate::launch_owner::has_active_launch(current)
470 || current.attempt.as_ref().map(|a| a.number)
471 != previous.attempt.as_ref().map(|a| a.number)
472 {
473 return Err("A different resume already reserved this execution".into());
474 }
475 }
476
477 if let Some(idx) = existing_index {
478 records[idx] = record.clone();
480 } else {
481 records.push(record.clone());
483 }
484
485 self.write_lino_records(&records)
487 .map_err(|e| e.to_string())?;
488
489 if self.use_links {
491 self.write_links_record(record);
492 }
493
494 Ok(())
495 }
496
497 pub fn patch_attempt(
499 &self,
500 uuid: &str,
501 number: u64,
502 fields: Value,
503 ) -> Result<Option<ExecutionRecord>, String> {
504 let mut lock = LockManager::new(self.lock_file_path.clone());
505 if !lock.acquire(LOCK_TIMEOUT_MS) {
506 return Err("Failed to acquire lock for attempt metadata".into());
507 }
508 let mut records = self.read_lino_records();
509 let Some(current) = records
510 .iter_mut()
511 .find(|r| r.uuid == uuid && r.attempt.as_ref().is_some_and(|a| a.number == number))
512 else {
513 return Ok(None);
514 };
515 let mut value =
516 serde_json::to_value(current.attempt.as_ref().unwrap()).map_err(|e| e.to_string())?;
517 if let Some(fields) = fields.as_object() {
518 for (key, field) in fields {
519 value[key] = field.clone();
520 }
521 }
522 current.attempt = Some(serde_json::from_value(value).map_err(|e| e.to_string())?);
523 let updated = current.clone();
524 self.write_lino_records(&records)
525 .map_err(|e| e.to_string())?;
526 if self.use_links {
527 self.write_links_record(&updated);
528 }
529 Ok(Some(updated))
530 }
531
532 pub fn get(&self, identifier: &str) -> Option<ExecutionRecord> {
534 let records = self.read_lino_records();
535 records
536 .iter()
537 .find(|r| r.uuid == identifier)
538 .cloned()
539 .or_else(|| {
540 records
541 .iter()
542 .find(|r| {
543 r.options.get("sessionName").and_then(|v| v.as_str()) == Some(identifier)
544 })
545 .cloned()
546 })
547 .or_else(|| {
548 records.into_iter().find(|r| {
552 r.options
553 .get("sessionNameHistory")
554 .and_then(|v| v.as_array())
555 .is_some_and(|history| {
556 history.iter().any(|name| name.as_str() == Some(identifier))
557 })
558 })
559 })
560 }
561
562 pub fn get_all(&self) -> Vec<ExecutionRecord> {
564 self.read_lino_records()
565 }
566 pub fn get_by_status(&self, status: ExecutionStatus) -> Vec<ExecutionRecord> {
568 self.read_lino_records()
569 .into_iter()
570 .filter(|r| r.status == status)
571 .collect()
572 }
573 pub fn get_executing(&self) -> Vec<ExecutionRecord> {
575 self.get_by_status(ExecutionStatus::Executing)
576 }
577 pub fn get_recent(&self, limit: usize) -> Vec<ExecutionRecord> {
579 let mut records = self.read_lino_records();
580 records.sort_by(|a, b| b.start_time.cmp(&a.start_time));
581 records.truncate(limit);
582 records
583 }
584 pub fn cleanup_stale(&self, options: CleanupOptions) -> CleanupResult {
586 let max_age_ms = options.max_age_ms.unwrap_or(24 * 60 * 60 * 1000);
587 let dry_run = options.dry_run;
588 let mut result = CleanupResult {
589 cleaned: 0,
590 records: Vec::new(),
591 errors: Vec::new(),
592 };
593
594 let records = self.read_lino_records();
595 let mut stale_records: Vec<ExecutionRecord> = Vec::new();
596
597 for record in records
598 .iter()
599 .filter(|r| r.status == ExecutionStatus::Executing)
600 {
601 if crate::launch_owner::has_active_launch(record) {
602 continue;
603 }
604 let mut is_stale = false;
605
606 #[cfg(unix)]
608 if let Some(pid) = record.pid {
609 if record.platform == std::env::consts::OS
610 && unsafe { libc::kill(pid as i32, 0) } != 0
611 {
612 is_stale = true;
613 self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
614 }
615 }
616
617 if !is_stale {
619 if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
620 let age_ms =
621 (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
622 if age_ms > max_age_ms as i64 {
623 is_stale = true;
624 self.log(&format!(
625 "Stale: {} ({}min old)",
626 record.uuid,
627 age_ms / 60000
628 ));
629 }
630 }
631 }
632
633 if is_stale {
634 stale_records.push(record.clone());
635 }
636 }
637
638 result.records = stale_records.clone();
639
640 if !dry_run && !stale_records.is_empty() {
641 let mut lock = LockManager::new(self.lock_file_path.clone());
642 if !lock.acquire(LOCK_TIMEOUT_MS) {
643 result.errors.push("Failed to acquire lock".to_string());
644 return result;
645 }
646
647 let mut current = self.read_lino_records();
648 for stale in &stale_records {
649 if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
650 current[i].status = ExecutionStatus::Executed;
651 current[i].exit_code = Some(-1);
652 current[i].end_time = None;
656 current[i].stale_detected_at = Some(chrono::Utc::now().to_rfc3339());
657 result.cleaned += 1;
658 }
659 }
660
661 if let Err(e) = self.write_lino_records(¤t) {
662 result.errors.push(format!("Cleanup error: {}", e));
663 } else {
664 self.log(&format!("Cleaned {} stale records", result.cleaned));
665 }
666 } else if dry_run {
667 result.cleaned = stale_records.len();
668 }
669
670 result
671 }
672
673 pub fn delete(&self, uuid: &str) -> Result<bool, String> {
675 let mut lock = LockManager::new(self.lock_file_path.clone());
676
677 if !lock.acquire(LOCK_TIMEOUT_MS) {
678 return Err("Failed to acquire lock for database write".to_string());
679 }
680
681 let records = self.read_lino_records();
682 let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();
683
684 if filtered.len() == records.len() {
685 return Ok(false); }
687
688 self.write_lino_records(&filtered)
689 .map_err(|e| e.to_string())?;
690
691 if self.use_links {
693 self.delete_links_record(uuid);
694 }
695
696 Ok(true)
697 }
698
699 pub fn clear(&self) -> Result<(), String> {
701 let mut lock = LockManager::new(self.lock_file_path.clone());
702
703 if !lock.acquire(LOCK_TIMEOUT_MS) {
704 return Err("Failed to acquire lock for database write".to_string());
705 }
706
707 self.write_lino_records(&[]).map_err(|e| e.to_string())?;
708
709 if self.use_links && self.links_db_path.exists() {
711 let _ = fs::remove_file(&self.links_db_path);
712 }
713
714 Ok(())
715 }
716
717 pub fn verify_consistency(&self) -> ConsistencyResult {
719 let mut result = ConsistencyResult {
720 consistent: true,
721 lino_count: 0,
722 links_count: 0,
723 errors: Vec::new(),
724 };
725
726 let lino_records = self.read_lino_records();
728 result.lino_count = lino_records.len();
729
730 if !self.use_links {
731 result
732 .errors
733 .push("clink not installed - links database not available".to_string());
734 return result;
735 }
736
737 match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
739 Ok(output) => {
740 let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
742 let uuids: std::collections::HashSet<_> = re
743 .captures_iter(&output)
744 .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
745 .collect();
746 result.links_count = uuids.len();
747
748 if result.lino_count != result.links_count {
750 result.consistent = false;
751 result.errors.push(format!(
752 "Record count mismatch: lino={}, links={}",
753 result.lino_count, result.links_count
754 ));
755 }
756
757 for record in &lino_records {
759 if !uuids.contains(&record.uuid.to_lowercase()) {
760 result.consistent = false;
761 result.errors.push(format!(
762 "Record {} missing from links database",
763 record.uuid
764 ));
765 }
766 }
767 }
768 Err(e) => {
769 result
770 .errors
771 .push(format!("Failed to query links database: {}", e));
772 result.consistent = false;
773 }
774 }
775
776 result
777 }
778
779 pub fn get_stats(&self) -> ExecutionStats {
781 let records = self.read_lino_records();
782 let executing = records
783 .iter()
784 .filter(|r| r.status == ExecutionStatus::Executing)
785 .count();
786 let executed = records
787 .iter()
788 .filter(|r| r.status == ExecutionStatus::Executed)
789 .count();
790 let successful = records
791 .iter()
792 .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
793 .count();
794 let failed = records
795 .iter()
796 .filter(|r| {
797 r.status == ExecutionStatus::Executed
798 && r.exit_code.map(|c| c != 0).unwrap_or(false)
799 })
800 .count();
801
802 ExecutionStats {
803 total: records.len(),
804 executing,
805 executed,
806 successful,
807 failed,
808 clink_available: self.use_links,
809 lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
810 links_db_path: self.links_db_path.to_string_lossy().to_string(),
811 }
812 }
813
814 pub fn app_folder(&self) -> &Path {
816 &self.app_folder
817 }
818}
819
820impl Default for ExecutionStore {
821 fn default() -> Self {
822 Self::new()
823 }
824}
825
826#[derive(Debug)]
828pub struct ConsistencyResult {
829 pub consistent: bool,
830 pub lino_count: usize,
831 pub links_count: usize,
832 pub errors: Vec<String>,
833}
834
835#[derive(Debug, Default)]
837pub struct CleanupOptions {
838 pub max_age_ms: Option<u64>,
839 pub dry_run: bool,
840}
841
842#[derive(Debug)]
844pub struct CleanupResult {
845 pub cleaned: usize,
846 pub records: Vec<ExecutionRecord>,
847 pub errors: Vec<String>,
848}
849
850#[derive(Debug)]
852pub struct ExecutionStats {
853 pub total: usize,
854 pub executing: usize,
855 pub executed: usize,
856 pub successful: usize,
857 pub failed: usize,
858 pub clink_available: bool,
859 pub lino_db_path: String,
860 pub links_db_path: String,
861}
862
863#[cfg(test)]
864#[path = "execution_store_cases.rs"]
865mod tests;