Skip to content

feat(cluster): control plane on the default runtime + formation fixes (v0.9 W0/C-1, #405) #450

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TinDang97 merged 2 commits into from
Aug 7, 2026
Merged

feat(cluster): control plane on the default runtime + formation fixes (v0.9 W0/C-1, #405) #450

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d573036
feat(cluster): control plane on the default runtime + formation fixes…
TinDang97 Aug 7, 2026
fd1ab67
fix(cluster): review round — MEET idempotence, fail-loud bus bind, po…
TinDang97 Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]

### Added
- **Cluster mode now works on the default (monoio) runtime — v0.9 W0/C-1
(#405).** The cluster control plane (bus listener, gossip ticker, failover
election) runs on a dedicated `cluster-ctl` std thread hosting a
current-thread tokio runtime, on BOTH server runtimes. Before this the
monoio startup path never spawned the bus or the ticker: `--cluster-enabled`
accepted `CLUSTER MEET` but no peer ever learned anything. Under tokio the
control plane previously shared the listener runtime with the accept loop
and every connection, so gossip starved under load (observed as PFAIL
detection stalling); the dedicated thread fixes that too. The unused,
untested monoio duplicates of the bus/gossip/election code were deleted —
one control-plane implementation on both runtimes. New e2e suite
`tests/cluster_formation.rs` proves a real 3-node cluster forms via
MEET + gossip and flags a killed node, per runtime.

### Fixed
- **Cluster formation actually converges (pre-existing, both runtimes).**
A 3-node cluster could never complete its mesh: (1) `CLUSTER MEET`'s
random-id placeholder was never retired when the peer's handshake arrived
under its real id, double-counting every met peer (`known_nodes` 5 in a
3-node cluster); (2) gossip sections were only consumed for PFAIL/FAIL
reports, so nodes MEET-ed into a common peer never learned about EACH
OTHER; (3) nodes adopted from rumors started with `pong_recv_ms = 0`,
which `check_failure_states` skips — a rumored node that died before
first direct contact could NEVER be marked PFAIL. Placeholders are now
retired on handshake (same-address, different-id), healthy rumors are
adopted (with self/known-address guards), and adoption stamps a freshness
baseline so the staleness clock always runs. Review round: `CLUSTER MEET`
is idempotent by ADDRESS (repeats no longer stack one placeholder per
call) and refuses the node's own address; a cluster-bus bind failure now
aborts startup loudly instead of leaving a node serving clients while
invisible to every peer; `--cluster-enabled` refuses ports > 55535 (bus
port would wrap past 65535); a `cluster-ctl` thread panic aborts the
process via the same hook that guards shard threads.
- **The AOF now compacts itself (#433): Redis-parity automatic rewrite.**
`--auto-aof-rewrite-percentage` (default 100, `0` disables) and
`--auto-aof-rewrite-min-size` (default `64mb`, size strings accepted)
Expand Down
6 changes: 5 additions & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,11 @@ flume = "0.12"
atomic-waker = "1"
# Base features (rt, net) are always available for the admin HTTP server.
# runtime-tokio adds the full feature set (multi-thread, io-util, signal, etc.).
tokio = { version = "1", features = ["rt", "net", "macros"] }
# Base features include "time" + "io-util" for the cluster control plane
# (bus + gossip + election), which is tokio-native on BOTH runtimes — under
# monoio it runs on a dedicated std thread with a current-thread runtime
# (v0.9 C-1, #405).
tokio = { version = "1", features = ["rt", "net", "macros", "time", "io-util"] }
tokio-util = { version = "0.7", features = ["codec"], optional = true }
clap = { version = "4", features = ["derive", "env"] }
tracing = "0.1"
Expand Down
210 changes: 9 additions & 201 deletions src/cluster/bus.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,7 @@ use std::net::SocketAddr;
use std::sync::{Arc, RwLock};

use crate::runtime::cancel::CancellationToken;
#[cfg(feature = "runtime-monoio")]
use monoio::io::{AsyncReadRent, AsyncWriteRentExt};
#[cfg(feature = "runtime-tokio")]
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[cfg(feature = "runtime-tokio")]
use tokio::net::{TcpListener, TcpStream};
use tracing::{debug, info, warn};

Expand Down Expand Up @@ -46,18 +42,20 @@ pub(crate) const MAX_GOSSIP_FRAME_LEN: usize = 64 * 1024;
///
/// Spawns a new task for each incoming peer connection.
/// Should be spawned on the listener runtime as a separate task.
#[cfg(feature = "runtime-tokio")]
///
/// Takes an already-bound listener: binding happens in the caller so a bind
/// failure (EADDRINUSE, bad bind address) aborts startup loudly instead of
/// leaving a node that serves clients while being invisible to every peer.
pub async fn run_cluster_bus(
bind: &str,
cluster_port: u16,
listener: TcpListener,
self_addr: SocketAddr,
cluster_state: Arc<RwLock<ClusterState>>,
shutdown: CancellationToken,
vote_tx: SharedVoteTx,
) -> anyhow::Result<()> {
let addr = format!("{}:{}", bind, cluster_port);
let listener = TcpListener::bind(&addr).await?;
info!("Cluster bus listening on {}", addr);
) {
if let Ok(addr) = listener.local_addr() {
info!("Cluster bus listening on {}", addr);
}

loop {
tokio::select! {
Expand Down Expand Up @@ -85,15 +83,13 @@ pub async fn run_cluster_bus(
}
}
}
Ok(())
}

/// Handle a single cluster peer connection.
///
/// Reads length-prefixed gossip messages in a loop.
/// For PING/MEET: responds with PONG.
/// For PONG: merges into our state.
#[cfg(feature = "runtime-tokio")]
async fn handle_cluster_peer(
mut stream: TcpStream,
peer_addr: SocketAddr,
Expand Down Expand Up @@ -201,191 +197,3 @@ async fn handle_cluster_peer(
}
}
}

/// Read exactly `total` bytes from a monoio TcpStream using ownership-based I/O.
///
/// Monoio has no `read_exact` — we loop on `stream.read()` accumulating bytes.
#[cfg(feature = "runtime-monoio")]
pub(crate) async fn monoio_read_exact(
stream: &mut monoio::net::TcpStream,
total: usize,
) -> std::io::Result<Vec<u8>> {
let mut result = Vec::with_capacity(total);
while result.len() < total {
let remaining = total - result.len();
let buf = vec![0u8; remaining];
let (res, buf) = stream.read(buf).await;
let n = res?;
if n == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"EOF",
));
}
result.extend_from_slice(&buf[..n]);
}
Ok(result)
}

/// Run the cluster bus listener loop (monoio variant).
///
/// Uses monoio::net::TcpListener and monoio::select!/monoio::spawn.
#[cfg(feature = "runtime-monoio")]
pub async fn run_cluster_bus(
bind: &str,
cluster_port: u16,
self_addr: SocketAddr,
cluster_state: Arc<RwLock<ClusterState>>,
shutdown: CancellationToken,
vote_tx: SharedVoteTx,
) -> anyhow::Result<()> {
let addr = format!("{}:{}", bind, cluster_port);
let listener = monoio::net::TcpListener::bind(&addr)?;
info!("Cluster bus listening on {}", addr);

loop {
monoio::select! {
result = listener.accept() => {
match result {
Ok((stream, peer_addr)) => {
let cs = cluster_state.clone();
let tok = shutdown.child_token();
let sa = self_addr;
let vtx = vote_tx.clone();
monoio::spawn(async move {
if let Err(e) = handle_cluster_peer(stream, peer_addr, sa, cs, tok, vtx).await {
debug!("Cluster peer {} error: {}", peer_addr, e);
}
});
}
Err(e) => {
warn!("Cluster bus accept error: {}", e);
}
}
}
_ = shutdown.cancelled() => {
info!("Cluster bus shutting down");
break;
}
}
}
Ok(())
}

/// Handle a single cluster peer connection (monoio variant).
///
/// Uses ownership-based I/O via monoio_read_exact helper and AsyncWriteRentExt.
#[cfg(feature = "runtime-monoio")]
async fn handle_cluster_peer(
mut stream: monoio::net::TcpStream,
peer_addr: SocketAddr,
self_addr: SocketAddr,
cluster_state: Arc<RwLock<ClusterState>>,
shutdown: CancellationToken,
vote_tx: SharedVoteTx,
) -> anyhow::Result<()> {
loop {
// Read 4-byte length prefix with shutdown check
let len_data = monoio::select! {
result = monoio_read_exact(&mut stream, 4) => {
result.map_err(|e| anyhow::anyhow!(e))?
}
_ = shutdown.cancelled() => return Ok(()),
};
let len_buf: [u8; 4] = len_data[..4].try_into().unwrap();
let msg_len = u32::from_be_bytes(len_buf) as usize;
if msg_len > MAX_GOSSIP_FRAME_LEN {
anyhow::bail!("gossip message too large: {} bytes", msg_len);
}

// Read message body — #10: bound with a timeout + shutdown-cancel so a
// client that sends the length prefix but stalls the body cannot pin
// the task/socket forever or dodge graceful shutdown.
let buf = monoio::select! {
result = monoio_read_exact(&mut stream, msg_len) => {
result.map_err(|e| anyhow::anyhow!(e))?
}
_ = shutdown.cancelled() => return Ok(()),
_ = monoio::time::sleep(GOSSIP_BODY_READ_TIMEOUT) => {
anyhow::bail!(
"gossip body read from {} timed out after {}s",
peer_addr,
GOSSIP_BODY_READ_TIMEOUT.as_secs()
);
}
};

// Deserialize
let msg = match deserialize_gossip(&buf) {
Ok(m) => m,
Err(e) => {
warn!("Bad gossip from {}: {}", peer_addr, e);
continue;
}
};

match msg.msg_type {
GossipMsgType::Ping | GossipMsgType::Meet => {
// Merge their state into ours
{
let mut cs = cluster_state.write().unwrap();
merge_gossip_into_state(&mut cs, &msg);
}
// Respond with PONG
let pong = {
let cs = cluster_state.read().unwrap();
build_message(&cs, self_addr, GossipMsgType::Pong)
};
let pong_bytes = serialize_gossip(&pong);
let len = (pong_bytes.len() as u32).to_be_bytes();
let len_vec = len.to_vec();
let (wr, _) = stream.write_all(len_vec).await;
wr.map_err(|e| anyhow::anyhow!(e))?;
let (wr, _) = stream.write_all(pong_bytes).await;
wr.map_err(|e| anyhow::anyhow!(e))?;
}
GossipMsgType::Pong => {
let mut cs = cluster_state.write().unwrap();
merge_gossip_into_state(&mut cs, &msg);
}
GossipMsgType::FailoverAuthRequest => {
let sender_id = std::str::from_utf8(&msg.sender_node_id)
.unwrap_or("")
.trim_end_matches('\0')
.to_string();
let request_epoch = msg.config_epoch;
let voted = {
let mut cs = cluster_state.write().unwrap();
crate::cluster::failover::handle_failover_auth_request(
&mut cs,
&sender_id,
request_epoch,
)
};
if voted {
let ack = {
let cs = cluster_state.read().unwrap();
build_message(&cs, self_addr, GossipMsgType::FailoverAuthAck)
};
let ack_bytes = serialize_gossip(&ack);
let len = (ack_bytes.len() as u32).to_be_bytes();
let len_vec = len.to_vec();
let (wr, _) = stream.write_all(len_vec).await;
let _ = wr;
let (wr, _) = stream.write_all(ack_bytes).await;
let _ = wr;
}
}
GossipMsgType::FailoverAuthAck => {
let sender_id = std::str::from_utf8(&msg.sender_node_id)
.unwrap_or("")
.trim_end_matches('\0')
.to_string();
debug!("Received failover ACK from {}", sender_id);
if let Some(tx) = vote_tx.lock().as_ref() {
let _ = tx.send(sender_id);
}
}
}
}
}
57 changes: 54 additions & 3 deletions src/cluster/command.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,13 +269,25 @@ pub fn handle_cluster_meet(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -> Fr
.parse()
.unwrap_or_else(|_| "127.0.0.1:0".parse().unwrap());

// Generate a placeholder node ID (will be replaced by gossip handshake)
// Generate a placeholder node ID (retired by the gossip handshake, which
// re-registers the peer under its real id)
use crate::replication::state::generate_repl_id;
let peer_id = generate_repl_id();

let mut state = cs.write().unwrap();
if !state.nodes.contains_key(&peer_id) {
let node = ClusterNode::new(peer_id.clone(), addr, NodeFlags::Master, 0);
if state.my_node().addr == addr {
return Frame::Error(Bytes::from_static(b"ERR Can't MEET myself"));
}
// Idempotence by ADDRESS, not by the (fresh-random) placeholder id:
// repeated MEETs for one address — or a MEET for a peer we already
// know — must not stack a new placeholder per call.
if !state.nodes.values().any(|n| n.addr == addr) {
let mut node = ClusterNode::new(peer_id.clone(), addr, NodeFlags::Master, 0);
// Freshness baseline: check_failure_states skips pong_recv_ms == 0
// entries, so MEET-ing a dead address would otherwise leave an
// entry that never goes PFAIL. The handshake replaces this
// placeholder (and its clock) with the peer's real identity.
node.pong_recv_ms = crate::cluster::gossip::now_ms();
state.nodes.insert(peer_id, node);
}
Frame::SimpleString(Bytes::from_static(b"OK"))
Expand Down Expand Up @@ -754,6 +766,45 @@ mod tests {
assert_eq!(cs.read().unwrap().nodes.len(), 2);
}

/// Repeated CLUSTER MEET for one address must not stack placeholders:
/// the placeholder id is fresh-random per call, so idempotence has to
/// key on the address.
#[test]
fn test_cluster_meet_is_idempotent_by_address() {
let cs = make_cs();
let args = vec![
Frame::BulkString(bytes::Bytes::from_static(b"MEET")),
Frame::BulkString(bytes::Bytes::from_static(b"192.168.1.2")),
Frame::BulkString(bytes::Bytes::from_static(b"6380")),
];
for _ in 0..3 {
let result = handle_cluster_command(&args, &cs, "127.0.0.1:6379".parse().unwrap());
assert!(matches!(result, Frame::SimpleString(_)));
}
assert_eq!(
cs.read().unwrap().nodes.len(),
2,
"one placeholder, not three"
);
}

/// MEET-ing our own advertised address is refused.
#[test]
fn test_cluster_meet_self_is_error() {
let cs = make_cs();
let args = vec![
Frame::BulkString(bytes::Bytes::from_static(b"MEET")),
Frame::BulkString(bytes::Bytes::from_static(b"127.0.0.1")),
Frame::BulkString(bytes::Bytes::from_static(b"6379")),
];
let result = handle_cluster_command(&args, &cs, "127.0.0.1:6379".parse().unwrap());
assert!(
matches!(result, Frame::Error(_)),
"self-MEET must be an error"
);
assert_eq!(cs.read().unwrap().nodes.len(), 1);
}

/// Helper: create a ClusterState where this node is a replica of a FAIL master.
fn make_replica_with_fail_master() -> Arc<RwLock<ClusterState>> {
let my_id = "a".repeat(40);
Expand Down
Loading
Loading