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 4c30a88a0 feat(server-ng): add cluster.heartbeat_timeout for VSR
liveness window (#3738)
4c30a88a0 is described below
commit 4c30a88a0f9eb811a6263d83636e5c8ebed06e10
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Thu Jul 23 15:49:14 2026 +0200
feat(server-ng): add cluster.heartbeat_timeout for VSR liveness window
(#3738)
---
core/configs/src/lib.rs | 4 +-
core/configs/src/server_ng_config/cluster.rs | 619 +++++++++++++++++++++
core/configs/src/server_ng_config/defaults.rs | 57 +-
core/configs/src/server_ng_config/mod.rs | 1 +
core/configs/src/server_ng_config/server_ng.rs | 2 +-
core/consensus/src/impls.rs | 8 +
core/consensus/src/lib.rs | 1 +
core/consensus/src/vsr_timeout.rs | 15 +-
core/integration/tests/server/message_retrieval.rs | 13 +
core/server-ng/config.toml | 7 +
core/server-ng/src/bootstrap.rs | 42 +-
core/server-ng/src/cluster_meta.rs | 2 +-
core/server-ng/src/http.rs | 2 +-
core/server-ng/src/http/error.rs | 2 +-
core/server-ng/src/partition_helpers.rs | 1 +
15 files changed, 758 insertions(+), 18 deletions(-)
diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs
index c9322f002..5c82dbd3a 100644
--- a/core/configs/src/lib.rs
+++ b/core/configs/src/lib.rs
@@ -30,6 +30,6 @@ pub use server_config::{
tcp, validators, websocket,
};
pub use server_ng_config::{
- COMPONENT_NG, message_bus, metadata as ng_metadata, quic as ng_quic,
server_ng,
- sharding as ng_sharding, tcp as ng_tcp, websocket as ng_websocket,
+ COMPONENT_NG, cluster as ng_cluster, message_bus, metadata as ng_metadata,
quic as ng_quic,
+ server_ng, sharding as ng_sharding, tcp as ng_tcp, websocket as
ng_websocket,
};
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
new file mode 100644
index 000000000..b8dc2d7f4
--- /dev/null
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -0,0 +1,619 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Server-ng cluster schema.
+//!
+//! Field shape mirrors the legacy [`crate::cluster::ClusterConfig`]
+//! plus the ng-only `heartbeat_timeout` knob; the type is forked into
+//! `server_ng_config` so server-ng can evolve its cluster surface
+//! (VSR consensus tunables) independently of the legacy server.
+
+use super::defaults::SERVER_NG_CONFIG;
+use crate::ConfigurationError;
+use configs::ConfigEnv;
+use iggy_common::{IggyDuration, Validatable};
+use serde::{Deserialize, Serialize};
+use serde_with::{DisplayFromStr, serde_as};
+use std::time::Duration;
+
+/// The primary heartbeats roughly every second (`PING_TICKS`); a window at
+/// or below one ping interval would elect on every scheduling hiccup.
+pub const MIN_CLUSTER_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Length floor for the replica-auth PSK, in raw bytes. The 32-byte MAC key
+/// is KDF-derived from these bytes at use-site, so any encoding clearing this
+/// length is accepted.
+const MIN_SHARED_SECRET_LEN: usize = 32;
+
+/// serde fallback for configs written before the field existed; the value
+/// itself lives in `core/server-ng/config.toml` like every other default.
+fn default_heartbeat_timeout() -> IggyDuration {
+ SERVER_NG_CONFIG.cluster.heartbeat_timeout.parse().unwrap()
+}
+
+#[serde_as]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterConfig {
+ pub enabled: bool,
+ pub name: String,
+ /// Backup-side liveness window for a plane's primary. A replica that
+ /// sees no primary traffic for this long starts a view change
+ /// (`normal_heartbeat_timeout`). Raise it on oversubscribed hosts where
+ /// scheduling stalls fake primary death;
sub-`MIN_CLUSTER_HEARTBEAT_TIMEOUT`
+ /// values (including the `0` / `disabled` / `unlimited` sentinels, which
+ /// all parse to zero) are rejected at boot.
+ #[serde(default = "default_heartbeat_timeout")]
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub heartbeat_timeout: IggyDuration,
+ /// Full roster of cluster members. Intended to be byte-identical across
+ /// every node so operators ship one config. The running node's identity
+ /// is supplied out-of-band via the `--replica-id` CLI flag, which
+ /// selects the entry in this list that describes the current node.
+ #[serde(default)]
+ pub nodes: Vec<ClusterNodeConfig>,
+ /// Replica-to-replica authentication settings (PSK + BLAKE3 handshake).
+ #[serde(default)]
+ pub auth: ClusterAuthConfig,
+ /// Replica-to-replica TLS settings for the consensus (`tcp_replica`) port.
+ #[serde(default)]
+ pub tls: ClusterTlsConfig,
+}
+
+/// Replica-to-replica authentication for the consensus (`tcp_replica`) port.
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterAuthConfig {
+ /// When true, every replica peer must complete the authenticated handshake
+ /// or be rejected, and [`Self::shared_secret`] is mandatory. When false
+ /// (default) the replica handshake stays in legacy unauthenticated mode
and
+ /// `shared_secret` is not used for authentication. A configured non-empty
+ /// `shared_secret` must still meet the 32-byte minimum whenever the
cluster
+ /// is enabled (a short value fails boot even with auth off).
+ ///
+ /// Enabling auth is a coordinated-restart change, and not the only one:
the
+ /// consensus `cluster_id` is derived from `ClusterConfig::name`
+ /// unconditionally, so a mixed-version roster fails to connect regardless
of
+ /// this flag. Flip every node in one restart.
+ #[serde(default)]
+ pub enabled: bool,
+ /// Cluster-wide pre-shared key for replica-to-replica authentication.
+ ///
+ /// At least 32 bytes of CSPRNG output, byte-identical across every node.
+ /// Provisioned out-of-band, normally via `IGGY_CLUSTER_AUTH_SHARED_SECRET`
+ /// rather than the on-disk config.
+ // skip_serializing keeps the PSK out of the runtime `current_config.toml`
+ // (and the `ServerConfig` diagnostic snapshot that cats it). The live
+ // secret is read from env / on-disk config at boot, never from the
+ // snapshot, so it must never be persisted there. Deserialize is retained.
+ #[serde(default, skip_serializing)]
+ #[config_env(secret)]
+ pub shared_secret: String,
+}
+
+/// Replica-to-replica TLS for the consensus (`tcp_replica`) port.
+///
+/// Mirrors the legacy [`crate::tcp::TcpTlsConfig`] shape plus `ca_file`:
+/// the replica plane DIALS its peers (a TLS client role the
+/// client-facing server plane never has), so the dialer needs a trust
+/// anchor to verify the acceptor's certificate against.
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterTlsConfig {
+ /// When true every replica connection is wrapped in TLS (1.3 only)
+ /// before the replica handshake runs. Requires `cluster.auth.enabled`:
+ /// TLS carries no client certificates, so it authenticates the
+ /// acceptor only; the PSK handshake authenticates the peer while TLS
+ /// supplies confidentiality. Enabling is a coordinated-restart
+ /// change: a TLS dialer cannot talk to a plaintext acceptor or vice
+ /// versa. Flip every node in one restart.
+ #[serde(default)]
+ pub enabled: bool,
+ /// When true the node auto-generates a self-signed certificate at
+ /// boot and the dialer accepts ANY peer certificate. With the
+ /// default `false`, `cert_file` / `key_file` / `ca_file` are all
+ /// required.
+ #[serde(default)]
+ pub self_signed: bool,
+ /// PEM certificate chain presented by this node's acceptor side.
+ #[serde(default)]
+ pub cert_file: String,
+ /// PEM private key matching `cert_file`.
+ #[serde(default)]
+ pub key_file: String,
+ /// PEM trust anchor(s) the dialer verifies peer certificates
+ /// against. Unused when `self_signed` is true.
+ #[serde(default)]
+ pub ca_file: String,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+pub struct ClusterNodeConfig {
+ pub name: String,
+ pub ip: String,
+ /// Numeric replica ID for VSR consensus (0-based).
+ ///
+ /// Must be unique across [`ClusterConfig::nodes`] and strictly less than
+ /// `nodes.len()`. Validated by [`ClusterConfig::validate`].
+ pub replica_id: u8,
+ pub ports: TransportPorts,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Default, ConfigEnv)]
+pub struct TransportPorts {
+ pub tcp: Option<u16>,
+ pub quic: Option<u16>,
+ pub http: Option<u16>,
+ pub websocket: Option<u16>,
+ /// Dedicated port for replica-to-replica consensus traffic.
+ pub tcp_replica: Option<u16>,
+}
+
+impl Validatable<ConfigurationError> for ClusterConfig {
+ fn validate(&self) -> Result<(), ConfigurationError> {
+ if !self.enabled {
+ return Ok(());
+ }
+
+ if self.name.trim().is_empty() {
+ eprintln!("Invalid cluster configuration: cluster name cannot be
empty");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ // `0` / `disabled` / `unlimited` all parse to a zero duration and
+ // land here too: there is no way to switch the liveness window off.
+ if self.heartbeat_timeout.get_duration() <
MIN_CLUSTER_HEARTBEAT_TIMEOUT {
+ eprintln!(
+ "Invalid cluster configuration: cluster.heartbeat_timeout '{}'
must be at least {}s \
+ (the primary heartbeats every second; a shorter window elects
on every hiccup)",
+ self.heartbeat_timeout,
+ MIN_CLUSTER_HEARTBEAT_TIMEOUT.as_secs()
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ if self.nodes.is_empty() {
+ eprintln!(
+ "Invalid cluster configuration: cluster.nodes must contain at
least one entry when cluster is enabled"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ // VSR needs every replica to have a stable, unique id strictly
+ // less than the total replica count. Duplicate ids would split the
+ // cluster into two replicas claiming the same slot; out-of-range
+ // ids never win a primary election. Both are unrecoverable at
+ // runtime - fail fast at startup.
+ let total_replicas = u8::try_from(self.nodes.len()).map_err(|_| {
+ eprintln!("Invalid cluster configuration: more than 255 replicas
is unsupported");
+ ConfigurationError::InvalidConfigurationValue
+ })?;
+
+ let mut seen_ids = std::collections::HashSet::new();
+ let mut seen_names = std::collections::HashSet::new();
+ let mut used_endpoints = std::collections::HashSet::new();
+
+ for node in &self.nodes {
+ if node.name.trim().is_empty() {
+ eprintln!("Invalid cluster configuration: node name cannot be
empty");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ if node.ip.trim().is_empty() {
+ eprintln!(
+ "Invalid cluster configuration: IP cannot be empty for
node '{}'",
+ node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ if !seen_names.insert(node.name.clone()) {
+ eprintln!(
+ "Invalid cluster configuration: duplicate node name '{}'
found",
+ node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ if node.replica_id >= total_replicas {
+ eprintln!(
+ "Invalid cluster configuration: replica_id {} for node
'{}' must be < total replica count {total_replicas}",
+ node.replica_id, node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ if !seen_ids.insert(node.replica_id) {
+ eprintln!(
+ "Invalid cluster configuration: duplicate replica_id {}
(two nodes claim the same slot)",
+ node.replica_id
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let port_list = [
+ ("TCP", node.ports.tcp),
+ ("QUIC", node.ports.quic),
+ ("HTTP", node.ports.http),
+ ("WebSocket", node.ports.websocket),
+ ("TCP_REPLICA", node.ports.tcp_replica),
+ ];
+
+ for (name, port_opt) in &port_list {
+ if let Some(port) = port_opt {
+ if *port == 0 {
+ eprintln!(
+ "Invalid cluster configuration: {} port cannot be
0 for node '{}'",
+ name, node.name
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let endpoint = format!("{}:{}", node.ip, port);
+ if !used_endpoints.insert(endpoint.clone()) {
+ eprintln!(
+ "Invalid cluster configuration: port conflict -
{endpoint} is already bound (node '{}', transport {name})",
+ node.name
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ }
+ }
+ }
+
+ // Replica-auth PSK (only reached when the cluster is enabled; the
early
+ // return above skips these while it is disabled). When auth is enabled
+ // the key is mandatory; any configured key must clear the length
floor -
+ // a typo guard that fires with auth off too, though only while the
+ // cluster itself is enabled.
+ let secret_len = self.auth.shared_secret.len();
+ if self.auth.enabled && self.auth.shared_secret.is_empty() {
+ eprintln!(
+ "Invalid cluster configuration: cluster.auth.shared_secret
must be set when cluster.auth.enabled is true"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if !self.auth.shared_secret.is_empty() && secret_len <
MIN_SHARED_SECRET_LEN {
+ eprintln!(
+ "Invalid cluster configuration: cluster.auth.shared_secret
must be >= {MIN_SHARED_SECRET_LEN} bytes"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ // Replica TLS. Both cert modes run one-directional TLS (no client
+ // certificate anywhere), so TLS only authenticates the acceptor to
+ // the dialer; peer authentication comes solely from the PSK
+ // handshake. Without it any TLS-capable host could register as a
+ // replica - require auth in both modes. CA mode (the default)
+ // additionally needs all three PEM paths: cert/key for this node's
+ // acceptor side, ca_file as the dialer's trust anchor.
+ if self.tls.enabled {
+ if !self.auth.enabled {
+ eprintln!(
+ "Invalid cluster configuration: cluster.tls.enabled = true
requires cluster.auth.enabled = true (TLS authenticates the acceptor only; the
PSK handshake authenticates the peer)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if !self.tls.self_signed {
+ for (field, value) in [
+ ("cert_file", &self.tls.cert_file),
+ ("key_file", &self.tls.key_file),
+ ("ca_file", &self.tls.ca_file),
+ ] {
+ if value.trim().is_empty() {
+ eprintln!(
+ "Invalid cluster configuration:
cluster.tls.{field} must be set when cluster.tls.enabled = true and self_signed
= false"
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn shared_secret_is_never_serialized() {
+ // Regression guard: the runtime current_config.toml (and the
+ // ServerConfig diagnostic snapshot that cats it) are produced by
+ // serializing this struct, so the PSK must not survive serialize.
+ // skip_serializing is format-agnostic, so a JSON dump proves the toml
+ // path too.
+ let config = ClusterConfig {
+ enabled: true,
+ name: "iggy-cluster".to_owned(),
+ heartbeat_timeout: default_heartbeat_timeout(),
+ nodes: Vec::new(),
+ auth: ClusterAuthConfig {
+ enabled: true,
+ shared_secret: "current-psk-MUST-NOT-be-persisted".to_owned(),
+ },
+ tls: ClusterTlsConfig::default(),
+ };
+ let serialized = serde_json::to_string(&config).expect("serialize
cluster config");
+ assert!(
+ !serialized.contains("MUST-NOT-be-persisted"),
+ "PSK leaked into serialized config: {serialized}"
+ );
+ assert!(
+ !serialized.contains("shared_secret"),
+ "shared_secret field present in serialized config: {serialized}"
+ );
+ }
+}
+
+#[cfg(test)]
+mod cluster_validate_tests {
+ use super::*;
+
+ fn node(name: &str, id: u8) -> ClusterNodeConfig {
+ ClusterNodeConfig {
+ name: name.to_string(),
+ ip: "127.0.0.1".to_string(),
+ replica_id: id,
+ ports: TransportPorts::default(),
+ }
+ }
+
+ fn cfg(nodes: Vec<ClusterNodeConfig>) -> ClusterConfig {
+ ClusterConfig {
+ enabled: true,
+ name: "iggy-cluster".to_string(),
+ heartbeat_timeout: default_heartbeat_timeout(),
+ nodes,
+ auth: ClusterAuthConfig::default(),
+ tls: ClusterTlsConfig::default(),
+ }
+ }
+
+ #[test]
+ fn validate_rejects_sub_minimum_heartbeat_timeout() {
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.heartbeat_timeout = IggyDuration::new(Duration::from_millis(500));
+ assert!(c.validate().is_err());
+ // The "disabled" / "unlimited" sentinels collapse to zero and must
+ // be rejected the same way.
+ c.heartbeat_timeout = IggyDuration::new(Duration::ZERO);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_empty_nodes() {
+ let c = cfg(vec![]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_duplicate_replica_ids() {
+ let c = cfg(vec![node("n1", 0), node("n2", 0)]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_duplicate_names() {
+ let c = cfg(vec![node("n1", 0), node("n1", 1)]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_out_of_range_replica_id() {
+ // 2 nodes total, so id 2 is out of range.
+ let c = cfg(vec![node("n1", 0), node("n2", 2)]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_unique_contiguous_replica_ids() {
+ let c = cfg(vec![node("n1", 0), node("n2", 1), node("n3", 2)]);
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_skips_checks_when_disabled() {
+ let mut c = cfg(vec![]);
+ c.enabled = false;
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_duplicate_tcp_replica_port() {
+ let ports = TransportPorts {
+ tcp: None,
+ quic: None,
+ http: None,
+ websocket: None,
+ tcp_replica: Some(9090),
+ };
+ let mut n1 = node("n1", 0);
+ n1.ports = ports.clone();
+ let mut n2 = node("n2", 1);
+ n2.ports = ports;
+ let c = cfg(vec![n1, n2]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_cross_transport_port_reuse() {
+ let mut n1 = node("n1", 0);
+ n1.ports = TransportPorts {
+ tcp: Some(8090),
+ quic: None,
+ http: Some(8090),
+ websocket: None,
+ tcp_replica: None,
+ };
+ let c = cfg(vec![n1]);
+ assert!(
+ c.validate().is_err(),
+ "same port on TCP and HTTP of the same node must be rejected"
+ );
+ }
+
+ #[test]
+ fn validate_accepts_same_port_on_different_ips() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "127.0.0.1".to_string();
+ n1.ports = TransportPorts {
+ tcp: Some(8090),
+ quic: None,
+ http: None,
+ websocket: None,
+ tcp_replica: None,
+ };
+ let mut n2 = node("n2", 1);
+ n2.ip = "127.0.0.2".to_string();
+ n2.ports = TransportPorts {
+ tcp: Some(8090),
+ quic: None,
+ http: None,
+ websocket: None,
+ tcp_replica: None,
+ };
+ let c = cfg(vec![n1, n2]);
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_zero_tcp_replica_port() {
+ let ports = TransportPorts {
+ tcp: None,
+ quic: None,
+ http: None,
+ websocket: None,
+ tcp_replica: Some(0),
+ };
+ let mut n1 = node("n1", 0);
+ n1.ports = ports;
+ let c = cfg(vec![n1]);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_empty_secret_when_auth_disabled() {
+ // Default: no secret, auth off -> legacy mode, must pass.
+ let c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_missing_secret_when_auth_enabled() {
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.auth.enabled = true;
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_short_secret_when_auth_enabled() {
+ 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 - 1);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_short_secret_even_when_auth_disabled() {
+ // Typo guard: a configured-but-short key fails even with auth off.
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN - 1);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_valid_secret_when_auth_enabled() {
+ 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);
+ assert!(c.validate().is_ok());
+ }
+
+ fn tls_files() -> ClusterTlsConfig {
+ ClusterTlsConfig {
+ enabled: true,
+ self_signed: false,
+ cert_file: "cert.pem".to_string(),
+ key_file: "key.pem".to_string(),
+ ca_file: "ca.pem".to_string(),
+ }
+ }
+
+ #[test]
+ fn validate_rejects_tls_ca_mode_with_missing_files() {
+ // Auth on so the failure exercises the file check, not the auth gate.
+ for missing in ["cert_file", "key_file", "ca_file"] {
+ 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.tls = tls_files();
+ match missing {
+ "cert_file" => c.tls.cert_file.clear(),
+ "key_file" => c.tls.key_file.clear(),
+ _ => c.tls.ca_file.clear(),
+ }
+ assert!(c.validate().is_err(), "missing {missing} must be
rejected");
+ }
+ }
+
+ #[test]
+ fn validate_rejects_tls_self_signed_without_auth() {
+ // Accept-any certificate without the PSK handshake = MITM-able.
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.tls = ClusterTlsConfig {
+ enabled: true,
+ self_signed: true,
+ ..ClusterTlsConfig::default()
+ };
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_tls_self_signed_with_auth() {
+ 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.tls = ClusterTlsConfig {
+ enabled: true,
+ self_signed: true,
+ ..ClusterTlsConfig::default()
+ };
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_tls_ca_mode_without_auth() {
+ // TLS never authenticates the dialer (no client certificates);
+ // only the PSK handshake does, so it is mandatory with TLS on.
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.tls = tls_files();
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_tls_ca_mode_with_auth() {
+ 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.tls = tls_files();
+ assert!(c.validate().is_ok());
+ }
+}
diff --git a/core/configs/src/server_ng_config/defaults.rs
b/core/configs/src/server_ng_config/defaults.rs
index 1c9b20483..58887a70f 100644
--- a/core/configs/src/server_ng_config/defaults.rs
+++ b/core/configs/src/server_ng_config/defaults.rs
@@ -17,16 +17,19 @@
//! `Default` impls for the server-ng config surface.
//!
-//! Sections that fork (`tcp`, `websocket`, `quic`, `message_bus`) have
-//! their own `Default` impls here, sourced from
+//! Sections that fork (`tcp`, `websocket`, `quic`, `cluster`,
+//! `message_bus`) have their own `Default` impls here, sourced from
//! `core/server-ng/config.toml` via [`SERVER_NG_CONFIG`]. Sections that
-//! still reuse legacy types (`http`, `cluster`, `system`, `telemetry`,
+//! still reuse legacy types (`http`, `system`, `telemetry`,
//! `consumer_group`, `data_maintenance`, `message_saver`,
//! `personal_access_token`, `heartbeat`) delegate to the legacy
//! `Default` impls; overrides land at the consumer level once
//! [`super::server_ng::ServerNgConfig::load`] is wired into server-ng's
//! bootstrap.
+use super::cluster::{
+ ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig,
TransportPorts,
+};
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
@@ -34,7 +37,6 @@ use super::server_ng::NgSystemConfig;
use super::server_ng::{ExtraConfig, ServerNgConfig};
use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig};
use super::websocket::{WebSocketConfig, WebSocketTlsConfig};
-use crate::server_config::cluster::ClusterConfig;
use crate::server_config::http::HttpConfig;
use crate::server_config::server::{
ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig,
MessageSaverConfig,
@@ -69,6 +71,53 @@ impl Default for ServerNgConfig {
}
}
+impl Default for ClusterConfig {
+ fn default() -> ClusterConfig {
+ ClusterConfig {
+ enabled: SERVER_NG_CONFIG.cluster.enabled,
+ name: SERVER_NG_CONFIG.cluster.name.parse().unwrap(),
+ heartbeat_timeout:
SERVER_NG_CONFIG.cluster.heartbeat_timeout.parse().unwrap(),
+ nodes: SERVER_NG_CONFIG
+ .cluster
+ .nodes
+ .iter()
+ .map(|node| ClusterNodeConfig {
+ name: node.name.parse().unwrap(),
+ ip: node.ip.parse().unwrap(),
+ replica_id: u8::try_from(node.replica_id).expect(
+ "static_toml replica_id must fit in u8 (0..=255); \
+ fix core/server-ng/config.toml",
+ ),
+ ports: TransportPorts {
+ tcp: Some(u16::try_from(node.ports.tcp).expect(
+ "static_toml cluster.nodes.ports.tcp must fit in
u16 (0..=65535); \
+ fix core/server-ng/config.toml",
+ )),
+ quic: Some(u16::try_from(node.ports.quic).expect(
+ "static_toml cluster.nodes.ports.quic must fit in
u16 (0..=65535); \
+ fix core/server-ng/config.toml",
+ )),
+ http: Some(u16::try_from(node.ports.http).expect(
+ "static_toml cluster.nodes.ports.http must fit in
u16 (0..=65535); \
+ fix core/server-ng/config.toml",
+ )),
+ websocket:
Some(u16::try_from(node.ports.websocket).expect(
+ "static_toml cluster.nodes.ports.websocket must
fit in u16 (0..=65535); \
+ fix core/server-ng/config.toml",
+ )),
+ tcp_replica:
Some(u16::try_from(node.ports.tcp_replica).expect(
+ "static_toml cluster.nodes.ports.tcp_replica must
fit in u16 (0..=65535); \
+ fix core/server-ng/config.toml",
+ )),
+ },
+ })
+ .collect(),
+ auth: ClusterAuthConfig::default(),
+ tls: ClusterTlsConfig::default(),
+ }
+ }
+}
+
impl Default for MetadataConfig {
fn default() -> MetadataConfig {
// Read from the embedded TOML so the Default impl and the on-disk
diff --git a/core/configs/src/server_ng_config/mod.rs
b/core/configs/src/server_ng_config/mod.rs
index 16de8fdea..159d0c2f5 100644
--- a/core/configs/src/server_ng_config/mod.rs
+++ b/core/configs/src/server_ng_config/mod.rs
@@ -27,6 +27,7 @@
//! wiring PR for `core/server-ng`'s bootstrap and the message_bus crate's
//! runtime type is a separate change.
+pub mod cluster;
pub mod defaults;
pub mod displays;
pub mod message_bus;
diff --git a/core/configs/src/server_ng_config/server_ng.rs
b/core/configs/src/server_ng_config/server_ng.rs
index e60bcc894..3cd8f5376 100644
--- a/core/configs/src/server_ng_config/server_ng.rs
+++ b/core/configs/src/server_ng_config/server_ng.rs
@@ -16,13 +16,13 @@
// under the License.
use super::COMPONENT_NG;
+use super::cluster::ClusterConfig;
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
use super::quic::QuicConfig;
use super::tcp::TcpConfig;
use super::websocket::WebSocketConfig;
use crate::ConfigurationError;
-use crate::server_config::cluster::ClusterConfig;
use crate::server_config::http::HttpConfig;
use crate::server_config::server::{
ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig,
MessageSaverConfig,
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 250b06d8b..8869f870f 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -926,6 +926,14 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
}
}
+ /// Override the normal-heartbeat (primary liveness) window, in consensus
+ /// ticks. Sized from `[cluster] heartbeat_timeout` by the runtime. Must
+ /// run before `init` / `init_as_backup`: the override discards any
+ /// countdown already in flight.
+ pub fn set_normal_heartbeat_ticks(&self, ticks: u64) {
+ self.timeouts.borrow_mut().set_normal_heartbeat_ticks(ticks);
+ }
+
pub fn init(&self) {
self.status.set(Status::Normal);
let mut timeouts = self.timeouts.borrow_mut();
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index 4ae31c5df..9c68f17ac 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -162,3 +162,4 @@ pub use observability::*;
mod view_change_quorum;
pub use view_change_quorum::*;
mod vsr_timeout;
+pub use vsr_timeout::TimeoutManager;
diff --git a/core/consensus/src/vsr_timeout.rs
b/core/consensus/src/vsr_timeout.rs
index eb7076a11..50e6482c8 100644
--- a/core/consensus/src/vsr_timeout.rs
+++ b/core/consensus/src/vsr_timeout.rs
@@ -133,13 +133,15 @@ impl TimeoutManager {
const PING_TICKS: u64 = 100;
const PREPARE_TICKS: u64 = 25;
const COMMIT_MESSAGE_TICKS: u64 = 50;
- const NORMAL_HEARTBEAT_TICKS: u64 = 500;
+ /// Public so the runtime can pin its config default (`[cluster]
+ /// heartbeat_timeout`) against this built-in with a static assert.
+ pub const NORMAL_HEARTBEAT_TICKS: u64 = 500;
const START_VIEW_CHANGE_MESSAGE_TICKS: u64 = 50;
const VIEW_CHANGE_STATUS_TICKS: u64 = 500;
const DO_VIEW_CHANGE_MESSAGE_TICKS: u64 = 50;
const REQUEST_START_VIEW_MESSAGE_TICKS: u64 = 100;
- // TODO: add #[must_use] -- constructor, discarding is always a bug.
+ #[must_use]
pub fn new(replica_id: u128) -> Self {
Self {
ping: Timeout::new(replica_id, Self::PING_TICKS),
@@ -162,6 +164,14 @@ impl TimeoutManager {
}
}
+ /// Override the backup's primary-liveness window (`[cluster]
+ /// heartbeat_timeout`). Replaces the timeout object, so any countdown
+ /// already in flight is discarded: call before the replica starts
+ /// ticking (i.e. before `init` / `init_as_backup`).
+ pub const fn set_normal_heartbeat_ticks(&mut self, ticks: u64) {
+ self.normal_heartbeat = Timeout::new(self.normal_heartbeat.id, ticks);
+ }
+
/// Tick all timeouts
/// This is the first phase of the two-phase tick-based timeout mechanism.
/// 2nd phase is checking which timeouts have fired and calling the
appropriate handlers.
@@ -181,6 +191,7 @@ impl TimeoutManager {
self.get(kind).fired()
}
+ #[must_use]
pub const fn get(&self, kind: TimeoutKind) -> &Timeout {
match kind {
TimeoutKind::Ping => &self.ping,
diff --git a/core/integration/tests/server/message_retrieval.rs
b/core/integration/tests/server/message_retrieval.rs
index 100b34e89..ad417cb0d 100644
--- a/core/integration/tests/server/message_retrieval.rs
+++ b/core/integration/tests/server/message_retrieval.rs
@@ -84,6 +84,19 @@ fn build_server_config(
"true".to_string(),
);
extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(),
"true".to_string());
+ // Under vsr these scenarios exercise retrieval, not failover, yet run
+ // on the default 3-node cluster. Under a parallel nextest run the box
+ // is oversubscribed and scheduling stalls exceed the 5s default
+ // liveness window, so backups elect a new primary mid-scenario and the
+ // client session dies with it (observed stalls reach ~20s; 60s rides
+ // them out). The knob exists only on server-ng: the legacy flavor's
+ // strict env provider aborts boot on unknown IGGY_ vars, so the gate
+ // is load-bearing.
+ #[cfg(feature = "vsr")]
+ extra_envs.insert(
+ "IGGY_CLUSTER_HEARTBEAT_TIMEOUT".to_string(),
+ "60s".to_string(),
+ );
TestServerConfig::builder().extra_envs(extra_envs).build()
}
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 6ed7195a1..832e6b33e 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -577,6 +577,13 @@ enabled = false
# This prevents accidental cross-cluster communication.
name = "iggy-cluster"
+# Backup-side liveness window for a consensus plane's primary (duration).
+# A replica that sees no primary traffic for this long starts a view change.
+# Raise it on oversubscribed hosts where scheduling stalls fake primary
+# death. Must be at least "2s": the primary heartbeats every second, and a
+# window at or below one ping interval would elect on every hiccup.
+heartbeat_timeout = "5s"
+
# Replica-to-replica authentication (PSK + BLAKE3 keyed-MAC handshake).
[cluster.auth]
# When true, every replica peer must complete the authenticated handshake or be
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 7d815d29e..b7d578a3f 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -957,6 +957,7 @@ async fn shard_main(
topology.replica_count,
Rc::clone(&bus),
config.metadata.prepare_queue_depth,
+ cluster_heartbeat_ticks(config),
);
(Some(consensus), Some(journal), snapshot)
} else {
@@ -1490,7 +1491,7 @@ fn build_cluster_roster(
name: config.cluster.name.clone(),
nodes: config.cluster.nodes.clone(),
self_ip: topology.client_listen_addr.ip().to_string(),
- self_ports: configs::cluster::TransportPorts {
+ self_ports: configs::ng_cluster::TransportPorts {
tcp: Some(topology.client_listen_addr.port()),
quic: topology.quic_listen_addr.map(|addr| addr.port()),
http: http_port,
@@ -1689,6 +1690,14 @@ const _: () = assert!(
configs::ng_metadata::DEFAULT_METADATA_JOURNAL_SLOTS
== journal::prepare_journal::DEFAULT_SLOT_COUNT
);
+/// `[cluster] heartbeat_timeout` in consensus ticks, floored at one tick.
+/// Every consensus group (metadata and per-partition planes alike) gets the
+/// same window: the failure it guards against - a primary that stopped
+/// heartbeating - is host-level, not per-plane.
+pub(crate) fn cluster_heartbeat_ticks(config: &ServerNgConfig) -> u64 {
+ let window = config.cluster.heartbeat_timeout.get_duration().as_millis();
+ u64::try_from((window /
shard::CONSENSUS_TICK_INTERVAL.as_millis()).max(1)).unwrap_or(u64::MAX)
+}
#[allow(clippy::too_many_arguments)]
fn restore_metadata_consensus(
@@ -1700,6 +1709,7 @@ fn restore_metadata_consensus(
replica_count: u8,
bus: Rc<IggyMessageBus>,
prepare_queue_depth: usize,
+ normal_heartbeat_ticks: u64,
) -> VsrConsensus<Rc<IggyMessageBus>> {
let mut consensus = VsrConsensus::new(
cluster_id,
@@ -1712,6 +1722,7 @@ fn restore_metadata_consensus(
// in-flight prepares and drain as prepares commit.
LocalPipeline::with_capacities(prepare_queue_depth,
prepare_queue_depth * 2),
);
+ consensus.set_normal_heartbeat_ticks(normal_heartbeat_ticks);
let last_header = journal
.last_op()
@@ -1835,6 +1846,7 @@ async fn load_partition(
bus,
LocalPipeline::new(),
);
+ consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config));
// A recovered partition lost its consensus state with the process: the
// partition journal is in-memory and segments carry no op numbers, so
// this replica cannot know the group's (op, commit). In a cluster it
@@ -2101,7 +2113,7 @@ fn resolve_optional_listener_addr(
}
fn resolve_cluster_client_addrs(
- self_node: &configs::cluster::ClusterNodeConfig,
+ self_node: &configs::ng_cluster::ClusterNodeConfig,
default_client_addr: SocketAddr,
default_ws_addr: Option<SocketAddr>,
default_quic_addr: Option<SocketAddr>,
@@ -2128,10 +2140,10 @@ fn resolve_cluster_client_addrs(
}
fn resolve_cluster_optional_addr(
- self_node: &configs::cluster::ClusterNodeConfig,
+ self_node: &configs::ng_cluster::ClusterNodeConfig,
context: &'static str,
default_addr: Option<SocketAddr>,
- port_selector: impl Fn(&configs::cluster::TransportPorts) -> Option<u16>,
+ port_selector: impl Fn(&configs::ng_cluster::TransportPorts) ->
Option<u16>,
) -> Result<Option<SocketAddr>, ServerNgError> {
let Some(default_addr) = default_addr else {
return Ok(None);
@@ -2141,7 +2153,7 @@ fn resolve_cluster_optional_addr(
}
fn resolve_cluster_replica_peers(
- nodes: &[configs::cluster::ClusterNodeConfig],
+ nodes: &[configs::ng_cluster::ClusterNodeConfig],
self_replica_id: u8,
) -> Result<Vec<(u8, SocketAddr)>, ServerNgError> {
let mut peers = Vec::with_capacity(nodes.len().saturating_sub(1));
@@ -2198,7 +2210,7 @@ async fn start_tcp_runtime(
// caller of this function.
if config.http.enabled {
let http_addr = parse_socket_addr("http.address",
&config.http.address)?;
- let self_ports = configs::cluster::TransportPorts {
+ let self_ports = configs::ng_cluster::TransportPorts {
tcp: config
.tcp
.enabled
@@ -3053,6 +3065,24 @@ const fn operation_triggers_partition_reconcile(op:
Operation) -> bool {
mod tests {
use super::*;
+ #[test]
+ fn default_cluster_heartbeat_timeout_matches_consensus_constant() {
+ // The config default lives in core/server-ng/config.toml (a string,
+ // so no static assert can pin it); keep it in lockstep with the
+ // built-in the simulator and un-configured replicas run on.
+ let config_default = configs::ng_cluster::ClusterConfig::default()
+ .heartbeat_timeout
+ .get_duration()
+ .as_millis();
+ let built_in =
u128::from(consensus::TimeoutManager::NORMAL_HEARTBEAT_TICKS)
+ * shard::CONSENSUS_TICK_INTERVAL.as_millis();
+ assert_eq!(
+ config_default, built_in,
+ "[cluster] heartbeat_timeout default drifted from \
+ TimeoutManager::NORMAL_HEARTBEAT_TICKS"
+ );
+ }
+
#[test]
fn shutdown_on_drop_armed_flips_flag() {
let flag = Arc::new(AtomicBool::new(false));
diff --git a/core/server-ng/src/cluster_meta.rs
b/core/server-ng/src/cluster_meta.rs
index 1be045fec..9e969f3df 100644
--- a/core/server-ng/src/cluster_meta.rs
+++ b/core/server-ng/src/cluster_meta.rs
@@ -28,7 +28,7 @@
//! leader, but the full roster is still returned). The self-synthesized single
//! node is the cluster-disabled fallback, shared by both callers.
-use configs::cluster::{ClusterNodeConfig, TransportPorts};
+use configs::ng_cluster::{ClusterNodeConfig, TransportPorts};
use iggy_common::{
ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus,
TransportEndpoints,
};
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 26aab3afc..37a5535b4 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -50,8 +50,8 @@ use axum::extract::{DefaultBodyLimit, Request};
use axum::http::{HeaderName, HeaderValue, Method};
use axum::middleware::{Next, from_fn};
use axum::routing::{delete, get, post, put};
-use configs::cluster::{ClusterConfig, TransportPorts};
use configs::http::{HttpConfig, HttpCorsConfig};
+use configs::ng_cluster::{ClusterConfig, TransportPorts};
use configs::server_ng::NgSystemConfig;
use iggy_common::IggyError;
use message_bus::client_listener;
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index a001d16fa..63fbe935a 100644
--- a/core/server-ng/src/http/error.rs
+++ b/core/server-ng/src/http/error.rs
@@ -491,7 +491,7 @@ pub(in crate::http) fn primary_redirect_location(
mod tests {
use super::*;
- use configs::cluster::{ClusterNodeConfig, TransportPorts};
+ use configs::ng_cluster::{ClusterNodeConfig, TransportPorts};
const READ_PATH: &str = "/streams?consistency=linearizable";
fn node(replica_id: u8, ip: &str, http: Option<u16>) -> ClusterNodeConfig {
diff --git a/core/server-ng/src/partition_helpers.rs
b/core/server-ng/src/partition_helpers.rs
index 923330e76..e7c33ecd9 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -495,6 +495,7 @@ pub async fn build_partition_fresh(
bus,
LocalPipeline::new(),
);
+
consensus.set_normal_heartbeat_ticks(crate::bootstrap::cluster_heartbeat_ticks(config));
// A partition directory that already holds segment bytes is a RESTART
// materialization, not a fresh create: this replica's group state died
// with the process, so claiming view-0 primaryship would heartbeat
