laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge master into refactor-hotpath

Browse Source 
Bring the runtime peer-list refresh and opt-in mDNS LAN discovery work
on master into the receive-path RejectReason / reloadable-config
integration branch. Code files auto-merge clean; the only conflict is
the CHANGELOG Unreleased section, resolved as the union of both sets of
entries.
This commit is contained in:
Johnathan Corgan
2026-05-30 01:43:31 +00:00
parent d672ed865f 7d7b551ca1
commit 3c5d9fd4f2
17 changed files with 1812 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+30
View File
@@ -25,6 +25,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
direct counters at the bloom / discovery / forwarding sites are
retained alongside the new dispatch while the rollout is in progress;
a later change collapses the duplicate increment.
- `Node::update_peers` for runtime peer-list refresh, returning an
`UpdatePeersOutcome` summarizing added, removed, and retained peers.
Re-derives active peer connections from a new peer configuration
without dropping links to peers that remain in the set.
`PeerAddress` gains a `seen_at_ms` recency field (with
`with_seen_at_ms`) used to prefer more recently observed addresses.
- Opt-in mDNS / DNS-SD LAN discovery for sub-second pairing of peers on
the same local link, without a relay or NAT-traversal roundtrip.
Disabled by default; operators enable it with
`node.discovery.lan.enabled: true`. Configurable service type and an
optional `node.discovery.lan.scope` that isolates discovery to peers
sharing the same private-network scope. The advertised UDP port is
chosen from a non-bootstrap operational UDP transport using a stable
selector, so it is deterministic across restarts.
- `pool_inbound` and `pool_outbound` counters on the TCP and Tor
transport stats (`TcpStats`, `TorStats`). Per-direction accounting
is updated at every pool-insert and receive-loop-exit site, plus on
@@ -41,6 +55,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- `complete_rekey_msg2` now returns the remote peer's startup epoch
alongside the new Noise session, so the rekey path can detect a peer
restart and clear stale session state.
- Active-peer path selection now sorts address candidates by recency
(`seen_at_ms`), preferring the most recently observed address when
racing concurrent path probes.
- Per-tick work budgets bound the connection churn done in a single
node tick: `MAX_DISCOVERY_CONNECTS_PER_TICK`,
`MAX_RETRY_CONNECTIONS_PER_TICK`, and
`MAX_PARALLEL_PATH_CANDIDATES_PER_PEER`. Work beyond a tick's budget
is deferred to the next tick rather than discarded.
- Nostr discovery startup is now non-blocking. `Node::start` no
longer waits for relay connect, subscribe, or initial advert
publish before returning. A slow or unreachable relay no longer
@@ -154,6 +179,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- A stale FSP (session-layer) session is now cleared when a peer
restart is detected during FMP rekey or cross-connection promotion.
Previously the old session could linger after the peer came back
with a new startup epoch, leaving the session-layer map out of sync
with the freshly promoted peer.
- TCP and Tor `max_inbound_connections` admission cap is now compared
against the per-direction inbound count (`pool_inbound`) rather than
the combined pool size. Outbound connect-on-send connections share
Cargo.lock
Generated
+141 -10
View File
@@ -1083,6 +1083,7 @@ dependencies = [
"hex",
"hkdf",
"libc",
"mdns-sd",
"nostr",
"nostr-sdk",
"portable-atomic",
@@ -1116,6 +1117,17 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
[[package]]
name = "flume"
version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
dependencies = [
"futures-core",
"futures-sink",
"spin",
]
[[package]]
name = "fnv"
version = "1.0.7"
@@ -1515,6 +1527,16 @@ dependencies = [
"icu_properties",
]
[[package]]
name = "if-addrs"
version = "0.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b"
dependencies = [
"libc",
"windows-sys 0.61.2",
]
[[package]]
name = "indexmap"
version = "2.14.0"
@@ -1774,6 +1796,21 @@ dependencies = [
"regex-automata",
]
[[package]]
name = "mdns-sd"
version = "0.19.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2bb8ce26633738d98ffcef71ec58bff967c6675be50229823c2835f6316e67e"
dependencies = [
"fastrand",
"flume",
"if-addrs",
"log",
"mio",
"socket-pktinfo",
"socket2",
]
[[package]]
name = "memchr"
version = "2.8.0"
@@ -3018,6 +3055,17 @@ version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
[[package]]
name = "socket-pktinfo"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "927136cc2ae6a1b0e66ac6b1210902b75c3f726db004a73bc18686dcd0dcd22f"
dependencies = [
"libc",
"socket2",
"windows-sys 0.60.2",
]
[[package]]
name = "socket2"
version = "0.6.3"
@@ -3028,6 +3076,15 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "spin"
version = "0.9.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
dependencies = [
"lock_api",
]
[[package]]
name = "stable_deref_trait"
version = "1.2.1"
@@ -3924,7 +3981,7 @@ version = "0.52.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
dependencies = [
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
@@ -3933,7 +3990,16 @@ version = "0.59.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
dependencies = [
"windows-targets",
"windows-targets 0.52.6",
]
[[package]]
name = "windows-sys"
version = "0.60.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
dependencies = [
"windows-targets 0.53.5",
]
[[package]]
@@ -3951,14 +4017,31 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
dependencies = [
"windows_aarch64_gnullvm",
"windows_aarch64_msvc",
"windows_i686_gnu",
"windows_i686_gnullvm",
"windows_i686_msvc",
"windows_x86_64_gnu",
"windows_x86_64_gnullvm",
"windows_x86_64_msvc",
"windows_aarch64_gnullvm 0.52.6",
"windows_aarch64_msvc 0.52.6",
"windows_i686_gnu 0.52.6",
"windows_i686_gnullvm 0.52.6",
"windows_i686_msvc 0.52.6",
"windows_x86_64_gnu 0.52.6",
"windows_x86_64_gnullvm 0.52.6",
"windows_x86_64_msvc 0.52.6",
]
[[package]]
name = "windows-targets"
version = "0.53.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
dependencies = [
"windows-link",
"windows_aarch64_gnullvm 0.53.1",
"windows_aarch64_msvc 0.53.1",
"windows_i686_gnu 0.53.1",
"windows_i686_gnullvm 0.53.1",
"windows_i686_msvc 0.53.1",
"windows_x86_64_gnu 0.53.1",
"windows_x86_64_gnullvm 0.53.1",
"windows_x86_64_msvc 0.53.1",
]
[[package]]
@@ -3967,48 +4050,96 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
[[package]]
name = "windows_aarch64_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
[[package]]
name = "windows_aarch64_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
[[package]]
name = "windows_i686_gnu"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
[[package]]
name = "windows_i686_gnu"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
[[package]]
name = "windows_i686_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
[[package]]
name = "windows_i686_gnullvm"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
[[package]]
name = "windows_i686_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
[[package]]
name = "windows_i686_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
[[package]]
name = "windows_x86_64_gnu"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
[[package]]
name = "windows_x86_64_gnu"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
[[package]]
name = "windows_x86_64_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
[[package]]
name = "windows_x86_64_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
[[package]]
name = "winreg"
version = "0.55.0"
Cargo.toml
+1
View File
@@ -29,6 +29,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tokio = { version = "1", features = ["rt", "macros", "signal", "sync", "net", "time", "process", "io-util"] }
futures = "0.3"
simple-dns = "0.11.2"
mdns-sd = "0.19"
socket2 = { version = "0.6.2", features = ["all"] }
tokio-socks = "0.5"
portable-atomic = { version = "1", features = ["std"] }
src/config/node.rs
+10
View File
@@ -219,6 +219,12 @@ pub struct DiscoveryConfig {
/// Nostr-mediated overlay endpoint discovery.
#[serde(default = "DiscoveryConfig::default_nostr")]
pub nostr: NostrDiscoveryConfig,
/// mDNS / DNS-SD peer discovery on the local link. Identity surface
/// is a strict subset of what `nostr.advertise` already publishes
/// publicly, so there's no marginal privacy cost; the latency win
/// for same-LAN peers is large (sub-second pairing, no relay).
#[serde(default = "DiscoveryConfig::default_lan")]
pub lan: crate::discovery::lan::LanDiscoveryConfig,
}
impl Default for DiscoveryConfig {
@@ -231,6 +237,7 @@ impl Default for DiscoveryConfig {
backoff_max_secs: 0,
forward_min_interval_secs: 2,
nostr: NostrDiscoveryConfig::default(),
lan: crate::discovery::lan::LanDiscoveryConfig::default(),
}
}
}
@@ -257,6 +264,9 @@ impl DiscoveryConfig {
fn default_nostr() -> NostrDiscoveryConfig {
NostrDiscoveryConfig::default()
}
fn default_lan() -> crate::discovery::lan::LanDiscoveryConfig {
crate::discovery::lan::LanDiscoveryConfig::default()
}
}
/// Nostr advert discovery policy.
src/config/peer.rs
+29
View File
@@ -45,8 +45,28 @@ pub struct PeerAddress {
/// are tried first.
#[serde(default = "default_priority")]
pub priority: u8,
/// Wall-clock observation timestamp (Unix ms) for ranking by recency.
///
/// `None` means "no freshness signal", typically an operator-edited
/// static config. The dialer sorts candidates by this field descending
/// so freshly observed overlay or runtime hints can be tried before stale
/// static addresses. This field is runtime-only and is ignored when
/// comparing peer-address lists for config changes.
#[serde(default, skip_serializing_if = "Option::is_none", skip_deserializing)]
pub seen_at_ms: Option<u64>,
}
impl PartialEq for PeerAddress {
fn eq(&self, other: &Self) -> bool {
self.transport == other.transport
&& self.addr == other.addr
&& self.priority == other.priority
}
}
impl Eq for PeerAddress {}
fn default_priority() -> u8 {
100
}
@@ -62,6 +82,7 @@ impl PeerAddress {
transport: transport.into(),
addr: addr.into(),
priority: default_priority(),
seen_at_ms: None,
}
}
@@ -75,8 +96,16 @@ impl PeerAddress {
transport: transport.into(),
addr: addr.into(),
priority,
seen_at_ms: None,
}
}
/// Tag this address with a freshness timestamp for source-agnostic
/// candidate ranking.
pub fn with_seen_at_ms(mut self, seen_at_ms: u64) -> Self {
self.seen_at_ms = Some(seen_at_ms);
self
}
}
/// Configuration for a known peer.
src/discovery.rs
+1
View File
@@ -6,6 +6,7 @@
//! it hands the live socket and selected remote endpoint to FIPS so the
//! existing Noise/FMP transport path can take over.
pub mod lan;
pub mod nostr;
use crate::config::UdpConfig;
src/discovery/lan/mod.rs
+368
View File
@@ -0,0 +1,368 @@
//! LAN peer discovery via mDNS / DNS-SD (RFC 6762 / RFC 6763).
//!
//! Publishes a `_fips._udp.local.` service advert carrying our `npub` and
//! optional discovery scope on the local link, and concurrently browses for the
//! same service type to learn peers reachable on the same broadcast
//! domain. The result is sub-second peer pairing without any Nostr-relay
//! roundtrip, STUN observation, or NAT traversal — the observed
//! endpoint is by construction routable from the consumer's LAN.
//!
//! ## Trust model
//!
//! mDNS adverts are unauthenticated: anyone on the LAN can multicast a
//! TXT carrying `npub=...`. Identity is still proven end-to-end by the
//! Noise XX handshake the Node initiates against the observed endpoint
//! — a spoofed advert with another peer's npub fails the handshake and
//! is silently dropped. Treat the mDNS advert as a routing hint, not as
//! identity. LAN discovery is link-local mDNS only. It is not a Nostr advert
//! and does not leave the broadcast domain unless the operator's LAN bridges
//! mDNS.
//!
//! ## Scope filtering
//!
//! When a `discovery_scope` is configured, the advert carries it in a
//! `scope=<name>` TXT entry and the browser only surfaces peers with a
//! matching scope. Nodes on the same physical LAN but configured for
//! different mesh networks don't cross-feed each other.
use std::collections::HashMap;
use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc;
use std::time::Instant;
use mdns_sd::{ScopedIp, ServiceDaemon, ServiceEvent, ServiceInfo};
use thiserror::Error;
use tokio::sync::Mutex;
use tracing::{debug, info, warn};
use crate::Identity;
/// DNS-SD service type for the FIPS LAN advert. RFC 6763 §4.1.2: must
/// end with `.local.`. The `_udp` is the IP transport, not the upper
/// protocol — both UDP and TCP FIPS endpoints announce under the same
/// service type because the link-layer punch/handshake travels over UDP
/// either way.
pub const SERVICE_TYPE: &str = "_fips._udp.local.";
/// TXT key carrying the bech32-encoded npub of the publishing node.
pub const TXT_KEY_NPUB: &str = "npub";
/// TXT key carrying the publishing node's `discovery_scope`, if any.
pub const TXT_KEY_SCOPE: &str = "scope";
/// TXT key carrying the FIPS protocol version (matches the Nostr advert
/// `PROTOCOL_VERSION`).
pub const TXT_KEY_VERSION: &str = "v";
#[derive(Debug, Error)]
pub enum LanDiscoveryError {
#[error("mDNS daemon init failed: {0}")]
Daemon(String),
#[error("mDNS register failed: {0}")]
Register(String),
#[error("mDNS browse failed: {0}")]
Browse(String),
#[error("no advertised UDP port — start a UDP transport first")]
NoAdvertisedPort,
#[error("LAN discovery disabled in config")]
Disabled,
}
/// A peer we learned about via mDNS. Identity is unverified at this
/// point; the Node initiates a Noise XX handshake against `addr` to
/// confirm `npub` actually controls the matching private key.
#[derive(Debug, Clone)]
pub struct LanDiscoveredPeer {
pub npub: String,
pub scope: Option<String>,
pub addr: SocketAddr,
pub observed_at: Instant,
}
/// Browser-side events surfaced by `LanDiscovery::drain_events`.
#[derive(Debug, Clone)]
pub enum LanEvent {
Discovered(LanDiscoveredPeer),
}
/// Runtime configuration for the mDNS responder + browser.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LanDiscoveryConfig {
/// Master switch. Default: `false` — LAN discovery is opt-in. Operators
/// who want sub-second same-LAN pairing enable it via
/// `node.discovery.lan.enabled: true`. Default-off avoids reintroducing
/// a per-LAN identity broadcast on nodes that have deliberately disabled
/// other discovery channels, and avoids any multicast surprise on upgrade.
#[serde(default = "LanDiscoveryConfig::default_enabled")]
pub enabled: bool,
/// Overridable service type, primarily so integration tests can run
/// multiple isolated services on the same loopback interface.
#[serde(default = "LanDiscoveryConfig::default_service_type")]
pub service_type: String,
/// Optional application/network scope carried in the LAN-only TXT
/// record. Browsers that set a scope ignore adverts for other scopes.
///
/// This is intentionally separate from Nostr discovery's public `app`
/// tag so applications can keep relay-visible adverts generic while
/// still isolating LAN discovery per private network.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub scope: Option<String>,
}
impl Default for LanDiscoveryConfig {
fn default() -> Self {
Self {
enabled: Self::default_enabled(),
service_type: Self::default_service_type(),
scope: None,
}
}
}
impl LanDiscoveryConfig {
fn default_enabled() -> bool {
false
}
fn default_service_type() -> String {
SERVICE_TYPE.to_string()
}
}
/// Running mDNS responder + browser bound to the node's UDP advert port.
pub struct LanDiscovery {
daemon: ServiceDaemon,
own_npub: String,
instance_fullname: String,
events_rx: Mutex<tokio::sync::mpsc::UnboundedReceiver<LanEvent>>,
event_pump: tokio::task::JoinHandle<()>,
}
impl LanDiscovery {
/// Start the mDNS responder and browser.
///
/// `advertised_port` is the UDP port the operational UDP transport
/// is bound to — peers receiving our advert will initiate Noise XX
/// against that port. `scope` mirrors the Nostr discovery scope and
/// is used to filter the browser stream.
pub async fn start(
identity: &Identity,
scope: Option<String>,
advertised_port: u16,
config: LanDiscoveryConfig,
) -> Result<Arc<Self>, LanDiscoveryError> {
if !config.enabled {
return Err(LanDiscoveryError::Disabled);
}
if advertised_port == 0 {
return Err(LanDiscoveryError::NoAdvertisedPort);
}
let daemon = ServiceDaemon::new().map_err(|e| LanDiscoveryError::Daemon(e.to_string()))?;
let npub = identity.npub();
// mDNS DNS labels are capped at 63 bytes. 16 bech32 chars of npub
// give 80 bits of effective entropy — collisions on a single LAN
// are vanishingly unlikely. Prefixed for human-readable logs.
let label_npub = &npub[..16.min(npub.len())];
let instance_name = format!("fips-{label_npub}");
let host_name = format!("{instance_name}.local.");
let mut props: HashMap<String, String> = HashMap::new();
props.insert(TXT_KEY_NPUB.to_string(), npub.clone());
if let Some(s) = scope.as_deref()
&& !s.is_empty()
{
props.insert(TXT_KEY_SCOPE.to_string(), s.to_string());
}
props.insert(
TXT_KEY_VERSION.to_string(),
super::nostr::PROTOCOL_VERSION.to_string(),
);
// host_ipv4 is set to "127.0.0.1" *and* enable_addr_auto() is
// called: the loopback seed makes the advert resolve for
// same-host peers (and same-host integration tests) while the
// auto-flag still appends every non-loopback interface address
// mdns-sd discovers. Belt-and-braces because addr_auto alone
// skips loopback by default on some platforms.
let service_info = ServiceInfo::new(
&config.service_type,
&instance_name,
&host_name,
"127.0.0.1",
advertised_port,
Some(props),
)
.map_err(|e| LanDiscoveryError::Register(e.to_string()))?
.enable_addr_auto();
let instance_fullname = service_info.get_fullname().to_string();
daemon
.register(service_info)
.map_err(|e| LanDiscoveryError::Register(e.to_string()))?;
let browse_rx = daemon
.browse(&config.service_type)
.map_err(|e| LanDiscoveryError::Browse(e.to_string()))?;
let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel();
let own_npub = npub.clone();
let scope_filter = scope.clone().filter(|s| !s.is_empty());
let event_pump = tokio::spawn(async move {
// mdns-sd browse returns a flume::Receiver; pump until the
// daemon shuts down and the channel closes.
loop {
let event = match browse_rx.recv_async().await {
Ok(e) => e,
Err(_) => break,
};
match event {
ServiceEvent::ServiceResolved(info) => {
let mut peer_npub: Option<String> = None;
let mut peer_scope: Option<String> = None;
for prop in info.get_properties().iter() {
match prop.key() {
TXT_KEY_NPUB => {
peer_npub = Some(prop.val_str().to_string());
}
TXT_KEY_SCOPE => {
peer_scope = Some(prop.val_str().to_string());
}
_ => {}
}
}
let Some(peer_npub) = peer_npub else {
debug!(
instance = info.get_fullname(),
"lan: skip advert without npub TXT"
);
continue;
};
if peer_npub == own_npub {
// Our own advert echoed back on a loopback
// or multi-homed interface.
continue;
}
if scope_filter.is_some() && scope_filter != peer_scope {
debug!(
npub = %short(&peer_npub),
their_scope = ?peer_scope,
our_scope = ?scope_filter,
"lan: skip cross-scope advert"
);
continue;
}
let port = info.get_port();
if port == 0 {
continue;
}
let observed_at = Instant::now();
// mdns-sd may report multiple interface IPs for
// a multi-homed responder. Surface all routable
// candidates — the Node side filters/dedups and
// only dials addresses compatible with an active
// UDP socket family. IPv6 link-local addresses
// require an interface scope; preserve it when
// mdns-sd provides one, and skip unusable
// scope-less link-local records.
for scoped in info.get_addresses() {
let Some(addr) = socket_addr_from_scoped_ip(scoped, port) else {
debug!(
npub = %short(&peer_npub),
addr = %scoped.to_ip_addr(),
"lan: skip scope-less IPv6 link-local advert"
);
continue;
};
if events_tx
.send(LanEvent::Discovered(LanDiscoveredPeer {
npub: peer_npub.clone(),
scope: peer_scope.clone(),
addr,
observed_at,
}))
.is_err()
{
return;
}
}
}
ServiceEvent::ServiceRemoved(_, fullname) => {
debug!(fullname = %fullname, "lan: service removed");
}
other => {
debug!(?other, "lan: mDNS event");
}
}
}
});
info!(
instance = %instance_fullname,
port = advertised_port,
scope = ?scope,
"lan: mDNS discovery started"
);
Ok(Arc::new(Self {
daemon,
own_npub: npub,
instance_fullname,
events_rx: Mutex::new(events_rx),
event_pump,
}))
}
/// Bech32 npub published by this node.
pub fn own_npub(&self) -> &str {
&self.own_npub
}
/// Drain pending browser events. Called once per Node tick.
pub async fn drain_events(&self) -> Vec<LanEvent> {
let mut rx = self.events_rx.lock().await;
let mut events = Vec::new();
while let Ok(event) = rx.try_recv() {
events.push(event);
}
events
}
/// Tear down the responder, browser, and event pump.
pub async fn shutdown(self: &Arc<Self>) {
if let Err(e) = self.daemon.unregister(&self.instance_fullname) {
warn!(error = %e, "lan: unregister failed");
}
if let Err(e) = self.daemon.shutdown() {
warn!(error = %e, "lan: daemon shutdown failed");
}
self.event_pump.abort();
}
}
fn short(npub: &str) -> &str {
let end = 16.min(npub.len());
&npub[..end]
}
fn socket_addr_from_scoped_ip(scoped: &ScopedIp, port: u16) -> Option<SocketAddr> {
match scoped {
ScopedIp::V4(v4) => Some(SocketAddr::V4(SocketAddrV4::new(*v4.addr(), port))),
ScopedIp::V6(v6) => {
let ip = *v6.addr();
let scope_id = v6.scope_id().index;
if ipv6_is_unicast_link_local(ip) && scope_id == 0 {
return None;
}
Some(SocketAddr::V6(SocketAddrV6::new(ip, port, 0, scope_id)))
}
_ => None,
}
}
fn ipv6_is_unicast_link_local(ip: std::net::Ipv6Addr) -> bool {
(ip.segments()[0] & 0xffc0) == 0xfe80
}
#[cfg(test)]
mod tests;
src/discovery/lan/tests.rs
+213
View File
@@ -0,0 +1,213 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::time::Duration;
use crate::Identity;
use mdns_sd::ScopedIp;
use super::{LanDiscovery, LanDiscoveryConfig, LanEvent};
/// Distinct service type per test run so concurrent cargo-test workers
/// on the same machine don't cross-feed each other's adverts via the
/// shared 224.0.0.251 multicast group. The trailing `.local.` is
/// required by RFC 6763 — mdns-sd will reject anything else.
fn isolated_service_type(tag: &str) -> String {
let rand: u32 = rand::random();
format!("_fipstest-{tag}-{rand:08x}._udp.local.")
}
fn config_for(service_type: String) -> LanDiscoveryConfig {
LanDiscoveryConfig {
enabled: true,
service_type,
scope: None,
}
}
#[test]
fn scoped_ipv4_advert_becomes_socket_addr() {
let scoped = ScopedIp::from(IpAddr::V4(Ipv4Addr::new(192, 168, 178, 91)));
let addr = super::socket_addr_from_scoped_ip(&scoped, 51820);
assert_eq!(addr, Some(SocketAddr::from(([192, 168, 178, 91], 51820))));
}
#[test]
fn scope_less_ipv6_link_local_advert_is_skipped() {
let scoped = ScopedIp::from(IpAddr::V6("fe80::32c5:99ff:fea7:5fe9".parse().unwrap()));
assert!(super::socket_addr_from_scoped_ip(&scoped, 51820).is_none());
}
#[test]
fn non_link_local_ipv6_advert_is_preserved() {
let scoped = ScopedIp::from(IpAddr::V6(Ipv6Addr::LOCALHOST));
let addr = super::socket_addr_from_scoped_ip(&scoped, 51820);
assert_eq!(addr, Some("[::1]:51820".parse().unwrap()));
}
async fn wait_for_peer(
discovery: &LanDiscovery,
expected_npub: &str,
timeout: Duration,
) -> Option<super::LanDiscoveredPeer> {
let deadline = tokio::time::Instant::now() + timeout;
while tokio::time::Instant::now() < deadline {
for event in discovery.drain_events().await {
let LanEvent::Discovered(peer) = event;
if peer.npub == expected_npub {
return Some(peer);
}
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
None
}
/// Two LanDiscovery instances on isolated service types — `a` browses
/// only its own type and never sees `b`, and vice versa. Sanity check
/// that the scope-isolation defense works (we'd lose isolation if mdns-
/// sd ever leaked across service types).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn isolated_service_types_do_not_cross_feed() {
let identity_a = Identity::generate();
let identity_b = Identity::generate();
let service_a = isolated_service_type("isolated-a");
let service_b = isolated_service_type("isolated-b");
let lan_a = LanDiscovery::start(
&identity_a,
Some("scope-x".to_string()),
61001,
config_for(service_a.clone()),
)
.await
.expect("start a");
let lan_b = LanDiscovery::start(
&identity_b,
Some("scope-x".to_string()),
61002,
config_for(service_b.clone()),
)
.await
.expect("start b");
// Give mDNS multicast time to settle, then confirm neither side saw
// the other (different service type isolates them).
tokio::time::sleep(Duration::from_secs(2)).await;
let saw_b_from_a = wait_for_peer(
&lan_a,
identity_b.npub().as_str(),
Duration::from_millis(500),
)
.await
.is_some();
let saw_a_from_b = wait_for_peer(
&lan_b,
identity_a.npub().as_str(),
Duration::from_millis(500),
)
.await
.is_some();
lan_a.shutdown().await;
lan_b.shutdown().await;
assert!(!saw_b_from_a, "isolated service types must not cross-feed");
assert!(!saw_a_from_b, "isolated service types must not cross-feed");
}
/// Two LanDiscovery instances on the same service type and the same
/// scope: each should observe the other's advert within a few seconds.
/// Exercises the responder + browser + TXT plumbing end-to-end.
///
/// Ignored by default: relies on multicast-loopback semantics that
/// vary across macOS/Linux/Windows when two `ServiceDaemon` instances
/// run in the same process. Real cross-host LAN deployment exercises
/// the same code path correctly — verify with `cargo test -- --ignored
/// matched_scope_peers_observe_each_other` on a setup where this
/// matters, or via end-to-end integration with two daemons.
#[ignore]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn matched_scope_peers_observe_each_other() {
let identity_a = Identity::generate();
let identity_b = Identity::generate();
let service = isolated_service_type("matched");
let lan_a = LanDiscovery::start(
&identity_a,
Some("scope-shared".to_string()),
61101,
config_for(service.clone()),
)
.await
.expect("start a");
let lan_b = LanDiscovery::start(
&identity_b,
Some("scope-shared".to_string()),
61102,
config_for(service.clone()),
)
.await
.expect("start b");
// Loopback mDNS resolution on macOS/Linux takes a moment.
let observed_b =
wait_for_peer(&lan_a, identity_b.npub().as_str(), Duration::from_secs(10)).await;
let observed_a =
wait_for_peer(&lan_b, identity_a.npub().as_str(), Duration::from_secs(10)).await;
lan_a.shutdown().await;
lan_b.shutdown().await;
let observed_b = observed_b.expect("a must see b");
let observed_a = observed_a.expect("b must see a");
assert_eq!(observed_b.scope.as_deref(), Some("scope-shared"));
assert_eq!(observed_a.scope.as_deref(), Some("scope-shared"));
assert_eq!(observed_b.addr.port(), 61102);
assert_eq!(observed_a.addr.port(), 61101);
}
/// Different scopes on the same service type must be filtered out by
/// the browser — peer in scope X does not surface to a browser in
/// scope Y, even if both adverts arrive on the same multicast group.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cross_scope_advert_is_filtered() {
let identity_a = Identity::generate();
let identity_b = Identity::generate();
let service = isolated_service_type("cross-scope");
let lan_a = LanDiscovery::start(
&identity_a,
Some("scope-a".to_string()),
61201,
config_for(service.clone()),
)
.await
.expect("start a");
let lan_b = LanDiscovery::start(
&identity_b,
Some("scope-b".to_string()),
61202,
config_for(service.clone()),
)
.await
.expect("start b");
tokio::time::sleep(Duration::from_secs(3)).await;
let saw_b = wait_for_peer(
&lan_a,
identity_b.npub().as_str(),
Duration::from_millis(500),
)
.await;
lan_a.shutdown().await;
lan_b.shutdown().await;
assert!(saw_b.is_none(), "cross-scope advert must be filtered");
}
src/lib.rs
+1 -1
View File
@@ -67,4 +67,4 @@ pub use peer::{
};
// Re-export node types
pub use node::{Node, NodeError, NodeState};
pub use node::{Node, NodeError, NodeState, UpdatePeersOutcome};
src/node/handlers/discovery.rs
+14 -3
View File
@@ -12,6 +12,8 @@ use crate::transport::{TransportAddr, TransportId};
use crate::{NodeAddr, PeerIdentity};
use tracing::{debug, info, trace, warn};
const MAX_RECENT_DISCOVERY_REQUESTS: usize = 4096;
impl Node {
/// Handle an incoming LookupRequest from a peer.
///
@@ -37,6 +39,7 @@ impl Node {
};
let now_ms = Self::now_ms();
self.purge_expired_requests(now_ms);
// Dedup: drop if we've already seen this request_id.
// Also serves as loop protection — tree routing is loop-free,
@@ -53,13 +56,21 @@ impl Node {
return;
}
if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS {
debug!(
request_id = request.request_id,
from = %self.peer_display_name(from),
recent_requests = self.recent_requests.len(),
max_recent_requests = MAX_RECENT_DISCOVERY_REQUESTS,
"Discovery request dedup cache full, dropping LookupRequest"
);
return;
}
// Record for reverse-path forwarding and dedup
self.recent_requests
.insert(request.request_id, RecentRequest::new(*from, now_ms));
// Lazy purge expired entries
self.purge_expired_requests(now_ms);
// Are we the target?
if request.target == *self.node_addr() {
self.stats_mut().discovery.req_target_is_us += 1;
src/node/handlers/handshake.rs
+45 -4
View File
@@ -698,8 +698,15 @@ impl Node {
// Complete the rekey handshake on the ActivePeer
if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
match peer.complete_rekey_msg2(noise_msg2) {
Ok(session) => {
Ok((session, remote_epoch)) => {
let our_index = peer.rekey_our_index().unwrap_or(header.receiver_idx);
let remote_epoch_changed = matches!(
(peer.remote_epoch(), remote_epoch),
(Some(old), Some(new)) if old != new
);
if remote_epoch.is_some() {
peer.set_remote_epoch(remote_epoch);
}
peer.set_pending_session(session, our_index, header.sender_idx);
if let Some(transport_id) = peer.transport_id() {
@@ -707,6 +714,19 @@ impl Node {
.insert((transport_id, our_index.as_u32()), peer_node_addr);
}
if remote_epoch_changed {
if self.sessions.remove(&peer_node_addr).is_some() {
debug!(
peer = %display_name,
"Cleared stale FSP session after peer restart during FMP rekey"
);
}
info!(
peer = %display_name,
"Peer restart detected during FMP rekey, replacing stale endpoint session"
);
}
debug!(
peer = %display_name,
new_our_index = %our_index,
@@ -1083,9 +1103,15 @@ impl Node {
if let Some(existing_peer) = self.peers.get(&peer_node_addr) {
let existing_link_id = existing_peer.link_id();
// Determine which connection wins
let this_wins =
cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound);
let remote_epoch_changed = matches!((existing_peer.remote_epoch(), remote_epoch), (Some(old), Some(new)) if old != new);
// Determine which connection wins. A peer restart (different
// startup epoch) is not a normal cross-connection: the old link
// and FSP sessions are cryptographically stale, so the freshly
// authenticated connection must replace them regardless of the
// tie-breaker direction.
let this_wins = remote_epoch_changed
|| cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound);
if this_wins {
// This connection wins, replace the existing peer
@@ -1111,6 +1137,21 @@ impl Node {
let _ = self.index_allocator.free(old_idx);
}
if remote_epoch_changed {
if self.sessions.remove(&peer_node_addr).is_some() {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Cleared stale FSP session after peer restart during promotion"
);
}
info!(
peer = %self.peer_display_name(&peer_node_addr),
winner_link = %link_id,
loser_link = %loser_link_id,
"Peer restart detected during promotion, replacing stale active peer"
);
}
self.seed_path_mtu_for_link_peer(&peer_node_addr, transport_id, &current_addr);
let mut new_peer = ActivePeer::with_session(
src/node/handlers/rx_loop.rs
+1
View File
@@ -260,6 +260,7 @@ impl Node {
self.reload_host_map().await;
self.poll_pending_connects().await;
self.poll_nostr_discovery().await;
self.poll_lan_discovery().await;
self.resend_pending_handshakes(now_ms).await;
self.resend_pending_rekeys(now_ms).await;
self.resend_pending_session_handshakes(now_ms).await;
src/node/lifecycle.rs
+684 -53
View File
@@ -14,14 +14,185 @@ use crate::protocol::{Disconnect, DisconnectReason};
use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface};
use crate::{NodeAddr, PeerIdentity};
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::thread;
use std::time::Duration;
use tracing::{debug, info, warn};
const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2;
const MAX_PARALLEL_PATH_CANDIDATES_PER_PEER: usize = 4;
const MAX_DISCOVERY_CONNECTS_PER_TICK: usize = 16;
fn socket_addr_families_compatible(local: SocketAddr, remote: SocketAddr) -> bool {
matches!(
(local, remote),
(SocketAddr::V4(_), SocketAddr::V4(_)) | (SocketAddr::V6(_), SocketAddr::V6(_))
)
}
impl Node {
/// Replace the runtime peer list.
///
/// Newly added auto-connect peers are dialed immediately, removed peers
/// are dropped from retry bookkeeping, and existing peers get fresh
/// address hints without tearing down an active link. If an existing peer
/// is already connected and a new concrete candidate appears, FIPS starts
/// an alternate handshake in parallel; promotion switches only after that
/// handshake authenticates.
pub async fn update_peers(
&mut self,
new_peers: Vec<PeerConfig>,
) -> Result<crate::node::UpdatePeersOutcome, NodeError> {
let mut new_by_addr: HashMap<NodeAddr, PeerConfig> =
HashMap::with_capacity(new_peers.len());
for peer in new_peers {
let identity =
PeerIdentity::from_npub(&peer.npub).map_err(|e| NodeError::InvalidPeerNpub {
npub: peer.npub.clone(),
reason: e.to_string(),
})?;
new_by_addr.insert(*identity.node_addr(), peer);
}
let current_by_addr: HashMap<NodeAddr, PeerConfig> = self
.config
.peers()
.iter()
.filter_map(|peer| {
PeerIdentity::from_npub(&peer.npub)
.ok()
.map(|identity| (*identity.node_addr(), peer.clone()))
})
.collect();
let new_addrs: HashSet<_> = new_by_addr.keys().copied().collect();
let current_addrs: HashSet<_> = current_by_addr.keys().copied().collect();
let removed: Vec<_> = current_addrs.difference(&new_addrs).copied().collect();
let added: Vec<_> = new_addrs.difference(&current_addrs).copied().collect();
let kept: Vec<_> = new_addrs.intersection(&current_addrs).copied().collect();
let mut outcome = crate::node::UpdatePeersOutcome::default();
let mut refresh_configs = Vec::new();
for node_addr in &removed {
if self.retry_pending.remove(node_addr).is_some() {
debug!(
peer = %self.peer_display_name(node_addr),
"Dropping retry entry for peer removed from runtime peer list"
);
}
self.peer_aliases.remove(node_addr);
outcome.removed += 1;
}
for node_addr in &kept {
let new_peer = &new_by_addr[node_addr];
let current_peer = &current_by_addr[node_addr];
let changed = new_peer.addresses != current_peer.addresses
|| new_peer.alias != current_peer.alias
|| new_peer.connect_policy != current_peer.connect_policy
|| new_peer.auto_reconnect != current_peer.auto_reconnect
|| new_peer.via_nostr != current_peer.via_nostr;
if changed {
outcome.updated += 1;
if let Some(state) = self.retry_pending.get_mut(node_addr) {
state.peer_config = new_peer.clone();
state.retry_after_ms = Self::now_ms();
}
if let Some(alias) = new_peer.alias.clone() {
self.peer_aliases.insert(*node_addr, alias);
}
} else {
outcome.unchanged += 1;
}
if new_peer.is_auto_connect() && (!new_peer.addresses.is_empty() || new_peer.via_nostr)
{
refresh_configs.push(new_peer.clone());
}
}
let added_configs: Vec<_> = added
.iter()
.map(|node_addr| new_by_addr[node_addr].clone())
.collect();
self.config.peers = new_by_addr.into_values().collect();
for peer_config in added_configs {
outcome.added += 1;
let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else {
continue;
};
let name = peer_config
.alias
.clone()
.unwrap_or_else(|| identity.short_npub());
self.peer_aliases.insert(*identity.node_addr(), name);
self.register_identity(*identity.node_addr(), identity.pubkey_full());
if peer_config.is_auto_connect()
&& let Err(err) = self.initiate_peer_connection(&peer_config).await
{
debug!(
npub = %peer_config.npub,
error = %err,
"Failed to initiate connection for newly added runtime peer"
);
self.schedule_retry(*identity.node_addr(), Self::now_ms());
}
}
for peer_config in refresh_configs {
let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else {
continue;
};
let node_addr = *identity.node_addr();
if self.peers.contains_key(&node_addr) {
match self
.try_active_peer_alternative_addresses(&peer_config, identity)
.await
{
Ok(true) => debug!(
peer = %self.peer_display_name(&node_addr),
"Started alternate-path handshake for active peer"
),
Ok(false) => {}
Err(err) => debug!(
npub = %peer_config.npub,
error = %err,
"Active peer alternate-path refresh did not start"
),
}
} else {
match self.initiate_peer_connection(&peer_config).await {
Ok(()) => {
if let Some(state) = self.retry_pending.get_mut(&node_addr) {
state.peer_config = peer_config;
state.retry_after_ms = Self::now_ms().saturating_add(
self.config.node.rate_limit.handshake_timeout_secs * 1000,
);
}
}
Err(err) => {
debug!(
npub = %peer_config.npub,
error = %err,
"Refreshed peer addresses did not initiate a direct connection"
);
self.schedule_retry(node_addr, Self::now_ms());
}
}
}
}
Ok(outcome)
}
/// Initiate connections to configured static peers.
///
/// For each peer configured with AutoConnect policy, creates a link and
@@ -142,6 +313,50 @@ impl Node {
})
}
fn is_connecting_to_peer_on_path(
&self,
peer_node_addr: &NodeAddr,
transport_id: TransportId,
remote_addr: &TransportAddr,
) -> bool {
self.connections.values().any(|conn| {
conn.expected_identity()
.map(|id| id.node_addr() == peer_node_addr)
.unwrap_or(false)
&& conn.transport_id() == Some(transport_id)
&& conn.source_addr() == Some(remote_addr)
}) || self.pending_connects.iter().any(|pending| {
pending.peer_identity.node_addr() == peer_node_addr
&& pending.transport_id == transport_id
&& pending.remote_addr == *remote_addr
})
}
/// Find a UDP transport whose bound socket can send to `remote_addr`.
///
/// LAN discovery can surface both IPv4 and IPv6 addresses for the same
/// service. A wildcard IPv4 socket cannot send to an IPv6 link-local
/// target, and vice versa, so callers must choose by socket family rather
/// than by transport type alone.
fn find_udp_transport_for_remote_addr(
&self,
remote_addr: SocketAddr,
) -> Option<(TransportId, SocketAddr)> {
self.transports
.iter()
.filter(|(id, handle)| {
handle.transport_type().name == "udp"
&& handle.is_operational()
&& !self.bootstrap_transports.contains(id)
})
.filter_map(|(id, handle)| {
let local_addr = handle.local_addr()?;
socket_addr_families_compatible(local_addr, remote_addr)
.then_some((*id, local_addr))
})
.min_by_key(|(id, _)| id.as_u32())
}
/// Initiate a connection to a peer on a specific transport and address.
///
/// For connectionless transports (UDP, Ethernet): allocates a link, starts
@@ -340,6 +555,9 @@ impl Node {
pub(super) async fn poll_transport_discovery(&mut self) {
// Collect discoveries first to avoid borrow conflict with self
let mut to_connect = Vec::new();
let mut queued_per_peer: HashMap<NodeAddr, usize> = HashMap::new();
let mut connect_budget = self.discovery_connect_budget();
let mut skipped_budget = 0usize;
for (transport_id, transport) in &self.transports {
if !transport.is_operational() {
@@ -366,29 +584,80 @@ impl Node {
if node_addr == *self.identity.node_addr() {
continue;
}
// Skip if already connected
let candidate_transport_id = *transport_id;
let remote_addr = peer.addr;
if self.peers.contains_key(&node_addr) {
let transport_name = transport.transport_type().name;
let candidate = PeerAddress::new(transport_name, remote_addr.to_string());
if self.active_peer_candidate_is_fresh_enough_to_skip(
&node_addr,
std::slice::from_ref(&candidate),
) {
continue;
}
// Skip if connection already in progress
let connecting = self.connections.values().any(|c| {
c.expected_identity()
.map(|id| id.node_addr() == &node_addr)
.unwrap_or(false)
});
if connecting {
if self.is_connecting_to_peer_on_path(
&node_addr,
candidate_transport_id,
&remote_addr,
) {
continue;
}
let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0);
if connect_budget == 0
|| self
.path_candidate_attempt_budget(&node_addr)
.saturating_sub(queued_for_peer)
== 0
{
skipped_budget = skipped_budget.saturating_add(1);
continue;
}
to_connect.push((candidate_transport_id, remote_addr, identity, true));
*queued_per_peer.entry(node_addr).or_default() += 1;
connect_budget = connect_budget.saturating_sub(1);
continue;
}
to_connect.push((*transport_id, peer.addr, identity));
if self.is_connecting_to_peer_on_path(
&node_addr,
candidate_transport_id,
&remote_addr,
) {
continue;
}
let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0);
if connect_budget == 0
|| self
.path_candidate_attempt_budget(&node_addr)
.saturating_sub(queued_for_peer)
== 0
{
skipped_budget = skipped_budget.saturating_add(1);
continue;
}
to_connect.push((candidate_transport_id, remote_addr, identity, false));
*queued_per_peer.entry(node_addr).or_default() += 1;
connect_budget = connect_budget.saturating_sub(1);
}
}
for (transport_id, remote_addr, identity) in to_connect {
if skipped_budget > 0 {
debug!(
skipped = skipped_budget,
queued = to_connect.len(),
"Transport discovery connect budget exhausted"
);
}
for (transport_id, remote_addr, identity, active_refresh) in to_connect {
info!(
peer = %self.peer_display_name(identity.node_addr()),
transport_id = %transport_id,
remote_addr = %remote_addr,
active_refresh,
"Auto-connecting to discovered peer"
);
if let Err(e) = self
@@ -593,6 +862,90 @@ impl Node {
self.queue_open_discovery_retries(&bootstrap).await;
}
/// Resolve the LAN-only discovery scope. Applications with explicit
/// connectivity config can set `node.discovery.lan.scope` without
/// changing the public Nostr discovery `app` tag. The older fallback
/// extracts a scope from the Nostr app tag used by default scoped
/// discovery.
pub(super) fn lan_discovery_scope(&self) -> Option<String> {
if let Some(scope) = self.config.node.discovery.lan.scope.as_deref() {
let scope = scope.trim();
if !scope.is_empty() {
return Some(scope.to_string());
}
}
let app = self.config.node.discovery.nostr.app.trim();
if app.is_empty() {
return None;
}
if let Some(rest) = app.strip_prefix("fips-overlay-v1:") {
let scope = rest.trim();
if scope.is_empty() {
None
} else {
Some(scope.to_string())
}
} else {
Some(app.to_string())
}
}
/// Drain mDNS-discovered peers and initiate Noise XX handshakes.
/// The handshake itself is the authentication — a spoofed mDNS advert
/// with someone else's npub fails the XX exchange and is dropped.
pub(super) async fn poll_lan_discovery(&mut self) {
let Some(runtime) = self.lan_discovery.clone() else {
return;
};
let events = runtime.drain_events().await;
if events.is_empty() {
return;
}
for event in events {
let crate::discovery::lan::LanEvent::Discovered(peer) = event;
let Some((transport_id, local_addr)) =
self.find_udp_transport_for_remote_addr(peer.addr)
else {
debug!(
addr = %peer.addr,
"lan: skip discovered peer with no compatible UDP transport"
);
continue;
};
let identity = match crate::PeerIdentity::from_npub(&peer.npub) {
Ok(id) => id,
Err(err) => {
debug!(npub = %peer.npub, error = %err, "lan: skip bad npub");
continue;
}
};
let peer_node_addr = *identity.node_addr();
let remote_addr = crate::transport::TransportAddr::from_string(&peer.addr.to_string());
if self.peers.contains_key(&peer_node_addr)
|| self.is_connecting_to_peer(&peer_node_addr)
{
continue;
}
info!(
npub = %identity.short_npub(),
addr = %peer.addr,
local_addr = %local_addr,
"lan: initiating handshake to discovered peer"
);
if let Err(err) = self
.initiate_connection(transport_id, remote_addr, identity)
.await
{
debug!(
npub = %peer.npub,
error = %err,
"lan: failed to initiate connection to discovered peer"
);
}
}
}
/// Poll pending transport connects and initiate handshakes for ready ones.
///
/// Called from the tick handler. For each pending connect, queries the
@@ -793,6 +1146,47 @@ impl Node {
}
}
// mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even
// when Nostr is disabled, since it gives us sub-second pairing
// on the same link without any relay or NAT-traversal roundtrip.
if self.config.node.discovery.lan.enabled {
// Advertise the port of a non-bootstrap operational UDP transport.
// Bootstrap transports must be excluded (they are not the node's
// listening data-plane socket), and a stable selector (lowest
// TransportId) is used so the advertised port is deterministic
// across restarts rather than dependent on HashMap iteration
// order. This mirrors find_udp_transport_for_remote_addr.
let advertised_udp_port = self
.transports
.iter()
.filter(|(id, h)| {
h.transport_type().name == "udp"
&& h.is_operational()
&& !self.bootstrap_transports.contains(id)
})
.filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port())))
.min_by_key(|(id, _)| id.as_u32())
.map(|(_, port)| port)
.unwrap_or(0);
let scope = self.lan_discovery_scope();
match crate::discovery::lan::LanDiscovery::start(
&self.identity,
scope,
advertised_udp_port,
self.config.node.discovery.lan.clone(),
)
.await
{
Ok(runtime) => {
self.lan_discovery = Some(runtime);
info!("LAN mDNS discovery enabled");
}
Err(err) => {
debug!(error = %err, "LAN mDNS discovery not started");
}
}
}
// Connect to static peers before TUN is active
// This allows handshake messages to be sent before we start accepting packets
self.initiate_peer_connections().await;
@@ -1078,6 +1472,13 @@ impl Node {
warn!(error = %e, "Failed to shutdown Nostr overlay discovery");
}
// Tear down LAN mDNS responder + browser. Best-effort: the
// OS will eventually time the advert out via its TTL even if
// we don't get a clean unregister out before the daemon exits.
if let Some(lan) = self.lan_discovery.take() {
lan.shutdown().await;
}
// Shutdown transports (they're packet producers)
let transport_ids: Vec<_> = self.transports.keys().cloned().collect();
for transport_id in transport_ids {
@@ -1228,8 +1629,10 @@ impl Node {
.max()
.unwrap_or(100)
.saturating_add(1);
let seen_at_ms = Self::now_ms();
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, next_priority)
let Some(candidate) =
Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms)
else {
continue;
};
@@ -1251,17 +1654,27 @@ impl Node {
fn overlay_endpoint_to_peer_address(
endpoint: &OverlayEndpointAdvert,
priority: u8,
seen_at_ms: u64,
) -> Option<PeerAddress> {
let transport = match endpoint.transport {
OverlayTransportKind::Udp => "udp",
OverlayTransportKind::Tcp => "tcp",
OverlayTransportKind::Tor => "tor",
};
Some(PeerAddress::with_priority(
transport,
endpoint.addr.clone(),
priority,
))
Some(
PeerAddress::with_priority(transport, endpoint.addr.clone(), priority)
.with_seen_at_ms(seen_at_ms),
)
}
async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool {
let Some(bootstrap) = self.nostr_discovery.clone() else {
debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address");
return false;
};
bootstrap.request_connect(peer_config.clone()).await;
info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt");
true
}
async fn attempt_peer_address_list(
@@ -1271,18 +1684,28 @@ impl Node {
allow_bootstrap_nat: bool,
addresses: &[PeerAddress],
) -> Result<(), NodeError> {
let peer_node_addr = *peer_identity.node_addr();
let mut attempted = 0usize;
let max_attempts = self.path_candidate_attempt_budget(&peer_node_addr);
if max_attempts == 0 {
return Err(NodeError::NoTransportForType(format!(
"no outbound slots available for {}",
peer_config.npub
)));
}
for addr in addresses {
if attempted >= max_attempts {
break;
}
if addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat") {
if !allow_bootstrap_nat {
continue;
}
let Some(bootstrap) = self.nostr_discovery.clone() else {
debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address");
if self.request_nostr_bootstrap(peer_config).await {
attempted = attempted.saturating_add(1);
}
continue;
};
bootstrap.request_connect(peer_config.clone()).await;
info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt");
return Ok(());
}
let (transport_id, remote_addr) = if addr.transport == "ethernet" {
@@ -1320,7 +1743,22 @@ impl Node {
continue;
}
} else {
let tid = match self.find_transport_for_type(&addr.transport) {
let tid = if addr.transport == "udp"
&& let Ok(remote_socket_addr) = addr.addr.parse::<SocketAddr>()
{
match self.find_udp_transport_for_remote_addr(remote_socket_addr) {
Some((id, _)) => id,
None => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
"No compatible operational UDP transport for address"
);
continue;
}
}
} else {
match self.find_transport_for_type(&addr.transport) {
Some(id) => id,
None => {
debug!(
@@ -1330,15 +1768,20 @@ impl Node {
);
continue;
}
}
};
(tid, TransportAddr::from_string(&addr.addr))
};
if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) {
continue;
}
match self
.initiate_connection(transport_id, remote_addr, peer_identity)
.await
{
Ok(()) => return Ok(()),
Ok(()) => attempted = attempted.saturating_add(1),
Err(e @ NodeError::AccessDenied(_)) => return Err(e),
Err(e) => {
debug!(
@@ -1351,6 +1794,10 @@ impl Node {
}
}
if attempted > 0 {
return Ok(());
}
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
peer_config.npub
@@ -1427,6 +1874,21 @@ impl Node {
}
if configured_npubs.contains(&npub) {
if let Ok(peer_identity) = PeerIdentity::from_npub(&npub) {
let node_addr = *peer_identity.node_addr();
if !self.peers.contains_key(&node_addr)
&& !self.is_connecting_to_peer(&node_addr)
&& let Some(state) = self.retry_pending.get_mut(&node_addr)
&& state.retry_after_ms > now_ms
{
state.retry_after_ms = now_ms;
debug!(
peer = %peer_identity.short_npub(),
caller = %caller,
"open-discovery sweep: fresh configured-peer advert expedited retry"
);
}
}
skipped_configured = skipped_configured.saturating_add(1);
continue;
}
@@ -1467,8 +1929,10 @@ impl Node {
let mut addresses = Vec::new();
let mut priority = 120u8;
let seen_at_ms = Self::now_ms();
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, priority)
let Some(candidate) =
Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms)
else {
continue;
};
@@ -1607,6 +2071,61 @@ impl Node {
connection_slots.min(peer_slots)
}
fn outbound_handshake_slots(&self) -> usize {
let used = self
.connections
.len()
.saturating_add(self.pending_connects.len());
if self.max_connections == 0 {
usize::MAX
} else {
self.max_connections.saturating_sub(used)
}
}
fn outbound_link_slots(&self) -> usize {
if self.max_links == 0 {
usize::MAX
} else {
self.max_links.saturating_sub(self.links.len())
}
}
fn path_candidate_attempt_budget(&self, peer_node_addr: &NodeAddr) -> usize {
if !self.peers.contains_key(peer_node_addr)
&& self.max_peers > 0
&& self.peers.len() >= self.max_peers
{
return 0;
}
let in_flight_for_peer = self
.connections
.values()
.filter(|conn| {
conn.expected_identity()
.map(|identity| identity.node_addr() == peer_node_addr)
.unwrap_or(false)
})
.count()
.saturating_add(
self.pending_connects
.iter()
.filter(|pending| pending.peer_identity.node_addr() == peer_node_addr)
.count(),
);
self.outbound_handshake_slots()
.min(self.outbound_link_slots())
.min(MAX_PARALLEL_PATH_CANDIDATES_PER_PEER.saturating_sub(in_flight_for_peer))
}
fn discovery_connect_budget(&self) -> usize {
self.outbound_handshake_slots()
.min(self.outbound_link_slots())
.min(MAX_DISCOVERY_CONNECTS_PER_TICK)
}
fn open_discovery_enqueue_budget(&self, configured_npubs: &HashSet<String>) -> usize {
let current_open_discovery_pending = self
.retry_pending
@@ -1841,40 +2360,22 @@ impl Node {
return Ok(());
}
// Static-first dialing: avoid delaying configured address attempts on
// advert fetch/network latency.
let static_addresses = self.static_peer_addresses(peer_config);
if self
.attempt_peer_address_list(
peer_config,
peer_identity,
allow_bootstrap_nat,
&static_addresses,
)
.await
.is_ok()
{
return Ok(());
let candidates = self.peer_address_candidates(peer_config).await;
if candidates.is_empty() {
return Err(NodeError::NoTransportForType(format!(
"no addresses known for {}",
peer_config.npub
)));
}
{
let fallback = self
.nostr_peer_fallback_addresses(peer_config, &static_addresses)
.await;
if !fallback.is_empty()
&& self
.attempt_peer_address_list(
peer_config,
peer_identity,
allow_bootstrap_nat,
&fallback,
)
if self
.attempt_peer_address_list(peer_config, peer_identity, allow_bootstrap_nat, &candidates)
.await
.is_ok()
{
return Ok(());
}
}
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
@@ -1882,6 +2383,136 @@ impl Node {
)))
}
async fn try_active_peer_alternative_addresses(
&mut self,
peer_config: &PeerConfig,
peer_identity: PeerIdentity,
) -> Result<bool, NodeError> {
let peer_node_addr = *peer_identity.node_addr();
let candidates = self.peer_address_candidates(peer_config).await;
if candidates.is_empty() {
return Err(NodeError::NoTransportForType(format!(
"no addresses known for {}",
peer_config.npub
)));
}
let concrete: Vec<_> = candidates
.into_iter()
.filter(|addr| !(addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat")))
.collect();
let has_alternative = concrete
.iter()
.any(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr));
let attempt_candidates: Vec<_> = if has_alternative {
concrete
.into_iter()
.filter(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr))
.collect()
} else if self.active_peer_needs_same_path_refresh(&peer_node_addr) {
concrete
} else {
Vec::new()
};
if attempt_candidates.is_empty() {
return Ok(false);
}
self.attempt_peer_address_list(peer_config, peer_identity, false, &attempt_candidates)
.await?;
Ok(true)
}
async fn peer_address_candidates(&self, peer_config: &PeerConfig) -> Vec<PeerAddress> {
let static_addresses = self.static_peer_addresses(peer_config);
let overlay_addresses = self
.nostr_peer_fallback_addresses(peer_config, &static_addresses)
.await;
let mut candidates = Vec::with_capacity(overlay_addresses.len() + static_addresses.len());
for addr in overlay_addresses.into_iter().chain(static_addresses) {
if !candidates.iter().any(|existing: &PeerAddress| {
existing.transport == addr.transport && existing.addr == addr.addr
}) {
candidates.push(addr);
}
}
candidates.sort_by(|a, b| match (a.seen_at_ms, b.seen_at_ms) {
(Some(a_ts), Some(b_ts)) => b_ts.cmp(&a_ts),
(Some(_), None) => std::cmp::Ordering::Less,
(None, Some(_)) => std::cmp::Ordering::Greater,
(None, None) => std::cmp::Ordering::Equal,
});
candidates
}
pub(in crate::node) fn active_peer_candidate_is_fresh_enough_to_skip(
&self,
peer_node_addr: &NodeAddr,
candidates: &[PeerAddress],
) -> bool {
if !self.active_peer_matches_any_candidate(peer_node_addr, candidates) {
return false;
}
!self.active_peer_needs_same_path_refresh(peer_node_addr)
}
fn active_peer_needs_same_path_refresh(&self, peer_node_addr: &NodeAddr) -> bool {
let Some(peer) = self.peers.get(peer_node_addr) else {
return false;
};
let stale_after_ms = self
.config
.node
.heartbeat_interval_secs
.saturating_mul(1000)
.max(1000);
peer.idle_time(Self::now_ms()) > stale_after_ms
}
fn active_peer_matches_any_candidate(
&self,
peer_node_addr: &NodeAddr,
candidates: &[PeerAddress],
) -> bool {
candidates
.iter()
.any(|candidate| self.active_peer_matches_candidate(peer_node_addr, candidate))
}
fn active_peer_matches_candidate(
&self,
peer_node_addr: &NodeAddr,
candidate: &PeerAddress,
) -> bool {
let Some(peer) = self.peers.get(peer_node_addr) else {
return false;
};
let Some(current_addr) = peer.current_addr() else {
return false;
};
if peer
.transport_id()
.map(|id| self.bootstrap_transports.contains(&id))
.unwrap_or(false)
{
return false;
}
let current_addr = current_addr.to_string();
let current_transport = peer
.transport_id()
.and_then(|id| self.transports.get(&id))
.map(|transport| transport.transport_type().name);
candidate.addr == current_addr
&& current_transport
.map(|transport| transport == candidate.transport)
.unwrap_or(true)
}
// === Control API methods ===
/// Connect to a peer via the control API.
src/node/mod.rs
+20
View File
@@ -202,6 +202,19 @@ impl fmt::Display for NodeState {
}
}
/// Reports what changed when replacing the runtime peer list.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UpdatePeersOutcome {
/// Peers present in the new list but not the previous list.
pub added: usize,
/// Peers removed from the previous list.
pub removed: usize,
/// Existing peers whose configured behavior changed.
pub updated: usize,
/// Existing peers whose comparable config did not change.
pub unchanged: usize,
}
/// Recent request tracking for dedup and reverse-path forwarding.
///
/// When a LookupRequest is forwarded through a node, the node stores the
@@ -453,6 +466,11 @@ pub struct Node {
/// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers.
nostr_discovery: Option<Arc<crate::discovery::nostr::NostrDiscovery>>,
/// mDNS / DNS-SD responder + browser for local-link peer discovery.
/// Identity is unverified at this layer — the Noise XX handshake
/// initiated against an mDNS-observed endpoint is what proves the
/// peer holds the matching private key.
lan_discovery: Option<Arc<crate::discovery::lan::LanDiscovery>>,
/// Wall-clock ms when Nostr discovery successfully started, used to
/// schedule the one-shot startup advert sweep after a settle delay.
/// `None` until discovery comes up; remains `None` if discovery is
@@ -669,6 +687,7 @@ impl Node {
retry_pending: HashMap::new(),
nostr_discovery: None,
nostr_discovery_started_at_ms: None,
lan_discovery: None,
startup_open_discovery_sweep_done: false,
bootstrap_transports: HashSet::new(),
bootstrap_transport_npubs: HashMap::new(),
@@ -812,6 +831,7 @@ impl Node {
retry_pending: HashMap::new(),
nostr_discovery: None,
nostr_discovery_started_at_ms: None,
lan_discovery: None,
startup_open_discovery_sweep_done: false,
bootstrap_transports: HashSet::new(),
bootstrap_transport_npubs: HashMap::new(),
src/node/retry.rs
+11 -1
View File
@@ -11,6 +11,7 @@ use crate::identity::NodeAddr;
use tracing::{debug, info, warn};
// MAX_BACKOFF_MS is now derived from config: node.retry.max_backoff_secs * 1000
const MAX_RETRY_CONNECTIONS_PER_TICK: usize = 16;
/// Tracks retry state for a peer across connection attempts.
pub struct RetryState {
@@ -248,8 +249,17 @@ impl Node {
.filter(|(_, state)| now_ms >= state.retry_after_ms)
.map(|(addr, _)| *addr)
.collect();
let deferred = due.len().saturating_sub(MAX_RETRY_CONNECTIONS_PER_TICK);
if deferred > 0 {
debug!(
due = due.len(),
processing = MAX_RETRY_CONNECTIONS_PER_TICK,
deferred,
"Retry processing budget exhausted; deferring remaining peers"
);
}
for node_addr in due {
for node_addr in due.into_iter().take(MAX_RETRY_CONNECTIONS_PER_TICK) {
// Peer may have connected inbound while we waited
if self.peers.contains_key(&node_addr) {
self.retry_pending.remove(&node_addr);
src/node/tests/unit.rs
+218
View File
@@ -97,6 +97,58 @@ async fn test_nat_bootstrap_failure_falls_back_to_direct_udp_address() {
}
}
#[tokio::test]
async fn test_try_peer_addresses_races_all_concrete_udp_candidates() {
let peer_identity = Identity::generate();
let mut node = make_node();
let (packet_tx, packet_rx) = packet_channel(64);
node.packet_tx = Some(packet_tx.clone());
node.packet_rx = Some(packet_rx);
let transport_id = TransportId::new(1);
let mut udp = UdpTransport::new(
transport_id,
Some("main".to_string()),
crate::config::UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
..Default::default()
},
packet_tx,
);
udp.start_async().await.unwrap();
node.transports
.insert(transport_id, TransportHandle::Udp(udp));
let peer_config = crate::config::PeerConfig {
npub: peer_identity.npub(),
alias: None,
addresses: vec![
crate::config::PeerAddress::with_priority("udp", "127.0.0.1:9", 1),
crate::config::PeerAddress::with_priority("udp", "127.0.0.1:10", 2),
],
connect_policy: crate::config::ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
};
let peer_identity = PeerIdentity::from_npub(&peer_config.npub).unwrap();
node.try_peer_addresses(&peer_config, peer_identity, false)
.await
.unwrap();
let mut addrs = node
.connections
.values()
.filter_map(|conn| conn.source_addr().and_then(|addr| addr.as_str()))
.collect::<Vec<_>>();
addrs.sort();
assert_eq!(addrs, vec!["127.0.0.1:10", "127.0.0.1:9"]);
for transport in node.transports.values_mut() {
transport.stop().await.ok();
}
}
#[tokio::test]
async fn test_node_state_transitions() {
let mut node = make_node();
@@ -712,6 +764,48 @@ fn test_schedule_retry_increments() {
assert_eq!(state.retry_after_ms, 11_000 + 20_000);
}
/// Retry processing is paced so a large due set cannot start every
/// handshake candidate in one maintenance tick.
#[tokio::test]
async fn test_process_pending_retries_is_budgeted_per_tick() {
let mut node = make_node();
let mut addrs = Vec::new();
for _ in 0..20 {
let identity = Identity::generate();
let npub = identity.npub();
let peer_identity = PeerIdentity::from_npub(&npub).unwrap();
let node_addr = *peer_identity.node_addr();
node.retry_pending.insert(
node_addr,
crate::node::retry::RetryState {
peer_config: crate::config::PeerConfig::new(npub, "udp", "10.0.0.2:2121"),
retry_count: 0,
retry_after_ms: 0,
reconnect: true,
expires_at_ms: None,
},
);
addrs.push(node_addr);
}
node.process_pending_retries(1).await;
let processed = addrs
.iter()
.filter(|addr| {
node.retry_pending
.get(addr)
.is_some_and(|state| state.retry_count > 0)
})
.count();
let deferred = addrs.len().saturating_sub(processed);
assert_eq!(processed, 16);
assert_eq!(deferred, 4);
assert_eq!(node.retry_pending.len(), 20);
}
/// Test that auto-connect peers retry indefinitely (never exhaust).
#[test]
fn test_schedule_retry_auto_connect_never_exhausts() {
@@ -864,6 +958,130 @@ async fn test_try_peer_addresses_skips_connecting_peer() {
);
}
#[test]
fn active_peer_same_path_discovery_skips_fresh_peer() {
let mut node = make_node();
let peer_full = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full());
let peer_node_addr = *peer_identity.node_addr();
let transport_id = TransportId::new(1);
let current_addr = TransportAddr::from_string("127.0.0.1:9");
let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), Node::now_ms());
active_peer.set_current_addr(transport_id, current_addr.clone());
node.peers.insert(peer_node_addr, active_peer);
let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9");
assert!(node.active_peer_candidate_is_fresh_enough_to_skip(
&peer_node_addr,
std::slice::from_ref(&candidate),
));
}
#[test]
fn active_peer_same_path_discovery_refreshes_stale_peer() {
let mut node = make_node();
let peer_full = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full());
let peer_node_addr = *peer_identity.node_addr();
let transport_id = TransportId::new(1);
let current_addr = TransportAddr::from_string("127.0.0.1:9");
let stale_at = Node::now_ms().saturating_sub(
node.config
.node
.heartbeat_interval_secs
.saturating_add(1)
.saturating_mul(1000),
);
let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), stale_at);
active_peer.set_current_addr(transport_id, current_addr.clone());
node.peers.insert(peer_node_addr, active_peer);
let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9");
assert!(!node.active_peer_candidate_is_fresh_enough_to_skip(
&peer_node_addr,
std::slice::from_ref(&candidate),
));
}
#[tokio::test]
async fn update_peers_races_new_alternative_without_dropping_active_peer() {
let mut node = make_node();
let (packet_tx, packet_rx) = packet_channel(64);
node.packet_tx = Some(packet_tx.clone());
node.packet_rx = Some(packet_rx);
let transport_id = TransportId::new(1);
let mut udp = UdpTransport::new(
transport_id,
Some("main".to_string()),
crate::config::UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
..Default::default()
},
packet_tx,
);
udp.start_async().await.unwrap();
node.transports
.insert(transport_id, TransportHandle::Udp(udp));
let peer_full = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full());
let peer_node_addr = *peer_identity.node_addr();
let current_addr = TransportAddr::from_string("127.0.0.1:9");
let new_addr = TransportAddr::from_string("127.0.0.1:10");
let old_link_id = LinkId::new(7);
let mut active_peer = ActivePeer::new(peer_identity, old_link_id, Node::now_ms());
active_peer.set_current_addr(transport_id, current_addr.clone());
node.peers.insert(peer_node_addr, active_peer);
node.links.insert(
old_link_id,
Link::connectionless(
old_link_id,
transport_id,
current_addr.clone(),
LinkDirection::Outbound,
Duration::from_millis(100),
),
);
let old_peer = crate::config::PeerConfig {
npub: peer_full.npub(),
alias: None,
addresses: vec![crate::config::PeerAddress::new("udp", "127.0.0.1:9")],
connect_policy: crate::config::ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
};
let new_peer = crate::config::PeerConfig {
addresses: vec![
crate::config::PeerAddress::new("udp", "127.0.0.1:9"),
crate::config::PeerAddress::new("udp", "127.0.0.1:10"),
],
..old_peer.clone()
};
node.config.peers = vec![old_peer];
let outcome = node.update_peers(vec![new_peer]).await.unwrap();
assert_eq!(outcome.updated, 1);
assert_eq!(node.peer_count(), 1, "existing link must stay live");
assert_eq!(node.connection_count(), 1);
assert_eq!(
node.connections
.values()
.next()
.and_then(|conn| conn.source_addr()),
Some(&new_addr)
);
let active = node.get_peer(&peer_node_addr).unwrap();
assert_eq!(active.link_id(), old_link_id);
assert_eq!(active.current_addr(), Some(&current_addr));
for transport in node.transports.values_mut() {
transport.stop().await.ok();
}
}
#[tokio::test]
async fn test_nostr_traversal_failure_skips_connected_peer() {
let mut node = make_node();
src/peer/active.rs
+13 -2
View File
@@ -593,6 +593,13 @@ impl ActivePeer {
self.remote_epoch
}
/// Update the remote peer's startup epoch after a successful in-place
/// rekey. Initial handshakes set this through `with_session`, but recovery
/// rekeys also exchange epochs and must keep restart detection current.
pub(crate) fn set_remote_epoch(&mut self, remote_epoch: Option<[u8; 8]>) {
self.remote_epoch = remote_epoch;
}
// === Tree Accessors ===
/// Get the peer's tree coordinates, if known.
@@ -1099,7 +1106,10 @@ impl ActivePeer {
/// Takes the stored handshake state, reads msg2, and returns the
/// completed NoiseSession. Clears the handshake-related fields but
/// leaves rekey_our_index for set_pending_session to use.
pub fn complete_rekey_msg2(&mut self, msg2_bytes: &[u8]) -> Result<NoiseSession, NoiseError> {
pub fn complete_rekey_msg2(
&mut self,
msg2_bytes: &[u8],
) -> Result<(NoiseSession, Option<[u8; 8]>), NoiseError> {
let mut hs = self
.rekey_handshake
.take()
@@ -1109,13 +1119,14 @@ impl ActivePeer {
})?;
hs.read_message_2(msg2_bytes)?;
let remote_epoch = hs.remote_epoch();
let session = hs.into_session()?;
// Clear msg1 resend state
self.rekey_msg1 = None;
self.rekey_msg1_next_resend = 0;
Ok(session)
Ok((session, remote_epoch))
}
/// Check if msg1 needs resending.