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

CHANGELOG.md

@@ -7,6 +7,33 @@ 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.
 - **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)

Cargo.toml

@@ -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"

src/cluster/bus.rs

@@ -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};
 
@@ -46,7 +42,6 @@ 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")]
 pub async fn run_cluster_bus(
     bind: &str,
     cluster_port: u16,
@@ -93,7 +88,6 @@ pub async fn run_cluster_bus(
 /// 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,
@@ -201,191 +195,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);
-                }
-            }
-        }
-    }
-}

src/cluster/command.rs

@@ -275,7 +275,12 @@ pub fn handle_cluster_meet(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -> Fr
 
     let mut state = cs.write().unwrap();
     if !state.nodes.contains_key(&peer_id) {
-        let node = ClusterNode::new(peer_id.clone(), addr, NodeFlags::Master, 0);
+        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"))