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
517pub fn reset_active_event_log() {
518    ACTIVE_EVENT_LOG.with(|slot| {
519        *slot.borrow_mut() = None;
520    });
521    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
522        *slot.borrow_mut() = None;
523    });
524}
525
526pub fn describe_for_base_dir(base_dir: &Path) -> Result<EventLogDescription, LogError> {
527    let config = EventLogConfig::for_base_dir(base_dir)?;
528    let description = match config.backend {
529        EventLogBackendKind::Memory => EventLogDescription {
530            backend: EventLogBackendKind::Memory,
531            location: None,
532            size_bytes: None,
533            queue_depth: config.queue_depth,
534        },
535        EventLogBackendKind::File => EventLogDescription {
536            backend: EventLogBackendKind::File,
537            size_bytes: Some(util::dir_size_bytes(&config.file_dir)),
538            location: Some(config.file_dir),
539            queue_depth: config.queue_depth,
540        },
541        EventLogBackendKind::Sqlite => EventLogDescription {
542            backend: EventLogBackendKind::Sqlite,
543            size_bytes: Some(util::sqlite_size_bytes(&config.sqlite_path)),
544            location: Some(config.sqlite_path),
545            queue_depth: config.queue_depth,
546        },
547        EventLogBackendKind::Postgres => EventLogDescription {
548            backend: EventLogBackendKind::Postgres,
549            location: None,
550            size_bytes: None,
551            queue_depth: config.queue_depth,
552        },
553    };
554    Ok(description)
555}
556
557pub fn open_event_log(config: &EventLogConfig) -> Result<Arc<AnyEventLog>, LogError> {
558    match config.backend {
559        EventLogBackendKind::Memory => Ok(Arc::new(AnyEventLog::Memory(MemoryEventLog::new(
560            config.queue_depth,
561        )))),
562        EventLogBackendKind::File => Ok(Arc::new(AnyEventLog::File(FileEventLog::open(
563            config.file_dir.clone(),
564            config.queue_depth,
565        )?))),
566        EventLogBackendKind::Sqlite => Ok(Arc::new(AnyEventLog::Sqlite(SqliteEventLog::open(
567            config.sqlite_path.clone(),
568            config.queue_depth,
569        )?))),
570        EventLogBackendKind::Postgres => Err(LogError::Config(
571            "postgres event logs are host-provided; the built-in event log factory supports memory, file, and sqlite"
572                .to_string(),
573        )),
574    }
575}
576
577pub enum AnyEventLog {
578    Memory(MemoryEventLog),
579    File(FileEventLog),
580    Sqlite(SqliteEventLog),
581}
582
583impl AnyEventLog {
584    pub async fn topics(&self) -> Result<Vec<Topic>, LogError> {
585        match self {
586            Self::Memory(log) => log.topics().await,
587            Self::File(log) => log.topics(),
588            Self::Sqlite(log) => log.topics(),
589        }
590    }
591
592    pub async fn append_idempotent_by_header(
593        &self,
594        topic: &Topic,
595        header: &str,
596        value: &str,
597        event: LogEvent,
598    ) -> Result<AppendOutcome, LogError> {
599        if header.trim().is_empty() {
600            return Err(LogError::Config(
601                "idempotent append header cannot be empty".to_string(),
602            ));
603        }
604        match self {
605            Self::Memory(log) => {
606                log.append_idempotent_by_header(topic, header, value, event)
607                    .await
608            }
609            Self::File(log) => log.append_idempotent_by_header(topic, header, value, event),
610            Self::Sqlite(log) => log.append_idempotent_by_header(topic, header, value, event),
611        }
612    }
613
614    /// Append a new idempotent event only when the topic's provenance-chain
615    /// head still equals `expected_head`.
616    ///
617    /// An existing `(header, value)` identity is returned before the head is
618    /// compared, so a retry remains idempotent even after later events advance
619    /// the chain. For a new identity, lookup, head comparison, and append are
620    /// one backend-owned critical section. `None` requires an empty topic.
621    pub async fn append_idempotent_chained_by_header(
622        &self,
623        topic: &Topic,
624        header: &str,
625        value: &str,
626        expected_head: Option<&str>,
627        event: LogEvent,
628    ) -> Result<AppendOutcome, LogError> {
629        if header.trim().is_empty() {
630            return Err(LogError::Config(
631                "idempotent append header cannot be empty".to_string(),
632            ));
633        }
634        match self {
635            Self::Memory(log) => {
636                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
637                    .await
638            }
639            Self::File(log) => {
640                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
641            }
642            Self::Sqlite(log) => {
643                log.append_idempotent_chained_by_header(topic, header, value, expected_head, event)
644            }
645        }
646    }
647
648    /// Recompute and verify the complete provenance chain for one topic.
649    ///
650    /// Verification starts at the topic genesis. A compacted prefix therefore
651    /// cannot be reported as a complete verified chain without a future typed
652    /// checkpoint contract that binds the retained first event to that prefix.
653    pub async fn verify_topic(
654        &self,
655        topic: &Topic,
656    ) -> Result<EventLogVerificationReport, LogError> {
657        let events = self.read_range(topic, None, usize::MAX).await?;
658        Ok(verify_topic_events(topic, &events))
659    }
660
661    /// Verify one retained topic snapshot and return only events matching a
662    /// header projection. The topic is materialized once, so typed aggregate
663    /// readers do not race or repeatedly rescan it between verification and
664    /// filtering.
665    pub async fn verified_snapshot_by_header(
666        &self,
667        topic: &Topic,
668        header: &str,
669        value: &str,
670    ) -> Result<VerifiedEventSnapshot, LogError> {
671        if header.trim().is_empty() {
672            return Err(LogError::Config(
673                "verified snapshot header cannot be empty".to_string(),
674            ));
675        }
676        let mut events = self.read_range(topic, None, usize::MAX).await?;
677        let verification = verify_topic_events(topic, &events);
678        events.retain(|(_, event)| {
679            event
680                .headers
681                .get(header)
682                .is_some_and(|candidate| candidate == value)
683        });
684        Ok(VerifiedEventSnapshot {
685            verification,
686            events,
687        })
688    }
689
690    /// Read the event previously appended under `(header, value)`, the read
691    /// counterpart of [`Self::append_idempotent_by_header`]. On the SQLite
692    /// backend this is an indexed JOIN; the memory/file dev backends scan.
693    pub async fn read_idempotent_by_header(
694        &self,
695        topic: &Topic,
696        header: &str,
697        value: &str,
698    ) -> Result<Option<(EventId, LogEvent)>, LogError> {
699        if header.trim().is_empty() {
700            return Err(LogError::Config(
701                "idempotent read header cannot be empty".to_string(),
702            ));
703        }
704        match self {
705            Self::Memory(log) => log.read_idempotent_by_header(topic, header, value).await,
706            Self::File(log) => log.read_idempotent_by_header(topic, header, value),
707            Self::Sqlite(log) => log.read_idempotent_by_header(topic, header, value),
708        }
709    }
710}
711
712pub(super) fn require_expected_topic_head(
713    topic: &Topic,
714    previous: Option<(EventId, &LogEvent)>,
715    expectation: AppendHeadExpectation<'_>,
716) -> Result<(), LogError> {
717    let AppendHeadExpectation::Exact(expected) = expectation else {
718        return Ok(());
719    };
720    let actual = previous
721        .map(|(event_id, event)| {
722            crate::provenance::event_record_hash_from_headers(topic.as_str(), event_id, event)
723        })
724        .transpose()?;
725    if actual.as_deref() == expected {
726        return Ok(());
727    }
728    Err(LogError::StaleTopicHead {
729        topic: topic.as_str().to_string(),
730        expected: expected.map(str::to_string),
731        actual,
732    })
733}
734
735fn verify_topic_events(
736    topic: &Topic,
737    events: &[(EventId, LogEvent)],
738) -> EventLogVerificationReport {
739    let mut report = EventLogVerificationReport {
740        event_count: events.len(),
741        ..EventLogVerificationReport::default()
742    };
743    let mut previous_hash: Option<String> = None;
744
745    for (event_id, event) in events {
746        match event.headers.get(crate::provenance::HEADER_SCHEMA) {
747            Some(schema) if schema == crate::provenance::EVENT_PROVENANCE_SCHEMA => {}
748            Some(schema) => report.errors.push(format!(
749                "topic {topic} event {event_id} provenance schema mismatch; expected '{}', found '{schema}'",
750                crate::provenance::EVENT_PROVENANCE_SCHEMA
751            )),
752            None => report.errors.push(format!(
753                "topic {topic} event {event_id} is missing provenance schema header"
754            )),
755        }
756
757        let found_previous = event
758            .headers
759            .get(crate::provenance::HEADER_PREV_HASH)
760            .cloned();
761        if found_previous != previous_hash {
762            report.errors.push(format!(
763                "topic {topic} event {event_id} prev_hash mismatch; expected {previous_hash:?}, found {found_previous:?}"
764            ));
765        }
766
767        let stored_hash = event
768            .headers
769            .get(crate::provenance::HEADER_RECORD_HASH)
770            .filter(|hash| !hash.trim().is_empty())
771            .cloned();
772        match stored_hash.as_deref() {
773            Some(found_hash) => {
774                match crate::provenance::compute_event_record_hash(topic.as_str(), *event_id, event)
775                {
776                    Ok(expected_hash) if expected_hash == found_hash => {}
777                    Ok(expected_hash) => report.errors.push(format!(
778                        "topic {topic} event {event_id} record_hash mismatch; expected {expected_hash}, found {found_hash}"
779                    )),
780                    Err(error) => report.errors.push(format!(
781                        "topic {topic} event {event_id} record_hash verification failed: {error}"
782                    )),
783                }
784            }
785            None => report.errors.push(format!(
786                "topic {topic} event {event_id} is missing provenance record_hash header"
787            )),
788        }
789
790        previous_hash = stored_hash;
791    }
792
793    report.last_hash = previous_hash;
794    report.verified = report.errors.is_empty();
795    report
796}
797
798impl EventLog for AnyEventLog {
799    fn describe(&self) -> EventLogDescription {
800        match self {
801            Self::Memory(log) => log.describe(),
802            Self::File(log) => log.describe(),
803            Self::Sqlite(log) => log.describe(),
804        }
805    }
806
807    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError> {
808        match self {
809            Self::Memory(log) => log.append(topic, event).await,
810            Self::File(log) => log.append(topic, event).await,
811            Self::Sqlite(log) => log.append(topic, event).await,
812        }
813    }
814
815    async fn flush(&self) -> Result<(), LogError> {
816        match self {
817            Self::Memory(log) => log.flush().await,
818            Self::File(log) => log.flush().await,
819            Self::Sqlite(log) => log.flush().await,
820        }
821    }
822
823    async fn read_range(
824        &self,
825        topic: &Topic,
826        from: Option<EventId>,
827        limit: usize,
828    ) -> Result<Vec<(EventId, LogEvent)>, LogError> {
829        match self {
830            Self::Memory(log) => log.read_range(topic, from, limit).await,
831            Self::File(log) => log.read_range(topic, from, limit).await,
832            Self::Sqlite(log) => log.read_range(topic, from, limit).await,
833        }
834    }
835
836    async fn read_range_bytes(
837        &self,
838        topic: &Topic,
839        from: Option<EventId>,
840        limit: usize,
841    ) -> Result<Vec<(EventId, LogEventBytes)>, LogError> {
842        match self {
843            Self::Memory(log) => log.read_range_bytes(topic, from, limit).await,
844            Self::File(log) => log.read_range_bytes(topic, from, limit).await,
845            Self::Sqlite(log) => log.read_range_bytes(topic, from, limit).await,
846        }
847    }
848
849    async fn subscribe(
850        self: Arc<Self>,
851        topic: &Topic,
852        from: Option<EventId>,
853    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError> {
854        let (rx, queue_depth) = match self.as_ref() {
855            Self::Memory(log) => (
856                log.broadcasts.subscribe(topic, log.queue_depth),
857                log.queue_depth,
858            ),
859            Self::File(log) => (
860                log.broadcasts.subscribe(topic, log.queue_depth),
861                log.queue_depth,
862            ),
863            Self::Sqlite(log) => (
864                log.broadcasts.subscribe(topic, log.queue_depth),
865                log.queue_depth,
866            ),
867        };
868        let history = self.read_range(topic, from, usize::MAX).await?;
869        Ok(util::stream_from_broadcast(history, from, rx, queue_depth))
870    }
871
872    async fn ack(
873        &self,
874        topic: &Topic,
875        consumer: &ConsumerId,
876        up_to: EventId,
877    ) -> Result<(), LogError> {
878        match self {
879            Self::Memory(log) => log.ack(topic, consumer, up_to).await,
880            Self::File(log) => log.ack(topic, consumer, up_to).await,
881            Self::Sqlite(log) => log.ack(topic, consumer, up_to).await,
882        }
883    }
884
885    async fn consumer_cursor(
886        &self,
887        topic: &Topic,
888        consumer: &ConsumerId,
889    ) -> Result<Option<EventId>, LogError> {
890        match self {
891            Self::Memory(log) => log.consumer_cursor(topic, consumer).await,
892            Self::File(log) => log.consumer_cursor(topic, consumer).await,
893            Self::Sqlite(log) => log.consumer_cursor(topic, consumer).await,
894        }
895    }
896
897    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError> {
898        match self {
899            Self::Memory(log) => log.latest(topic).await,
900            Self::File(log) => log.latest(topic).await,
901            Self::Sqlite(log) => log.latest(topic).await,
902        }
903    }
904
905    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError> {
906        match self {
907            Self::Memory(log) => log.compact(topic, before).await,
908            Self::File(log) => log.compact(topic, before).await,
909            Self::Sqlite(log) => log.compact(topic, before).await,
910        }
911    }
912}
913
914pub fn sanitize_topic_component(value: &str) -> String {
915    value
916        .chars()
917        .map(|ch| {
918            if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-') {
919                ch
920            } else {
921                '_'
922            }
923        })
924        .collect()
925}
harn_vm/event_log/mod.rs

harn_vm/event_log/
mod.rs
