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
