From 15dd1b337f1540e769980ba05b7d421b4b697ac5 Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Sun, 21 Dec 2025 21:56:05 +0100 Subject: [PATCH] feat: Add `Store` module for persistent KV state, update `lib.rs` and documentation including a development journal and architecture details. --- Cargo.toml | 1 + docs/architecture.md | 191 ++++++++++++++++++--- docs/roadmap.md | 9 + lattice-core/Cargo.toml | 1 + lattice-core/src/lib.rs | 3 + lattice-core/src/store.rs | 342 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 525 insertions(+), 22 deletions(-) create mode 100644 lattice-core/src/store.rs diff --git a/Cargo.toml b/Cargo.toml index 7e98ab5..19dd5bb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ bytes = "1" dirs = "5" blake3 = "1" hex = "0.4" +redb = "2" # Testing tokio-test = "0.4" diff --git a/docs/architecture.md b/docs/architecture.md index 05ac616..cfb47f3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -2,20 +2,38 @@ ## Ideas -- SigChains Ed25519-signed, hash-chained append-only logs per node. Trust via local signature verification. -- Log-Based State: KV store derived by replaying entries. Watermarks enable safe log pruning + snapshots. -- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect—converges mathematically. -- Full Replication: All nodes keep all logs until watermark consensus, then prune and snapshot. +**Core:** +- SigChains: Ed25519-signed, hash-chained append-only logs per node. +- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect. +- Full Replication: All nodes keep all logs until watermark consensus, then prune. + +**State:** +- Log-Based State: KV store derived from entries. Watermarks enable pruning + snapshots. +- Merkle-ized State: state.db as Merkle tree. O(1) sync checks, efficient diffing, light clients. +- DAG Conflict Resolution: Entries track ancestry. Forks merge on next write. Tips only in state.db. +- KV Snapshots: Point-in-time snapshots for log pruning, fast bootstrap, time travel. + +**Operations:** +- Atomic Batch Writes: Multiple key updates as single entry. +- Conditional Updates (CAS): Update only if current value matches expected hash. + +**CRDTs:** +- LWW-Register: Last-writer-wins for single values. +- LWW-Element-Set: Set with add/remove, element present if add > remove timestamp. ## Concepts -- Transitive Pairing. Nodes can introduce new nodes to the mesh. +- Transitive Pairing: Nodes can introduce new nodes to the mesh. +- Multi-Mesh: A node can participate in multiple meshes (clusters). Each mesh is a group of nodes sharing data. +- Manifest Store: Joining a mesh means joining a special KV store of type "manifest" that defines the mesh membership. The manifest contains node info (`/nodes/{pubkey}/...`). ## Stack - rust - iroh - prost protocol buffers +- redb (embedded KV store) +- rustyline (interactive CLI) ### Bootstrap @@ -58,11 +76,62 @@ Future: ### Data Model -- Keys are flat strings using path conventions (e.g., `/nodes/{pubkey}`, `/config/sync/interval`). -- Prefix queries via string matching (sorted map enables efficient range scans). -- State is computed by replaying `Put`/`Delete` operations from all authors. +- Multiple KV stores supported, identified by `store_id` (UUID). +- Keys: Arbitrary byte arrays (`Vec`), sorted lexicographically. +- Values: Arbitrary byte arrays (`Vec`). +- Each store defines its own key/value format — applications know their schema. +- Logs are per `(store_id, author_id)` tuple. +- State is maintained by tracking the "frontier" (tips) of the causal graph for each key. - Entry ordering: by HLC timestamp, then by author ID as tiebreaker. -- Conflicts resolved by last-write-wins (using the ordering above). + +**Sync vs Causality:** +- Vector Clocks track log coverage ("I have entries from Node A up to seq 50") — syncing files. +- DAG Parents track data causality ("This value replaces that value") — resolving key conflicts. + +#### DAG Conflict Resolution + +Instead of simple LWW where newest timestamp blindly overwrites, every entry tracks its ancestry: + +**Data Model:** +- Each entry includes `parent_hashes` — references to the entries it supersedes +- History forms a DAG (directed acyclic graph), not a linear chain +- state.db stores only "tips" (heads) of the graph per key + +**Life Cycle:** + +1. **Write (normal):** New entry points to previous entry's hash as parent. History is a straight line. + +2. **Write (concurrent/offline):** Two nodes edit same key independently, both pointing to same old parent. History forks into two branches. + +3. **Read (forked):** System sees multiple valid values. Uses deterministic rule (highest HLC, then author_id tiebreaker) to return one "winner". No error thrown. + +4. **Merge (healing):** Next write to that key cites both existing branches as parents. Fork merges back to single tip. + +**Example: Partial Write (Branch Extension)** + +``` +Initial: Heads = {A, B} where A(ts:100), B(ts:105). Read winner = B. + +Offline node C wakes up, only knows A (hasn't seen B). +C writes "v3" with parent = [A]. + +Result: Heads = {C, B}. Conflict shifted, not resolved. + C(ts:110) > B(ts:105), so C wins reads. + + ┌──> [A] ──> [C:110] +[Root]─┤ + └──> [B:105] + +Later: A synced node writes D with parents = [C, B]. +Result: Heads = {D}. Fork merged. +``` + +This preserves B's work even though C never saw it. Naive LWW would lose B forever. + +#### Store Consistency Modes + +- **Eventually consistent**: Default. Writes accepted locally, sync happens async. Fast, offline-capable. +- **Strictly consistent**: Writes require quorum acknowledgment before commit. Slower, requires connectivity. ### Timestamps (Hybrid Logical Clocks) @@ -86,27 +155,80 @@ Authors apply their own entries through the standard receive path to ensure cons Each node stores logs as one file per author: ``` -data/ -├── identity.key # Local node's Ed25519 private key -├── logs/ -│ └── {author_id_hex}.log # Append-only SignedEntry stream per author -└── state.db # redb: KV snapshot + vector clocks + indexes +~/.local/share/lattice/ +├── identity.key # Ed25519 private key +├── stores/ +│ └── {store_uuid}/ +│ ├── logs/ +│ │ └── {author_id_hex}.log # Append-only SignedEntry stream +│ └── state.db # redb: KV snapshot + frontiers +└── meta.db # redb: global metadata (known stores, peers) ``` -- Logs: Append-only binary files per author, containing serialized `SignedEntry` messages. -- State DB (redb): Combined KV state, vector clocks, and indexes. Updated as entries are applied. +- Logs: Append-only binary files per `(store, author)`, containing serialized `SignedEntry` messages. +- State DB (redb): Per-store KV state and frontiers. Updated as entries are applied. -#### state.db Tables (redb) +#### state.db Tables (per store, redb) ``` -Table Key Value Purpose +Table Key Value Purpose ───────────────────────────────────────────────────────────────────────────── -kv String (path) Vec Replicated key-value data -vector_clocks [u8; 32] (author_id) (u64 seq, [u8; 32] hash) Track sync state + chain verification -entry_index (author_id, seq) u64 (offset) Fast entry lookup by position -meta String Vec System metadata (own_seq, watermark, etc.) +kv Vec (key) Vec Current tips for each key +applied_frontiers [u8; 32] (author_id) (u64 seq, [u8; 32] hash) What's applied to this store +meta Vec Vec Store metadata (incl. merkle_root) ``` +`HeadInfo: { value: Vec, hlc: u64, author: [u8;32], hash: [u8;32] }` + +Note: KV stores multiple heads per key to support DAG conflict resolution. Reads pick winner deterministically. + +#### meta.db Tables (global, redb) + +``` +Table Key Value Purpose +───────────────────────────────────────────────────────────────────────────── +stores UUID (store_id) StoreInfo Known stores this node participates in +meta String Vec Global metadata (node_id, etc.) +``` + +StoreInfo: `{ type: "manifest" | "data", name, created_at, ... }` +- Manifest stores define mesh membership via KV entries (`/nodes/{pubkey}/...`) +- Data stores hold application data +- Node's list of manifest store IDs = meshes it belongs to + +#### In-Memory Structures + +- log_frontiers: `HashMap` — rebuilt from log files on startup + +### Operation Flow (put/delete) + +``` +1. User calls put("/key", value) + │ + ▼ +2. SigChain.create_entry() + - Build Entry with parent_hashes (current tips for key) + - Sign it → SignedEntry + │ + ▼ +3. Append to log + Gossip (critical path) + - Write to author's log file + - Update log_frontiers (in-memory) + - Broadcast to peers + │ + ▼ +4. Apply to state.db (background) + - Update kv heads (merge parent tips into new tip) + - Update applied_frontiers + - Update merkle_root hash +``` + +Fast path (1-3): durable + distributed. Background (4): queryable state. + +### Read Flow (get) + +`get(key)` reads directly from local state.db. Reads are eventually consistent — if state.db lags behind the log, the read may return slightly stale data. + ### Watermarks - Nodes gossip their watermarks periodically (throttled). @@ -114,6 +236,11 @@ meta String Vec System metada - All nodes keep all logs (own + others) for redundancy until watermark consensus. - Once all peers have acknowledged entries, they can be pruned and replaced by the snapshot. - If a node is offline too long, it re-bootstraps with a fresh snapshot when it reconnects. +- Note: Consider preserving logs longer than required for redundancy — enables time travel (view state at any point in history). + +**Pruning and DAG Parents:** +- If a new entry references a parent that was pruned, accept it only if strictly newer than snapshot timestamp. +- Snapshots act as the base; entries referencing parents older than snapshot are roots relative to that snapshot. ### Rich CRDTs (Future) @@ -157,3 +284,23 @@ message MergeOp { ``` Recommendation: Use Put/Delete for 90% of data. Add CRDT primitives only when needed (concurrent counters, lists) rather than a scripting language. + +## Open Questions + +### Permissions + +Write permissions are enforceable cryptographically: +- Every entry is signed by author +- Nodes verify signature before accepting +- Manifest defines allowed writers: `/nodes/{pubkey}/role` = `writer` | `reader` +- Entries from non-writers are rejected + +Read permissions are not enforceable: +- Sharing a store = granting read access +- Encryption adds a layer but doesn't solve revocation (once you have the key, you can read past data) +- True revocation is impossible — you can't "unread" data + +Practical model: +- Share store = grant read +- Write access defined in manifest +- Read-only nodes replicate and verify but can't contribute entries \ No newline at end of file diff --git a/docs/roadmap.md b/docs/roadmap.md index ba32453..bbb468a 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -29,6 +29,10 @@ Current code assumes single store. Changes needed: - [ ] Store → per-store state.db, not global - [ ] Log paths → `stores/{uuid}/logs/{author}.log` - [ ] Add global meta.db for stores table +- [ ] Proto: SignedEntry/messages need store_id (UUID) +- [ ] Proto: Entry needs `parent_hashes` for DAG (not just `prev_hash`) +- [ ] Store: keys/values → `Vec` (binary, not String) +- [ ] Store: KV table value → `Vec` for DAG heads - [ ] CLI → `create-store`, `list-stores`, `use ` --- @@ -74,3 +78,8 @@ Current code assumes single store. Changes needed: - Snapshots for fast bootstrap - FUSE filesystem mount - Note: FUSE requires u64 inode numbers → maintain `BiMap` in redb +- Merkle-ized State + - state.db as Merkle tree with signed root hash + - O(1) sync checks (compare root), efficient binary-search diffing + - Light clients: fetch value + Merkle proof, verify without full state + - Trade-off: write amplification, requires deterministic tree (Patricia Trie / Merkle Search Tree) diff --git a/lattice-core/Cargo.toml b/lattice-core/Cargo.toml index 7696c22..77c4dd9 100644 --- a/lattice-core/Cargo.toml +++ b/lattice-core/Cargo.toml @@ -14,6 +14,7 @@ bytes = { workspace = true } dirs = { workspace = true } blake3 = { workspace = true } hex = { workspace = true } +redb = { workspace = true } [build-dependencies] prost-build = { workspace = true } diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index 888e5dd..88f6b54 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -11,6 +11,7 @@ //! - **DataDir**: Platform-specific data directory paths //! - **SignedEntry**: Entry creation, signing, and verification //! - **Log**: Append-only log file I/O +//! - **Store**: Persistent KV state from log replay pub mod node; pub mod sigchain; @@ -22,6 +23,7 @@ pub mod proto; pub mod data_dir; pub mod signed_entry; pub mod log; +pub mod store; // Constants /// Maximum size of a serialized SignedEntry (16 MB) @@ -36,3 +38,4 @@ pub use clock::{Clock, SystemClock, MockClock}; pub use data_dir::DataDir; pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signed_entry}; pub use log::{append_entry, read_entries, LogReader}; +pub use store::Store; diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs new file mode 100644 index 0000000..1bee710 --- /dev/null +++ b/lattice-core/src/store.rs @@ -0,0 +1,342 @@ +//! Store - persistent KV state from log replay +//! +//! Uses redb for efficient embedded storage. +//! Tables: +//! - kv: String → Vec (replicated key-value data) +//! - meta: String → Vec (system metadata: own_seq, last_hash, etc.) + +use crate::log::{read_entries, LogError}; +use crate::proto::{operation, Entry, SignedEntry}; +use prost::Message; +use redb::{Database, ReadableTable, TableDefinition}; +use std::path::Path; +use thiserror::Error; + +// Table definitions +const KV_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("kv"); +const META_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("meta"); + +// Meta keys +const META_LAST_SEQ: &str = "last_seq"; +const META_LAST_HASH: &str = "last_hash"; + +/// Errors that can occur during store operations +#[derive(Error, Debug)] +pub enum StoreError { + #[error("Database error: {0}")] + Database(#[from] redb::DatabaseError), + + #[error("Table error: {0}")] + Table(#[from] redb::TableError), + + #[error("Transaction error: {0}")] + Transaction(#[from] redb::TransactionError), + + #[error("Commit error: {0}")] + Commit(#[from] redb::CommitError), + + #[error("Storage error: {0}")] + Storage(#[from] redb::StorageError), + + #[error("Log error: {0}")] + Log(#[from] LogError), + + #[error("Decode error: {0}")] + Decode(#[from] prost::DecodeError), +} + +/// Persistent store for KV state +pub struct Store { + db: Database, +} + +impl Store { + /// Open or create a store at the given path + pub fn open(path: impl AsRef) -> Result { + let db = Database::create(path)?; + + // Ensure tables exist + let write_txn = db.begin_write()?; + { + let _ = write_txn.open_table(KV_TABLE)?; + let _ = write_txn.open_table(META_TABLE)?; + } + write_txn.commit()?; + + Ok(Self { db }) + } + + /// Replay a log file and apply all entries to the store + pub fn replay_log(&self, log_path: impl AsRef) -> Result { + let entries = read_entries(log_path)?; + let mut count = 0u64; + + for signed_entry in entries { + self.apply_entry(&signed_entry)?; + count += 1; + } + + Ok(count) + } + + /// Apply a single signed entry to the store + pub fn apply_entry(&self, signed_entry: &SignedEntry) -> Result<(), StoreError> { + let entry = Entry::decode(&signed_entry.entry_bytes[..])?; + + let write_txn = self.db.begin_write()?; + { + let mut kv_table = write_txn.open_table(KV_TABLE)?; + let mut meta_table = write_txn.open_table(META_TABLE)?; + + // Apply operations + for op in entry.ops { + if let Some(op_type) = op.op_type { + match op_type { + operation::OpType::Put(put) => { + kv_table.insert(put.key.as_str(), put.value.as_slice())?; + } + operation::OpType::Delete(del) => { + kv_table.remove(del.key.as_str())?; + } + } + } + } + + // Update meta + meta_table.insert(META_LAST_SEQ, &entry.seq.to_le_bytes()[..])?; + + // Compute and store hash of this entry + let hash: [u8; 32] = blake3::hash(&signed_entry.entry_bytes).into(); + meta_table.insert(META_LAST_HASH, &hash[..])?; + } + write_txn.commit()?; + + Ok(()) + } + + /// Get a value by key + pub fn get(&self, key: &str) -> Result>, StoreError> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(KV_TABLE)?; + + Ok(table.get(key)?.map(|v| v.value().to_vec())) + } + + /// Put a value (use SigChain.create_entry for proper signing) + /// This is a low-level method for direct writes + pub fn put(&self, key: &str, value: &[u8]) -> Result<(), StoreError> { + let write_txn = self.db.begin_write()?; + { + let mut table = write_txn.open_table(KV_TABLE)?; + table.insert(key, value)?; + } + write_txn.commit()?; + Ok(()) + } + + /// Delete a key + pub fn delete(&self, key: &str) -> Result { + let write_txn = self.db.begin_write()?; + let removed; + { + let mut table = write_txn.open_table(KV_TABLE)?; + removed = table.remove(key)?.is_some(); + } + write_txn.commit()?; + Ok(removed) + } + + /// Get the last applied sequence number + pub fn last_seq(&self) -> Result { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(META_TABLE)?; + + Ok(table.get(META_LAST_SEQ)? + .map(|v| { + let bytes: [u8; 8] = v.value().try_into().unwrap_or([0u8; 8]); + u64::from_le_bytes(bytes) + }) + .unwrap_or(0)) + } + + /// Get the hash of the last applied entry + pub fn last_hash(&self) -> Result<[u8; 32], StoreError> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(META_TABLE)?; + + Ok(table.get(META_LAST_HASH)? + .map(|v| { + let bytes: [u8; 32] = v.value().try_into().unwrap_or([0u8; 32]); + bytes + }) + .unwrap_or([0u8; 32])) + } + + /// Set a meta value + pub fn set_meta(&self, key: &str, value: &[u8]) -> Result<(), StoreError> { + let write_txn = self.db.begin_write()?; + { + let mut table = write_txn.open_table(META_TABLE)?; + table.insert(key, value)?; + } + write_txn.commit()?; + Ok(()) + } + + /// Get a meta value + pub fn get_meta(&self, key: &str) -> Result>, StoreError> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(META_TABLE)?; + + Ok(table.get(key)?.map(|v| v.value().to_vec())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::clock::MockClock; + use crate::hlc::HLC; + use crate::log::append_entry; + use crate::node::Node; + use crate::signed_entry::EntryBuilder; + use std::env::temp_dir; + use std::path::PathBuf; + + fn temp_paths(name: &str) -> (PathBuf, PathBuf) { + let base = temp_dir().join(format!("lattice_store_test_{}", name)); + (base.with_extension("db"), base.with_extension("log")) + } + + #[test] + fn test_open_store() { + let (db_path, _) = temp_paths("open"); + std::fs::remove_file(&db_path).ok(); + + let store = Store::open(&db_path).unwrap(); + + assert_eq!(store.last_seq().unwrap(), 0); + assert_eq!(store.last_hash().unwrap(), [0u8; 32]); + + std::fs::remove_file(&db_path).ok(); + } + + #[test] + fn test_put_get() { + let (db_path, _) = temp_paths("put_get"); + std::fs::remove_file(&db_path).ok(); + + let store = Store::open(&db_path).unwrap(); + + store.put("key1", b"value1").unwrap(); + store.put("key2", b"value2").unwrap(); + + assert_eq!(store.get("key1").unwrap(), Some(b"value1".to_vec())); + assert_eq!(store.get("key2").unwrap(), Some(b"value2".to_vec())); + assert_eq!(store.get("key3").unwrap(), None); + + std::fs::remove_file(&db_path).ok(); + } + + #[test] + fn test_delete() { + let (db_path, _) = temp_paths("delete"); + std::fs::remove_file(&db_path).ok(); + + let store = Store::open(&db_path).unwrap(); + + store.put("key", b"value").unwrap(); + assert!(store.delete("key").unwrap()); + assert_eq!(store.get("key").unwrap(), None); + assert!(!store.delete("key").unwrap()); // Already gone + + std::fs::remove_file(&db_path).ok(); + } + + #[test] + fn test_replay_log() { + let (db_path, log_path) = temp_paths("replay"); + std::fs::remove_file(&db_path).ok(); + std::fs::remove_file(&log_path).ok(); + + let node = Node::generate(); + let clock = MockClock::new(1000); + + // Create log entries + let mut prev_hash = [0u8; 32]; + for i in 1..=3 { + let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock)) + .prev_hash(prev_hash.to_vec()) + .put(format!("/key/{}", i), format!("value{}", i).into_bytes()) + .sign(&node); + + // Compute hash for next entry + let bytes = entry.encode_to_vec(); + prev_hash = blake3::hash(&bytes).into(); + + append_entry(&log_path, &entry).unwrap(); + } + + // Replay + let store = Store::open(&db_path).unwrap(); + let count = store.replay_log(&log_path).unwrap(); + + assert_eq!(count, 3); + assert_eq!(store.last_seq().unwrap(), 3); + assert_eq!(store.get("/key/1").unwrap(), Some(b"value1".to_vec())); + assert_eq!(store.get("/key/2").unwrap(), Some(b"value2".to_vec())); + assert_eq!(store.get("/key/3").unwrap(), Some(b"value3".to_vec())); + + std::fs::remove_file(&db_path).ok(); + std::fs::remove_file(&log_path).ok(); + } + + #[test] + fn test_replay_with_delete() { + let (db_path, log_path) = temp_paths("replay_delete"); + std::fs::remove_file(&db_path).ok(); + std::fs::remove_file(&log_path).ok(); + + let node = Node::generate(); + let clock = MockClock::new(1000); + + // Entry 1: put key + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .prev_hash([0u8; 32].to_vec()) + .put("/key", b"value".to_vec()) + .sign(&node); + append_entry(&log_path, &entry1).unwrap(); + + // Entry 2: delete key + let hash1: [u8; 32] = blake3::hash(&entry1.encode_to_vec()).into(); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock)) + .prev_hash(hash1.to_vec()) + .delete("/key") + .sign(&node); + append_entry(&log_path, &entry2).unwrap(); + + // Replay + let store = Store::open(&db_path).unwrap(); + store.replay_log(&log_path).unwrap(); + + assert_eq!(store.get("/key").unwrap(), None); + assert_eq!(store.last_seq().unwrap(), 2); + + std::fs::remove_file(&db_path).ok(); + std::fs::remove_file(&log_path).ok(); + } + + #[test] + fn test_meta_accessors() { + let (db_path, _) = temp_paths("meta"); + std::fs::remove_file(&db_path).ok(); + + let store = Store::open(&db_path).unwrap(); + + store.set_meta("custom_key", b"custom_value").unwrap(); + assert_eq!(store.get_meta("custom_key").unwrap(), Some(b"custom_value".to_vec())); + assert_eq!(store.get_meta("missing").unwrap(), None); + + std::fs::remove_file(&db_path).ok(); + } +}