use libp2p::{Multiaddr, PeerId};
use parking_lot::RwLock;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tracing::{debug, info, warn};
pub const DEFAULT_IPFS_BOOTSTRAP_PEERS: &[&str] = &[
"/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
];
#[derive(Debug, Clone)]
pub struct BootstrapConfig {
pub max_retries: u32,
pub initial_backoff: Duration,
pub max_backoff: Duration,
pub backoff_multiplier: f64,
pub circuit_breaker_threshold: u32,
pub circuit_breaker_timeout: Duration,
pub re_bootstrap_interval: Duration,
}
impl Default for BootstrapConfig {
fn default() -> Self {
Self {
max_retries: 5,
initial_backoff: Duration::from_secs(1),
max_backoff: Duration::from_secs(60),
backoff_multiplier: 2.0,
circuit_breaker_threshold: 3,
circuit_breaker_timeout: Duration::from_secs(300),
re_bootstrap_interval: Duration::from_secs(300),
}
}
}
#[derive(Debug, Clone)]
struct BootstrapPeerState {
addr: Multiaddr,
attempts: u32,
consecutive_failures: u32,
last_attempt: Option<Instant>,
last_success: Option<Instant>,
circuit_open: bool,
circuit_opened_at: Option<Instant>,
current_backoff: Duration,
}
impl BootstrapPeerState {
fn new(addr: Multiaddr, initial_backoff: Duration) -> Self {
Self {
addr,
attempts: 0,
consecutive_failures: 0,
last_attempt: None,
last_success: None,
circuit_open: false,
circuit_opened_at: None,
current_backoff: initial_backoff,
}
}
fn should_retry(&self, config: &BootstrapConfig) -> bool {
if self.circuit_open {
if let Some(opened_at) = self.circuit_opened_at {
if opened_at.elapsed() < config.circuit_breaker_timeout {
return false;
}
}
}
if self.consecutive_failures >= config.max_retries {
return false;
}
if let Some(last) = self.last_attempt {
if last.elapsed() < self.current_backoff {
return false;
}
}
true
}
fn record_attempt(&mut self) {
self.attempts += 1;
self.last_attempt = Some(Instant::now());
}
fn record_success(&mut self, initial_backoff: Duration) {
self.consecutive_failures = 0;
self.last_success = Some(Instant::now());
self.circuit_open = false;
self.circuit_opened_at = None;
self.current_backoff = initial_backoff;
}
fn record_failure(&mut self, config: &BootstrapConfig) {
self.consecutive_failures += 1;
let new_backoff =
Duration::from_secs_f64(self.current_backoff.as_secs_f64() * config.backoff_multiplier);
self.current_backoff = new_backoff.min(config.max_backoff);
if self.consecutive_failures >= config.circuit_breaker_threshold {
self.circuit_open = true;
self.circuit_opened_at = Some(Instant::now());
warn!(
"Circuit breaker opened for peer {} after {} failures",
self.addr, self.consecutive_failures
);
}
}
}
pub struct BootstrapManager {
config: BootstrapConfig,
peers: Arc<RwLock<HashMap<String, BootstrapPeerState>>>,
connected: Arc<RwLock<Vec<PeerId>>>,
}
impl BootstrapManager {
pub fn new(config: BootstrapConfig) -> Self {
Self {
config,
peers: Arc::new(RwLock::new(HashMap::new())),
connected: Arc::new(RwLock::new(Vec::new())),
}
}
pub fn add_peer(&self, addr: Multiaddr) {
let key = addr.to_string();
let mut peers = self.peers.write();
peers
.entry(key)
.or_insert_with(|| BootstrapPeerState::new(addr, self.config.initial_backoff));
}
pub fn add_peers_from_strings(&self, addrs: &[String]) {
for addr_str in addrs {
if let Ok(addr) = addr_str.parse::<Multiaddr>() {
self.add_peer(addr);
} else {
warn!("Invalid bootstrap peer address: {}", addr_str);
}
}
}
pub fn add_default_peers(&self) {
for addr_str in DEFAULT_IPFS_BOOTSTRAP_PEERS {
if let Ok(addr) = addr_str.parse::<Multiaddr>() {
self.add_peer(addr);
}
}
info!(
"Added {} default IPFS bootstrap peers",
DEFAULT_IPFS_BOOTSTRAP_PEERS.len()
);
}
pub fn get_peers_to_dial(&self) -> Vec<Multiaddr> {
let peers = self.peers.read();
peers
.values()
.filter(|state| state.should_retry(&self.config))
.map(|state| state.addr.clone())
.collect()
}
pub fn record_dial_attempt(&self, addr: &Multiaddr) {
let key = addr.to_string();
let mut peers = self.peers.write();
if let Some(state) = peers.get_mut(&key) {
state.record_attempt();
debug!("Recorded dial attempt for {}", addr);
}
}
pub fn record_connection_success(&self, addr: &Multiaddr, peer_id: PeerId) {
let key = addr.to_string();
let mut peers = self.peers.write();
if let Some(state) = peers.get_mut(&key) {
state.record_success(self.config.initial_backoff);
info!(
"Successfully connected to bootstrap peer {} ({})",
addr, peer_id
);
}
let mut connected = self.connected.write();
if !connected.contains(&peer_id) {
connected.push(peer_id);
}
}
pub fn record_connection_failure(&self, addr: &Multiaddr) {
let key = addr.to_string();
let mut peers = self.peers.write();
if let Some(state) = peers.get_mut(&key) {
state.record_failure(&self.config);
warn!(
"Failed to connect to bootstrap peer {}, backoff: {:?}",
addr, state.current_backoff
);
}
}
pub fn record_disconnection(&self, peer_id: &PeerId) {
let mut connected = self.connected.write();
connected.retain(|p| p != peer_id);
}
pub fn has_sufficient_connections(&self, min_peers: usize) -> bool {
self.connected.read().len() >= min_peers
}
pub fn connected_count(&self) -> usize {
self.connected.read().len()
}
pub fn stats(&self) -> BootstrapStats {
let peers = self.peers.read();
let connected = self.connected.read();
let total_attempts: u32 = peers.values().map(|s| s.attempts).sum();
let total_failures: u32 = peers.values().map(|s| s.consecutive_failures).sum();
let open_circuits = peers.values().filter(|s| s.circuit_open).count();
BootstrapStats {
total_peers: peers.len(),
connected_peers: connected.len(),
total_attempts,
total_failures,
open_circuits,
}
}
pub fn reset_circuit_breaker(&self, addr: &Multiaddr) {
let key = addr.to_string();
let mut peers = self.peers.write();
if let Some(state) = peers.get_mut(&key) {
state.circuit_open = false;
state.circuit_opened_at = None;
state.consecutive_failures = 0;
state.current_backoff = self.config.initial_backoff;
info!("Reset circuit breaker for {}", addr);
}
}
pub fn reset_all_circuit_breakers(&self) {
let mut peers = self.peers.write();
for state in peers.values_mut() {
state.circuit_open = false;
state.circuit_opened_at = None;
state.consecutive_failures = 0;
state.current_backoff = self.config.initial_backoff;
}
info!("Reset all circuit breakers");
}
}
impl Default for BootstrapManager {
fn default() -> Self {
Self::new(BootstrapConfig::default())
}
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct BootstrapStats {
pub total_peers: usize,
pub connected_peers: usize,
pub total_attempts: u32,
pub total_failures: u32,
pub open_circuits: usize,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_bootstrap_config_default() {
let config = BootstrapConfig::default();
assert_eq!(config.max_retries, 5);
assert_eq!(config.initial_backoff, Duration::from_secs(1));
}
#[test]
fn test_bootstrap_manager_add_peer() {
let manager = BootstrapManager::default();
let addr: Multiaddr = "/ip4/127.0.0.1/tcp/4001"
.parse()
.expect("test: valid multiaddr should parse");
manager.add_peer(addr.clone());
let peers = manager.get_peers_to_dial();
assert_eq!(peers.len(), 1);
assert_eq!(peers[0], addr);
}
#[test]
fn test_bootstrap_manager_backoff() {
let config = BootstrapConfig {
initial_backoff: Duration::from_millis(10),
max_backoff: Duration::from_secs(1),
backoff_multiplier: 2.0,
..Default::default()
};
let manager = BootstrapManager::new(config);
let addr: Multiaddr = "/ip4/127.0.0.1/tcp/4001"
.parse()
.expect("test: valid multiaddr should parse");
manager.add_peer(addr.clone());
manager.record_dial_attempt(&addr);
manager.record_connection_failure(&addr);
let peers = manager.get_peers_to_dial();
assert!(peers.is_empty());
std::thread::sleep(Duration::from_millis(25));
let peers = manager.get_peers_to_dial();
assert_eq!(peers.len(), 1);
}
#[test]
fn test_bootstrap_manager_circuit_breaker() {
let config = BootstrapConfig {
initial_backoff: Duration::from_millis(1),
circuit_breaker_threshold: 2,
circuit_breaker_timeout: Duration::from_secs(1),
..Default::default()
};
let manager = BootstrapManager::new(config);
let addr: Multiaddr = "/ip4/127.0.0.1/tcp/4001"
.parse()
.expect("test: valid multiaddr should parse");
manager.add_peer(addr.clone());
manager.record_dial_attempt(&addr);
manager.record_connection_failure(&addr);
std::thread::sleep(Duration::from_millis(5));
manager.record_dial_attempt(&addr);
manager.record_connection_failure(&addr);
let peers = manager.get_peers_to_dial();
assert!(peers.is_empty());
let stats = manager.stats();
assert_eq!(stats.open_circuits, 1);
}
#[test]
fn test_bootstrap_manager_success_resets() {
let config = BootstrapConfig {
initial_backoff: Duration::from_millis(1),
..Default::default()
};
let manager = BootstrapManager::new(config);
let addr: Multiaddr = "/ip4/127.0.0.1/tcp/4001"
.parse()
.expect("test: valid multiaddr should parse");
let peer_id = PeerId::random();
manager.add_peer(addr.clone());
manager.record_dial_attempt(&addr);
manager.record_connection_failure(&addr);
manager.record_connection_success(&addr, peer_id);
assert!(manager.has_sufficient_connections(1));
assert_eq!(manager.connected_count(), 1);
}
}