diff --git a/CHANGELOG.md b/CHANGELOG.md index 05da0c909..e5487e070 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/Cargo.toml b/Cargo.toml index 81f9bf6b0..cdc04548c 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/cluster/bus.rs b/src/cluster/bus.rs index 3c775f4a9..18dd47b9a 100644 --- a/src/cluster/bus.rs +++ b/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,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>, 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! { @@ -85,7 +83,6 @@ pub async fn run_cluster_bus( } } } - Ok(()) } /// Handle a single cluster peer connection. @@ -93,7 +90,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 +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> { - 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>, - 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>, - 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); - } - } - } - } -} diff --git a/src/cluster/command.rs b/src/cluster/command.rs index d834c4d63..e362aa03e 100644 --- a/src/cluster/command.rs +++ b/src/cluster/command.rs @@ -269,13 +269,25 @@ pub fn handle_cluster_meet(args: &[Frame], cs: &Arc>) -> 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")) @@ -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> { let my_id = "a".repeat(40); diff --git a/src/cluster/failover.rs b/src/cluster/failover.rs index 149cf2d04..ef24bf255 100644 --- a/src/cluster/failover.rs +++ b/src/cluster/failover.rs @@ -17,12 +17,8 @@ use std::net::SocketAddr; use std::sync::{Arc, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; -#[cfg(feature = "runtime-monoio")] -use monoio::io::AsyncWriteRentExt; use rand::RngExt; -#[cfg(feature = "runtime-tokio")] use tokio::io::AsyncWriteExt; -#[cfg(feature = "runtime-tokio")] use tokio::net::TcpStream; use tracing::{info, warn}; @@ -188,7 +184,6 @@ pub fn compute_failover_delay(replica_rank: u32) -> u64 { /// to all masters, collects votes, and promotes on majority. /// /// Spawned when this replica detects its master is FAIL. -#[cfg(feature = "runtime-tokio")] pub async fn run_election_task( cluster_state: Arc>, self_addr: SocketAddr, @@ -310,140 +305,6 @@ pub async fn run_election_task( } } -/// Async election task (monoio variant): waits the jittered delay, sends -/// FailoverAuthRequest to all masters, collects votes, and promotes on majority. -/// -/// Uses monoio::time::sleep, monoio::spawn, monoio::select!, and ownership I/O. -#[cfg(feature = "runtime-monoio")] -pub async fn run_election_task( - cluster_state: Arc>, - self_addr: SocketAddr, - _my_repl_offset: u64, - vote_rx: crate::runtime::channel::MpscReceiver, -) { - // Compute delay (rank 0 for now; multi-replica ranking is future work) - let replica_rank = 0u32; - let delay = compute_failover_delay(replica_rank); - - // Set state to WaitingDelay - { - let mut cs = cluster_state.write().unwrap(); - cs.failover_state = FailoverState::WaitingDelay { - start_ms: now_ms(), - delay_ms: delay, - }; - } - - monoio::time::sleep(std::time::Duration::from_millis(delay)).await; - - // Increment epoch and build FailoverAuthRequest - let (new_epoch, quorum, master_addrs) = { - let mut cs = cluster_state.write().unwrap(); - cs.epoch += 1; - let new_epoch = cs.epoch; - let quorum = cs.quorum(); - let my_id = cs.node_id.clone(); - - // Collect bus addresses of all known masters - let addrs: Vec = cs - .nodes - .values() - .filter(|n| n.node_id != my_id && matches!(n.flags, NodeFlags::Master)) - .map(|n| SocketAddr::new(n.addr.ip(), n.bus_port)) - .collect(); - - cs.failover_state = FailoverState::WaitingVotes { - epoch: new_epoch, - votes_received: 1, // self-vote - votes_needed: quorum, - }; - - (new_epoch, quorum, addrs) - }; - - // Build the auth request message - let auth_msg = { - let cs = cluster_state.read().unwrap(); - let mut msg = build_message(&cs, self_addr, GossipMsgType::FailoverAuthRequest); - msg.config_epoch = new_epoch; - msg - }; - - // Send to all masters - let data = serialize_gossip(&auth_msg); - for addr in &master_addrs { - let data = data.clone(); - let addr = *addr; - monoio::spawn(async move { - if let Ok(mut stream) = monoio::net::TcpStream::connect(addr).await { - let len = (data.len() as u32).to_be_bytes(); - let len_vec = len.to_vec(); - let (wr, _) = stream.write_all(len_vec).await; - if wr.is_ok() { - let _ = stream.write_all(data).await; - } - } - }); - } - - info!( - "Failover election started: epoch={}, sent auth request to {} masters, need {} votes", - new_epoch, - master_addrs.len(), - quorum - ); - - // Collect votes with 5-second timeout - let mut votes: u32 = 1; // self-vote - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - - loop { - if votes >= quorum { - break; - } - let remaining = deadline.saturating_duration_since(std::time::Instant::now()); - if remaining.is_zero() { - break; - } - monoio::select! { - result = vote_rx.recv_async() => { - match result { - Ok(_voter_id) => { - votes += 1; - let mut cs = cluster_state.write().unwrap(); - if let FailoverState::WaitingVotes { - ref mut votes_received, - .. - } = cs.failover_state - { - *votes_received = votes; - } - } - Err(_) => break, - } - } - _ = monoio::time::sleep(remaining) => break, - } - } - - if votes >= quorum { - info!( - "Failover election won: epoch={}, votes={}/{}", - new_epoch, votes, quorum - ); - let mut cs = cluster_state.write().unwrap(); - check_and_initiate_failover(&mut cs, _my_repl_offset); - cs.failover_state = FailoverState::None; - } else { - warn!( - "Failover election timed out: epoch={}, votes={}/{}", - new_epoch, votes, quorum - ); - let mut cs = cluster_state.write().unwrap(); - cs.failover_state = FailoverState::None; - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/cluster/gossip.rs b/src/cluster/gossip.rs index 41da2b384..15ce4eafd 100644 --- a/src/cluster/gossip.rs +++ b/src/cluster/gossip.rs @@ -10,11 +10,7 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; 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::TcpStream; use tracing::warn; @@ -194,7 +190,7 @@ pub fn deserialize_gossip(data: &[u8]) -> Result { } /// Return current unix milliseconds. -fn now_ms() -> u64 { +pub(crate) fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() @@ -276,16 +272,19 @@ pub fn merge_gossip_into_state(state: &mut ClusterState, msg: &GossipMessage) { return; } - // Update or insert sender node - let entry = state.nodes.entry(node_id_str.clone()).or_insert_with(|| { + let sender_addr = { let ip_str = std::str::from_utf8(&msg.sender_ip) .unwrap_or("127.0.0.1") .trim_end_matches('\0'); let ip: IpAddr = ip_str.parse().unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST)); - let addr = SocketAddr::new(ip, msg.sender_port); + SocketAddr::new(ip, msg.sender_port) + }; + + // Update or insert sender node + let entry = state.nodes.entry(node_id_str.clone()).or_insert_with(|| { ClusterNode::new( node_id_str.clone(), - addr, + sender_addr, NodeFlags::Master, msg.config_epoch, ) @@ -296,21 +295,34 @@ pub fn merge_gossip_into_state(state: &mut ClusterState, msg: &GossipMessage) { entry.slots = msg.sender_slots.clone(); } + // C-1 formation fix: CLUSTER MEET registers the peer under a RANDOM + // placeholder id ("replaced by the handshake") — but the handshake lands + // HERE under the peer's real id, so the placeholder must die now or every + // met peer is double-counted forever (observed as known_nodes 5 in a + // 3-node cluster). The same rule retires a stale entry for a node that + // restarted at the same address under a fresh id. + let self_id = state.node_id.clone(); + state + .nodes + .retain(|id, n| *id == node_id_str || *id == self_id || n.addr != sender_addr); + state.messages_received += 1; - // Process gossip sections for PFAIL/FAIL reports. - // When a peer reports another node as PFAIL (flags=2) or FAIL (flags=3), - // record the reporter in the target node's pfail_reports. + // Process gossip sections. Failure rumors (flags=2 PFAIL / flags=3 FAIL) + // feed the pfail consensus; healthy rumors teach us nodes we have never + // talked to directly — without that, two nodes MEET-ed into a third never + // learn about EACH OTHER and the mesh cannot complete. + let my_addr = state.nodes.get(&self_id).map(|n| n.addr); for section in &msg.gossip_sections { let section_flags = section.flags; + let target_node_id = std::str::from_utf8(§ion.node_id) + .unwrap_or("") + .trim_end_matches('\0') + .to_string(); + if target_node_id.is_empty() || target_node_id == state.node_id { + continue; + } if section_flags == 2 || section_flags == 3 { - let target_node_id = std::str::from_utf8(§ion.node_id) - .unwrap_or("") - .trim_end_matches('\0') - .to_string(); - if target_node_id.is_empty() || target_node_id == state.node_id { - continue; - } let reporter_id = node_id_str.clone(); let now = now_ms(); @@ -321,6 +333,42 @@ pub fn merge_gossip_into_state(state: &mut ClusterState, msg: &GossipMessage) { // Check if majority consensus reached for PFAIL->FAIL crate::cluster::failover::try_mark_fail_with_consensus(state, &target_node_id); + } else if !state.nodes.contains_key(&target_node_id) { + // C-1 formation fix: adopt healthy unknown nodes from rumors. + // Skip rumors without a usable address, rumors pointing at our + // own address (a peer's not-yet-resolved MEET placeholder for + // us), and addresses we already know under another id (the real + // entry wins; placeholder rumors die out on direct contact). + let ip_str = std::str::from_utf8(§ion.ip) + .unwrap_or("") + .trim_end_matches('\0'); + let Ok(ip) = ip_str.parse::() else { + continue; + }; + let addr = SocketAddr::new(ip, section.port); + if section.port == 0 + || Some(addr) == my_addr + || state.nodes.values().any(|n| n.addr == addr) + { + continue; + } + // Flags from rumors are advisory; the node starts as Master and + // direct contact refreshes liveness (replica linkage propagates + // via the failover path, not rumors). + let mut node = ClusterNode::new( + target_node_id.clone(), + addr, + NodeFlags::Master, + section.epoch, + ); + node.bus_port = section.bus_port; + // Freshness baseline: `check_failure_states` skips entries with + // pong_recv_ms == 0, so a rumored node that dies before our + // first direct contact would otherwise NEVER go PFAIL on this + // node (observed as one survivor permanently not flagging a + // killed peer). Adoption time starts the staleness clock. + node.pong_recv_ms = now_ms(); + state.nodes.insert(target_node_id, node); } } } @@ -353,7 +401,6 @@ pub fn check_failure_states(state: &mut ClusterState, node_timeout_ms: u64) { /// /// Runs as a separate async task on the listener runtime (NOT on shard threads). /// Also monitors for master FAIL and spawns election task for replicas. -#[cfg(feature = "runtime-tokio")] pub async fn run_gossip_ticker( self_addr: SocketAddr, cluster_state: Arc>, @@ -482,151 +529,122 @@ pub async fn run_gossip_ticker( } } -/// Background gossip ticker (monoio variant): sends PING to a random peer every 100ms. -/// -/// Uses `loop { monoio::time::sleep() }` instead of `tokio::time::interval`. -/// Uses `monoio::spawn` for election task and ping senders. -/// Uses ownership-based I/O for all TCP reads/writes. -#[cfg(feature = "runtime-monoio")] -pub async fn run_gossip_ticker( - self_addr: SocketAddr, - cluster_state: Arc>, - node_timeout_ms: u64, - shutdown: CancellationToken, - vote_tx: SharedVoteTx, - repl_state: std::sync::Arc>, -) { - let mut election_spawned = false; - // Bound outstanding PING probes (see the tokio variant): one probe at a - // time, rotating across peers, so a dead peer can't leak a task per tick. - let ping_in_flight = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let mut ping_rotation: usize = 0; - loop { - monoio::select! { - _ = monoio::time::sleep(Duration::from_millis(100)) => { - // Pick a random peer to PING - let (target_addr, ping_msg) = { - let mut cs = cluster_state.write().unwrap(); - check_failure_states(&mut cs, node_timeout_ms); +#[cfg(test)] +mod tests { + use super::*; - // Reset election_spawned when failover state returns to None - if election_spawned && cs.failover_state == FailoverState::None { - election_spawned = false; - } + fn id40(c: u8) -> String { + String::from_utf8(vec![c; 40]).unwrap() + } - // Check if we should initiate failover (replica with FAIL master) - if !election_spawned { - let my_flags = cs.my_node().flags.clone(); - if let NodeFlags::Replica { ref master_id } = my_flags { - let master_is_fail = cs.nodes.get(master_id) - .map(|n| matches!(n.flags, NodeFlags::Fail)) - .unwrap_or(false); - if master_is_fail && cs.failover_state == FailoverState::None { - election_spawned = true; - let (tx, rx) = crate::runtime::channel::mpsc_unbounded(); - { - let mut guard = vote_tx.lock(); - *guard = Some(tx); - } - let cs_election = cluster_state.clone(); - let sa = self_addr; - let offset = repl_state.read().total_offset(); - let vtx = vote_tx.clone(); - monoio::spawn(async move { - crate::cluster::failover::run_election_task( - cs_election, sa, offset, rx, - ).await; - // Clear vote_tx when election ends - *vtx.lock() = None; - }); - } - } - } + fn test_msg_from(node_id: &str, port: u16, sections: Vec) -> GossipMessage { + let mut sender_node_id = [0u8; 40]; + sender_node_id.copy_from_slice(node_id.as_bytes()); + let mut sender_ip = [0u8; 16]; + sender_ip[..9].copy_from_slice(b"127.0.0.1"); + GossipMessage { + msg_type: GossipMsgType::Pong, + sender_node_id, + sender_slots: Box::new([0u8; 2048]), + config_epoch: 0, + sender_ip, + sender_port: port, + sender_bus_port: port + 10000, + gossip_sections: sections, + } + } - // Rotate across peers instead of always pinging the first. - let peer_count = cs - .nodes - .values() - .filter(|n| n.node_id != cs.node_id) - .count(); - let target = if peer_count == 0 { - None - } else { - let idx = ping_rotation % peer_count; - ping_rotation = ping_rotation.wrapping_add(1); - cs.nodes - .values() - .filter(|n| n.node_id != cs.node_id) - .nth(idx) - .map(|n| SocketAddr::new(n.addr.ip(), n.bus_port)) - }; - let msg = build_message(&cs, self_addr, GossipMsgType::Ping); - cs.messages_sent += 1; - (target, msg) - }; - if let Some(target_addr) = target_addr { - // Skip if the previous probe hasn't resolved yet — bounds - // outstanding tasks to one even against a dead peer, and a - // sleep-race timeout cancels a hung connect/read. - if !ping_in_flight.swap(true, std::sync::atomic::Ordering::AcqRel) { - let cs = cluster_state.clone(); - let inflight = ping_in_flight.clone(); - let probe_timeout = - Duration::from_millis((node_timeout_ms / 2).max(100)); - monoio::spawn(async move { - monoio::select! { - _ = async { - if let Ok(mut stream) = monoio::net::TcpStream::connect(target_addr).await { - let data = serialize_gossip(&ping_msg); - let len = (data.len() as u32).to_be_bytes(); - let len_vec = len.to_vec(); - let (wr, _) = stream.write_all(len_vec).await; - if wr.is_err() { return; } - let (wr, _) = stream.write_all(data).await; - if wr.is_err() { return; } - // Read PONG response. Exact-read loops: - // a single `read()` can return short, and - // dropping a fragmented-but-valid PONG - // leaves `pong_recv_ms` stale, pushing a - // healthy peer toward PFAIL. Length is - // capped like the bus listener (the prefix - // is peer-controlled and sizes an alloc). - if let Ok(len_buf) = - crate::cluster::bus::monoio_read_exact(&mut stream, 4).await - { - let pong_len = u32::from_be_bytes([ - len_buf[0], len_buf[1], len_buf[2], len_buf[3], - ]) as usize; - if pong_len <= crate::cluster::bus::MAX_GOSSIP_FRAME_LEN - && let Ok(pong_buf) = - crate::cluster::bus::monoio_read_exact( - &mut stream, - pong_len, - ) - .await - && let Ok(pong) = deserialize_gossip(&pong_buf) - { - let mut cs2 = cs.write().unwrap(); - merge_gossip_into_state(&mut cs2, &pong); - } - } - } - } => {} - _ = monoio::time::sleep(probe_timeout) => {} - } - inflight.store(false, std::sync::atomic::Ordering::Release); - }); - } - } - } - _ = shutdown.cancelled() => break, + fn test_section_for(node_id: &str, port: u16, flags: u16) -> GossipSection { + let mut id = [0u8; 40]; + id.copy_from_slice(node_id.as_bytes()); + let mut ip = [0u8; 16]; + ip[..9].copy_from_slice(b"127.0.0.1"); + GossipSection { + node_id: id, + ip, + port, + bus_port: port + 10000, + flags, + epoch: 0, + ping_sent_ms: 0, + pong_recv_ms: 0, } } -} -#[cfg(test)] -mod tests { - use super::*; + /// C-1 formation fix: the random-id placeholder CLUSTER MEET creates must + /// be retired when the peer's handshake arrives under its real id from + /// the same address — otherwise every met peer is double-counted forever. + #[test] + fn handshake_retires_meet_placeholder() { + let self_addr: SocketAddr = "127.0.0.1:7000".parse().unwrap(); + let mut state = ClusterState::new(id40(b'a'), self_addr); + let peer_addr: SocketAddr = "127.0.0.1:7001".parse().unwrap(); + state.nodes.insert( + id40(b'p'), + ClusterNode::new(id40(b'p'), peer_addr, NodeFlags::Master, 0), + ); + assert_eq!(state.nodes.len(), 2); + + let real = id40(b'b'); + merge_gossip_into_state(&mut state, &test_msg_from(&real, 7001, vec![])); + + assert!(state.nodes.contains_key(&real), "real peer entry missing"); + assert!( + !state.nodes.contains_key(&id40(b'p')), + "MEET placeholder at the peer's addr must be retired by the handshake" + ); + assert_eq!(state.nodes.len(), 2, "self + real peer, nothing else"); + } + + /// C-1 formation fix: a healthy gossip section about a node we have never + /// talked to must be adopted — this is the only way two nodes MEET-ed + /// into a third learn about each other. + #[test] + fn healthy_rumor_adds_unknown_node() { + let self_addr: SocketAddr = "127.0.0.1:7000".parse().unwrap(); + let mut state = ClusterState::new(id40(b'a'), self_addr); + + let peer = id40(b'b'); + let third = id40(b'c'); + let msg = test_msg_from(&peer, 7001, vec![test_section_for(&third, 7002, 0)]); + merge_gossip_into_state(&mut state, &msg); + + let adopted = state.nodes.get(&third).expect("rumored node not adopted"); + assert_eq!( + adopted.addr, + "127.0.0.1:7002".parse::().unwrap() + ); + assert_eq!(adopted.bus_port, 17002); + assert!( + adopted.pong_recv_ms > 0, + "adopted nodes need a freshness baseline or they can never go PFAIL" + ); + assert_eq!(state.nodes.len(), 3, "self + sender + rumored third"); + } + + /// A rumor pointing at OUR OWN address (a peer's not-yet-resolved MEET + /// placeholder for us) must not create a phantom self entry, and a rumor + /// about an address we already know under another id must not duplicate it. + #[test] + fn rumor_about_known_addr_is_ignored() { + let self_addr: SocketAddr = "127.0.0.1:7000".parse().unwrap(); + let mut state = ClusterState::new(id40(b'a'), self_addr); + + let peer = id40(b'b'); + let msg = test_msg_from( + &peer, + 7001, + vec![ + test_section_for(&id40(b'q'), 7000, 0), // our own addr + test_section_for(&id40(b'r'), 7001, 0), // sender's addr, other id + ], + ); + merge_gossip_into_state(&mut state, &msg); + + assert!(!state.nodes.contains_key(&id40(b'q'))); + assert!(!state.nodes.contains_key(&id40(b'r'))); + assert_eq!(state.nodes.len(), 2, "self + sender only"); + } /// CLUSTER-10: serialize then deserialize a PING produces identical GossipMessage. #[test] diff --git a/src/main.rs b/src/main.rs index 46a8f668b..e8ee30e76 100644 --- a/src/main.rs +++ b/src/main.rs @@ -109,10 +109,11 @@ fn main() -> anyhow::Result<()> { default_hook(info); let thread = std::thread::current(); let name = thread.name().unwrap_or(""); - if name.starts_with("shard-") { + if name.starts_with("shard-") || name == "cluster-ctl" { eprintln!( - "FATAL: shard thread '{name}' panicked; aborting the whole \ - process rather than serving with a dead shard" + "FATAL: thread '{name}' panicked; aborting the whole \ + process rather than serving with a dead shard or a \ + dead cluster control plane" ); std::process::abort(); } @@ -1009,6 +1010,17 @@ fn main() -> anyhow::Result<()> { // Cluster mode initialization let cluster_state: Option>> = if config.cluster_enabled { + // Redis convention: cluster bus port = port + 10000. Refuse ports + // where that wraps past u16::MAX instead of silently binding the + // bus on a truncated port no peer would ever compute. + if config.port.checked_add(10000).is_none() { + eprintln!( + "REFUSING TO START: --cluster-enabled with --port {} — the cluster \ + bus port (port + 10000) exceeds 65535. Use a port <= 55535.", + config.port + ); + std::process::exit(2); + } moon::cluster::CLUSTER_ENABLED.store(true, std::sync::atomic::Ordering::Relaxed); let self_addr: std::net::SocketAddr = format!("{}:{}", config.bind, config.port) .parse() @@ -1891,10 +1903,48 @@ fn main() -> anyhow::Result<()> { let listener_cancel = cancel_token.clone(); + // v0.9 C-1 (#405): the cluster control plane (bus + gossip + election) is + // tokio-native and runs on a dedicated `cluster-ctl` std thread hosting a + // current-thread tokio runtime — on BOTH server runtimes. Under monoio + // there is no tokio runtime to share; under tokio, sharing the listener + // runtime made gossip compete with the accept loop and every connection, + // which starved the 100ms ticker under load (observed as PFAIL detection + // stalling in the formation e2e). The control plane is not + // latency-critical, and monoio's !Send task model makes a port high-risk + // for zero payoff. + if let Some(ref cs) = cluster_state { + let ctl_cs = cs.clone(); + let ctl_bind = config.bind.clone(); + let ctl_port = config.port; + let ctl_node_timeout = config.cluster_node_timeout; + let ctl_cancel = cancel_token.child_token(); + let ctl_repl_state = repl_state.clone(); + std::thread::Builder::new() + .name("cluster-ctl".to_string()) + .spawn(move || { + // O5: escape the shard-core mask this thread would otherwise + // inherit from its spawning thread. + moon::shard::numa::pin_current_aux_thread("cluster-ctl"); + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("failed to build cluster control-plane runtime"); + // Runs until the shutdown token cancels the gossip ticker. + rt.block_on(run_cluster_control_plane( + ctl_cs, + ctl_bind, + ctl_port, + ctl_node_timeout, + ctl_cancel, + ctl_repl_state, + )); + }) + .expect("failed to spawn cluster control-plane thread"); + } + // Run the sharded listener on the main thread. // Under tokio: uses current_thread runtime with tokio::spawn for background tasks. - // Under monoio: uses monoio RuntimeFactory with simplified startup (cluster/gossip - // not yet supported under monoio). + // Under monoio: uses monoio RuntimeFactory (auto-save on its own thread). #[cfg(feature = "runtime-tokio")] { let listener_rt = tokio::runtime::Builder::new_current_thread() @@ -1925,57 +1975,6 @@ fn main() -> anyhow::Result<()> { } } - // Start cluster bus and gossip ticker when cluster mode is enabled - if let Some(ref cs) = cluster_state { - let cluster_port = (config.port as u32 + 10000) as u16; - let cs_clone = cs.clone(); - let bus_cancel = cancel_token.child_token(); - let bind2 = config.bind.clone(); - let self_addr: std::net::SocketAddr = - format!("{}:{}", config.bind, config.port).parse().unwrap(); - - // Shared vote channel: gossip ticker sets sender when election starts, - // bus handler forwards FailoverAuthAck votes through it. - let failover_vote_tx: moon::cluster::bus::SharedVoteTx = - std::sync::Arc::new(parking_lot::Mutex::new(None)); - - let bus_vote_tx = failover_vote_tx.clone(); - tokio::spawn(async move { - if let Err(e) = moon::cluster::bus::run_cluster_bus( - &bind2, - cluster_port, - self_addr, - cs_clone, - bus_cancel, - bus_vote_tx, - ) - .await - { - tracing::error!("Cluster bus error: {}", e); - } - }); - - let cs_gossip = cs.clone(); - let gossip_cancel = cancel_token.child_token(); - let node_timeout = config.cluster_node_timeout; - let self_addr2: std::net::SocketAddr = - format!("{}:{}", config.bind, config.port).parse().unwrap(); - let gossip_vote_tx = failover_vote_tx.clone(); - let gossip_repl_state = repl_state.clone(); - tokio::spawn(async move { - moon::cluster::gossip::run_gossip_ticker( - self_addr2, - cs_gossip, - node_timeout, - gossip_cancel, - gossip_vote_tx, - gossip_repl_state, - ) - .await; - }); - info!("Cluster bus and gossip ticker started"); - } - // The central tokio listener plain-binds the port (no SO_REUSEPORT, // see listener::run_sharded), which makes EVERY per-shard SO_REUSEPORT // bind fail with EADDRINUSE — both the io_uring multishot path and the @@ -2006,9 +2005,6 @@ fn main() -> anyhow::Result<()> { #[cfg(feature = "runtime-monoio")] { - // Monoio listener: simplified startup. Cluster bus and gossip not yet - // supported under monoio. - // Auto-save runs on a dedicated thread (same pattern as AOF writer). if config.save.is_some() { let rules = moon::persistence::auto_save::parse_save_rules(&config.save); @@ -2082,6 +2078,81 @@ fn main() -> anyhow::Result<()> { Ok(()) } +/// Run the cluster control plane — bus listener, gossip ticker, and (via the +/// ticker) failover elections — on the CURRENT tokio runtime, until the +/// cancellation token fires. +/// +/// v0.9 C-1 (#405): the control plane is tokio-native on BOTH runtimes. The +/// tokio server spawns this onto its listener runtime; the monoio server runs +/// it on a dedicated `cluster-ctl` std thread hosting a current-thread tokio +/// runtime. All I/O here is gossip-rate (100ms ticks), never on shard threads. +async fn run_cluster_control_plane( + cluster_state: std::sync::Arc>, + bind: String, + port: u16, + node_timeout: u64, + cancel_token: moon::runtime::cancel::CancellationToken, + repl_state: std::sync::Arc>, +) { + // Overflow is refused at startup (cluster init hard-exits on ports > + // 55535); the else arm is unreachable defense. + let Some(cluster_port) = port.checked_add(10000) else { + tracing::error!("Cluster control plane disabled: port {port} + 10000 overflows u16"); + return; + }; + let self_addr: std::net::SocketAddr = match format!("{bind}:{port}").parse() { + Ok(a) => a, + Err(e) => { + tracing::error!( + "Cluster control plane disabled: invalid bind address {bind}:{port}: {e}" + ); + return; + } + }; + + // Bind the bus BEFORE spawning anything, and abort on failure: a cluster + // node whose bus is not listening keeps serving clients while being + // invisible to every peer — fail loud instead of running half-alive. + let bus_addr = format!("{bind}:{cluster_port}"); + let listener = match tokio::net::TcpListener::bind(&bus_addr).await { + Ok(l) => l, + Err(e) => { + eprintln!( + "FATAL: cluster bus failed to bind {bus_addr}: {e} — a cluster \ + node without its bus is invisible to every peer; exiting" + ); + std::process::exit(1); + } + }; + + // Shared vote channel: gossip ticker sets sender when election starts, + // bus handler forwards FailoverAuthAck votes through it. + let failover_vote_tx: moon::cluster::bus::SharedVoteTx = + std::sync::Arc::new(parking_lot::Mutex::new(None)); + + let bus_cs = cluster_state.clone(); + let bus_cancel = cancel_token.child_token(); + let bus_vote_tx = failover_vote_tx.clone(); + tokio::spawn(moon::cluster::bus::run_cluster_bus( + listener, + self_addr, + bus_cs, + bus_cancel, + bus_vote_tx, + )); + + info!("Cluster bus and gossip ticker started"); + moon::cluster::gossip::run_gossip_ticker( + self_addr, + cluster_state, + node_timeout, + cancel_token.child_token(), + failover_vote_tx, + repl_state, + ) + .await; +} + /// Resolve the automatic shard count, optionally capped to the empirical /// knee of 2 when `MOON_AUTO_SHARDS_CONSERVATIVE=1` is set. /// diff --git a/tests/cluster_formation.rs b/tests/cluster_formation.rs new file mode 100644 index 000000000..290f27ba5 --- /dev/null +++ b/tests/cluster_formation.rs @@ -0,0 +1,336 @@ +//! v0.9 W0/C-1 (#405): the cluster control plane must run on the DEFAULT +//! runtime. A 3-node cluster started with `--cluster-enabled` has to form via +//! CLUSTER MEET + gossip regardless of which runtime the binary was built +//! with — before C-1, the monoio startup path never spawned the cluster bus +//! or the gossip ticker, so MEET wrote local state that no peer ever learned +//! about. +//! +//! The load-bearing assertion is on nodes 2 and 3: node 1 knows all three +//! from its own MEET commands, but nodes 2/3 only reach +//! `cluster_known_nodes:3` when the bus carries node 1's gossip pings to +//! them (node 2 learns of node 3 — and vice versa — exclusively through +//! gossip payloads). + +mod common; + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +/// Reserve a port whose cluster-bus sibling (port + 10000) is also free — +/// the bus binds `port + 10000` unconditionally in cluster mode. +fn reserve_cluster_ports(n: usize) -> Vec { + // OS-assigned ports land in the ephemeral range (49152+ on macOS), where + // `port + 10000` overflows past 65535 — probe a low range explicitly. + // All reservations (client port AND its +10000 bus sibling) stay bound + // until every node's pair is chosen, so picks can't collide with each + // other. Spread starts by pid so parallel test processes don't fight + // over the same slots. + // Tests in this file run in parallel threads of ONE process: give each + // reservation call its own sub-range so a later test can't race the + // window between a fleet's reservation-drop and its servers' binds. + static NEXT_RANGE: std::sync::atomic::AtomicU16 = std::sync::atomic::AtomicU16::new(0); + let range = NEXT_RANGE.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let start = 21000 + (std::process::id() as u16 % 1500) * 7 + range * 4000; + let mut held: Vec = Vec::new(); + let mut ports = Vec::new(); + for candidate in (start..40000).step_by(3) { + let Ok(l) = TcpListener::bind(("127.0.0.1", candidate)) else { + continue; + }; + let Ok(bus) = TcpListener::bind(("127.0.0.1", candidate + 10000)) else { + continue; + }; + held.push(l); + held.push(bus); + ports.push(candidate); + if ports.len() == n { + return ports; + } + } + panic!("could not reserve {n} ports with free +10000 siblings"); +} + +/// Kills the whole fleet on drop so a failed assertion never leaks servers +/// (leaked-moon gotcha: orphans spin CPU and poison later benches). +struct Fleet(Vec); + +impl Drop for Fleet { + fn drop(&mut self) { + for child in &mut self.0 { + common::sigkill(child); + } + } +} + +fn spawn_cluster_node(dir: &std::path::Path, port: u16, extra: &[&str]) -> Child { + Command::new(common::find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--dir", + dir.to_str().unwrap(), + "--disk-free-min-pct", + "0", + "--cluster-enabled", + ]) + .args(extra) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") +} + +fn connect_retry(port: u16) -> TcpStream { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + match TcpStream::connect(("127.0.0.1", port)) { + Ok(s) => return s, + Err(e) => { + assert!( + std::time::Instant::now() < deadline, + "connect to 127.0.0.1:{port} kept failing: {e}" + ); + std::thread::sleep(Duration::from_millis(50)); + } + } + } +} + +/// Read one full RESP reply (simple line or bulk string). +fn command_reply(stream: &mut TcpStream, cmd: &str) -> String { + stream.write_all(cmd.as_bytes()).expect("write cmd"); + stream + .set_read_timeout(Some(Duration::from_secs(30))) + .expect("set timeout"); + let mut buf = Vec::new(); + let mut chunk = [0u8; 65536]; + loop { + let n = stream.read(&mut chunk).expect("read reply"); + assert!(n > 0, "connection closed mid-reply"); + buf.extend_from_slice(&chunk[..n]); + if buf.starts_with(b"$") { + if let Some(pos) = buf.iter().position(|&b| b == b'\n') { + let len: usize = std::str::from_utf8(&buf[1..pos - 1]) + .unwrap() + .trim() + .parse() + .unwrap(); + if buf.len() >= pos + 1 + len + 2 { + break; + } + } + } else if buf.ends_with(b"\r\n") { + break; + } + } + String::from_utf8_lossy(&buf).into_owned() +} + +fn known_nodes(stream: &mut TcpStream) -> usize { + let info = command_reply(stream, "CLUSTER INFO\r\n"); + info.lines() + .find_map(|l| l.strip_prefix("cluster_known_nodes:")) + .and_then(|v| v.trim().parse().ok()) + .unwrap_or_else(|| panic!("no cluster_known_nodes in reply: {info}")) +} + +fn myid(stream: &mut TcpStream) -> String { + let reply = command_reply(stream, "CLUSTER MYID\r\n"); + let id = reply + .lines() + .nth(1) + .expect("CLUSTER MYID bulk payload") + .trim() + .to_string(); + assert_eq!(id.len(), 40, "unexpected MYID reply: {reply}"); + id +} + +/// Spawn a 3-node fleet, MEET nodes 2/3 from node 1, and wait for the mesh +/// to complete: every node's CLUSTER NODES must list all three REAL node ids. +/// +/// Identity convergence (not just `cluster_known_nodes:3`) is the criterion: +/// a rumor can adopt a peer under a not-yet-resolved placeholder id at the +/// right address, which satisfies the count while the real id is still only +/// resolvable by direct handshake. +fn form_three_node_cluster( + dirs: &[tempfile::TempDir], + extra: &[&str], +) -> (Fleet, Vec, Vec, Vec) { + let ports = reserve_cluster_ports(3); + + let fleet = Fleet( + dirs.iter() + .zip(&ports) + .map(|(d, &p)| spawn_cluster_node(d.path(), p, extra)) + .collect(), + ); + + let mut conns: Vec = ports.iter().map(|&p| connect_retry(p)).collect(); + for c in &mut conns { + let pong = command_reply(c, "PING\r\n"); + assert_eq!(pong, "+PONG\r\n"); + } + + // Meet nodes 2 and 3 into the cluster from node 1 only. Node 2 and node 3 + // never hear about each other directly — only gossip can complete the mesh. + for &p in &ports[1..] { + let r = command_reply(&mut conns[0], &format!("CLUSTER MEET 127.0.0.1 {p}\r\n")); + assert!(r.starts_with("+OK"), "CLUSTER MEET failed: {r}"); + } + + let ids: Vec = conns.iter_mut().map(myid).collect(); + + // Gossip ticks every 100ms; give a loaded runner plenty of slack. + let deadline = std::time::Instant::now() + Duration::from_secs(30); + loop { + let resolved: Vec = conns + .iter_mut() + .map(|c| { + let nodes = command_reply(c, "CLUSTER NODES\r\n"); + ids.iter() + .filter(|id| nodes.lines().any(|l| l.starts_with(id.as_str()))) + .count() + }) + .collect(); + if resolved.iter().all(|&n| n == 3) { + break; + } + assert!( + std::time::Instant::now() < deadline, + "cluster never formed: real ids resolved per node = {resolved:?} \ + (expected [3, 3, 3]; nodes 2/3 stuck below 3 means the cluster \ + bus / gossip ticker is not running on this runtime)" + ); + std::thread::sleep(Duration::from_millis(200)); + } + + // Identity convergence implies exact membership: any placeholder at one + // of the three addresses is retired by the direct handshake that + // resolved the real id there. + let counts: Vec = conns.iter_mut().map(known_nodes).collect(); + assert_eq!(counts, vec![3, 3, 3], "phantom entries survived formation"); + + (fleet, conns, ports, ids) +} + +#[test] +fn three_node_cluster_forms_via_meet_and_gossip() { + let dirs: Vec = (0..3) + .map(|_| tempfile::tempdir().expect("tempdir")) + .collect(); + let (fleet, mut conns, ports, _ids) = form_three_node_cluster(&dirs, &[]); + + // Bus traffic must actually have flowed on the seed node. + let info = command_reply(&mut conns[0], "CLUSTER INFO\r\n"); + let sent: u64 = info + .lines() + .find_map(|l| l.strip_prefix("cluster_stats_messages_sent:")) + .and_then(|v| v.trim().parse().ok()) + .expect("cluster_stats_messages_sent present"); + assert!(sent > 0, "no cluster bus messages were sent: {info}"); + + // MEET is idempotent by address: repeating one must not stack a fresh + // placeholder (the placeholder id is random per call). + let r = command_reply( + &mut conns[0], + &format!("CLUSTER MEET 127.0.0.1 {}\r\n", ports[1]), + ); + assert!(r.starts_with("+OK"), "repeat MEET failed: {r}"); + assert_eq!( + known_nodes(&mut conns[0]), + 3, + "repeat MEET stacked a placeholder" + ); + + // MEET-ing our own advertised address is refused. + let r = command_reply( + &mut conns[0], + &format!("CLUSTER MEET 127.0.0.1 {}\r\n", ports[0]), + ); + assert!(r.starts_with("-ERR"), "self-MEET must error: {r}"); + + drop(fleet); +} + +/// A cluster node whose bus port (port + 10000) is already taken must abort +/// loudly at startup — not serve clients while invisible to every peer. +#[test] +fn occupied_bus_port_aborts_startup() { + let port = reserve_cluster_ports(1)[0]; + let _bus_blocker = TcpListener::bind(("127.0.0.1", port + 10000)).expect("occupy bus port"); + let dir = tempfile::tempdir().expect("tempdir"); + let mut child = spawn_cluster_node(dir.path(), port, &[]); + + let deadline = std::time::Instant::now() + Duration::from_secs(15); + loop { + match child.try_wait().expect("try_wait") { + Some(status) => { + assert!( + !status.success(), + "startup must abort with a nonzero status, got {status}" + ); + break; + } + None => { + if std::time::Instant::now() >= deadline { + common::sigkill(&mut child); + panic!("node kept running with an occupied cluster bus port"); + } + std::thread::sleep(Duration::from_millis(100)); + } + } + } +} + +/// Failure detection through the control plane: killing one node of a formed +/// 3-node cluster must get it flagged (pfail) by both survivors within the +/// node timeout. Hard FAIL needs quorum ≥ 2 EXTERNAL reporters, which two +/// survivors of three masters cannot reach — full FAIL/election e2e lands +/// with C-3's replica legs. +#[test] +fn killed_node_is_flagged_by_survivors() { + let dirs: Vec = (0..3) + .map(|_| tempfile::tempdir().expect("tempdir")) + .collect(); + let (mut fleet, mut conns, _ports, ids) = + form_three_node_cluster(&dirs, &["--cluster-node-timeout", "3000"]); + + // Node 3's id — formation guarantees both survivors already resolved it. + let victim_id = ids[2].clone(); + + common::sigkill(&mut fleet.0[2]); + + // node_timeout is 3s and gossip ticks every 100ms; poll each survivor's + // CLUSTER NODES for the victim's flags token to become pfail (or fail, + // should quorum semantics ever start counting the local observation). + let deadline = std::time::Instant::now() + Duration::from_secs(30); + 'outer: loop { + let mut flagged = 0; + for c in conns.iter_mut().take(2) { + let nodes = command_reply(c, "CLUSTER NODES\r\n"); + let flags = nodes + .lines() + .find(|l| l.starts_with(&victim_id)) + .and_then(|l| l.split_whitespace().nth(2)) + .unwrap_or(""); + if flags.split(',').any(|f| f == "pfail" || f == "fail") { + flagged += 1; + } + } + if flagged == 2 { + break 'outer; + } + assert!( + std::time::Instant::now() < deadline, + "survivors never flagged the killed node ({flagged}/2 saw pfail/fail)" + ); + std::thread::sleep(Duration::from_millis(200)); + } + + drop(fleet); +}