nils/lattice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Refactor node initialization and info retrieval, enhance store actor lifecycle, and refine store replay logic.

Browse Source
This commit is contained in:
nils
2025-12-22 02:59:55 +01:00
parent a1f134eb02
commit c2d4219320
5 changed files with 118 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lattice-cli/src/commands.rs
+51 -1
View File
@@ -97,6 +97,14 @@ pub fn commands() -> Vec<Command> {
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
}
lattice-cli/src/main.rs
+4 -8
View File
@@ -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<StoreHandle> = match node.open_root_store() {
Ok(Some((h, open_info))) => {
lattice-cli/src/node.rs
+46 -32
View File
@@ -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<Uuid>,
pub stores: Vec<Uuid>,
}
@@ -63,11 +61,10 @@ impl LatticeNodeBuilder {
Self { data_dir: DataDir::default() }
}
pub fn build(self) -> Result<(LatticeNode, NodeInfo), NodeError> {
pub fn build(self) -> Result<LatticeNode, NodeError> {
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<crate::store_actor::StoreCmd>,
#[allow(dead_code)]
actor_handle: std::thread::JoinHandle<()>,
actor_handle: Option<std::thread::JoinHandle<()>>,
}
impl StoreHandle {
@@ -248,6 +242,16 @@ impl StoreHandle {
.map_err(NodeError::Store)
}
pub fn author_state(&self, author: &[u8; 32]) -> Result<Option<lattice_core::proto::AuthorState>, 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<u64, NodeError> {
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");
lattice-cli/src/store_actor.rs
-2
View File
@@ -49,7 +49,6 @@ pub enum StoreCmd {
pub enum StoreActorError {
Store(StoreError),
SigChain(SigChainError),
ChannelClosed,
}
impl From<StoreError> 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"),
}
}
}
lattice-core/src/store.rs
+17 -12
View File
@@ -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<Path>) -> Result<u64, StoreError> {
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<bool, 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);
@@ -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");