This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch cleanup_fixes in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dc88bc49beb51d969f0772e8580f0457b2490650 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Jul 29 10:59:47 2026 +0200 feat(server-ng): harden listener, auth, shutdown, and disk-poll paths --- core/configs/src/server_ng_config/cluster.rs | 63 +++++++ core/configs/src/server_ng_config/sharding.rs | 74 ++++++++ core/message_bus/src/client_listener/mod.rs | 5 +- core/message_bus/src/client_listener/quic.rs | 24 +-- core/message_bus/src/framing.rs | 8 +- core/message_bus/src/installer/replica.rs | 2 +- core/message_bus/src/replica/auth.rs | 106 +++++++++++- core/message_bus/src/replica/handshake.rs | 21 ++- core/message_bus/src/socket_opts.rs | 59 ++++--- core/message_bus/tests/common/mod.rs | 7 +- core/message_bus/tests/replica_auth_handshake.rs | 28 +++ core/partitions/src/iggy_partition.rs | 88 +++++++++- core/partitions/src/poll_plan.rs | 57 ++++-- core/server-ng/config.toml | 15 ++ core/server-ng/src/auth.rs | 100 +++++++---- core/server-ng/src/bootstrap.rs | 211 +++++++++++++---------- core/server-ng/src/dispatch.rs | 16 +- core/server-ng/src/server_error.rs | 16 +- 18 files changed, 682 insertions(+), 218 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 613f4bf7c..62ae0ae22 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -299,6 +299,20 @@ pub struct ClusterAuthConfig { #[serde(default, skip_serializing)] #[config_env(secret)] pub shared_secret: String, + /// Retiring pre-shared key, accepted for VERIFICATION only during a key + /// rotation window; every MAC this node produces uses [`Self::shared_secret`]. + /// + /// Enables rolling PSK rotation without an auth outage, three rolls: + /// 1. every node gets `shared_secret = old, previous_shared_secret = new`; + /// 2. every node gets `shared_secret = new, previous_shared_secret = old`; + /// 3. every node gets `shared_secret = new` alone, closing the window. + /// + /// Leave empty (default) outside a rotation. Same 32-byte minimum and + /// provisioning rules as `shared_secret` + /// (`IGGY_CLUSTER_AUTH_PREVIOUS_SHARED_SECRET`). + #[serde(default, skip_serializing)] + #[config_env(secret)] + pub previous_shared_secret: String, } /// Replica-to-replica TLS for the consensus (`tcp_replica`) port. @@ -860,6 +874,24 @@ impl Validatable<ConfigurationError> for ClusterConfig { ); return Err(ConfigurationError::InvalidConfigurationValue); } + // Rotation window key: same typo guard as the primary, plus a + // distinctness check - a window equal to the primary means the + // operator rolled the config without changing the key, so the + // "rotation" would silently be a no-op. + if !self.auth.previous_shared_secret.is_empty() { + if self.auth.previous_shared_secret.len() < MIN_SHARED_SECRET_LEN { + eprintln!( + "Invalid cluster configuration: cluster.auth.previous_shared_secret must be >= {MIN_SHARED_SECRET_LEN} bytes" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.auth.previous_shared_secret == self.auth.shared_secret { + eprintln!( + "Invalid cluster configuration: cluster.auth.previous_shared_secret must differ from cluster.auth.shared_secret (an identical window is a no-op rotation)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + } // Replica TLS. Both cert modes run one-directional TLS (no client // certificate anywhere), so TLS only authenticates the acceptor to @@ -923,6 +955,7 @@ mod tests { auth: ClusterAuthConfig { enabled: true, shared_secret: "current-psk-MUST-NOT-be-persisted".to_owned(), + previous_shared_secret: "retiring-psk-MUST-NOT-be-persisted".to_owned(), }, tls: ClusterTlsConfig::default(), }; @@ -1520,6 +1553,36 @@ mod cluster_validate_tests { assert!(c.validate().is_ok()); } + #[test] + fn validate_accepts_valid_rotation_window() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.auth.enabled = true; + c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN); + c.auth.previous_shared_secret = "b".repeat(MIN_SHARED_SECRET_LEN); + assert!(c.validate().is_ok()); + } + + #[test] + fn validate_rejects_short_previous_secret() { + // Same typo guard as the primary key. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.auth.enabled = true; + c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN); + c.auth.previous_shared_secret = "b".repeat(MIN_SHARED_SECRET_LEN - 1); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_rotation_window_equal_to_primary() { + // An identical window is a no-op rotation: the operator rolled the + // config without changing the key. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.auth.enabled = true; + c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN); + c.auth.previous_shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN); + assert!(c.validate().is_err()); + } + fn tls_files() -> ClusterTlsConfig { ClusterTlsConfig { enabled: true, diff --git a/core/configs/src/server_ng_config/sharding.rs b/core/configs/src/server_ng_config/sharding.rs index 1f2e3f419..f5c2257d4 100644 --- a/core/configs/src/server_ng_config/sharding.rs +++ b/core/configs/src/server_ng_config/sharding.rs @@ -54,6 +54,11 @@ pub const SHUTDOWN_DRAIN_TIMEOUT_MAX: Duration = Duration::from_secs(600); /// at 5s so Ctrl-C latency stays bounded regardless of config. pub const SHUTDOWN_POLL_INTERVAL_MAX: Duration = Duration::from_secs(5); +/// Hard upper bound on `shutdown_join_timeout`. Comfortably above the +/// drain cap so a full drain always fits inside the join budget, while +/// still guaranteeing process exit against a pathological config typo. +pub const SHUTDOWN_JOIN_TIMEOUT_MAX: Duration = Duration::from_secs(900); + /// Hard upper bound on `reconcile_periodic_interval`. A tick longer /// than ~30s makes post-failure recovery latency operator-visible; the /// cap reins in pathological typos without disturbing reasonable @@ -117,6 +122,15 @@ pub struct ShardingConfig { #[serde_as(as = "DisplayFromStr")] #[config_env(leaf)] pub shutdown_poll_interval: IggyDuration, + /// Hard wall-clock deadline for joining shard threads at process + /// exit. A shard whose pump or listener wedges past this budget is + /// abandoned with an error log instead of blocking exit forever. + /// Must be at least `shutdown_drain_timeout` (abandoning a shard + /// mid-drain would interrupt its WAL fsync / replica drain) and at + /// most [`SHUTDOWN_JOIN_TIMEOUT_MAX`]. + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub shutdown_join_timeout: IggyDuration, /// Safety-tick cadence for the partition reconciliation loop; the /// reconciler also wakes immediately on every /// `LifecycleFrame::MetadataCommitTick` from shard 0. This periodic @@ -146,6 +160,12 @@ impl Default for ShardingConfig { .shutdown_poll_interval .parse() .unwrap(), + shutdown_join_timeout: SERVER_NG_CONFIG + .system + .sharding + .shutdown_join_timeout + .parse() + .unwrap(), reconcile_periodic_interval: SERVER_NG_CONFIG .system .sharding @@ -218,6 +238,25 @@ impl Validatable<ConfigurationError> for ShardingConfig { return Err(ConfigurationError::InvalidConfigurationValue); } + let join = self.shutdown_join_timeout.get_duration(); + if join < drain { + eprintln!( + "Invalid sharding configuration: shutdown_join_timeout {:?} must be >= \ + shutdown_drain_timeout {:?} (a join budget shorter than the drain abandons \ + shards mid-drain, interrupting the WAL fsync / replica drain)", + join, drain + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if join > SHUTDOWN_JOIN_TIMEOUT_MAX { + eprintln!( + "Invalid sharding configuration: shutdown_join_timeout {:?} exceeds the {:?} \ + cap (an unbounded join budget wedges process exit on a stuck shard)", + join, SHUTDOWN_JOIN_TIMEOUT_MAX + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + let reconcile = self.reconcile_periodic_interval.get_duration(); if reconcile.is_zero() { eprintln!( @@ -303,6 +342,38 @@ mod tests { assert!(cfg.validate().is_err()); } + #[test] + fn join_shorter_than_drain_is_rejected() { + // A join budget under the drain would abandon shards mid-drain. + let cfg = ShardingConfig { + shutdown_drain_timeout: IggyDuration::new(Duration::from_secs(10)), + shutdown_join_timeout: IggyDuration::new(Duration::from_secs(5)), + ..ShardingConfig::default() + }; + assert!(cfg.validate().is_err()); + } + + #[test] + fn over_cap_join_is_rejected() { + let cfg = ShardingConfig { + shutdown_join_timeout: IggyDuration::new( + SHUTDOWN_JOIN_TIMEOUT_MAX + Duration::from_secs(1), + ), + ..ShardingConfig::default() + }; + assert!(cfg.validate().is_err()); + } + + #[test] + fn join_equal_to_drain_is_accepted() { + let cfg = ShardingConfig { + shutdown_drain_timeout: IggyDuration::new(Duration::from_secs(10)), + shutdown_join_timeout: IggyDuration::new(Duration::from_secs(10)), + ..ShardingConfig::default() + }; + assert!(cfg.validate().is_ok()); + } + // Guards the single source of truth: the server-ng sharding defaults // resolve from the embedded server-ng TOML, not hard-coded Rust values. #[test] @@ -321,6 +392,7 @@ mod tests { assert_eq!(sharding.inbox_capacity, 1024); assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap()); assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap()); + assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap()); assert_eq!(sharding.reconcile_periodic_interval, "1 s".parse().unwrap()); } @@ -338,6 +410,7 @@ mod tests { assert_eq!(sharding.inbox_capacity, 1024); assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap()); assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap()); + assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap()); assert_eq!(sharding.reconcile_periodic_interval, "1 s".parse().unwrap()); } @@ -352,6 +425,7 @@ mod tests { assert_eq!(sharding.inbox_capacity, 1024); assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap()); assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap()); + assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap()); assert_eq!(sharding.reconcile_periodic_interval, "1 s".parse().unwrap()); } } diff --git a/core/message_bus/src/client_listener/mod.rs b/core/message_bus/src/client_listener/mod.rs index 8c65a6a16..1ed966bd2 100644 --- a/core/message_bus/src/client_listener/mod.rs +++ b/core/message_bus/src/client_listener/mod.rs @@ -104,8 +104,9 @@ pub mod wss; /// with the `TcpSocket` builder; this preserves the prior `nodelay(true)` /// bind. `SO_REUSEADDR` is set so a restarted server can rebind the port /// while a previous client connection lingers in `TIME_WAIT` (matching the -/// replica and QUIC listeners); `SO_REUSEPORT` is intentionally not set: -/// only shard 0 binds the client listeners (see each caller). +/// replica and TLS listeners; QUIC sets no reuse flag since UDP has no +/// `TIME_WAIT`); `SO_REUSEPORT` is intentionally not set: only shard 0 +/// binds the client listeners (see each caller). /// /// # Errors /// diff --git a/core/message_bus/src/client_listener/quic.rs b/core/message_bus/src/client_listener/quic.rs index 1d64af514..63cb6ac78 100644 --- a/core/message_bus/src/client_listener/quic.rs +++ b/core/message_bus/src/client_listener/quic.rs @@ -65,27 +65,13 @@ pub fn bind( server_config: ServerConfig, ) -> Result<(Endpoint, SocketAddr), IggyError> { // QUIC remains shard-0 terminal, so nothing here needs the load balancing - // `SO_REUSEPORT` exists for; it only widens what the bind accepts. - // - // TODO: work out whether this listener needs `SO_REUSEPORT` at all. Dropping - // it would make a stale process holding the port a loud bind failure instead - // of two live sockets silently splitting inbound datagrams. + // `SO_REUSEPORT` exists for. No reuse flag is set at all: UDP has no + // TIME_WAIT to rebind over, and on Linux `SO_REUSEADDR` alone lets a + // second unicast UDP socket bind the same port and silently steal inbound + // datagrams. A stale process holding the port must be a loud + // `EADDRINUSE` instead. let socket = Socket::new(Domain::for_address(addr), Type::DGRAM, Some(Protocol::UDP)) .map_err(|e| IggyError::IoError(e.to_string()))?; - socket - .set_reuse_address(true) - .map_err(|e| IggyError::IoError(e.to_string()))?; - // Match the gate in `core/message_bus/src/socket_opts.rs`: `set_reuse_port` - // is absent on illumos/solaris/cygwin, so the cfg must exclude them or - // QUIC bind fails to compile on those targets while TCP intentionally - // skips. - #[cfg(all( - unix, - not(any(target_os = "illumos", target_os = "solaris", target_os = "cygwin")) - ))] - socket - .set_reuse_port(true) - .map_err(|e| IggyError::IoError(e.to_string()))?; socket .bind(&addr.into()) .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?; diff --git a/core/message_bus/src/framing.rs b/core/message_bus/src/framing.rs index 5587a8726..39ce24d81 100644 --- a/core/message_bus/src/framing.rs +++ b/core/message_bus/src/framing.rs @@ -91,8 +91,12 @@ pub async fn write_message<S: AsyncWriteExt>( /// contexts (e.g. `select!`) must accept the resulting framing error /// and tear the connection down. /// -/// See `tcp_tls::run_pump` rustdoc and TODO for the resumable-framing -/// fix path. +/// Every current caller is a non-cancellable context: the plaintext TCP +/// reader loops with no `select!` (shutdown arrives via the `SHUT_RD` +/// watchdog), QUIC reads one frame per dedicated bidi stream, and the +/// replica handshake is sequential. A pump that must `select!` over the +/// read instead uses its own resumable accumulator; see +/// `transports::tcp_tls::run_pump` and its `TlsPumpState`. /// /// # Errors /// diff --git a/core/message_bus/src/installer/replica.rs b/core/message_bus/src/installer/replica.rs index 40eb4bbc0..bf8a344b7 100644 --- a/core/message_bus/src/installer/replica.rs +++ b/core/message_bus/src/installer/replica.rs @@ -244,7 +244,7 @@ pub fn install_replica_outbound( .await } Some(tls) => { - let Some(peer_name) = tls.peer_names.get(usize::from(peer_id)) else { + let Some(peer_name) = tls.peer_names.get(&peer_id) else { warn!( peer = %peer, "no TLS peer name for replica id; dropping dialed connection" diff --git a/core/message_bus/src/replica/auth.rs b/core/message_bus/src/replica/auth.rs index 122727787..b455f69ee 100644 --- a/core/message_bus/src/replica/auth.rs +++ b/core/message_bus/src/replica/auth.rs @@ -237,9 +237,24 @@ pub struct Transcript { /// `Option<ReplicaAuth>`; `None` means auth is disabled and the handshake stays /// in legacy unauthenticated mode. `Some` means auth is enabled and enforced - /// a peer that does not complete the handshake is rejected. +/// +/// # Key rotation +/// +/// A second, verify-only subkey ([`Self::with_previous_secret`]) widens +/// verification to `{current, previous}` while every MAC this node PRODUCES +/// uses the current key. That makes a rolling PSK rotation possible in two +/// coordinated rolls with no auth outage: +/// +/// 1. roll every node to `shared_secret = old, previous_shared_secret = new` +/// (all nodes still sign with the old key; the new key is pre-trusted); +/// 2. roll every node to `shared_secret = new, previous_shared_secret = old` +/// (mid-roll mixes verify because each side trusts the other's signer); +/// 3. roll every node to `shared_secret = new` alone, closing the window. #[derive(Clone)] pub struct ReplicaAuth { key: Zeroizing<[u8; 32]>, + /// Rotation acceptance window: verified against, never signed with. + previous_key: Option<Zeroizing<[u8; 32]>>, } impl ReplicaAuth { @@ -248,35 +263,63 @@ impl ReplicaAuth { pub fn new(secret_material: &[u8]) -> Self { Self { key: Zeroizing::new(blake3::derive_key(KEY_CONTEXT, secret_material)), + previous_key: None, } } + /// Additionally accept MACs keyed by `secret_material` (the retiring + /// PSK) during a rotation window. Verification-only: outbound MACs + /// always use the key from [`Self::new`]. See the type-level rotation + /// procedure. + #[must_use] + pub fn with_previous_secret(mut self, secret_material: &[u8]) -> Self { + self.previous_key = Some(Zeroizing::new(blake3::derive_key( + KEY_CONTEXT, + secret_material, + ))); + self + } + /// MAC the acceptor sends in the `ReplicaChallenge` frame. #[must_use] pub fn acceptor_mac(&self, transcript: &Transcript) -> [u8; MAC_LEN] { - *self.mac(DIR_ACCEPTOR, transcript).as_bytes() + *Self::mac(&self.key, DIR_ACCEPTOR, transcript).as_bytes() } /// MAC the dialer sends in the finish frame. #[must_use] pub fn dialer_mac(&self, transcript: &Transcript) -> [u8; MAC_LEN] { - *self.mac(DIR_DIALER, transcript).as_bytes() + *Self::mac(&self.key, DIR_DIALER, transcript).as_bytes() } /// Verify the acceptor's MAC in constant time. #[must_use] pub fn verify_acceptor_mac(&self, transcript: &Transcript, received: &[u8; MAC_LEN]) -> bool { - self.mac(DIR_ACCEPTOR, transcript) == Hash::from_bytes(*received) + self.verify(DIR_ACCEPTOR, transcript, received) } /// Verify the dialer's MAC in constant time. #[must_use] pub fn verify_dialer_mac(&self, transcript: &Transcript, received: &[u8; MAC_LEN]) -> bool { - self.mac(DIR_DIALER, transcript) == Hash::from_bytes(*received) + self.verify(DIR_DIALER, transcript, received) + } + + /// Accept a MAC produced under the current key or, during a rotation + /// window, under the previous key. Each comparison is constant-time + /// (`blake3::Hash` equality); trying the second key only reveals WHICH + /// cluster key signed, which any PSK holder already knows. + fn verify(&self, dir: u8, transcript: &Transcript, received: &[u8; MAC_LEN]) -> bool { + let received = Hash::from_bytes(*received); + if Self::mac(&self.key, dir, transcript) == received { + return true; + } + self.previous_key + .as_ref() + .is_some_and(|previous_key| Self::mac(previous_key, dir, transcript) == received) } - fn mac(&self, dir: u8, transcript: &Transcript) -> Hash { - let mut hasher = blake3::Hasher::new_keyed(&self.key); + fn mac(key: &[u8; 32], dir: u8, transcript: &Transcript) -> Hash { + let mut hasher = blake3::Hasher::new_keyed(key); hasher.update(DOMAIN_TAG); hasher.update(&transcript.cluster_id.to_le_bytes()); hasher.update(&[transcript.dialer_id, transcript.acceptor_id]); @@ -322,6 +365,57 @@ mod tests { assert!(!attacker.verify_dialer_mac(&t, &signer.dialer_mac(&t))); } + const NEW_SECRET: &[u8] = b"fedcba9876543210fedcba9876543210"; + + #[test] + fn rotation_window_accepts_both_keys_in_both_directions() { + // Mid-roll shape: the peer signs with either the retiring or the new + // PSK; a node carrying the window must verify both, in both handshake + // directions. + let old_signer = ReplicaAuth::new(SECRET); + let new_signer = ReplicaAuth::new(NEW_SECRET); + let window = ReplicaAuth::new(NEW_SECRET).with_previous_secret(SECRET); + let t = transcript(); + assert!(window.verify_dialer_mac(&t, &old_signer.dialer_mac(&t))); + assert!(window.verify_dialer_mac(&t, &new_signer.dialer_mac(&t))); + assert!(window.verify_acceptor_mac(&t, &old_signer.acceptor_mac(&t))); + assert!(window.verify_acceptor_mac(&t, &new_signer.acceptor_mac(&t))); + } + + #[test] + fn rotation_window_signs_with_current_key_only() { + // The previous key must never sign: a window node's MAC verifies + // under the current key alone, not under the retiring one. + let window = ReplicaAuth::new(NEW_SECRET).with_previous_secret(SECRET); + let old_only = ReplicaAuth::new(SECRET); + let new_only = ReplicaAuth::new(NEW_SECRET); + let t = transcript(); + let mac = window.dialer_mac(&t); + assert!(new_only.verify_dialer_mac(&t, &mac)); + assert!(!old_only.verify_dialer_mac(&t, &mac)); + } + + #[test] + fn closed_window_rejects_the_retired_key() { + // Step 3 of the rotation: dropping previous_shared_secret must stop + // accepting MACs keyed by the retired PSK. + let old_signer = ReplicaAuth::new(SECRET); + let closed = ReplicaAuth::new(NEW_SECRET); + let t = transcript(); + assert!(!closed.verify_dialer_mac(&t, &old_signer.dialer_mac(&t))); + assert!(!closed.verify_acceptor_mac(&t, &old_signer.acceptor_mac(&t))); + } + + #[test] + fn rotation_window_still_rejects_a_third_key() { + // The window widens acceptance to exactly {current, previous}; any + // other key keeps failing. + let window = ReplicaAuth::new(NEW_SECRET).with_previous_secret(SECRET); + let attacker = ReplicaAuth::new(b"ffffffffffffffffffffffffffffffff"); + let t = transcript(); + assert!(!window.verify_dialer_mac(&t, &attacker.dialer_mac(&t))); + } + #[test] fn direction_is_not_reflectable() { // The acceptor's MAC must not validate as a dialer MAC (reflection). diff --git a/core/message_bus/src/replica/handshake.rs b/core/message_bus/src/replica/handshake.rs index 4b5a4c5f6..226766604 100644 --- a/core/message_bus/src/replica/handshake.rs +++ b/core/message_bus/src/replica/handshake.rs @@ -54,8 +54,12 @@ //! trusted L2 boundary (cluster-local VPC, dedicated private subnet, //! encrypted overlay such as `WireGuard`, or an air-gapped management //! network). -// -// TODO(hubcio): follow-up - dual-key rotation acceptance window. +//! +//! Rotating the PSK is NOT a coordinated-restart change: a verify-only +//! acceptance window for the retiring key +//! (`cluster.auth.previous_shared_secret`) lets a rolling three-step +//! rotation keep every mid-roll handshake verifiable; see the +//! [`ReplicaAuth`] rustdoc for the procedure. use crate::framing; use crate::replica::auth::{self, ChannelBinding, HandshakeStatus, ReplicaAuth, Transcript}; @@ -64,6 +68,7 @@ use compio::io::{AsyncRead, AsyncWrite}; use iggy_binary_protocol::{Command2, HEADER_SIZE}; use iggy_common::IggyError; use rustls::pki_types::ServerName; +use std::collections::HashMap; use std::mem::size_of; use std::rc::Rc; use std::sync::Arc; @@ -90,9 +95,11 @@ pub struct ReplicaHandshakeCtx { /// `server` drives the acceptor (TLS server) role on a delegated inbound /// connection; `client` drives the dialer role on a delegated outbound /// one. `peer_names` maps a replica id to the name the dialer presents -/// in SNI and verifies the peer's certificate against; in self-signed -/// mode the client config's verifier accepts any certificate and only -/// the PSK handshake authenticates the peer, so the name only feeds SNI. +/// in SNI and verifies the peer's certificate against; keyed explicitly +/// by id (not roster position) so sparse ids from dynamic replica join +/// cannot verify against another peer's name. In self-signed mode the +/// client config's verifier accepts any certificate and only the PSK +/// handshake authenticates the peer, so the name only feeds SNI. /// /// Both configs are TLS 1.3 only with the `iggy-replica` ALPN /// ([`crate::transports::tls::REPLICA_ALPN`]); a client-plane TLS @@ -101,8 +108,8 @@ pub struct ReplicaHandshakeCtx { pub struct ReplicaTlsCtx { pub server: Arc<rustls::ServerConfig>, pub client: Arc<rustls::ClientConfig>, - /// Indexed by replica id; same length as the roster. - pub peer_names: Vec<ServerName<'static>>, + /// Keyed by replica id; one entry per roster node. + pub peer_names: HashMap<u8, ServerName<'static>>, } /// Run the acceptor half on a delegated inbound stream and return the diff --git a/core/message_bus/src/socket_opts.rs b/core/message_bus/src/socket_opts.rs index 7a85c1032..4c9a2ff94 100644 --- a/core/message_bus/src/socket_opts.rs +++ b/core/message_bus/src/socket_opts.rs @@ -45,22 +45,20 @@ pub fn apply_nodelay_for_connection(stream: &TcpStream) -> io::Result<()> { SockRef::from(stream).set_tcp_nodelay(true) } -/// Build a TCP listener that tolerates another socket already holding the port. +/// Build a TCP listener that tolerates `TIME_WAIT` leftovers and bound-but-idle +/// reservation sockets, but refuses a live listener on the same port. /// -/// Only shard 0 binds the real listener, so nothing here needs the load -/// balancing `SO_REUSEPORT` exists for; it only widens what the bind accepts. -/// -/// TODO: work out whether these listeners need `SO_REUSEPORT` at all. Dropping -/// it would make a stale process holding the port a loud bind failure instead -/// of two live listeners silently splitting the accept queue. +/// `SO_REUSEADDR` is all that takes: it rebinds over `TIME_WAIT` remnants of a +/// previous process's connections and over a bound-not-listening reservation +/// socket, while a second bind against an already-LISTENing socket still fails +/// with `EADDRINUSE`. `SO_REUSEPORT` is deliberately NOT set: only shard 0 +/// binds the real listener, so nothing here needs the load balancing it +/// exists for, and setting it would let a stale process holding the port +/// silently split the accept queue with this one instead of failing the bind +/// loudly. pub fn bind_reusable_tcp_listener(addr: SocketAddr) -> io::Result<TcpListener> { let socket = Socket::new(Domain::for_address(addr), Type::STREAM, Some(Protocol::TCP))?; socket.set_reuse_address(true)?; - #[cfg(all( - unix, - not(any(target_os = "illumos", target_os = "solaris", target_os = "cygwin")) - ))] - socket.set_reuse_port(true)?; socket.bind(&addr.into())?; socket.listen(libc::SOMAXCONN)?; socket.set_nonblocking(true)?; @@ -76,8 +74,7 @@ mod tests { use socket2::{Domain, Protocol, Socket, Type}; use std::net::{Ipv4Addr, SocketAddr}; - #[test] - fn reusable_tcp_listener_can_bind_over_reserved_port() { + fn reserve_socket() -> (Socket, SocketAddr) { let reserve_addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0); let socket = Socket::new( Domain::for_address(reserve_addr), @@ -86,20 +83,20 @@ mod tests { ) .expect("reserve socket"); socket.set_reuse_address(true).expect("reserve reuseaddr"); - #[cfg(all( - unix, - not(any(target_os = "illumos", target_os = "solaris", target_os = "cygwin")) - ))] - socket.set_reuse_port(true).expect("reserve reuseport"); - // Bound, not listening: a port held without joining the accept group, - // which is what the reuse flags have to tolerate and what a plain bind - // would reject. socket.bind(&reserve_addr.into()).expect("reserve bind"); let addr = socket .local_addr() .expect("reserve local addr") .as_socket() .expect("socket addr"); + (socket, addr) + } + + #[test] + fn reusable_tcp_listener_can_bind_over_reserved_port() { + // Bound, not listening: a port held without joining the accept group, + // which `SO_REUSEADDR` has to tolerate and a plain bind would reject. + let (_reserve, addr) = reserve_socket(); let runtime = Runtime::new().expect("runtime"); runtime.enter(|| { @@ -108,4 +105,22 @@ mod tests { assert_eq!(listener.local_addr().expect("listener addr"), addr); }); } + + #[test] + fn reusable_tcp_listener_refuses_port_with_live_listener() { + // A LISTENing socket on the port is a stale process still serving it. + // Without `SO_REUSEPORT` the bind must fail loudly instead of joining + // the accept group and silently splitting inbound connections. + let (listening, addr) = reserve_socket(); + listening.listen(1).expect("reserve listen"); + + let runtime = Runtime::new().expect("runtime"); + runtime.enter(|| { + let result = bind_reusable_tcp_listener(addr); + assert!( + result.is_err(), + "bind over a live listener must fail with EADDRINUSE" + ); + }); + } } diff --git a/core/message_bus/tests/common/mod.rs b/core/message_bus/tests/common/mod.rs index e7c7633d1..b90a1c65c 100644 --- a/core/message_bus/tests/common/mod.rs +++ b/core/message_bus/tests/common/mod.rs @@ -133,7 +133,12 @@ pub fn self_signed_replica_tls_ctx(replica_count: u8) -> Rc<ReplicaTlsCtx> { .with_no_client_auth(); client.alpn_protocols = vec![REPLICA_ALPN.to_vec()]; let peer_names = (0..replica_count) - .map(|_| ServerName::try_from("localhost").expect("static server name")) + .map(|replica_id| { + ( + replica_id, + ServerName::try_from("localhost").expect("static server name"), + ) + }) .collect(); Rc::new(ReplicaTlsCtx { server: Arc::new(server), diff --git a/core/message_bus/tests/replica_auth_handshake.rs b/core/message_bus/tests/replica_auth_handshake.rs index 1720800af..44e48930c 100644 --- a/core/message_bus/tests/replica_auth_handshake.rs +++ b/core/message_bus/tests/replica_auth_handshake.rs @@ -129,6 +129,34 @@ async fn wrong_key_rejects_peer() { acceptor.shutdown(Duration::from_secs(2)).await; } +#[compio::test] +async fn rotation_window_registers_mid_roll_peers() { + // Step-2 mid-roll of a PSK rotation: the acceptor already signs with the + // new key, the dialer still signs with the old one, and each carries the + // other key in its verify-only window. The mutual handshake must complete + // in both directions. + let acceptor = Rc::new(IggyMessageBus::new(0)); + let addr = spawn_acceptor( + &acceptor, + Some(ReplicaAuth::new(SECRET_B).with_previous_secret(SECRET_A)), + ) + .await; + + let dialer = Rc::new(IggyMessageBus::new(0)); + spawn_dialer( + &dialer, + Some(ReplicaAuth::new(SECRET_A).with_previous_secret(SECRET_B)), + addr, + ) + .await; + + wait_until(|| dialer.replicas().contains(1), Duration::from_secs(2)).await; + wait_until(|| acceptor.replicas().contains(0), Duration::from_secs(2)).await; + + dialer.shutdown(Duration::from_secs(2)).await; + acceptor.shutdown(Duration::from_secs(2)).await; +} + #[compio::test] async fn enforcement_rejects_legacy_peer() { // Acceptor requires auth; dialer speaks the legacy (no-nonce) protocol. diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 8d2c3a4bf..1652c72ec 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -22,8 +22,8 @@ use crate::log::SegmentedLog; use crate::messages_writer::MessagesWriter; use crate::offset_storage::{delete_persisted_offset, persist_offset, persist_offset_max}; use crate::poll_plan::{ - AutoCommitCtx, AutoCommitTarget, DiskReadPlan, DiskSegment, LastPolledCtx, PollPlan, PollTier, - ResidentTailSnapshot, + AutoCommitCtx, AutoCommitTarget, DiskReadPlan, DiskSegment, LastPolledCtx, + PartitionDirResolution, PollPlan, PollTier, ResidentTailSnapshot, }; use crate::segment::Segment; use crate::types::RepairSession; @@ -825,7 +825,7 @@ where }) .collect(); let disk = DiskReadPlan { - partition_dir: self.partition_dir(), + partition_dir: self.partition_dir_resolution(), segments, start_position, namespace_raw: self.namespace().inner(), @@ -1041,6 +1041,26 @@ where }) } + /// [`Self::partition_dir`] upgraded with the reason a dir is absent, so a + /// disk poll can tell file-less (simulated) storage from a live partition + /// whose dir is transiently unresolvable mid-rotation. Storage readers, + /// unlike writers, survive segment sealing, so any present reader or + /// writer proves file-backed data exists behind the missing dir. + fn partition_dir_resolution(&self) -> PartitionDirResolution { + if let Some(dir) = self.partition_dir() { + return PartitionDirResolution::Resolved(dir); + } + let file_backed = + self.log.storages().iter().any(|storage| { + storage.messages_reader.is_some() || storage.messages_writer.is_some() + }); + if file_backed { + PartitionDirResolution::Unresolvable + } else { + PartitionDirResolution::NoFiles + } + } + fn has_persisted_segment_bytes(&self) -> bool { self.log .segments() @@ -3884,7 +3904,7 @@ mod tests { // First segment claims persisted bytes but its file is absent, so the // open exhausts retries -> the walk must fault-close before segment two. let plan = DiskReadPlan { - partition_dir: Some(partition_dir), + partition_dir: PartitionDirResolution::Resolved(partition_dir), segments: vec![ DiskSegment { start_offset: 0, @@ -3967,7 +3987,7 @@ mod tests { } let plan = DiskReadPlan { - partition_dir: Some(partition_dir), + partition_dir: PartitionDirResolution::Resolved(partition_dir), segments: vec![ DiskSegment { start_offset: 0, @@ -3998,6 +4018,64 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A simulated (file-less) partition has no segment files by design, so a + /// disk poll with no dir must stay `Empty`: the caller then serves the + /// resident journal tier, the sim's only tier. + #[compio::test] + async fn read_disk_serves_journal_when_partition_has_no_files() { + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::NoFiles, + segments: vec![DiskSegment { + start_offset: 0, + persisted: 512, + }], + start_position: 0, + namespace_raw: IggyNamespace::new(1, 1, 0).inner(), + }; + + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 10, + ceiling: u64::MAX, + }) + .await; + + assert!( + matches!(outcome, DiskReadOutcome::Empty), + "file-less (simulated) storage must serve the journal tier, not fault", + ); + } + + /// A live partition whose dir is transiently unresolvable (mid-rotation) + /// may hold disk-resident data the walk cannot reach; the poll must + /// fault-close instead of letting the journal-forward skip those offsets. + #[compio::test] + async fn read_disk_faults_closed_when_partition_dir_unresolvable() { + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Unresolvable, + segments: vec![DiskSegment { + start_offset: 0, + persisted: 512, + }], + start_position: 0, + namespace_raw: IggyNamespace::new(1, 1, 0).inner(), + }; + + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 10, + ceiling: u64::MAX, + }) + .await; + + assert!( + matches!(outcome, DiskReadOutcome::Faulted), + "unresolvable dir over file-backed data must fault-close, not serve the journal", + ); + } + fn repair_config() -> PartitionsConfig { PartitionsConfig { messages_required_to_save: 1, diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 4792760d3..7c0e5afab 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -41,11 +41,28 @@ use std::sync::Arc; use std::sync::atomic::Ordering; use tracing::warn; +/// Where the disk tier's segment files live, resolved at plan-build time. +/// The two dir-less cases must stay distinct: only a genuinely file-less +/// partition may fall forward to the journal tier, while file-backed data +/// behind an unresolvable dir must fail closed (see [`DiskReadOutcome`]). +pub enum PartitionDirResolution { + /// The canonical partition directory to open segment files from. + Resolved(String), + /// No file-backed storage exists (simulated in-memory persistence): + /// there are no files to read and the journal tier is the only tier. + NoFiles, + /// File-backed storage exists but no directory was resolvable right now + /// (a live partition mid-rotation whose sealed segments dropped their + /// writer). Disk-resident data may be temporarily hidden, so the read + /// fails closed instead of letting the journal-forward skip it. + Unresolvable, +} + /// Owned, borrow-free inputs for the disk tier of a poll (see module docs). /// Segment files are re-opened by path because sealed segments drop their /// writer at rotation. pub struct DiskReadPlan { - pub(crate) partition_dir: Option<String>, + pub(crate) partition_dir: PartitionDirResolution, /// Segments to walk, snapshotted from the poll's starting segment onward /// (see `build_poll_plan`); `start_position` is the byte offset into the /// first one. @@ -385,23 +402,27 @@ impl DiskReadPlan { if count == 0 || self.segments.is_empty() { return DiskReadOutcome::Empty; } - let Some(partition_dir) = self.partition_dir.as_deref() else { - // Simulated in-memory persistence, or no writer was resolvable - // (e.g. mid-rotation): no files to read. This is not an IO fault on - // present data, so it is `Empty`: the caller serves the resident - // journal tier (the sim's only tier) without skipping anything. - // TODO(hubcio): a live partition mid-rotation can also land here - // with disk-resident-but-unresolvable data; the journal-forward - // could then skip those offsets. Distinguish sim/no-files (Empty) - // from a transiently-unresolvable writer (Faulted, fail-closed). - warn!( - target: "iggy.partitions.diag", - plane = "partitions", - namespace_raw = self.namespace_raw, - segment_count = self.segments.len(), - "disk poll: no partition dir to resolve segment files; serving journal tier" - ); - return DiskReadOutcome::Empty; + let partition_dir = match &self.partition_dir { + PartitionDirResolution::Resolved(dir) => dir.as_str(), + // Simulated in-memory persistence: no files exist, so this is not + // an IO fault on present data. `Empty` lets the caller serve the + // resident journal tier (the sim's only tier) without skipping + // anything. + PartitionDirResolution::NoFiles => return DiskReadOutcome::Empty, + // File-backed data exists but the dir was unresolvable at plan + // time (mid-rotation). Fail-closed like an IO fault: the + // journal-forward would splice resident ops over the hidden + // disk-resident offsets. A later poll resolves the dir again. + PartitionDirResolution::Unresolvable => { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace_raw, + segment_count = self.segments.len(), + "disk poll: file-backed partition has no resolvable dir; failing closed" + ); + return DiskReadOutcome::Faulted; + } }; // `start_position` applies to the first snapshotted segment; each later diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index c9ef5aa47..050d880e3 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -684,6 +684,15 @@ enabled = false # logs, never persisted) over storing it on disk. Ignored when enabled = false. shared_secret = "" +# Retiring pre-shared key, accepted for verification only during a rolling key +# rotation (this node keeps signing with shared_secret). Rotate in three rolls: +# 1) shared_secret = old + previous_shared_secret = new on every node, +# 2) shared_secret = new + previous_shared_secret = old on every node, +# 3) shared_secret = new alone. Leave empty outside a rotation. Same length +# floor and env-var preference as shared_secret +# (IGGY_CLUSTER_AUTH_PREVIOUS_SHARED_SECRET). +previous_shared_secret = "" + # Replica-to-replica TLS for the consensus (tcp_replica) port. [cluster.tls] # When true every replica connection is wrapped in TLS 1.3 (ALPN @@ -796,6 +805,12 @@ shutdown_drain_timeout = "10 s" # scheduler overhead. Must be less than or equal to shutdown_drain_timeout. shutdown_poll_interval = "50 ms" +# Hard wall-clock deadline for joining shard threads at process exit. A +# shard whose pump or listener wedges past this budget is abandoned with +# an error log instead of blocking exit forever. Must be at least +# shutdown_drain_timeout, or shards would be abandoned mid-drain. +shutdown_join_timeout = "30 s" + # Safety-tick cadence for the partition reconciliation loop. The reconciler # also wakes on every metadata commit from shard 0, so this only covers # dropped wake-ups and the initial post-bootstrap convergence window. diff --git a/core/server-ng/src/auth.rs b/core/server-ng/src/auth.rs index 61c67ec42..65dc83ab9 100644 --- a/core/server-ng/src/auth.rs +++ b/core/server-ng/src/auth.rs @@ -18,17 +18,18 @@ //! Local credential verification and login/register completion. //! //! Verifies password + PAT credentials locally, then runs the consensus -//! `Register` proposal on the metadata owner; also builds the terminal -//! login-failure replies. +//! `Register` proposal on the metadata owner; terminal failures are +//! surfaced as typed `Eviction` frames, transient ones as +//! `TransientNotCommitted` replay hints. use crate::bootstrap::{ShellBus, ShellShard}; -use crate::dispatch::submit_register_on_owner; +use crate::dispatch::{send_login_eviction, submit_register_on_owner}; use crate::login_register::LoginRegisterError; -use crate::responses::{build_empty_reply, build_login_register_reply, current_metadata_commit}; +use crate::responses::{build_login_register_reply, current_metadata_commit}; use crate::session_manager::{ClientSdkInfo, SessionManager}; use consensus::{MetadataHandle, build_result_rejection_reply}; use iggy_binary_protocol::PrepareHeader; -use iggy_binary_protocol::{ClientVersionInfo, RequestHeader}; +use iggy_binary_protocol::{ClientVersionInfo, EvictionReason, RequestHeader}; use iggy_common::defaults::{ MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH, }; @@ -267,7 +268,8 @@ where Ok(()) } -/// Decide whether a failed login/register gets a terminal reply or silence. +/// Decide whether a failed login/register gets a terminal eviction or a +/// transient replay hint. /// /// A transient consensus failure ([`LoginRegisterError::is_terminal`] is /// `false`) means the cluster could not commit *right now* (a freshly booted @@ -277,9 +279,9 @@ where /// break the replay. /// /// Terminal auth errors (`InvalidCredentials` / `InvalidToken` / -/// `UserInactive` / `Session`) fast-fail with an empty reply rather than make -/// the client wait for a timeout. (TODO: ship a typed `Eviction` frame once -/// the SDK eviction decoder lands on every transport.) +/// `UserInactive` / `Session`) fast-fail with a typed `Eviction` frame so the +/// SDK surfaces the real reason (every frame transport decodes +/// `Command2::Eviction`) instead of a decode error or a timeout. #[allow(clippy::future_not_send)] pub(crate) async fn surface_login_failure<B, MJ, S>( shard: &Rc<ShellShard<B, MJ, S>>, @@ -293,7 +295,13 @@ pub(crate) async fn surface_login_failure<B, MJ, S>( S: 'static, { if error.is_terminal() { - send_login_failure_reply(shard, transport_client_id, request_header).await; + send_login_eviction( + shard, + transport_client_id, + request_header.client, + eviction_reason_for(error), + ) + .await; } else { // Transient consensus failure (not-caught-up / not-primary / pipeline // full): send the explicit `TransientNotCommitted` frame instead of @@ -341,32 +349,50 @@ async fn send_login_transient_reply<B, MJ, S>( } } -/// Empty Reply on a terminal failed Register. The SDK only decodes -/// `Command2::Reply`; an empty body fails `LoginRegisterResponse` decoding -/// fast instead of hanging until the socket read timeout. Only call for -/// terminal errors -- see [`surface_login_failure`]. -#[allow(clippy::future_not_send)] -pub(crate) async fn send_login_failure_reply<B, MJ, S>( - shard: &Rc<ShellShard<B, MJ, S>>, - transport_client_id: u128, - request_header: &RequestHeader, -) where - B: ShellBus, - MJ: JournalHandle + 'static, - MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>, - S: 'static, -{ - let commit = current_metadata_commit(shard); - let reply = build_empty_reply(request_header, transport_client_id, 0, commit); - if let Err(error) = shard - .bus - .send_to_client(transport_client_id, reply.into_generic().into_frozen()) - .await - { - warn!( - transport_client_id, - error = %error, - "failed to send login-failure reply" - ); +/// Wire reason for a terminal login/register failure. Session-level +/// rejections (including the non-retryable submit refusal, where the +/// presented client id belongs to another user) collapse to +/// `SessionError`; the SDK maps it to `Unauthenticated`. +const fn eviction_reason_for(error: &LoginRegisterError) -> EvictionReason { + match error { + LoginRegisterError::InvalidCredentials => EvictionReason::InvalidCredentials, + LoginRegisterError::InvalidToken => EvictionReason::InvalidToken, + LoginRegisterError::UserInactive => EvictionReason::UserInactive, + _ => EvictionReason::SessionError, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use iggy_common::{IggyError, eviction_reason_to_error}; + + /// The reasons this module emits must map back to the credential + /// errors the SDK is expected to surface (the shared + /// `eviction_reason_to_error` grading both ends). + #[test] + fn terminal_login_errors_map_to_typed_sdk_errors() { + let cases = [ + ( + eviction_reason_for(&LoginRegisterError::InvalidCredentials), + IggyError::InvalidCredentials, + ), + ( + eviction_reason_for(&LoginRegisterError::InvalidToken), + IggyError::InvalidPersonalAccessToken, + ), + ( + eviction_reason_for(&LoginRegisterError::UserInactive), + IggyError::Unauthenticated, + ), + ]; + for (reason, expected) in cases { + let error = eviction_reason_to_error(reason, 0, 0); + assert_eq!( + error.as_code(), + expected.as_code(), + "reason {reason:?} must surface as {expected:?}" + ); + } } } diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index aed10ffb0..573b24f44 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -94,6 +94,7 @@ use shard::{ }; use shard_allocator::{ShardAllocator, ShardInfo}; use std::cell::RefCell; +use std::collections::HashMap; use std::env; use std::net::{IpAddr, SocketAddr}; use std::path::{Path, PathBuf}; @@ -101,7 +102,7 @@ use std::rc::{Rc, Weak}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use tracing::{error, info, warn}; const SHARD_REPLICA_ID: u8 = 0; @@ -214,10 +215,12 @@ pub type ServerNgShard = ShellShard<Rc<IggyMessageBus>, PrepareJournal, IggySnap /// /// Carries the cross-thread shutdown flag and one OS-thread `JoinHandle` /// per shard. The caller flips the flag via [`Self::install_ctrlc_handler`] -/// and then drains every shard via [`Self::join_all`]. +/// and then drains every shard via [`Self::join_all`], bounded by +/// `join_timeout` (`system.sharding.shutdown_join_timeout`). pub struct ShardHandles { shutdown_flag: Arc<AtomicBool>, shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerNgError>>)>, + join_timeout: Duration, } impl ShardHandles { @@ -245,27 +248,48 @@ impl ShardHandles { }) } - /// Drain every shard thread. Each shard's outcome is logged - /// (`info` on clean exit, `error` on Err or panic). If any shard - /// failed, returns every failure together as - /// [`ServerNgError::ShardJoinFailures`] so the operator sees the - /// full set rather than just the first. + /// Drain every shard thread under one shared `shutdown_join_timeout` + /// deadline. Each shard's outcome is logged (`info` on clean exit, + /// `error` on Err, panic, or wedge). If any shard failed, returns + /// every failure together as [`ServerNgError::ShardJoinFailures`] so + /// the operator sees the full set rather than just the first. + /// + /// A shard whose thread is still running at the deadline is + /// abandoned (its `JoinHandle` dropped, the OS thread left to die + /// with the process) and reported as + /// [`ShardJoinFailureKind::Wedged`]: a wedged pump or listener must + /// not block process exit forever. /// /// # Errors /// /// Returns [`ServerNgError::ShardJoinFailures`] if any shard - /// returned a `Result::Err` or panicked. The variant carries every - /// per-shard failure (`ShardJoinFailureKind::Error` or - /// `ShardJoinFailureKind::Panic`) in shard-id order so the caller - /// does not need to read the trace log to discover late-failing shards. + /// returned a `Result::Err`, panicked, or wedged past the deadline. + /// The variant carries every per-shard failure in shard-id order so + /// the caller does not need to read the trace log to discover + /// late-failing shards. pub fn join_all(self) -> Result<(), ServerNgError> { let mut failures: Vec<ShardJoinFailure> = Vec::new(); + let deadline = Instant::now() + self.join_timeout; // Shards run thread-per-core with compio's blocking fallback pool // disabled, so an io_uring opcode the kernel lacks aborts every shard // with the same panic. Surface the actionable diagnostic once. let mut io_uring_diagnostic_shown = false; for (shard_id, handle) in self.shard_threads { - match handle.join() { + let Some(joined) = join_with_deadline(handle, deadline) else { + error!( + shard_id, + waited = ?self.join_timeout, + "shard thread still running at the shutdown join deadline; abandoning it" + ); + failures.push(ShardJoinFailure { + shard_id, + kind: ShardJoinFailureKind::Wedged { + waited: self.join_timeout, + }, + }); + continue; + }; + match joined { Ok(Ok(())) => { info!(shard_id, "shard thread exited cleanly"); } @@ -301,6 +325,30 @@ impl ShardHandles { } } +/// Poll cadence for the bounded shard joins. Coarse enough to cost +/// nothing during a normal drain, fine enough that exit latency past +/// the last shard's return stays imperceptible. +const JOIN_POLL_INTERVAL: Duration = Duration::from_millis(25); + +/// Join `handle` unless `deadline` passes first. `None` means the +/// thread was still running at the deadline and the handle was dropped +/// (the OS thread keeps running detached; process exit reaps it). +/// `JoinHandle` has no timed join, so this polls `is_finished` at +/// [`JOIN_POLL_INTERVAL`]; the closing `join()` on a finished thread +/// returns immediately. +fn join_with_deadline( + handle: thread::JoinHandle<Result<(), ServerNgError>>, + deadline: Instant, +) -> Option<thread::Result<Result<(), ServerNgError>>> { + while !handle.is_finished() { + if Instant::now() >= deadline { + return None; + } + thread::sleep(JOIN_POLL_INTERVAL); + } + Some(handle.join()) +} + /// Best-effort extraction of the panic message from a /// `Box<dyn Any + Send>` returned by `JoinHandle::join`. Tries the two /// payload shapes the standard library guarantees (`&'static str` and @@ -316,73 +364,44 @@ fn panic_payload_to_string(payload: &(dyn std::any::Any + Send)) -> String { "<panic payload not String/&str>".to_string() } -/// Joins survivor shard threads after a partial-spawn failure without -/// panicking the bootstrap thread on `pthread_create` EAGAIN. -/// -/// Bare `thread::spawn` panics on EAGAIN, which is the most likely OS -/// state on this path since the parent `Builder::spawn` already failed -/// for the same reason. A panic would unwind `bootstrap()` while -/// survivor shard threads keep driving their compio runtimes and -/// `io_uring` rings, orphaning them across process exit. -/// -/// Uses `thread::Builder::spawn` and hands each survivor over via a -/// one-shot `sync_channel(1)` so an `Err` drops the rx (not the -/// survivor `JoinHandle`), letting us fall back to a sequential -/// `survivor.join()` instead. Once one cleanup spawn fails, treats the -/// OS as exhausted and routes every remaining survivor straight to the -/// sequential pool to avoid re-trying spawn. -/// -/// This routine bounds CPU/IO via the survivor's own -/// `shutdown_drain_timeout` (driven by each shard's watchdog after -/// `shutdown_flag` is set by the caller), not via a wall-clock -/// deadline here. If a survivor's `shard_main` blocks past the drain -/// window without observing the flag, this join hangs - that scenario -/// is the same surface as the deferred watchdog-detach gap and is not -/// addressed by this helper. +/// Joins survivor shard threads after a partial-spawn failure, bounded +/// by the same `shutdown_join_timeout` budget as the normal exit path. /// -/// TODO(hubcio): no hard time limit on shard shutdown here. If a -/// survivor's `shard_main` never returns, `survivor.join()` blocks -/// forever. +/// Polls every survivor's `is_finished` in one loop instead of spawning +/// per-survivor joiner threads: the likely OS state on this path is +/// `pthread_create` EAGAIN (the parent spawn just failed with it), so +/// nothing here may create threads, and polling drains all survivors in +/// parallel anyway. A survivor still running at the deadline is +/// abandoned with an error log so the failed bootstrap can surface its +/// spawn error instead of hanging on a wedged shard. fn join_partial_shard_survivors( shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerNgError>>)>, + join_timeout: Duration, ) { - let mut joiners = Vec::with_capacity(shard_threads.len()); - let mut sequential_join: Vec<thread::JoinHandle<Result<(), ServerNgError>>> = Vec::new(); - let mut spawn_exhausted = false; - for (sid, survivor) in shard_threads { - if spawn_exhausted { - sequential_join.push(survivor); - continue; - } - let (handover_tx, handover_rx) = - std::sync::mpsc::sync_channel::<thread::JoinHandle<Result<(), ServerNgError>>>(1); - match thread::Builder::new() - .name(format!("shard-{sid}-cleanup")) - .spawn(move || { - if let Ok(survivor) = handover_rx.recv() { - let _ = survivor.join(); - } - }) { - Ok(joiner) => { - let _ = handover_tx.send(survivor); - joiners.push(joiner); - } - Err(spawn_err) => { - warn!( - error = %spawn_err, - shard_id = sid, - "cleanup helper thread spawn failed; falling back to sequential survivor join" - ); - spawn_exhausted = true; - sequential_join.push(survivor); + let deadline = Instant::now() + join_timeout; + let mut remaining = shard_threads; + loop { + let mut still_running = Vec::with_capacity(remaining.len()); + for (shard_id, survivor) in remaining { + if survivor.is_finished() { + let _ = survivor.join(); + info!(shard_id, "survivor shard thread drained"); + } else { + still_running.push((shard_id, survivor)); } } + remaining = still_running; + if remaining.is_empty() || Instant::now() >= deadline { + break; + } + thread::sleep(JOIN_POLL_INTERVAL); } - for joiner in joiners { - let _ = joiner.join(); - } - for survivor in sequential_join { - let _ = survivor.join(); + for (shard_id, _survivor) in remaining { + error!( + shard_id, + waited = ?join_timeout, + "survivor shard thread still running at the shutdown join deadline; abandoning it" + ); } } @@ -741,7 +760,10 @@ pub fn bootstrap( drop(metadata_bundle_rx); drop(ready_tx); drop(ready_rx); - join_partial_shard_survivors(shard_threads); + join_partial_shard_survivors( + shard_threads, + config.system.sharding.shutdown_join_timeout.get_duration(), + ); return Err(ServerNgError::ShardSpawnFailed { shard_id, source }); } }; @@ -765,6 +787,7 @@ pub fn bootstrap( Ok(ShardHandles { shutdown_flag, shard_threads, + join_timeout: config.system.sharding.shutdown_join_timeout.get_duration(), }) } @@ -2990,18 +3013,23 @@ async fn start_client_listeners( /// Build the replica auth context from cluster config. Returns `None` when the /// cluster or replica auth is disabled, keeping the handshake in legacy mode. -/// Only the derived MAC key is carried onward in [`ReplicaAuth`]; the raw secret -/// (masked in config logs via `config_env(secret)`) is read here only to derive -/// that key. `ClusterConfig::validate` guarantees a non-empty secret whenever -/// both `cluster.enabled` and `cluster.auth.enabled` are set (validate -/// early-returns `Ok` while `cluster.enabled` is false). +/// Only the derived MAC keys are carried onward in [`ReplicaAuth`]; the raw +/// secrets (masked in config logs via `config_env(secret)`) are read here only +/// to derive them. A non-empty `previous_shared_secret` opens the verify-only +/// rotation acceptance window (see the [`ReplicaAuth`] rustdoc for the rolling +/// rotation procedure). `ClusterConfig::validate` guarantees a non-empty +/// secret whenever both `cluster.enabled` and `cluster.auth.enabled` are set +/// (validate early-returns `Ok` while `cluster.enabled` is false). fn load_replica_auth(config: &ServerNgConfig) -> Option<ReplicaAuth> { if !config.cluster.enabled || !config.cluster.auth.enabled { return None; } - Some(ReplicaAuth::new( - config.cluster.auth.shared_secret.as_bytes(), - )) + let auth = ReplicaAuth::new(config.cluster.auth.shared_secret.as_bytes()); + let previous_shared_secret = &config.cluster.auth.previous_shared_secret; + if previous_shared_secret.is_empty() { + return Some(auth); + } + Some(auth.with_previous_secret(previous_shared_secret.as_bytes())) } /// Build the replica TLS context from cluster config. Returns `None` when @@ -3079,17 +3107,15 @@ fn load_replica_tls_ctx( }; client.alpn_protocols = vec![REPLICA_ALPN.to_vec()]; - // Replica ids form a bijection onto 0..nodes.len() (validated at - // boot), so sorting by id yields a Vec indexable by replica id. - // TODO(hubcio): dynamic replica join will break this positional - // indexing (sparse ids silently map to the wrong SNI/verify name); - // key peer names by replica id explicitly before supporting it. - let mut roster: Vec<_> = config.cluster.nodes.iter().collect(); - roster.sort_unstable_by_key(|node| node.replica_id); - let peer_names = roster + // Keyed by replica id, never by roster position: sparse ids (dynamic + // replica join) would make a positional lookup verify against another + // peer's SNI name. + let peer_names = config + .cluster + .nodes .iter() .map(|node| { - ServerName::try_from(node.ip.clone()).map_err(|error| { + let name = ServerName::try_from(node.ip.clone()).map_err(|error| { credential_error(std::io::Error::new( std::io::ErrorKind::InvalidInput, format!( @@ -3097,9 +3123,10 @@ fn load_replica_tls_ctx( node.name, node.ip ), )) - }) + })?; + Ok((node.replica_id, name)) }) - .collect::<Result<Vec<_>, _>>()?; + .collect::<Result<HashMap<_, _>, ServerNgError>>()?; Ok(Some(ReplicaTlsCtx { server: Arc::new(server), diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 3c7b7cf6b..425e6348d 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -25,8 +25,8 @@ mod authz; use crate::auth::{ - complete_login_register, send_login_failure_reply, surface_login_failure, - verify_login_credentials, verify_pat_credentials, + complete_login_register, surface_login_failure, verify_login_credentials, + verify_pat_credentials, }; use crate::bootstrap::{ShellBus, ShellShard, ShellShardHandle}; use crate::cluster_meta::ClusterRoster; @@ -2700,9 +2700,15 @@ async fn handle_login_register_request<B, MJ, S>( warn!( transport_client_id, - "dropping register request with unsupported payload shape" + "rejecting register request with unsupported payload shape" ); - send_login_failure_reply(shard, transport_client_id, request.header()).await; + send_login_eviction( + shard, + transport_client_id, + request.header().client, + EvictionReason::MalformedLogin, + ) + .await; } /// Best-effort login-rejection eviction. Terminal one-way frame; a gone @@ -2711,7 +2717,7 @@ async fn handle_login_register_request<B, MJ, S>( /// metadata shard and zeroed elsewhere -- the SDK only reads the reason, /// plus the protocol window on `IncompatibleProtocol`. #[allow(clippy::future_not_send)] -async fn send_login_eviction<B, MJ, S>( +pub(crate) async fn send_login_eviction<B, MJ, S>( shard: &Rc<ShellShard<B, MJ, S>>, transport_client_id: u128, vsr_client_id: u128, diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs index b19f26778..5da767272 100644 --- a/core/server-ng/src/server_error.rs +++ b/core/server-ng/src/server_error.rs @@ -206,7 +206,14 @@ pub struct ShardJoinFailure { #[derive(Debug)] pub enum ShardJoinFailureKind { Error(Box<ServerNgError>), - Panic { message: String }, + Panic { + message: String, + }, + /// The shard thread never finished inside `shutdown_join_timeout` + /// and was abandoned so process exit is not blocked forever. + Wedged { + waited: std::time::Duration, + }, } fn format_shard_failures(failures: &[ShardJoinFailure]) -> String { @@ -223,6 +230,13 @@ fn format_shard_failures(failures: &[ShardJoinFailure]) -> String { ShardJoinFailureKind::Panic { message } => { let _ = write!(out, "shard {} panicked: {message}", failure.shard_id); } + ShardJoinFailureKind::Wedged { waited } => { + let _ = write!( + out, + "shard {} wedged: thread still running after {waited:?}, abandoned", + failure.shard_id + ); + } } } out
