diff --git a/Cargo.toml b/Cargo.toml index 19dd5bb..645ced9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "lattice-core", "lattice-net", "lattice-store", + "lattice-cli", ] [workspace.package] @@ -16,6 +17,10 @@ license = "MIT" lattice-core = { path = "lattice-core" } lattice-net = { path = "lattice-net" } lattice-store = { path = "lattice-store" } +lattice-cli = { path = "lattice-cli" } + +# CLI +rustyline = "17" # Networking (Iroh) iroh = "0.95" diff --git a/docs/roadmap.md b/docs/roadmap.md index bbb468a..63f53fc 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -12,7 +12,7 @@ - [x] Log file I/O (append, read, hash verification) - [x] SigChain (validate entries before appending) - [x] Store (redb) — `kv` + `meta` tables, log replay -- [ ] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit` +- [x] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit` ### Success Criteria @@ -30,20 +30,39 @@ Current code assumes single store. Changes needed: - [ ] 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 ` --- +## Milestone 1.5: DAG Conflict Resolution ← NEXT + +**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 + +### 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 + +--- + ## Milestone 2: Two-Node Sync **Goal:** Two nodes can sync their logs over the network. ### Deliverables -- [ ] Store: add `applied_frontiers` table (sync state per author) - [ ] VectorClock module (diff, merge, missing entries) - [ ] Sync protocol (push missing entries) - [ ] Iroh integration (peer discovery, connection) diff --git a/lattice-cli/Cargo.toml b/lattice-cli/Cargo.toml new file mode 100644 index 0000000..8020d9d --- /dev/null +++ b/lattice-cli/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "lattice-cli" +description = "Interactive CLI for Lattice" +version.workspace = true +edition.workspace = true +license.workspace = true + +[[bin]] +name = "lattice" +path = "src/main.rs" + +[dependencies] +lattice-core = { workspace = true } +rustyline = { workspace = true } +hex = { workspace = true } +thiserror = { workspace = true } +shlex = "1" diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs new file mode 100644 index 0000000..7141633 --- /dev/null +++ b/lattice-cli/src/commands.rs @@ -0,0 +1,152 @@ +//! CLI command handlers (presentation layer) + +use crate::node::LatticeNode; +use std::time::Instant; + +/// Command handler function type +pub type Handler = fn(&mut LatticeNode, &[String]); + +/// Command definition +pub struct Command { + pub name: &'static str, + pub args: &'static str, + pub description: &'static str, + pub min_args: usize, + pub max_args: usize, + pub handler: Handler, +} + +/// Build the command registry +pub fn commands() -> Vec { + vec![ + Command { + name: "put", + args: " ", + description: "Store a key-value pair", + min_args: 2, + max_args: 2, + handler: cmd_put, + }, + Command { + name: "get", + args: "", + description: "Retrieve a value by key", + min_args: 1, + max_args: 1, + handler: cmd_get, + }, + Command { + name: "delete", + args: "", + description: "Delete a key", + min_args: 1, + max_args: 1, + handler: cmd_delete, + }, + Command { + name: "list", + args: "[-v]", + description: "List all key-value pairs (-v for verbose)", + min_args: 0, + max_args: 1, + handler: cmd_list, + }, + Command { + name: "status", + args: "", + description: "Show node statistics", + min_args: 0, + max_args: 0, + handler: cmd_status, + }, + Command { + name: "help", + args: "", + description: "Show this help message", + min_args: 0, + max_args: 0, + handler: cmd_help, + }, + ] +} + +/// Print help from the command registry +fn cmd_help(_node: &mut LatticeNode, _args: &[String]) { + println!("\nLattice Commands:"); + for cmd in commands() { + if cmd.args.is_empty() { + println!(" {:<18} {}", cmd.name, cmd.description); + } else { + println!(" {} {:<10} {}", cmd.name, cmd.args, cmd.description); + } + } + println!(" quit Exit the CLI"); + println!("\nTip: Use quotes for values with spaces: put \"my key\" \"hello world\"\n"); +} + +fn cmd_status(node: &mut LatticeNode, _args: &[String]) { + let status = node.status(); + println!("--- Node Status ---"); + println!("Node ID: {}", status.node_id); + println!("Data Dir: {}", status.data_dir); + println!("Log Sequence: {}", status.log_seq); + println!("Applied Entries: {}", status.applied_seq); + println!("-------------------"); +} + +fn cmd_put(node: &mut LatticeNode, args: &[String]) { + let start = Instant::now(); + match node.put(&args[0], args[1].as_bytes()) { + Ok(seq) => println!("OK (seq: {}, time: {:.2?})", seq, start.elapsed()), + Err(e) => eprintln!("Error: {}", e), + } +} + +fn cmd_get(node: &mut LatticeNode, args: &[String]) { + let start = Instant::now(); + match node.get(&args[0]) { + Ok(Some(value)) => { + println!("{}", format_value(&value)); + println!("({:.2?})", start.elapsed()); + } + Ok(None) => println!("(nil)"), + Err(e) => eprintln!("Error: {}", e), + } +} + +fn cmd_delete(node: &mut LatticeNode, args: &[String]) { + let start = Instant::now(); + match node.delete(&args[0]) { + Ok(seq) => println!("OK (seq: {}, time: {:.2?})", seq, start.elapsed()), + Err(e) => eprintln!("Error: {}", e), + } +} + +fn cmd_list(node: &mut LatticeNode, args: &[String]) { + let verbose = args.first().map(|a| a == "-v").unwrap_or(false); + let start = Instant::now(); + match node.list() { + Ok(entries) => { + if entries.is_empty() { + println!("(empty)"); + return; + } + for (key, value) in &entries { + if verbose { + println!("{} = {} ({} bytes)", key, format_value(value), value.len()); + } else { + println!("{} = {}", key, format_value(value)); + } + } + println!("({} keys, {:.2?})", entries.len(), start.elapsed()); + } + Err(e) => eprintln!("Error: {}", e), + } +} + +fn format_value(value: &[u8]) -> String { + match std::str::from_utf8(value) { + Ok(s) => s.to_string(), + Err(_) => format!("0x{}", hex::encode(value)), + } +} diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs new file mode 100644 index 0000000..607178a --- /dev/null +++ b/lattice-cli/src/main.rs @@ -0,0 +1,91 @@ +//! Lattice Interactive CLI + +mod node; +mod commands; + +use node::LatticeNodeBuilder; +use rustyline::error::ReadlineError; +use rustyline::DefaultEditor; + +fn main() { + println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION")); + println!("Type 'help' for commands, 'quit' to exit.\n"); + + let mut node = match LatticeNodeBuilder::new().build() { + Ok((n, info)) => { + println!("Node ID: {}", info.node_id); + println!("Data: {}", info.data_path); + if info.is_new { + println!("Status: New identity created"); + } else if info.entries_replayed > 0 { + println!("Replay: {} log entries applied", info.entries_replayed); + } + println!(); + n + } + Err(e) => { + eprintln!("Failed to initialize node: {}", e); + eprintln!("Hint: If data is corrupted, remove the data directory and restart."); + return; + } + }; + + let mut rl = DefaultEditor::new().expect("Failed to create editor"); + let cmds = commands::commands(); + + loop { + match rl.readline("lattice> ") { + Ok(line) => { + let line = line.trim(); + if line.is_empty() { + continue; + } + let _ = rl.add_history_entry(line); + + let args = match shlex::split(line) { + Some(a) => a, + None => { + println!("Error: mismatched quotes"); + continue; + } + }; + + let cmd_name = match args.first() { + Some(c) => c.as_str(), + None => continue, + }; + + // Handle quit specially + if cmd_name == "quit" || cmd_name == "exit" { + println!("Goodbye!"); + break; + } + + // Look up command in registry + match cmds.iter().find(|c| c.name == cmd_name) { + Some(cmd) => { + let cmd_args = &args[1..]; + if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args { + if cmd.min_args == cmd.max_args { + println!("Usage: {} {}", cmd.name, cmd.args); + } else { + println!("Usage: {} {} (got {} args)", cmd.name, cmd.args, cmd_args.len()); + } + } else { + (cmd.handler)(&mut node, cmd_args); + } + } + None => println!("Unknown command: '{}'. Type 'help' for commands.", cmd_name), + } + } + Err(ReadlineError::Interrupted | ReadlineError::Eof) => { + println!("Goodbye!"); + break; + } + Err(err) => { + eprintln!("Error: {:?}", err); + break; + } + } + } +} diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs new file mode 100644 index 0000000..6cb0a9d --- /dev/null +++ b/lattice-cli/src/node.rs @@ -0,0 +1,269 @@ +//! Lattice Node API +//! +//! A programmatic interface to a local Lattice node. + +use lattice_core::{ + DataDir, EntryBuilder, Node, SigChain, Store, + hlc::HLC, + log::LogError, + sigchain::SigChainError, + store::StoreError, +}; +use std::path::Path; +use thiserror::Error; + +/// Errors that can occur during node operations +#[derive(Error, Debug)] +pub enum NodeError { + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + + #[error("Store error: {0}")] + Store(#[from] StoreError), + + #[error("SigChain error: {0}")] + SigChain(#[from] SigChainError), + + #[error("Log error: {0}")] + Log(#[from] LogError), + + #[error("Node error: {0}")] + Node(#[from] lattice_core::node::NodeError), +} + +/// Info returned when building a node +pub struct NodeInfo { + pub node_id: String, + pub data_path: String, + pub is_new: bool, + pub entries_replayed: u64, +} + +/// Status information about the node +pub struct NodeStatus { + pub node_id: String, + pub data_dir: String, + pub log_seq: u64, + pub applied_seq: u64, +} + +/// Builder for creating a fully initialized LatticeNode +pub struct LatticeNodeBuilder { + data_dir: DataDir, +} + +impl LatticeNodeBuilder { + /// Create a builder with the default data directory + pub fn new() -> Self { + Self { + data_dir: DataDir::default(), + } + } + + /// Build and initialize the node + pub fn build(self) -> Result<(LatticeNode, NodeInfo), NodeError> { + // Create directories + self.data_dir.ensure_dirs()?; + + // Load or create node identity + let key_path = self.data_dir.identity_key(); + let is_new = !key_path.exists(); + let node = if key_path.exists() { + Node::load(&key_path)? + } else { + let node = Node::generate(); + node.save(&key_path)?; + node + }; + + let author_id_hex = hex::encode(node.public_key_bytes()); + + // Load or create sigchain + let log_path = self.data_dir.log_file(&author_id_hex); + let sigchain = if log_path.exists() { + SigChain::from_log(&log_path, node.public_key_bytes())? + } else { + SigChain::new(&log_path, node.public_key_bytes()) + }; + + // Open store and replay log + let store = Store::open(self.data_dir.state_db())?; + let entries_replayed = if log_path.exists() { + store.replay_log(&log_path)? + } else { + 0 + }; + + let info = NodeInfo { + node_id: hex::encode(node.public_key_bytes()), + data_path: self.data_dir.base().display().to_string(), + is_new, + entries_replayed, + }; + + Ok((LatticeNode { + data_dir: self.data_dir, + node, + sigchain, + store, + }, info)) + } +} + +impl Default for LatticeNodeBuilder { + fn default() -> Self { + Self::new() + } +} + +/// A fully initialized Lattice node +/// +/// Use `LatticeNodeBuilder` to create an instance. +pub struct LatticeNode { + data_dir: DataDir, + node: Node, + sigchain: SigChain, + store: Store, +} + +impl LatticeNode { + /// Get the node's public key as hex + pub fn node_id(&self) -> String { + hex::encode(self.node.public_key_bytes()) + } + + /// Get the path to the data directory + pub fn data_path(&self) -> &Path { + self.data_dir.base() + } + + /// Get the current status of the node + pub fn status(&self) -> NodeStatus { + NodeStatus { + node_id: self.node_id(), + data_dir: self.data_dir.base().display().to_string(), + log_seq: self.sigchain.len(), + applied_seq: self.store.last_seq().unwrap_or(0), + } + } + + /// Put a key-value pair + pub fn put(&mut self, key: &str, value: &[u8]) -> Result { + let entry = EntryBuilder::new(self.sigchain.next_seq(), HLC::now()) + .prev_hash(self.sigchain.last_hash().to_vec()) + .put(key, value.to_vec()) + .sign(&self.node); + + self.commit_entry(entry) + } + + /// Get a value by key + pub fn get(&self, key: &str) -> Result>, NodeError> { + Ok(self.store.get(key)?) + } + + /// List all key-value pairs + pub fn list(&self) -> Result)>, NodeError> { + Ok(self.store.list_all()?) + } + + /// Delete a key + pub fn delete(&mut self, key: &str) -> Result { + let entry = EntryBuilder::new(self.sigchain.next_seq(), HLC::now()) + .prev_hash(self.sigchain.last_hash().to_vec()) + .delete(key) + .sign(&self.node); + + self.commit_entry(entry) + } + + /// Commit a signed entry: append to log via sigchain, then apply to store + fn commit_entry(&mut self, entry: lattice_core::proto::SignedEntry) -> Result { + self.sigchain.append(&entry)?; + self.store.apply_entry(&entry)?; + + Ok(self.sigchain.len()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::env::temp_dir; + + fn temp_data_dir(name: &str) -> DataDir { + let path = temp_dir().join(format!("lattice_node_test_{}", name)); + // Clean up from previous runs + let _ = std::fs::remove_dir_all(&path); + DataDir::new(path) + } + + #[test] + fn test_put_survives_restart() { + let data_dir = temp_data_dir("restart"); + + // First session: put a value + { + let (mut node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to create node"); + + node.put("/test/key", b"hello").expect("put failed"); + assert_eq!(node.get("/test/key").unwrap(), Some(b"hello".to_vec())); + } + + // Second session: value should still be there + { + let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to create node on restart"); + + assert_eq!(node.get("/test/key").unwrap(), Some(b"hello".to_vec())); + assert_eq!(node.status().log_seq, 1); + } + + // Cleanup + let _ = std::fs::remove_dir_all(data_dir.base()); + } + + #[test] + fn test_log_replay_after_db_deletion() { + let data_dir = temp_data_dir("replay"); + + // First session: put some values + { + let (mut node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to create node"); + + node.put("/key1", b"value1").expect("put failed"); + node.put("/key2", b"value2").expect("put failed"); + node.delete("/key1").expect("delete failed"); + } + + // Delete state.db but keep the log + let db_path = data_dir.state_db(); + std::fs::remove_file(&db_path).expect("Failed to delete state.db"); + assert!(!db_path.exists(), "state.db should be deleted"); + + // Third session: log should be replayed to reconstruct state + { + let (node, info) = LatticeNodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to rebuild node from log"); + + // Should have replayed 3 entries + assert_eq!(info.entries_replayed, 3); + // key1 was deleted + assert_eq!(node.get("/key1").unwrap(), None); + // key2 should still exist + assert_eq!(node.get("/key2").unwrap(), Some(b"value2".to_vec())); + // log seq should be 3 (put, put, delete) + assert_eq!(node.status().log_seq, 3); + } + + // Cleanup + let _ = std::fs::remove_dir_all(data_dir.base()); + } +} + diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index 1bee710..24ed473 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -66,50 +66,67 @@ impl Store { Ok(Self { db }) } - /// Replay a log file and apply all entries to the store + /// Replay a log file and apply all entries to the store (batched) 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; + if entries.is_empty() { + return Ok(0); } - 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())?; - } + for signed_entry in &entries { + Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut meta_table)?; + } + } + write_txn.commit()?; + + Ok(entries.len() as u64) + } + + /// Apply a single signed entry to the store + pub fn apply_entry(&self, signed_entry: &SignedEntry) -> Result<(), StoreError> { + 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)?; + } + write_txn.commit()?; + Ok(()) + } + + /// 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]>, + ) -> Result<(), StoreError> { + let entry = Entry::decode(&signed_entry.entry_bytes[..])?; + + // 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()?; + + // 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[..])?; Ok(()) } @@ -122,28 +139,17 @@ impl Store { 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)?; + /// List all key-value pairs + pub fn list_all(&self) -> Result)>, 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())); } - 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) + Ok(result) } /// Get the last applied sequence number @@ -222,33 +228,33 @@ mod tests { } #[test] - fn test_put_get() { - let (db_path, _) = temp_paths("put_get"); + fn test_apply_entry() { + let (db_path, _) = temp_paths("apply_entry"); std::fs::remove_file(&db_path).ok(); let store = Store::open(&db_path).unwrap(); + let node = Node::generate(); + let clock = MockClock::new(1000); - store.put("key1", b"value1").unwrap(); - store.put("key2", b"value2").unwrap(); + // 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.get("key2").unwrap(), Some(b"value2".to_vec())); - assert_eq!(store.get("key3").unwrap(), None); + assert_eq!(store.get("/key1").unwrap(), Some(b"value1".to_vec())); + assert_eq!(store.last_seq().unwrap(), 1); - std::fs::remove_file(&db_path).ok(); - } - - #[test] - fn test_delete() { - let (db_path, _) = temp_paths("delete"); - std::fs::remove_file(&db_path).ok(); + // 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(); - 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 + assert_eq!(store.get("/key1").unwrap(), None); + assert_eq!(store.last_seq().unwrap(), 2); std::fs::remove_file(&db_path).ok(); }