From 86c49dfb043f851575acf689faa3620ee9b14b8e Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 1 Feb 2026 00:50:47 +0000 Subject: [PATCH] Implement PeerSlot architecture with two-phase peer lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refactor peer management into two distinct phases: PeerConnection (handshake phase): - Indexed by LinkId (identity unknown for inbound connections) - HandshakeState: AwaitingHello → SentHello → SentAuth → AwaitingAuthAck → Complete - Tracks expected identity, retry count, timing ActivePeer (authenticated phase): - Indexed by NodeId (verified identity) - ConnectivityState: Connected → Stale → Reconnecting → Disconnected - Holds tree state, bloom filter, routing data Cross-connection handling: - Deterministic tie-breaker: smaller node_id's OUTBOUND wins - PromotionResult enum for promotion outcomes - Both nodes reach same conclusion independently Node refactoring: - Split storage: connections (by LinkId) + peers (by NodeId) - Add addr_to_link reverse lookup for packet dispatch - promote_connection() handles promotion with cross-connection detection --- src/lib.rs | 5 +- src/node.rs | 466 ++++++++++++++++++++++----- src/peer.rs | 691 ----------------------------------------- src/peer/active.rs | 518 ++++++++++++++++++++++++++++++ src/peer/connection.rs | 434 ++++++++++++++++++++++++++ src/peer/mod.rs | 385 +++++++++++++++++++++++ 6 files changed, 1725 insertions(+), 774 deletions(-) delete mode 100644 src/peer.rs create mode 100644 src/peer/active.rs create mode 100644 src/peer/connection.rs create mode 100644 src/peer/mod.rs diff --git a/src/lib.rs b/src/lib.rs index 6b72d86..1b7c1e5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -49,7 +49,10 @@ pub use protocol::{ pub use cache::{CacheEntry, CacheError, CacheStats, CachedCoords, CoordCache, RouteCache}; // Re-export peer types -pub use peer::{Peer, PeerError, PeerState, UpstreamPeer}; +pub use peer::{ + cross_connection_winner, ActivePeer, ConnectivityState, HandshakeState, PeerConnection, + PeerError, PeerSlot, PromotionResult, +}; // Re-export node types pub use node::{Node, NodeError, NodeState}; diff --git a/src/node.rs b/src/node.rs index e9a35dc..711df84 100644 --- a/src/node.rs +++ b/src/node.rs @@ -7,7 +7,9 @@ use crate::bloom::BloomState; use crate::cache::CoordCache; use crate::config::PeerConfig; -use crate::peer::Peer; +use crate::peer::{ + cross_connection_winner, ActivePeer, PeerConnection, PromotionResult, +}; use crate::transport::{ packet_channel, Link, LinkDirection, LinkId, PacketRx, PacketTx, TransportAddr, TransportHandle, TransportId, @@ -44,15 +46,24 @@ pub enum NodeError { #[error("link not found: {0}")] LinkNotFound(LinkId), + #[error("connection not found: {0}")] + ConnectionNotFound(LinkId), + #[error("peer not found: {0:?}")] PeerNotFound(NodeId), #[error("peer already exists: {0:?}")] PeerAlreadyExists(NodeId), + #[error("connection already exists for link: {0}")] + ConnectionAlreadyExists(LinkId), + #[error("invalid peer npub '{npub}': {reason}")] InvalidPeerNpub { npub: String, reason: String }, + #[error("max connections exceeded: {max}")] + MaxConnectionsExceeded { max: usize }, + #[error("max peers exceeded: {max}")] MaxPeersExceeded { max: usize }, @@ -114,9 +125,21 @@ impl fmt::Display for NodeState { } } +/// Key for addr_to_link reverse lookup. +type AddrKey = (TransportId, TransportAddr); + /// A running FIPS node instance. /// /// This is the top-level container holding all node state. +/// +/// ## Peer Lifecycle +/// +/// Peers go through two phases: +/// 1. **Connection phase** (`connections`): Handshake in progress, indexed by LinkId +/// 2. **Active phase** (`peers`): Authenticated, indexed by NodeId +/// +/// The `addr_to_link` map enables dispatching incoming packets to the right +/// connection before authentication completes. pub struct Node { // === Identity === /// This node's cryptographic identity. @@ -150,6 +173,8 @@ pub struct Node { transports: HashMap, /// Active links. links: HashMap, + /// Reverse lookup: (transport_id, remote_addr) -> link_id. + addr_to_link: HashMap, // === Packet Channel === /// Packet sender for transports. @@ -157,11 +182,19 @@ pub struct Node { /// Packet receiver (for event loop). packet_rx: Option, - // === Peers === + // === Connections (Handshake Phase) === + /// Pending connections (handshake in progress). + /// Indexed by LinkId since we don't know the peer's identity yet. + connections: HashMap, + + // === Peers (Active Phase) === /// Authenticated peers. - peers: HashMap, + /// Indexed by NodeId (verified identity). + peers: HashMap, // === Resource Limits === + /// Maximum connections (0 = unlimited). + max_connections: usize, /// Maximum peers (0 = unlimited). max_peers: usize, /// Maximum links (0 = unlimited). @@ -227,9 +260,12 @@ impl Node { coord_cache: CoordCache::with_defaults(), transports: HashMap::new(), links: HashMap::new(), + addr_to_link: HashMap::new(), packet_tx: None, packet_rx: None, + connections: HashMap::new(), peers: HashMap::new(), + max_connections: 256, max_peers: 128, max_links: 256, next_link_id: 1, @@ -273,9 +309,12 @@ impl Node { coord_cache: CoordCache::with_defaults(), transports: HashMap::new(), links: HashMap::new(), + addr_to_link: HashMap::new(), packet_tx: None, packet_rx: None, + connections: HashMap::new(), peers: HashMap::new(), + max_connections: 256, max_peers: 128, max_links: 256, next_link_id: 1, @@ -374,7 +413,7 @@ impl Node { let peer_node_id = *peer_identity.node_id(); - // Check if peer already exists + // Check if peer already exists (fully authenticated) if self.peers.contains_key(&peer_node_id) { debug!( npub = %peer_config.npub, @@ -383,6 +422,20 @@ impl Node { return Ok(()); } + // Check if connection already in progress to this peer + let already_connecting = self.connections.values().any(|conn| { + conn.expected_identity() + .map(|id| id.node_id() == &peer_node_id) + .unwrap_or(false) + }); + if already_connecting { + debug!( + npub = %peer_config.npub, + "Connection already in progress, skipping" + ); + return Ok(()); + } + // Try addresses in priority order until one works for addr in peer_config.addresses_by_priority() { // Find a transport matching this address type @@ -407,16 +460,23 @@ impl Node { let link = Link::connectionless( link_id, transport_id, - remote_addr, + remote_addr.clone(), LinkDirection::Outbound, Duration::from_millis(100), // Base RTT estimate for UDP ); self.links.insert(link_id, link); - // Create peer in Connecting state - let mut peer = Peer::discovered(peer_identity.clone(), link_id); - peer.set_connecting(); + // Add reverse lookup for packet dispatch + self.addr_to_link + .insert((transport_id, remote_addr), link_id); + + // Create connection in handshake phase (outbound knows expected identity) + let current_time_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let connection = PeerConnection::outbound(link_id, peer_identity.clone(), current_time_ms); let alias_display = peer_config .alias @@ -431,7 +491,7 @@ impl Node { info!(" addr: {}", addr.addr); info!(" link_id: {}", link_id); - self.peers.insert(peer_node_id, peer); + self.connections.insert(link_id, connection); // Successfully initiated connection via this address return Ok(()); @@ -531,7 +591,12 @@ impl Node { // === Resource Limits === - /// Set the maximum number of peers. + /// Set the maximum number of connections (handshake phase). + pub fn set_max_connections(&mut self, max: usize) { + self.max_connections = max; + } + + /// Set the maximum number of peers (authenticated). pub fn set_max_peers(&mut self, max: usize) { self.max_peers = max; } @@ -543,6 +608,11 @@ impl Node { // === Counts === + /// Number of pending connections (handshake in progress). + pub fn connection_count(&self) -> usize { + self.connections.len() + } + /// Number of authenticated peers. pub fn peer_count(&self) -> usize { self.peers.len() @@ -601,7 +671,12 @@ impl Node { if self.max_links > 0 && self.links.len() >= self.max_links { return Err(NodeError::MaxLinksExceeded { max: self.max_links }); } - self.links.insert(link.link_id(), link); + let link_id = link.link_id(); + let transport_id = link.transport_id(); + let remote_addr = link.remote_addr().clone(); + + self.links.insert(link_id, link); + self.addr_to_link.insert((transport_id, remote_addr), link_id); Ok(()) } @@ -615,9 +690,21 @@ impl Node { self.links.get_mut(link_id) } + /// Find link ID by transport address. + pub fn find_link_by_addr(&self, transport_id: TransportId, addr: &TransportAddr) -> Option { + self.addr_to_link.get(&(transport_id, addr.clone())).copied() + } + /// Remove a link. pub fn remove_link(&mut self, link_id: &LinkId) -> Option { - self.links.remove(link_id) + if let Some(link) = self.links.remove(link_id) { + // Clean up reverse lookup + let key = (link.transport_id(), link.remote_addr().clone()); + self.addr_to_link.remove(&key); + Some(link) + } else { + None + } } /// Iterate over all links. @@ -625,41 +712,158 @@ impl Node { self.links.values() } - // === Peer Management === + // === Connection Management (Handshake Phase) === - /// Add an authenticated peer. - pub fn add_peer(&mut self, peer: Peer) -> Result<(), NodeError> { - let node_id = *peer.node_id(); + /// Add a pending connection. + pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> { + let link_id = connection.link_id(); - if self.peers.contains_key(&node_id) { - return Err(NodeError::PeerAlreadyExists(node_id)); + if self.connections.contains_key(&link_id) { + return Err(NodeError::ConnectionAlreadyExists(link_id)); } - if self.max_peers > 0 && self.peers.len() >= self.max_peers { - return Err(NodeError::MaxPeersExceeded { max: self.max_peers }); + if self.max_connections > 0 && self.connections.len() >= self.max_connections { + return Err(NodeError::MaxConnectionsExceeded { + max: self.max_connections, + }); } - self.peers.insert(node_id, peer); + self.connections.insert(link_id, connection); Ok(()) } + /// Get a connection by LinkId. + pub fn get_connection(&self, link_id: &LinkId) -> Option<&PeerConnection> { + self.connections.get(link_id) + } + + /// Get a mutable connection by LinkId. + pub fn get_connection_mut(&mut self, link_id: &LinkId) -> Option<&mut PeerConnection> { + self.connections.get_mut(link_id) + } + + /// Remove a connection. + pub fn remove_connection(&mut self, link_id: &LinkId) -> Option { + self.connections.remove(link_id) + } + + /// Iterate over all connections. + pub fn connections(&self) -> impl Iterator { + self.connections.values() + } + + /// Promote a connection to active peer after successful authentication. + /// + /// Handles cross-connection detection and resolution using tie-breaker rules. + pub fn promote_connection( + &mut self, + link_id: LinkId, + verified_identity: PeerIdentity, + current_time_ms: u64, + ) -> Result { + // Remove the connection from pending + let connection = self + .connections + .remove(&link_id) + .ok_or(NodeError::ConnectionNotFound(link_id))?; + + let peer_node_id = *verified_identity.node_id(); + let is_outbound = connection.is_outbound(); + + // Check for cross-connection + if let Some(existing_peer) = self.peers.get(&peer_node_id) { + let existing_link_id = existing_peer.link_id(); + + // Determine which connection wins + let this_wins = cross_connection_winner( + self.identity.node_id(), + &peer_node_id, + is_outbound, + ); + + if this_wins { + // This connection wins, replace the existing peer + let old_peer = self.peers.remove(&peer_node_id).unwrap(); + let loser_link_id = old_peer.link_id(); + + // Create new active peer with stats from handshake + let new_peer = ActivePeer::with_stats( + verified_identity, + link_id, + current_time_ms, + connection.link_stats().clone(), + ); + + self.peers.insert(peer_node_id, new_peer.clone()); + + info!( + node_id = %peer_node_id, + winner_link = %link_id, + loser_link = %loser_link_id, + "Cross-connection resolved: this connection won" + ); + + Ok(PromotionResult::CrossConnectionWon { + loser_link_id, + peer: new_peer, + }) + } else { + // This connection loses, keep existing + info!( + node_id = %peer_node_id, + winner_link = %existing_link_id, + loser_link = %link_id, + "Cross-connection resolved: this connection lost" + ); + + Ok(PromotionResult::CrossConnectionLost { + winner_link_id: existing_link_id, + }) + } + } else { + // No cross-connection, normal promotion + if self.max_peers > 0 && self.peers.len() >= self.max_peers { + return Err(NodeError::MaxPeersExceeded { max: self.max_peers }); + } + + let new_peer = ActivePeer::with_stats( + verified_identity, + link_id, + current_time_ms, + connection.link_stats().clone(), + ); + + self.peers.insert(peer_node_id, new_peer.clone()); + + info!( + node_id = %peer_node_id, + link_id = %link_id, + "Connection promoted to active peer" + ); + + Ok(PromotionResult::Promoted(new_peer)) + } + } + + // === Peer Management (Active Phase) === + /// Get a peer by NodeId. - pub fn get_peer(&self, node_id: &NodeId) -> Option<&Peer> { + pub fn get_peer(&self, node_id: &NodeId) -> Option<&ActivePeer> { self.peers.get(node_id) } /// Get a mutable peer by NodeId. - pub fn get_peer_mut(&mut self, node_id: &NodeId) -> Option<&mut Peer> { + pub fn get_peer_mut(&mut self, node_id: &NodeId) -> Option<&mut ActivePeer> { self.peers.get_mut(node_id) } /// Remove a peer. - pub fn remove_peer(&mut self, node_id: &NodeId) -> Option { + pub fn remove_peer(&mut self, node_id: &NodeId) -> Option { self.peers.remove(node_id) } /// Iterate over all peers. - pub fn peers(&self) -> impl Iterator { + pub fn peers(&self) -> impl Iterator { self.peers.values() } @@ -668,14 +872,14 @@ impl Node { self.peers.keys() } - /// Iterate over all active peers. - pub fn active_peers(&self) -> impl Iterator { - self.peers.values().filter(|p| p.state().is_active()) + /// Iterate over peers that can send traffic. + pub fn sendable_peers(&self) -> impl Iterator { + self.peers.values().filter(|p| p.can_send()) } - /// Number of active peers. - pub fn active_peer_count(&self) -> usize { - self.peers.values().filter(|p| p.state().is_active()).count() + /// Number of peers that can send traffic. + pub fn sendable_peer_count(&self) -> usize { + self.peers.values().filter(|p| p.can_send()).count() } // === Routing (stubs) === @@ -683,13 +887,13 @@ impl Node { /// Find next hop for a destination (stub). /// /// Returns the peer that minimizes tree distance to the destination. - pub fn find_next_hop(&self, _dest_node_id: &NodeId) -> Option<&Peer> { + pub fn find_next_hop(&self, _dest_node_id: &NodeId) -> Option<&ActivePeer> { // Stub: would implement greedy tree routing None } /// Check if a destination is in any peer's bloom filter. - pub fn destination_in_filters(&self, dest: &NodeId) -> Vec<&Peer> { + pub fn destination_in_filters(&self, dest: &NodeId) -> Vec<&ActivePeer> { self.peers.values().filter(|p| p.may_reach(dest)).collect() } @@ -791,7 +995,7 @@ impl Node { info!( state = %self.state, transports = self.transports.len(), - peers = self.peers.len(), + connections = self.connections.len(), "Node started" ); Ok(()) @@ -875,6 +1079,7 @@ impl fmt::Debug for Node { .field("node_id", self.node_id()) .field("state", &self.state) .field("is_leaf_only", &self.is_leaf_only) + .field("connections", &self.connection_count()) .field("peers", &self.peer_count()) .field("links", &self.link_count()) .field("transports", &self.transport_count()) @@ -900,12 +1105,18 @@ mod tests { NodeId::from_bytes(bytes) } + fn make_peer_identity() -> PeerIdentity { + let identity = Identity::generate(); + PeerIdentity::from_pubkey(identity.pubkey()) + } + #[test] fn test_node_creation() { let node = make_node(); assert_eq!(node.state(), NodeState::Created); assert_eq!(node.peer_count(), 0); + assert_eq!(node.connection_count(), 0); assert_eq!(node.link_count(), 0); assert!(!node.is_leaf_only()); } @@ -984,8 +1195,17 @@ mod tests { assert!(node.get_link(&link_id).is_some()); + // Test addr_to_link lookup + assert_eq!( + node.find_link_by_addr(TransportId::new(1), &TransportAddr::from_string("test")), + Some(link_id) + ); + node.remove_link(&link_id); assert_eq!(node.link_count(), 0); + + // Lookup should be gone + assert!(node.find_link_by_addr(TransportId::new(1), &TransportAddr::from_string("test")).is_none()); } #[test] @@ -993,12 +1213,12 @@ mod tests { let mut node = make_node(); node.set_max_links(2); - for _ in 0..2 { + for i in 0..2 { let link_id = node.allocate_link_id(); let link = Link::connectionless( link_id, TransportId::new(1), - TransportAddr::from_string("test"), + TransportAddr::from_string(&format!("test{}", i)), LinkDirection::Outbound, Duration::from_millis(50), ); @@ -1009,7 +1229,7 @@ mod tests { let link = Link::connectionless( link_id, TransportId::new(1), - TransportAddr::from_string("test"), + TransportAddr::from_string("test_extra"), LinkDirection::Outbound, Duration::from_millis(50), ); @@ -1019,36 +1239,102 @@ mod tests { } #[test] - fn test_node_peer_management() { + fn test_node_connection_management() { let mut node = make_node(); - let peer_identity = Identity::generate(); - let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); - let peer = Peer::discovered(peer_pub, LinkId::new(1)); - let peer_node_id = *peer.node_id(); + let identity = make_peer_identity(); + let link_id = LinkId::new(1); + let conn = PeerConnection::outbound(link_id, identity, 1000); - node.add_peer(peer).unwrap(); - assert_eq!(node.peer_count(), 1); + node.add_connection(conn).unwrap(); + assert_eq!(node.connection_count(), 1); - assert!(node.get_peer(&peer_node_id).is_some()); + assert!(node.get_connection(&link_id).is_some()); - node.remove_peer(&peer_node_id); - assert_eq!(node.peer_count(), 0); + node.remove_connection(&link_id); + assert_eq!(node.connection_count(), 0); } #[test] - fn test_node_peer_duplicate() { + fn test_node_connection_duplicate() { let mut node = make_node(); - let peer_identity = Identity::generate(); - let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); - let peer1 = Peer::discovered(peer_pub, LinkId::new(1)); - let peer2 = Peer::discovered(peer_pub, LinkId::new(2)); + let identity = make_peer_identity(); + let link_id = LinkId::new(1); + let conn1 = PeerConnection::outbound(link_id, identity.clone(), 1000); + let conn2 = PeerConnection::outbound(link_id, identity, 2000); - node.add_peer(peer1).unwrap(); - let result = node.add_peer(peer2); + node.add_connection(conn1).unwrap(); + let result = node.add_connection(conn2); - assert!(matches!(result, Err(NodeError::PeerAlreadyExists(_)))); + assert!(matches!(result, Err(NodeError::ConnectionAlreadyExists(_)))); + } + + #[test] + fn test_node_promote_connection() { + let mut node = make_node(); + + let identity = make_peer_identity(); + let node_id = *identity.node_id(); + let link_id = LinkId::new(1); + let conn = PeerConnection::outbound(link_id, identity.clone(), 1000); + + node.add_connection(conn).unwrap(); + assert_eq!(node.connection_count(), 1); + assert_eq!(node.peer_count(), 0); + + let result = node.promote_connection(link_id, identity, 2000).unwrap(); + + assert!(matches!(result, PromotionResult::Promoted(_))); + assert_eq!(node.connection_count(), 0); + assert_eq!(node.peer_count(), 1); + + let peer = node.get_peer(&node_id).unwrap(); + assert_eq!(peer.authenticated_at(), 2000); + } + + #[test] + fn test_node_cross_connection_resolution() { + let mut node = make_node(); + + // First connection and promotion (becomes active peer) + let identity = make_peer_identity(); + let node_id = *identity.node_id(); + let link_id1 = LinkId::new(1); + let conn1 = PeerConnection::outbound(link_id1, identity.clone(), 1000); + + node.add_connection(conn1).unwrap(); + node.promote_connection(link_id1, identity.clone(), 1500).unwrap(); + + assert_eq!(node.peer_count(), 1); + assert_eq!(node.get_peer(&node_id).unwrap().link_id(), link_id1); + + // Second connection (simulates cross-connection scenario) + let link_id2 = LinkId::new(2); + let conn2 = PeerConnection::inbound(link_id2, 2000); + + node.add_connection(conn2).unwrap(); + + // Promote second connection - tie-breaker determines outcome + let result = node.promote_connection(link_id2, identity, 2500).unwrap(); + + // One connection should win, one should lose + match result { + PromotionResult::CrossConnectionWon { loser_link_id, .. } => { + assert_eq!(loser_link_id, link_id1); + assert_eq!(node.get_peer(&node_id).unwrap().link_id(), link_id2); + } + PromotionResult::CrossConnectionLost { winner_link_id } => { + assert_eq!(winner_link_id, link_id1); + assert_eq!(node.get_peer(&node_id).unwrap().link_id(), link_id1); + } + PromotionResult::Promoted(_) => { + panic!("Expected cross-connection, got normal promotion"); + } + } + + // Still only one peer + assert_eq!(node.peer_count(), 1); } #[test] @@ -1056,18 +1342,24 @@ mod tests { let mut node = make_node(); node.set_max_peers(2); - for _ in 0..2 { - let peer_identity = Identity::generate(); - let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); - let peer = Peer::discovered(peer_pub, LinkId::new(1)); - node.add_peer(peer).unwrap(); + // Add two peers via promotion + for i in 0..2 { + let identity = make_peer_identity(); + let link_id = LinkId::new(i as u64 + 1); + let conn = PeerConnection::outbound(link_id, identity.clone(), 1000); + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, identity, 2000).unwrap(); } - let peer_identity = Identity::generate(); - let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); - let peer = Peer::discovered(peer_pub, LinkId::new(1)); + assert_eq!(node.peer_count(), 2); - let result = node.add_peer(peer); + // Third should fail + let identity = make_peer_identity(); + let link_id = LinkId::new(3); + let conn = PeerConnection::outbound(link_id, identity.clone(), 3000); + node.add_connection(conn).unwrap(); + + let result = node.promote_connection(link_id, identity, 4000); assert!(matches!(result, Err(NodeError::MaxPeersExceeded { .. }))); } @@ -1107,28 +1399,38 @@ mod tests { } #[test] - fn test_node_active_peers() { + fn test_node_sendable_peers() { let mut node = make_node(); - // Add a discovered peer - let peer_identity1 = Identity::generate(); - let peer_pub1 = crate::PeerIdentity::from_pubkey(peer_identity1.pubkey()); - let peer1 = Peer::discovered(peer_pub1, LinkId::new(1)); - node.add_peer(peer1).unwrap(); + // Add a healthy peer + let identity1 = make_peer_identity(); + let node_id1 = *identity1.node_id(); + let link_id1 = LinkId::new(1); + let conn1 = PeerConnection::outbound(link_id1, identity1.clone(), 1000); + node.add_connection(conn1).unwrap(); + node.promote_connection(link_id1, identity1, 2000).unwrap(); - // Add an active peer - let peer_identity2 = Identity::generate(); - let peer_pub2 = crate::PeerIdentity::from_pubkey(peer_identity2.pubkey()); - let mut peer2 = Peer::discovered(peer_pub2, LinkId::new(2)); - peer2.set_active(1000); - let peer2_id = *peer2.node_id(); - node.add_peer(peer2).unwrap(); + // Add another peer and mark it stale (still sendable) + let identity2 = make_peer_identity(); + let link_id2 = LinkId::new(2); + let conn2 = PeerConnection::outbound(link_id2, identity2.clone(), 1000); + node.add_connection(conn2).unwrap(); + node.promote_connection(link_id2, identity2, 2000).unwrap(); - assert_eq!(node.peer_count(), 2); - assert_eq!(node.active_peer_count(), 1); + // Add a third peer and mark it disconnected (not sendable) + let identity3 = make_peer_identity(); + let node_id3 = *identity3.node_id(); + let link_id3 = LinkId::new(3); + let conn3 = PeerConnection::outbound(link_id3, identity3.clone(), 1000); + node.add_connection(conn3).unwrap(); + node.promote_connection(link_id3, identity3, 2000).unwrap(); + node.get_peer_mut(&node_id3).unwrap().mark_disconnected(); - let active: Vec<_> = node.active_peers().collect(); - assert_eq!(active.len(), 1); - assert_eq!(active[0].node_id(), &peer2_id); + assert_eq!(node.peer_count(), 3); + assert_eq!(node.sendable_peer_count(), 2); + + let sendable: Vec<_> = node.sendable_peers().collect(); + assert_eq!(sendable.len(), 2); + assert!(sendable.iter().any(|p| p.node_id() == &node_id1)); } } diff --git a/src/peer.rs b/src/peer.rs deleted file mode 100644 index 7e46355..0000000 --- a/src/peer.rs +++ /dev/null @@ -1,691 +0,0 @@ -//! Peer Management Entities -//! -//! Structures for tracking authenticated remote FIPS nodes. A Peer -//! represents an authenticated connection to another node in the mesh. - -use crate::bloom::BloomFilter; -use crate::transport::{LinkId, LinkStats}; -use crate::tree::{ParentDeclaration, TreeCoordinate}; -use crate::{FipsAddress, NodeId, PeerIdentity}; -use secp256k1::XOnlyPublicKey; -use std::fmt; -use thiserror::Error; - -/// Errors related to peer operations. -#[derive(Debug, Error)] -pub enum PeerError { - #[error("peer not authenticated")] - NotAuthenticated, - - #[error("peer not found: {0:?}")] - NotFound(NodeId), - - #[error("peer already exists: {0:?}")] - AlreadyExists(NodeId), - - #[error("peer state invalid for operation: expected {expected}, got {actual}")] - InvalidState { expected: &'static str, actual: PeerState }, - - #[error("peer disconnected")] - Disconnected, -} - -/// Peer lifecycle state. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum PeerState { - /// Known via discovery or config, no link yet. - Discovered, - /// Link establishment in progress (connection-oriented transports). - Connecting, - /// FIPS authentication handshake in progress. - Authenticating, - /// Fully integrated peer. - Active, - /// Was active, now disconnected. - Disconnected, -} - -impl PeerState { - /// Check if the peer is fully operational. - pub fn is_active(&self) -> bool { - matches!(self, PeerState::Active) - } - - /// Check if peer can receive data. - pub fn can_send(&self) -> bool { - matches!(self, PeerState::Active) - } - - /// Check if this is a terminal state. - pub fn is_terminal(&self) -> bool { - matches!(self, PeerState::Disconnected) - } - - /// Check if the peer is in the process of connecting. - pub fn is_connecting(&self) -> bool { - matches!(self, PeerState::Connecting | PeerState::Authenticating) - } -} - -impl fmt::Display for PeerState { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let s = match self { - PeerState::Discovered => "discovered", - PeerState::Connecting => "connecting", - PeerState::Authenticating => "authenticating", - PeerState::Active => "active", - PeerState::Disconnected => "disconnected", - }; - write!(f, "{}", s) - } -} - -/// An authenticated remote FIPS node. -#[derive(Clone, Debug)] -pub struct Peer { - // === Identity === - /// Cryptographic identity (includes pubkey, node_id, address). - identity: PeerIdentity, - - // === Connection === - /// Link used to reach this peer. - link_id: LinkId, - /// Current lifecycle state. - state: PeerState, - - // === Spanning Tree === - /// Their latest parent declaration. - declaration: Option, - /// Their path to root. - ancestry: Option, - - // === Bloom Filter === - /// What's reachable through them (inbound filter). - inbound_filter: Option, - /// Their filter's sequence number. - filter_sequence: u64, - /// Remaining propagation hops on their filter. - filter_ttl: u8, - /// When we received their last filter (Unix milliseconds). - filter_received_at: u64, - /// Whether we owe them a filter update. - pending_filter_update: bool, - - // === Statistics === - /// Link statistics. - link_stats: LinkStats, - /// When this peer was first connected (Unix milliseconds). - connected_at: Option, - /// When this peer was last seen (any activity, Unix milliseconds). - last_seen: u64, -} - -impl Peer { - /// Create a new peer in Discovered state. - pub fn discovered(identity: PeerIdentity, link_id: LinkId) -> Self { - Self { - identity, - link_id, - state: PeerState::Discovered, - declaration: None, - ancestry: None, - inbound_filter: None, - filter_sequence: 0, - filter_ttl: 0, - filter_received_at: 0, - pending_filter_update: false, - link_stats: LinkStats::new(), - connected_at: None, - last_seen: 0, - } - } - - /// Create a new peer from a public key. - pub fn from_pubkey(pubkey: XOnlyPublicKey, link_id: LinkId) -> Self { - Self::discovered(PeerIdentity::from_pubkey(pubkey), link_id) - } - - // === Identity Accessors === - - /// Get the peer's identity. - pub fn identity(&self) -> &PeerIdentity { - &self.identity - } - - /// Get the peer's NodeId. - pub fn node_id(&self) -> &NodeId { - self.identity.node_id() - } - - /// Get the peer's FIPS address. - pub fn address(&self) -> &FipsAddress { - self.identity.address() - } - - /// Get the peer's public key. - pub fn pubkey(&self) -> XOnlyPublicKey { - self.identity.pubkey() - } - - /// Get the peer's npub string. - pub fn npub(&self) -> String { - self.identity.npub() - } - - // === Connection Accessors === - - /// Get the link ID. - pub fn link_id(&self) -> LinkId { - self.link_id - } - - /// Get the current state. - pub fn state(&self) -> PeerState { - self.state - } - - /// Check if the peer is active. - pub fn is_active(&self) -> bool { - self.state.is_active() - } - - /// Check if the peer can receive data. - pub fn can_send(&self) -> bool { - self.state.can_send() - } - - // === Tree Accessors === - - /// Get the peer's tree coordinates, if known. - pub fn coords(&self) -> Option<&TreeCoordinate> { - self.ancestry.as_ref() - } - - /// Get the peer's parent declaration, if known. - pub fn declaration(&self) -> Option<&ParentDeclaration> { - self.declaration.as_ref() - } - - /// Check if this peer has a known tree position. - pub fn has_tree_position(&self) -> bool { - self.declaration.is_some() && self.ancestry.is_some() - } - - // === Filter Accessors === - - /// Get the peer's inbound filter, if known. - pub fn inbound_filter(&self) -> Option<&BloomFilter> { - self.inbound_filter.as_ref() - } - - /// Get the filter sequence number. - pub fn filter_sequence(&self) -> u64 { - self.filter_sequence - } - - /// Get the filter TTL. - pub fn filter_ttl(&self) -> u8 { - self.filter_ttl - } - - /// Check if this peer's filter is stale. - pub fn filter_is_stale(&self, current_time_ms: u64, stale_threshold_ms: u64) -> bool { - if self.filter_received_at == 0 { - return true; - } - current_time_ms.saturating_sub(self.filter_received_at) > stale_threshold_ms - } - - /// Check if a destination might be reachable through this peer. - pub fn may_reach(&self, node_id: &NodeId) -> bool { - match &self.inbound_filter { - Some(filter) => filter.contains(node_id), - None => false, - } - } - - /// Check if we need to send this peer a filter update. - pub fn needs_filter_update(&self) -> bool { - self.pending_filter_update - } - - // === Statistics Accessors === - - /// Get link statistics. - pub fn link_stats(&self) -> &LinkStats { - &self.link_stats - } - - /// Get mutable link statistics. - pub fn link_stats_mut(&mut self) -> &mut LinkStats { - &mut self.link_stats - } - - /// Get when this peer was connected. - pub fn connected_at(&self) -> Option { - self.connected_at - } - - /// Get when this peer was last seen. - pub fn last_seen(&self) -> u64 { - self.last_seen - } - - /// Time since last activity. - pub fn idle_time(&self, current_time_ms: u64) -> u64 { - if self.last_seen == 0 { - return u64::MAX; - } - current_time_ms.saturating_sub(self.last_seen) - } - - /// Connection duration. - pub fn connection_duration(&self, current_time_ms: u64) -> Option { - self.connected_at - .map(|t| current_time_ms.saturating_sub(t)) - } - - // === State Transitions === - - /// Transition to Connecting state. - pub fn set_connecting(&mut self) { - self.state = PeerState::Connecting; - } - - /// Transition to Authenticating state. - pub fn set_authenticating(&mut self) { - self.state = PeerState::Authenticating; - } - - /// Transition to Active state. - pub fn set_active(&mut self, current_time_ms: u64) { - self.state = PeerState::Active; - self.connected_at = Some(current_time_ms); - self.last_seen = current_time_ms; - } - - /// Transition to Disconnected state. - pub fn set_disconnected(&mut self) { - self.state = PeerState::Disconnected; - } - - /// Update last seen timestamp. - pub fn touch(&mut self, current_time_ms: u64) { - self.last_seen = current_time_ms; - } - - // === Tree Updates === - - /// Update peer's tree position. - pub fn update_tree_position( - &mut self, - declaration: ParentDeclaration, - ancestry: TreeCoordinate, - current_time_ms: u64, - ) { - self.declaration = Some(declaration); - self.ancestry = Some(ancestry); - self.last_seen = current_time_ms; - } - - /// Clear peer's tree position. - pub fn clear_tree_position(&mut self) { - self.declaration = None; - self.ancestry = None; - } - - // === Filter Updates === - - /// Update peer's inbound filter. - pub fn update_filter( - &mut self, - filter: BloomFilter, - sequence: u64, - ttl: u8, - current_time_ms: u64, - ) { - self.inbound_filter = Some(filter); - self.filter_sequence = sequence; - self.filter_ttl = ttl; - self.filter_received_at = current_time_ms; - self.last_seen = current_time_ms; - } - - /// Clear peer's inbound filter. - pub fn clear_filter(&mut self) { - self.inbound_filter = None; - self.filter_sequence = 0; - self.filter_ttl = 0; - self.filter_received_at = 0; - } - - /// Mark that we need to send this peer a filter update. - pub fn mark_filter_update_needed(&mut self) { - self.pending_filter_update = true; - } - - /// Clear the pending filter update flag. - pub fn clear_filter_update_needed(&mut self) { - self.pending_filter_update = false; - } - - // === Link Updates === - - /// Update the link ID (e.g., on reconnect). - pub fn set_link_id(&mut self, link_id: LinkId) { - self.link_id = link_id; - } -} - -/// Simplified peer for leaf-only nodes. -/// -/// Leaf-only nodes maintain a single upstream peer without tree state -/// or Bloom filter management. -#[derive(Clone, Debug)] -pub struct UpstreamPeer { - /// Peer identity. - identity: PeerIdentity, - /// Link to upstream. - link_id: LinkId, - /// Lifecycle state (auth lifecycle only). - state: PeerState, - /// Link statistics. - link_stats: LinkStats, - /// When connected. - connected_at: Option, - /// Last activity. - last_seen: u64, -} - -impl UpstreamPeer { - /// Create a new upstream peer. - pub fn new(identity: PeerIdentity, link_id: LinkId) -> Self { - Self { - identity, - link_id, - state: PeerState::Discovered, - link_stats: LinkStats::new(), - connected_at: None, - last_seen: 0, - } - } - - /// Create from public key. - pub fn from_pubkey(pubkey: XOnlyPublicKey, link_id: LinkId) -> Self { - Self::new(PeerIdentity::from_pubkey(pubkey), link_id) - } - - /// Get the identity. - pub fn identity(&self) -> &PeerIdentity { - &self.identity - } - - /// Get the node ID. - pub fn node_id(&self) -> &NodeId { - self.identity.node_id() - } - - /// Get the FIPS address. - pub fn address(&self) -> &FipsAddress { - self.identity.address() - } - - /// Get the link ID. - pub fn link_id(&self) -> LinkId { - self.link_id - } - - /// Get the state. - pub fn state(&self) -> PeerState { - self.state - } - - /// Check if active. - pub fn is_active(&self) -> bool { - self.state.is_active() - } - - /// Get link statistics. - pub fn link_stats(&self) -> &LinkStats { - &self.link_stats - } - - /// Get mutable link statistics. - pub fn link_stats_mut(&mut self) -> &mut LinkStats { - &mut self.link_stats - } - - /// Set connecting state. - pub fn set_connecting(&mut self) { - self.state = PeerState::Connecting; - } - - /// Set authenticating state. - pub fn set_authenticating(&mut self) { - self.state = PeerState::Authenticating; - } - - /// Set active state. - pub fn set_active(&mut self, current_time_ms: u64) { - self.state = PeerState::Active; - self.connected_at = Some(current_time_ms); - self.last_seen = current_time_ms; - } - - /// Set disconnected state. - pub fn set_disconnected(&mut self) { - self.state = PeerState::Disconnected; - } - - /// Update last seen. - pub fn touch(&mut self, current_time_ms: u64) { - self.last_seen = current_time_ms; - } - - /// Get connected timestamp. - pub fn connected_at(&self) -> Option { - self.connected_at - } - - /// Get last seen timestamp. - pub fn last_seen(&self) -> u64 { - self.last_seen - } - - /// Set link ID. - pub fn set_link_id(&mut self, link_id: LinkId) { - self.link_id = link_id; - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::Identity; - - fn make_peer() -> Peer { - let identity = Identity::generate(); - let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); - Peer::discovered(peer_identity, LinkId::new(1)) - } - - fn make_node_id(val: u8) -> NodeId { - let mut bytes = [0u8; 32]; - bytes[0] = val; - NodeId::from_bytes(bytes) - } - - fn make_coords(ids: &[u8]) -> TreeCoordinate { - TreeCoordinate::new(ids.iter().map(|&v| make_node_id(v)).collect()).unwrap() - } - - // ===== PeerState Tests ===== - - #[test] - fn test_peer_state_properties() { - assert!(!PeerState::Discovered.is_active()); - assert!(!PeerState::Connecting.is_active()); - assert!(!PeerState::Authenticating.is_active()); - assert!(PeerState::Active.is_active()); - assert!(!PeerState::Disconnected.is_active()); - - assert!(PeerState::Connecting.is_connecting()); - assert!(PeerState::Authenticating.is_connecting()); - assert!(!PeerState::Active.is_connecting()); - - assert!(PeerState::Disconnected.is_terminal()); - assert!(!PeerState::Active.is_terminal()); - } - - // ===== Peer Tests ===== - - #[test] - fn test_peer_state_transitions() { - let mut peer = make_peer(); - - assert_eq!(peer.state(), PeerState::Discovered); - assert!(!peer.is_active()); - - peer.set_connecting(); - assert_eq!(peer.state(), PeerState::Connecting); - - peer.set_authenticating(); - assert_eq!(peer.state(), PeerState::Authenticating); - - peer.set_active(1000); - assert_eq!(peer.state(), PeerState::Active); - assert!(peer.is_active()); - assert_eq!(peer.connected_at(), Some(1000)); - - peer.set_disconnected(); - assert_eq!(peer.state(), PeerState::Disconnected); - assert!(peer.state().is_terminal()); - } - - #[test] - fn test_peer_filter_stale() { - let mut peer = make_peer(); - - // No filter received yet - assert!(peer.filter_is_stale(1000, 500)); - - // Update filter - peer.update_filter(BloomFilter::new(), 1, 2, 1000); - - // Not stale yet - assert!(!peer.filter_is_stale(1200, 500)); - - // Stale after threshold - assert!(peer.filter_is_stale(1600, 500)); - } - - #[test] - fn test_peer_may_reach() { - let mut peer = make_peer(); - let target = make_node_id(42); - - // No filter yet - assert!(!peer.may_reach(&target)); - - // Add filter with target - let mut filter = BloomFilter::new(); - filter.insert(&target); - peer.update_filter(filter, 1, 2, 0); - - assert!(peer.may_reach(&target)); - } - - #[test] - fn test_peer_tree_position() { - let mut peer = make_peer(); - - assert!(!peer.has_tree_position()); - assert!(peer.coords().is_none()); - assert!(peer.declaration().is_none()); - - let node = make_node_id(1); - let parent = make_node_id(2); - let decl = ParentDeclaration::new(node, parent, 1, 1000); - let coords = make_coords(&[1, 2, 0]); - - peer.update_tree_position(decl, coords, 2000); - - assert!(peer.has_tree_position()); - assert!(peer.coords().is_some()); - assert!(peer.declaration().is_some()); - assert_eq!(peer.last_seen(), 2000); - } - - #[test] - fn test_peer_filter_update_flag() { - let mut peer = make_peer(); - - assert!(!peer.needs_filter_update()); - - peer.mark_filter_update_needed(); - assert!(peer.needs_filter_update()); - - peer.clear_filter_update_needed(); - assert!(!peer.needs_filter_update()); - } - - #[test] - fn test_peer_idle_time() { - let mut peer = make_peer(); - - // No activity yet - assert_eq!(peer.idle_time(1000), u64::MAX); - - peer.touch(500); - assert_eq!(peer.idle_time(1000), 500); - assert_eq!(peer.idle_time(500), 0); - } - - #[test] - fn test_peer_connection_duration() { - let mut peer = make_peer(); - - // Not connected - assert!(peer.connection_duration(1000).is_none()); - - peer.set_active(500); - assert_eq!(peer.connection_duration(1000), Some(500)); - } - - // ===== UpstreamPeer Tests ===== - - #[test] - fn test_upstream_peer_state_transitions() { - let identity = Identity::generate(); - let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); - let mut upstream = UpstreamPeer::new(peer_identity, LinkId::new(1)); - - assert!(!upstream.is_active()); - assert_eq!(upstream.state(), PeerState::Discovered); - - upstream.set_connecting(); - assert_eq!(upstream.state(), PeerState::Connecting); - - upstream.set_authenticating(); - assert_eq!(upstream.state(), PeerState::Authenticating); - - upstream.set_active(1000); - assert!(upstream.is_active()); - assert_eq!(upstream.connected_at(), Some(1000)); - - upstream.set_disconnected(); - assert!(!upstream.is_active()); - } - - #[test] - fn test_upstream_peer_touch() { - let identity = Identity::generate(); - let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); - let mut upstream = UpstreamPeer::new(peer_identity, LinkId::new(1)); - - assert_eq!(upstream.last_seen(), 0); - - upstream.touch(1000); - assert_eq!(upstream.last_seen(), 1000); - } -} diff --git a/src/peer/active.rs b/src/peer/active.rs new file mode 100644 index 0000000..7f64068 --- /dev/null +++ b/src/peer/active.rs @@ -0,0 +1,518 @@ +//! Active Peer (Authenticated Phase) +//! +//! Represents a fully authenticated peer after successful Noise handshake. +//! ActivePeer holds tree state, Bloom filter, and routing information. + +use crate::bloom::BloomFilter; +use crate::transport::{LinkId, LinkStats}; +use crate::tree::{ParentDeclaration, TreeCoordinate}; +use crate::{FipsAddress, NodeId, PeerIdentity}; +use secp256k1::XOnlyPublicKey; +use std::fmt; + +/// Connectivity state for an active peer. +/// +/// This is simpler than the full PeerState since authentication is complete. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ConnectivityState { + /// Peer is fully connected and responsive. + Connected, + /// Peer hasn't been heard from recently (potential timeout). + Stale, + /// Connection lost, attempting to reconnect. + Reconnecting, + /// Peer has been explicitly disconnected. + Disconnected, +} + +impl ConnectivityState { + /// Check if the peer is usable for sending traffic. + pub fn can_send(&self) -> bool { + matches!(self, ConnectivityState::Connected | ConnectivityState::Stale) + } + + /// Check if this is a terminal state requiring cleanup. + pub fn is_terminal(&self) -> bool { + matches!(self, ConnectivityState::Disconnected) + } + + /// Check if peer is fully healthy. + pub fn is_healthy(&self) -> bool { + matches!(self, ConnectivityState::Connected) + } +} + +impl fmt::Display for ConnectivityState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + ConnectivityState::Connected => "connected", + ConnectivityState::Stale => "stale", + ConnectivityState::Reconnecting => "reconnecting", + ConnectivityState::Disconnected => "disconnected", + }; + write!(f, "{}", s) + } +} + +/// A fully authenticated remote FIPS node. +/// +/// Created only after successful Noise KK handshake. The identity is +/// cryptographically verified at this point. +#[derive(Clone, Debug)] +pub struct ActivePeer { + // === Identity (Verified) === + /// Cryptographic identity (verified via handshake). + identity: PeerIdentity, + + // === Connection === + /// Link used to reach this peer. + link_id: LinkId, + /// Current connectivity state. + connectivity: ConnectivityState, + + // === Spanning Tree === + /// Their latest parent declaration. + declaration: Option, + /// Their path to root. + ancestry: Option, + + // === Bloom Filter === + /// What's reachable through them (inbound filter). + inbound_filter: Option, + /// Their filter's sequence number. + filter_sequence: u64, + /// Remaining propagation hops on their filter. + filter_ttl: u8, + /// When we received their last filter (Unix milliseconds). + filter_received_at: u64, + /// Whether we owe them a filter update. + pending_filter_update: bool, + + // === Statistics === + /// Link statistics. + link_stats: LinkStats, + /// When this peer was authenticated (Unix milliseconds). + authenticated_at: u64, + /// When this peer was last seen (any activity, Unix milliseconds). + last_seen: u64, +} + +impl ActivePeer { + /// Create a new active peer from verified identity. + /// + /// Called after successful authentication handshake. + pub fn new(identity: PeerIdentity, link_id: LinkId, authenticated_at: u64) -> Self { + Self { + identity, + link_id, + connectivity: ConnectivityState::Connected, + declaration: None, + ancestry: None, + inbound_filter: None, + filter_sequence: 0, + filter_ttl: 0, + filter_received_at: 0, + pending_filter_update: true, // Send filter on new connection + link_stats: LinkStats::new(), + authenticated_at, + last_seen: authenticated_at, + } + } + + /// Create from verified identity with existing link stats. + /// + /// Used when promoting from PeerConnection, preserving handshake stats. + pub fn with_stats( + identity: PeerIdentity, + link_id: LinkId, + authenticated_at: u64, + link_stats: LinkStats, + ) -> Self { + let mut peer = Self::new(identity, link_id, authenticated_at); + peer.link_stats = link_stats; + peer + } + + // === Identity Accessors === + + /// Get the peer's verified identity. + pub fn identity(&self) -> &PeerIdentity { + &self.identity + } + + /// Get the peer's NodeId. + pub fn node_id(&self) -> &NodeId { + self.identity.node_id() + } + + /// Get the peer's FIPS address. + pub fn address(&self) -> &FipsAddress { + self.identity.address() + } + + /// Get the peer's public key. + pub fn pubkey(&self) -> XOnlyPublicKey { + self.identity.pubkey() + } + + /// Get the peer's npub string. + pub fn npub(&self) -> String { + self.identity.npub() + } + + // === Connection Accessors === + + /// Get the link ID. + pub fn link_id(&self) -> LinkId { + self.link_id + } + + /// Get the connectivity state. + pub fn connectivity(&self) -> ConnectivityState { + self.connectivity + } + + /// Check if peer can receive traffic. + pub fn can_send(&self) -> bool { + self.connectivity.can_send() + } + + /// Check if peer is fully healthy. + pub fn is_healthy(&self) -> bool { + self.connectivity.is_healthy() + } + + /// Check if peer is disconnected. + pub fn is_disconnected(&self) -> bool { + self.connectivity.is_terminal() + } + + // === Tree Accessors === + + /// Get the peer's tree coordinates, if known. + pub fn coords(&self) -> Option<&TreeCoordinate> { + self.ancestry.as_ref() + } + + /// Get the peer's parent declaration, if known. + pub fn declaration(&self) -> Option<&ParentDeclaration> { + self.declaration.as_ref() + } + + /// Check if this peer has a known tree position. + pub fn has_tree_position(&self) -> bool { + self.declaration.is_some() && self.ancestry.is_some() + } + + // === Filter Accessors === + + /// Get the peer's inbound filter, if known. + pub fn inbound_filter(&self) -> Option<&BloomFilter> { + self.inbound_filter.as_ref() + } + + /// Get the filter sequence number. + pub fn filter_sequence(&self) -> u64 { + self.filter_sequence + } + + /// Get the filter TTL. + pub fn filter_ttl(&self) -> u8 { + self.filter_ttl + } + + /// Check if this peer's filter is stale. + pub fn filter_is_stale(&self, current_time_ms: u64, stale_threshold_ms: u64) -> bool { + if self.filter_received_at == 0 { + return true; + } + current_time_ms.saturating_sub(self.filter_received_at) > stale_threshold_ms + } + + /// Check if a destination might be reachable through this peer. + pub fn may_reach(&self, node_id: &NodeId) -> bool { + match &self.inbound_filter { + Some(filter) => filter.contains(node_id), + None => false, + } + } + + /// Check if we need to send this peer a filter update. + pub fn needs_filter_update(&self) -> bool { + self.pending_filter_update + } + + // === Statistics Accessors === + + /// Get link statistics. + pub fn link_stats(&self) -> &LinkStats { + &self.link_stats + } + + /// Get mutable link statistics. + pub fn link_stats_mut(&mut self) -> &mut LinkStats { + &mut self.link_stats + } + + /// When this peer was authenticated. + pub fn authenticated_at(&self) -> u64 { + self.authenticated_at + } + + /// When this peer was last seen. + pub fn last_seen(&self) -> u64 { + self.last_seen + } + + /// Time since last activity. + pub fn idle_time(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.last_seen) + } + + /// Connection duration since authentication. + pub fn connection_duration(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.authenticated_at) + } + + // === State Updates === + + /// Update last seen timestamp. + pub fn touch(&mut self, current_time_ms: u64) { + self.last_seen = current_time_ms; + // If we were stale, receiving traffic makes us connected again + if self.connectivity == ConnectivityState::Stale { + self.connectivity = ConnectivityState::Connected; + } + } + + /// Mark peer as stale (no recent traffic). + pub fn mark_stale(&mut self) { + if self.connectivity == ConnectivityState::Connected { + self.connectivity = ConnectivityState::Stale; + } + } + + /// Mark peer as reconnecting. + pub fn mark_reconnecting(&mut self) { + self.connectivity = ConnectivityState::Reconnecting; + } + + /// Mark peer as disconnected. + pub fn mark_disconnected(&mut self) { + self.connectivity = ConnectivityState::Disconnected; + } + + /// Mark peer as connected (e.g., after successful reconnect). + pub fn mark_connected(&mut self, current_time_ms: u64) { + self.connectivity = ConnectivityState::Connected; + self.last_seen = current_time_ms; + } + + /// Update the link ID (e.g., on reconnect). + pub fn set_link_id(&mut self, link_id: LinkId) { + self.link_id = link_id; + } + + // === Tree Updates === + + /// Update peer's tree position. + pub fn update_tree_position( + &mut self, + declaration: ParentDeclaration, + ancestry: TreeCoordinate, + current_time_ms: u64, + ) { + self.declaration = Some(declaration); + self.ancestry = Some(ancestry); + self.last_seen = current_time_ms; + } + + /// Clear peer's tree position. + pub fn clear_tree_position(&mut self) { + self.declaration = None; + self.ancestry = None; + } + + // === Filter Updates === + + /// Update peer's inbound filter. + pub fn update_filter( + &mut self, + filter: BloomFilter, + sequence: u64, + ttl: u8, + current_time_ms: u64, + ) { + self.inbound_filter = Some(filter); + self.filter_sequence = sequence; + self.filter_ttl = ttl; + self.filter_received_at = current_time_ms; + self.last_seen = current_time_ms; + } + + /// Clear peer's inbound filter. + pub fn clear_filter(&mut self) { + self.inbound_filter = None; + self.filter_sequence = 0; + self.filter_ttl = 0; + self.filter_received_at = 0; + } + + /// Mark that we need to send this peer a filter update. + pub fn mark_filter_update_needed(&mut self) { + self.pending_filter_update = true; + } + + /// Clear the pending filter update flag. + pub fn clear_filter_update_needed(&mut self) { + self.pending_filter_update = false; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Identity; + + fn make_peer_identity() -> PeerIdentity { + let identity = Identity::generate(); + PeerIdentity::from_pubkey(identity.pubkey()) + } + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::new(ids.iter().map(|&v| make_node_id(v)).collect()).unwrap() + } + + #[test] + fn test_connectivity_state_properties() { + assert!(ConnectivityState::Connected.can_send()); + assert!(ConnectivityState::Stale.can_send()); + assert!(!ConnectivityState::Reconnecting.can_send()); + assert!(!ConnectivityState::Disconnected.can_send()); + + assert!(ConnectivityState::Connected.is_healthy()); + assert!(!ConnectivityState::Stale.is_healthy()); + + assert!(ConnectivityState::Disconnected.is_terminal()); + assert!(!ConnectivityState::Connected.is_terminal()); + } + + #[test] + fn test_active_peer_creation() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity.clone(), LinkId::new(1), 1000); + + assert_eq!(peer.identity().node_id(), identity.node_id()); + assert_eq!(peer.link_id(), LinkId::new(1)); + assert!(peer.is_healthy()); + assert!(peer.can_send()); + assert_eq!(peer.authenticated_at(), 1000); + assert!(peer.needs_filter_update()); // New peers need filter + } + + #[test] + fn test_connectivity_transitions() { + let identity = make_peer_identity(); + let mut peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + assert!(peer.is_healthy()); + + peer.mark_stale(); + assert_eq!(peer.connectivity(), ConnectivityState::Stale); + assert!(peer.can_send()); // Stale can still send + + // Traffic received brings back to connected + peer.touch(2000); + assert!(peer.is_healthy()); + + peer.mark_reconnecting(); + assert!(!peer.can_send()); + + peer.mark_connected(3000); + assert!(peer.is_healthy()); + + peer.mark_disconnected(); + assert!(peer.is_disconnected()); + assert!(!peer.can_send()); + } + + #[test] + fn test_tree_position() { + let identity = make_peer_identity(); + let mut peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + assert!(!peer.has_tree_position()); + assert!(peer.coords().is_none()); + + let node = make_node_id(1); + let parent = make_node_id(2); + let decl = ParentDeclaration::new(node, parent, 1, 1000); + let coords = make_coords(&[1, 2, 0]); + + peer.update_tree_position(decl, coords, 2000); + + assert!(peer.has_tree_position()); + assert!(peer.coords().is_some()); + assert_eq!(peer.last_seen(), 2000); + } + + #[test] + fn test_bloom_filter() { + let identity = make_peer_identity(); + let mut peer = ActivePeer::new(identity, LinkId::new(1), 1000); + let target = make_node_id(42); + + assert!(!peer.may_reach(&target)); + assert!(peer.filter_is_stale(2000, 500)); + + let mut filter = BloomFilter::new(); + filter.insert(&target); + peer.update_filter(filter, 1, 2, 1500); + + assert!(peer.may_reach(&target)); + assert!(!peer.filter_is_stale(1800, 500)); + assert!(peer.filter_is_stale(2500, 500)); + } + + #[test] + fn test_timing() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + assert_eq!(peer.connection_duration(2000), 1000); + assert_eq!(peer.idle_time(2000), 1000); + } + + #[test] + fn test_filter_update_flag() { + let identity = make_peer_identity(); + let mut peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + assert!(peer.needs_filter_update()); // New peer + + peer.clear_filter_update_needed(); + assert!(!peer.needs_filter_update()); + + peer.mark_filter_update_needed(); + assert!(peer.needs_filter_update()); + } + + #[test] + fn test_with_stats() { + let identity = make_peer_identity(); + let mut stats = LinkStats::new(); + stats.record_sent(100); + stats.record_recv(200, 500); + + let peer = ActivePeer::with_stats(identity, LinkId::new(1), 1000, stats); + + assert_eq!(peer.link_stats().packets_sent, 1); + assert_eq!(peer.link_stats().packets_recv, 1); + } +} diff --git a/src/peer/connection.rs b/src/peer/connection.rs new file mode 100644 index 0000000..a69cbd3 --- /dev/null +++ b/src/peer/connection.rs @@ -0,0 +1,434 @@ +//! Peer Connection (Handshake Phase) +//! +//! Represents an in-progress connection before authentication completes. +//! PeerConnection tracks the Noise handshake state and transitions to +//! ActivePeer upon successful authentication. + +use crate::transport::{LinkDirection, LinkId, LinkStats}; +use crate::PeerIdentity; +use std::fmt; + +/// Handshake protocol state machine. +/// +/// For Noise KK pattern: +/// - Initiator: SentHello → AwaitingAuth → Complete +/// - Responder: AwaitingHello → SentAuth → Complete +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum HandshakeState { + /// Waiting for initial Hello from remote (responder role). + AwaitingHello, + /// Sent Hello, waiting for Auth response (initiator role). + SentHello, + /// Received Hello, sent Auth, waiting for AuthAck (responder role). + SentAuth, + /// Sent Auth, waiting for AuthAck (initiator role). + AwaitingAuthAck, + /// Handshake completed successfully. + Complete, + /// Handshake failed. + Failed, +} + +impl HandshakeState { + /// Check if handshake is still in progress. + pub fn is_in_progress(&self) -> bool { + matches!( + self, + HandshakeState::AwaitingHello + | HandshakeState::SentHello + | HandshakeState::SentAuth + | HandshakeState::AwaitingAuthAck + ) + } + + /// Check if handshake completed successfully. + pub fn is_complete(&self) -> bool { + matches!(self, HandshakeState::Complete) + } + + /// Check if handshake failed. + pub fn is_failed(&self) -> bool { + matches!(self, HandshakeState::Failed) + } + + /// Check if we are the initiator (sent first message). + pub fn is_initiator(&self) -> bool { + matches!( + self, + HandshakeState::SentHello | HandshakeState::AwaitingAuthAck + ) + } + + /// Check if we are the responder (received first message). + pub fn is_responder(&self) -> bool { + matches!( + self, + HandshakeState::AwaitingHello | HandshakeState::SentAuth + ) + } +} + +impl fmt::Display for HandshakeState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + HandshakeState::AwaitingHello => "awaiting_hello", + HandshakeState::SentHello => "sent_hello", + HandshakeState::SentAuth => "sent_auth", + HandshakeState::AwaitingAuthAck => "awaiting_auth_ack", + HandshakeState::Complete => "complete", + HandshakeState::Failed => "failed", + }; + write!(f, "{}", s) + } +} + +/// A connection in the handshake phase, before authentication completes. +/// +/// For outbound connections, we know the expected peer identity from config. +/// For inbound connections, we learn the identity during the handshake. +#[derive(Clone, Debug)] +pub struct PeerConnection { + // === Link Reference === + /// The link carrying this connection. + link_id: LinkId, + + /// Connection direction (we initiated or they initiated). + direction: LinkDirection, + + // === Handshake State === + /// Current handshake state. + handshake_state: HandshakeState, + + /// Expected peer identity (known for outbound, learned for inbound). + /// None until we receive their public key in the handshake. + expected_identity: Option, + + // === Noise Session State === + // TODO: Add actual Noise protocol state when implementing crypto + // noise_state: Option, + + // === Timing === + /// When the connection attempt started (Unix milliseconds). + started_at: u64, + + /// When the last handshake message was sent/received. + last_activity: u64, + + /// Number of retries attempted. + retry_count: u32, + + // === Statistics === + /// Link statistics during handshake. + link_stats: LinkStats, +} + +impl PeerConnection { + /// Create a new outbound connection (we are initiating). + /// + /// For outbound, we know who we're trying to reach from configuration. + pub fn outbound( + link_id: LinkId, + expected_identity: PeerIdentity, + current_time_ms: u64, + ) -> Self { + Self { + link_id, + direction: LinkDirection::Outbound, + handshake_state: HandshakeState::SentHello, + expected_identity: Some(expected_identity), + started_at: current_time_ms, + last_activity: current_time_ms, + retry_count: 0, + link_stats: LinkStats::new(), + } + } + + /// Create a new inbound connection (they are initiating). + /// + /// For inbound, we don't know who they are until they identify in handshake. + pub fn inbound(link_id: LinkId, current_time_ms: u64) -> Self { + Self { + link_id, + direction: LinkDirection::Inbound, + handshake_state: HandshakeState::AwaitingHello, + expected_identity: None, + started_at: current_time_ms, + last_activity: current_time_ms, + retry_count: 0, + link_stats: LinkStats::new(), + } + } + + // === Accessors === + + /// Get the link ID. + pub fn link_id(&self) -> LinkId { + self.link_id + } + + /// Get the connection direction. + pub fn direction(&self) -> LinkDirection { + self.direction + } + + /// Get the handshake state. + pub fn handshake_state(&self) -> HandshakeState { + self.handshake_state + } + + /// Get the expected/learned peer identity, if known. + pub fn expected_identity(&self) -> Option<&PeerIdentity> { + self.expected_identity.as_ref() + } + + /// Check if this is an outbound connection. + pub fn is_outbound(&self) -> bool { + self.direction == LinkDirection::Outbound + } + + /// Check if this is an inbound connection. + pub fn is_inbound(&self) -> bool { + self.direction == LinkDirection::Inbound + } + + /// Check if handshake is in progress. + pub fn is_in_progress(&self) -> bool { + self.handshake_state.is_in_progress() + } + + /// Check if handshake completed. + pub fn is_complete(&self) -> bool { + self.handshake_state.is_complete() + } + + /// Check if handshake failed. + pub fn is_failed(&self) -> bool { + self.handshake_state.is_failed() + } + + /// When the connection started. + pub fn started_at(&self) -> u64 { + self.started_at + } + + /// When the last activity occurred. + pub fn last_activity(&self) -> u64 { + self.last_activity + } + + /// Connection duration so far. + pub fn duration(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.started_at) + } + + /// Time since last activity. + pub fn idle_time(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.last_activity) + } + + /// Number of retries. + pub fn retry_count(&self) -> u32 { + self.retry_count + } + + /// Get link statistics. + pub fn link_stats(&self) -> &LinkStats { + &self.link_stats + } + + /// Get mutable link statistics. + pub fn link_stats_mut(&mut self) -> &mut LinkStats { + &mut self.link_stats + } + + // === State Transitions === + + /// Record that we sent a Hello message (initiator). + pub fn mark_hello_sent(&mut self, current_time_ms: u64) { + self.handshake_state = HandshakeState::SentHello; + self.last_activity = current_time_ms; + } + + /// Record that we received a Hello and learned peer identity. + pub fn mark_hello_received(&mut self, identity: PeerIdentity, current_time_ms: u64) { + self.expected_identity = Some(identity); + self.last_activity = current_time_ms; + } + + /// Record that we sent Auth response (responder). + pub fn mark_auth_sent(&mut self, current_time_ms: u64) { + self.handshake_state = HandshakeState::SentAuth; + self.last_activity = current_time_ms; + } + + /// Record that we're awaiting AuthAck (initiator). + pub fn mark_awaiting_auth_ack(&mut self, current_time_ms: u64) { + self.handshake_state = HandshakeState::AwaitingAuthAck; + self.last_activity = current_time_ms; + } + + /// Mark handshake as complete. + pub fn mark_complete(&mut self, current_time_ms: u64) { + self.handshake_state = HandshakeState::Complete; + self.last_activity = current_time_ms; + } + + /// Mark handshake as failed. + pub fn mark_failed(&mut self) { + self.handshake_state = HandshakeState::Failed; + } + + /// Increment retry counter. + pub fn increment_retry(&mut self) { + self.retry_count += 1; + } + + /// Update last activity timestamp. + pub fn touch(&mut self, current_time_ms: u64) { + self.last_activity = current_time_ms; + } + + // === Validation === + + /// Check if the connection has timed out. + pub fn is_timed_out(&self, current_time_ms: u64, timeout_ms: u64) -> bool { + self.idle_time(current_time_ms) > timeout_ms + } + + /// Check if max retries exceeded. + pub fn max_retries_exceeded(&self, max_retries: u32) -> bool { + self.retry_count >= max_retries + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Identity; + + fn make_peer_identity() -> PeerIdentity { + let identity = Identity::generate(); + PeerIdentity::from_pubkey(identity.pubkey()) + } + + #[test] + fn test_handshake_state_properties() { + assert!(HandshakeState::AwaitingHello.is_in_progress()); + assert!(HandshakeState::SentHello.is_in_progress()); + assert!(HandshakeState::SentAuth.is_in_progress()); + assert!(HandshakeState::AwaitingAuthAck.is_in_progress()); + assert!(!HandshakeState::Complete.is_in_progress()); + assert!(!HandshakeState::Failed.is_in_progress()); + + assert!(HandshakeState::Complete.is_complete()); + assert!(HandshakeState::Failed.is_failed()); + + assert!(HandshakeState::SentHello.is_initiator()); + assert!(HandshakeState::AwaitingAuthAck.is_initiator()); + assert!(HandshakeState::AwaitingHello.is_responder()); + assert!(HandshakeState::SentAuth.is_responder()); + } + + #[test] + fn test_outbound_connection() { + let identity = make_peer_identity(); + let conn = PeerConnection::outbound(LinkId::new(1), identity.clone(), 1000); + + assert!(conn.is_outbound()); + assert!(!conn.is_inbound()); + assert_eq!(conn.handshake_state(), HandshakeState::SentHello); + assert!(conn.expected_identity().is_some()); + assert_eq!(conn.started_at(), 1000); + assert_eq!(conn.retry_count(), 0); + } + + #[test] + fn test_inbound_connection() { + let conn = PeerConnection::inbound(LinkId::new(2), 2000); + + assert!(conn.is_inbound()); + assert!(!conn.is_outbound()); + assert_eq!(conn.handshake_state(), HandshakeState::AwaitingHello); + assert!(conn.expected_identity().is_none()); + assert_eq!(conn.started_at(), 2000); + } + + #[test] + fn test_outbound_handshake_flow() { + let identity = make_peer_identity(); + let mut conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + + // Initial state: SentHello + assert_eq!(conn.handshake_state(), HandshakeState::SentHello); + assert!(conn.is_in_progress()); + + // Received response, awaiting auth ack + conn.mark_awaiting_auth_ack(1100); + assert_eq!(conn.handshake_state(), HandshakeState::AwaitingAuthAck); + assert!(conn.is_in_progress()); + + // Complete + conn.mark_complete(1200); + assert!(conn.is_complete()); + assert!(!conn.is_in_progress()); + } + + #[test] + fn test_inbound_handshake_flow() { + let mut conn = PeerConnection::inbound(LinkId::new(2), 2000); + + // Initial state: AwaitingHello + assert_eq!(conn.handshake_state(), HandshakeState::AwaitingHello); + + // Received Hello, learned identity + let identity = make_peer_identity(); + conn.mark_hello_received(identity, 2100); + assert!(conn.expected_identity().is_some()); + + // Sent Auth response + conn.mark_auth_sent(2200); + assert_eq!(conn.handshake_state(), HandshakeState::SentAuth); + + // Complete + conn.mark_complete(2300); + assert!(conn.is_complete()); + } + + #[test] + fn test_connection_timing() { + let identity = make_peer_identity(); + let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + + assert_eq!(conn.duration(1500), 500); + assert_eq!(conn.idle_time(1500), 500); + assert!(!conn.is_timed_out(1500, 1000)); + assert!(conn.is_timed_out(2500, 1000)); + } + + #[test] + fn test_retry_tracking() { + let identity = make_peer_identity(); + let mut conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + + assert_eq!(conn.retry_count(), 0); + assert!(!conn.max_retries_exceeded(3)); + + conn.increment_retry(); + conn.increment_retry(); + conn.increment_retry(); + + assert_eq!(conn.retry_count(), 3); + assert!(conn.max_retries_exceeded(3)); + } + + #[test] + fn test_connection_failure() { + let identity = make_peer_identity(); + let mut conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + + conn.mark_failed(); + assert!(conn.is_failed()); + assert!(!conn.is_in_progress()); + assert!(!conn.is_complete()); + } +} diff --git a/src/peer/mod.rs b/src/peer/mod.rs new file mode 100644 index 0000000..bbd65fe --- /dev/null +++ b/src/peer/mod.rs @@ -0,0 +1,385 @@ +//! Peer Management +//! +//! Two-phase peer lifecycle: +//! 1. **PeerConnection** - Handshake phase, before identity is verified +//! 2. **ActivePeer** - Authenticated phase, after successful Noise handshake +//! +//! The PeerSlot enum represents either phase, enabling unified storage +//! while maintaining type safety for phase-specific operations. + +mod active; +mod connection; + +pub use active::{ActivePeer, ConnectivityState}; +pub use connection::{HandshakeState, PeerConnection}; + +use crate::transport::LinkId; +use crate::NodeId; +use std::fmt; +use thiserror::Error; + +// ============================================================================ +// Errors +// ============================================================================ + +/// Errors related to peer operations. +#[derive(Debug, Error)] +pub enum PeerError { + #[error("peer not authenticated")] + NotAuthenticated, + + #[error("peer not found: {0:?}")] + NotFound(NodeId), + + #[error("connection not found: {0}")] + ConnectionNotFound(LinkId), + + #[error("peer already exists: {0:?}")] + AlreadyExists(NodeId), + + #[error("handshake failed: {0}")] + HandshakeFailed(String), + + #[error("handshake timeout")] + HandshakeTimeout, + + #[error("identity mismatch: expected {expected:?}, got {actual:?}")] + IdentityMismatch { expected: NodeId, actual: NodeId }, + + #[error("peer disconnected")] + Disconnected, + + #[error("max connections exceeded: {max}")] + MaxConnectionsExceeded { max: usize }, + + #[error("max peers exceeded: {max}")] + MaxPeersExceeded { max: usize }, +} + +// ============================================================================ +// Cross-Connection Handling +// ============================================================================ + +/// Result of attempting to promote a connection to active peer. +/// +/// When a handshake completes, we may discover that we already have a +/// connection to this peer (cross-connection). The tie-breaker rule +/// determines which connection survives. +#[derive(Debug)] +pub enum PromotionResult { + /// New peer created successfully. + Promoted(ActivePeer), + + /// Cross-connection detected. This connection lost the tie-breaker + /// and should be closed. + CrossConnectionLost { + /// The link that won (existing connection). + winner_link_id: LinkId, + }, + + /// Cross-connection detected. This connection won the tie-breaker. + /// The existing connection was replaced. + CrossConnectionWon { + /// The link that lost (previous connection, now closed). + loser_link_id: LinkId, + /// The new active peer. + peer: ActivePeer, + }, +} + +impl PromotionResult { + /// Get the active peer if promotion succeeded. + pub fn peer(&self) -> Option<&ActivePeer> { + match self { + PromotionResult::Promoted(peer) => Some(peer), + PromotionResult::CrossConnectionWon { peer, .. } => Some(peer), + PromotionResult::CrossConnectionLost { .. } => None, + } + } + + /// Check if this connection should be closed. + pub fn should_close_this_connection(&self) -> bool { + matches!(self, PromotionResult::CrossConnectionLost { .. }) + } + + /// Get the link that should be closed, if any. + pub fn link_to_close(&self) -> Option { + match self { + PromotionResult::CrossConnectionLost { .. } => None, // Caller's link + PromotionResult::CrossConnectionWon { loser_link_id, .. } => Some(*loser_link_id), + PromotionResult::Promoted(_) => None, + } + } +} + +/// Determine winner of cross-connection tie-breaker. +/// +/// Rule: The node with the smaller node_id prefers its OUTBOUND connection. +/// This is deterministic and symmetric: both nodes will reach the same conclusion. +/// +/// # Arguments +/// * `our_node_id` - Our node's ID +/// * `their_node_id` - The peer's node ID +/// * `this_is_outbound` - Whether the connection being evaluated is our outbound +/// +/// # Returns +/// `true` if this connection should win (survive), `false` if it should close. +pub fn cross_connection_winner( + our_node_id: &NodeId, + their_node_id: &NodeId, + this_is_outbound: bool, +) -> bool { + let we_are_smaller = our_node_id < their_node_id; + + // Smaller node's outbound wins + // If we're smaller: our outbound wins, our inbound loses + // If they're smaller: our outbound loses, our inbound wins + if we_are_smaller { + this_is_outbound + } else { + !this_is_outbound + } +} + +// ============================================================================ +// PeerSlot +// ============================================================================ + +/// A slot in the peer table, representing either connection or active phase. +#[derive(Clone, Debug)] +pub enum PeerSlot { + /// Connection in handshake phase. + Connecting(PeerConnection), + /// Authenticated peer. + Active(ActivePeer), +} + +impl PeerSlot { + /// Create a new connecting slot (outbound). + pub fn outbound(conn: PeerConnection) -> Self { + PeerSlot::Connecting(conn) + } + + /// Create a new connecting slot (inbound). + pub fn inbound(conn: PeerConnection) -> Self { + PeerSlot::Connecting(conn) + } + + /// Create a new active slot. + pub fn active(peer: ActivePeer) -> Self { + PeerSlot::Active(peer) + } + + /// Check if this is a connecting slot. + pub fn is_connecting(&self) -> bool { + matches!(self, PeerSlot::Connecting(_)) + } + + /// Check if this is an active slot. + pub fn is_active(&self) -> bool { + matches!(self, PeerSlot::Active(_)) + } + + /// Get the link ID for this slot. + pub fn link_id(&self) -> LinkId { + match self { + PeerSlot::Connecting(conn) => conn.link_id(), + PeerSlot::Active(peer) => peer.link_id(), + } + } + + /// Get as connection reference, if connecting. + pub fn as_connection(&self) -> Option<&PeerConnection> { + match self { + PeerSlot::Connecting(conn) => Some(conn), + PeerSlot::Active(_) => None, + } + } + + /// Get as mutable connection reference, if connecting. + pub fn as_connection_mut(&mut self) -> Option<&mut PeerConnection> { + match self { + PeerSlot::Connecting(conn) => Some(conn), + PeerSlot::Active(_) => None, + } + } + + /// Get as active peer reference, if active. + pub fn as_active(&self) -> Option<&ActivePeer> { + match self { + PeerSlot::Active(peer) => Some(peer), + PeerSlot::Connecting(_) => None, + } + } + + /// Get as mutable active peer reference, if active. + pub fn as_active_mut(&mut self) -> Option<&mut ActivePeer> { + match self { + PeerSlot::Active(peer) => Some(peer), + PeerSlot::Connecting(_) => None, + } + } + + /// Get the known node_id, if any. + /// + /// For connections, this is the expected identity (may be None for inbound). + /// For active peers, this is always known. + pub fn node_id(&self) -> Option<&NodeId> { + match self { + PeerSlot::Connecting(conn) => conn.expected_identity().map(|id| id.node_id()), + PeerSlot::Active(peer) => Some(peer.node_id()), + } + } +} + +impl fmt::Display for PeerSlot { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + PeerSlot::Connecting(conn) => { + write!(f, "connecting(link={}, state={})", conn.link_id(), conn.handshake_state()) + } + PeerSlot::Active(peer) => { + write!(f, "active(node={:?}, link={})", peer.node_id(), peer.link_id()) + } + } + } +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::LinkId; + use crate::{Identity, PeerIdentity}; + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + fn make_peer_identity() -> PeerIdentity { + let identity = Identity::generate(); + PeerIdentity::from_pubkey(identity.pubkey()) + } + + #[test] + fn test_cross_connection_smaller_node_wins_outbound() { + let node_a = make_node_id(1); // smaller + let node_b = make_node_id(2); // larger + + // Node A's perspective + assert!(cross_connection_winner(&node_a, &node_b, true)); // A's outbound wins + assert!(!cross_connection_winner(&node_a, &node_b, false)); // A's inbound loses + + // Node B's perspective + assert!(!cross_connection_winner(&node_b, &node_a, true)); // B's outbound loses + assert!(cross_connection_winner(&node_b, &node_a, false)); // B's inbound wins + } + + #[test] + fn test_cross_connection_symmetric() { + let node_a = make_node_id(1); + let node_b = make_node_id(2); + + // A's outbound = B's inbound + let a_outbound_wins = cross_connection_winner(&node_a, &node_b, true); + let b_inbound_wins = cross_connection_winner(&node_b, &node_a, false); + assert_eq!(a_outbound_wins, b_inbound_wins); + + // A's inbound = B's outbound + let a_inbound_wins = cross_connection_winner(&node_a, &node_b, false); + let b_outbound_wins = cross_connection_winner(&node_b, &node_a, true); + assert_eq!(a_inbound_wins, b_outbound_wins); + + // Exactly one survives + assert!(a_outbound_wins != a_inbound_wins); + } + + #[test] + fn test_peer_slot_connecting() { + let identity = make_peer_identity(); + let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + let slot = PeerSlot::Connecting(conn); + + assert!(slot.is_connecting()); + assert!(!slot.is_active()); + assert!(slot.as_connection().is_some()); + assert!(slot.as_active().is_none()); + assert_eq!(slot.link_id(), LinkId::new(1)); + } + + #[test] + fn test_peer_slot_active() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(2), 2000); + let slot = PeerSlot::Active(peer); + + assert!(!slot.is_connecting()); + assert!(slot.is_active()); + assert!(slot.as_connection().is_none()); + assert!(slot.as_active().is_some()); + assert_eq!(slot.link_id(), LinkId::new(2)); + } + + #[test] + fn test_promotion_result_promoted() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(1), 1000); + let result = PromotionResult::Promoted(peer); + + assert!(result.peer().is_some()); + assert!(!result.should_close_this_connection()); + assert!(result.link_to_close().is_none()); + } + + #[test] + fn test_promotion_result_cross_lost() { + let result = PromotionResult::CrossConnectionLost { + winner_link_id: LinkId::new(1), + }; + + assert!(result.peer().is_none()); + assert!(result.should_close_this_connection()); + assert!(result.link_to_close().is_none()); // Caller closes their own + } + + #[test] + fn test_promotion_result_cross_won() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(2), 2000); + let result = PromotionResult::CrossConnectionWon { + loser_link_id: LinkId::new(1), + peer, + }; + + assert!(result.peer().is_some()); + assert!(!result.should_close_this_connection()); + assert_eq!(result.link_to_close(), Some(LinkId::new(1))); + } + + #[test] + fn test_peer_slot_node_id() { + // Outbound connection knows expected identity + let identity = make_peer_identity(); + let expected_node_id = *identity.node_id(); + let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); + let slot = PeerSlot::Connecting(conn); + assert_eq!(slot.node_id(), Some(&expected_node_id)); + + // Inbound connection doesn't know identity yet + let conn_inbound = PeerConnection::inbound(LinkId::new(2), 2000); + let slot_inbound = PeerSlot::Connecting(conn_inbound); + assert!(slot_inbound.node_id().is_none()); + + // Active peer always knows identity + let identity2 = make_peer_identity(); + let active_node_id = *identity2.node_id(); + let peer = ActivePeer::new(identity2, LinkId::new(3), 3000); + let slot_active = PeerSlot::Active(peer); + assert_eq!(slot_active.node_id(), Some(&active_node_id)); + } +}