use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
pub const METADATA_QUEUE_WAIT_BUCKETS_MS: [u64; 8] = [10, 50, 100, 250, 500, 1_000, 2_000, 5_000];
pub const METADATA_FETCH_BUCKETS_MS: [u64; 7] = [250, 500, 1_000, 2_000, 4_000, 6_000, 10_000];
pub const METADATA_SIZE_BUCKETS_BYTES: [u64; 8] = [
16 * 1024,
32 * 1024,
64 * 1024,
128 * 1024,
256 * 1024,
512 * 1024,
1024 * 1024,
10 * 1024 * 1024,
];
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FixedHistogramSnapshot {
pub bounds: Vec<u64>,
pub counts: Vec<u64>,
pub overflow: u64,
pub count: u64,
pub sum: u64,
}
impl FixedHistogramSnapshot {
pub fn percentile(&self, percentile: f64) -> Option<u64> {
if self.count == 0 {
return None;
}
let rank = ((self.count as f64 * percentile.clamp(0.0, 1.0)).ceil() as u64).max(1);
let mut cumulative = 0u64;
for (bound, count) in self.bounds.iter().zip(&self.counts) {
cumulative = cumulative.saturating_add(*count);
if cumulative >= rank {
return Some(*bound);
}
}
self.bounds.last().copied()
}
}
struct AtomicFixedHistogram<const N: usize> {
bounds: [u64; N],
counts: [AtomicU64; N],
overflow: AtomicU64,
count: AtomicU64,
sum: AtomicU64,
}
impl<const N: usize> AtomicFixedHistogram<N> {
fn new(bounds: [u64; N]) -> Self {
Self {
bounds,
counts: std::array::from_fn(|_| AtomicU64::new(0)),
overflow: AtomicU64::new(0),
count: AtomicU64::new(0),
sum: AtomicU64::new(0),
}
}
fn record(&self, value: u64) {
if let Some(index) = self.bounds.iter().position(|bound| value <= *bound) {
self.counts[index].fetch_add(1, Ordering::Relaxed);
} else {
self.overflow.fetch_add(1, Ordering::Relaxed);
}
self.count.fetch_add(1, Ordering::Relaxed);
self.sum.fetch_add(value, Ordering::Relaxed);
}
fn snapshot(&self) -> FixedHistogramSnapshot {
FixedHistogramSnapshot {
bounds: self.bounds.to_vec(),
counts: self
.counts
.iter()
.map(|count| count.load(Ordering::Relaxed))
.collect(),
overflow: self.overflow.load(Ordering::Relaxed),
count: self.count.load(Ordering::Relaxed),
sum: self.sum.load(Ordering::Relaxed),
}
}
}
impl<const N: usize> Default for AtomicFixedHistogram<N> {
fn default() -> Self {
Self::new([0; N])
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DhtObservabilitySnapshot {
pub udp_rx_packets: u64,
pub udp_rx_bytes: u64,
pub udp_tx_packets: u64,
pub udp_tx_bytes: u64,
pub inbound_ping: u64,
pub inbound_find_node: u64,
pub inbound_get_peers: u64,
pub inbound_announce_peer: u64,
pub inbound_other: u64,
pub response_normal: u64,
pub response_fallback: u64,
pub response_rate_limited: u64,
pub response_send_failed: u64,
pub announce_accepted: u64,
pub announce_invalid_token: u64,
pub announce_filtered: u64,
pub node_admitted: u64,
pub node_replaced: u64,
pub node_dropped_duplicate: u64,
pub node_dropped_rate_limited: u64,
pub node_dropped_invalid: u64,
pub metadata_bytes_downloaded: u64,
pub metadata_failure_timeout: u64,
pub metadata_failure_connect: u64,
pub metadata_failure_no_extension: u64,
pub metadata_failure_send: u64,
pub metadata_failure_size_limit: u64,
pub metadata_failure_sha1: u64,
pub metadata_failure_parse: u64,
pub metadata_failure_other: u64,
pub peer_cache_timeout_hits: u64,
pub peer_cache_connect_hits: u64,
pub queue_wait_ms: FixedHistogramSnapshot,
pub fetch_duration_ms: FixedHistogramSnapshot,
pub metadata_size_bytes: FixedHistogramSnapshot,
}
#[derive(Clone, Default)]
pub struct DhtRuntimeStats {
inner: Arc<DhtRuntimeStatsInner>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DhtRuntimeSnapshot {
pub hashes_received: u64,
pub hash_ingress_dropped: u64,
pub hash_ingress_queue_depth: usize,
pub hash_ingress_queue_capacity: usize,
pub crawl_priority_queue_depth: usize,
pub crawl_priority_queue_capacity: usize,
pub crawl_discovery_queue_depth: usize,
pub crawl_discovery_queue_capacity: usize,
pub metadata_queue_depth: usize,
pub metadata_queue_max: usize,
pub metadata_in_flight: usize,
pub metadata_queue_inserted: u64,
pub metadata_queue_deduplicated: u64,
pub metadata_queue_evicted: u64,
pub metadata_queue_stale: u64,
pub metadata_races_started: u64,
pub metadata_peer_candidates: u64,
pub metadata_live_peer_joins: u64,
pub metadata_peer_canceled: u64,
pub peer_lookup_requested: u64,
pub peer_lookup_started: u64,
pub peer_lookup_rate_limited: u64,
pub peer_lookup_empty: u64,
pub peer_lookup_queries: u64,
pub peer_lookup_responses: u64,
pub peer_lookup_timeouts: u64,
pub peer_lookup_send_failures: u64,
pub peer_lookup_peers_found: u64,
pub peer_lookup_response_dropped: u64,
pub peer_lookup_output_dropped: u64,
pub metadata_peer_attempts: u64,
pub metadata_peer_succeeded: u64,
pub metadata_peer_failed: u64,
pub metadata_peer_timeouts: u64,
pub metadata_connect_failed: u64,
pub metadata_no_extension: u64,
pub metadata_peer_failure_cache_hits: u64,
pub metadata_peer_failure_cache_entries: usize,
pub node_pool_size: usize,
pub node_pool_capacity: usize,
pub node_pool_low_watermark: usize,
pub find_node_in_flight: usize,
pub find_node_in_flight_max: usize,
pub find_node_effective_rate_per_sec: u32,
pub queries_new: u64,
pub queries_revisit: u64,
pub queries_bootstrap: u64,
pub responses: u64,
pub unmatched_responses: u64,
pub timeouts: u64,
pub send_failures: u64,
pub crawl_events_dropped_response: u64,
pub crawl_events_dropped_discovered: u64,
pub udp_received: u64,
pub udp_queue_full: u64,
pub udp_invalid: u64,
pub udp_responses_rate_limited: u64,
pub udp_responses_priority_reserved: u64,
}
struct DhtRuntimeStatsInner {
hashes_received: AtomicU64,
hash_ingress_dropped: AtomicU64,
hash_ingress_queue_depth: AtomicUsize,
hash_ingress_queue_capacity: usize,
crawl_priority_queue_depth: AtomicUsize,
crawl_priority_queue_capacity: usize,
crawl_discovery_queue_depth: AtomicUsize,
crawl_discovery_queue_capacity: usize,
metadata_queue_depth: AtomicUsize,
metadata_queue_max: usize,
metadata_in_flight: AtomicUsize,
metadata_queue_inserted: AtomicU64,
metadata_queue_deduplicated: AtomicU64,
metadata_queue_evicted: AtomicU64,
metadata_queue_stale: AtomicU64,
metadata_races_started: AtomicU64,
metadata_peer_candidates: AtomicU64,
metadata_live_peer_joins: AtomicU64,
metadata_peer_canceled: AtomicU64,
peer_lookup_requested: AtomicU64,
peer_lookup_started: AtomicU64,
peer_lookup_rate_limited: AtomicU64,
peer_lookup_empty: AtomicU64,
peer_lookup_queries: AtomicU64,
peer_lookup_responses: AtomicU64,
peer_lookup_timeouts: AtomicU64,
peer_lookup_send_failures: AtomicU64,
peer_lookup_peers_found: AtomicU64,
peer_lookup_response_dropped: AtomicU64,
peer_lookup_output_dropped: AtomicU64,
metadata_peer_attempts: AtomicU64,
metadata_peer_succeeded: AtomicU64,
metadata_peer_failed: AtomicU64,
metadata_peer_timeouts: AtomicU64,
metadata_connect_failed: AtomicU64,
metadata_no_extension: AtomicU64,
metadata_peer_failure_cache_hits: AtomicU64,
metadata_peer_failure_cache_entries: AtomicUsize,
node_pool_size: AtomicUsize,
node_pool_capacity: usize,
node_pool_low_watermark: usize,
find_node_in_flight: AtomicUsize,
find_node_in_flight_max: usize,
find_node_effective_rate_per_sec: AtomicU32,
queries_new: AtomicU64,
queries_revisit: AtomicU64,
queries_bootstrap: AtomicU64,
responses: AtomicU64,
unmatched_responses: AtomicU64,
timeouts: AtomicU64,
send_failures: AtomicU64,
crawl_events_dropped_response: AtomicU64,
crawl_events_dropped_discovered: AtomicU64,
udp_received: AtomicU64,
udp_queue_full: AtomicU64,
udp_invalid: AtomicU64,
udp_responses_rate_limited: AtomicU64,
udp_responses_priority_reserved: AtomicU64,
udp_rx_bytes: AtomicU64,
udp_tx_packets: AtomicU64,
udp_tx_bytes: AtomicU64,
inbound_ping: AtomicU64,
inbound_find_node: AtomicU64,
inbound_get_peers: AtomicU64,
inbound_announce_peer: AtomicU64,
inbound_other: AtomicU64,
response_normal: AtomicU64,
response_send_failed: AtomicU64,
announce_accepted: AtomicU64,
announce_invalid_token: AtomicU64,
announce_filtered: AtomicU64,
node_admitted: AtomicU64,
node_replaced: AtomicU64,
node_dropped_duplicate: AtomicU64,
node_dropped_rate_limited: AtomicU64,
node_dropped_invalid: AtomicU64,
metadata_bytes_downloaded: AtomicU64,
metadata_failure_timeout: AtomicU64,
metadata_failure_connect: AtomicU64,
metadata_failure_no_extension: AtomicU64,
metadata_failure_send: AtomicU64,
metadata_failure_size_limit: AtomicU64,
metadata_failure_sha1: AtomicU64,
metadata_failure_parse: AtomicU64,
metadata_failure_other: AtomicU64,
peer_cache_timeout_hits: AtomicU64,
peer_cache_connect_hits: AtomicU64,
queue_wait_ms: AtomicFixedHistogram<8>,
fetch_duration_ms: AtomicFixedHistogram<7>,
metadata_size_bytes: AtomicFixedHistogram<8>,
}
impl Default for DhtRuntimeStatsInner {
fn default() -> Self {
Self {
queue_wait_ms: AtomicFixedHistogram::new(METADATA_QUEUE_WAIT_BUCKETS_MS),
fetch_duration_ms: AtomicFixedHistogram::new(METADATA_FETCH_BUCKETS_MS),
metadata_size_bytes: AtomicFixedHistogram::new(METADATA_SIZE_BUCKETS_BYTES),
hashes_received: AtomicU64::new(0),
hash_ingress_dropped: AtomicU64::new(0),
hash_ingress_queue_depth: AtomicUsize::new(0),
hash_ingress_queue_capacity: 0,
crawl_priority_queue_depth: AtomicUsize::new(0),
crawl_priority_queue_capacity: 0,
crawl_discovery_queue_depth: AtomicUsize::new(0),
crawl_discovery_queue_capacity: 0,
metadata_queue_depth: AtomicUsize::new(0),
metadata_queue_max: 0,
metadata_in_flight: AtomicUsize::new(0),
metadata_queue_inserted: AtomicU64::new(0),
metadata_queue_deduplicated: AtomicU64::new(0),
metadata_queue_evicted: AtomicU64::new(0),
metadata_queue_stale: AtomicU64::new(0),
metadata_races_started: AtomicU64::new(0),
metadata_peer_candidates: AtomicU64::new(0),
metadata_live_peer_joins: AtomicU64::new(0),
metadata_peer_canceled: AtomicU64::new(0),
peer_lookup_requested: AtomicU64::new(0),
peer_lookup_started: AtomicU64::new(0),
peer_lookup_rate_limited: AtomicU64::new(0),
peer_lookup_empty: AtomicU64::new(0),
peer_lookup_queries: AtomicU64::new(0),
peer_lookup_responses: AtomicU64::new(0),
peer_lookup_timeouts: AtomicU64::new(0),
peer_lookup_send_failures: AtomicU64::new(0),
peer_lookup_peers_found: AtomicU64::new(0),
peer_lookup_response_dropped: AtomicU64::new(0),
peer_lookup_output_dropped: AtomicU64::new(0),
metadata_peer_attempts: AtomicU64::new(0),
metadata_peer_succeeded: AtomicU64::new(0),
metadata_peer_failed: AtomicU64::new(0),
metadata_peer_timeouts: AtomicU64::new(0),
metadata_connect_failed: AtomicU64::new(0),
metadata_no_extension: AtomicU64::new(0),
metadata_peer_failure_cache_hits: AtomicU64::new(0),
metadata_peer_failure_cache_entries: AtomicUsize::new(0),
node_pool_size: AtomicUsize::new(0),
node_pool_capacity: 0,
node_pool_low_watermark: 0,
find_node_in_flight: AtomicUsize::new(0),
find_node_in_flight_max: 0,
find_node_effective_rate_per_sec: AtomicU32::new(0),
queries_new: AtomicU64::new(0),
queries_revisit: AtomicU64::new(0),
queries_bootstrap: AtomicU64::new(0),
responses: AtomicU64::new(0),
unmatched_responses: AtomicU64::new(0),
timeouts: AtomicU64::new(0),
send_failures: AtomicU64::new(0),
crawl_events_dropped_response: AtomicU64::new(0),
crawl_events_dropped_discovered: AtomicU64::new(0),
udp_received: AtomicU64::new(0),
udp_queue_full: AtomicU64::new(0),
udp_invalid: AtomicU64::new(0),
udp_responses_rate_limited: AtomicU64::new(0),
udp_responses_priority_reserved: AtomicU64::new(0),
udp_rx_bytes: AtomicU64::new(0),
udp_tx_packets: AtomicU64::new(0),
udp_tx_bytes: AtomicU64::new(0),
inbound_ping: AtomicU64::new(0),
inbound_find_node: AtomicU64::new(0),
inbound_get_peers: AtomicU64::new(0),
inbound_announce_peer: AtomicU64::new(0),
inbound_other: AtomicU64::new(0),
response_normal: AtomicU64::new(0),
response_send_failed: AtomicU64::new(0),
announce_accepted: AtomicU64::new(0),
announce_invalid_token: AtomicU64::new(0),
announce_filtered: AtomicU64::new(0),
node_admitted: AtomicU64::new(0),
node_replaced: AtomicU64::new(0),
node_dropped_duplicate: AtomicU64::new(0),
node_dropped_rate_limited: AtomicU64::new(0),
node_dropped_invalid: AtomicU64::new(0),
metadata_bytes_downloaded: AtomicU64::new(0),
metadata_failure_timeout: AtomicU64::new(0),
metadata_failure_connect: AtomicU64::new(0),
metadata_failure_no_extension: AtomicU64::new(0),
metadata_failure_send: AtomicU64::new(0),
metadata_failure_size_limit: AtomicU64::new(0),
metadata_failure_sha1: AtomicU64::new(0),
metadata_failure_parse: AtomicU64::new(0),
metadata_failure_other: AtomicU64::new(0),
peer_cache_timeout_hits: AtomicU64::new(0),
peer_cache_connect_hits: AtomicU64::new(0),
}
}
}
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct DhtRuntimeLimits {
pub metadata_queue: usize,
pub node_pool: usize,
pub node_pool_low_watermark: usize,
pub find_node_in_flight: usize,
pub initial_find_node_rate: u32,
pub hash_ingress_queue: usize,
pub crawl_priority_queue: usize,
pub crawl_discovery_queue: usize,
}
impl DhtRuntimeStats {
pub(crate) fn with_limits(limits: DhtRuntimeLimits) -> Self {
Self {
inner: Arc::new(DhtRuntimeStatsInner {
metadata_queue_max: limits.metadata_queue,
node_pool_capacity: limits.node_pool,
node_pool_low_watermark: limits.node_pool_low_watermark,
find_node_in_flight_max: limits.find_node_in_flight,
find_node_effective_rate_per_sec: AtomicU32::new(limits.initial_find_node_rate),
hash_ingress_queue_capacity: limits.hash_ingress_queue,
crawl_priority_queue_capacity: limits.crawl_priority_queue,
crawl_discovery_queue_capacity: limits.crawl_discovery_queue,
..DhtRuntimeStatsInner::default()
}),
}
}
pub fn snapshot(&self) -> DhtRuntimeSnapshot {
let inner = &self.inner;
DhtRuntimeSnapshot {
hashes_received: inner.hashes_received.load(Ordering::Relaxed),
hash_ingress_dropped: inner.hash_ingress_dropped.load(Ordering::Relaxed),
hash_ingress_queue_depth: inner.hash_ingress_queue_depth.load(Ordering::Relaxed),
hash_ingress_queue_capacity: inner.hash_ingress_queue_capacity,
crawl_priority_queue_depth: inner.crawl_priority_queue_depth.load(Ordering::Relaxed),
crawl_priority_queue_capacity: inner.crawl_priority_queue_capacity,
crawl_discovery_queue_depth: inner.crawl_discovery_queue_depth.load(Ordering::Relaxed),
crawl_discovery_queue_capacity: inner.crawl_discovery_queue_capacity,
metadata_queue_depth: inner.metadata_queue_depth.load(Ordering::Relaxed),
metadata_queue_max: inner.metadata_queue_max,
metadata_in_flight: inner.metadata_in_flight.load(Ordering::Relaxed),
metadata_queue_inserted: inner.metadata_queue_inserted.load(Ordering::Relaxed),
metadata_queue_deduplicated: inner.metadata_queue_deduplicated.load(Ordering::Relaxed),
metadata_queue_evicted: inner.metadata_queue_evicted.load(Ordering::Relaxed),
metadata_queue_stale: inner.metadata_queue_stale.load(Ordering::Relaxed),
metadata_races_started: inner.metadata_races_started.load(Ordering::Relaxed),
metadata_peer_candidates: inner.metadata_peer_candidates.load(Ordering::Relaxed),
metadata_live_peer_joins: inner.metadata_live_peer_joins.load(Ordering::Relaxed),
metadata_peer_canceled: inner.metadata_peer_canceled.load(Ordering::Relaxed),
peer_lookup_requested: inner.peer_lookup_requested.load(Ordering::Relaxed),
peer_lookup_started: inner.peer_lookup_started.load(Ordering::Relaxed),
peer_lookup_rate_limited: inner.peer_lookup_rate_limited.load(Ordering::Relaxed),
peer_lookup_empty: inner.peer_lookup_empty.load(Ordering::Relaxed),
peer_lookup_queries: inner.peer_lookup_queries.load(Ordering::Relaxed),
peer_lookup_responses: inner.peer_lookup_responses.load(Ordering::Relaxed),
peer_lookup_timeouts: inner.peer_lookup_timeouts.load(Ordering::Relaxed),
peer_lookup_send_failures: inner.peer_lookup_send_failures.load(Ordering::Relaxed),
peer_lookup_peers_found: inner.peer_lookup_peers_found.load(Ordering::Relaxed),
peer_lookup_response_dropped: inner
.peer_lookup_response_dropped
.load(Ordering::Relaxed),
peer_lookup_output_dropped: inner.peer_lookup_output_dropped.load(Ordering::Relaxed),
metadata_peer_attempts: inner.metadata_peer_attempts.load(Ordering::Relaxed),
metadata_peer_succeeded: inner.metadata_peer_succeeded.load(Ordering::Relaxed),
metadata_peer_failed: inner.metadata_peer_failed.load(Ordering::Relaxed),
metadata_peer_timeouts: inner.metadata_peer_timeouts.load(Ordering::Relaxed),
metadata_connect_failed: inner.metadata_connect_failed.load(Ordering::Relaxed),
metadata_no_extension: inner.metadata_no_extension.load(Ordering::Relaxed),
metadata_peer_failure_cache_hits: inner
.metadata_peer_failure_cache_hits
.load(Ordering::Relaxed),
metadata_peer_failure_cache_entries: inner
.metadata_peer_failure_cache_entries
.load(Ordering::Relaxed),
node_pool_size: inner.node_pool_size.load(Ordering::Relaxed),
node_pool_capacity: inner.node_pool_capacity,
node_pool_low_watermark: inner.node_pool_low_watermark,
find_node_in_flight: inner.find_node_in_flight.load(Ordering::Relaxed),
find_node_in_flight_max: inner.find_node_in_flight_max,
find_node_effective_rate_per_sec: inner
.find_node_effective_rate_per_sec
.load(Ordering::Relaxed),
queries_new: inner.queries_new.load(Ordering::Relaxed),
queries_revisit: inner.queries_revisit.load(Ordering::Relaxed),
queries_bootstrap: inner.queries_bootstrap.load(Ordering::Relaxed),
responses: inner.responses.load(Ordering::Relaxed),
unmatched_responses: inner.unmatched_responses.load(Ordering::Relaxed),
timeouts: inner.timeouts.load(Ordering::Relaxed),
send_failures: inner.send_failures.load(Ordering::Relaxed),
crawl_events_dropped_response: inner
.crawl_events_dropped_response
.load(Ordering::Relaxed),
crawl_events_dropped_discovered: inner
.crawl_events_dropped_discovered
.load(Ordering::Relaxed),
udp_received: inner.udp_received.load(Ordering::Relaxed),
udp_queue_full: inner.udp_queue_full.load(Ordering::Relaxed),
udp_invalid: inner.udp_invalid.load(Ordering::Relaxed),
udp_responses_rate_limited: inner.udp_responses_rate_limited.load(Ordering::Relaxed),
udp_responses_priority_reserved: inner
.udp_responses_priority_reserved
.load(Ordering::Relaxed),
}
}
pub fn observability_snapshot(&self) -> DhtObservabilitySnapshot {
let inner = &self.inner;
DhtObservabilitySnapshot {
udp_rx_packets: inner.udp_received.load(Ordering::Relaxed),
udp_rx_bytes: inner.udp_rx_bytes.load(Ordering::Relaxed),
udp_tx_packets: inner.udp_tx_packets.load(Ordering::Relaxed),
udp_tx_bytes: inner.udp_tx_bytes.load(Ordering::Relaxed),
inbound_ping: inner.inbound_ping.load(Ordering::Relaxed),
inbound_find_node: inner.inbound_find_node.load(Ordering::Relaxed),
inbound_get_peers: inner.inbound_get_peers.load(Ordering::Relaxed),
inbound_announce_peer: inner.inbound_announce_peer.load(Ordering::Relaxed),
inbound_other: inner.inbound_other.load(Ordering::Relaxed),
response_normal: inner.response_normal.load(Ordering::Relaxed),
response_fallback: inner
.udp_responses_priority_reserved
.load(Ordering::Relaxed),
response_rate_limited: inner.udp_responses_rate_limited.load(Ordering::Relaxed),
response_send_failed: inner.response_send_failed.load(Ordering::Relaxed),
announce_accepted: inner.announce_accepted.load(Ordering::Relaxed),
announce_invalid_token: inner.announce_invalid_token.load(Ordering::Relaxed),
announce_filtered: inner.announce_filtered.load(Ordering::Relaxed),
node_admitted: inner.node_admitted.load(Ordering::Relaxed),
node_replaced: inner.node_replaced.load(Ordering::Relaxed),
node_dropped_duplicate: inner.node_dropped_duplicate.load(Ordering::Relaxed),
node_dropped_rate_limited: inner.node_dropped_rate_limited.load(Ordering::Relaxed),
node_dropped_invalid: inner.node_dropped_invalid.load(Ordering::Relaxed),
metadata_bytes_downloaded: inner.metadata_bytes_downloaded.load(Ordering::Relaxed),
metadata_failure_timeout: inner.metadata_failure_timeout.load(Ordering::Relaxed),
metadata_failure_connect: inner.metadata_failure_connect.load(Ordering::Relaxed),
metadata_failure_no_extension: inner
.metadata_failure_no_extension
.load(Ordering::Relaxed),
metadata_failure_send: inner.metadata_failure_send.load(Ordering::Relaxed),
metadata_failure_size_limit: inner.metadata_failure_size_limit.load(Ordering::Relaxed),
metadata_failure_sha1: inner.metadata_failure_sha1.load(Ordering::Relaxed),
metadata_failure_parse: inner.metadata_failure_parse.load(Ordering::Relaxed),
metadata_failure_other: inner.metadata_failure_other.load(Ordering::Relaxed),
peer_cache_timeout_hits: inner.peer_cache_timeout_hits.load(Ordering::Relaxed),
peer_cache_connect_hits: inner.peer_cache_connect_hits.load(Ordering::Relaxed),
queue_wait_ms: inner.queue_wait_ms.snapshot(),
fetch_duration_ms: inner.fetch_duration_ms.snapshot(),
metadata_size_bytes: inner.metadata_size_bytes.snapshot(),
}
}
pub(crate) fn hash_received(&self) {
self.inner.hashes_received.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn hash_ingress_dropped(&self) {
self.inner
.hash_ingress_dropped
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn set_hash_ingress_queue_depth(&self, depth: usize) {
self.inner
.hash_ingress_queue_depth
.store(depth, Ordering::Relaxed);
}
pub(crate) fn set_crawl_priority_queue_depth(&self, depth: usize) {
self.inner
.crawl_priority_queue_depth
.store(depth, Ordering::Relaxed);
}
pub(crate) fn set_crawl_discovery_queue_depth(&self, depth: usize) {
self.inner
.crawl_discovery_queue_depth
.store(depth, Ordering::Relaxed);
}
pub(crate) fn set_metadata_queue(&self, depth: usize, in_flight: usize) {
self.inner
.metadata_queue_depth
.store(depth, Ordering::Relaxed);
self.inner
.metadata_in_flight
.store(in_flight, Ordering::Relaxed);
}
pub(crate) fn metadata_queue_deduplicated(&self) {
self.inner
.metadata_queue_deduplicated
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_queue_inserted(&self) {
self.inner
.metadata_queue_inserted
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_queue_evicted(&self) {
self.inner
.metadata_queue_evicted
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_queue_stale(&self, count: usize) {
self.inner
.metadata_queue_stale
.fetch_add(count as u64, Ordering::Relaxed);
}
pub(crate) fn metadata_race_started(&self) {
self.inner
.metadata_races_started
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_candidate(&self) {
self.inner
.metadata_peer_candidates
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_live_peer_join(&self) {
self.inner
.metadata_live_peer_joins
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_canceled(&self, count: usize) {
self.inner
.metadata_peer_canceled
.fetch_add(count as u64, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_requested(&self) {
self.inner
.peer_lookup_requested
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_started(&self) {
self.inner
.peer_lookup_started
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_rate_limited(&self) {
self.inner
.peer_lookup_rate_limited
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_empty(&self) {
self.inner.peer_lookup_empty.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_query(&self) {
self.inner
.peer_lookup_queries
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_response(&self) {
self.inner
.peer_lookup_responses
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_timeout(&self) {
self.inner
.peer_lookup_timeouts
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_send_failed(&self) {
self.inner
.peer_lookup_send_failures
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_peer_found(&self) {
self.inner
.peer_lookup_peers_found
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_response_dropped(&self) {
self.inner
.peer_lookup_response_dropped
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_lookup_output_dropped(&self) {
self.inner
.peer_lookup_output_dropped
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_attempt(&self) {
self.inner
.metadata_peer_attempts
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_succeeded(&self) {
self.inner
.metadata_peer_succeeded
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_failed(&self) {
self.inner
.metadata_peer_failed
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_timeout(&self) {
self.inner
.metadata_peer_timeouts
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_connect_failed(&self) {
self.inner
.metadata_connect_failed
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_no_extension(&self) {
self.inner
.metadata_no_extension
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_peer_failure_cache_hit(&self) {
self.inner
.metadata_peer_failure_cache_hits
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn set_metadata_peer_failure_cache_entries(&self, count: usize) {
self.inner
.metadata_peer_failure_cache_entries
.store(count, Ordering::Relaxed);
}
pub(crate) fn set_node_pool_size(&self, size: usize) {
self.inner.node_pool_size.store(size, Ordering::Relaxed);
}
pub(crate) fn set_find_node_in_flight(&self, count: usize) {
self.inner
.find_node_in_flight
.store(count, Ordering::Relaxed);
}
pub(crate) fn set_find_node_effective_rate(&self, rate: u32) {
self.inner
.find_node_effective_rate_per_sec
.store(rate, Ordering::Relaxed);
}
pub(crate) fn query_new(&self) {
self.inner.queries_new.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn query_revisit(&self) {
self.inner.queries_revisit.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn query_bootstrap(&self) {
self.inner.queries_bootstrap.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn response(&self) {
self.inner.responses.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn unmatched_response(&self) {
self.inner
.unmatched_responses
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn timeout(&self) {
self.inner.timeouts.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn send_failure(&self) {
self.inner.send_failures.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn crawl_event_dropped_response(&self) {
self.inner
.crawl_events_dropped_response
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn crawl_event_dropped_discovered(&self) {
self.inner
.crawl_events_dropped_discovered
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn udp_received(&self) {
self.inner.udp_received.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn udp_received_bytes(&self, bytes: usize) {
self.inner
.udp_rx_bytes
.fetch_add(bytes as u64, Ordering::Relaxed);
}
pub(crate) fn udp_sent(&self, bytes: usize) {
self.inner.udp_tx_packets.fetch_add(1, Ordering::Relaxed);
self.inner
.udp_tx_bytes
.fetch_add(bytes as u64, Ordering::Relaxed);
}
pub(crate) fn inbound_query(&self, query: &str) {
let counter = match query {
"ping" => &self.inner.inbound_ping,
"find_node" => &self.inner.inbound_find_node,
"get_peers" => &self.inner.inbound_get_peers,
"announce_peer" => &self.inner.inbound_announce_peer,
_ => &self.inner.inbound_other,
};
counter.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn response_normal(&self) {
self.inner.response_normal.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn response_send_failed(&self) {
self.inner
.response_send_failed
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn announce_accepted(&self) {
self.inner.announce_accepted.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn announce_invalid_token(&self) {
self.inner
.announce_invalid_token
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn announce_filtered(&self) {
self.inner.announce_filtered.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn node_admitted(&self) {
self.inner.node_admitted.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn node_replaced(&self) {
self.inner.node_replaced.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn node_dropped_duplicate(&self) {
self.inner
.node_dropped_duplicate
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn node_dropped_rate_limited(&self) {
self.inner
.node_dropped_rate_limited
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn node_dropped_invalid(&self) {
self.inner
.node_dropped_invalid
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_bytes_downloaded(&self, bytes: usize) {
self.inner
.metadata_bytes_downloaded
.fetch_add(bytes as u64, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_timeout(&self) {
self.inner
.metadata_failure_timeout
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_connect(&self) {
self.inner
.metadata_failure_connect
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_no_extension(&self) {
self.inner
.metadata_failure_no_extension
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_send(&self) {
self.inner
.metadata_failure_send
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_size_limit(&self) {
self.inner
.metadata_failure_size_limit
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_sha1(&self) {
self.inner
.metadata_failure_sha1
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_parse(&self) {
self.inner
.metadata_failure_parse
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn metadata_failure_other(&self) {
self.inner
.metadata_failure_other
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_cache_hit_timeout(&self) {
self.inner
.peer_cache_timeout_hits
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn peer_cache_hit_connect(&self) {
self.inner
.peer_cache_connect_hits
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn observe_metadata_queue_wait(&self, millis: u64) {
self.inner.queue_wait_ms.record(millis);
}
pub(crate) fn observe_metadata_fetch_duration(&self, millis: u64) {
self.inner.fetch_duration_ms.record(millis);
}
pub(crate) fn observe_metadata_size(&self, bytes: usize) {
self.inner.metadata_size_bytes.record(bytes as u64);
}
pub(crate) fn udp_queue_full(&self) {
self.inner.udp_queue_full.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn udp_invalid(&self) {
self.inner.udp_invalid.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn udp_response_rate_limited(&self) {
self.inner
.udp_responses_rate_limited
.fetch_add(1, Ordering::Relaxed);
}
pub(crate) fn udp_response_priority_reserved(&self) {
self.inner
.udp_responses_priority_reserved
.fetch_add(1, Ordering::Relaxed);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn cloned_handle_updates_one_snapshot() {
let stats = DhtRuntimeStats::with_limits(DhtRuntimeLimits {
metadata_queue: 100,
node_pool: 1_000,
node_pool_low_watermark: 10,
find_node_in_flight: 512,
initial_find_node_rate: 200,
hash_ingress_queue: 20,
crawl_priority_queue: 30,
crawl_discovery_queue: 40,
});
let writer = stats.clone();
writer.hash_received();
writer.hash_ingress_dropped();
writer.set_hash_ingress_queue_depth(11);
writer.set_crawl_priority_queue_depth(12);
writer.set_crawl_discovery_queue_depth(13);
writer.set_metadata_queue(42, 7);
writer.metadata_queue_inserted();
writer.metadata_queue_deduplicated();
writer.metadata_queue_evicted();
writer.metadata_queue_stale(3);
writer.metadata_race_started();
writer.metadata_peer_candidate();
writer.metadata_live_peer_join();
writer.metadata_peer_canceled(2);
writer.peer_lookup_requested();
writer.peer_lookup_started();
writer.peer_lookup_rate_limited();
writer.peer_lookup_empty();
writer.peer_lookup_query();
writer.peer_lookup_response();
writer.peer_lookup_timeout();
writer.peer_lookup_send_failed();
writer.peer_lookup_peer_found();
writer.peer_lookup_response_dropped();
writer.peer_lookup_output_dropped();
writer.metadata_peer_attempt();
writer.metadata_peer_succeeded();
writer.metadata_peer_failed();
writer.metadata_peer_timeout();
writer.metadata_connect_failed();
writer.metadata_no_extension();
writer.metadata_peer_failure_cache_hit();
writer.set_metadata_peer_failure_cache_entries(9);
writer.set_node_pool_size(321);
writer.set_find_node_in_flight(12);
writer.set_find_node_effective_rate(150);
writer.query_new();
writer.query_revisit();
writer.query_bootstrap();
writer.response();
writer.unmatched_response();
writer.timeout();
writer.send_failure();
writer.crawl_event_dropped_response();
writer.crawl_event_dropped_discovered();
writer.udp_received();
writer.udp_queue_full();
writer.udp_invalid();
writer.udp_response_rate_limited();
writer.udp_response_priority_reserved();
assert_eq!(
stats.snapshot(),
DhtRuntimeSnapshot {
hashes_received: 1,
hash_ingress_dropped: 1,
hash_ingress_queue_depth: 11,
hash_ingress_queue_capacity: 20,
crawl_priority_queue_depth: 12,
crawl_priority_queue_capacity: 30,
crawl_discovery_queue_depth: 13,
crawl_discovery_queue_capacity: 40,
metadata_queue_depth: 42,
metadata_queue_max: 100,
metadata_in_flight: 7,
metadata_queue_inserted: 1,
metadata_queue_deduplicated: 1,
metadata_queue_evicted: 1,
metadata_queue_stale: 3,
metadata_races_started: 1,
metadata_peer_candidates: 1,
metadata_live_peer_joins: 1,
metadata_peer_canceled: 2,
peer_lookup_requested: 1,
peer_lookup_started: 1,
peer_lookup_rate_limited: 1,
peer_lookup_empty: 1,
peer_lookup_queries: 1,
peer_lookup_responses: 1,
peer_lookup_timeouts: 1,
peer_lookup_send_failures: 1,
peer_lookup_peers_found: 1,
peer_lookup_response_dropped: 1,
peer_lookup_output_dropped: 1,
metadata_peer_attempts: 1,
metadata_peer_succeeded: 1,
metadata_peer_failed: 1,
metadata_peer_timeouts: 1,
metadata_connect_failed: 1,
metadata_no_extension: 1,
metadata_peer_failure_cache_hits: 1,
metadata_peer_failure_cache_entries: 9,
node_pool_size: 321,
node_pool_capacity: 1_000,
node_pool_low_watermark: 10,
find_node_in_flight: 12,
find_node_in_flight_max: 512,
find_node_effective_rate_per_sec: 150,
queries_new: 1,
queries_revisit: 1,
queries_bootstrap: 1,
responses: 1,
unmatched_responses: 1,
timeouts: 1,
send_failures: 1,
crawl_events_dropped_response: 1,
crawl_events_dropped_discovered: 1,
udp_received: 1,
udp_queue_full: 1,
udp_invalid: 1,
udp_responses_rate_limited: 1,
udp_responses_priority_reserved: 1,
}
);
}
#[test]
fn observability_snapshot_tracks_fixed_histograms_and_categories() {
let stats = DhtRuntimeStats::default();
stats.udp_received();
stats.udp_received_bytes(128);
stats.udp_sent(64);
stats.inbound_query("ping");
stats.inbound_query("unknown");
stats.node_admitted();
stats.metadata_bytes_downloaded(1_024);
stats.metadata_failure_parse();
stats.observe_metadata_queue_wait(75);
stats.observe_metadata_queue_wait(600);
stats.observe_metadata_fetch_duration(1_500);
stats.observe_metadata_size(70_000);
let snapshot = stats.observability_snapshot();
assert_eq!(snapshot.udp_rx_packets, 1);
assert_eq!(snapshot.udp_rx_bytes, 128);
assert_eq!(snapshot.udp_tx_packets, 1);
assert_eq!(snapshot.udp_tx_bytes, 64);
assert_eq!(snapshot.inbound_ping, 1);
assert_eq!(snapshot.inbound_other, 1);
assert_eq!(snapshot.node_admitted, 1);
assert_eq!(snapshot.metadata_bytes_downloaded, 1_024);
assert_eq!(snapshot.metadata_failure_parse, 1);
assert_eq!(snapshot.queue_wait_ms.percentile(0.50), Some(100));
assert_eq!(snapshot.queue_wait_ms.percentile(0.95), Some(1_000));
assert_eq!(snapshot.fetch_duration_ms.percentile(0.95), Some(2_000));
assert_eq!(snapshot.metadata_size_bytes.percentile(0.50), Some(131_072));
}
}