use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use frame_core::component::ComponentId;
use frame_core::event::{
EventReceiveError, LifecycleEventKind, LifecycleState, LifecycleSubscription,
};
use frame_core::registry::ComponentRegistry;
use serde::Serialize;
use serde_json::Value;
use crate::error::HostError;
const TRUTH_EVENT_BUFFER: usize = 256;
const TRUTH_WAIT_QUANTUM: Duration = Duration::from_secs(3600);
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordedTransition {
pub sequence: u64,
pub component_id: String,
pub from: Option<LifecycleState>,
pub to: LifecycleState,
pub at_epoch_ms: u128,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordedFact {
pub at_epoch_ms: u128,
pub body: Value,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AppTruthSnapshot {
pub generated_at_epoch_ms: u128,
pub current: HashMap<String, LifecycleState>,
pub transitions: Vec<RecordedTransition>,
pub facts: Vec<RecordedFact>,
}
#[derive(Debug, Default)]
struct TruthState {
transitions: Vec<RecordedTransition>,
current: HashMap<ComponentId, LifecycleState>,
facts: Vec<RecordedFact>,
}
#[derive(Debug)]
pub struct AppTruth {
state: Mutex<TruthState>,
}
impl AppTruth {
#[must_use]
pub fn empty() -> Arc<Self> {
Arc::new(Self {
state: Mutex::new(TruthState::default()),
})
}
pub fn spawn(
registry: &ComponentRegistry,
expected: HashSet<ComponentId>,
) -> Result<(Arc<Self>, JoinHandle<()>), HostError> {
let capacity =
NonZeroUsize::new(TRUTH_EVENT_BUFFER).ok_or(HostError::SynchronizationPoisoned)?;
let subscription = registry.subscribe(capacity)?;
let truth = Self::empty();
let recorder = Arc::clone(&truth);
let join = thread::Builder::new()
.name("frame-host-app-truth".to_owned())
.spawn(move || record_stream(&recorder, &subscription, &expected))
.map_err(|source| HostError::TruthRecorderSpawn { source })?;
Ok((truth, join))
}
pub fn record_fact(&self, body: Value) {
let at_epoch_ms = epoch_ms();
match self.state.lock() {
Ok(mut state) => state.facts.push(RecordedFact { at_epoch_ms, body }),
Err(_) => {
tracing::error!(
"app-truth synchronization poisoned while recording an announced fact; the \
fact is real (already accepted by the announcer, and still published on the \
bus) but will not appear in the /frame/app/status.json snapshot"
);
}
}
}
#[must_use]
pub fn snapshot(&self) -> AppTruthSnapshot {
let generated_at_epoch_ms = epoch_ms();
let Ok(state) = self.state.lock() else {
tracing::error!(
"app-truth synchronization poisoned while assembling a snapshot; serving an \
empty snapshot rather than a fabricated one"
);
return AppTruthSnapshot {
generated_at_epoch_ms,
current: HashMap::new(),
transitions: Vec::new(),
facts: Vec::new(),
};
};
AppTruthSnapshot {
generated_at_epoch_ms,
current: state
.current
.iter()
.map(|(id, lifecycle_state)| (id.to_string(), *lifecycle_state))
.collect(),
transitions: state.transitions.clone(),
facts: state.facts.clone(),
}
}
fn record_transition(
&self,
sequence: u64,
component_id: ComponentId,
from: Option<LifecycleState>,
to: LifecycleState,
) {
let at_epoch_ms = epoch_ms();
match self.state.lock() {
Ok(mut state) => {
state.transitions.push(RecordedTransition {
sequence,
component_id: component_id.to_string(),
from,
to,
at_epoch_ms,
});
state.current.insert(component_id, to);
}
Err(_) => {
tracing::error!(
component = %component_id,
?to,
"app-truth synchronization poisoned while recording a lifecycle transition; \
it is real (already published on the registry's own stream) but will not \
appear in the /frame/app/status.json snapshot"
);
}
}
}
}
fn record_stream(
truth: &AppTruth,
subscription: &LifecycleSubscription,
expected: &HashSet<ComponentId>,
) {
let mut reported_lag = 0;
let mut removed: HashSet<ComponentId> = HashSet::new();
loop {
let event = match subscription.recv_timeout(TRUTH_WAIT_QUANTUM) {
Ok(event) => event,
Err(EventReceiveError::Timeout) => continue,
Err(EventReceiveError::Closed) => {
tracing::info!("lifecycle event stream closed; app-truth recorder exiting");
return;
}
Err(EventReceiveError::Poisoned) => {
tracing::error!(
"lifecycle event stream synchronization poisoned; app-truth recorder exiting"
);
return;
}
};
let lagged = subscription.lagged_events();
if lagged > reported_lag {
tracing::warn!(
dropped = lagged - reported_lag,
total_dropped = lagged,
"app-truth recorder overflowed its bounded relay buffer; oldest events were \
dropped (the retained snapshot itself has no cap — only this in-flight relay \
does)"
);
reported_lag = lagged;
}
if let LifecycleEventKind::Transition { from, to } = event.kind {
truth.record_transition(event.sequence, event.component_id, from, to);
if to == LifecycleState::Removed {
removed.insert(event.component_id);
if removed.is_superset(expected) {
tracing::info!(
"every application component removed; app-truth recorder exiting"
);
return;
}
}
}
}
}
fn epoch_ms() -> u128 {
let Ok(duration) = SystemTime::now().duration_since(UNIX_EPOCH) else {
tracing::error!(
"host wall clock reports a time before the Unix epoch; recording 0 rather than \
fabricating a timestamp"
);
return 0;
};
duration.as_millis()
}
#[cfg(test)]
mod tests {
use super::{AppTruth, epoch_ms};
use frame_core::component::ComponentId;
use frame_core::event::LifecycleState;
fn id() -> ComponentId {
ComponentId::derive("frame.host", "app-truth-test")
}
#[test]
fn epoch_ms_is_monotonic_enough_to_order_two_calls() {
let first = epoch_ms();
std::thread::sleep(std::time::Duration::from_millis(2));
let second = epoch_ms();
assert!(
second >= first,
"wall-clock reads must not run backwards in a tight loop"
);
}
#[test]
fn empty_snapshot_has_no_transitions_or_facts() {
let truth = AppTruth::empty();
let snapshot = truth.snapshot();
assert!(snapshot.transitions.is_empty());
assert!(snapshot.facts.is_empty());
assert!(snapshot.current.is_empty());
}
#[test]
fn recorded_transition_updates_current_and_appends_history() {
let truth = AppTruth::empty();
truth.record_transition(0, id(), None, LifecycleState::Registered);
truth.record_transition(
1,
id(),
Some(LifecycleState::Registered),
LifecycleState::Starting,
);
let snapshot = truth.snapshot();
assert_eq!(snapshot.transitions.len(), 2);
assert_eq!(snapshot.transitions[0].sequence, 0);
assert!(snapshot.transitions[0].from.is_none());
assert_eq!(snapshot.transitions[1].sequence, 1);
assert_eq!(
snapshot.current.get(&id().to_string()),
Some(&LifecycleState::Starting),
"current must fold to the LATEST transition, not the first"
);
}
#[test]
fn recorded_fact_is_retained_verbatim() {
let truth = AppTruth::empty();
truth.record_fact(serde_json::json!({ "entity": "e-1" }));
let snapshot = truth.snapshot();
assert_eq!(snapshot.facts.len(), 1);
assert_eq!(snapshot.facts[0].body["entity"], "e-1");
}
}