From 57c2906b105ee4600eb520e3f9af9e5abfbf1ad5 Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Mon, 22 Dec 2025 01:56:26 +0100 Subject: [PATCH] feat: Implement DAG-based conflict resolution with binary keys and multi-head CLI display --- docs/architecture.md | 5 +- docs/roadmap.md | 25 +- lattice-cli/src/commands.rs | 72 ++- lattice-cli/src/node.rs | 44 +- lattice-core/src/lib.rs | 1 + lattice-core/src/proto.rs | 3 +- lattice-core/src/sigchain.rs | 2 +- lattice-core/src/signed_entry.rs | 19 +- lattice-core/src/store.rs | 987 +++++++++++++++++++++++++------ proto/lattice.proto | 32 +- 10 files changed, 950 insertions(+), 240 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index ca4dc9e..d548b7d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -303,4 +303,7 @@ Read permissions are not enforceable: 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 +- Read-only nodes replicate and verify but can't contribute entries + +Future: +- Capability-based permissions: Explore finer-grained write access (e.g., per-key or per-prefix permissions) via capabilities. Exact mechanism TBD. \ No newline at end of file diff --git a/docs/roadmap.md b/docs/roadmap.md index cbabdc2..561a8a2 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -33,26 +33,27 @@ --- -## Milestone 1.5: DAG Conflict Resolution ← NEXT +## Milestone 1.5: DAG Conflict Resolution **Goal:** Upgrade store from simple LWW to DAG-based conflict resolution per architecture.md. ### Deliverables -- [ ] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality, separate from sigchain `prev_hash`) -- [ ] Store: keys → `Vec` (binary, not String) -- [ ] Store: KV table schema → `Vec → Vec` where `HeadInfo = { value, hlc, author, hash }` -- [ ] Store: `applied_frontiers` table → `author_id → (seq, hash)` per author -- [ ] Store: `apply_entry` → track multiple heads, merge parent tips into new tip -- [ ] Store: `get` → deterministic winner from heads (highest HLC, author_id tiebreaker) -- [ ] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry +- [x] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality) +- [x] Proto: Add `HeadInfo` message for multi-head storage +- [x] Store: KV table schema → `Vec → Vec` +- [x] Store: `apply_entry` → track multiple heads, merge parent tips +- [x] Store: `get` → deterministic winner (highest HLC, author tiebreaker) +- [x] Store: `get_heads` → inspect all heads for a key +- [x] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry +- [x] CLI: Show conflict indicator when multiple heads ### Success Criteria -- Concurrent writes to same key create multiple heads -- Reads return deterministic winner -- Next write citing both heads merges fork to single tip -- All existing tests still pass +- [x] Concurrent writes to same key create multiple heads +- [x] Reads return deterministic winner +- [x] Next write citing both heads merges fork to single tip +- [x] All existing tests still pass (71 tests) --- diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index 141c9fd..599fce5 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -67,10 +67,10 @@ pub fn commands() -> Vec { }, Command { name: "get", - args: "", + args: " [-v]", description: "Retrieve a value by key", min_args: 1, - max_args: 1, + max_args: 2, handler: cmd_get, }, Command { @@ -240,7 +240,7 @@ fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> return CommandResult::Ok; }; let start = Instant::now(); - match h.put(&args[0], args[1].as_bytes()) { + match h.put(args[0].as_bytes(), args[1].as_bytes()) { Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), Err(e) => eprintln!("Error: {}", e), } @@ -252,14 +252,48 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; }; + let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false); let start = Instant::now(); - match h.get(&args[0]) { - Ok(Some(v)) => { - println!("{}", format_value(&v)); - println!("({:.2?})", start.elapsed()); + let key = args[0].as_bytes(); + + if verbose { + // Show all heads + match h.get_heads(key) { + Ok(heads) if heads.is_empty() => println!("(nil)"), + Ok(heads) => { + for (i, head) in heads.iter().enumerate() { + let winner = if i == 0 { "→" } else { " " }; + let tombstone = if head.tombstone { "⊗" } else { "" }; + let author_short = hex::encode(&head.author).chars().take(8).collect::(); + if head.tombstone { + println!("{} {} (deleted) (hlc:{}, author:{})", + winner, tombstone, head.hlc, author_short); + } else { + println!("{} {} (hlc:{}, author:{})", + winner, format_value(&head.value), head.hlc, author_short); + } + } + if heads.len() > 1 { + println!("⚠ {} heads (conflict)", heads.len()); + } + println!("({:.2?})", start.elapsed()); + } + Err(e) => eprintln!("Error: {}", e), + } + } else { + match h.get(key) { + Ok(Some(v)) => { + let heads = h.get_heads(key).unwrap_or_default(); + if heads.len() > 1 { + println!("{} (⚠ {} heads)", format_value(&v), heads.len()); + } else { + println!("{}", format_value(&v)); + } + println!("({:.2?})", start.elapsed()); + } + Ok(None) => println!("(nil)"), + Err(e) => eprintln!("Error: {}", e), } - Ok(None) => println!("(nil)"), - Err(e) => eprintln!("Error: {}", e), } CommandResult::Ok } @@ -270,7 +304,7 @@ fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) return CommandResult::Ok; }; let start = Instant::now(); - match h.delete(&args[0]) { + match h.delete(args[0].as_bytes()) { Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), Err(e) => eprintln!("Error: {}", e), } @@ -290,10 +324,24 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) - println!("(empty)"); } else { for (k, v) in &entries { + let key_str = format_value(k); if verbose { - println!("{} = {} ({} bytes)", k, format_value(v), v.len()); + // Show all heads for this key + let heads = h.get_heads(k).unwrap_or_default(); + println!("{}:", key_str); + for (i, head) in heads.iter().enumerate() { + let winner = if i == 0 { "→" } else { " " }; + let author_short = hex::encode(&head.author).chars().take(8).collect::(); + if head.tombstone { + println!(" {} ⊗ (deleted) (hlc:{}, author:{})", + winner, head.hlc, author_short); + } else { + println!(" {} {} (hlc:{}, author:{})", + winner, format_value(&head.value), head.hlc, author_short); + } + } } else { - println!("{} = {}", k, format_value(v)); + println!("{} = {}", key_str, format_value(v)); } } println!("({} keys, {:.2?})", entries.len(), start.elapsed()); diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs index eb8eba1..99871d2 100644 --- a/lattice-cli/src/node.rs +++ b/lattice-cli/src/node.rs @@ -189,11 +189,15 @@ pub struct StoreHandle { impl StoreHandle { pub fn id(&self) -> Uuid { self.store_id } - pub fn get(&self, key: &str) -> Result>, NodeError> { + pub fn get(&self, key: &[u8]) -> Result>, NodeError> { Ok(self.store.get(key)?) } - pub fn list(&self) -> Result)>, NodeError> { + pub fn get_heads(&self, key: &[u8]) -> Result, NodeError> { + Ok(self.store.get_heads(key)?) + } + + pub fn list(&self) -> Result, Vec)>, NodeError> { Ok(self.store.list_all()?) } @@ -202,18 +206,29 @@ impl StoreHandle { } pub fn applied_seq(&self) -> Result { - Ok(self.store.last_seq()?) + let author = self.node.public_key_bytes(); + Ok(self.store.author_state(&author)? + .map(|s| s.seq) + .unwrap_or(0)) } - pub fn put(&self, key: &str, value: &[u8]) -> Result { - self.commit_entry(|b| b.put(key, value.to_vec())) + pub fn put(&self, key: &[u8], value: &[u8]) -> Result { + // Get current heads for this key to cite as parents + let heads = self.store.get_heads(key)?; + let parent_hashes: Vec> = heads.iter().map(|h| h.hash.clone()).collect(); + + self.commit_entry(parent_hashes, |b| b.put(key.to_vec(), value.to_vec())) } - pub fn delete(&self, key: &str) -> Result { - self.commit_entry(|b| b.delete(key)) + pub fn delete(&self, key: &[u8]) -> Result { + // Get current heads for this key to cite as parents + let heads = self.store.get_heads(key)?; + let parent_hashes: Vec> = heads.iter().map(|h| h.hash.clone()).collect(); + + self.commit_entry(parent_hashes, |b| b.delete(key.to_vec())) } - fn commit_entry(&self, build: F) -> Result + fn commit_entry(&self, parent_hashes: Vec>, build: F) -> Result where F: FnOnce(EntryBuilder) -> EntryBuilder, { @@ -223,7 +238,8 @@ impl StoreHandle { let builder = EntryBuilder::new(seq, HLC::now()) .store_id(self.store_id.as_bytes().to_vec()) - .prev_hash(prev_hash.to_vec()); + .prev_hash(prev_hash.to_vec()) + .parent_hashes(parent_hashes); let entry = build(builder).sign(&self.node); sigchain.append(&entry)?; @@ -261,8 +277,8 @@ mod tests { assert!(stores.contains(&store_id)); let (handle, _) = node.open_store(store_id).expect("Failed to open store"); - handle.put("/key", b"value").expect("put failed"); - assert_eq!(handle.get("/key").unwrap(), Some(b"value".to_vec())); + handle.put(b"/key", b"value").expect("put failed"); + assert_eq!(handle.get(b"/key").unwrap(), Some(b"value".to_vec())); let _ = std::fs::remove_dir_all(data_dir.base()); } @@ -279,12 +295,12 @@ mod tests { let store_b = node.create_store().expect("create B"); let (handle_a, _) = node.open_store(store_a).expect("open A"); - handle_a.put("/key", b"from A").expect("put A"); + handle_a.put(b"/key", b"from A").expect("put A"); let (handle_b, _) = node.open_store(store_b).expect("open B"); - assert_eq!(handle_b.get("/key").unwrap(), None); + assert_eq!(handle_b.get(b"/key").unwrap(), None); - assert_eq!(handle_a.get("/key").unwrap(), Some(b"from A".to_vec())); + assert_eq!(handle_a.get(b"/key").unwrap(), Some(b"from A".to_vec())); let _ = std::fs::remove_dir_all(data_dir.base()); } diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index c431a0b..2b820b2 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -41,4 +41,5 @@ pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signe pub use log::{append_entry, read_entries, LogReader}; pub use store::Store; pub use meta_store::MetaStore; +pub use proto::HeadInfo; pub use uuid::Uuid; diff --git a/lattice-core/src/proto.rs b/lattice-core/src/proto.rs index 18139ca..e42a125 100644 --- a/lattice-core/src/proto.rs +++ b/lattice-core/src/proto.rs @@ -33,6 +33,7 @@ mod tests { version: 1, store_id: vec![1u8; 16], prev_hash: vec![0u8; 32], + parent_hashes: vec![], seq: 5, timestamp: Some(Hlc { wall_time: 1000, @@ -41,7 +42,7 @@ mod tests { ops: vec![ Operation { op_type: Some(operation::OpType::Put(PutOp { - key: "/nodes/abc".to_string(), + key: b"/nodes/abc".to_vec(), value: b"hello".to_vec(), })), }, diff --git a/lattice-core/src/sigchain.rs b/lattice-core/src/sigchain.rs index f3ff1fe..d81fb5b 100644 --- a/lattice-core/src/sigchain.rs +++ b/lattice-core/src/sigchain.rs @@ -451,7 +451,7 @@ mod tests { let ops = vec![ Operation { op_type: Some(operation::OpType::Put(PutOp { - key: "/test".to_string(), + key: b"/test".to_vec(), value: b"hello".to_vec(), })), }, diff --git a/lattice-core/src/signed_entry.rs b/lattice-core/src/signed_entry.rs index 030c51f..eff9b08 100644 --- a/lattice-core/src/signed_entry.rs +++ b/lattice-core/src/signed_entry.rs @@ -34,6 +34,7 @@ pub struct EntryBuilder { version: u32, store_id: Vec, prev_hash: Vec, + parent_hashes: Vec>, seq: u64, timestamp: HLC, ops: Vec, @@ -44,8 +45,9 @@ impl EntryBuilder { pub fn new(seq: u64, timestamp: HLC) -> Self { Self { version: 1, - store_id: Vec::new(), // Empty = legacy single-store - prev_hash: vec![0u8; 32], // Genesis or will be set + store_id: Vec::new(), + prev_hash: vec![0u8; 32], + parent_hashes: Vec::new(), seq, timestamp, ops: Vec::new(), @@ -58,14 +60,20 @@ impl EntryBuilder { self } - /// Set the previous entry hash (for chaining) + /// Set the previous entry hash (for sigchain linking) pub fn prev_hash(mut self, hash: impl Into>) -> Self { self.prev_hash = hash.into(); self } + /// Set the parent hashes (for DAG ancestry) + pub fn parent_hashes(mut self, hashes: Vec>) -> Self { + self.parent_hashes = hashes; + self + } + /// Add a Put operation - pub fn put(mut self, key: impl Into, value: impl Into>) -> Self { + pub fn put(mut self, key: impl Into>, value: impl Into>) -> Self { self.ops.push(Operation { op_type: Some(operation::OpType::Put(PutOp { key: key.into(), @@ -76,7 +84,7 @@ impl EntryBuilder { } /// Add a Delete operation - pub fn delete(mut self, key: impl Into) -> Self { + pub fn delete(mut self, key: impl Into>) -> Self { self.ops.push(Operation { op_type: Some(operation::OpType::Delete(DeleteOp { key: key.into(), @@ -97,6 +105,7 @@ impl EntryBuilder { version: self.version, store_id: self.store_id, prev_hash: self.prev_hash, + parent_hashes: self.parent_hashes, seq: self.seq, timestamp: Some(Hlc { wall_time: self.timestamp.wall_time, diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index 24ed473..7618d9e 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -1,24 +1,22 @@ -//! Store - persistent KV state from log replay +//! Store - persistent KV state with DAG-based conflict resolution //! //! Uses redb for efficient embedded storage. //! Tables: -//! - kv: String → Vec (replicated key-value data) -//! - meta: String → Vec (system metadata: own_seq, last_hash, etc.) +//! - kv: Vec → HeadList (multi-head DAG tips per key) +//! - meta: String → Vec (system metadata: last_seq, last_hash, etc.) +//! - author: [u8; 32] → AuthorState (per-author replay tracking) use crate::log::{read_entries, LogError}; -use crate::proto::{operation, Entry, SignedEntry}; +use crate::proto::{operation, AuthorState, Entry, HeadInfo, HeadList, SignedEntry}; +use crate::signed_entry::hash_signed_entry; 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"; +const KV_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("kv"); +const AUTHOR_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("author"); /// Errors that can occur during store operations #[derive(Error, Debug)] @@ -45,7 +43,7 @@ pub enum StoreError { Decode(#[from] prost::DecodeError), } -/// Persistent store for KV state +/// Persistent store for KV state with DAG conflict resolution pub struct Store { db: Database, } @@ -59,7 +57,7 @@ impl Store { let write_txn = db.begin_write()?; { let _ = write_txn.open_table(KV_TABLE)?; - let _ = write_txn.open_table(META_TABLE)?; + let _ = write_txn.open_table(AUTHOR_TABLE)?; } write_txn.commit()?; @@ -76,10 +74,10 @@ impl Store { 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)?; + let mut author_table = write_txn.open_table(AUTHOR_TABLE)?; for signed_entry in &entries { - Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut meta_table)?; + Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut author_table)?; } } write_txn.commit()?; @@ -92,8 +90,8 @@ impl Store { 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)?; - Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut meta_table)?; + let mut author_table = write_txn.open_table(AUTHOR_TABLE)?; + Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut author_table)?; } write_txn.commit()?; Ok(()) @@ -102,99 +100,153 @@ impl Store { /// Internal: apply operations from a signed entry to tables fn apply_ops_to_tables( signed_entry: &SignedEntry, - kv_table: &mut redb::Table<&str, &[u8]>, - meta_table: &mut redb::Table<&str, &[u8]>, + kv_table: &mut redb::Table<&[u8], &[u8]>, + author_table: &mut redb::Table<&[u8], &[u8]>, ) -> Result<(), StoreError> { let entry = Entry::decode(&signed_entry.entry_bytes[..])?; + let entry_hash = hash_signed_entry(signed_entry); + let entry_hlc = entry.timestamp.as_ref().map(|t| (t.wall_time << 16) | t.counter as u64).unwrap_or(0); + let author: [u8; 32] = signed_entry.author_id.clone().try_into().unwrap_or([0u8; 32]); + + // Check if entry was already applied (per-author seq check) + if let Some(author_state_bytes) = author_table.get(&author[..])? { + if let Ok(author_state) = AuthorState::decode(author_state_bytes.value()) { + if entry.seq <= author_state.seq { + return Ok(()); // Already applied, skip + } + } + } - // 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())?; + let new_head = HeadInfo { + value: put.value, + hlc: entry_hlc, + author: author.to_vec(), + hash: entry_hash.to_vec(), + tombstone: false, + }; + Self::apply_head(kv_table, &put.key, new_head, &entry.parent_hashes)?; } operation::OpType::Delete(del) => { - kv_table.remove(del.key.as_str())?; + let tombstone = HeadInfo { + value: vec![], + hlc: entry_hlc, + author: author.to_vec(), + hash: entry_hash.to_vec(), + tombstone: true, + }; + Self::apply_head(kv_table, &del.key, tombstone, &entry.parent_hashes)?; } } } } - // Update meta - meta_table.insert(META_LAST_SEQ, &entry.seq.to_le_bytes()[..])?; - - // Hash the full SignedEntry (includes signature), not just entry_bytes - let hash: [u8; 32] = blake3::hash(&signed_entry.encode_to_vec()).into(); - meta_table.insert(META_LAST_HASH, &hash[..])?; + // Update per-author state + let author_state = AuthorState { + seq: entry.seq, + hash: entry_hash.to_vec(), + log_offset: 0, // TODO: track actual log offset + }; + author_table.insert(&author[..], author_state.encode_to_vec().as_slice())?; Ok(()) } - /// Get a value by key - pub fn get(&self, key: &str) -> Result>, StoreError> { + /// Apply a new head to a key, removing ancestor heads (idempotent) + fn apply_head( + kv_table: &mut redb::Table<&[u8], &[u8]>, + key: &[u8], + new_head: HeadInfo, + parent_hashes: &[Vec], + ) -> Result<(), StoreError> { + let mut heads = match kv_table.get(key)? { + Some(v) => HeadList::decode(v.value()).map(|h| h.heads).unwrap_or_default(), + None => Vec::new(), + }; + + // Idempotency: skip if this entry was already applied + if heads.iter().any(|h| h.hash == new_head.hash) { + return Ok(()); + } + + // Remove any heads that are ancestors (their hash is in parent_hashes) + heads.retain(|h| !parent_hashes.iter().any(|p| p == &h.hash)); + + // Add new head + heads.push(new_head); + + let encoded = HeadList { heads }.encode_to_vec(); + kv_table.insert(key, encoded.as_slice())?; + Ok(()) + } + + /// Get a value by key (returns deterministic winner from heads, None if tombstone) + pub fn get(&self, key: &[u8]) -> 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())) + match table.get(key)? { + Some(v) => { + let heads = HeadList::decode(v.value())?.heads; + match Self::pick_winner(&heads) { + Some(winner) if winner.tombstone => Ok(None), + Some(winner) => Ok(Some(winner.value.clone())), + None => Ok(None), + } + } + None => Ok(None), + } } - /// List all key-value pairs - pub fn list_all(&self) -> Result)>, StoreError> { + /// Get all heads for a key (for conflict inspection) + pub fn get_heads(&self, key: &[u8]) -> Result, StoreError> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(KV_TABLE)?; + + match table.get(key)? { + Some(v) => Ok(HeadList::decode(v.value())?.heads), + None => Ok(Vec::new()), + } + } + + /// Pick deterministic winner from heads: highest HLC, then highest author bytes + fn pick_winner(heads: &[HeadInfo]) -> Option<&HeadInfo> { + heads.iter().max_by(|a, b| { + match a.hlc.cmp(&b.hlc) { + std::cmp::Ordering::Equal => a.author.cmp(&b.author), + ord => ord, + } + }) + } + + /// List all key-value pairs (winner values only) + pub fn list_all(&self) -> Result, Vec)>, StoreError> { let read_txn = self.db.begin_read()?; let table = read_txn.open_table(KV_TABLE)?; let mut result = Vec::new(); for entry in table.iter()? { let (key, value) = entry?; - result.push((key.value().to_string(), value.value().to_vec())); + let heads = HeadList::decode(value.value())?.heads; + if let Some(winner) = Self::pick_winner(&heads) { + result.push((key.value().to_vec(), winner.value.clone())); + } } Ok(result) } - /// Get the last applied sequence number - pub fn last_seq(&self) -> Result { + /// Get author state for a specific author + pub fn author_state(&self, author: &[u8; 32]) -> Result, StoreError> { let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(META_TABLE)?; + let table = read_txn.open_table(AUTHOR_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)?; + match table.get(&author[..])? { + Some(v) => Ok(AuthorState::decode(v.value()).ok()), + None => Ok(None), } - 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())) } } @@ -203,146 +255,701 @@ 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")) + fn temp_db_path(name: &str) -> std::path::PathBuf { + temp_dir().join(format!("lattice_dag_store_test_{}.db", name)) } - #[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(); - } + const TEST_STORE: [u8; 16] = [1u8; 16]; #[test] - fn test_apply_entry() { - let (db_path, _) = temp_paths("apply_entry"); - std::fs::remove_file(&db_path).ok(); + fn test_single_write_one_head() { + let path = temp_db_path("single_write"); + let _ = std::fs::remove_file(&path); - let store = Store::open(&db_path).unwrap(); + let store = Store::open(&path).unwrap(); let node = Node::generate(); let clock = MockClock::new(1000); - // Create and apply a put entry - let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock)) - .prev_hash([0u8; 32].to_vec()) - .put("/key1", b"value1".to_vec()) - .sign(&node); - store.apply_entry(&entry1).unwrap(); - - assert_eq!(store.get("/key1").unwrap(), Some(b"value1".to_vec())); - assert_eq!(store.last_seq().unwrap(), 1); - - // Create and apply a delete entry - let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock)) - .prev_hash([0u8; 32].to_vec()) // simplified for test - .delete("/key1") - .sign(&node); - store.apply_entry(&entry2).unwrap(); - - assert_eq!(store.get("/key1").unwrap(), None); - assert_eq!(store.last_seq().unwrap(), 2); - - 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)) + let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) .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(); + store.apply_entry(&entry).unwrap(); - // Replay - let store = Store::open(&db_path).unwrap(); - store.replay_log(&log_path).unwrap(); + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 1); + assert_eq!(heads[0].value, b"value"); - 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(); + let _ = std::fs::remove_file(&path); } #[test] - fn test_meta_accessors() { - let (db_path, _) = temp_paths("meta"); - std::fs::remove_file(&db_path).ok(); + fn test_deterministic_winner() { + // Test pick_winner logic directly (no store needed) + let heads = HeadList { + heads: vec![ + HeadInfo { + value: b"older".to_vec(), + hlc: 100, + author: [1u8; 32].to_vec(), + hash: [1u8; 32].to_vec(), + tombstone: false, + }, + HeadInfo { + value: b"newer".to_vec(), + hlc: 200, + author: [2u8; 32].to_vec(), + hash: [2u8; 32].to_vec(), + tombstone: false, + }, + ], + }; - let store = Store::open(&db_path).unwrap(); + let winner = Store::pick_winner(&heads.heads).unwrap(); + assert_eq!(winner.value, b"newer"); // Higher HLC wins + } + + #[test] + fn test_concurrent_writes_multiple_heads() { + let path = temp_db_path("concurrent"); + let _ = std::fs::remove_file(&path); - 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); + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + let clock = MockClock::new(1000); - std::fs::remove_file(&db_path).ok(); + // First write + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .parent_hashes(vec![]) // No parent + .put("/key", b"v1".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + + // Second write with SAME parent (simulates concurrent/offline write) + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry1).to_vec()) + .parent_hashes(vec![]) // Also no parent (doesn't know about entry1) + .put("/key", b"v2".to_vec()) + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + // Should have TWO heads now + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 2); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_merge_write_single_head() { + let path = temp_db_path("merge"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + + // Create two heads + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/key", b"v1".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry1).to_vec()) + .put("/key", b"v2".to_vec()) + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + assert_eq!(store.get_heads(b"/key").unwrap().len(), 2); + + // Merge write citing BOTH heads as parents + let hash1 = hash_signed_entry(&entry1); + let hash2 = hash_signed_entry(&entry2); + let clock3 = MockClock::new(3000); + let entry3 = EntryBuilder::new(3, HLC::now_with_clock(&clock3)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash2.to_vec()) + .parent_hashes(vec![hash1.to_vec(), hash2.to_vec()]) + .put("/key", b"merged".to_vec()) + .sign(&node); + store.apply_entry(&entry3).unwrap(); + + // Should now have ONE head + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 1); + assert_eq!(heads[0].value, b"merged"); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_delete_preserves_concurrent_heads() { + let path = temp_db_path("delete_concurrent"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + + // Create two concurrent heads + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/key", b"v1".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry1).to_vec()) + // No parent_hashes = concurrent write + .put("/key", b"v2".to_vec()) + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + assert_eq!(store.get_heads(b"/key").unwrap().len(), 2); + + // Delete citing only entry1 as parent + let hash1 = hash_signed_entry(&entry1); + let clock3 = MockClock::new(3000); + let entry3 = EntryBuilder::new(3, HLC::now_with_clock(&clock3)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash1.to_vec()) + .parent_hashes(vec![hash1.to_vec()]) // Only cites entry1 + .delete("/key") + .sign(&node); + store.apply_entry(&entry3).unwrap(); + + // entry2 should survive (wasn't cited as parent), plus tombstone head + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 2, "Expected tombstone + v2, got {}", heads.len()); + + // One should be a tombstone, one should be v2 + let has_tombstone = heads.iter().any(|h| h.tombstone); + let has_v2 = heads.iter().any(|h| h.value == b"v2"); + assert!(has_tombstone); + assert!(has_v2); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_delete_all_heads_removes_key() { + let path = temp_db_path("delete_all"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + + // Create a single head + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/key", b"value".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + + assert!(store.get(b"/key").unwrap().is_some()); + + // Delete citing the only head + let hash1 = hash_signed_entry(&entry1); + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash1.to_vec()) + .parent_hashes(vec![hash1.to_vec()]) + .delete("/key") + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + // Key should show as deleted (tombstone wins) + assert!(store.get(b"/key").unwrap().is_none()); + + // Should have one tombstone head + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 1); + assert!(heads[0].tombstone); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_concurrent_delete_and_put() { + // This test demonstrates that concurrent delete and put should both exist as heads + // Scenario: + // 1. Initial: K = v1 (head H1) + // 2. Alice (offline): Delete K citing H1 + // 3. Bob (offline): Put K = v2 citing H1 (doesn't know about delete) + // 4. Result: Should have 2 heads (tombstone + v2), not just v2 + + let path = temp_db_path("concurrent_delete_put"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let alice = Node::generate(); + let bob = Node::generate(); + + // Initial state: K = v1 + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put(b"/key", b"v1".to_vec()) + .sign(&alice); + store.apply_entry(&entry1).unwrap(); + let h1 = hash_signed_entry(&entry1); + + // Alice deletes K citing H1 + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(h1.to_vec()) + .parent_hashes(vec![h1.to_vec()]) + .delete(b"/key") + .sign(&alice); + store.apply_entry(&entry2).unwrap(); + + // Bob (concurrently) puts K = v2 citing H1 (doesn't know about Alice's delete) + let clock3 = MockClock::new(2500); + let entry3 = EntryBuilder::new(1, HLC::now_with_clock(&clock3)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) // Bob's own chain + .parent_hashes(vec![h1.to_vec()]) // Cites H1 as parent + .put(b"/key", b"v2".to_vec()) + .sign(&bob); + store.apply_entry(&entry3).unwrap(); + + // Should have 2 heads: Alice's tombstone and Bob's v2 + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 2, "Expected 2 heads (tombstone + put), got {}", heads.len()); + + // One should be a tombstone, one should be v2 + let has_tombstone = heads.iter().any(|h| h.tombstone); + let has_v2 = heads.iter().any(|h| h.value == b"v2"); + assert!(has_tombstone, "Expected a tombstone head"); + assert!(has_v2, "Expected a v2 head"); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_two_authors_diverged_then_merge() { + // Scenario: + // 1. Alice creates K = v1 (head H1) + // 2. Bob (offline, doesn't see H1) creates K = v2 (head H2) + // 3. Result: 2 heads (conflict) + // 4. Charlie (sees both) creates K = v3 citing H1 and H2 + // 5. Result: 1 head (merged) + + let path = temp_db_path("two_authors_merge"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let alice = Node::generate(); + let bob = Node::generate(); + let charlie = Node::generate(); + + // Alice creates K = v1 + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put(b"/key", b"alice_v1".to_vec()) + .sign(&alice); + store.apply_entry(&entry1).unwrap(); + let h1 = hash_signed_entry(&entry1); + + // Bob (offline, no parent_hashes) creates K = v2 + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(1, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + // No parent_hashes = concurrent/diverged + .put(b"/key", b"bob_v2".to_vec()) + .sign(&bob); + store.apply_entry(&entry2).unwrap(); + let h2 = hash_signed_entry(&entry2); + + // Should have 2 heads now + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 2, "Expected 2 diverged heads"); + + // Verify deterministic winner (higher HLC wins) + let value = store.get(b"/key").unwrap().unwrap(); + assert_eq!(value, b"bob_v2"); // Bob has higher HLC (2000 > 1000) + + // Charlie merges by citing both H1 and H2 + let clock3 = MockClock::new(3000); + let entry3 = EntryBuilder::new(1, HLC::now_with_clock(&clock3)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .parent_hashes(vec![h1.to_vec(), h2.to_vec()]) + .put(b"/key", b"charlie_merged".to_vec()) + .sign(&charlie); + store.apply_entry(&entry3).unwrap(); + + // Should have 1 head now (merged) + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 1, "Expected 1 merged head"); + assert_eq!(heads[0].value, b"charlie_merged"); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_apply_entry_is_idempotent() { + // Applying the same entry twice should not duplicate the head + // This is critical for log replay and network message deduplication + + let path = temp_db_path("idempotent"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put(b"/key", b"value".to_vec()) + .sign(&node); + + // Apply once + store.apply_entry(&entry1).unwrap(); + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1); + + // Apply again (e.g., log replay or duplicate message) + store.apply_entry(&entry1).unwrap(); + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1, "Duplicate entry should not create duplicate head"); + + // Apply a third time for good measure + store.apply_entry(&entry1).unwrap(); + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_sequential_writes_then_replay() { + // Simulates: put a=1, put a=2, then replay from log + // After replay, should have only 1 head (the latest) + + let path = temp_db_path("sequential_replay"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = Node::generate(); + + // First write: a = 1 + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .parent_hashes(vec![]) // No parents for first write + .put(b"/key", b"1".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + let h1 = hash_signed_entry(&entry1); + + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1); + + // Second write: a = 2, citing h1 as parent + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(h1.to_vec()) + .parent_hashes(vec![h1.to_vec()]) // Cites h1 + .put(b"/key", b"2".to_vec()) + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1, "After put 2, should have 1 head"); + + // Now simulate log replay: clear state and re-apply both entries + drop(store); + let _ = std::fs::remove_file(&path); + let store = Store::open(&path).unwrap(); + + // Check what parent_hashes entry2 actually has + let decoded_entry2 = Entry::decode(&entry2.entry_bytes[..]).unwrap(); + eprintln!("Entry2 parent_hashes: {:?}", decoded_entry2.parent_hashes); + eprintln!("H1: {:?}", h1); + + // Replay entry1 + store.apply_entry(&entry1).unwrap(); + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1, "After replay entry1"); + + // Replay entry2 + store.apply_entry(&entry2).unwrap(); + let heads = store.get_heads(b"/key").unwrap(); + assert_eq!(heads.len(), 1, "After replay entry2, should have 1 head, got {}: {:?}", + heads.len(), heads.iter().map(|h| String::from_utf8_lossy(&h.value)).collect::>()); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn test_replay_to_existing_state_no_duplicates() { + use crate::sigchain::SigChain; + + // This simulates: put a=1, put a=2, then restart and replay from log + // The replay should skip already-applied entries + + let state_path = temp_db_path("replay_existing_state"); + let log_path = temp_db_path("replay_existing_log"); + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + + let store = Store::open(&state_path).unwrap(); + let node = Node::generate(); + let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); + + // First write: a = 1 + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .parent_hashes(vec![]) + .put(b"/key", b"1".to_vec()) + .sign(&node); + sigchain.append(&entry1).unwrap(); + store.apply_entry(&entry1).unwrap(); + let h1 = hash_signed_entry(&entry1); + + // Second write: a = 2, citing h1 as parent + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(h1.to_vec()) + .parent_hashes(vec![h1.to_vec()]) + .put(b"/key", b"2".to_vec()) + .sign(&node); + sigchain.append(&entry2).unwrap(); + store.apply_entry(&entry2).unwrap(); + + assert_eq!(store.get_heads(b"/key").unwrap().len(), 1, "Before restart"); + let author = node.public_key_bytes(); + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 2, "author seq should be 2"); + + // Simulate restart: reopen state.db (persisted) and replay log + drop(store); + drop(sigchain); + + let store = Store::open(&state_path).unwrap(); // Reopen existing state + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 2, "author seq persisted"); + + // Replay log - apply_head skips entries whose parents don't exist + let replayed = store.replay_log(&log_path).unwrap(); + assert_eq!(replayed, 2, "Replayed 2 entries from log"); + + let final_heads = store.get_heads(b"/key").unwrap(); + assert_eq!(final_heads.len(), 1, + "After replay, should have 1 head, got {}: {:?}", + final_heads.len(), + final_heads.iter().map(|h| String::from_utf8_lossy(&h.value)).collect::>()); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + } + + #[test] + fn test_fast_resume_on_restart() { + use crate::sigchain::SigChain; + + // Fast resume: entries already applied are skipped based on per-author seq + let state_path = temp_db_path("fast_resume_state"); + let log_path = temp_db_path("fast_resume_log"); + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + + let store = Store::open(&state_path).unwrap(); + let node = Node::generate(); + let author = node.public_key_bytes(); + let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); + + // Apply 3 entries with proper chaining + for i in 1u64..=3 { + let clock = MockClock::new(i * 1000); + let prev = sigchain.last_hash().to_vec(); + let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(prev) + .put(format!("/key{}", i).as_bytes(), format!("v{}", i).into_bytes()) + .sign(&node); + sigchain.append(&entry).unwrap(); + store.apply_entry(&entry).unwrap(); + } + + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3); + assert_eq!(store.get_heads(b"/key3").unwrap().len(), 1); + + // Restart and replay - should skip all entries + drop(store); + drop(sigchain); + + let store = Store::open(&state_path).unwrap(); + let replayed = store.replay_log(&log_path).unwrap(); + + // All 3 entries were replayed but skipped (seq check) + assert_eq!(replayed, 3, "Replayed 3 entries from log"); + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3, "seq unchanged"); + assert_eq!(store.get_heads(b"/key3").unwrap().len(), 1, "heads unchanged"); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + } + + #[test] + fn test_partial_replay_after_crash() { + use crate::sigchain::SigChain; + + // Simulates: log has 5 entries, state.db only has first 3 applied (crash) + // Replay should apply entries 4 and 5 + let state_path = temp_db_path("partial_replay_state"); + let log_path = temp_db_path("partial_replay_log"); + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + + let store = Store::open(&state_path).unwrap(); + let node = Node::generate(); + let author = node.public_key_bytes(); + let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); + + // Write 5 entries to log with proper chaining + for i in 1u64..=5 { + let clock = MockClock::new(i * 1000); + let prev = sigchain.last_hash().to_vec(); + let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(prev) + .put(format!("/key{}", i).as_bytes(), format!("v{}", i).into_bytes()) + .sign(&node); + sigchain.append(&entry).unwrap(); + + // Only apply first 3 to state.db (simulating crash after 3rd) + if i <= 3 { + store.apply_entry(&entry).unwrap(); + } + } + + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3); + assert!(store.get_heads(b"/key4").unwrap().is_empty(), "key4 not applied yet"); + + // Simulate restart and replay + drop(store); + drop(sigchain); + + let store = Store::open(&state_path).unwrap(); + let replayed = store.replay_log(&log_path).unwrap(); + + assert_eq!(replayed, 5, "Replayed 5 entries from log"); + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 5, "seq updated to 5"); + assert_eq!(store.get_heads(b"/key4").unwrap().len(), 1, "key4 now applied"); + assert_eq!(store.get_heads(b"/key5").unwrap().len(), 1, "key5 now applied"); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&log_path); + } + + #[test] + fn test_state_db_rollback_and_replay() { + use crate::sigchain::SigChain; + + // Simulates: + // 1. Apply entries 1-3 + // 2. Copy state.db (backup) + // 3. Apply entries 4-5 + // 4. Restore state.db from backup + // 5. Restart and replay - should apply entries 4-5 + + let state_path = temp_db_path("rollback_state"); + let backup_path = temp_db_path("rollback_backup"); + let log_path = temp_db_path("rollback_log"); + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&backup_path); + let _ = std::fs::remove_file(&log_path); + + let store = Store::open(&state_path).unwrap(); + let node = Node::generate(); + let author = node.public_key_bytes(); + let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); + + // Apply first 3 entries + for i in 1u64..=3 { + let clock = MockClock::new(i * 1000); + let prev = sigchain.last_hash().to_vec(); + let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(prev) + .put(format!("/key{}", i).as_bytes(), format!("v{}", i).into_bytes()) + .sign(&node); + sigchain.append(&entry).unwrap(); + store.apply_entry(&entry).unwrap(); + } + + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3); + + // Close and backup state.db + drop(store); + std::fs::copy(&state_path, &backup_path).unwrap(); + + // Reopen and apply entries 4-5 + let store = Store::open(&state_path).unwrap(); + for i in 4u64..=5 { + let clock = MockClock::new(i * 1000); + let prev = sigchain.last_hash().to_vec(); + let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(prev) + .put(format!("/key{}", i).as_bytes(), format!("v{}", i).into_bytes()) + .sign(&node); + sigchain.append(&entry).unwrap(); + store.apply_entry(&entry).unwrap(); + } + + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 5); + assert_eq!(store.get_heads(b"/key5").unwrap().len(), 1); + + // Now restore state.db from backup (simulating crash/rollback) + drop(store); + drop(sigchain); + std::fs::copy(&backup_path, &state_path).unwrap(); + + // Restart and replay + let store = Store::open(&state_path).unwrap(); + + // State should be at seq 3 (restored from backup) + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3, "Restored to seq 3"); + assert!(store.get_heads(b"/key4").unwrap().is_empty(), "key4 not in restored state"); + + // Replay log - should apply entries 4 and 5 + let replayed = store.replay_log(&log_path).unwrap(); + assert_eq!(replayed, 5, "Replayed 5 entries from log"); + + // Now seq should be 5 and keys 4-5 should exist + assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 5, "seq updated to 5"); + assert_eq!(store.get_heads(b"/key4").unwrap().len(), 1, "key4 now applied"); + assert_eq!(store.get_heads(b"/key5").unwrap().len(), 1, "key5 now applied"); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(&backup_path); + let _ = std::fs::remove_file(&log_path); } } diff --git a/proto/lattice.proto b/proto/lattice.proto index 10c5431..4219427 100644 --- a/proto/lattice.proto +++ b/proto/lattice.proto @@ -24,14 +24,38 @@ message Entry { bytes store_id = 6; // Ordering Metadata - bytes prev_hash = 2; // Link to previous entry (32 bytes) + bytes prev_hash = 2; // Link to previous sigchain entry (32 bytes) uint64 seq = 3; // Monotonic sequence number HLC timestamp = 4; // Hybrid Logical Clock + // DAG ancestry: hashes of entries this supersedes (separate from sigchain) + repeated bytes parent_hashes = 7; + // The Batch of Operations repeated Operation ops = 5; } +// HeadInfo: a tip/head in the DAG for a key +message HeadInfo { + bytes value = 1; // The value at this head + uint64 hlc = 2; // Combined HLC for ordering (wall_time_ms << 16 | counter) + bytes author = 3; // Author's public key (32 bytes) + bytes hash = 4; // Hash of the SignedEntry that created this head + bool tombstone = 5; // True if this head represents a delete +} + +// HeadList: wrapper for storing multiple heads per key in state.db +message HeadList { + repeated HeadInfo heads = 1; +} + +// AuthorState: tracks last applied entry per author for replay optimization +message AuthorState { + uint64 seq = 1; // Last applied seq for this author's sigchain + bytes hash = 2; // Hash of last applied entry + uint64 log_offset = 3; // Byte offset in log file for fast resume +} + // Hybrid Logical Clock message HLC { uint64 wall_time = 1; // Unix timestamp (ms) @@ -49,12 +73,12 @@ message Operation { } message PutOp { - string key = 1; - bytes value = 2; // Raw bytes allows storing images, JSON, binary, etc. + bytes key = 1; + bytes value = 2; } message DeleteOp { - string key = 1; + bytes key = 1; } // 4. The Sync Handshake (Vector Clocks)