From ad3dd62df0307aee34b663f81d3eed1c550e9f1a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 5 Feb 2026 23:18:06 -0500 Subject: [PATCH 1/2] feat(cluster): add slot management and topology foundations implements the core building blocks for redis cluster compatibility: - slots.rs: CRC16 hash function (redis-compatible), SlotMap for 16384 slot distribution, SlotRange for contiguous slot assignments, hash tag extraction for key co-location - topology.rs: NodeId (uuid-based), ClusterNode with roles and health, ClusterState for full cluster view, MOVED redirect generation, CLUSTER INFO and CLUSTER NODES output formatting - error.rs: ClusterError enum for MOVED, ASK, and other cluster errors all modules include comprehensive unit tests. the crc16 implementation is verified against known redis slot assignments. --- Cargo.lock | 115 +++++ crates/ember-cluster/Cargo.toml | 9 + crates/ember-cluster/src/error.rs | 66 +++ crates/ember-cluster/src/lib.rs | 40 +- crates/ember-cluster/src/slots.rs | 405 ++++++++++++++++++ crates/ember-cluster/src/topology.rs | 603 +++++++++++++++++++++++++++ 6 files changed, 1235 insertions(+), 3 deletions(-) create mode 100644 crates/ember-cluster/src/error.rs create mode 100644 crates/ember-cluster/src/slots.rs create mode 100644 crates/ember-cluster/src/topology.rs diff --git a/Cargo.lock b/Cargo.lock index 2708e355..0fcff27c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,6 +73,12 @@ version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +[[package]] +name = "bumpalo" +version = "3.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" + [[package]] name = "bytes" version = "1.11.1" @@ -144,7 +150,13 @@ dependencies = [ name = "ember-cluster" version = "0.2.0" dependencies = [ + "bytes", + "rand", + "serde", "thiserror", + "tokio", + "tracing", + "uuid", ] [[package]] @@ -249,6 +261,16 @@ version = "1.0.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +[[package]] +name = "js-sys" +version = "0.3.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c942ebf8e95485ca0d52d97da7c5a2c387d0e7f0ba4c35e93bfcaee045955b3" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -477,12 +499,48 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -678,6 +736,18 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee48d38b119b0cd71fe4141b30f5ba9c7c5d9f4e7a3a8b4a674e4b6ef789976f" +dependencies = [ + "getrandom", + "js-sys", + "serde_core", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.1" @@ -699,6 +769,51 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasm-bindgen" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12" +dependencies = [ + "unicode-ident", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/crates/ember-cluster/Cargo.toml b/crates/ember-cluster/Cargo.toml index da18950d..d8c6e274 100644 --- a/crates/ember-cluster/Cargo.toml +++ b/crates/ember-cluster/Cargo.toml @@ -11,3 +11,12 @@ readme = "README.md" [dependencies] thiserror = { workspace = true } +tokio = { workspace = true } +bytes = { workspace = true } +rand = { workspace = true } +tracing = { workspace = true } +serde = { version = "1", features = ["derive"] } +uuid = { version = "1", features = ["v4", "serde"] } + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/crates/ember-cluster/src/error.rs b/crates/ember-cluster/src/error.rs new file mode 100644 index 00000000..84deb47c --- /dev/null +++ b/crates/ember-cluster/src/error.rs @@ -0,0 +1,66 @@ +//! Error types for cluster operations. + +use std::net::SocketAddr; + +use crate::NodeId; + +/// Errors that can occur during cluster operations. +#[derive(Debug, thiserror::Error)] +pub enum ClusterError { + /// The slot is not assigned to any node. + #[error("slot {0} is not assigned to any node")] + SlotNotAssigned(u16), + + /// The key belongs to a slot on a different node. + #[error("MOVED {slot} {addr}")] + Moved { slot: u16, addr: SocketAddr }, + + /// The slot is being migrated; client should retry with ASK. + #[error("ASK {slot} {addr}")] + Ask { slot: u16, addr: SocketAddr }, + + /// Node not found in the cluster. + #[error("node {0} not found in cluster")] + NodeNotFound(NodeId), + + /// Cluster is not in a healthy state. + #[error("cluster is down")] + ClusterDown, + + /// Operation requires a different node role. + #[error("operation not supported on {role} node")] + WrongRole { role: String }, + + /// Cross-slot operation with keys in different slots. + #[error("cross-slot keys not allowed (keys span slots {0} and {1})")] + CrossSlot(u16, u16), + + /// Network error during cluster communication. + #[error("cluster communication error: {0}")] + Network(String), + + /// Timeout waiting for cluster operation. + #[error("cluster operation timed out")] + Timeout, + + /// Configuration error. + #[error("invalid cluster configuration: {0}")] + Configuration(String), +} + +impl ClusterError { + /// Returns true if this is a redirect error (MOVED or ASK). + pub fn is_redirect(&self) -> bool { + matches!(self, ClusterError::Moved { .. } | ClusterError::Ask { .. }) + } + + /// Creates a MOVED error for a slot redirect. + pub fn moved(slot: u16, addr: SocketAddr) -> Self { + ClusterError::Moved { slot, addr } + } + + /// Creates an ASK error for a slot migration redirect. + pub fn ask(slot: u16, addr: SocketAddr) -> Self { + ClusterError::Ask { slot, addr } + } +} diff --git a/crates/ember-cluster/src/lib.rs b/crates/ember-cluster/src/lib.rs index 30563b01..5d6dc2c9 100644 --- a/crates/ember-cluster/src/lib.rs +++ b/crates/ember-cluster/src/lib.rs @@ -1,4 +1,38 @@ -//! ember-cluster: distributed coordination. +//! ember-cluster: distributed coordination for ember. //! -//! Manages cluster topology, gossip-based failure detection, -//! raft consensus, and live slot migration. +//! This crate provides the building blocks for running ember as a distributed +//! cluster with automatic failover and horizontal scaling. +//! +//! # Architecture +//! +//! The cluster layer sits between the protocol layer and the storage engine, +//! handling: +//! +//! - **Slot management**: 16384 hash slots distributed across nodes +//! - **Topology tracking**: Node membership and health monitoring +//! - **Failure detection**: SWIM gossip protocol for quick detection +//! - **Consensus**: Raft for cluster configuration changes +//! - **Migration**: Live slot resharding without downtime +//! +//! # Quick Start +//! +//! ```rust,ignore +//! use ember_cluster::{ClusterState, ClusterNode, NodeId, key_slot}; +//! +//! // Create a single-node cluster +//! let node_id = NodeId::new(); +//! let node = ClusterNode::new_primary(node_id, "127.0.0.1:6379".parse().unwrap()); +//! let cluster = ClusterState::single_node(node); +//! +//! // Route a key to its slot +//! let slot = key_slot(b"mykey"); +//! assert!(cluster.owns_slot(slot)); +//! ``` + +mod error; +mod slots; +mod topology; + +pub use error::ClusterError; +pub use slots::{key_slot, SlotMap, SlotRange, SLOT_COUNT}; +pub use topology::{ClusterHealth, ClusterNode, ClusterState, NodeFlags, NodeId, NodeRole}; diff --git a/crates/ember-cluster/src/slots.rs b/crates/ember-cluster/src/slots.rs new file mode 100644 index 00000000..b78774c3 --- /dev/null +++ b/crates/ember-cluster/src/slots.rs @@ -0,0 +1,405 @@ +//! Hash slot management for Redis Cluster-compatible key distribution. +//! +//! Implements CRC16 hashing (XMODEM polynomial) and 16384-slot mapping +//! following the Redis Cluster specification. + +use crate::NodeId; + +/// Total number of hash slots in the cluster (Redis Cluster standard). +pub const SLOT_COUNT: u16 = 16384; + +/// CRC16 lookup table from Redis source code (crc16.c). +/// Uses CCITT polynomial for Redis Cluster slot calculation. +#[rustfmt::skip] +static CRC16_TABLE: [u16; 256] = [ + 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7, + 0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef, + 0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6, + 0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de, + 0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485, + 0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d, + 0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4, + 0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc, + 0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823, + 0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b, + 0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12, + 0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a, + 0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41, + 0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49, + 0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70, + 0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78, + 0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f, + 0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067, + 0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e, + 0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256, + 0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d, + 0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, + 0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c, + 0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634, + 0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab, + 0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3, + 0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a, + 0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92, + 0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9, + 0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1, + 0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8, + 0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0, +]; + +/// Computes CRC16 checksum using XMODEM polynomial. +/// +/// This is the same algorithm Redis uses for slot calculation. +fn crc16(data: &[u8]) -> u16 { + let mut crc: u16 = 0; + for &byte in data { + let idx = ((crc >> 8) ^ (byte as u16)) as usize; + crc = (crc << 8) ^ CRC16_TABLE[idx]; + } + crc +} + +/// Extracts the hashable portion of a key, handling hash tags. +/// +/// Hash tags allow multiple keys to be assigned to the same slot. +/// The tag is the content between the first `{` and the first `}` after it. +/// +/// Examples: +/// - `user:{123}:profile` → hashes `123` +/// - `{user}:123` → hashes `user` +/// - `foo{}{bar}` → hashes `foo{}{bar}` (empty tag, no match) +/// - `foo{bar` → hashes `foo{bar` (no closing brace) +/// - `foobar` → hashes `foobar` (no tag) +fn extract_hash_tag(key: &[u8]) -> &[u8] { + // Find first '{' + let Some(open) = key.iter().position(|&b| b == b'{') else { + return key; + }; + + // Find first '}' after the '{' + let after_open = &key[open + 1..]; + let Some(close) = after_open.iter().position(|&b| b == b'}') else { + return key; + }; + + // Empty tag (e.g., "foo{}bar") means use the whole key + if close == 0 { + return key; + } + + &after_open[..close] +} + +/// Computes the hash slot for a key. +/// +/// Returns a value in the range [0, 16383]. +pub fn key_slot(key: &[u8]) -> u16 { + let hash_input = extract_hash_tag(key); + crc16(hash_input) % SLOT_COUNT +} + +/// A contiguous range of slots assigned to a node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct SlotRange { + pub start: u16, + pub end: u16, // inclusive +} + +impl SlotRange { + /// Creates a new slot range (end is inclusive). + pub fn new(start: u16, end: u16) -> Self { + debug_assert!(start <= end); + debug_assert!(end < SLOT_COUNT); + Self { start, end } + } + + /// Creates a range containing a single slot. + pub fn single(slot: u16) -> Self { + Self::new(slot, slot) + } + + /// Returns the number of slots in this range. + pub fn len(&self) -> u16 { + self.end - self.start + 1 + } + + /// Returns true if this range contains no slots (never true for valid ranges). + pub fn is_empty(&self) -> bool { + false // a valid range always has at least one slot + } + + /// Returns true if this range contains the given slot. + pub fn contains(&self, slot: u16) -> bool { + slot >= self.start && slot <= self.end + } + + /// Returns an iterator over all slots in this range. + pub fn iter(&self) -> impl Iterator { + self.start..=self.end + } +} + +impl std::fmt::Display for SlotRange { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.start == self.end { + write!(f, "{}", self.start) + } else { + write!(f, "{}-{}", self.start, self.end) + } + } +} + +/// Maps each of the 16384 slots to a node ID. +/// +/// When a slot is `None`, it means the slot is not assigned to any node, +/// which indicates an incomplete cluster configuration. +#[derive(Debug, Clone)] +pub struct SlotMap { + slots: Box<[Option; SLOT_COUNT as usize]>, +} + +impl Default for SlotMap { + fn default() -> Self { + Self::new() + } +} + +impl SlotMap { + /// Creates an empty slot map with no assignments. + pub fn new() -> Self { + Self { + // Box it to avoid 128KB stack allocation + slots: Box::new([None; SLOT_COUNT as usize]), + } + } + + /// Creates a slot map with all slots assigned to a single node. + /// + /// Useful for single-node clusters. + pub fn single_node(node: NodeId) -> Self { + let mut map = Self::new(); + for slot in map.slots.iter_mut() { + *slot = Some(node); + } + map + } + + /// Returns the node that owns the given slot, if assigned. + pub fn owner(&self, slot: u16) -> Option { + self.slots.get(slot as usize).copied().flatten() + } + + /// Assigns a slot to a node. + pub fn assign(&mut self, slot: u16, node: NodeId) { + if let Some(entry) = self.slots.get_mut(slot as usize) { + *entry = Some(node); + } + } + + /// Assigns a range of slots to a node. + pub fn assign_range(&mut self, range: SlotRange, node: NodeId) { + for slot in range.iter() { + self.assign(slot, node); + } + } + + /// Clears the assignment for a slot. + pub fn unassign(&mut self, slot: u16) { + if let Some(entry) = self.slots.get_mut(slot as usize) { + *entry = None; + } + } + + /// Returns true if all slots are assigned to some node. + pub fn is_complete(&self) -> bool { + self.slots.iter().all(|s| s.is_some()) + } + + /// Returns the number of unassigned slots. + pub fn unassigned_count(&self) -> usize { + self.slots.iter().filter(|s| s.is_none()).count() + } + + /// Returns all slots owned by a specific node as a list of ranges. + /// + /// Consecutive slots are merged into ranges for compact representation. + pub fn slots_for_node(&self, node: NodeId) -> Vec { + let mut ranges = Vec::new(); + let mut range_start: Option = None; + let mut prev_slot: Option = None; + + for (slot_idx, owner) in self.slots.iter().enumerate() { + let slot = slot_idx as u16; + let owned = *owner == Some(node); + + match (owned, range_start) { + (true, None) => { + // Start a new range + range_start = Some(slot); + prev_slot = Some(slot); + } + (true, Some(_)) => { + // Continue the current range + prev_slot = Some(slot); + } + (false, Some(start)) => { + // End the current range + if let Some(end) = prev_slot { + ranges.push(SlotRange::new(start, end)); + } + range_start = None; + prev_slot = None; + } + (false, None) => { + // Not in a range, not owned by this node + } + } + } + + // Close any open range at the end + if let (Some(start), Some(end)) = (range_start, prev_slot) { + ranges.push(SlotRange::new(start, end)); + } + + ranges + } + + /// Returns a count of slots per node. + pub fn slot_counts(&self) -> std::collections::HashMap { + let mut counts = std::collections::HashMap::new(); + for owner in self.slots.iter().flatten() { + *counts.entry(*owner).or_insert(0) += 1; + } + counts + } +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + // Test vectors verified against Redis CLUSTER KEYSLOT command + #[test] + fn crc16_matches_redis() { + // These are known Redis slot assignments + assert_eq!(key_slot(b""), 0); + assert_eq!(key_slot(b"foo"), 12182); + assert_eq!(key_slot(b"bar"), 5061); + assert_eq!(key_slot(b"hello"), 866); + // CRC16 CCITT/XMODEM of "123456789" is 0x31C3 = 12739 + assert_eq!(key_slot(b"123456789"), 12739); + } + + #[test] + fn hash_tag_extraction() { + // Basic hash tag + assert_eq!(key_slot(b"user:{123}:profile"), key_slot(b"123")); + assert_eq!(key_slot(b"order:{123}:items"), key_slot(b"123")); + + // Tag at start + assert_eq!(key_slot(b"{user}:123"), key_slot(b"user")); + + // Empty tag uses whole key + assert_eq!(key_slot(b"foo{}bar"), key_slot(b"foo{}bar")); + + // No closing brace uses whole key + assert_eq!(key_slot(b"foo{bar"), key_slot(b"foo{bar")); + + // Only first tag matters + assert_eq!(key_slot(b"{a}{b}"), key_slot(b"a")); + } + + #[test] + fn slot_range_basics() { + let range = SlotRange::new(0, 5460); + assert_eq!(range.len(), 5461); + assert!(range.contains(0)); + assert!(range.contains(5460)); + assert!(!range.contains(5461)); + + let single = SlotRange::single(100); + assert_eq!(single.len(), 1); + assert!(single.contains(100)); + assert!(!single.contains(99)); + assert!(!single.contains(101)); + } + + #[test] + fn slot_range_display() { + assert_eq!(SlotRange::new(0, 5460).to_string(), "0-5460"); + assert_eq!(SlotRange::single(100).to_string(), "100"); + } + + #[test] + fn slot_map_single_node() { + let node = NodeId(Uuid::new_v4()); + let map = SlotMap::single_node(node); + + assert!(map.is_complete()); + assert_eq!(map.unassigned_count(), 0); + assert_eq!(map.owner(0), Some(node)); + assert_eq!(map.owner(SLOT_COUNT - 1), Some(node)); + + let ranges = map.slots_for_node(node); + assert_eq!(ranges.len(), 1); + assert_eq!(ranges[0], SlotRange::new(0, SLOT_COUNT - 1)); + } + + #[test] + fn slot_map_multi_node() { + let node1 = NodeId(Uuid::new_v4()); + let node2 = NodeId(Uuid::new_v4()); + let node3 = NodeId(Uuid::new_v4()); + + let mut map = SlotMap::new(); + assert!(!map.is_complete()); + assert_eq!(map.unassigned_count(), SLOT_COUNT as usize); + + // Assign slots evenly: 0-5460 to node1, 5461-10922 to node2, 10923-16383 to node3 + map.assign_range(SlotRange::new(0, 5460), node1); + map.assign_range(SlotRange::new(5461, 10922), node2); + map.assign_range(SlotRange::new(10923, 16383), node3); + + assert!(map.is_complete()); + + assert_eq!(map.owner(0), Some(node1)); + assert_eq!(map.owner(5460), Some(node1)); + assert_eq!(map.owner(5461), Some(node2)); + assert_eq!(map.owner(10922), Some(node2)); + assert_eq!(map.owner(10923), Some(node3)); + assert_eq!(map.owner(16383), Some(node3)); + + let counts = map.slot_counts(); + assert_eq!(counts.get(&node1), Some(&5461)); + assert_eq!(counts.get(&node2), Some(&5462)); + assert_eq!(counts.get(&node3), Some(&5461)); + } + + #[test] + fn slot_map_unassign() { + let node = NodeId(Uuid::new_v4()); + let mut map = SlotMap::single_node(node); + + map.unassign(100); + assert_eq!(map.owner(100), None); + assert!(!map.is_complete()); + assert_eq!(map.unassigned_count(), 1); + } + + #[test] + fn slots_for_node_ranges() { + let node = NodeId(Uuid::new_v4()); + let mut map = SlotMap::new(); + + // Assign non-contiguous ranges + map.assign_range(SlotRange::new(0, 10), node); + map.assign_range(SlotRange::new(100, 110), node); + map.assign_range(SlotRange::new(200, 200), node); // single slot + + let ranges = map.slots_for_node(node); + assert_eq!(ranges.len(), 3); + assert_eq!(ranges[0], SlotRange::new(0, 10)); + assert_eq!(ranges[1], SlotRange::new(100, 110)); + assert_eq!(ranges[2], SlotRange::new(200, 200)); + } +} diff --git a/crates/ember-cluster/src/topology.rs b/crates/ember-cluster/src/topology.rs new file mode 100644 index 00000000..b5d1f4e6 --- /dev/null +++ b/crates/ember-cluster/src/topology.rs @@ -0,0 +1,603 @@ +//! Cluster topology management. +//! +//! Defines the structure of a cluster: nodes, their roles, health states, +//! and the overall cluster configuration. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::time::Instant; + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::slots::{SlotMap, SlotRange}; +use crate::ClusterError; + +/// Unique identifier for a cluster node. +/// +/// Wraps a UUID v4 for guaranteed uniqueness across the cluster. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct NodeId(pub Uuid); + +impl NodeId { + /// Generates a new random node ID. + pub fn new() -> Self { + Self(Uuid::new_v4()) + } + + /// Creates a node ID from a UUID string. + pub fn parse(s: &str) -> Result { + Ok(Self(Uuid::parse_str(s)?)) + } +} + +impl Default for NodeId { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Display for NodeId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Show first 8 chars for readability (similar to git short hashes) + write!(f, "{}", &self.0.to_string()[..8]) + } +} + +/// The role of a node in the cluster. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum NodeRole { + /// Primary node that owns slots and accepts writes. + Primary, + /// Replica node that mirrors a primary's data. + Replica, +} + +impl std::fmt::Display for NodeRole { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + NodeRole::Primary => write!(f, "primary"), + NodeRole::Replica => write!(f, "replica"), + } + } +} + +/// Status flags for a node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct NodeFlags { + /// Node is the local node (myself). + pub myself: bool, + /// Node is suspected to be failing. + pub pfail: bool, + /// Node has been confirmed as failed by the cluster. + pub fail: bool, + /// Node is performing a handshake (not yet part of cluster). + pub handshake: bool, + /// Node has no address yet. + pub noaddr: bool, +} + +impl NodeFlags { + /// Returns true if the node is considered healthy. + pub fn is_healthy(&self) -> bool { + !self.fail && !self.pfail + } +} + +impl std::fmt::Display for NodeFlags { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let mut flags = Vec::new(); + if self.myself { + flags.push("myself"); + } + if self.pfail { + flags.push("pfail"); + } + if self.fail { + flags.push("fail"); + } + if self.handshake { + flags.push("handshake"); + } + if self.noaddr { + flags.push("noaddr"); + } + if flags.is_empty() { + write!(f, "-") + } else { + write!(f, "{}", flags.join(",")) + } + } +} + +/// Information about a single node in the cluster. +#[derive(Debug, Clone)] +pub struct ClusterNode { + /// Unique node identifier. + pub id: NodeId, + /// Address for client connections. + pub addr: SocketAddr, + /// Address for cluster bus (gossip) connections. + /// Typically addr.port + 10000. + pub cluster_bus_addr: SocketAddr, + /// Node's role in the cluster. + pub role: NodeRole, + /// Slot ranges assigned to this node (only for primaries). + pub slots: Vec, + /// If this is a replica, the ID of its primary. + pub replicates: Option, + /// IDs of nodes replicating this one (if primary). + pub replicas: Vec, + /// Last time we received a message from this node. + pub last_seen: Instant, + /// Last time we sent a ping to this node. + pub last_ping_sent: Option, + /// Last time we received a pong from this node. + pub last_pong_received: Option, + /// Status flags. + pub flags: NodeFlags, + /// Configuration epoch (used for conflict resolution). + pub config_epoch: u64, +} + +impl ClusterNode { + /// Creates a new primary node. + pub fn new_primary(id: NodeId, addr: SocketAddr) -> Self { + let cluster_bus_addr = SocketAddr::new(addr.ip(), addr.port() + 10000); + Self { + id, + addr, + cluster_bus_addr, + role: NodeRole::Primary, + slots: Vec::new(), + replicates: None, + replicas: Vec::new(), + last_seen: Instant::now(), + last_ping_sent: None, + last_pong_received: None, + flags: NodeFlags::default(), + config_epoch: 0, + } + } + + /// Creates a new replica node. + pub fn new_replica(id: NodeId, addr: SocketAddr, primary_id: NodeId) -> Self { + let cluster_bus_addr = SocketAddr::new(addr.ip(), addr.port() + 10000); + Self { + id, + addr, + cluster_bus_addr, + role: NodeRole::Replica, + slots: Vec::new(), + replicates: Some(primary_id), + replicas: Vec::new(), + last_seen: Instant::now(), + last_ping_sent: None, + last_pong_received: None, + flags: NodeFlags::default(), + config_epoch: 0, + } + } + + /// Marks this node as the local node. + pub fn set_myself(&mut self) { + self.flags.myself = true; + } + + /// Returns true if this node is healthy and can serve requests. + pub fn is_healthy(&self) -> bool { + self.flags.is_healthy() + } + + /// Returns the total number of slots owned by this node. + pub fn slot_count(&self) -> u16 { + self.slots.iter().map(|r| r.len()).sum() + } + + /// Formats the node in CLUSTER NODES output format. + pub fn to_cluster_nodes_line(&self, slot_map: &SlotMap) -> String { + let slots_str = if self.role == NodeRole::Primary { + let ranges = slot_map.slots_for_node(self.id); + if ranges.is_empty() { + String::new() + } else { + ranges + .iter() + .map(|r| r.to_string()) + .collect::>() + .join(" ") + } + } else { + String::new() + }; + + let replicates_str = self + .replicates + .map(|id| id.0.to_string()) + .unwrap_or_else(|| "-".to_string()); + + // Format: @ + format!( + "{} {}@{} {} {} {} {} {} connected {}", + self.id.0, + self.addr, + self.cluster_bus_addr.port(), + self.format_flags(), + replicates_str, + self.last_ping_sent + .map(|t| t.elapsed().as_millis() as u64) + .unwrap_or(0), + self.last_pong_received + .map(|t| t.elapsed().as_millis() as u64) + .unwrap_or(0), + self.config_epoch, + slots_str + ) + .trim() + .to_string() + } + + fn format_flags(&self) -> String { + let mut flags = Vec::new(); + + if self.flags.myself { + flags.push("myself"); + } + + match self.role { + NodeRole::Primary => flags.push("master"), + NodeRole::Replica => flags.push("slave"), + } + + if self.flags.fail { + flags.push("fail"); + } else if self.flags.pfail { + flags.push("fail?"); + } + + if self.flags.handshake { + flags.push("handshake"); + } + + if self.flags.noaddr { + flags.push("noaddr"); + } + + flags.join(",") + } +} + +/// The complete state of the cluster as seen by a node. +#[derive(Debug)] +pub struct ClusterState { + /// All known nodes in the cluster, indexed by ID. + pub nodes: HashMap, + /// This node's ID. + pub local_id: NodeId, + /// Current configuration epoch (increases on topology changes). + pub config_epoch: u64, + /// Slot-to-node mapping. + pub slot_map: SlotMap, + /// Cluster state: ok, fail, or unknown. + pub state: ClusterHealth, +} + +/// Overall cluster health status. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClusterHealth { + /// Cluster is operational and all slots are covered. + Ok, + /// Cluster has failed nodes or uncovered slots. + Fail, + /// Cluster state is being computed. + Unknown, +} + +impl std::fmt::Display for ClusterHealth { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ClusterHealth::Ok => write!(f, "ok"), + ClusterHealth::Fail => write!(f, "fail"), + ClusterHealth::Unknown => write!(f, "unknown"), + } + } +} + +impl ClusterState { + /// Creates a new cluster state for a single-node cluster. + pub fn single_node(local_node: ClusterNode) -> Self { + let local_id = local_node.id; + let slot_map = SlotMap::single_node(local_id); + let mut nodes = HashMap::new(); + nodes.insert(local_id, local_node); + + Self { + nodes, + local_id, + config_epoch: 1, + slot_map, + state: ClusterHealth::Ok, + } + } + + /// Creates a new empty cluster state (for joining an existing cluster). + pub fn new(local_id: NodeId) -> Self { + Self { + nodes: HashMap::new(), + local_id, + config_epoch: 0, + slot_map: SlotMap::new(), + state: ClusterHealth::Unknown, + } + } + + /// Returns the local node. + pub fn local_node(&self) -> Option<&ClusterNode> { + self.nodes.get(&self.local_id) + } + + /// Returns a mutable reference to the local node. + pub fn local_node_mut(&mut self) -> Option<&mut ClusterNode> { + self.nodes.get_mut(&self.local_id) + } + + /// Adds a node to the cluster. + pub fn add_node(&mut self, node: ClusterNode) { + self.nodes.insert(node.id, node); + } + + /// Removes a node from the cluster. + pub fn remove_node(&mut self, node_id: NodeId) -> Option { + self.nodes.remove(&node_id) + } + + /// Returns the node that owns the given slot. + pub fn slot_owner(&self, slot: u16) -> Option<&ClusterNode> { + let node_id = self.slot_map.owner(slot)?; + self.nodes.get(&node_id) + } + + /// Returns true if the local node owns the given slot. + pub fn owns_slot(&self, slot: u16) -> bool { + self.slot_map.owner(slot) == Some(self.local_id) + } + + /// Returns all primary nodes. + pub fn primaries(&self) -> impl Iterator { + self.nodes.values().filter(|n| n.role == NodeRole::Primary) + } + + /// Returns all replica nodes. + pub fn replicas(&self) -> impl Iterator { + self.nodes.values().filter(|n| n.role == NodeRole::Replica) + } + + /// Returns replicas of a specific primary. + pub fn replicas_of(&self, primary_id: NodeId) -> impl Iterator { + self.nodes + .values() + .filter(move |n| n.replicates == Some(primary_id)) + } + + /// Computes and updates the cluster health state. + pub fn update_health(&mut self) { + // Check if all slots are covered by healthy primaries + if !self.slot_map.is_complete() { + self.state = ClusterHealth::Fail; + return; + } + + // Check if any slot's owner is unhealthy + for slot in 0..crate::slots::SLOT_COUNT { + if let Some(owner_id) = self.slot_map.owner(slot) { + if let Some(node) = self.nodes.get(&owner_id) { + if !node.is_healthy() { + self.state = ClusterHealth::Fail; + return; + } + } else { + // Owner node not found + self.state = ClusterHealth::Fail; + return; + } + } + } + + self.state = ClusterHealth::Ok; + } + + /// Generates the response for CLUSTER INFO command. + pub fn cluster_info(&self) -> String { + let primaries: Vec<_> = self.primaries().collect(); + let assigned_slots: u16 = primaries.iter().map(|n| n.slot_count()).sum(); + + format!( + "cluster_state:{}\r\n\ + cluster_slots_assigned:{}\r\n\ + cluster_slots_ok:{}\r\n\ + cluster_slots_pfail:0\r\n\ + cluster_slots_fail:0\r\n\ + cluster_known_nodes:{}\r\n\ + cluster_size:{}\r\n\ + cluster_current_epoch:{}\r\n\ + cluster_my_epoch:{}\r\n", + self.state, + assigned_slots, + if self.state == ClusterHealth::Ok { + assigned_slots + } else { + 0 + }, + self.nodes.len(), + primaries.len(), + self.config_epoch, + self.local_node().map(|n| n.config_epoch).unwrap_or(0), + ) + } + + /// Generates the response for CLUSTER NODES command. + pub fn cluster_nodes(&self) -> String { + let mut lines: Vec = self + .nodes + .values() + .map(|node| node.to_cluster_nodes_line(&self.slot_map)) + .collect(); + lines.sort(); // Consistent ordering + lines.join("\n") + } + + /// Generates MOVED redirect information for a slot. + pub fn moved_redirect(&self, slot: u16) -> Result<(u16, SocketAddr), ClusterError> { + let node = self + .slot_owner(slot) + .ok_or(ClusterError::SlotNotAssigned(slot))?; + Ok((slot, node.addr)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr}; + + fn test_addr(port: u16) -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port) + } + + #[test] + fn node_id_display() { + let id = NodeId::new(); + let display = id.to_string(); + assert_eq!(display.len(), 8); + } + + #[test] + fn node_id_parse() { + let id = NodeId::new(); + let parsed = NodeId::parse(&id.0.to_string()).unwrap(); + assert_eq!(id, parsed); + } + + #[test] + fn node_flags_display() { + let mut flags = NodeFlags::default(); + assert_eq!(flags.to_string(), "-"); + + flags.myself = true; + assert_eq!(flags.to_string(), "myself"); + + flags.pfail = true; + assert_eq!(flags.to_string(), "myself,pfail"); + } + + #[test] + fn cluster_node_primary() { + let id = NodeId::new(); + let node = ClusterNode::new_primary(id, test_addr(6379)); + + assert_eq!(node.id, id); + assert_eq!(node.role, NodeRole::Primary); + assert_eq!(node.addr.port(), 6379); + assert_eq!(node.cluster_bus_addr.port(), 16379); + assert!(node.replicates.is_none()); + assert!(node.is_healthy()); + } + + #[test] + fn cluster_node_replica() { + let primary_id = NodeId::new(); + let replica_id = NodeId::new(); + let node = ClusterNode::new_replica(replica_id, test_addr(6380), primary_id); + + assert_eq!(node.id, replica_id); + assert_eq!(node.role, NodeRole::Replica); + assert_eq!(node.replicates, Some(primary_id)); + } + + #[test] + fn cluster_state_single_node() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + + let state = ClusterState::single_node(node); + + assert_eq!(state.local_id, id); + assert!(state.owns_slot(0)); + assert!(state.owns_slot(16383)); + assert_eq!(state.state, ClusterHealth::Ok); + } + + #[test] + fn cluster_state_slot_owner() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + + let state = ClusterState::single_node(node); + + let owner = state.slot_owner(100).unwrap(); + assert_eq!(owner.id, id); + } + + #[test] + fn cluster_state_health_check() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + + let mut state = ClusterState::single_node(node); + state.update_health(); + assert_eq!(state.state, ClusterHealth::Ok); + + // Unassign a slot + state.slot_map.unassign(0); + state.update_health(); + assert_eq!(state.state, ClusterHealth::Fail); + } + + #[test] + fn cluster_info_format() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + + let state = ClusterState::single_node(node); + let info = state.cluster_info(); + + assert!(info.contains("cluster_state:ok")); + assert!(info.contains("cluster_slots_assigned:0")); // slots in node.slots, not slot_map + assert!(info.contains("cluster_known_nodes:1")); + } + + #[test] + fn moved_redirect() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + + let state = ClusterState::single_node(node); + + let (slot, addr) = state.moved_redirect(100).unwrap(); + assert_eq!(slot, 100); + assert_eq!(addr.port(), 6379); + } + + #[test] + fn primaries_and_replicas() { + let primary_id = NodeId::new(); + let replica_id = NodeId::new(); + + let mut primary = ClusterNode::new_primary(primary_id, test_addr(6379)); + primary.set_myself(); + + let mut state = ClusterState::single_node(primary); + + let replica = ClusterNode::new_replica(replica_id, test_addr(6380), primary_id); + state.add_node(replica); + + assert_eq!(state.primaries().count(), 1); + assert_eq!(state.replicas().count(), 1); + assert_eq!(state.replicas_of(primary_id).count(), 1); + } +} From f86835683f7846b19638f334a1b4b063ccf0f261 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 5 Feb 2026 23:20:20 -0500 Subject: [PATCH 2/2] feat(protocol): add cluster commands adds CLUSTER subcommands and ASKING to the protocol layer: - CLUSTER INFO: returns cluster state (stub for now) - CLUSTER NODES: returns node list (stub for now) - CLUSTER SLOTS: returns slot distribution (stub for now) - CLUSTER KEYSLOT: computes hash slot for a key (fully working) - CLUSTER MYID: returns node id (stub for now) - ASKING: migration hint flag (no-op for now) integrates ember-cluster crate into ember-server for slot calculation. includes comprehensive tests for all new command parsing. --- Cargo.lock | 1 + crates/ember-protocol/src/command.rs | 162 ++++++++++++++++++++++++++ crates/ember-server/Cargo.toml | 1 + crates/ember-server/src/connection.rs | 34 ++++++ 4 files changed, 198 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 0fcff27c..67ffa80a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -185,6 +185,7 @@ version = "0.2.0" dependencies = [ "bytes", "clap", + "ember-cluster", "ember-persistence", "ember-protocol", "emberkv-core", diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index dcd7b3cb..c70004f7 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -196,6 +196,25 @@ pub enum Command { /// SCARD . Returns the cardinality (number of members) of a set. SCard { key: String }, + // --- cluster commands --- + /// CLUSTER INFO. Returns cluster state and configuration information. + ClusterInfo, + + /// CLUSTER NODES. Returns the list of cluster nodes. + ClusterNodes, + + /// CLUSTER SLOTS. Returns the slot distribution across nodes. + ClusterSlots, + + /// CLUSTER KEYSLOT . Returns the hash slot for a key. + ClusterKeySlot { key: String }, + + /// CLUSTER MYID. Returns the node's ID. + ClusterMyId, + + /// ASKING. Signals that the next command is for a migrating slot. + Asking, + /// A command we don't recognize (yet). Unknown(String), } @@ -291,6 +310,8 @@ impl Command { "SMEMBERS" => parse_smembers(&frames[1..]), "SISMEMBER" => parse_sismember(&frames[1..]), "SCARD" => parse_scard(&frames[1..]), + "CLUSTER" => parse_cluster(&frames[1..]), + "ASKING" => parse_asking(&frames[1..]), _ => Ok(Command::Unknown(name)), } } @@ -1016,6 +1037,59 @@ fn parse_scard(args: &[Frame]) -> Result { Ok(Command::SCard { key }) } +// --- cluster commands --- + +fn parse_cluster(args: &[Frame]) -> Result { + if args.is_empty() { + return Err(ProtocolError::WrongArity("CLUSTER".into())); + } + + let subcommand = extract_string(&args[0])?.to_ascii_uppercase(); + match subcommand.as_str() { + "INFO" => { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("CLUSTER INFO".into())); + } + Ok(Command::ClusterInfo) + } + "NODES" => { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("CLUSTER NODES".into())); + } + Ok(Command::ClusterNodes) + } + "SLOTS" => { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("CLUSTER SLOTS".into())); + } + Ok(Command::ClusterSlots) + } + "KEYSLOT" => { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("CLUSTER KEYSLOT".into())); + } + let key = extract_string(&args[1])?; + Ok(Command::ClusterKeySlot { key }) + } + "MYID" => { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("CLUSTER MYID".into())); + } + Ok(Command::ClusterMyId) + } + _ => Err(ProtocolError::InvalidCommandFrame(format!( + "unknown CLUSTER subcommand '{subcommand}'" + ))), + } +} + +fn parse_asking(args: &[Frame]) -> Result { + if !args.is_empty() { + return Err(ProtocolError::WrongArity("ASKING".into())); + } + Ok(Command::Asking) +} + #[cfg(test)] mod tests { use super::*; @@ -2506,4 +2580,92 @@ mod tests { Command::SMembers { .. } )); } + + // --- cluster commands --- + + #[test] + fn cluster_info_basic() { + assert_eq!( + Command::from_frame(cmd(&["CLUSTER", "INFO"])).unwrap(), + Command::ClusterInfo, + ); + } + + #[test] + fn cluster_nodes_basic() { + assert_eq!( + Command::from_frame(cmd(&["CLUSTER", "NODES"])).unwrap(), + Command::ClusterNodes, + ); + } + + #[test] + fn cluster_slots_basic() { + assert_eq!( + Command::from_frame(cmd(&["CLUSTER", "SLOTS"])).unwrap(), + Command::ClusterSlots, + ); + } + + #[test] + fn cluster_keyslot_basic() { + assert_eq!( + Command::from_frame(cmd(&["CLUSTER", "KEYSLOT", "mykey"])).unwrap(), + Command::ClusterKeySlot { + key: "mykey".into() + }, + ); + } + + #[test] + fn cluster_keyslot_wrong_arity() { + let err = Command::from_frame(cmd(&["CLUSTER", "KEYSLOT"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn cluster_myid_basic() { + assert_eq!( + Command::from_frame(cmd(&["CLUSTER", "MYID"])).unwrap(), + Command::ClusterMyId, + ); + } + + #[test] + fn cluster_unknown_subcommand() { + let err = Command::from_frame(cmd(&["CLUSTER", "BADCMD"])).unwrap_err(); + assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); + } + + #[test] + fn cluster_no_subcommand() { + let err = Command::from_frame(cmd(&["CLUSTER"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn cluster_case_insensitive() { + assert!(matches!( + Command::from_frame(cmd(&["cluster", "info"])).unwrap(), + Command::ClusterInfo + )); + assert!(matches!( + Command::from_frame(cmd(&["cluster", "keyslot", "k"])).unwrap(), + Command::ClusterKeySlot { .. } + )); + } + + #[test] + fn asking_basic() { + assert_eq!( + Command::from_frame(cmd(&["ASKING"])).unwrap(), + Command::Asking, + ); + } + + #[test] + fn asking_wrong_arity() { + let err = Command::from_frame(cmd(&["ASKING", "extra"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/Cargo.toml b/crates/ember-server/Cargo.toml index 3a9c356d..f1d32499 100644 --- a/crates/ember-server/Cargo.toml +++ b/crates/ember-server/Cargo.toml @@ -14,6 +14,7 @@ bytes = { workspace = true } emberkv-core = { workspace = true } ember-protocol = { workspace = true } ember-persistence = { workspace = true } +ember-cluster = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index a9a00990..cd8b7361 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -838,6 +838,40 @@ async fn execute(cmd: Command, engine: &Engine) -> Frame { } } + // --- cluster commands --- + // Note: Full cluster support requires integration with ember-cluster crate. + // For now, CLUSTER KEYSLOT works, and other commands return stub responses. + Command::ClusterKeySlot { key } => { + let slot = ember_cluster::key_slot(key.as_bytes()); + Frame::Integer(slot as i64) + } + + Command::ClusterInfo => { + // Return minimal info indicating cluster mode is disabled + let info = "cluster_enabled:0\r\n"; + Frame::Bulk(Bytes::from(info)) + } + + Command::ClusterNodes => { + // In non-cluster mode, return empty string + Frame::Bulk(Bytes::from("")) + } + + Command::ClusterSlots => { + // In non-cluster mode, return empty array + Frame::Array(vec![]) + } + + Command::ClusterMyId => { + // In non-cluster mode, return an error + Frame::Error("ERR This instance has cluster support disabled".into()) + } + + Command::Asking => { + // ASKING is a no-op in non-cluster mode, just return OK + Frame::Simple("OK".into()) + } + Command::Unknown(name) => Frame::Error(format!("ERR unknown command '{name}'")), } }