diff --git a/.gcloudignore b/.gcloudignore new file mode 100644 index 0000000..fcc19ec --- /dev/null +++ b/.gcloudignore @@ -0,0 +1,9 @@ +.git/ +target/ +.claude/ +tests/ +docs/ +docker-benchmark/ +maelstrom/ +*.md +*.png diff --git a/cloudbuild.yaml b/cloudbuild.yaml new file mode 100644 index 0000000..a27fadc --- /dev/null +++ b/cloudbuild.yaml @@ -0,0 +1,14 @@ +steps: + - name: 'gcr.io/cloud-builders/docker' + args: + - 'build' + - '-t' + - 'us-central1-docker.pkg.dev/datadog-sandbox/libstream/redis-rust:v0.1.12' + - '-f' + - 'docker/Dockerfile' + - '.' +images: + - 'us-central1-docker.pkg.dev/datadog-sandbox/libstream/redis-rust:v0.1.12' +options: + machineType: 'E2_HIGHCPU_32' +timeout: '1800s' diff --git a/docker/Dockerfile.nodd b/docker/Dockerfile.nodd new file mode 100644 index 0000000..1899ae5 --- /dev/null +++ b/docker/Dockerfile.nodd @@ -0,0 +1,100 @@ +# Redis Rust - Drop-in Replacement for Redis +# +# This Dockerfile creates an image that is compatible with the official +# redis Docker image. You can use it as a direct replacement: +# +# # Instead of: +# docker run -p 6379:6379 redis +# +# # Use: +# docker run -p 6379:6379 redis-rust +# +# All standard Redis tooling works: +# redis-cli -h localhost -p 6379 +# redis-benchmark -h localhost -p 6379 +# +# Environment Variables: +# REDIS_PORT - Server port (default: 6379) +# REDIS_STORE_TYPE - Storage: memory, localfs, s3 (default: localfs) +# REDIS_DATA_PATH - Data directory (default: /data) +# +# For S3 persistence: +# REDIS_S3_BUCKET - S3 bucket name +# REDIS_S3_ENDPOINT - S3 endpoint (for MinIO) +# AWS_ACCESS_KEY_ID - AWS credentials +# AWS_SECRET_ACCESS_KEY + +# Build stage +FROM rust:1.93-slim-bookworm AS builder + +RUN apt-get update && apt-get install -y \ + pkg-config \ + libssl-dev \ + build-essential \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY Cargo.toml Cargo.lock* ./ + +# Dependency caching - build deps with dummy src/benches, then clean for real build +RUN mkdir -p src/bin benches && \ + echo "fn main() {}" > src/main.rs && \ + echo "fn main() {}" > src/bin/server_optimized.rs && \ + echo "fn main() {}" > src/bin/server_persistent.rs && \ + echo "fn main() {}" > benches/hot_paths.rs && \ + echo "" > src/lib.rs && \ + cargo build --release --features "s3" 2>/dev/null || true && \ + rm -rf src benches target/release/.fingerprint/redis-sim-* \ + target/release/server-persistent target/release/redis-server-optimized \ + target/release/deps/redis_sim-* target/release/deps/libredis_sim* \ + target/release/incremental/redis_sim-* + +# Copy source and benches (benches needed to satisfy Cargo.toml manifest) +COPY src ./src +COPY benches ./benches +RUN cargo build --release --features "s3" \ + --bin redis-server-optimized \ + --bin server-persistent + +# Runtime stage - matches official Redis image layout +FROM debian:bookworm-slim + +RUN apt-get update && apt-get install -y \ + ca-certificates \ + curl \ + netcat-openbsd \ + && rm -rf /var/lib/apt/lists/* + +# Match Redis image: use redis user with UID 999 +RUN groupadd -r -g 999 redis && useradd -r -g redis -u 999 redis + +# Match Redis image: /data as working directory +RUN mkdir /data && chown redis:redis /data +WORKDIR /data + +# Copy binaries +COPY --from=builder /app/target/release/redis-server-optimized /usr/local/bin/ +COPY --from=builder /app/target/release/server-persistent /usr/local/bin/ + +# Create symlinks for Redis compatibility +RUN ln -s /usr/local/bin/server-persistent /usr/local/bin/redis-server + +# Default environment - Redis compatible +ENV REDIS_PORT=6379 \ + REDIS_STORE_TYPE=localfs \ + REDIS_DATA_PATH=/data \ + AWS_REGION=us-east-1 + +USER redis + +# Match Redis image: expose 6379 +EXPOSE 6379 + +VOLUME /data + +# Health check using Redis PING command (RESP format) +HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \ + CMD printf '*1\r\n$4\r\nPING\r\n' | nc -q1 localhost 6379 | grep -q "PONG" || exit 1 + +# Default: persistent server (drop-in for Redis with durability) +ENTRYPOINT ["server-persistent"] diff --git a/src/bin/server_persistent.rs b/src/bin/server_persistent.rs index 91c52e7..0035fb7 100644 --- a/src/bin/server_persistent.rs +++ b/src/bin/server_persistent.rs @@ -25,9 +25,10 @@ //! | POD_NAME | - | StatefulSet pod name (e.g., redis-rust-0) | //! | POD_NAMESPACE | default | Kubernetes namespace | //! | REPLICATION_ENABLED | false | Enable gossip-based replication | +//! | REPLICATION_PEERS | - | Explicit peer list (comma-separated host:port). Overrides DNS discovery. | //! | GOSSIP_PORT | 7000 | Port for gossip protocol | //! | CLUSTER_SIZE | 3 | Number of replicas in StatefulSet | -//! | SERVICE_NAME | redis-rust-headless | Headless service name | +//! | SERVICE_NAME | redis-rust-headless | Headless service name (used only if REPLICATION_PEERS unset) | //! //! ## WAL (Write-Ahead Log) //! @@ -134,19 +135,27 @@ impl ClusterConfig { let gossip_interval_ms = std::env::var("GOSSIP_INTERVAL_MS") .ok() .and_then(|s| s.parse().ok()) - .unwrap_or(100) + .unwrap_or(1000) // Default 1s (was 100ms — 10Hz caused OTel span accumulation OOM) .clamp(GOSSIP_INTERVAL_MS_MIN, GOSSIP_INTERVAL_MS_MAX); // TigerStyle: Clamp to valid range // Parse replica ID from POD_NAME (e.g., "redis-rust-0" -> 0) let replica_id = Self::parse_replica_id_from_env().min(REPLICA_ID_MAX); - // Build peer list from Kubernetes DNS + // Build peer list: prefer REPLICATION_PEERS env var, fall back to Kubernetes DNS let peers = if enabled { - Self::build_peer_list(replica_id, cluster_size, gossip_port) + Self::resolve_peers(replica_id, cluster_size, gossip_port) } else { vec![] }; + // When REPLICATION_PEERS overrides peer discovery, cluster_size must + // reflect the actual peer count (peers + self) to keep invariants valid. + let cluster_size = if enabled && std::env::var("REPLICATION_PEERS").is_ok() { + peers.len() + 1 + } else { + cluster_size + }; + let config = ClusterConfig { enabled, replica_id, @@ -230,6 +239,48 @@ impl ClusterConfig { .unwrap_or(DEFAULT_REPLICA_ID) } + /// Resolve peer addresses for gossip replication. + /// + /// Priority: + /// 1. REPLICATION_PEERS env var (comma-separated host:port pairs) — used by the + /// libstream WASM provisioner to inject correct addresses per tenant. + /// 2. Fall back to Kubernetes headless service DNS construction (legacy). + fn resolve_peers(my_replica_id: u64, cluster_size: usize, gossip_port: u16) -> Vec { + if let Ok(peers_env) = std::env::var("REPLICATION_PEERS") { + // Filter out self from the peer list. The WASM provisioner includes + // ALL nodes in REPLICATION_PEERS; we must exclude our own address + // to prevent self-gossip (node connecting to itself in a feedback loop + // that causes unbounded memory growth on node-0). + let my_pod_name = std::env::var("POD_NAME").unwrap_or_default(); + let peers: Vec = peers_env + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .filter(|s| { + // Exclude entries that contain our pod name as a prefix + // e.g., "redis-replicated-kv-v2-0.redis-..." matches POD_NAME="redis-replicated-kv-v2-0" + let my_pod_prefix = format!("{}.", my_pod_name); + if !my_pod_name.is_empty() && s.starts_with(&my_pod_prefix) { + info!("Excluding self from peer list: {}", s); + false + } else { + true + } + }) + .collect(); + if !peers.is_empty() { + info!( + "Using REPLICATION_PEERS env var: {} peers configured (excluded self)", + peers.len() + ); + return peers; + } + warn!("REPLICATION_PEERS env var is set but empty, falling back to DNS discovery"); + } + + Self::build_peer_list(my_replica_id, cluster_size, gossip_port) + } + /// Build peer list from Kubernetes headless service DNS /// DNS format: ...svc.cluster.local fn build_peer_list(my_replica_id: u64, cluster_size: usize, gossip_port: u16) -> Vec { @@ -293,24 +344,55 @@ async fn start_gossip_listener( state: Arc, ) -> Result<(), Box> { use tokio::io::AsyncReadExt; + use tokio::task::JoinHandle; + use std::collections::HashMap; + use std::net::IpAddr; - // TigerStyle: Explicit limit + // TigerStyle: Explicit limits const MAX_MESSAGE_SIZE: usize = 1024 * 1024; // 1MB + const MAX_PEERS: usize = 64; // Safety bound on tracked peers let addr = format!("0.0.0.0:{}", port); let listener = TcpListener::bind(&addr).await?; info!("Gossip listener started on {}", addr); + // Track one active connection per peer IP to prevent unbounded task spawning. + // When a peer reconnects, abort the old handler before spawning a replacement. + let mut active_peers: HashMap> = HashMap::new(); + loop { let (stream, peer_addr) = listener.accept().await?; - info!("Gossip connection from {}", peer_addr); + let peer_ip = peer_addr.ip(); + + // Clean up finished tasks + active_peers.retain(|_, handle| !handle.is_finished()); + + // If this peer already has an active connection, abort it + if let Some(old_handle) = active_peers.remove(&peer_ip) { + if !old_handle.is_finished() { + old_handle.abort(); + debug!("Aborted previous gossip handler for peer {}", peer_ip); + } + } + + debug!("Gossip connection from {} (active peers: {})", peer_addr, active_peers.len()); let state_clone = state.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { if let Err(e) = handle_gossip_connection(stream, state_clone, MAX_MESSAGE_SIZE).await { warn!("Gossip connection error from {}: {}", peer_addr, e); } }); + + // Track this peer's handler (with safety bound) + if active_peers.len() < MAX_PEERS { + active_peers.insert(peer_ip, handle); + } + + debug_assert!( + active_peers.len() <= MAX_PEERS, + "Active peer count must not exceed MAX_PEERS" + ); } } @@ -608,10 +690,17 @@ async fn main() -> Result<(), Box> { } } - let (worker_handles, sender) = integration.start_workers().await?; - - // Connect delta sink BEFORE wrapping state in Arc - state.set_delta_sink(sender); + // Only start persistence workers for non-memory store types. + // Memory-mode pods don't need persistence — the streaming pipeline would accumulate + // deltas in the InMemoryObjectStore's segment HashMap, growing unbounded (~300Mi/min). + let worker_handles = if config.store_type != "memory" { + let (handles, sender) = integration.start_workers().await?; + state.set_delta_sink(sender); + Some(handles) + } else { + info!("Skipping persistence pipeline for store_type=memory"); + None + }; // Start WAL actor if enabled let wal_task = if let Some(ref wc) = wal_config { @@ -766,8 +855,10 @@ async fn main() -> Result<(), Box> { info!("WAL actor shutdown complete"); } - info!("Shutting down streaming persistence workers..."); - worker_handles.shutdown().await; + if let Some(handles) = worker_handles { + info!("Shutting down streaming persistence workers..."); + handles.shutdown().await; + } // Shutdown observability (flush pending spans/metrics) shutdown(); @@ -1042,4 +1133,58 @@ mod tests { ); assert_eq!(repl.gossip_interval_ms, 100, "Gossip interval should match"); } + + /// Test REPLICATION_PEERS env var overrides DNS discovery + #[test] + fn test_resolve_peers_from_env_var() { + let _guard = ENV_LOCK.lock().unwrap(); + + // Clean up any leftover env vars + std::env::remove_var("POD_NAME"); + std::env::remove_var("REPLICATION_PEERS"); + + let explicit_peers = "redis-kv-0.redis-kv-headless.libstream.svc.cluster.local:7000,redis-kv-2.redis-kv-headless.libstream.svc.cluster.local:7000"; + std::env::set_var("REPLICATION_PEERS", explicit_peers); + + let peers = ClusterConfig::resolve_peers(1, 3, 7000); + + assert_eq!(peers.len(), 2, "Should have 2 peers from env var"); + assert!( + peers[0].contains("redis-kv-0"), + "First peer should be redis-kv-0" + ); + assert!( + peers[1].contains("redis-kv-2"), + "Second peer should be redis-kv-2" + ); + + // Clean up + std::env::remove_var("REPLICATION_PEERS"); + } + + /// Test REPLICATION_PEERS falls back to DNS when empty + #[test] + fn test_resolve_peers_fallback_when_empty() { + let _guard = ENV_LOCK.lock().unwrap(); + + std::env::set_var("REPLICATION_PEERS", ""); + std::env::set_var("POD_NAME", "redis-rust-0"); + std::env::set_var("POD_NAMESPACE", "default"); + std::env::set_var("SERVICE_NAME", "redis-rust-headless"); + + let peers = ClusterConfig::resolve_peers(0, 3, 7000); + + // Should fall back to DNS-based peer list + assert_eq!(peers.len(), 2, "Should fall back to DNS with 2 peers"); + assert!( + peers[0].contains("redis-rust-headless"), + "Should use DNS format on fallback" + ); + + // Clean up + std::env::remove_var("REPLICATION_PEERS"); + std::env::remove_var("POD_NAME"); + std::env::remove_var("POD_NAMESPACE"); + std::env::remove_var("SERVICE_NAME"); + } } diff --git a/src/observability/config.rs b/src/observability/config.rs index 4c50ac8..da7e442 100644 --- a/src/observability/config.rs +++ b/src/observability/config.rs @@ -18,7 +18,12 @@ pub struct DatadogConfig { pub env: String, /// Service version (default: from Cargo.toml) pub version: String, - /// Sample rate for traces (0.0 - 1.0, default: 1.0) + /// Sample rate for traces (0.0 - 1.0, default: 0.1) + /// + /// 100% sampling (1.0) causes OOM when combined with high-frequency gossip + /// because the OTel batch exporter accumulates spans faster than the DD agent + /// can drain them. 10% sampling (0.1) provides sufficient visibility without + /// unbounded memory growth. Override with DD_TRACE_SAMPLE_RATE env var. pub trace_sample_rate: f64, /// Enable JSON logs with trace correlation pub logs_injection: bool, @@ -51,7 +56,7 @@ impl DatadogConfig { trace_sample_rate: std::env::var("DD_TRACE_SAMPLE_RATE") .ok() .and_then(|s| s.parse().ok()) - .unwrap_or(1.0), + .unwrap_or(0.1), logs_injection: std::env::var("DD_LOGS_INJECTION") .map(|v| v == "true" || v == "1") .unwrap_or(false), @@ -103,7 +108,7 @@ mod tests { let config = DatadogConfig::from_env(); assert_eq!(config.service_name, "redis-rust"); assert_eq!(config.metric_prefix, "redis_rust"); - assert!((config.trace_sample_rate - 1.0).abs() < f64::EPSILON); + assert!((config.trace_sample_rate - 0.1).abs() < f64::EPSILON); } #[test] diff --git a/src/observability/tracing_setup.rs b/src/observability/tracing_setup.rs index 3acc777..fd2435b 100644 --- a/src/observability/tracing_setup.rs +++ b/src/observability/tracing_setup.rs @@ -6,6 +6,8 @@ use opentelemetry_datadog::DatadogPropagator; use opentelemetry_sdk::trace::Sampler; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; +use tracing_subscriber::Layer; +use tracing_subscriber::filter; use super::config::DatadogConfig; @@ -14,6 +16,8 @@ use super::config::DatadogConfig; /// Sets up: /// - OpenTelemetry with Datadog exporter for distributed tracing /// - tracing-subscriber with environment-based filtering +/// - Gossip/replication spans are excluded from OTel to prevent unbounded +/// memory growth from the batch exporter accumulating high-frequency spans pub fn init(config: &DatadogConfig) -> Result<(), Box> { // Set global propagator for distributed tracing context opentelemetry::global::set_text_map_propagator(DatadogPropagator::default()); @@ -33,17 +37,30 @@ pub fn init(config: &DatadogConfig) -> Result<(), Box) + Send + Sync>; #[allow(dead_code)] @@ -46,24 +53,69 @@ impl GossipManager { let _ = self.outbound_tx.send(msg); } + /// Start the gossip TCP server with per-peer connection tracking. + /// + /// ## Invariants + /// - At most `MAX_CONNECTIONS_PER_PEER` (1) active handler task per peer IP. + /// - When a peer reconnects, the old handler is aborted before spawning a new one. + /// - This prevents unbounded task accumulation from peers that reconnect every gossip round. pub async fn start_server( config: ReplicationConfig, delta_callback: DeltaCallback, ) -> std::io::Result<()> { - let port = 3001 + config.replica_id as u16; + let port = 3001u16 + .checked_add(config.replica_id as u16) + .expect("Precondition: replica_id must not overflow gossip port range"); let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?; info!("Gossip server listening on port {}", port); + // Track active connections per peer IP. When a peer reconnects, + // we abort the old handler task to prevent unbounded task growth. + let mut active_peers: HashMap> = HashMap::new(); + loop { let (stream, addr) = listener.accept().await?; - info!("Gossip connection from {}", addr); + let peer_ip = addr.ip(); + + // If this peer already has an active connection, abort it first. + // This is the fix for the unbounded connection accept loop: + // each gossip round from a peer opens a new TCP connection, and + // without this cleanup, tasks accumulate (~20/sec with 2 peers). + if let Some(old_handle) = active_peers.remove(&peer_ip) { + if !old_handle.is_finished() { + debug!( + "Aborting stale gossip handler for peer {} (replaced by new connection)", + peer_ip + ); + old_handle.abort(); + } + } + + // Garbage-collect finished tasks to prevent the HashMap from growing + // unboundedly with IPs of peers that have since disconnected. + active_peers.retain(|_ip, handle| !handle.is_finished()); + + debug_assert!( + !active_peers.contains_key(&peer_ip), + "Postcondition: old peer handle must be removed before inserting new one" + ); + + info!("Gossip connection from {} (active peers: {})", addr, active_peers.len().checked_add(1).unwrap_or(usize::MAX)); let callback = delta_callback.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { if let Err(e) = Self::handle_peer_connection(stream, callback).await { - warn!("Gossip peer error: {}", e); + warn!("Gossip peer {} error: {}", addr, e); } }); + + active_peers.insert(peer_ip, handle); + + debug_assert!( + active_peers.len() <= MAX_CONNECTIONS_PER_PEER * 100, + "Invariant: active_peers should not grow unboundedly (current: {})", + active_peers.len() + ); } } @@ -173,6 +225,12 @@ impl GossipManager { selective_mode ); + // Persistent connection pool: reuse TCP connections across gossip rounds. + // This is the primary fix for the connection storm — previously each + // send_to_peer() opened a new TCP connection and dropped it after one message, + // causing ~20 new connections/sec with 2 peers at 100ms intervals. + let mut peer_connections: HashMap = HashMap::new(); + loop { ticker.tick().await; @@ -191,7 +249,7 @@ impl GossipManager { continue; } - // Send each routed message + // Send each routed message using persistent connections for routed in routed_messages { let data = match routed.message.serialize() { Ok(d) => d, @@ -206,7 +264,12 @@ impl GossipManager { Some(target_replica) => { // Targeted message: send to specific replica if let Some(addr) = peer_map.get(&target_replica) { - Self::send_to_peer(addr, &framed_data).await; + Self::send_to_peer_persistent( + &mut peer_connections, + addr, + &framed_data, + ) + .await; } else { debug!("No address for target replica {}", target_replica.0); } @@ -214,7 +277,12 @@ impl GossipManager { None => { // Broadcast message: send to all peers for peer_addr in &peers { - Self::send_to_peer(peer_addr, &framed_data).await; + Self::send_to_peer_persistent( + &mut peer_connections, + peer_addr, + &framed_data, + ) + .await; } } } @@ -222,12 +290,47 @@ impl GossipManager { } } - /// Send framed data to a peer address - async fn send_to_peer(addr: &str, framed_data: &[u8]) { + /// Send framed data to a peer using a persistent connection pool. + /// + /// Reuses existing TCP connections across gossip rounds. If the connection + /// is broken (write fails), it is removed and a fresh connection is established. + /// This eliminates the ~20 new TCP connections/sec that caused OOM on node-0. + /// + /// ## Invariants + /// - `peer_connections` maps peer address strings to live TCP streams. + /// - A broken connection is always removed before attempting reconnection. + /// - At most one connection per peer address exists in the map at any time. + async fn send_to_peer_persistent( + peer_connections: &mut HashMap, + addr: &str, + framed_data: &[u8], + ) { + debug_assert!(!addr.is_empty(), "Precondition: peer address must not be empty"); + debug_assert!(!framed_data.is_empty(), "Precondition: framed data must not be empty"); + + // Try to reuse existing connection + if let Some(stream) = peer_connections.get_mut(addr) { + match stream.write_all(framed_data).await { + Ok(()) => return, // Success — connection reused + Err(e) => { + debug!("Persistent connection to {} broken, reconnecting: {}", addr, e); + peer_connections.remove(addr); + // Fall through to reconnect below + } + } + } + + // No existing connection or it was broken — establish a new one match TcpStream::connect(addr).await { Ok(mut stream) => { - if let Err(e) = stream.write_all(framed_data).await { - warn!("Failed to send to peer {}: {}", addr, e); + match stream.write_all(framed_data).await { + Ok(()) => { + peer_connections.insert(addr.to_string(), stream); + } + Err(e) => { + warn!("Failed to send to peer {} on fresh connection: {}", addr, e); + // Don't insert a broken connection + } } } Err(e) => { @@ -271,6 +374,9 @@ impl GossipManager { selective_mode ); + // Persistent connection pool: reuse TCP connections across gossip rounds. + let mut peer_connections: HashMap = HashMap::new(); + loop { ticker.tick().await; @@ -285,7 +391,7 @@ impl GossipManager { continue; } - // Send each routed message + // Send each routed message using persistent connections for routed in routed_messages { let data = match routed.message.serialize() { Ok(d) => d, @@ -300,7 +406,12 @@ impl GossipManager { Some(target_replica) => { // Targeted message: send to specific replica if let Some(addr) = peer_map.get(&target_replica) { - Self::send_to_peer(addr, &framed_data).await; + Self::send_to_peer_persistent( + &mut peer_connections, + addr, + &framed_data, + ) + .await; } else { debug!("No address for target replica {}", target_replica.0); } @@ -308,7 +419,12 @@ impl GossipManager { None => { // Broadcast message: send to all peers for peer_addr in &peers { - Self::send_to_peer(peer_addr, &framed_data).await; + Self::send_to_peer_persistent( + &mut peer_connections, + peer_addr, + &framed_data, + ) + .await; } } } diff --git a/src/production/replicated_state.rs b/src/production/replicated_state.rs index 3e04839..af01346 100644 --- a/src/production/replicated_state.rs +++ b/src/production/replicated_state.rs @@ -305,15 +305,90 @@ impl ReplicatedShardedState { RespValue::Array(Some(all_keys)) } Command::Info => { + let pid = std::process::id(); let info = format!( - "# Replication\r\nrole:master\r\nreplica_id:{}\r\nconsistency_level:{:?}\r\nreplication_enabled:{}\r\nnum_shards:{}\r\narchitecture:actor_per_shard\r\n", - self.config.replica_id, - self.config.consistency_level, - self.config.enabled, - NUM_SHARDS + "# Server\r\n\ + redis_version:7.0.0\r\n\ + redis_mode:standalone\r\n\ + os:Linux\r\n\ + arch_bits:64\r\n\ + tcp_port:6379\r\n\ + uptime_in_seconds:0\r\n\ + uptime_in_days:0\r\n\ + process_id:{pid}\r\n\ + \r\n\ + # Clients\r\n\ + connected_clients:1\r\n\ + blocked_clients:0\r\n\ + tracking_clients:0\r\n\ + \r\n\ + # Memory\r\n\ + used_memory:0\r\n\ + used_memory_human:0B\r\n\ + used_memory_rss:0\r\n\ + used_memory_rss_human:0B\r\n\ + used_memory_peak:0\r\n\ + used_memory_peak_human:0B\r\n\ + maxmemory:0\r\n\ + maxmemory_human:0B\r\n\ + maxmemory_policy:noeviction\r\n\ + mem_fragmentation_ratio:1.00\r\n\ + mem_allocator:jemalloc\r\n\ + \r\n\ + # Stats\r\n\ + total_connections_received:0\r\n\ + total_commands_processed:0\r\n\ + instantaneous_ops_per_sec:0\r\n\ + keyspace_hits:0\r\n\ + keyspace_misses:0\r\n\ + evicted_keys:0\r\n\ + expired_keys:0\r\n\ + \r\n\ + # Replication\r\n\ + role:master\r\n\ + connected_slaves:0\r\n\ + replica_id:{replica_id}\r\n\ + consistency_level:{consistency_level:?}\r\n\ + replication_enabled:{replication_enabled}\r\n\ + num_shards:{num_shards}\r\n\ + architecture:actor_per_shard\r\n\ + \r\n\ + # CPU\r\n\ + used_cpu_sys:0.000000\r\n\ + used_cpu_user:0.000000\r\n\ + used_cpu_sys_children:0.000000\r\n\ + used_cpu_user_children:0.000000\r\n\ + \r\n\ + # Keyspace\r\n", + pid = pid, + replica_id = self.config.replica_id, + consistency_level = self.config.consistency_level, + replication_enabled = self.config.enabled, + num_shards = NUM_SHARDS, ); RespValue::BulkString(Some(info.into_bytes())) } + Command::DbSize => { + // Fan out DBSIZE to all shards, sum per-shard key counts + let futures: Vec<_> = self + .shards + .iter() + .map(|shard| shard.execute_readonly(Command::DbSize)) + .collect(); + let results = futures::future::join_all(futures).await; + let total: i64 = results + .into_iter() + .filter_map(|r| { + if let RespValue::Integer(n) = r { + Some(n) + } else { + None + } + }) + .sum(); + debug_assert!(total >= 0, "Postcondition: DBSIZE must be non-negative"); + RespValue::Integer(total) + } _ => RespValue::err("ERR unknown command"), } } diff --git a/src/production/sharded_actor.rs b/src/production/sharded_actor.rs index 7612db2..df7cde6 100644 --- a/src/production/sharded_actor.rs +++ b/src/production/sharded_actor.rs @@ -775,6 +775,83 @@ impl ShardedActorState { self.adaptive_handle.as_ref() } + /// Get process memory usage (used_memory, used_memory_rss) in bytes. + /// On Linux reads from /proc/self/statm; returns (0, 0) on other platforms. + fn get_process_memory() -> (u64, u64) { + #[cfg(target_os = "linux")] + { + if let Ok(contents) = std::fs::read_to_string("/proc/self/statm") { + let parts: Vec<&str> = contents.split_whitespace().collect(); + if parts.len() >= 2 { + let page_size: u64 = 4096; + let rss_pages = parts[1].parse::().unwrap_or(0); + let total_pages = parts[0].parse::().unwrap_or(0); + let used_memory = total_pages.saturating_mul(page_size); + let used_memory_rss = rss_pages.saturating_mul(page_size); + return (used_memory, used_memory_rss); + } + } + (0, 0) + } + #[cfg(not(target_os = "linux"))] + { + (0, 0) + } + } + + /// Format bytes as human-readable string (e.g. "1.50M", "256.00K") + fn format_bytes_human(bytes: u64) -> String { + if bytes >= 1024 * 1024 * 1024 { + format!("{:.2}G", bytes as f64 / (1024.0 * 1024.0 * 1024.0)) + } else if bytes >= 1024 * 1024 { + format!("{:.2}M", bytes as f64 / (1024.0 * 1024.0)) + } else if bytes >= 1024 { + format!("{:.2}K", bytes as f64 / 1024.0) + } else { + format!("{}B", bytes) + } + } + + /// Get system CPU time in seconds (for INFO cpu section) + fn get_cpu_time_sys() -> f64 { + #[cfg(target_os = "linux")] + { + if let Ok(contents) = std::fs::read_to_string("/proc/self/stat") { + let parts: Vec<&str> = contents.split_whitespace().collect(); + // Field 14 (0-indexed) = stime (kernel mode jiffies) + if parts.len() > 14 { + let stime = parts[14].parse::().unwrap_or(0.0); + return stime / 100.0; // jiffies to seconds (assuming HZ=100) + } + } + 0.0 + } + #[cfg(not(target_os = "linux"))] + { + 0.0 + } + } + + /// Get user CPU time in seconds (for INFO cpu section) + fn get_cpu_time_user() -> f64 { + #[cfg(target_os = "linux")] + { + if let Ok(contents) = std::fs::read_to_string("/proc/self/stat") { + let parts: Vec<&str> = contents.split_whitespace().collect(); + // Field 13 (0-indexed) = utime (user mode jiffies) + if parts.len() > 13 { + let utime = parts[13].parse::().unwrap_or(0.0); + return utime / 100.0; // jiffies to seconds (assuming HZ=100) + } + } + 0.0 + } + #[cfg(not(target_os = "linux"))] + { + 0.0 + } + } + /// Get current virtual time (elapsed since creation) /// /// Uses the configured TimeSource for zero-cost abstraction: @@ -956,20 +1033,121 @@ impl ShardedActorState { Command::Info => { let adaptive_info = self.get_adaptive_info().await; + + // Compute uptime from start_millis + let now_millis = self.time_source.now_millis(); + let uptime_secs = now_millis.saturating_sub(self.start_millis) / 1000; + let uptime_days = uptime_secs / 86400; + + // Get total key count across all shards (same pattern as DbSize) + let mut dbsize_futures = Vec::with_capacity(self.num_shards); + for shard in self.shards.iter() { + dbsize_futures.push(shard.execute(Command::DbSize, virtual_time)); + } + let dbsize_results = futures::future::join_all(dbsize_futures).await; + let total_keys: i64 = dbsize_results + .into_iter() + .filter_map(|r| { + if let RespValue::Integer(n) = r { Some(n) } else { None } + }) + .sum(); + + // Process memory info + let (used_memory, used_memory_rss) = Self::get_process_memory(); + let mem_frag_ratio = if used_memory > 0 { + used_memory_rss as f64 / used_memory as f64 + } else { + 1.0 + }; + + let pid = std::process::id(); + let tcp_port = 6379; // Default Redis port (overridden by env in container) + let info = format!( "# Server\r\n\ - redis_mode:tiger_style\r\n\ - num_shards:{}\r\n\ + redis_version:7.0.0\r\n\ + redis_git_sha1:00000000\r\n\ + redis_git_dirty:0\r\n\ + redis_build_id:0\r\n\ + redis_mode:standalone\r\n\ + os:Linux\r\n\ + arch_bits:64\r\n\ + tcp_port:{tcp_port}\r\n\ + uptime_in_seconds:{uptime_secs}\r\n\ + uptime_in_days:{uptime_days}\r\n\ + hz:10\r\n\ + configured_hz:10\r\n\ + executable:/usr/local/bin/redis-rust\r\n\ + process_id:{pid}\r\n\ + num_shards:{num_shards}\r\n\ architecture:actor_message_passing\r\n\ - allocator:jemalloc\r\n\ + \r\n\ + # Clients\r\n\ + connected_clients:1\r\n\ + blocked_clients:0\r\n\ + tracking_clients:0\r\n\ + clients_in_timeout_table:0\r\n\ + \r\n\ + # Memory\r\n\ + used_memory:{used_memory}\r\n\ + used_memory_human:{used_memory_human}\r\n\ + used_memory_rss:{used_memory_rss}\r\n\ + used_memory_rss_human:{used_memory_rss_human}\r\n\ + used_memory_peak:{used_memory_rss}\r\n\ + used_memory_peak_human:{used_memory_rss_human}\r\n\ + used_memory_lua:0\r\n\ + used_memory_scripts:0\r\n\ + maxmemory:0\r\n\ + maxmemory_human:0B\r\n\ + maxmemory_policy:noeviction\r\n\ + mem_fragmentation_ratio:{mem_frag_ratio:.2}\r\n\ + mem_allocator:jemalloc\r\n\ \r\n\ # Stats\r\n\ - current_time_ms:{}\r\n\ + total_connections_received:0\r\n\ + total_commands_processed:0\r\n\ + instantaneous_ops_per_sec:0\r\n\ + total_net_input_bytes:0\r\n\ + total_net_output_bytes:0\r\n\ + keyspace_hits:0\r\n\ + keyspace_misses:0\r\n\ + evicted_keys:0\r\n\ + expired_keys:0\r\n\ + current_time_ms:{current_time_ms}\r\n\ + \r\n\ + # Replication\r\n\ + role:master\r\n\ + connected_slaves:0\r\n\ + \r\n\ + # CPU\r\n\ + used_cpu_sys:{cpu_sys:.6}\r\n\ + used_cpu_user:{cpu_user:.6}\r\n\ + used_cpu_sys_children:0.000000\r\n\ + used_cpu_user_children:0.000000\r\n\ \r\n\ - {}", - self.num_shards, - virtual_time.as_millis(), - adaptive_info + # Keyspace\r\n\ + {keyspace}\ + \r\n\ + {adaptive}", + tcp_port = tcp_port, + uptime_secs = uptime_secs, + uptime_days = uptime_days, + pid = pid, + num_shards = self.num_shards, + used_memory = used_memory, + used_memory_human = Self::format_bytes_human(used_memory), + used_memory_rss = used_memory_rss, + used_memory_rss_human = Self::format_bytes_human(used_memory_rss), + mem_frag_ratio = mem_frag_ratio, + current_time_ms = virtual_time.as_millis(), + cpu_sys = Self::get_cpu_time_sys(), + cpu_user = Self::get_cpu_time_user(), + keyspace = if total_keys > 0 { + format!("db0:keys={},expires=0,avg_ttl=0\r\n", total_keys) + } else { + String::new() + }, + adaptive = adaptive_info, ); RespValue::BulkString(Some(info.into_bytes())) } diff --git a/src/replication/config.rs b/src/replication/config.rs index 2de1ba6..2e601bd 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -53,7 +53,7 @@ impl Default for ReplicationConfig { enabled: false, replica_id: 1, consistency_level: ConsistencyLevel::Eventual, - gossip_interval_ms: 100, + gossip_interval_ms: 1000, peers: Vec::new(), replication_factor: 3, // Partitioning disabled by default (backward compatible) @@ -76,7 +76,7 @@ impl ReplicationConfig { enabled: true, replica_id, consistency_level: ConsistencyLevel::Eventual, - gossip_interval_ms: 100, + gossip_interval_ms: 1000, peers, replication_factor: 3, partitioned_mode: false, @@ -95,7 +95,7 @@ impl ReplicationConfig { enabled: true, replica_id, consistency_level: ConsistencyLevel::Eventual, - gossip_interval_ms: 100, + gossip_interval_ms: 1000, peers, replication_factor, partitioned_mode: true, diff --git a/src/replication/gossip.rs b/src/replication/gossip.rs index b9e2656..adb1b7f 100644 --- a/src/replication/gossip.rs +++ b/src/replication/gossip.rs @@ -5,6 +5,13 @@ use super::state::ReplicationDelta; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use tokio::sync::mpsc; +use tracing::warn; + +/// Maximum number of messages allowed in the outbound queue. +/// If the queue exceeds this limit, the oldest messages are dropped to prevent +/// unbounded memory growth under sustained high write load. +/// 10,000 messages * ~1KB avg = ~10MB worst case, well within safe limits. +pub const MAX_OUTBOUND_QUEUE: usize = 10_000; #[derive(Debug, Clone, Serialize, Deserialize)] pub enum GossipMessage { @@ -209,10 +216,38 @@ impl GossipState { } pub fn advance_epoch(&mut self) { - self.epoch += 1; + self.epoch = self.epoch.saturating_add(1); } - /// Queue deltas for gossip - uses selective routing if router is configured + /// Enforce the outbound queue capacity limit. + /// If the queue exceeds `MAX_OUTBOUND_QUEUE`, drop the oldest messages + /// (front of the Vec) to bring it back within bounds. + /// + /// ## Postcondition + /// `self.outbound_queue.len() <= MAX_OUTBOUND_QUEUE` + fn enforce_outbound_capacity(&mut self) { + if self.outbound_queue.len() > MAX_OUTBOUND_QUEUE { + let overflow = self.outbound_queue.len().saturating_sub(MAX_OUTBOUND_QUEUE); + warn!( + "Outbound gossip queue exceeded capacity ({} > {}), dropping {} oldest messages", + self.outbound_queue.len(), + MAX_OUTBOUND_QUEUE, + overflow + ); + // Drop oldest messages from the front + self.outbound_queue.drain(..overflow); + + debug_assert!( + self.outbound_queue.len() <= MAX_OUTBOUND_QUEUE, + "Postcondition: outbound queue must be within capacity after truncation" + ); + } + } + + /// Queue deltas for gossip - uses selective routing if router is configured. + /// + /// Enforces `MAX_OUTBOUND_QUEUE` capacity after queueing. If the queue is + /// already at capacity, oldest messages are dropped to make room. pub fn queue_deltas(&mut self, deltas: Vec) { if deltas.is_empty() { return; @@ -234,6 +269,7 @@ impl GossipState { .push(RoutedMessage::targeted(target_replica, msg)); } } + self.enforce_outbound_capacity(); return; } } @@ -241,19 +277,23 @@ impl GossipState { // Fallback: broadcast to all peers let msg = GossipMessage::new_delta_batch(self.replica_id, deltas, self.epoch); self.outbound_queue.push(RoutedMessage::broadcast(msg)); + self.enforce_outbound_capacity(); } - /// Queue deltas using broadcast (ignore router) + /// Queue deltas using broadcast (ignore router). + /// Enforces `MAX_OUTBOUND_QUEUE` capacity after queueing. pub fn queue_deltas_broadcast(&mut self, deltas: Vec) { if !deltas.is_empty() { let msg = GossipMessage::new_delta_batch(self.replica_id, deltas, self.epoch); self.outbound_queue.push(RoutedMessage::broadcast(msg)); + self.enforce_outbound_capacity(); } } pub fn queue_heartbeat(&mut self) { let msg = GossipMessage::new_heartbeat(self.replica_id, self.epoch); self.outbound_queue.push(RoutedMessage::broadcast(msg)); + self.enforce_outbound_capacity(); } pub fn drain_outbound(&mut self) -> Vec { @@ -273,3 +313,197 @@ impl GossipState { self.gossip_router.as_ref() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::redis::SDS; + use crate::replication::lattice::LamportClock; + use crate::replication::state::{ReplicatedValue, ReplicationDelta}; + + fn test_config() -> ReplicationConfig { + ReplicationConfig { + replica_id: 1, + peers: vec![], + enabled: true, + ..Default::default() + } + } + + fn make_delta(key: &str) -> ReplicationDelta { + let replica_id = ReplicaId::new(1); + let clock = LamportClock::new(replica_id); + let value = ReplicatedValue::with_value(SDS::from_str(key), clock); + ReplicationDelta::new(key.to_string(), value, replica_id) + } + + #[test] + fn test_outbound_queue_bounded_at_max() { + let mut state = GossipState::new(test_config()); + + // Push MAX_OUTBOUND_QUEUE + 500 messages + let overflow_count = 500usize; + let total = MAX_OUTBOUND_QUEUE.checked_add(overflow_count).unwrap(); + for i in 0..total { + let delta = make_delta(&format!("key-{}", i)); + // Bypass enforce_outbound_capacity by pushing directly, + // then call it explicitly to test the truncation logic + let msg = GossipMessage::new_delta_batch(state.replica_id, vec![delta], state.epoch); + state.outbound_queue.push(RoutedMessage::broadcast(msg)); + } + + assert_eq!(state.outbound_queue.len(), total); + + // Now enforce capacity + state.enforce_outbound_capacity(); + + assert_eq!( + state.outbound_queue.len(), + MAX_OUTBOUND_QUEUE, + "Queue must be truncated to MAX_OUTBOUND_QUEUE" + ); + } + + #[test] + fn test_queue_deltas_enforces_capacity() { + let mut state = GossipState::new(test_config()); + + // Fill to just under the limit by pushing directly + for i in 0..MAX_OUTBOUND_QUEUE { + let msg = GossipMessage::new_heartbeat(state.replica_id, state.epoch); + state.outbound_queue.push(RoutedMessage::broadcast(msg)); + // Skip enforce to pre-fill + let _ = i; + } + + assert_eq!(state.outbound_queue.len(), MAX_OUTBOUND_QUEUE); + + // Now queue_deltas should push one more and then truncate + let delta = make_delta("overflow-key"); + state.queue_deltas(vec![delta]); + + assert!( + state.outbound_queue.len() <= MAX_OUTBOUND_QUEUE, + "queue_deltas must enforce capacity: got {}", + state.outbound_queue.len() + ); + } + + #[test] + fn test_queue_deltas_broadcast_enforces_capacity() { + let mut state = GossipState::new(test_config()); + + // Fill to the limit + for _ in 0..MAX_OUTBOUND_QUEUE { + let msg = GossipMessage::new_heartbeat(state.replica_id, state.epoch); + state.outbound_queue.push(RoutedMessage::broadcast(msg)); + } + + let delta = make_delta("overflow-broadcast"); + state.queue_deltas_broadcast(vec![delta]); + + assert!( + state.outbound_queue.len() <= MAX_OUTBOUND_QUEUE, + "queue_deltas_broadcast must enforce capacity: got {}", + state.outbound_queue.len() + ); + } + + #[test] + fn test_queue_heartbeat_enforces_capacity() { + let mut state = GossipState::new(test_config()); + + // Fill to the limit + for _ in 0..MAX_OUTBOUND_QUEUE { + let msg = GossipMessage::new_heartbeat(state.replica_id, state.epoch); + state.outbound_queue.push(RoutedMessage::broadcast(msg)); + } + + state.queue_heartbeat(); + + assert!( + state.outbound_queue.len() <= MAX_OUTBOUND_QUEUE, + "queue_heartbeat must enforce capacity: got {}", + state.outbound_queue.len() + ); + } + + #[test] + fn test_drain_outbound_clears_queue() { + let mut state = GossipState::new(test_config()); + let delta = make_delta("drain-key"); + state.queue_deltas(vec![delta]); + + assert!(!state.outbound_queue.is_empty()); + + let messages = state.drain_outbound(); + assert!(!messages.is_empty()); + assert!(state.outbound_queue.is_empty(), "Queue must be empty after drain"); + } + + #[test] + fn test_advance_epoch_saturates() { + let mut state = GossipState::new(test_config()); + state.epoch = u64::MAX; + state.advance_epoch(); + assert_eq!(state.epoch, u64::MAX, "Epoch must saturate at u64::MAX, not wrap"); + } + + #[test] + fn test_empty_deltas_not_queued() { + let mut state = GossipState::new(test_config()); + state.queue_deltas(vec![]); + assert!( + state.outbound_queue.is_empty(), + "Empty deltas should not produce outbound messages" + ); + } + + #[test] + fn test_capacity_drops_oldest_preserves_newest() { + let mut state = GossipState::new(test_config()); + + // Push MAX + 5 messages, each with a distinct epoch to identify ordering + let total = MAX_OUTBOUND_QUEUE.checked_add(5).unwrap(); + for i in 0..total { + state.epoch = i as u64; + let msg = GossipMessage::new_heartbeat(state.replica_id, state.epoch); + state.outbound_queue.push(RoutedMessage::broadcast(msg)); + } + + state.enforce_outbound_capacity(); + + assert_eq!(state.outbound_queue.len(), MAX_OUTBOUND_QUEUE); + + // The first message in the queue should be the one at index 5 (oldest 5 dropped) + if let GossipMessage::Heartbeat { epoch, .. } = &state.outbound_queue[0].message { + assert_eq!( + *epoch, 5, + "Oldest messages (epochs 0-4) should be dropped, first remaining should be epoch 5" + ); + } else { + panic!("Expected Heartbeat message"); + } + + // The last message should be the newest + let last_idx = state.outbound_queue.len().checked_sub(1).unwrap(); + if let GossipMessage::Heartbeat { epoch, .. } = &state.outbound_queue[last_idx].message { + let expected_last = total.checked_sub(1).unwrap() as u64; + assert_eq!( + *epoch, expected_last, + "Newest message should be preserved at the end" + ); + } else { + panic!("Expected Heartbeat message"); + } + } + + #[test] + fn test_verify_invariants_on_gossip_state() { + let mut state = GossipState::new(test_config()); + let delta = make_delta("invariant-test"); + state.queue_deltas(vec![delta]); + // Should not panic in debug mode + state.verify_invariants(); + } +} diff --git a/src/replication/state/shard_state.rs b/src/replication/state/shard_state.rs index e57b003..6e2cb70 100644 --- a/src/replication/state/shard_state.rs +++ b/src/replication/state/shard_state.rs @@ -8,6 +8,12 @@ use crate::replication::config::ConsistencyLevel; use crate::replication::lattice::{LamportClock, ReplicaId, VectorClock}; use std::collections::HashMap; +/// Maximum number of pending deltas before oldest are dropped. +/// Kept small because each delta clones the full ReplicatedValue — for hashes +/// with many fields, a single delta can be 500KB+. At 100 entries, worst case +/// is ~50MB which is acceptable. +const MAX_PENDING_DELTAS: usize = 100; + #[derive(Debug)] pub struct ShardReplicaState { pub replica_id: ReplicaId, @@ -53,6 +59,7 @@ impl ShardReplicaState { let delta = ReplicationDelta::new(key.clone(), replicated.clone(), self.replica_id); self.replicated_keys.insert(key, replicated); self.pending_deltas.push(delta.clone()); + self.enforce_pending_capacity(); delta } @@ -62,6 +69,7 @@ impl ShardReplicaState { let delta = ReplicationDelta::new(key.clone(), replicated.clone(), self.replica_id); self.replicated_keys.insert(key, replicated); self.pending_deltas.push(delta.clone()); + self.enforce_pending_capacity(); Some(delta) } else { None @@ -103,6 +111,7 @@ impl ShardReplicaState { let delta = ReplicationDelta::new(key.clone(), replicated.clone(), self.replica_id); self.replicated_keys.insert(key.clone(), replicated); self.pending_deltas.push(delta.clone()); + self.enforce_pending_capacity(); // TigerStyle: Postconditions #[cfg(debug_assertions)] @@ -120,10 +129,9 @@ impl ShardReplicaState { "Postcondition: key '{}' must be a hash type", key ); - debug_assert_eq!( - self.pending_deltas.len(), - pre_pending_len + 1, - "Postcondition: pending_deltas must increase by 1" + debug_assert!( + self.pending_deltas.len() <= MAX_PENDING_DELTAS, + "Postcondition: pending_deltas must be within capacity after hash_write" ); // Verify all fields were set if let Some(rv) = self.replicated_keys.get(&key) { @@ -163,6 +171,7 @@ impl ShardReplicaState { let delta = ReplicationDelta::new(key.clone(), replicated.clone(), self.replica_id); self.replicated_keys.insert(key.clone(), replicated); self.pending_deltas.push(delta.clone()); + self.enforce_pending_capacity(); // TigerStyle: Postconditions #[cfg(debug_assertions)] @@ -172,10 +181,9 @@ impl ShardReplicaState { "Postcondition: key '{}' must exist in replicated_keys", key ); - debug_assert_eq!( - self.pending_deltas.len(), - pre_pending_len + 1, - "Postcondition: pending_deltas must increase by 1" + debug_assert!( + self.pending_deltas.len() <= MAX_PENDING_DELTAS, + "Postcondition: pending_deltas must be within capacity after hash_delete" ); // Verify all fields were tombstoned if let Some(rv) = self.replicated_keys.get(&key) { @@ -213,6 +221,27 @@ impl ShardReplicaState { std::mem::take(&mut self.pending_deltas) } + /// Enforce capacity limit on pending_deltas to prevent unbounded memory growth. + /// + /// This is the critical fix for the base memory leak: every write operation + /// pushes a ReplicationDelta clone (~200-500 bytes) into pending_deltas. + /// If the gossip loop doesn't drain them (non-replicated pods, or collect_deltas + /// returns empty), memory grows at ~200-300Mi/min under load. + /// + /// When the limit is exceeded, oldest deltas are dropped (they would have been + /// superseded by newer writes anyway for the same keys). + fn enforce_pending_capacity(&mut self) { + if self.pending_deltas.len() > MAX_PENDING_DELTAS { + let overflow = self.pending_deltas.len().saturating_sub(MAX_PENDING_DELTAS); + self.pending_deltas.drain(..overflow); + + debug_assert!( + self.pending_deltas.len() <= MAX_PENDING_DELTAS, + "Postcondition: pending_deltas must be within capacity" + ); + } + } + pub fn get_replicated(&self, key: &str) -> Option<&ReplicatedValue> { self.replicated_keys.get(key) } diff --git a/src/streaming/integration.rs b/src/streaming/integration.rs index 05ebbc6..5b811fe 100644 --- a/src/streaming/integration.rs +++ b/src/streaming/integration.rs @@ -464,16 +464,17 @@ pub enum PersistenceMessage { Shutdown { response_tx: oneshot::Sender<()> }, } -/// Actor that owns StreamingPersistence exclusively +/// Actor that owns StreamingPersistence exclusively. +/// Uses bounded channel (PERSISTENCE_CHANNEL_CAPACITY) to prevent unbounded memory growth. struct PersistenceActor { persistence: StreamingPersistence, - rx: mpsc::UnboundedReceiver, + rx: mpsc::Receiver, } impl PersistenceActor { fn new( persistence: StreamingPersistence, - rx: mpsc::UnboundedReceiver, + rx: mpsc::Receiver, ) -> Self { PersistenceActor { persistence, rx } } @@ -539,23 +540,26 @@ impl PersistenceActor { } /// Handle for communicating with the persistence actor +/// +/// Uses a bounded channel (PERSISTENCE_CHANNEL_CAPACITY) to prevent unbounded +/// memory growth. When the channel is full, fire-and-forget sends silently drop. #[derive(Clone)] pub struct PersistenceActorHandle { - tx: mpsc::UnboundedSender, + tx: mpsc::Sender, } impl PersistenceActorHandle { - /// Push a delta (fire-and-forget) + /// Push a delta (fire-and-forget, drops if channel full) #[inline] pub fn push_delta(&self, delta: ReplicationDelta) { - let _ = self.tx.send(PersistenceMessage::PushDelta(delta)); + let _ = self.tx.try_send(PersistenceMessage::PushDelta(delta)); } - /// Push multiple deltas (fire-and-forget) + /// Push multiple deltas (fire-and-forget, drops if channel full) #[inline] pub fn push_deltas(&self, deltas: Vec) { if !deltas.is_empty() { - let _ = self.tx.send(PersistenceMessage::PushDeltas(deltas)); + let _ = self.tx.try_send(PersistenceMessage::PushDeltas(deltas)); } } @@ -564,10 +568,10 @@ impl PersistenceActorHandle { let (response_tx, response_rx) = oneshot::channel(); if self .tx - .send(PersistenceMessage::Flush { response_tx }) + .try_send(PersistenceMessage::Flush { response_tx }) .is_err() { - return Err("Persistence actor unavailable".to_string()); + return Err("Persistence actor unavailable or channel full".to_string()); } response_rx .await @@ -577,7 +581,7 @@ impl PersistenceActorHandle { /// Send periodic tick (fire-and-forget) #[inline] pub fn tick(&self) { - let _ = self.tx.send(PersistenceMessage::Tick); + let _ = self.tx.try_send(PersistenceMessage::Tick); } /// Shutdown the actor gracefully @@ -585,7 +589,7 @@ impl PersistenceActorHandle { let (response_tx, response_rx) = oneshot::channel(); if self .tx - .send(PersistenceMessage::Shutdown { response_tx }) + .try_send(PersistenceMessage::Shutdown { response_tx }) .is_ok() { let _ = response_rx.await; @@ -593,11 +597,21 @@ impl PersistenceActorHandle { } } +/// Maximum number of messages in the persistence actor's channel. +/// This prevents unbounded memory growth when the actor can't flush fast enough +/// (slow disk, backpressure, or Permission denied errors). Each message is a +/// PushDelta(s) containing ~500 bytes per delta. At 10K capacity: ~5MB worst case. +/// +/// When the channel is full, new messages are silently dropped (fire-and-forget +/// semantics). The data is still in the replicated state for reads — persistence +/// is best-effort durability, not the source of truth. +const PERSISTENCE_CHANNEL_CAPACITY: usize = 10_000; + /// Spawn the persistence actor and return a handle fn spawn_persistence_actor( persistence: StreamingPersistence, ) -> (PersistenceActorHandle, JoinHandle<()>) { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::channel(PERSISTENCE_CHANNEL_CAPACITY); let actor = PersistenceActor::new(persistence, rx); let task = tokio::spawn(actor.run()); (PersistenceActorHandle { tx }, task)