From 4b22586a2aea1e328577ff253f3e0a451cdcf720 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Fri, 6 Feb 2026 08:35:12 -0500 Subject: [PATCH] chore: phase 4 audit fixes and documentation updates fixes: - log warnings when gossip event channel sends fail (was silently ignored) - remove semantically incorrect SlotRange::is_empty() method - add clippy allow for len_without_is_empty with proper documentation documentation: - update README.md: phase 4 complete, 574 tests, ~18k LOC - rewrite ember-cluster README with full feature list and usage - add invariant documentation to SlotRange --- README.md | 6 +-- crates/ember-cluster/README.md | 71 ++++++++++++++++++++++++++---- crates/ember-cluster/src/gossip.rs | 36 +++++++++++++-- crates/ember-cluster/src/slots.rs | 21 +++++---- 4 files changed, 111 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 339bac0b..f5c16d9b 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ crates/ ember-core/ core engine (keyspace, types, sharding) ember-protocol/ RESP3 wire protocol ember-persistence/ AOF and snapshot durability - ember-cluster/ raft, gossip, slot management (wip) + ember-cluster/ raft consensus, gossip, slot management, migration ember-cli/ interactive CLI tool ``` @@ -138,10 +138,10 @@ ember uses a shared-nothing, thread-per-core design inspired by [Dragonfly](http | 1 | foundation (protocol, engine, expiration) | ✅ complete | | 2 | persistence (AOF, snapshots, recovery) | ✅ complete | | 3 | data types (sorted sets, lists, hashes, sets) | ✅ complete | -| 4 | clustering (raft, gossip, slots) | 🚧 not started | +| 4 | clustering (raft, gossip, slots, migration) | ✅ complete | | 5 | developer experience (observability, CLI, clients) | 🚧 not started | -**current**: 49 commands, 487 tests, ~6k lines of code +**current**: 62 commands, 574 tests, ~18k lines of code ## security diff --git a/crates/ember-cluster/README.md b/crates/ember-cluster/README.md index 76b340ba..4ba80e2f 100644 --- a/crates/ember-cluster/README.md +++ b/crates/ember-cluster/README.md @@ -1,15 +1,70 @@ # ember-cluster -distributed coordination for [ember](https://github.com/kacy/ember). will handle cluster topology, gossip-based failure detection, raft consensus, and live slot migration. +distributed coordination for [ember](https://github.com/kacy/ember). provides cluster topology management, gossip-based failure detection, raft consensus, and live slot migration. -> this crate is a stub — cluster support is planned for a future phase. +## features -## planned features +- **slot management** — 16384 hash slots with CRC16 hashing (Redis Cluster compatible) +- **topology tracking** — node identity, roles (primary/replica), health status +- **SWIM gossip** — failure detection with configurable probe intervals and suspicion timeouts +- **raft consensus** — cluster configuration changes via [openraft](https://github.com/datafuselabs/openraft) +- **live migration** — slot resharding without downtime, MOVED/ASK redirects -- hash slot mapping (16384 slots) with `MOVED`/`ASK` redirects -- SWIM gossip protocol for failure detection -- raft consensus (via `openraft`) for leader election and log replication -- live slot migration for rebalancing without downtime +## modules + +| module | description | +|--------|-------------| +| `slots` | CRC16 hash function, slot-to-node mapping, slot ranges | +| `topology` | `NodeId`, `ClusterNode`, `ClusterState`, health tracking | +| `gossip` | SWIM protocol engine, membership events, probe management | +| `raft` | openraft integration, cluster commands, state machine | +| `migration` | migration state machine, batch streaming, key tracking | +| `message` | binary wire format for gossip messages | +| `error` | cluster-specific error types with MOVED/ASK support | + +## usage + +```rust +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)); +``` + +## cluster commands + +the following CLUSTER commands are supported at the protocol layer: + +| command | description | +|---------|-------------| +| `CLUSTER INFO` | cluster state and configuration | +| `CLUSTER NODES` | list of cluster nodes | +| `CLUSTER SLOTS` | slot distribution across nodes | +| `CLUSTER KEYSLOT ` | compute slot for a key | +| `CLUSTER MYID` | local node's ID | +| `CLUSTER MEET ` | add a node to the cluster | +| `CLUSTER ADDSLOTS ...` | assign slots to local node | +| `CLUSTER DELSLOTS ...` | remove slots from local node | +| `CLUSTER SETSLOT IMPORTING/MIGRATING/NODE/STABLE` | migration control | +| `CLUSTER FORGET ` | remove a node | +| `CLUSTER REPLICATE ` | become a replica | +| `CLUSTER FAILOVER [FORCE\|TAKEOVER]` | manual failover | +| `CLUSTER COUNTKEYSINSLOT ` | key count in slot | +| `CLUSTER GETKEYSINSLOT ` | get keys in slot | +| `ASKING` | follow ASK redirect | +| `MIGRATE ` | migrate a key | + +## design notes + +- **in-memory raft log** — persistence is deferred; cluster state survives via gossip +- **single raft group** — all cluster config changes go through one raft group (simpler than multi-raft) +- **server integration pending** — modules are complete but not yet wired into ember-server ## related crates @@ -19,4 +74,4 @@ distributed coordination for [ember](https://github.com/kacy/ember). will handle | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cli](../ember-cli) | interactive command-line client (WIP) | +| [ember-cli](../ember-cli) | interactive command-line client | diff --git a/crates/ember-cluster/src/gossip.rs b/crates/ember-cluster/src/gossip.rs index 029acefd..a04a8668 100644 --- a/crates/ember-cluster/src/gossip.rs +++ b/crates/ember-cluster/src/gossip.rs @@ -388,7 +388,14 @@ impl GossipEngine { if member.state != MemberStatus::Alive { member.state = MemberStatus::Alive; member.state_change = Instant::now(); - let _ = self.event_tx.send(GossipEvent::MemberAlive(*node)).await; + if self + .event_tx + .send(GossipEvent::MemberAlive(*node)) + .await + .is_err() + { + warn!("event channel closed, cannot send MemberAlive event"); + } } } } else { @@ -453,7 +460,14 @@ impl GossipEngine { { member.state = MemberStatus::Dead; member.state_change = Instant::now(); - let _ = self.event_tx.send(GossipEvent::MemberFailed(*node)).await; + if self + .event_tx + .send(GossipEvent::MemberFailed(*node)) + .await + .is_err() + { + warn!("event channel closed, cannot send MemberFailed event"); + } } } } @@ -466,7 +480,14 @@ impl GossipEngine { if member.state != MemberStatus::Left { member.state = MemberStatus::Left; member.state_change = Instant::now(); - let _ = self.event_tx.send(GossipEvent::MemberLeft(*node)).await; + if self + .event_tx + .send(GossipEvent::MemberLeft(*node)) + .await + .is_err() + { + warn!("event channel closed, cannot send MemberLeft event"); + } } } } @@ -479,7 +500,14 @@ impl GossipEngine { if member.state == MemberStatus::Suspect { member.state = MemberStatus::Alive; member.state_change = Instant::now(); - let _ = self.event_tx.send(GossipEvent::MemberAlive(node)).await; + if self + .event_tx + .send(GossipEvent::MemberAlive(node)) + .await + .is_err() + { + warn!("event channel closed, cannot send MemberAlive event"); + } } } } diff --git a/crates/ember-cluster/src/slots.rs b/crates/ember-cluster/src/slots.rs index b78774c3..7f52147f 100644 --- a/crates/ember-cluster/src/slots.rs +++ b/crates/ember-cluster/src/slots.rs @@ -98,6 +98,11 @@ pub fn key_slot(key: &[u8]) -> u16 { } /// A contiguous range of slots assigned to a node. +/// +/// # Invariants +/// +/// A valid `SlotRange` always satisfies `start <= end`, meaning it contains +/// at least one slot. This is enforced by debug assertions in the constructor. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct SlotRange { pub start: u16, @@ -106,9 +111,13 @@ pub struct SlotRange { impl SlotRange { /// Creates a new slot range (end is inclusive). + /// + /// # Panics + /// + /// Debug-panics if `start > end` or if `end >= SLOT_COUNT`. pub fn new(start: u16, end: u16) -> Self { - debug_assert!(start <= end); - debug_assert!(end < SLOT_COUNT); + debug_assert!(start <= end, "SlotRange requires start <= end"); + debug_assert!(end < SLOT_COUNT, "slot must be < {SLOT_COUNT}"); Self { start, end } } @@ -117,16 +126,12 @@ impl SlotRange { Self::new(slot, slot) } - /// Returns the number of slots in this range. + /// Returns the number of slots in this range (always >= 1 for valid ranges). + #[allow(clippy::len_without_is_empty)] 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