This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 53a7ff99e feat(server-ng): harden listener, auth, shutdown, and
disk-poll paths (#3774)
53a7ff99e is described below
commit 53a7ff99e94a9bb587e6195e8b30f5556660babf
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Fri Jul 31 16:43:00 2026 +0200
feat(server-ng): harden listener, auth, shutdown, and disk-poll paths
(#3774)
---
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 | 285 ++++++++++++++++-------
core/server-ng/src/dispatch.rs | 16 +-
core/server-ng/src/server_error.rs | 16 +-
18 files changed, 758 insertions(+), 216 deletions(-)
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
index d2d097549..22f750402 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 5aa6d075b..c118e8d02 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
@@ -799,6 +808,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 d80e59213..d99815414 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,60 @@ 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
+ /// Drain every shard thread. This is the main thread's park for the
+ /// server's whole lifetime, so shards are awaited WITHOUT any time
+ /// bound while the server runs; the `shutdown_join_timeout` clock
+ /// only starts once the cross-thread shutdown flag flips (Ctrl-C or
+ /// a shard failure). 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 when the post-shutdown
+ /// deadline passes 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();
+ // Armed on the first poll that observes the shutdown flag, shared
+ // across all shards: one budget covers the whole drain, not one
+ // budget per shard.
+ let mut deadline: Option<Instant> = None;
// 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_until_shutdown_deadline(
+ handle,
+ &self.shutdown_flag,
+ self.join_timeout,
+ &mut 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 +337,41 @@ 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`, waiting indefinitely while the server runs. The
+/// `join_timeout` clock starts only when `shutdown_flag` is observed set
+/// (arming the caller-shared `deadline` once, so all shards drain under
+/// ONE budget); a running server parked here for hours must never be
+/// mistaken for a wedged shard. `None` means the thread was still
+/// running at the post-shutdown 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_until_shutdown_deadline(
+ handle: thread::JoinHandle<Result<(), ServerNgError>>,
+ shutdown_flag: &AtomicBool,
+ join_timeout: Duration,
+ deadline: &mut Option<Instant>,
+) -> Option<thread::Result<Result<(), ServerNgError>>> {
+ while !handle.is_finished() {
+ if deadline.is_none() && shutdown_flag.load(Ordering::Relaxed) {
+ *deadline = Some(Instant::now() + join_timeout);
+ }
+ if let Some(deadline) = deadline
+ && 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 +387,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.
+/// Joins survivor shard threads after a partial-spawn failure, bounded
+/// by the same `shutdown_join_timeout` budget as the normal exit path.
///
-/// 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.
-///
-/// 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 +783,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 +810,7 @@ pub fn bootstrap(
Ok(ShardHandles {
shutdown_flag,
shard_threads,
+ join_timeout:
config.system.sharding.shutdown_join_timeout.get_duration(),
})
}
@@ -3023,18 +3069,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
@@ -3112,17 +3163,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!(
@@ -3130,9 +3179,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),
@@ -3921,6 +3971,61 @@ mod tests {
assert_eq!(addrs.http, None);
}
+ /// Regression: the shutdown-join deadline must arm at SHUTDOWN, not
+ /// at boot. The original bound measured from `join_all` entry, so any
+ /// healthy server outliving `shutdown_join_timeout` (30s default) was
+ /// abandoned as "wedged" and the process exited - every BDD run died
+ /// at t+30s while the test container was still compiling.
+ #[test]
+ fn join_waits_unbounded_while_the_server_runs() {
+ let shutdown_flag = AtomicBool::new(false);
+ // Thread outlives a deliberately tiny join budget; with the flag
+ // clear the budget must never even arm.
+ let handle = thread::spawn(|| -> Result<(), ServerNgError> {
+ thread::sleep(Duration::from_millis(300));
+ Ok(())
+ });
+ let mut deadline = None;
+ let joined = join_until_shutdown_deadline(
+ handle,
+ &shutdown_flag,
+ Duration::from_millis(20),
+ &mut deadline,
+ );
+ assert!(
+ matches!(joined, Some(Ok(Ok(())))),
+ "a running server must be awaited indefinitely, not abandoned as
wedged"
+ );
+ assert!(
+ deadline.is_none(),
+ "the join deadline must not arm before the shutdown flag flips"
+ );
+ }
+
+ #[test]
+ fn join_abandons_a_wedged_shard_after_the_shutdown_deadline() {
+ let shutdown_flag = AtomicBool::new(true);
+ // Never finishes: stands in for a wedged pump. The thread leaks
+ // into the test process, which exits right after.
+ let handle = thread::spawn(|| -> Result<(), ServerNgError> {
+ loop {
+ thread::sleep(Duration::from_secs(1));
+ }
+ });
+ let mut deadline = None;
+ let joined = join_until_shutdown_deadline(
+ handle,
+ &shutdown_flag,
+ Duration::from_millis(100),
+ &mut deadline,
+ );
+ assert!(
+ joined.is_none(),
+ "a shard still running past the post-shutdown budget must be
abandoned"
+ );
+ assert!(deadline.is_some(), "the deadline arms once the flag is set");
+ }
+
#[test]
fn cluster_tcp_addr_takes_port_from_roster() {
// Same rule as the other transports: the roster owns the port so
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index b6bea8cd8..3d0179b05 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;
@@ -2711,9 +2711,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
@@ -2722,7 +2728,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 61cd1adf4..262170b81 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -210,7 +210,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 {
@@ -227,6 +234,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
