laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 00:04:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement PeerSlot architecture with two-phase peer lifecycle

Browse Source 
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
This commit is contained in:
Johnathan Corgan
2026-02-01 00:50:47 +00:00
parent a5a62b3768
commit 86c49dfb04
6 changed files with 1725 additions and 774 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib.rs
+4 -1
View File
@@ -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};
src/node.rs
+384 -82
View File
@@ -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<TransportId, TransportHandle>,
/// Active links.
links: HashMap<LinkId, Link>,
/// Reverse lookup: (transport_id, remote_addr) -> link_id.
addr_to_link: HashMap<AddrKey, LinkId>,
// === Packet Channel ===
/// Packet sender for transports.
@@ -157,11 +182,19 @@ pub struct Node {
/// Packet receiver (for event loop).
packet_rx: Option<PacketRx>,
// === Peers ===
// === Connections (Handshake Phase) ===
/// Pending connections (handshake in progress).
/// Indexed by LinkId since we don't know the peer's identity yet.
connections: HashMap<LinkId, PeerConnection>,
// === Peers (Active Phase) ===
/// Authenticated peers.
peers: HashMap<NodeId, Peer>,
/// Indexed by NodeId (verified identity).
peers: HashMap<NodeId, ActivePeer>,
// === 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<LinkId> {
self.addr_to_link.get(&(transport_id, addr.clone())).copied()
}
/// Remove a link.
pub fn remove_link(&mut self, link_id: &LinkId) -> Option<Link> {
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<PeerConnection> {
self.connections.remove(link_id)
}
/// Iterate over all connections.
pub fn connections(&self) -> impl Iterator<Item = &PeerConnection> {
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<PromotionResult, NodeError> {
// 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<Peer> {
pub fn remove_peer(&mut self, node_id: &NodeId) -> Option<ActivePeer> {
self.peers.remove(node_id)
}
/// Iterate over all peers.
pub fn peers(&self) -> impl Iterator<Item = &Peer> {
pub fn peers(&self) -> impl Iterator<Item = &ActivePeer> {
self.peers.values()
}
@@ -668,14 +872,14 @@ impl Node {
self.peers.keys()
}
/// Iterate over all active peers.
pub fn active_peers(&self) -> impl Iterator<Item = &Peer> {
self.peers.values().filter(|p| p.state().is_active())
/// Iterate over peers that can send traffic.
pub fn sendable_peers(&self) -> impl Iterator<Item = &ActivePeer> {
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));
}
}
src/peer.rs
-691
View File
@@ -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<ParentDeclaration>,
/// Their path to root.
ancestry: Option<TreeCoordinate>,
// === Bloom Filter ===
/// What's reachable through them (inbound filter).
inbound_filter: Option<BloomFilter>,
/// 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<u64>,
/// 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<u64> {
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<u64> {
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<u64>,
/// 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<u64> {
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);
}
}
src/peer/active.rs
+518
View File
@@ -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<ParentDeclaration>,
/// Their path to root.
ancestry: Option<TreeCoordinate>,
// === Bloom Filter ===
/// What's reachable through them (inbound filter).
inbound_filter: Option<BloomFilter>,
/// 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);
}
}
src/peer/connection.rs
+434
View File
@@ -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<PeerIdentity>,
// === Noise Session State ===
// TODO: Add actual Noise protocol state when implementing crypto
// noise_state: Option<NoiseSession>,
// === 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());
}
}
src/peer/mod.rs
+385
View File
@@ -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<LinkId> {
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));
}
}