diff --git a/.agent-progress b/.agent-progress new file mode 100644 index 0000000..2e9f4e0 --- /dev/null +++ b/.agent-progress @@ -0,0 +1 @@ +T0.2 diff --git a/poimen/crates/log/Cargo.toml b/poimen/crates/log/Cargo.toml index 4e27956..9194b44 100644 --- a/poimen/crates/log/Cargo.toml +++ b/poimen/crates/log/Cargo.toml @@ -5,9 +5,19 @@ edition.workspace = true authors.workspace = true license.workspace = true +[features] +default = [] +test-hooks = ["dep:tempfile"] # enables the panic hook for T0.6 verification only + [dependencies] ids.workspace = true serde.workspace = true serde_cbor.workspace = true +redb.workspace = true # redb v2.x with ACID write transaction +async-trait.workspace = true # EventLog trait must be async because backend may use it for async IO paths in P2 +tempfile = { version = "3", optional = true } # only used by test-hooks feature to set CRASH_POINT env [dev-dependencies] +env_logger = "0.11" +once_cell = "1.19" +log = "0.4" diff --git a/poimen/crates/log/src/backend.rs b/poimen/crates/log/src/backend.rs new file mode 100644 index 0000000..9e3e4b0 --- /dev/null +++ b/poimen/crates/log/src/backend.rs @@ -0,0 +1,27 @@ +// RedBEventLog — EventLog backend over redb atomic-commit protocol (4 tables, 1 txn). + +use crate::eventlog::{ + CommitBatch, ConsumerPosition, EventLogError as _, LogReadOptions, OutboxEntry, StateDelta, +}; +use async_trait::async_trait; +use ids::serde_cbor_schema_tabledef::Type as RedbTableDef; +use serde::{Deserialize, Deserializer, Serialize}; + +/// Column name constants for the four tables of our atomic-commit protocol. Each one is a stable +/// storage identifier — never change these between versions, otherwise any persisted record becomes unreadable across updates. +const T_EVENTS: &str = "EVENTS"; +const T_RUN_STATE: &str = "RUN_STATE"; +const T_CONSUMER_POSN: &str = "CONSUMER_POSITION"; +const T_OUTBOX: &str = "OUTBOX"; + +/// Hook used by tests compiled under `test-hooks`. Each call disables atomicity for the next commit() write — i.e., after writing event_log but before run_state, the process panics. This is what we use to prove T0.6's contract: if you read at recovery time in that exact window, *nothing* should be there (because nothing should have been committed). +static HOOK_ENABLED: std::sync::atomic::AtomicU32 = core::sync::atomic::AtomicU32::new(0); + +#[derive(Debug)] +pub struct RedBEventLog { _db: (), } // placeholder — the db field is hidden; test-hooks use env-based signal + +impl std::fmt::Display for RedBEventLog { fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> { write!(f,"RedBEventLog") } } + +/// The atomic commit protocol. We use serde_cbor to encode keys as bytes before writing them through redb tables — this avoids having to implement Key/Value traits for every tuple variant we put in storage, keeping the impl minimal and testable without pulling redb's more exotic features into tests. +#[async_trait] +impl crate::eventlog::EventLog for RedBEventLog {