raftstore: fix failing to restart after tikv OOM crash

components/engine_rocks/src/options.rs

@@ -17,6 +17,7 @@ impl From<engine_traits::ReadOptions> for RocksReadOptions {
  fn from(opts: engine_traits::ReadOptions) -> Self {
  let mut r = RawReadOptions::default();
  r.fill_cache(opts.fill_cache());
+ r.set_read_tier(opts.read_tier() as i32);
  RocksReadOptions(r)
  }
 }

components/engine_traits/src/options.rs

@@ -2,9 +2,18 @@
 use std::ops::Bound;
 use tikv_util::keybuilder::KeyBuilder;
 
+#[repr(i32)]
+#[derive(Clone, Copy)]
+pub enum ReadTier {
+ ReadAllTier = 0,
+ BlockCacheTier = 1,
+ PersistedTier = 2,
+}
+
 #[derive(Clone)]
 pub struct ReadOptions {
  fill_cache: bool,
+ read_tier: ReadTier,
 }
 
 impl ReadOptions {
@@ -21,11 +30,24 @@ impl ReadOptions {
  pub fn set_fill_cache(&mut self, v: bool) {
  self.fill_cache = v;
  }
+
+ #[inline]
+ pub fn read_tier(&self) -> ReadTier {
+ self.read_tier
+ }
+
+ #[inline]
+ pub fn set_read_tier(&mut self, v: ReadTier) {
+ self.read_tier = v;
+ }
 }
 
 impl Default for ReadOptions {
  fn default() -> ReadOptions {
- ReadOptions { fill_cache: true }
+ ReadOptions {
+ fill_cache: true,
+ read_tier: ReadTier::ReadAllTier, // all tier
+ }
  }
 }
 

components/raftstore/src/store/fsm/peer.rs

@@ -11,7 +11,8 @@ use batch_system::{BasicMailbox, Fsm};
 use collections::HashMap;
 use engine_traits::CF_RAFT;
 use engine_traits::{
- Engines, KvEngine, RaftEngine, RaftLogBatch, SSTMetaInfo, WriteBatch, WriteBatchExt,
+ Engines, KvEngine, RaftEngine, RaftLogBatch, ReadOptions, ReadTier, SSTMetaInfo, WriteBatch,
+ WriteBatchExt,
 };
 use error_code::ErrorCodeExt;
 use fail::fail_point;
@@ -131,6 +132,9 @@ where
  max_inflight_msgs: usize,
 
  trace: PeerMemoryTrace,
+
+ // read applied idx from sst for GC
+ check_truncated_idx_for_gc: bool,
 }
 
 pub struct BatchRaftCmdRequestBuilder<E>
@@ -231,6 +235,7 @@ where
  ),
  max_inflight_msgs: cfg.raft_max_inflight_msgs,
  trace: PeerMemoryTrace::default(),
+ check_truncated_idx_for_gc: cfg.disable_kv_wal,
  }),
  ))
  }
@@ -276,6 +281,7 @@ where
  ),
  max_inflight_msgs: cfg.raft_max_inflight_msgs,
  trace: PeerMemoryTrace::default(),
+ check_truncated_idx_for_gc: cfg.disable_kv_wal,
  }),
  ))
  }
@@ -2212,18 +2218,50 @@ where
  }
  }
 
+ fn get_flushed_truncated_idx(&mut self, region_id: u64) -> u64 {
+ let state_key = keys::apply_state_key(region_id);
+ let mut opts = ReadOptions::new();
+ opts.set_read_tier(ReadTier::PersistedTier);
+ let value = self
+ .ctx
+ .engines
+ .kv
+ .get_value_cf_opt(&opts, CF_RAFT, &state_key)
+ .unwrap();
+ if value.is_none() {
+ return 0;
+ }
+ let mut m = RaftApplyState::default();
+ m.merge_from_bytes(&value.unwrap()).unwrap();
+ cmp::min(m.get_truncated_state().get_index(), m.applied_index)
+ }
+
  fn on_ready_compact_log(&mut self, first_index: u64, state: RaftTruncatedState) {
  let total_cnt = self.fsm.peer.last_applying_idx - first_index;
+
+ let mut index = state.get_index();
+ if self.fsm.check_truncated_idx_for_gc {
+ let last_truncated_idx =
+ self.get_flushed_truncated_idx(self.fsm.peer.get_store().get_region_id());
+ if last_truncated_idx != 0 && last_truncated_idx - 1 < index {
+ debug!("on_ready_compact_log update index.";
+ "region" => self.fsm.peer.get_store().get_region_id(),
+ "original" => index,
+ "new value" => last_truncated_idx-1);
+ index = last_truncated_idx - 1;
+ }
+ }
  // the size of current CompactLog command can be ignored.
  let remain_cnt = self.fsm.peer.last_applying_idx - state.get_index() - 1;
  self.fsm.peer.raft_log_size_hint =
  self.fsm.peer.raft_log_size_hint * remain_cnt / total_cnt;
- let compact_to = state.get_index() + 1;
+ let compact_to = index + 1;
  let task = RaftlogGcTask::gc(
  self.fsm.peer.get_store().get_region_id(),
  self.fsm.peer.last_compacted_idx,
  compact_to,
  );
+ let start_idx = self.fsm.peer.last_compacted_idx;
  self.fsm.peer.last_compacted_idx = compact_to;
  self.fsm.peer.mut_store().compact_to(compact_to);
  if let Err(e) = self.ctx.raftlog_gc_scheduler.schedule(task) {
@@ -2233,6 +2271,14 @@ where
  "peer_id" => self.fsm.peer_id(),
  "err" => %e,
  );
+ } else {
+ debug!(
+ "successfully to schedule compact task";
+ "region_id" => self.fsm.region_id(),
+ "peer_id" => self.fsm.peer_id(),
+ "start_idx" => start_idx,
+ "end_idx" => compact_to,
+ );
  }
  }
 

components/raftstore/src/store/mod.rs

@@ -49,9 +49,9 @@ pub use self::region_snapshot::{RegionIterator, RegionSnapshot};
 pub use self::replication_mode::{GlobalReplicationState, StoreGroup};
 pub use self::snap::{
  check_abort, copy_snapshot,
- snap_io::{apply_sst_cf_file, build_sst_cf_file},
+ snap_io::{apply_sst_cf_file, build_sst_cf_file_list},
  ApplyOptions, Error as SnapError, SnapEntry, SnapKey, SnapManager, SnapManagerBuilder,
- Snapshot, SnapshotStatistics,
+ Snapshot, SnapshotStatistics,CfFile,
 };
 pub use self::transport::{CasualRouter, ProposalRouter, StoreRouter, Transport};
 pub use self::util::{RegionReadProgress, RegionReadProgressRegistry};