fix(server): eliminate startup panics and add pub/sub pattern length cap

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<TypeConfig>, 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<ClusterResponse, RaftProposalError> {
+    pub async fn propose(&self, cmd: ClusterCommand) -> Result<ClusterResponse, RaftProposalError> {
         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()));
     }

crates/ember-protocol/src/command.rs

@@ -1588,7 +1588,7 @@ fn parse_cluster(args: &[Frame]) -> Result<Command, ProtocolError> {
         }
         "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(_)));
     }
 

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<PathBuf>,
-    ) -> (Self, mpsc::Receiver<GossipEvent>) {
+    ) -> Result<(Self, mpsc::Receiver<GossipEvent>), 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<NodeId> =
-            data.nodes.keys().filter_map(|k| NodeId::parse(k).ok()).collect();
+        let raft_ids: std::collections::HashSet<NodeId> = data
+            .nodes
+            .keys()
+            .filter_map(|k| NodeId::parse(k).ok())
+            .collect();
         let to_remove: Vec<NodeId> = 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,15 +1084,28 @@ 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.
     fn test_coordinator_bootstrapped() -> (ClusterCoordinator, mpsc::Receiver<GossipEvent>) {
         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;
 

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()),

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);

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 {

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)
         };