From 57ecbffaed95b0de3ccb58ee891942f3fb595aed Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Mon, 22 Dec 2025 22:11:20 +0100 Subject: [PATCH] feat: introduce `NodeIdentity` and `store_actor` in `lattice-core`, and implement `mesh` networking in `lattice-net` while removing `unicast`. --- lattice-cli/Cargo.toml | 1 - lattice-cli/src/commands.rs | 102 +-- lattice-cli/src/main.rs | 11 +- lattice-cli/src/node.rs | 551 --------------- lattice-core/Cargo.toml | 3 + lattice-core/src/causal_iter.rs | 12 +- lattice-core/src/lib.rs | 9 +- lattice-core/src/log.rs | 22 +- lattice-core/src/meta_store.rs | 23 + lattice-core/src/node.rs | 662 ++++++++++++++---- lattice-core/src/node_identity.rs | 252 +++++++ lattice-core/src/sigchain.rs | 22 +- lattice-core/src/signed_entry.rs | 20 +- lattice-core/src/store.rs | 66 +- .../src/store_actor.rs | 23 +- lattice-net/Cargo.toml | 1 + lattice-net/src/lib.rs | 4 +- lattice-net/src/mesh/mod.rs | 13 + .../src/mesh/protocol.rs | 11 +- .../src/mesh/server.rs | 11 +- .../src => lattice-net/src/mesh}/sync.rs | 46 +- lattice-net/src/unicast.rs | 3 - 22 files changed, 984 insertions(+), 884 deletions(-) delete mode 100644 lattice-cli/src/node.rs create mode 100644 lattice-core/src/node_identity.rs rename {lattice-cli => lattice-core}/src/store_actor.rs (94%) create mode 100644 lattice-net/src/mesh/mod.rs rename lattice-cli/src/sync_protocol.rs => lattice-net/src/mesh/protocol.rs (88%) rename lattice-cli/src/accept_handler.rs => lattice-net/src/mesh/server.rs (94%) rename {lattice-cli/src => lattice-net/src/mesh}/sync.rs (83%) delete mode 100644 lattice-net/src/unicast.rs diff --git a/lattice-cli/Cargo.toml b/lattice-cli/Cargo.toml index 758f5ed..5338563 100644 --- a/lattice-cli/Cargo.toml +++ b/lattice-cli/Cargo.toml @@ -17,7 +17,6 @@ hex = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } shlex = "1" -hostname = "0.4" serde_json = "1" iroh = { workspace = true } prost = { workspace = true } diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index c87f89f..8a927f7 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -1,7 +1,7 @@ //! CLI command handlers -use crate::node::{LatticeNode, StoreHandle, PeerStatus}; -use lattice_core::Uuid; +use lattice_core::{Node, StoreHandle}; +use lattice_core::{Uuid, PeerStatus}; use lattice_net::LatticeEndpoint; use chrono::DateTime; use std::time::Instant; @@ -21,7 +21,7 @@ fn block_async(f: F) -> F::Output { }) } -pub type Handler = fn(&LatticeNode, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult; +pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult; pub struct Command { pub name: &'static str, @@ -167,11 +167,11 @@ pub fn commands() -> Vec { // --- Store management --- -fn cmd_init(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { match block_async(node.init()) { Ok((store_id, handle)) => { println!("Initialized with root store: {}", store_id); - println!("Node pubkey stored in /nodes/{}/info", hex::encode(node.node_id())); + println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id())); CommandResult::SwitchTo(handle) } Err(e) => { @@ -181,7 +181,7 @@ fn cmd_init(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option< } } -fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { match node.create_store() { Ok(store_id) => { println!("Created store: {}", store_id); @@ -203,7 +203,7 @@ fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: } } -fn cmd_use_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store_id = match Uuid::parse_str(&args[0]) { Ok(id) => id, Err(_) => { @@ -229,7 +229,7 @@ fn cmd_use_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Op } } -fn cmd_list_stores(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { let stores = match node.list_stores() { Ok(s) => s, Err(e) => { @@ -252,7 +252,7 @@ fn cmd_list_stores(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: O // --- Info --- -fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { println!("\nCommands:"); for cmd in commands() { if cmd.args.is_empty() { @@ -266,7 +266,7 @@ fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option CommandResult::Ok } -fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_status(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { println!("Node ID: {}", hex::encode(node.node_id())); println!("Data: {}", node.data_path().display()); match node.root_store() { @@ -320,7 +320,7 @@ fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option // --- KV --- -fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -333,7 +333,7 @@ fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<& CommandResult::Ok } -fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -384,7 +384,7 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<& CommandResult::Ok } -fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -397,7 +397,7 @@ fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Optio CommandResult::Ok } -fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -448,7 +448,7 @@ 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>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { @@ -492,7 +492,7 @@ fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: // --- Peer management --- -fn cmd_invite(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_invite(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { @@ -510,22 +510,27 @@ fn cmd_invite(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option } }; - // Write /nodes/{pubkey}/info with inviter info - let info_key = format!("/nodes/{}/info", pubkey_hex); + // Write peer info as separate keys let inviter_hex = hex::encode(node.node_id()); let added_at = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0); - let info = serde_json::json!({ - "added_by": inviter_hex, - "added_at": added_at - }); - match block_async(store.put(info_key.as_bytes(), info.to_string().as_bytes())) { + let added_by_key = format!("/nodes/{}/added_by", pubkey_hex); + match block_async(store.put(added_by_key.as_bytes(), inviter_hex.as_bytes())) { Ok(_) => {} Err(e) => { - eprintln!("Error writing info: {}", e); + eprintln!("Error writing added_by: {}", e); + return CommandResult::Ok; + } + } + + let added_at_key = format!("/nodes/{}/added_at", pubkey_hex); + match block_async(store.put(added_at_key.as_bytes(), added_at.to_string().as_bytes())) { + Ok(_) => {} + Err(e) => { + eprintln!("Error writing added_at: {}", e); return CommandResult::Ok; } } @@ -541,12 +546,13 @@ fn cmd_invite(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option } println!("Invited peer: {}", pubkey_hex); - println!(" /nodes/{}/info", pubkey_hex); + println!(" /nodes/{}/added_by", pubkey_hex); + println!(" /nodes/{}/added_at", pubkey_hex); println!(" /nodes/{}/status = {} (will become active after sync)", pubkey_hex, PeerStatus::Invited.as_str()); CommandResult::Ok } -fn cmd_peers(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_peers(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { @@ -586,23 +592,27 @@ fn cmd_peers(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option std::collections::HashMap::new(); for (pubkey, status) in &peers { - // Try to get info for name/added_at - let info_key = format!("/nodes/{}/info", pubkey); - let mut name = String::new(); - let mut added = String::new(); + // Try to get name and added_at from separate keys + let name_key = format!("/nodes/{}/name", pubkey); + let added_at_key = format!("/nodes/{}/added_at", pubkey); - if let Ok(Some(info_bytes)) = block_async(store.get(info_key.as_bytes())) { - if let Ok(info) = serde_json::from_slice::(&info_bytes) { - if let Some(n) = info.get("name").and_then(|v| v.as_str()) { - name = n.to_string(); - } - if let Some(ts) = info.get("added_at").and_then(|v| v.as_u64()) { - if let Some(dt) = DateTime::from_timestamp(ts as i64, 0) { - added = dt.format("%Y-%m-%d").to_string(); - } + let name = match block_async(store.get(name_key.as_bytes())) { + Ok(Some(bytes)) => String::from_utf8_lossy(&bytes).to_string(), + _ => String::new(), + }; + + let added = match block_async(store.get(added_at_key.as_bytes())) { + Ok(Some(bytes)) => { + if let Ok(ts) = String::from_utf8_lossy(&bytes).parse::() { + DateTime::from_timestamp(ts, 0) + .map(|dt| dt.format("%Y-%m-%d").to_string()) + .unwrap_or_default() + } else { + String::new() } } - } + _ => String::new(), + }; by_status.entry(*status) .or_default() @@ -631,7 +641,7 @@ fn cmd_peers(_node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option CommandResult::Ok } -fn cmd_remove(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_remove(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { @@ -683,7 +693,7 @@ fn cmd_remove(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option CommandResult::Ok } -fn cmd_join(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let endpoint = match endpoint { Some(ep) => ep, None => { @@ -707,7 +717,7 @@ fn cmd_join(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&L println!("Joining mesh via {}...", peer_id.fmt_short()); - match block_async(crate::sync::join_mesh(node, endpoint, peer_id)) { + match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) { Ok(handle) => { println!("Joined mesh! Use 'sync' command to sync entries."); CommandResult::SwitchTo(handle) @@ -719,7 +729,7 @@ fn cmd_join(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&L } } -fn cmd_sync(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let endpoint = match endpoint { Some(ep) => ep, None => { @@ -738,7 +748,7 @@ fn cmd_sync(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&L if args.is_empty() { // Sync with all active peers - match block_async(crate::sync::sync_all(node, endpoint, store)) { + match block_async(lattice_net::sync_all(node, endpoint, store)) { Ok(results) => { if results.is_empty() { println!("No peers to sync with."); @@ -760,7 +770,7 @@ fn cmd_sync(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&L }; println!("Syncing with {}...", peer_id.fmt_short()); - match block_async(crate::sync::sync_with_peer(node, endpoint, store, peer_id)) { + match block_async(lattice_net::sync_with_peer(node, endpoint, store, peer_id)) { Ok(result) => { println!("Sync complete! Applied {} entries (peer sent {})", result.entries_applied, result.entries_sent_by_peer); diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index 2637591..4ef29b0 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -1,15 +1,10 @@ //! Lattice Interactive CLI -mod accept_handler; -mod node; mod commands; -mod store_actor; -mod sync_protocol; -mod sync; -use accept_handler::spawn_accept_loop; +use lattice_net::spawn_accept_loop; use commands::CommandResult; -use node::{LatticeNodeBuilder, StoreHandle}; +use lattice_core::{NodeBuilder, StoreHandle}; use rustyline::error::ReadlineError; use rustyline::DefaultEditor; use std::sync::Arc; @@ -20,7 +15,7 @@ async fn main() { println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION")); println!("Type 'help' for commands, 'quit' to exit.\n"); - let node = match LatticeNodeBuilder::new().build() { + let node = match NodeBuilder::new().build() { Ok(n) => n, Err(e) => { eprintln!("Failed to initialize: {}", e); diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs deleted file mode 100644 index fd3f33a..0000000 --- a/lattice-cli/src/node.rs +++ /dev/null @@ -1,551 +0,0 @@ -//! Local Lattice node API with multi-store support - -use lattice_core::{ - DataDir, MetaStore, Node, SigChain, Store, Uuid, - log::LogError, - meta_store::MetaStoreError, - sigchain::SigChainError, - store::StoreError, -}; -use std::path::Path; -use std::rc::Rc; -use thiserror::Error; - -#[derive(Error, Debug)] -pub enum NodeError { - #[error("IO error: {0}")] - Io(#[from] std::io::Error), - - #[error("Store error: {0}")] - Store(#[from] StoreError), - - #[error("MetaStore error: {0}")] - MetaStore(#[from] MetaStoreError), - - #[error("SigChain error: {0}")] - SigChain(#[from] SigChainError), - - #[error("Log error: {0}")] - Log(#[from] LogError), - - #[error("Node error: {0}")] - Node(#[from] lattice_core::node::NodeError), - - #[error("Already initialized")] - AlreadyInitialized, - - #[error("Channel closed")] - ChannelClosed, - - #[error("Actor error: {0}")] - Actor(String), -} - -/// Peer status values used across the system -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum PeerStatus { - /// Peer has been invited but hasn't joined yet - Invited, - /// Peer is active and can sync - Active, - /// Peer has been removed from the mesh - Removed, -} - -impl PeerStatus { - pub fn as_str(&self) -> &'static str { - match self { - PeerStatus::Invited => "invited", - PeerStatus::Active => "active", - PeerStatus::Removed => "removed", - } - } - - pub fn from_str(s: &str) -> Option { - match s { - "invited" => Some(PeerStatus::Invited), - "active" => Some(PeerStatus::Active), - "removed" => Some(PeerStatus::Removed), - _ => None, - } - } -} - -pub struct NodeInfo { - pub node_id: String, - pub data_path: String, - pub stores: Vec, -} - -pub struct StoreInfo { - pub store_id: Uuid, - pub entries_replayed: u64, -} - -pub struct LatticeNodeBuilder { - pub data_dir: DataDir, -} - -impl LatticeNodeBuilder { - pub fn new() -> Self { - Self { data_dir: DataDir::default() } - } - - pub fn build(self) -> Result { - self.data_dir.ensure_dirs()?; - - let key_path = self.data_dir.identity_key(); - let node = if key_path.exists() { - Node::load(&key_path)? - } else { - let node = Node::generate(); - node.save(&key_path)?; - node - }; - - let meta = MetaStore::open(self.data_dir.meta_db())?; - - Ok(LatticeNode { - data_dir: self.data_dir, - node: Rc::new(node), - meta, - }) - } -} - -impl Default for LatticeNodeBuilder { - fn default() -> Self { Self::new() } -} - -/// A local Lattice node (manages identity and store registry) -pub struct LatticeNode { - data_dir: DataDir, - node: Rc, - meta: MetaStore, -} - -impl LatticeNode { - 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() - } - - /// Get the secret key bytes for Iroh integration (same Ed25519 key) - pub fn secret_key_bytes(&self) -> [u8; 32] { - self.node.secret_key_bytes() - } - - pub fn data_path(&self) -> &Path { - self.data_dir.base() - } - - /// Get the root store ID - pub fn root_store(&self) -> Result, NodeError> { - Ok(self.meta.root_store()?) - } - /// Open the root store if set - pub fn open_root_store(&self) -> Result, NodeError> { - match self.meta.root_store()? { - Some(id) => Ok(Some(self.open_store(id)?)), - None => Ok(None), - } - } - - /// Initialize the node with a root store (fails if already initialized). - /// Writes the node's pubkey to `/nodes/{pubkey}/info` in the root store. - pub async fn init(&self) -> Result<(Uuid, StoreHandle), NodeError> { - if self.meta.root_store()?.is_some() { - return Err(NodeError::AlreadyInitialized); - } - let store_id = self.create_store()?; - self.meta.set_root_store(store_id)?; - - // Open the store and write our node info - let (handle, _) = self.open_store(store_id)?; - let pubkey_hex = hex::encode(self.node.public_key_bytes()); - let key = format!("/nodes/{}/info", pubkey_hex); - - // Store node metadata: name (hostname), added_at (timestamp) - let hostname = hostname::get() - .map(|s| s.to_string_lossy().to_string()) - .unwrap_or_else(|_| "unknown".to_string()); - let added_at = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis()) - .unwrap_or(0); - let info = serde_json::json!({ - "name": hostname, - "added_at": added_at - }); - handle.put(key.as_bytes(), info.to_string().as_bytes()).await?; - - // Write status = active - let status_key = format!("/nodes/{}/status", pubkey_hex); - handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?; - - Ok((store_id, handle)) - } - - pub fn list_stores(&self) -> Result, NodeError> { - Ok(self.meta.list_stores()?) - } - - pub fn create_store(&self) -> Result { - let store_id = Uuid::new_v4(); - self.create_store_internal(store_id) - } - - /// Create a store with a specific UUID (for joining existing mesh) - pub fn create_store_with_uuid(&self, store_id: Uuid) -> Result { - self.create_store_internal(store_id) - } - - /// Set a store as the root store - pub fn set_root_store(&self, store_id: Uuid) -> Result<(), NodeError> { - self.meta.set_root_store(store_id)?; - Ok(()) - } - - fn create_store_internal(&self, store_id: Uuid) -> Result { - self.data_dir.ensure_store_dirs(store_id)?; - let _ = Store::open(self.data_dir.store_state_db(store_id))?; - self.meta.add_store(store_id)?; - Ok(store_id) - } - - pub fn open_store(&self, store_id: Uuid) -> Result<(StoreHandle, StoreInfo), NodeError> { - self.data_dir.ensure_store_dirs(store_id)?; - - let author_id_hex = hex::encode(self.node.public_key_bytes()); - let log_path = self.data_dir.store_log_file(store_id, &author_id_hex); - - let sigchain = if log_path.exists() { - SigChain::from_log(&log_path, *store_id.as_bytes(), self.node.public_key_bytes())? - } else { - SigChain::new(&log_path, *store_id.as_bytes(), self.node.public_key_bytes()) - }; - - let store = Store::open(self.data_dir.store_state_db(store_id))?; - let entries_replayed = if log_path.exists() { - store.replay_log(&log_path)? - } else { - 0 - }; - - let info = StoreInfo { store_id, entries_replayed }; - - // Spawn actor thread - actor owns store, sigchain, and node copy - let (tx, actor_handle) = crate::store_actor::spawn_store_actor( - store_id, - store, - sigchain, - (*self.node).clone(), - ); - - let handle = StoreHandle { - store_id, - tx, - actor_handle: Some(actor_handle), - }; - - Ok((handle, info)) - } -} - -/// A handle to a specific store - wraps channel to actor thread -pub struct StoreHandle { - store_id: Uuid, - tx: tokio::sync::mpsc::Sender, - actor_handle: Option>, -} - -impl Clone for StoreHandle { - fn clone(&self) -> Self { - Self { - store_id: self.store_id, - tx: self.tx.clone(), - actor_handle: None, // Clones don't own the actor thread - } - } -} - -impl StoreHandle { - pub fn id(&self) -> Uuid { self.store_id } - - pub async fn get(&self, key: &[u8]) -> Result>, NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::Get { key: key.to_vec(), resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn get_heads(&self, key: &[u8]) -> Result, NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::GetHeads { key: key.to_vec(), resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn list(&self) -> Result, Vec)>, NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::List { resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn log_seq(&self) -> u64 { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - let _ = self.tx.send(StoreCmd::LogSeq { resp: resp_tx }).await; - resp_rx.await.unwrap_or(0) - } - - pub async fn applied_seq(&self) -> Result { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::AppliedSeq { resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn author_state(&self, author: &[u8; 32]) -> Result, NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn sync_state(&self) -> Result { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::SyncState { resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn read_entries_after(&self, author: &[u8; 32], from_hash: Option<[u8; 32]>) -> Result, NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::ReadEntriesAfter { author: *author, from_hash, resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn apply_entry(&self, entry: lattice_core::proto::SignedEntry) -> Result<(), NodeError> { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::ApplyEntry { entry, resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(NodeError::Store) - } - - pub async fn put(&self, key: &[u8], value: &[u8]) -> Result { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::Put { key: key.to_vec(), value: value.to_vec(), resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(|e| NodeError::Actor(e.to_string())) - } - - pub async fn delete(&self, key: &[u8]) -> Result { - use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::Delete { key: key.to_vec(), resp: resp_tx }).await - .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.await - .map_err(|_| NodeError::ChannelClosed)? - .map_err(|e| NodeError::Actor(e.to_string())) - } - -} - -impl Drop for StoreHandle { - fn drop(&mut self) { - // Only send shutdown if we own the actor (non-cloned handle) - if let Some(handle) = self.actor_handle.take() { - let _ = self.tx.try_send(crate::store_actor::StoreCmd::Shutdown); - let _ = handle.join(); - } - // Clones (actor_handle = None) don't send shutdown - actor keeps running - } -} - -#[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)); - let _ = std::fs::remove_dir_all(&path); - DataDir::new(path) - } - - #[tokio::test] - async fn test_create_and_open_store() { - let data_dir = temp_data_dir("meta_store"); - - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("Failed to create node"); - - assert!(node.info().stores.is_empty()); - - let store_id = node.create_store().expect("Failed to create store"); - - // Verify it's in the list - let stores = node.list_stores().expect("list failed"); - assert!(stores.contains(&store_id)); - - let (handle, _) = node.open_store(store_id).expect("Failed to open store"); - handle.put(b"/key", b"value").await.expect("put failed"); - assert_eq!(handle.get(b"/key").await.unwrap(), Some(b"value".to_vec())); - - let _ = std::fs::remove_dir_all(data_dir.base()); - } - - #[tokio::test] - async fn test_store_isolation() { - let data_dir = temp_data_dir("meta_isolation"); - - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("Failed to create node"); - - let store_a = node.create_store().expect("create A"); - let store_b = node.create_store().expect("create B"); - - let (handle_a, _) = node.open_store(store_a).expect("open A"); - handle_a.put(b"/key", b"from A").await.expect("put A"); - - let (handle_b, _) = node.open_store(store_b).expect("open B"); - assert_eq!(handle_b.get(b"/key").await.unwrap(), None); - - assert_eq!(handle_a.get(b"/key").await.unwrap(), Some(b"from A".to_vec())); - - let _ = std::fs::remove_dir_all(data_dir.base()); - } - - #[tokio::test] - async fn test_init_creates_root_store() { - let data_dir = temp_data_dir("init_root"); - - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("create node"); - - // Initially no root store - assert!(node.root_store().unwrap().is_none()); - - // Init creates root store - let (root_id, _handle) = node.init().await.expect("init failed"); - assert_eq!(node.root_store().unwrap(), Some(root_id)); - - let _ = std::fs::remove_dir_all(data_dir.base()); - } - - #[tokio::test] - async fn test_duplicate_init_fails() { - let data_dir = temp_data_dir("init_dup"); - - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("create node"); - - node.init().await.expect("first init"); - - // Second init should fail - match node.init().await { - Ok(_) => panic!("Expected AlreadyInitialized error"), - Err(e) => match e { - NodeError::AlreadyInitialized => (), - _ => panic!("Expected AlreadyInitialized, got {:?}", e), - }, - } - - let _ = std::fs::remove_dir_all(data_dir.base()); - } - - #[tokio::test] - async fn test_root_store_in_info_after_init() { - let data_dir = temp_data_dir("init_info"); - - // First session: init - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("create node"); - let (root_id, _) = node.init().await.expect("init"); - drop(node); // End first session - - // Second session: root_store should persist - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("reload node"); - - assert_eq!(node.root_store().unwrap(), Some(root_id)); - - let _ = std::fs::remove_dir_all(data_dir.base()); - } - - #[tokio::test] - async fn test_idempotent_put_and_delete() { - let data_dir = temp_data_dir("idempotent"); - - let node = LatticeNodeBuilder { data_dir: data_dir.clone() } - .build() - .expect("create node"); - let (_, store) = node.init().await.expect("init"); - - // Get baseline seq after init - let baseline = store.log_seq().await; - - // Put twice with same value - second should be idempotent - let seq1 = store.put(b"/key", b"value").await.expect("put 1"); - assert_eq!(seq1, baseline + 1); - - let seq2 = store.put(b"/key", b"value").await.expect("put 2"); - assert_eq!(seq2, baseline + 1, "Second put should be idempotent (no new entry)"); - - assert_eq!(store.log_seq().await, baseline + 1); - - // Delete twice - second should be idempotent - let seq3 = store.delete(b"/key").await.expect("delete 1"); - assert_eq!(seq3, baseline + 2); - - let seq4 = store.delete(b"/key").await.expect("delete 2"); - assert_eq!(seq4, baseline + 2, "Second delete should be idempotent (no new entry)"); - - assert_eq!(store.log_seq().await, baseline + 2); - - let _ = std::fs::remove_dir_all(data_dir.base()); - } -} diff --git a/lattice-core/Cargo.toml b/lattice-core/Cargo.toml index 6e7cd2d..9967988 100644 --- a/lattice-core/Cargo.toml +++ b/lattice-core/Cargo.toml @@ -16,6 +16,9 @@ blake3 = { workspace = true } hex = { workspace = true } redb = { workspace = true } uuid = { workspace = true } +tokio = { workspace = true } +hostname = "0.4" +serde_json = "1" [build-dependencies] prost-build = { workspace = true } diff --git a/lattice-core/src/causal_iter.rs b/lattice-core/src/causal_iter.rs index b49a436..df5a924 100644 --- a/lattice-core/src/causal_iter.rs +++ b/lattice-core/src/causal_iter.rs @@ -102,10 +102,10 @@ mod tests { use super::*; use crate::hlc::HLC; use crate::clock::MockClock; - use crate::node::Node; + use crate::node_identity::NodeIdentity; use crate::signed_entry::EntryBuilder; - fn make_entry(node: &Node, seq: u64, clock_ms: u64) -> SignedEntry { + fn make_entry(node: &NodeIdentity, seq: u64, clock_ms: u64) -> SignedEntry { let clock = MockClock::new(clock_ms); EntryBuilder::new(seq, HLC::now_with_clock(&clock)) .store_id(vec![0u8; 16]) @@ -122,7 +122,7 @@ mod tests { #[test] fn test_single_queue() { - let node = Node::generate(); + let node = NodeIdentity::generate(); let entries: VecDeque<_> = vec![ make_entry(&node, 1, 1000), make_entry(&node, 2, 2000), @@ -135,8 +135,8 @@ mod tests { #[test] fn test_merge_multiple_queues() { - let node_a = Node::generate(); - let node_b = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); // Author A: entries at time 1000, 3000 let queue_a: VecDeque<_> = vec![ @@ -167,7 +167,7 @@ mod tests { #[test] fn test_many_authors() { // Test with 10 authors to verify heap behavior - let nodes: Vec<_> = (0..10).map(|_| Node::generate()).collect(); + let nodes: Vec<_> = (0..10).map(|_| NodeIdentity::generate()).collect(); let queues: Vec> = nodes.iter().enumerate().map(|(i, node)| { vec![make_entry(node, 1, (i * 100 + 50) as u64)].into() }).collect(); diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index ff24a06..b960cce 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -1,7 +1,7 @@ //! Lattice Core //! //! Core types for the Lattice distributed mesh: -//! - **Node**: Identity with Ed25519 keypair +//! - **NodeIdentity**: Cryptographic identity with Ed25519 keypair //! - **SigChain**: Append-only cryptographically signed log //! - **Entry**: Atomic operations in the log //! - **SyncState**: Per-author sequence tracking for reconciliation @@ -14,6 +14,7 @@ //! - **Store**: Persistent KV state from log replay //! - **CausalIter**: Merge-sort iterator for HLC-ordered sync +pub mod node_identity; pub mod node; pub mod sigchain; pub mod entry; @@ -27,12 +28,14 @@ pub mod log; pub mod store; pub mod meta_store; pub mod causal_iter; +pub mod store_actor; // Constants /// Maximum size of a serialized SignedEntry (16 MB) pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024; -pub use node::Node; +pub use node_identity::{NodeIdentity, PeerStatus}; +pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError}; pub use sigchain::{SigChain, SigChainManager}; pub use entry::Entry; pub use sync_state::{SyncState, AuthorInfo, MissingRange}; @@ -46,4 +49,4 @@ pub use meta_store::MetaStore; pub use proto::HeadInfo; pub use uuid::Uuid; pub use causal_iter::CausalEntryIter; - +pub use store_actor::{StoreActor, StoreCmd, StoreActorError, spawn_store_actor}; diff --git a/lattice-core/src/log.rs b/lattice-core/src/log.rs index eab2d73..3c0eb3d 100644 --- a/lattice-core/src/log.rs +++ b/lattice-core/src/log.rs @@ -207,7 +207,7 @@ mod tests { use super::*; use crate::clock::MockClock; use crate::hlc::HLC; - use crate::node::Node; + use crate::node_identity::NodeIdentity; use crate::signed_entry::EntryBuilder; use std::env::temp_dir; @@ -226,7 +226,7 @@ mod tests { let path = temp_log_path("single_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let hlc = HLC::now_with_clock(&clock); @@ -248,7 +248,7 @@ mod tests { let path = temp_log_path("multiple_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); for i in 1..=5 { @@ -269,7 +269,7 @@ mod tests { let path = temp_log_path("after_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let mut entries = Vec::new(); @@ -301,7 +301,7 @@ mod tests { let path = temp_log_path("not_found_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -321,7 +321,7 @@ mod tests { let path = temp_log_path("reader_hash_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -368,7 +368,7 @@ mod tests { let path = temp_log_path("corrupted_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -399,7 +399,7 @@ mod tests { let path = temp_log_path("truncated_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -430,7 +430,7 @@ mod tests { let path = temp_log_path("too_large_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); // Create payload larger than MAX_ENTRY_SIZE @@ -455,7 +455,7 @@ mod tests { let path = temp_log_path("boundary_last_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -506,7 +506,7 @@ mod tests { let path = temp_log_path("corruption_middle_v6"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); // Write 3 entries diff --git a/lattice-core/src/meta_store.rs b/lattice-core/src/meta_store.rs index cda2178..cb5881d 100644 --- a/lattice-core/src/meta_store.rs +++ b/lattice-core/src/meta_store.rs @@ -13,6 +13,7 @@ const STORES_TABLE: TableDefinition<&[u8], u64> = TableDefinition::new("stores") const META_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("meta"); const META_ROOT_STORE: &str = "root_store"; +const META_NAME: &str = "name"; #[derive(Error, Debug)] pub enum MetaStoreError { @@ -106,6 +107,28 @@ impl MetaStore { write_txn.commit()?; Ok(()) } + + /// Get the node's display name + pub fn name(&self) -> Result, MetaStoreError> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(META_TABLE)?; + + match table.get(META_NAME)? { + Some(value) => Ok(Some(String::from_utf8_lossy(value.value()).to_string())), + None => Ok(None), + } + } + + /// Set the node's display name + pub fn set_name(&self, name: &str) -> Result<(), MetaStoreError> { + let write_txn = self.db.begin_write()?; + { + let mut table = write_txn.open_table(META_TABLE)?; + table.insert(META_NAME, name.as_bytes())?; + } + write_txn.commit()?; + Ok(()) + } } #[cfg(test)] diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index e7d58d6..dab82b4 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -1,133 +1,401 @@ -//! Node identity and cryptographic keys -//! -//! Each node has an Ed25519 keypair: -//! - Private key: stored locally in `identity.key` (never replicated) -//! - Public key: serves as the node's identity (32 bytes) +//! Local Lattice node API with multi-store support -use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey}; -use rand::rngs::OsRng; -use std::fs; -use std::io::{self, Read, Write}; +use crate::{ + DataDir, MetaStore, NodeIdentity, PeerStatus, SigChain, Store, Uuid, + log::LogError, + meta_store::MetaStoreError, + sigchain::SigChainError, + store::StoreError, + spawn_store_actor, StoreCmd, + node_identity::NodeError as IdentityError, +}; use std::path::Path; +use std::rc::Rc; use thiserror::Error; -/// Errors that can occur during node operations #[derive(Error, Debug)] pub enum NodeError { #[error("IO error: {0}")] - Io(#[from] io::Error), + Io(#[from] std::io::Error), - #[error("Invalid key length: expected 32 bytes, got {0}")] - InvalidKeyLength(usize), + #[error("Store error: {0}")] + Store(#[from] StoreError), - #[error("Invalid signature")] - InvalidSignature, + #[error("MetaStore error: {0}")] + MetaStore(#[from] MetaStoreError), + + #[error("SigChain error: {0}")] + SigChain(#[from] SigChainError), + + #[error("Log error: {0}")] + Log(#[from] LogError), + + #[error("Node error: {0}")] + Node(#[from] IdentityError), + + #[error("Already initialized")] + AlreadyInitialized, + + #[error("Channel closed")] + ChannelClosed, + + #[error("Actor error: {0}")] + Actor(String), } -/// A node in the Lattice mesh. -/// -/// Each node has an Ed25519 keypair used for signing sigchain entries -/// and establishing trust within the network. -#[derive(Clone)] +pub struct NodeInfo { + pub node_id: String, + pub data_path: String, + pub stores: Vec, +} + +pub struct StoreInfo { + pub store_id: Uuid, + pub entries_replayed: u64, +} + +pub struct NodeBuilder { + pub data_dir: DataDir, +} + +impl NodeBuilder { + pub fn new() -> Self { + Self { data_dir: DataDir::default() } + } + + 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() { + NodeIdentity::load(&key_path)? + } else { + let node = NodeIdentity::generate(); + node.save(&key_path)?; + node + }; + + let meta = MetaStore::open(self.data_dir.meta_db())?; + + // Set hostname on first creation + if is_new { + let hostname = hostname::get() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|_| "unknown".to_string()); + let _ = meta.set_name(&hostname); + } + + Ok(Node { + data_dir: self.data_dir, + node: Rc::new(node), + meta, + }) + } +} + +impl Default for NodeBuilder { + fn default() -> Self { Self::new() } +} + +/// A local Lattice node (manages identity and store registry) pub struct Node { - signing_key: SigningKey, + data_dir: DataDir, + node: Rc, + meta: MetaStore, } impl Node { - /// Generate a new node with a random keypair. - pub fn generate() -> Self { - let signing_key = SigningKey::generate(&mut OsRng); - Self { signing_key } - } - - /// Create a node from an existing signing key. - pub fn from_signing_key(signing_key: SigningKey) -> Self { - Self { signing_key } - } - - /// Load a node's identity from a key file, or generate and save if it doesn't exist. - pub fn load_or_generate(path: impl AsRef) -> Result { - let path = path.as_ref(); - if path.exists() { - Self::load(path) - } else { - let node = Self::generate(); - node.save(path)?; - Ok(node) + 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(), } } - /// Load a node's identity from a key file. - pub fn load(path: impl AsRef) -> Result { - let mut file = fs::File::open(path)?; - let mut bytes = Vec::new(); - file.read_to_end(&mut bytes)?; - - if bytes.len() != 32 { - return Err(NodeError::InvalidKeyLength(bytes.len())); - } - - let key_bytes: [u8; 32] = bytes.try_into().unwrap(); - let signing_key = SigningKey::from_bytes(&key_bytes); - Ok(Self { signing_key }) + pub fn node_id(&self) -> [u8; 32] { + self.node.public_key_bytes() } - /// Save the node's private key to a file. - pub fn save(&self, path: impl AsRef) -> Result<(), NodeError> { - let path = path.as_ref(); + /// Get the secret key bytes for Iroh integration (same Ed25519 key) + pub fn secret_key_bytes(&self) -> [u8; 32] { + self.node.secret_key_bytes() + } + + pub fn data_path(&self) -> &Path { + self.data_dir.base() + } + + /// Get the node's display name (from meta.db, set on creation) + pub fn name(&self) -> Option { + self.meta.name().ok().flatten() + } + + /// Set the node's display name. + /// Updates meta.db and if a store handle is provided, also updates /nodes/{pubkey}/name + pub async fn set_name(&self, name: &str, store: Option<&StoreHandle>) -> Result<(), NodeError> { + // Update meta.db + self.meta.set_name(name)?; - // Create parent directories if they don't exist - if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; + // If store provided, update there too + if let Some(handle) = store { + let pubkey_hex = hex::encode(self.node.public_key_bytes()); + let name_key = format!("/nodes/{}/name", pubkey_hex); + handle.put(name_key.as_bytes(), name.as_bytes()).await?; } - let mut file = fs::File::create(path)?; - file.write_all(self.signing_key.as_bytes())?; Ok(()) } - /// Get the node's public key (identity). - pub fn public_key(&self) -> VerifyingKey { - self.signing_key.verifying_key() + /// Get the root store ID + pub fn root_store(&self) -> Result, NodeError> { + Ok(self.meta.root_store()?) + } + /// Open the root store if set + pub fn open_root_store(&self) -> Result, NodeError> { + match self.meta.root_store()? { + Some(id) => Ok(Some(self.open_store(id)?)), + None => Ok(None), + } } - /// Get the node's public key as bytes (32 bytes). - pub fn public_key_bytes(&self) -> [u8; 32] { - self.signing_key.verifying_key().to_bytes() + /// Initialize the node with a root store (fails if already initialized). + /// Writes the node's pubkey to `/nodes/{pubkey}/info` in the root store. + pub async fn init(&self) -> Result<(Uuid, StoreHandle), NodeError> { + if self.meta.root_store()?.is_some() { + return Err(NodeError::AlreadyInitialized); + } + let store_id = self.create_store()?; + self.meta.set_root_store(store_id)?; + + // Open the store and write our node info as separate keys + let (handle, _) = self.open_store(store_id)?; + let pubkey_hex = hex::encode(self.node.public_key_bytes()); + + // Store node metadata as separate keys + if let Some(name) = self.name() { + let name_key = format!("/nodes/{}/name", pubkey_hex); + handle.put(name_key.as_bytes(), name.as_bytes()).await?; + } + + let added_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let added_at_key = format!("/nodes/{}/added_at", pubkey_hex); + handle.put(added_at_key.as_bytes(), added_at.to_string().as_bytes()).await?; + + // Write status = active + let status_key = format!("/nodes/{}/status", pubkey_hex); + handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?; + + Ok((store_id, handle)) } - /// Get the signing key for creating signatures. - pub fn signing_key(&self) -> &SigningKey { - &self.signing_key + pub fn list_stores(&self) -> Result, NodeError> { + Ok(self.meta.list_stores()?) } - /// Get the secret key bytes (32 bytes) for Iroh integration. - /// WARNING: Handle with care - this exposes the private key material. - pub fn secret_key_bytes(&self) -> [u8; 32] { - self.signing_key.to_bytes() + pub fn create_store(&self) -> Result { + let store_id = Uuid::new_v4(); + self.create_store_internal(store_id) + } + + /// Create a store with a specific UUID (for joining existing mesh) + pub fn create_store_with_uuid(&self, store_id: Uuid) -> Result { + self.create_store_internal(store_id) + } + + /// Set a store as the root store + pub fn set_root_store(&self, store_id: Uuid) -> Result<(), NodeError> { + self.meta.set_root_store(store_id)?; + Ok(()) + } + + fn create_store_internal(&self, store_id: Uuid) -> Result { + self.data_dir.ensure_store_dirs(store_id)?; + let _ = Store::open(self.data_dir.store_state_db(store_id))?; + self.meta.add_store(store_id)?; + Ok(store_id) } - /// Sign a message. - pub fn sign(&self, message: &[u8]) -> Signature { - self.signing_key.sign(message) + pub fn open_store(&self, store_id: Uuid) -> Result<(StoreHandle, StoreInfo), NodeError> { + self.data_dir.ensure_store_dirs(store_id)?; + + let author_id_hex = hex::encode(self.node.public_key_bytes()); + let log_path = self.data_dir.store_log_file(store_id, &author_id_hex); + + let sigchain = if log_path.exists() { + SigChain::from_log(&log_path, *store_id.as_bytes(), self.node.public_key_bytes())? + } else { + SigChain::new(&log_path, *store_id.as_bytes(), self.node.public_key_bytes()) + }; + + let store = Store::open(self.data_dir.store_state_db(store_id))?; + let entries_replayed = if log_path.exists() { + store.replay_log(&log_path)? + } else { + 0 + }; + + let info = StoreInfo { store_id, entries_replayed }; + + // Spawn actor thread - actor owns store, sigchain, and node copy + let (tx, actor_handle) = spawn_store_actor( + store_id, + store, + sigchain, + (*self.node).clone(), + ); + + let handle = StoreHandle { + store_id, + tx, + actor_handle: Some(actor_handle), + }; + + Ok((handle, info)) + } +} + +/// A handle to a specific store - wraps channel to actor thread +pub struct StoreHandle { + store_id: Uuid, + tx: tokio::sync::mpsc::Sender, + actor_handle: Option>, +} + +impl Clone for StoreHandle { + fn clone(&self) -> Self { + Self { + store_id: self.store_id, + tx: self.tx.clone(), + actor_handle: None, // Clones don't own the actor thread + } + } +} + +impl StoreHandle { + pub fn id(&self) -> Uuid { self.store_id } + + pub async fn get(&self, key: &[u8]) -> Result>, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Get { key: key.to_vec(), resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) } - /// Verify a signature against this node's public key. - pub fn verify(&self, message: &[u8], signature: &Signature) -> Result<(), NodeError> { - self.public_key() - .verify(message, signature) - .map_err(|_| NodeError::InvalidSignature) + pub async fn get_heads(&self, key: &[u8]) -> Result, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::GetHeads { key: key.to_vec(), resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) } - /// Verify a signature using a raw public key. - pub fn verify_with_key( - public_key: &VerifyingKey, - message: &[u8], - signature: &Signature, - ) -> Result<(), NodeError> { - public_key - .verify(message, signature) - .map_err(|_| NodeError::InvalidSignature) + pub async fn list(&self) -> Result, Vec)>, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::List { resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn log_seq(&self) -> u64 { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + let _ = self.tx.send(StoreCmd::LogSeq { resp: resp_tx }).await; + resp_rx.await.unwrap_or(0) + } + + pub async fn applied_seq(&self) -> Result { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::AppliedSeq { resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn author_state(&self, author: &[u8; 32]) -> Result, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn sync_state(&self) -> Result { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::SyncState { resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn read_entries_after(&self, author: &[u8; 32], from_hash: Option<[u8; 32]>) -> Result, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::ReadEntriesAfter { author: *author, from_hash, resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn apply_entry(&self, entry: crate::proto::SignedEntry) -> Result<(), NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::ApplyEntry { entry, resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn put(&self, key: &[u8], value: &[u8]) -> Result { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Put { key: key.to_vec(), value: value.to_vec(), resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(|e| NodeError::Actor(e.to_string())) + } + + pub async fn delete(&self, key: &[u8]) -> Result { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Delete { key: key.to_vec(), resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(|e| NodeError::Actor(e.to_string())) + } + +} + +impl Drop for StoreHandle { + fn drop(&mut self) { + // Only send shutdown if we own the actor (non-cloned handle) + if let Some(handle) = self.actor_handle.take() { + let _ = self.tx.try_send(StoreCmd::Shutdown); + let _ = handle.join(); + } + // Clones (actor_handle = None) don't send shutdown - actor keeps running } } @@ -136,87 +404,183 @@ mod tests { use super::*; use std::env::temp_dir; - #[test] - fn test_generate() { - let node = Node::generate(); - let pk = node.public_key_bytes(); - assert_eq!(pk.len(), 32); + fn temp_data_dir(name: &str) -> DataDir { + let path = temp_dir().join(format!("lattice_node_test_{}", name)); + let _ = std::fs::remove_dir_all(&path); + DataDir::new(path) } - #[test] - fn test_sign_and_verify() { - let node = Node::generate(); - let message = b"hello lattice"; + #[tokio::test] + async fn test_create_and_open_store() { + let data_dir = temp_data_dir("meta_store"); - let signature = node.sign(message); - assert!(node.verify(message, &signature).is_ok()); + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to create node"); + + assert!(node.info().stores.is_empty()); + + let store_id = node.create_store().expect("Failed to create store"); + + // Verify it's in the list + let stores = node.list_stores().expect("list failed"); + assert!(stores.contains(&store_id)); + + let (handle, _) = node.open_store(store_id).expect("Failed to open store"); + handle.put(b"/key", b"value").await.expect("put failed"); + assert_eq!(handle.get(b"/key").await.unwrap(), Some(b"value".to_vec())); + + let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_verify_wrong_message() { - let node = Node::generate(); - let signature = node.sign(b"original"); + #[tokio::test] + async fn test_store_isolation() { + let data_dir = temp_data_dir("meta_isolation"); - assert!(node.verify(b"tampered", &signature).is_err()); + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("Failed to create node"); + + let store_a = node.create_store().expect("create A"); + let store_b = node.create_store().expect("create B"); + + let (handle_a, _) = node.open_store(store_a).expect("open A"); + handle_a.put(b"/key", b"from A").await.expect("put A"); + + let (handle_b, _) = node.open_store(store_b).expect("open B"); + assert_eq!(handle_b.get(b"/key").await.unwrap(), None); + + assert_eq!(handle_a.get(b"/key").await.unwrap(), Some(b"from A".to_vec())); + + let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_verify_with_different_key() { - let node1 = Node::generate(); - let node2 = Node::generate(); + #[tokio::test] + async fn test_init_creates_root_store() { + let data_dir = temp_data_dir("init_root"); - let signature = node1.sign(b"message"); - assert!(node2.verify(b"message", &signature).is_err()); + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + + // Initially no root store + assert!(node.root_store().unwrap().is_none()); + + // Init creates root store + let (root_id, _handle) = node.init().await.expect("init failed"); + assert_eq!(node.root_store().unwrap(), Some(root_id)); + + let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_save_and_load() { - let temp_path = temp_dir().join("lattice_test_identity.key"); + #[tokio::test] + async fn test_duplicate_init_fails() { + let data_dir = temp_data_dir("init_dup"); - // Generate and save - let node1 = Node::generate(); - let pk1 = node1.public_key_bytes(); - node1.save(&temp_path).unwrap(); + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); - // Load and verify same key - let node2 = Node::load(&temp_path).unwrap(); - let pk2 = node2.public_key_bytes(); + node.init().await.expect("first init"); - assert_eq!(pk1, pk2); + // Second init should fail + match node.init().await { + Ok(_) => panic!("Expected AlreadyInitialized error"), + Err(e) => match e { + NodeError::AlreadyInitialized => (), + _ => panic!("Expected AlreadyInitialized, got {:?}", e), + }, + } - // Cleanup - fs::remove_file(&temp_path).ok(); + let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_load_or_generate() { - let temp_path = temp_dir().join("lattice_test_identity2.key"); + #[tokio::test] + async fn test_root_store_in_info_after_init() { + let data_dir = temp_data_dir("init_info"); - // Remove if exists - fs::remove_file(&temp_path).ok(); + // First session: init + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + let (root_id, _) = node.init().await.expect("init"); + drop(node); // End first session - // First call: generates - let node1 = Node::load_or_generate(&temp_path).unwrap(); - let pk1 = node1.public_key_bytes(); + // Second session: root_store should persist + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("reload node"); - // Second call: loads existing - let node2 = Node::load_or_generate(&temp_path).unwrap(); - let pk2 = node2.public_key_bytes(); + assert_eq!(node.root_store().unwrap(), Some(root_id)); - assert_eq!(pk1, pk2); - - // Cleanup - fs::remove_file(&temp_path).ok(); + let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_verify_with_key_static() { - let node = Node::generate(); - let pk = node.public_key(); - let message = b"test message"; + #[tokio::test] + async fn test_idempotent_put_and_delete() { + let data_dir = temp_data_dir("idempotent"); - let signature = node.sign(message); + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + let (_, store) = node.init().await.expect("init"); - assert!(Node::verify_with_key(&pk, message, &signature).is_ok()); + // Get baseline seq after init + let baseline = store.log_seq().await; + + // Put twice with same value - second should be idempotent + let seq1 = store.put(b"/key", b"value").await.expect("put 1"); + assert_eq!(seq1, baseline + 1); + + let seq2 = store.put(b"/key", b"value").await.expect("put 2"); + assert_eq!(seq2, baseline + 1, "Second put should be idempotent (no new entry)"); + + assert_eq!(store.log_seq().await, baseline + 1); + + // Delete twice - second should be idempotent + let seq3 = store.delete(b"/key").await.expect("delete 1"); + assert_eq!(seq3, baseline + 2); + + let seq4 = store.delete(b"/key").await.expect("delete 2"); + assert_eq!(seq4, baseline + 2, "Second delete should be idempotent (no new entry)"); + + assert_eq!(store.log_seq().await, baseline + 2); + + let _ = std::fs::remove_dir_all(data_dir.base()); + } + + #[tokio::test] + async fn test_set_name_updates_store() { + let data_dir = temp_data_dir("set_name"); + + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + + // Set initial name + assert!(node.name().is_some()); + let initial_name = node.name().unwrap(); + + // Init creates root store + let (_, store) = node.init().await.expect("init"); + + // Verify initial name is in store + let pubkey_hex = hex::encode(node.node_id()); + let name_key = format!("/nodes/{}/name", pubkey_hex); + let stored_name = store.get(name_key.as_bytes()).await.unwrap(); + assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec())); + + // Change name + let new_name = "my-custom-name"; + node.set_name(new_name, Some(&store)).await.expect("set_name"); + + // Verify meta.db updated + assert_eq!(node.name(), Some(new_name.to_string())); + + // Verify store updated + let stored_name = store.get(name_key.as_bytes()).await.unwrap(); + assert_eq!(stored_name, Some(new_name.as_bytes().to_vec())); + + let _ = std::fs::remove_dir_all(data_dir.base()); } } diff --git a/lattice-core/src/node_identity.rs b/lattice-core/src/node_identity.rs new file mode 100644 index 0000000..985c396 --- /dev/null +++ b/lattice-core/src/node_identity.rs @@ -0,0 +1,252 @@ +//! Node identity and cryptographic keys +//! +//! Each node has an Ed25519 keypair: +//! - Private key: stored locally in `identity.key` (never replicated) +//! - Public key: serves as the node's identity (32 bytes) + +use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey}; +use rand::rngs::OsRng; +use std::fs; +use std::io::{self, Read, Write}; +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] io::Error), + + #[error("Invalid key length: expected 32 bytes, got {0}")] + InvalidKeyLength(usize), + + #[error("Invalid signature")] + InvalidSignature, +} + +/// A node in the Lattice mesh. +/// +/// Each node has an Ed25519 keypair used for signing sigchain entries +/// and establishing trust within the network. +#[derive(Clone)] +pub struct NodeIdentity { + signing_key: SigningKey, +} + +impl NodeIdentity { + /// Generate a new node with a random keypair. + pub fn generate() -> Self { + let signing_key = SigningKey::generate(&mut OsRng); + Self { signing_key } + } + + /// Create a node from an existing signing key. + pub fn from_signing_key(signing_key: SigningKey) -> Self { + Self { signing_key } + } + + /// Load a node's identity from a key file, or generate and save if it doesn't exist. + pub fn load_or_generate(path: impl AsRef) -> Result { + let path = path.as_ref(); + if path.exists() { + Self::load(path) + } else { + let node = Self::generate(); + node.save(path)?; + Ok(node) + } + } + + /// Load a node's identity from a key file. + pub fn load(path: impl AsRef) -> Result { + let mut file = fs::File::open(path)?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes)?; + + if bytes.len() != 32 { + return Err(NodeError::InvalidKeyLength(bytes.len())); + } + + let key_bytes: [u8; 32] = bytes.try_into().unwrap(); + let signing_key = SigningKey::from_bytes(&key_bytes); + Ok(Self { signing_key }) + } + + /// Save the node's private key to a file. + pub fn save(&self, path: impl AsRef) -> Result<(), NodeError> { + let path = path.as_ref(); + + // Create parent directories if they don't exist + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + + let mut file = fs::File::create(path)?; + file.write_all(self.signing_key.as_bytes())?; + Ok(()) + } + + /// Get the node's public key (identity). + pub fn public_key(&self) -> VerifyingKey { + self.signing_key.verifying_key() + } + + /// Get the node's public key as bytes (32 bytes). + pub fn public_key_bytes(&self) -> [u8; 32] { + self.signing_key.verifying_key().to_bytes() + } + + /// Get the signing key for creating signatures. + pub fn signing_key(&self) -> &SigningKey { + &self.signing_key + } + + /// Get the secret key bytes (32 bytes) for Iroh integration. + /// WARNING: Handle with care - this exposes the private key material. + pub fn secret_key_bytes(&self) -> [u8; 32] { + self.signing_key.to_bytes() + } + + /// Sign a message. + pub fn sign(&self, message: &[u8]) -> Signature { + self.signing_key.sign(message) + } + + /// Verify a signature against this node's public key. + pub fn verify(&self, message: &[u8], signature: &Signature) -> Result<(), NodeError> { + self.public_key() + .verify(message, signature) + .map_err(|_| NodeError::InvalidSignature) + } + + /// Verify a signature using a raw public key. + pub fn verify_with_key( + public_key: &VerifyingKey, + message: &[u8], + signature: &Signature, + ) -> Result<(), NodeError> { + public_key + .verify(message, signature) + .map_err(|_| NodeError::InvalidSignature) + } +} + +/// Peer status values used across the system +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum PeerStatus { + /// Peer has been invited but hasn't joined yet + Invited, + /// Peer is active and can sync + Active, + /// Peer has been removed from the mesh + Removed, +} + +impl PeerStatus { + pub fn as_str(&self) -> &'static str { + match self { + PeerStatus::Invited => "invited", + PeerStatus::Active => "active", + PeerStatus::Removed => "removed", + } + } + + pub fn from_str(s: &str) -> Option { + match s { + "invited" => Some(PeerStatus::Invited), + "active" => Some(PeerStatus::Active), + "removed" => Some(PeerStatus::Removed), + _ => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::env::temp_dir; + + #[test] + fn test_generate() { + let node = NodeIdentity::generate(); + let pk = node.public_key_bytes(); + assert_eq!(pk.len(), 32); + } + + #[test] + fn test_sign_and_verify() { + let node = NodeIdentity::generate(); + let message = b"hello lattice"; + + let signature = node.sign(message); + assert!(node.verify(message, &signature).is_ok()); + } + + #[test] + fn test_verify_wrong_message() { + let node = NodeIdentity::generate(); + let signature = node.sign(b"original"); + + assert!(node.verify(b"tampered", &signature).is_err()); + } + + #[test] + fn test_verify_with_different_key() { + let node1 = NodeIdentity::generate(); + let node2 = NodeIdentity::generate(); + + let signature = node1.sign(b"message"); + assert!(node2.verify(b"message", &signature).is_err()); + } + + #[test] + fn test_save_and_load() { + let temp_path = temp_dir().join("lattice_test_identity.key"); + + // Generate and save + let node1 = NodeIdentity::generate(); + let pk1 = node1.public_key_bytes(); + node1.save(&temp_path).unwrap(); + + // Load and verify same key + let node2 = NodeIdentity::load(&temp_path).unwrap(); + let pk2 = node2.public_key_bytes(); + + assert_eq!(pk1, pk2); + + // Cleanup + fs::remove_file(&temp_path).ok(); + } + + #[test] + fn test_load_or_generate() { + let temp_path = temp_dir().join("lattice_test_identity2.key"); + + // Remove if exists + fs::remove_file(&temp_path).ok(); + + // First call: generates + let node1 = NodeIdentity::load_or_generate(&temp_path).unwrap(); + let pk1 = node1.public_key_bytes(); + + // Second call: loads existing + let node2 = NodeIdentity::load_or_generate(&temp_path).unwrap(); + let pk2 = node2.public_key_bytes(); + + assert_eq!(pk1, pk2); + + // Cleanup + fs::remove_file(&temp_path).ok(); + } + + #[test] + fn test_verify_with_key_static() { + let node = NodeIdentity::generate(); + let pk = node.public_key(); + let message = b"test message"; + + let signature = node.sign(message); + + assert!(NodeIdentity::verify_with_key(&pk, message, &signature).is_ok()); + } +} diff --git a/lattice-core/src/sigchain.rs b/lattice-core/src/sigchain.rs index 2b56155..a2602d1 100644 --- a/lattice-core/src/sigchain.rs +++ b/lattice-core/src/sigchain.rs @@ -4,7 +4,7 @@ //! before appending (correct seq, prev_hash, valid signature) and persists to disk. use crate::log::{append_entry, read_entries, LogError}; -use crate::node::Node; +use crate::node_identity::NodeIdentity; use crate::proto::{Entry, SignedEntry}; use crate::signed_entry::{hash_signed_entry, verify_signed_entry}; use prost::Message; @@ -229,7 +229,7 @@ impl SigChain { } /// Create and append a new entry using the node's key - pub fn create_entry(&mut self, node: &Node, ops: Vec) -> Result { + pub fn create_entry(&mut self, node: &NodeIdentity, ops: Vec) -> Result { use crate::clock::SystemClock; use crate::hlc::HLC; use crate::signed_entry::EntryBuilder; @@ -322,7 +322,7 @@ mod tests { use super::*; use crate::clock::MockClock; use crate::hlc::HLC; - use crate::node::Node; + use crate::node_identity::NodeIdentity; use crate::proto::{operation, Operation, PutOp}; use crate::signed_entry::EntryBuilder; use std::env::temp_dir; @@ -352,7 +352,7 @@ mod tests { let path = temp_log_path("append"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut chain = SigChain::new(&path, TEST_STORE, author); @@ -377,7 +377,7 @@ mod tests { let path = temp_log_path("multiple"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut chain = SigChain::new(&path, TEST_STORE, author); let clock = MockClock::new(1000); @@ -402,7 +402,7 @@ mod tests { let path = temp_log_path("from_log"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let clock = MockClock::new(1000); @@ -433,7 +433,7 @@ mod tests { let path = temp_log_path("wrong_seq"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut chain = SigChain::new(&path, TEST_STORE, author); let clock = MockClock::new(1000); @@ -457,7 +457,7 @@ mod tests { let path = temp_log_path("wrong_prev"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut chain = SigChain::new(&path, TEST_STORE, author); let clock = MockClock::new(1000); @@ -489,7 +489,7 @@ mod tests { let path = temp_log_path("wrong_author"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let other_author = [99u8; 32]; // Different author let mut chain = SigChain::new(&path, TEST_STORE, other_author); let clock = MockClock::new(1000); @@ -513,7 +513,7 @@ mod tests { let path = temp_log_path("create"); std::fs::remove_file(&path).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut chain = SigChain::new(&path, TEST_STORE, author); @@ -545,7 +545,7 @@ mod tests { std::fs::remove_file(&path_a).ok(); std::fs::remove_file(&path_b).ok(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let clock = MockClock::new(1000); diff --git a/lattice-core/src/signed_entry.rs b/lattice-core/src/signed_entry.rs index eff9b08..1e466bf 100644 --- a/lattice-core/src/signed_entry.rs +++ b/lattice-core/src/signed_entry.rs @@ -7,7 +7,7 @@ //! - Computing entry hashes for prev_hash linking use crate::hlc::HLC; -use crate::node::{Node, NodeError}; +use crate::node_identity::{NodeIdentity, NodeError}; use crate::proto::{Entry, Hlc, Operation, PutOp, DeleteOp, SignedEntry, operation}; use ed25519_dalek::{Signature, VerifyingKey}; use prost::Message; @@ -116,14 +116,14 @@ impl EntryBuilder { } /// Build and sign the entry, returning a SignedEntry - pub fn sign(self, node: &Node) -> SignedEntry { + pub fn sign(self, node: &NodeIdentity) -> SignedEntry { let entry = self.build(); sign_entry(&entry, node) } } /// Sign an Entry to create a SignedEntry -pub fn sign_entry(entry: &Entry, node: &Node) -> SignedEntry { +pub fn sign_entry(entry: &Entry, node: &NodeIdentity) -> SignedEntry { let entry_bytes = entry.encode_to_vec(); let signature = node.sign(&entry_bytes); @@ -152,7 +152,7 @@ pub fn verify_signed_entry(signed: &SignedEntry) -> Result { let signature = Signature::from_bytes(&sig_bytes); // Verify - Node::verify_with_key(&public_key, &signed.entry_bytes, &signature)?; + NodeIdentity::verify_with_key(&public_key, &signed.entry_bytes, &signature)?; // Decode entry let entry = Entry::decode(&signed.entry_bytes[..])?; @@ -192,7 +192,7 @@ mod tests { #[test] fn test_sign_and_verify() { - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let hlc = HLC::now_with_clock(&clock); @@ -211,7 +211,7 @@ mod tests { #[test] fn test_verify_tampered_fails() { - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let hlc = HLC::now_with_clock(&clock); @@ -227,8 +227,8 @@ mod tests { #[test] fn test_verify_wrong_key_fails() { - let node1 = Node::generate(); - let node2 = Node::generate(); + let node1 = NodeIdentity::generate(); + let node2 = NodeIdentity::generate(); let clock = MockClock::new(1000); let hlc = HLC::now_with_clock(&clock); @@ -244,7 +244,7 @@ mod tests { #[test] fn test_hash_signed_entry() { - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let hlc = HLC::now_with_clock(&clock); @@ -262,7 +262,7 @@ mod tests { #[test] fn test_prev_hash_chaining() { - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); // First entry diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index 87d399f..7175b9c 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -324,7 +324,7 @@ mod tests { use super::*; use crate::clock::MockClock; use crate::hlc::HLC; - use crate::node::Node; + use crate::node_identity::NodeIdentity; use crate::signed_entry::EntryBuilder; use std::env::temp_dir; @@ -341,7 +341,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock)) @@ -391,7 +391,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock = MockClock::new(1000); // First write @@ -426,7 +426,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); // Create two heads let clock1 = MockClock::new(1000); @@ -473,7 +473,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); // Create two concurrent heads let clock1 = MockClock::new(1000); @@ -525,7 +525,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); // Create a single head let clock1 = MockClock::new(1000); @@ -573,8 +573,8 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let alice = Node::generate(); - let bob = Node::generate(); + let alice = NodeIdentity::generate(); + let bob = NodeIdentity::generate(); // Initial state: K = v1 let clock1 = MockClock::new(1000); @@ -632,9 +632,9 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let alice = Node::generate(); - let bob = Node::generate(); - let charlie = Node::generate(); + let alice = NodeIdentity::generate(); + let bob = NodeIdentity::generate(); + let charlie = NodeIdentity::generate(); // Alice creates K = v1 let clock1 = MockClock::new(1000); @@ -692,7 +692,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let clock1 = MockClock::new(1000); let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) @@ -725,7 +725,7 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); // First write: a = 1 let clock1 = MockClock::new(1000); @@ -788,7 +788,7 @@ mod tests { let _ = std::fs::remove_file(&log_path); let store = Store::open(&state_path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); // First write: a = 1 @@ -850,7 +850,7 @@ mod tests { let _ = std::fs::remove_file(&log_path); let store = Store::open(&state_path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); @@ -898,7 +898,7 @@ mod tests { let _ = std::fs::remove_file(&log_path); let store = Store::open(&state_path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); @@ -957,7 +957,7 @@ mod tests { let _ = std::fs::remove_file(&log_path); let store = Store::open(&state_path).unwrap(); - let node = Node::generate(); + let node = NodeIdentity::generate(); let author = node.public_key_bytes(); let mut sigchain = SigChain::new(&log_path, TEST_STORE, node.public_key_bytes()); @@ -1125,7 +1125,7 @@ mod tests { // Node A writes some entries let store_a = Store::open(&path_a).unwrap(); - let node_a = Node::generate(); + let node_a = NodeIdentity::generate(); // Write 3 entries on node A for i in 1u64..=3 { @@ -1196,8 +1196,8 @@ mod tests { let store_a = Store::open(&path_a).unwrap(); let store_b = Store::open(&path_b).unwrap(); - let node_a = Node::generate(); - let node_b = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); // Node A writes entries for i in 1u64..=2 { @@ -1283,9 +1283,9 @@ mod tests { let store_a = Store::open(&path_a).unwrap(); let store_b = Store::open(&path_b).unwrap(); let store_c = Store::open(&path_c).unwrap(); - let node_a = Node::generate(); - let node_b = Node::generate(); - let node_c = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); + let node_c = NodeIdentity::generate(); // Each node writes one entry let entry_a = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(1000))) @@ -1369,8 +1369,8 @@ mod tests { let store_a = Store::open(&path_a).unwrap(); let store_b = Store::open(&path_b).unwrap(); - let node_a = Node::generate(); - let node_b = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); // Both nodes write to the SAME key with different values // Use same HLC to force conflict (tie-break on author) @@ -1435,8 +1435,8 @@ mod tests { let _ = std::fs::remove_file(&path); let store = Store::open(&path).unwrap(); - let node_low = Node::generate(); - let node_high = Node::generate(); + let node_low = NodeIdentity::generate(); + let node_high = NodeIdentity::generate(); // Determine which node has "higher" author bytes let (high_node, low_node) = if node_high.public_key_bytes() > node_low.public_key_bytes() { @@ -1494,9 +1494,9 @@ mod tests { let store_d = Store::open(&path_d).unwrap(); // Create 3 nodes (virtual peers) - let node_a = Node::generate(); - let node_b = Node::generate(); - let node_c = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); + let node_c = NodeIdentity::generate(); let clock = MockClock::new(1000); @@ -1609,9 +1609,9 @@ mod tests { let store = Store::open(&path).unwrap(); // Create 3 nodes - let node_a = Node::generate(); - let node_b = Node::generate(); - let node_c = Node::generate(); + let node_a = NodeIdentity::generate(); + let node_b = NodeIdentity::generate(); + let node_c = NodeIdentity::generate(); let clock = MockClock::new(1000); diff --git a/lattice-cli/src/store_actor.rs b/lattice-core/src/store_actor.rs similarity index 94% rename from lattice-cli/src/store_actor.rs rename to lattice-core/src/store_actor.rs index 041fe68..8e3c06e 100644 --- a/lattice-cli/src/store_actor.rs +++ b/lattice-core/src/store_actor.rs @@ -1,11 +1,14 @@ //! Store Actor - dedicated thread that owns Store and processes commands via channel -use lattice_core::{ - EntryBuilder, HeadInfo, Node, SigChain, SigChainManager, Store, Uuid, +use crate::{ + EntryBuilder, HeadInfo, NodeIdentity, SigChain, SigChainManager, Store, Uuid, hlc::HLC, proto::AuthorState, sigchain::SigChainError, store::StoreError, + sync_state::SyncState, + proto::SignedEntry, + log, }; use tokio::sync::{mpsc, oneshot}; use std::thread::{self, JoinHandle}; @@ -44,15 +47,15 @@ pub enum StoreCmd { }, // Sync-related commands SyncState { - resp: oneshot::Sender>, + resp: oneshot::Sender>, }, ReadEntriesAfter { author: [u8; 32], from_hash: Option<[u8; 32]>, - resp: oneshot::Sender, StoreError>>, + resp: oneshot::Sender, StoreError>>, }, ApplyEntry { - entry: lattice_core::proto::SignedEntry, + entry: SignedEntry, resp: oneshot::Sender>, }, Shutdown, @@ -92,7 +95,7 @@ pub struct StoreActor { store_id: Uuid, store: Store, chain_manager: SigChainManager, // Manages all authors' sigchains - node: Node, + node: NodeIdentity, rx: mpsc::Receiver, } @@ -102,7 +105,7 @@ impl StoreActor { store_id: Uuid, store: Store, sigchain: SigChain, - node: Node, + node: NodeIdentity, rx: mpsc::Receiver, ) -> Self { // Derive logs_dir from sigchain's log file path @@ -243,7 +246,7 @@ impl StoreActor { &self, author: &[u8; 32], from_hash: Option<[u8; 32]>, - ) -> Result, StoreError> { + ) -> Result, StoreError> { // Build log path for this author let author_hex = hex::encode(author); let log_path = self.chain_manager.logs_dir().join(format!("{}.log", author_hex)); @@ -253,7 +256,7 @@ impl StoreActor { } // Use lattice_core's read_entries_after - lattice_core::log::read_entries_after(&log_path, from_hash) + log::read_entries_after(&log_path, from_hash) .map_err(StoreError::from) } } @@ -264,7 +267,7 @@ pub fn spawn_store_actor( store_id: Uuid, store: Store, sigchain: SigChain, - node: Node, + node: NodeIdentity, ) -> (mpsc::Sender, JoinHandle<()>) { let (tx, rx) = mpsc::channel(32); let actor = StoreActor::new(store_id, store, sigchain, node, rx); diff --git a/lattice-net/Cargo.toml b/lattice-net/Cargo.toml index f605899..50045f5 100644 --- a/lattice-net/Cargo.toml +++ b/lattice-net/Cargo.toml @@ -16,6 +16,7 @@ tracing = { workspace = true } bytes = { workspace = true } tokio-util = { workspace = true } futures-util = { workspace = true } +hex = { workspace = true } [dev-dependencies] tokio-test = { workspace = true } diff --git a/lattice-net/src/lib.rs b/lattice-net/src/lib.rs index 9d85534..6202b96 100644 --- a/lattice-net/src/lib.rs +++ b/lattice-net/src/lib.rs @@ -5,15 +5,17 @@ //! - **Gossip**: Broadcasting changes across the mesh //! - **Unicast**: Point-to-point communication for reconciliation //! - **Framing**: Length-delimited message framing for QUIC streams +//! - **Mesh**: Peer-to-peer join and sync operations pub mod endpoint; pub mod gossip; -pub mod unicast; pub mod framing; +pub mod mesh; pub use endpoint::{LatticeEndpoint, PublicKey}; pub use framing::{MessageSink, MessageStream}; pub use lattice_core::proto::{SyncRequest, SyncResponse, SyncEntry, SyncDone, SyncState, Frontier}; +pub use mesh::{spawn_accept_loop, join_mesh, sync_with_peer, sync_all, SyncResult}; /// Parse a PublicKey (NodeId) from hex or base32 string pub fn parse_node_id(s: &str) -> Result { diff --git a/lattice-net/src/mesh/mod.rs b/lattice-net/src/mesh/mod.rs new file mode 100644 index 0000000..341136d --- /dev/null +++ b/lattice-net/src/mesh/mod.rs @@ -0,0 +1,13 @@ +//! Mesh networking - peer-to-peer join and sync operations +//! +//! - **server**: Accept incoming connections and handle join/sync requests +//! - **sync**: Outgoing join and sync operations +//! - **protocol**: Shared send/receive entry logic + +mod server; +mod sync; +mod protocol; + +pub use server::spawn_accept_loop; +pub use sync::{join_mesh, sync_with_peer, sync_all, SyncResult}; +pub use protocol::{send_missing_entries, receive_entries}; diff --git a/lattice-cli/src/sync_protocol.rs b/lattice-net/src/mesh/protocol.rs similarity index 88% rename from lattice-cli/src/sync_protocol.rs rename to lattice-net/src/mesh/protocol.rs index daa950b..96e8632 100644 --- a/lattice-cli/src/sync_protocol.rs +++ b/lattice-net/src/mesh/protocol.rs @@ -1,12 +1,9 @@ -//! Sync Protocol - shared logic for bidirectional sync -//! -//! Provides reusable functions for sending and receiving entries during sync. -//! Used by both accept_handler (incoming sync) and sync (outgoing sync). +//! Protocol - shared logic for bidirectional sync entry exchange -use crate::node::StoreHandle; +use crate::{MessageSink, MessageStream}; +use lattice_core::{StoreHandle, CausalEntryIter}; use lattice_core::proto::{peer_message, PeerMessage, SignedEntry}; use lattice_core::sync_state::SyncState; -use lattice_net::{MessageSink, MessageStream}; use prost::Message; use std::collections::VecDeque; @@ -33,7 +30,7 @@ pub async fn send_missing_entries( // Stream entries in HLC (causal) order let mut entries_sent = 0u64; - for entry in lattice_core::CausalEntryIter::new(author_entries) { + for entry in CausalEntryIter::new(author_entries) { let sync_msg = PeerMessage { message: Some(peer_message::Message::SyncEntry(lattice_core::proto::SyncEntry { signed_entry: entry.encode_to_vec(), diff --git a/lattice-cli/src/accept_handler.rs b/lattice-net/src/mesh/server.rs similarity index 94% rename from lattice-cli/src/accept_handler.rs rename to lattice-net/src/mesh/server.rs index fb65ca8..4436272 100644 --- a/lattice-cli/src/accept_handler.rs +++ b/lattice-net/src/mesh/server.rs @@ -1,12 +1,13 @@ -//! Accept handler for incoming Iroh connections +//! Server - handle incoming peer connections for join and sync -use lattice_net::{MessageSink, MessageStream}; -use crate::node::{StoreHandle, PeerStatus}; +use crate::{MessageSink, MessageStream}; +use lattice_core::{StoreHandle, PeerStatus}; use iroh::Endpoint; use iroh::endpoint::Connection; use std::sync::Arc; use tokio::sync::RwLock; use lattice_core::proto::{PeerMessage, peer_message, JoinResponse}; +use super::protocol; /// Spawn the accept loop for incoming connections. pub fn spawn_accept_loop( @@ -151,11 +152,11 @@ async fn handle_sync_request( .map(|s| lattice_core::sync_state::SyncState::from_proto(&s)) .unwrap_or_default(); - let entries_sent = crate::sync_protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await?; + let entries_sent = protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await?; println!("[Sync] Sent {} entries, now receiving from peer...", entries_sent); // 3. Receive entries from requester (bidirectional) - let (entries_applied, _) = crate::sync_protocol::receive_entries(&mut stream, store).await?; + let (entries_applied, _) = protocol::receive_entries(&mut stream, store).await?; sink.finish().await?; diff --git a/lattice-cli/src/sync.rs b/lattice-net/src/mesh/sync.rs similarity index 83% rename from lattice-cli/src/sync.rs rename to lattice-net/src/mesh/sync.rs index e33bed8..3bea46d 100644 --- a/lattice-cli/src/sync.rs +++ b/lattice-net/src/mesh/sync.rs @@ -1,12 +1,10 @@ -//! Sync networking operations for LatticeNode -//! -//! Provides async methods for joining meshes and syncing with peers. +//! Sync - outgoing mesh join and sync operations -use lattice_net::{MessageSink, MessageStream}; -use crate::node::{LatticeNode, NodeError, StoreHandle, PeerStatus}; +use crate::{MessageSink, MessageStream, LatticeEndpoint, parse_node_id}; +use lattice_core::{Node, NodeError, StoreHandle, PeerStatus}; use lattice_core::proto::{peer_message, PeerMessage, JoinRequest, SignedEntry}; -use lattice_net::LatticeEndpoint; use prost::Message; +use super::protocol; /// Result of a sync operation with a peer pub struct SyncResult { @@ -18,7 +16,7 @@ pub struct SyncResult { /// Returns the new StoreHandle on success. /// After joining, automatically syncs with the peer to get initial data. pub async fn join_mesh( - node: &LatticeNode, + node: &Node, endpoint: &LatticeEndpoint, peer_id: iroh::PublicKey, ) -> Result { @@ -72,6 +70,14 @@ pub async fn join_mesh( } } + // Write our name to the store (separate key, not JSON) + // Note: inviter sets our status to 'active' via server.rs + let pubkey_hex = hex::encode(node.node_id()); + if let Some(name) = node.name() { + let name_key = format!("/nodes/{}/name", pubkey_hex); + let _ = handle.put(name_key.as_bytes(), name.as_bytes()).await; + } + Ok(handle) } _ => Err(NodeError::Actor("Unexpected response message type".to_string())), @@ -81,7 +87,7 @@ pub async fn join_mesh( /// Sync with a specific peer (bidirectional). /// Both sides exchange states and send missing entries to each other. pub async fn sync_with_peer( - node: &LatticeNode, + node: &Node, endpoint: &LatticeEndpoint, store: &StoreHandle, peer_id: iroh::PublicKey, @@ -144,30 +150,12 @@ pub async fn sync_with_peer( } // 3. Send entries peer is missing (using shared protocol) - let entries_sent = crate::sync_protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await + let entries_sent = protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await .map_err(|e| NodeError::Actor(e))?; sink.finish().await .map_err(|e| NodeError::Actor(format!("Failed to finish: {}", e)))?; - // Update own node info if we applied entries - if entries_applied > 0 { - let pubkey_hex = hex::encode(node.node_id()); - let info_key = format!("/nodes/{}/info", pubkey_hex); - let info_val = serde_json::json!({ - "name": hostname::get().map(|h| h.to_string_lossy().to_string()).unwrap_or_default(), - "added_at": std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() - }).to_string(); - let _ = store.put(info_key.as_bytes(), info_val.as_bytes()).await; - - // Set own status to 'active' (we're now a fully synced peer) - let status_key = format!("/nodes/{}/status", pubkey_hex); - let _ = store.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await; - } - println!("[Sync] Applied {} entries, sent {} entries", entries_applied, entries_sent); Ok(SyncResult { @@ -178,7 +166,7 @@ pub async fn sync_with_peer( /// Sync with all active peers from the store. pub async fn sync_all( - node: &LatticeNode, + node: &Node, endpoint: &LatticeEndpoint, store: &StoreHandle, ) -> Result, NodeError> { @@ -193,7 +181,7 @@ pub async fn sync_all( if key_str.ends_with("/status") && value == PeerStatus::Active.as_str().as_bytes() { if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) { if pubkey != my_pubkey { - if let Ok(id) = lattice_net::parse_node_id(pubkey) { + if let Ok(id) = parse_node_id(pubkey) { peer_ids.push(id); } } diff --git a/lattice-net/src/unicast.rs b/lattice-net/src/unicast.rs deleted file mode 100644 index 39b6da2..0000000 --- a/lattice-net/src/unicast.rs +++ /dev/null @@ -1,3 +0,0 @@ -//! Unicast communication for direct peer-to-peer messaging - -// TODO: Implement unicast using iroh