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 90d66a13c60bc035e941a49fae77dd853bdeee17 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 15:55:01 2026 +0200 feat(server-ng): make VSR view-change timing configurable VSR view-change retransmit intervals, the stalled-view-change status backstop, the request-start-view retransmit interval, and the recovering-replica probe-attempt ceiling were hardcoded consensus constants, so tuning view-change and recovery timing for a deployment's network or cluster size meant recompiling. Surface them as four [cluster] knobs following the existing config pattern: defaults live in the embedded config.toml, convert to consensus ticks at bootstrap, and apply to every consensus group. One knob drives both the StartViewChange and DoViewChange retransmit timers, equal by design. Boot rejects the zero sentinels and enforces that the status backstop spans at least 4x the retransmit interval, so a few dropped view-change messages retransmit rather than escalating a progressing view change into a fresh cluster-wide election. The probe ceiling is bounded to 1..=100 as a typo guard. --- core/configs/src/server_ng_config/cluster.rs | 231 ++++++++++++++++++++++++++ core/configs/src/server_ng_config/defaults.rs | 16 ++ core/consensus/src/impls.rs | 50 +++++- core/consensus/src/vsr_timeout.rs | 51 +++++- core/server-ng/config.toml | 23 +++ core/server-ng/src/bootstrap.rs | 130 ++++++++++++++- core/server-ng/src/partition_helpers.rs | 5 + 7 files changed, 499 insertions(+), 7 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 5ca8669c3..4fdb87b91 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -43,6 +43,24 @@ pub const MIN_CLUSTER_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(2); /// delayed broadcast never trips a view change on a healthy primary. const MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO: u32 = 4; +/// The view-change status backstop (`view_change_status_timeout`) must span at +/// least this many retransmit intervals (`view_change_retransmit_interval`), so +/// a few dropped `StartViewChange` / `DoViewChange` messages retransmit rather +/// than escalating a progressing view change into a fresh cluster-wide election. +const MIN_STATUS_TO_RETRANSMIT_RATIO: u32 = 4; + +/// Default recovering-replica probe-attempt ceiling. Duplicated here rather +/// than imported so `core/configs` keeps off a build-time edge onto +/// `core/consensus` (mirroring [`super::partition`]); `core/server-ng`'s +/// bootstrap static-asserts it equal to `consensus::PROBE_ATTEMPTS_MAX`. +pub const DEFAULT_VIEW_PROBE_ATTEMPTS_MAX: u32 = 5; + +/// Upper bound on `view_probe_attempts_max`. A recovering replica probes once +/// per `request_start_view_retransmit_interval`, so hundreds of attempts would +/// stall the election fallback for minutes on a full-cluster restart; this is a +/// typo guard, not a sizing endorsement. +const MAX_VIEW_PROBE_ATTEMPTS: u32 = 100; + /// 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. @@ -74,6 +92,42 @@ fn default_prepare_retransmit_interval() -> IggyDuration { .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_view_change_retransmit_interval() -> IggyDuration { + SERVER_NG_CONFIG + .cluster + .view_change_retransmit_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_view_change_status_timeout() -> IggyDuration { + SERVER_NG_CONFIG + .cluster + .view_change_status_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_request_start_view_retransmit_interval() -> IggyDuration { + SERVER_NG_CONFIG + .cluster + .request_start_view_retransmit_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_view_probe_attempts_max() -> u32 { + SERVER_NG_CONFIG.cluster.view_probe_attempts_max as u32 +} + #[serde_as] #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] #[serde(deny_unknown_fields)] @@ -110,6 +164,41 @@ pub struct ClusterConfig { #[serde_as(as = "DisplayFromStr")] #[config_env(leaf)] pub prepare_retransmit_interval: IggyDuration, + /// How often a plane retransmits its `StartViewChange` / `DoViewChange` + /// while a view change is running. Lower converges a healthy election + /// faster at the cost of replica traffic. Sizes both consensus view-change + /// retransmit timers, which are deliberately equal. Zero (and the `0` / + /// `disabled` / `unlimited` sentinels, which all parse to zero) is rejected + /// at boot. + #[serde(default = "default_view_change_retransmit_interval")] + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub view_change_retransmit_interval: IggyDuration, + /// Backstop for a stalled view change: one that does not conclude within + /// this window escalates to a fresh cluster-wide election. Must span + /// several `view_change_retransmit_interval`s so a few dropped view-change + /// messages retransmit rather than escalate: boot rejects + /// `view_change_status_timeout < MIN_STATUS_TO_RETRANSMIT_RATIO * + /// view_change_retransmit_interval`. Zero (and the `0` / `disabled` / + /// `unlimited` sentinels, which all parse to zero) is rejected at boot. + #[serde(default = "default_view_change_status_timeout")] + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub view_change_status_timeout: IggyDuration, + /// How often a recovering or view-change backup re-requests the current + /// view's `StartView` from its primary (`RequestStartView`). Sizes the + /// consensus `RequestStartView` timer. Zero (and the `0` / `disabled` / + /// `unlimited` sentinels, which all parse to zero) is rejected at boot. + #[serde(default = "default_request_start_view_retransmit_interval")] + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub request_start_view_retransmit_interval: IggyDuration, + /// How many consecutive unanswered `RequestStartView` probes a recovering + /// replica tolerates before it falls back to an election (a full-cluster + /// restart leaves nobody settled to answer). Must be >= 1 and <= + /// `MAX_VIEW_PROBE_ATTEMPTS`. + #[serde(default = "default_view_probe_attempts_max")] + pub view_probe_attempts_max: u32, /// 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 @@ -274,6 +363,75 @@ impl Validatable<ConfigurationError> for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } + // The three view-change timers each size a consensus timer that has to + // advance; `0` / `disabled` / `unlimited` all collapse to zero and + // would stall it - reject them. + if self + .view_change_retransmit_interval + .get_duration() + .is_zero() + { + eprintln!( + "Invalid cluster configuration: cluster.view_change_retransmit_interval must be \ + nonzero (it drives StartViewChange / DoViewChange retransmission)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.view_change_status_timeout.get_duration().is_zero() { + eprintln!( + "Invalid cluster configuration: cluster.view_change_status_timeout must be nonzero \ + (it backstops a stalled view change)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self + .request_start_view_retransmit_interval + .get_duration() + .is_zero() + { + eprintln!( + "Invalid cluster configuration: cluster.request_start_view_retransmit_interval \ + must be nonzero (it drives RequestStartView retransmission)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + + // The status backstop must span several retransmits so a few dropped + // view-change messages retransmit rather than escalating a progressing + // view change into a fresh cluster-wide election. + let min_status = self + .view_change_retransmit_interval + .get_duration() + .saturating_mul(MIN_STATUS_TO_RETRANSMIT_RATIO); + if self.view_change_status_timeout.get_duration() < min_status { + eprintln!( + "Invalid cluster configuration: cluster.view_change_status_timeout '{}' must be at \ + least {}x cluster.view_change_retransmit_interval '{}' so a stalled view change \ + retransmits before it escalates to an election", + self.view_change_status_timeout, + MIN_STATUS_TO_RETRANSMIT_RATIO, + self.view_change_retransmit_interval + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + + // A recovering replica needs at least one probe before it may give up + // and elect; the ceiling is a typo guard (see MAX_VIEW_PROBE_ATTEMPTS). + if self.view_probe_attempts_max == 0 { + eprintln!( + "Invalid cluster configuration: cluster.view_probe_attempts_max must be >= 1" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.view_probe_attempts_max > MAX_VIEW_PROBE_ATTEMPTS { + eprintln!( + "Invalid cluster configuration: cluster.view_probe_attempts_max ({}) exceeds the \ + maximum ({MAX_VIEW_PROBE_ATTEMPTS})", + self.view_probe_attempts_max + ); + 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" @@ -433,6 +591,11 @@ mod tests { heartbeat_timeout: default_heartbeat_timeout(), commit_broadcast_interval: default_commit_broadcast_interval(), prepare_retransmit_interval: default_prepare_retransmit_interval(), + view_change_retransmit_interval: default_view_change_retransmit_interval(), + view_change_status_timeout: default_view_change_status_timeout(), + request_start_view_retransmit_interval: default_request_start_view_retransmit_interval( + ), + view_probe_attempts_max: default_view_probe_attempts_max(), nodes: Vec::new(), auth: ClusterAuthConfig { enabled: true, @@ -472,6 +635,11 @@ mod cluster_validate_tests { heartbeat_timeout: default_heartbeat_timeout(), commit_broadcast_interval: default_commit_broadcast_interval(), prepare_retransmit_interval: default_prepare_retransmit_interval(), + view_change_retransmit_interval: default_view_change_retransmit_interval(), + view_change_status_timeout: default_view_change_status_timeout(), + request_start_view_retransmit_interval: default_request_start_view_retransmit_interval( + ), + view_probe_attempts_max: default_view_probe_attempts_max(), nodes, auth: ClusterAuthConfig::default(), tls: ClusterTlsConfig::default(), @@ -524,6 +692,69 @@ mod cluster_validate_tests { assert!(c.validate().is_ok()); } + #[test] + fn validate_rejects_zero_view_change_retransmit_interval() { + // `0` / `disabled` / `unlimited` all collapse to zero and stall the + // view-change retransmit timers. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_change_retransmit_interval = IggyDuration::new(Duration::ZERO); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_zero_view_change_status_timeout() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_change_status_timeout = IggyDuration::new(Duration::ZERO); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_zero_request_start_view_retransmit_interval() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.request_start_view_retransmit_interval = IggyDuration::new(Duration::ZERO); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_view_change_status_below_retransmit_ratio() { + // 3s is nonzero but still < 4x the 1s retransmit, so the ratio rule is + // what rejects here, not the zero check. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_change_retransmit_interval = IggyDuration::new(Duration::from_secs(1)); + c.view_change_status_timeout = IggyDuration::new(Duration::from_secs(3)); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_accepts_view_change_status_at_retransmit_ratio() { + // Exactly 4x the retransmit interval must pass. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_change_retransmit_interval = IggyDuration::new(Duration::from_secs(1)); + c.view_change_status_timeout = IggyDuration::new(Duration::from_secs(4)); + assert!(c.validate().is_ok()); + } + + #[test] + fn validate_rejects_zero_view_probe_attempts_max() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_probe_attempts_max = 0; + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_view_probe_attempts_above_ceiling() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_probe_attempts_max = MAX_VIEW_PROBE_ATTEMPTS + 1; + assert!(c.validate().is_err()); + } + + #[test] + fn validate_accepts_view_probe_attempts_at_ceiling() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.view_probe_attempts_max = MAX_VIEW_PROBE_ATTEMPTS; + 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 a48e9f8fd..a0a49b3bd 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -91,6 +91,22 @@ impl Default for ClusterConfig { .prepare_retransmit_interval .parse() .unwrap(), + view_change_retransmit_interval: SERVER_NG_CONFIG + .cluster + .view_change_retransmit_interval + .parse() + .unwrap(), + view_change_status_timeout: SERVER_NG_CONFIG + .cluster + .view_change_status_timeout + .parse() + .unwrap(), + request_start_view_retransmit_interval: SERVER_NG_CONFIG + .cluster + .request_start_view_retransmit_interval + .parse() + .unwrap(), + view_probe_attempts_max: SERVER_NG_CONFIG.cluster.view_probe_attempts_max as u32, nodes: SERVER_NG_CONFIG .cluster .nodes diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index a7188e94d..f969cb022 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -830,8 +830,14 @@ where /// Tracks start view change messages received from all replicas (including self) start_view_change_from_all_replicas: RefCell<BitSet<u32>>, /// Consecutive unanswered `RequestStartView` probes while Recovering; - /// at [`PROBE_ATTEMPTS_MAX`] the replica falls back to an election. + /// at the `probe_attempts_max` ceiling the replica falls back to an + /// election. probe_attempts: Cell<u32>, + /// Probe-attempt ceiling backing the fall-back-to-election decision, + /// seeded from [`PROBE_ATTEMPTS_MAX`] and overridable by the runtime via + /// `[cluster] view_probe_attempts_max`. The simulator and tests keep the + /// built-in default. + probe_attempts_max: Cell<u32>, /// Tracks DVC messages received (only used by primary candidate) /// Stores metadata; actual log comes from message @@ -934,6 +940,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { loopback_queue: RefCell::new(VecDeque::with_capacity(prepare_queue_max)), start_view_change_from_all_replicas: RefCell::new(BitSet::with_capacity(REPLICAS_MAX)), probe_attempts: Cell::new(0), + probe_attempts_max: Cell::new(PROBE_ATTEMPTS_MAX), do_view_change_from_all_replicas: RefCell::new(dvc_quorum_array_empty()), do_view_change_quorum: Cell::new(false), sent_own_start_view_change: Cell::new(false), @@ -968,6 +975,45 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.timeouts.borrow_mut().set_prepare_ticks(ticks); } + /// Override the view-change retransmit interval (`StartViewChange` and + /// `DoViewChange`, kept equal), in consensus ticks. Sized from `[cluster] + /// view_change_retransmit_interval` by the runtime. Must run before `init` + /// / `init_as_backup`: the override discards any countdown already in + /// flight. + pub fn set_view_change_retransmit_ticks(&self, ticks: u64) { + self.timeouts + .borrow_mut() + .set_view_change_retransmit_ticks(ticks); + } + + /// Override the view-change status backstop, in consensus ticks. Sized from + /// `[cluster] view_change_status_timeout` by the runtime. Must run before + /// `init` / `init_as_backup`: the override discards any countdown already + /// in flight. + pub fn set_view_change_status_ticks(&self, ticks: u64) { + self.timeouts + .borrow_mut() + .set_view_change_status_ticks(ticks); + } + + /// Override the request-start-view retransmit interval, in consensus ticks. + /// Sized from `[cluster] request_start_view_retransmit_interval` by the + /// runtime. Must run before `init` / `init_as_backup`: the override + /// discards any countdown already in flight. + pub fn set_request_start_view_ticks(&self, ticks: u64) { + self.timeouts + .borrow_mut() + .set_request_start_view_ticks(ticks); + } + + /// Override the recovering-replica probe-attempt ceiling before it falls + /// back to an election. Sized from `[cluster] view_probe_attempts_max` by + /// the runtime; the simulator and tests keep [`PROBE_ATTEMPTS_MAX`]. Must + /// run before `init` / `init_as_backup`. + pub fn set_probe_attempts_max(&self, max: u32) { + self.probe_attempts_max.set(max); + } + pub fn init(&self) { self.status.set(Status::Normal); let mut timeouts = self.timeouts.borrow_mut(); @@ -1481,7 +1527,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // primary answers well before the fallback fires. let attempts = self.probe_attempts.get() + 1; self.probe_attempts.set(attempts); - if attempts >= PROBE_ATTEMPTS_MAX { + if attempts >= self.probe_attempts_max.get() { self.finish_view_probe(); actions.extend( self.start_election(plane, ViewChangeReason::ViewProbeUnanswered), diff --git a/core/consensus/src/vsr_timeout.rs b/core/consensus/src/vsr_timeout.rs index 958644836..aa5d5f6e0 100644 --- a/core/consensus/src/vsr_timeout.rs +++ b/core/consensus/src/vsr_timeout.rs @@ -137,10 +137,23 @@ impl TimeoutManager { /// 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; + /// Public so the runtime can pin its `[cluster] + /// view_change_retransmit_interval` config default against this built-in. + /// Deliberately equal to [`Self::DO_VIEW_CHANGE_MESSAGE_TICKS`]: one config + /// knob drives both retransmit timers. + pub const START_VIEW_CHANGE_MESSAGE_TICKS: u64 = 50; + /// Public so the runtime can pin its `[cluster] view_change_status_timeout` + /// config default against this built-in. + pub const VIEW_CHANGE_STATUS_TICKS: u64 = 500; + /// Public so the runtime can pin its `[cluster] + /// view_change_retransmit_interval` config default against this built-in. + /// Deliberately equal to [`Self::START_VIEW_CHANGE_MESSAGE_TICKS`]: one + /// config knob drives both retransmit timers. + pub const DO_VIEW_CHANGE_MESSAGE_TICKS: u64 = 50; + /// Public so the runtime can pin its `[cluster] + /// request_start_view_retransmit_interval` config default against this + /// built-in. + pub const REQUEST_START_VIEW_MESSAGE_TICKS: u64 = 100; #[must_use] pub fn new(replica_id: u128) -> Self { @@ -188,6 +201,36 @@ impl TimeoutManager { self.prepare = Timeout::new(self.prepare.id, ticks); } + /// Override the view-change retransmit interval (`[cluster] + /// view_change_retransmit_interval`), in consensus ticks. Drives BOTH the + /// `StartViewChange` and `DoViewChange` retransmit timers - kept equal by + /// design so a view change retransmits both messages at one cadence. + /// Replaces the timeout objects, so any countdown already in flight is + /// discarded: call before the replica starts ticking (i.e. before `init`). + pub const fn set_view_change_retransmit_ticks(&mut self, ticks: u64) { + self.start_view_change_message = Timeout::new(self.start_view_change_message.id, ticks); + self.do_view_change_message = Timeout::new(self.do_view_change_message.id, ticks); + } + + /// Override the view-change status backstop (`[cluster] + /// view_change_status_timeout`), in consensus ticks. A view change stalled + /// this long escalates to a fresh cluster-wide election. 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_view_change_status_ticks(&mut self, ticks: u64) { + self.view_change_status = Timeout::new(self.view_change_status.id, ticks); + } + + /// Override the request-start-view retransmit interval (`[cluster] + /// request_start_view_retransmit_interval`), in consensus ticks: how often + /// a recovering or view-change backup re-requests the current view's + /// `StartView`. 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_request_start_view_ticks(&mut self, ticks: u64) { + self.request_start_view_message = Timeout::new(self.request_start_view_message.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. diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index fe342c55b..2cd08855e 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -611,6 +611,29 @@ commit_broadcast_interval = "500ms" # more replica traffic; must be nonzero. Drives the consensus Prepare timer. prepare_retransmit_interval = "250ms" +# How often a replica retransmits its StartViewChange / DoViewChange while a +# view change is in progress (duration). Lower values converge a healthy +# election faster at the cost of more replica traffic; must be nonzero. Drives +# both consensus view-change retransmit timers. +view_change_retransmit_interval = "500ms" + +# Backstop for a stalled view change (duration): one that does not conclude +# within this window escalates to a fresh cluster-wide election. Must be nonzero +# and at least 4x view_change_retransmit_interval, so a few dropped view-change +# messages retransmit rather than prematurely escalate. +view_change_status_timeout = "5s" + +# How often a recovering or view-change backup re-requests the current view's +# StartView from its primary (duration); must be nonzero. Drives the consensus +# RequestStartView timer. +request_start_view_retransmit_interval = "1s" + +# How many consecutive unanswered RequestStartView probes a recovering replica +# tolerates before falling back to an election (integer). A full-cluster restart +# leaves nobody settled to answer, so the replica elects on its recovered log. +# Must be between 1 and 100. +view_probe_attempts_max = 5 + # 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 d86235e4c..4d9064848 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -960,6 +960,10 @@ async fn shard_main( cluster_heartbeat_ticks(config), commit_broadcast_ticks(config), prepare_retransmit_ticks(config), + view_change_retransmit_ticks(config), + view_change_status_ticks(config), + request_start_view_ticks(config), + config.cluster.view_probe_attempts_max, recovery_barrier_deadline(config.cluster.heartbeat_timeout.get_duration()), ); (Some(consensus), Some(journal), snapshot) @@ -1697,6 +1701,8 @@ const _: () = assert!( configs::ng_partition::DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH == consensus::PIPELINE_PREPARE_QUEUE_MAX ); +const _: () = + assert!(configs::ng_cluster::DEFAULT_VIEW_PROBE_ATTEMPTS_MAX == consensus::PROBE_ATTEMPTS_MAX); /// 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 { @@ -1717,7 +1723,10 @@ pub(crate) fn cluster_heartbeat_ticks(config: &ServerNgConfig) -> u64 { /// worst-case recovery is dominated by the heartbeat-independent term - the /// `ViewChangeStatus` backstop plus election ceremony and suffix recommit, /// empirically ~7s - so the scaled value must never fall under this or a -/// fast-heartbeat cluster would 503 legitimate reads mid-recovery. +/// fast-heartbeat cluster would 503 legitimate reads mid-recovery. That +/// backstop is now the configurable `[cluster] view_change_status_timeout` (5s +/// default), so a deployment that raises it far past the default stretches this +/// heartbeat-independent term beyond what the floor assumes. const RECOVERY_BARRIER_DEADLINE_FLOOR: Duration = Duration::from_secs(15); /// Heartbeat multiplier for the recovery deadline: a slower heartbeat stretches @@ -1752,6 +1761,37 @@ pub(crate) fn prepare_retransmit_ticks(config: &ServerNgConfig) -> u64 { duration_to_ticks(config.cluster.prepare_retransmit_interval.get_duration()) } +/// `[cluster] view_change_retransmit_interval` in consensus ticks: how often a +/// replica retransmits its `StartViewChange` / `DoViewChange` during a view +/// change. Applied to every consensus group, matching `cluster_heartbeat_ticks`. +pub(crate) fn view_change_retransmit_ticks(config: &ServerNgConfig) -> u64 { + duration_to_ticks( + config + .cluster + .view_change_retransmit_interval + .get_duration(), + ) +} + +/// `[cluster] view_change_status_timeout` in consensus ticks: the stalled +/// view-change backstop before escalating to a fresh election. Applied to every +/// consensus group, matching `cluster_heartbeat_ticks`. +pub(crate) fn view_change_status_ticks(config: &ServerNgConfig) -> u64 { + duration_to_ticks(config.cluster.view_change_status_timeout.get_duration()) +} + +/// `[cluster] request_start_view_retransmit_interval` in consensus ticks: how +/// often a recovering or view-change backup re-requests the current `StartView`. +/// Applied to every consensus group, matching `cluster_heartbeat_ticks`. +pub(crate) fn request_start_view_ticks(config: &ServerNgConfig) -> u64 { + duration_to_ticks( + config + .cluster + .request_start_view_retransmit_interval + .get_duration(), + ) +} + #[allow(clippy::too_many_arguments)] fn restore_metadata_consensus( journal: &PrepareJournal, @@ -1765,6 +1805,10 @@ fn restore_metadata_consensus( normal_heartbeat_ticks: u64, commit_message_ticks: u64, prepare_ticks: u64, + view_change_retransmit_ticks: u64, + view_change_status_ticks: u64, + request_start_view_ticks: u64, + probe_attempts_max: u32, recovery_deadline: Duration, ) -> VsrConsensus<Rc<IggyMessageBus>> { let mut consensus = VsrConsensus::new( @@ -1781,6 +1825,10 @@ fn restore_metadata_consensus( consensus.set_normal_heartbeat_ticks(normal_heartbeat_ticks); consensus.set_commit_message_ticks(commit_message_ticks); consensus.set_prepare_ticks(prepare_ticks); + consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks); + consensus.set_view_change_status_ticks(view_change_status_ticks); + consensus.set_request_start_view_ticks(request_start_view_ticks); + consensus.set_probe_attempts_max(probe_attempts_max); let last_header = journal .last_op() @@ -1911,6 +1959,10 @@ async fn load_partition( 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)); + consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks(config)); + consensus.set_view_change_status_ticks(view_change_status_ticks(config)); + consensus.set_request_start_view_ticks(request_start_view_ticks(config)); + consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max); // 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 @@ -3236,6 +3288,82 @@ mod tests { ); } + #[test] + fn default_view_change_retransmit_interval_matches_consensus_constant() { + // The config default lives in core/server-ng/config.toml (a string, so + // no static assert can pin it). One knob drives both view-change + // retransmit timers, which are equal by design, so pin it against both. + let config_default = configs::ng_cluster::ClusterConfig::default() + .view_change_retransmit_interval + .get_duration() + .as_millis(); + let start_view_change = + u128::from(consensus::TimeoutManager::START_VIEW_CHANGE_MESSAGE_TICKS) + * shard::CONSENSUS_TICK_INTERVAL.as_millis(); + let do_view_change = u128::from(consensus::TimeoutManager::DO_VIEW_CHANGE_MESSAGE_TICKS) + * shard::CONSENSUS_TICK_INTERVAL.as_millis(); + assert_eq!( + config_default, start_view_change, + "[cluster] view_change_retransmit_interval default drifted from \ + TimeoutManager::START_VIEW_CHANGE_MESSAGE_TICKS" + ); + assert_eq!( + config_default, do_view_change, + "[cluster] view_change_retransmit_interval default drifted from \ + TimeoutManager::DO_VIEW_CHANGE_MESSAGE_TICKS" + ); + } + + #[test] + fn default_view_change_status_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() + .view_change_status_timeout + .get_duration() + .as_millis(); + let built_in = u128::from(consensus::TimeoutManager::VIEW_CHANGE_STATUS_TICKS) + * shard::CONSENSUS_TICK_INTERVAL.as_millis(); + assert_eq!( + config_default, built_in, + "[cluster] view_change_status_timeout default drifted from \ + TimeoutManager::VIEW_CHANGE_STATUS_TICKS" + ); + } + + #[test] + fn default_request_start_view_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() + .request_start_view_retransmit_interval + .get_duration() + .as_millis(); + let built_in = u128::from(consensus::TimeoutManager::REQUEST_START_VIEW_MESSAGE_TICKS) + * shard::CONSENSUS_TICK_INTERVAL.as_millis(); + assert_eq!( + config_default, built_in, + "[cluster] request_start_view_retransmit_interval default drifted from \ + TimeoutManager::REQUEST_START_VIEW_MESSAGE_TICKS" + ); + } + + #[test] + fn default_view_probe_attempts_max_matches_consensus_constant() { + // Belt and suspenders with the static assert above: that pins the + // duplicated configs-crate literal, this pins the shipped config.toml + // value the simulator and un-configured replicas run on. + let config_default = configs::ng_cluster::ClusterConfig::default().view_probe_attempts_max; + assert_eq!( + config_default, + consensus::PROBE_ATTEMPTS_MAX, + "[cluster] view_probe_attempts_max default drifted from \ + consensus::PROBE_ATTEMPTS_MAX" + ); + } + #[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 8bccd5521..f6a10f349 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -501,6 +501,11 @@ pub async fn build_partition_fresh( 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)); + consensus + .set_view_change_retransmit_ticks(crate::bootstrap::view_change_retransmit_ticks(config)); + consensus.set_view_change_status_ticks(crate::bootstrap::view_change_status_ticks(config)); + consensus.set_request_start_view_ticks(crate::bootstrap::request_start_view_ticks(config)); + consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max); // 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
