decdn-node 0.1.0

//! Routing-table bucket refresh (ADR 022 §Routing Table — "Bucket
//! refresh interval | 1 hour").
//!
//! Every routing-table bucket holds peers at a specific XOR-distance
//! prefix from `self.node_id`. As nodes leave the network those
//! peers' entries silently die — without refresh, a long-lived node's
//! buckets fill with stale entries and lookup convergence degrades.
//!
//! The fix per ADR 022 §Routing Table is to periodically issue a
//! `FindNode(random_id_in_bucket)` to one peer in each bucket, every
//! hour. Live peers in that distance shell respond with their own
//! K-closest, populating the bucket with fresh entries.
//!
//! Implementation: a single tokio task wakes on a 1-hour ticker and
//! refreshes *every* non-empty bucket per tick (see
//! [`BUCKET_REFRESH_TICK`] for the rate-budget argument). For each
//! non-empty bucket it synthesises a random `NodeId` whose XOR
//! distance to `self` falls in that bucket's range and runs
//! `FindNode` against the bucket's most-recently-seen peer; the
//! per-bucket RPCs fan out in parallel.
//!
//! This is a "best-effort hygiene" task: a refresh that fails (peer
//! unreachable, timeout) is silently ignored — the next tick will try
//! the next bucket. Failure has no operator-actionable signal.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use iroh::{Endpoint, EndpointAddr, PublicKey};
use rand::RngExt;
use tokio::sync::oneshot;

use crate::dht::chain_projection::with_lock;
use crate::dht::client;
use crate::dht::routing::{KEYSPACE_BITS, NodeId, RoutingTable, bucket_index};

/// Bucket-refresh tick interval. ADR 022 §Routing Table specifies "1
/// hour" as the refresh interval **per bucket** — we run *every*
/// non-empty bucket per tick so each populated bucket gets the
/// ADR-mandated hourly refresh. At realistic scale (≤30 populated
/// buckets even in a 500-node network) this is ≤30 `FindNode` RPCs
/// per hour, comfortably within the routing-table peers' own per-IP
/// rate limit (100 req/s by default).
pub const BUCKET_REFRESH_TICK: Duration = Duration::from_hours(1);

/// Long-running task: every [`BUCKET_REFRESH_TICK`] runs a
/// `FindNode(random_id_in_bucket)` against each non-empty bucket's
/// most-recently-seen peer in parallel. Exits on `stop_rx`.
///
/// `refresh_clock` is stamped with the wall-clock microseconds at which
/// each refresh pass completes, so the `admin_v1_status` health view
/// (issue #741) can report the network-wide "last refreshed" time. Since
/// a pass touches *all* non-empty buckets at once, a single timestamp is
/// the honest representation — there is no per-bucket refresh time. It
/// starts at 0 ("never refreshed") and is updated every tick regardless
/// of whether any bucket was populated, so it tracks "the refresh task is
/// alive and ran at T" rather than "a bucket received fresh entries".
///
/// Each pass also counts its failed bucket refreshes into
/// `decdn_dht_bucket_refresh_failures_total` and publishes the table's size
/// as `decdn_dht_routing_table_size`.
pub async fn run_bucket_refresh(
    endpoint: Endpoint,
    self_id: PublicKey,
    routing: Arc<Mutex<RoutingTable>>,
    mut stop_rx: oneshot::Receiver<()>,
    interval: Duration,
    refresh_clock: Arc<AtomicU64>,
    metrics: Arc<crate::metrics::Metrics>,
) {
    let self_node_id = NodeId::from_bytes(*self_id.as_bytes());
    let mut ticker = tokio::time::interval(interval);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ticker.tick().await; // burn first tick — bucket refresh isn't needed at t=0

    loop {
        tokio::select! {
            biased;
            _ = &mut stop_rx => {
                tracing::debug!("dht bucket-refresh: shutdown signal received");
                return;
            }
            _ = ticker.tick() => {
                let failures =
                    refresh_all_non_empty_buckets(&endpoint, self_node_id, &routing).await;
                for _ in 0..failures {
                    metrics.dht_bucket_refresh_failure();
                }
                metrics.dht_routing_table_size(
                    with_lock(&routing, "dht routing table", |table| table.len()),
                );
                // Stamp after the pass completes so `admin_v1_status`
                // reports the time refresh last *finished*. `Relaxed` is
                // sufficient: the admin reader only needs the latest value
                // eventually, with no ordering dependency on other state.
                refresh_clock.store(now_us(), Ordering::Relaxed);
            }
        }
    }
}

/// Wall-clock now in microseconds since `UNIX_EPOCH`. Falls back to 0 if
/// the host clock is set before the epoch (panic-free per the workspace
/// anti-panic policy), logging at `error` level on that path — same
/// treatment as [`crate::handlers::dht`]'s `now_us`, since a 0 stamp here
/// makes `admin_v1_status` report `last_refresh=never` even though a pass
/// completed, so the misconfiguration must be diagnosable from the logs.
/// (The byte-identical-but-silent copy in [`crate::dht::publish`] only
/// feeds scheduling, where the fallback has no operator-visible effect.)
fn now_us() -> u64 {
    match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
        Ok(d) => u64::try_from(d.as_micros()).unwrap_or(u64::MAX),
        Err(err) => {
            tracing::error!(
                error = %err,
                "wall-clock before UNIX_EPOCH; bucket-refresh timestamp falls back to 0, so \
                 admin_v1_status will report last_refresh=never despite refreshes running. \
                 Set the host clock."
            );
            0
        }
    }
}

/// Refresh every non-empty bucket once. Snapshots the (`bucket_index`,
/// peer, target) tuples under a single short lock acquire, then fans
/// the network requests out in parallel.
///
/// Returns how many bucket refreshes failed: a `FindNode` RPC error, a
/// routing-table peer that is not a valid public key, or a panicked refresh
/// task.
async fn refresh_all_non_empty_buckets(
    endpoint: &Endpoint,
    self_node_id: NodeId,
    routing: &Arc<Mutex<RoutingTable>>,
) -> usize {
    let picks: Vec<BucketPick> = with_lock(routing, "dht routing table", |table| {
        all_non_empty_picks(table)
    });
    if picks.is_empty() {
        return 0;
    }
    let mut handles = Vec::with_capacity(picks.len());
    for pick in picks {
        let endpoint_cloned = endpoint.clone();
        let routing_cloned = Arc::clone(routing);
        handles.push(tokio::spawn(async move {
            refresh_one_bucket(&endpoint_cloned, self_node_id, &routing_cloned, pick).await
        }));
    }
    let mut failures = 0;
    for h in handles {
        // A panicked refresh task counts as a failed refresh.
        if !h.await.unwrap_or(false) {
            failures += 1;
        }
    }
    failures
}

/// Snapshot all non-empty buckets as `BucketPick`s. Returns one pick
/// per non-empty bucket; the caller fans the corresponding `FindNode`
/// RPCs out in parallel.
fn all_non_empty_picks(table: &RoutingTable) -> Vec<BucketPick> {
    (0..KEYSPACE_BITS)
        .filter_map(|idx| {
            let peer = bucket_iter(table, idx).last().copied()?;
            let target = random_target_in_bucket(table.self_id(), idx);
            Some(BucketPick {
                bucket_index: idx,
                peer,
                target,
            })
        })
        .collect()
}

/// Run a single bucket's refresh: `FindNode(target)` against the
/// bucket's most-recently-seen peer, then insert any closer-nodes the
/// responder returns. The pick is computed upstream in
/// [`all_non_empty_picks`] so the routing-table lock is held only
/// briefly under the snapshot.
async fn refresh_one_bucket(
    endpoint: &Endpoint,
    self_node_id: NodeId,
    routing: &Arc<Mutex<RoutingTable>>,
    pick: BucketPick,
) -> bool {
    let BucketPick {
        bucket_index,
        peer,
        target,
    } = pick;
    let target_pk = match PublicKey::from_bytes(peer.as_bytes()) {
        Ok(k) => k,
        Err(e) => {
            tracing::warn!(
                peer = %peer,
                error = %e,
                "dht bucket-refresh: routing-table peer not a valid public key"
            );
            return false;
        }
    };
    let addr = EndpointAddr::new(target_pk);
    match client::find_node(endpoint, addr, target, self_node_id).await {
        Ok(resp) => {
            // Insert any new peers from the response into the routing
            // table. The bucket's freshness is updated by virtue of
            // these inserts moving entries to MRU. These are
            // response-learned ids (not the authenticated peer), inserted
            // as Kademlia probe candidates by design — distinct from the
            // `AuthenticatedNodeId`-gated recency refresh in the handler.
            with_lock(routing, "dht routing table", |table| {
                for nid in resp.closer_nodes.into_inner() {
                    if nid != self_node_id {
                        table.insert(nid);
                    }
                }
            });
            true
        }
        Err(e) => {
            tracing::debug!(
                bucket = bucket_index,
                error = %e,
                "dht bucket-refresh: FindNode failed; will retry on next tick"
            );
            false
        }
    }
}

struct BucketPick {
    bucket_index: usize,
    /// The peer we'll send the `FindNode` to — the bucket's
    /// most-recently-seen entry (highest chance of responding).
    peer: NodeId,
    /// A random `NodeId` whose XOR distance to `self_id` falls in
    /// `bucket_index`'s prefix range — used as the `target` of the
    /// `FindNode` so the responder returns peers from that distance
    /// shell.
    target: NodeId,
}

/// Build a single non-empty bucket's pick. Used by tests; the
/// production loop uses [`all_non_empty_picks`] which collects every
/// non-empty bucket per tick.
#[cfg(test)]
fn pick_bucket(table: &RoutingTable, bucket: usize) -> Option<BucketPick> {
    let peer = bucket_iter(table, bucket).last().copied()?;
    let target = random_target_in_bucket(table.self_id(), bucket);
    Some(BucketPick {
        bucket_index: bucket,
        peer,
        target,
    })
}

/// Iterate over the peers in `bucket_index` by filtering `iter_peers`
/// on the distance computation. Marginal cost compared to exposing a
/// dedicated `bucket(idx)` accessor on `RoutingTable`, but keeps the
/// `RoutingTable` API minimal.
fn bucket_iter(table: &RoutingTable, target_bucket: usize) -> impl Iterator<Item = &NodeId> {
    let self_id = *table.self_id();
    table.iter_peers().filter(move |peer| {
        let d = crate::dht::routing::xor_distance(&self_id, peer);
        bucket_index(&d) == Some(target_bucket)
    })
}

/// Produce a random `NodeId` whose XOR-distance to `self_id` falls
/// inside `bucket_index`'s prefix range. Used as the `target` of the
/// refresh `FindNode` so the responder returns peers from that
/// distance shell.
fn random_target_in_bucket(self_id: &NodeId, bucket_index: usize) -> NodeId {
    // bucket_index = KEYSPACE_BITS - 1 - bit_from_msb  →  bit_from_msb = 255 - bucket_index
    let bit_from_msb = KEYSPACE_BITS.saturating_sub(1).saturating_sub(bucket_index);
    let byte_idx = bit_from_msb / 8;
    let bit_within = bit_from_msb % 8;
    let mut target = *self_id.as_bytes();
    // Flip the bucket's prefix bit (guarantees the XOR distance has
    // its highest set bit at this position).
    let flip_mask = 1u8 << (7 - bit_within);
    if let Some(b) = target.get_mut(byte_idx) {
        *b ^= flip_mask;
    }
    // Randomise the lower-order bits so the target isn't predictable.
    let mut rng = rand::rng();
    for slot in target.iter_mut().skip(byte_idx + 1) {
        *slot = rng.random();
    }
    NodeId::from_bytes(target)
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::indexing_slicing
)]
mod tests;