Fix status redaction and leader election lease validity

crates/leader-election/src/elector.rs

@@ -269,7 +269,7 @@ impl<L: Lock> LeaderElector<L> {
             let record = self.build_record(now).await;
             match self.lock.update(record.clone()).await {
                 Ok(()) => {
-                    self.update_observed(record, now).await;
+                    self.set_observed_record(record, now).await;
                     return true;
                 }
                 Err(e) => {
@@ -282,7 +282,7 @@ impl<L: Lock> LeaderElector<L> {
         // Slow path: Get current record
         let old_record = match self.lock.get().await {
             Ok(Some(r)) => {
-                self.update_observed(r.clone(), now).await;
+                self.observe_record(r.clone(), now).await;
                 r
             }
             Ok(None) => {
@@ -291,7 +291,7 @@ impl<L: Lock> LeaderElector<L> {
                 return match self.lock.create(record.clone()).await {
                     Ok(()) => {
                         debug!(identity = %self.config.identity, "created new lease");
-                        self.update_observed(record, now).await;
+                        self.set_observed_record(record, now).await;
                         true
                     }
                     Err(e) => {
@@ -308,7 +308,7 @@ impl<L: Lock> LeaderElector<L> {
 
         // Check if we can take over
         if !old_record.holder_identity.is_empty()
-            && self.is_lease_valid_with(&old_record, now)
+            && self.is_observed_lease_valid(&old_record, now).await
             && !self.is_leader_with(&old_record)
         {
             // Lease is held by someone else and still valid — give up this attempt
@@ -328,7 +328,7 @@ impl<L: Lock> LeaderElector<L> {
 
         match self.lock.update(record.clone()).await {
             Ok(()) => {
-                self.update_observed(record, now).await;
+                self.set_observed_record(record, now).await;
                 true
             }
             Err(e) => {
@@ -403,11 +403,24 @@ impl<L: Lock> LeaderElector<L> {
         }
     }
 
-    /// Check if a specific record's lease is still valid relative to now.
-    fn is_lease_valid_with(&self, record: &LeaderElectionRecord, now: DateTime<Utc>) -> bool {
-        let duration = Duration::from_secs(record.lease_duration_seconds as u64);
-        // Use renew_time as the basis for validity (the last time the lease was refreshed)
-        now < record.renew_time + duration
+    /// Check if a specific record's lease is still valid based on local observation time.
+    async fn is_observed_lease_valid(
+        &self,
+        record: &LeaderElectionRecord,
+        now: DateTime<Utc>,
+    ) -> bool {
+        let observed = self.observed.read().await;
+        if observed.record.as_ref() != Some(record) {
+            return false;
+        }
+
+        match observed.observed_time {
+            Some(observed_time) => {
+                let duration = Duration::from_secs(record.lease_duration_seconds as u64);
+                now < observed_time + duration
+            }
+            None => false,
+        }
     }
 
     /// Build a new election record for this identity.
@@ -446,8 +459,19 @@ impl<L: Lock> LeaderElector<L> {
             .unwrap_or(false)
     }
 
-    /// Update observed state after a successful get or update.
-    async fn update_observed(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
+    /// Observe a record fetched from the lock backend.
+    async fn observe_record(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
+        let mut observed = self.observed.write().await;
+        if observed.record.as_ref() != Some(&record) {
+            observed.record = Some(record);
+            observed.observed_time = Some(now);
+        } else if observed.observed_time.is_none() {
+            observed.observed_time = Some(now);
+        }
+    }
+
+    /// Record a successful local create/update acknowledged by the lock backend.
+    async fn set_observed_record(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
         let mut observed = self.observed.write().await;
         observed.record = Some(record);
         observed.observed_time = Some(now);
@@ -608,25 +632,57 @@ mod tests {
         assert!(!elector.is_leader_with(&record_other));
     }
 
-    #[test]
-    fn test_is_lease_valid_with() {
+    #[tokio::test]
+    async fn test_observed_lease_valid_uses_local_observed_time() {
         let config = test_config("node-1");
         let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap();
         let now = Utc::now();
 
-        // Valid: renew_time + duration is in the future
         let record = LeaderElectionRecord {
-            holder_identity: "node-1".to_string(),
+            holder_identity: "node-2".to_string(),
+            lease_duration_seconds: 15,
+            acquire_time: now - chrono::Duration::seconds(30),
+            renew_time: now - chrono::Duration::seconds(30),
+            leader_transitions: 0,
+        };
+        elector.observe_record(record.clone(), now).await;
+
+        assert!(elector.is_observed_lease_valid(&record, now).await);
+        assert!(
+            elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(14))
+                .await
+        );
+        assert!(
+            !elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(16))
+                .await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_same_observed_record_does_not_refresh_observed_time() {
+        let config = test_config("node-1");
+        let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap();
+        let now = Utc::now();
+
+        let record = LeaderElectionRecord {
+            holder_identity: "node-2".to_string(),
             lease_duration_seconds: 15,
             acquire_time: now,
             renew_time: now,
             leader_transitions: 0,
         };
-        assert!(elector.is_lease_valid_with(&record, now));
-
-        // Expired: check time is past renew_time + duration
-        let future = now + chrono::Duration::seconds(20);
-        assert!(!elector.is_lease_valid_with(&record, future));
+        elector.observe_record(record.clone(), now).await;
+        elector
+            .observe_record(record.clone(), now + chrono::Duration::seconds(10))
+            .await;
+
+        assert!(
+            !elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(16))
+                .await
+        );
     }
 
     // ─── Dummy lock for unit tests ────────────────────────────────────

crates/leader-election/src/record.rs

@@ -17,7 +17,7 @@
 use chrono::{DateTime, Utc};
 
 /// Record stored in the Lease object, representing the current election state.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct LeaderElectionRecord {
     /// Identity of the current leader (empty if released).
     pub holder_identity: String,

crates/leader-election/tests/integration_tests.rs

@@ -30,6 +30,7 @@ use kube_leader_election::{
 // ─── Test Helpers ───────────────────────────────────────────────────────────
 
 /// Fake lock with controllable behavior for testing.
+#[derive(Clone)]
 struct FakeLock {
     identity: String,
     record: Arc<RwLock<Option<LeaderElectionRecord>>>,
@@ -116,6 +117,10 @@ impl MockClock {
             now: Arc::new(RwLock::new(time)),
         }
     }
+
+    async fn set(&self, time: DateTime<Utc>) {
+        *self.now.write().await = time;
+    }
 }
 
 impl Clock for MockClock {
@@ -318,13 +323,15 @@ async fn test_acquire_no_holder() {
     );
 }
 
-/// Test 3: Acquire when lease is expired.
+/// Test 3: Acquire only after the locally observed lease window expires.
 #[tokio::test]
 async fn test_acquire_expired_lease() {
     let lock = FakeLock::new("node-2");
+    let now = Utc::now();
 
-    // Set an expired lease held by another node
-    let expired_time = Utc::now() - chrono::Duration::seconds(30);
+    // The remote renew_time is old, but a candidate must not use that
+    // remote wall clock to take over immediately.
+    let expired_time = now - chrono::Duration::seconds(30);
     let expired_record = LeaderElectionRecord {
         holder_identity: "node-1".to_string(),
         lease_duration_seconds: 15,
@@ -334,7 +341,8 @@ async fn test_acquire_expired_lease() {
     };
     lock.set_record(expired_record).await;
 
-    let clock = MockClock::new(Utc::now());
+    let clock = MockClock::new(now);
+    let test_clock = clock.clone();
     let config = test_config("node-2");
     let callbacks = Arc::new(TestCallbacks::new());
 
@@ -345,13 +353,21 @@ async fn test_acquire_expired_lease() {
     let cb = callbacks.clone();
     let handle = tokio::spawn(async move { elector.run(SharedCallbacks(cb), cancel_clone).await });
 
-    tokio::time::sleep(Duration::from_millis(200)).await;
+    tokio::time::sleep(Duration::from_millis(150)).await;
+    assert_eq!(
+        callbacks.started_count().await,
+        0,
+        "should not acquire immediately from stale remote renew_time"
+    );
+
+    test_clock.set(now + chrono::Duration::seconds(16)).await;
+    tokio::time::sleep(Duration::from_millis(250)).await;
     cancel.cancel();
 
     let result = tokio::time::timeout(Duration::from_secs(2), handle).await;
     assert!(result.is_ok(), "elector should complete");
 
-    // node-2 should have acquired the expired lease
+    // node-2 should acquire after the local observed lease duration elapses.
     assert_eq!(
         callbacks.started_count().await,
         1,
@@ -400,6 +416,79 @@ async fn test_acquire_active_lease() {
     let _ = tokio::time::timeout(Duration::from_secs(2), handle).await;
 }
 
+#[tokio::test]
+async fn test_clock_skewed_candidate_waits_while_remote_record_changes() {
+    let lock = FakeLock::new("shared");
+    let remote_start = Utc::now();
+    let candidate_start = remote_start + chrono::Duration::seconds(60);
+    let leader_record = LeaderElectionRecord {
+        holder_identity: "node-1".to_string(),
+        lease_duration_seconds: 15,
+        acquire_time: remote_start,
+        renew_time: remote_start,
+        leader_transitions: 0,
+    };
+    lock.set_record(leader_record.clone()).await;
+
+    let clock = MockClock::new(candidate_start);
+    let test_clock = clock.clone();
+    let mut config = test_config("node-2");
+    config.release_on_cancel = false;
+    let callbacks = Arc::new(TestCallbacks::new());
+
+    let elector = LeaderElector::new(config, lock.clone(), clock).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(150)).await;
+    assert_eq!(
+        callbacks.started_count().await,
+        0,
+        "skewed candidate should not acquire immediately"
+    );
+
+    for step in 1..=3 {
+        test_clock
+            .set(candidate_start + chrono::Duration::seconds(step * 10))
+            .await;
+        lock.set_record(LeaderElectionRecord {
+            renew_time: remote_start + chrono::Duration::seconds(step * 10),
+            ..leader_record.clone()
+        })
+        .await;
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert_eq!(
+            callbacks.started_count().await,
+            0,
+            "candidate should not acquire while renewals keep changing the record"
+        );
+    }
+
+    test_clock
+        .set(candidate_start + chrono::Duration::seconds(46))
+        .await;
+    tokio::time::sleep(Duration::from_millis(250)).await;
+    cancel.cancel();
+
+    let result = tokio::time::timeout(Duration::from_secs(2), handle).await;
+    assert!(result.is_ok(), "elector should complete");
+    assert_eq!(
+        callbacks.started_count().await,
+        1,
+        "candidate should acquire after renewals stop and local observation expires"
+    );
+
+    let final_record = lock
+        .get()
+        .await
+        .expect("fake lock get should succeed")
+        .expect("record should exist");
+    assert_eq!(final_record.holder_identity, "node-2");
+}
+
 /// Test 5: Successful renewal.
 #[tokio::test]
 async fn test_renew_success() {