harn_vm/event_log/
mod.rs

1use std::cell::RefCell;
2use std::collections::BTreeMap;
3use std::fmt;
4use std::path::{Path, PathBuf};
5use std::str::FromStr;
6use std::sync::Arc;
7
8use bytes::Bytes;
9use futures::stream::BoxStream;
10use serde::{Deserialize, Serialize};
11
12use crate::runtime_limits::RuntimeLimits;
13
14mod file;
15mod memory;
16mod sqlite;
17mod util;
18
19#[cfg(test)]
20pub(crate) use util::pin_test_occurred_at_ms;
21
22#[cfg(test)]
23mod tests;
24
25pub use file::FileEventLog;
26pub use memory::MemoryEventLog;
27pub use sqlite::SqliteEventLog;
28
29pub type EventId = u64;
30
31/// Topic the LLM observability writer appends `provider_call_response`
32/// (and sibling transcript) events to. Read by `harn portal` and
33/// `harn usage`; kept here so every reader/writer shares one literal.
34pub const HARN_LLM_TRANSCRIPT_TOPIC: &str = "agent.transcript.llm";
35
36pub const HARN_EVENT_LOG_BACKEND_ENV: &str = "HARN_EVENT_LOG_BACKEND";
37pub const HARN_EVENT_LOG_DIR_ENV: &str = "HARN_EVENT_LOG_DIR";
38pub const HARN_EVENT_LOG_SQLITE_PATH_ENV: &str = "HARN_EVENT_LOG_SQLITE_PATH";
39pub const HARN_EVENT_LOG_QUEUE_DEPTH_ENV: &str = "HARN_EVENT_LOG_QUEUE_DEPTH";
40
41#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
42pub struct Topic(String);
43
44impl Topic {
45    pub fn new(value: impl Into<String>) -> Result<Self, LogError> {
46        let value = value.into();
47        if value.is_empty() {
48            return Err(LogError::InvalidTopic("topic cannot be empty".to_string()));
49        }
50        if !value
51            .chars()
52            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-'))
53        {
54            return Err(LogError::InvalidTopic(format!(
55                "topic '{value}' contains unsupported characters"
56            )));
57        }
58        Ok(Self(value))
59    }
60
61    pub fn as_str(&self) -> &str {
62        &self.0
63    }
64}
65
66impl fmt::Display for Topic {
67    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68        self.0.fmt(f)
69    }
70}
71
72impl FromStr for Topic {
73    type Err = LogError;
74
75    fn from_str(s: &str) -> Result<Self, Self::Err> {
76        Self::new(s)
77    }
78}
79
80#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
81pub struct ConsumerId(String);
82
83impl ConsumerId {
84    pub fn new(value: impl Into<String>) -> Result<Self, LogError> {
85        let value = value.into();
86        if value.trim().is_empty() {
87            return Err(LogError::InvalidConsumer(
88                "consumer id cannot be empty".to_string(),
89            ));
90        }
91        Ok(Self(value))
92    }
93
94    pub fn as_str(&self) -> &str {
95        &self.0
96    }
97}
98
99impl fmt::Display for ConsumerId {
100    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
101        self.0.fmt(f)
102    }
103}
104
105#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
106#[serde(rename_all = "snake_case")]
107pub enum EventLogBackendKind {
108    Memory,
109    File,
110    Sqlite,
111    Postgres,
112}
113
114impl fmt::Display for EventLogBackendKind {
115    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
116        match self {
117            Self::Memory => write!(f, "memory"),
118            Self::File => write!(f, "file"),
119            Self::Sqlite => write!(f, "sqlite"),
120            Self::Postgres => write!(f, "postgres"),
121        }
122    }
123}
124
125impl FromStr for EventLogBackendKind {
126    type Err = LogError;
127
128    fn from_str(value: &str) -> Result<Self, Self::Err> {
129        match value.trim().to_ascii_lowercase().as_str() {
130            "memory" => Ok(Self::Memory),
131            "file" => Ok(Self::File),
132            "sqlite" => Ok(Self::Sqlite),
133            "postgres" => Ok(Self::Postgres),
134            other => Err(LogError::Config(format!(
135                "unsupported event log backend '{other}'"
136            ))),
137        }
138    }
139}
140
141#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
142pub struct LogEvent {
143    pub kind: String,
144    pub payload: serde_json::Value,
145    #[serde(default)]
146    pub headers: BTreeMap<String, String>,
147    pub occurred_at_ms: i64,
148}
149
150impl LogEvent {
151    pub fn new(kind: impl Into<String>, payload: serde_json::Value) -> Self {
152        Self {
153            kind: kind.into(),
154            payload,
155            headers: BTreeMap::new(),
156            occurred_at_ms: util::now_ms(),
157        }
158    }
159
160    pub fn with_headers(mut self, headers: BTreeMap<String, String>) -> Self {
161        self.headers = headers;
162        self
163    }
164
165    /// Apply the unified redaction policy to this event's headers and
166    /// payload. Backends are intentionally left unaware of redaction —
167    /// emitters that need scrubbed events call this before append, and
168    /// readers that materialize events for display can apply the policy
169    /// again as defense in depth.
170    pub fn redact_in_place(&mut self, policy: &crate::redact::RedactionPolicy) {
171        self.headers = policy.redact_headers(&self.headers);
172        policy.redact_json_in_place(&mut self.payload);
173    }
174}
175
176/// Serialized event payload form for large read paths.
177///
178/// `payload` contains the original JSON bytes for backends that can expose
179/// them directly. Callers that only need to forward or hash the payload can
180/// avoid materializing a `serde_json::Value`; callers that need structured
181/// access can opt in with `payload_json`.
182#[derive(Clone, Debug, PartialEq, Eq)]
183pub struct LogEventBytes {
184    pub kind: String,
185    pub payload: Bytes,
186    pub headers: BTreeMap<String, String>,
187    pub occurred_at_ms: i64,
188}
189
190impl LogEventBytes {
191    pub fn payload_json(&self) -> Result<serde_json::Value, LogError> {
192        serde_json::from_slice(&self.payload)
193            .map_err(|error| LogError::Serde(format!("event log payload parse error: {error}")))
194    }
195
196    pub fn into_log_event(self) -> Result<LogEvent, LogError> {
197        Ok(LogEvent {
198            kind: self.kind,
199            payload: serde_json::from_slice(&self.payload).map_err(|error| {
200                LogError::Serde(format!("event log payload parse error: {error}"))
201            })?,
202            headers: self.headers,
203            occurred_at_ms: self.occurred_at_ms,
204        })
205    }
206}
207
208impl TryFrom<LogEvent> for LogEventBytes {
209    type Error = LogError;
210
211    fn try_from(event: LogEvent) -> Result<Self, Self::Error> {
212        let payload = serde_json::to_vec(&event.payload)
213            .map_err(|error| LogError::Serde(format!("event log payload encode error: {error}")))?;
214        Ok(Self {
215            kind: event.kind,
216            payload: Bytes::from(payload),
217            headers: event.headers,
218            occurred_at_ms: event.occurred_at_ms,
219        })
220    }
221}
222
223#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
224pub struct CompactReport {
225    pub removed: usize,
226    pub remaining: usize,
227    pub latest: Option<EventId>,
228    pub checkpointed: bool,
229}
230
231#[derive(Clone, Debug, PartialEq, Eq)]
232pub struct AppendOutcome {
233    pub event_id: EventId,
234    pub event: LogEvent,
235    pub inserted: bool,
236}
237
238#[derive(Clone, Copy, Debug, PartialEq, Eq)]
239pub(super) enum AppendHeadExpectation<'a> {
240    Any,
241    Exact(Option<&'a str>),
242}
243
244#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
245pub struct EventLogVerificationReport {
246    pub verified: bool,
247    pub errors: Vec<String>,
248    pub event_count: usize,
249    pub last_hash: Option<String>,
250}
251
252#[derive(Clone, Debug, PartialEq, Eq)]
253pub struct VerifiedEventSnapshot {
254    pub verification: EventLogVerificationReport,
255    pub events: Vec<(EventId, LogEvent)>,
256}
257
258#[derive(Clone, Debug, PartialEq, Eq)]
259pub struct EventLogDescription {
260    pub backend: EventLogBackendKind,
261    pub location: Option<PathBuf>,
262    pub size_bytes: Option<u64>,
263    pub queue_depth: usize,
264}
265
266#[derive(Debug)]
267pub enum LogError {
268    Config(String),
269    InvalidTopic(String),
270    InvalidConsumer(String),
271    Io(String),
272    Serde(String),
273    Sqlite(String),
274    ConsumerLagged(EventId),
275    StaleTopicHead {
276        topic: String,
277        expected: Option<String>,
278        actual: Option<String>,
279    },
280}
281
282impl fmt::Display for LogError {
283    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
284        match self {
285            Self::Config(message)
286            | Self::InvalidTopic(message)
287            | Self::InvalidConsumer(message)
288            | Self::Io(message)
289            | Self::Serde(message)
290            | Self::Sqlite(message) => message.fmt(f),
291            Self::ConsumerLagged(last_id) => {
292                write!(f, "subscriber lagged behind after event {last_id}")
293            }
294            Self::StaleTopicHead {
295                topic,
296                expected,
297                actual,
298            } => write!(
299                f,
300                "event log topic '{topic}' chain head mismatch; expected {expected:?}, found {actual:?}"
301            ),
302        }
303    }
304}
305
306impl std::error::Error for LogError {}
307
308#[allow(async_fn_in_trait)]
309pub trait EventLog: Send + Sync {
310    fn describe(&self) -> EventLogDescription;
311
312    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError>;
313
314    async fn flush(&self) -> Result<(), LogError>;
315
316    /// Read events strictly after `from`. `None` starts from the
317    /// beginning of the topic.
318    async fn read_range(
319        &self,
320        topic: &Topic,
321        from: Option<EventId>,
322        limit: usize,
323    ) -> Result<Vec<(EventId, LogEvent)>, LogError>;
324
325    async fn read_range_bytes(
326        &self,
327        topic: &Topic,
328        from: Option<EventId>,
329        limit: usize,
330    ) -> Result<Vec<(EventId, LogEventBytes)>, LogError> {
331        let events = self.read_range(topic, from, limit).await?;
332        events
333            .into_iter()
334            .map(|(event_id, event)| Ok((event_id, event.try_into()?)))
335            .collect()
336    }
337
338    /// `async fn` keeps the ergonomic generic surface; the boxed stream
339    /// preserves dyn-dispatch for callers that store `Arc<dyn EventLog>`.
340    async fn subscribe(
341        self: Arc<Self>,
342        topic: &Topic,
343        from: Option<EventId>,
344    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError>;
345
346    async fn ack(
347        &self,
348        topic: &Topic,
349        consumer: &ConsumerId,
350        up_to: EventId,
351    ) -> Result<(), LogError>;
352
353    async fn consumer_cursor(
354        &self,
355        topic: &Topic,
356        consumer: &ConsumerId,
357    ) -> Result<Option<EventId>, LogError>;
358
359    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError>;
360
361    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError>;
362}
363
364fn backend_from_env() -> Result<EventLogBackendKind, LogError> {
365    Ok(std::env::var(HARN_EVENT_LOG_BACKEND_ENV)
366        .ok()
367        .map(|value| value.parse())
368        .transpose()?
369        .unwrap_or(EventLogBackendKind::Sqlite))
370}
371
372fn queue_depth_from_env() -> usize {
373    std::env::var(HARN_EVENT_LOG_QUEUE_DEPTH_ENV)
374        .ok()
375        .and_then(|value| value.parse::<usize>().ok())
376        .unwrap_or(RuntimeLimits::DEFAULT.default_event_log_queue_depth)
377        .max(1)
378}
379
380#[derive(Clone, Debug)]
381pub struct EventLogConfig {
382    pub backend: EventLogBackendKind,
383    pub file_dir: PathBuf,
384    pub sqlite_path: PathBuf,
385    pub queue_depth: usize,
386}
387
388impl EventLogConfig {
389    pub fn for_base_dir(base_dir: &Path) -> Result<Self, LogError> {
390        let file_dir = match std::env::var(HARN_EVENT_LOG_DIR_ENV) {
391            Ok(value) if !value.trim().is_empty() => util::resolve_path(base_dir, &value),
392            _ => crate::runtime_paths::event_log_dir(base_dir),
393        };
394        let sqlite_path = match std::env::var(HARN_EVENT_LOG_SQLITE_PATH_ENV) {
395            Ok(value) if !value.trim().is_empty() => util::resolve_path(base_dir, &value),
396            _ => crate::runtime_paths::event_log_sqlite_path(base_dir),
397        };
398
399        Ok(Self {
400            backend: backend_from_env()?,
401            file_dir,
402            sqlite_path,
403            queue_depth: queue_depth_from_env(),
404        })
405    }
406
407    /// Config for a state root the caller resolved itself.
408    ///
409    /// Backend and queue depth still come from the environment — those are
410    /// operator tuning knobs that apply to whichever log the process opens.
411    /// The *location* does not: it is taken from `state_root` verbatim, so
412    /// neither `HARN_STATE_DIR` nor the `HARN_EVENT_LOG_*` path overrides can
413    /// redirect a caller that already knows where its state lives. Use this
414    /// whenever the state root is a parameter rather than ambient
415    /// configuration — see [`install_default_at_state_root`].
416    pub fn for_state_root(state_root: &Path) -> Result<Self, LogError> {
417        Ok(Self {
418            backend: backend_from_env()?,
419            file_dir: crate::runtime_paths::event_log_dir_at_state_root(state_root),
420            sqlite_path: crate::runtime_paths::event_log_sqlite_path_at_state_root(state_root),
421            queue_depth: queue_depth_from_env(),
422        })
423    }
424
425    pub fn location(&self) -> Option<PathBuf> {
426        match self.backend {
427            EventLogBackendKind::Memory => None,
428            EventLogBackendKind::File => Some(self.file_dir.clone()),
429            EventLogBackendKind::Sqlite => Some(self.sqlite_path.clone()),
430            EventLogBackendKind::Postgres => None,
431        }
432    }
433}
434
435thread_local! {
436    static ACTIVE_EVENT_LOG: RefCell<Option<Arc<AnyEventLog>>> = const { RefCell::new(None) };
437    static PENDING_DEFAULT_EVENT_LOG: RefCell<Option<EventLogConfig>> = const { RefCell::new(None) };
438}
439
440pub fn install_default_for_base_dir(base_dir: &Path) -> Result<Arc<AnyEventLog>, LogError> {
441    install_active(&EventLogConfig::for_base_dir(base_dir)?)
442}
443
444/// Open `config` and make it this thread's active log, clearing any pending
445/// lazy default so it cannot later replace what the caller just installed.
446fn install_active(config: &EventLogConfig) -> Result<Arc<AnyEventLog>, LogError> {
447    let log = open_event_log(config)?;
448    ACTIVE_EVENT_LOG.with(|slot| {
449        *slot.borrow_mut() = Some(log.clone());
450    });
451    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
452        *slot.borrow_mut() = None;
453    });
454    Ok(log)
455}
456
457/// Install the active event log at a state root the caller resolved itself.
458///
459/// The `_for_base_dir` variants derive their location through
460/// [`crate::runtime_paths::state_root`], so an absolute `HARN_STATE_DIR` in the
461/// environment silently replaces the `base_dir` they were handed. Callers that
462/// were *told* where their state lives — an orchestrator with a configured
463/// state dir, an embedder running isolated VMs side by side — must not be
464/// redirected that way, and must not have to publish their path to the whole
465/// process to be obeyed.
466pub fn install_default_at_state_root(state_root: &Path) -> Result<Arc<AnyEventLog>, LogError> {
467    install_active(&EventLogConfig::for_state_root(state_root)?)
468}
469
470pub fn install_lazy_default_for_base_dir(base_dir: &Path) -> Result<(), LogError> {
471    let config = EventLogConfig::for_base_dir(base_dir)?;
472    let has_active = ACTIVE_EVENT_LOG.with(|slot| slot.borrow().is_some());
473    if !has_active {
474        PENDING_DEFAULT_EVENT_LOG.with(|slot| {
475            *slot.borrow_mut() = Some(config);
476        });
477    }
478    Ok(())
479}
480
481pub fn install_memory_for_current_thread(queue_depth: usize) -> Arc<AnyEventLog> {
482    let log = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(queue_depth.max(1))));
483    ACTIVE_EVENT_LOG.with(|slot| {
484        *slot.borrow_mut() = Some(log.clone());
485    });
486    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
487        *slot.borrow_mut() = None;
488    });
489    log
490}
491
492pub fn install_active_event_log(log: Arc<AnyEventLog>) -> Arc<AnyEventLog> {
493    ACTIVE_EVENT_LOG.with(|slot| {
494        *slot.borrow_mut() = Some(log.clone());
495    });
496    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
497        *slot.borrow_mut() = None;
498    });
499    log
500}
501
502pub fn active_event_log() -> Option<Arc<AnyEventLog>> {
503    if let Some(log) = ACTIVE_EVENT_LOG.with(|slot| slot.borrow().clone()) {
504        return Some(log);
505    }
506
507    let config = PENDING_DEFAULT_EVENT_LOG.with(|slot| slot.borrow_mut().take())?;
508    match open_event_log(&config) {
509        Ok(log) => Some(install_active_event_log(log)),
510        Err(error) => {
511            crate::events::log_warn("event_log.init", &error.to_string());
512            None
513        }
514    }
515}
516
517/// Swap the active event log handle. Paired with `AmbientExecutionScope`'s
518/// per-poll swap.
519///
520/// The captured value is an `Arc`, so a subtask and its parent append to ONE
521/// log. Capturing the log by value would give each worker thread its own
522/// in-memory log that nothing ever reads.
523pub(crate) fn swap_active_event_log(next: Option<Arc<AnyEventLog>>) -> Option<Arc<AnyEventLog>> {
524    ACTIVE_EVENT_LOG.with(|slot| std::mem::replace(&mut *slot.borrow_mut(), next))
525}
526
527pub fn reset_active_event_log() {
528    ACTIVE_EVENT_LOG.with(|slot| {
529        *slot.borrow_mut() = None;
530    });
531    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
532        *slot.borrow_mut() = None;
533    });
534}
535
536pub fn describe_for_base_dir(base_dir: &Path) -> Result<EventLogDescription, LogError> {
537    let config = EventLogConfig::for_base_dir(base_dir)?;
538    let description = match config.backend {
539        EventLogBackendKind::Memory => EventLogDescription {
540            backend: EventLogBackendKind::Memory,
541            location: None,
542            size_bytes: None,
543            queue_depth: config.queue_depth,
544        },
545        EventLogBackendKind::File => EventLogDescription {
546            backend: EventLogBackendKind::File,
547            size_bytes: Some(util::dir_size_bytes(&config.file_dir)),
548            location: Some(config.file_dir),
549            queue_depth: config.queue_depth,
550        },
551        EventLogBackendKind::Sqlite => EventLogDescription {
552            backend: EventLogBackendKind::Sqlite,
553            size_bytes: Some(util::sqlite_size_bytes(&config.sqlite_path)),
554            location: Some(config.sqlite_path),
555            queue_depth: config.queue_depth,
556        },
557        EventLogBackendKind::Postgres => EventLogDescription {
558            backend: EventLogBackendKind::Postgres,
559            location: None,
560            size_bytes: None,
561            queue_depth: config.queue_depth,
562        },
563    };
564    Ok(description)
565}
566
567pub fn open_event_log(config: &EventLogConfig) -> Result<Arc<AnyEventLog>, LogError> {
568    match config.backend {
569        EventLogBackendKind::Memory => Ok(Arc::new(AnyEventLog::Memory(MemoryEventLog::new(
570            config.queue_depth,
571        )))),
572        EventLogBackendKind::File => Ok(Arc::new(AnyEventLog::File(FileEventLog::open(
573            config.file_dir.clone(),
574            config.queue_depth,
575        )?))),
576        EventLogBackendKind::Sqlite => Ok(Arc::new(AnyEventLog::Sqlite(SqliteEventLog::open(
577            config.sqlite_path.clone(),
578            config.queue_depth,
579        )?))),
580        EventLogBackendKind::Postgres => Err(LogError::Config(
581            "postgres event logs are host-provided; the built-in event log factory supports memory, file, and sqlite"
582                .to_string(),
583        )),
584    }
585}
586
587pub enum AnyEventLog {
588    Memory(MemoryEventLog),
589    File(FileEventLog),
590    Sqlite(SqliteEventLog),
591}
592
593impl AnyEventLog {
594    pub async fn topics(&self) -> Result<Vec<Topic>, LogError> {
595        match self {
596            Self::Memory(log) => log.topics().await,
597            Self::File(log) => log.topics(),
598            Self::Sqlite(log) => log.topics(),
599        }
600    }
601
602    pub async fn append_idempotent_by_header(
603        &self,
604        topic: &Topic,
605        header: &str,
606        value: &str,
607        event: LogEvent,
608    ) -> Result<AppendOutcome, LogError> {
609        if header.trim().is_empty() {
610            return Err(LogError::Config(
611                "idempotent append header cannot be empty".to_string(),
612            ));
613        }
614        match self {
615            Self::Memory(log) => {
616                log.append_idempotent_by_header(topic, header, value, event)
617                    .await
618            }
619            Self::File(log) => log.append_idempotent_by_header(topic, header, value, event),
620            Self::Sqlite(log) => log.append_idempotent_by_header(topic, header, value, event),
621        }
622    }
623
624    /// Append a new idempotent event only when the topic's provenance-chain
625    /// head still equals `expected_head`.
626    ///
627    /// An existing `(header, value)` identity is returned before the head is
628    /// compared, so a retry remains idempotent even after later events advance
629    /// the chain. For a new identity, lookup, head comparison, and append are
630    /// one backend-owned critical section. `None` requires an empty topic.
631    pub async fn append_idempotent_chained_by_header(
632        &self,
633        topic: &Topic,
634        header: &str,
635        value: &str,
636        expected_head: Option<&str>,
637        event: LogEvent,
638    ) -> Result<AppendOutcome, LogError> {
639        if header.trim().is_empty() {
640            return Err(LogError::Config(
641                "idempotent append header cannot be empty".to_string(),
642            ));
643        }
644        match self {
645            Self::Memory(log) => {
646                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
647                    .await
648            }
649            Self::File(log) => {
650                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
651            }
652            Self::Sqlite(log) => {
653                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
654            }
655        }
656    }
657
658    /// Recompute and verify the complete provenance chain for one topic.
659    ///
660    /// Verification starts at the topic genesis. A compacted prefix therefore
661    /// cannot be reported as a complete verified chain without a future typed
662    /// checkpoint contract that binds the retained first event to that prefix.
663    pub async fn verify_topic(
664        &self,
665        topic: &Topic,
666    ) -> Result<EventLogVerificationReport, LogError> {
667        let events = self.read_range(topic, None, usize::MAX).await?;
668        Ok(verify_topic_events(topic, &events))
669    }
670
671    /// Verify one retained topic snapshot and return only events matching a
672    /// header projection. The topic is materialized once, so typed aggregate
673    /// readers do not race or repeatedly rescan it between verification and
674    /// filtering.
675    pub async fn verified_snapshot_by_header(
676        &self,
677        topic: &Topic,
678        header: &str,
679        value: &str,
680    ) -> Result<VerifiedEventSnapshot, LogError> {
681        if header.trim().is_empty() {
682            return Err(LogError::Config(
683                "verified snapshot header cannot be empty".to_string(),
684            ));
685        }
686        let mut events = self.read_range(topic, None, usize::MAX).await?;
687        let verification = verify_topic_events(topic, &events);
688        events.retain(|(_, event)| {
689            event
690                .headers
691                .get(header)
692                .is_some_and(|candidate| candidate == value)
693        });
694        Ok(VerifiedEventSnapshot {
695            verification,
696            events,
697        })
698    }
699
700    /// Read the event previously appended under `(header, value)`, the read
701    /// counterpart of [`Self::append_idempotent_by_header`]. On the SQLite
702    /// backend this is an indexed JOIN; the memory/file dev backends scan.
703    pub async fn read_idempotent_by_header(
704        &self,
705        topic: &Topic,
706        header: &str,
707        value: &str,
708    ) -> Result<Option<(EventId, LogEvent)>, LogError> {
709        if header.trim().is_empty() {
710            return Err(LogError::Config(
711                "idempotent read header cannot be empty".to_string(),
712            ));
713        }
714        match self {
715            Self::Memory(log) => log.read_idempotent_by_header(topic, header, value).await,
716            Self::File(log) => log.read_idempotent_by_header(topic, header, value),
717            Self::Sqlite(log) => log.read_idempotent_by_header(topic, header, value),
718        }
719    }
720}
721
722pub(super) fn require_expected_topic_head(
723    topic: &Topic,
724    previous: Option<(EventId, &LogEvent)>,
725    expectation: AppendHeadExpectation<'_>,
726) -> Result<(), LogError> {
727    let AppendHeadExpectation::Exact(expected) = expectation else {
728        return Ok(());
729    };
730    let actual = previous
731        .map(|(event_id, event)| {
732            crate::provenance::event_record_hash_from_headers(topic.as_str(), event_id, event)
733        })
734        .transpose()?;
735    if actual.as_deref() == expected {
736        return Ok(());
737    }
738    Err(LogError::StaleTopicHead {
739        topic: topic.as_str().to_string(),
740        expected: expected.map(str::to_string),
741        actual,
742    })
743}
744
745fn verify_topic_events(
746    topic: &Topic,
747    events: &[(EventId, LogEvent)],
748) -> EventLogVerificationReport {
749    let mut report = EventLogVerificationReport {
750        event_count: events.len(),
751        ..EventLogVerificationReport::default()
752    };
753    let mut previous_hash: Option<String> = None;
754
755    for (event_id, event) in events {
756        match event.headers.get(crate::provenance::HEADER_SCHEMA) {
757            Some(schema) if schema == crate::provenance::EVENT_PROVENANCE_SCHEMA => {}
758            Some(schema) => report.errors.push(format!(
759                "topic {topic} event {event_id} provenance schema mismatch; expected '{}', found '{schema}'",
760                crate::provenance::EVENT_PROVENANCE_SCHEMA
761            )),
762            None => report.errors.push(format!(
763                "topic {topic} event {event_id} is missing provenance schema header"
764            )),
765        }
766
767        let found_previous = event
768            .headers
769            .get(crate::provenance::HEADER_PREV_HASH)
770            .cloned();
771        if found_previous != previous_hash {
772            report.errors.push(format!(
773                "topic {topic} event {event_id} prev_hash mismatch; expected {previous_hash:?}, found {found_previous:?}"
774            ));
775        }
776
777        let stored_hash = event
778            .headers
779            .get(crate::provenance::HEADER_RECORD_HASH)
780            .filter(|hash| !hash.trim().is_empty())
781            .cloned();
782        match stored_hash.as_deref() {
783            Some(found_hash) => {
784                match crate::provenance::compute_event_record_hash(topic.as_str(), *event_id, event)
785                {
786                    Ok(expected_hash) if expected_hash == found_hash => {}
787                    Ok(expected_hash) => report.errors.push(format!(
788                        "topic {topic} event {event_id} record_hash mismatch; expected {expected_hash}, found {found_hash}"
789                    )),
790                    Err(error) => report.errors.push(format!(
791                        "topic {topic} event {event_id} record_hash verification failed: {error}"
792                    )),
793                }
794            }
795            None => report.errors.push(format!(
796                "topic {topic} event {event_id} is missing provenance record_hash header"
797            )),
798        }
799
800        previous_hash = stored_hash;
801    }
802
803    report.last_hash = previous_hash;
804    report.verified = report.errors.is_empty();
805    report
806}
807
808impl EventLog for AnyEventLog {
809    fn describe(&self) -> EventLogDescription {
810        match self {
811            Self::Memory(log) => log.describe(),
812            Self::File(log) => log.describe(),
813            Self::Sqlite(log) => log.describe(),
814        }
815    }
816
817    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError> {
818        match self {
819            Self::Memory(log) => log.append(topic, event).await,
820            Self::File(log) => log.append(topic, event).await,
821            Self::Sqlite(log) => log.append(topic, event).await,
822        }
823    }
824
825    async fn flush(&self) -> Result<(), LogError> {
826        match self {
827            Self::Memory(log) => log.flush().await,
828            Self::File(log) => log.flush().await,
829            Self::Sqlite(log) => log.flush().await,
830        }
831    }
832
833    async fn read_range(
834        &self,
835        topic: &Topic,
836        from: Option<EventId>,
837        limit: usize,
838    ) -> Result<Vec<(EventId, LogEvent)>, LogError> {
839        match self {
840            Self::Memory(log) => log.read_range(topic, from, limit).await,
841            Self::File(log) => log.read_range(topic, from, limit).await,
842            Self::Sqlite(log) => log.read_range(topic, from, limit).await,
843        }
844    }
845
846    async fn read_range_bytes(
847        &self,
848        topic: &Topic,
849        from: Option<EventId>,
850        limit: usize,
851    ) -> Result<Vec<(EventId, LogEventBytes)>, LogError> {
852        match self {
853            Self::Memory(log) => log.read_range_bytes(topic, from, limit).await,
854            Self::File(log) => log.read_range_bytes(topic, from, limit).await,
855            Self::Sqlite(log) => log.read_range_bytes(topic, from, limit).await,
856        }
857    }
858
859    async fn subscribe(
860        self: Arc<Self>,
861        topic: &Topic,
862        from: Option<EventId>,
863    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError> {
864        let (rx, queue_depth) = match self.as_ref() {
865            Self::Memory(log) => (
866                log.broadcasts.subscribe(topic, log.queue_depth),
867                log.queue_depth,
868            ),
869            Self::File(log) => (
870                log.broadcasts.subscribe(topic, log.queue_depth),
871                log.queue_depth,
872            ),
873            Self::Sqlite(log) => (
874                log.broadcasts.subscribe(topic, log.queue_depth),
875                log.queue_depth,
876            ),
877        };
878        let history = self.read_range(topic, from, usize::MAX).await?;
879        Ok(util::stream_from_broadcast(history, from, rx, queue_depth))
880    }
881
882    async fn ack(
883        &self,
884        topic: &Topic,
885        consumer: &ConsumerId,
886        up_to: EventId,
887    ) -> Result<(), LogError> {
888        match self {
889            Self::Memory(log) => log.ack(topic, consumer, up_to).await,
890            Self::File(log) => log.ack(topic, consumer, up_to).await,
891            Self::Sqlite(log) => log.ack(topic, consumer, up_to).await,
892        }
893    }
894
895    async fn consumer_cursor(
896        &self,
897        topic: &Topic,
898        consumer: &ConsumerId,
899    ) -> Result<Option<EventId>, LogError> {
900        match self {
901            Self::Memory(log) => log.consumer_cursor(topic, consumer).await,
902            Self::File(log) => log.consumer_cursor(topic, consumer).await,
903            Self::Sqlite(log) => log.consumer_cursor(topic, consumer).await,
904        }
905    }
906
907    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError> {
908        match self {
909            Self::Memory(log) => log.latest(topic).await,
910            Self::File(log) => log.latest(topic).await,
911            Self::Sqlite(log) => log.latest(topic).await,
912        }
913    }
914
915    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError> {
916        match self {
917            Self::Memory(log) => log.compact(topic, before).await,
918            Self::File(log) => log.compact(topic, before).await,
919            Self::Sqlite(log) => log.compact(topic, before).await,
920        }
921    }
922}
923
924pub fn sanitize_topic_component(value: &str) -> String {
925    value
926        .chars()
927        .map(|ch| {
928            if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-') {
929                ch
930            } else {
931                '_'
932            }
933        })
934        .collect()
935}
harn_vm/event_log/mod.rs

harn_vm/event_log/
mod.rs
