task: T0.5 (implementer)

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"

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"] }

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"] }

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<bool, StorageWriteError>;
+
+    /// 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<u64, StorageWriteError>;
+
+    /// Bulk-append `events`. Returns `(last_lsn, num_inserted)` tuple in success 
+    /// case; empty batches return an error.  
+    async fn batch_append(&self, events: Vec<EventIntent>) -> 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<Vec<ExportRecord>, 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<u64>;
+
+}  
+
+/// 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")),  
+        }
+    }
+}

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<Self> 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<Self> 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) 
+}
+
+

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<u8>,
+}
+
+/// 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<u8>,
+}
+
+/// 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,
+}