From 3e39f34383d71ad4282f9af0e8aef2d9a0695df8 Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Tue, 23 Dec 2025 02:51:28 +0100 Subject: [PATCH] feat: integrate iroh-gossip for mesh networking, using ALPN for protocol routing and node events to manage gossip topics. --- docs/roadmap.md | 43 ++++- lattice-cli/src/main.rs | 1 + lattice-core/src/lib.rs | 2 +- lattice-core/src/node.rs | 48 ++++- lattice-core/src/store_actor.rs | 20 ++- lattice-net/Cargo.toml | 3 + lattice-net/src/endpoint.rs | 5 +- lattice-net/src/lib.rs | 2 +- lattice-net/src/mesh/server.rs | 310 +++++++++++++++++++++++++++++--- 9 files changed, 388 insertions(+), 46 deletions(-) diff --git a/docs/roadmap.md b/docs/roadmap.md index 196923b..2ce6d8d 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -146,26 +146,53 @@ ### Deliverables -**Phase 1: LatticeServer Refactor** +**Phase 1: LatticeServer Refactor** ✓ - [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] Encapsulate accept loop inside `LatticeServer` (via Router + ProtocolHandler) - [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 - [ ] Track last sync time per peer -**Phase 2: Gossip Protocol** -- [ ] Proto: `GossipAnnounce` message with author + latest seq + HLC -- [ ] `LatticeServer::spawn_gossip_loop` for periodic announcements -- [ ] On receiving announce: detect missing entries, trigger sync -- [ ] Track last-seen per peer for staleness detection +**Phase 2: Gossip Protocol** ✓ (iroh-gossip) +- [x] Router handles both `lattice-sync/1` and `/iroh-gossip/1` ALPNs +- [x] `NodeEvent::RootStoreActivated` emitted when root store opens +- [x] Auto-join gossip topic on root store activation +- [x] Broadcast local entries to gossip topic on commit +- [x] Receive gossip entries and apply to store +- [x] Topic ID via `blake3::hash("lattice/{store_id}")` +- [ ] Gossip bootstrap peers from `/peers/` (needs Prefix Watch) + +**Next: Prefix Watch (reactive store updates)** +- [ ] `store.watch_prefix(prefix) -> Receiver` +- [ ] `WatchEvent::Put { key, value }` / `WatchEvent::Delete { key }` +- [ ] StoreActor tracks watchers per prefix, emits on matching put/delete +- [ ] LatticeServer uses `/peers/` watch to update gossip bootstrap peers dynamically +- [ ] Enables reactive patterns: config changes, presence, app-level subscriptions + +--- + +## Technical Debt + +**Logging** +- [ ] Replace `println!`/`eprintln!` with `tracing` crate (`tracing::info!`, `tracing::error!`) +- Standard in Rust async ecosystem, used by Iroh internally + +**Lifecycle Management (Zombie Tasks)** +- [ ] Spawned infinite loops (`spawn_node_event_listener`, `spawn_entry_forward_loop`, gossip receive loop) keep running if `LatticeServer` is dropped +- [ ] Use `tokio_util::sync::CancellationToken` or keep `JoinHandle`s for graceful shutdown + +**Error Handling** +- [ ] Replace `Result<..., String>` with `anyhow::Result` or define `LatticeNetError` enum +- String errors make it hard to handle specific failure cases --- ## Future +- offline nodes should not delay sync +- sync command should transitive sync all peers - Gossip: - gossip new entries to peers - backfill missing entries from peers (how do peers notice missing entries?) diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index f00d157..898ed2f 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -51,6 +51,7 @@ async fn main() { } else { println!("Root: {}", open_info.store_id); } + node.root_store().await.as_ref().cloned() } Ok(None) => { diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index 9953bb6..a9b38fc 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -35,7 +35,7 @@ pub mod store_actor; pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024; pub use node_identity::{NodeIdentity, PeerStatus}; -pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, PeerInfo, JoinAcceptance}; +pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, NodeEvent, PeerInfo, JoinAcceptance}; pub use sigchain::{SigChain, SigChainManager}; pub use entry::Entry; pub use sync_state::{SyncState, AuthorInfo, MissingRange}; diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index c558ac9..e968fe4 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -8,9 +8,11 @@ use crate::{ store::StoreError, spawn_store_actor, StoreCmd, node_identity::NodeError as IdentityError, + proto::SignedEntry, }; use std::path::Path; use thiserror::Error; +use tokio::sync::broadcast; #[derive(Error, Debug)] pub enum NodeError { @@ -67,6 +69,13 @@ pub struct PeerInfo { pub status: PeerStatus, } +/// Events emitted by Node for interested listeners (e.g., LatticeServer) +#[derive(Clone, Debug)] +pub enum NodeEvent { + /// Root store was activated (opened or set) + RootStoreActivated(StoreHandle), +} + pub struct NodeBuilder { pub data_dir: DataDir, } @@ -90,7 +99,6 @@ impl NodeBuilder { }; let meta = MetaStore::open(self.data_dir.meta_db())?; - // Set hostname on first creation if is_new { let hostname = hostname::get() @@ -98,12 +106,16 @@ impl NodeBuilder { .unwrap_or_else(|_| "unknown".to_string()); let _ = meta.set_name(&hostname); } + + // Create event channel + let (event_tx, _) = broadcast::channel(16); Ok(Node { data_dir: self.data_dir, node: std::sync::Arc::new(node), meta, root_store: tokio::sync::RwLock::new(None), + event_tx, }) } } @@ -118,6 +130,7 @@ pub struct Node { node: std::sync::Arc, meta: MetaStore, root_store: tokio::sync::RwLock>, + event_tx: broadcast::Sender, } impl Node { @@ -132,6 +145,11 @@ impl Node { pub fn node_id(&self) -> [u8; 32] { self.node.public_key_bytes() } + + /// Subscribe to node events (e.g., root store activation) + pub fn subscribe_events(&self) -> broadcast::Receiver { + self.event_tx.subscribe() + } /// Get the secret key bytes for Iroh integration (same Ed25519 key) pub fn secret_key_bytes(&self) -> [u8; 32] { @@ -184,7 +202,13 @@ impl Node { match self.meta.root_store()? { Some(id) => { let (handle, info) = self.open_store(id).await?; + + // Emit event for listeners (send clone, keep original) + let _ = self.event_tx.send(NodeEvent::RootStoreActivated(handle.clone())); + + // Store original handle (owns actor thread) *self.root_store.write().await = Some(handle); + Ok(Some(info)) } None => Ok(None), @@ -234,14 +258,18 @@ impl Node { self.create_store_with_uuid(store_id)?; self.meta.set_root_store(store_id)?; - // Open and cache the handle + // Open and cache the handle (original stays in cache) let (handle, _) = self.open_store(store_id).await?; - *self.root_store.write().await = Some(handle.clone()); + let handle_clone = handle.clone(); + *self.root_store.write().await = Some(handle); + + // Emit event for listeners + let _ = self.event_tx.send(NodeEvent::RootStoreActivated(handle_clone.clone())); // Publish our name to the store let _ = self.publish_name().await; - Ok(handle) + Ok(handle_clone) } // --- Peer Management --- @@ -471,17 +499,19 @@ impl Node { let info = StoreInfo { store_id, entries_replayed }; // Spawn actor thread - actor owns store, sigchain, and node copy - let (tx, actor_handle) = spawn_store_actor( + let (tx, entry_tx, actor_handle) = spawn_store_actor( store_id, store, sigchain, (*self.node).clone(), ); + // Store the entry sender for gossip let handle = StoreHandle { store_id, tx, actor_handle: Some(actor_handle), + entry_tx, }; Ok((handle, info)) @@ -489,10 +519,12 @@ impl Node { } /// A handle to a specific store - wraps channel to actor thread +#[derive(Debug)] pub struct StoreHandle { store_id: Uuid, tx: tokio::sync::mpsc::Sender, actor_handle: Option>, + entry_tx: broadcast::Sender, } impl Clone for StoreHandle { @@ -501,12 +533,18 @@ impl Clone for StoreHandle { store_id: self.store_id, tx: self.tx.clone(), actor_handle: None, // Clones don't own the actor thread + entry_tx: self.entry_tx.clone(), } } } impl StoreHandle { pub fn id(&self) -> Uuid { self.store_id } + + /// Subscribe to receive entries as they're committed locally + pub fn subscribe_entries(&self) -> broadcast::Receiver { + self.entry_tx.subscribe() + } pub async fn get(&self, key: &[u8]) -> Result>, NodeError> { use StoreCmd; diff --git a/lattice-core/src/store_actor.rs b/lattice-core/src/store_actor.rs index de58b86..27a4155 100644 --- a/lattice-core/src/store_actor.rs +++ b/lattice-core/src/store_actor.rs @@ -10,7 +10,7 @@ use crate::{ proto::SignedEntry, log, }; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, broadcast}; use std::thread::{self, JoinHandle}; /// Commands sent to the store actor @@ -103,9 +103,11 @@ impl std::error::Error for StoreActorError {} pub struct StoreActor { store_id: Uuid, store: Store, - chain_manager: SigChainManager, // Manages all authors' sigchains + chain_manager: SigChainManager, node: NodeIdentity, rx: mpsc::Receiver, + /// Broadcast sender for emitting entries after they're committed locally + entry_tx: broadcast::Sender, } impl StoreActor { @@ -116,6 +118,7 @@ impl StoreActor { sigchain: SigChain, node: NodeIdentity, rx: mpsc::Receiver, + entry_tx: broadcast::Sender, ) -> Self { // Derive logs_dir from sigchain's log file path let logs_dir = sigchain.log_path() @@ -134,6 +137,7 @@ impl StoreActor { chain_manager, node, rx, + entry_tx, } } @@ -253,6 +257,9 @@ impl StoreActor { let sigchain = self.chain_manager.get_or_create(local_author); sigchain.append(&entry)?; self.store.apply_entry(&entry)?; + + // Broadcast the entry to listeners (for gossip) + let _ = self.entry_tx.send(entry.clone()); Ok(seq) } @@ -276,16 +283,17 @@ impl StoreActor { } } -/// Spawn a store actor in a new thread, returns (sender, join_handle) +/// Spawn a store actor in a new thread, returns (cmd_tx, entry_tx, join_handle) /// Uses std::thread since redb is blocking pub fn spawn_store_actor( store_id: Uuid, store: Store, sigchain: SigChain, node: NodeIdentity, -) -> (mpsc::Sender, JoinHandle<()>) { +) -> (mpsc::Sender, broadcast::Sender, JoinHandle<()>) { let (tx, rx) = mpsc::channel(32); - let actor = StoreActor::new(store_id, store, sigchain, node, rx); + let (entry_tx, _entry_rx) = broadcast::channel(64); + let actor = StoreActor::new(store_id, store, sigchain, node, rx, entry_tx.clone()); let handle = thread::spawn(move || actor.run()); - (tx, handle) + (tx, entry_tx, handle) } diff --git a/lattice-net/Cargo.toml b/lattice-net/Cargo.toml index 50045f5..c10a919 100644 --- a/lattice-net/Cargo.toml +++ b/lattice-net/Cargo.toml @@ -17,6 +17,9 @@ bytes = { workspace = true } tokio-util = { workspace = true } futures-util = { workspace = true } hex = { workspace = true } +blake3.workspace = true +anyhow = "1.0.100" +futures-lite = "2.6.1" [dev-dependencies] tokio-test = { workspace = true } diff --git a/lattice-net/src/endpoint.rs b/lattice-net/src/endpoint.rs index b4bf0f1..1a32706 100644 --- a/lattice-net/src/endpoint.rs +++ b/lattice-net/src/endpoint.rs @@ -28,7 +28,10 @@ impl LatticeEndpoint { let endpoint = Endpoint::builder() .secret_key(secret_key) - .alpns(vec![LATTICE_ALPN.to_vec()]) + .alpns(vec![ + LATTICE_ALPN.to_vec(), + iroh_gossip::ALPN.to_vec(), // Also accept gossip protocol + ]) .discovery(mdns) // Add mDNS on top of default DNS .bind() .await?; diff --git a/lattice-net/src/lib.rs b/lattice-net/src/lib.rs index bc0690f..1180f7a 100644 --- a/lattice-net/src/lib.rs +++ b/lattice-net/src/lib.rs @@ -12,7 +12,7 @@ pub mod gossip; pub mod framing; pub mod mesh; -pub use endpoint::{LatticeEndpoint, PublicKey}; +pub use endpoint::{LatticeEndpoint, PublicKey, LATTICE_ALPN}; pub use framing::{MessageSink, MessageStream}; pub use lattice_core::proto::{SyncRequest, SyncResponse, SyncEntry, SyncDone, SyncState, Frontier}; pub use mesh::{LatticeServer, SyncResult}; diff --git a/lattice-net/src/mesh/server.rs b/lattice-net/src/mesh/server.rs index dbcfe3b..ed786c8 100644 --- a/lattice-net/src/mesh/server.rs +++ b/lattice-net/src/mesh/server.rs @@ -1,10 +1,16 @@ //! Server - LatticeServer for mesh networking -use crate::{MessageSink, MessageStream, LatticeEndpoint, parse_node_id}; -use lattice_core::{Node, NodeError, PeerStatus, Uuid, StoreHandle}; +use crate::{MessageSink, MessageStream, LatticeEndpoint, parse_node_id, LATTICE_ALPN}; +use lattice_core::{Node, NodeError, NodeEvent, PeerStatus, Uuid, StoreHandle}; use iroh::endpoint::Connection; +use iroh::protocol::{Router, ProtocolHandler, AcceptError}; +use iroh_gossip::Gossip; use std::sync::Arc; -use lattice_core::proto::{PeerMessage, peer_message, JoinRequest, JoinResponse}; +use std::collections::HashMap; +use tokio::sync::RwLock; +use futures_util::StreamExt; +use lattice_core::proto::{PeerMessage, peer_message, JoinRequest, JoinResponse, SignedEntry}; +use prost::Message; use super::protocol; /// Result of a sync operation with a peer @@ -13,11 +19,41 @@ pub struct SyncResult { pub entries_sent_by_peer: u64, } -/// LatticeServer wraps Node + Endpoint and provides mesh networking methods. -/// Spawns accept loop on creation to handle incoming connections. +/// LatticeServer wraps Node + Endpoint + Gossip and provides mesh networking methods. +/// Uses Router to handle incoming connections for both sync and gossip protocols. pub struct LatticeServer { node: Arc, endpoint: LatticeEndpoint, + gossip: Gossip, + #[allow(dead_code)] + router: Router, + /// Gossip senders per store topic + gossip_senders: Arc>>, +} + +/// Protocol handler for lattice sync connections +struct SyncProtocol { + node: Arc, +} + +impl std::fmt::Debug for SyncProtocol { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SyncProtocol").finish() + } +} + + +impl ProtocolHandler for SyncProtocol { + fn accept(&self, conn: Connection) -> impl std::future::Future> + Send { + let node = self.node.clone(); + Box::pin(async move { + if let Err(e) = handle_connection(node, conn).await { + eprintln!("[Accept] Error: {}", e); + // Log error but return Ok - protocol handled the connection + } + Ok(()) + }) + } } impl LatticeServer { @@ -25,14 +61,175 @@ impl LatticeServer { 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)) + Self::new(node, endpoint).await } - /// 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 + /// Create a new LatticeServer with existing endpoint. + pub async fn new(node: Arc, endpoint: LatticeEndpoint) -> Result { + // Create gossip instance + let gossip = Gossip::builder().spawn(endpoint.endpoint().clone()); + + // Create sync protocol handler + let sync_protocol = SyncProtocol { node: node.clone() }; + + // Create router to handle both protocols + let router = Router::builder(endpoint.endpoint().clone()) + .accept(LATTICE_ALPN, sync_protocol) + .accept(iroh_gossip::ALPN, gossip.clone()) + .spawn(); + + let server = Self { + node, + endpoint, + gossip, + router, + gossip_senders: Arc::new(RwLock::new(HashMap::new())), + }; + server.spawn_node_event_listener(); + + // If root store is already open, start gossip for it + if let Some(store) = (*server.node.root_store().await).clone() { + println!("[Gossip] Root store already open, starting gossip..."); + server.join_gossip_topic(store.id()).await?; + server.spawn_entry_forward_loop(store); + } + + Ok(server) + } + + /// Spawn a listener for Node events (auto-starts gossip when root store is activated) + fn spawn_node_event_listener(&self) { + let mut event_rx = self.node.subscribe_events(); + let gossip_senders = self.gossip_senders.clone(); + let gossip = self.gossip.clone(); + let node = self.node.clone(); + + tokio::spawn(async move { + while let Ok(event) = event_rx.recv().await { + match event { + NodeEvent::RootStoreActivated(store) => { + println!("[Gossip] Root store activated: {}, starting gossip...", store.id()); + + let store_id = store.id(); + + // Get bootstrap peers from node's peer list + let bootstrap_peers: Vec = match node.list_peers().await { + Ok(peers) => { + peers.iter() + .filter(|p| p.status == PeerStatus::Active) + .filter_map(|p| parse_node_id(&p.pubkey).ok()) + .collect() + } + Err(e) => { + eprintln!("[Gossip] Failed to list peers: {}, using empty list", e); + Vec::new() + } + }; + println!("[Gossip] Bootstrap peers: {}", bootstrap_peers.len()); + + // Topic ID from hash of "lattice/{store_id}" for namespacing + let topic_bytes = blake3::hash(format!("lattice/{}", store_id).as_bytes()); + let topic_id = iroh_gossip::TopicId::from_bytes(*topic_bytes.as_bytes()); + + // Use subscribe (non-blocking) - peers will connect when they sync + // subscribe_and_join would block waiting for peers we can't reach yet + match gossip.subscribe(topic_id, bootstrap_peers).await { + Ok(sub) => { + let (sender, receiver) = sub.split(); + + // Store sender + gossip_senders.write().await.insert(store_id, sender); + + // Spawn receive loop + let store_recv = store.clone(); + tokio::spawn(async move { + let mut receiver = receiver; + println!("[Gossip] Receive loop started for topic {:?}", topic_id); + + while let Some(event) = futures_util::StreamExt::next(&mut receiver).await { + match event { + Ok(iroh_gossip::api::Event::Received(msg)) => { + println!("[Gossip] Received {} bytes", msg.content.len()); + if let Ok(entry) = SignedEntry::decode(&msg.content[..]) { + if let Err(e) = store_recv.apply_entry(entry).await { + eprintln!("[Gossip] Failed to apply entry: {}", e); + } else { + println!("[Gossip] Applied entry successfully"); + } + } + } + Ok(other) => { + println!("[Gossip] Event: {:?}", other); + } + Err(e) => { + eprintln!("[Gossip] Error: {}", e); + } + } + } + }); + + // Spawn entry forward loop + let gossip_senders = gossip_senders.clone(); + let mut entry_rx = store.subscribe_entries(); + tokio::spawn(async move { + println!("[Gossip] Entry forward loop started for store {}", store_id); + while let Ok(entry) = entry_rx.recv().await { + let senders = gossip_senders.read().await; + if let Some(sender) = senders.get(&store_id) { + let bytes = entry.encode_to_vec(); + println!("[Gossip] Broadcasting {} bytes", bytes.len()); + let _ = sender.broadcast(bytes.into()).await; + } + } + }); + + println!("[Gossip] Gossip started for store {}", store_id); + } + Err(e) => { + eprintln!("[Gossip] Failed to subscribe to topic: {}", e); + } + } + } + } + } + }); + } + + /// Start gossip for a store (call after store is opened) + pub async fn start_gossip_for_store(&self, store: StoreHandle) -> Result<(), String> { + println!("[Gossip] Starting gossip for store {}", store.id()); + self.join_gossip_topic(store.id()).await?; + self.spawn_entry_forward_loop(store); + Ok(()) + } + + /// Spawn a loop that forwards local store entry broadcasts to gossip + fn spawn_entry_forward_loop(&self, store: StoreHandle) { + let store_id = store.id(); + let gossip_senders = self.gossip_senders.clone(); + let mut entry_rx = store.subscribe_entries(); + + println!("[Gossip] Starting entry forward loop for store {}", store_id); + + tokio::spawn(async move { + while let Ok(entry) = entry_rx.recv().await { + println!("[Gossip] Received local entry, forwarding to gossip..."); + // Forward to gossip sender + let senders = gossip_senders.read().await; + if let Some(sender) = senders.get(&store_id) { + let bytes = entry.encode_to_vec(); + println!("[Gossip] Broadcasting {} bytes to topic {}", bytes.len(), store_id); + if let Err(e) = sender.broadcast(bytes.into()).await { + eprintln!("[Gossip] Failed to broadcast entry: {}", e); + } else { + println!("[Gossip] Broadcast successful"); + } + } else { + eprintln!("[Gossip] No gossip sender for store {}", store_id); + } + } + println!("[Gossip] Entry forward loop ended for store {}", store_id); + }); } /// Access the underlying node @@ -44,28 +241,87 @@ impl LatticeServer { pub fn endpoint(&self) -> &LatticeEndpoint { &self.endpoint } + - /// Spawn the accept loop for incoming connections. - fn spawn_accept_loop(&self) { + /// Join gossip topic for a store (subscribes and spawns receive loop) + pub async fn join_gossip_topic(&self, store_id: Uuid) -> Result<(), String> { + // Get active peers to bootstrap gossip + let peers = self.node.list_peers().await + .map_err(|e| format!("Failed to list peers: {}", e))?; + + let bootstrap_peers: Vec = peers.iter() + .filter(|p| p.status == PeerStatus::Active) + .filter_map(|p| parse_node_id(&p.pubkey).ok()) + .collect(); + + println!("[Gossip] Joining topic {} with {} bootstrap peers", store_id, bootstrap_peers.len()); + + // Topic ID from store UUID bytes (padded to 32 bytes) + // Topic ID from hash of "lattice/{store_id}" for namespacing + let topic_bytes = blake3::hash(format!("lattice/{}", store_id).as_bytes()); + let topic_id = iroh_gossip::TopicId::from_bytes(*topic_bytes.as_bytes()); + + // Subscribe to topic + let (sender, mut receiver) = self.gossip.subscribe(topic_id, bootstrap_peers).await + .map_err(|e| format!("Failed to subscribe to gossip topic: {}", e))? + .split(); + + // Store sender for broadcasting + { + let mut senders = self.gossip_senders.write().await; + senders.insert(store_id, sender); + } + + // Spawn receive loop let node = self.node.clone(); - let endpoint = self.endpoint.endpoint().clone(); + let topic = topic_id; 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); + // StreamExt imported at module level + println!("[Gossip] Receive loop started for topic {:?}", topic); + + while let Some(event) = receiver.next().await { + match event { + Ok(iroh_gossip::api::Event::Received(message)) => { + println!("[Gossip] Received gossip message: {} bytes", message.content.len()); + // Decode SignedEntry and apply + match SignedEntry::decode(&message.content[..]) { + Ok(entry) => { + println!("[Gossip] Decoded entry, applying..."); + // Find store and apply entry + if let Some(store) = (*node.root_store().await).clone() { + if let Err(e) = store.apply_entry(entry.into()).await { + eprintln!("[Gossip] Failed to apply entry: {}", e); + } else { + println!("[Gossip] Entry applied successfully"); + } + } else { + eprintln!("[Gossip] No root store to apply entry to"); } - }); + } + Err(e) => eprintln!("[Gossip] Failed to decode entry: {}", e), } - Err(e) => eprintln!("[Accept] Handshake error: {:?}", e), } + Ok(other) => { + println!("[Gossip] Other event: {:?}", other); + } + Err(e) => eprintln!("[Gossip] Receive error: {}", e), } } + println!("[Gossip] Receive loop ended for topic"); }); + + Ok(()) + } + + /// Broadcast an entry to all gossip subscribers for a store + pub async fn broadcast_entry(&self, store_id: Uuid, entry: &SignedEntry) -> Result<(), String> { + let senders = self.gossip_senders.read().await; + if let Some(sender) = senders.get(&store_id) { + let bytes = entry.encode_to_vec(); + sender.broadcast(bytes.into()).await + .map_err(|e| format!("Gossip broadcast failed: {}", e))?; + } + Ok(()) } /// Join an existing mesh by connecting to a peer. @@ -163,6 +419,7 @@ impl LatticeServer { pub async fn sync_all(&self, store: &StoreHandle) -> Result, NodeError> { let peers = self.node.list_peers().await?; let mut results = Vec::new(); + let my_pubkey = self.endpoint.public_key(); for peer in peers { if peer.status != PeerStatus::Active { @@ -177,6 +434,11 @@ impl LatticeServer { } }; + // Skip self + if peer_id == my_pubkey { + continue; + } + println!("[Sync] Syncing with {}...", peer_id.fmt_short()); match self.sync_with_peer(store, peer_id).await { Ok(result) => {