diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 5c077193..d7442e89 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1356,6 +1356,15 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + [[package]] name = "num-bigint" version = "0.4.8" @@ -2063,7 +2072,7 @@ dependencies = [ "soroban-env-macros", "soroban-wasmi", "static_assertions", - "stellar-xdr 27.0.0", + "stellar-xdr", "wasmparser", ] @@ -2125,7 +2134,7 @@ dependencies = [ "quote", "serde", "serde_json", - "stellar-xdr 27.0.0", + "stellar-xdr", "syn 2.0.119", ] @@ -2183,7 +2192,7 @@ dependencies = [ "soroban-env-common", "soroban-spec", "soroban-spec-rust", - "stellar-xdr 27.0.0", + "stellar-xdr", "syn 2.0.119", ] @@ -2195,7 +2204,7 @@ checksum = "473404322827b285cbcd87517f365986bd63af7842c78b2a86ee061715fda61e" dependencies = [ "base64", "sha2", - "stellar-xdr 27.0.0", + "stellar-xdr", "thiserror 1.0.69", "wasmparser", ] @@ -2211,7 +2220,7 @@ dependencies = [ "quote", "sha2", "soroban-spec", - "stellar-xdr 27.0.0", + "stellar-xdr", "syn 2.0.119", "thiserror 1.0.69", ] @@ -2283,13 +2292,14 @@ dependencies = [ "serde_json", "soroban-sdk", "stellar-insights", - "stellar-xdr 28.0.0", + "stellar-xdr", "thiserror 2.0.20", "tokio", "tokio-tungstenite", "tracing", "tracing-futures", "tracing-opentelemetry", + "tracing-subscriber", ] [[package]] @@ -2332,20 +2342,6 @@ dependencies = [ "stellar-strkey 0.0.13", ] -[[package]] -name = "stellar-xdr" -version = "28.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f93d09ff8b9f919b084f664003c4c546ac66a76affd5429460dbe29f4b326f8e" -dependencies = [ - "crate-git-revision 0.0.9", - "escape-bytes", - "ethnum", - "hex", - "sha2", - "stellar-strkey 0.0.13", -] - [[package]] name = "strsim" version = "0.11.1" @@ -2589,9 +2585,12 @@ version = "0.3.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" dependencies = [ + "nu-ansi-term", "sharded-slab", + "smallvec", "thread_local", "tracing-core", + "tracing-log", ] [[package]] diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 5f15f0b4..6064f712 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -17,7 +17,7 @@ lazy_static = "1.4" async-trait = "0.1" chrono = "0.4" hex = "0.4" -stellar-xdr = "=28.0.0" +stellar-xdr = { version = "=27.0.0", default-features = false, features = ["alloc", "std"] } tracing = "0.1.44" tracing-futures = "0.2.5" opentelemetry = "0.32.0" @@ -28,3 +28,4 @@ opentelemetry-stdout = "0.32.0" [dev-dependencies] soroban-sdk = { version = "=27.0.6", features = ["testutils"] } stellar-insights = { path = "../contracts/stellar_insights", features = ["testutils"] } +tracing-subscriber = { version = "0.3", features = ["registry"] } diff --git a/backend/src/realtime/connection.rs b/backend/src/realtime/connection.rs index 47581f72..daee350d 100644 --- a/backend/src/realtime/connection.rs +++ b/backend/src/realtime/connection.rs @@ -1,48 +1,209 @@ -use tokio::sync::mpsc::{self, UnboundedSender}; -use tokio_tungstenite::tungstenite::protocol::Message; -use tokio_tungstenite::{connect_async, tungstenite::Error}; -use futures_util::{SinkExt, StreamExt}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; -use tokio::sync::Mutex; -use crate::realtime::policy::handle_overflow; +use tokio::sync::{mpsc::{self, UnboundedReceiver, UnboundedSender}, Mutex}; +use tokio_tungstenite::tungstenite::protocol::Message; + +use crate::realtime::policy::{ + ConnectionHealthState, ConnectionPolicyTracker, PolicyDecision, QueueDepthTrend, + SlowConsumerPolicyConfig, +}; pub type ConnectionId = String; +/// Active WebSocket client connection with isolated queue depth tracking and proactive slow-consumer policy. pub struct Connection { pub id: ConnectionId, pub sender: UnboundedSender, - pub queue_depth: Arc>, + pub queue_depth: Arc, + pub policy_tracker: Arc>, } impl Connection { + /// Creates a new connection with default policy and unbound channel. pub fn new(id: ConnectionId) -> Self { + Self::with_policy(id, SlowConsumerPolicyConfig::default()) + } + + /// Creates a new connection with specified policy configuration. + pub fn with_policy(id: ConnectionId, config: SlowConsumerPolicyConfig) -> Self { let (sender, _receiver) = mpsc::unbounded_channel(); Self { id, sender, - queue_depth: Arc::new(Mutex::new(0)), + queue_depth: Arc::new(AtomicUsize::new(0)), + policy_tracker: Arc::new(Mutex::new(ConnectionPolicyTracker::new(config))), + } + } + + /// Creates a connection wrapping an existing sender. + pub fn new_with_sender( + id: ConnectionId, + sender: UnboundedSender, + config: SlowConsumerPolicyConfig, + ) -> Self { + Self { + id, + sender, + queue_depth: Arc::new(AtomicUsize::new(0)), + policy_tracker: Arc::new(Mutex::new(ConnectionPolicyTracker::new(config))), } } + /// Creates a connected pair of `(Connection, ConnectionReceiver)` with automatic queue depth accounting. + pub fn create_channel(id: ConnectionId) -> (Self, ConnectionReceiver) { + Self::create_channel_with_policy(id, SlowConsumerPolicyConfig::default()) + } + + /// Creates a connected pair with custom policy. + pub fn create_channel_with_policy( + id: ConnectionId, + config: SlowConsumerPolicyConfig, + ) -> (Self, ConnectionReceiver) { + let (sender, receiver) = mpsc::unbounded_channel(); + let queue_depth = Arc::new(AtomicUsize::new(0)); + let policy_tracker = Arc::new(Mutex::new(ConnectionPolicyTracker::new(config))); + + let connection = Self { + id: id.clone(), + sender, + queue_depth: queue_depth.clone(), + policy_tracker: policy_tracker.clone(), + }; + + let connection_receiver = ConnectionReceiver { + id, + receiver, + queue_depth, + policy_tracker, + }; + + (connection, connection_receiver) + } + + /// Non-blocking, isolated send method evaluated against the proactive slow-consumer policy. pub async fn send(&self, message: Message) -> Result<(), String> { - let mut depth = self.queue_depth.lock().await; - *depth += 1; + let current_depth = self.queue_depth.load(Ordering::Relaxed); - // If queue depth exceeds a threshold, apply overflow policy - if *depth > 100 { - let _ = handle_overflow(&self.id).await; - return Err("Queue full".to_string()); - } + let decision = { + let mut tracker = self.policy_tracker.lock().await; + tracker.record_queue_depth(current_depth) + }; - match self.sender.send(message) { - Ok(_) => { - *depth -= 1; - Ok(()) + match decision { + PolicyDecision::Allow | PolicyDecision::AllowDegraded => { + self.queue_depth.fetch_add(1, Ordering::SeqCst); + match self.sender.send(message) { + Ok(_) => Ok(()), + Err(_) => { + self.queue_depth.fetch_sub(1, Ordering::SeqCst); + let mut tracker = self.policy_tracker.lock().await; + tracker.record_drop(); + Err("Failed to send message: receiver disconnected".to_string()) + } + } } - Err(_) => { - *depth -= 1; - Err("Failed to send message".to_string()) + PolicyDecision::QuarantineDrop => { + Err("Connection quarantined: message dropped to protect realtime fanout latency".to_string()) } + PolicyDecision::Evict => { + Err("Connection evicted: exceeded queue depth limit".to_string()) + } + } + } + + /// Current queue depth. + pub fn queue_depth(&self) -> usize { + self.queue_depth.load(Ordering::Relaxed) + } + + /// Manually decrement queue depth when a message is processed externally. + pub fn decrement_queue_depth(&self) { + if self.queue_depth.load(Ordering::Relaxed) > 0 { + self.queue_depth.fetch_sub(1, Ordering::SeqCst); } } + + /// Returns current health state. + pub async fn health_state(&self) -> ConnectionHealthState { + let tracker = self.policy_tracker.lock().await; + tracker.health_state() + } + + /// Returns whether the connection is currently quarantined. + pub async fn is_quarantined(&self) -> bool { + let tracker = self.policy_tracker.lock().await; + tracker.is_quarantined() + } + + /// Returns whether the connection should be evicted from the registry. + pub async fn should_evict(&self) -> bool { + let tracker = self.policy_tracker.lock().await; + tracker.should_evict() + } + + /// Returns the trend in queue depth. + pub async fn trend(&self) -> QueueDepthTrend { + let tracker = self.policy_tracker.lock().await; + tracker.calculate_trend() + } +} + +/// Paired receiver that drains messages and automatically decrements the queue depth tracker. +pub struct ConnectionReceiver { + pub id: ConnectionId, + pub receiver: UnboundedReceiver, + pub queue_depth: Arc, + pub policy_tracker: Arc>, +} + +impl ConnectionReceiver { + /// Asynchronously receives the next message, draining queue depth and recording telemetry. + pub async fn recv(&mut self) -> Option { + let msg = self.receiver.recv().await?; + if self.queue_depth.load(Ordering::Relaxed) > 0 { + self.queue_depth.fetch_sub(1, Ordering::SeqCst); + } + let current_depth = self.queue_depth.load(Ordering::Relaxed); + let mut tracker = self.policy_tracker.lock().await; + tracker.record_drain(1); + tracker.record_queue_depth(current_depth); + Some(msg) + } + + /// Non-blocking receive attempt. + pub fn try_recv(&mut self) -> Result { + let msg = self.receiver.try_recv()?; + if self.queue_depth.load(Ordering::Relaxed) > 0 { + self.queue_depth.fetch_sub(1, Ordering::SeqCst); + } + Ok(msg) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_connection_send_and_receive() { + let (conn, mut rx) = Connection::create_channel("test-conn".to_string()); + assert_eq!(conn.queue_depth(), 0); + + conn.send(Message::text("hello")).await.unwrap(); + assert_eq!(conn.queue_depth(), 1); + + let received = rx.recv().await.unwrap(); + assert_eq!(received, Message::text("hello")); + assert_eq!(conn.queue_depth(), 0); + } + + #[tokio::test] + async fn test_connection_eviction_on_receiver_drop() { + let (conn, rx) = Connection::create_channel("test-conn".to_string()); + drop(rx); + + let result = conn.send(Message::text("hello")).await; + assert!(result.is_err()); + assert!(conn.should_evict().await); + } } \ No newline at end of file diff --git a/backend/src/realtime/fanout.rs b/backend/src/realtime/fanout.rs index 1b7a609d..8dd75e7b 100644 --- a/backend/src/realtime/fanout.rs +++ b/backend/src/realtime/fanout.rs @@ -1,25 +1,186 @@ -use crate::realtime::ConnectionRegistry; +use std::sync::Arc; +use std::time::Duration; +use futures_util::stream::{self, StreamExt}; use tokio_tungstenite::tungstenite::protocol::Message; + use crate::realtime::policy::handle_overflow; +use crate::realtime::{Connection, ConnectionId, ConnectionRegistry}; + +/// Configuration parameters for realtime message fanout. +#[derive(Debug, Clone)] +pub struct FanoutConfig { + /// Maximum concurrent sends allowed in flight. + pub max_concurrency: usize, + /// Per-connection timeout for delivery before isolating. + pub per_connection_timeout: Option, +} + +impl Default for FanoutConfig { + fn default() -> Self { + Self { + max_concurrency: 64, + per_connection_timeout: Some(Duration::from_millis(50)), + } + } +} +/// Telemetry summary returned after a fanout operation completes. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct FanoutSummary { + pub total_connections: usize, + pub delivered: usize, + pub quarantined_dropped: usize, + pub failed_evicted: usize, +} + +enum DeliveryResult { + Delivered, + QuarantinedDropped, + Evict(ConnectionId), +} + +/// Fans out a message to all connected clients in the registry with bounded concurrency and per-connection isolation. pub async fn fanout_message( registry: &ConnectionRegistry, message: Message, ) -> Result<(), String> { - let connections = registry.get_all().await; + fanout_message_with_concurrency(registry, message, 64).await +} + +/// Fans out a message with an explicit concurrency bound. +pub async fn fanout_message_with_concurrency( + registry: &ConnectionRegistry, + message: Message, + max_concurrency: usize, +) -> Result<(), String> { + let config = FanoutConfig { + max_concurrency, + per_connection_timeout: None, + }; + fanout_message_detailed(registry, &message, &config) + .await + .map(|_| ()) +} + +/// Detailed message fanout returning delivery metrics and offloading slow-consumer eviction to a background task. +pub async fn fanout_message_detailed( + registry: &ConnectionRegistry, + message: &Message, + config: &FanoutConfig, +) -> Result { + let connections = registry.get_all_connections().await; + let total_connections = connections.len(); + + if total_connections == 0 { + return Ok(FanoutSummary::default()); + } + + let concurrency = config.max_concurrency.max(1); + + // Deliver concurrently across connections bounded by `concurrency` + let delivery_results: Vec = stream::iter(connections) + .map(|conn: Arc| { + let msg = message.clone(); + let timeout_opt = config.per_connection_timeout; + async move { + let send_future = conn.send(msg); + + let result = match timeout_opt { + Some(timeout) => match tokio::time::timeout(timeout, send_future).await { + Ok(send_res) => send_res, + Err(_) => Err("Send timed out".to_string()), + }, + None => send_future.await, + }; - for (id, sender) in connections { - match sender.send(message.clone()) { - Ok(_) => { - // Successfully sent to this connection + match result { + Ok(_) => DeliveryResult::Delivered, + Err(_) => { + let should_evict = conn.should_evict().await; + if should_evict { + DeliveryResult::Evict(conn.id.clone()) + } else if conn.is_quarantined().await { + DeliveryResult::QuarantinedDropped + } else { + DeliveryResult::Evict(conn.id.clone()) + } + } + } } - Err(_) => { - // If the connection's queue is full, handle overflow - let _ = handle_overflow(&id).await; - registry.remove(&id).await; + }) + .buffer_unordered(concurrency) + .collect() + .await; + + let mut summary = FanoutSummary { + total_connections, + delivered: 0, + quarantined_dropped: 0, + failed_evicted: 0, + }; + + let mut to_evict = Vec::new(); + + for result in delivery_results { + match result { + DeliveryResult::Delivered => summary.delivered += 1, + DeliveryResult::QuarantinedDropped => summary.quarantined_dropped += 1, + DeliveryResult::Evict(id) => { + summary.failed_evicted += 1; + to_evict.push(id); } } } - Ok(()) + // Offload eviction and overflow handling off the hot path + if !to_evict.is_empty() { + let registry_clone = registry.clone(); + tokio::spawn(async move { + for id in &to_evict { + let _ = handle_overflow(id).await; + } + registry_clone.remove_batch(&to_evict).await; + }); + } + + Ok(summary) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_empty_fanout() { + let registry = ConnectionRegistry::new(); + let summary = fanout_message_detailed(®istry, &Message::text("test"), &FanoutConfig::default()) + .await + .unwrap(); + assert_eq!(summary, FanoutSummary::default()); + } + + #[tokio::test] + async fn test_healthy_fanout() { + let registry = ConnectionRegistry::new(); + let mut rxs = Vec::new(); + + for i in 0..10 { + let (conn, rx) = Connection::create_channel(format!("conn-{}", i)); + registry.add_connection(Arc::new(conn)).await; + rxs.push(rx); + } + + let summary = fanout_message_detailed(®istry, &Message::text("hello"), &FanoutConfig::default()) + .await + .unwrap(); + + assert_eq!(summary.total_connections, 10); + assert_eq!(summary.delivered, 10); + assert_eq!(summary.failed_evicted, 0); + + for mut rx in rxs { + let msg = rx.recv().await.unwrap(); + assert_eq!(msg, Message::text("hello")); + } + } } \ No newline at end of file diff --git a/backend/src/realtime/mod.rs b/backend/src/realtime/mod.rs index 91a699d7..a6606807 100644 --- a/backend/src/realtime/mod.rs +++ b/backend/src/realtime/mod.rs @@ -2,38 +2,105 @@ pub mod connection; pub mod fanout; pub mod policy; -use std::sync::Arc; -use tokio::sync::Mutex; +pub use connection::{Connection, ConnectionReceiver}; +pub use fanout::{ + fanout_message, fanout_message_detailed, fanout_message_with_concurrency, FanoutConfig, + FanoutSummary, +}; +pub use policy::{ + ConnectionHealthState, ConnectionPolicyTracker, PolicyDecision, QueueDepthTrend, + SlowConsumerPolicyConfig, +}; + use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::RwLock; use tokio_tungstenite::tungstenite::protocol::Message; pub type ConnectionId = String; pub type ConnectionSender = tokio::sync::mpsc::UnboundedSender; +/// Thread-safe registry of active WebSocket client connections. #[derive(Clone)] pub struct ConnectionRegistry { - pub connections: Arc>>, + pub connections: Arc>>>, +} + +impl Default for ConnectionRegistry { + fn default() -> Self { + Self::new() + } } impl ConnectionRegistry { + /// Creates an empty connection registry. pub fn new() -> Self { Self { - connections: Arc::new(Mutex::new(HashMap::new())), + connections: Arc::new(RwLock::new(HashMap::new())), } } + /// Registers a connection by raw sender channel, wrapping it in a managed `Connection`. pub async fn add(&self, id: ConnectionId, sender: ConnectionSender) { - let mut guard = self.connections.lock().await; - guard.insert(id, sender); + let conn = Arc::new(Connection::new_with_sender( + id.clone(), + sender, + SlowConsumerPolicyConfig::default(), + )); + let mut guard = self.connections.write().await; + guard.insert(id, conn); + } + + /// Registers a pre-configured `Arc`. + pub async fn add_connection(&self, connection: Arc) { + let mut guard = self.connections.write().await; + guard.insert(connection.id.clone(), connection); } - pub async fn remove(&self, id: &str) { - let mut guard = self.connections.lock().await; - guard.remove(id); + /// Removes a connection by ID. + pub async fn remove(&self, id: &str) -> Option> { + let mut guard = self.connections.write().await; + guard.remove(id) } + /// Removes a batch of connections in a single write-lock acquisition. + pub async fn remove_batch(&self, ids: &[ConnectionId]) { + let mut guard = self.connections.write().await; + for id in ids { + guard.remove(id); + } + } + + /// Retrieves an individual connection if registered. + pub async fn get(&self, id: &str) -> Option> { + let guard = self.connections.read().await; + guard.get(id).cloned() + } + + /// Backward-compatible method returning snapshot of `(id, sender)` pairs. pub async fn get_all(&self) -> Vec<(ConnectionId, ConnectionSender)> { - let guard = self.connections.lock().await; - guard.iter().map(|(k, v)| (k.clone(), v.clone())).collect() + let guard = self.connections.read().await; + guard + .iter() + .map(|(k, v)| (k.clone(), v.sender.clone())) + .collect() + } + + /// Returns a snapshot of all active `Arc` handles. + pub async fn get_all_connections(&self) -> Vec> { + let guard = self.connections.read().await; + guard.values().cloned().collect() + } + + /// Returns the number of currently registered connections. + pub async fn len(&self) -> usize { + let guard = self.connections.read().await; + guard.len() + } + + /// Returns whether the registry is empty. + pub async fn is_empty(&self) -> bool { + let guard = self.connections.read().await; + guard.is_empty() } } diff --git a/backend/src/realtime/policy.rs b/backend/src/realtime/policy.rs index 373c1f35..263d3949 100644 --- a/backend/src/realtime/policy.rs +++ b/backend/src/realtime/policy.rs @@ -1,29 +1,424 @@ +use std::collections::VecDeque; +use std::time::{Duration, Instant}; use log::{info, warn}; use prometheus::{IntCounter, IntGauge, register_int_counter, register_int_gauge}; +use serde::{Deserialize, Serialize}; lazy_static::lazy_static! { - static ref SLOW_CONSUMER_COUNT: IntCounter = register_int_counter!( + pub static ref SLOW_CONSUMER_COUNT: IntCounter = register_int_counter!( "slow_consumer_disconnects_total", "Total number of slow consumer disconnections" ).unwrap(); - static ref QUEUE_DEPTH_GAUGE: IntGauge = register_int_gauge!( + pub static ref QUARANTINED_CONSUMER_COUNT: IntCounter = register_int_counter!( + "quarantined_consumers_total", + "Total number of slow consumer quarantines" + ).unwrap(); + + pub static ref DROPPED_MESSAGES_COUNT: IntCounter = register_int_counter!( + "dropped_messages_total", + "Total number of messages dropped due to slow consumers" + ).unwrap(); + + pub static ref QUEUE_DEPTH_GAUGE: IntGauge = register_int_gauge!( "connection_queue_depth", "Current queue depth per connection" ).unwrap(); } +/// Health state of a connection in the realtime subsystem. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum ConnectionHealthState { + /// Normal operation: healthy drain rate and low queue depth. + Healthy, + /// Queue depth is elevated or growing; monitored for quarantine. + Degraded, + /// Connection is quarantined: messages are dropped to isolate healthy connections. + Quarantined, + /// Connection exceeded maximum queue depth or dropped message limits and is evicted. + Evicted, +} + +/// Directional trend of queue depth over a sliding observation window. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum QueueDepthTrend { + /// Queue is empty or stable at a low depth. + Stable, + /// Consumer is draining faster than ingestion. + Draining, + /// Queue depth is increasing over successive samples. + Growing, + /// Queue is full or backed up with no drain progress. + Stalled, +} + +/// Decision made by the policy tracker for an incoming message delivery attempt. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PolicyDecision { + /// Allow message delivery normally. + Allow, + /// Allow message delivery, but connection is flagged as degraded. + AllowDegraded, + /// Drop message to prevent head-of-line blocking while connection is quarantined. + QuarantineDrop, + /// Evict connection immediately. + Evict, +} + +/// Single queue depth sample timestamped for trend analysis. +#[derive(Debug, Clone, Copy)] +pub struct QueueSample { + pub timestamp: Instant, + pub depth: usize, +} + +/// Configuration parameters for the proactive slow-consumer policy. +#[derive(Debug, Clone)] +pub struct SlowConsumerPolicyConfig { + /// Maximum allowable queue depth before immediate eviction. + pub max_queue_depth: usize, + /// Queue depth threshold where connection becomes Degraded. + pub warning_queue_depth: usize, + /// Number of consecutive high/growing samples required before quarantining. + pub consecutive_high_samples_threshold: usize, + /// Size of the trailing queue depth observation window. + pub sample_window_size: usize, + /// Leaky bucket leak rate in tokens (burst units) per second. + pub leak_rate_per_sec: f64, + /// Leaky bucket burst capacity to prevent false-positive evictions during transient traffic bursts. + pub bucket_capacity: f64, + /// Cooldown duration in quarantine before recovery is permitted. + pub quarantine_cooldown: Duration, + /// Maximum consecutive dropped messages in quarantine before hard eviction. + pub max_quarantined_drops: usize, +} + +impl Default for SlowConsumerPolicyConfig { + fn default() -> Self { + Self { + max_queue_depth: 100, + warning_queue_depth: 25, + consecutive_high_samples_threshold: 3, + sample_window_size: 10, + leak_rate_per_sec: 10.0, + bucket_capacity: 30.0, + quarantine_cooldown: Duration::from_millis(500), + max_quarantined_drops: 20, + } + } +} + +/// Per-connection tracker maintaining queue depth trend, leaky bucket state, and health lifecycle. +#[derive(Debug)] +pub struct ConnectionPolicyTracker { + config: SlowConsumerPolicyConfig, + history: VecDeque, + state: ConnectionHealthState, + consecutive_high_samples: usize, + quarantined_at: Option, + quarantined_drops: usize, + leaky_bucket_level: f64, + last_bucket_update: Instant, + total_sent: u64, + total_dropped: u64, +} + +impl ConnectionPolicyTracker { + pub fn new(config: SlowConsumerPolicyConfig) -> Self { + let now = Instant::now(); + Self { + config, + history: VecDeque::with_capacity(16), + state: ConnectionHealthState::Healthy, + consecutive_high_samples: 0, + quarantined_at: None, + quarantined_drops: 0, + leaky_bucket_level: 0.0, + last_bucket_update: now, + total_sent: 0, + total_dropped: 0, + } + } + + /// Evaluates current queue depth and returns a policy decision for the connection. + pub fn record_queue_depth(&mut self, current_depth: usize) -> PolicyDecision { + let now = Instant::now(); + + // 1. Update leaky bucket + let elapsed_secs = now.duration_since(self.last_bucket_update).as_secs_f64(); + self.leaky_bucket_level = (self.leaky_bucket_level - elapsed_secs * self.config.leak_rate_per_sec).max(0.0); + self.last_bucket_update = now; + + // 2. Record historical sample + if self.history.len() >= self.config.sample_window_size { + self.history.pop_front(); + } + self.history.push_back(QueueSample { + timestamp: now, + depth: current_depth, + }); + + // 3. Immediate hard eviction check + if current_depth >= self.config.max_queue_depth { + self.state = ConnectionHealthState::Evicted; + return PolicyDecision::Evict; + } + + // 4. Handle quarantined state + if self.state == ConnectionHealthState::Quarantined { + if let Some(q_time) = self.quarantined_at { + // If queue has drained back to healthy levels and cooldown elapsed, recover + if current_depth <= self.config.warning_queue_depth / 2 + && now.duration_since(q_time) >= self.config.quarantine_cooldown + { + info!("Connection recovered from quarantine (queue depth: {})", current_depth); + self.state = ConnectionHealthState::Healthy; + self.quarantined_at = None; + self.quarantined_drops = 0; + self.consecutive_high_samples = 0; + self.leaky_bucket_level = 0.0; + self.total_sent += 1; + return PolicyDecision::Allow; + } + } + + // Still quarantined + if self.quarantined_drops >= self.config.max_quarantined_drops { + self.state = ConnectionHealthState::Evicted; + return PolicyDecision::Evict; + } + + self.quarantined_drops += 1; + self.total_dropped += 1; + DROPPED_MESSAGES_COUNT.inc(); + return PolicyDecision::QuarantineDrop; + } + + // 5. Evaluate depth & growth trend for healthy / degraded + let trend = self.calculate_trend(); + + if current_depth >= self.config.warning_queue_depth { + self.consecutive_high_samples += 1; + self.leaky_bucket_level = (self.leaky_bucket_level + 1.0).min(self.config.bucket_capacity + 10.0); + + // Proactive quarantine trigger: sustained high samples or bucket overflow with growing trend + let should_quarantine = self.consecutive_high_samples >= self.config.consecutive_high_samples_threshold + || (self.leaky_bucket_level >= self.config.bucket_capacity && trend == QueueDepthTrend::Growing); + + if should_quarantine { + warn!( + "Connection transitioned to Quarantined (depth: {}, trend: {:?}, consecutive: {})", + current_depth, trend, self.consecutive_high_samples + ); + self.state = ConnectionHealthState::Quarantined; + self.quarantined_at = Some(now); + self.quarantined_drops = 1; + self.total_dropped += 1; + QUARANTINED_CONSUMER_COUNT.inc(); + DROPPED_MESSAGES_COUNT.inc(); + return PolicyDecision::QuarantineDrop; + } + + self.state = ConnectionHealthState::Degraded; + self.total_sent += 1; + return PolicyDecision::AllowDegraded; + } + + // Low depth + self.consecutive_high_samples = 0; + self.state = ConnectionHealthState::Healthy; + self.total_sent += 1; + PolicyDecision::Allow + } + + /// Records an explicit message drop. + pub fn record_drop(&mut self) -> PolicyDecision { + self.total_dropped += 1; + DROPPED_MESSAGES_COUNT.inc(); + if self.state == ConnectionHealthState::Quarantined { + self.quarantined_drops += 1; + if self.quarantined_drops >= self.config.max_quarantined_drops { + self.state = ConnectionHealthState::Evicted; + return PolicyDecision::Evict; + } + return PolicyDecision::QuarantineDrop; + } + self.state = ConnectionHealthState::Evicted; + PolicyDecision::Evict + } + + /// Records drain progress when the consumer consumes messages. + pub fn record_drain(&mut self, amount: usize) { + if let Some(last) = self.history.back_mut() { + last.depth = last.depth.saturating_sub(amount); + } + self.leaky_bucket_level = (self.leaky_bucket_level - amount as f64).max(0.0); + } + + /// Computes queue depth trend across recorded samples. + pub fn calculate_trend(&self) -> QueueDepthTrend { + if self.history.len() < 2 { + return QueueDepthTrend::Stable; + } + + let last = self.history.back().unwrap().depth; + if last == 0 { + return QueueDepthTrend::Stable; + } + if last >= self.config.max_queue_depth { + return QueueDepthTrend::Stalled; + } + + let n = self.history.len(); + let prev = self.history[n - 2].depth; + + if last < prev { + QueueDepthTrend::Draining + } else if last > prev { + QueueDepthTrend::Growing + } else { + let first = self.history.front().unwrap().depth; + if last < first { + QueueDepthTrend::Draining + } else if last > first { + QueueDepthTrend::Growing + } else { + QueueDepthTrend::Stable + } + } + } + + pub fn health_state(&self) -> ConnectionHealthState { + self.state + } + + pub fn is_quarantined(&self) -> bool { + self.state == ConnectionHealthState::Quarantined + } + + pub fn should_evict(&self) -> bool { + self.state == ConnectionHealthState::Evicted + } + + pub fn quarantined_drops(&self) -> usize { + self.quarantined_drops + } + + pub fn total_sent(&self) -> u64 { + self.total_sent + } + + pub fn total_dropped(&self) -> u64 { + self.total_dropped + } + + pub fn config(&self) -> &SlowConsumerPolicyConfig { + &self.config + } +} + +/// Handles slow consumer overflow off the hot path. pub async fn handle_overflow(connection_id: &str) -> Result<(), String> { warn!("Slow consumer detected: {} - disconnecting", connection_id); - - // Increment the slow consumer counter SLOW_CONSUMER_COUNT.inc(); - - // Log the disconnection reason info!("Disconnected slow consumer: {}", connection_id); - - // Optionally emit a metric for queue depth QUEUE_DEPTH_GAUGE.set(0); + Ok(()) +} +/// Handles quarantine transition for logging and metrics. +pub async fn handle_quarantine(connection_id: &str) -> Result<(), String> { + warn!("Slow consumer quarantined: {}", connection_id); + QUARANTINED_CONSUMER_COUNT.inc(); Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_policy_tracker_healthy_flow() { + let mut tracker = ConnectionPolicyTracker::new(SlowConsumerPolicyConfig::default()); + assert_eq!(tracker.health_state(), ConnectionHealthState::Healthy); + + for depth in [1, 2, 3, 2, 1, 0] { + let decision = tracker.record_queue_depth(depth); + assert_eq!(decision, PolicyDecision::Allow); + assert_eq!(tracker.health_state(), ConnectionHealthState::Healthy); + } + } + + #[test] + fn test_policy_tracker_trend_detection() { + let mut tracker = ConnectionPolicyTracker::new(SlowConsumerPolicyConfig::default()); + tracker.record_queue_depth(2); + tracker.record_queue_depth(5); + tracker.record_queue_depth(10); + assert_eq!(tracker.calculate_trend(), QueueDepthTrend::Growing); + + tracker.record_queue_depth(4); + assert_eq!(tracker.calculate_trend(), QueueDepthTrend::Draining); + } + + #[test] + fn test_policy_tracker_quarantine_and_recovery() { + let config = SlowConsumerPolicyConfig { + warning_queue_depth: 10, + consecutive_high_samples_threshold: 2, + quarantine_cooldown: Duration::from_millis(5), + ..Default::default() + }; + let mut tracker = ConnectionPolicyTracker::new(config); + + // Sample 1 at high depth: degraded + let d1 = tracker.record_queue_depth(12); + assert_eq!(d1, PolicyDecision::AllowDegraded); + assert_eq!(tracker.health_state(), ConnectionHealthState::Degraded); + + // Sample 2 at high depth: quarantined + let d2 = tracker.record_queue_depth(15); + assert_eq!(d2, PolicyDecision::QuarantineDrop); + assert_eq!(tracker.health_state(), ConnectionHealthState::Quarantined); + assert!(tracker.is_quarantined()); + + // Wait cooldown and drain + std::thread::sleep(Duration::from_millis(10)); + let d3 = tracker.record_queue_depth(2); + assert_eq!(d3, PolicyDecision::Allow); + assert_eq!(tracker.health_state(), ConnectionHealthState::Healthy); + } + + #[test] + fn test_policy_tracker_max_depth_eviction() { + let config = SlowConsumerPolicyConfig { + max_queue_depth: 50, + ..Default::default() + }; + let mut tracker = ConnectionPolicyTracker::new(config); + let decision = tracker.record_queue_depth(50); + assert_eq!(decision, PolicyDecision::Evict); + assert!(tracker.should_evict()); + } + + #[test] + fn test_transient_burst_does_not_evict() { + let config = SlowConsumerPolicyConfig { + warning_queue_depth: 20, + consecutive_high_samples_threshold: 3, + bucket_capacity: 50.0, + ..Default::default() + }; + let mut tracker = ConnectionPolicyTracker::new(config); + + // Transient single burst + let d1 = tracker.record_queue_depth(25); + assert_eq!(d1, PolicyDecision::AllowDegraded); + + // Immediately drained + tracker.record_drain(20); + let d2 = tracker.record_queue_depth(5); + assert_eq!(d2, PolicyDecision::Allow); + assert_eq!(tracker.health_state(), ConnectionHealthState::Healthy); + } } \ No newline at end of file diff --git a/backend/tests/connected_trace_test.rs b/backend/tests/connected_trace_test.rs index 9edb080d..70a0b38e 100644 --- a/backend/tests/connected_trace_test.rs +++ b/backend/tests/connected_trace_test.rs @@ -1,12 +1,12 @@ use opentelemetry::trace::{TraceContextExt, TracerProvider as _}; -use opentelemetry_sdk::trace::TracerProvider; +use opentelemetry_sdk::trace::SdkTracerProvider; use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::Registry; use stellar_insights_backend::observability::trace_context::{spawn_with_trace, TracedMessage}; fn setup_tracer() { - let provider = TracerProvider::builder().build(); + let provider = SdkTracerProvider::builder().build(); let tracer = provider.tracer("test"); let telemetry = tracing_opentelemetry::layer().with_tracer(tracer); let subscriber = Registry::default().with(telemetry); diff --git a/backend/tests/slow_consumer_test.rs b/backend/tests/slow_consumer_test.rs index 3be6ba7b..fca53a33 100644 --- a/backend/tests/slow_consumer_test.rs +++ b/backend/tests/slow_consumer_test.rs @@ -1,43 +1,239 @@ #[cfg(test)] mod tests { - use stellar_insights_backend::realtime::{fanout::fanout_message, ConnectionRegistry}; - use tokio::time::{sleep, Duration}; + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use stellar_insights_backend::realtime::{ + fanout_message, fanout_message_detailed, Connection, ConnectionHealthState, + ConnectionRegistry, FanoutConfig, QueueDepthTrend, SlowConsumerPolicyConfig, + }; + use tokio::time::sleep; use tokio_tungstenite::tungstenite::protocol::Message; + /// Acceptance Criterion 1: + /// A stalled/slow single connection cannot measurably delay delivery to any other healthy connection. + /// Demonstrates that p99 latency to N-1 healthy connections remains low (< 15ms) even when a stalled + /// connection has a full buffer / does not drain. + #[tokio::test] + async fn test_stalled_connection_does_not_delay_healthy_connections() { + let registry = ConnectionRegistry::new(); + let healthy_count = 50; + let mut healthy_receivers = Vec::new(); + + // 1. Set up healthy consumers with active draining tasks + for i in 0..healthy_count { + let id = format!("healthy-{}", i); + let (conn, rx) = Connection::create_channel(id); + registry.add_connection(Arc::new(conn)).await; + healthy_receivers.push(rx); + } + + // 2. Set up 1 deliberately stalled consumer (policy with low warning depth, receiver never drains) + let stalled_id = "stalled-consumer".to_string(); + let stalled_config = SlowConsumerPolicyConfig { + warning_queue_depth: 2, + consecutive_high_samples_threshold: 2, + max_queue_depth: 5, + max_quarantined_drops: 3, + ..Default::default() + }; + let (stalled_conn, _stalled_rx) = Connection::create_channel_with_policy(stalled_id.clone(), stalled_config); + let stalled_conn_arc = Arc::new(stalled_conn); + registry.add_connection(stalled_conn_arc.clone()).await; + + // 3. Measure fanout latency over multiple iterations + let iterations = 20; + let mut latencies = Vec::new(); + + for seq in 0..iterations { + let msg = Message::text(format!("analytics-event-{}", seq)); + let start = Instant::now(); + let summary = fanout_message_detailed(®istry, &msg, &FanoutConfig::default()) + .await + .expect("fanout should succeed"); + let elapsed = start.elapsed(); + latencies.push(elapsed); + + // Healthy consumers must always receive their message immediately + assert!( + summary.delivered >= healthy_count, + "All healthy connections must receive messages (delivered: {}, healthy: {})", + summary.delivered, + healthy_count + ); + } + + // 4. Verify healthy receivers received all messages without loss + for mut rx in healthy_receivers { + for seq in 0..iterations { + let msg = rx.recv().await.expect("healthy receiver must get message"); + assert_eq!(msg, Message::text(format!("analytics-event-{}", seq))); + } + } + + // 5. Verify p99 latency to healthy connections is flat and sub-millisecond range + latencies.sort(); + let p99_latency = latencies[(latencies.len() as f64 * 0.95) as usize]; + assert!( + p99_latency < Duration::from_millis(15), + "p99 fanout latency should be < 15ms even with stalled consumer, got {:?}", + p99_latency + ); + + // 6. Verify the stalled consumer was quarantined and marked for eviction + assert!( + stalled_conn_arc.is_quarantined().await || stalled_conn_arc.should_evict().await, + "Stalled consumer must be quarantined or evicted by proactive policy" + ); + } + + /// Acceptance Criterion 2: + /// Per-connection message ordering is strictly preserved. + #[tokio::test] + async fn test_per_connection_message_ordering_preserved() { + let registry = ConnectionRegistry::new(); + let client_count = 10; + let message_count = 100; + let mut receiver_handles = Vec::new(); + + for i in 0..client_count { + let (conn, mut rx) = Connection::create_channel(format!("ordered-client-{}", i)); + registry.add_connection(Arc::new(conn)).await; + + let handle = tokio::spawn(async move { + let mut received = Vec::new(); + for _ in 0..message_count { + if let Some(msg) = rx.recv().await { + received.push(msg); + } + } + received + }); + receiver_handles.push(handle); + } + + // Fan out sequential messages + for seq in 0..message_count { + let msg = Message::text(format!("seq-{}", seq)); + fanout_message(®istry, msg).await.expect("fanout should succeed"); + } + + // Verify each connection received messages in exact sequential order + for handle in receiver_handles { + let received = handle.await.expect("receiver task should complete"); + assert_eq!(received.len(), message_count); + for (seq, msg) in received.iter().enumerate() { + assert_eq!( + *msg, + Message::text(format!("seq-{}", seq)), + "Message order violated" + ); + } + } + } + + /// Acceptance Criterion 3: + /// Concurrency is bounded (no unbounded task spawning per fanout call). #[tokio::test] - async fn test_slow_consumer_does_not_affect_others() { + async fn test_bounded_concurrency_fanout() { let registry = ConnectionRegistry::new(); + let total_clients = 128; + let concurrency_bound = 16; + let mut receivers = Vec::new(); + + for i in 0..total_clients { + let (conn, rx) = Connection::create_channel(format!("bounded-client-{}", i)); + registry.add_connection(Arc::new(conn)).await; + receivers.push(rx); + } + + let config = FanoutConfig { + max_concurrency: concurrency_bound, + per_connection_timeout: None, + }; + + let summary = fanout_message_detailed(®istry, &Message::text("bounded-test"), &config) + .await + .expect("bounded fanout should succeed"); + + assert_eq!(summary.total_connections, total_clients); + assert_eq!(summary.delivered, total_clients); + + for mut rx in receivers { + let msg = rx.recv().await.expect("message delivered"); + assert_eq!(msg, Message::text("bounded-test")); + } + } + + /// Acceptance Criterion 4: + /// Documented, tested proactive slow-consumer detection and quarantine policy in policy.rs. + #[tokio::test] + async fn test_proactive_slow_consumer_quarantine_and_recovery() { + let policy_config = SlowConsumerPolicyConfig { + warning_queue_depth: 3, + consecutive_high_samples_threshold: 2, + max_queue_depth: 10, + quarantine_cooldown: Duration::from_millis(50), + max_quarantined_drops: 5, + ..Default::default() + }; + + let (conn, mut rx) = Connection::create_channel_with_policy("monitored-client".to_string(), policy_config); + + // Send 1: seen depth 0 -> Healthy, queue becomes 1 + conn.send(Message::text("m1")).await.unwrap(); + assert_eq!(conn.health_state().await, ConnectionHealthState::Healthy); + + // Send 2: seen depth 1 -> Healthy, queue becomes 2 + conn.send(Message::text("m2")).await.unwrap(); + assert_eq!(conn.health_state().await, ConnectionHealthState::Healthy); + + // Send 3: seen depth 2 -> Healthy, queue becomes 3 + conn.send(Message::text("m3")).await.unwrap(); + assert_eq!(conn.health_state().await, ConnectionHealthState::Healthy); + + // Send 4: seen depth 3 (>= warning depth 3, sample 1) -> Degraded, queue becomes 4 + conn.send(Message::text("m4")).await.unwrap(); + assert_eq!(conn.health_state().await, ConnectionHealthState::Degraded); + assert_eq!(conn.trend().await, QueueDepthTrend::Growing); + + // Send 5: seen depth 4 (>= warning depth 3, sample 2 >= consecutive threshold 2) -> Quarantined + let send_res5 = conn.send(Message::text("m5")).await; + assert!(send_res5.is_err(), "Quarantined connection should drop message"); + assert!(conn.is_quarantined().await); + assert_eq!(conn.health_state().await, ConnectionHealthState::Quarantined); + + // Drain messages to recover + while let Ok(msg) = rx.try_recv() { + let _ = msg; + } + assert_eq!(conn.queue_depth(), 0); + + // Allow quarantine cooldown to pass + sleep(Duration::from_millis(60)).await; + + // Next send recovers to Healthy + let recover_send = conn.send(Message::text("m6")).await; + assert!(recover_send.is_ok(), "Connection should recover from quarantine after draining"); + assert_eq!(conn.health_state().await, ConnectionHealthState::Healthy); + } + + /// Acceptance Criterion 5: + /// Legacy interface compatibility with raw ConnectionRegistry::add. + #[tokio::test] + async fn test_legacy_add_and_get_all_compatibility() { + let registry = ConnectionRegistry::new(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + registry.add("legacy-conn".to_string(), tx).await; + assert_eq!(registry.len().await, 1); + + let all = registry.get_all().await; + assert_eq!(all.len(), 1); + assert_eq!(all[0].0, "legacy-conn"); - // Create N well-behaved subscribers - let mut well_behaved_handles = vec![]; - for i in 0..5 { - let id = format!("well-behaved-{}", i); - registry.add(id.clone(), tokio::sync::mpsc::unbounded_channel().0).await; - let registry_clone = registry.clone(); - well_behaved_handles.push(tokio::spawn(async move { - let _ = fanout_message(®istry_clone, Message::text("test")).await; - })); - } - - // Create 1 artificially slow subscriber - let slow_id = "slow-consumer".to_string(); - registry.add(slow_id.clone(), tokio::sync::mpsc::unbounded_channel().0).await; - let registry_clone = registry.clone(); - let slow_handle = tokio::spawn(async move { - // Simulate a slow consumer by pausing - sleep(Duration::from_millis(500)).await; - let _ = fanout_message(®istry_clone, Message::text("test")).await; - }); - - // Wait for all tasks to complete - for handle in well_behaved_handles { - let _ = handle.await; - } - let _ = slow_handle.await; - - // Assert that the slow consumer was disconnected (queue overflow handled) - // and that well-behaved consumers received all messages without loss. - // This is a placeholder assertion; real tests would check actual message delivery. - assert!(true); + fanout_message(®istry, Message::text("legacy-msg")).await.unwrap(); + let received = rx.recv().await.unwrap(); + assert_eq!(received, Message::text("legacy-msg")); } }