start_command/
execution_attempt.rs

1//! Attempt boundaries and evidence for executions sharing one UUID/log.
2use chrono::{DateTime, Utc};
3use serde::{Deserialize, Serialize};
4use serde_json::{json, Value};
5use std::fs;
6use std::io::Read;
7use std::path::PathBuf;
8
9use crate::execution_store::ExecutionRecord;
10use crate::isolation::isolation_log::{append_log_file, read_log_tail_from};
11
12#[derive(Debug, Clone, Serialize, Deserialize)]
13#[serde(rename_all = "camelCase")]
14pub struct ExecutionAttempt {
15    pub number: u64,
16    pub started_at: String,
17    pub log_offset: Option<u64>,
18    pub last_output_at: Option<String>,
19    pub launch_accepted_at: Option<String>,
20    pub watcher_attached_at: Option<String>,
21    pub mode: String,
22    pub previous_session_name: Option<String>,
23    pub session_name: Option<String>,
24    #[serde(default, skip_serializing_if = "Option::is_none")]
25    pub watcher_error: Option<String>,
26}
27
28pub fn create_attempt(
29    record: &ExecutionRecord,
30    mode: &str,
31    session_name: &str,
32) -> ExecutionAttempt {
33    let counter = |key| record.options.get(key).and_then(Value::as_u64).unwrap_or(0);
34    let previous = record
35        .attempt
36        .as_ref()
37        .map(|a| a.number)
38        .unwrap_or(counter("resumeCount") + counter("recoveryAttempts") + 1);
39    let log_offset = if record.log_path.is_empty() {
40        None
41    } else {
42        match fs::metadata(&record.log_path) {
43            Ok(metadata) => Some(metadata.len()),
44            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Some(0),
45            Err(_) => None,
46        }
47    };
48    ExecutionAttempt {
49        number: previous + 1,
50        started_at: Utc::now().to_rfc3339(),
51        log_offset,
52        last_output_at: None,
53        launch_accepted_at: None,
54        watcher_attached_at: None,
55        mode: mode.into(),
56        previous_session_name: record
57            .options
58            .get("sessionName")
59            .and_then(Value::as_str)
60            .map(str::to_string),
61        session_name: Some(session_name.into()),
62        watcher_error: None,
63    }
64}
65
66pub fn archive_attempt(record: &mut ExecutionRecord) {
67    read_attempt_activity(record);
68    let value = serde_json::to_value(&*record).unwrap_or_default();
69    let mut snapshot = record.attempt.as_ref().and_then(|a| serde_json::to_value(a).ok())
70        .unwrap_or_else(|| json!({ "number": record.options.get("resumeCount").and_then(Value::as_u64).unwrap_or(0)
71            + record.options.get("recoveryAttempts").and_then(Value::as_u64).unwrap_or(0) + 1,
72            "startedAt": record.options.get("resumedAt").cloned().unwrap_or(json!(record.start_time)), "logOffset": null }));
73    for key in [
74        "command",
75        "status",
76        "exitCode",
77        "endTime",
78        "endTimeSource",
79        "observedAt",
80        "staleDetectedAt",
81        "containerStartedAt",
82        "exitReason",
83        "oomKilled",
84        "memoryExhausted",
85        "memoryExhaustedReason",
86        "cgroupMemory",
87    ] {
88        if let Some(field) = value.get(key) {
89            snapshot[key] = field.clone();
90        }
91    }
92    for key in ["sessionName", "containerId", "containerError"] {
93        snapshot[key] = record.options.get(key).cloned().unwrap_or(Value::Null);
94    }
95    record.attempt_history.push(snapshot);
96    record.end_time_source = None;
97    record.observed_at = None;
98    record.stale_detected_at = None;
99    record.container_started_at = None;
100    record.exit_reason = None;
101    record.oom_killed = None;
102    record.memory_exhausted = None;
103    record.memory_exhausted_reason = None;
104    record.cgroup_memory = None;
105    record.options.remove("containerError");
106}
107
108pub fn append_lifecycle(record: &ExecutionRecord, event: &str, details: Value) {
109    if let Some(attempt) = &record.attempt {
110        if record.log_path.is_empty() {
111            return;
112        }
113        let mut entry = serde_json::to_value(attempt).unwrap_or_default();
114        entry["event"] = json!(event);
115        entry["at"] = json!(Utc::now().to_rfc3339());
116        entry["uuid"] = json!(record.uuid);
117        entry["attemptNumber"] = json!(attempt.number);
118        if let Some(details) = details.as_object() {
119            for (key, value) in details {
120                entry[key] = value.clone();
121            }
122        }
123        append_log_file(
124            &PathBuf::from(&record.log_path),
125            &format!("\n[Start Command Lifecycle] {}\n", entry),
126        );
127    }
128}
129
130pub fn read_attempt_log_tail(record: &ExecutionRecord, bytes: u64) -> Option<String> {
131    let offset = match &record.attempt {
132        Some(attempt) => attempt.log_offset?,
133        None => 0,
134    };
135    read_log_tail_from(&record.log_path, bytes, offset)
136}
137
138pub fn activity_path(log_path: &str, number: u64) -> String {
139    format!("{}.attempt-{}.activity", log_path, number)
140}
141
142pub fn read_attempt_activity(record: &mut ExecutionRecord) {
143    let Some(attempt) = record.attempt.as_mut() else {
144        return;
145    };
146    if record.log_path.is_empty() {
147        return;
148    }
149    let Ok(file) = fs::File::open(activity_path(&record.log_path, attempt.number)) else {
150        return;
151    };
152    let mut text = String::new();
153    if file.take(128).read_to_string(&mut text).is_err() {
154        return;
155    }
156    let time = DateTime::parse_from_rfc3339(text.trim());
157    let since = DateTime::parse_from_rfc3339(&attempt.started_at);
158    if let (Ok(time), Ok(since)) = (time, since) {
159        if time >= since {
160            attempt.last_output_at = Some(text.trim().into());
161        }
162    }
163}
start_command/execution_attempt.rs

start_command/
execution_attempt.rs
