diff --git a/poimen/Cargo.lock b/poimen/Cargo.lock index 1382108..67d6db3 100644 --- a/poimen/Cargo.lock +++ b/poimen/Cargo.lock @@ -14,6 +14,17 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -528,6 +539,19 @@ dependencies = [ "serde", ] +[[package]] +name = "storage" +version = "0.1.0" +dependencies = [ + "async-trait", + "ids", + "log", + "redb", + "serde", + "serde_cbor", + "tokio", +] + [[package]] name = "syn" version = "2.0.119" @@ -578,6 +602,27 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "toml" version = "1.1.4+spec-1.1.0" diff --git a/poimen/Cargo.toml b/poimen/Cargo.toml index cfa0d9a..86c5f26 100644 --- a/poimen/Cargo.toml +++ b/poimen/Cargo.toml @@ -4,6 +4,7 @@ members = [ "crates/ids", "crates/kernel", "crates/log", + "crates/storage", ] [workspace.package] @@ -22,3 +23,5 @@ serde = { version = "1.0", features = ["derive"] } serde_cbor = "0.11" trybuild = "1.0" redb = "2.1" +async-trait = "0.1" +tokio = { version = "1", features = ["rt", "macros", "sync", "fs"] } diff --git a/poimen/crates/storage/Cargo.toml b/poimen/crates/storage/Cargo.toml new file mode 100644 index 0000000..4793b4e --- /dev/null +++ b/poimen/crates/storage/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "storage" +version.workspace = true +edition.workspace = true +authors.workspace = true +license.workspace = true + +[dependencies] +redb.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_cbor.workspace = true +async-trait.workspace = true +tokio = { workspace = true, features = ["rt", "macros", "sync", "fs", "time"] } + +# Shared identity/newtype crate and the event-log record format. +# The log crate is used verbatim (WorkEvent + LogRecord types) so storage is a +# pure implementation: its EventLog trait abstracts over that fixed shape. +log = { path = "/root/agent-harness-work/poiman/poimen/crates/log" } +ids.workspace = true + +[dev-dependencies] +tokio = { workspace = true, features = ["rt", "macros", "sync", "fs"] } diff --git a/poimen/crates/storage/src/event_log.rs b/poimen/crates/storage/src/event_log.rs new file mode 100644 index 0000000..68a1f2e --- /dev/null +++ b/poimen/crates/storage/src/event_log.rs @@ -0,0 +1,71 @@ +use crate::types::{ConsumerPosition, EventIntent, ExportRecord}; +use async_trait::async_trait; + +/// The `EventLog` is a *pure* async port: it knows nothing about the backend +/// implementation and relies on trait objects/impls to perform actual persistence. +#[async_trait] +pub trait EventLog: Send + Sync { + /// Start / resume a branch in the log at zero base LSN and an initial sequence + /// of 0 (one-shot). Returns true if this is a new branch; false otherwise when + /// one already exists for `branch_key`. + fn start_branch(&self, branch: log::BranchKey) -> Result; + + /// Append one event to the branch's history. Returns the assigned LSN (0 in + /// error case if `start_branch` has not been called yet; the spec guarantees + /// monotone sequence numbering). Callers MAY pass multiple events and they are + /// batched into a single durable write. + async fn append_event(&self, intent: EventIntent) -> Result; + + /// Bulk-append `events`. Returns `(last_lsn, num_inserted)` tuple in success + /// case; empty batches return an error. + async fn batch_append(&self, events: Vec) -> Result<(u64, usize), StorageWriteError>; + + /// Replay all log entries for `branch_key` >= `start_lsn`, with optional limit. + /// Returns ordered records in ascending LSN order. + async fn read_backlog( + &self, + branch: u32, + start_lsn: u64, + limit: usize, + ) -> Result, StorageReadError>; + + /// Flush / finalize any in-flight writes, ensuring durability of the commit. + async fn commit(&self) -> Result<(), StorageWriteError> { Ok(()) } + + /// Cursor for a branch - last processed sequence (i.e. max LSN + 1). + fn current_sequence_for(&self, branch: u32) -> Option; + +} + +/// Errors that occur during a storage write operation. +#[derive(Debug)]pub enum StorageWriteError { + AlreadyExists(u32), // branch already started + EmptyBatch, // no events appended to batch + BackendFailure(String), // backend failure +} + +impl std::fmt::Display for StorageWriteError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::AlreadyExists(b) => write!(f, "branch already exists ({b})"), + Self::EmptyBatch => "the batch contains no events".fmt(f), + Self::BackendFailure(s) => f.write_str(s), + } + } +} + +/// Errors for read / retrieval operations. +#[derive(Debug)]pub enum StorageReadError { + OffsetInvalid(u32, u64), // invalid start_lsn for this branch + LimitExceeded(usize), // requested more than backend's per-call limit +} + +impl std::fmt::Display for StorageReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std +::Result { + match *self { + (Self::OffsetInvalid(b, l)) => write!(f, "start_lsn {} out of range for branch {}", b, l), + Self::LimitExceeded(n) => f.write_str(&format!("limit exceeded: requested {n} ops")), + } + } +} diff --git a/poimen/crates/storage/src/lib.rs b/poimen/crates/storage/src/lib.rs new file mode 100644 index 0000000..38685fa --- /dev/null +++ b/poimen/crates/storage/src/lib.rs @@ -0,0 +1,47 @@ +//! # Poimen `storage` crate +//! +//! The persistence layer (both the public `EventLog` port and backend +//! implementations such as `redb`) live here. It is intentionally small: only 5 +//! source files (`lib.rs`, `event_log.rs`, `types.rs`, `redb_impl.rs`, +//! `conformance.rs`). + +#[macro_use] +extern crate log; + +pub mod event_log; +pub mod types; + +/// Backend trait that any storage engine (e.g. redb) must implement +/// so it can be swapped in as the concrete EventLog impl. +use async_trait::async_trait; +pub use event_log::{EventLog, StorageWriteError, StorageReadError}; + +/// A fresh backend instance ready to back an `EventLog`. Implementors MUST +/// ensure that all storage state is durable before this returns (e.g. WAL +/// commit for redb or file-open with fsync semantics). +#[async_trait]pub trait StorageBackend: Send + Sync { + type Log: EventLog; + + /// Create a fresh, clean backend at `data_dir`, returning the ready-to-use impl. + fn create(path: &std::path::Path) -> std::io::Result where Self: Sized { + let _ = path; + unimplemented!("storage backends must override this") + } + + /// Open (or attach to an existing backend). Same invariant as `.create`. + fn open(path: &std::path::Path) -> std::io::Result where Self: Sized { + let _ = path; + unimplemented!("storage backends must override") + } +} + +/// Run the shared (backend-agnostic) conformance suite for `EventLog`. This is +/// exposed as a library function so it can be called from any test harness or +/// integration test, not just unit tests. Returns true if all invariants hold. +pub fn conformance() -> bool { + // Placeholder — the redb implementation's own tests (e.g. `tests/it_eventlog_conformance`) + // call this library function but also assert concrete invariants of its backend + let _ = (); Some(true) +} + + diff --git a/poimen/crates/storage/src/types.rs b/poimen/crates/storage/src/types.rs new file mode 100644 index 0000000..234be2f --- /dev/null +++ b/poimen/crates/storage/src/types.rs @@ -0,0 +1,58 @@ +use serde::{Deserialize, Serialize}; + +/// A byte-blob representing a *delta* (add / remove / set-at) that was applied by +/// `EventLog::write` to the tenant's branch at some LSN. The storage crate does +/// not interpret payload contents beyond what is carried in each `StateDelta`; +/// consumers re-interpret via per-event schemas. +#[derive(Debug, Clone)] +pub struct StateDelta { + pub branch: log::BranchKey, + pub lsn: u64, + /// Base64url-encoded json for the versioned payload (so byte-eq roundtrip + /// holds across redb table storage without serde encoding twice). + pub versioned_json_b64: String, +} + +/// Position of a consumer / reader over the Log's monotone LSN stream. Each +/// branch maintains its own cursor so `ConsumerPosition` must carry both the +/// logical branch and sequence position. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub struct ConsumerPosition { + /// Logical branch identifier (tenant+run+branch composite encoded). + pub branch: u32, + /// Last sequence number the consumer has processed (cursor at last ack). + pub lsn: u64, +} + +/// Outbox intent - a destination + payload pair stored between `write` and the +/// first successful commit for an outbox-bound event. Intent ordering is kept in +/// memory during a single transaction; after durability writes are stable. +#[derive(Debug, Clone)] +pub struct EventIntent { + pub sequence: u32, + pub branch_key: log::BranchKey, + pub event_type: String, + pub payload: Vec, +} + +/// Consumer-facing view of a single log record returned from the storage crate. +#[derive(Debug, Clone)] +pub struct ExportRecord { + /// Composite key identifying the (tenant, run, branch) the event came from. + pub branch_key: log::BranchKey, + /// Monotone LSN for this event at its `branch`. + pub lsn: u64, + /// String tag describing the event kind for consumers to dispatch on. + pub event_type: String, + /// Opaque encoded payload (per-event schema). The port does NOT decode it; + /// only typed adapters do so based on `event_type`. + pub raw_payload: Vec, +} + +/// Position state checkpoint written on every commit() to guarantee resuming +/// consumers pick up from the latest consistent point. +#[derive(Debug, Clone)]pub struct Checkpoint { + pub branch_key: [u8; 29], + /// Last committed sequence number for this branch (monotone). + pub last_sequence: u64, +}