diff --git a/crates/ember-cluster/src/raft.rs b/crates/ember-cluster/src/raft.rs index dc531103..61cf4800 100644 --- a/crates/ember-cluster/src/raft.rs +++ b/crates/ember-cluster/src/raft.rs @@ -11,7 +11,6 @@ use std::net::SocketAddr; use std::ops::RangeBounds; use std::sync::Arc; - use openraft::error::{ ClientWriteError, InstallSnapshotError, NetworkError, RPCError, RaftError, Unreachable, }; @@ -22,9 +21,8 @@ use openraft::raft::{ }; use openraft::storage::{Adaptor, LogState, RaftLogReader, RaftSnapshotBuilder, Snapshot}; use openraft::{ - BasicNode, Config, Entry, EntryPayload, LogId, OptionalSend, Raft, RaftStorage, - RaftTypeConfig, ServerState, SnapshotMeta, StorageError, StorageIOError, StoredMembership, - Vote, + BasicNode, Config, Entry, EntryPayload, LogId, OptionalSend, Raft, RaftStorage, RaftTypeConfig, + ServerState, SnapshotMeta, StorageError, StorageIOError, StoredMembership, Vote, }; use serde::{Deserialize, Serialize}; use tokio::net::{TcpListener, TcpStream}; @@ -703,9 +701,9 @@ pub(crate) fn spawn_raft_listener(raft: Raft, bind_addr: SocketAddr) snapshot: Box::new(Cursor::new(data)), }; match raft.install_full_snapshot(vote, snapshot).await { - Ok(r) => RaftRpcResponse::InstallSnapshot( - InstallSnapshotResponse { vote: r.vote }, - ), + Ok(r) => RaftRpcResponse::InstallSnapshot(InstallSnapshotResponse { + vote: r.vote, + }), Err(e) => { debug!("install_snapshot error: {e}"); return; @@ -779,8 +777,14 @@ impl RaftNode { let (log_store, state_machine) = Adaptor::new(Arc::clone(&storage)); - let raft = Raft::new(local_raft_id, config, RaftNetworkFactory, log_store, state_machine) - .await?; + let raft = Raft::new( + local_raft_id, + config, + RaftNetworkFactory, + log_store, + state_machine, + ) + .await?; spawn_raft_listener(raft.clone(), raft_addr); @@ -814,10 +818,7 @@ impl RaftNode { /// /// Blocks until the entry is committed and applied to the state machine /// on a quorum of nodes. Returns `NotLeader` if this node is not the leader. - pub async fn propose( - &self, - cmd: ClusterCommand, - ) -> Result { + pub async fn propose(&self, cmd: ClusterCommand) -> Result { match self.raft.client_write(cmd).await { Ok(resp) => Ok(resp.data), Err(e) => match e { @@ -866,7 +867,7 @@ pub fn raft_id_from_node_id(node_id: NodeId) -> u64 { } fn io_error(msg: &str) -> std::io::Error { - std::io::Error::new(std::io::ErrorKind::Other, msg) + std::io::Error::other(msg) } #[cfg(test)] @@ -1214,7 +1215,10 @@ mod tests { s.apply_to_state_machine(&[entry]).await.unwrap(); - assert!(rx.changed().await.is_ok(), "watch channel should have fired"); + assert!( + rx.changed().await.is_ok(), + "watch channel should have fired" + ); let data = rx.borrow(); assert!(data.nodes.contains_key(&node_id.as_key())); } diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index dfe750e2..8cd6000d 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -1588,7 +1588,7 @@ fn parse_cluster(args: &[Frame]) -> Result { } "ADDSLOTSRANGE" => { // arguments are pairs: start1 end1 [start2 end2 ...] - if args.len() < 3 || (args.len() - 1) % 2 != 0 { + if args.len() < 3 || !(args.len() - 1).is_multiple_of(2) { return Err(ProtocolError::WrongArity("CLUSTER ADDSLOTSRANGE".into())); } let mut ranges = Vec::new(); @@ -4357,16 +4357,14 @@ mod tests { #[test] fn cluster_addslotsrange_invalid_range() { - let err = - Command::from_frame(cmd(&["CLUSTER", "ADDSLOTSRANGE", "100", "50"])).unwrap_err(); + let err = Command::from_frame(cmd(&["CLUSTER", "ADDSLOTSRANGE", "100", "50"])).unwrap_err(); assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); } #[test] fn cluster_addslotsrange_wrong_arity() { // odd number of slot args - let err = - Command::from_frame(cmd(&["CLUSTER", "ADDSLOTSRANGE", "0"])).unwrap_err(); + let err = Command::from_frame(cmd(&["CLUSTER", "ADDSLOTSRANGE", "0"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs index 2c7f8cb3..15eda5c6 100644 --- a/crates/ember-server/src/cluster.rs +++ b/crates/ember-server/src/cluster.rs @@ -56,20 +56,27 @@ impl ClusterCoordinator { /// /// Returns the coordinator and a receiver for gossip events that /// should be consumed by a background task. + /// + /// # Errors + /// + /// Returns an error if `bind_addr.port() + gossip_config.gossip_port_offset` overflows u16. pub fn new( local_id: NodeId, bind_addr: SocketAddr, gossip_config: GossipConfig, bootstrap: bool, data_dir: Option, - ) -> (Self, mpsc::Receiver) { + ) -> Result<(Self, mpsc::Receiver), String> { let (event_tx, event_rx) = mpsc::channel(256); let port_offset = gossip_config.gossip_port_offset; - let gossip_port = bind_addr - .port() - .checked_add(port_offset) - .expect("gossip port offset overflows u16"); + let gossip_port = bind_addr.port().checked_add(port_offset).ok_or_else(|| { + format!( + "gossip port overflow: {} + {} exceeds u16 range", + bind_addr.port(), + port_offset + ) + })?; let gossip_addr = SocketAddr::new(bind_addr.ip(), gossip_port); let gossip = GossipEngine::new(local_id, gossip_addr, gossip_config, event_tx); @@ -97,7 +104,7 @@ impl ClusterCoordinator { raft_node: std::sync::OnceLock::new(), }; - (coordinator, event_rx) + Ok((coordinator, event_rx)) } /// Restores a cluster coordinator from a previously saved `nodes.conf`. @@ -116,10 +123,13 @@ impl ClusterCoordinator { let (event_tx, event_rx) = mpsc::channel(256); - let gossip_port = bind_addr - .port() - .checked_add(port_offset) - .expect("gossip port offset overflows u16"); + let gossip_port = bind_addr.port().checked_add(port_offset).ok_or_else(|| { + ConfigParseError::InvalidAddress(format!( + "gossip port overflow: {} + {} exceeds u16 range", + bind_addr.port(), + port_offset + )) + })?; let gossip_addr = SocketAddr::new(bind_addr.ip(), gossip_port); let mut gossip = GossipEngine::new(local_id, gossip_addr, gossip_config, event_tx); @@ -202,18 +212,18 @@ impl ClusterCoordinator { Ok(a) => a, Err(_) => continue, }; - let node = ClusterNode::new_primary_with_offset( - node_id, - addr, - self.gossip_port_offset, - ); + let node = + ClusterNode::new_primary_with_offset(node_id, addr, self.gossip_port_offset); state.add_node(node); } } // remove nodes that raft has dropped (never remove ourselves) - let raft_ids: std::collections::HashSet = - data.nodes.keys().filter_map(|k| NodeId::parse(k).ok()).collect(); + let raft_ids: std::collections::HashSet = data + .nodes + .keys() + .filter_map(|k| NodeId::parse(k).ok()) + .collect(); let to_remove: Vec = state .nodes .keys() @@ -226,8 +236,7 @@ impl ClusterCoordinator { // reconcile slot assignments from the raft slot map for slot in 0..SLOT_COUNT { - let raft_owner = - data.slots.get(&slot).and_then(|k| NodeId::parse(k).ok()); + let raft_owner = data.slots.get(&slot).and_then(|k| NodeId::parse(k).ok()); let current_owner = state.slot_map.owner(slot); if raft_owner != current_owner { match raft_owner { @@ -937,8 +946,7 @@ impl ClusterCoordinator { addr: raft_addr.to_string(), }; let handle = raft.raft_handle(); - if let Ok(_) = - handle.add_learner(raft_id, node, true).await + if handle.add_learner(raft_id, node, true).await.is_ok() { let m = handle.metrics().borrow().clone(); let mut new_members: std::collections::BTreeSet< @@ -1076,7 +1084,7 @@ mod tests { let local_id = NodeId::new(); let addr: SocketAddr = "127.0.0.1:6379".parse().unwrap(); let config = GossipConfig::default(); - ClusterCoordinator::new(local_id, addr, config, false, None) + ClusterCoordinator::new(local_id, addr, config, false, None).unwrap() } /// Creates a test coordinator bootstrapped with all 16384 slots. @@ -1084,7 +1092,20 @@ mod tests { let local_id = NodeId::new(); let addr: SocketAddr = "127.0.0.1:6379".parse().unwrap(); let config = GossipConfig::default(); - ClusterCoordinator::new(local_id, addr, config, true, None) + ClusterCoordinator::new(local_id, addr, config, true, None).unwrap() + } + + #[test] + fn new_rejects_port_overflow() { + // port 65000 + offset 2000 = 67000, which overflows u16 (max 65535) + let local_id = NodeId::new(); + let addr: SocketAddr = "127.0.0.1:65000".parse().unwrap(); + let config = GossipConfig { + gossip_port_offset: 2000, + ..GossipConfig::default() + }; + let result = ClusterCoordinator::new(local_id, addr, config, false, None); + assert!(result.is_err(), "expected port overflow error"); } #[tokio::test] @@ -1397,7 +1418,8 @@ mod tests { let addr: SocketAddr = "127.0.0.1:6379".parse().unwrap(); let config = GossipConfig::default(); let (coord, _rx) = - ClusterCoordinator::new(local_id, addr, config, true, Some(dir.path().to_path_buf())); + ClusterCoordinator::new(local_id, addr, config, true, Some(dir.path().to_path_buf())) + .unwrap(); // add some slots and save coord.save_config().await; @@ -1422,7 +1444,8 @@ mod tests { config.clone(), true, Some(dir.path().to_path_buf()), - ); + ) + .unwrap(); coord.save_config().await; diff --git a/crates/ember-server/src/concurrent_handler.rs b/crates/ember-server/src/concurrent_handler.rs index 97b7d839..ddbeeb4c 100644 --- a/crates/ember-server/src/concurrent_handler.rs +++ b/crates/ember-server/src/concurrent_handler.rs @@ -283,7 +283,12 @@ async fn execute_concurrent( TtlResult::Milliseconds(ms) => Frame::Integer(ms as i64), TtlResult::NoExpiry => Frame::Integer(-1), TtlResult::NotFound => Frame::Integer(-2), - TtlResult::Seconds(s) => Frame::Integer(s as i64 * 1000), + TtlResult::Seconds(s) => { + // convert seconds → milliseconds, capping at i64::MAX to + // avoid overflow for pathologically large TTL values + let ms = s.saturating_mul(1000).min(i64::MAX as u64); + Frame::Integer(ms as i64) + } }, Command::Ping(None) => Frame::Simple("PONG".into()), diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index dde36fb5..a546e124 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -513,7 +513,14 @@ fn handle_sub_command( .serialize(out); continue; } - let rx = pubsub.psubscribe(&pat); + // psubscribe returns None if the pattern exceeds its internal + // length cap — this is a backstop; the check above should have + // already rejected oversized patterns. + let Some(rx) = pubsub.psubscribe(&pat) else { + Frame::Error(format!("ERR pattern too long ({} bytes)", pat.len())) + .serialize(out); + continue; + }; pattern_rxs.insert(pat.clone(), rx); let count = channel_rxs.len() + pattern_rxs.len(); serialize_sub_response(b"psubscribe", &pat, count, out); diff --git a/crates/ember-server/src/grpc.rs b/crates/ember-server/src/grpc.rs index aa9cb8e0..c5fb11b2 100644 --- a/crates/ember-server/src/grpc.rs +++ b/crates/ember-server/src/grpc.rs @@ -2021,7 +2021,9 @@ impl EmberCache for EmberService { channel_rxs.push((ch.clone(), pubsub.subscribe(ch))); } for pat in &req.patterns { - pattern_rxs.push((pat.clone(), pubsub.psubscribe(pat))); + if let Some(rx) = pubsub.psubscribe(pat) { + pattern_rxs.push((pat.clone(), rx)); + } } tokio::spawn(async move { diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index 15bf627c..0d7c02ac 100644 --- a/crates/ember-server/src/main.rs +++ b/crates/ember-server/src/main.rs @@ -22,7 +22,7 @@ use std::path::PathBuf; use std::sync::Arc; use clap::Parser; -use ember_cluster::{GossipConfig, NodeId, RaftNode, RaftStorage, raft_id_from_node_id}; +use ember_cluster::{raft_id_from_node_id, GossipConfig, NodeId, RaftNode, RaftStorage}; use ember_core::ShardPersistenceConfig; use tracing::info; #[cfg(feature = "protobuf")] @@ -451,9 +451,13 @@ async fn main() { let data = std::fs::read_to_string(&conf_path) .unwrap_or_else(|e| exit_err(format!("failed to read nodes.conf: {e}"))); - let (coord, rx) = - ClusterCoordinator::from_config(&data, addr, gossip_config, cluster_data_dir.clone()) - .unwrap_or_else(|e| exit_err(format!("failed to parse nodes.conf: {e}"))); + let (coord, rx) = ClusterCoordinator::from_config( + &data, + addr, + gossip_config, + cluster_data_dir.clone(), + ) + .unwrap_or_else(|e| exit_err(format!("failed to parse nodes.conf: {e}"))); info!("cluster mode: restored from nodes.conf"); let id = coord.local_id(); @@ -466,7 +470,8 @@ async fn main() { gossip_config, true, Some(cluster_data_dir.clone()), - ); + ) + .unwrap_or_else(|e| exit_err(format!("error: {e}"))); info!("cluster mode: bootstrapped with all 16384 slots"); (coord, rx, local_id, true) } else { @@ -477,7 +482,8 @@ async fn main() { gossip_config, false, Some(cluster_data_dir.clone()), - ); + ) + .unwrap_or_else(|e| exit_err(format!("error: {e}"))); info!("cluster mode: waiting for CLUSTER MEET"); (coord, rx, local_id, false) }; diff --git a/crates/ember-server/src/pubsub.rs b/crates/ember-server/src/pubsub.rs index f3cff8f1..8471f971 100644 --- a/crates/ember-server/src/pubsub.rs +++ b/crates/ember-server/src/pubsub.rs @@ -10,6 +10,12 @@ use bytes::Bytes; use dashmap::DashMap; use tokio::sync::broadcast; +/// Maximum allowed byte length for a pub/sub pattern. +/// +/// Longer patterns provide no real-world value and allow clients to +/// force repeated glob-match work on every PUBLISH call. +const MAX_PATTERN_LEN: usize = 512; + /// Maximum number of buffered messages per subscription before /// slow consumers start missing messages. This is per-channel, /// not global — a subscriber that falls behind on one busy channel @@ -68,8 +74,13 @@ impl PubSubManager { /// Subscribe to a glob pattern. Returns a receiver for messages /// matching the pattern. - pub fn psubscribe(&self, pattern: &str) -> broadcast::Receiver { - self.subscribe_to(&self.patterns, pattern) + /// + /// Returns `None` if the pattern exceeds `MAX_PATTERN_LEN` bytes. + pub fn psubscribe(&self, pattern: &str) -> Option> { + if pattern.len() > MAX_PATTERN_LEN { + return None; + } + Some(self.subscribe_to(&self.patterns, pattern)) } /// Unsubscribe from a pattern. Returns true if the pattern existed @@ -382,7 +393,7 @@ mod tests { #[test] fn pattern_subscribe_and_publish() { let mgr = PubSubManager::new(); - let mut rx = mgr.psubscribe("news.*"); + let mut rx = mgr.psubscribe("news.*").unwrap(); let count = mgr.publish("news.sports", Bytes::from("goal!")); assert_eq!(count, 1); @@ -400,7 +411,7 @@ mod tests { fn exact_and_pattern_both_receive() { let mgr = PubSubManager::new(); let mut rx_exact = mgr.subscribe("news.sports"); - let mut rx_pattern = mgr.psubscribe("news.*"); + let mut rx_pattern = mgr.psubscribe("news.*").unwrap(); let count = mgr.publish("news.sports", Bytes::from("goal!")); assert_eq!(count, 2); @@ -427,9 +438,22 @@ mod tests { let _rx1 = mgr.subscribe("a"); let _rx2 = mgr.subscribe("b"); - let _rx3 = mgr.psubscribe("c.*"); + let _rx3 = mgr.psubscribe("c.*").unwrap(); assert_eq!(mgr.total_subscriptions(), 3); assert_eq!(mgr.channel_names(None).len(), 2); assert_eq!(mgr.active_patterns(), 1); } + + #[test] + fn psubscribe_rejects_oversized_pattern() { + let mgr = PubSubManager::new(); + let long_pattern = "*".repeat(MAX_PATTERN_LEN + 1); + assert!( + mgr.psubscribe(&long_pattern).is_none(), + "oversized pattern should be rejected" + ); + // a pattern right at the limit is allowed + let ok_pattern = "*".repeat(MAX_PATTERN_LEN); + assert!(mgr.psubscribe(&ok_pattern).is_some()); + } }