pub mod harness_adapt;
pub mod harness_metrics;
pub mod observability;
pub mod tool_receipts;
pub use observability::{
evaluate_alerts, summarize, summarize_log, Alert, AlertKind, AlertThresholds, MetricsSummary,
};
use car_secrets::{
atomic_replace_private_file, create_private_file, open_private_append, revalidate_private_file,
revalidate_private_path,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, HashSet, VecDeque};
use std::fs;
use std::future::Future;
use std::io::{BufRead, BufReader, BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::{mpsc, Arc, Condvar, Mutex, Weak};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};
use uuid::Uuid;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EventLogStats {
pub events: usize,
pub spans: usize,
pub approx_event_bytes: usize,
pub approx_span_bytes: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
RunStarted,
RunCancellationRequested,
RunCancellationResult,
ProposalReceived,
ProposalCompleted,
ActionValidated,
ActionRejected,
ActionExecuting,
ActionSucceeded,
ActionFailed,
ActionSkipped,
ActionRetrying,
ActionDeduplicated,
PolicyViolation,
StateChanged,
StateSnapshot,
StateCommitted,
StateRollback,
SkillDistilled,
SkillEvolved,
SkillDeprecated,
EvolutionTriggered,
CandidatePromoted,
CandidateRejected,
Consolidated,
ProactiveMemoryMaintained,
ProactiveMemoryIntervention,
ReplanAttempted,
ReplanProposalReceived,
ReplanRejected,
ReplanExhausted,
VoiceFastTurnStarted,
VoiceFastTurnEnded,
VoiceSidecarResolved,
VoiceSidecarFailed,
VoiceSidecarTimedOut,
VoiceTurnCancelled,
VoiceBridgePlayed,
GateAccepted,
GateRejected,
SessionScope,
PermissionDecision,
ApprovalRecorded,
BranchDecision,
AlternativeRejected,
InferenceMetered,
TransactionConflict,
AdmissionGateDecision,
ToolReceiptHallucination,
GoalEvaluated,
TurnCompleted,
RunCompleted,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum SpanStatus {
Ok,
Error,
Unset,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Span {
pub trace_id: String,
pub span_id: String,
pub parent_span_id: Option<String>,
pub name: String,
pub start_time: DateTime<Utc>,
pub end_time: Option<DateTime<Utc>>,
pub status: SpanStatus,
pub attributes: HashMap<String, Value>,
}
pub mod metric_keys {
pub const DURATION_MS: &str = "duration_ms";
pub const TOKENS_IN: &str = "tokens_in";
pub const TOKENS_OUT: &str = "tokens_out";
pub const COST_USD: &str = "cost_usd";
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
pub struct Metrics {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub duration_ms: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tokens_in: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tokens_out: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cost_usd: Option<f64>,
}
impl Metrics {
pub fn latency(duration_ms: f64) -> Self {
Self {
duration_ms: Some(duration_ms),
..Default::default()
}
}
pub fn inference(tokens_in: u64, tokens_out: u64, cost_usd: Option<f64>) -> Self {
Self {
duration_ms: None,
tokens_in: Some(tokens_in),
tokens_out: Some(tokens_out),
cost_usd,
}
}
pub fn with_duration(mut self, duration_ms: f64) -> Self {
self.duration_ms = Some(duration_ms);
self
}
fn merge_into(&self, data: &mut HashMap<String, Value>) {
if let Some(d) = self.duration_ms {
data.insert(metric_keys::DURATION_MS.into(), Value::from(d));
}
if let Some(t) = self.tokens_in {
data.insert(metric_keys::TOKENS_IN.into(), Value::from(t));
}
if let Some(t) = self.tokens_out {
data.insert(metric_keys::TOKENS_OUT.into(), Value::from(t));
}
if let Some(c) = self.cost_usd {
data.insert(metric_keys::COST_USD.into(), Value::from(c));
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Event {
pub kind: EventKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub run_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub client_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub policy_session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub proposal_id: Option<String>,
#[serde(default)]
pub data: HashMap<String, Value>,
#[serde(default = "Utc::now")]
pub timestamp: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prev_hash: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub hash: Option<String>,
}
impl Event {
pub fn duration_ms(&self) -> Option<f64> {
self.data
.get(metric_keys::DURATION_MS)
.and_then(Value::as_f64)
}
pub fn tokens_in(&self) -> Option<u64> {
self.data
.get(metric_keys::TOKENS_IN)
.and_then(Value::as_u64)
}
pub fn tokens_out(&self) -> Option<u64> {
self.data
.get(metric_keys::TOKENS_OUT)
.and_then(Value::as_u64)
}
pub fn cost_usd(&self) -> Option<f64> {
self.data.get(metric_keys::COST_USD).and_then(Value::as_f64)
}
pub fn metrics(&self) -> Metrics {
Metrics {
duration_ms: self.duration_ms(),
tokens_in: self.tokens_in(),
tokens_out: self.tokens_out(),
cost_usd: self.cost_usd(),
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
pub struct MetricsTotals {
pub duration_ms: f64,
pub tokens_in: u64,
pub tokens_out: u64,
pub tokens: u64,
pub cost_usd: f64,
pub metered_events: usize,
}
pub fn metrics_totals_of(events: &[Event]) -> MetricsTotals {
let mut totals = MetricsTotals::default();
for ev in events {
let m = ev.metrics();
let mut metered = false;
if let Some(d) = m.duration_ms {
totals.duration_ms += d;
metered = true;
}
if let Some(t) = m.tokens_in {
totals.tokens_in = totals.tokens_in.saturating_add(t);
metered = true;
}
if let Some(t) = m.tokens_out {
totals.tokens_out = totals.tokens_out.saturating_add(t);
metered = true;
}
if let Some(c) = m.cost_usd {
totals.cost_usd += c;
metered = true;
}
if metered {
totals.metered_events += 1;
}
}
totals.tokens = totals.tokens_in.saturating_add(totals.tokens_out);
totals
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct AgentCost {
pub agent: String,
pub calls: u64,
pub tokens_in: u64,
pub tokens_out: u64,
pub cost_usd: f64,
}
pub fn cost_by_agent_of(events: &[Event]) -> Vec<AgentCost> {
use std::collections::BTreeMap;
let mut map: BTreeMap<String, AgentCost> = BTreeMap::new();
for e in events {
if e.kind != EventKind::InferenceMetered {
continue;
}
let agent = e
.data
.get("agent")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string();
let entry = map.entry(agent.clone()).or_insert_with(|| AgentCost {
agent,
..Default::default()
});
entry.calls += 1;
entry.tokens_in = entry.tokens_in.saturating_add(e.tokens_in().unwrap_or(0));
entry.tokens_out = entry.tokens_out.saturating_add(e.tokens_out().unwrap_or(0));
entry.cost_usd += e.cost_usd().unwrap_or(0.0);
}
map.into_values().collect()
}
enum JournalMessage {
Async(String),
Critical {
line: String,
known_existing: bool,
ack: JournalAcknowledgement,
},
#[cfg(test)]
Shutdown,
}
pub const MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_CRITICAL_ACKNOWLEDGEMENT_CAPACITY: usize = 64;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JournalFailurePoint {
AsyncWrite,
Write,
Flush,
Fsync,
HoldAcknowledgement,
}
#[derive(Debug, Clone, Default)]
pub struct JournalFailureInjector {
failures: Arc<Mutex<VecDeque<JournalFailurePoint>>>,
held_acknowledgements: Arc<Mutex<Vec<JournalAcknowledgement>>>,
}
impl JournalFailureInjector {
pub fn fail_next(&self, point: JournalFailurePoint) {
self.failures
.lock()
.expect("journal failure injector mutex poisoned")
.push_back(point);
}
fn take(&self, point: JournalFailurePoint) -> bool {
let mut failures = self
.failures
.lock()
.expect("journal failure injector mutex poisoned");
if failures.front() == Some(&point) {
failures.pop_front();
true
} else {
false
}
}
fn hold_acknowledgement(&self, ack: JournalAcknowledgement) {
self.held_acknowledgements
.lock()
.expect("journal held-acknowledgement mutex poisoned")
.push(ack);
}
#[doc(hidden)]
pub fn held_acknowledgement_count(&self) -> usize {
self.held_acknowledgements
.lock()
.expect("journal held-acknowledgement mutex poisoned")
.len()
}
#[doc(hidden)]
pub fn release_held_acknowledgements(&self) {
let acknowledgements: Vec<_> = self
.held_acknowledgements
.lock()
.expect("journal held-acknowledgement mutex poisoned")
.drain(..)
.collect();
for acknowledgement in acknowledgements {
acknowledgement.send(Ok(()));
}
}
}
#[derive(Debug)]
enum CriticalPreAcceptanceError {
WriterUnavailable,
WriterStopped,
CoordinatorUnavailable(String),
CapacityExhausted { capacity: usize },
InvalidAcknowledgementTimeout { requested: Duration },
}
impl std::fmt::Display for CriticalPreAcceptanceError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::WriterUnavailable => write!(formatter, "journal writer thread is unavailable"),
Self::WriterStopped => {
write!(
formatter,
"journal writer stopped before accepting critical append"
)
}
Self::CoordinatorUnavailable(reason) => write!(
formatter,
"critical acknowledgement coordinator is unavailable: {reason}"
),
Self::CapacityExhausted { capacity } => write!(
formatter,
"critical acknowledgement capacity is exhausted ({capacity} in flight)"
),
Self::InvalidAcknowledgementTimeout { requested } => write!(
formatter,
"critical acknowledgement timeout must be between 1ns and {}ms, got {}ms",
MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT.as_millis(),
requested.as_millis()
),
}
}
}
#[derive(Debug)]
enum CriticalPostAcceptanceError {
DurabilityFailure(String),
AcknowledgementTimedOut { timeout: Duration },
CoordinatorStopped,
}
impl std::fmt::Display for CriticalPostAcceptanceError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::DurabilityFailure(reason) => write!(formatter, "{reason}"),
Self::AcknowledgementTimedOut { timeout } => write!(
formatter,
"journal writer did not acknowledge within {}ms",
timeout.as_millis()
),
Self::CoordinatorStopped => {
write!(
formatter,
"acknowledgement coordinator stopped after enqueue"
)
}
}
}
}
struct AsyncAcknowledgementState {
terminal: bool,
result: Option<Result<(), CriticalPostAcceptanceError>>,
waker: Option<Waker>,
}
struct AsyncAcknowledgementEntry {
id: u64,
deadline: Instant,
timeout: Duration,
manager: Weak<AsyncAcknowledgementManagerInner>,
state: Mutex<AsyncAcknowledgementState>,
}
impl AsyncAcknowledgementEntry {
fn complete(&self, result: Result<(), CriticalPostAcceptanceError>) {
self.complete_deciding(|| result);
}
fn complete_writer(&self, result: std::io::Result<()>) {
self.complete_deciding(|| {
if Instant::now() >= self.deadline {
Err(CriticalPostAcceptanceError::AcknowledgementTimedOut {
timeout: self.timeout,
})
} else {
result.map_err(|error| {
CriticalPostAcceptanceError::DurabilityFailure(error.to_string())
})
}
});
}
fn complete_deciding(&self, decide: impl FnOnce() -> Result<(), CriticalPostAcceptanceError>) {
let waker = {
let mut state = self
.state
.lock()
.expect("journal async-acknowledgement mutex poisoned");
if state.terminal {
return;
}
state.terminal = true;
state.result = Some(decide());
state.waker.take()
};
if let Some(manager) = self.manager.upgrade() {
manager.remove(self.id);
}
if let Some(waker) = waker {
waker.wake();
}
}
fn cancel_preacceptance(&self) {
{
let mut state = self
.state
.lock()
.expect("journal async-acknowledgement mutex poisoned");
if state.terminal {
return;
}
state.terminal = true;
}
if let Some(manager) = self.manager.upgrade() {
manager.remove(self.id);
}
}
}
struct AsyncAcknowledgement {
entry: Arc<AsyncAcknowledgementEntry>,
}
impl Future for AsyncAcknowledgement {
type Output = Result<(), CriticalPostAcceptanceError>;
fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
let mut state = self
.entry
.state
.lock()
.expect("journal async-acknowledgement mutex poisoned");
match state.result.take() {
Some(result) => Poll::Ready(result),
None => {
state.waker = Some(context.waker().clone());
Poll::Pending
}
}
}
}
struct AsyncAcknowledgementManagerState {
entries: HashMap<u64, Arc<AsyncAcknowledgementEntry>>,
next_id: u64,
shutting_down: bool,
}
struct AsyncAcknowledgementManagerInner {
capacity: usize,
state: Mutex<AsyncAcknowledgementManagerState>,
changed: Condvar,
#[cfg(test)]
expiry_barrier: Mutex<Option<AsyncAcknowledgementExpiryBarrier>>,
}
#[cfg(test)]
#[derive(Clone)]
struct AsyncAcknowledgementExpiryBarrier {
removed: Arc<std::sync::Barrier>,
release: Arc<std::sync::Barrier>,
}
#[cfg(test)]
impl AsyncAcknowledgementExpiryBarrier {
fn new() -> Self {
Self {
removed: Arc::new(std::sync::Barrier::new(2)),
release: Arc::new(std::sync::Barrier::new(2)),
}
}
fn pause_after_removal(&self) {
self.removed.wait();
self.release.wait();
}
fn wait_until_removed(&self) {
self.removed.wait();
}
fn allow_timeout_completion(&self) {
self.release.wait();
}
}
impl AsyncAcknowledgementManagerInner {
fn remove(&self, id: u64) {
let removed = self
.state
.lock()
.expect("journal acknowledgement-manager mutex poisoned")
.entries
.remove(&id)
.is_some();
if removed {
self.changed.notify_all();
}
}
}
struct AsyncAcknowledgementManager {
inner: Arc<AsyncAcknowledgementManagerInner>,
worker: Mutex<Option<thread::JoinHandle<()>>>,
}
impl AsyncAcknowledgementManager {
fn new(capacity: usize) -> Self {
Self {
inner: Arc::new(AsyncAcknowledgementManagerInner {
capacity,
state: Mutex::new(AsyncAcknowledgementManagerState {
entries: HashMap::new(),
next_id: 0,
shutting_down: false,
}),
changed: Condvar::new(),
#[cfg(test)]
expiry_barrier: Mutex::new(None),
}),
worker: Mutex::new(None),
}
}
#[cfg(test)]
fn pause_next_expiry_after_removal(&self, barrier: AsyncAcknowledgementExpiryBarrier) {
*self
.inner
.expiry_barrier
.lock()
.expect("journal acknowledgement expiry-barrier mutex poisoned") = Some(barrier);
}
fn ensure_worker(&self) -> Result<(), CriticalPreAcceptanceError> {
let mut worker = self
.worker
.lock()
.expect("journal acknowledgement-worker mutex poisoned");
if worker.is_some() {
return Ok(());
}
let inner = self.inner.clone();
let handle = thread::Builder::new()
.name("car-eventlog-critical-ack".into())
.spawn(move || async_acknowledgement_timer_loop(inner))
.map_err(|error| {
CriticalPreAcceptanceError::CoordinatorUnavailable(error.to_string())
})?;
*worker = Some(handle);
Ok(())
}
fn reserve(
&self,
timeout: Duration,
) -> Result<AsyncAcknowledgementReservation, CriticalPreAcceptanceError> {
if timeout.is_zero() || timeout > MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT {
return Err(CriticalPreAcceptanceError::InvalidAcknowledgementTimeout {
requested: timeout,
});
}
self.ensure_worker()?;
let mut state = self
.inner
.state
.lock()
.expect("journal acknowledgement-manager mutex poisoned");
if state.shutting_down {
return Err(CriticalPreAcceptanceError::CoordinatorUnavailable(
"coordinator is shutting down".to_string(),
));
}
if state.entries.len() >= self.inner.capacity {
return Err(CriticalPreAcceptanceError::CapacityExhausted {
capacity: self.inner.capacity,
});
}
let id = loop {
let candidate = state.next_id;
state.next_id = state.next_id.wrapping_add(1);
if !state.entries.contains_key(&candidate) {
break candidate;
}
};
let entry = Arc::new(AsyncAcknowledgementEntry {
id,
deadline: Instant::now() + timeout,
timeout,
manager: Arc::downgrade(&self.inner),
state: Mutex::new(AsyncAcknowledgementState {
terminal: false,
result: None,
waker: None,
}),
});
state.entries.insert(id, entry.clone());
drop(state);
self.inner.changed.notify_all();
Ok(AsyncAcknowledgementReservation {
entry,
preaccepted: true,
})
}
fn shutdown(&self) {
let entries = {
let mut state = self
.inner
.state
.lock()
.expect("journal acknowledgement-manager mutex poisoned");
state.shutting_down = true;
let entries = state
.entries
.drain()
.map(|(_, entry)| entry)
.collect::<Vec<_>>();
self.inner.changed.notify_all();
entries
};
for entry in entries {
entry.complete(Err(CriticalPostAcceptanceError::CoordinatorStopped));
}
if let Some(worker) = self
.worker
.lock()
.expect("journal acknowledgement-worker mutex poisoned")
.take()
{
let _ = worker.join();
}
}
}
fn async_acknowledgement_timer_loop(inner: Arc<AsyncAcknowledgementManagerInner>) {
loop {
let expired = {
let mut state = inner
.state
.lock()
.expect("journal acknowledgement-manager mutex poisoned");
loop {
if state.shutting_down {
return;
}
let now = Instant::now();
let expired_ids: Vec<_> = state
.entries
.iter()
.filter_map(|(id, entry)| (entry.deadline <= now).then_some(*id))
.collect();
if !expired_ids.is_empty() {
break expired_ids
.into_iter()
.filter_map(|id| state.entries.remove(&id))
.collect::<Vec<_>>();
}
if let Some(deadline) = state.entries.values().map(|entry| entry.deadline).min() {
let wait = deadline.saturating_duration_since(now);
let (next, _) = inner
.changed
.wait_timeout(state, wait)
.expect("journal acknowledgement-manager mutex poisoned");
state = next;
} else {
state = inner
.changed
.wait(state)
.expect("journal acknowledgement-manager mutex poisoned");
}
}
};
#[cfg(test)]
if !expired.is_empty() {
if let Some(barrier) = inner
.expiry_barrier
.lock()
.expect("journal acknowledgement expiry-barrier mutex poisoned")
.take()
{
barrier.pause_after_removal();
}
}
for entry in expired {
entry.complete(Err(CriticalPostAcceptanceError::AcknowledgementTimedOut {
timeout: entry.timeout,
}));
}
}
}
struct AsyncAcknowledgementReservation {
entry: Arc<AsyncAcknowledgementEntry>,
preaccepted: bool,
}
impl AsyncAcknowledgementReservation {
fn sender(&self) -> AsyncAcknowledgementSender {
AsyncAcknowledgementSender {
entry: self.entry.clone(),
}
}
fn into_future(mut self) -> AsyncAcknowledgement {
self.preaccepted = false;
AsyncAcknowledgement {
entry: self.entry.clone(),
}
}
}
impl Drop for AsyncAcknowledgementReservation {
fn drop(&mut self) {
if self.preaccepted {
self.entry.cancel_preacceptance();
}
}
}
struct AsyncAcknowledgementSender {
entry: Arc<AsyncAcknowledgementEntry>,
}
impl AsyncAcknowledgementSender {
fn send(&self, result: std::io::Result<()>) {
self.entry.complete_writer(result);
}
}
enum JournalAcknowledgement {
Sync(mpsc::SyncSender<std::io::Result<()>>),
Async(AsyncAcknowledgementSender),
}
impl JournalAcknowledgement {
fn send(&self, result: std::io::Result<()>) {
match self {
Self::Sync(sender) => {
let _ = sender.send(result);
}
Self::Async(sender) => sender.send(result),
}
}
}
impl std::fmt::Debug for JournalAcknowledgement {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Sync(_) => formatter.write_str("JournalAcknowledgement::Sync"),
Self::Async(_) => formatter.write_str("JournalAcknowledgement::Async"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CriticalAppendError {
Rejected { reason: String },
DurabilityUnknown { reason: String },
}
impl CriticalAppendError {
pub fn is_retry_safe(&self) -> bool {
matches!(self, Self::DurabilityUnknown { .. })
}
}
impl std::fmt::Display for CriticalAppendError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Rejected { reason } => {
write!(formatter, "critical journal append rejected: {reason}")
}
Self::DurabilityUnknown { reason } => write!(
formatter,
"critical journal durability is unknown; retry the exact event safely: {reason}"
),
}
}
}
impl std::error::Error for CriticalAppendError {}
struct JournalWriter {
tx: Option<mpsc::Sender<JournalMessage>>,
handle: Option<thread::JoinHandle<()>>,
acknowledgements: AsyncAcknowledgementManager,
}
impl JournalWriter {
fn spawn(path: PathBuf) -> Self {
Self::spawn_with_injectors(path, JournalFailureInjector::default(), None)
}
fn spawn_with_injector(path: PathBuf, failures: JournalFailureInjector) -> Self {
Self::spawn_with_injectors(path, failures, None)
}
fn spawn_with_private_path_injector(
path: PathBuf,
failures: car_secrets::PrivatePathDurabilityFailureInjector,
) -> Self {
Self::spawn_with_injectors(path, JournalFailureInjector::default(), Some(failures))
}
fn spawn_with_injectors(
path: PathBuf,
failures: JournalFailureInjector,
private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
) -> Self {
Self::spawn_with_injectors_and_ack_capacity(
path,
failures,
private_path_failures,
DEFAULT_CRITICAL_ACKNOWLEDGEMENT_CAPACITY,
)
}
fn spawn_with_injectors_and_ack_capacity(
path: PathBuf,
failures: JournalFailureInjector,
private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
acknowledgement_capacity: usize,
) -> Self {
let (tx, rx) = mpsc::channel::<JournalMessage>();
let acknowledgements = AsyncAcknowledgementManager::new(acknowledgement_capacity);
match thread::Builder::new()
.name("car-eventlog-journal".into())
.spawn(move || journal_loop(path, rx, failures, private_path_failures))
{
Ok(handle) => Self {
tx: Some(tx),
handle: Some(handle),
acknowledgements,
},
Err(e) => {
tracing::warn!(error = %e, "car-eventlog: failed to spawn journal writer thread — journaling disabled for this log");
Self {
tx: None,
handle: None,
acknowledgements,
}
}
}
}
fn send(&self, line: String) {
if let Some(tx) = &self.tx {
let _ = tx.send(JournalMessage::Async(line));
}
}
fn enqueue_critical_sync(
&self,
line: String,
known_existing: bool,
) -> Result<mpsc::Receiver<std::io::Result<()>>, CriticalPreAcceptanceError> {
let tx = self
.tx
.as_ref()
.ok_or(CriticalPreAcceptanceError::WriterUnavailable)?;
let (ack_tx, ack_rx) = mpsc::sync_channel(0);
tx.send(JournalMessage::Critical {
line,
known_existing,
ack: JournalAcknowledgement::Sync(ack_tx),
})
.map_err(|_| CriticalPreAcceptanceError::WriterStopped)?;
Ok(ack_rx)
}
fn reserve_async_acknowledgement(
&self,
acknowledgement_timeout: Duration,
) -> Result<AsyncAcknowledgementReservation, CriticalPreAcceptanceError> {
if self.tx.is_none() {
return Err(CriticalPreAcceptanceError::WriterUnavailable);
}
self.acknowledgements.reserve(acknowledgement_timeout)
}
fn enqueue_critical_async(
&self,
line: String,
known_existing: bool,
reservation: AsyncAcknowledgementReservation,
) -> Result<AsyncAcknowledgement, CriticalPreAcceptanceError> {
let tx = self
.tx
.as_ref()
.ok_or(CriticalPreAcceptanceError::WriterUnavailable)?;
tx.send(JournalMessage::Critical {
line,
known_existing,
ack: JournalAcknowledgement::Async(reservation.sender()),
})
.map_err(|_| CriticalPreAcceptanceError::WriterStopped)?;
Ok(reservation.into_future())
}
#[cfg(test)]
fn remove_sender_for_test(&mut self) {
self.tx.take();
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
#[cfg(test)]
fn stop_receiver_for_test(&mut self) {
if let Some(tx) = &self.tx {
let _ = tx.send(JournalMessage::Shutdown);
}
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
}
impl Drop for JournalWriter {
fn drop(&mut self) {
self.acknowledgements.shutdown();
self.tx.take();
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
}
}
fn journal_loop(
path: PathBuf,
rx: mpsc::Receiver<JournalMessage>,
failures: JournalFailureInjector,
private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
) {
let mut writer: Option<std::fs::File> = None;
let mut written_critical_lines = HashSet::new();
let mut blocked_critical: Option<String> = None;
let mut failed_async = VecDeque::new();
let mut after_blocked_critical = VecDeque::new();
while let Ok(message) = rx.recv() {
let existed_before_open = path.exists();
let open = || match private_path_failures.as_ref() {
Some(failures) => {
car_secrets::open_private_append_with_failure_injector(&path, failures)
}
None => open_private_append(&path),
};
match message {
#[cfg(test)]
JournalMessage::Shutdown => break,
JournalMessage::Async(line) => {
if blocked_critical.is_some() {
after_blocked_critical.push_back(line);
continue;
}
if !failed_async.is_empty() {
failed_async.push_back(line);
continue;
}
if writer.is_none() {
writer = open().ok();
}
let result = match writer.as_mut() {
Some(file) => append_journal_line(
&path,
file,
&line,
false,
false,
existed_before_open,
&failures,
&mut written_critical_lines,
),
None => Err(std::io::Error::other("cannot open journal file")),
};
if let Err(error) = result {
failed_async.push_back(line);
tracing::warn!(path = %path.display(), %error, "car-eventlog: asynchronous journal append failed and is awaiting ordered retry");
}
continue;
}
JournalMessage::Critical {
line,
known_existing,
ack,
} => {
if blocked_critical
.as_deref()
.is_some_and(|pending| pending != line)
{
ack.send(Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"another critical journal row is awaiting durability",
)));
continue;
}
if writer.is_none() {
writer = open().ok();
}
while let Some(pending) = failed_async.front() {
let replay = match writer.as_mut() {
Some(file) => append_journal_line(
&path,
file,
pending,
false,
false,
existed_before_open,
&failures,
&mut written_critical_lines,
),
None => Err(std::io::Error::other("cannot open journal file")),
};
match replay {
Ok(()) => {
failed_async.pop_front();
}
Err(error) => {
blocked_critical = Some(line.clone());
ack.send(Err(std::io::Error::new(
error.kind(),
format!("prior asynchronous journal row is not durable: {error}"),
)));
break;
}
}
}
if !failed_async.is_empty() {
continue;
}
let result = match writer.as_mut() {
Some(file) => append_journal_line(
&path,
file,
&line,
true,
known_existing,
existed_before_open,
&failures,
&mut written_critical_lines,
),
None => Err(std::io::Error::other("cannot open journal file")),
};
match result {
Ok(()) => {
blocked_critical = None;
while let Some(queued) = after_blocked_critical.pop_front() {
if let Some(file) = writer.as_mut() {
if let Err(error) = append_journal_line(
&path,
file,
&queued,
false,
false,
true,
&failures,
&mut written_critical_lines,
) {
failed_async.push_back(queued);
failed_async.append(&mut after_blocked_critical);
tracing::warn!(path = %path.display(), %error, "car-eventlog: queued asynchronous append failed after critical recovery and is awaiting ordered retry");
break;
}
}
}
if failures.take(JournalFailurePoint::HoldAcknowledgement) {
failures.hold_acknowledgement(ack);
} else {
ack.send(Ok(()));
}
}
Err(error) => {
blocked_critical = Some(line);
ack.send(Err(error));
}
}
continue;
}
};
}
if let Some(mut writer) = writer {
let _ = writer.flush();
}
}
fn append_journal_line(
path: &Path,
file: &mut std::fs::File,
line: &str,
critical: bool,
known_existing: bool,
existed_before_open: bool,
failures: &JournalFailureInjector,
written_critical_lines: &mut HashSet<String>,
) -> std::io::Result<()> {
revalidate_private_path(path, file)?;
let already_written = known_existing || written_critical_lines.contains(line);
if !already_written {
if (!critical && failures.take(JournalFailurePoint::AsyncWrite))
|| (critical && failures.take(JournalFailurePoint::Write))
{
return Err(std::io::Error::from_raw_os_error(28)); }
let mut bytes = Vec::with_capacity(line.len() + 2);
let len = file.seek(SeekFrom::End(0))?;
if len > 0 {
file.seek(SeekFrom::End(-1))?;
let mut tail = [0u8; 1];
file.read_exact(&mut tail)?;
if tail[0] != b'\n' {
bytes.push(b'\n');
}
}
bytes.extend_from_slice(line.as_bytes());
bytes.push(b'\n');
if let Err(error) = file.write_all(&bytes) {
let _ = file.set_len(len);
let _ = file.seek(SeekFrom::End(0));
return Err(error);
}
if critical {
written_critical_lines.insert(line.to_string());
}
}
if critical && failures.take(JournalFailurePoint::Flush) {
return Err(std::io::Error::other("injected journal flush failure"));
}
file.flush()?;
if critical {
if failures.take(JournalFailurePoint::Fsync) {
return Err(std::io::Error::other("injected journal fsync failure"));
}
file.sync_all()?;
if !existed_before_open {
sync_journal_parent(path)?;
}
}
revalidate_private_path(path, file)
}
#[cfg(not(target_os = "windows"))]
fn sync_journal_parent(path: &Path) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
std::fs::File::open(parent)?.sync_all()?;
}
Ok(())
}
#[cfg(target_os = "windows")]
fn sync_journal_parent(_path: &Path) -> std::io::Result<()> {
Ok(())
}
pub struct EventLog {
events: Vec<Event>,
spans: Vec<Span>,
journal: Option<JournalWriter>,
hash_chaining: bool,
last_hash: Option<String>,
retention: Option<RetentionPolicy>,
journal_path: Option<PathBuf>,
journal_lines: usize,
trimmed_events: u64,
cumulative_cost_usd: f64,
active_binding: Option<EventBinding>,
critical_pending: HashSet<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct EventBinding {
run_id: String,
client_id: String,
policy_session_id: Option<String>,
}
struct PreparedCriticalAppend {
existing_index: Option<usize>,
event: Option<Event>,
line: String,
known_existing: bool,
}
const JOURNAL_COMPACT_MIN_EXCESS: usize = 1024;
fn event_digest(
prev_hash: &str,
kind: &EventKind,
run_id: Option<&str>,
client_id: Option<&str>,
policy_session_id: Option<&str>,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: &HashMap<String, Value>,
timestamp: &DateTime<Utc>,
) -> String {
use sha2::{Digest, Sha256};
let mut sorted: Vec<(&String, &Value)> = data.iter().collect();
sorted.sort_by(|a, b| a.0.cmp(b.0));
let data_canon: String = sorted
.iter()
.map(|(k, v)| format!("{k}={}", v))
.collect::<Vec<_>>()
.join("\u{1f}");
let kind_str = serde_json::to_string(kind).unwrap_or_default();
let mut hasher = Sha256::new();
hasher.update(prev_hash.as_bytes());
hasher.update(b"\x1e");
hasher.update(kind_str.as_bytes());
if run_id.is_some() || client_id.is_some() || policy_session_id.is_some() {
hasher.update(b"\x1d");
hasher.update(run_id.unwrap_or("").as_bytes());
hasher.update(b"\x1f");
hasher.update(client_id.unwrap_or("").as_bytes());
hasher.update(b"\x1f");
hasher.update(policy_session_id.unwrap_or("").as_bytes());
}
hasher.update(b"\x1e");
hasher.update(action_id.unwrap_or("").as_bytes());
hasher.update(b"\x1e");
hasher.update(proposal_id.unwrap_or("").as_bytes());
hasher.update(b"\x1e");
hasher.update(data_canon.as_bytes());
hasher.update(b"\x1e");
hasher.update(timestamp.to_rfc3339().as_bytes());
let digest = hasher.finalize();
digest.iter().map(|b| format!("{b:02x}")).collect()
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct RetentionPolicy {
#[serde(default)]
pub max_events: Option<usize>,
#[serde(default)]
pub max_age_secs: Option<i64>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct EventQuery {
#[serde(default)]
pub kinds: Vec<EventKind>,
#[serde(default)]
pub action_id: Option<String>,
#[serde(default)]
pub proposal_id: Option<String>,
#[serde(default)]
pub since: Option<DateTime<Utc>>,
#[serde(default)]
pub until: Option<DateTime<Utc>>,
#[serde(default)]
pub data_matches: std::collections::HashMap<String, String>,
#[serde(default)]
pub limit: Option<usize>,
}
fn data_value_matches(v: &Value, want: &str) -> bool {
match v {
Value::String(s) => s == want,
Value::Null => false,
other => *other == want,
}
}
impl EventQuery {
pub fn matches(&self, e: &Event) -> bool {
if !self.kinds.is_empty() && !self.kinds.contains(&e.kind) {
return false;
}
if let Some(aid) = &self.action_id {
if e.action_id.as_deref() != Some(aid.as_str()) {
return false;
}
}
if let Some(pid) = &self.proposal_id {
if e.proposal_id.as_deref() != Some(pid.as_str()) {
return false;
}
}
if let Some(since) = self.since {
if e.timestamp < since {
return false;
}
}
if let Some(until) = self.until {
if e.timestamp >= until {
return false;
}
}
for (k, want) in &self.data_matches {
match e.data.get(k) {
Some(v) if data_value_matches(v, want) => {}
_ => return false,
}
}
true
}
}
impl EventLog {
pub fn new() -> Self {
Self {
events: Vec::new(),
spans: Vec::new(),
journal: None,
hash_chaining: false,
last_hash: None,
retention: None,
journal_path: None,
journal_lines: 0,
trimmed_events: 0,
cumulative_cost_usd: 0.0,
active_binding: None,
critical_pending: HashSet::new(),
}
}
pub fn with_journal(path: PathBuf) -> Self {
Self {
events: Vec::new(),
spans: Vec::new(),
journal: Some(JournalWriter::spawn(path.clone())),
hash_chaining: false,
last_hash: None,
retention: None,
journal_path: Some(path),
journal_lines: 0,
trimmed_events: 0,
cumulative_cost_usd: 0.0,
active_binding: None,
critical_pending: HashSet::new(),
}
}
pub fn with_journal_failure_injector(path: PathBuf, failures: JournalFailureInjector) -> Self {
let mut log = Self::with_journal(path.clone());
log.journal = Some(JournalWriter::spawn_with_injector(path, failures));
log
}
#[cfg(test)]
fn with_journal_failure_injector_and_ack_capacity(
path: PathBuf,
failures: JournalFailureInjector,
acknowledgement_capacity: usize,
) -> Self {
let mut log = Self::with_journal(path.clone());
log.journal = Some(JournalWriter::spawn_with_injectors_and_ack_capacity(
path,
failures,
None,
acknowledgement_capacity,
));
log
}
pub fn with_private_path_failure_injector(
path: PathBuf,
failures: car_secrets::PrivatePathDurabilityFailureInjector,
) -> Self {
let mut log = Self::with_journal(path.clone());
log.journal = Some(JournalWriter::spawn_with_private_path_injector(
path, failures,
));
log
}
pub fn bind_run(&mut self, run_id: &str, client_id: &str) -> Result<(), String> {
if run_id.is_empty() || client_id.is_empty() {
return Err("active journal binding requires non-empty run_id and client_id".into());
}
match &self.active_binding {
Some(binding) if binding.run_id == run_id && binding.client_id == client_id => Ok(()),
Some(binding) => Err(format!(
"journal is already bound to run_id `{}` and client_id `{}`",
binding.run_id, binding.client_id
)),
None => {
self.active_binding = Some(EventBinding {
run_id: run_id.to_string(),
client_id: client_id.to_string(),
policy_session_id: None,
});
Ok(())
}
}
}
pub fn bind_policy_session(&mut self, policy_session_id: &str) -> Result<(), String> {
if policy_session_id.is_empty() {
return Err("policy_session_id must be non-empty".into());
}
let binding = self
.active_binding
.as_mut()
.ok_or_else(|| "cannot bind a policy session without an active run".to_string())?;
match binding.policy_session_id.as_deref() {
Some(existing) if existing != policy_session_id => Err(format!(
"journal proposal is already bound to policy_session_id `{existing}`"
)),
_ => {
binding.policy_session_id = Some(policy_session_id.to_string());
Ok(())
}
}
}
pub fn clear_policy_session(&mut self, policy_session_id: &str) -> Result<(), String> {
let binding = self
.active_binding
.as_mut()
.ok_or_else(|| "cannot clear a policy session without an active run".to_string())?;
if binding.policy_session_id.as_deref() != Some(policy_session_id) {
return Err("policy_session_id does not match the active journal binding".into());
}
binding.policy_session_id = None;
Ok(())
}
pub fn clear_run_binding(&mut self, run_id: &str, client_id: &str) -> Result<(), String> {
let binding = self
.active_binding
.as_ref()
.ok_or_else(|| "journal has no active run binding".to_string())?;
if binding.run_id != run_id || binding.client_id != client_id {
return Err("run_id/client_id does not match the active journal binding".into());
}
if binding.policy_session_id.is_some() {
return Err(
"cannot clear an active run while a proposal policy session is bound".into(),
);
}
self.active_binding = None;
Ok(())
}
pub fn active_run_binding(&self) -> Option<(&str, &str, Option<&str>)> {
self.active_binding.as_ref().map(|binding| {
(
binding.run_id.as_str(),
binding.client_id.as_str(),
binding.policy_session_id.as_deref(),
)
})
}
pub fn with_hash_chaining(mut self) -> Self {
self.enable_hash_chaining();
self
}
pub fn enable_hash_chaining(&mut self) {
self.hash_chaining = true;
if self.last_hash.is_none() {
self.last_hash = self.events.last().and_then(|e| e.hash.clone());
}
}
pub fn hash_chaining_enabled(&self) -> bool {
self.hash_chaining
}
pub fn append(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: HashMap<String, Value>,
) -> &Event {
let timestamp = Utc::now();
let (prev_hash, hash) = if self.hash_chaining {
let prev = self.last_hash.clone().unwrap_or_default();
let binding = self.active_binding.as_ref();
let h = event_digest(
&prev,
&kind,
binding.map(|b| b.run_id.as_str()),
binding.map(|b| b.client_id.as_str()),
binding.and_then(|b| b.policy_session_id.as_deref()),
action_id,
proposal_id,
&data,
×tamp,
);
self.last_hash = Some(h.clone());
(Some(prev), Some(h))
} else {
(None, None)
};
let event = Event {
kind,
run_id: self.active_binding.as_ref().map(|b| b.run_id.clone()),
client_id: self.active_binding.as_ref().map(|b| b.client_id.clone()),
policy_session_id: self
.active_binding
.as_ref()
.and_then(|b| b.policy_session_id.clone()),
action_id: action_id.map(|s| s.to_string()),
proposal_id: proposal_id.map(|s| s.to_string()),
data,
timestamp,
prev_hash,
hash,
};
if let Some(journal) = &self.journal {
if let Ok(json) = serde_json::to_string(&event) {
journal.send(json);
self.journal_lines += 1;
}
}
if let Some(c) = event.cost_usd() {
self.cumulative_cost_usd += c;
}
self.events.push(event);
if let Some(max) = self.retention.as_ref().and_then(|p| p.max_events) {
if self.events.len() > max {
let removed = truncate_vec_keep_last(&mut self.events, max);
self.trimmed_events += removed as u64;
self.maybe_compact_journal();
}
}
self.events.last().unwrap()
}
fn prepare_critical_append(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: HashMap<String, Value>,
) -> Result<PreparedCriticalAppend, String> {
let binding = self.active_binding.as_ref().ok_or_else(|| {
"critical lifecycle event requires an authenticated run binding".to_string()
})?;
let existing = self.events.iter().position(|event| {
event.kind == kind
&& event.run_id.as_deref() == Some(binding.run_id.as_str())
&& event.client_id.as_deref() == Some(binding.client_id.as_str())
&& event.policy_session_id.as_deref() == binding.policy_session_id.as_deref()
&& event.action_id.as_deref() == action_id
&& event.proposal_id.as_deref() == proposal_id
&& event.data == data
});
if existing.is_none() && !self.critical_pending.is_empty() {
return Err(
"another critical lifecycle event is awaiting an exact durability retry".into(),
);
}
if let Some(index) = existing {
let line = serde_json::to_string(&self.events[index]).map_err(|e| e.to_string())?;
return Ok(PreparedCriticalAppend {
existing_index: Some(index),
event: None,
known_existing: !self.critical_pending.contains(&line),
line,
});
}
let timestamp = Utc::now();
let (prev_hash, hash) = if self.hash_chaining {
let prev = self.last_hash.clone().unwrap_or_default();
let hash = event_digest(
&prev,
&kind,
Some(binding.run_id.as_str()),
Some(binding.client_id.as_str()),
binding.policy_session_id.as_deref(),
action_id,
proposal_id,
&data,
×tamp,
);
(Some(prev), Some(hash))
} else {
(None, None)
};
let event = Event {
kind,
run_id: Some(binding.run_id.clone()),
client_id: Some(binding.client_id.clone()),
policy_session_id: binding.policy_session_id.clone(),
action_id: action_id.map(str::to_string),
proposal_id: proposal_id.map(str::to_string),
data,
timestamp,
prev_hash,
hash,
};
let line = serde_json::to_string(&event).map_err(|error| error.to_string())?;
Ok(PreparedCriticalAppend {
existing_index: None,
event: Some(event),
line,
known_existing: false,
})
}
fn commit_prepared_critical(&mut self, prepared: PreparedCriticalAppend) -> (usize, String) {
let index = match prepared.existing_index {
Some(index) => index,
None => {
let event = prepared
.event
.expect("new critical append must carry its prepared event");
if self.hash_chaining {
self.last_hash = event.hash.clone();
}
self.events.push(event);
self.journal_lines += 1;
self.events.len() - 1
}
};
(index, prepared.line)
}
pub fn append_critical(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: HashMap<String, Value>,
) -> Result<&Event, String> {
if self.journal.is_none() {
return Err("critical lifecycle event requires an enabled journal".to_string());
}
let prepared = self.prepare_critical_append(kind, action_id, proposal_id, data)?;
let acknowledgement = self
.journal
.as_ref()
.expect("journal presence checked above")
.enqueue_critical_sync(prepared.line.clone(), prepared.known_existing)
.map_err(|error| error.to_string())?;
let (index, line) = self.commit_prepared_critical(prepared);
self.critical_pending.insert(line.clone());
let result = acknowledgement
.recv()
.map_err(|_| {
std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"journal writer stopped before critical acknowledgement",
)
})
.and_then(|result| result);
match result {
Ok(()) => {
self.critical_pending.remove(&line);
Ok(&self.events[index])
}
Err(error) => {
self.critical_pending.insert(line);
Err(format!("critical journal append was not durable: {error}"))
}
}
}
pub fn append_critical_bounded(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: HashMap<String, Value>,
acknowledgement_timeout: Duration,
) -> Result<&Event, CriticalAppendError> {
if acknowledgement_timeout.is_zero()
|| acknowledgement_timeout > MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT
{
return Err(CriticalAppendError::Rejected {
reason: CriticalPreAcceptanceError::InvalidAcknowledgementTimeout {
requested: acknowledgement_timeout,
}
.to_string(),
});
}
let prepared = self
.prepare_critical_append(kind, action_id, proposal_id, data)
.map_err(|reason| CriticalAppendError::Rejected { reason })?;
let acknowledgement = self
.journal
.as_ref()
.ok_or_else(|| CriticalAppendError::Rejected {
reason: "critical lifecycle event requires an enabled journal".to_string(),
})?
.enqueue_critical_sync(prepared.line.clone(), prepared.known_existing)
.map_err(|error| CriticalAppendError::Rejected {
reason: error.to_string(),
})?;
let (index, line) = self.commit_prepared_critical(prepared);
self.critical_pending.insert(line.clone());
let result = match acknowledgement.recv_timeout(acknowledgement_timeout) {
Ok(result) => result.map_err(|error| error.to_string()),
Err(mpsc::RecvTimeoutError::Timeout) => Err(format!(
"journal writer did not acknowledge within {}ms",
acknowledgement_timeout.as_millis()
)),
Err(mpsc::RecvTimeoutError::Disconnected) => {
Err("journal writer stopped before critical acknowledgement".to_string())
}
};
match result {
Ok(()) => {
self.critical_pending.remove(&line);
Ok(&self.events[index])
}
Err(reason) => Err(CriticalAppendError::DurabilityUnknown { reason }),
}
}
pub async fn append_critical_async(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
data: HashMap<String, Value>,
acknowledgement_timeout: Duration,
) -> Result<&Event, CriticalAppendError> {
let reservation = self
.journal
.as_ref()
.ok_or_else(|| CriticalAppendError::Rejected {
reason: "critical lifecycle event requires an enabled journal".to_string(),
})?
.reserve_async_acknowledgement(acknowledgement_timeout)
.map_err(|error| CriticalAppendError::Rejected {
reason: error.to_string(),
})?;
let prepared = self
.prepare_critical_append(kind, action_id, proposal_id, data)
.map_err(|reason| CriticalAppendError::Rejected { reason })?;
let acknowledgement = self
.journal
.as_ref()
.expect("journal presence checked before reservation")
.enqueue_critical_async(prepared.line.clone(), prepared.known_existing, reservation)
.map_err(|error| CriticalAppendError::Rejected {
reason: error.to_string(),
})?;
let (index, line) = self.commit_prepared_critical(prepared);
self.critical_pending.insert(line.clone());
match acknowledgement.await {
Ok(()) => {
self.critical_pending.remove(&line);
Ok(&self.events[index])
}
Err(error) => Err(CriticalAppendError::DurabilityUnknown {
reason: error.to_string(),
}),
}
}
pub fn verify_chain(&self) -> Result<usize, usize> {
let mut prev = String::new();
let mut verified = 0usize;
let mut chain_started = false;
for (i, ev) in self.events.iter().enumerate() {
let Some(stored) = &ev.hash else {
if chain_started {
return Err(i);
}
continue;
};
let recorded_prev = ev.prev_hash.clone().unwrap_or_default();
if chain_started && recorded_prev != prev {
return Err(i);
}
let recomputed = event_digest(
&recorded_prev,
&ev.kind,
ev.run_id.as_deref(),
ev.client_id.as_deref(),
ev.policy_session_id.as_deref(),
ev.action_id.as_deref(),
ev.proposal_id.as_deref(),
&ev.data,
&ev.timestamp,
);
if &recomputed != stored {
return Err(i);
}
prev = stored.clone();
chain_started = true;
verified += 1;
}
Ok(verified)
}
pub fn append_metered(
&mut self,
kind: EventKind,
action_id: Option<&str>,
proposal_id: Option<&str>,
mut data: HashMap<String, Value>,
metrics: Metrics,
) -> &Event {
metrics.merge_into(&mut data);
self.append(kind, action_id, proposal_id, data)
}
pub fn metrics_totals(&self) -> MetricsTotals {
metrics_totals_of(&self.events)
}
pub fn cost_by_agent(&self) -> Vec<AgentCost> {
cost_by_agent_of(&self.events)
}
pub fn events(&self) -> &[Event] {
&self.events
}
pub fn len(&self) -> usize {
self.events.len()
}
pub fn span_len(&self) -> usize {
self.spans.len()
}
pub fn is_empty(&self) -> bool {
self.events.is_empty()
}
pub fn stats(&self) -> EventLogStats {
EventLogStats {
events: self.events.len(),
spans: self.spans.len(),
approx_event_bytes: approx_json_bytes(&self.events),
approx_span_bytes: approx_json_bytes(&self.spans),
}
}
pub fn truncate_events_keep_last(&mut self, keep_last: usize) -> usize {
let removed = truncate_vec_keep_last(&mut self.events, keep_last);
self.trimmed_events += removed as u64;
if removed > 0 {
self.maybe_compact_journal();
}
removed
}
pub fn truncate_spans_keep_last(&mut self, keep_last: usize) -> usize {
truncate_vec_keep_last(&mut self.spans, keep_last)
}
pub fn clear(&mut self) -> EventLogStats {
let removed = self.stats();
self.trimmed_events += removed.events as u64;
self.events.clear();
self.events.shrink_to_fit();
self.spans.clear();
self.spans.shrink_to_fit();
removed
}
pub fn trimmed_events(&self) -> u64 {
self.trimmed_events
}
pub fn cumulative_cost_usd(&self) -> f64 {
self.cumulative_cost_usd
}
pub fn journal_size_bytes(&self) -> Option<u64> {
let path = self.journal_path.as_ref()?;
fs::metadata(path).ok().map(|m| m.len())
}
fn maybe_compact_journal(&mut self) {
if self.journal_path.is_none() {
return;
}
let excess = self.journal_lines.saturating_sub(self.events.len());
if excess >= JOURNAL_COMPACT_MIN_EXCESS && excess.saturating_mul(4) >= self.journal_lines {
self.compact_journal();
}
}
pub fn compact_journal(&mut self) -> bool {
let Some(path) = self.journal_path.clone() else {
return false;
};
let compacted_lines: HashSet<String> = self
.events
.iter()
.filter_map(|event| serde_json::to_string(event).ok())
.collect();
self.journal = None;
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("journal");
let tmp = path.with_file_name(format!(".{file_name}.compact-{}.tmp", Uuid::new_v4()));
let rewrite = (|| -> std::io::Result<()> {
let file = create_private_file(&tmp)?;
let mut writer = BufWriter::new(file);
for ev in &self.events {
let line = serde_json::to_string(ev).map_err(std::io::Error::other)?;
writeln!(writer, "{line}")?;
}
writer.flush()?;
let file = writer.into_inner().map_err(|error| error.into_error())?;
file.sync_all()?;
revalidate_private_file(&file)?;
drop(file);
atomic_replace_private_file(&tmp, &path)
})();
let ok = match rewrite {
Ok(()) => {
self.journal_lines = self.events.len();
self.critical_pending
.retain(|line| !compacted_lines.contains(line));
true
}
Err(e) => {
let _ = fs::remove_file(&tmp);
tracing::warn!(
path = %path.display(), error = %e,
"car-eventlog: journal compaction failed — journal keeps growing until the next successful compaction"
);
false
}
};
self.journal = Some(JournalWriter::spawn(path));
ok
}
pub fn query(&self, query: &EventQuery) -> Vec<&Event> {
let mut out: Vec<&Event> = self.events.iter().filter(|e| query.matches(e)).collect();
out.reverse(); if let Some(limit) = query.limit.filter(|l| *l > 0) {
out.truncate(limit);
}
out
}
pub fn set_retention(&mut self, policy: Option<RetentionPolicy>) {
self.retention = policy;
}
pub fn retention(&self) -> Option<&RetentionPolicy> {
self.retention.as_ref()
}
pub fn enforce_retention(&mut self, policy: &RetentionPolicy, now: DateTime<Utc>) -> usize {
let before = self.events.len();
if let Some(age) = policy.max_age_secs {
let cutoff = now - chrono::Duration::seconds(age);
self.events.retain(|e| e.timestamp >= cutoff);
}
if let Some(max) = policy.max_events {
truncate_vec_keep_last(&mut self.events, max);
}
let removed = before.saturating_sub(self.events.len());
self.trimmed_events += removed as u64;
if removed > 0 {
self.maybe_compact_journal();
}
removed
}
pub fn filter(&self, kind: Option<&EventKind>, action_id: Option<&str>) -> Vec<&Event> {
self.events
.iter()
.filter(|e| {
if let Some(k) = kind {
if &e.kind != k {
return false;
}
}
if let Some(aid) = action_id {
if e.action_id.as_deref() != Some(aid) {
return false;
}
}
true
})
.collect()
}
pub fn begin_span(
&mut self,
name: &str,
trace_id: &str,
parent_span_id: Option<&str>,
attributes: HashMap<String, Value>,
) -> String {
let span_id = Uuid::new_v4().to_string();
let span = Span {
trace_id: trace_id.to_string(),
span_id: span_id.clone(),
parent_span_id: parent_span_id.map(|s| s.to_string()),
name: name.to_string(),
start_time: Utc::now(),
end_time: None,
status: SpanStatus::Unset,
attributes,
};
self.spans.push(span);
span_id
}
pub fn end_span(&mut self, span_id: &str, status: SpanStatus) {
if let Some(span) = self.spans.iter_mut().find(|s| s.span_id == span_id) {
span.end_time = Some(Utc::now());
span.status = status;
}
}
pub fn spans(&self) -> Vec<Span> {
self.spans.clone()
}
pub fn export_traces(&self) -> String {
let mut traces: HashMap<&str, Vec<&Span>> = HashMap::new();
for span in &self.spans {
traces.entry(span.trace_id.as_str()).or_default().push(span);
}
let resource_spans: Vec<Value> = traces.into_values().map(|spans| {
let scope_spans = spans
.iter()
.map(|s| {
let mut span_obj = serde_json::json!({
"traceId": s.trace_id,
"spanId": s.span_id,
"name": s.name,
"startTimeUnixNano": s.start_time.timestamp_nanos_opt().unwrap_or(0).to_string(),
"status": {
"code": match s.status {
SpanStatus::Ok => 1,
SpanStatus::Error => 2,
SpanStatus::Unset => 0,
}
},
"attributes": s.attributes.iter().map(|(k, v)| {
serde_json::json!({
"key": k,
"value": { "stringValue": v.to_string() }
})
}).collect::<Vec<_>>(),
});
if let Some(ref parent) = s.parent_span_id {
span_obj.as_object_mut().unwrap().insert(
"parentSpanId".to_string(),
Value::from(parent.as_str()),
);
}
if let Some(end) = s.end_time {
span_obj.as_object_mut().unwrap().insert(
"endTimeUnixNano".to_string(),
Value::from(end.timestamp_nanos_opt().unwrap_or(0).to_string()),
);
}
span_obj
})
.collect::<Vec<_>>();
serde_json::json!({
"resource": {
"attributes": [
{ "key": "service.name", "value": { "stringValue": "car-runtime" } }
]
},
"scopeSpans": [{
"scope": { "name": "car-eventlog" },
"spans": scope_spans
}]
})
})
.collect();
serde_json::to_string(&serde_json::json!({
"resourceSpans": resource_spans
}))
.unwrap_or_else(|_| "{}".to_string())
}
pub fn load(path: &Path) -> std::io::Result<Self> {
Self::load_with_writer(path, JournalWriter::spawn(path.to_path_buf()))
}
pub fn load_read_only(path: &Path) -> std::io::Result<Self> {
Self::load_from_journal(path, None, false)
}
#[doc(hidden)]
pub fn load_with_journal_failure_injector(
path: &Path,
failures: JournalFailureInjector,
) -> std::io::Result<Self> {
Self::load_with_writer(
path,
JournalWriter::spawn_with_injector(path.to_path_buf(), failures),
)
}
fn load_with_writer(path: &Path, writer: JournalWriter) -> std::io::Result<Self> {
Self::load_from_journal(path, Some(writer), true)
}
fn load_from_journal(
path: &Path,
writer: Option<JournalWriter>,
repair_torn_tail: bool,
) -> std::io::Result<Self> {
let file = fs::File::open(path)?;
let mut reader = BufReader::new(file);
let mut events = Vec::new();
let mut event_lines = Vec::new();
let mut line_bytes = Vec::new();
let mut line_number = 0usize;
let mut torn_tail_line = None;
loop {
line_bytes.clear();
let bytes_read = reader.read_until(b'\n', &mut line_bytes)?;
if bytes_read == 0 {
break;
}
line_number += 1;
let terminated = line_bytes.ends_with(b"\n");
if line_bytes.iter().all(|byte| byte.is_ascii_whitespace()) {
if !terminated {
torn_tail_line = Some(line_number);
}
continue;
}
match serde_json::from_slice::<Event>(&line_bytes) {
Ok(event) => {
events.push(event);
event_lines.push(line_number);
if !terminated {
torn_tail_line = Some(line_number);
}
}
Err(_) if !terminated => {
torn_tail_line = Some(line_number);
}
Err(error) => {
return Err(invalid_journal_data(path, line_number, error));
}
}
if !terminated {
break;
}
}
drop(reader);
let last_hash = events.last().and_then(|e| e.hash.clone());
let hash_chaining = last_hash.is_some();
let cumulative_cost_usd = events.iter().filter_map(Event::cost_usd).sum();
let journal_lines = events.len();
let loaded = Self {
events,
spans: Vec::new(),
journal: writer,
hash_chaining,
last_hash,
retention: None,
journal_path: repair_torn_tail.then(|| path.to_path_buf()),
journal_lines,
trimmed_events: 0,
cumulative_cost_usd,
active_binding: None,
critical_pending: HashSet::new(),
};
if let Err(index) = loaded.verify_chain() {
let source_line = event_lines.get(index).copied().unwrap_or(index + 1);
return Err(invalid_journal_data(
path,
source_line,
"hash chain integrity check failed",
));
}
if let Some(line_number) = torn_tail_line {
if !repair_torn_tail {
return Err(torn_journal_tail(path, line_number));
}
rewrite_loaded_journal(path, &loaded.events)?;
}
Ok(loaded)
}
}
fn torn_journal_tail(path: &Path, line_number: usize) -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
format!(
"event journal torn tail: path={} line={} reason=unterminated final record",
path.display(),
line_number
),
)
}
fn invalid_journal_data(
path: &Path,
line_number: usize,
reason: impl std::fmt::Display,
) -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"event journal corruption: path={} line={} reason={reason}",
path.display(),
line_number
),
)
}
fn rewrite_loaded_journal(path: &Path, events: &[Event]) -> std::io::Result<()> {
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("journal");
let temp = path.with_file_name(format!(".{file_name}.recover-{}.tmp", Uuid::new_v4()));
let result = (|| {
let file = create_private_file(&temp)?;
let mut output = BufWriter::new(file);
for event in events {
let line = serde_json::to_string(event).map_err(std::io::Error::other)?;
writeln!(output, "{line}")?;
}
output.flush()?;
let file = output.into_inner().map_err(|error| error.into_error())?;
file.sync_all()?;
revalidate_private_file(&file)?;
drop(file);
atomic_replace_private_file(&temp, path)
})();
if result.is_err() {
let _ = fs::remove_file(&temp);
}
result
}
fn approx_json_bytes<T: Serialize>(value: &T) -> usize {
serde_json::to_vec(value)
.map(|bytes| bytes.len())
.unwrap_or(0)
}
fn truncate_vec_keep_last<T>(items: &mut Vec<T>, keep_last: usize) -> usize {
let len = items.len();
if len <= keep_last {
return 0;
}
let removed = len - keep_last;
items.drain(..removed);
items.shrink_to_fit();
removed
}
impl Default for EventLog {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
struct ThreadWake(std::thread::Thread);
impl std::task::Wake for ThreadWake {
fn wake(self: Arc<Self>) {
self.0.unpark();
}
fn wake_by_ref(self: &Arc<Self>) {
self.0.unpark();
}
}
fn test_waker() -> std::task::Waker {
std::task::Waker::from(Arc::new(ThreadWake(std::thread::current())))
}
fn block_on_test_future<F: std::future::Future>(future: F) -> F::Output {
let waker = test_waker();
let mut context = std::task::Context::from_waker(&waker);
let mut future = Box::pin(future);
loop {
match future.as_mut().poll(&mut context) {
std::task::Poll::Ready(output) => return output,
std::task::Poll::Pending => std::thread::park(),
}
}
}
#[test]
fn append_and_read() {
let mut log = EventLog::new();
log.append(
EventKind::ProposalReceived,
None,
Some("p1"),
[("source".to_string(), Value::from("test"))].into(),
);
assert_eq!(log.len(), 1);
assert_eq!(log.events()[0].kind, EventKind::ProposalReceived);
}
#[test]
fn query_filters_by_kind_data_and_time() {
let mut log = EventLog::new();
log.append(
EventKind::PermissionDecision,
Some("a1"),
None,
[
("caller".to_string(), Value::from("alice")),
("tool".to_string(), Value::from("shell")),
]
.into(),
);
log.append(
EventKind::PermissionDecision,
Some("a2"),
None,
[
("caller".to_string(), Value::from("bob")),
("tool".to_string(), Value::from("shell")),
]
.into(),
);
log.append(
EventKind::StateChanged,
Some("a3"),
None,
Default::default(),
);
let q = EventQuery {
kinds: vec![EventKind::PermissionDecision],
..Default::default()
};
assert_eq!(log.query(&q).len(), 2);
let q = EventQuery {
data_matches: [("caller".to_string(), "alice".to_string())].into(),
..Default::default()
};
let hits = log.query(&q);
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].action_id.as_deref(), Some("a1"));
let q = EventQuery {
kinds: vec![EventKind::PermissionDecision],
data_matches: [("tool".to_string(), "shell".to_string())].into(),
limit: Some(1),
..Default::default()
};
let hits = log.query(&q);
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].action_id.as_deref(), Some("a2"));
}
#[test]
fn cost_by_agent_folds_metered_events() {
let mut log = EventLog::new();
log.append_metered(
EventKind::InferenceMetered,
None,
None,
[("agent".to_string(), Value::from("researcher"))].into(),
Metrics {
tokens_in: Some(100),
tokens_out: Some(50),
cost_usd: Some(2.0),
..Default::default()
},
);
log.append_metered(
EventKind::InferenceMetered,
None,
None,
[("agent".to_string(), Value::from("researcher"))].into(),
Metrics {
tokens_in: Some(10),
tokens_out: Some(5),
cost_usd: Some(0.2),
..Default::default()
},
);
log.append_metered(
EventKind::InferenceMetered,
None,
None,
[("agent".to_string(), Value::from("coordinator"))].into(),
Metrics {
cost_usd: Some(0.5),
..Default::default()
},
);
let report = log.cost_by_agent();
assert_eq!(report.len(), 2);
assert_eq!(report[0].agent, "coordinator");
assert_eq!(report[0].cost_usd, 0.5);
assert_eq!(report[1].agent, "researcher");
assert_eq!(report[1].calls, 2);
assert_eq!(report[1].tokens_in, 110);
assert_eq!(report[1].tokens_out, 55);
assert!((report[1].cost_usd - 2.2).abs() < 1e-9);
}
#[test]
fn auto_retention_caps_event_count() {
let mut log = EventLog::new();
log.set_retention(Some(RetentionPolicy {
max_events: Some(3),
max_age_secs: None,
}));
for i in 0..10 {
log.append(
EventKind::StateChanged,
Some(&format!("a{i}")),
None,
Default::default(),
);
}
assert_eq!(log.len(), 3);
assert_eq!(log.events()[0].action_id.as_deref(), Some("a7"));
assert_eq!(log.events()[2].action_id.as_deref(), Some("a9"));
}
#[test]
fn enforce_retention_drops_old_by_age() {
let mut log = EventLog::new();
log.append(
EventKind::StateChanged,
Some("old"),
None,
Default::default(),
);
log.events[0].timestamp = Utc::now() - chrono::Duration::seconds(3600);
log.append(
EventKind::StateChanged,
Some("fresh"),
None,
Default::default(),
);
let removed = log.enforce_retention(
&RetentionPolicy {
max_events: None,
max_age_secs: Some(60),
},
Utc::now(),
);
assert_eq!(removed, 1);
assert_eq!(log.len(), 1);
assert_eq!(log.events()[0].action_id.as_deref(), Some("fresh"));
}
#[test]
fn retention_trims_are_counted() {
let mut log = EventLog::new();
log.set_retention(Some(RetentionPolicy {
max_events: Some(2),
max_age_secs: None,
}));
for i in 0..5 {
log.append(
EventKind::StateChanged,
Some(&format!("a{i}")),
None,
Default::default(),
);
}
assert_eq!(log.trimmed_events(), 3);
assert_eq!(log.truncate_events_keep_last(1), 1);
assert_eq!(log.trimmed_events(), 4);
log.clear();
assert_eq!(log.trimmed_events(), 5);
}
#[test]
fn cumulative_cost_is_monotonic_across_trims_and_reload() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("cost.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
log.set_retention(Some(RetentionPolicy {
max_events: Some(1),
max_age_secs: None,
}));
for _ in 0..4 {
log.append_metered(
EventKind::InferenceMetered,
None,
None,
Default::default(),
Metrics {
cost_usd: Some(2.5),
..Default::default()
},
);
}
assert_eq!(log.len(), 1);
assert!((log.cumulative_cost_usd() - 10.0).abs() < 1e-9);
}
let reloaded = EventLog::load(&journal).unwrap();
assert!((reloaded.cumulative_cost_usd() - 10.0).abs() < 1e-9);
}
#[test]
fn journal_compaction_rewrites_to_retained_set() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("compact.jsonl");
let keep = 16usize;
let total = keep + JOURNAL_COMPACT_MIN_EXCESS + 8;
{
let mut log = EventLog::with_journal(journal.clone());
log.set_retention(Some(RetentionPolicy {
max_events: Some(keep),
max_age_secs: None,
}));
for i in 0..total {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
None,
HashMap::new(),
);
}
assert_eq!(log.len(), keep);
assert!(log.journal_size_bytes().unwrap_or(0) > 0);
}
let reloaded = EventLog::load(&journal).unwrap();
assert!(
reloaded.len() < total,
"journal must have been compacted (got {} lines)",
reloaded.len()
);
assert_eq!(
reloaded.events().last().unwrap().action_id.as_deref(),
Some(format!("a{}", total - 1).as_str())
);
}
#[test]
fn compact_journal_preserves_hash_chain_of_retained_tail() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("chained.jsonl");
{
let mut log = EventLog::with_journal(journal.clone()).with_hash_chaining();
for i in 0..20 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
None,
HashMap::new(),
);
}
log.truncate_events_keep_last(5);
assert!(log.compact_journal(), "compaction must succeed");
log.append(
EventKind::ActionSucceeded,
Some("post"),
None,
HashMap::new(),
);
}
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 6);
assert_eq!(reloaded.verify_chain(), Ok(6), "retained tail must verify");
assert_eq!(reloaded.events()[0].action_id.as_deref(), Some("a15"));
assert_eq!(reloaded.events()[5].action_id.as_deref(), Some("post"));
}
#[test]
fn compact_journal_without_journal_is_noop() {
let mut log = EventLog::new();
log.append(EventKind::StateChanged, Some("a"), None, Default::default());
assert!(!log.compact_journal());
assert_eq!(log.journal_size_bytes(), None);
}
#[test]
fn chaining_off_by_default_no_hashes() {
let mut log = EventLog::new();
log.append(
EventKind::ActionSucceeded,
Some("a1"),
Some("p1"),
HashMap::new(),
);
assert!(!log.hash_chaining_enabled());
assert!(log.events()[0].hash.is_none());
assert!(log.events()[0].prev_hash.is_none());
assert_eq!(log.verify_chain(), Ok(0));
}
#[test]
fn hash_chain_verifies_clean_log() {
let mut log = EventLog::new().with_hash_chaining();
for i in 0..5 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
Some("p"),
[("i".to_string(), Value::from(i))].into(),
);
}
assert!(log.events().iter().all(|e| e.hash.is_some()));
assert_eq!(log.verify_chain(), Ok(5));
assert_eq!(log.events()[0].prev_hash.as_deref(), Some(""));
for w in log.events().windows(2) {
assert_eq!(w[1].prev_hash, w[0].hash);
}
}
#[test]
fn tampering_with_data_breaks_chain() {
let mut log = EventLog::new().with_hash_chaining();
for i in 0..4 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
Some("p"),
[("v".to_string(), Value::from(i))].into(),
);
}
assert_eq!(log.verify_chain(), Ok(4));
log.events[2].data.insert("v".to_string(), Value::from(999));
assert_eq!(log.verify_chain(), Err(2));
}
#[test]
fn deleting_an_event_breaks_chain() {
let mut log = EventLog::new().with_hash_chaining();
for i in 0..4 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
Some("p"),
HashMap::new(),
);
}
log.events.remove(1);
assert_eq!(log.verify_chain(), Err(1));
}
#[test]
fn chain_survives_serialize_roundtrip() {
let mut log = EventLog::new().with_hash_chaining();
for i in 0..3 {
log.append(
EventKind::PermissionDecision,
Some(&format!("a{i}")),
Some("p"),
[
("decision".to_string(), Value::from("allow")),
("nested".to_string(), serde_json::json!({"z": 1, "a": 2})),
]
.into(),
);
}
let lines: Vec<String> = log
.events()
.iter()
.map(|e| serde_json::to_string(e).unwrap())
.collect();
let mut rebuilt = EventLog::new();
for line in &lines {
rebuilt.events.push(serde_json::from_str(line).unwrap());
}
assert_eq!(rebuilt.verify_chain(), Ok(3));
}
#[test]
fn chain_survives_journal_load_and_append() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("chain.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
log.enable_hash_chaining();
log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
log.append(EventKind::ActionSucceeded, Some("a2"), None, HashMap::new());
}
{
let mut log = EventLog::load(&journal).unwrap();
assert!(
log.hash_chaining_enabled(),
"loading a chained tail re-enables chaining"
);
log.append(EventKind::ActionSucceeded, Some("a3"), None, HashMap::new());
assert_eq!(log.verify_chain(), Ok(3), "post-load append stays chained");
}
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 3);
assert_eq!(reloaded.verify_chain(), Ok(3));
let plain = dir.path().join("plain.jsonl");
{
let mut log = EventLog::with_journal(plain.clone());
log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
}
let loaded = EventLog::load(&plain).unwrap();
assert!(!loaded.hash_chaining_enabled(), "unchained tail stays off");
}
#[test]
fn metered_event_carries_metrics_in_data() {
let mut log = EventLog::new();
log.append_metered(
EventKind::ActionSucceeded,
Some("a1"),
Some("p1"),
[("tool".to_string(), Value::from("search"))].into(),
Metrics::inference(120, 45, Some(0.0012)).with_duration(83.0),
);
let ev = &log.events()[0];
assert_eq!(ev.data.get("tool").unwrap(), "search");
assert_eq!(ev.duration_ms(), Some(83.0));
assert_eq!(ev.tokens_in(), Some(120));
assert_eq!(ev.tokens_out(), Some(45));
assert_eq!(ev.cost_usd(), Some(0.0012));
}
#[test]
fn metrics_totals_sum_across_events() {
let mut log = EventLog::new();
log.append_metered(
EventKind::ActionSucceeded,
Some("a1"),
None,
HashMap::new(),
Metrics::latency(50.0),
);
log.append_metered(
EventKind::ActionSucceeded,
Some("a2"),
None,
HashMap::new(),
Metrics::inference(100, 20, Some(0.5)).with_duration(70.0),
);
log.append(EventKind::ProposalReceived, None, None, HashMap::new());
let t = log.metrics_totals();
assert_eq!(t.duration_ms, 120.0);
assert_eq!(t.tokens_in, 100);
assert_eq!(t.tokens_out, 20);
assert_eq!(t.tokens, 120);
assert_eq!(t.cost_usd, 0.5);
assert_eq!(t.metered_events, 2);
}
#[test]
fn metrics_totals_counts_raw_appended_duration_key() {
let mut log = EventLog::new();
log.append(
EventKind::ActionSucceeded,
Some("a1"),
None,
[(metric_keys::DURATION_MS.to_string(), Value::from(42.0))].into(),
);
let t = log.metrics_totals();
assert_eq!(t.duration_ms, 42.0);
assert_eq!(t.metered_events, 1);
}
#[test]
fn new_telemetry_event_kinds_serialize_snake_case() {
let json = serde_json::to_string(&EventKind::BranchDecision).unwrap();
assert_eq!(json, "\"branch_decision\"");
let json = serde_json::to_string(&EventKind::AlternativeRejected).unwrap();
assert_eq!(json, "\"alternative_rejected\"");
let json = serde_json::to_string(&EventKind::InferenceMetered).unwrap();
assert_eq!(json, "\"inference_metered\"");
}
#[test]
fn filter_by_kind() {
let mut log = EventLog::new();
log.append(
EventKind::ProposalReceived,
None,
Some("p1"),
HashMap::new(),
);
log.append(
EventKind::ActionValidated,
Some("a1"),
Some("p1"),
HashMap::new(),
);
log.append(
EventKind::ActionSucceeded,
Some("a1"),
Some("p1"),
HashMap::new(),
);
let validated = log.filter(Some(&EventKind::ActionValidated), None);
assert_eq!(validated.len(), 1);
}
#[test]
fn filter_by_action_id() {
let mut log = EventLog::new();
log.append(EventKind::ActionValidated, Some("a1"), None, HashMap::new());
log.append(EventKind::ActionValidated, Some("a2"), None, HashMap::new());
let a1_events = log.filter(None, Some("a1"));
assert_eq!(a1_events.len(), 1);
}
#[test]
fn journal_write_and_reload() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("events.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
log.append(
EventKind::ProposalReceived,
None,
Some("p1"),
HashMap::new(),
);
log.append(
EventKind::ActionSucceeded,
Some("a1"),
Some("p1"),
HashMap::new(),
);
}
assert!(journal.exists());
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 2);
assert_eq!(reloaded.events()[0].kind, EventKind::ProposalReceived);
assert_eq!(reloaded.events()[1].kind, EventKind::ActionSucceeded);
}
#[test]
fn load_rejects_newline_terminated_corrupt_middle_before_later_terminal() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("corrupt-middle.jsonl");
let started = serde_json::json!({
"kind": "run_started",
"run_id": "run-corrupt",
"client_id": "client-1",
"data": {"agent_id": "daily-continuity-newsroom"},
"timestamp": "2026-08-30T09:30:00Z"
});
let completed = serde_json::json!({
"kind": "run_completed",
"run_id": "run-corrupt",
"client_id": "client-1",
"data": {
"completion_digest": "must-not-be-trusted",
"termination": {"kind": "outcome", "status": "success", "outcome": {}}
},
"timestamp": "2026-08-30T10:00:00Z"
});
fs::write(
&journal,
format!("{started}\n{{this-is-not-json}}\n{completed}\n"),
)
.unwrap();
let error = match EventLog::load(&journal) {
Ok(_) => panic!("newline-terminated middle corruption must fail closed"),
Err(error) => error,
};
assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
assert!(error.to_string().contains("line=2"), "{error}");
}
#[test]
fn load_repairs_crash_torn_final_record_before_append() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("torn-tail.jsonl");
let started = serde_json::json!({
"kind": "run_started",
"run_id": "run-torn",
"client_id": "client-1",
"data": {"agent_id": "daily-continuity-newsroom"},
"timestamp": "2026-08-30T09:30:00Z"
});
let mut journal_file = create_private_file(&journal).unwrap();
write!(journal_file, "{started}\n{{\"kind\":\"run_completed\"").unwrap();
drop(journal_file);
{
let mut loaded = EventLog::load(&journal).expect("torn final row is recoverable");
assert_eq!(loaded.len(), 1);
loaded.append(
EventKind::ProposalReceived,
None,
Some("proposal-after-recovery"),
HashMap::new(),
);
}
let bytes = fs::read_to_string(&journal).unwrap();
assert!(!bytes.contains("{\"kind\":\"run_completed\""), "{bytes}");
assert!(bytes.ends_with('\n'));
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 2);
assert_eq!(reloaded.events()[0].kind, EventKind::RunStarted);
assert_eq!(reloaded.events()[1].kind, EventKind::ProposalReceived);
}
#[test]
fn load_read_only_rejects_a_torn_tail_without_modifying_the_journal() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("read-only-torn-tail.jsonl");
let started = serde_json::json!({
"kind": "run_started",
"run_id": "run-torn",
"client_id": "client-1",
"data": {"agent_id": "daily-continuity-newsroom"},
"timestamp": "2026-08-30T09:30:00Z"
});
fs::write(&journal, format!("{started}\n{{\"kind\":\"run_completed\"")).unwrap();
let before = fs::read(&journal).unwrap();
let error = match EventLog::load_read_only(&journal) {
Ok(_) => panic!("read-only loading must expose a torn final row"),
Err(error) => error,
};
assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof);
assert!(error.to_string().contains("event journal torn tail"));
assert!(error.to_string().contains("line=2"));
assert_eq!(fs::read(&journal).unwrap(), before);
}
#[test]
fn load_rejects_existing_hash_chain_tampering() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("tampered-chain.jsonl");
{
let mut log = EventLog::with_journal(journal.clone()).with_hash_chaining();
log.append(
EventKind::RunStarted,
None,
None,
[(
"agent_id".to_string(),
Value::from("daily-continuity-newsroom"),
)]
.into(),
);
log.append(EventKind::RunCompleted, None, None, HashMap::new());
}
let mut rows: Vec<Value> = fs::read_to_string(&journal)
.unwrap()
.lines()
.map(|line| serde_json::from_str(line).unwrap())
.collect();
rows[0]["data"]["agent_id"] = Value::from("tampered-agent");
fs::write(
&journal,
rows.iter()
.map(Value::to_string)
.collect::<Vec<_>>()
.join("\n")
+ "\n",
)
.unwrap();
let error = match EventLog::load(&journal) {
Ok(_) => panic!("hash-chain tampering must fail closed during load"),
Err(error) => error,
};
assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
assert!(error.to_string().contains("hash chain"), "{error}");
assert!(error.to_string().contains("line=1"), "{error}");
}
#[cfg(unix)]
fn unix_mode(path: &Path) -> u32 {
use std::os::unix::fs::PermissionsExt;
fs::symlink_metadata(path).unwrap().permissions().mode() & 0o777
}
#[cfg(unix)]
#[test]
fn car_owned_journal_and_created_parents_are_private() {
let root = tempfile::tempdir().unwrap();
let parent = root.path().join("eventlogs").join("session");
let journal = parent.join("events.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
}
assert_eq!(unix_mode(&root.path().join("eventlogs")), 0o700);
assert_eq!(unix_mode(&parent), 0o700);
assert_eq!(unix_mode(&journal), 0o600);
}
#[cfg(unix)]
#[test]
fn append_hardens_preexisting_owned_permissive_journal() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("events.jsonl");
fs::write(&journal, b"").unwrap();
fs::set_permissions(&journal, fs::Permissions::from_mode(0o644)).unwrap();
{
let mut log = EventLog::with_journal(journal.clone());
log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
}
assert_eq!(unix_mode(&journal), 0o600);
}
#[cfg(unix)]
#[test]
fn journal_refuses_symlink_and_hardlink_destinations() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let victim = dir.path().join("victim");
fs::write(&victim, b"unchanged").unwrap();
for journal in [dir.path().join("symlink"), dir.path().join("hardlink")] {
if journal.ends_with("symlink") {
symlink(&victim, &journal).unwrap();
} else {
fs::hard_link(&victim, &journal).unwrap();
}
{
let mut log = EventLog::with_journal(journal);
log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
}
assert_eq!(fs::read(&victim).unwrap(), b"unchanged");
}
}
#[cfg(unix)]
#[test]
fn journal_stops_if_the_opened_path_is_substituted() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("events.jsonl");
let moved = dir.path().join("moved.jsonl");
let mut log = EventLog::with_journal(journal.clone());
log.append(EventKind::StateChanged, Some("first"), None, HashMap::new());
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
while fs::metadata(&journal).map_or(true, |metadata| metadata.len() == 0) {
assert!(
std::time::Instant::now() < deadline,
"first event was not persisted"
);
std::thread::yield_now();
}
fs::rename(&journal, &moved).unwrap();
let _substitute = create_private_file(&journal).unwrap();
log.append(
EventKind::StateChanged,
Some("second"),
None,
HashMap::new(),
);
drop(log);
assert_eq!(EventLog::load(&moved).unwrap().len(), 1);
assert_eq!(fs::metadata(&journal).unwrap().len(), 0);
}
#[cfg(unix)]
#[test]
fn compaction_preserves_private_mode_and_leaves_no_temp_name() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("events.jsonl");
let mut log = EventLog::with_journal(journal.clone());
for index in 0..4 {
log.append(
EventKind::StateChanged,
Some(&format!("a{index}")),
None,
HashMap::new(),
);
}
log.truncate_events_keep_last(2);
assert!(log.compact_journal());
drop(log);
assert_eq!(unix_mode(&journal), 0o600);
let names: Vec<_> = fs::read_dir(dir.path())
.unwrap()
.map(|entry| entry.unwrap().file_name())
.collect();
assert_eq!(names, vec![journal.file_name().unwrap()]);
}
#[cfg(unix)]
#[test]
fn historical_world_readable_journal_can_be_loaded_without_mutation() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("historical.jsonl");
let event = Event {
kind: EventKind::StateChanged,
run_id: None,
client_id: None,
policy_session_id: None,
action_id: Some("historical".into()),
proposal_id: None,
data: HashMap::new(),
timestamp: Utc::now(),
prev_hash: None,
hash: None,
};
fs::write(
&journal,
format!("{}\n", serde_json::to_string(&event).unwrap()),
)
.unwrap();
fs::set_permissions(&journal, fs::Permissions::from_mode(0o644)).unwrap();
let loaded = EventLog::load(&journal).unwrap();
assert_eq!(loaded.len(), 1);
drop(loaded);
assert_eq!(unix_mode(&journal), 0o644);
}
#[test]
fn journal_not_created_without_appends() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("no-events.jsonl");
{
let _log = EventLog::with_journal(journal.clone());
}
assert!(
!journal.exists(),
"journal file must not be created when nothing is appended"
);
}
#[test]
fn journal_preserves_order_and_count_under_burst() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("burst.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
for i in 0..500 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{i}")),
None,
HashMap::new(),
);
}
}
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 500, "no events lost");
for (i, event) in reloaded.events().iter().enumerate() {
assert_eq!(
event.action_id.as_deref(),
Some(format!("a{i}").as_str()),
"order preserved at {i}"
);
}
}
#[test]
fn unopenable_journal_is_best_effort_not_fatal() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("a-directory");
fs::create_dir(&journal).unwrap();
let mut log = EventLog::with_journal(journal);
log.append(
EventKind::ProposalReceived,
None,
Some("p1"),
HashMap::new(),
);
log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
assert_eq!(
log.len(),
2,
"in-memory log unaffected by an unwritable journal"
);
}
#[test]
fn load_then_append_preserves_existing_and_adds() {
let dir = tempfile::tempdir().unwrap();
let journal = dir.path().join("resume.jsonl");
{
let mut log = EventLog::with_journal(journal.clone());
log.append(
EventKind::ProposalReceived,
None,
Some("p1"),
HashMap::new(),
);
}
{
let mut log = EventLog::load(&journal).unwrap();
assert_eq!(log.len(), 1);
log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
}
let reloaded = EventLog::load(&journal).unwrap();
assert_eq!(reloaded.len(), 2, "append-mode preserved the loaded line");
assert_eq!(reloaded.events()[0].kind, EventKind::ProposalReceived);
assert_eq!(reloaded.events()[1].kind, EventKind::ActionSucceeded);
}
#[test]
fn event_kind_serializes_snake_case() {
assert_eq!(
serde_json::to_string(&EventKind::ProposalReceived).unwrap(),
"\"proposal_received\""
);
assert_eq!(
serde_json::to_string(&EventKind::StateSnapshot).unwrap(),
"\"state_snapshot\""
);
}
#[test]
fn stats_truncate_and_clear_release_retained_entries() {
let mut log = EventLog::new();
for idx in 0..5 {
log.append(
EventKind::ActionSucceeded,
Some(&format!("a{idx}")),
Some("p1"),
[("payload".to_string(), Value::from("x".repeat(16)))].into(),
);
log.begin_span("action.tool_call", "trace", None, HashMap::new());
}
let stats = log.stats();
assert_eq!(stats.events, 5);
assert_eq!(stats.spans, 5);
assert!(stats.approx_event_bytes > 0);
assert!(stats.approx_span_bytes > 0);
assert_eq!(log.truncate_events_keep_last(2), 3);
assert_eq!(log.truncate_spans_keep_last(1), 4);
assert_eq!(log.len(), 2);
assert_eq!(log.span_len(), 1);
assert_eq!(log.events()[0].action_id.as_deref(), Some("a3"));
let removed = log.clear();
assert_eq!(removed.events, 2);
assert_eq!(removed.spans, 1);
assert_eq!(log.len(), 0);
assert_eq!(log.span_len(), 0);
}
#[test]
fn span_begin_end_lifecycle() {
let mut log = EventLog::new();
let trace_id = "trace-1".to_string();
let span_id = log.begin_span(
"test.operation",
&trace_id,
None,
[("key".to_string(), Value::from("value"))].into(),
);
let spans = log.spans();
assert_eq!(spans.len(), 1);
assert_eq!(spans[0].name, "test.operation");
assert_eq!(spans[0].trace_id, "trace-1");
assert!(spans[0].parent_span_id.is_none());
assert!(spans[0].end_time.is_none());
assert_eq!(spans[0].status, SpanStatus::Unset);
log.end_span(&span_id, SpanStatus::Ok);
let spans = log.spans();
assert!(spans[0].end_time.is_some());
assert_eq!(spans[0].status, SpanStatus::Ok);
}
#[test]
fn span_parent_child_relationship() {
let mut log = EventLog::new();
let trace_id = "trace-2".to_string();
let parent_id = log.begin_span("parent.op", &trace_id, None, HashMap::new());
let child_id = log.begin_span("child.op", &trace_id, Some(&parent_id), HashMap::new());
let spans = log.spans();
assert_eq!(spans.len(), 2);
let child = spans.iter().find(|s| s.span_id == child_id).unwrap();
assert_eq!(child.parent_span_id.as_deref(), Some(parent_id.as_str()));
assert_eq!(child.trace_id, trace_id);
let parent = spans.iter().find(|s| s.span_id == parent_id).unwrap();
assert!(parent.parent_span_id.is_none());
}
#[test]
fn export_traces_produces_valid_json() {
let mut log = EventLog::new();
let trace_id = "trace-3".to_string();
let root = log.begin_span(
"proposal.execute",
&trace_id,
None,
[("proposal_id".to_string(), Value::from("p1"))].into(),
);
let child = log.begin_span(
"action.tool_call",
&trace_id,
Some(&root),
[("tool".to_string(), Value::from("read_file"))].into(),
);
log.end_span(&child, SpanStatus::Ok);
log.end_span(&root, SpanStatus::Ok);
let json_str = log.export_traces();
let parsed: Value =
serde_json::from_str(&json_str).expect("export_traces must produce valid JSON");
let resource_spans = parsed["resourceSpans"].as_array().unwrap();
assert_eq!(resource_spans.len(), 1);
let scope_spans = &resource_spans[0]["scopeSpans"][0]["spans"];
let spans_arr = scope_spans.as_array().unwrap();
assert_eq!(spans_arr.len(), 2);
for span in spans_arr {
assert!(span.get("traceId").is_some());
assert!(span.get("spanId").is_some());
assert!(span.get("name").is_some());
assert!(span.get("startTimeUnixNano").is_some());
assert!(span.get("endTimeUnixNano").is_some());
assert!(span.get("status").is_some());
}
let child_span = spans_arr
.iter()
.find(|s| s["name"] == "action.tool_call")
.unwrap();
assert!(child_span.get("parentSpanId").is_some());
}
#[test]
fn span_status_set_on_error() {
let mut log = EventLog::new();
let trace_id = "trace-4".to_string();
let span_id = log.begin_span("failing.op", &trace_id, None, HashMap::new());
log.end_span(&span_id, SpanStatus::Error);
let spans = log.spans();
assert_eq!(spans[0].status, SpanStatus::Error);
assert!(spans[0].end_time.is_some());
}
#[test]
fn active_run_binding_stamps_every_new_event_and_rejects_conflicts() {
let mut log = EventLog::new();
log.bind_run("run-a", "client-a")
.expect("first active run binds");
log.bind_policy_session("policy-session-a")
.expect("CAR-minted policy session binds inside the run");
log.append(
EventKind::ProposalReceived,
None,
Some("same-proposal"),
HashMap::new(),
);
log.append(
EventKind::ActionSucceeded,
Some("action-a"),
Some("same-proposal"),
HashMap::new(),
);
for event in log.events() {
assert_eq!(event.run_id.as_deref(), Some("run-a"));
assert_eq!(event.client_id.as_deref(), Some("client-a"));
assert_eq!(event.policy_session_id.as_deref(), Some("policy-session-a"));
}
assert!(log.bind_run("run-b", "client-a").is_err());
assert!(log.bind_run("run-a", "client-b").is_err());
assert!(log.clear_run_binding("run-b", "client-a").is_err());
assert_eq!(
log.active_run_binding(),
Some(("run-a", "client-a", Some("policy-session-a")))
);
log.clear_policy_session("policy-session-a")
.expect("exact policy session clears");
log.clear_run_binding("run-a", "client-a")
.expect("exact active run clears");
log.append(
EventKind::ProposalReceived,
None,
Some("unbound-legacy"),
HashMap::new(),
);
let legacy = log.events().last().unwrap();
assert!(legacy.run_id.is_none());
assert!(legacy.client_id.is_none());
assert!(legacy.policy_session_id.is_none());
}
#[test]
fn historical_event_without_binding_fields_still_deserializes() {
let historical = r#"{"kind":"proposal_received","proposal_id":"p-old","data":{},"timestamp":"2026-01-02T03:04:05Z"}"#;
let event: Event = serde_json::from_str(historical).expect("historical event replays");
assert!(event.run_id.is_none());
assert!(event.client_id.is_none());
assert!(event.policy_session_id.is_none());
}
#[test]
fn async_acknowledgement_timeout_wins_after_expiry_removal() {
let manager = AsyncAcknowledgementManager::new(1);
let barrier = AsyncAcknowledgementExpiryBarrier::new();
manager.pause_next_expiry_after_removal(barrier.clone());
let reservation = manager.reserve(Duration::from_millis(10)).unwrap();
let sender = reservation.sender();
let acknowledgement = reservation.into_future();
barrier.wait_until_removed();
sender.send(Ok(()));
let result = block_on_test_future(acknowledgement);
barrier.allow_timeout_completion();
manager.shutdown();
assert!(matches!(
result,
Err(CriticalPostAcceptanceError::AcknowledgementTimedOut { .. })
));
}
#[test]
fn async_acknowledgement_capacity_is_atomic_under_concurrent_reservation() {
let manager = Arc::new(AsyncAcknowledgementManager::new(1));
let barrier = Arc::new(std::sync::Barrier::new(3));
let handles: Vec<_> = (0..2)
.map(|_| {
let manager = manager.clone();
let barrier = barrier.clone();
std::thread::spawn(move || {
let reservation = manager.reserve(MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT);
barrier.wait();
reservation
})
})
.collect();
barrier.wait();
let results: Vec<_> = handles
.into_iter()
.map(|handle| handle.join().unwrap())
.collect();
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
assert_eq!(
results
.iter()
.filter(|result| {
matches!(
result,
Err(CriticalPreAcceptanceError::CapacityExhausted { capacity: 1 })
)
})
.count(),
1
);
drop(results);
manager.shutdown();
}
#[test]
fn async_acknowledgement_completed_before_future_construction_is_observed() {
let manager = AsyncAcknowledgementManager::new(1);
let reservation = manager.reserve(Duration::from_secs(1)).unwrap();
reservation.sender().send(Ok(()));
let acknowledgement = reservation.into_future();
assert!(block_on_test_future(acknowledgement).is_ok());
manager.shutdown();
}
#[test]
fn journal_writer_drop_completes_pending_async_acknowledgement() {
let dir = tempfile::tempdir().unwrap();
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
let writer = JournalWriter::spawn_with_injector(
dir.path().join("pending-ack-shutdown.jsonl"),
failures.clone(),
);
let reservation = writer
.reserve_async_acknowledgement(MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT)
.unwrap();
let acknowledgement = writer
.enqueue_critical_async("{}".to_string(), false, reservation)
.unwrap();
let deadline = Instant::now() + Duration::from_secs(2);
while failures.held_acknowledgement_count() == 0 {
assert!(
Instant::now() < deadline,
"writer never retained the pending acknowledgement"
);
std::thread::yield_now();
}
drop(writer);
let result = block_on_test_future(acknowledgement);
assert!(matches!(
result,
Err(CriticalPostAcceptanceError::CoordinatorStopped)
));
failures.release_held_acknowledgements();
}
#[test]
fn prepare_failure_releases_async_acknowledgement_capacity() {
let dir = tempfile::tempdir().unwrap();
let mut log = EventLog::with_journal_failure_injector_and_ack_capacity(
dir.path().join("prepare-failure-capacity.jsonl"),
JournalFailureInjector::default(),
1,
);
let data = HashMap::from([("completion_digest".to_string(), Value::from("7".repeat(64)))]);
let error = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data.clone(),
Duration::from_secs(1),
))
.expect_err("an unbound critical event must fail during preparation");
assert!(matches!(error, CriticalAppendError::Rejected { .. }));
log.bind_run("run-after-prepare-failure", "client-after-prepare-failure")
.unwrap();
block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data,
Duration::from_secs(1),
))
.expect("the failed preparation must release the only acknowledgement slot");
}
#[test]
fn async_critical_preacceptance_failures_reject_without_fabricating_events() {
let data = HashMap::from([("completion_digest".to_string(), Value::from("f".repeat(64)))]);
let mut no_journal = EventLog::new();
no_journal
.bind_run("run-no-writer", "client-no-writer")
.unwrap();
let error = block_on_test_future(no_journal.append_critical_async(
EventKind::RunCompleted,
None,
None,
data.clone(),
Duration::from_millis(100),
))
.expect_err("a missing writer must reject before acceptance");
assert!(matches!(error, CriticalAppendError::Rejected { .. }));
assert!(!error.is_retry_safe());
assert!(no_journal.events().is_empty());
assert!(no_journal.critical_pending.is_empty());
let dir = tempfile::tempdir().unwrap();
let unavailable_path = dir.path().join("sender-missing.jsonl");
let mut unavailable = EventLog::with_journal(unavailable_path.clone());
unavailable
.bind_run("run-sender-missing", "client-sender-missing")
.unwrap();
unavailable
.journal
.as_mut()
.unwrap()
.remove_sender_for_test();
let error = block_on_test_future(unavailable.append_critical_async(
EventKind::RunCompleted,
None,
None,
data.clone(),
Duration::from_millis(100),
))
.expect_err("a missing writer sender must reject before event construction");
assert!(matches!(error, CriticalAppendError::Rejected { .. }));
assert!(unavailable.events().is_empty());
assert!(unavailable.critical_pending.is_empty());
assert!(!unavailable_path.exists());
let stopped_path = dir.path().join("receiver-stopped.jsonl");
let mut stopped = EventLog::with_journal(stopped_path.clone());
stopped
.bind_run("run-receiver-stopped", "client-receiver-stopped")
.unwrap();
stopped.journal.as_mut().unwrap().stop_receiver_for_test();
let error = block_on_test_future(stopped.append_critical_async(
EventKind::RunCompleted,
None,
None,
data,
Duration::from_millis(100),
))
.expect_err("a stopped writer receiver must reject a failed enqueue");
assert!(matches!(error, CriticalAppendError::Rejected { .. }));
assert!(stopped.events().is_empty());
assert!(stopped.critical_pending.is_empty());
assert!(!stopped_path.exists());
}
#[test]
fn async_critical_rejects_excessive_acknowledgement_duration_before_enqueue() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("excessive-ack-duration.jsonl");
let mut log = EventLog::with_journal(path.clone());
log.bind_run("run-excessive-ack", "client-excessive-ack")
.unwrap();
let error = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
HashMap::new(),
MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT + Duration::from_millis(1),
))
.expect_err("an excessive acknowledgement duration must reject");
assert!(matches!(error, CriticalAppendError::Rejected { .. }));
assert!(log.events().is_empty());
assert!(log.critical_pending.is_empty());
assert!(!path.exists());
}
#[test]
fn async_critical_capacity_exhaustion_rejects_before_enqueue_and_exact_retry_unblocks() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("critical-ack-capacity.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
let mut log = EventLog::with_journal_failure_injector_and_ack_capacity(
path.clone(),
failures.clone(),
1,
);
log.bind_run("run-capacity", "client-capacity").unwrap();
let first_data =
HashMap::from([("completion_digest".to_string(), Value::from("1".repeat(64)))]);
let second_data =
HashMap::from([("completion_digest".to_string(), Value::from("2".repeat(64)))]);
let mut first = Box::pin(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
first_data.clone(),
MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
));
let waker = test_waker();
let mut context = std::task::Context::from_waker(&waker);
assert!(matches!(
first.as_mut().poll(&mut context),
std::task::Poll::Pending
));
drop(first);
assert_eq!(log.events().len(), 1);
assert_eq!(log.critical_pending.len(), 1);
for attempt in 0..100 {
let error = block_on_test_future(log.append_critical_async(
EventKind::ProposalCompleted,
None,
Some("capacity-rejected"),
second_data.clone(),
Duration::from_millis(100),
))
.expect_err("exhausted acknowledgement capacity must reject");
let CriticalAppendError::Rejected { reason } = error else {
panic!("attempt {attempt} was not a pre-enqueue rejection");
};
assert!(
reason.contains("acknowledgement capacity is exhausted"),
"attempt {attempt} bypassed capacity admission: {reason}"
);
assert_eq!(log.events().len(), 1);
assert_eq!(log.critical_pending.len(), 1);
}
let hold_deadline = std::time::Instant::now() + Duration::from_secs(2);
while failures.held_acknowledgement_count() == 0 {
assert!(
std::time::Instant::now() < hold_deadline,
"writer never reached the held acknowledgement"
);
std::thread::yield_now();
}
let pending_line = serde_json::to_string(&log.events()[0]).unwrap();
assert_eq!(
fs::read_to_string(&path).unwrap(),
format!("{pending_line}\n")
);
failures.release_held_acknowledgements();
let original_timestamp = log.events()[0].timestamp;
let retried = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
first_data,
Duration::from_millis(500),
))
.expect("the exact cancelled row must reconcile pending state");
assert_eq!(retried.timestamp, original_timestamp);
assert!(log.critical_pending.is_empty());
block_on_test_future(log.append_critical_async(
EventKind::ProposalCompleted,
None,
Some("capacity-rejected"),
second_data,
Duration::from_millis(500),
))
.expect("a distinct row is allowed after exact reconciliation");
drop(log);
let loaded = EventLog::load(&path).unwrap();
assert_eq!(loaded.events().len(), 2);
assert_eq!(loaded.events()[0].kind, EventKind::RunCompleted);
assert_eq!(loaded.events()[1].kind, EventKind::ProposalCompleted);
}
#[test]
fn async_critical_never_acknowledged_row_remains_exactly_retryable() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("critical-never-acknowledged.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
log.bind_run("run-never-ack", "client-never-ack").unwrap();
let data = HashMap::from([("completion_digest".to_string(), Value::from("9".repeat(64)))]);
let error = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data.clone(),
Duration::from_millis(20),
))
.expect_err("the first acknowledgement is retained forever");
assert!(matches!(
error,
CriticalAppendError::DurabilityUnknown { .. }
));
let original = serde_json::to_string(&log.events()[0]).unwrap();
assert!(log.critical_pending.contains(&original));
let retried = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data,
Duration::from_millis(500),
))
.expect("the exact row must reconcile without the first acknowledgement");
assert_eq!(serde_json::to_string(retried).unwrap(), original);
assert!(log.critical_pending.is_empty());
assert_eq!(
failures.held_acknowledgement_count(),
1,
"the original acknowledgement must remain unsent"
);
drop(log);
assert_eq!(fs::read_to_string(&path).unwrap(), format!("{original}\n"));
assert_eq!(EventLog::load(&path).unwrap().events().len(), 1);
}
#[test]
fn bounded_sync_critical_timeout_is_exactly_retryable() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("critical-bounded-sync-timeout.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
log.bind_run("run-bounded-sync", "client-bounded-sync")
.unwrap();
let data = HashMap::from([("completion_digest".to_string(), Value::from("7".repeat(64)))]);
let error = log
.append_critical_bounded(
EventKind::RunCompleted,
None,
None,
data.clone(),
Duration::from_millis(20),
)
.expect_err("the retained acknowledgement must hit the exact sync bound");
assert_eq!(
error,
CriticalAppendError::DurabilityUnknown {
reason: "journal writer did not acknowledge within 20ms".to_string()
}
);
assert!(error.is_retry_safe());
let pending = log
.critical_pending
.iter()
.next()
.expect("the exact timed-out row remains pending")
.clone();
assert_eq!(pending, serde_json::to_string(&log.events()[0]).unwrap());
let retried = log
.append_critical_bounded(
EventKind::RunCompleted,
None,
None,
data,
Duration::from_millis(500),
)
.expect("an exact retry must reconcile without the held acknowledgement");
assert_eq!(serde_json::to_string(retried).unwrap(), pending);
assert!(log.critical_pending.is_empty());
assert_eq!(failures.held_acknowledgement_count(), 1);
drop(log);
assert_eq!(fs::read_to_string(&path).unwrap(), format!("{pending}\n"));
}
#[test]
fn async_critical_append_bounds_unknown_ack_and_preserves_exact_retry() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("critical-unknown-ack.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
log.bind_run("run-unknown-ack", "client-unknown-ack")
.unwrap();
let data = HashMap::from([("completion_digest".to_string(), Value::from("e".repeat(64)))]);
let started = std::time::Instant::now();
let error = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data.clone(),
std::time::Duration::from_millis(40),
))
.expect_err("a retained acknowledgement must become durability-unknown");
assert!(
started.elapsed() < std::time::Duration::from_millis(500),
"the async acknowledgement wait exceeded its bounded allowance"
);
assert!(matches!(
error,
CriticalAppendError::DurabilityUnknown { .. }
));
assert!(error.is_retry_safe());
assert!(error
.to_string()
.contains("did not acknowledge within 40ms"));
let pending = log
.critical_pending
.iter()
.next()
.expect("the exact unacknowledged row remains pending")
.clone();
assert_eq!(log.critical_pending.len(), 1);
assert_eq!(pending, serde_json::to_string(&log.events()[0]).unwrap());
let disk_row = format!("{pending}\n");
let writer_deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
while fs::read_to_string(&path).unwrap_or_default() != disk_row {
assert!(
std::time::Instant::now() < writer_deadline,
"the writer never durably accepted the held-acknowledgement row"
);
std::thread::yield_now();
}
while failures.held_acknowledgement_count() == 0 {
assert!(
std::time::Instant::now() < writer_deadline,
"the writer never retained the late acknowledgement"
);
std::thread::yield_now();
}
assert_eq!(failures.held_acknowledgement_count(), 1);
failures.release_held_acknowledgements();
let original_timestamp = log.events()[0].timestamp;
let distinct_error = block_on_test_future(log.append_critical_async(
EventKind::ProposalCompleted,
None,
Some("different-terminal"),
HashMap::new(),
Duration::from_millis(100),
))
.expect_err("a different critical row must not bypass exact retry");
assert!(matches!(
distinct_error,
CriticalAppendError::Rejected { .. }
));
assert_eq!(log.events().len(), 1);
let retried = block_on_test_future(log.append_critical_async(
EventKind::RunCompleted,
None,
None,
data,
std::time::Duration::from_secs(1),
))
.expect("an identical retry must finish the retained row");
assert_eq!(retried.timestamp, original_timestamp);
assert!(log.critical_pending.is_empty());
block_on_test_future(log.append_critical_async(
EventKind::ProposalCompleted,
None,
Some("different-terminal"),
HashMap::new(),
Duration::from_millis(500),
))
.expect("a different row is allowed after exact retry reconciliation");
drop(log);
let loaded = EventLog::load(&path).unwrap();
assert_eq!(loaded.events().len(), 2);
assert_eq!(serde_json::to_string(&loaded.events()[0]).unwrap(), pending);
}
#[test]
fn critical_append_failures_retry_same_row_and_fsync_prior_async_events() {
for point in [
JournalFailurePoint::Write,
JournalFailurePoint::Flush,
JournalFailurePoint::Fsync,
] {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join(format!("critical-{point:?}.jsonl"));
let failures = JournalFailureInjector::default();
failures.fail_next(point);
let mut log = EventLog::with_journal_failure_injector(path.clone(), failures);
log.bind_run("run-critical", "client-critical").unwrap();
log.append(
EventKind::ProposalReceived,
None,
Some("proposal-critical"),
HashMap::new(),
);
let data =
HashMap::from([("completion_digest".to_string(), Value::from("a".repeat(64)))]);
assert!(
log.append_critical(EventKind::RunCompleted, None, None, data.clone())
.is_err(),
"{point:?} failure must not acknowledge"
);
log.append(
EventKind::ActionSucceeded,
Some("after-pending-terminal"),
Some("proposal-critical"),
HashMap::new(),
);
log.append_critical(EventKind::RunCompleted, None, None, data)
.expect("retry finishes the exact critical row");
drop(log);
let loaded = EventLog::load(&path).unwrap();
let events = loaded.events();
assert_eq!(events[0].kind, EventKind::ProposalReceived);
assert_eq!(events[1].kind, EventKind::RunCompleted);
assert_eq!(events[2].kind, EventKind::ActionSucceeded);
assert_eq!(
events
.iter()
.filter(|event| event.kind == EventKind::RunCompleted)
.count(),
1,
"{point:?} retry must not duplicate a terminal"
);
}
}
#[test]
fn compacting_a_failed_critical_row_makes_its_retry_idempotent() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("critical-compact-retry.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::Fsync);
let mut log =
EventLog::with_journal_failure_injector(path.clone(), failures).with_hash_chaining();
log.bind_run("run-critical-compact", "client-critical-compact")
.unwrap();
log.append(
EventKind::ProposalReceived,
None,
Some("proposal-critical-compact"),
HashMap::new(),
);
let data = HashMap::from([("completion_digest".to_string(), Value::from("c".repeat(64)))]);
assert!(
log.append_critical(EventKind::RunCompleted, None, None, data.clone())
.is_err(),
"the injected fsync failure must leave the exact terminal pending"
);
assert!(
log.compact_journal(),
"compaction persists the in-memory pending terminal"
);
log.append_critical(EventKind::RunCompleted, None, None, data)
.expect("identical retry recognizes the compacted terminal as durable");
drop(log);
let loaded = EventLog::load(&path).unwrap();
assert_eq!(
loaded
.events()
.iter()
.filter(|event| event.kind == EventKind::RunCompleted)
.count(),
1,
"writer respawn must not duplicate the compacted terminal"
);
assert_eq!(loaded.events()[0].kind, EventKind::ProposalReceived);
assert_eq!(loaded.events()[1].kind, EventKind::RunCompleted);
assert_eq!(loaded.verify_chain(), Ok(2));
}
#[test]
fn critical_append_cannot_ack_until_failed_prior_async_row_is_replayed() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("prior-async-failure.jsonl");
let failures = JournalFailureInjector::default();
failures.fail_next(JournalFailurePoint::AsyncWrite);
failures.fail_next(JournalFailurePoint::AsyncWrite);
let mut log = EventLog::with_journal_failure_injector(path.clone(), failures);
log.bind_run("run-ordered", "client-ordered").unwrap();
log.append(
EventKind::ActionSucceeded,
Some("action-ordered"),
Some("proposal-ordered"),
HashMap::new(),
);
let data = HashMap::from([("completion_digest".to_string(), Value::from("b".repeat(64)))]);
assert!(
log.append_critical(
EventKind::ProposalCompleted,
None,
Some("proposal-ordered"),
data.clone()
)
.is_err(),
"a terminal must not acknowledge while an earlier row is still missing"
);
log.append_critical(
EventKind::ProposalCompleted,
None,
Some("proposal-ordered"),
data,
)
.expect("retry repairs the prior row before acknowledging the terminal");
drop(log);
let loaded = EventLog::load(&path).unwrap();
assert_eq!(loaded.events().len(), 2);
assert_eq!(loaded.events()[0].kind, EventKind::ActionSucceeded);
assert_eq!(loaded.events()[1].kind, EventKind::ProposalCompleted);
}
#[test]
fn first_use_parent_sync_failure_blocks_terminal_until_prior_async_replays() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("nested").join("first-use.jsonl");
let failures = car_secrets::PrivatePathDurabilityFailureInjector::default();
failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
let mut log = EventLog::with_private_path_failure_injector(path.clone(), failures);
log.bind_run("run-first-use", "client-first-use").unwrap();
log.append(
EventKind::ActionSucceeded,
Some("action-first-use"),
Some("proposal-first-use"),
HashMap::new(),
);
let data = HashMap::from([("completion_digest".to_string(), Value::from("d".repeat(64)))]);
assert!(log
.append_critical(
EventKind::ProposalCompleted,
None,
Some("proposal-first-use"),
data.clone(),
)
.is_err());
log.append_critical(
EventKind::ProposalCompleted,
None,
Some("proposal-first-use"),
data,
)
.expect("retry durably replays async row then exact terminal");
drop(log);
let loaded = EventLog::load(&path).unwrap();
assert_eq!(loaded.events().len(), 2);
assert_eq!(loaded.events()[0].kind, EventKind::ActionSucceeded);
assert_eq!(loaded.events()[1].kind, EventKind::ProposalCompleted);
}
}