This is an automated email from the ASF dual-hosted git repository.

hubcio pushed a commit to branch feat/iggy-server-hardening
in repository https://gitbox.apache.org/repos/asf/iggy.git

commit ca21c2bc54476a094a6484f10b3c50659632173f
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Fri Jul 24 11:48:04 2026 +0200

    feat(server-ng): add commit broadcast and prepare retransmit intervals
    
    The primary's CommitMessage broadcast (500ms) and the backup
    prepare retransmit (250ms) were fixed tick counts in consensus,
    untunable without a rebuild. Both interact with the
    cluster.heartbeat_timeout knob shipped earlier: the broadcast is
    the liveness feed heartbeat_timeout watches, so mismatched values
    can false-trigger view changes.
    
    Surface both as [cluster] durations following the
    heartbeat_timeout pattern: defaults in the embedded config.toml,
    converted to consensus ticks at bootstrap, applied before init on
    all three consensus construction paths. Validation rejects zero
    and enforces heartbeat_timeout >= 4x commit_broadcast_interval so
    a liveness window always spans several broadcasts. Lockstep tests
    pin the TOML defaults to the consensus constants.
    
    Also delete the dead ping timeout mechanism: it never fed any
    liveness decision since CommitMessage broadcast took that role,
    and the surviving code misled readers into thinking ping was the
    heartbeat.
---
 core/configs/src/server_ng_config/cluster.rs  | 132 +++++++++++++++++++++++++-
 core/configs/src/server_ng_config/defaults.rs |  10 ++
 core/consensus/src/impls.rs                   |  16 ++++
 core/consensus/src/vsr_timeout.rs             |  32 +++++--
 core/server-ng/config.toml                    |  17 +++-
 core/server-ng/src/bootstrap.rs               |  76 +++++++++++++--
 core/server-ng/src/partition_helpers.rs       |   2 +
 7 files changed, 264 insertions(+), 21 deletions(-)

diff --git a/core/configs/src/server_ng_config/cluster.rs 
b/core/configs/src/server_ng_config/cluster.rs
index b8dc2d7f4..5ca8669c3 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -30,10 +30,19 @@ 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.
+/// Absolute floor for the backup liveness window, independent of the
+/// commit-broadcast rate. The primary signals liveness through its commit
+/// broadcast (`commit_broadcast_interval`, 500ms by default); 2s spans several
+/// broadcasts, so a single delayed one never elects. The per-config
+/// `MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO` check scales the same headroom
+/// when the broadcast interval is retuned.
 pub const MIN_CLUSTER_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(2);
 
+/// The backup liveness window (`heartbeat_timeout`) must span at least this
+/// many commit broadcasts (`commit_broadcast_interval`), so one dropped or
+/// delayed broadcast never trips a view change on a healthy primary.
+const MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO: u32 = 4;
+
 /// 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.
@@ -45,6 +54,26 @@ fn default_heartbeat_timeout() -> IggyDuration {
     SERVER_NG_CONFIG.cluster.heartbeat_timeout.parse().unwrap()
 }
 
+/// 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_commit_broadcast_interval() -> IggyDuration {
+    SERVER_NG_CONFIG
+        .cluster
+        .commit_broadcast_interval
+        .parse()
+        .unwrap()
+}
+
+/// 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_prepare_retransmit_interval() -> IggyDuration {
+    SERVER_NG_CONFIG
+        .cluster
+        .prepare_retransmit_interval
+        .parse()
+        .unwrap()
+}
+
 #[serde_as]
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 #[serde(deny_unknown_fields)]
@@ -61,6 +90,26 @@ pub struct ClusterConfig {
     #[serde_as(as = "DisplayFromStr")]
     #[config_env(leaf)]
     pub heartbeat_timeout: IggyDuration,
+    /// How often the primary broadcasts its commit point to every backup, the
+    /// cluster's primary-liveness signal. Each broadcast resets the backups'
+    /// `heartbeat_timeout` window, so that window must span several 
broadcasts:
+    /// boot rejects `heartbeat_timeout < 
MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO
+    /// * commit_broadcast_interval`. Sizes the consensus `CommitMessage` 
timer.
+    /// Zero (and the `0` / `disabled` / `unlimited` sentinels, which all parse
+    /// to zero) is rejected at boot.
+    #[serde(default = "default_commit_broadcast_interval")]
+    #[serde_as(as = "DisplayFromStr")]
+    #[config_env(leaf)]
+    pub commit_broadcast_interval: IggyDuration,
+    /// How often the primary retransmits prepares a backup has not yet acked.
+    /// Lower recovers faster from a dropped prepare at the cost of replica
+    /// traffic. Sizes the consensus `Prepare` timer. Zero (and the `0` /
+    /// `disabled` / `unlimited` sentinels, which all parse to zero) is 
rejected
+    /// at boot.
+    #[serde(default = "default_prepare_retransmit_interval")]
+    #[serde_as(as = "DisplayFromStr")]
+    #[config_env(leaf)]
+    pub prepare_retransmit_interval: 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
@@ -180,13 +229,51 @@ impl Validatable<ConfigurationError> for ClusterConfig {
         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)",
+                 (the primary signals liveness through its commit broadcast; a 
shorter window \
+                 elects on every scheduling hiccup)",
                 self.heartbeat_timeout,
                 MIN_CLUSTER_HEARTBEAT_TIMEOUT.as_secs()
             );
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
 
+        // The commit broadcast is the cluster's liveness feed and the prepare
+        // retransmit its recovery timer; both size consensus timers that have
+        // to advance. `0` / `disabled` / `unlimited` all collapse to a zero
+        // duration, which would stall the timer - reject them.
+        if self.commit_broadcast_interval.get_duration().is_zero() {
+            eprintln!(
+                "Invalid cluster configuration: 
cluster.commit_broadcast_interval must be nonzero \
+                 (it drives the primary's liveness broadcast)"
+            );
+            return Err(ConfigurationError::InvalidConfigurationValue);
+        }
+        if self.prepare_retransmit_interval.get_duration().is_zero() {
+            eprintln!(
+                "Invalid cluster configuration: 
cluster.prepare_retransmit_interval must be \
+                 nonzero (it drives prepare retransmission)"
+            );
+            return Err(ConfigurationError::InvalidConfigurationValue);
+        }
+
+        // The liveness window must span several commit broadcasts so a single
+        // delayed broadcast never trips a view change on a healthy primary.
+        let min_heartbeat = self
+            .commit_broadcast_interval
+            .get_duration()
+            .saturating_mul(MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO);
+        if self.heartbeat_timeout.get_duration() < min_heartbeat {
+            eprintln!(
+                "Invalid cluster configuration: cluster.heartbeat_timeout '{}' 
must be at least \
+                 {}x cluster.commit_broadcast_interval '{}' so the liveness 
window spans several \
+                 broadcasts",
+                self.heartbeat_timeout,
+                MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO,
+                self.commit_broadcast_interval
+            );
+            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"
@@ -344,6 +431,8 @@ mod tests {
             enabled: true,
             name: "iggy-cluster".to_owned(),
             heartbeat_timeout: default_heartbeat_timeout(),
+            commit_broadcast_interval: default_commit_broadcast_interval(),
+            prepare_retransmit_interval: default_prepare_retransmit_interval(),
             nodes: Vec::new(),
             auth: ClusterAuthConfig {
                 enabled: true,
@@ -381,6 +470,8 @@ mod cluster_validate_tests {
             enabled: true,
             name: "iggy-cluster".to_string(),
             heartbeat_timeout: default_heartbeat_timeout(),
+            commit_broadcast_interval: default_commit_broadcast_interval(),
+            prepare_retransmit_interval: default_prepare_retransmit_interval(),
             nodes,
             auth: ClusterAuthConfig::default(),
             tls: ClusterTlsConfig::default(),
@@ -398,6 +489,41 @@ mod cluster_validate_tests {
         assert!(c.validate().is_err());
     }
 
+    #[test]
+    fn validate_rejects_zero_commit_broadcast_interval() {
+        // `0` / `disabled` / `unlimited` all collapse to zero and stall the
+        // liveness broadcast.
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.commit_broadcast_interval = IggyDuration::new(Duration::ZERO);
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_zero_prepare_retransmit_interval() {
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.prepare_retransmit_interval = IggyDuration::new(Duration::ZERO);
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_heartbeat_below_commit_broadcast_ratio() {
+        // 3s clears the absolute 2s floor but is still < 4x the 1s broadcast,
+        // so the ratio rule is what rejects here, not the floor.
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.heartbeat_timeout = IggyDuration::new(Duration::from_secs(3));
+        c.commit_broadcast_interval = 
IggyDuration::new(Duration::from_secs(1));
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn validate_accepts_heartbeat_at_commit_broadcast_ratio() {
+        // Exactly 4x the broadcast (and above the 2s floor) must pass.
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.heartbeat_timeout = IggyDuration::new(Duration::from_secs(4));
+        c.commit_broadcast_interval = 
IggyDuration::new(Duration::from_secs(1));
+        assert!(c.validate().is_ok());
+    }
+
     #[test]
     fn validate_rejects_empty_nodes() {
         let c = cfg(vec![]);
diff --git a/core/configs/src/server_ng_config/defaults.rs 
b/core/configs/src/server_ng_config/defaults.rs
index 58887a70f..40e47d68b 100644
--- a/core/configs/src/server_ng_config/defaults.rs
+++ b/core/configs/src/server_ng_config/defaults.rs
@@ -77,6 +77,16 @@ impl Default for 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(),
+            commit_broadcast_interval: SERVER_NG_CONFIG
+                .cluster
+                .commit_broadcast_interval
+                .parse()
+                .unwrap(),
+            prepare_retransmit_interval: SERVER_NG_CONFIG
+                .cluster
+                .prepare_retransmit_interval
+                .parse()
+                .unwrap(),
             nodes: SERVER_NG_CONFIG
                 .cluster
                 .nodes
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index b041b0df8..d81c308f7 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -946,6 +946,22 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> 
VsrConsensus<B, P> {
         self.timeouts.borrow_mut().set_normal_heartbeat_ticks(ticks);
     }
 
+    /// Override the primary's commit-broadcast interval, in consensus ticks.
+    /// Sized from `[cluster] commit_broadcast_interval` by the runtime. Must
+    /// run before `init` / `init_as_backup`: the override discards any
+    /// countdown already in flight.
+    pub fn set_commit_message_ticks(&self, ticks: u64) {
+        self.timeouts.borrow_mut().set_commit_message_ticks(ticks);
+    }
+
+    /// Override the primary's prepare-retransmit interval, in consensus ticks.
+    /// Sized from `[cluster] prepare_retransmit_interval` by the runtime. Must
+    /// run before `init` / `init_as_backup`: the override discards any
+    /// countdown already in flight.
+    pub fn set_prepare_ticks(&self, ticks: u64) {
+        self.timeouts.borrow_mut().set_prepare_ticks(ticks);
+    }
+
     pub fn init(&self) {
         self.status.set(Status::Normal);
         let mut timeouts = self.timeouts.borrow_mut();
diff --git a/core/consensus/src/vsr_timeout.rs 
b/core/consensus/src/vsr_timeout.rs
index 50e6482c8..958644836 100644
--- a/core/consensus/src/vsr_timeout.rs
+++ b/core/consensus/src/vsr_timeout.rs
@@ -99,7 +99,6 @@ impl Timeout {
 #[allow(unused)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum TimeoutKind {
-    Ping,
     Prepare,
     CommitMessage,
     NormalHeartbeat,
@@ -115,7 +114,6 @@ pub enum TimeoutKind {
 #[allow(unused)]
 #[derive(Debug)]
 pub struct TimeoutManager {
-    ping: Timeout,
     prepare: Timeout,
     commit_message: Timeout,
     normal_heartbeat: Timeout,
@@ -130,9 +128,12 @@ pub struct TimeoutManager {
 impl TimeoutManager {
     // Timeout durations in ticks (10ms per tick).
     // TODO define 10ms per tick in a separate constant.
-    const PING_TICKS: u64 = 100;
-    const PREPARE_TICKS: u64 = 25;
-    const COMMIT_MESSAGE_TICKS: u64 = 50;
+    /// Public so the runtime can pin its `[cluster] 
prepare_retransmit_interval`
+    /// config default against this built-in.
+    pub const PREPARE_TICKS: u64 = 25;
+    /// Public so the runtime can pin its `[cluster] commit_broadcast_interval`
+    /// config default against this built-in.
+    pub const COMMIT_MESSAGE_TICKS: u64 = 50;
     /// 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;
@@ -144,7 +145,6 @@ impl TimeoutManager {
     #[must_use]
     pub fn new(replica_id: u128) -> Self {
         Self {
-            ping: Timeout::new(replica_id, Self::PING_TICKS),
             prepare: Timeout::new(replica_id, Self::PREPARE_TICKS),
             commit_message: Timeout::new(replica_id, 
Self::COMMIT_MESSAGE_TICKS),
             normal_heartbeat: Timeout::new(replica_id, 
Self::NORMAL_HEARTBEAT_TICKS),
@@ -172,11 +172,26 @@ impl TimeoutManager {
         self.normal_heartbeat = Timeout::new(self.normal_heartbeat.id, ticks);
     }
 
+    /// Override the primary's commit-broadcast interval (`[cluster]
+    /// commit_broadcast_interval`). Replaces the timeout object, so any
+    /// countdown already in flight is discarded: call before the replica
+    /// starts ticking (i.e. before `init`).
+    pub const fn set_commit_message_ticks(&mut self, ticks: u64) {
+        self.commit_message = Timeout::new(self.commit_message.id, ticks);
+    }
+
+    /// Override the primary's prepare-retransmit interval (`[cluster]
+    /// prepare_retransmit_interval`). Replaces the timeout object, so any
+    /// countdown already in flight is discarded: call before the replica
+    /// starts ticking (i.e. before `init`).
+    pub const fn set_prepare_ticks(&mut self, ticks: u64) {
+        self.prepare = Timeout::new(self.prepare.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.
     pub const fn tick(&mut self) {
-        self.ping.tick();
         self.prepare.tick();
         self.commit_message.tick();
         self.normal_heartbeat.tick();
@@ -194,7 +209,6 @@ impl TimeoutManager {
     #[must_use]
     pub const fn get(&self, kind: TimeoutKind) -> &Timeout {
         match kind {
-            TimeoutKind::Ping => &self.ping,
             TimeoutKind::Prepare => &self.prepare,
             TimeoutKind::CommitMessage => &self.commit_message,
             TimeoutKind::NormalHeartbeat => &self.normal_heartbeat,
@@ -207,7 +221,6 @@ impl TimeoutManager {
 
     pub const fn get_mut(&mut self, kind: TimeoutKind) -> &mut Timeout {
         match kind {
-            TimeoutKind::Ping => &mut self.ping,
             TimeoutKind::Prepare => &mut self.prepare,
             TimeoutKind::CommitMessage => &mut self.commit_message,
             TimeoutKind::NormalHeartbeat => &mut self.normal_heartbeat,
@@ -232,7 +245,6 @@ impl TimeoutManager {
 
     pub fn backoff(&mut self, kind: TimeoutKind) {
         let timeout = match kind {
-            TimeoutKind::Ping => &mut self.ping,
             TimeoutKind::Prepare => &mut self.prepare,
             TimeoutKind::CommitMessage => &mut self.commit_message,
             TimeoutKind::NormalHeartbeat => &mut self.normal_heartbeat,
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 832e6b33e..08fea06d4 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -580,10 +580,23 @@ 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.
+# death. Must be at least "2s" and at least 4x commit_broadcast_interval: the
+# primary signals liveness through its commit broadcast, and the window must
+# span several broadcasts so one delayed broadcast never trips an election.
 heartbeat_timeout = "5s"
 
+# How often the primary broadcasts its commit point to every backup (duration).
+# This is the cluster's liveness signal: each broadcast resets every backup's
+# heartbeat_timeout window and carries the latest commit point forward. Must be
+# nonzero and, with heartbeat_timeout, satisfy heartbeat_timeout >= 4x this
+# value. Drives the consensus CommitMessage timer.
+commit_broadcast_interval = "500ms"
+
+# How often the primary retransmits prepares that backups have not yet acked
+# (duration). Lower values recover faster from a dropped prepare at the cost of
+# more replica traffic; must be nonzero. Drives the consensus Prepare timer.
+prepare_retransmit_interval = "250ms"
+
 # 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 b7d578a3f..3212c944d 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -958,6 +958,8 @@ async fn shard_main(
                 Rc::clone(&bus),
                 config.metadata.prepare_queue_depth,
                 cluster_heartbeat_ticks(config),
+                commit_broadcast_ticks(config),
+                prepare_retransmit_ticks(config),
             );
             (Some(consensus), Some(journal), snapshot)
         } else {
@@ -1690,13 +1692,33 @@ 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.
+/// Convert a consensus-timer interval to whole ticks, floored at one tick so a
+/// sub-tick value still fires and saturated on overflow.
+fn duration_to_ticks(interval: Duration) -> u64 {
+    let ticks = interval.as_millis() / 
shard::CONSENSUS_TICK_INTERVAL.as_millis();
+    u64::try_from(ticks.max(1)).unwrap_or(u64::MAX)
+}
+
+/// `[cluster] heartbeat_timeout` in consensus ticks. 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)
+    duration_to_ticks(config.cluster.heartbeat_timeout.get_duration())
+}
+
+/// `[cluster] commit_broadcast_interval` in consensus ticks: how often the
+/// primary broadcasts its commit point, the cluster's liveness feed. Applied
+/// to every consensus group, matching `cluster_heartbeat_ticks`.
+pub(crate) fn commit_broadcast_ticks(config: &ServerNgConfig) -> u64 {
+    duration_to_ticks(config.cluster.commit_broadcast_interval.get_duration())
+}
+
+/// `[cluster] prepare_retransmit_interval` in consensus ticks: how often the
+/// primary retransmits un-acked prepares. Applied to every consensus group,
+/// matching `cluster_heartbeat_ticks`.
+pub(crate) fn prepare_retransmit_ticks(config: &ServerNgConfig) -> u64 {
+    
duration_to_ticks(config.cluster.prepare_retransmit_interval.get_duration())
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -1710,6 +1732,8 @@ fn restore_metadata_consensus(
     bus: Rc<IggyMessageBus>,
     prepare_queue_depth: usize,
     normal_heartbeat_ticks: u64,
+    commit_message_ticks: u64,
+    prepare_ticks: u64,
 ) -> VsrConsensus<Rc<IggyMessageBus>> {
     let mut consensus = VsrConsensus::new(
         cluster_id,
@@ -1723,6 +1747,8 @@ fn restore_metadata_consensus(
         LocalPipeline::with_capacities(prepare_queue_depth, 
prepare_queue_depth * 2),
     );
     consensus.set_normal_heartbeat_ticks(normal_heartbeat_ticks);
+    consensus.set_commit_message_ticks(commit_message_ticks);
+    consensus.set_prepare_ticks(prepare_ticks);
 
     let last_header = journal
         .last_op()
@@ -1847,6 +1873,8 @@ async fn load_partition(
         LocalPipeline::new(),
     );
     consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config));
+    consensus.set_commit_message_ticks(commit_broadcast_ticks(config));
+    consensus.set_prepare_ticks(prepare_retransmit_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
@@ -3083,6 +3111,42 @@ mod tests {
         );
     }
 
+    #[test]
+    fn default_commit_broadcast_interval_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()
+            .commit_broadcast_interval
+            .get_duration()
+            .as_millis();
+        let built_in = 
u128::from(consensus::TimeoutManager::COMMIT_MESSAGE_TICKS)
+            * shard::CONSENSUS_TICK_INTERVAL.as_millis();
+        assert_eq!(
+            config_default, built_in,
+            "[cluster] commit_broadcast_interval default drifted from \
+             TimeoutManager::COMMIT_MESSAGE_TICKS"
+        );
+    }
+
+    #[test]
+    fn default_prepare_retransmit_interval_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()
+            .prepare_retransmit_interval
+            .get_duration()
+            .as_millis();
+        let built_in = u128::from(consensus::TimeoutManager::PREPARE_TICKS)
+            * shard::CONSENSUS_TICK_INTERVAL.as_millis();
+        assert_eq!(
+            config_default, built_in,
+            "[cluster] prepare_retransmit_interval default drifted from \
+             TimeoutManager::PREPARE_TICKS"
+        );
+    }
+
     #[test]
     fn shutdown_on_drop_armed_flips_flag() {
         let flag = Arc::new(AtomicBool::new(false));
diff --git a/core/server-ng/src/partition_helpers.rs 
b/core/server-ng/src/partition_helpers.rs
index e7c33ecd9..aad42b9d4 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -496,6 +496,8 @@ pub async fn build_partition_fresh(
         LocalPipeline::new(),
     );
     
consensus.set_normal_heartbeat_ticks(crate::bootstrap::cluster_heartbeat_ticks(config));
+    
consensus.set_commit_message_ticks(crate::bootstrap::commit_broadcast_ticks(config));
+    
consensus.set_prepare_ticks(crate::bootstrap::prepare_retransmit_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

Reply via email to