diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index 599fce5..3cb3550 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -97,6 +97,14 @@ pub fn commands() -> Vec { max_args: 0, handler: cmd_status, }, + Command { + name: "author-state", + args: "[author-hex]", + description: "Show author state (default: self)", + min_args: 0, + max_args: 1, + handler: cmd_author_state, + }, Command { name: "help", args: "", @@ -215,7 +223,7 @@ fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) } fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { - println!("Node ID: {}", node.node_id()); + println!("Node ID: {}", hex::encode(node.node_id())); println!("Data: {}", node.data_path().display()); match node.root_store() { Ok(Some(id)) => println!("Root: {}", id), @@ -355,3 +363,45 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) - fn format_value(v: &[u8]) -> String { std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v))) } + +fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult { + let store = match store { + Some(s) => s, + None => { + eprintln!("Error: no store selected"); + return CommandResult::Ok; + } + }; + + // Get author: from arg or default to self + let author_bytes: [u8; 32] = if args.is_empty() { + node.node_id() + } else { + let hex_str = args[0].trim_start_matches("0x"); + match hex::decode(hex_str) { + Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), + Ok(bytes) => { + eprintln!("Error: author must be 32 bytes, got {}", bytes.len()); + return CommandResult::Ok; + } + Err(e) => { + eprintln!("Error: invalid hex: {}", e); + return CommandResult::Ok; + } + } + }; + + match store.author_state(&author_bytes) { + Ok(Some(state)) => { + println!("Author: {}", hex::encode(&author_bytes)); + println!(" seq: {}", state.seq); + println!(" hash: {}", hex::encode(&state.hash)); + println!(" log_offset: {}", state.log_offset); + } + Ok(None) => { + println!("No state for author: {}", hex::encode(&author_bytes)); + } + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index 15123da..48589df 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -13,25 +13,21 @@ fn main() { println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION")); println!("Type 'help' for commands, 'quit' to exit.\n"); - let (node, info) = match LatticeNodeBuilder::new().build() { - Ok(result) => result, + let node = match LatticeNodeBuilder::new().build() { + Ok(n) => n, Err(e) => { eprintln!("Failed to initialize: {}", e); return; } }; + let info = node.info(); println!("Node ID: {}", info.node_id); println!("Data: {}", info.data_path); - if info.is_new { - println!("Status: New identity created"); - } else if !info.stores.is_empty() { + if !info.stores.is_empty() { println!("Stores: {}", info.stores.len()); } - if let Some(root) = info.root_store { - println!("Root: {}", root); - } let mut current_store: Option = match node.open_root_store() { Ok(Some((h, open_info))) => { diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs index 9ea34c2..fcec953 100644 --- a/lattice-cli/src/node.rs +++ b/lattice-cli/src/node.rs @@ -44,8 +44,6 @@ pub enum NodeError { pub struct NodeInfo { pub node_id: String, pub data_path: String, - pub is_new: bool, - pub root_store: Option, pub stores: Vec, } @@ -63,11 +61,10 @@ impl LatticeNodeBuilder { Self { data_dir: DataDir::default() } } - pub fn build(self) -> Result<(LatticeNode, NodeInfo), NodeError> { + pub fn build(self) -> Result { self.data_dir.ensure_dirs()?; 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 { @@ -77,22 +74,12 @@ impl LatticeNodeBuilder { }; let meta = MetaStore::open(self.data_dir.meta_db())?; - let root_store = meta.root_store()?; - let stores = meta.list_stores()?; - let info = NodeInfo { - node_id: hex::encode(node.public_key_bytes()), - data_path: self.data_dir.base().display().to_string(), - is_new, - root_store, - stores, - }; - - Ok((LatticeNode { + Ok(LatticeNode { data_dir: self.data_dir, node: Rc::new(node), meta, - }, info)) + }) } } @@ -108,8 +95,16 @@ pub struct LatticeNode { } impl LatticeNode { - pub fn node_id(&self) -> String { - hex::encode(self.node.public_key_bytes()) + pub fn info(&self) -> NodeInfo { + NodeInfo { + node_id: hex::encode(self.node.public_key_bytes()), + data_path: self.data_dir.base().display().to_string(), + stores: self.meta.list_stores().unwrap_or_default(), + } + } + + pub fn node_id(&self) -> [u8; 32] { + self.node.public_key_bytes() } pub fn data_path(&self) -> &Path { @@ -183,7 +178,7 @@ impl LatticeNode { let handle = StoreHandle { store_id, tx, - actor_handle, + actor_handle: Some(actor_handle), }; Ok((handle, info)) @@ -194,8 +189,7 @@ impl LatticeNode { pub struct StoreHandle { store_id: Uuid, tx: std::sync::mpsc::Sender, - #[allow(dead_code)] - actor_handle: std::thread::JoinHandle<()>, + actor_handle: Option>, } impl StoreHandle { @@ -248,6 +242,16 @@ impl StoreHandle { .map_err(NodeError::Store) } + pub fn author_state(&self, author: &[u8; 32]) -> Result, NodeError> { + use crate::store_actor::StoreCmd; + let (resp_tx, resp_rx) = std::sync::mpsc::channel(); + self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }) + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.recv() + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + pub fn put(&self, key: &[u8], value: &[u8]) -> Result { use crate::store_actor::StoreCmd; let (resp_tx, resp_rx) = std::sync::mpsc::channel(); @@ -269,6 +273,16 @@ impl StoreHandle { } } +impl Drop for StoreHandle { + fn drop(&mut self) { + // Send shutdown command and wait for actor to finish + let _ = self.tx.send(crate::store_actor::StoreCmd::Shutdown); + if let Some(handle) = self.actor_handle.take() { + let _ = handle.join(); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -284,11 +298,11 @@ mod tests { fn test_create_and_open_store() { let data_dir = temp_data_dir("meta_store"); - let (node, info) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("Failed to create node"); - assert!(info.stores.is_empty()); + assert!(node.info().stores.is_empty()); let store_id = node.create_store().expect("Failed to create store"); @@ -307,7 +321,7 @@ mod tests { fn test_store_isolation() { let data_dir = temp_data_dir("meta_isolation"); - let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("Failed to create node"); @@ -329,12 +343,12 @@ mod tests { fn test_init_creates_root_store() { let data_dir = temp_data_dir("init_root"); - let (node, info) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); // Initially no root store - assert!(info.root_store.is_none()); + assert!(node.root_store().unwrap().is_none()); // Init creates root store let root_id = node.init().expect("init failed"); @@ -347,7 +361,7 @@ mod tests { fn test_duplicate_init_fails() { let data_dir = temp_data_dir("init_dup"); - let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); @@ -368,18 +382,18 @@ mod tests { // First session: init let root_id = { - let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); node.init().expect("init") }; - // Second session: root_store should be in info - let (_, info) = LatticeNodeBuilder { data_dir: data_dir.clone() } + // Second session: root_store should persist + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("reload node"); - assert_eq!(info.root_store, Some(root_id)); + assert_eq!(node.root_store().unwrap(), Some(root_id)); let _ = std::fs::remove_dir_all(data_dir.base()); } @@ -388,7 +402,7 @@ mod tests { fn test_idempotent_put_and_delete() { let data_dir = temp_data_dir("idempotent"); - let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() } + let node = LatticeNodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); let store_id = node.init().expect("init"); diff --git a/lattice-cli/src/store_actor.rs b/lattice-cli/src/store_actor.rs index 6533033..12a3191 100644 --- a/lattice-cli/src/store_actor.rs +++ b/lattice-cli/src/store_actor.rs @@ -49,7 +49,6 @@ pub enum StoreCmd { pub enum StoreActorError { Store(StoreError), SigChain(SigChainError), - ChannelClosed, } impl From for StoreActorError { @@ -69,7 +68,6 @@ impl std::fmt::Display for StoreActorError { match self { StoreActorError::Store(e) => write!(f, "Store error: {}", e), StoreActorError::SigChain(e) => write!(f, "SigChain error: {}", e), - StoreActorError::ChannelClosed => write!(f, "Channel closed"), } } } diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index fea574a..df9c6b9 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -65,6 +65,7 @@ impl Store { } /// Replay a log file and apply all entries to the store (batched) + /// Returns the number of newly applied entries (skipped entries not counted) pub fn replay_log(&self, log_path: impl AsRef) -> Result { let entries = read_entries(log_path)?; if entries.is_empty() { @@ -72,17 +73,20 @@ impl Store { } let write_txn = self.db.begin_write()?; + let mut applied = 0u64; { let mut kv_table = write_txn.open_table(KV_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 author_table)?; + if Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut author_table)? { + applied += 1; + } } } write_txn.commit()?; - Ok(entries.len() as u64) + Ok(applied) } /// Apply a single signed entry to the store @@ -98,11 +102,12 @@ impl Store { } /// Internal: apply operations from a signed entry to tables + /// Returns true if applied, false if skipped (already applied) fn apply_ops_to_tables( signed_entry: &SignedEntry, kv_table: &mut redb::Table<&[u8], &[u8]>, author_table: &mut redb::Table<&[u8], &[u8]>, - ) -> Result<(), StoreError> { + ) -> Result { 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); @@ -112,7 +117,7 @@ impl Store { 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 + return Ok(false); // Already applied, skip } } } @@ -152,7 +157,7 @@ impl Store { }; author_table.insert(&author[..], author_state.encode_to_vec().as_slice())?; - Ok(()) + Ok(true) } /// Apply a new head to a key, removing ancestor heads (idempotent) @@ -772,9 +777,9 @@ mod tests { 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 + // Replay log - entries already applied, skip all let replayed = store.replay_log(&log_path).unwrap(); - assert_eq!(replayed, 2, "Replayed 2 entries from log"); + assert_eq!(replayed, 0, "0 new entries (all skipped)"); let final_heads = store.get_heads(b"/key").unwrap(); assert_eq!(final_heads.len(), 1, @@ -824,8 +829,8 @@ mod tests { 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"); + // All 3 entries were read but skipped (already applied) + assert_eq!(replayed, 0, "0 new entries (all skipped)"); assert_eq!(store.author_state(&author).unwrap().unwrap().seq, 3, "seq unchanged"); assert_eq!(store.get_heads(b"/key3").unwrap().len(), 1, "heads unchanged"); @@ -876,7 +881,7 @@ mod tests { 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!(replayed, 2, "Only 2 new entries applied (3 skipped)"); 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"); @@ -956,9 +961,9 @@ mod tests { 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 + // Replay log - should apply entries 4 and 5 (skip 1-3) let replayed = store.replay_log(&log_path).unwrap(); - assert_eq!(replayed, 5, "Replayed 5 entries from log"); + assert_eq!(replayed, 2, "Only 2 new entries applied (3 skipped)"); // 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");