From 9d4495b3d77f6788752b95e3700651b5be9ae284 Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Tue, 23 Dec 2025 00:31:54 +0100 Subject: [PATCH] feat: Refactor mesh server to use `Arc`, introduce `JoinAcceptance` for mesh joins, and enhance store listing with prefix filtering and deleted entry inclusion. --- docs/architecture.md | 10 +- docs/roadmap.md | 36 +++++-- lattice-cli/src/main.rs | 18 +--- lattice-cli/src/node_commands.rs | 6 +- lattice-cli/src/store_commands.rs | 21 +++- lattice-core/src/lib.rs | 2 +- lattice-core/src/node.rs | 164 +++++++++++++++++++++--------- lattice-core/src/node_identity.rs | 8 +- lattice-core/src/store.rs | 81 ++++++++++++++- lattice-core/src/store_actor.rs | 13 ++- lattice-net/src/mesh/server.rs | 129 +++++++++++------------ lattice-net/src/mesh/sync.rs | 51 +++------- proto/lattice.proto | 8 +- 13 files changed, 341 insertions(+), 206 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 3b427c5..b3be0b1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -59,14 +59,14 @@ Networking modes: - Identified by their Ed25519 public key. - Private key stored locally in `identity.key` (not replicated). - Node data stored in KV: - - `/nodes/{pubkey}/info` = static metadata (name, added_by, added_at) - - `/nodes/{pubkey}/status` = `active` | `dormant` | `disabled` + - `/nodes/{pubkey}/name` = display name + - `/nodes/{pubkey}/added_at` = timestamp when added + - `/nodes/{pubkey}/status` = `invited` | `active` | `dormant` (removal deletes keys) - `/nodes/{pubkey}/role` = `server` | `device` (optional, hints sync priority) - - `/nodes/{pubkey}/iroh` = Iroh NodeId for network connection - Peer invitation flow: 1. Inviter runs `invite ` → writes `/nodes/{peer}/info` + `/status` 2. Inviter shares their Iroh NodeId out-of-band (QR code, link, text) - 3. Invited peer runs `connect ` → syncs with inviter + 3. Invited peer runs `join ` → syncs with inviter 4. Sync pulls `/nodes/{self}/info` + `/status` → peer is authorized 5. `connect` implicitly adds inviter to peer's `/nodes/*` (mutual awareness) - Accepting = syncing. The invited peer discovers authorization by receiving the entries. @@ -181,7 +181,7 @@ Each node stores logs as one file per author: Table Key Value Purpose ───────────────────────────────────────────────────────────────────────────── kv Vec (key) Vec Current tips for each key -applied_frontiers [u8; 32] (author_id) (u64 seq, [u8; 32] hash) What's applied to this store +AUTHOR_TABLE [u8; 32] (author_id) (u64 seq, [u8; 32] hash) Per-author frontier tracking meta Vec Vec Store metadata (incl. merkle_root) ``` diff --git a/docs/roadmap.md b/docs/roadmap.md index 9255f7b..f8fd136 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -127,10 +127,7 @@ - [x] Entry ordering: Per-author streaming is correct (hash chain per author, HLC for cross-author). - [x] Multi-head sync fixed: SyncState now tracks HashSet of head hashes per author. - [x] Sync entry ordering: Entries sent in HLC order (merge-sort across authors) to ensure causal order. - -*Background Sync:* -- [ ] Periodic sync with known peers -- [ ] Track last sync time per peer +- [x] `join_mesh` doesn't populate `node.root_store`: Fixed with `complete_join` method. ### Success Criteria @@ -138,26 +135,43 @@ - Works offline-first (sync when connected) **Post-M2 Refactoring:** -- [ ] Unify `node.rs` from `lattice-cli` and `lattice-core` -- [ ] Use prost for node status in store +- [x] Unify `node.rs` from `lattice-cli` and `lattice-core` +- [x] Move network code to `lattice-net` --- ## Milestone 3: Multi-Node Mesh -**Goal:** N nodes form a gossip mesh with watermark consensus. +**Goal:** N nodes form a gossip mesh for real-time sync. ### Deliverables -- [ ] Gossip protocol -- [ ] Watermark tracking & log pruning -- [ ] Node invitation (sigchain membership) -- [ ] Conflict detection (LWW resolution) +**Phase 1: LatticeServer Refactor** +- [ ] `LatticeServer` struct in `lattice-net` wrapping `Arc` + `Endpoint` +- [ ] Move `join_mesh`, `sync_with_peer`, `sync_all` to `LatticeServer` methods +- [ ] Encapsulate `spawn_accept_loop` inside `LatticeServer` +- [ ] CLI uses `LatticeServer` instead of raw `Node` + `Endpoint` +- [ ] Route sync command through `LatticeServer` (not raw functions) +- [ ] Integration test: invite → join → sync end-to-end +- [ ] Periodic background sync with known peers +- [ ] Track last sync time per peer + +**Phase 2: Gossip Protocol** +- [ ] Proto: `GossipAnnounce` message with author + latest seq + HLC +- [ ] `LatticeServer::spawn_gossip_loop` for periodic announcements +- [ ] On receiving announce: detect missing entries, trigger sync +- [ ] Track last-seen per peer for staleness detection --- ## Future +- remove_peer should be a transactional operation on store +- Watermark tracking & log pruning + - Track minimum confirmed seq per author across all peers + - Log pruning: remove entries below watermark +- Multi-KV-Store sync +- Optimized sync on join. Only transfer current watermark state, then sync missing entries. This would allow pruning. Might need snapshot support in KV store. - Mobile (iOS/Android) clients - Key rotation - Secure storage (Keychain, TPM) diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index 6deb276..e858a73 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -10,7 +10,6 @@ use lattice_core::{NodeBuilder, StoreHandle}; use rustyline::error::ReadlineError; use rustyline::DefaultEditor; use std::sync::Arc; -use tokio::sync::RwLock; #[tokio::main] async fn main() { @@ -18,7 +17,7 @@ async fn main() { println!("Type 'help' for commands, 'quit' to exit.\n"); let node = match NodeBuilder::new().build() { - Ok(n) => n, + Ok(n) => Arc::new(n), Err(e) => { eprintln!("Failed to initialize: {}", e); return; @@ -37,12 +36,9 @@ async fn main() { } }; - // Shared store handle for accept loop (updated when store is opened/changed) - let shared_store: Arc>> = Arc::new(RwLock::new(None)); - // Spawn accept loop for incoming connections if let Some(ref ep) = endpoint { - spawn_accept_loop(ep.endpoint().clone(), shared_store.clone()); + spawn_accept_loop(node.clone(), ep.endpoint().clone()); } let info = node.info(); @@ -53,18 +49,14 @@ async fn main() { println!("Stores: {}", info.stores.len()); } - let mut current_store: Option = match node.open_root_store() { + let mut current_store: Option = match node.open_root_store().await { Ok(Some(open_info)) => { if open_info.entries_replayed > 0 { println!("Root: {} (replayed {})", open_info.store_id, open_info.entries_replayed); } else { println!("Root: {}", open_info.store_id); } - let h = node.root_store().as_ref().cloned(); - if let Some(ref handle) = h { - *shared_store.write().await = Some(handle.clone()); - } - h + node.root_store().await.as_ref().cloned() } Ok(None) => { println!("Status: Not initialized (use 'init')"); @@ -111,8 +103,6 @@ async fn main() { match (cmd.handler)(&node, current_store.as_ref(), endpoint.as_ref(), cmd_args) { CommandResult::Ok => {} CommandResult::SwitchTo(h) => { - // Update shared store for accept loop - *shared_store.write().await = Some(h.clone()); current_store = Some(h); } CommandResult::Quit => break, diff --git a/lattice-cli/src/node_commands.rs b/lattice-cli/src/node_commands.rs index c1470ae..f412186 100644 --- a/lattice-cli/src/node_commands.rs +++ b/lattice-cli/src/node_commands.rs @@ -31,7 +31,7 @@ fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Lattic Ok(store_id) => { println!("Initialized with root store: {}", store_id); println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id())); - match node.root_store().as_ref() { + match block_async(node.root_store()).as_ref() { Some(h) => CommandResult::SwitchTo(h.clone()), None => CommandResult::Ok, } @@ -179,8 +179,8 @@ fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Latti by_status.entry(peer.status).or_default().push(peer); } - // Print grouped by status in order: active, invited, removed - let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed]; + // Print grouped by status in order: active, invited, dormant + let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Dormant]; for status in &status_order { if let Some(peer_list) = by_status.get(status) { println!("\n[{}] ({}):", status.as_str(), peer_list.len()); diff --git a/lattice-cli/src/store_commands.rs b/lattice-cli/src/store_commands.rs index f46e39a..23830bc 100644 --- a/lattice-cli/src/store_commands.rs +++ b/lattice-cli/src/store_commands.rs @@ -11,7 +11,7 @@ pub fn store_commands() -> Vec { Command { name: "put", args: " ", desc: "Store a key-value pair", group: "store", min_args: 2, max_args: 2, handler: cmd_put as Handler }, Command { name: "get", args: " [-v]", desc: "Get value for key", group: "store", min_args: 1, max_args: 2, handler: cmd_get as Handler }, Command { name: "delete", args: "", desc: "Delete a key", group: "store", min_args: 1, max_args: 1, handler: cmd_delete as Handler }, - Command { name: "list", args: "[-v]", desc: "List all keys", group: "store", min_args: 0, max_args: 1, handler: cmd_list as Handler }, + Command { name: "list", args: "[prefix] [-v]", desc: "List keys (optionally filtered by prefix)", group: "store", min_args: 0, max_args: 2, handler: cmd_list as Handler }, Command { name: "author-state", args: "[pubkey]", desc: "Show author sync state", group: "store", min_args: 0, max_args: 1, handler: cmd_author_state as Handler }, ] } @@ -26,7 +26,7 @@ fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option println!("Log Seq: {}", block_async(h.log_seq())); println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0)); - let all = block_async(h.list()).unwrap_or_default(); + let all = block_async(h.list(false)).unwrap_or_default(); println!("Keys: {}", all.len()); // Show log directory size @@ -120,9 +120,19 @@ fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Lattic println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; }; - let verbose = args.first().map(|a| a == "-v").unwrap_or(false); + + // Parse args: [prefix] [-v] + let verbose = args.iter().any(|a| a == "-v"); + let prefix = args.iter().find(|a| *a != "-v").cloned(); + let start = Instant::now(); - match block_async(h.list()) { + let result = if let Some(p) = &prefix { + block_async(h.list_by_prefix(p.as_bytes(), verbose)) + } else { + block_async(h.list(verbose)) + }; + + match result { Ok(entries) => { if entries.is_empty() { println!("(empty)"); @@ -154,7 +164,8 @@ fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Lattic } } } - println!("({} keys, {:.2?})", entries.len(), start.elapsed()); + let prefix_str = prefix.as_ref().map(|p| format!(" (prefix: {})", p)).unwrap_or_default(); + println!("({} keys{}, {:.2?})", entries.len(), prefix_str, start.elapsed()); } } Err(e) => eprintln!("Error: {}", e), diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index 2bd2fb6..9953bb6 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -35,7 +35,7 @@ pub mod store_actor; pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024; pub use node_identity::{NodeIdentity, PeerStatus}; -pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, PeerInfo}; +pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, PeerInfo, JoinAcceptance}; pub use sigchain::{SigChain, SigChainManager}; pub use entry::Entry; pub use sync_state::{SyncState, AuthorInfo, MissingRange}; diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index 6bc8327..4297f2b 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -10,7 +10,6 @@ use crate::{ node_identity::NodeError as IdentityError, }; use std::path::Path; -use std::rc::Rc; use thiserror::Error; #[derive(Error, Debug)] @@ -54,6 +53,11 @@ pub struct StoreInfo { pub entries_replayed: u64, } +/// Result of accepting a peer's join request +pub struct JoinAcceptance { + pub store_id: Uuid, +} + /// Information about a peer in the mesh pub struct PeerInfo { pub pubkey: String, @@ -97,9 +101,9 @@ impl NodeBuilder { Ok(Node { data_dir: self.data_dir, - node: Rc::new(node), + node: std::sync::Arc::new(node), meta, - root_store: std::cell::RefCell::new(None), + root_store: tokio::sync::RwLock::new(None), }) } } @@ -111,9 +115,9 @@ impl Default for NodeBuilder { /// A local Lattice node (manages identity and store registry) pub struct Node { data_dir: DataDir, - node: Rc, + node: std::sync::Arc, meta: MetaStore, - root_store: std::cell::RefCell>, + root_store: tokio::sync::RwLock>, } impl Node { @@ -146,16 +150,21 @@ impl Node { /// Set the node's display name. /// Updates meta.db and if root store is open, also updates /nodes/{pubkey}/name pub async fn set_name(&self, name: &str) -> Result<(), NodeError> { - // Update meta.db self.meta.set_name(name)?; - - // If root store is open, update there too - if let Some(handle) = self.root_store.borrow().as_ref() { - 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?; + self.publish_name().await + } + + /// Publish this node's name from meta.db to the root store. + /// Used after joining a mesh to announce ourselves. + pub async fn publish_name(&self) -> Result<(), NodeError> { + if let Some(name) = self.name() { + let guard = self.root_store.read().await; + if let Some(handle) = guard.as_ref() { + 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?; + } } - Ok(()) } @@ -165,17 +174,17 @@ impl Node { } /// Get reference to the cached root store handle (if open) - pub fn root_store(&self) -> std::cell::Ref<'_, Option> { - self.root_store.borrow() + pub async fn root_store(&self) -> tokio::sync::RwLockReadGuard<'_, Option> { + self.root_store.read().await } /// Open the root store if set. Node owns the handle internally. /// Returns StoreInfo on success, or None if no root store is set. - pub fn open_root_store(&self) -> Result, NodeError> { + pub async fn open_root_store(&self) -> Result, NodeError> { match self.meta.root_store()? { Some(id) => { let (handle, info) = self.open_store(id)?; - *self.root_store.borrow_mut() = Some(handle); + *self.root_store.write().await = Some(handle); Ok(Some(info)) } None => Ok(None), @@ -213,17 +222,34 @@ impl Node { handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?; // Store the handle - node owns it - *self.root_store.borrow_mut() = Some(handle); + *self.root_store.write().await = Some(handle); Ok(store_id) } + /// Complete joining a mesh - creates store with given UUID, sets as root, caches handle. + /// Called after receiving store_id from peer's JoinResponse. + pub async fn complete_join(&self, store_id: Uuid) -> Result { + // Create local store with that UUID + self.create_store_with_uuid(store_id)?; + self.meta.set_root_store(store_id)?; + + // Open and cache the handle + let (handle, _) = self.open_store(store_id)?; + *self.root_store.write().await = Some(handle.clone()); + + // Publish our name to the store + let _ = self.publish_name().await; + + Ok(handle) + } + // --- Peer Management --- /// Invite a peer to the mesh. Writes their info with status = invited. pub async fn invite_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> { - let store = self.root_store.borrow(); - let store = store.as_ref() + let guard = self.root_store.read().await; + let store = guard.as_ref() .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; let pubkey_hex = hex::encode(pubkey); @@ -251,11 +277,11 @@ impl Node { /// List all peers in the mesh with their info pub async fn list_peers(&self) -> Result, NodeError> { - let store = self.root_store.borrow(); - let store = store.as_ref() + let guard = self.root_store.read().await; + let store = guard.as_ref() .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; - let all = store.list().await?; + let all = store.list(false).await?; // Collect unique pubkeys with status let mut peers_map: std::collections::HashMap = std::collections::HashMap::new(); @@ -299,10 +325,10 @@ impl Node { Ok(peers) } - /// Remove a peer from the mesh (sets status to removed) + /// Remove a peer from the mesh (deletes all their /nodes/{pubkey}/* keys) pub async fn remove_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> { - let store = self.root_store.borrow(); - let store = store.as_ref() + let guard = self.root_store.read().await; + let store = guard.as_ref() .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; let pubkey_hex = hex::encode(pubkey); @@ -312,28 +338,26 @@ impl Node { return Err(NodeError::Actor("Cannot remove yourself".to_string())); } - // Check if peer exists - let status_key = format!("/nodes/{}/status", pubkey_hex); - match store.get(status_key.as_bytes()).await? { - Some(status) if status == PeerStatus::Removed.as_str().as_bytes() => { - return Err(NodeError::Actor("Peer already removed".to_string())); - } - None => { - return Err(NodeError::Actor("Peer not found".to_string())); - } - _ => {} + // Find all keys for this peer using prefix search + let prefix = format!("/nodes/{}/", pubkey_hex); + let keys = store.list_by_prefix(prefix.as_bytes(), false).await?; + + if keys.is_empty() { + return Err(NodeError::Actor("Peer not found".to_string())); } - // Set status to removed - store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes()).await?; + // Delete all found keys + for (key, _) in keys { + store.delete(&key).await?; + } Ok(()) } /// Get a peer's status pub async fn get_peer_status(&self, pubkey: &[u8; 32]) -> Result, NodeError> { - let store = self.root_store.borrow(); - let store = store.as_ref() + let guard = self.root_store.read().await; + let store = guard.as_ref() .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; let pubkey_hex = hex::encode(pubkey); @@ -347,6 +371,44 @@ impl Node { None => Ok(None), } } + + /// Set a peer's status + pub async fn set_peer_status(&self, pubkey: &[u8; 32], status: PeerStatus) -> Result<(), NodeError> { + let guard = self.root_store.read().await; + let store = guard.as_ref() + .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; + + let pubkey_hex = hex::encode(pubkey); + let status_key = format!("/nodes/{}/status", pubkey_hex); + store.put(status_key.as_bytes(), status.as_str().as_bytes()).await?; + Ok(()) + } + + /// Verify a peer has one of the expected statuses + pub async fn verify_peer_status(&self, pubkey: &[u8; 32], expected: &[PeerStatus]) -> Result<(), NodeError> { + match self.get_peer_status(pubkey).await? { + Some(status) if expected.contains(&status) => Ok(()), + Some(status) => Err(NodeError::Actor(format!( + "Peer status is '{:?}', expected one of {:?}", status, expected + ))), + None => Err(NodeError::Actor("Peer not found".to_string())), + } + } + + /// Accept a peer's join request - verifies they're invited, sets active, returns join info + pub async fn accept_join(&self, pubkey: &[u8; 32]) -> Result { + // Verify peer is invited + self.verify_peer_status(pubkey, &[PeerStatus::Invited]).await?; + + // Get root store ID + let store_id = self.meta.root_store()? + .ok_or_else(|| NodeError::Actor("No root store configured".to_string()))?; + + // Set peer status to active + self.set_peer_status(pubkey, PeerStatus::Active).await?; + + Ok(JoinAcceptance { store_id }) + } pub fn list_stores(&self) -> Result, NodeError> { Ok(self.meta.list_stores()?) @@ -454,10 +516,20 @@ impl StoreHandle { .map_err(NodeError::Store) } - pub async fn list(&self) -> Result, Vec)>, NodeError> { + pub async fn list(&self, include_deleted: bool) -> Result, Vec)>, NodeError> { use StoreCmd; let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - self.tx.send(StoreCmd::List { resp: resp_tx }).await + self.tx.send(StoreCmd::List { include_deleted, resp: resp_tx }).await + .map_err(|_| NodeError::ChannelClosed)?; + resp_rx.await + .map_err(|_| NodeError::ChannelClosed)? + .map_err(NodeError::Store) + } + + pub async fn list_by_prefix(&self, prefix: &[u8], include_deleted: bool) -> Result, Vec)>, NodeError> { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::ListByPrefix { prefix: prefix.to_vec(), include_deleted, resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; resp_rx.await .map_err(|_| NodeError::ChannelClosed)? @@ -627,7 +699,7 @@ mod tests { .expect("create node"); // Initially no root store - assert!(node.root_store().is_none()); + assert!(node.root_store().await.is_none()); // Init creates root store let root_id = node.init().await.expect("init failed"); @@ -687,7 +759,7 @@ mod tests { .build() .expect("create node"); node.init().await.expect("init"); - let store = node.root_store(); + let store = node.root_store().await; let store = store.as_ref().unwrap(); // Get baseline seq after init @@ -733,7 +805,7 @@ mod tests { let pubkey_hex = hex::encode(node.node_id()); let name_key = format!("/nodes/{}/name", pubkey_hex); { - let store = node.root_store(); + let store = node.root_store().await; let store = store.as_ref().unwrap(); let stored_name = store.get(name_key.as_bytes()).await.unwrap(); assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec())); @@ -748,7 +820,7 @@ mod tests { // Verify store updated { - let store = node.root_store(); + let store = node.root_store().await; let store = store.as_ref().unwrap(); let stored_name = store.get(name_key.as_bytes()).await.unwrap(); assert_eq!(stored_name, Some(new_name.as_bytes().to_vec())); diff --git a/lattice-core/src/node_identity.rs b/lattice-core/src/node_identity.rs index 985c396..7ad0083 100644 --- a/lattice-core/src/node_identity.rs +++ b/lattice-core/src/node_identity.rs @@ -138,8 +138,8 @@ pub enum PeerStatus { Invited, /// Peer is active and can sync Active, - /// Peer has been removed from the mesh - Removed, + /// Peer is temporarily inactive + Dormant, } impl PeerStatus { @@ -147,7 +147,7 @@ impl PeerStatus { match self { PeerStatus::Invited => "invited", PeerStatus::Active => "active", - PeerStatus::Removed => "removed", + PeerStatus::Dormant => "dormant", } } @@ -155,7 +155,7 @@ impl PeerStatus { match s { "invited" => Some(PeerStatus::Invited), "active" => Some(PeerStatus::Active), - "removed" => Some(PeerStatus::Removed), + "dormant" => Some(PeerStatus::Dormant), _ => None, } } diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index 7175b9c..01d82a9 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -247,16 +247,36 @@ impl Store { } /// List all key-value pairs (winner values only) - pub fn list_all(&self) -> Result, Vec)>, StoreError> { + /// If include_deleted is true, includes tombstoned entries + pub fn list_all(&self, include_deleted: bool) -> Result, Vec)>, StoreError> { + self.list_by_prefix(&[], include_deleted) + } + + /// List all key-value pairs matching a prefix (winner values only) + /// Uses efficient range query on redb's sorted B-tree + /// If include_deleted is true, includes tombstoned entries + pub fn list_by_prefix(&self, prefix: &[u8], include_deleted: bool) -> Result, Vec)>, StoreError> { let read_txn = self.db.begin_read()?; let table = read_txn.open_table(KV_TABLE)?; let mut result = Vec::new(); - for entry in table.iter()? { + + // Use range query: from prefix to first key that doesn't match + for entry in table.range(prefix..)? { let (key, value) = entry?; + let key_bytes = key.value(); + + // Stop when we've passed the prefix + if !key_bytes.starts_with(prefix) { + break; + } + let heads = HeadList::decode(value.value())?.heads; if let Some(winner) = Self::pick_winner(&heads) { - result.push((key.value().to_vec(), winner.value.clone())); + // Skip tombstones unless include_deleted is true + if include_deleted || !winner.tombstone { + result.push((key_bytes.to_vec(), winner.value.clone())); + } } } Ok(result) @@ -1672,4 +1692,59 @@ mod tests { let _ = std::fs::remove_file(&path); } + + #[test] + fn test_list_by_prefix_filters_tombstones() { + let path = temp_db_path("list_tombstones"); + let _ = std::fs::remove_file(&path); + + let store = Store::open(&path).unwrap(); + let node = NodeIdentity::generate(); + + // Create a key under /test/ prefix + let clock1 = MockClock::new(1000); + let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock1)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/test/key1", b"value1".to_vec()) + .sign(&node); + store.apply_entry(&entry1).unwrap(); + + // Create another key + let clock2 = MockClock::new(2000); + let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock2)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry1).to_vec()) + .put("/test/key2", b"value2".to_vec()) + .sign(&node); + store.apply_entry(&entry2).unwrap(); + + // Delete key1 + let clock3 = MockClock::new(3000); + let entry3 = EntryBuilder::new(3, HLC::now_with_clock(&clock3)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry2).to_vec()) + .parent_hashes(vec![hash_signed_entry(&entry1).to_vec()]) + .delete(b"/test/key1") + .sign(&node); + store.apply_entry(&entry3).unwrap(); + + // list_by_prefix without include_deleted should only show key2 + let entries = store.list_by_prefix(b"/test/", false).unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].0, b"/test/key2"); + + // list_by_prefix with include_deleted should show both (key1 as tombstone) + let entries_all = store.list_by_prefix(b"/test/", true).unwrap(); + assert_eq!(entries_all.len(), 2); + + // Verify list_all also respects the flag + let all_entries = store.list_all(false).unwrap(); + assert_eq!(all_entries.len(), 1); + + let all_entries_incl_deleted = store.list_all(true).unwrap(); + assert_eq!(all_entries_incl_deleted.len(), 2); + + let _ = std::fs::remove_file(&path); + } } diff --git a/lattice-core/src/store_actor.rs b/lattice-core/src/store_actor.rs index 7b0bf7d..de58b86 100644 --- a/lattice-core/src/store_actor.rs +++ b/lattice-core/src/store_actor.rs @@ -24,6 +24,12 @@ pub enum StoreCmd { resp: oneshot::Sender, StoreError>>, }, List { + include_deleted: bool, + resp: oneshot::Sender, Vec)>, StoreError>>, + }, + ListByPrefix { + prefix: Vec, + include_deleted: bool, resp: oneshot::Sender, Vec)>, StoreError>>, }, Put { @@ -142,8 +148,11 @@ impl StoreActor { StoreCmd::GetHeads { key, resp } => { let _ = resp.send(self.store.get_heads(&key)); } - StoreCmd::List { resp } => { - let _ = resp.send(self.store.list_all()); + StoreCmd::List { include_deleted, resp } => { + let _ = resp.send(self.store.list_all(include_deleted)); + } + StoreCmd::ListByPrefix { prefix, include_deleted, resp } => { + let _ = resp.send(self.store.list_by_prefix(&prefix, include_deleted)); } StoreCmd::Put { key, value, resp } => { let result = self.do_put(&key, &value); diff --git a/lattice-net/src/mesh/server.rs b/lattice-net/src/mesh/server.rs index 4436272..aeceea6 100644 --- a/lattice-net/src/mesh/server.rs +++ b/lattice-net/src/mesh/server.rs @@ -1,27 +1,26 @@ //! Server - handle incoming peer connections for join and sync use crate::{MessageSink, MessageStream}; -use lattice_core::{StoreHandle, PeerStatus}; +use lattice_core::{Node, PeerStatus, Uuid}; 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( + node: Arc, endpoint: Endpoint, - shared_store: Arc>>, ) { tokio::spawn(async move { loop { if let Some(incoming) = endpoint.accept().await { match incoming.await { Ok(conn) => { - let store = shared_store.clone(); + let node = node.clone(); tokio::spawn(async move { - if let Err(e) = handle_connection(conn, store).await { + if let Err(e) = handle_connection(node, conn).await { eprintln!("[Accept] Error: {}", e); } }); @@ -35,104 +34,89 @@ pub fn spawn_accept_loop( /// Handle a single incoming connection async fn handle_connection( + node: Arc, conn: Connection, - shared_store: Arc>>, ) -> Result<(), String> { let remote_id = conn.remote_id(); let remote_hex = hex::encode(remote_id.as_bytes()); println!("\n[Incoming] {} (ALPN: {})", remote_id.fmt_short(), String::from_utf8_lossy(conn.alpn())); - let store = { - let guard = shared_store.read().await; - match &*guard { - Some(s) => s.clone(), - None => return Err("No store available".to_string()), - } - }; + // Parse remote pubkey + let remote_pubkey: [u8; 32] = hex::decode(&remote_hex) + .map_err(|_| "Invalid pubkey hex")? + .try_into() + .map_err(|_| "Invalid pubkey length")?; let (send, recv) = conn.accept_bi().await .map_err(|e| format!("Accept stream error: {}", e))?; - // Wrap in framed message streams - let mut sink = MessageSink::new(send); + let sink = MessageSink::new(send); let mut stream = MessageStream::new(recv); - // Read first message + // Read first message to determine request type let msg = stream.recv().await? .ok_or_else(|| "Peer closed stream".to_string())?; match msg.message { Some(peer_message::Message::JoinRequest(req)) => { - // For join: verify peer is invited - verify_peer_status(&store, &remote_hex, PeerStatusCheck::Exactly(PeerStatus::Invited)).await?; - println!("[Peer] Verified as invited"); - - println!("[Join] Got JoinRequest from {}", hex::encode(&req.node_pubkey)); - - let resp = PeerMessage { - message: Some(peer_message::Message::JoinResponse(JoinResponse { - store_uuid: store.id().as_bytes().to_vec(), - inviter_pubkey: vec![], - })), - }; - sink.send(&resp).await?; - sink.finish().await?; - - // Set peer status to 'active' now that they've joined - let status_key = format!("/nodes/{}/status", remote_hex); - if let Err(e) = store.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await { - eprintln!("[Join] Warning: Failed to set peer status: {}", e); - } - - println!("[Join] Sent JoinResponse, peer now active"); - Ok(()) + handle_join_request(&node, &remote_pubkey, req, sink).await } Some(peer_message::Message::SyncRequest(req)) => { - // For sync: verify peer is active (or invited for first sync after join) - verify_peer_status(&store, &remote_hex, PeerStatusCheck::ActiveOrInvited).await?; - println!("[Peer] Verified for sync"); - - handle_sync_request(sink, stream, req, &store).await + handle_sync_request(&node, &remote_pubkey, req, sink, stream).await } _ => Err("Unexpected message type".to_string()), } } -/// Expected peer status check mode -#[derive(Debug, Clone, Copy)] -enum PeerStatusCheck { - Exactly(PeerStatus), - ActiveOrInvited, -} - -/// Verify a peer has the expected status -async fn verify_peer_status(store: &StoreHandle, remote_hex: &str, expected: PeerStatusCheck) -> Result<(), String> { - let status_key = format!("/nodes/{}/status", remote_hex); - let status = match store.get(status_key.as_bytes()).await { - Ok(Some(s)) => String::from_utf8_lossy(&s).to_string(), - Ok(None) => return Err(format!("Peer not found")), - Err(e) => return Err(format!("Error checking peer status: {}", e)), - }; +/// Handle a join request from an invited peer +async fn handle_join_request( + node: &Node, + remote_pubkey: &[u8; 32], + req: lattice_core::proto::JoinRequest, + mut sink: MessageSink, +) -> Result<(), String> { + println!("[Join] Got JoinRequest from {}", hex::encode(&req.node_pubkey)); - let valid = match expected { - PeerStatusCheck::Exactly(ps) => status == ps.as_str(), - PeerStatusCheck::ActiveOrInvited => status == PeerStatus::Active.as_str() || status == PeerStatus::Invited.as_str(), - }; + // Accept the join - verifies invited, sets active, returns store ID + let acceptance = node.accept_join(remote_pubkey).await + .map_err(|e| e.to_string())?; - if valid { - Ok(()) - } else { - Err(format!("Peer status is '{}', expected {:?}", status, expected)) - } + let resp = PeerMessage { + message: Some(peer_message::Message::JoinResponse(JoinResponse { + store_uuid: acceptance.store_id.as_bytes().to_vec(), + inviter_pubkey: vec![], + })), + }; + sink.send(&resp).await?; + sink.finish().await?; + + println!("[Join] Sent JoinResponse, peer now active"); + Ok(()) } /// Handle a sync request - bidirectional exchange of entries async fn handle_sync_request( + node: &Node, + remote_pubkey: &[u8; 32], + peer_request: lattice_core::proto::SyncRequest, mut sink: MessageSink, mut stream: MessageStream, - peer_request: lattice_core::proto::SyncRequest, - store: &StoreHandle, ) -> Result<(), String> { + // Verify peer is active (allowed to sync) + node.verify_peer_status(remote_pubkey, &[PeerStatus::Active]).await + .map_err(|e| e.to_string())?; + println!("[Sync] Verified peer as active"); + + // Parse store_id from request + let store_id = Uuid::from_slice(&peer_request.store_id) + .map_err(|_| "Invalid store_id in SyncRequest".to_string())?; + + println!("[Sync] Received SyncRequest for store {}", store_id); + + // Open the requested store + let (store, _info) = node.open_store(store_id) + .map_err(|e| format!("Failed to open store {}: {}", store_id, e))?; + println!("[Sync] Received SyncRequest"); // Get our sync state @@ -142,6 +126,7 @@ async fn handle_sync_request( // 1. Send our sync state as response let resp = PeerMessage { message: Some(peer_message::Message::SyncResponse(lattice_core::proto::SyncResponse { + store_id: store.id().as_bytes().to_vec(), state: Some(my_state.to_proto()), })), }; @@ -152,11 +137,11 @@ async fn handle_sync_request( .map(|s| lattice_core::sync_state::SyncState::from_proto(&s)) .unwrap_or_default(); - let entries_sent = 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, _) = protocol::receive_entries(&mut stream, store).await?; + let (entries_applied, _) = protocol::receive_entries(&mut stream, &store).await?; sink.finish().await?; diff --git a/lattice-net/src/mesh/sync.rs b/lattice-net/src/mesh/sync.rs index bf2f7e5..2ddff7c 100644 --- a/lattice-net/src/mesh/sync.rs +++ b/lattice-net/src/mesh/sync.rs @@ -52,11 +52,8 @@ pub async fn join_mesh( let store_uuid = lattice_core::Uuid::from_slice(&resp.store_uuid) .map_err(|_| NodeError::Actor("Invalid UUID from peer".to_string()))?; - // Create local store with that UUID - node.create_store_with_uuid(store_uuid)?; - node.set_root_store(store_uuid)?; - - let (handle, _) = node.open_store(store_uuid)?; + // Complete join - creates store, sets as root, caches handle + let handle = node.complete_join(store_uuid).await?; // Immediately sync with the peer to get initial data println!("[Join] Syncing with peer to get initial data..."); @@ -66,18 +63,9 @@ pub async fn join_mesh( } Err(e) => { eprintln!("[Join] Warning: Initial sync failed: {}", e); - // Don't fail join, just warn - peer might not have data yet } } - // 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())), @@ -109,6 +97,7 @@ pub async fn sync_with_peer( // 1. Send SyncRequest with our state (don't finish yet - we'll send entries later) let req = PeerMessage { message: Some(peer_message::Message::SyncRequest(lattice_core::proto::SyncRequest { + store_id: store.id().as_bytes().to_vec(), state: Some(my_state.to_proto()), full_sync: false, })), @@ -163,7 +152,7 @@ pub async fn sync_with_peer( }) } -/// Sync with all active peers from the store. +/// Sync with all active peers from the node. pub async fn sync_all( node: &Node, endpoint: &LatticeEndpoint, @@ -171,34 +160,22 @@ pub async fn sync_all( ) -> Result, NodeError> { let my_pubkey = hex::encode(node.node_id()); - // Get all active peers (invited peers haven't joined yet) - let all_entries = store.list().await?; - let mut peer_ids = Vec::new(); + // Get all active peers using node.list_peers() + let peers = node.list_peers().await?; + let mut results = Vec::new(); - for (key, value) in &all_entries { - let key_str = String::from_utf8_lossy(key); - 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) = parse_node_id(pubkey) { - peer_ids.push(id); + for peer in peers { + if peer.status == PeerStatus::Active && peer.pubkey != my_pubkey { + if let Ok(peer_id) = parse_node_id(&peer.pubkey) { + match sync_with_peer(endpoint, store, peer_id).await { + Ok(result) => results.push(result), + Err(e) => { + eprintln!("Sync with {} failed: {}", peer_id.fmt_short(), e); } } } } } - // Sync with each peer - let mut results = Vec::new(); - for peer_id in peer_ids { - match sync_with_peer(endpoint, store, peer_id).await { - Ok(result) => results.push(result), - Err(e) => { - // Log error but continue with other peers - eprintln!("Sync with {} failed: {}", peer_id.fmt_short(), e); - } - } - } - Ok(results) } diff --git a/proto/lattice.proto b/proto/lattice.proto index ff9bc71..f25a81b 100644 --- a/proto/lattice.proto +++ b/proto/lattice.proto @@ -111,12 +111,14 @@ message JoinResponse { // 7. Sync Protocol Messages (bidirectional sync after join) message SyncRequest { - SyncState state = 1; // Sender's sync state (for incremental sync) - bool full_sync = 2; // If true, request all entries (for join) + bytes store_id = 1; // Store UUID to sync (16 bytes) + SyncState state = 2; // Sender's sync state (for incremental sync) + bool full_sync = 3; // If true, request all entries (for join) } message SyncResponse { - SyncState state = 1; // Responder's sync state + bytes store_id = 1; // Store UUID being synced (16 bytes) + SyncState state = 2; // Responder's sync state } message SyncEntry {