nodedb-cluster 0.1.0

// SPDX-License-Identifier: BUSL-1.1

//! [`NexarTransport`] struct, constructors, and basic accessors.

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, RwLock};
use std::time::Duration;

use nodedb_types::config::tuning::ClusterTransportTuning;
use tracing::info;

use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, RetryPolicy};
use crate::error::{ClusterError, Result};
use crate::transport::auth_context::AuthContext;
use crate::transport::config;
use crate::transport::credentials::{self, TransportCredentials};
use crate::transport::server::{NoopIdentityStore, PeerIdentityStore};

/// QUIC-based Raft transport with retry and circuit breaker.
///
/// Implements [`RaftTransport`] for outbound RPCs and provides [`serve`]
/// for inbound RPC handling. Thread-safe — wrap in `Arc` for shared use.
///
/// Resilience features:
/// - **Retry**: Transient transport failures are retried with exponential backoff.
/// - **Circuit breaker**: Peers with consecutive failures are fast-failed until cooldown.
/// - **Connection eviction**: Stale connections are evicted on failure and re-established on retry.
///
/// [`RaftTransport`]: nodedb_raft::transport::RaftTransport
/// [`serve`]: Self::serve
pub struct NexarTransport {
    pub(super) node_id: u64,
    pub(super) listener: nexar::TransportListener,
    pub(super) client_config: quinn::ClientConfig,
    /// Cached connections to peers. Stale connections are replaced on next use.
    pub(super) peers: RwLock<HashMap<u64, quinn::Connection>>,
    /// Known peer addresses for connection establishment.
    pub(super) peer_addrs: RwLock<HashMap<u64, SocketAddr>>,
    pub(super) rpc_timeout: Duration,
    pub(super) circuit_breaker: Arc<CircuitBreaker>,
    pub(super) retry_policy: RetryPolicy,
    /// MAC key + per-peer sequence trackers. Shared with every spawned
    /// per-connection / per-stream task via `Arc::clone`.
    pub(super) auth: Arc<AuthContext>,
    /// SPKI pin for this node's own TLS leaf certificate.  `None` when
    /// running in insecure transport mode.  Transmitted in `JoinRequest`
    /// so remote peers can pin our identity.
    local_spki_pin: Option<[u8; 32]>,
    /// Agreed wire version per connection, keyed on `conn.stable_id()`.
    /// Populated by `perform_version_handshake_client` after the
    /// per-connection handshake completes; evicted on connection drop.
    ///
    /// Cluster-plane only: this cache is consulted by send-path callers
    /// inside the cluster transport and never crosses the SPSC bridge
    /// into the Data Plane.
    pub(super) agreed_versions: RwLock<HashMap<usize, crate::wire_version::WireVersion>>,
}

impl NexarTransport {
    /// Create a new transport bound to the given address.
    ///
    /// `creds` selects channel-level authentication — see
    /// [`TransportCredentials`]. Uses `ClusterTransportTuning::default()`
    /// for all QUIC and RPC settings. Prefer [`NexarTransport::with_tuning`]
    /// in production to read values from the server's `TuningConfig`.
    ///
    /// Uses [`NoopIdentityStore`] for TLS-layer SPKI pinning, which accepts
    /// all CA-signed certs in the bootstrap window. For production clusters
    /// with an active topology, use [`NexarTransport::with_tuning_and_identity`].
    pub fn new(node_id: u64, listen_addr: SocketAddr, creds: TransportCredentials) -> Result<Self> {
        Self::with_timeout(node_id, listen_addr, config::DEFAULT_RPC_TIMEOUT, creds)
    }

    /// Create a new transport with a custom RPC timeout.
    ///
    /// `creds` selects channel-level authentication. Uses
    /// `ClusterTransportTuning::default()` for all QUIC settings.
    /// Uses [`NoopIdentityStore`] for TLS-layer SPKI pinning.
    pub fn with_timeout(
        node_id: u64,
        listen_addr: SocketAddr,
        rpc_timeout: Duration,
        creds: TransportCredentials,
    ) -> Result<Self> {
        let defaults = ClusterTransportTuning::default();
        Self::build(
            node_id,
            listen_addr,
            rpc_timeout,
            &defaults,
            creds,
            Arc::new(NoopIdentityStore),
        )
    }

    /// Create a new transport using values from `ClusterTransportTuning`.
    ///
    /// All QUIC parameters (streams, windows, keep-alive, idle timeout) and
    /// the RPC timeout are read from `tuning`. `creds` selects channel-level
    /// authentication. Use this in production so that operators can override
    /// defaults via the `[tuning.cluster_transport]` section of `config.toml`.
    ///
    /// Uses [`NoopIdentityStore`] for TLS-layer SPKI pinning. For a fully
    /// pinned production cluster, use [`NexarTransport::with_tuning_and_identity`].
    pub fn with_tuning(
        node_id: u64,
        listen_addr: SocketAddr,
        tuning: &ClusterTransportTuning,
        creds: TransportCredentials,
    ) -> Result<Self> {
        let rpc_timeout = Duration::from_secs(tuning.rpc_timeout_secs);
        Self::build(
            node_id,
            listen_addr,
            rpc_timeout,
            tuning,
            creds,
            Arc::new(NoopIdentityStore),
        )
    }

    /// Create a new transport using tuning and an explicit identity store for
    /// TLS-layer SPKI/SPIFFE pinning.
    ///
    /// Pass the cluster topology (wrapped as `Arc<dyn PeerIdentityStore>`) to
    /// enable per-node cert pinning at the TLS handshake layer. The pinning is
    /// additive: the WebPki chain verifier (CA trust, expiry, CRL) still runs
    /// first, and the SPKI check fires only when a topology entry exists for
    /// the connecting peer's cert.
    pub fn with_tuning_and_identity(
        node_id: u64,
        listen_addr: SocketAddr,
        tuning: &ClusterTransportTuning,
        creds: TransportCredentials,
        identity_store: Arc<dyn PeerIdentityStore>,
    ) -> Result<Self> {
        let rpc_timeout = Duration::from_secs(tuning.rpc_timeout_secs);
        Self::build(
            node_id,
            listen_addr,
            rpc_timeout,
            tuning,
            creds,
            identity_store,
        )
    }

    /// Internal assembly shared by every constructor.
    fn build(
        node_id: u64,
        listen_addr: SocketAddr,
        rpc_timeout: Duration,
        tuning: &ClusterTransportTuning,
        creds: TransportCredentials,
        identity_store: Arc<dyn PeerIdentityStore>,
    ) -> Result<Self> {
        let (server_config, client_config) = match &creds {
            TransportCredentials::Mtls(tls) => (
                config::make_raft_server_config_mtls(tls, tuning, Arc::clone(&identity_store))?,
                config::make_raft_client_config_mtls(tls, tuning, Arc::clone(&identity_store))?,
            ),
            TransportCredentials::Insecure => {
                credentials::announce_insecure_transport(node_id);
                (
                    config::make_raft_server_config(tuning)?,
                    config::make_raft_client_config(tuning)?,
                )
            }
        };

        let local_spki_pin = match &creds {
            TransportCredentials::Mtls(tls) => Some(tls.spki_pin),
            TransportCredentials::Insecure => None,
        };

        let auth = Arc::new(AuthContext::from_credentials(node_id, &creds));

        let listener = nexar::TransportListener::bind_with_config(listen_addr, server_config)
            .map_err(|e| ClusterError::Transport {
                detail: format!("bind {listen_addr}: {e}"),
            })?;

        info!(
            node_id,
            addr = %listener.local_addr(),
            rpc_timeout_ms = rpc_timeout.as_millis() as u64,
            mtls = !creds.is_insecure(),
            "raft transport bound"
        );

        Ok(Self {
            node_id,
            listener,
            client_config,
            peers: RwLock::new(HashMap::new()),
            peer_addrs: RwLock::new(HashMap::new()),
            rpc_timeout,
            circuit_breaker: Arc::new(CircuitBreaker::new(CircuitBreakerConfig::default())),
            retry_policy: RetryPolicy::default(),
            auth,
            local_spki_pin,
            agreed_versions: RwLock::new(HashMap::new()),
        })
    }

    /// Accessor used by the serve / send paths when spawning per-connection
    /// tasks that need the shared auth state.
    pub(super) fn auth(&self) -> &Arc<AuthContext> {
        &self.auth
    }

    /// Access the circuit breaker (for observability / testing and
    /// for subsystems that need to share the same breaker instance).
    pub fn circuit_breaker(&self) -> &Arc<CircuitBreaker> {
        &self.circuit_breaker
    }

    /// The local address this transport is listening on.
    pub fn local_addr(&self) -> SocketAddr {
        self.listener.local_addr()
    }

    /// This node's ID.
    pub fn node_id(&self) -> u64 {
        self.node_id
    }

    /// The cluster MAC key carried by this transport. SWIM subsystem
    /// uses it to authenticate UDP datagrams on the same key material.
    pub fn mac_key(&self) -> crate::rpc_codec::MacKey {
        self.auth.mac_key.clone()
    }

    /// SHA-256 SPKI pin for this node's own TLS leaf certificate.
    /// `None` in insecure transport mode.
    pub fn local_spki_pin(&self) -> Option<[u8; 32]> {
        self.local_spki_pin
    }

    /// Return the negotiated [`WireVersion`] for the connection identified by
    /// `stable_id`. Returns `None` if the handshake has not yet completed for
    /// this connection (e.g. first RPC not yet sent, or connection was evicted).
    pub fn agreed_version_for(&self, stable_id: usize) -> Option<crate::wire_version::WireVersion> {
        let versions = self
            .agreed_versions
            .read()
            .unwrap_or_else(|p| p.into_inner());
        versions.get(&stable_id).copied()
    }

    /// Store the agreed wire version for a connection (called after the
    /// per-connection handshake completes on the client side).
    pub(super) fn store_agreed_version(
        &self,
        stable_id: usize,
        version: crate::wire_version::WireVersion,
    ) {
        let mut versions = self
            .agreed_versions
            .write()
            .unwrap_or_else(|p| p.into_inner());
        versions.insert(stable_id, version);
    }

    /// Remove the cached agreed version for an evicted connection.
    pub(super) fn evict_agreed_version(&self, stable_id: usize) {
        let mut versions = self
            .agreed_versions
            .write()
            .unwrap_or_else(|p| p.into_inner());
        versions.remove(&stable_id);
    }

    /// Accept a raw incoming QUIC connection without the wire-version handshake
    /// or any RPC dispatch. Intended for tests that need to inspect or send
    /// deliberately malformed handshake frames server-side.
    #[doc(hidden)]
    pub async fn accept_raw(&self) -> crate::error::Result<quinn::Connection> {
        self.listener
            .accept()
            .await
            .map_err(|e| crate::error::ClusterError::Transport {
                detail: format!("accept_raw: {e}"),
            })
    }

    /// Establish a raw QUIC connection to `addr` without registering it as a
    /// named peer and without running the wire-version handshake. Intended for
    /// tests that need to send deliberately malformed handshake frames.
    #[doc(hidden)]
    pub async fn connect_raw(
        &self,
        addr: std::net::SocketAddr,
    ) -> crate::error::Result<quinn::Connection> {
        self.listener
            .endpoint()
            .connect_with(
                self.client_config.clone(),
                addr,
                crate::transport::config::SNI_HOSTNAME,
            )
            .map_err(|e| crate::error::ClusterError::Transport {
                detail: format!("connect_raw to {addr}: {e}"),
            })?
            .await
            .map_err(|e| crate::error::ClusterError::Transport {
                detail: format!("connect_raw handshake with {addr}: {e}"),
            })
    }

    /// Snapshot of every peer the transport has addresses cached for,
    /// with a per-peer `connected` flag for whether a nexar client is
    /// currently held in the pool. Sorted by peer id so
    /// `/cluster/debug/transport` output is deterministic.
    pub fn peer_snapshot(&self) -> Vec<TransportPeerSnapshot> {
        let addrs = self.peer_addrs.read().unwrap_or_else(|p| p.into_inner());
        let peers = self.peers.read().unwrap_or_else(|p| p.into_inner());
        let mut out: Vec<TransportPeerSnapshot> = addrs
            .iter()
            .map(|(id, addr)| TransportPeerSnapshot {
                peer_id: *id,
                addr: addr.to_string(),
                connected: peers.contains_key(id),
            })
            .collect();
        out.sort_by_key(|p| p.peer_id);
        out
    }
}

/// Per-peer view emitted by [`NexarTransport::peer_snapshot`].
#[derive(Debug, Clone, serde::Serialize)]
pub struct TransportPeerSnapshot {
    pub peer_id: u64,
    pub addr: String,
    /// True when a nexar client is currently held in the connection
    /// cache. False means either we've never connected or the client
    /// was evicted.
    pub connected: bool,
}