diff --git a/docs/roadmap.md b/docs/roadmap.md index f8fd136..196923b 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -147,10 +147,10 @@ ### Deliverables **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` +- [x] `LatticeServer` struct in `lattice-net` wrapping `Arc` + `Endpoint` +- [x] Move `join_mesh`, `sync_with_peer`, `sync_all` to `LatticeServer` methods +- [x] Encapsulate `spawn_accept_loop` inside `LatticeServer` +- [x] 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 @@ -166,6 +166,11 @@ ## Future +- Gossip: + - gossip new entries to peers + - backfill missing entries from peers (how do peers notice missing entries?) + - snapshots for kv store + - prune using consensus watermark - remove_peer should be a transactional operation on store - Watermark tracking & log pruning - Track minimum confirmed seq per author across all peers diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index c50769e..40ba5e2 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -1,7 +1,7 @@ //! CLI command handlers use lattice_core::{Node, StoreHandle}; -use lattice_net::LatticeEndpoint; +use lattice_net::LatticeServer; /// Result of a command that may switch stores or exit pub enum CommandResult { @@ -18,7 +18,7 @@ pub fn block_async(f: F) -> F::Output { tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(f)) } -pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult; +pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeServer>, &[String]) -> CommandResult; pub struct Command { pub name: &'static str, @@ -53,7 +53,7 @@ pub fn commands() -> Vec { cmds } -fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { let cmds = commands(); let mut last_group = ""; for cmd in &cmds { @@ -73,7 +73,7 @@ fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Latti CommandResult::Ok } -fn cmd_quit(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_quit(_node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { println!("Goodbye!"); CommandResult::Quit } diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index e858a73..f00d157 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -4,7 +4,7 @@ mod commands; mod node_commands; mod store_commands; -use lattice_net::spawn_accept_loop; +use lattice_net::LatticeServer; use commands::CommandResult; use lattice_core::{NodeBuilder, StoreHandle}; use rustyline::error::ReadlineError; @@ -24,11 +24,11 @@ async fn main() { } }; - // Start Iroh endpoint using same Ed25519 identity - let endpoint = match lattice_net::LatticeEndpoint::new(node.secret_key_bytes()).await { - Ok(ep) => { - println!("Iroh: {} (listening)", ep.public_key().fmt_short()); - Some(ep) + // Create LatticeServer (creates endpoint and spawns accept loop internally) + let server = match LatticeServer::new_from_node(node.clone()).await { + Ok(s) => { + println!("Iroh: {} (listening)", s.endpoint().public_key().fmt_short()); + Some(s) } Err(e) => { eprintln!("Warning: Iroh failed to start: {}", e); @@ -36,11 +36,6 @@ async fn main() { } }; - // Spawn accept loop for incoming connections - if let Some(ref ep) = endpoint { - spawn_accept_loop(node.clone(), ep.endpoint().clone()); - } - let info = node.info(); println!("Node ID: {}", info.node_id); println!("Data: {}", info.data_path); @@ -100,7 +95,7 @@ async fn main() { if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args { println!("Usage: {} {}", cmd.name, cmd.args); } else { - match (cmd.handler)(&node, current_store.as_ref(), endpoint.as_ref(), cmd_args) { + match (cmd.handler)(&node, current_store.as_ref(), server.as_ref(), cmd_args) { CommandResult::Ok => {} CommandResult::SwitchTo(h) => { current_store = Some(h); diff --git a/lattice-cli/src/node_commands.rs b/lattice-cli/src/node_commands.rs index 222d714..adf0e55 100644 --- a/lattice-cli/src/node_commands.rs +++ b/lattice-cli/src/node_commands.rs @@ -2,7 +2,7 @@ use crate::commands::{block_async, Command, CommandResult, Handler}; use lattice_core::{Node, StoreHandle, PeerStatus, Uuid}; -use lattice_net::LatticeEndpoint; +use lattice_net::LatticeServer; use chrono::DateTime; use std::time::Instant; @@ -26,7 +26,7 @@ pub fn node_commands() -> Vec { // --- Store management --- -fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { match block_async(node.init()) { Ok(store_id) => { println!("Initialized with root store: {}", store_id); @@ -43,7 +43,7 @@ fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Lattic } } -fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { match node.create_store() { Ok(store_id) => { println!("Created store: {}", store_id); @@ -65,7 +65,7 @@ fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option } } -fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let store_id = match Uuid::parse_str(&args[0]) { Ok(id) => id, Err(_) => { @@ -91,7 +91,7 @@ fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&L } } -fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { let stores = match node.list_stores() { Ok(s) => s, Err(e) => { @@ -114,7 +114,7 @@ fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<& // --- Info --- -fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { println!("Node ID: {}", hex::encode(node.node_id())); if let Some(name) = node.name() { println!("Name: {}", name); @@ -138,7 +138,7 @@ fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option< // --- Peer management --- -fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let pubkey_hex = &args[0]; let pubkey: [u8; 32] = match hex::decode(pubkey_hex) { Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), @@ -158,7 +158,7 @@ fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Latt CommandResult::Ok } -fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { let peers = match block_async(node.list_peers()) { Ok(p) => p, Err(e) => { @@ -204,7 +204,7 @@ fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Latti CommandResult::Ok } -fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let pubkey_hex = &args[0]; let pubkey: [u8; 32] = match hex::decode(pubkey_hex) { Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), @@ -223,9 +223,9 @@ fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&Latt // --- Networking --- -fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let endpoint = match endpoint { - Some(ep) => ep, +fn cmd_join(_node: &Node, store: Option<&StoreHandle>, server: Option<&LatticeServer>, args: &[String]) -> CommandResult { + let server = match server { + Some(s) => s, None => { eprintln!("Iroh endpoint not started."); return CommandResult::Ok; @@ -247,7 +247,7 @@ fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeE println!("Joining mesh via {}...", peer_id.fmt_short()); - match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) { + match block_async(server.join_mesh(peer_id)) { Ok(handle) => { println!("Joined mesh! Use 'sync' command to sync entries."); CommandResult::SwitchTo(handle) @@ -259,9 +259,9 @@ fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeE } } -fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let endpoint = match endpoint { - Some(ep) => ep, +fn cmd_sync(_node: &Node, store: Option<&StoreHandle>, server: Option<&LatticeServer>, args: &[String]) -> CommandResult { + let server = match server { + Some(s) => s, None => { eprintln!("Iroh endpoint not started."); return CommandResult::Ok; @@ -278,7 +278,7 @@ fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeE if args.is_empty() { // Sync with all active peers - match block_async(lattice_net::sync_all(node, endpoint, store)) { + match block_async(server.sync_all(store)) { Ok(results) => { if results.is_empty() { println!("No peers to sync with."); @@ -300,7 +300,7 @@ fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeE }; println!("Syncing with {}...", peer_id.fmt_short()); - match block_async(lattice_net::sync_with_peer(endpoint, store, peer_id)) { + match block_async(server.sync_with_peer(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/store_commands.rs b/lattice-cli/src/store_commands.rs index 23830bc..d5adfe3 100644 --- a/lattice-cli/src/store_commands.rs +++ b/lattice-cli/src/store_commands.rs @@ -2,7 +2,7 @@ use crate::commands::{block_async, Command, CommandResult, Handler}; use lattice_core::{Node, StoreHandle}; -use lattice_net::LatticeEndpoint; +use lattice_net::LatticeServer; use std::time::Instant; pub fn store_commands() -> Vec { @@ -16,7 +16,7 @@ pub fn store_commands() -> Vec { ] } -fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { +fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -38,7 +38,7 @@ fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option CommandResult::Ok } -fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -51,7 +51,7 @@ fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Lattice CommandResult::Ok } -fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -102,7 +102,7 @@ fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Lattice CommandResult::Ok } -fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -115,7 +115,7 @@ fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Latt CommandResult::Ok } -fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let Some(h) = store else { println!("No store selected. Use 'init' or 'use '"); return CommandResult::Ok; @@ -173,7 +173,7 @@ fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&Lattic CommandResult::Ok } -fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { +fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index d12df9c..c558ac9 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -840,4 +840,108 @@ mod tests { let _ = std::fs::remove_dir_all(data_dir.base()); } + + #[tokio::test] + async fn test_invite_peer() { + let data_dir = temp_data_dir("invite_peer"); + + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + + // Init first + node.init().await.expect("init"); + + // Invite a peer + let peer_pubkey = [0u8; 32]; // Dummy pubkey + node.invite_peer(&peer_pubkey).await.expect("invite"); + + // Verify peer is Invited + let peers = node.list_peers().await.expect("list_peers"); + let invited = peers.iter().find(|p| p.pubkey == hex::encode(peer_pubkey)); + assert!(invited.is_some(), "Should find invited peer"); + assert_eq!(invited.unwrap().status, PeerStatus::Invited); + + let _ = std::fs::remove_dir_all(data_dir.base()); + } + + #[tokio::test] + async fn test_accept_join() { + let data_dir = temp_data_dir("accept_join"); + + let node = NodeBuilder { data_dir: data_dir.clone() } + .build() + .expect("create node"); + + // Init first + let store_id = node.init().await.expect("init"); + + // Invite a peer + let peer_pubkey = [1u8; 32]; // Dummy pubkey + node.invite_peer(&peer_pubkey).await.expect("invite"); + + // Accept the join + let acceptance = node.accept_join(&peer_pubkey).await.expect("accept_join"); + assert_eq!(acceptance.store_id, store_id); + + // Peer should now be Active + let peers = node.list_peers().await.expect("list_peers"); + let peer = peers.iter().find(|p| p.pubkey == hex::encode(peer_pubkey)); + assert!(peer.is_some(), "Should find peer"); + assert_eq!(peer.unwrap().status, PeerStatus::Active); + + let _ = std::fs::remove_dir_all(data_dir.base()); + } + + #[tokio::test] + async fn test_invite_join_sync_flow() { + // Node A: creator, Node B: joiner + let data_dir_a = temp_data_dir("flow_a"); + let data_dir_b = temp_data_dir("flow_b"); + + let node_a = NodeBuilder { data_dir: data_dir_a.clone() } + .build() + .expect("create node A"); + let node_b = NodeBuilder { data_dir: data_dir_b.clone() } + .build() + .expect("create node B"); + + // Step 1: Node A initializes + let store_id = node_a.init().await.expect("A init"); + let store_a = node_a.root_store().await; + let store_a = store_a.as_ref().expect("A has root store"); + + // Step 2: A invites B + let b_pubkey: [u8; 32] = node_b.node_id().try_into().unwrap(); + node_a.invite_peer(&b_pubkey).await.expect("invite B"); + + // Verify B is invited + let peers = node_a.list_peers().await.expect("list peers"); + assert!(peers.iter().any(|p| p.status == PeerStatus::Invited)); + + // Step 3: B "joins" (complete_join simulates receiving JoinResponse) + let store_b = node_b.complete_join(store_id).await.expect("B join"); + + // Verify B has the same store ID + assert_eq!(store_b.id(), store_id); + + // Step 4: A writes data + store_a.put(b"/key", b"from A").await.expect("A put"); + + // Step 5: B writes data independently + store_b.put(b"/key", b"from B").await.expect("B put"); + + // Each store has its own local state (not synced yet) + let a_val = store_a.get(b"/key").await.expect("A get").unwrap(); + let b_val = store_b.get(b"/key").await.expect("B get").unwrap(); + + // A sees "from A" (its own write wins locally) + assert_eq!(a_val, b"from A".to_vec()); + // B sees "from B" (its own write wins locally) + assert_eq!(b_val, b"from B".to_vec()); + + // Cleanup + let _ = std::fs::remove_dir_all(data_dir_a.base()); + let _ = std::fs::remove_dir_all(data_dir_b.base()); + } } diff --git a/lattice-net/src/lib.rs b/lattice-net/src/lib.rs index 6202b96..bc0690f 100644 --- a/lattice-net/src/lib.rs +++ b/lattice-net/src/lib.rs @@ -15,7 +15,7 @@ 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}; +pub use mesh::{LatticeServer, 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 index 341136d..7eda3be 100644 --- a/lattice-net/src/mesh/mod.rs +++ b/lattice-net/src/mesh/mod.rs @@ -1,13 +1,10 @@ //! Mesh networking - peer-to-peer join and sync operations //! -//! - **server**: Accept incoming connections and handle join/sync requests -//! - **sync**: Outgoing join and sync operations +//! - **server**: LatticeServer for mesh networking (join, sync, accept loop) //! - **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 server::{LatticeServer, SyncResult}; pub use protocol::{send_missing_entries, receive_entries}; diff --git a/lattice-net/src/mesh/server.rs b/lattice-net/src/mesh/server.rs index c93ca29..dbcfe3b 100644 --- a/lattice-net/src/mesh/server.rs +++ b/lattice-net/src/mesh/server.rs @@ -1,37 +1,199 @@ -//! Server - handle incoming peer connections for join and sync +//! Server - LatticeServer for mesh networking -use crate::{MessageSink, MessageStream}; -use lattice_core::{Node, PeerStatus, Uuid}; -use iroh::Endpoint; +use crate::{MessageSink, MessageStream, LatticeEndpoint, parse_node_id}; +use lattice_core::{Node, NodeError, PeerStatus, Uuid, StoreHandle}; use iroh::endpoint::Connection; use std::sync::Arc; -use lattice_core::proto::{PeerMessage, peer_message, JoinResponse}; +use lattice_core::proto::{PeerMessage, peer_message, JoinRequest, JoinResponse}; use super::protocol; -/// Spawn the accept loop for incoming connections. -pub fn spawn_accept_loop( +/// Result of a sync operation with a peer +pub struct SyncResult { + pub entries_applied: u64, + pub entries_sent_by_peer: u64, +} + +/// LatticeServer wraps Node + Endpoint and provides mesh networking methods. +/// Spawns accept loop on creation to handle incoming connections. +pub struct LatticeServer { node: Arc, - endpoint: Endpoint, -) { - tokio::spawn(async move { - loop { - if let Some(incoming) = endpoint.accept().await { - match incoming.await { - Ok(conn) => { - let node = node.clone(); - tokio::spawn(async move { - if let Err(e) = handle_connection(node, conn).await { - eprintln!("[Accept] Error: {}", e); - } - }); + endpoint: LatticeEndpoint, +} + +impl LatticeServer { + /// Create a new LatticeServer from just a Node (creates endpoint internally). + pub async fn new_from_node(node: Arc) -> Result { + let endpoint = LatticeEndpoint::new(node.secret_key_bytes()).await + .map_err(|e| format!("Failed to create endpoint: {}", e))?; + Ok(Self::new(node, endpoint)) + } + + /// Create a new LatticeServer with existing endpoint and spawn the accept loop. + pub fn new(node: Arc, endpoint: LatticeEndpoint) -> Self { + let server = Self { node, endpoint }; + server.spawn_accept_loop(); + server + } + + /// Access the underlying node + pub fn node(&self) -> &Node { + &self.node + } + + /// Access the underlying endpoint + pub fn endpoint(&self) -> &LatticeEndpoint { + &self.endpoint + } + + /// Spawn the accept loop for incoming connections. + fn spawn_accept_loop(&self) { + let node = self.node.clone(); + let endpoint = self.endpoint.endpoint().clone(); + tokio::spawn(async move { + loop { + if let Some(incoming) = endpoint.accept().await { + match incoming.await { + Ok(conn) => { + let node = node.clone(); + tokio::spawn(async move { + if let Err(e) = handle_connection(node, conn).await { + eprintln!("[Accept] Error: {}", e); + } + }); + } + Err(e) => eprintln!("[Accept] Handshake error: {:?}", e), } - Err(e) => eprintln!("[Accept] Handshake error: {:?}", e), } } + }); + } + + /// Join an existing mesh by connecting to a peer. + pub async fn join_mesh(&self, peer_id: iroh::PublicKey) -> Result { + let conn = self.endpoint.connect(peer_id).await + .map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?; + + let (send, recv) = conn.open_bi().await + .map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?; + + let mut sink = MessageSink::new(send); + let mut stream = MessageStream::new(recv); + + // Send JoinRequest + let req = PeerMessage { + message: Some(peer_message::Message::JoinRequest(JoinRequest { + node_pubkey: self.node.node_id().to_vec(), + })), + }; + sink.send(&req).await.map_err(|e| NodeError::Actor(e))?; + sink.finish().await.map_err(|e| NodeError::Actor(e))?; + + // Receive JoinResponse + let msg = stream.recv().await + .map_err(|e| NodeError::Actor(e))? + .ok_or_else(|| NodeError::Actor("Peer closed stream".to_string()))?; + + match msg.message { + Some(peer_message::Message::JoinResponse(resp)) => { + let store_uuid = lattice_core::Uuid::from_slice(&resp.store_uuid) + .map_err(|_| NodeError::Actor("Invalid UUID from peer".to_string()))?; + + let handle = self.node.complete_join(store_uuid).await?; + + // Sync with peer to get initial data + println!("[Join] Syncing with peer to get initial data..."); + if let Ok(result) = self.sync_with_peer(&handle, peer_id).await { + println!("[Join] Initial sync complete: {} entries", result.entries_applied); + } + + Ok(handle) + } + _ => Err(NodeError::Actor("Unexpected response".to_string())), } - }); + } + + /// Sync with a specific peer. + pub async fn sync_with_peer(&self, store: &StoreHandle, peer_id: iroh::PublicKey) -> Result { + let conn = self.endpoint.connect(peer_id).await + .map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?; + + let (send, recv) = conn.open_bi().await + .map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?; + + let mut sink = MessageSink::new(send); + let mut stream = MessageStream::new(recv); + + let my_state = store.sync_state().await?; + + // Send SyncRequest + 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, + })), + }; + sink.send(&req).await.map_err(|e| NodeError::Actor(e))?; + + // Receive SyncResponse + let resp_msg = stream.recv().await.map_err(|e| NodeError::Actor(e))? + .ok_or_else(|| NodeError::Actor("Peer closed stream".to_string()))?; + + let peer_state = match resp_msg.message { + Some(peer_message::Message::SyncResponse(resp)) => { + resp.state.map(|s| lattice_core::sync_state::SyncState::from_proto(&s)) + .unwrap_or_default() + } + _ => return Err(NodeError::Actor("Expected SyncResponse".to_string())), + }; + + // Exchange entries + let _entries_sent = protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await + .map_err(|e| NodeError::Actor(e))?; + + let (entries_applied, entries_sent_by_peer) = protocol::receive_entries(&mut stream, store).await + .map_err(|e| NodeError::Actor(e))?; + + sink.finish().await.map_err(|e| NodeError::Actor(e))?; + + Ok(SyncResult { entries_applied, entries_sent_by_peer }) + } + + /// Sync with all active peers. + pub async fn sync_all(&self, store: &StoreHandle) -> Result, NodeError> { + let peers = self.node.list_peers().await?; + let mut results = Vec::new(); + + for peer in peers { + if peer.status != PeerStatus::Active { + continue; + } + + let peer_id = match parse_node_id(&peer.pubkey) { + Ok(id) => id, + Err(e) => { + eprintln!("[Sync] Failed to parse peer {}: {}", peer.pubkey, e); + continue; + } + }; + + println!("[Sync] Syncing with {}...", peer_id.fmt_short()); + match self.sync_with_peer(store, peer_id).await { + Ok(result) => { + println!("[Sync] Applied {} entries", result.entries_applied); + results.push(result); + } + Err(e) => eprintln!("[Sync] Failed: {}", e), + } + } + + Ok(results) + } } +// --- Connection handling --- +// --- Connection handling --- + /// Handle a single incoming connection async fn handle_connection( node: Arc, diff --git a/lattice-net/src/mesh/sync.rs b/lattice-net/src/mesh/sync.rs deleted file mode 100644 index 2ddff7c..0000000 --- a/lattice-net/src/mesh/sync.rs +++ /dev/null @@ -1,181 +0,0 @@ -//! Sync - outgoing mesh join and sync operations - -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 prost::Message; -use super::protocol; - -/// Result of a sync operation with a peer -pub struct SyncResult { - pub entries_applied: u64, - pub entries_sent_by_peer: u64, -} - -/// Join an existing mesh by connecting to a peer. -/// Returns the new StoreHandle on success. -/// After joining, automatically syncs with the peer to get initial data. -pub async fn join_mesh( - node: &Node, - endpoint: &LatticeEndpoint, - peer_id: iroh::PublicKey, -) -> Result { - // Connect to peer - let conn = endpoint.connect(peer_id).await - .map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?; - - // Open stream - let (send, recv) = conn.open_bi().await - .map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?; - - let mut sink = MessageSink::new(send); - let mut stream = MessageStream::new(recv); - - // Send JoinRequest - let req = PeerMessage { - message: Some(peer_message::Message::JoinRequest(JoinRequest { - node_pubkey: node.node_id().to_vec(), - })), - }; - sink.send(&req).await - .map_err(|e| NodeError::Actor(format!("Failed to send: {}", e)))?; - sink.finish().await - .map_err(|e| NodeError::Actor(format!("Failed to finish: {}", e)))?; - - // Receive JoinResponse - let msg = stream.recv().await - .map_err(|e| NodeError::Actor(format!("Recv error: {}", e)))? - .ok_or_else(|| NodeError::Actor("Peer closed stream".to_string()))?; - - match msg.message { - Some(peer_message::Message::JoinResponse(resp)) => { - let store_uuid = lattice_core::Uuid::from_slice(&resp.store_uuid) - .map_err(|_| NodeError::Actor("Invalid UUID from peer".to_string()))?; - - // 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..."); - match sync_with_peer(endpoint, &handle, peer_id).await { - Ok(result) => { - println!("[Join] Initial sync complete: applied {} entries", result.entries_applied); - } - Err(e) => { - eprintln!("[Join] Warning: Initial sync failed: {}", e); - } - } - - Ok(handle) - } - _ => Err(NodeError::Actor("Unexpected response message type".to_string())), - } -} - -/// Sync with a specific peer (bidirectional). -/// Both sides exchange states and send missing entries to each other. -pub async fn sync_with_peer( - endpoint: &LatticeEndpoint, - store: &StoreHandle, - peer_id: iroh::PublicKey, -) -> Result { - - // Connect - let conn = endpoint.connect(peer_id).await - .map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?; - - // Open stream - let (send, recv) = conn.open_bi().await - .map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?; - - let mut sink = MessageSink::new(send); - let mut stream = MessageStream::new(recv); - - // Get our sync state - let my_state = store.sync_state().await?; - - // 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, - })), - }; - sink.send(&req).await - .map_err(|e| NodeError::Actor(format!("Failed to send: {}", e)))?; - - // 2. Receive SyncResponse (peer's state) and entries until SyncDone - let mut entries_applied = 0u64; - let mut entries_sent_by_peer = 0u64; - let mut peer_state = lattice_core::sync_state::SyncState::default(); - - loop { - match stream.recv().await { - Ok(Some(msg)) => match msg.message { - Some(peer_message::Message::SyncResponse(resp)) => { - // Peer's sync state - we'll use this to compute what to send - if let Some(s) = resp.state { - peer_state = lattice_core::sync_state::SyncState::from_proto(&s); - } - } - Some(peer_message::Message::SyncEntry(entry)) => { - if let Ok(signed) = SignedEntry::decode(&entry.signed_entry[..]) { - if store.apply_entry(signed).await.is_ok() { - entries_applied += 1; - } - } - } - Some(peer_message::Message::SyncDone(done)) => { - entries_sent_by_peer = done.entries_sent; - break; - } - _ => {} - } - Ok(None) => break, - Err(_) => break, - } - } - - // 3. Send entries peer is missing (using shared protocol) - 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)))?; - - println!("[Sync] Applied {} entries, sent {} entries", entries_applied, entries_sent); - - Ok(SyncResult { - entries_applied, - entries_sent_by_peer, - }) -} - -/// Sync with all active peers from the node. -pub async fn sync_all( - node: &Node, - endpoint: &LatticeEndpoint, - store: &StoreHandle, -) -> Result, NodeError> { - let my_pubkey = hex::encode(node.node_id()); - - // Get all active peers using node.list_peers() - let peers = node.list_peers().await?; - let mut results = Vec::new(); - - 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); - } - } - } - } - } - - Ok(results) -}