Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
4.1 kB · 117 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118use 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),}