tranquil-pds/crates/tranquil-ripple/src/engine.rs

use crate::cache::RippleCache;
use crate::config::RippleConfig;
use crate::crdt::ShardedCrdtStore;
use crate::eviction::MemoryBudget;
use crate::gossip::{GossipEngine, PeerId};
use crate::metrics;
use crate::rate_limiter::RippleRateLimiter;
use crate::transport::{ClusterKey, Transport};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tranquil_infra::{Cache, DistributedRateLimiter};

pub struct RippleEngine;

impl RippleEngine {
    pub async fn start(
        config: RippleConfig,
        shutdown: CancellationToken,
    ) -> Result<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>, SocketAddr), RippleStartError>
    {
        let store = Arc::new(ShardedCrdtStore::new(config.machine_id));

        let cluster_key = match config.cluster_key.as_deref() {
            Some(secret) => Some(ClusterKey::new(secret)),
            None => {
                if !config.bind_addr.ip().is_loopback() && !config.allow_insecure {
                    return Err(RippleStartError::Config(format!(
                        "ripple is bound to non-loopback {} without RIPPLE_CLUSTER_KEY. Set the cluster key to authenticate peers, or set RIPPLE_ALLOW_INSECURE=true to bind unauthenticated on a trusted network",
                        config.bind_addr
                    )));
                }
                None
            }
        };
        let (transport, incoming_rx) =
            Transport::bind(config.bind_addr, cluster_key, shutdown.clone())
                .await
                .map_err(|e| RippleStartError::Bind(e.to_string()))?;

        let transport = Arc::new(transport);

        let bound_addr = transport.local_addr();
        let generation = u32::try_from(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs()
                % u64::from(u32::MAX),
        )
        .unwrap_or(0);
        let local_id = PeerId {
            addr: bound_addr,
            machine_id: config.machine_id,
            generation,
        };

        let gossip = GossipEngine::new(transport, store.clone(), local_id);

        let gossip_handle = gossip.spawn(
            config.seed_peers,
            config.gossip_interval_ms,
            incoming_rx,
            shutdown.clone(),
        );

        let budget = MemoryBudget::new(config.cache_max_bytes);
        let store_for_eviction = store.clone();
        let eviction_shutdown = shutdown.clone();
        let eviction_handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
            loop {
                tokio::select! {
                    _ = eviction_shutdown.cancelled() => break,
                    _ = interval.tick() => {
                        budget.enforce(&store_for_eviction);
                    }
                }
            }
        });

        let shutdown_for_monitor = shutdown.clone();
        tokio::spawn(async move {
            shutdown_for_monitor.cancelled().await;
            let gossip_result = gossip_handle.await;
            let eviction_result = eviction_handle.await;
            if let Err(e) = gossip_result {
                tracing::error!(error = %e, "gossip task panicked");
            }
            if let Err(e) = eviction_result {
                tracing::error!(error = %e, "eviction task panicked");
            }
        });

        let cache: Arc<dyn Cache> = Arc::new(RippleCache::new(store.clone()));
        let rate_limiter: Arc<dyn DistributedRateLimiter> = Arc::new(RippleRateLimiter::new(store));

        metrics::describe_metrics();

        tracing::info!(
            bind = %bound_addr,
            machine_id = config.machine_id,
            max_cache_mb = config.cache_max_bytes / (1024 * 1024),
            "ripple engine started"
        );

        Ok((cache, rate_limiter, bound_addr))
    }
}

#[derive(Debug, thiserror::Error)]
pub enum RippleStartError {
    #[error("failed to bind transport: {0}")]
    Bind(String),
    #[error("configuration error: {0}")]
    Config(String),
}