use std::sync::Arc;
use crate::{clock::Clock, run_store::RunStore};
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(
tag = "event",
rename_all = "camelCase",
rename_all_fields = "camelCase"
)]
pub enum Event {
ImageBuilt {
image_identifier: String,
image_name: String,
},
Launched {
image_identifier: String,
endpoint: String,
region: String,
},
Exec {
exec_id: String,
exit_code: Option<i32>,
truncated: bool,
writers_may_be_alive: Option<bool>,
},
HookObserved { hook: String, fired_at: u64 },
Suspended,
Resumed,
Terminated {
terminate_accepted: bool,
undeleted: Vec<String>,
},
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Record {
pub seq: u64,
pub at: u64,
#[serde(flatten)]
pub event: Event,
}
#[derive(Clone, Debug)]
pub struct History {
vm_id: String,
store: Arc<dyn RunStore>,
clock: Arc<dyn Clock>,
}
impl History {
pub fn from_store(vm_id: &str, store: Arc<dyn RunStore>, clock: Arc<dyn Clock>) -> Self {
Self {
vm_id: vm_id.to_string(),
store,
clock,
}
}
pub fn append(&self, event: Event) {
let seq = self
.store
.read_history(&self.vm_id)
.map(|text| text.lines().count() as u64)
.unwrap_or(0);
let record = Record {
seq,
at: self.clock.unix_now().as_secs(),
event,
};
let Ok(line) = serde_json::to_string(&record) else {
return;
};
let _ = self.store.append_history(&self.vm_id, &line);
}
}
pub fn append_unseen_hooks(
history: &History,
observed: &[crate::protocol::health::HookObservation],
) {
if observed.is_empty() {
return;
}
let mut seen: std::collections::HashSet<(String, u64)> =
read_events(history.store.as_ref(), &history.vm_id)
.iter()
.filter(|event| event["event"] == "hookObserved")
.filter_map(|event| {
Some((
event["hook"].as_str()?.to_string(),
event["firedAt"].as_u64()?,
))
})
.collect();
for observation in observed {
if seen.insert((observation.hook.clone(), observation.fired_at)) {
history.append(Event::HookObserved {
hook: observation.hook.clone(),
fired_at: observation.fired_at,
});
}
}
}
pub fn read_events(store: &dyn RunStore, vm_id: &str) -> Vec<serde_json::Value> {
let Ok(text) = store.read_history(vm_id) else {
return Vec::new();
};
text.lines()
.enumerate()
.map(|(index, line)| {
serde_json::from_str::<serde_json::Value>(line)
.ok()
.filter(serde_json::Value::is_object)
.unwrap_or_else(|| {
serde_json::json!({
"line": index + 1,
"unreadable": line,
})
})
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testing::{MemoryRunStore, TestClock};
use serde_json::json;
use std::time::Duration;
#[test]
fn appends_preserve_old_lines_and_record_sequence_clock_and_event() {
let store = Arc::new(MemoryRunStore::default());
let original = "{\"seq\":0,\"at\":12,\"event\":\"resumed\"}\n";
store
.histories
.lock()
.unwrap()
.insert("vm".into(), original.into());
let clock = Arc::new(TestClock::new());
let now = clock.unix_now().as_secs();
let history = History::from_store("vm", store.clone(), clock.clone());
history.append(Event::Suspended);
clock.advance(Duration::from_secs(7));
history.append(Event::Terminated {
terminate_accepted: true,
undeleted: vec!["image".into()],
});
assert!(store.read_history("vm").unwrap().starts_with(original));
assert_eq!(
read_events(store.as_ref(), "vm"),
vec![
json!({"seq":0,"at":12,"event":"resumed"}),
json!({"seq":1,"at":now,"event":"suspended"}),
json!({"seq":2,"at":now+7,"event":"terminated","terminateAccepted":true,"undeleted":["image"]}),
]
);
}
#[test]
fn reading_preserves_objects_and_reports_each_malformed_line() {
let store = MemoryRunStore::default();
assert!(read_events(&store, "missing").is_empty());
store
.histories
.lock()
.unwrap()
.insert("vm".into(), "{\"seq\":0}\nbroken\nfalse\n\n".into());
assert_eq!(
read_events(&store, "vm"),
vec![
json!({"seq":0}),
json!({"line":2,"unreadable":"broken"}),
json!({"line":3,"unreadable":"false"}),
json!({"line":4,"unreadable":""}),
]
);
}
#[test]
fn observed_hooks_deduplicate_against_prior_lines_and_the_same_poll() {
let store = Arc::new(MemoryRunStore::default());
let history = History::from_store("vm", store.clone(), Arc::new(TestClock::new()));
append_unseen_hooks(&history, &[]);
assert!(store.histories.lock().unwrap().is_empty());
history.append(Event::HookObserved {
hook: "ready".into(),
fired_at: 100,
});
let observed = [
crate::protocol::health::HookObservation {
handler: None,
hook: "ready".into(),
fired_at: 100,
},
crate::protocol::health::HookObservation {
handler: None,
hook: "resume".into(),
fired_at: 101,
},
crate::protocol::health::HookObservation {
handler: None,
hook: "resume".into(),
fired_at: 101,
},
crate::protocol::health::HookObservation {
handler: None,
hook: "ready".into(),
fired_at: 102,
},
];
append_unseen_hooks(&history, &observed);
append_unseen_hooks(&history, &observed);
let events = read_events(store.as_ref(), "vm");
assert_eq!(events.len(), 3);
assert_eq!(
events
.iter()
.map(|event| (&event["hook"], &event["firedAt"]))
.collect::<Vec<_>>(),
vec![
(&json!("ready"), &json!(100)),
(&json!("resume"), &json!(101)),
(&json!("ready"), &json!(102))
]
);
}
}