Change: only purge logs that are in snapshot

Let `snapshot+logs` be a complete state of a raft node.

The Assumption before is `state_machine+logs` is a complete state of a
raft node. This requires state machine to persist the state every time
applying a log, which would be an innecessary overhead.

- Change: remove ENV config entries. Do not let a lib be affected by
  environment variables.

- Change: remove `Config.keep_unsnapshoted_log`: now by default, logs
  not included in snapshot won't be deleted.

  Rename `Config.max_applied_log_to_keep` to `max_in_snapshot_log_to_keep`.

openraft/src/config/config.rs

@@ -83,75 +83,56 @@ fn parse_snapshot_policy(src: &str) -> Result<SnapshotPolicy, ConfigError> {
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub struct Config {
     /// The application specific name of this Raft cluster
-    #[clap(long, env = "RAFT_CLUSTER_NAME", default_value = "foo")]
+    #[clap(long, default_value = "foo")]
     pub cluster_name: String,
 
     /// The minimum election timeout in milliseconds
-    #[clap(long, env = "RAFT_ELECTION_TIMEOUT_MIN", default_value = "150")]
+    #[clap(long, default_value = "150")]
     pub election_timeout_min: u64,
 
     /// The maximum election timeout in milliseconds
-    #[clap(long, env = "RAFT_ELECTION_TIMEOUT_MAX", default_value = "300")]
+    #[clap(long, default_value = "300")]
     pub election_timeout_max: u64,
 
     /// The heartbeat interval in milliseconds at which leaders will send heartbeats to followers
-    #[clap(long, env = "RAFT_HEARTBEAT_INTERVAL", default_value = "50")]
+    #[clap(long, default_value = "50")]
     pub heartbeat_interval: u64,
 
     /// The timeout for sending a snapshot segment, in millisecond
-    #[clap(long, env = "RAFT_INSTALL_SNAPSHOT_TIMEOUT", default_value = "200")]
+    #[clap(long, default_value = "200")]
     pub install_snapshot_timeout: u64,
 
     /// The maximum number of entries per payload allowed to be transmitted during replication
     ///
     /// If this is too low, it will take longer for the nodes to be brought up to
     /// consistency with the rest of the cluster.
-    #[clap(long, env = "RAFT_MAX_PAYLOAD_ENTRIES", default_value = "300")]
+    #[clap(long, default_value = "300")]
     pub max_payload_entries: u64,
 
     /// The distance behind in log replication a follower must fall before it is considered lagging
     ///
     /// Once a replication stream transition into line-rate state, the target node will be considered safe to join a
     /// cluster.
-    #[clap(long, env = "RAFT_REPLICATION_LAG_THRESHOLD", default_value = "1000")]
+    #[clap(long, default_value = "1000")]
     pub replication_lag_threshold: u64,
 
     /// The snapshot policy to use for a Raft node.
     #[clap(
         long,
-        env = "RAFT_SNAPSHOT_POLICY",
         default_value = "since_last:5000",
         parse(try_from_str=parse_snapshot_policy)
     )]
     pub snapshot_policy: SnapshotPolicy,
 
-    /// Whether to keep `applied_log`s that are not included by snapshots.
-    ///
-    /// If your application may rebuild it's state machine from snapshots,
-    /// please set this to true.
-    ///
-    /// By default, `OpenRaft` purges `applied_log`s from time to time regardless of snapshots, because it assumes once
-    /// logs are `applied` to the state machine, logs are persisted on disk.
-    ///
-    /// If an implementation does not persist data when `RaftStorage::apply_to_state_machine()` returns, and just
-    /// relies on `snapshot` to rebuild the state machine when the next time it restarts, the application must always
-    /// set `keep_unsnapshoted_log` to `true`, so that only logs that are already included in a snapshot will be
-    /// purged.
-    //
-    // This is another way to implement:
-    // #[clap(long, env = "RAFT_KEEP_UNSNAPSHOTED_LOG",
-    //        default_value_t = false,
-    //        value_parser=clap::value_parser!(bool))]
-    #[clap(long, env = "RAFT_KEEP_UNSNAPSHOTED_LOG")]
-    pub keep_unsnapshoted_log: bool,
-
     /// The maximum snapshot chunk size allowed when transmitting snapshots (in bytes)
-    #[clap(long, env = "RAFT_SNAPSHOT_MAX_CHUNK_SIZE", default_value = "3MiB", parse(try_from_str=parse_bytes_with_unit))]
+    #[clap(long, default_value = "3MiB", parse(try_from_str=parse_bytes_with_unit))]
     pub snapshot_max_chunk_size: u64,
 
-    /// The maximum number of applied logs to keep before purging
-    #[clap(long, env = "RAFT_MAX_APPLIED_LOG_TO_KEEP", default_value = "1000")]
-    pub max_applied_log_to_keep: u64,
+    /// The maximum number of logs to keep that are already included in **snapshot**.
+    ///
+    /// Logs that are not in snapshot will never be purged.
+    #[clap(long, default_value = "1000")]
+    pub max_in_snapshot_log_to_keep: u64,
 
     /// The minimal number of applied logs to purge in a batch.
     #[clap(long, default_value = "1")]

openraft/src/config/config_test.rs

@@ -56,9 +56,8 @@ fn test_build() -> anyhow::Result<()> {
         "--max-payload-entries=201",
         "--replication-lag-threshold=202",
         "--snapshot-policy=since_last:203",
-        "--keep-unsnapshoted-log",
         "--snapshot-max-chunk-size=204",
-        "--max-applied-log-to-keep=205",
+        "--max-in-snapshot-log-to-keep=205",
         "--purge-batch-size=207",
     ])?;
 
@@ -70,25 +69,13 @@ fn test_build() -> anyhow::Result<()> {
     assert_eq!(201, config.max_payload_entries);
     assert_eq!(202, config.replication_lag_threshold);
     assert_eq!(SnapshotPolicy::LogsSinceLast(203), config.snapshot_policy);
-    assert_eq!(true, config.keep_unsnapshoted_log);
     assert_eq!(204, config.snapshot_max_chunk_size);
-    assert_eq!(205, config.max_applied_log_to_keep);
+    assert_eq!(205, config.max_in_snapshot_log_to_keep);
     assert_eq!(207, config.purge_batch_size);
 
     Ok(())
 }
 
-#[test]
-fn test_config_keep_unsnapshoted_log() -> anyhow::Result<()> {
-    let config = Config::build(&["foo", "--keep-unsnapshoted-log"])?;
-    assert_eq!(true, config.keep_unsnapshoted_log);
-
-    let config = Config::build(&["foo"])?;
-    assert_eq!(false, config.keep_unsnapshoted_log);
-
-    Ok(())
-}
-
 #[test]
 fn test_config_enable_tick() -> anyhow::Result<()> {
     let config = Config::build(&["foo", "--enable-tick=false"])?;

openraft/src/core/raft_core.rs

@@ -250,9 +250,8 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
         self.storage.save_vote(&state.vote).await?;
 
         self.engine = Engine::new(self.id, &state, EngineConfig {
-            max_applied_log_to_keep: self.config.max_applied_log_to_keep,
+            max_in_snapshot_log_to_keep: self.config.max_in_snapshot_log_to_keep,
             purge_batch_size: self.config.purge_batch_size,
-            keep_unsnapshoted_log: self.config.keep_unsnapshoted_log,
         });
 
         // Fetch the most recent snapshot in the system.
@@ -850,7 +849,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
         // TODO: add building-session id to identify different building
         match result {
             SnapshotResult::Ok(meta) => {
-                self.engine.update_snapshot(meta);
+                self.engine.finish_building_snapshot(meta);
                 self.run_engine_commands::<Entry<C>>(&[]).await?;
             }
             SnapshotResult::StorageError(sto_err) => {
@@ -868,6 +867,8 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
     /// If force is True, it will skip the threshold check and start creating snapshot as demanded.
     #[tracing::instrument(level = "debug", skip(self))]
     pub(crate) async fn trigger_log_compaction_if_needed(&mut self, force: bool) {
+        tracing::debug!("trigger_log_compaction_if_needed: force: {}", force);
+
         if let SnapshotState::None = self.snapshot_state {
             // Continue.
         } else {

openraft/src/core/replication.rs

@@ -4,6 +4,12 @@ pub(crate) fn snapshot_is_within_half_of_threshold(
     last_log_index: &u64,
     threshold: &u64,
 ) -> bool {
+    tracing::debug!(
+        "snapshot_last_indeix:{}, last_log_index:{}, threshold: {}",
+        snapshot_last_index,
+        last_log_index,
+        threshold
+    );
     // Calculate distance from actor's last log index.
     let distance_from_line = last_log_index.saturating_sub(*snapshot_last_index);
 

openraft/src/engine/calc_purge_upto_test.rs

@@ -24,7 +24,8 @@ fn eng() -> Engine<u64, ()> {
 
 #[test]
 fn test_calc_purge_upto() -> anyhow::Result<()> {
-    // last_purged_log_id, committed, max_keep, want
+    // last_purged_log_id, last_snapshot_log_id, max_keep, want
+    // last_applied should not affect the purge
     let cases = vec![
         //
         (None, None, 0, None),
@@ -53,71 +54,23 @@ fn test_calc_purge_upto() -> anyhow::Result<()> {
         (Some(log_id(1, 2)), Some(log_id(3, 4)), 5, None),
     ];
 
-    for (last_purged, committed, max_keep, want) in cases {
+    for (last_purged, snapshot_last_log_id, max_keep, want) in cases {
         let mut eng = eng();
-        eng.config.keep_unsnapshoted_log = false;
-        eng.config.max_applied_log_to_keep = max_keep;
+        eng.config.max_in_snapshot_log_to_keep = max_keep;
         eng.config.purge_batch_size = 1;
 
         if let Some(last_purged) = last_purged {
             eng.state.log_ids.purge(&last_purged);
         }
-        eng.state.committed = committed;
+        eng.snapshot_meta.last_log_id = snapshot_last_log_id;
         let got = eng.calc_purge_upto();
 
         assert_eq!(
             want, got,
-            "case: last_purged: {:?}, last_applied: {:?}, max_keep: {}",
-            last_purged, committed, max_keep
+            "case: last_purged: {:?}, snapshot_last_log_id: {:?}, max_keep: {}",
+            last_purged, snapshot_last_log_id, max_keep
         );
     }
 
     Ok(())
 }
-
-#[test]
-// in this test, keep_unsnapshoted_log is set to true.
-// logs being purged should at most the last that was in the snapshot.
-fn test_keep_unsnapshoted() -> anyhow::Result<()> {
-    let cases = vec![
-        // last_deleted, last_applied, last_snapshoted, max_keep, want
-        // empty test
-        (None, None, None, 0, None),
-        (None, None, None, 1, None),
-        // nothing in snapshot
-        (Some(log_id(1, 1)), Some(log_id(2, 2)), None, 0, None),
-        (Some(log_id(1, 1)), Some(log_id(5, 5)), None, 0, None),
-        // snapshot kept up
-        (None, Some(log_id(5, 5)), Some(log_id(3, 4)), 0, Some(log_id(3, 4))),
-        (
-            Some(log_id(1, 1)),
-            Some(log_id(5, 5)),
-            Some(log_id(5, 5)),
-            0,
-            Some(log_id(5, 5)),
-        ),
-    ];
-
-    for (purged, committed, snapshot, keep, want) in cases {
-        let mut eng = eng();
-        eng.config.keep_unsnapshoted_log = true;
-        eng.config.max_applied_log_to_keep = keep;
-        eng.config.purge_batch_size = 1;
-
-        if let Some(last_purged) = purged {
-            eng.state.log_ids.purge(&last_purged);
-        }
-        eng.state.committed = committed;
-        eng.snapshot_meta.last_log_id = snapshot;
-
-        let got = eng.calc_purge_upto();
-
-        assert_eq!(
-            want, got,
-            "case: purged: {:?}, committed: {:?}, snapshot: {:?}, keep: {}",
-            purged, committed, snapshot, keep
-        )
-    }
-
-    Ok(())
-}

openraft/src/engine/engine_impl.rs

@@ -33,22 +33,17 @@ use crate::Vote;
 #[derive(PartialEq, Eq)]
 pub(crate) struct EngineConfig {
     /// The maximum number of applied logs to keep before purging.
-    pub(crate) max_applied_log_to_keep: u64,
+    pub(crate) max_in_snapshot_log_to_keep: u64,
 
     /// The minimal number of applied logs to purge in a batch.
     pub(crate) purge_batch_size: u64,
-
-    /// whether to keep applied log that are not included in snapshots.
-    /// false by default
-    pub(crate) keep_unsnapshoted_log: bool,
 }
 
 impl Default for EngineConfig {
     fn default() -> Self {
         Self {
-            max_applied_log_to_keep: 1000,
+            max_in_snapshot_log_to_keep: 1000,
             purge_batch_size: 256,
-            keep_unsnapshoted_log: false,
         }
     }
 }
@@ -433,7 +428,6 @@ where
                 already_committed: prev_committed,
                 upto: committed.unwrap(),
             });
-            self.purge_applied_log();
         }
     }
 
@@ -540,13 +534,13 @@ where
         }
     }
 
-    /// Purge applied log if needed.
+    /// Purge logs that are already in snapshot if needed.
     ///
-    /// `max_applied_log_to_keep` specifies the number of applied logs to keep.
-    /// `max_applied_log_to_keep==0` means every applied log can be purged.
+    /// `max_in_snapshot_log_to_keep` specifies the number of logs already included in snapshot to keep.
+    /// `max_in_snapshot_log_to_keep==0` means to purge every log stored in snapshot.
     // NOTE: simple method, not tested.
     #[tracing::instrument(level = "debug", skip_all)]
-    pub(crate) fn purge_applied_log(&mut self) {
+    pub(crate) fn purge_in_snapshot_log(&mut self) {
         if let Some(purge_upto) = self.calc_purge_upto() {
             self.purge_log(purge_upto);
         }
@@ -562,28 +556,21 @@ where
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) fn calc_purge_upto(&mut self) -> Option<LogId<NID>> {
         let st = &self.state;
-        let last_applied = &st.committed;
-        let max_keep = self.config.max_applied_log_to_keep;
+        let max_keep = self.config.max_in_snapshot_log_to_keep;
         let batch_size = self.config.purge_batch_size;
 
-        let mut purge_end = last_applied.next_index().saturating_sub(max_keep);
-
-        if self.config.keep_unsnapshoted_log {
-            let idx = self.snapshot_meta.last_log_id.next_index();
-            tracing::debug!("the very last log included in snapshots: {}", idx);
-            purge_end = idx.min(purge_end);
-        }
+        let purge_end = self.snapshot_meta.last_log_id.next_index().saturating_sub(max_keep);
 
         tracing::debug!(
-            last_applied = debug(last_applied),
+            snapshot_last_log_id = debug(self.snapshot_meta.last_log_id),
             max_keep,
             "try purge: (-oo, {})",
             purge_end
         );
 
         if st.last_purged_log_id().next_index() + batch_size > purge_end {
             tracing::debug!(
-                last_applied = debug(last_applied),
+                snapshot_last_log_id = debug(self.snapshot_meta.last_log_id),
                 max_keep,
                 last_purged_log_id = display(st.last_purged_log_id().summary()),
                 batch_size,
@@ -640,6 +627,7 @@ where
         //   membership.log_id       = (10, 5);
         //   local_effective.log_id = (2, 10);
         if effective.log_id.index() <= m.log_id.index() {
+            // TODO: if effective membership changes, call `update_repliation()`
             effective = m;
         }
 
@@ -743,7 +731,6 @@ where
                 already_committed: prev_committed,
                 upto: self.state.committed.unwrap(),
             });
-            self.purge_applied_log();
         }
     }
 
@@ -801,6 +788,18 @@ where
         self.purge_log(snap_last_log_id)
     }
 
+    #[tracing::instrument(level = "debug", skip_all)]
+    pub(crate) fn finish_building_snapshot(&mut self, meta: SnapshotMeta<NID, N>) {
+        tracing::info!("finish_building_snapshot: {:?}", meta);
+
+        let updated = self.update_snapshot(meta);
+        if !updated {
+            return;
+        }
+
+        self.purge_in_snapshot_log();
+    }
+
     /// Update engine state when a new snapshot is built or installed.
     ///
     /// Engine records only the metadata of a snapshot. Snapshot data is stored by RaftStorage implementation.