diff --git a/crates/leader-election/README.md b/crates/leader-election/README.md index a4d0497..6a7d9fe 100644 --- a/crates/leader-election/README.md +++ b/crates/leader-election/README.md @@ -74,7 +74,9 @@ elector.run(MyCallbacks, cancel).await; | `retry_period` | 2s | Interval between acquire/renew attempts (with jitter) | | `release_on_cancel` | `true` | Whether to release the lease when context is cancelled | -**Constraint:** `lease_duration > renew_deadline > retry_period × 1.2` +**Constraints:** `lease_duration > renew_deadline > retry_period × 1.2`. Kubernetes persists +`lease_duration` as whole `i32` seconds, so the truncated value must be positive and remain +greater than `renew_deadline`. ## RBAC diff --git a/crates/leader-election/src/config.rs b/crates/leader-election/src/config.rs index dfa239f..219e262 100644 --- a/crates/leader-election/src/config.rs +++ b/crates/leader-election/src/config.rs @@ -18,7 +18,10 @@ use std::time::Duration; /// Configuration for the leader elector. /// -/// Constraint: `lease_duration > renew_deadline > retry_period * 1.2` +/// Timing constraint: `lease_duration > renew_deadline > retry_period * 1.2`. +/// +/// Kubernetes persists the lease duration as whole `i32` seconds. The truncated value must be +/// positive and remain greater than `renew_deadline`. #[derive(Debug, Clone)] pub struct LeaderElectorConfig { /// Unique identity for this instance (typically pod name or hostname). diff --git a/crates/leader-election/src/elector.rs b/crates/leader-election/src/elector.rs index 1d23bde..db69868 100644 --- a/crates/leader-election/src/elector.rs +++ b/crates/leader-election/src/elector.rs @@ -34,6 +34,8 @@ use crate::state::{LeaderElectorHandle, LeaderState}; /// The leader elector — drives the acquire/renew/release loop against a Lock backend. pub struct LeaderElector { config: LeaderElectorConfig, + /// Whole-second lease duration validated for the Kubernetes wire format. + lease_duration_seconds: i32, lock: L, observed: Arc>, clock: Box, @@ -42,8 +44,9 @@ pub struct LeaderElector { impl LeaderElector { /// Create a new LeaderElector with the given configuration, lock, and clock. /// - /// Returns `Error::InvalidConfig` if the timing constraints are violated: - /// `lease_duration > renew_deadline > retry_period * 1.2` + /// Returns `Error::InvalidConfig` if the timing constraints are violated or if + /// `lease_duration` cannot be represented as positive whole `i32` seconds. The persisted + /// whole-second duration must also remain greater than `renew_deadline`. pub fn new( config: LeaderElectorConfig, lock: L, @@ -73,8 +76,37 @@ impl LeaderElector { }); } + let lease_duration_seconds = + i32::try_from(config.lease_duration.as_secs()).map_err(|_| Error::InvalidConfig { + message: format!( + "lease_duration ({:?}) must fit in the Kubernetes Lease int32 seconds field", + config.lease_duration + ), + })?; + if lease_duration_seconds == 0 { + return Err(Error::InvalidConfig { + message: format!( + "lease_duration ({:?}) must be at least one second", + config.lease_duration + ), + }); + } + + // Kubernetes stores only whole seconds. Re-check the timing constraint after + // truncation so the persisted lease cannot expire before the renew deadline. + let persisted_lease_duration = Duration::from_secs(config.lease_duration.as_secs()); + if persisted_lease_duration <= config.renew_deadline { + return Err(Error::InvalidConfig { + message: format!( + "lease_duration rounded to whole seconds ({persisted_lease_duration:?}) must be greater than renew_deadline ({:?})", + config.renew_deadline + ), + }); + } + Ok(Self { config, + lease_duration_seconds, lock, observed: Arc::new(RwLock::new(ObservedState::default())), clock: Box::new(clock), @@ -396,8 +428,7 @@ impl LeaderElector { let observed = self.observed.read().await; match (observed.record.as_ref(), observed.observed_time) { (Some(record), Some(obs_time)) => { - let duration = Duration::from_secs(record.lease_duration_seconds as u64); - now < obs_time + duration + Self::is_lease_valid_at(obs_time, record.lease_duration_seconds, now) } _ => false, } @@ -416,13 +447,34 @@ impl LeaderElector { match observed.observed_time { Some(observed_time) => { - let duration = Duration::from_secs(record.lease_duration_seconds as u64); - now < observed_time + duration + Self::is_lease_valid_at(observed_time, record.lease_duration_seconds, now) } None => false, } } + /// Evaluate a lease duration without unchecked integer casts or panicking date arithmetic. + fn is_lease_valid_at( + observed_time: DateTime, + lease_duration_seconds: i32, + now: DateTime, + ) -> bool { + if lease_duration_seconds <= 0 { + return false; + } + + let Some(duration) = chrono::Duration::try_seconds(i64::from(lease_duration_seconds)) + else { + return false; + }; + + // A positive lease whose deadline exceeds Chrono's representable range must not be + // expired early: doing so could allow two participants to lead concurrently. + observed_time + .checked_add_signed(duration) + .is_none_or(|deadline| now < deadline) + } + /// Build a new election record for this identity. async fn build_record(&self, now: DateTime) -> LeaderElectionRecord { let observed = self.observed.read().await; @@ -439,7 +491,7 @@ impl LeaderElector { LeaderElectionRecord { holder_identity: self.config.identity.clone(), - lease_duration_seconds: self.config.lease_duration.as_secs() as i32, + lease_duration_seconds: self.lease_duration_seconds, acquire_time, renew_time: now, leader_transitions: observed @@ -599,6 +651,58 @@ mod tests { assert!(matches!(result, Err(Error::InvalidConfig { .. }))); } + #[test] + fn test_config_validation_lease_shorter_than_one_second() { + let config = LeaderElectorConfig { + identity: "node-1".to_string(), + lease_duration: Duration::from_millis(900), + renew_deadline: Duration::from_millis(500), + retry_period: Duration::from_millis(100), + release_on_cancel: true, + }; + let result = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock); + assert!(matches!(result, Err(Error::InvalidConfig { .. }))); + } + + #[test] + fn test_config_validation_uses_persisted_whole_seconds() { + let config = LeaderElectorConfig { + identity: "node-1".to_string(), + lease_duration: Duration::from_millis(1500), + renew_deadline: Duration::from_millis(1200), + retry_period: Duration::from_millis(500), + release_on_cancel: true, + }; + let result = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock); + assert!(matches!(result, Err(Error::InvalidConfig { .. }))); + } + + #[test] + fn test_config_validation_lease_exceeds_wire_format() { + let config = LeaderElectorConfig { + identity: "node-1".to_string(), + lease_duration: Duration::from_secs(u64::from(i32::MAX.unsigned_abs()) + 1), + renew_deadline: Duration::from_secs(10), + retry_period: Duration::from_secs(2), + release_on_cancel: true, + }; + let result = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock); + assert!(matches!(result, Err(Error::InvalidConfig { .. }))); + } + + #[test] + fn test_config_validation_accepts_wire_format_maximum() { + let config = LeaderElectorConfig { + identity: "node-1".to_string(), + lease_duration: Duration::from_secs(u64::from(i32::MAX.unsigned_abs())), + renew_deadline: Duration::from_secs(10), + retry_period: Duration::from_secs(2), + release_on_cancel: true, + }; + let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap(); + assert_eq!(elector.lease_duration_seconds, i32::MAX); + } + #[test] fn test_jittered_retry_period() { let config = test_config("node-1"); @@ -685,6 +789,55 @@ mod tests { ); } + #[tokio::test] + async fn test_non_positive_lease_durations_are_expired() { + let config = test_config("node-1"); + let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap(); + let now = Utc::now(); + + for lease_duration_seconds in [i32::MIN, -1, 0] { + let record = LeaderElectionRecord { + holder_identity: "node-2".to_string(), + lease_duration_seconds, + acquire_time: now, + renew_time: now, + leader_transitions: 0, + }; + elector.observe_record(record.clone(), now).await; + + assert!(!elector.is_lease_valid(now).await); + assert!(!elector.is_observed_lease_valid(&record, now).await); + } + } + + #[test] + fn test_lease_validity_boundaries_do_not_panic() { + let observed_time = Utc::now(); + + assert!(LeaderElector::::is_lease_valid_at( + observed_time, + 1, + observed_time + )); + assert!(!LeaderElector::::is_lease_valid_at( + observed_time, + 1, + observed_time + chrono::Duration::seconds(1) + )); + assert!(LeaderElector::::is_lease_valid_at( + observed_time, + i32::MAX, + observed_time + )); + + let maximum_time = DateTime::::MAX_UTC; + assert!(LeaderElector::::is_lease_valid_at( + maximum_time, + 1, + maximum_time + )); + } + // ─── Dummy lock for unit tests ──────────────────────────────────── struct DummyLock { diff --git a/crates/leader-election/src/lock/lease.rs b/crates/leader-election/src/lock/lease.rs index 8acbb7b..1862f05 100644 --- a/crates/leader-election/src/lock/lease.rs +++ b/crates/leader-election/src/lock/lease.rs @@ -257,4 +257,20 @@ mod tests { }; assert!(spec_to_record(&spec).is_none()); } + + #[test] + fn spec_to_record_preserves_duration_for_core_validation() { + for lease_duration_seconds in [i32::MIN, -1, 0, i32::MAX] { + let spec = LeaseSpec { + holder_identity: Some("pod-1".into()), + lease_duration_seconds: Some(lease_duration_seconds), + acquire_time: Some(MicroTime(Utc::now())), + renew_time: Some(MicroTime(Utc::now())), + lease_transitions: Some(0), + }; + + let record = spec_to_record(&spec).expect("complete lease should produce a record"); + assert_eq!(record.lease_duration_seconds, lease_duration_seconds); + } + } } diff --git a/crates/leader-election/tests/integration_tests.rs b/crates/leader-election/tests/integration_tests.rs index bee4e1c..ef6fdbe 100644 --- a/crates/leader-election/tests/integration_tests.rs +++ b/crates/leader-election/tests/integration_tests.rs @@ -416,6 +416,40 @@ async fn test_acquire_active_lease() { let _ = tokio::time::timeout(Duration::from_secs(2), handle).await; } +#[tokio::test] +async fn test_acquire_non_positive_lease_without_panicking() { + for lease_duration_seconds in [i32::MIN, -1, 0] { + let lock = FakeLock::new("node-2"); + let now = Utc::now(); + lock.set_record(LeaderElectionRecord { + holder_identity: "node-1".to_string(), + lease_duration_seconds, + acquire_time: now, + renew_time: now, + leader_transitions: 0, + }) + .await; + + let callbacks = Arc::new(TestCallbacks::new()); + let elector = LeaderElector::new(test_config("node-2"), lock, MockClock::new(now)).unwrap(); + let cancel = CancellationToken::new(); + let cancel_clone = cancel.clone(); + let cb = callbacks.clone(); + let handle = + tokio::spawn(async move { elector.run(SharedCallbacks(cb), cancel_clone).await }); + + tokio::time::sleep(Duration::from_millis(250)).await; + cancel.cancel(); + + let run_result = tokio::time::timeout(Duration::from_secs(2), handle) + .await + .expect("elector should complete") + .expect("elector task should not panic"); + assert!(run_result.is_ok()); + assert_eq!(callbacks.started_count().await, 1); + } +} + #[tokio::test] async fn test_clock_skewed_candidate_waits_while_remote_record_changes() { let lock = FakeLock::new("shared");