dynomite/embed/

server.rs

//! Embedding-API server runtime.
//!
//! [`Server`] owns the runtime data structures (cluster pool,
//! dispatcher, stats, event bus, hooks). [`Server::start`] spawns
//! the background tasks (gossip, stats aggregator, optional TCP
//! listeners) on the current tokio runtime and returns a
//! [`ServerHandle`].
//!
//! The handle exposes the public control surface documented in
//! `docs/book/src/embedding/server.md`. It is `Clone + Send +
//! Sync`; multiple consumers may hold one. Dropping the last
//! handle does not shut the server down - only
//! [`ServerHandle::shutdown`] does.

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock, Weak};

use parking_lot::Mutex;
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{Duration, MissedTickBehavior};
use tokio_util::sync::CancellationToken;

use crate::cluster::dispatch::{ClusterDispatcher, DispatchPlan};
use crate::cluster::peer::{Peer, PeerEndpoint, PeerState};
use crate::cluster::pool::{PoolConfig, ServerPool};
use crate::conf::{ConfDynSeed, ConfPool, Config, DataStore};
use crate::embed::error::EmbedError;
use crate::embed::events::{CloseReason, ConnRoleTag, EventBus, EventStream, ServerEvent};
use crate::embed::hooks::{
    CryptoProvider, Datastore, LoggingMetricsSink, MetricsSink, SeedsProvider,
};
use crate::embed::snapshots::{DatacenterSnapshot, PeerSnapshot, RingSnapshot};
use crate::events::EventManager;
use crate::hashkit::DynToken;
use crate::io::reactor::{ConnRole, TcpTransport};
use crate::msg::Msg;
use crate::net::client::client_loop;
use crate::net::{ClientHandler, Conn};
use crate::stats::{
    describe_stats, MetricSpec, PoolField, PoolStats, ServerField, ServerStats, ServiceInfo,
    Snapshot, Stats,
};

/// Bag of optional hook overrides supplied to a builder.
///
/// All fields are private so adding new hooks (for example, a
/// `Box<dyn TransportListener>` slot) is not a SemVer-breaking
/// change. Construct a `ServerHooks` only through
/// [`crate::embed::ServerBuilder`] setters; read it back through
/// the accessor methods below.
#[non_exhaustive]
pub struct ServerHooks {
    pub(crate) datastore: Option<Box<dyn Datastore>>,
    pub(crate) seeds: Option<Box<dyn SeedsProvider>>,
    pub(crate) crypto: Option<Box<dyn CryptoProvider>>,
    pub(crate) metrics: Option<Box<dyn MetricsSink>>,
}

impl ServerHooks {
    /// Borrow the configured [`Datastore`], if any.
    #[must_use]
    pub fn datastore(&self) -> Option<&dyn Datastore> {
        self.datastore.as_deref()
    }

    /// Borrow the configured [`SeedsProvider`], if any.
    #[must_use]
    pub fn seeds(&self) -> Option<&dyn SeedsProvider> {
        self.seeds.as_deref()
    }

    /// Borrow the configured [`CryptoProvider`], if any.
    #[must_use]
    pub fn crypto(&self) -> Option<&dyn CryptoProvider> {
        self.crypto.as_deref()
    }

    /// Borrow the configured [`MetricsSink`], if any.
    #[must_use]
    pub fn metrics(&self) -> Option<&dyn MetricsSink> {
        self.metrics.as_deref()
    }
}

impl std::fmt::Debug for ServerHooks {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServerHooks")
            .field("datastore", &self.datastore.is_some())
            .field("seeds", &self.seeds.is_some())
            .field("crypto", &self.crypto.is_some())
            .field("metrics", &self.metrics.is_some())
            .finish()
    }
}

// -------- in-process registry -------------------------------------------

/// Process-wide registry for in-process clusters.
///
/// When `inject_request` produces a `Replicas` plan that targets
/// a peer co-located in the same process, the embed runtime
/// looks up the target via this registry and forwards directly
/// without the dnode wire path. This keeps the in-process tests
/// from depending on real network I/O while preserving the
/// production code path semantics (compute plan, deliver to
/// target peers).
fn registry() -> &'static Mutex<HashMap<SocketAddr, Weak<ServerInner>>> {
    static R: OnceLock<Mutex<HashMap<SocketAddr, Weak<ServerInner>>>> = OnceLock::new();
    R.get_or_init(|| Mutex::new(HashMap::new()))
}

fn registry_register(addr: SocketAddr, server: &Arc<ServerInner>) {
    registry().lock().insert(addr, Arc::downgrade(server));
}

fn registry_remove(addr: SocketAddr) {
    registry().lock().remove(&addr);
}

fn registry_lookup(addr: SocketAddr) -> Option<Arc<ServerInner>> {
    registry().lock().get(&addr).and_then(Weak::upgrade)
}

// -------- inner state ----------------------------------------------------

/// Generation counter for [`ServerEvent::ConfigReloaded`] /
/// [`RingSnapshot`].
fn next_generation() -> u64 {
    static G: AtomicU64 = AtomicU64::new(0);
    G.fetch_add(1, Ordering::Relaxed)
}

/// Connection-id allocator.
fn next_conn_id() -> u64 {
    static C: AtomicU64 = AtomicU64::new(0);
    C.fetch_add(1, Ordering::Relaxed)
}

/// Inner state shared by every clone of a [`ServerHandle`].
pub(crate) struct ServerInner {
    pool: Arc<ServerPool>,
    dispatcher: Arc<ClusterDispatcher>,
    client_proto: DataStore,
    stats: Arc<Stats>,
    snapshot_cache: Arc<Mutex<Snapshot>>,
    bus: EventBus,
    events: Arc<EventManager>,
    datastore: Arc<dyn Datastore>,
    seeds: Box<dyn SeedsProvider>,
    metrics: Box<dyn MetricsSink>,
    crypto: Option<Box<dyn CryptoProvider>>,
    listen_addr: Option<SocketAddr>,
    dyn_listen_addr: Option<SocketAddr>,
    cancel: CancellationToken,
    pool_name: String,
    config: Mutex<ConfPool>,
    generation: AtomicU64,
    command_extension: Option<Arc<dyn crate::embed::CommandExtension>>,
}

impl std::fmt::Debug for ServerInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServerInner")
            .field("pool_name", &self.pool_name)
            .field("listen", &self.listen_addr)
            .field("dyn_listen", &self.dyn_listen_addr)
            .finish_non_exhaustive()
    }
}

impl ServerInner {
    fn dispatch_local(&self, req: Msg) -> super::hooks::BoxFuture<'_, Result<Msg, EmbedError>> {
        let fut = self.datastore.dispatch(req);
        Box::pin(async move {
            let rsp = fut.await.map_err(|e| EmbedError::Inject(e.to_string()))?;
            Ok(rsp)
        })
    }
}

// -------- Server (configured-but-not-running) ----------------------------

/// Configured-but-not-running server.
///
/// Build one with [`ServerBuilder`](crate::embed::ServerBuilder),
/// then call [`Server::start`] to spawn the background tasks.
#[derive(Debug)]
pub struct Server {
    pool_name: String,
    pool: ConfPool,
    cluster: Arc<ServerPool>,
    hooks: ServerHooks,
    stats: Arc<Stats>,
    command_extension: Option<Arc<dyn crate::embed::CommandExtension>>,
}

impl Server {
    /// Used internally by [`crate::embed::ServerBuilder::build`].
    pub(crate) fn from_pool(
        pool_name: String,
        pool: ConfPool,
        hooks: ServerHooks,
        command_extension: Option<Arc<dyn crate::embed::CommandExtension>>,
    ) -> Self {
        let pool_cfg = PoolConfig::from_conf(&pool_name, &pool);
        let local_peer = build_local_peer(&pool, &pool_cfg);
        let mut peers = vec![local_peer];
        if let Some(seeds) = pool.dyn_seeds.as_ref() {
            let start = u32::try_from(peers.len()).unwrap_or(0);
            peers.extend(peers_from_seeds(&pool_cfg, seeds, start));
        }
        let server_pool_arc = Arc::new(ServerPool::new(pool_cfg.clone(), peers));
        server_pool_arc.preselect_remote_racks();

        let stats = Arc::new(Stats::new(
            ServiceInfo {
                source: pool_cfg.name.clone(),
                version: env!("CARGO_PKG_VERSION").to_string(),
                rack: pool_cfg.rack.clone(),
                dc: pool_cfg.dc.clone(),
            },
            PoolStats::new(&pool_cfg.name),
            ServerStats::new("backend"),
        ));

        Self {
            pool_name,
            pool,
            cluster: server_pool_arc,
            hooks,
            stats,
            command_extension,
        }
    }

    /// Borrow the [`crate::embed::CommandExtension`]
    /// this server was built with, if any.
    #[must_use]
    pub fn command_extension(&self) -> Option<&Arc<dyn crate::embed::CommandExtension>> {
        self.command_extension.as_ref()
    }

    /// The pool name configured in the YAML / builder.
    #[must_use]
    pub fn pool_name(&self) -> &str {
        &self.pool_name
    }

    /// Spawn background tasks on the current tokio runtime and
    /// return a [`ServerHandle`].
    ///
    /// `start` is non-blocking. The returned handle is `Clone +
    /// Send + Sync`.
    ///
    /// # Client plane and in-process traffic
    ///
    /// A client connecting to the bound `listen:` socket is
    /// served end to end: the request is parsed, routed through
    /// the cluster dispatcher, and for a request that resolves to
    /// the local node the configured [`Datastore`] hook produces
    /// the reply -- the same hook
    /// [`ServerHandle::inject_request`] uses for in-process
    /// traffic. `inject_request` remains available for callers
    /// that prefer to drive the engine without a socket hop.
    ///
    /// The `dyn_listen:` (peer-plane) socket binds for address
    /// reporting; embedded multi-node setups forward between
    /// in-process nodes through the in-process registry, so a raw
    /// cross-process peer connection to `dyn_listen:` is accepted
    /// and closed. Cross-process peer serving is provided by the
    /// `dynomited` binary.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use dynomite::embed::ServerBuilder;
    /// use dynomite::conf::DataStore;
    /// # tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap().block_on(async {
    /// let server = ServerBuilder::new("dyn_o_mite")
    ///     .listen("127.0.0.1:0".parse().unwrap())
    ///     .dyn_listen("127.0.0.1:0".parse().unwrap())
    ///     .data_store(DataStore::Valkey)
    ///     .servers(vec![dynomite::conf::ConfServer::parse("127.0.0.1:6379:1").unwrap()])
    ///     .tokens_str("0")
    ///     .enable_gossip(false)
    ///     .build()
    ///     .unwrap();
    /// let handle = server.start().await.unwrap();
    /// handle.shutdown().await.unwrap();
    /// # });
    /// ```
    pub async fn start(self) -> Result<ServerHandle, EmbedError> {
        let Server {
            pool_name,
            pool,
            cluster,
            mut hooks,
            stats,
            command_extension,
        } = self;

        let datastore: Arc<dyn Datastore> = hooks
            .datastore
            .take()
            .expect("invariant: builder always populates datastore")
            .into();

        let mut dispatcher = ClusterDispatcher::new(cluster.clone());
        if let Some(ext) = command_extension.as_ref() {
            dispatcher = dispatcher.with_command_extension(ext.clone());
        }
        // Serve client-plane requests that route to the local node
        // through the same Datastore hook inject_request uses, so a
        // connection accepted on the bound listen: socket is fully
        // served (parse -> route -> reply) rather than closed.
        dispatcher = dispatcher.with_local_datastore(Arc::clone(&datastore));
        let dispatcher = Arc::new(dispatcher);
        // Resolve the wire protocol the client proxy parses. The
        // pool stores it as the numeric data_store selector;
        // default to Valkey (RESP) when unset, matching the proxy
        // default.
        let client_proto = pool
            .data_store
            .and_then(|n| DataStore::from_int(n).ok())
            .unwrap_or(DataStore::Valkey);
        let bus = EventBus::new(64);
        let events = Arc::new(EventManager::new(64));
        let cancel = CancellationToken::new();
        let snapshot_cache = Arc::new(Mutex::new(Snapshot::default()));

        let seeds = hooks
            .seeds
            .take()
            .expect("invariant: builder always populates seeds");
        let metrics = hooks
            .metrics
            .take()
            .unwrap_or_else(|| Box::new(LoggingMetricsSink::new(pool_name.clone())));

        let transport = pool.transport.unwrap_or_default();
        let (listen_listener, listen_addr) =
            bind_listener(pool.listen.as_ref(), transport, &pool).await?;
        let (dyn_listener, dyn_listen_addr) =
            bind_listener(pool.dyn_listen.as_ref(), transport, &pool).await?;

        let inner = Arc::new(ServerInner {
            pool: cluster.clone(),
            dispatcher,
            client_proto,
            stats: stats.clone(),
            snapshot_cache: snapshot_cache.clone(),
            bus: bus.clone(),
            events: events.clone(),
            datastore,
            seeds,
            metrics,
            crypto: hooks.crypto,
            listen_addr,
            dyn_listen_addr,
            cancel: cancel.clone(),
            pool_name: pool_name.clone(),
            config: Mutex::new(pool.clone()),
            generation: AtomicU64::new(0),
            command_extension,
        });

        // Register self in the in-process registry so peer
        // forwarding can find this node.
        if let Some(addr) = inner.dyn_listen_addr {
            registry_register(addr, &inner);
        }

        // Spawn background tasks.
        let mut tasks: Vec<JoinHandle<()>> = Vec::new();
        tasks.push(tokio::spawn(stats_loop(inner.clone(), pool.stats_interval)));
        tasks.push(tokio::spawn(metrics_loop(inner.clone())));
        if pool.enable_gossip.unwrap_or(false) {
            tasks.push(tokio::spawn(gossip_loop(
                inner.clone(),
                Duration::from_millis(
                    u64::try_from(pool.gos_interval.unwrap_or(1_000)).unwrap_or(1_000),
                ),
            )));
        }
        if let (Some(listener), Some(addr)) = (listen_listener, listen_addr) {
            tasks.push(tokio::spawn(accept_loop(
                inner.clone(),
                listener,
                addr,
                ConnRoleTag::Proxy,
            )));
        }
        if let (Some(listener), Some(addr)) = (dyn_listener, dyn_listen_addr) {
            tasks.push(tokio::spawn(accept_loop(
                inner.clone(),
                listener,
                addr,
                ConnRoleTag::DnodeProxy,
            )));
        }

        Ok(ServerHandle {
            inner,
            tasks: Arc::new(Mutex::new(tasks)),
        })
    }
}

// -------- ServerHandle ---------------------------------------------------

/// Cloneable handle to a running [`Server`].
///
/// The handle is the public control surface: shutdown, reload,
/// stats, events, request injection, topology snapshots.
#[derive(Clone)]
pub struct ServerHandle {
    inner: Arc<ServerInner>,
    tasks: Arc<Mutex<Vec<JoinHandle<()>>>>,
}

impl std::fmt::Debug for ServerHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServerHandle")
            .field("pool", &self.inner.pool_name)
            .field("listen", &self.inner.listen_addr)
            .field("dyn_listen", &self.inner.dyn_listen_addr)
            .finish_non_exhaustive()
    }
}

impl ServerHandle {
    /// Borrow the configured [`CryptoProvider`], if one was
    /// plugged. The default builder does not install one.
    #[must_use]
    pub fn crypto_provider(&self) -> Option<&dyn CryptoProvider> {
        self.inner.crypto.as_deref()
    }

    /// Borrow the [`crate::embed::CommandExtension`] installed
    /// at build time, if any.
    #[must_use]
    pub fn command_extension(&self) -> Option<Arc<dyn crate::embed::CommandExtension>> {
        self.inner.command_extension.as_ref().map(Arc::clone)
    }

    /// Local listen address (post-bind), if a `listen:` was
    /// configured.
    #[must_use]
    pub fn listen_addr(&self) -> Option<SocketAddr> {
        self.inner.listen_addr
    }

    /// Local dnode listen address (post-bind), if a
    /// `dyn_listen:` was configured.
    #[must_use]
    pub fn dyn_listen_addr(&self) -> Option<SocketAddr> {
        self.inner.dyn_listen_addr
    }

    /// Subscribe to the [`ServerEvent`] broadcast.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use dynomite::embed::ServerBuilder;
    /// # use dynomite::conf::DataStore;
    /// # tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap().block_on(async {
    /// let server = ServerBuilder::new("p")
    ///     .listen("127.0.0.1:0".parse().unwrap())
    ///     .dyn_listen("127.0.0.1:0".parse().unwrap())
    ///     .data_store(DataStore::Valkey)
    ///     .servers(vec![dynomite::conf::ConfServer::parse("127.0.0.1:6379:1").unwrap()])
    ///     .tokens_str("0")
    ///     .build().unwrap();
    /// let handle = server.start().await.unwrap();
    /// let _events = handle.subscribe_events();
    /// handle.shutdown().await.unwrap();
    /// # });
    /// ```
    #[must_use]
    pub fn subscribe_events(&self) -> EventStream {
        self.inner.bus.subscribe()
    }

    /// Borrow the [`EventManager`] handle that publishes
    /// structured cluster-event payloads (`PeerUp`, `PeerDown`,
    /// `GossipRoundComplete`, etc.).
    ///
    /// Returned as `Arc` so embedders can hold a long-lived
    /// reference and pass it across crate boundaries (notably to
    /// `dyniak`, which publishes AAE start/complete events).
    #[must_use]
    pub fn events(&self) -> Arc<EventManager> {
        Arc::clone(&self.inner.events)
    }

    /// Latest stats snapshot.
    #[must_use]
    pub fn stats(&self) -> Snapshot {
        self.inner.stats.snapshot()
    }

    /// Borrow the live [`Stats`] aggregator as a shared handle.
    ///
    /// Returns the same `Arc<Stats>` the runtime publishes to,
    /// so an embedder can read counters as they update without
    /// going through the periodic snapshot path. Useful for
    /// in-process metrics integrations that prefer a pull model
    /// (Prometheus scrape, OpenTelemetry pull readers) over the
    /// push-on-flush path exposed by [`MetricsSink`].
    ///
    /// The returned handle is `Clone`; callers may share it
    /// across threads. The aggregator outlives the handle for as
    /// long as the [`ServerHandle`] tree is alive.
    ///
    /// [`MetricsSink`]: crate::embed::MetricsSink
    #[must_use]
    pub fn stats_handle(&self) -> Arc<Stats> {
        Arc::clone(&self.inner.stats)
    }

    /// Manifest of every metric the engine emits.
    #[must_use]
    pub fn describe_stats(&self) -> Vec<MetricSpec> {
        let _ = describe_stats(); // ensure the module is exercised
        let mut out: Vec<MetricSpec> = Vec::new();
        out.extend(crate::stats::POOL_CODEC.iter().copied());
        out.extend(crate::stats::SERVER_CODEC.iter().copied());
        out
    }

    /// Snapshot of every peer in the cluster.
    #[must_use]
    pub fn peers(&self) -> Vec<PeerSnapshot> {
        self.inner
            .pool
            .peers()
            .read()
            .iter()
            .map(PeerSnapshot::from)
            .collect()
    }

    /// Snapshot of every datacenter and its racks.
    #[must_use]
    pub fn datacenters(&self) -> Vec<DatacenterSnapshot> {
        self.inner
            .pool
            .datacenters()
            .read()
            .iter()
            .map(DatacenterSnapshot::from)
            .collect()
    }

    /// Snapshot of the token ring.
    #[must_use]
    pub fn ring(&self) -> RingSnapshot {
        let dcs = self.inner.pool.datacenters().read();
        let mut entries: Vec<(DynToken, u32)> = Vec::new();
        for dc in dcs.iter() {
            for rack in dc.racks() {
                for c in rack.continuums() {
                    entries.push((c.token.clone(), c.peer_idx));
                }
            }
        }
        RingSnapshot {
            entries,
            generation: self.inner.generation.load(Ordering::Relaxed),
        }
    }

    /// Inject a parsed request as if it had arrived from a
    /// client. Returns the response message.
    ///
    /// This is the test/embedding entry point that bypasses the
    /// proxy listener. The dispatcher computes a routing plan; if
    /// the plan resolves to the local datastore, the
    /// [`Datastore::dispatch`]
    /// hook is invoked. If the plan resolves to a remote peer
    /// co-located in the same process (via the in-process
    /// registry), the request is forwarded to that peer's
    /// datastore and its response is returned.
    pub async fn inject_request(&self, req: Msg) -> Result<Msg, EmbedError> {
        // Use the parsed key (if any) for routing; if there are
        // no keys, the dispatcher's `LocalDatastore` short-circuit
        // applies.
        let key: Vec<u8> = req
            .keys()
            .first()
            .map(|kp| kp.tag_bytes().to_vec())
            .unwrap_or_default();
        let plan = self.inner.dispatcher.plan(&req, &key);
        match plan {
            DispatchPlan::Drop => {
                // Drop: respond with an empty message of the same
                // id so callers can correlate.
                let mut rsp = Msg::new(req.id(), crate::msg::MsgType::Unknown, false);
                rsp.set_parent_id(req.id());
                Ok(rsp)
            }
            DispatchPlan::NoTargets => Err(EmbedError::Inject(
                "cluster has no quorum-eligible targets".into(),
            )),
            DispatchPlan::LocalDatastore => self.inner.dispatch_local(req).await,
            DispatchPlan::Replicas { targets, .. } => {
                self.inner.stats.pool_incr_by(PoolField::ForwardError, 0); // touch counter table
                self.inner.stats.server_incr(ServerField::ReadRequests);
                // Snapshot the targets' addresses outside any
                // lock so we don't hold the peer-list read guard
                // across the awaits below.
                let resolved: Vec<(bool, Option<SocketAddr>)> = {
                    let peers = self.inner.pool.peers().read();
                    targets
                        .iter()
                        .map(|t| {
                            let peer = peers.get(t.peer_idx as usize);
                            let is_local = peer.is_some_and(crate::cluster::peer::Peer::is_local);
                            let addr = peer.and_then(|p| p.endpoint().pname().parse().ok());
                            (is_local, addr)
                        })
                        .collect()
                };
                let mut last_err: Option<EmbedError> = None;
                for (is_local, addr) in resolved {
                    if is_local {
                        return self.inner.dispatch_local(req).await;
                    }
                    if let Some(addr) = addr {
                        if let Some(remote) = registry_lookup(addr) {
                            let mut forwarded = Msg::new(req.id(), req.ty(), true);
                            forwarded.set_parent_id(req.parent_id());
                            match remote.datastore.dispatch(forwarded).await {
                                Ok(rsp) => return Ok(rsp),
                                Err(e) => last_err = Some(EmbedError::Inject(e.to_string())),
                            }
                        }
                    }
                }
                if let Some(e) = last_err {
                    return Err(e);
                }
                // No registered remote peer; fall back to the
                // local datastore so the embed surface stays
                // useful in mixed deployments.
                self.inner.dispatch_local(req).await
            }
        }
    }

    /// Reload the configuration. Validates the new [`Config`]
    /// before any state is touched; on failure the running config
    /// is left untouched.
    ///
    /// This is the in-process equivalent of SIGHUP: instead of
    /// re-reading the YAML from disk, the embedding API takes the
    /// replacement `Config` directly.
    pub async fn reload(&self, mut cfg: Config) -> Result<(), EmbedError> {
        cfg.finalize();
        cfg.validate()?;
        let new_pool = cfg.pool().clone();
        // Update the stored pool so subsequent .stats() / .peers()
        // calls reflect the live config. The dispatcher's pool
        // structure is left untouched here; a future stage will
        // rebuild it under a write lock.
        *self.inner.config.lock() = new_pool;
        let gen_id = next_generation();
        self.inner.generation.store(gen_id, Ordering::Relaxed);
        self.inner
            .bus
            .send(ServerEvent::ConfigReloaded { generation: gen_id });
        // Yield once so the broadcast send is observable on the
        // calling task's scheduler before the function returns.
        tokio::task::yield_now().await;
        Ok(())
    }

    /// Graceful shutdown.
    ///
    /// Cancels every background task, deregisters from the
    /// in-process peer registry, drains the join set, and
    /// returns. Idempotent: calling `shutdown` after a previous
    /// `shutdown` (or after [`ServerHandle::join`] returned)
    /// completes successfully without doing any work.
    pub async fn shutdown(&self) -> Result<(), EmbedError> {
        self.inner.cancel.cancel();
        if let Some(addr) = self.inner.dyn_listen_addr {
            registry_remove(addr);
        }
        let drained: Vec<JoinHandle<()>> = std::mem::take(&mut *self.tasks.lock());
        for t in drained {
            // Tasks observe `cancel` and return promptly.
            let _ = t.await;
        }
        Ok(())
    }

    /// Wait for every background task to complete.
    ///
    /// Unlike [`ServerHandle::shutdown`], `join` does not request
    /// cancellation: it parks the caller until the runtime tasks
    /// (gossip, stats aggregator, accept loops, metrics flusher)
    /// finish on their own. The intended use is the
    /// "start-then-park" pattern in long-running embedders that
    /// drive shutdown from a separate signal handler:
    ///
    /// ```no_run
    /// # use dynomite::embed::ServerBuilder;
    /// # use dynomite::conf::DataStore;
    /// # tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap().block_on(async {
    /// let handle = ServerBuilder::new("p")
    ///     .listen("127.0.0.1:0".parse().unwrap())
    ///     .dyn_listen("127.0.0.1:0".parse().unwrap())
    ///     .data_store(DataStore::Valkey)
    ///     .servers(vec![dynomite::conf::ConfServer::parse("127.0.0.1:6379:1").unwrap()])
    ///     .tokens_str("0")
    ///     .build().unwrap()
    ///     .start().await.unwrap();
    /// let shutdown = handle.clone();
    /// tokio::spawn(async move {
    ///     // Replace with `tokio::signal::ctrl_c().await.unwrap();`
    ///     // in a real embedder.
    ///     shutdown.shutdown().await.unwrap();
    /// });
    /// handle.join().await;
    /// # });
    /// ```
    ///
    /// `join` resolves once every task spawned by
    /// [`Server::start`] has returned. Calling `join` repeatedly
    /// is safe; the second call returns immediately because the
    /// task set is drained on the first.
    pub async fn join(&self) {
        let drained: Vec<JoinHandle<()>> = std::mem::take(&mut *self.tasks.lock());
        for t in drained {
            let _ = t.await;
        }
    }
}

// -------- helpers ---------------------------------------------------------

fn build_local_peer(pool: &ConfPool, cfg: &PoolConfig) -> Peer {
    let dyn_listen = pool.dyn_listen.as_ref().map_or_else(
        || ("127.0.0.1".to_string(), 0u16),
        |l| (l.name().to_string(), l.port()),
    );
    let tokens = pool
        .tokens
        .as_ref()
        .map(|tl| {
            tl.components()
                .iter()
                .map(|c| DynToken::from_u32(c.digits.parse::<u32>().unwrap_or(0)))
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let mut peer = Peer::new(
        0,
        PeerEndpoint::tcp(dyn_listen.0, dyn_listen.1),
        cfg.rack.clone(),
        cfg.dc.clone(),
        if tokens.is_empty() {
            vec![DynToken::from_u32(0)]
        } else {
            tokens
        },
        true,
        true,
        false,
    );
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_secs());
    peer.set_state(PeerState::Normal, now_secs);
    peer
}

fn peers_from_seeds(cfg: &PoolConfig, seeds: &[ConfDynSeed], start_idx: u32) -> Vec<Peer> {
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_secs());
    seeds
        .iter()
        .enumerate()
        .map(|(i, s)| {
            let tokens: Vec<DynToken> = s
                .tokens()
                .components()
                .iter()
                .map(|c| DynToken::from_u32(c.digits.parse::<u32>().unwrap_or(0)))
                .collect();
            let idx_off = u32::try_from(i).unwrap_or(0);
            let mut p = Peer::new(
                start_idx + idx_off,
                PeerEndpoint::tcp(s.host().to_string(), s.port()),
                s.rack().to_string(),
                s.dc().to_string(),
                if tokens.is_empty() {
                    vec![DynToken::from_u32(0)]
                } else {
                    tokens
                },
                false,
                s.dc() == cfg.dc,
                false,
            );
            p.set_state(PeerState::Normal, now_secs);
            p
        })
        .collect()
}

// Unused legacy helpers removed.

async fn bind_listener(
    listen: Option<&crate::conf::ConfListen>,
    transport: crate::conf::Transport,
    #[cfg_attr(not(feature = "quic"), allow(unused_variables))] pool: &crate::conf::ConfPool,
) -> Result<(Option<TcpListener>, Option<SocketAddr>), EmbedError> {
    let Some(l) = listen else {
        return Ok((None, None));
    };
    let host = l.name();
    let port = l.port();
    if host.is_empty() {
        return Ok((None, None));
    }
    let addr_str = format!("{host}:{port}");
    let Ok(_addr) = addr_str.parse::<SocketAddr>() else {
        return Ok((None, None));
    };
    match transport {
        crate::conf::Transport::Tcp => {
            let listener = TcpListener::bind(&addr_str).await?;
            let local = listener.local_addr()?;
            Ok((Some(listener), Some(local)))
        }
        #[cfg(feature = "quic")]
        crate::conf::Transport::Quic => {
            // The embedded server's accept loop closes every
            // accepted client (see `accept_loop` below): it logs
            // and closes. We still bind the QUIC
            // listener up-front so the embedder sees a clean
            // bind error if the cert / key paths are bad,
            // matching the TCP path's eager-bind contract;
            // dropping the listener here is fine because
            // production QUIC traffic flows through the
            // `dynomited` binary, not the embedded harness.
            let cert = pool
                .quic_cert_file
                .as_ref()
                .and_then(|p| p.to_str().map(str::to_owned))
                .ok_or_else(|| {
                    EmbedError::Build(
                        "transport: quic requires quic_cert_file (UTF-8 path)".to_string(),
                    )
                })?;
            let key = pool
                .quic_key_file
                .as_ref()
                .and_then(|p| p.to_str().map(str::to_owned))
                .ok_or_else(|| {
                    EmbedError::Build(
                        "transport: quic requires quic_key_file (UTF-8 path)".to_string(),
                    )
                })?;
            let cfg = crate::net::QuicConfig::server_with_cert_paths(cert, key);
            let parsed: SocketAddr = addr_str.parse().map_err(|e: std::net::AddrParseError| {
                EmbedError::Build(format!("invalid quic listen address {addr_str}: {e}"))
            })?;
            let listener = crate::net::QuicListener::bind(parsed, cfg).await?;
            let local = listener.local_addr();
            // The QUIC listener is dropped on this scope exit;
            // the in-process embedder does not drive a
            // QUIC accept loop. Return `None` for the listener
            // so the existing TCP-flavoured accept_loop is
            // skipped, but report the local address for
            // observability.
            drop(listener);
            Ok((None, Some(local)))
        }
        #[cfg(not(feature = "quic"))]
        crate::conf::Transport::Quic => Err(EmbedError::Build(
            "transport: quic requires the engine's `quic` Cargo feature".to_string(),
        )),
    }
}

// removed unused try_bind helper

async fn accept_loop(
    inner: Arc<ServerInner>,
    listener: TcpListener,
    addr: SocketAddr,
    role: ConnRoleTag,
) {
    loop {
        tokio::select! {
            biased;
            () = inner.cancel.cancelled() => return,
            res = listener.accept() => {
                let Ok((sock, peer)) = res else { return };
                let conn_id = next_conn_id();
                inner.bus.send(ServerEvent::ConnectionAccepted {
                    conn_id,
                    role,
                    local_addr: Some(addr),
                });
                if matches!(role, ConnRoleTag::Proxy) {
                    // Client plane: serve the connection through the
                    // same per-client pipeline the standalone proxy
                    // uses (parse -> route via the dispatcher ->
                    // write the reply). This is the wire entry point
                    // for an embedded node; `inject_request` remains
                    // available for in-process traffic.
                    //
                    // Small Redis/memcache requests should not
                    // wait on Nagle; a peer that closed before
                    // the option applies is harmless.
                    let _ = sock.set_nodelay(true);
                    let dispatcher: Arc<dyn crate::net::dispatcher::Dispatcher> =
                        inner.dispatcher.clone();
                    let client_proto = inner.client_proto;
                    let bus = inner.bus.clone();
                    tokio::spawn(async move {
                        let transport = Box::new(TcpTransport::new(sock, ConnRole::Client));
                        let conn = Conn::new(transport, ConnRole::Client);
                        let (tx, rx) = mpsc::channel(64);
                        let handler = ClientHandler::new(dispatcher, tx, client_proto);
                        let outcome = client_loop(conn, handler, rx).await;
                        let close_reason = match outcome {
                            Ok(()) => CloseReason::PeerEof,
                            Err(_) => CloseReason::IoError,
                        };
                        bus.send(ServerEvent::ConnectionClosed {
                            conn_id,
                            reason: close_reason,
                        });
                    });
                } else {
                    // Peer plane: cross-process dnode serving is
                    // driven by the `dynomited` binary. Embedded
                    // multi-node setups forward between in-process
                    // nodes through the in-process registry (see
                    // `registry_register`), so the embedded
                    // dyn_listen socket is bound for address
                    // reporting and accepts then closes a raw
                    // cross-process peer connection.
                    tracing::debug!(
                        listen = %addr,
                        peer = %peer,
                        role = ?role,
                        conn_id,
                        "embedded dyn_listen accepted a cross-process peer; embedded \
                         multi-node forwarding uses the in-process registry. Closing \
                         connection."
                    );
                    let bus = inner.bus.clone();
                    let cancel = inner.cancel.clone();
                    tokio::spawn(async move {
                        let _ = sock; // drop on close
                        let close_reason = if cancel.is_cancelled() {
                            CloseReason::LocalClose
                        } else {
                            CloseReason::PeerEof
                        };
                        bus.send(ServerEvent::ConnectionClosed {
                            conn_id,
                            reason: close_reason,
                        });
                    });
                }
            }
        }
    }
}

async fn stats_loop(inner: Arc<ServerInner>, interval_ms: Option<i64>) {
    let interval =
        Duration::from_millis(u64::try_from(interval_ms.unwrap_or(1_000)).unwrap_or(1_000));
    let mut ticker = tokio::time::interval(interval);
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            biased;
            () = inner.cancel.cancelled() => return,
            _ = ticker.tick() => {
                let snap = inner.stats.snapshot();
                *inner.snapshot_cache.lock() = snap;
            }
        }
    }
}

async fn metrics_loop(inner: Arc<ServerInner>) {
    let interval = inner.metrics.flush_interval();
    let mut ticker = tokio::time::interval(interval.max(Duration::from_millis(50)));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            biased;
            () = inner.cancel.cancelled() => return,
            _ = ticker.tick() => {
                let snap = inner.snapshot_cache.lock().clone();
                let _ = inner.metrics.emit(&snap).await;
            }
        }
    }
}

async fn gossip_loop(inner: Arc<ServerInner>, interval: Duration) {
    let mut ticker = tokio::time::interval(interval.max(Duration::from_millis(20)));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
    let mut round: u64 = 0;
    let mut known: std::collections::HashSet<(String, u16)> = std::collections::HashSet::new();
    {
        let peers = inner.pool.peers().read();
        for p in peers.iter() {
            known.insert((p.endpoint().host().to_string(), p.endpoint().port()));
        }
    }
    loop {
        tokio::select! {
            biased;
            () = inner.cancel.cancelled() => return,
            _ = ticker.tick() => {
                let round_started = std::time::Instant::now();
                let round_ts = std::time::SystemTime::now();
                round += 1;
                let seeds = inner.seeds.fetch().unwrap_or_default();
                let mut added: u32 = 0;
                {
                    let mut peers = inner.pool.peers().write();
                    let now_secs = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map_or(0, |d| d.as_secs());
                    let cfg = inner.pool.config().clone();
                    for seed in &seeds {
                        let key = (seed.host().to_string(), seed.port());
                        if known.contains(&key) {
                            continue;
                        }
                        let next_idx = u32::try_from(peers.len()).unwrap_or(u32::MAX);
                        let tokens: Vec<DynToken> = seed
                            .tokens()
                            .components()
                            .iter()
                            .map(|c| DynToken::from_u32(c.digits.parse::<u32>().unwrap_or(0)))
                            .collect();
                        let mut p = Peer::new(
                            next_idx,
                            PeerEndpoint::tcp(seed.host().to_string(), seed.port()),
                            seed.rack().to_string(),
                            seed.dc().to_string(),
                            if tokens.is_empty() {
                                vec![DynToken::from_u32(0)]
                            } else {
                                tokens
                            },
                            false,
                            seed.dc() == cfg.dc,
                            false,
                        );
                        p.set_state(PeerState::Normal, now_secs);
                        peers.push(p);
                        known.insert(key);
                        added += 1;
                    }
                }
                if added > 0 {
                    // Rebuild the topology + ring under fresh
                    // datacenters so the dispatcher sees the new
                    // peers.
                    rebuild_topology(&inner.pool);
                    let peers_now = inner.pool.peers().read();
                    for (idx, p) in peers_now.iter().enumerate().rev().take(added as usize) {
                        let _ = idx;
                        inner.bus.send(ServerEvent::PeerUp(p.idx()));
                        inner.events.publish(crate::events::ClusterEvent::PeerUp {
                            peer_id: p.idx(),
                            dc: p.dc().to_string(),
                            ts: round_ts,
                        });
                    }
                    inner.events.publish(crate::events::ClusterEvent::RingChanged {
                        tag: "seed-discovery".to_string(),
                        ts: round_ts,
                    });
                }
                let count = u32::try_from(inner.pool.peers().read().len()).unwrap_or(u32::MAX);
                inner.bus.send(ServerEvent::GossipRound { round, peers: count });
                inner.events.publish(crate::events::ClusterEvent::GossipRoundComplete {
                    duration: round_started.elapsed(),
                    peers_seen: count as usize,
                    ts: round_ts,
                });
                inner.stats.pool_incr(PoolField::StatsCount);
            }
        }
    }
}

fn rebuild_topology(pool: &Arc<ServerPool>) {
    use crate::cluster::Datacenter;
    let peers = pool.peers().read();
    let mut new_dcs: Vec<Datacenter> = Vec::new();
    for p in peers.iter() {
        let idx = if let Some(i) = new_dcs.iter().position(|d| d.name() == p.dc()) {
            i
        } else {
            new_dcs.push(Datacenter::new(p.dc().to_string()));
            new_dcs.len() - 1
        };
        new_dcs[idx].upsert_rack(p.rack().to_string());
    }
    drop(peers);
    {
        let mut dcs = pool.datacenters().write();
        *dcs = new_dcs;
    }
    pool.rebuild_ring();
    pool.preselect_remote_racks();
}

// Internal helpers (`shutdown_signal`, `_instant_now`) removed:
// they were public-but-doc(hidden) escape hatches that risked
// SemVer commitments. The `oneshot` channel was not used
// outside the (deleted) test helper, and `Instant` was kept in
// scope only to silence an unused-import warning. Both have
// been dropped.