diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..e734e7c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,6 @@ +# Changelog + +## v0.4.1 — 2026-09-02 + +- `federation-net 0.3.1`: recover timed-out or failed Mainline-DHT rendezvous clients and expose network health to applications. +- `music-dht 0.4.1`: retain temporarily unreachable contacts and serialize duplicate dial attempts with bounded backoff. diff --git a/Cargo.lock b/Cargo.lock index 8b4f7ce..38a2286 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -837,7 +837,7 @@ checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" [[package]] name = "federation-net" -version = "0.3.0" +version = "0.3.1" dependencies = [ "anyhow", "blake3", @@ -2073,7 +2073,7 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.4.0" +version = "0.4.1" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 5c2f024..54c25ed 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,8 +13,8 @@ rust-version = "1.97" repository = "https://gt.hexor.cy/ab/frid" [workspace.dependencies] -federation-net = { path = "crates/federation-net", version = "0.3.0" } -music-dht = { path = "crates/music-dht", version = "0.4.0" } +federation-net = { path = "crates/federation-net", version = "0.3.1" } +music-dht = { path = "crates/music-dht", version = "0.4.1" } iroh = "1" iroh-base = "1" iroh-tickets = "1" diff --git a/crates/federation-net/Cargo.toml b/crates/federation-net/Cargo.toml index b1b9990..4c8e160 100644 --- a/crates/federation-net/Cargo.toml +++ b/crates/federation-net/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "federation-net" -version = "0.3.0" +version = "0.3.1" description = "Generic peer-to-peer networking engine built on Iroh" readme = "README.md" documentation = "https://docs.rs/federation-net" diff --git a/crates/federation-net/src/engine.rs b/crates/federation-net/src/engine.rs index 6a191bf..1adfd9b 100644 --- a/crates/federation-net/src/engine.rs +++ b/crates/federation-net/src/engine.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use std::fmt; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak}; -use std::time::Duration; +use std::time::{Duration, Instant}; use iroh::endpoint::{Connection, RecvStream, SendStream, VarInt, presets}; use iroh::protocol::{AcceptError, ProtocolHandler, Router}; @@ -19,6 +19,7 @@ use tracing::{debug, info, warn}; use crate::config::NetworkConfig; use crate::error::{NetworkError, Result}; use crate::event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver}; +use crate::health::{NetworkHealthSnapshot, NetworkHealthState}; use crate::identity; use crate::protocol::{ ALPN, Handshake, HandshakeAck, HandshakeErrorCode, MAX_HANDSHAKE_FRAME_SIZE, @@ -50,6 +51,12 @@ const REJECT_LINGER: Duration = Duration::from_secs(3); const SHUTDOWN_TASK_GRACE: Duration = Duration::from_secs(5); /// Upper bound of the rendezvous round delay while no peer is connected yet. const RENDEZVOUS_LONELY_INTERVAL: Duration = Duration::from_secs(15); +/// A stuck Mainline-DHT operation must not stall discovery forever. +const RENDEZVOUS_ROUND_TIMEOUT: Duration = Duration::from_secs(30); +/// Recreate the Mainline-DHT client after this many consecutive failures. +const RENDEZVOUS_REBUILD_AFTER_FAILURES: u32 = 2; +/// Recommend an application-owned full restart after sustained discovery failure. +const RENDEZVOUS_RESTART_AFTER_FAILURES: u32 = 4; /// Capacity of the queue of accepted-but-not-yet-consumed incoming byte /// streams, per stream protocol. A full queue delays the handshake ack of /// further incoming streams (natural backpressure). @@ -260,6 +267,27 @@ struct StreamAcceptorSlot { receiver: Option>, } +#[derive(Debug)] +struct RendezvousHealth { + last_success: Option, + consecutive_failures: u32, + consecutive_peer_dial_failures: u32, + client_restarts: u64, + last_error: Option, +} + +impl RendezvousHealth { + fn new() -> Self { + Self { + last_success: None, + consecutive_failures: 0, + consecutive_peer_dial_failures: 0, + client_restarts: 0, + last_error: None, + } + } +} + fn alpn_display(alpn: &[u8]) -> String { String::from_utf8_lossy(alpn).into_owned() } @@ -276,6 +304,7 @@ struct Shared { tasks: Mutex>, next_generation: AtomicU64, shutting_down: AtomicBool, + rendezvous_health: Mutex>, /// Broadcasts the start of the shutdown to long-running background /// loops so they can stop promptly instead of being aborted. shutdown_signal: watch::Sender, @@ -294,6 +323,97 @@ impl Shared { } } + fn record_rendezvous_success(&self) { + if let Some(health) = lock(&self.rendezvous_health).as_mut() { + health.last_success = Some(Instant::now()); + health.consecutive_failures = 0; + health.last_error = None; + } + } + + fn record_rendezvous_failure(&self, error: String) -> u32 { + let mut guard = lock(&self.rendezvous_health); + let Some(health) = guard.as_mut() else { + return 0; + }; + health.consecutive_failures = health.consecutive_failures.saturating_add(1); + health.last_error = Some(error); + health.consecutive_failures + } + + fn record_rendezvous_restart(&self) { + if let Some(health) = lock(&self.rendezvous_health).as_mut() { + health.client_restarts = health.client_restarts.saturating_add(1); + } + } + + fn record_peer_connection_success(&self) { + if let Some(health) = lock(&self.rendezvous_health).as_mut() { + health.consecutive_peer_dial_failures = 0; + } + } + + fn record_rendezvous_peer_dials(&self, candidates: usize, attempts: usize, successes: usize) { + let mut guard = lock(&self.rendezvous_health); + let Some(health) = guard.as_mut() else { + return; + }; + if candidates == 0 || successes > 0 { + health.consecutive_peer_dial_failures = 0; + } else if attempts > 0 { + health.consecutive_peer_dial_failures = + health.consecutive_peer_dial_failures.saturating_add(1); + health.last_error = Some(format!( + "failed to connect to all {attempts} peer(s) discovered by rendezvous" + )); + } + } + + fn health(&self) -> NetworkHealthSnapshot { + let connected_peers = lock(&self.peers).len(); + let guard = lock(&self.rendezvous_health); + let Some(health) = guard.as_ref() else { + return NetworkHealthSnapshot { + state: if connected_peers == 0 { + NetworkHealthState::Discovering + } else { + NetworkHealthState::Healthy + }, + connected_peers, + rendezvous_enabled: false, + consecutive_rendezvous_failures: 0, + consecutive_peer_dial_failures: 0, + rendezvous_restarts: 0, + last_rendezvous_success_ago: None, + last_rendezvous_error: None, + restart_recommended: false, + }; + }; + let restart_recommended = connected_peers == 0 + && (health.consecutive_failures >= RENDEZVOUS_RESTART_AFTER_FAILURES + || health.consecutive_peer_dial_failures >= RENDEZVOUS_RESTART_AFTER_FAILURES); + let state = if restart_recommended { + NetworkHealthState::Degraded + } else if health.consecutive_failures > 0 || health.consecutive_peer_dial_failures > 0 { + NetworkHealthState::Recovering + } else if health.last_success.is_some() { + NetworkHealthState::Healthy + } else { + NetworkHealthState::Discovering + }; + NetworkHealthSnapshot { + state, + connected_peers, + rendezvous_enabled: true, + consecutive_rendezvous_failures: health.consecutive_failures, + consecutive_peer_dial_failures: health.consecutive_peer_dial_failures, + rendezvous_restarts: health.client_restarts, + last_rendezvous_success_ago: health.last_success.map(|at| at.elapsed()), + last_rendezvous_error: health.last_error.clone(), + restart_recommended, + } + } + /// Delivers an event to the application. /// /// The channel is bounded; if it is full this awaits until the @@ -335,6 +455,7 @@ impl Shared { generation, }; let replaced = lock(&self.peers).insert(peer_id, state); + self.record_peer_connection_success(); if let Some(old) = replaced { debug!(peer = %peer_id, "replacing existing connection"); old.connection @@ -786,39 +907,92 @@ impl Shared { /// Failures of a single round or dial are logged and retried on the next /// round; the loop only ends when the engine shuts down (the task is /// aborted). -async fn rendezvous_loop( - shared: Arc>, - client: RendezvousClient, - config: RendezvousConfig, -) { +async fn rendezvous_loop(shared: Arc>, config: RendezvousConfig) { let mut shutdown = shared.shutdown_signal.subscribe(); // Give the endpoint a moment to learn its relay and direct addresses so // the very first published record is already dialable. let _ = timeout(shared.config.request_timeout, shared.endpoint.online()).await; let self_id = shared.endpoint.id(); + let mut client = None; loop { if shared.is_shutting_down() { return; } + if client.is_none() { + match RendezvousClient::new(shared.config.network_id, &config) { + Ok(new_client) => client = Some(new_client), + Err(err) => { + let failures = shared.record_rendezvous_failure(err.to_string()); + warn!(error = %err, failures, "failed to start peer rendezvous; retrying"); + } + } + } let round = async { + let Some(active_client) = client.as_ref() else { + return; + }; let addr = shared.endpoint.addr(); let self_addr = (!addr.is_empty()).then_some(addr); - match client.round(self_addr, now_ms()).await { - Ok(peers) => { - for peer_addr in peers { + if self_addr.is_none() { + let failures = shared.record_rendezvous_failure( + "local endpoint has no dialable address".to_string(), + ); + debug!(failures, "rendezvous waiting for a dialable local address"); + return; + } + match timeout( + RENDEZVOUS_ROUND_TIMEOUT, + active_client.round(self_addr, now_ms()), + ) + .await + { + Ok(Ok(round)) => { + let candidates = round.peers.len(); + let mut attempts = 0; + let mut successes = 0; + for peer_addr in round.peers { let peer = peer_addr.id; if peer == self_id || lock(&shared.peers).contains_key(&peer) { continue; } debug!(peer = %peer, "rendezvous discovered a peer; connecting"); - if let Err(err) = shared.connect_to_addr(peer_addr).await { - // Stale entries (peers that left) fail here; they - // age out of the record by TTL. - debug!(peer = %peer, error = %err, "rendezvous connect attempt failed"); + attempts += 1; + match shared.connect_to_addr(peer_addr).await { + Ok(_) => successes += 1, + Err(err) => { + // Stale entries (peers that left) fail here; + // they age out of the record by TTL. + debug!(peer = %peer, error = %err, "rendezvous connect attempt failed"); + } } } + if let Some(error) = round.publish_error { + let failures = shared.record_rendezvous_failure(error.clone()); + debug!(%error, failures, "rendezvous publish failed"); + } else { + shared.record_rendezvous_success(); + } + shared.record_rendezvous_peer_dials(candidates, attempts, successes); } - Err(err) => debug!(error = %err, "rendezvous round failed"), + Ok(Err(err)) => { + let failures = shared.record_rendezvous_failure(err.to_string()); + debug!(error = %err, failures, "rendezvous round failed"); + } + Err(_) => { + let failures = shared.record_rendezvous_failure(format!( + "rendezvous round timed out after {}s", + RENDEZVOUS_ROUND_TIMEOUT.as_secs() + )); + warn!(failures, "rendezvous round timed out"); + } + } + let failures = lock(&shared.rendezvous_health) + .as_ref() + .map_or(0, |health| health.consecutive_failures); + if failures >= RENDEZVOUS_REBUILD_AFTER_FAILURES { + client = None; + shared.record_rendezvous_restart(); + info!(failures, "rebuilding the rendezvous client"); } }; tokio::select! { @@ -960,6 +1134,7 @@ impl NetworkEngine { config: NetworkConfig, secret_key: SecretKey, ) -> Result<(Self, NetworkEventReceiver)> { + let rendezvous_enabled = config.rendezvous.is_some(); let endpoint = Endpoint::builder(presets::N0) .secret_key(secret_key) .bind() @@ -992,6 +1167,7 @@ impl NetworkEngine { tasks: Mutex::new(JoinSet::new()), next_generation: AtomicU64::new(0), shutting_down: AtomicBool::new(false), + rendezvous_health: Mutex::new(rendezvous_enabled.then(RendezvousHealth::new)), shutdown_signal: watch::Sender::new(false), }); let handler = FederationProtocol { @@ -1015,17 +1191,10 @@ impl NetworkEngine { let router = router_builder.spawn(); *lock(&shared.router) = Some(router); if let Some(rendezvous) = shared.config.rendezvous.clone() { - match RendezvousClient::new(shared.config.network_id, &rendezvous) { - Ok(client) => { - let loop_shared = shared.clone(); - shared.spawn_task(async move { - rendezvous_loop(loop_shared, client, rendezvous).await; - }); - } - // Rendezvous is a convenience; the engine stays usable via - // tickets even when the DHT client cannot start. - Err(err) => warn!(error = %err, "peer rendezvous disabled"), - } + let loop_shared = shared.clone(); + shared.spawn_task(async move { + rendezvous_loop(loop_shared, rendezvous).await; + }); } info!( endpoint_id = %endpoint_id, @@ -1257,6 +1426,11 @@ impl NetworkEngine { lock(&self.shared.peers).keys().copied().collect() } + /// Returns a point-in-time snapshot of transport and rendezvous health. + pub fn health(&self) -> NetworkHealthSnapshot { + self.shared.health() + } + /// Returns `true` if there is an active connection to `peer`. pub fn is_connected(&self, peer: EndpointId) -> bool { lock(&self.shared.peers).contains_key(&peer) diff --git a/crates/federation-net/src/health.rs b/crates/federation-net/src/health.rs new file mode 100644 index 0000000..57ebd45 --- /dev/null +++ b/crates/federation-net/src/health.rs @@ -0,0 +1,62 @@ +//! Point-in-time health information for the transport and rendezvous loop. + +use std::fmt; +use std::time::Duration; + +/// Coarse health state of a running network engine. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NetworkHealthState { + /// The engine is starting or has no discovery mechanism configured. + Discovering, + /// Transport and rendezvous maintenance are operating normally. + Healthy, + /// A recent rendezvous operation failed and is being retried. + Recovering, + /// Repeated rendezvous failures indicate that the engine should be restarted. + Degraded, +} + +impl NetworkHealthState { + /// Returns a compact stable label for logs and user interfaces. + pub fn as_str(self) -> &'static str { + match self { + Self::Discovering => "discovering", + Self::Healthy => "healthy", + Self::Recovering => "recovering", + Self::Degraded => "degraded", + } + } +} + +impl fmt::Display for NetworkHealthState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Point-in-time health snapshot of a running network engine. +/// +/// Zero connected peers is not itself an error: a network may legitimately +/// contain only one online node. [`Self::restart_recommended`] is therefore +/// based on repeated discovery failures as well as the absence of peers. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NetworkHealthSnapshot { + /// Coarse state suitable for logs and user interfaces. + pub state: NetworkHealthState, + /// Number of currently connected peers. + pub connected_peers: usize, + /// Whether Mainline-DHT rendezvous discovery is enabled. + pub rendezvous_enabled: bool, + /// Consecutive failed or timed-out rendezvous rounds. + pub consecutive_rendezvous_failures: u32, + /// Consecutive rendezvous rounds in which every discovered peer dial failed. + pub consecutive_peer_dial_failures: u32, + /// Number of times the rendezvous client was rebuilt after failures. + pub rendezvous_restarts: u64, + /// Time elapsed since the most recent successful rendezvous round. + pub last_rendezvous_success_ago: Option, + /// Most recent rendezvous error, if a later successful round has not cleared it. + pub last_rendezvous_error: Option, + /// Whether an application-owned full service restart is advisable. + pub restart_recommended: bool, +} diff --git a/crates/federation-net/src/lib.rs b/crates/federation-net/src/lib.rs index 75ca714..38425e9 100644 --- a/crates/federation-net/src/lib.rs +++ b/crates/federation-net/src/lib.rs @@ -55,6 +55,7 @@ mod config; mod engine; mod error; mod event; +mod health; mod identity; mod protocol; mod rendezvous; @@ -71,6 +72,7 @@ pub use engine::{ }; pub use error::{NetworkError, Result}; pub use event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver}; +pub use health::{NetworkHealthSnapshot, NetworkHealthState}; pub use protocol::{ALPN, NetworkId, PROTOCOL_VERSION, SchemaId}; pub use rendezvous::RENDEZVOUS_RECORD_VERSION; pub use rendezvous::{DEFAULT_RENDEZVOUS_ENTRY_TTL, DEFAULT_RENDEZVOUS_INTERVAL, RendezvousConfig}; diff --git a/crates/federation-net/src/rendezvous.rs b/crates/federation-net/src/rendezvous.rs index 55dfa0a..2f7882a 100644 --- a/crates/federation-net/src/rendezvous.rs +++ b/crates/federation-net/src/rendezvous.rs @@ -27,7 +27,7 @@ use iroh::{EndpointAddr, EndpointId}; use mainline::async_dht::AsyncDht; use mainline::{Dht, MutableItem, SigningKey}; use serde::{Deserialize, Serialize}; -use tracing::{debug, warn}; +use tracing::debug; use crate::error::{NetworkError, Result}; use crate::protocol::NetworkId; @@ -171,6 +171,12 @@ pub(crate) struct RendezvousClient { entry_ttl: Duration, } +/// Result of one read/merge/publish cycle. +pub(crate) struct RendezvousRound { + pub(crate) peers: Vec, + pub(crate) publish_error: Option, +} + impl RendezvousClient { /// Binds a mainline DHT client for the rendezvous record of `network_id`. pub(crate) fn new(network_id: NetworkId, config: &RendezvousConfig) -> Result { @@ -201,7 +207,7 @@ impl RendezvousClient { &self, self_addr: Option, now_ms: u64, - ) -> Result> { + ) -> Result { let public_key = self.key.verifying_key().to_bytes(); let mut items = self.dht.get_mutable(&public_key, None, None); let mut seen = Vec::new(); @@ -222,21 +228,28 @@ impl RendezvousClient { let publish = self_entry.is_some(); let merged = merge_entries(seen, self_entry, now_ms, self.entry_ttl); - if publish { + let publish_error = if publish { let encoded = encode_record_capped(merged.clone())?; // Strictly newer than every instance seen this round; concurrent // writers race, but merging on read makes lost updates benign. let item = MutableItem::new(self.key.clone(), &encoded, max_seq + 1, None); - if let Err(err) = self.dht.put_mutable(item, None).await { - warn!(error = %err, "failed to publish the rendezvous record"); - } - } + self.dht + .put_mutable(item, None) + .await + .err() + .map(|err| format!("failed to publish the rendezvous record: {err}")) + } else { + None + }; - Ok(merged - .into_iter() - .map(|entry| entry.addr) - .filter(|addr| Some(addr.id) != self_id) - .collect()) + Ok(RendezvousRound { + peers: merged + .into_iter() + .map(|entry| entry.addr) + .filter(|addr| Some(addr.id) != self_id) + .collect(), + publish_error, + }) } } diff --git a/crates/music-dht/Cargo.toml b/crates/music-dht/Cargo.toml index 2a084e0..20d69b0 100644 --- a/crates/music-dht/Cargo.toml +++ b/crates/music-dht/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "music-dht" -version = "0.4.0" +version = "0.4.1" description = "Distributed music library search: a Kademlia-style DHT on top of federation-net" readme = "README.md" documentation = "https://docs.rs/music-dht" diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs index 77dea57..d203976 100644 --- a/crates/music-dht/src/lib.rs +++ b/crates/music-dht/src/lib.rs @@ -133,6 +133,7 @@ pub use service::{ // Re-exported types from the transport layer that appear in this API. pub use federation_net::{ - ByteStream, ByteStreamConnectionStats, ConnectionPathKind, EndpointAddr, EndpointId, NetworkId, - PeerTicket, RecvStream, RendezvousConfig, SecretKey, SendStream, StreamAcceptor, + ByteStream, ByteStreamConnectionStats, ConnectionPathKind, EndpointAddr, EndpointId, + NetworkHealthSnapshot, NetworkHealthState, NetworkId, PeerTicket, RecvStream, RendezvousConfig, + SecretKey, SendStream, StreamAcceptor, }; diff --git a/crates/music-dht/src/node.rs b/crates/music-dht/src/node.rs index a597f3a..f3b3aa1 100644 --- a/crates/music-dht/src/node.rs +++ b/crates/music-dht/src/node.rs @@ -36,8 +36,8 @@ fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { const DIAL_BACKOFF_BASE: Duration = Duration::from_secs(30); /// Upper bound of the dial backoff. const DIAL_BACKOFF_MAX: Duration = Duration::from_secs(10 * 60); -/// Consecutive failed dials after which a contact is evicted entirely. -const DIAL_FAILURES_BEFORE_EVICT: u32 = 5; +/// Consecutive failed dials after which a contact is considered cold. +const DIAL_FAILURES_BEFORE_COLD: u32 = 5; /// Dial-failure state of one currently unreachable contact. #[derive(Debug, Clone, Copy)] @@ -97,6 +97,8 @@ pub(crate) struct Node { exchange_sent: Mutex>, /// Contacts that recently failed to dial, with their backoff state. dial_failures: Mutex>, + /// Serializes on-demand connection attempts to the same contact. + dial_locks: Mutex>>>, events: Mutex>>, /// Set once the post-startup republish has been triggered. initial_republish_done: AtomicBool, @@ -123,6 +125,7 @@ impl Node { hello_sent: Mutex::new(HashSet::new()), exchange_sent: Mutex::new(HashSet::new()), dial_failures: Mutex::new(HashMap::new()), + dial_locks: Mutex::new(HashMap::new()), events: Mutex::new(Some(events)), initial_republish_done: AtomicBool::new(false), shutting_down: AtomicBool::new(false), @@ -478,6 +481,18 @@ impl Node { if self.engine.is_connected(contact.peer_id) { return Ok(contact.peer_id); } + let dial_lock = lock(&self.dial_locks) + .entry(contact.peer_id) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) + .clone(); + let _dial_guard = dial_lock.lock().await; + self.ensure_running()?; + if self.engine.is_connected(contact.peer_id) { + return Ok(contact.peer_id); + } + if self.dial_backoff_active(&contact.peer_id, now_ms()) { + return Err(MusicDhtError::Timeout); + } let ticket: PeerTicket = contact .ticket .parse() @@ -493,7 +508,7 @@ impl Node { Ok(peer) } Err(err) => { - self.note_dial_failure(contact).await; + self.note_dial_failure(contact); Err(err) } } @@ -519,35 +534,36 @@ impl Node { } } - /// Records a failed dial; after [`DIAL_FAILURES_BEFORE_EVICT`] failures - /// in a row the contact is dropped from the routing table and the - /// database (gossip re-adds it with a clean slate if it comes back). - async fn note_dial_failure(&self, contact: &NodeContact) { + /// Records a failed dial and increases its bounded retry backoff. + /// + /// The contact remains in memory and durable storage: temporary network + /// loss or system sleep must not erase the only route back to a peer. + fn note_dial_failure(&self, contact: &NodeContact) { let consecutive = { let mut failures = lock(&self.dial_failures); let failure = failures.entry(contact.peer_id).or_insert(DialFailure { consecutive: 0, last_attempt_ms: 0, }); - failure.consecutive += 1; + failure.consecutive = failure.consecutive.saturating_add(1); failure.last_attempt_ms = now_ms(); failure.consecutive }; - if consecutive < DIAL_FAILURES_BEFORE_EVICT { + if consecutive == DIAL_FAILURES_BEFORE_COLD { + info!( + peer = %contact.peer_id, + consecutive, + backoff_s = dial_backoff(consecutive).as_secs(), + "DHT contact remains unavailable; retaining it for later recovery" + ); + } else { debug!( peer = %contact.peer_id, consecutive, backoff_s = dial_backoff(consecutive).as_secs(), "dial failed; backing off" ); - return; } - lock(&self.dial_failures).remove(&contact.peer_id); - lock(&self.routing).remove(&contact.peer_id); - if let Err(err) = self.db.delete_known_peer(contact.peer_id).await { - warn!(error = %err, "failed to delete evicted peer from the database"); - } - info!(peer = %contact.peer_id, "evicted unreachable DHT contact"); } /// Sends one request and awaits its response, cleaning up the pending diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs index 979f693..2c4905a 100644 --- a/crates/music-dht/src/service.rs +++ b/crates/music-dht/src/service.rs @@ -5,8 +5,8 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use federation_net::{ - ByteStream, EndpointAddr, EndpointId, NetworkConfig, NetworkEngine, NetworkId, PeerTicket, - SchemaId, SecretKey, Signature, StreamAcceptor, + ByteStream, EndpointAddr, EndpointId, NetworkConfig, NetworkEngine, NetworkHealthSnapshot, + NetworkId, PeerTicket, SchemaId, SecretKey, Signature, StreamAcceptor, }; use tokio::sync::mpsc; use tokio::task::JoinHandle; @@ -375,6 +375,11 @@ impl MusicDhtService { self.node.engine.connected_peers() } + /// Returns a point-in-time snapshot of transport and rendezvous health. + pub fn network_health(&self) -> NetworkHealthSnapshot { + self.node.engine.health() + } + /// Returns `true` if a transport connection to `peer` is open. pub fn is_connected(&self, peer: EndpointId) -> bool { self.node.engine.is_connected(peer)