From e942da49fffc78db61d45d6847f24cb48aa5a27a Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Mon, 22 Dec 2025 21:11:34 +0100 Subject: [PATCH] feat: Implement Iroh-based peer networking, join protocol, and bidirectional store synchronization. --- Cargo.toml | 5 +- docs/architecture.md | 9 +- docs/roadmap.md | 45 +++- lattice-cli/Cargo.toml | 6 + lattice-cli/src/accept_handler.rs | 165 +++++++++++++ lattice-cli/src/commands.rs | 392 ++++++++++++++++++++++++++++-- lattice-cli/src/main.rs | 36 ++- lattice-cli/src/node.rs | 98 +++++++- lattice-cli/src/store_actor.rs | 92 ++++++- lattice-cli/src/sync.rs | 217 +++++++++++++++++ lattice-cli/src/sync_protocol.rs | 89 +++++++ lattice-core/src/causal_iter.rs | 189 ++++++++++++++ lattice-core/src/lib.rs | 6 +- lattice-core/src/node.rs | 6 + lattice-core/src/sigchain.rs | 69 ++++++ lattice-core/src/store.rs | 203 ++++++++++++++++ lattice-core/src/sync_state.rs | 184 ++++++++++++-- lattice-net/Cargo.toml | 3 + lattice-net/src/endpoint.rs | 57 +++++ lattice-net/src/framing.rs | 63 +++++ lattice-net/src/lib.rs | 13 + proto/lattice.proto | 43 +++- 22 files changed, 1933 insertions(+), 57 deletions(-) create mode 100644 lattice-cli/src/accept_handler.rs create mode 100644 lattice-cli/src/sync.rs create mode 100644 lattice-cli/src/sync_protocol.rs create mode 100644 lattice-core/src/causal_iter.rs create mode 100644 lattice-net/src/endpoint.rs create mode 100644 lattice-net/src/framing.rs diff --git a/Cargo.toml b/Cargo.toml index ccd8d2b..2a298b4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ lattice-cli = { path = "lattice-cli" } rustyline = "17" # Networking (Iroh) -iroh = "0.95" +iroh = { version = "0.95", features = ["discovery-local-network"] } iroh-gossip = "0.95" # Cryptography @@ -37,6 +37,8 @@ prost-build = "0.13" # Async runtime tokio = { version = "1", features = ["full"] } +tokio-util = { version = "0.7", features = ["codec"] } +futures-util = "0.3" # Utilities thiserror = "2" @@ -47,6 +49,7 @@ blake3 = "1" hex = "0.4" redb = "2" uuid = { version = "1", features = ["v4"] } +chrono = "0.4" # Testing tokio-test = "0.4" diff --git a/docs/architecture.md b/docs/architecture.md index d548b7d..3b427c5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -62,7 +62,14 @@ Networking modes: - `/nodes/{pubkey}/info` = static metadata (name, added_by, added_at) - `/nodes/{pubkey}/status` = `active` | `dormant` | `disabled` - `/nodes/{pubkey}/role` = `server` | `device` (optional, hints sync priority) -- Inviting a node = writing entries to `/nodes/{pubkey}/...`. + - `/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 + 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. - Liveness: Each node tracks `last_seen` locally (from watermark gossip). UI alerts if a peer hasn't been seen for threshold (e.g., 30 days). User decides to mark dormant/disabled. - Status effects: - `active`: Normal sync participant, blocks watermark until acknowledged. diff --git a/docs/roadmap.md b/docs/roadmap.md index 0be7c30..f50b4c3 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -94,17 +94,54 @@ - [x] Multi-store sync test: compute diff, fetch entries, apply, verify same state **Phase 2: Iroh Integration** + +*Completed:* - [x] Node info in root store on init: `/nodes/{pubkey}/info` + `/status` -- [ ] Iroh integration (peer discovery, connection) -- [ ] Sync protocol (push missing entries over network) -- [ ] CLI: `peers`, `connect`/`join` commands -- [ ] Background sync task (tokio::spawn) +- [x] CLI: `invite ` to authorize peers +- [x] CLI: `peers` to list known nodes (with name/added_at info, sorted) +- [x] CLI: `remove ` to remove a peer +- [x] Iroh endpoint on startup (same Ed25519 key, mDNS + DNS discovery) +- [x] CLI: `join ` - connects to peer, verifies invited +- [x] Peer verification via `/nodes/{pubkey}/status` check + +*Join Protocol (new→existing):* ✓ +- [x] Proto: `JoinRequest` / `JoinResponse` with store UUID +- [x] Accept handler sends root store UUID in response +- [x] Join command creates empty store with received UUID (no writes until sync) + +*Sync Protocol (bidirectional):* ✓ +- [x] Proto: `PeerMessage` wrapper with `oneof` for message type discrimination +- [x] `framing.rs` with `MessageSink`/`MessageStream` using `LengthDelimitedCodec` +- [x] Proto: `SyncRequest`/`SyncResponse` using `SyncState` +- [x] `Store::read_entries_after(hash)` to fetch log chunks +- [x] Accept handler: receive SyncState, compute diff, send missing entries +- [x] Sync command: receive entries, apply to store via `apply_entry` +- [x] CLI: `sync [nodeid]` command (syncs with all active peers if no nodeid) +- [x] After sync: node updates own `/nodes/{pubkey}/info` with hostname + +*Cleanup*: +- [x] Move core logic from cmd_join and cmd_sync out of commands.rs (now in `sync.rs`) +- [x] Add 'invited' state: invite sets 'invited', peer sets 'active' after sync + +*Regressions:* +- [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 ### Success Criteria - Node A writes, Node B syncs, both have same state - Works offline-first (sync when connected) +**Post-M2 Refactoring:** +- [ ] Unify `node.rs` from `lattice-cli` and `lattice-core` +- [ ] Move `Store` code into `lattice-store` crate +- [ ] Move `StoreActor` code into `lattice-store` crate + --- ## Milestone 3: Multi-Node Mesh diff --git a/lattice-cli/Cargo.toml b/lattice-cli/Cargo.toml index 02cbb53..758f5ed 100644 --- a/lattice-cli/Cargo.toml +++ b/lattice-cli/Cargo.toml @@ -11,6 +11,7 @@ path = "src/main.rs" [dependencies] lattice-core = { workspace = true } +lattice-net = { workspace = true } rustyline = { workspace = true } hex = { workspace = true } thiserror = { workspace = true } @@ -18,3 +19,8 @@ tokio = { workspace = true } shlex = "1" hostname = "0.4" serde_json = "1" +iroh = { workspace = true } +prost = { workspace = true } +chrono = { workspace = true } +tokio-util = { version = "0.7", features = ["codec"] } +futures-util = "0.3" diff --git a/lattice-cli/src/accept_handler.rs b/lattice-cli/src/accept_handler.rs new file mode 100644 index 0000000..fb65ca8 --- /dev/null +++ b/lattice-cli/src/accept_handler.rs @@ -0,0 +1,165 @@ +//! Accept handler for incoming Iroh connections + +use lattice_net::{MessageSink, MessageStream}; +use crate::node::{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}; + +/// Spawn the accept loop for incoming connections. +pub fn spawn_accept_loop( + 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(); + tokio::spawn(async move { + if let Err(e) = handle_connection(conn, store).await { + eprintln!("[Accept] Error: {}", e); + } + }); + } + Err(e) => eprintln!("[Accept] Handshake error: {:?}", e), + } + } + } + }); +} + +/// Handle a single incoming connection +async fn handle_connection( + 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()), + } + }; + + 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 mut stream = MessageStream::new(recv); + + // Read first message + 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(()) + } + 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 + } + _ => 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)), + }; + + let valid = match expected { + PeerStatusCheck::Exactly(ps) => status == ps.as_str(), + PeerStatusCheck::ActiveOrInvited => status == PeerStatus::Active.as_str() || status == PeerStatus::Invited.as_str(), + }; + + if valid { + Ok(()) + } else { + Err(format!("Peer status is '{}', expected {:?}", status, expected)) + } +} + +/// Handle a sync request - bidirectional exchange of entries +async fn handle_sync_request( + mut sink: MessageSink, + mut stream: MessageStream, + peer_request: lattice_core::proto::SyncRequest, + store: &StoreHandle, +) -> Result<(), String> { + println!("[Sync] Received SyncRequest"); + + // Get our sync state + let my_state = store.sync_state().await + .map_err(|e| format!("Failed to get sync state: {}", e))?; + + // 1. Send our sync state as response + let resp = PeerMessage { + message: Some(peer_message::Message::SyncResponse(lattice_core::proto::SyncResponse { + state: Some(my_state.to_proto()), + })), + }; + sink.send(&resp).await?; + + // 2. Send entries peer is missing + let peer_state = peer_request.state + .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?; + 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?; + + sink.finish().await?; + + println!("[Sync] Applied {} entries from peer", entries_applied); + Ok(()) +} + diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index 5431182..c87f89f 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -1,7 +1,9 @@ //! CLI command handlers -use crate::node::{LatticeNode, StoreHandle}; +use crate::node::{LatticeNode, StoreHandle, PeerStatus}; use lattice_core::Uuid; +use lattice_net::LatticeEndpoint; +use chrono::DateTime; use std::time::Instant; /// Result of a command that may switch stores @@ -19,7 +21,7 @@ fn block_async(f: F) -> F::Output { }) } -pub type Handler = fn(&LatticeNode, Option<&StoreHandle>, &[String]) -> CommandResult; +pub type Handler = fn(&LatticeNode, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult; pub struct Command { pub name: &'static str, @@ -112,6 +114,46 @@ pub fn commands() -> Vec { max_args: 1, handler: cmd_author_state, }, + Command { + name: "invite", + args: "", + description: "Invite a peer node (writes to root store)", + min_args: 1, + max_args: 1, + handler: cmd_invite, + }, + Command { + name: "peers", + args: "", + description: "List known peers from root store", + min_args: 0, + max_args: 0, + handler: cmd_peers, + }, + Command { + name: "remove", + args: "", + description: "Remove a peer (set status to removed)", + min_args: 1, + max_args: 1, + handler: cmd_remove, + }, + Command { + name: "join", + args: "", + description: "Join a mesh by connecting to a peer (requires no local store)", + min_args: 1, + max_args: 1, + handler: cmd_join, + }, + Command { + name: "sync", + args: "[nodeid]", + description: "Sync entries with a peer (or all peers if none specified)", + min_args: 0, + max_args: 1, + handler: cmd_sync, + }, Command { name: "help", args: "", @@ -125,7 +167,7 @@ pub fn commands() -> Vec { // --- Store management --- -fn cmd_init(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { +fn cmd_init(node: &LatticeNode, _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); @@ -139,7 +181,7 @@ fn cmd_init(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) } } -fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { +fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { match node.create_store() { Ok(store_id) => { println!("Created store: {}", store_id); @@ -161,7 +203,7 @@ fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[S } } -fn cmd_use_store(node: &LatticeNode, _store: Option<&StoreHandle>, args: &[String]) -> CommandResult { +fn cmd_use_store(node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store_id = match Uuid::parse_str(&args[0]) { Ok(id) => id, Err(_) => { @@ -187,7 +229,7 @@ fn cmd_use_store(node: &LatticeNode, _store: Option<&StoreHandle>, args: &[Strin } } -fn cmd_list_stores(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { +fn cmd_list_stores(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { let stores = match node.list_stores() { Ok(s) => s, Err(e) => { @@ -210,7 +252,7 @@ fn cmd_list_stores(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[Str // --- Info --- -fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { +fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { println!("\nCommands:"); for cmd in commands() { if cmd.args.is_empty() { @@ -224,7 +266,7 @@ fn cmd_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) CommandResult::Ok } -fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) -> CommandResult { +fn cmd_status(node: &LatticeNode, 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() { @@ -236,6 +278,40 @@ fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) println!("Store: {}", h.id()); println!("Log Seq: {}", block_async(h.log_seq())); println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0)); + + // Show sync state summary + if let Ok(sync_state) = block_async(h.sync_state()) { + let authors = sync_state.authors(); + let total_entries: u64 = authors.values().map(|a| a.seq).sum(); + let num_authors = authors.len(); + println!("Authors: {} ({} total entries)", num_authors, total_entries); + + // Show per-author details + for (author, info) in authors { + println!(" {}...: seq={}, heads={}", + hex::encode(&author[..6]), + info.seq, + info.heads.len()); + } + } + + // Show log directory size + let logs_dir = node.data_path().join("stores").join(h.id().to_string()).join("logs"); + if logs_dir.exists() { + let mut total_size = 0u64; + let mut file_count = 0; + if let Ok(entries) = std::fs::read_dir(&logs_dir) { + for entry in entries.flatten() { + if let Ok(meta) = entry.metadata() { + if meta.is_file() { + total_size += meta.len(); + file_count += 1; + } + } + } + } + println!("Logs: {} files, {} bytes", file_count, total_size); + } } else { println!("Store: (none)"); } @@ -244,7 +320,7 @@ fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) // --- KV --- -fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult { +fn cmd_put(_node: &LatticeNode, 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; @@ -257,7 +333,7 @@ fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult::Ok } -fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult { +fn cmd_get(_node: &LatticeNode, 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; @@ -308,7 +384,7 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult::Ok } -fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult { +fn cmd_delete(_node: &LatticeNode, 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; @@ -321,7 +397,7 @@ fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) CommandResult::Ok } -fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult { +fn cmd_list(_node: &LatticeNode, 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; @@ -351,7 +427,13 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) - } } } else { - println!("{} = {}", key_str, format_value(v)); + // Check for multiple heads + let heads = block_async(h.get_heads(k)).unwrap_or_default(); + if heads.len() > 1 { + println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len()); + } else { + println!("{} = {}", key_str, format_value(v)); + } } } println!("({} keys, {:.2?})", entries.len(), start.elapsed()); @@ -366,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>, args: &[String]) -> CommandResult { +fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { let store = match store { Some(s) => s, None => { @@ -407,3 +489,285 @@ fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, args: &[Str } CommandResult::Ok } + +// --- Peer management --- + +fn cmd_invite(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let store = match store { + Some(s) => s, + None => { + eprintln!("Not in a store. Run 'use' or 'init' first."); + return CommandResult::Ok; + } + }; + + let pubkey_hex = &args[0]; + let _peer_pubkey = match hex::decode(pubkey_hex) { + Ok(bytes) if bytes.len() == 32 => bytes, + _ => { + eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)"); + return CommandResult::Ok; + } + }; + + // Write /nodes/{pubkey}/info with inviter info + let info_key = format!("/nodes/{}/info", pubkey_hex); + 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())) { + Ok(_) => {} + Err(e) => { + eprintln!("Error writing info: {}", e); + return CommandResult::Ok; + } + } + + // Write /nodes/{pubkey}/status = invited (becomes active after sync) + let status_key = format!("/nodes/{}/status", pubkey_hex); + match block_async(store.put(status_key.as_bytes(), PeerStatus::Invited.as_str().as_bytes())) { + Ok(_) => {} + Err(e) => { + eprintln!("Error writing status: {}", e); + return CommandResult::Ok; + } + } + + println!("Invited peer: {}", pubkey_hex); + println!(" /nodes/{}/info", 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 { + let store = match store { + Some(s) => s, + None => { + eprintln!("Not in a store. Run 'use' or 'init' first."); + return CommandResult::Ok; + } + }; + + // List all keys under /nodes/ + let all = match block_async(store.list()) { + Ok(v) => v, + Err(e) => { + eprintln!("Error listing: {}", e); + return CommandResult::Ok; + } + }; + + // Collect unique pubkeys with status + let mut peers: std::collections::HashMap = std::collections::HashMap::new(); + for (key, value) in &all { + let key_str = String::from_utf8_lossy(key); + if key_str.ends_with("/status") { + if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) { + let status_str = String::from_utf8_lossy(value); + if let Some(status) = PeerStatus::from_str(&status_str) { + peers.insert(pubkey.to_string(), status); + } + } + } + } + + if peers.is_empty() { + println!("No peers found."); + } else { + // Group peers by status + let mut by_status: std::collections::HashMap> = + 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(); + + 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(); + } + } + } + } + + by_status.entry(*status) + .or_default() + .push((pubkey.clone(), name, added)); + } + + // Print grouped by status in order: active, invited, removed + let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed]; + for status in &status_order { + if let Some(peer_list) = by_status.get(status) { + println!("\n[{}] ({}):", status.as_str(), peer_list.len()); + let mut sorted = peer_list.clone(); + sorted.sort_by(|a, b| a.0.cmp(&b.0)); + for (pubkey, name, added) in &sorted { + let info_str = match (name.is_empty(), added.is_empty()) { + (false, false) => format!(" {} ({})", name, added), + (false, true) => format!(" {}", name), + (true, false) => format!(" ({})", added), + _ => String::new(), + }; + println!(" {}{}", pubkey, info_str); + } + } + } + } + CommandResult::Ok +} + +fn cmd_remove(node: &LatticeNode, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let store = match store { + Some(s) => s, + None => { + eprintln!("Not in a store. Run 'use' or 'init' first."); + return CommandResult::Ok; + } + }; + + let pubkey_hex = &args[0]; + + // Validate pubkey format (should be 64 hex chars) + if pubkey_hex.len() != 64 || !pubkey_hex.chars().all(|c| c.is_ascii_hexdigit()) { + eprintln!("Invalid pubkey: expected 64 hex characters"); + return CommandResult::Ok; + } + + // Prevent self-removal + let my_pubkey = hex::encode(node.node_id()); + if pubkey_hex == &my_pubkey { + eprintln!("Cannot remove yourself."); + return CommandResult::Ok; + } + + // Check if peer exists + let status_key = format!("/nodes/{}/status", pubkey_hex); + match block_async(store.get(status_key.as_bytes())) { + Ok(Some(status)) => { + if status == PeerStatus::Removed.as_str().as_bytes() { + println!("Peer {} is already removed.", &pubkey_hex[..10]); + return CommandResult::Ok; + } + } + Ok(None) => { + eprintln!("Peer {} not found.", &pubkey_hex[..10]); + return CommandResult::Ok; + } + Err(e) => { + eprintln!("Error checking peer: {}", e); + return CommandResult::Ok; + } + } + + // Set status to removed + match block_async(store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes())) { + Ok(_) => println!("Removed peer: {}...", &pubkey_hex[..10]), + Err(e) => eprintln!("Error removing peer: {}", e), + } + + CommandResult::Ok +} + +fn cmd_join(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let endpoint = match endpoint { + Some(ep) => ep, + None => { + eprintln!("Iroh endpoint not started."); + return CommandResult::Ok; + } + }; + + if store.is_some() { + eprintln!("Already initialized. Use 'sync' to sync with peers."); + return CommandResult::Ok; + } + + let peer_id = match lattice_net::parse_node_id(&args[0]) { + Ok(id) => id, + Err(e) => { + eprintln!("Invalid node ID: {}", e); + return CommandResult::Ok; + } + }; + + println!("Joining mesh via {}...", peer_id.fmt_short()); + + match block_async(crate::sync::join_mesh(node, endpoint, peer_id)) { + Ok(handle) => { + println!("Joined mesh! Use 'sync' command to sync entries."); + CommandResult::SwitchTo(handle) + } + Err(e) => { + eprintln!("Join failed: {}", e); + CommandResult::Ok + } + } +} + +fn cmd_sync(node: &LatticeNode, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let endpoint = match endpoint { + Some(ep) => ep, + None => { + eprintln!("Iroh endpoint not started."); + return CommandResult::Ok; + } + }; + + let store = match store { + Some(s) => s, + None => { + eprintln!("No store open. Use 'init' or 'join' first."); + return CommandResult::Ok; + } + }; + + if args.is_empty() { + // Sync with all active peers + match block_async(crate::sync::sync_all(node, endpoint, store)) { + Ok(results) => { + if results.is_empty() { + println!("No peers to sync with."); + } else { + let total: u64 = results.iter().map(|r| r.entries_applied).sum(); + println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len()); + } + } + Err(e) => eprintln!("Sync failed: {}", e), + } + } else { + // Sync with specific peer + let peer_id = match lattice_net::parse_node_id(&args[0]) { + Ok(id) => id, + Err(e) => { + eprintln!("Invalid node ID: {}", e); + return CommandResult::Ok; + } + }; + + println!("Syncing with {}...", peer_id.fmt_short()); + match block_async(crate::sync::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); + } + Err(e) => eprintln!("Sync failed: {}", e), + } + } + + CommandResult::Ok +} diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index ea7c151..2637591 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -1,13 +1,19 @@ //! 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 commands::CommandResult; use node::{LatticeNodeBuilder, StoreHandle}; use rustyline::error::ReadlineError; use rustyline::DefaultEditor; +use std::sync::Arc; +use tokio::sync::RwLock; #[tokio::main] async fn main() { @@ -22,6 +28,26 @@ 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) + } + Err(e) => { + eprintln!("Warning: Iroh failed to start: {}", e); + None + } + }; + + // 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()); + } + let info = node.info(); println!("Node ID: {}", info.node_id); println!("Data: {}", info.data_path); @@ -37,6 +63,8 @@ async fn main() { } else { println!("Root: {}", open_info.store_id); } + // Update shared store for accept loop + *shared_store.write().await = Some(h.clone()); Some(h) } Ok(None) => { @@ -86,9 +114,13 @@ 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(), cmd_args) { + match (cmd.handler)(&node, current_store.as_ref(), endpoint.as_ref(), cmd_args) { CommandResult::Ok => {} - CommandResult::SwitchTo(h) => current_store = Some(h), + CommandResult::SwitchTo(h) => { + // Update shared store for accept loop + *shared_store.write().await = Some(h.clone()); + current_store = Some(h); + } } } } diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs index 280f4fd..fd3f33a 100644 --- a/lattice-cli/src/node.rs +++ b/lattice-cli/src/node.rs @@ -41,6 +41,36 @@ pub enum NodeError { 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, @@ -107,6 +137,11 @@ impl LatticeNode { 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() } @@ -153,7 +188,7 @@ impl LatticeNode { // Write status = active let status_key = format!("/nodes/{}/status", pubkey_hex); - handle.put(status_key.as_bytes(), b"active").await?; + handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?; Ok((store_id, handle)) } @@ -164,6 +199,21 @@ impl LatticeNode { 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)?; @@ -216,6 +266,16 @@ pub struct StoreHandle { 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 } @@ -276,6 +336,36 @@ impl StoreHandle { .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(); @@ -300,12 +390,12 @@ impl StoreHandle { impl Drop for StoreHandle { fn drop(&mut self) { - // Send shutdown command (non-blocking) and wait for actor to finish - // Use try_send to avoid panic in async context - let _ = self.tx.try_send(crate::store_actor::StoreCmd::Shutdown); + // 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 } } diff --git a/lattice-cli/src/store_actor.rs b/lattice-cli/src/store_actor.rs index 2bcc43d..041fe68 100644 --- a/lattice-cli/src/store_actor.rs +++ b/lattice-cli/src/store_actor.rs @@ -1,7 +1,7 @@ //! Store Actor - dedicated thread that owns Store and processes commands via channel use lattice_core::{ - EntryBuilder, HeadInfo, Node, SigChain, Store, Uuid, + EntryBuilder, HeadInfo, Node, SigChain, SigChainManager, Store, Uuid, hlc::HLC, proto::AuthorState, sigchain::SigChainError, @@ -42,6 +42,19 @@ pub enum StoreCmd { author: [u8; 32], resp: oneshot::Sender, StoreError>>, }, + // Sync-related commands + SyncState { + resp: oneshot::Sender>, + }, + ReadEntriesAfter { + author: [u8; 32], + from_hash: Option<[u8; 32]>, + resp: oneshot::Sender, StoreError>>, + }, + ApplyEntry { + entry: lattice_core::proto::SignedEntry, + resp: oneshot::Sender>, + }, Shutdown, } @@ -74,11 +87,11 @@ impl std::fmt::Display for StoreActorError { impl std::error::Error for StoreActorError {} -/// The store actor - runs in its own thread, owns Store and SigChain +/// The store actor - runs in its own thread, owns Store and SigChainManager pub struct StoreActor { store_id: Uuid, store: Store, - sigchain: SigChain, + chain_manager: SigChainManager, // Manages all authors' sigchains node: Node, rx: mpsc::Receiver, } @@ -92,10 +105,21 @@ impl StoreActor { node: Node, rx: mpsc::Receiver, ) -> Self { + // Derive logs_dir from sigchain's log file path + let logs_dir = sigchain.log_path() + .parent() + .map(|p| p.to_path_buf()) + .unwrap_or_default(); + + // Create chain manager and register the local node's sigchain + let mut chain_manager = SigChainManager::new(&logs_dir, *store_id.as_bytes()); + let local_author = node.public_key_bytes(); + chain_manager.get_or_create(local_author); // Pre-initialize local chain + Self { store_id, store, - sigchain, + chain_manager, node, rx, } @@ -124,7 +148,11 @@ impl StoreActor { let _ = resp.send(result); } StoreCmd::LogSeq { resp } => { - let _ = resp.send(self.sigchain.len()); + let local_author = self.node.public_key_bytes(); + let len = self.chain_manager.get(&local_author) + .map(|c| c.len()) + .unwrap_or(0); + let _ = resp.send(len); } StoreCmd::AppliedSeq { resp } => { let author = self.node.public_key_bytes(); @@ -135,6 +163,25 @@ impl StoreActor { StoreCmd::AuthorState { author, resp } => { let _ = resp.send(self.store.author_state(&author)); } + StoreCmd::SyncState { resp } => { + let _ = resp.send(self.store.sync_state()); + } + StoreCmd::ReadEntriesAfter { author, from_hash, resp } => { + // Read entries from the log file for this author + let result = self.do_read_entries_after(&author, from_hash); + let _ = resp.send(result); + } + StoreCmd::ApplyEntry { entry, resp } => { + // Use SigChainManager to append to the correct author's log + if let Err(e) = self.chain_manager.append_entry(&entry) { + let _ = resp.send(Err(StoreError::from(e))); + continue; + } + + // Then apply to store + let result = self.store.apply_entry(&entry); + let _ = resp.send(result); + } StoreCmd::Shutdown => { break; } @@ -147,7 +194,8 @@ impl StoreActor { // Idempotency check (pure function) if !Store::needs_put(&heads, value) { - return Ok(self.sigchain.len()); // Idempotent, no new entry + let local_author = self.node.public_key_bytes(); + return Ok(self.chain_manager.get(&local_author).map(|c| c.len()).unwrap_or(0)); } let parent_hashes: Vec> = heads.iter().map(|h| h.hash.clone()).collect(); @@ -159,7 +207,8 @@ impl StoreActor { // Idempotency check (pure function) if !Store::needs_delete(&heads) { - return Ok(self.sigchain.len()); // Idempotent, no new entry + let local_author = self.node.public_key_bytes(); + return Ok(self.chain_manager.get(&local_author).map(|c| c.len()).unwrap_or(0)); } let parent_hashes: Vec> = heads.iter().map(|h| h.hash.clone()).collect(); @@ -170,8 +219,11 @@ impl StoreActor { where F: FnOnce(EntryBuilder) -> EntryBuilder, { - let seq = self.sigchain.len() + 1; - let prev_hash = self.sigchain.last_hash(); + let local_author = self.node.public_key_bytes(); + let sigchain = self.chain_manager.get_or_create(local_author); + + let seq = sigchain.len() + 1; + let prev_hash = *sigchain.last_hash(); let builder = EntryBuilder::new(seq, HLC::now()) .store_id(self.store_id.as_bytes().to_vec()) @@ -179,11 +231,31 @@ impl StoreActor { .parent_hashes(parent_hashes); let entry = build(builder).sign(&self.node); - self.sigchain.append(&entry)?; + // Append to local sigchain + let sigchain = self.chain_manager.get_or_create(local_author); + sigchain.append(&entry)?; self.store.apply_entry(&entry)?; Ok(seq) } + + fn do_read_entries_after( + &self, + author: &[u8; 32], + from_hash: Option<[u8; 32]>, + ) -> 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)); + + if !log_path.exists() { + return Ok(Vec::new()); // No log file for this author + } + + // Use lattice_core's read_entries_after + lattice_core::log::read_entries_after(&log_path, from_hash) + .map_err(StoreError::from) + } } /// Spawn a store actor in a new thread, returns (sender, join_handle) diff --git a/lattice-cli/src/sync.rs b/lattice-cli/src/sync.rs new file mode 100644 index 0000000..e33bed8 --- /dev/null +++ b/lattice-cli/src/sync.rs @@ -0,0 +1,217 @@ +//! Sync networking operations for LatticeNode +//! +//! Provides async methods for joining meshes and syncing with peers. + +use lattice_net::{MessageSink, MessageStream}; +use crate::node::{LatticeNode, NodeError, StoreHandle, PeerStatus}; +use lattice_core::proto::{peer_message, PeerMessage, JoinRequest, SignedEntry}; +use lattice_net::LatticeEndpoint; +use prost::Message; + +/// 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: &LatticeNode, + 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()))?; + + // 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)?; + + // Immediately sync with the peer to get initial data + println!("[Join] Syncing with peer to get initial data..."); + match sync_with_peer(node, 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); + // Don't fail join, just warn - peer might not have data yet + } + } + + 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( + node: &LatticeNode, + 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 { + 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 = crate::sync_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 { + entries_applied, + entries_sent_by_peer, + }) +} + +/// Sync with all active peers from the store. +pub async fn sync_all( + node: &LatticeNode, + endpoint: &LatticeEndpoint, + store: &StoreHandle, +) -> 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(); + + 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) = lattice_net::parse_node_id(pubkey) { + peer_ids.push(id); + } + } + } + } + } + + // Sync with each peer + let mut results = Vec::new(); + for peer_id in peer_ids { + match sync_with_peer(node, 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/lattice-cli/src/sync_protocol.rs b/lattice-cli/src/sync_protocol.rs new file mode 100644 index 0000000..daa950b --- /dev/null +++ b/lattice-cli/src/sync_protocol.rs @@ -0,0 +1,89 @@ +//! 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). + +use crate::node::StoreHandle; +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; + +/// Send entries that peer is missing based on state diff. +/// Returns (entries_sent, optional_error). +pub async fn send_missing_entries( + sink: &mut MessageSink, + store: &StoreHandle, + my_state: &SyncState, + peer_state: &SyncState, +) -> Result { + let missing = peer_state.diff(my_state); + + // Build queues for each author's entries + let mut author_entries: Vec> = Vec::new(); + for range in missing { + let from_hash = if range.from_hash == [0u8; 32] { None } else { Some(range.from_hash) }; + let entries = store.read_entries_after(&range.author, from_hash).await + .map_err(|e| format!("Failed to read entries: {}", e))?; + if !entries.is_empty() { + author_entries.push(entries.into()); + } + } + + // Stream entries in HLC (causal) order + let mut entries_sent = 0u64; + for entry in lattice_core::CausalEntryIter::new(author_entries) { + let sync_msg = PeerMessage { + message: Some(peer_message::Message::SyncEntry(lattice_core::proto::SyncEntry { + signed_entry: entry.encode_to_vec(), + hash: vec![], + })), + }; + sink.send(&sync_msg).await?; + entries_sent += 1; + } + + // Send SyncDone + let done = PeerMessage { + message: Some(peer_message::Message::SyncDone(lattice_core::proto::SyncDone { + entries_sent, + })), + }; + sink.send(&done).await?; + + Ok(entries_sent) +} + +/// Receive and apply entries until SyncDone is received. +/// Returns (entries_applied, entries_reported_by_peer). +pub async fn receive_entries( + stream: &mut MessageStream, + store: &StoreHandle, +) -> Result<(u64, u64), String> { + let mut entries_applied = 0u64; + let mut entries_reported = 0u64; + + loop { + match stream.recv().await { + Ok(Some(msg)) => match msg.message { + 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_reported = done.entries_sent; + break; + } + _ => {} + } + Ok(None) => break, + Err(_) => break, + } + } + + Ok((entries_applied, entries_reported)) +} diff --git a/lattice-core/src/causal_iter.rs b/lattice-core/src/causal_iter.rs new file mode 100644 index 0000000..b49a436 --- /dev/null +++ b/lattice-core/src/causal_iter.rs @@ -0,0 +1,189 @@ +//! Causal Entry Iterator - yields entries in HLC (causal) order +//! +//! Implements merge-sort streaming across multiple author queues using a min-heap, +//! ensuring entries are returned in correct causal order for sync. +//! Complexity: O(N log K) where N = total entries, K = number of authors. + +use crate::proto::{Entry, SignedEntry}; +use prost::Message; +use std::cmp::Ordering; +use std::collections::{BinaryHeap, VecDeque}; + +/// A heap entry that wraps an author queue index and the HLC of its front entry. +/// Uses Reverse for min-heap behavior (lowest HLC first). +struct HeapEntry { + hlc: (u64, u32), + queue_idx: usize, +} + +impl PartialEq for HeapEntry { + fn eq(&self, other: &Self) -> bool { + self.hlc == other.hlc + } +} + +impl Eq for HeapEntry {} + +impl PartialOrd for HeapEntry { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for HeapEntry { + fn cmp(&self, other: &Self) -> Ordering { + // Reverse order for min-heap (BinaryHeap is max-heap by default) + other.hlc.cmp(&self.hlc) + } +} + +/// Iterator that yields SignedEntry in HLC (causal) order. +/// +/// Takes multiple VecDeques (one per author) and yields entries +/// from lowest to highest HLC, ensuring causal ordering for sync. +/// Uses a min-heap for O(log K) per-entry overhead instead of O(K) linear scan. +pub struct CausalEntryIter { + queues: Vec>, + heap: BinaryHeap, +} + +impl CausalEntryIter { + /// Create a new iterator from a list of entry queues (one per author) + pub fn new(queues: Vec>) -> Self { + let mut heap = BinaryHeap::with_capacity(queues.len()); + + // Initialize heap with the front entry from each non-empty queue + for (idx, queue) in queues.iter().enumerate() { + if let Some(entry) = queue.front() { + heap.push(HeapEntry { + hlc: Self::get_hlc(entry), + queue_idx: idx, + }); + } + } + + Self { queues, heap } + } + + /// Extract HLC (wall_time, counter) from a SignedEntry + fn get_hlc(entry: &SignedEntry) -> (u64, u32) { + Entry::decode(&entry.entry_bytes[..]) + .ok() + .and_then(|e| e.timestamp) + .map(|t| (t.wall_time, t.counter)) + .unwrap_or((0, 0)) + } +} + +impl Iterator for CausalEntryIter { + type Item = SignedEntry; + + fn next(&mut self) -> Option { + // Pop the queue with lowest HLC + let HeapEntry { queue_idx, .. } = self.heap.pop()?; + + // Remove entry from that queue + let entry = self.queues[queue_idx].pop_front()?; + + // If queue still has entries, push its new front back to heap + if let Some(next_entry) = self.queues[queue_idx].front() { + self.heap.push(HeapEntry { + hlc: Self::get_hlc(next_entry), + queue_idx, + }); + } + + Some(entry) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hlc::HLC; + use crate::clock::MockClock; + use crate::node::Node; + use crate::signed_entry::EntryBuilder; + + fn make_entry(node: &Node, 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]) + .prev_hash(vec![0u8; 32]) + .put(b"/test".to_vec(), format!("seq{}", seq).into_bytes()) + .sign(node) + } + + #[test] + fn test_empty_iter() { + let iter = CausalEntryIter::new(vec![]); + assert_eq!(iter.count(), 0); + } + + #[test] + fn test_single_queue() { + let node = Node::generate(); + let entries: VecDeque<_> = vec![ + make_entry(&node, 1, 1000), + make_entry(&node, 2, 2000), + ].into(); + + let iter = CausalEntryIter::new(vec![entries]); + let result: Vec<_> = iter.collect(); + assert_eq!(result.len(), 2); + } + + #[test] + fn test_merge_multiple_queues() { + let node_a = Node::generate(); + let node_b = Node::generate(); + + // Author A: entries at time 1000, 3000 + let queue_a: VecDeque<_> = vec![ + make_entry(&node_a, 1, 1000), + make_entry(&node_a, 2, 3000), + ].into(); + + // Author B: entries at time 2000 + let queue_b: VecDeque<_> = vec![ + make_entry(&node_b, 1, 2000), + ].into(); + + let iter = CausalEntryIter::new(vec![queue_a, queue_b]); + let result: Vec<_> = iter.collect(); + + // Should be in HLC order: 1000, 2000, 3000 + assert_eq!(result.len(), 3); + + // Verify order by checking HLC values + let hlcs: Vec<_> = result.iter() + .map(|e| CausalEntryIter::get_hlc(e)) + .collect(); + assert_eq!(hlcs[0].0, 1000); + assert_eq!(hlcs[1].0, 2000); + assert_eq!(hlcs[2].0, 3000); + } + + #[test] + fn test_many_authors() { + // Test with 10 authors to verify heap behavior + let nodes: Vec<_> = (0..10).map(|_| Node::generate()).collect(); + let queues: Vec> = nodes.iter().enumerate().map(|(i, node)| { + vec![make_entry(node, 1, (i * 100 + 50) as u64)].into() + }).collect(); + + let iter = CausalEntryIter::new(queues); + let result: Vec<_> = iter.collect(); + + assert_eq!(result.len(), 10); + + // Verify strictly increasing HLC order + let hlcs: Vec<_> = result.iter() + .map(|e| CausalEntryIter::get_hlc(e).0) + .collect(); + for window in hlcs.windows(2) { + assert!(window[0] < window[1], "HLCs should be strictly increasing"); + } + } +} + diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index 0759cf5..ff24a06 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -12,6 +12,7 @@ //! - **SignedEntry**: Entry creation, signing, and verification //! - **Log**: Append-only log file I/O //! - **Store**: Persistent KV state from log replay +//! - **CausalIter**: Merge-sort iterator for HLC-ordered sync pub mod node; pub mod sigchain; @@ -25,13 +26,14 @@ pub mod signed_entry; pub mod log; pub mod store; pub mod meta_store; +pub mod causal_iter; // Constants /// Maximum size of a serialized SignedEntry (16 MB) pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024; pub use node::Node; -pub use sigchain::SigChain; +pub use sigchain::{SigChain, SigChainManager}; pub use entry::Entry; pub use sync_state::{SyncState, AuthorInfo, MissingRange}; pub use hlc::HLC; @@ -43,3 +45,5 @@ pub use store::Store; pub use meta_store::MetaStore; pub use proto::HeadInfo; pub use uuid::Uuid; +pub use causal_iter::CausalEntryIter; + diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index 96958cc..e7d58d6 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -101,6 +101,12 @@ impl Node { &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) diff --git a/lattice-core/src/sigchain.rs b/lattice-core/src/sigchain.rs index d81fb5b..2b56155 100644 --- a/lattice-core/src/sigchain.rs +++ b/lattice-core/src/sigchain.rs @@ -146,6 +146,11 @@ impl SigChain { &self.last_hash } + /// Get the log file path + pub fn log_path(&self) -> &std::path::Path { + &self.log_path + } + /// Get the current length of the chain pub fn len(&self) -> u64 { self.next_seq - 1 @@ -248,6 +253,70 @@ impl SigChain { } } +/// Manages multiple SigChains (one per author) for a store. +/// Provides unified interface for appending entries from any author. +pub struct SigChainManager { + /// Directory containing log files (one per author) + logs_dir: PathBuf, + + /// Store UUID (16 bytes) + store_id: [u8; 16], + + /// Cache of loaded SigChains by author + chains: std::collections::HashMap<[u8; 32], SigChain>, +} + +impl SigChainManager { + /// Create a new manager for a store's logs directory + pub fn new(logs_dir: impl AsRef, store_id: [u8; 16]) -> Self { + Self { + logs_dir: logs_dir.as_ref().to_path_buf(), + store_id, + chains: std::collections::HashMap::new(), + } + } + + /// Get or create a SigChain for an author + pub fn get_or_create(&mut self, author: [u8; 32]) -> &mut SigChain { + self.chains.entry(author).or_insert_with(|| { + let author_hex = hex::encode(author); + let log_path = self.logs_dir.join(format!("{}.log", author_hex)); + + // Try to load existing log, or create new + SigChain::from_log(&log_path, self.store_id, author) + .unwrap_or_else(|_| SigChain::new(&log_path, self.store_id, author)) + }) + } + + /// Get the local node's sigchain (for creating new entries) + pub fn get(&self, author: &[u8; 32]) -> Option<&SigChain> { + self.chains.get(author) + } + + /// Append an entry to the appropriate author's log + /// This is the main entry point for all entry writes (from put, sync, etc.) + pub fn append_entry(&mut self, entry: &SignedEntry) -> Result<(), SigChainError> { + let author: [u8; 32] = entry.author_id.clone() + .try_into() + .map_err(|_| SigChainError::WrongAuthor { + expected: "32 bytes".to_string(), + got: format!("{} bytes", entry.author_id.len()), + })?; + + // For synced entries, we can't validate seq/prev_hash since they may arrive + // out of order. Just append to the log file directly. + let chain = self.get_or_create(author); + append_entry(chain.log_path(), entry)?; + + Ok(()) + } + + /// Get the logs directory path + pub fn logs_dir(&self) -> &Path { + &self.logs_dir + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/lattice-core/src/store.rs b/lattice-core/src/store.rs index 49fa255..87d399f 100644 --- a/lattice-core/src/store.rs +++ b/lattice-core/src/store.rs @@ -8,6 +8,7 @@ use crate::log::{read_entries, LogError}; use crate::proto::{operation, AuthorState, Entry, HeadInfo, HeadList, SignedEntry}; +use crate::sigchain::SigChainError; use crate::signed_entry::hash_signed_entry; use prost::Message; use redb::{Database, ReadableTable, TableDefinition}; @@ -41,6 +42,9 @@ pub enum StoreError { #[error("Decode error: {0}")] Decode(#[from] prost::DecodeError), + + #[error("Sigchain error: {0}")] + SigChain(#[from] SigChainError), } /// Persistent store for KV state with DAG conflict resolution @@ -1469,4 +1473,203 @@ mod tests { let _ = std::fs::remove_file(path); } + + /// Test case for multi-node sync: 3 nodes create multi-heads, then merge, then sync to new node. + /// + /// Scenario: + /// 1. Node A, B, C each write to key "/a" independently (creating 3 heads) + /// 2. Node A does a final put to merge all heads + /// 3. After merge, node A should have only 1 head + /// 4. Simulate sync to new node D using SyncState diff + /// 5. Node D should end up with same state as A (1 head, not 3) + #[test] + fn test_multinode_sync_after_merge() { + let path_a = temp_db_path("multinode_a"); + let path_d = temp_db_path("multinode_d"); + let _ = std::fs::remove_file(&path_a); + let _ = std::fs::remove_file(&path_d); + + // Create stores + let store_a = Store::open(&path_a).unwrap(); + 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 clock = MockClock::new(1000); + + // 1. Each node writes to "/a" independently (simulating offline concurrent writes) + // Node A: seq 1 + let entry_a = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_a".to_vec()) + .sign(&node_a); + store_a.apply_entry(&entry_a).unwrap(); + + // Node B: seq 1 (different author, same key - creates fork) + let entry_b = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_b".to_vec()) + .sign(&node_b); + store_a.apply_entry(&entry_b).unwrap(); + + // Node C: seq 1 (third author, same key - creates third fork) + let entry_c = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_c".to_vec()) + .sign(&node_c); + store_a.apply_entry(&entry_c).unwrap(); + + // After applying all 3 entries, store_a has 3 heads for "/a" + let heads_before_merge = store_a.get_heads(b"/a").unwrap(); + assert_eq!(heads_before_merge.len(), 3, "Should have 3 heads before merge"); + + // 2. Node A does a final put referencing all heads (merge) + // Get the hashes of all current heads as parent_hashes + let parent_hashes: Vec> = heads_before_merge.iter() + .map(|h| h.hash.clone()) + .collect(); + + let merge_entry = EntryBuilder::new(2, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_signed_entry(&entry_a).to_vec()) // Continues A's chain + .parent_hashes(parent_hashes) // References all heads + .put("/a", b"merged".to_vec()) + .sign(&node_a); + store_a.apply_entry(&merge_entry).unwrap(); + + // After merge, should have only 1 head + let heads_after_merge = store_a.get_heads(b"/a").unwrap(); + assert_eq!(heads_after_merge.len(), 1, "Should have 1 head after merge"); + assert_eq!(heads_after_merge[0].value, b"merged"); + + // 3. Get sync state from store_a + let sync_state_a = store_a.sync_state().unwrap(); + + println!("Store A sync state:"); + for (author, info) in sync_state_a.authors() { + println!(" author {:?}: seq={}, heads={:?}", + hex::encode(&author[..4]), info.seq, + info.heads.iter().map(|h| hex::encode(&h[..4])).collect::>()); + } + + // 4. Store D is empty, compute diff + let sync_state_d = store_d.sync_state().unwrap(); + let missing = sync_state_d.diff(&sync_state_a); + + println!("Missing ranges: {:?}", missing.len()); + for m in &missing { + println!(" author {:?}: from_seq={}, to_seq={}", + hex::encode(&m.author[..4]), m.from_seq, m.to_seq); + } + + // We should get missing ranges for all authors that have entries + assert!(!missing.is_empty(), "Should have missing entries to sync"); + + // 5. Apply all entries to store_d (simulating sync) + // In a real sync, we'd read entries from logs, but for this test, + // we just apply the same entries in order + store_d.apply_entry(&entry_a).unwrap(); + store_d.apply_entry(&entry_b).unwrap(); + store_d.apply_entry(&entry_c).unwrap(); + store_d.apply_entry(&merge_entry).unwrap(); + + // 6. Check state on store_d + let heads_d = store_d.get_heads(b"/a").unwrap(); + println!("Store D heads count: {}", heads_d.len()); + for (i, h) in heads_d.iter().enumerate() { + println!(" head[{}]: value={:?}, author={}", i, String::from_utf8_lossy(&h.value), hex::encode(&h.author[..4])); + } + + // BUG CHECK: Store D should have same state as Store A (1 head, not 3) + assert_eq!(heads_d.len(), 1, + "BUG: Store D should have 1 head (merged) but has {} heads", heads_d.len()); + assert_eq!(heads_d[0].value, b"merged"); + + let _ = std::fs::remove_file(&path_a); + let _ = std::fs::remove_file(&path_d); + } + + /// Test what happens when entries are applied in "wrong" order. + /// This simulates the real sync bug where: + /// - Sync iterates by author + /// - Author A's entries (including merge) are sent first + /// - Author B and C's entries are sent after + /// - The merge entry arrives BEFORE the entries it merges! + #[test] + fn test_multinode_sync_wrong_order() { + let path = temp_db_path("wrongorder"); + let _ = std::fs::remove_file(&path); + + 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 clock = MockClock::new(1000); + + // Create entries (same as before) + let entry_a = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_a".to_vec()) + .sign(&node_a); + + let entry_b = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_b".to_vec()) + .sign(&node_b); + + let entry_c = EntryBuilder::new(1, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash([0u8; 32].to_vec()) + .put("/a", b"from_c".to_vec()) + .sign(&node_c); + + // We need the hashes for parent_hashes - compute them + let hash_a = hash_signed_entry(&entry_a); + let hash_b = hash_signed_entry(&entry_b); + let hash_c = hash_signed_entry(&entry_c); + + let merge_entry = EntryBuilder::new(2, HLC::now_with_clock(&clock)) + .store_id(TEST_STORE.to_vec()) + .prev_hash(hash_a.to_vec()) + .parent_hashes(vec![hash_a.to_vec(), hash_b.to_vec(), hash_c.to_vec()]) + .put("/a", b"merged".to_vec()) + .sign(&node_a); + + // Apply in WRONG order: A's chain first (entry_a + merge), then B, then C + // This is what happens in sync when iterating by author + println!("Applying entry_a (A seq 1)..."); + store.apply_entry(&entry_a).unwrap(); + + println!("Applying merge_entry (A seq 2) BEFORE B and C..."); + store.apply_entry(&merge_entry).unwrap(); + + println!("Applying entry_b (B seq 1)..."); + store.apply_entry(&entry_b).unwrap(); + + println!("Applying entry_c (C seq 1)..."); + store.apply_entry(&entry_c).unwrap(); + + // Check final state + let heads = store.get_heads(b"/a").unwrap(); + println!("Final heads count: {}", heads.len()); + for (i, h) in heads.iter().enumerate() { + println!(" head[{}]: value={:?}", i, String::from_utf8_lossy(&h.value)); + } + + assert_eq!(heads.len(), 3, + "Wrong order application creates 3 heads (expected - sync handles ordering)"); + + let _ = std::fs::remove_file(&path); + } } diff --git a/lattice-core/src/sync_state.rs b/lattice-core/src/sync_state.rs index d14f9ac..a4d16bc 100644 --- a/lattice-core/src/sync_state.rs +++ b/lattice-core/src/sync_state.rs @@ -1,21 +1,33 @@ //! Sync state for causality tracking and reconciliation -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; /// Author ID type (32-byte Ed25519 public key) pub type Author = [u8; 32]; -/// Per-author sync information (seq + hash for resume). +/// Per-author sync information: seq + all head hashes. #[derive(Debug, Clone, PartialEq, Eq)] pub struct AuthorInfo { pub seq: u64, - pub hash: [u8; 32], + pub heads: HashSet<[u8; 32]>, // All head hashes for this author } -/// Sync state tracking per-author sequence numbers and hashes. +impl AuthorInfo { + pub fn new(seq: u64, hash: [u8; 32]) -> Self { + let mut heads = HashSet::new(); + heads.insert(hash); + Self { seq, heads } + } + + pub fn with_heads(seq: u64, heads: HashSet<[u8; 32]>) -> Self { + Self { seq, heads } + } +} + +/// Sync state tracking per-author sequence numbers and head hashes. /// /// Used during reconciliation to identify missing entries between peers. -/// Each author's highest seen sequence number and hash is tracked. +/// Tracks all head hashes per author to handle forks correctly. #[derive(Debug, Clone, Default)] pub struct SyncState { authors: HashMap, @@ -26,7 +38,7 @@ pub struct SyncState { pub struct MissingRange { pub author: Author, pub from_seq: u64, // exclusive - we have up to this - pub from_hash: [u8; 32], // hash to resume reading after + pub from_hash: [u8; 32], // hash to resume reading after (zero = start) pub to_seq: u64, // inclusive - peer has up to this } @@ -47,10 +59,32 @@ impl SyncState { pub fn seq(&self, author: &Author) -> u64 { self.authors.get(author).map(|i| i.seq).unwrap_or(0) } + + /// Get head hashes for an author (returns empty set if not present). + pub fn heads(&self, author: &Author) -> HashSet<[u8; 32]> { + self.authors.get(author).map(|i| i.heads.clone()).unwrap_or_default() + } - /// Set the info for an author. + /// Set the info for an author (single hash convenience method). pub fn set(&mut self, author: Author, seq: u64, hash: [u8; 32]) { - self.authors.insert(author, AuthorInfo { seq, hash }); + self.authors.insert(author, AuthorInfo::new(seq, hash)); + } + + /// Set the info for an author with multiple heads. + pub fn set_heads(&mut self, author: Author, seq: u64, heads: HashSet<[u8; 32]>) { + self.authors.insert(author, AuthorInfo::with_heads(seq, heads)); + } + + /// Add a head hash for an author (updates seq if higher). + pub fn add_head(&mut self, author: Author, seq: u64, hash: [u8; 32]) { + if let Some(info) = self.authors.get_mut(&author) { + info.heads.insert(hash); + if seq > info.seq { + info.seq = seq; + } + } else { + self.set(author, seq, hash); + } } /// Get all authors and their info. @@ -61,18 +95,33 @@ impl SyncState { /// Compute what entries we're missing compared to a peer's state. /// /// Returns ranges of entries we need from the peer. - /// Each range includes the hash to resume reading after. + /// Compares hash sets when seq matches to detect forks. pub fn diff(&self, peer: &SyncState) -> Vec { let mut missing = Vec::new(); for (author, peer_info) in peer.authors() { let my_seq = self.seq(author); - if peer_info.seq > my_seq { - // We need entries from my_seq+1 to peer_info.seq - // Use our hash (or zero if we have nothing) as resume point - let from_hash = self.get(author) - .map(|i| i.hash) - .unwrap_or([0u8; 32]); + let my_heads = self.heads(author); + + // We need entries if: + // 1. Peer's seq is higher than ours, OR + // 2. Peer's seq equals ours but they have heads we don't (fork) + let need_entries = if peer_info.seq > my_seq { + true + } else if peer_info.seq == my_seq && my_seq > 0 { + // Same seq - check for forks (different hashes at same seq) + peer_info.heads.iter().any(|h| !my_heads.contains(h)) + } else { + false + }; + + if need_entries { + // Request from our common ancestor (or start if we have nothing) + let from_hash = if my_heads.is_empty() { + [0u8; 32] + } else { + *my_heads.iter().next().unwrap() + }; missing.push(MissingRange { author: *author, @@ -86,15 +135,66 @@ impl SyncState { missing } - /// Merge another sync state into this one (take max seq per author). + /// Merge another sync state into this one (union of heads, max seq). pub fn merge(&mut self, other: &SyncState) { for (author, info) in other.authors() { - let my_seq = self.seq(author); - if info.seq > my_seq { - self.set(*author, info.seq, info.hash); + if let Some(my_info) = self.authors.get_mut(author) { + // Union heads + for h in &info.heads { + my_info.heads.insert(*h); + } + // Take max seq + if info.seq > my_info.seq { + my_info.seq = info.seq; + } + } else { + self.authors.insert(*author, info.clone()); } } } + + /// Convert to proto message for network transmission + pub fn to_proto(&self) -> crate::proto::SyncState { + let frontiers = self.authors.iter().map(|(author, info)| { + crate::proto::Frontier { + author_id: author.to_vec(), + max_seq: info.seq, + head_hashes: info.heads.iter().map(|h| h.to_vec()).collect(), + } + }).collect(); + crate::proto::SyncState { + frontiers, + sender_hlc: None, + } + } + + /// Create from proto message + pub fn from_proto(proto: &crate::proto::SyncState) -> Self { + let mut state = Self::new(); + for frontier in &proto.frontiers { + if frontier.author_id.len() == 32 { + let mut author = [0u8; 32]; + author.copy_from_slice(&frontier.author_id); + + let mut heads = HashSet::new(); + for hash_bytes in &frontier.head_hashes { + if hash_bytes.len() == 32 { + let mut hash = [0u8; 32]; + hash.copy_from_slice(hash_bytes); + heads.insert(hash); + } + } + + if heads.is_empty() { + // Fallback: empty hash if no heads provided + heads.insert([0u8; 32]); + } + + state.set_heads(author, frontier.max_seq, heads); + } + } + state + } } #[cfg(test)] @@ -176,4 +276,50 @@ mod tests { assert_eq!(a.seq(&author1), 10); // kept a's value assert_eq!(a.seq(&author2), 8); // took b's value } + + /// This test documents a known issue: SyncState tracks only ONE hash per author, + /// but with forks/multi-heads, there could be multiple branches. + /// + /// Scenario: + /// - Author writes entry1 (hash=A) + /// - Two peers independently write entry2 and entry3 (both have prev=A) + /// - Peer1 has: entry1 -> entry2 (seq=2, hash=B) + /// - Peer2 has: entry1 -> entry3 (seq=2, hash=C) + /// - When Peer3 syncs with Peer1, SyncState says "I need entries after hash=B" + /// - But Peer2 only has entries after hash=A, so Peer3 never gets entry3! + /// + #[test] + fn test_multihead_sync_inconsistency() { + // This is a conceptual test showing the problem + // In reality, both forks would have seq=2 but different hashes + // SyncState can only track one, so the other branch gets lost + + let mut peer1_state = SyncState::new(); + let mut peer2_state = SyncState::new(); + let new_peer_state = SyncState::new(); + + let author = [1u8; 32]; + + // Both peers have seq=2, but different hashes (different forks) + peer1_state.set(author, 2, [0xBB; 32]); // entry1 -> entry2 + peer2_state.set(author, 2, [0xCC; 32]); // entry1 -> entry3 + + // New peer syncs with peer1 first + let missing_from_peer1 = new_peer_state.diff(&peer1_state); + assert_eq!(missing_from_peer1.len(), 1); + assert_eq!(missing_from_peer1[0].to_seq, 2); + + // After applying peer1's entries, new peer has seq=2, hash=BB + let mut after_peer1 = new_peer_state.clone(); + after_peer1.set(author, 2, [0xBB; 32]); + + // Now sync with peer2 - BUG: new peer thinks it's up to date! + let missing_from_peer2 = after_peer1.diff(&peer2_state); + + // This assertion FAILS - we get empty missing even though peer2 has entry3! + // The bug: peer2's seq=2 equals our seq=2, so we think we're in sync + // But peer2's hash=0xCC != our hash=0xBB - they have different entries! + assert!(!missing_from_peer2.is_empty(), + "BUG: SyncState misses peer2's fork because seq numbers match"); + } } diff --git a/lattice-net/Cargo.toml b/lattice-net/Cargo.toml index 263239d..f605899 100644 --- a/lattice-net/Cargo.toml +++ b/lattice-net/Cargo.toml @@ -9,10 +9,13 @@ license.workspace = true lattice-core = { workspace = true } iroh = { workspace = true } iroh-gossip = { workspace = true } +prost = { workspace = true } tokio = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } bytes = { workspace = true } +tokio-util = { workspace = true } +futures-util = { workspace = true } [dev-dependencies] tokio-test = { workspace = true } diff --git a/lattice-net/src/endpoint.rs b/lattice-net/src/endpoint.rs new file mode 100644 index 0000000..b4bf0f1 --- /dev/null +++ b/lattice-net/src/endpoint.rs @@ -0,0 +1,57 @@ +//! Iroh endpoint for network connectivity +//! +//! Creates an Iroh endpoint from the node's Ed25519 secret key, +//! ensuring the same identity is used for both Lattice and Iroh. +//! +//! Discovery: Uses both DNS (default) and mDNS (local network) + +use iroh::{Endpoint, endpoint::{BindError, Connection, ConnectError}}; +use iroh::discovery::mdns::MdnsDiscovery; +pub use iroh::PublicKey; + +/// ALPN protocol identifier for Lattice sync +pub const LATTICE_ALPN: &[u8] = b"lattice-sync/1"; + +/// Wrapper around Iroh endpoint with Lattice integration +pub struct LatticeEndpoint { + endpoint: Endpoint, +} + +impl LatticeEndpoint { + /// Create a new endpoint from Ed25519 secret key bytes (from identity.key) + /// Enables both DNS discovery (internet) and mDNS discovery (local network) + pub async fn new(secret_key_bytes: [u8; 32]) -> Result { + let secret_key = iroh::SecretKey::from_bytes(&secret_key_bytes); + + // mDNS for local network discovery + let mdns = MdnsDiscovery::builder(); + + let endpoint = Endpoint::builder() + .secret_key(secret_key) + .alpns(vec![LATTICE_ALPN.to_vec()]) + .discovery(mdns) // Add mDNS on top of default DNS + .bind() + .await?; + Ok(Self { endpoint }) + } + + /// Get the public key (same as Lattice pubkey, can be shared with peers) + pub fn public_key(&self) -> PublicKey { + self.endpoint.secret_key().public() + } + + /// Connect to a peer by their public key + pub async fn connect(&self, peer: PublicKey) -> Result { + self.endpoint.connect(peer, LATTICE_ALPN).await + } + + /// Accept an incoming connection + pub async fn accept(&self) -> Option { + self.endpoint.accept().await + } + + /// Get the underlying endpoint + pub fn endpoint(&self) -> &Endpoint { + &self.endpoint + } +} diff --git a/lattice-net/src/framing.rs b/lattice-net/src/framing.rs new file mode 100644 index 0000000..f1faa20 --- /dev/null +++ b/lattice-net/src/framing.rs @@ -0,0 +1,63 @@ +//! Message framing for Iroh streams using tokio-util LengthDelimitedCodec +//! +//! Provides a clean interface for sending/receiving length-prefixed PeerMessage +//! over QUIC streams without manual buffer management. + +use futures_util::{SinkExt, StreamExt}; +use lattice_core::proto::PeerMessage; +use prost::Message; +use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec}; + +/// Framed writer for sending PeerMessage over an Iroh SendStream +pub struct MessageSink { + inner: FramedWrite, +} + +impl MessageSink { + pub fn new(stream: iroh::endpoint::SendStream) -> Self { + Self { + inner: FramedWrite::new(stream, LengthDelimitedCodec::new()), + } + } + + /// Send a PeerMessage (length-prefixed) + pub async fn send(&mut self, msg: &PeerMessage) -> Result<(), String> { + let bytes = msg.encode_to_vec(); + self.inner.send(bytes.into()).await + .map_err(|e| format!("Send error: {}", e)) + } + + /// Finish the stream (signal we're done sending) + pub async fn finish(self) -> Result<(), String> { + let mut stream = self.inner.into_inner(); + let _ = stream.finish(); + stream.stopped().await.ok(); + Ok(()) + } +} + +/// Framed reader for receiving PeerMessage from an Iroh RecvStream +pub struct MessageStream { + inner: FramedRead, +} + +impl MessageStream { + pub fn new(stream: iroh::endpoint::RecvStream) -> Self { + Self { + inner: FramedRead::new(stream, LengthDelimitedCodec::new()), + } + } + + /// Receive next PeerMessage (or None if stream closed) + pub async fn recv(&mut self) -> Result, String> { + match self.inner.next().await { + Some(Ok(bytes)) => { + PeerMessage::decode(&bytes[..]) + .map(Some) + .map_err(|e| format!("Decode error: {}", e)) + } + Some(Err(e)) => Err(format!("Read error: {}", e)), + None => Ok(None), + } + } +} diff --git a/lattice-net/src/lib.rs b/lattice-net/src/lib.rs index 1c4c36f..9d85534 100644 --- a/lattice-net/src/lib.rs +++ b/lattice-net/src/lib.rs @@ -1,8 +1,21 @@ //! Lattice Networking //! //! Networking layer using Iroh: +//! - **Endpoint**: Network identity and connection management //! - **Gossip**: Broadcasting changes across the mesh //! - **Unicast**: Point-to-point communication for reconciliation +//! - **Framing**: Length-delimited message framing for QUIC streams +pub mod endpoint; pub mod gossip; pub mod unicast; +pub mod framing; + +pub use endpoint::{LatticeEndpoint, PublicKey}; +pub use framing::{MessageSink, MessageStream}; +pub use lattice_core::proto::{SyncRequest, SyncResponse, SyncEntry, SyncDone, SyncState, Frontier}; + +/// Parse a PublicKey (NodeId) from hex or base32 string +pub fn parse_node_id(s: &str) -> Result { + s.parse().map_err(|e| format!("{}", e)) +} diff --git a/proto/lattice.proto b/proto/lattice.proto index 4219427..ff9bc71 100644 --- a/proto/lattice.proto +++ b/proto/lattice.proto @@ -90,7 +90,7 @@ message SyncState { message Frontier { bytes author_id = 1; // Ed25519 public key (32 bytes) uint64 max_seq = 2; // Highest sequence number seen from this author - bytes last_hash = 3; // Hash of the last entry (for chain verification) + repeated bytes head_hashes = 3; // All head hashes for this author } // 5. Log File Record (wrapper for storage) @@ -98,3 +98,44 @@ message LogRecord { bytes hash = 1; // BLAKE3 hash of entry_bytes (32 bytes) bytes entry_bytes = 2; // Serialized SignedEntry } + +// 6. Join Protocol Messages (new node joining existing mesh) +message JoinRequest { + bytes node_pubkey = 1; // Joining node's public key (32 bytes) +} + +message JoinResponse { + bytes store_uuid = 1; // Root store UUID (16 bytes) for new node to create + bytes inviter_pubkey = 2; // Inviter's public key for verification +} + +// 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) +} + +message SyncResponse { + SyncState state = 1; // Responder's sync state +} + +message SyncEntry { + bytes signed_entry = 1; // Serialized SignedEntry + bytes hash = 2; // Hash for verification +} + +message SyncDone { + uint64 entries_sent = 1; +} + +// 8. Peer Message Wrapper (for proper message type discrimination) +message PeerMessage { + oneof message { + JoinRequest join_request = 1; + JoinResponse join_response = 2; + SyncRequest sync_request = 3; + SyncResponse sync_response = 4; + SyncEntry sync_entry = 5; + SyncDone sync_done = 6; + } +}