feat(server-ng): harden listener, auth, shutdown, and disk-poll paths

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,

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());
     }
 }

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
 ///

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()))?;

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
 ///

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"