use std::collections::{HashMap, HashSet, VecDeque};
use std::net::SocketAddr;
use std::sync::{Arc, OnceLock, RwLock};
use std::time::Instant;
use crate::ring::reconcile::ReconcileActionDivergence;
use crate::ring::{PeerKeyLocation, SubscribedContractSnapshot};
use crate::router::Router;
use crate::transport::metrics::{TRANSPORT_METRICS, TransportSnapshot};
static NETWORK_STATUS: OnceLock<Arc<RwLock<NetworkStatus>>> = OnceLock::new();
static ROUTER: OnceLock<Arc<parking_lot::RwLock<Router>>> = OnceLock::new();
pub type SubscriptionProvider =
Arc<dyn Fn() -> Vec<SubscribedContractSnapshot> + Send + Sync + 'static>;
pub type GovernanceProvider = Arc<dyn Fn() -> GovernanceSnapshot + Send + Sync + 'static>;
pub type RingStatsProvider = Arc<dyn Fn() -> RingStatsSnapshot + Send + Sync + 'static>;
pub type BanListProvider = Arc<dyn Fn() -> BanListSnapshot + Send + Sync + 'static>;
pub type HostingProvider = Arc<dyn Fn() -> HostingSnapshot + Send + Sync + 'static>;
#[derive(Debug, Clone, Default)]
pub struct RingStatsSnapshot {
pub connection_count: u32,
pub hosted_contracts: u32,
pub peer_id: String,
pub own_pub_key: String,
pub updates_accepted: u64,
pub updates_rate_limited: u64,
pub updates_capacity_dropped: u64,
pub lattice_has_successor: bool,
pub lattice_has_predecessor: bool,
pub lattice_successor_distance: Option<f64>,
pub lattice_predecessor_distance: Option<f64>,
pub lattice_probes_issued: u64,
pub lattice_probe_improvements: u64,
}
static GOVERNANCE_PROVIDER: parking_lot::RwLock<Option<GovernanceProvider>> =
parking_lot::RwLock::new(None);
pub fn set_governance_provider(provider: GovernanceProvider) {
*GOVERNANCE_PROVIDER.write() = Some(provider);
}
static RING_STATS_PROVIDER: parking_lot::RwLock<Option<RingStatsProvider>> =
parking_lot::RwLock::new(None);
pub fn set_ring_stats_provider(provider: RingStatsProvider) {
*RING_STATS_PROVIDER.write() = Some(provider);
}
#[cfg(test)]
pub(crate) fn clear_ring_stats_provider() {
*RING_STATS_PROVIDER.write() = None;
}
#[cfg(test)]
#[allow(dead_code)] pub(crate) fn clear_governance_provider() {
*GOVERNANCE_PROVIDER.write() = None;
}
static BAN_LIST_PROVIDER: parking_lot::RwLock<Option<BanListProvider>> =
parking_lot::RwLock::new(None);
pub fn set_ban_list_provider(provider: BanListProvider) {
*BAN_LIST_PROVIDER.write() = Some(provider);
}
static HOSTING_PROVIDER: parking_lot::RwLock<Option<HostingProvider>> =
parking_lot::RwLock::new(None);
pub fn set_hosting_provider(provider: HostingProvider) {
*HOSTING_PROVIDER.write() = Some(provider);
}
static SUBSCRIPTION_PROVIDER: parking_lot::RwLock<Option<SubscriptionProvider>> =
parking_lot::RwLock::new(None);
pub fn set_router(router: Arc<parking_lot::RwLock<Router>>) {
#[allow(clippy::let_underscore_must_use)]
let _ = ROUTER.set(router);
}
pub(crate) fn get_router() -> Option<Arc<parking_lot::RwLock<Router>>> {
ROUTER.get().cloned()
}
pub fn set_subscription_provider(provider: SubscriptionProvider) {
*SUBSCRIPTION_PROVIDER.write() = Some(provider);
}
#[cfg(test)]
pub(crate) fn clear_subscription_provider() {
*SUBSCRIPTION_PROVIDER.write() = None;
}
pub struct NetworkStatus {
pub gateway_failures: Vec<GatewayFailure>,
pub connection_attempts: u32,
pub listening_port: u16,
pub started_at: Instant,
pub gateway_addresses: HashSet<SocketAddr>,
pub connected_peers: Vec<ConnectedPeer>,
pub version: String,
pub own_location: Option<f64>,
pub external_address: Option<SocketAddr>,
pub op_stats: OperationStats,
pub nat_stats: NatStats,
pub terminal_consult_stats: TerminalConsultStats,
pub upstream_divergence_stats: UpstreamDivergenceStats,
pub reconcile_shadow_collapse: ReconcileShadowStats,
pub reconcile_shadow_renewal: ReconcileShadowStats,
pub reconcile_shadow_inbound_unsubscribe: ReconcileShadowStats,
pub reconcile_shadow_connection_drop: ReconcileShadowStats,
pub reconcile_shadow_host_formation: ReconcileShadowStats,
}
#[derive(Default)]
pub struct UpstreamDivergenceStats {
pub comparisons: u64,
pub divergences: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReconcileShadowSite {
Collapse,
Renewal,
InboundUnsubscribe,
ConnectionDrop,
HostFormation,
}
#[derive(Default, Clone, Copy)]
pub struct ReconcileShadowStats {
pub comparisons: u64,
pub divergences: u64,
pub subscribe_diffs: u64,
pub renew_diffs: u64,
pub unsubscribe_diffs: u64,
pub collapse_diffs: u64,
pub announce_diffs: u64,
pub retract_diffs: u64,
pub reroot_search_diffs: u64,
}
#[derive(Default)]
pub struct TerminalConsultStats {
pub attempts: u64,
pub hits: u64,
pub resolved_found: u64,
pub still_not_found: u64,
}
pub struct ConnectedPeer {
pub address: SocketAddr,
pub is_gateway: bool,
pub location: Option<f64>,
pub connected_since: Instant,
pub peer_key_location: Option<PeerKeyLocation>,
}
#[derive(Default)]
pub struct OperationStats {
pub gets: (u32, u32),
pub puts: (u32, u32),
pub updates: (u32, u32),
pub subscribes: (u32, u32),
pub updates_received: u32,
pub put_probe_new_contract: (u32, u64),
pub put_probe_existing_mesh_delta: (u32, u64),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PutProbeCase {
NewContract,
ExistingMeshDelta,
}
const NAT_RECENT_WINDOW: usize = 20;
#[derive(Default)]
pub struct NatStats {
pub attempts: u32,
pub successes: u32,
recent: VecDeque<bool>,
}
impl NatStats {
pub fn record(&mut self, success: bool) {
self.attempts = self.attempts.saturating_add(1);
if success {
self.successes = self.successes.saturating_add(1);
}
if self.recent.len() >= NAT_RECENT_WINDOW {
self.recent.pop_front();
}
self.recent.push_back(success);
}
pub fn recent_attempts(&self) -> u32 {
self.recent.len() as u32
}
pub fn recent_successes(&self) -> u32 {
self.recent.iter().filter(|&&s| s).count() as u32
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[cfg_attr(test, derive(arbitrary::Arbitrary))]
pub enum OpType {
Get,
Put,
Update,
Subscribe,
}
impl OpType {
pub fn as_str(&self) -> &'static str {
match self {
OpType::Get => "GET",
OpType::Put => "PUT",
OpType::Update => "UPDATE",
OpType::Subscribe => "SUBSCRIBE",
}
}
}
impl std::fmt::Display for OpType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
pub struct GatewayFailure {
pub address: SocketAddr,
pub reason: FailureReason,
}
pub enum FailureReason {
VersionMismatch { local: String, gateway: String },
NatTraversalFailed,
Timeout,
Other(String),
}
pub fn init(listening_port: u16, gateway_addrs: HashSet<SocketAddr>, version: String) {
let status = NetworkStatus {
gateway_failures: Vec::new(),
connection_attempts: 0,
listening_port,
started_at: Instant::now(),
gateway_addresses: gateway_addrs,
connected_peers: Vec::new(),
version,
own_location: None,
external_address: None,
op_stats: OperationStats::default(),
nat_stats: NatStats::default(),
terminal_consult_stats: TerminalConsultStats::default(),
upstream_divergence_stats: UpstreamDivergenceStats::default(),
reconcile_shadow_collapse: ReconcileShadowStats::default(),
reconcile_shadow_renewal: ReconcileShadowStats::default(),
reconcile_shadow_inbound_unsubscribe: ReconcileShadowStats::default(),
reconcile_shadow_connection_drop: ReconcileShadowStats::default(),
reconcile_shadow_host_formation: ReconcileShadowStats::default(),
};
match NETWORK_STATUS.get() {
Some(existing) => {
if let Ok(mut guard) = existing.write() {
*guard = status;
}
}
None => {
#[allow(clippy::let_underscore_must_use)]
let _ = NETWORK_STATUS.set(Arc::new(RwLock::new(status)));
}
}
}
pub fn is_known_gateway(addr: &SocketAddr) -> bool {
NETWORK_STATUS
.get()
.and_then(|s| s.read().ok())
.is_some_and(|s| s.gateway_addresses.contains(addr))
}
pub fn record_gateway_failure(address: SocketAddr, reason: FailureReason) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.connection_attempts = s.connection_attempts.saturating_add(1);
if matches!(reason, FailureReason::NatTraversalFailed) {
s.nat_stats.record(false);
}
s.gateway_failures.push(GatewayFailure { address, reason });
if s.gateway_failures.len() > 20 {
s.gateway_failures.remove(0);
}
}
}
}
pub fn record_peer_connected(
addr: SocketAddr,
location: Option<f64>,
peer_key_location: Option<PeerKeyLocation>,
) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.connected_peers.retain(|p| p.address != addr);
let is_gateway = s.gateway_addresses.contains(&addr);
s.connected_peers.push(ConnectedPeer {
address: addr,
is_gateway,
location,
connected_since: Instant::now(),
peer_key_location,
});
s.gateway_failures.clear();
}
}
}
pub fn record_peer_disconnected(addr: SocketAddr) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.connected_peers.retain(|p| p.address != addr);
}
}
TRANSPORT_METRICS.remove_peer(addr);
}
pub fn record_op_result(op_type: OpType, success: bool) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
let counter = match op_type {
OpType::Get => &mut s.op_stats.gets,
OpType::Put => &mut s.op_stats.puts,
OpType::Update => &mut s.op_stats.updates,
OpType::Subscribe => &mut s.op_stats.subscribes,
};
if success {
counter.0 = counter.0.saturating_add(1);
} else {
counter.1 = counter.1.saturating_add(1);
}
}
}
}
pub fn record_update_received() {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.op_stats.updates_received = s.op_stats.updates_received.saturating_add(1);
}
}
}
pub fn record_put_bytes(case: PutProbeCase, bytes: u64) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
let counter = match case {
PutProbeCase::NewContract => &mut s.op_stats.put_probe_new_contract,
PutProbeCase::ExistingMeshDelta => &mut s.op_stats.put_probe_existing_mesh_delta,
};
counter.0 = counter.0.saturating_add(1);
counter.1 = counter.1.saturating_add(bytes);
}
}
}
pub fn record_terminal_consult_attempt() {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.terminal_consult_stats.attempts = s.terminal_consult_stats.attempts.saturating_add(1);
}
}
}
pub fn record_terminal_consult_hit() {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.terminal_consult_stats.hits = s.terminal_consult_stats.hits.saturating_add(1);
}
}
}
pub fn record_terminal_consult_resolved_found() {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.terminal_consult_stats.resolved_found =
s.terminal_consult_stats.resolved_found.saturating_add(1);
}
}
}
pub fn record_terminal_consult_still_not_found() {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.terminal_consult_stats.still_not_found =
s.terminal_consult_stats.still_not_found.saturating_add(1);
}
}
}
pub fn terminal_consult_counts() -> Option<(u64, u64, u64, u64)> {
let status = NETWORK_STATUS.get()?;
let s = status.read().ok()?;
let c = &s.terminal_consult_stats;
Some((c.attempts, c.hits, c.resolved_found, c.still_not_found))
}
pub fn record_upstream_divergence_comparison(diverged: bool) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.upstream_divergence_stats.comparisons =
s.upstream_divergence_stats.comparisons.saturating_add(1);
if diverged {
s.upstream_divergence_stats.divergences =
s.upstream_divergence_stats.divergences.saturating_add(1);
}
}
}
}
pub fn upstream_divergence_counts() -> Option<(u64, u64)> {
let status = NETWORK_STATUS.get()?;
let s = status.read().ok()?;
let d = &s.upstream_divergence_stats;
Some((d.comparisons, d.divergences))
}
pub fn record_reconcile_shadow_comparison(
site: ReconcileShadowSite,
divergence: ReconcileActionDivergence,
) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
let r = match site {
ReconcileShadowSite::Collapse => &mut s.reconcile_shadow_collapse,
ReconcileShadowSite::Renewal => &mut s.reconcile_shadow_renewal,
ReconcileShadowSite::InboundUnsubscribe => {
&mut s.reconcile_shadow_inbound_unsubscribe
}
ReconcileShadowSite::ConnectionDrop => &mut s.reconcile_shadow_connection_drop,
ReconcileShadowSite::HostFormation => &mut s.reconcile_shadow_host_formation,
};
r.comparisons = r.comparisons.saturating_add(1);
if divergence.any() {
r.divergences = r.divergences.saturating_add(1);
}
if divergence.subscribe {
r.subscribe_diffs = r.subscribe_diffs.saturating_add(1);
}
if divergence.renew {
r.renew_diffs = r.renew_diffs.saturating_add(1);
}
if divergence.unsubscribe {
r.unsubscribe_diffs = r.unsubscribe_diffs.saturating_add(1);
}
if divergence.collapse {
r.collapse_diffs = r.collapse_diffs.saturating_add(1);
}
if divergence.announce {
r.announce_diffs = r.announce_diffs.saturating_add(1);
}
if divergence.retract {
r.retract_diffs = r.retract_diffs.saturating_add(1);
}
if divergence.reroot_search {
r.reroot_search_diffs = r.reroot_search_diffs.saturating_add(1);
}
}
}
}
#[derive(Clone, Copy)]
pub struct ReconcileShadowSnapshot {
pub collapse: ReconcileShadowStats,
pub renewal: ReconcileShadowStats,
pub inbound_unsubscribe: ReconcileShadowStats,
pub connection_drop: ReconcileShadowStats,
pub host_formation: ReconcileShadowStats,
}
pub fn reconcile_shadow_counts() -> Option<ReconcileShadowSnapshot> {
let status = NETWORK_STATUS.get()?;
let s = status.read().ok()?;
Some(ReconcileShadowSnapshot {
collapse: s.reconcile_shadow_collapse,
renewal: s.reconcile_shadow_renewal,
inbound_unsubscribe: s.reconcile_shadow_inbound_unsubscribe,
connection_drop: s.reconcile_shadow_connection_drop,
host_formation: s.reconcile_shadow_host_formation,
})
}
pub fn record_nat_attempt(success: bool) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.nat_stats.record(success);
}
}
}
pub fn set_own_location(location: f64) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.own_location = Some(location);
}
}
}
pub fn set_external_address(addr: SocketAddr) {
if let Some(status) = NETWORK_STATUS.get() {
if let Ok(mut s) = status.write() {
s.external_address = Some(addr);
}
}
}
#[cfg(test)]
pub(crate) fn external_address() -> Option<SocketAddr> {
NETWORK_STATUS
.get()
.and_then(|status| status.read().ok().and_then(|s| s.external_address))
}
#[cfg(test)]
pub(crate) static TEST_GLOBAL_STATE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
pub struct NetworkStatusSnapshot {
pub failures: Vec<FailureSnapshot>,
pub connection_attempts: u32,
pub open_connections: u32,
pub elapsed_secs: u64,
pub listening_port: u16,
pub version: String,
pub own_location: Option<f64>,
pub external_address: Option<SocketAddr>,
pub peers: Vec<PeerSnapshot>,
pub contracts: Vec<ContractSnapshot>,
pub op_stats: OpStatsSnapshot,
pub nat_stats: NatStatsSnapshot,
pub gateway_only: bool,
pub bytes_uploaded: u64,
pub bytes_downloaded: u64,
pub health: HealthLevel,
pub ring_stats: RingStatsSnapshot,
pub transport_snapshot: TransportSnapshot,
#[allow(dead_code)] pub governance: GovernanceSnapshot,
pub ban_list: BanListSnapshot,
pub hosting: HostingSnapshot,
}
#[derive(Debug, Clone, Default)]
pub struct BanListSnapshot {
pub count: usize,
pub capacity_rejected_total: u64,
pub entries: Vec<BanListEntry>,
}
#[derive(Debug, Clone)]
pub struct BanListEntry {
pub instance_id: String,
pub reason: BanReasonSnapshot,
pub expires_in_secs: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BanReasonSnapshot {
AutoMad,
Operator,
}
#[allow(dead_code)] #[derive(Default)]
pub struct GovernanceSnapshot {
pub mode: GovernanceModeSnapshot,
pub contracts: Vec<ContractGovernanceEntry>,
pub observed_count: usize,
pub min_samples: usize,
pub norms: NetworkNorms,
pub last_tick_at: Option<tokio::time::Instant>,
pub state_by_id: std::collections::HashMap<String, GovernanceStateSnapshot>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum GovernanceModeSnapshot {
Off,
#[default]
DryRun,
Enforce,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GovernanceStateSnapshot {
Normal,
Borderline,
WouldEvict,
Evicted,
Banned,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GovernanceTransitionReasonSnapshot {
FirstSeen,
BorderlineEntered,
ThresholdCrossed,
Evicted,
BanTriggered,
Recovered,
BanLifted,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GovernanceSkipReasonSnapshot {
InsufficientSamples,
MadCollapsed,
NoExtractableRatios,
}
#[allow(dead_code)] #[derive(Debug, Clone)]
pub struct ContractGovernanceEntry {
pub instance_id: String,
pub instance_id_short: String,
pub state: GovernanceStateSnapshot,
pub cost_used: f64,
pub benefit_score: f64,
pub log_ratio: Option<f64>,
pub age_secs: u64,
pub last_transition_secs_ago: u64,
pub history: Vec<GovernanceTransitionEntry>,
}
#[allow(dead_code)] #[derive(Debug, Clone)]
pub struct GovernanceTransitionEntry {
pub secs_ago: u64,
pub from: GovernanceStateSnapshot,
pub to: GovernanceStateSnapshot,
pub reason: GovernanceTransitionReasonSnapshot,
}
#[allow(dead_code)] #[derive(Default, Debug, Clone)]
pub struct NetworkNorms {
pub median_log_ratio: Option<f64>,
pub mad: Option<f64>,
pub threshold: Option<f64>,
pub sample_size: usize,
pub capacity_ceiling_binding: bool,
pub skip_reason: Option<GovernanceSkipReasonSnapshot>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealthLevel {
Healthy,
Degraded,
Connecting,
Trouble,
}
pub struct FailureSnapshot {
pub address: SocketAddr,
pub reason_html: String,
}
pub struct PeerSnapshot {
pub address: SocketAddr,
pub is_gateway: bool,
pub location: Option<f64>,
pub connected_secs: u64,
pub peer_key_location: Option<PeerKeyLocation>,
pub bytes_sent: u64,
pub bytes_received: u64,
}
pub struct ContractSnapshot {
pub key_short: String,
pub key_full: String,
pub instance_id: String,
pub subscribed_secs: u64,
pub last_updated_secs: Option<u64>,
pub is_receiving_updates: bool,
pub in_use: bool,
}
#[derive(Default, Clone)]
pub struct HostingSnapshot {
pub budget_bytes: u64,
pub used_bytes: u64,
pub contract_count: u64,
pub budget_evictions_total: u64,
pub evictions_of_recently_read_total: u64,
pub contracts: Vec<HostedContractEntry>,
pub disk_state_bytes: Option<u64>,
pub disk_wasm_bytes: Option<u64>,
pub disk_compile_cache_bytes: Option<u64>,
pub disk_total_bytes: Option<u64>,
pub disk_budget_bytes: Option<u64>,
}
#[derive(Clone)]
pub struct HostedContractEntry {
pub key_full: String,
pub key_short: String,
pub keep_score: f64,
pub predicted_demand: f64,
pub size_bytes: u64,
pub read_count: u32,
pub eviction_eligible: bool,
}
#[derive(Default)]
pub struct OpStatsSnapshot {
pub gets: (u32, u32),
pub puts: (u32, u32),
pub updates: (u32, u32),
pub subscribes: (u32, u32),
pub updates_received: u32,
}
impl OpStatsSnapshot {
pub fn total(&self) -> u32 {
let sum = |pair: (u32, u32)| pair.0.saturating_add(pair.1);
sum(self.gets)
.saturating_add(sum(self.puts))
.saturating_add(sum(self.updates))
.saturating_add(sum(self.subscribes))
.saturating_add(self.updates_received)
}
}
#[derive(Default)]
pub struct NatStatsSnapshot {
pub attempts: u32,
pub successes: u32,
pub recent_attempts: u32,
pub recent_successes: u32,
}
pub fn get_snapshot() -> Option<NetworkStatusSnapshot> {
let status = NETWORK_STATUS.get()?;
let s = status.read().ok()?;
let now = Instant::now();
let failures = s
.gateway_failures
.iter()
.map(|f| FailureSnapshot {
address: f.address,
reason_html: match &f.reason {
FailureReason::VersionMismatch { local, gateway } => {
let local = html_escape(local);
let gateway = html_escape(gateway);
format!(
"<strong>Version mismatch</strong>: Your version: <code>{local}</code> \
— Gateway requires: <code>{gateway}</code><br>\
Run: <code>cargo install --force freenet --version {gateway}</code>"
)
}
FailureReason::NatTraversalFailed => {
let has_peer_connections = s.connected_peers.iter().any(|p| !p.is_gateway);
if has_peer_connections {
"<strong>NAT traversal failed</strong>: Could not connect to this \
peer. This is normal — not all NAT traversal attempts succeed."
.to_string()
} else {
format!(
"<strong>NAT traversal failed</strong>: Can't reach gateway. \
Check that UDP port <code>{}</code> is open in your firewall.",
s.listening_port
)
}
}
FailureReason::Timeout => {
"<strong>Connection timed out</strong>: Gateway did not respond.".to_string()
}
FailureReason::Other(msg) => {
format!("<strong>Connection failed</strong>: {}", html_escape(msg))
}
},
})
.collect();
let per_peer_metrics: HashMap<SocketAddr, (u64, u64)> = TRANSPORT_METRICS
.per_peer_snapshot()
.into_iter()
.map(|(addr, sent, recv)| (addr, (sent, recv)))
.collect();
let peers: Vec<PeerSnapshot> = s
.connected_peers
.iter()
.map(|p| {
let (sent, recv) = per_peer_metrics.get(&p.address).copied().unwrap_or((0, 0));
PeerSnapshot {
address: p.address,
is_gateway: p.is_gateway,
location: p.location,
connected_secs: now.duration_since(p.connected_since).as_secs(),
peer_key_location: p.peer_key_location.clone(),
bytes_sent: sent,
bytes_received: recv,
}
})
.collect();
let open_connections = peers.len() as u32;
let gateway_only = open_connections > 0 && peers.iter().all(|p| p.is_gateway);
let contracts: Vec<ContractSnapshot> = SUBSCRIPTION_PROVIDER
.read()
.as_ref()
.map(|provider| {
provider()
.into_iter()
.map(|c| {
let key_full = c.key.to_string();
let instance_id = c.key.id().to_string();
let key_short = if key_full.chars().count() > 12 {
let trunc: String = key_full.chars().take(12).collect();
format!("{trunc}...")
} else {
key_full.clone()
};
ContractSnapshot {
key_short,
key_full,
instance_id,
subscribed_secs: c.subscribed_secs,
last_updated_secs: c.last_updated_secs,
is_receiving_updates: c.is_receiving_updates,
in_use: c.in_use,
}
})
.collect()
})
.unwrap_or_default();
let elapsed_secs = s.started_at.elapsed().as_secs();
let has_version_mismatch = s
.gateway_failures
.iter()
.any(|f| matches!(f.reason, FailureReason::VersionMismatch { .. }));
let nat_all_failing = s.nat_stats.attempts > 0 && s.nat_stats.successes == 0;
let health = if has_version_mismatch || (open_connections == 0 && elapsed_secs > 60) {
HealthLevel::Trouble
} else if open_connections == 0 {
HealthLevel::Connecting
} else if gateway_only || nat_all_failing {
HealthLevel::Degraded
} else {
HealthLevel::Healthy
};
let bytes_uploaded = TRANSPORT_METRICS.cumulative_bytes_sent();
let bytes_downloaded = TRANSPORT_METRICS.cumulative_bytes_received();
let ring_stats = RING_STATS_PROVIDER
.read()
.as_ref()
.map(|provider| provider())
.unwrap_or_default();
let transport_snapshot = TRANSPORT_METRICS.read_snapshot();
Some(NetworkStatusSnapshot {
failures,
connection_attempts: s.connection_attempts,
open_connections,
elapsed_secs,
listening_port: s.listening_port,
version: s.version.clone(),
own_location: s.own_location,
external_address: s.external_address,
peers,
contracts,
op_stats: OpStatsSnapshot {
gets: s.op_stats.gets,
puts: s.op_stats.puts,
updates: s.op_stats.updates,
subscribes: s.op_stats.subscribes,
updates_received: s.op_stats.updates_received,
},
nat_stats: NatStatsSnapshot {
attempts: s.nat_stats.attempts,
successes: s.nat_stats.successes,
recent_attempts: s.nat_stats.recent_attempts(),
recent_successes: s.nat_stats.recent_successes(),
},
gateway_only,
bytes_uploaded,
bytes_downloaded,
health,
ring_stats,
transport_snapshot,
governance: GOVERNANCE_PROVIDER
.read()
.as_ref()
.map(|provider| provider())
.unwrap_or_default(),
ban_list: BAN_LIST_PROVIDER
.read()
.as_ref()
.map(|provider| provider())
.unwrap_or_default(),
hosting: HOSTING_PROVIDER
.read()
.as_ref()
.map(|provider| provider())
.unwrap_or_default(),
})
}
pub fn classify_transport_error(error_msg: &str) -> FailureReason {
if error_msg.contains("Version incompatibility") {
let local = error_msg
.split("Your client version:")
.nth(1)
.and_then(|s| s.split('\n').next())
.map(|s| s.trim().to_string())
.unwrap_or_default();
let gateway = error_msg
.split("Gateway version:")
.nth(1)
.and_then(|s| s.split('\n').next())
.map(|s| s.trim().to_string())
.unwrap_or_default();
FailureReason::VersionMismatch { local, gateway }
} else if error_msg.contains("max connection attempts reached") {
FailureReason::NatTraversalFailed
} else {
FailureReason::Other(error_msg.to_string())
}
}
pub(crate) fn html_escape(s: &str) -> String {
s.replace('&', "&")
.replace('<', "<")
.replace('>', ">")
.replace('"', """)
}
pub(crate) fn format_duration(secs: u64) -> String {
if secs < 60 {
format!("{secs}s")
} else if secs < 3600 {
let m = secs / 60;
let s = secs % 60;
if s == 0 {
format!("{m}m")
} else {
format!("{m}m {s}s")
}
} else {
let h = secs / 3600;
let m = (secs % 3600) / 60;
if m == 0 {
format!("{h}h")
} else {
format!("{h}h {m}m")
}
}
}
pub(crate) fn format_ago(secs: u64) -> String {
if secs < 5 {
"just now".to_string()
} else {
format!("{} ago", format_duration(secs))
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::{IpAddr, Ipv4Addr};
use super::TEST_GLOBAL_STATE_LOCK as TEST_MUTEX;
#[test]
fn test_classify_version_mismatch() {
let msg = "Version incompatibility with gateway\n Your client version: 0.1.133\n Gateway version: 0.1.135\n \n To fix this, update your Freenet client:\n cargo install --force freenet --version 0.1.135";
let reason = classify_transport_error(msg);
match reason {
FailureReason::VersionMismatch { local, gateway } => {
assert_eq!(local, "0.1.133");
assert_eq!(gateway, "0.1.135");
}
FailureReason::NatTraversalFailed
| FailureReason::Timeout
| FailureReason::Other(_) => panic!("Expected VersionMismatch"),
}
}
#[test]
fn test_classify_nat_traversal() {
let msg = "failed while establishing connection, reason: max connection attempts reached";
let reason = classify_transport_error(msg);
assert!(matches!(reason, FailureReason::NatTraversalFailed));
}
#[test]
fn test_classify_other() {
let msg = "some unknown error";
let reason = classify_transport_error(msg);
assert!(matches!(reason, FailureReason::Other(_)));
}
#[test]
fn test_snapshot_without_init_returns_none() {
let _ = get_snapshot();
}
#[test]
fn test_html_escape() {
assert_eq!(html_escape("<script>"), "<script>");
assert_eq!(html_escape("a&b"), "a&b");
}
#[test]
fn terminal_consult_counts_reflect_recorded_events() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31337, HashSet::new(), "test".to_string());
assert_eq!(
terminal_consult_counts(),
Some((0, 0, 0, 0)),
"counters start at zero after init"
);
record_terminal_consult_attempt();
record_terminal_consult_attempt();
record_terminal_consult_attempt();
record_terminal_consult_hit();
record_terminal_consult_hit();
record_terminal_consult_resolved_found();
record_terminal_consult_still_not_found();
record_terminal_consult_still_not_found();
record_terminal_consult_still_not_found();
record_terminal_consult_still_not_found();
assert_eq!(
terminal_consult_counts(),
Some((3, 2, 1, 4)),
"getter returns (attempts, hits, resolved_found, still_not_found)"
);
}
#[test]
fn upstream_divergence_counts_reflect_recorded_events() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31337, HashSet::new(), "test".to_string());
assert_eq!(
upstream_divergence_counts(),
Some((0, 0)),
"counters start at zero after init"
);
record_upstream_divergence_comparison(false);
record_upstream_divergence_comparison(true);
record_upstream_divergence_comparison(false);
record_upstream_divergence_comparison(true);
record_upstream_divergence_comparison(false);
assert_eq!(
upstream_divergence_counts(),
Some((5, 2)),
"getter returns (comparisons, divergences); every call bumps comparisons, \
only diverged calls bump divergences"
);
}
#[test]
fn reconcile_shadow_counts_reflect_recorded_events() {
use ReconcileShadowSite::*;
let _lock = TEST_MUTEX.lock().unwrap();
init(31337, HashSet::new(), "test".to_string());
let z = reconcile_shadow_counts().expect("counters present after init");
assert_eq!(
(
z.collapse.comparisons,
z.renewal.comparisons,
z.inbound_unsubscribe.comparisons,
z.connection_drop.comparisons,
z.host_formation.comparisons,
),
(0, 0, 0, 0, 0),
"all sites start at zero after init"
);
record_reconcile_shadow_comparison(Collapse, ReconcileActionDivergence::default());
record_reconcile_shadow_comparison(
Collapse,
ReconcileActionDivergence {
retract: true,
..Default::default()
},
);
record_reconcile_shadow_comparison(
Renewal,
ReconcileActionDivergence {
subscribe: true,
renew: true,
..Default::default()
},
);
record_reconcile_shadow_comparison(
ConnectionDrop,
ReconcileActionDivergence {
reroot_search: true,
..Default::default()
},
);
let c = reconcile_shadow_counts().expect("counters present");
assert_eq!(c.collapse.comparisons, 2, "collapse site: every call bumps");
assert_eq!(
c.collapse.divergences, 1,
"collapse site: one diverging call"
);
assert_eq!(c.collapse.retract_diffs, 1);
assert_eq!(c.collapse.subscribe_diffs, 0);
assert_eq!(
c.renewal.comparisons, 1,
"renewal site independent of collapse"
);
assert_eq!(c.renewal.subscribe_diffs, 1);
assert_eq!(c.renewal.renew_diffs, 1);
assert_eq!(c.renewal.retract_diffs, 0);
assert_eq!(c.connection_drop.comparisons, 1);
assert_eq!(c.connection_drop.reroot_search_diffs, 1);
assert_eq!(c.connection_drop.renew_diffs, 0);
assert_eq!(c.inbound_unsubscribe.comparisons, 0);
assert_eq!(c.host_formation.comparisons, 0);
}
#[test]
fn test_failure_snapshot_rendering() {
let _lock = TEST_MUTEX.lock().unwrap();
let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 31337);
let reason = FailureReason::VersionMismatch {
local: "0.1.1".to_string(),
gateway: "0.1.2".to_string(),
};
init(31338, HashSet::new(), "0.1.0".to_string());
record_gateway_failure(addr, reason);
let snap = get_snapshot().unwrap();
assert_eq!(snap.failures.len(), 1);
assert!(snap.failures[0].reason_html.contains("Version mismatch"));
assert!(snap.failures[0].reason_html.contains("0.1.1"));
}
#[test]
fn test_format_duration() {
assert_eq!(format_duration(0), "0s");
assert_eq!(format_duration(45), "45s");
assert_eq!(format_duration(60), "1m");
assert_eq!(format_duration(90), "1m 30s");
assert_eq!(format_duration(3600), "1h");
assert_eq!(format_duration(3660), "1h 1m");
assert_eq!(format_duration(7200), "2h");
}
#[test]
fn test_format_ago() {
assert_eq!(format_ago(2), "just now");
assert_eq!(format_ago(30), "30s ago");
assert_eq!(format_ago(120), "2m ago");
}
#[test]
fn test_gateway_only_detection() {
let _lock = TEST_MUTEX.lock().unwrap();
let gw_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(5, 9, 111, 215)), 31337);
let peer_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 12345);
init(31339, HashSet::new(), "0.1.148".to_string());
let status = NETWORK_STATUS.get().unwrap();
{
let mut s = status.write().unwrap();
s.gateway_addresses.insert(gw_addr);
s.connected_peers.clear();
s.gateway_failures.clear();
s.connected_peers.push(ConnectedPeer {
address: gw_addr,
is_gateway: true,
location: Some(0.5),
connected_since: Instant::now(),
peer_key_location: None,
});
}
let snap = get_snapshot().unwrap();
assert!(snap.gateway_only);
{
let mut s = status.write().unwrap();
s.connected_peers.push(ConnectedPeer {
address: peer_addr,
is_gateway: false,
location: Some(0.3),
connected_since: Instant::now(),
peer_key_location: None,
});
}
let snap = get_snapshot().unwrap();
assert!(!snap.gateway_only);
{
let mut s = status.write().unwrap();
s.connected_peers.retain(|p| p.address != peer_addr);
}
let snap = get_snapshot().unwrap();
assert!(snap.gateway_only);
{
let mut s = status.write().unwrap();
s.connected_peers.clear();
}
}
#[test]
fn test_op_stats_recording() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31340, HashSet::new(), "0.1.148".to_string());
if let Some(status) = NETWORK_STATUS.get() {
let mut s = status.write().unwrap();
s.op_stats = OperationStats::default();
}
record_op_result(OpType::Get, true);
record_op_result(OpType::Get, true);
record_op_result(OpType::Get, false);
record_op_result(OpType::Put, true);
record_op_result(OpType::Subscribe, true);
record_op_result(OpType::Subscribe, false);
record_op_result(OpType::Update, true);
record_op_result(OpType::Update, true);
record_op_result(OpType::Update, false);
let snap = get_snapshot().unwrap();
assert_eq!(snap.op_stats.gets, (2, 1));
assert_eq!(snap.op_stats.puts, (1, 0));
assert_eq!(snap.op_stats.subscribes, (1, 1));
assert_eq!(snap.op_stats.updates, (2, 1));
}
#[test]
fn test_subscription_tracking_no_provider_wired() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31341, HashSet::new(), "0.1.148".to_string());
clear_subscription_provider();
let snap = get_snapshot().unwrap();
assert!(snap.contracts.is_empty());
}
#[test]
fn test_subscription_provider_wired_renders_contracts() {
use freenet_stdlib::prelude::{ContractCode, ContractInstanceId, ContractKey};
let _lock = TEST_MUTEX.lock().unwrap();
init(31350, HashSet::new(), "0.1.148".to_string());
let code_hash = *ContractCode::from(vec![0u8; 32]).hash();
let key_a = ContractKey::from_id_and_code(ContractInstanceId::new([0xAA; 32]), code_hash);
let key_b = ContractKey::from_id_and_code(ContractInstanceId::new([0xBB; 32]), code_hash);
let snapshot = vec![
crate::ring::SubscribedContractSnapshot {
key: key_a,
subscribed_secs: 30,
last_updated_secs: Some(5),
is_receiving_updates: true,
in_use: true,
},
crate::ring::SubscribedContractSnapshot {
key: key_b,
subscribed_secs: 30,
last_updated_secs: None,
is_receiving_updates: true,
in_use: true,
},
];
set_subscription_provider(Arc::new(move || snapshot.clone()));
let snap = get_snapshot().unwrap();
assert_eq!(snap.contracts.len(), 2);
assert_eq!(snap.contracts[0].key_full, key_a.to_string());
assert_eq!(snap.contracts[0].subscribed_secs, 30);
assert_eq!(snap.contracts[0].last_updated_secs, Some(5));
assert!(snap.contracts[0].key_short.ends_with("..."));
assert_eq!(snap.contracts[0].key_short.chars().count(), 15);
set_subscription_provider(Arc::new(Vec::new));
let snap = get_snapshot().unwrap();
assert!(snap.contracts.is_empty());
clear_subscription_provider();
}
#[test]
fn test_nat_stats() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31342, HashSet::new(), "0.1.148".to_string());
if let Some(status) = NETWORK_STATUS.get() {
let mut s = status.write().unwrap();
s.nat_stats = NatStats::default();
}
record_nat_attempt(true);
record_nat_attempt(false);
record_nat_attempt(false);
let snap = get_snapshot().unwrap();
assert_eq!(snap.nat_stats.attempts, 3);
assert_eq!(snap.nat_stats.successes, 1);
}
#[test]
fn test_nat_failure_no_peers_shows_firewall_message() {
let _lock = TEST_MUTEX.lock().unwrap();
let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 31337);
init(31343, HashSet::new(), "0.1.148".to_string());
if let Some(status) = NETWORK_STATUS.get() {
let mut s = status.write().unwrap();
s.connected_peers.clear();
s.gateway_failures.clear();
}
record_gateway_failure(addr, FailureReason::NatTraversalFailed);
let snap = get_snapshot().unwrap();
let html = &snap.failures.last().unwrap().reason_html;
assert!(html.contains("NAT traversal failed"));
assert!(html.contains("Check that UDP port"));
assert!(html.contains("open in your firewall"));
}
#[test]
fn test_nat_failure_with_peer_connections_shows_benign_message() {
let _lock = TEST_MUTEX.lock().unwrap();
let peer_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 12345);
let fail_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 31337);
init(31344, HashSet::new(), "0.1.148".to_string());
if let Some(status) = NETWORK_STATUS.get() {
let mut s = status.write().unwrap();
s.connected_peers.clear();
s.gateway_failures.clear();
}
record_peer_connected(peer_addr, Some(0.5), None);
record_gateway_failure(fail_addr, FailureReason::NatTraversalFailed);
let snap = get_snapshot().unwrap();
let html = &snap.failures.last().unwrap().reason_html;
assert!(html.contains("NAT traversal failed"));
assert!(html.contains("This is normal"));
assert!(!html.contains("firewall"));
record_peer_disconnected(peer_addr);
}
#[test]
fn test_nat_failure_with_only_gateway_connections_shows_firewall_message() {
let _lock = TEST_MUTEX.lock().unwrap();
let gw_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(5, 9, 111, 215)), 31337);
let fail_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 31337);
init(31345, HashSet::new(), "0.1.148".to_string());
if let Some(status) = NETWORK_STATUS.get() {
let mut s = status.write().unwrap();
s.gateway_addresses.insert(gw_addr);
s.connected_peers.clear();
s.gateway_failures.clear();
}
record_peer_connected(gw_addr, None, None);
record_gateway_failure(fail_addr, FailureReason::NatTraversalFailed);
let snap = get_snapshot().unwrap();
let html = &snap.failures.last().unwrap().reason_html;
assert!(html.contains("NAT traversal failed"));
assert!(html.contains("firewall"));
assert!(!html.contains("This is normal"));
record_peer_disconnected(gw_addr);
}
#[test]
fn test_nat_stats_rolling_window() {
let mut nat = NatStats::default();
for _ in 0..10 {
nat.record(true);
}
for _ in 0..15 {
nat.record(false);
}
assert_eq!(nat.attempts, 25);
assert_eq!(nat.successes, 10);
assert_eq!(nat.recent_attempts(), 20);
assert_eq!(nat.recent_successes(), 5);
}
#[test]
fn test_record_peer_connected_updates_location() {
let _lock = TEST_MUTEX.lock().unwrap();
let peer_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 9999);
init(31339, HashSet::new(), "0.1.148".to_string());
record_peer_connected(peer_addr, None, None);
let snap = get_snapshot().unwrap();
let peer = snap.peers.iter().find(|p| p.address == peer_addr).unwrap();
assert_eq!(peer.location, None);
record_peer_connected(peer_addr, Some(0.42), None);
let snap = get_snapshot().unwrap();
let peer = snap.peers.iter().find(|p| p.address == peer_addr).unwrap();
assert_eq!(peer.location, Some(0.42));
record_peer_disconnected(peer_addr);
}
#[test]
fn test_nat_stats_rolling_window_all_failures() {
let mut nat = NatStats::default();
for _ in 0..25 {
nat.record(false);
}
assert_eq!(nat.recent_attempts(), 20);
assert_eq!(nat.recent_successes(), 0);
}
#[test]
fn init_populates_snapshot() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31337, HashSet::new(), "0.1.0-test".to_string());
let snap = get_snapshot();
assert!(snap.is_some(), "snapshot should be Some after init");
let s = snap.unwrap();
assert_eq!(s.version, "0.1.0-test");
assert_eq!(s.listening_port, 31337);
assert_eq!(s.open_connections, 0);
}
#[test]
fn init_overwrites_existing_global() {
let _lock = TEST_MUTEX.lock().unwrap();
init(40001, HashSet::new(), "first-version".to_string());
record_gateway_failure(
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 31337),
FailureReason::Timeout,
);
let first = get_snapshot().unwrap();
assert_eq!(first.version, "first-version");
assert_eq!(first.listening_port, 40001);
assert_eq!(
first.failures.len(),
1,
"failure recorded against the first tracker"
);
init(40002, HashSet::new(), "second-version".to_string());
let second = get_snapshot().unwrap();
assert_eq!(
second.version, "second-version",
"second init() version must win (first-write-wins would keep `first-version`)"
);
assert_eq!(second.listening_port, 40002);
assert!(
second.failures.is_empty(),
"second init() must reset accumulated gateway failures"
);
}
#[test]
fn ring_stats_provider_round_trip() {
let _lock = TEST_MUTEX.lock().unwrap();
init(31337, HashSet::new(), "0.1.0-test".to_string());
let snap = get_snapshot().unwrap();
assert_eq!(snap.ring_stats.connection_count, 0);
assert!(snap.ring_stats.own_pub_key.is_empty());
set_ring_stats_provider(Arc::new(|| RingStatsSnapshot {
connection_count: 42,
hosted_contracts: 7,
peer_id: "abc".to_string(),
own_pub_key: "test-key".to_string(),
updates_accepted: 1000,
updates_rate_limited: 13,
updates_capacity_dropped: 2,
..Default::default()
}));
let snap = get_snapshot().unwrap();
assert_eq!(snap.ring_stats.connection_count, 42);
assert_eq!(snap.ring_stats.peer_id, "abc");
assert_eq!(snap.ring_stats.own_pub_key, "test-key");
assert_eq!(snap.ring_stats.updates_accepted, 1000);
assert_eq!(snap.ring_stats.updates_rate_limited, 13);
assert_eq!(snap.ring_stats.updates_capacity_dropped, 2);
set_ring_stats_provider(Arc::new(|| RingStatsSnapshot {
connection_count: 99,
hosted_contracts: 1,
peer_id: "xyz".to_string(),
own_pub_key: "new-key".to_string(),
..Default::default()
}));
let snap = get_snapshot().unwrap();
assert_eq!(snap.ring_stats.connection_count, 99);
assert_eq!(snap.ring_stats.peer_id, "xyz");
assert_eq!(snap.ring_stats.own_pub_key, "new-key");
clear_ring_stats_provider();
}
}