Refactor: Add Leader Lease Duration into RaftState.vote

This commit enhances the management of leader lease durations by
integrating them directly into the `RaftState.vote` structure.
Previously, the leader lease was managed separately as a configuration
parameter, and `RaftState.vote` only recorded the timestamp of the last
received `AppendEntries` RPC.

- Store the lease duration together with the last update time in
  `RaftState.vote`, we simplify the process of expiration checking and
  improve the clarity of state logging.

- Dynamic Lease Management: This integration allows the lease duration
  to be reset dynamically, which is particularly useful when a
  leadership transfer is scheduled. This capability ensures that the
  leader's expiration can be immediately considered without delay.

- Runtime Adjustments: Leaders can now adjust their lease durations
  at runtime as needed, without requiring a configuration update. This
  flexibility supports more responsive and adaptable leadership
  management within the Raft cluster.

- Rename `UTime` to `Leased` and add field `lease` and related methods.

Author: drmingdrmer

openraft/src/engine/engine_impl.rs

@@ -7,7 +7,6 @@ use crate::core::raft_msg::AppendEntriesTx;
 use crate::core::raft_msg::ResultSender;
 use crate::core::sm;
 use crate::core::ServerState;
-use crate::display_ext::DisplayInstantExt;
 use crate::display_ext::DisplayOptionExt;
 use crate::display_ext::DisplaySliceExt;
 use crate::engine::engine_config::EngineConfig;
@@ -210,7 +209,7 @@ where C: RaftTypeConfig
             ..Default::default()
         };
         self.last_seen_vote = vote;
-        self.state.vote.update(C::now(), vote);
+        self.state.vote.update(C::now(), Duration::default(), vote);
         self.following_handler().do_append_entries(vec![entry]);
 
         // With the new config, start to elect to become leader
@@ -293,37 +292,22 @@ where C: RaftTypeConfig
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) fn handle_vote_req(&mut self, req: VoteRequest<C>) -> VoteResponse<C> {
         let now = C::now();
-        let lease = self.config.timer_config.leader_lease;
-        let local_vote = self.state.vote_ref();
-
-        // Make default vote-last-modified a low enough value, that expires leader lease.
-        let local_vote_utime =
-            self.state.vote_last_modified().unwrap_or_else(|| now - lease - Duration::from_millis(1));
+        let local_leased_vote = &self.state.vote;
 
         tracing::info!(req = display(&req), "Engine::handle_vote_req");
         tracing::info!(
-            my_vote = display(self.state.vote_ref()),
+            my_vote = display(&**local_leased_vote),
             my_last_log_id = display(self.state.last_log_id().display()),
+            lease = display(local_leased_vote.time_info(now)),
             "Engine::handle_vote_req"
         );
-        tracing::info!(
-            "now; {}, vote is updated at: {}, vote is updated before {:?}, leader lease({:?}) will expire after {:?}",
-            now.display(),
-            local_vote_utime.display(),
-            now - local_vote_utime,
-            lease,
-            local_vote_utime + lease - now
-        );
 
-        if local_vote.is_committed() {
+        if local_leased_vote.is_committed() {
             // Current leader lease has not yet expired, reject voting request
-            if now <= local_vote_utime + lease {
+            if !local_leased_vote.is_expired(now) {
                 tracing::info!(
-                    "reject vote-request: leader lease has not yet expire; now; {:?}, vote is update at: {:?}, leader lease({:?}) will expire after {:?}",
-                    now,
-                    local_vote_utime,
-                    lease,
-                    local_vote_utime + lease - now
+                    "reject vote-request: leader lease has not yet expire: {}",
+                    local_leased_vote.time_info(now)
                 );
 
                 return VoteResponse::new(self.state.vote_ref(), self.state.last_log_id().copied());

openraft/src/engine/handler/following_handler/append_entries_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 use pretty_assertions::assert_eq;
@@ -11,11 +12,11 @@ use crate::raft_state::IOId;
 use crate::raft_state::LogStateReader;
 use crate::testing::blank_ent;
 use crate::testing::log_id;
-use crate::utime::UTime;
+use crate::type_config::TypeConfigExt;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
-use crate::TokioInstant;
 use crate::Vote;
 
 fn m01() -> Membership<UTConfig> {
@@ -31,7 +32,11 @@ fn eng() -> Engine<UTConfig> {
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
     eng.config.id = 2;
-    eng.state.vote.update(TokioInstant::now(), Vote::new_committed(2, 1));
+    eng.state.vote.update(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.log_ids.append(log_id(1, 1, 1));
     eng.state.log_ids.append(log_id(2, 1, 3));
     eng.state.membership_state = MembershipState::new(
@@ -81,7 +86,11 @@ fn test_follower_append_entries_update_accepted() -> anyhow::Result<()> {
     // Update to a new Leader and smaller log id
     {
         // Assume this node's Leader becomes T3-N1
-        eng.state.vote = UTime::new(TokioInstant::now(), Vote::new_committed(3, 1));
+        eng.state.vote = Leased::new(
+            UTConfig::<()>::now(),
+            Duration::from_millis(500),
+            Vote::new_committed(3, 1),
+        );
         eng.following_handler().append_entries(Some(log_id(2, 1, 3)), vec![
             //
             blank_ent(3, 1, 4),

openraft/src/engine/handler/following_handler/commit_entries_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 
@@ -9,7 +10,7 @@ use crate::raft_state::IOId;
 use crate::raft_state::LogStateReader;
 use crate::testing::log_id;
 use crate::type_config::TypeConfigExt;
-use crate::utime::UTime;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
@@ -27,7 +28,11 @@ fn eng() -> Engine<UTConfig> {
     let mut eng = Engine::testing_default(0);
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
-    eng.state.vote = UTime::new(UTConfig::<()>::now(), Vote::new_committed(2, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.committed = Some(log_id(1, 1, 1));
     eng.state.membership_state = MembershipState::new(
         Arc::new(EffectiveMembership::new(Some(log_id(1, 1, 1)), m01())),

openraft/src/engine/handler/following_handler/do_append_entries_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 
@@ -11,7 +12,7 @@ use crate::raft_state::LogStateReader;
 use crate::testing::blank_ent;
 use crate::testing::log_id;
 use crate::type_config::TypeConfigExt;
-use crate::utime::UTime;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Entry;
 use crate::EntryPayload;
@@ -40,7 +41,11 @@ fn eng() -> Engine<UTConfig> {
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
     eng.config.id = 2;
-    eng.state.vote = UTime::new(UTConfig::<()>::now(), Vote::new_committed(2, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.log_ids.append(log_id(1, 1, 1));
     eng.state.log_ids.append(log_id(2, 1, 3));
     eng.state.membership_state = MembershipState::new(
@@ -54,7 +59,7 @@ fn eng() -> Engine<UTConfig> {
 #[test]
 fn test_follower_do_append_entries_no_membership_entries() -> anyhow::Result<()> {
     let mut eng = eng();
-    eng.state.vote = UTime::without_utime(Vote::new_committed(1, 1));
+    eng.state.vote = Leased::without_last_update(Vote::new_committed(1, 1));
 
     eng.following_handler().do_append_entries(vec![blank_ent(3, 1, 4)]);
 
@@ -96,7 +101,7 @@ fn test_follower_do_append_entries_one_membership_entry() -> anyhow::Result<()>
     // - Follower become Learner, since it is not in the new effective membership.
     let mut eng = eng();
     eng.config.id = 2; // make it a member, the become learner
-    eng.state.vote = UTime::without_utime(Vote::new_committed(1, 1));
+    eng.state.vote = Leased::without_last_update(Vote::new_committed(1, 1));
 
     eng.following_handler().do_append_entries(vec![blank_ent(3, 1, 4), Entry::<UTConfig> {
         log_id: log_id(3, 1, 5),
@@ -152,7 +157,7 @@ fn test_follower_do_append_entries_three_membership_entries() -> anyhow::Result<
     let mut eng = eng();
     eng.config.id = 5; // make it a learner, then become follower
     eng.state.server_state = eng.calc_server_state();
-    eng.state.vote = UTime::without_utime(Vote::new_committed(1, 1));
+    eng.state.vote = Leased::without_last_update(Vote::new_committed(1, 1));
 
     eng.following_handler().do_append_entries(vec![
         blank_ent(3, 1, 4),

openraft/src/engine/handler/following_handler/install_snapshot_test.rs

@@ -1,5 +1,6 @@
 use std::io::Cursor;
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 use pretty_assertions::assert_eq;
@@ -13,12 +14,12 @@ use crate::engine::LogIdList;
 use crate::raft_state::IOId;
 use crate::raft_state::LogStateReader;
 use crate::testing::log_id;
+use crate::type_config::TypeConfigExt;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::Snapshot;
 use crate::SnapshotMeta;
 use crate::StoredMembership;
-use crate::TokioInstant;
 use crate::Vote;
 
 fn m12() -> Membership<UTConfig> {
@@ -33,7 +34,11 @@ fn eng() -> Engine<UTConfig> {
     let mut eng = Engine::testing_default(0);
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
-    eng.state.vote.update(TokioInstant::now(), Vote::new_committed(2, 1));
+    eng.state.vote.update(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.committed = Some(log_id(4, 1, 5));
     eng.state.log_ids = LogIdList::new(vec![
         //
@@ -179,7 +184,11 @@ fn test_install_snapshot_conflict() -> anyhow::Result<()> {
         let mut eng = Engine::<UTConfig>::testing_default(0);
         eng.state.enable_validation(false); // Disable validation for incomplete state
 
-        eng.state.vote.update(TokioInstant::now(), Vote::new_committed(2, 1));
+        eng.state.vote.update(
+            UTConfig::<()>::now(),
+            Duration::from_millis(500),
+            Vote::new_committed(2, 1),
+        );
         eng.state.committed = Some(log_id(2, 1, 3));
         eng.state.log_ids = LogIdList::new(vec![
             //

openraft/src/engine/handler/following_handler/truncate_logs_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 
@@ -9,7 +10,7 @@ use crate::engine::LogIdList;
 use crate::raft_state::LogStateReader;
 use crate::testing::log_id;
 use crate::type_config::TypeConfigExt;
-use crate::utime::UTime;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
@@ -33,7 +34,11 @@ fn eng() -> Engine<UTConfig> {
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
     eng.config.id = 2;
-    eng.state.vote = UTime::new(UTConfig::<()>::now(), Vote::new_committed(2, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.log_ids = LogIdList::new(vec![
         log_id(2, 1, 2), //
         log_id(4, 1, 4),

openraft/src/engine/handler/following_handler/update_committed_membership_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 
@@ -7,7 +8,7 @@ use crate::engine::testing::UTConfig;
 use crate::engine::Engine;
 use crate::testing::log_id;
 use crate::type_config::TypeConfigExt;
-use crate::utime::UTime;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
@@ -28,7 +29,11 @@ fn m34() -> Membership<UTConfig> {
 fn eng() -> Engine<UTConfig> {
     let mut eng = Engine::testing_default(0);
     eng.config.id = 2;
-    eng.state.vote = UTime::new(UTConfig::<()>::now(), Vote::new_committed(2, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(2, 1),
+    );
     eng.state.membership_state = MembershipState::new(
         Arc::new(EffectiveMembership::new(Some(log_id(1, 1, 1)), m01())),
         Arc::new(EffectiveMembership::new(Some(log_id(2, 1, 3)), m23())),

openraft/src/engine/handler/leader_handler/append_entries_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 #[allow(unused_imports)]
@@ -19,14 +20,14 @@ use crate::raft_state::IOId;
 use crate::raft_state::LogStateReader;
 use crate::testing::blank_ent;
 use crate::testing::log_id;
-use crate::utime::UTime;
+use crate::type_config::TypeConfigExt;
+use crate::utime::Leased;
 use crate::vote::CommittedLeaderId;
 use crate::EffectiveMembership;
 use crate::Entry;
 use crate::LogId;
 use crate::Membership;
 use crate::MembershipState;
-use crate::TokioInstant;
 use crate::Vote;
 
 fn m01() -> Membership<UTConfig> {
@@ -52,7 +53,11 @@ fn eng() -> Engine<UTConfig> {
 
     eng.config.id = 1;
     eng.state.committed = Some(log_id(0, 1, 0));
-    eng.state.vote = UTime::new(TokioInstant::now(), Vote::new_committed(3, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(3, 1),
+    );
     eng.state.log_ids.append(log_id(1, 1, 1));
     eng.state.log_ids.append(log_id(2, 1, 3));
     eng.state.membership_state = MembershipState::new(

openraft/src/engine/handler/leader_handler/get_read_log_id_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 #[allow(unused_imports)]
@@ -11,11 +12,11 @@ use pretty_assertions::assert_str_eq;
 use crate::engine::testing::UTConfig;
 use crate::engine::Engine;
 use crate::testing::log_id;
-use crate::utime::UTime;
+use crate::type_config::TypeConfigExt;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
-use crate::TokioInstant;
 use crate::Vote;
 
 fn m01() -> Membership<UTConfig> {
@@ -32,7 +33,11 @@ fn eng() -> Engine<UTConfig> {
 
     eng.config.id = 1;
     eng.state.committed = Some(log_id(0, 1, 0));
-    eng.state.vote = UTime::new(TokioInstant::now(), Vote::new_committed(3, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(3, 1),
+    );
     eng.state.log_ids.append(log_id(1, 1, 1));
     eng.state.log_ids.append(log_id(2, 1, 3));
     eng.state.membership_state = MembershipState::new(

openraft/src/engine/handler/leader_handler/send_heartbeat_test.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use maplit::btreeset;
 #[allow(unused_imports)]
@@ -14,11 +15,11 @@ use crate::engine::Engine;
 use crate::progress::Inflight;
 use crate::progress::Progress;
 use crate::testing::log_id;
-use crate::utime::UTime;
+use crate::type_config::TypeConfigExt;
+use crate::utime::Leased;
 use crate::EffectiveMembership;
 use crate::Membership;
 use crate::MembershipState;
-use crate::TokioInstant;
 use crate::Vote;
 
 fn m01() -> Membership<UTConfig> {
@@ -35,7 +36,11 @@ fn eng() -> Engine<UTConfig> {
 
     eng.config.id = 1;
     eng.state.committed = Some(log_id(0, 1, 0));
-    eng.state.vote = UTime::new(TokioInstant::now(), Vote::new_committed(3, 1));
+    eng.state.vote = Leased::new(
+        UTConfig::<()>::now(),
+        Duration::from_millis(500),
+        Vote::new_committed(3, 1),
+    );
     eng.state.log_ids.append(log_id(1, 1, 1));
     eng.state.log_ids.append(log_id(2, 1, 3));
     eng.state.membership_state = MembershipState::new(