// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. use std::{ cmp, collections::VecDeque, mem, ops::{Deref, DerefMut}, sync::{Arc, atomic::Ordering}, time::Instant as StdInstant, }; use api_version::api_v2::is_whole_keyspace_range; use bytes::{Buf, Bytes}; use error_code::ErrorCodeExt; use fail::fail_point; use kvengine::{ CheckMergeResult, DEL_PREFIXES_KEY, IdVer, LARGE_NUM_COLUMNAR_TABLES_IN_SHARD, MANUAL_MAJOR_COMPACTION, MANUAL_MAJOR_COMPACTION_DISABLE, MANUAL_MAJOR_COMPACTION_ENABLE, MANUAL_MAJOR_COMPACTION_ENABLE_COLUMNAR, Shard, ShardRange, TERM_KEY, WRITE_CF_BOTTOM_LEVEL, ia::ia_auto_file::report_transitions, table::{ DataBound, InnerKey, fts::{TrackedFtsIndexMap, new_legacy_tracked_index_meta}, schema_file::SchemaFile, }, table_id::{has_consistent_tracked_fts_indexes, is_table_boundary_key}, }; use kvproto::{ import_sstpb::SwitchMode, kvrpcpb::{self, KeyRange}, metapb::{self, Region, RegionEpoch}, pdpb::{CheckPolicy, SplitReason}, raft_cmdpb::{ AdminCmdType, AdminRequest, CmdType, RaftCmdRequest, RaftCmdResponse, RaftRequestHeader, Request, StatusCmdType, StatusResponse, }, raft_serverpb::{ExtraMessageType, PeerState, RaftMessage}, }; use parking_lot::RwLockWriteGuard; use pd_client::BucketMeta; use protobuf::Message; use raft::{self, GetEntriesContext, Storage, eraftpb::MessageType}; use raft_proto::eraftpb; use raftstore::{ coprocessor::RegionChangeEvent, store::{ LocksStatus, fsm::ChangeObserver, metrics::LOAD_BASE_SPLIT_EVENT, util, worker_metrics::CHECK_SPLIT_COUNTER, }, }; use tikv_util::{ GLOBAL_SERVER_READINESS, box_err, codec::bytes::{decode_bytes, encode_bytes, encode_bytes_maybe_empty}, debug, error, info, metrics::{ServerIsBusyType, new_server_is_busy}, spawn_anonymous_thread_with, store::{find_peer, is_learner, region_on_same_stores}, sys::thread::StdThreadBuildWrapper, trace, txn_debug, warn, worker::{ScheduleError, Scheduler}, }; use trace_event::types::TraceContext; use tracker::GLOBAL_TRACKERS; use txn_types::{Key, TimeStamp, WriteBatchFlags}; use uuid::Uuid; use super::{ CustomRaftLog, RequestInspector, ResolveScheduleCheck, ResolveScheduleOutcome, ResolveTxnFileLocksTask, SchemaTask, StartupPeerState, WorkerType, }; use crate::{ DiscardReason, Error, MERGE_REGION_WITH_TXN_FILE_LOCKS_ERR_MSG, RaftStoreRouter, Result, store::{ ApplyMetrics, ApplyMsg, CasualMessage, ColdCampaignReason, ColdLeaderProbeTask, ColdRegionPolicy, ColdStoreLiveness, Config, CustomBuilder, CustomRaftLogType, Engines, MsgApplyResult, PEER_TICK_CHECK_LONG, PEER_TICK_FILE_GC_COMPACTION_CHECK, PEER_TICK_MAINTENANCE, PEER_TICK_PD_HEARTBEAT, PEER_TICK_RAFT, PEER_TICK_RAFT_LOG_GC, PEER_TICK_REACTIVATE_MEMORY_LOCK, PEER_TICK_SPLIT_CHECK, PdTask, PeerMsg, PersistReady, RAFT_INIT_LOG_INDEX, RaftApplyState, RaftCommand, RaftContext, ReadIndexContext, SignificantMsg, SnapState, SplitCheckScheduleOutcome, SplitCheckTask, SplitSource, StoreMeta, StoreMsg, Ticker, TrimOverBoundParameter, cmd_resp::{bind_term, message_error, new_error, new_with_key_error}, ingest::convert_sst, load_last_peer_state, metrics::{ COLD_CAMPAIGN_EVENT_COUNTER_VEC, INGEST_COMPACTION_PROPOSAL_FAILED_COUNTER, INGEST_TRIGGER_COMPACTION_COUNTER, STORE_INGEST_CONVERT_TASK_STATUS, TXN_FILE_ACTIVE_RESOLVE_SCHEDULE_TOTAL, record_txn_file_split_denied, }, msg::Callback, notify_req_region_removed, peer::{ColdPeerState, Peer, StaleState, TRANSFER_LEADER_COMMAND_REPLY_CTX}, peer_storage::RaftlogFetchResult, schema::{schema_file_is_matched_with_meta, shard_is_matched_with_meta}, util::{self as _util, KeysInfoFormatter}, write_engine_meta, write_engine_meta_diff, }, }; /// Limits the maximum number of regions returned by error. /// /// Another choice is using coprocessor batch limit, but 10 should be a good fit /// in most case. const _MAX_REGIONS_IN_ERROR: usize = 10; // Bound each peer's tick to about 8K tombstones because every version can have // both a raft state key and a region state key. Later ticks continue the GC. const OBSOLETE_VERSION_GC_BATCH_SIZE: usize = 4096; fn tracked_fts_indexes_from_snapshot(snapshot: &kvenginepb::Snapshot) -> TrackedFtsIndexMap { let mut tracked_indexes = TrackedFtsIndexMap::default(); for idx in snapshot.get_fts_index_metas() { let meta = idx.clone(); tracked_indexes .entry(meta.get_table_id()) .or_default() .insert(meta.get_index_id(), meta); } for idx in snapshot.get_fts_indexes() { tracked_indexes .entry(idx.get_table_id()) .or_default() .entry(idx.get_index_id()) .or_insert_with(|| { new_legacy_tracked_index_meta(idx.get_table_id(), idx.get_index_id()) }); } tracked_indexes } fn tracked_fts_indexes_from_shard_meta(shard_meta: &kvengine::ShardMeta) -> TrackedFtsIndexMap { let mut tracked_indexes = TrackedFtsIndexMap::default(); for indexes in shard_meta.fts_index_metas.values() { for meta in indexes.values() { let meta = meta.clone(); tracked_indexes .entry(meta.get_table_id()) .or_default() .insert(meta.get_index_id(), meta); } } for &(table_id, index_id) in &shard_meta.fts_indexes { tracked_indexes .entry(table_id) .or_default() .entry(index_id) .or_insert_with(|| new_legacy_tracked_index_meta(table_id, index_id)); } tracked_indexes } fn new_read_index_request( region_id: u64, region_epoch: RegionEpoch, peer: metapb::Peer, ) -> RaftCmdRequest { let mut header = RaftRequestHeader::default(); header.set_region_id(region_id); header.set_region_epoch(region_epoch); header.set_peer(peer); let mut request = Request::default(); request.set_cmd_type(CmdType::ReadIndex); let mut cmd = RaftCmdRequest::default(); cmd.set_header(header); cmd.mut_requests().push(request); cmd } fn intersect_raw_range(region: &Region, raw_range: &KeyRange) -> Option<(Vec, Vec)> { let encoded_start = encode_bytes_maybe_empty(raw_range.get_start_key()); let encoded_end = encode_bytes_maybe_empty(raw_range.get_end_key()); let start = cmp::max(encoded_start.as_slice(), region.get_start_key()).to_vec(); let end = if encoded_end.is_empty() { region.get_end_key().to_vec() } else if region.get_end_key().is_empty() { encoded_end } else { cmp::min(encoded_end.as_slice(), region.get_end_key()).to_vec() }; (end.is_empty() || start < end).then_some((start, end)) } #[derive(Debug, PartialEq, Eq)] enum BucketSplitDecision { SplitKey(Vec), CheckSst { start_key: Vec, end_key: Vec, }, NoCandidate, } fn decode_region_key(encoded_key: &[u8]) -> Option> { if encoded_key.is_empty() { return Some(Vec::new()); } let mut input = encoded_key; let key = decode_bytes(&mut input, false).ok()?; input.is_empty().then_some(key) } fn decide_bucket_split( region: &Region, bucket_meta: Option<&BucketMeta>, raw_range: &KeyRange, ) -> BucketSplitDecision { if !raw_range.get_start_key().is_empty() && raw_range.get_start_key() == raw_range.get_end_key() { return BucketSplitDecision::NoCandidate; } // A range recorded before the PD refresh may no longer overlap this // Region. Reject it before selecting from current buckets or SSTs. let Some((start, end)) = intersect_raw_range(region, raw_range) else { return BucketSplitDecision::NoCandidate; }; let Some(bucket_meta) = bucket_meta else { return check_sst_for_region(region); }; let keys = &bucket_meta.keys; if bucket_meta.region_id != region.get_id() || bucket_meta.region_epoch.get_version() != region.get_region_epoch().get_version() || keys.len() < 2 || keys.first().map(Vec::as_slice) != Some(region.get_start_key()) || keys.last().map(Vec::as_slice) != Some(region.get_end_key()) { return check_sst_for_region(region); } let ordered_before_end = keys[..keys.len() - 1] .windows(2) .all(|pair| pair[0] < pair[1]); let end_is_ordered = keys.last().is_some_and(Vec::is_empty) || keys[keys.len() - 2] < keys[keys.len() - 1]; if !ordered_before_end || !end_is_ordered { return check_sst_for_region(region); } // The first and last entries are Region bounds, not split candidates. let internal_keys = &keys[1..keys.len() - 1]; if end.is_empty() || start < end { let first = internal_keys.partition_point(|key| key.as_slice() <= start.as_slice()); let last = if end.is_empty() { internal_keys.len() } else { internal_keys.partition_point(|key| key.as_slice() < end.as_slice()) }; if first < last { return BucketSplitDecision::SplitKey( internal_keys[first + (last - first) / 2].clone(), ); } } // The Region is already CPU hot. If its preferred request span is too // narrow, use the middle current bucket boundary rather than skipping it. if let Some(split_key) = internal_keys.get(internal_keys.len() / 2) { return BucketSplitDecision::SplitKey(split_key.clone()); } check_sst_for_region(region) } fn check_sst_for_region(region: &Region) -> BucketSplitDecision { let (Some(start_key), Some(end_key)) = ( decode_region_key(region.get_start_key()), decode_region_key(region.get_end_key()), ) else { return BucketSplitDecision::NoCandidate; }; BucketSplitDecision::CheckSst { start_key, end_key } } fn is_read_only_raft_command(msg: &RaftCmdRequest) -> bool { !msg.get_requests().is_empty() && msg.get_requests().iter().all(|req| { matches!( req.get_cmd_type(), CmdType::Get | CmdType::Snap | CmdType::ReadIndex ) }) } // Keep partial single-keyspace ranges active because restore and ingest may // pre-split them while they are still empty. fn is_cold_eligible_range(range: &ShardRange) -> bool { is_whole_keyspace_range(&range.outer_start, &range.outer_end) || range.keyspace_id != range.get_end_keyspace_id() } // Cheap data-only filter run before the remaining cold safety checks. fn is_cold_data_candidate( range: &ShardRange, max_ts: u64, estimated_size: u64, max_ts_threshold_ms: u64, now_physical_ms: u64, ) -> bool { if !is_cold_eligible_range(range) { return false; } if max_ts == 0 && estimated_size == 0 { return true; } TimeStamp::new(max_ts) .physical() .saturating_add(max_ts_threshold_ms) < now_physical_ms } fn cold_state_reason(state: ColdPeerState) -> &'static str { match state { ColdPeerState::Regular => "ineligible", ColdPeerState::TempRegular => "temporary", ColdPeerState::Cold => "eligible", } } fn record_cold_campaign_attempt(reason: ColdCampaignReason) { let counters = match reason { ColdCampaignReason::ReadIndexTimeout => &COLD_CAMPAIGN_EVENT_COUNTER_VEC.read_index, ColdCampaignReason::LeaderUnknown => &COLD_CAMPAIGN_EVENT_COUNTER_VEC.leader_unknown, ColdCampaignReason::LeaderStoreDown => &COLD_CAMPAIGN_EVENT_COUNTER_VEC.store_down, }; counters.inc(); } fn should_send_cold_leader_heartbeat(ctx: &RaftContext, region_id: u64, sweep_id: u64) -> bool { let cold_tick_interval_ms = ctx .cfg .cold_region .region_heartbeat_interval .as_millis() .max(1); let phase_count = ctx .cfg .cold_region .leader_heartbeat_interval .as_millis() .div_ceil(cold_tick_interval_ms) .max(1); crc32c::crc32c(®ion_id.to_be_bytes()) as u64 % phase_count == sweep_id % phase_count } pub struct PeerFsm { pub(crate) peer: Peer, pub(crate) stopped: bool, pub(crate) ticker: Ticker, pub(crate) startup_state: Arc, /// Counter for reactivating in-memory pessimistic locks after leader /// transfer fails. pub(crate) reactivate_memory_lock_ticks: usize, cold_leader_probe_task: Option, } impl PeerFsm { // If we create the peer actively, like bootstrap/split/merge region, we should // use this function to create the peer. The region must contain the peer info // for this store. pub fn create( store_id: u64, cfg: &Config, engines: Engines, region: &metapb::Region, read_scheduler: Scheduler, ) -> Result { let meta_peer = match find_peer(region, store_id) { None => { return Err(box_err!( "find no peer for store {} in region {:?}", store_id, region )); } Some(peer) => peer.clone(), }; let peer = Peer::new(store_id, cfg, engines, region, meta_peer, read_scheduler)?; info!( "create peer"; "region" => peer.tag(), "peer_id" => peer.peer_id(), ); Ok(PeerFsm { peer, stopped: false, reactivate_memory_lock_ticks: 0, ticker: Ticker::new(cfg), startup_state: Arc::new(StartupPeerState::default()), cold_leader_probe_task: None, }) } // The peer can be created from another node with raft membership changes, and // we only know the region_id and peer_id when creating this replicated // peer, the region info will be retrieved later after restored snapshot. pub fn replicate( store_id: u64, cfg: &Config, engines: Engines, region_id: u64, region_epoch: metapb::RegionEpoch, peer: metapb::Peer, read_scheduler: Scheduler, ) -> Result { // We will remove tombstone key when apply snapshot let mut region = metapb::Region::default(); region.set_id(region_id); // Peer state key contains version, so that we have to set it to prevent // destroyed uninitialized peer from being created by raft msg again. region.set_region_epoch(region_epoch); let peer = Peer::new(store_id, cfg, engines, ®ion, peer, read_scheduler)?; info!( "replicate peer"; "region" => peer.tag(), "peer_id" => peer.peer_id(), ); Ok(PeerFsm { peer, stopped: false, reactivate_memory_lock_ticks: 0, ticker: Ticker::new(cfg), startup_state: Arc::new(StartupPeerState::default()), cold_leader_probe_task: None, }) } #[inline] pub fn region_id(&self) -> u64 { self.peer.region().get_id() } #[inline] pub(crate) fn get_peer(&self) -> &Peer { &self.peer } #[inline] pub fn peer_id(&self) -> u64 { self.peer.peer_id() } #[inline] pub fn stop(&mut self) { self.peer.clear_cold_state_for_removal(); self.cold_leader_probe_task = None; self.stopped = true; } pub(crate) fn maybe_report_startup_connected(&self, ctx: &mut RaftContext) { let connected = self.peer.is_cold() || self.peer.is_leader() || self.peer.leader_id() != raft::INVALID_ID; if GLOBAL_SERVER_READINESS .startup_peer_activation_completed .load(Ordering::Relaxed) || !connected || !self.startup_state.maybe_mark_connected() { return; } ctx.global .router .send_store(StoreMsg::PeerStartupConnected(self.region_id())); } pub(crate) fn refresh_cold_state(&mut self, ctx: &RaftContext) { let state = self.evaluate_cold_state(ctx); self.set_cold_state(ctx, state, cold_state_reason(state)); } fn initialize_cold_state(&mut self, ctx: &RaftContext) { let state = if ctx.cfg.cold_region.policy == ColdRegionPolicy::Disabled { ColdPeerState::Regular } else if let Some(shard) = ctx.global.engines.kv.get_shard(self.region_id()) { let estimated_size = shard.get_estimated_size(); if self.is_cold_candidate_with_shard(ctx, shard.as_ref(), estimated_size) && !self.peer.has_unexpired_vote_lease() { // election_elapsed is runtime-only. Expiring it does not // change vote handling when no recent leader is known, and // avoids activating every restored cold peer after restart. self.peer.saturate_election_elapsed(); } self.evaluate_cold_state_with_shard(ctx, shard.as_ref(), estimated_size) } else { ColdPeerState::Regular }; self.set_cold_state(ctx, state, cold_state_reason(state)); } fn update_cold_state_with_shard( &mut self, ctx: &RaftContext, shard: &Shard, estimated_size: u64, ) -> bool { let state = self.evaluate_cold_state_with_shard(ctx, shard, estimated_size); self.set_cold_state(ctx, state, cold_state_reason(state)); state == ColdPeerState::Cold } fn set_cold_state( &mut self, ctx: &RaftContext, state: ColdPeerState, reason: &'static str, ) -> bool { let previous_state = self.peer.cold_state(); self.peer.set_cold_state(state, reason); if previous_state != ColdPeerState::Cold && state == ColdPeerState::Cold && self.peer.take_region_meta_heartbeat_pending() && self.peer.is_leader() { self.peer.heartbeat_pd(ctx, true); } if previous_state != state { ctx.global.coprocessor_host.on_region_changed( self.peer.region(), RegionChangeEvent::UpdateColdState(state == ColdPeerState::Cold), self.peer.get_role(), ); } if state != ColdPeerState::Cold { self.cold_leader_probe_task = None; } if previous_state == ColdPeerState::Cold && state != ColdPeerState::Cold { self.ticker.reschedule_elapsed_peer_ticks(); } previous_state == ColdPeerState::TempRegular && state == ColdPeerState::Cold } pub(crate) fn cold_read_timeout_after(&self, cfg: &Config) -> Option { if !self.peer.is_cold() || self.peer.is_leader() || !self.peer.has_pending_reads() { return None; } self.peer .pending_read_timeout_after(cfg.raft_election_timeout()) } fn start_cold_leader_probe( &mut self, ctx: &mut RaftContext, campaign_reason: ColdCampaignReason, ) -> bool { if self.cold_leader_probe_task.is_some() || !self.peer.is_cold() || self.peer.is_leader() { return false; } let leader_id = self.peer.leader_id(); let Some(probe_id) = self.peer.send_cold_leader_probe(ctx) else { return false; }; let raft = &mut self.peer.raft_group.raft; raft.reset_randomized_election_timeout(); let timeout = ctx.cfg.raft_base_tick_interval.0 * raft.randomized_election_timeout() as u32; self.cold_leader_probe_task = Some(ColdLeaderProbeTask { deadline: StdInstant::now() + timeout, probe_id, leader_id, campaign_reason, }); true } fn observe_cold_leader_probe_response( &mut self, probe_id: Uuid, leader_id: u64, response_term: u64, ) -> bool { if self .cold_leader_probe_task .as_ref() .is_none_or(|task| task.probe_id != probe_id) || !self .peer .restore_cold_leader_from_probe(leader_id, response_term) { return false; } self.cold_leader_probe_task = None; true } pub(crate) fn cold_leader_probe_task( &mut self, liveness: &ColdStoreLiveness, ) -> Option { let task = self.cold_leader_probe_task?; if !self.peer.is_cold() || self.peer.is_leader() { self.cold_leader_probe_task = None; return None; } let leader_id = self.peer.leader_id(); let leader_unchanged = leader_id == raft::INVALID_ID || leader_id == task.leader_id; let reason_still_valid = match task.campaign_reason { ColdCampaignReason::LeaderStoreDown => self .cold_cached_leader_store_id() .is_some_and(|store_id| liveness.is_down(store_id)), ColdCampaignReason::LeaderUnknown | ColdCampaignReason::ReadIndexTimeout => true, }; if leader_unchanged && reason_still_valid { return Some(task); } self.cold_leader_probe_task = None; None } fn cold_tick_campaign_reason( &self, liveness: &ColdStoreLiveness, ) -> Option { if self.peer.leader_id() == raft::INVALID_ID { return Some(ColdCampaignReason::LeaderUnknown); } self.cold_cached_leader_store_id() .filter(|store_id| liveness.is_down(*store_id)) .map(|_| ColdCampaignReason::LeaderStoreDown) } fn evaluate_cold_state(&self, ctx: &RaftContext) -> ColdPeerState { if ctx.cfg.cold_region.policy == ColdRegionPolicy::Disabled { return ColdPeerState::Regular; } let Some(shard) = ctx.global.engines.kv.get_shard(self.region_id()) else { return ColdPeerState::Regular; }; let estimated_size = shard.get_estimated_size(); self.evaluate_cold_state_with_shard(ctx, shard.as_ref(), estimated_size) } fn has_cold_progress_obligation(&self) -> bool { let store = self.peer.get_store(); self.peer.need_campaign || self.peer.is_applying_snapshot() || self.peer.has_pending_snapshot() || self.peer.has_cold_tick_obligation() || store.snap_state != SnapState::Relax || !store.snapshot_not_ready_peers.borrow().is_empty() || store.applied_index() != store.commit_index() || (self.peer.is_leader() && store.commit_index() != store.last_index()) } fn is_cold_candidate_with_shard( &self, ctx: &RaftContext, shard: &Shard, estimated_size: u64, ) -> bool { ctx.cfg .cold_region .policy .allows_cold(self.peer.is_leader()) && is_cold_data_candidate( &shard.range, shard.get_max_ts(), estimated_size, ctx.cfg.cold_region.max_ts_threshold.as_millis(), TimeStamp::physical_now(), ) && !self.stopped && !self.startup_state.is_inactive() && !self.peer.pending_remove && self.peer.is_initialized() && shard.get_initial_flushed() && shard.data_all_persisted() } fn evaluate_cold_state_with_shard( &self, ctx: &RaftContext, shard: &Shard, estimated_size: u64, ) -> ColdPeerState { if matches!( self.peer.get_role(), raft::StateRole::PreCandidate | raft::StateRole::Candidate ) { return ColdPeerState::Regular; } if !self.is_cold_candidate_with_shard(ctx, shard, estimated_size) { return ColdPeerState::Regular; } // Lock reactivation needs regular scheduling, but not a Raft tick. if self.has_cold_progress_obligation() || self.peer.txn_ext.pessimistic_locks.read().status == LocksStatus::TransferringLeader { return ColdPeerState::TempRegular; } if !self.peer.is_election_elapsed_saturated() { return ColdPeerState::TempRegular; } ColdPeerState::Cold } pub(crate) fn cold_cached_leader_store_id(&self) -> Option { if !self.peer.is_cold() || self.peer.is_leader() { return None; } self.peer .get_peer_from_cache(self.peer.leader_id()) .map(|peer| peer.get_store_id()) } } pub(crate) struct PeerMsgHandler<'a> { pub(crate) fsm: &'a mut PeerFsm, pub(crate) ctx: &'a mut RaftContext, } impl Deref for PeerMsgHandler<'_> { type Target = PeerFsm; fn deref(&self) -> &Self::Target { self.fsm } } impl DerefMut for PeerMsgHandler<'_> { fn deref_mut(&mut self) -> &mut Self::Target { self.fsm } } /// A file-GC compaction candidate located by the leader tick. enum FileGcCandidate { /// L1+ candidate at this level number. L1Plus(usize), /// L0 candidate with this file id. L0(u64), } impl<'a> PeerMsgHandler<'a> { pub(crate) fn new(fsm: &'a mut PeerFsm, ctx: &'a mut RaftContext) -> PeerMsgHandler<'a> { PeerMsgHandler { fsm, ctx } } #[allow(clippy::vec_box)] pub fn handle_msgs(&mut self, msgs: &mut Vec>) { for m in msgs.drain(..) { match *m { PeerMsg::RaftMessage(msg) => { let msg_type = msg.get_message().get_msg_type(); let from_store = msg.get_from_peer().get_store_id(); if let Err(e) = self.on_raft_message(msg) { error!(%e; "handle raft message err"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "type" => ?msg_type, "from_store" => from_store, ); } } PeerMsg::RaftCommand(cmd) => { let propose_time = cmd.send_time.saturating_elapsed(); self.ctx .raft_metrics .propose_wait_time .observe(propose_time.as_secs_f64()); // Waterfall metrics: record when this read-only raft command finishes waiting // and is ready to be proposed. if self.ctx.raft_metrics.waterfall_metrics { cmd.callback.read_tracker().map(|tracker| { GLOBAL_TRACKERS.with_tracker(tracker, |t| { t.metrics.read_index_propose_wait_nanos = propose_time.as_nanos() as u64; }) }); } self.propose_raft_command(cmd.request, cmd.trace_ctx, cmd.callback, None); } PeerMsg::Tick => self.on_tick(), PeerMsg::ColdRegionTick(sweep_id) => self.on_cold_region_tick(sweep_id), PeerMsg::ColdReadIndexTimeout => self.on_cold_read_index_timeout(), PeerMsg::ColdLeaderProbeTimeout => self.on_cold_leader_probe_timeout(), PeerMsg::ApplyResult(res) => { self.on_apply_result(res); } PeerMsg::SignificantMsg(msg) => self.on_significant_msg(msg), PeerMsg::CasualMessage(msg) => self.on_casual_msg(msg), PeerMsg::Start => self.start(), PeerMsg::GenerateEngineChangeSet(cs, cb) => { self.on_generate_engine_change_set(cs, cb) } PeerMsg::ApplySnapshotResult(cs) => { self.on_apply_snapshot_result(cs); } PeerMsg::Persisted(ready) => { self.on_persisted(ready); } PeerMsg::PrepareChangeSetResult(res, peer_id) => { self.on_prepared_change_set(res, peer_id); } PeerMsg::PrepareCommitMergeResult(res, commit_index) => { self.on_prepared_commit_merge(res, commit_index); } PeerMsg::PrepareTxnFileResult { entry_index, peer_id, apply_generation, txn_file, } => { self.on_prepared_txn_file(entry_index, peer_id, apply_generation, txn_file); } PeerMsg::RaftlogFetched(fetched) => { self.on_raft_log_fetched(fetched.context, fetched.logs); } PeerMsg::WakeUp(_) => { unreachable!("wakeup idle peer should be handled in raft worker"); } PeerMsg::Idle(_) => { unreachable!("idle peer should be handled in raft idle worker"); } PeerMsg::StoreMsgForWakeUp(store_msg) => { self.ctx.global.router.send_store(store_msg); } } } } fn on_casual_msg(&mut self, msg: CasualMessage) { match msg { CasualMessage::SplitRegion { region_epoch, split_keys, callback, source, } => { self.on_prepare_split_region(region_epoch, split_keys, callback, source); } CasualMessage::HalfSplitRegion { region_epoch, policy, source, } => { self.on_schedule_half_split_region(region_epoch, policy, source); } CasualMessage::HalfSplitKeyRange { region_epoch, key_range, } => { self.on_schedule_half_split_key_range(region_epoch, key_range); } CasualMessage::DeletePrefix { region_version, prefix, callback, } => self.on_delete_prefix(region_version, prefix, callback), CasualMessage::IngestFiles { cs, callback } => self.on_ingest_files(cs, callback), CasualMessage::RestoreShard { cs, expected_conf_ver, callback, } => self.on_restore_shard(cs, expected_conf_ver, callback), CasualMessage::TriggerTrimOverBound(parameter) => { self.trigger_trim_over_bound(parameter.target_shard.unwrap().ver, parameter) } CasualMessage::MajorCompact { major_compact, columnar, callback, } => self.on_manual_major_compact(major_compact, columnar, callback), CasualMessage::UpdateSchemaFile(schema_file) => self.on_update_schema_file(schema_file), CasualMessage::CheckLeader { shard_ver, callback, } => self.on_check_leader(shard_ver, callback), CasualMessage::ClearColumnar { restore_version } => { self.on_clear_columnar(restore_version) } CasualMessage::ClearFtsIndex => self.on_clear_fts_index(), CasualMessage::TriggerRefreshShardStates => self.on_trigger_refresh_shard_states(), CasualMessage::ForceSwitchMemTable { current_size, callback, } => self.on_force_switch_mem_table(current_size, callback), } } fn on_tick(&mut self) { if self.fsm.stopped { return; } trace!( "tick"; "peer_id" => self.fsm.peer_id(), "tag" => self.peer.tag(), ); self.fsm.peer.adjust_cfg_if_changed(self.ctx); let region_id = self.region_id(); self.fsm.ticker.update_peer_ticker(&self.ctx.cfg, region_id); self.ticker.tick_clock(); if self.fsm.peer.is_cold() { self.fsm.refresh_cold_state(self.ctx); if self.fsm.peer.is_cold() { return; } } if self.ticker.is_on_tick(PEER_TICK_RAFT) { self.on_raft_base_tick(); } if self.ticker.is_on_tick(PEER_TICK_PD_HEARTBEAT) { self.on_pd_heartbeat_tick(); } if self.ticker.is_on_tick(PEER_TICK_SPLIT_CHECK) { self.on_split_region_check_tick(); if self.fsm.peer.is_cold() { return; } } if self.ticker.is_on_tick(PEER_TICK_MAINTENANCE) { self.on_maintenance_tick(); } if self.ticker.is_on_tick(PEER_TICK_RAFT_LOG_GC) { self.on_raft_log_gc_tick(); } if self.ticker.is_on_tick(PEER_TICK_CHECK_LONG) { self.on_check_long_tick(); } if self.ticker.is_on_tick(PEER_TICK_REACTIVATE_MEMORY_LOCK) { self.on_reactivate_memory_lock_tick(); } if self.ticker.is_on_tick(PEER_TICK_FILE_GC_COMPACTION_CHECK) { self.on_file_gc_compaction_check_tick(); } } fn on_cold_region_tick(&mut self, sweep_id: u64) { self.fsm.peer.adjust_cfg_if_changed(self.ctx); self.fsm.refresh_cold_state(self.ctx); if !self.fsm.peer.is_cold() { return; } self.gc_obsolete_state_versions(); if self.fsm.peer.is_leader() { self.fsm.peer.check_peers(); self.update_max_lag_metrics(); self.fsm.peer.heartbeat_pd(self.ctx, true); let has_lagging_replica = self.fsm.peer.has_lagging_replica(); if has_lagging_replica || should_send_cold_leader_heartbeat(self.ctx, self.fsm.region_id(), sweep_id) { self.fsm.peer.raft_group.raft.ping(); debug!( "cold leader sends raft heartbeat"; "tag" => self.fsm.peer.tag(), "peer_id" => self.fsm.peer_id(), "has_lagging_replica" => has_lagging_replica, ); } return; } self.on_check_peer_stale_state(false); if let Some(reason) = self .fsm .cold_tick_campaign_reason(&self.ctx.global.cold_store_liveness) { self.start_cold_recovery_probe(reason); } } fn cold_campaign_in_progress(&self) -> bool { matches!( self.fsm.peer.raft_group.raft.state, raft::StateRole::PreCandidate | raft::StateRole::Candidate ) } fn campaign_cold_peer(&mut self, reason: ColdCampaignReason) { let Some(campaign_result) = self.fsm.peer.cold_campaign() else { return; }; record_cold_campaign_attempt(reason); self.fsm.refresh_cold_state(self.ctx); if let Err(err) = campaign_result { warn!( "cold campaign failed"; "tag" => self.fsm.peer.tag(), "leader_id" => self.fsm.peer.leader_id(), "reason" => ?reason, "err" => ?err, ); } } fn start_cold_recovery_probe(&mut self, reason: ColdCampaignReason) { if self .fsm .cold_leader_probe_task(&self.ctx.global.cold_store_liveness) .is_some() { return; } if self.fsm.start_cold_leader_probe(self.ctx, reason) { debug!( "cold follower probes peers to discover leader"; "tag" => self.fsm.peer.tag(), "peer_id" => self.fsm.peer_id(), "reason" => ?reason, ); return; } if self.cold_campaign_in_progress() { return; } if self.fsm.peer.cold_campaign_allowed() { self.campaign_cold_peer(reason); } else { self.schedule_validate_peer(); } } fn start_request_driven_cold_campaign(&mut self) { if !self.fsm.peer.cold_campaign_allowed() { return; } // A foreground request should not wait for a passive probe when no // leader is known. self.fsm.cold_leader_probe_task = None; if self.cold_campaign_in_progress() { return; } debug!( "cold follower starts request-driven campaign"; "tag" => self.fsm.peer.tag(), "peer_id" => self.fsm.peer_id(), ); self.campaign_cold_peer(ColdCampaignReason::LeaderUnknown); } fn start_cold_request_recovery(&mut self) { if !self.fsm.peer.is_cold() || self.fsm.peer.is_leader() { return; } if self.fsm.peer.leader_id() == raft::INVALID_ID { self.start_request_driven_cold_campaign(); } else { self.start_cold_recovery_probe(ColdCampaignReason::LeaderUnknown); } } fn on_cold_read_index_timeout(&mut self) { self.fsm.refresh_cold_state(self.ctx); if !self.fsm.peer.is_cold() || self.fsm.peer.is_leader() || self .fsm .peer .pending_read_timeout_after(self.ctx.cfg.raft_election_timeout()) != Some(std::time::Duration::default()) { return; } if self.fsm.peer.cold_campaign_allowed() { self.fsm.cold_leader_probe_task = None; debug!( "cold follower read-index timed out, starts cold request-driven campaign"; "tag" => self.fsm.peer.tag(), "leader_id" => self.fsm.peer.leader_id(), ); self.campaign_cold_peer(ColdCampaignReason::ReadIndexTimeout); } let term = self.fsm.peer.term(); self.fsm .peer .pending_reads .clear_uncommitted(term, "cold read-index recovery timeout"); } fn on_cold_leader_probe_timeout(&mut self) { self.fsm.refresh_cold_state(self.ctx); let Some(task) = self .fsm .cold_leader_probe_task(&self.ctx.global.cold_store_liveness) else { return; }; // The message may refer to an older task that was replaced before the // raft worker applied the updated deadline. if task.deadline > StdInstant::now() { return; } debug!( "cold follower leader probe timed out"; "tag" => self.fsm.peer.tag(), "leader_id" => self.fsm.peer.leader_id(), "reason" => ?task.campaign_reason, ); // Consume the one-shot probe before campaigning so a delayed response // cannot restore an obsolete leader hint. self.fsm.cold_leader_probe_task = None; if self.fsm.peer.cold_campaign_allowed() { self.campaign_cold_peer(task.campaign_reason); } else { self.schedule_validate_peer(); } } fn start(&mut self) { let region_id = self.region_id(); self.ticker.schedule(PEER_TICK_RAFT); self.ticker .schedule_staggered(PEER_TICK_PD_HEARTBEAT, region_id); self.ticker.schedule(PEER_TICK_SPLIT_CHECK); self.ticker .schedule_staggered(PEER_TICK_MAINTENANCE, region_id); self.ticker.schedule(PEER_TICK_RAFT_LOG_GC); self.ticker .schedule_staggered(PEER_TICK_CHECK_LONG, region_id); self.ticker.schedule(PEER_TICK_FILE_GC_COMPACTION_CHECK); // A restored cold candidate with no unexpired Raft vote lease can // enter Cold directly. Recent leader contact keeps it TempRegular. self.fsm.initialize_cold_state(self.ctx); if GLOBAL_SERVER_READINESS .startup_peer_activation_completed .load(Ordering::Relaxed) || !self.fsm.startup_state.is_managed() { return; } self.ctx .global .router .send_store(StoreMsg::PeerStartupActivated(self.region_id())); } fn on_leader_callback(&mut self, cb: Callback) { let msg = new_read_index_request( self.region_id(), self.region().get_region_epoch().clone(), self.fsm.peer.peer.clone(), ); self.propose_raft_command(msg, TraceContext::default(), cb, None); } fn on_capture_change( &mut self, cmd: ChangeObserver, region_epoch: RegionEpoch, cb: Callback, can_apply: bool, ) { if can_apply { // Check peer. The leadership may change before this 2nd message, so // we should do a check again. if !self.peer.is_leader() { cb.invoke_with_response(new_error(Error::NotLeader( self.peer.region_id, self.fsm.peer.get_peer_from_cache(self.fsm.peer.leader_id()), ))); } else { self.ctx.apply_msgs.msgs.push(ApplyMsg::Change { cmd, region_epoch, cb, }); } return; } let region_id = self.region_id(); fail_point!("raft_on_capture_change"); let mut msg = new_read_index_request(region_id, region_epoch.clone(), self.fsm.peer.peer.clone()); // Allow to capture change even is in flashback state. // TODO: add a test case for this kind of situation. if self.region().is_in_flashback { let mut flags = WriteBatchFlags::from_bits_check(msg.get_header().get_flags()); flags.insert(WriteBatchFlags::FLASHBACK); msg.mut_header().set_flags(flags.bits()); } let raft_router = self.ctx.global.router.clone(); self.propose_raft_command( msg, TraceContext::default(), Callback::read(Box::new(move |resp| { // Return the error if resp.response.get_header().has_error() { cb.invoke_read(resp); return; } raft_router.significant_send( region_id, SignificantMsg::CaptureChange { cmd, region_epoch, callback: cb, can_apply: true, }, ); })), None, ); } fn on_significant_msg(&mut self, msg: SignificantMsg) { match msg { SignificantMsg::StoreUnreachable { store_id } => { if let Some(peer_id) = find_peer(self.region(), store_id).map(|p| p.get_id()) { if self.fsm.peer.is_leader() { self.fsm.peer.raft_group.report_unreachable(peer_id); } } } SignificantMsg::CaptureChange { cmd, region_epoch, callback, can_apply, } => self.on_capture_change(cmd, region_epoch, callback, can_apply), SignificantMsg::LeaderCallback(cb) => { self.on_leader_callback(cb); } } } #[inline] fn region_id(&self) -> u64 { self.fsm.peer.region().get_id() } #[inline] fn region(&self) -> &Region { self.fsm.peer.region() } #[inline] fn store_id(&self) -> u64 { self.fsm.peer.peer.get_store_id() } fn on_raft_base_tick(&mut self) { if self.peer.pending_remove { self.peer.mut_store().flush_cache_metrics(); return; } // Update the state whether the peer is pending on applying raft // logs if necesssary. self.on_check_peer_complete_apply_logs(); // If the peer is busy on apply and missing the last leader committed index, // it should propose a read index to check whether its lag is behind the leader. // It won't generate flooding fetching messages. This proposal will only be sent // out before it gets response and updates the `last_leader_committed_index`. self.try_to_fetch_committed_index(); self.ticker.schedule(PEER_TICK_RAFT); let temp_regular_leader_needs_raft_tick = self.fsm.peer.is_temp_regular() && self.fsm.peer.is_leader() && self.fsm.has_cold_progress_obligation(); let peer = &mut self.fsm.peer; // When having pending snapshot, if election timeout is met, it can't pass // the pending conf change check because first index has been updated to // a value that is larger than last index. if peer.is_applying_snapshot() || peer.has_pending_snapshot() { // need to check if snapshot is applied. return; } if peer.pending_merge_state.is_some() && peer.get_store().initial_flushed() { self.ctx .global .router .send_store(StoreMsg::CheckMerge(peer.region_id)); } let snapshot_not_ready_peers = &peer.get_store().snapshot_not_ready_peers; if !snapshot_not_ready_peers.borrow().is_empty() { let peers = snapshot_not_ready_peers.take(); if peer.is_leader() { let last_index = peer.get_store().last_index(); for peer_id in peers { if let Some(pr) = peer.raft_group.raft.mut_prs().get_mut(peer_id) { // Increase these peers' next_idx manually to append entries. // pr.next_idx should never equal to pr.matched, otherwise the follower // will keep ignoring the snapshot, never be able to catch up. pr.next_idx = cmp::max(last_index, pr.matched + 1); } } } } let raft_election_timeout_ticks = self.ctx.cfg.raft_election_timeout_ticks; peer.retry_pending_reads(raft_election_timeout_ticks); if peer.is_temp_regular() && (!peer.is_leader() || !temp_regular_leader_needs_raft_tick) { // Keep outer progress work running while quiescing Raft. The // saturated state remains TempRegular until the next split check. peer.advance_election_elapsed_without_raft_tick(); } else { peer.raft_group.tick(); if self.ctx.cfg.idle_worker_tick_slow && self.ctx.worker_type == WorkerType::Idle && !peer.is_leader() { // We increase the tick interval by 2 for Idle worker but want the election // timeout to be the same. So we tick the follower raft group again. peer.raft_group.tick(); } } peer.mut_store().flush_cache_metrics(); if peer.need_campaign { let _ = peer.raft_group.campaign(); peer.need_campaign = false; } peer.check_snapshot_ack_timeout(); } fn on_apply_result(&mut self, res: MsgApplyResult) { if res.peer_id != self.peer.peer_id() { warn!( "{} mismatch apply result peer_id {}", self.peer.tag(), res.peer_id ); return; } if self.peer.is_applying_snapshot() { debug!( "{} applying snapshot, skip handle apply results {:?}", self.peer.tag(), res.results ); return; } self.fsm .peer .record_encryption_pause_from_apply_result(res.encryption_paused_index); fail_point!("on_apply_res", |_| {}); if !self.fsm.peer.pending_apply_results.is_empty() { // Apply results should be handled in order but there is a pending one, so // delay to handle it. self.fsm.peer.pending_apply_results.push(res); return; } if !res.results.is_empty() { self.fsm.peer.pending_apply_results.push(res); // Some metadata change need to be handled by store FSM, so send the result to // it and store FSM will handle all pending results of the peer. let region_id = self.region_id(); let peer_id = self.peer_id(); self.ctx .global .router .send_store(StoreMsg::ApplyResult { region_id, peer_id }); return; } debug!( "async apply finish"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "res" => ?res, ); self.ctx .global .engine_total_bytes_written .fetch_add(res.metrics.written_bytes, Ordering::Relaxed); self.ctx .global .engine_total_keys_written .fetch_add(res.metrics.written_keys, Ordering::Relaxed); self.fsm.peer.post_apply(self.ctx, &res); if self.fsm.stopped {} } fn on_transfer_leader_msg(&mut self, msg: &eraftpb::Message) { // log_term is set by original leader, represents the term last log is written // in, which should be equal to the original leader's term. if msg.get_log_term() != self.fsm.peer.term() { return; } if self.fsm.peer.is_leader() { let from = match self.fsm.peer.get_peer_from_cache(msg.get_from()) { Some(p) => p, None => return, }; match self .fsm .peer .ready_to_transfer_leader(self.ctx, msg.get_index(), &from) { Some(reason) => { info!( "reject to transfer leader"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "to" => ?from, "reason" => reason, "index" => msg.get_index(), "last_index" => self.fsm.peer.get_store().last_index(), ); } None => { // Propose in-memory pessimistic locks before transferring leader. // Returns true if locks were proposed and transfer should be delayed. if self.propose_locks_before_transfer_leader(msg) { // If some pessimistic locks are just proposed, we propose another // TransferLeader command instead of transferring leader immediately. // This command acts as a fence - when the follower applies it, // everything before it (including locks) is committed. The follower // then sends back a reply with TRANSFER_LEADER_COMMAND_REPLY_CTX. info!( "propose transfer leader command"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "to" => ?from, ); // Use AdminCmdType::TransferLeader command which goes through raft // replication. When the target peer applies it, an ack is sent back. let mut cmd = RaftCmdRequest::default(); cmd.mut_header().set_region_id(self.region_id()); cmd.mut_header() .set_region_epoch(self.region().get_region_epoch().clone()); cmd.mut_header().set_peer(self.fsm.peer.peer.clone()); cmd.mut_header() .set_flags(WriteBatchFlags::TRANSFER_LEADER_PROPOSAL.bits()); cmd.mut_admin_request() .set_cmd_type(AdminCmdType::TransferLeader); cmd.mut_admin_request() .mut_transfer_leader() .set_peer(from.clone()); self.propose_raft_command( cmd, TraceContext::default(), Callback::None, None, ); } else { self.fsm.peer.transfer_leader(&from); } } } } else { self.fsm.peer.execute_transfer_leader(self.ctx, msg); } } /// Proposes in-memory pessimistic locks before transferring leader. /// /// Returns true if the transfer should be delayed (locks were proposed and /// we need to wait for PD retry). Returns false if we can proceed with /// transfer immediately. /// /// This prevents lock loss during leader transfer by ensuring all in-memory /// pessimistic locks are persisted before the new leader takes over. fn propose_locks_before_transfer_leader(&mut self, msg: &eraftpb::Message) -> bool { fail_point!("propose_locks_before_transfer_leader", |_| false); // Clone to make borrow checker happy when registering ticks. let txn_ext = self.fsm.peer.txn_ext.clone(); let mut pessimistic_locks = txn_ext.pessimistic_locks.write(); // Check if this is a reply to a TransferLeader command we proposed earlier. // If the locks are still in TransferringLeader status, we can proceed with // the transfer. if msg.get_context() == TRANSFER_LEADER_COMMAND_REPLY_CTX && pessimistic_locks.status == LocksStatus::TransferringLeader { return false; } // If not writable (e.g., already in TransferringLeader or other state), // the locks have already been proposed. But we still need to return true to // propose another TransferLeader command. Otherwise, some write requests that // have marked some locks as deleted will fail because raft rejects more // proposals. // It is OK to return true here if it's in other states like MergingRegion or // NotLeader. In those cases, the locks will fail to propose and nothing will // happen. if !pessimistic_locks.is_writable() { return true; } // Set status to TransferringLeader and register the reactivation tick. pessimistic_locks.status = LocksStatus::TransferringLeader; self.fsm.reactivate_memory_lock_ticks = 0; self.ticker.schedule(PEER_TICK_REACTIVATE_MEMORY_LOCK); // If there are no locks to propose, we can proceed with transfer. if pessimistic_locks.is_empty() { return false; } // Build a custom raft log to persist the pessimistic locks. // CSE uses CustomBuilder with CustomRaftLogType::PessimisticLock format // instead of Put requests. let mut builder = CustomBuilder::new(); builder.set_type(CustomRaftLogType::PessimisticLock); let mut lock_count = 0; { let pessimistic_locks = RwLockWriteGuard::downgrade(pessimistic_locks); fail_point!("invalidate_locks_before_transfer_leader"); for (key, (lock, deleted)) in &*pessimistic_locks { if *deleted { // Skip locks marked as deleted. The corresponding writes are already // in the raft pipeline and will be proposed. See PeerPessimisticLocks // comments for the safety argument. continue; } // Note: append_lock expects raw key (not encoded key). // Key::to_raw() decodes the key from its internal encoded form. let raw_key = key.to_raw().unwrap(); builder.append_lock(&raw_key, &lock.to_lock().to_bytes()); lock_count += 1; } } if lock_count == 0 { // All locks were marked as deleted. Propose a TransferLeader command to // ensure ordering with writes that removed them. return true; } let mut cmd = RaftCmdRequest::default(); cmd.set_custom_request(builder.build()); cmd.mut_header().set_region_id(self.region_id()); cmd.mut_header() .set_region_epoch(self.region().get_region_epoch().clone()); cmd.mut_header().set_peer(self.fsm.peer.peer.clone()); cmd.mut_header() .set_flags(WriteBatchFlags::TRANSFER_LEADER_PROPOSAL.bits()); info!( "propose locks before transferring leader"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "lock_count" => lock_count, ); // Propose the locks. self.propose_raft_command(cmd, TraceContext::default(), Callback::None, None); true } /// Handles the reactivation tick for in-memory pessimistic locks. /// /// This tick is scheduled when leader transfer is initiated. If the /// transfer fails (transferee doesn't become leader within the /// timeout), this reactivates the in-memory locks so they can be used /// again. fn on_reactivate_memory_lock_tick(&mut self) { let mut pessimistic_locks = self.fsm.peer.txn_ext.pessimistic_locks.write(); // Only handle this if we're the leader and in TransferringLeader status. if !self.fsm.peer.is_leader() || pessimistic_locks.status != LocksStatus::TransferringLeader { return; } self.fsm.reactivate_memory_lock_ticks += 1; let transferring_leader = self.fsm.peer.raft_group.raft.lead_transferee.is_some(); // If transfer is no longer in progress and we've waited long enough, // reactivate. if !transferring_leader && self.fsm.reactivate_memory_lock_ticks >= self.ctx.cfg.reactive_memory_lock_timeout_tick { info!( "reactivate in-memory pessimistic locks after transfer timeout"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "ticks" => self.fsm.reactivate_memory_lock_ticks, ); pessimistic_locks.status = LocksStatus::Normal; self.fsm.reactivate_memory_lock_ticks = 0; } else { // Continue checking. drop(pessimistic_locks); self.ticker.schedule(PEER_TICK_REACTIVATE_MEMORY_LOCK); } } fn on_raft_message(&mut self, mut msg: RaftMessage) -> Result<()> { let msg_debug = MsgDebug(&msg); debug!( "handle raft message"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "message" => %msg_debug, ); let msg_type = msg.get_message().get_msg_type(); if !matches!( msg_type, MessageType::MsgHeartbeat | MessageType::MsgHeartbeatResponse ) { self.peer.last_active_time = tikv_util::time::Instant::now_coarse(); } if !self.validate_raft_msg(&msg) { return Ok(()); } if self.fsm.peer.pending_remove || self.fsm.stopped || self.peer.is_applying_snapshot() { return Ok(()); } if msg.get_is_tombstone() { // we receive a message tells us to remove ourself. self.handle_gc_peer_msg(&msg); return Ok(()); } if msg.has_merge_target() { fail_point!("on_has_merge_target", |_| Ok(())); self.on_stale_merge( msg.get_merge_target().get_id(), msg.get_from_peer().get_id(), ); return Ok(()); } if self.check_msg(&msg) { return Ok(()); } // If this peer is restarting, it may lose some logs, so it should update // the `last_leader_committed_idx` with the commited index of the first // `MsgAppend` or `MsgHeartbeat` message or the committed index in // `MsgReadIndexResp` it received from leader. if self.fsm.peer.needs_update_last_leader_committed_idx() && (MessageType::MsgAppend == msg_type || MessageType::MsgHeartbeat == msg_type || MessageType::MsgReadIndexResp == msg_type) { let committed_index = cmp::max( msg.get_message().get_commit(), // from MsgAppend or MsgHeartbeat msg.get_message().get_index(), // from MsgReadIndexResp ); self.fsm .peer .update_last_leader_committed_idx(committed_index); } if msg.has_extra_msg() { self.on_extra_message(msg); return Ok(()); } let is_snapshot = msg.get_message().has_snapshot(); let regions_to_destroy = self.check_snapshot(&msg)?; let from_peer_id = msg.get_from_peer().get_id(); let cold_leader_probe_response = if msg_type == MessageType::MsgReadIndexResp { msg.get_message() .get_entries() .first() .and_then(|entry| ReadIndexContext::parse(entry.get_data()).ok()) .map(|ctx| (ctx.id, msg.get_message().get_term())) } else { None }; self.fsm.peer.insert_peer_cache(msg.take_from_peer()); let result = if msg_type == MessageType::MsgTransferLeader { self.on_transfer_leader_msg(msg.get_message()); Ok(()) } else { // This can be a message that sent when it's still a follower. Nevertheless, // it's meaningless to continue to handle the request as callbacks are cleared. if msg_type == MessageType::MsgReadIndex && self.fsm.peer.is_leader() && (msg.get_message().get_from() == raft::INVALID_ID || msg.get_message().get_from() == self.fsm.peer_id()) { self.ctx.raft_metrics.message_dropped.stale_msg.inc(); return Ok(()); } self.fsm.peer.step(self.ctx, msg.take_message()) }; if result.is_ok() && cold_leader_probe_response.is_some_and(|(probe_id, response_term)| { self.fsm .observe_cold_leader_probe_response(probe_id, from_peer_id, response_term) }) { debug!( "cold follower discovered leader by read-index probe"; "tag" => self.fsm.peer.tag(), "leader_id" => from_peer_id, ); } if is_snapshot && self.fsm.peer.has_pending_snapshot() { self.destroy_regions_for_snapshot(regions_to_destroy); } if self.fsm.peer.any_new_peer_catch_up(from_peer_id) { self.fsm.peer.heartbeat_pd(self.ctx, true); } result } fn on_extra_message(&mut self, mut msg: RaftMessage) { match msg.get_extra_msg().get_type() { ExtraMessageType::MsgWantRollbackMerge => { self.fsm.peer.maybe_add_want_rollback_merge_peer( msg.get_from_peer().get_id(), msg.get_extra_msg(), ); } ExtraMessageType::MsgCheckStalePeerResponse => { self.fsm.peer.on_check_stale_peer_response( msg.get_region_epoch().get_conf_ver(), msg.mut_extra_msg().take_check_peers().into(), ); } _ => {} // TODO } } // return false means the message is invalid, and can be ignored. pub(crate) fn validate_raft_msg(&mut self, msg: &RaftMessage) -> bool { let region_id = msg.get_region_id(); let to = msg.get_to_peer(); if to.get_store_id() != self.store_id() { warn!( "store not match, ignore it"; "tag" => self.peer.tag(), "region_id" => region_id, "to_store_id" => to.get_store_id(), "my_store_id" => self.store_id(), ); self.ctx .raft_metrics .message_dropped .mismatch_store_id .inc(); return false; } if !msg.has_region_epoch() { error!( "missing epoch in raft message, ignore it"; "tag" => self.peer.tag(), "region_id" => region_id, ); self.ctx .raft_metrics .message_dropped .mismatch_region_epoch .inc(); return false; } true } /// Checks if the message is sent to the correct peer. /// /// Returns true means that the message can be dropped silently. pub(crate) fn check_msg(&mut self, msg: &RaftMessage) -> bool { let from_epoch = msg.get_region_epoch(); let from_store_id = msg.get_from_peer().get_store_id(); // Let's consider following cases with three nodes [1, 2, 3] and 1 is leader: // - 1 removes 2, 2 may still send MsgAppendResponse to 1. // We should ignore this stale message and let 2 remove itself after // applying the ConfChange log. // - 2 is isolated, 1 removes 2. When 2 rejoins the cluster, 2 will // send stale MsgRequestVote to 1 and 3, at this time, we should tell 2 to gc // itself. // - 2 is isolated but can communicate with 3. 1 removes 3. // 2 will send stale MsgRequestVote to 3, 3 should ignore this message. // - 2 is isolated but can communicate with 3. 1 removes 2, then adds 4, remove // 3. // 2 will send stale MsgRequestVote to 3, 3 should tell 2 to gc itself. // - 2 is isolated. 1 adds 4, 5, 6, removes 3, 1. Now assume 4 is leader. // After 2 rejoins the cluster, 2 may send stale MsgRequestVote to 1 and 3, // 1 and 3 will ignore this message. Later 4 will send messages to 2 and 2 will // rejoin the raft group again. // - 2 is isolated. 1 adds 4, 5, 6, removes 3, 1. Now assume 4 is leader, and 4 // removes 2. // unlike case e, 2 will be stale forever. // TODO: for case f, if 2 is stale for a long time, 2 will communicate with pd // and pd will tell 2 is stale, so 2 can remove itself. if util::is_epoch_stale(from_epoch, self.fsm.peer.region().get_region_epoch()) && self.peer.is_initialized() && find_peer(self.fsm.peer.region(), from_store_id).is_none() { self.ctx .handle_stale_msg(msg, self.fsm.peer.region().get_region_epoch().clone(), None); return true; } if !self.peer.is_initialized() { match msg.get_message().get_msg_type() { MessageType::MsgRequestVote | MessageType::MsgRequestPreVote => { // An uninitialized peer may be replaced by split later, if we vote for one peer // and later replaced by split, we may vote for another peer on the same term, // break the raft protocol. return true; } _ => {} } } let target = msg.get_to_peer(); match target.get_id().cmp(&self.fsm.peer.peer_id()) { cmp::Ordering::Less => { info!( "target peer id is smaller, msg maybe stale"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "target_peer" => ?target, ); self.ctx.raft_metrics.message_dropped.stale_msg.inc(); true } cmp::Ordering::Greater => { self.maybe_destroy(); true } cmp::Ordering::Equal => false, } } pub(crate) fn handle_gc_peer_msg(&mut self, msg: &RaftMessage) { let from_epoch = msg.get_region_epoch(); if !util::is_epoch_stale(self.fsm.peer.region().get_region_epoch(), from_epoch) { return; } if self.fsm.peer.peer != *msg.get_to_peer() { info!( "receive stale gc message, ignore."; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); return; } // TODO: ask pd to guarantee we are stale now. info!( "receives gc message, trying to remove"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "to_peer" => ?msg.get_to_peer(), ); self.maybe_destroy(); } fn on_stale_merge(&mut self, target_region_id: u64, from_peer_id: u64) { if self.fsm.peer.pending_remove { return; } info!( "successful merge can't be continued, try to gc stale peer"; "tag" => self.peer.tag(), "from_peer_id" => from_peer_id, "to_peer_id" => self.fsm.peer_id(), "target_region_id" => target_region_id, "merge_state" => ?self.fsm.peer.pending_merge_state, ); // Because of the checking before proposing `PrepareMerge`, which is // no `CompactLog` proposal between the smallest commit index and the latest // index. If the merge succeed, all source peers are impossible in apply // snapshot state and must be initialized. // So `maybe_destroy` must succeed here. self.maybe_destroy(); } // Returns `Vec<(u64, bool)>` indicated (source_region_id, merge_to_this_peer) // if the `msg` doesn't contain a snapshot or this snapshot doesn't conflict // with any other snapshots or regions. Otherwise a `SnapKey` is returned. fn check_snapshot(&mut self, msg: &RaftMessage) -> Result> { if !msg.get_message().has_snapshot() { return Ok(vec![]); } // TODO(x) Ok(vec![]) } fn destroy_regions_for_snapshot(&mut self, regions_to_destroy: Vec<(u64, bool)>) { if regions_to_destroy.is_empty() {} // TODO(x) } /// Check if a request is valid if it has valid prepare_merge/commit_merge /// proposal. pub(crate) fn check_merge_proposal( &mut self, msg: &RaftCmdRequest, store_meta: Option<&mut StoreMeta>, ) -> Result<()> { if !msg.has_admin_request() { return Ok(()); } let admin_req = msg.get_admin_request(); let region = self.fsm.peer.get_preprocessed_region(); if admin_req.has_prepare_merge() { // Reject starting a merge while an `ENCRYPTION_META_KEY` update is still // in-flight. We want to prevent the case where the source becomes // newer than the target when prepare-merge is applied, because that // could mean the data on the source region has an encryption meta // rollback when the merge is committed (commit-merge keeps the // target properties for same-keyspace merges). // // We don't reject CommitMerge here because that wouldn't stop the target from // moving forward to a newer encryption meta anyway, and it's // harmless for the target to be newer. if let Some(pending_idx) = self.fsm.peer.pending_encryption_meta_change { let applied_idx = self.fsm.peer.get_store().applied_index(); if applied_idx < pending_idx { return Err(box_err!( "{} pending encryption meta change, applied_index {}, pending_index {}, skip merge", self.fsm.peer.tag(), applied_idx, pending_idx )); } } let store_meta = store_meta.unwrap(); // Just for simplicity, do not start region merge while in joint state if self.fsm.peer.in_joint_state() { return Err(box_err!( "{} region in joint state, can not propose merge command, command: {:?}", self.fsm.peer.tag(), msg.get_admin_request() )); } let target_region = admin_req.get_prepare_merge().get_target(); { match store_meta.region_map.get(target_region.get_id()) { Some(r) => { if r != target_region { return Err(box_err!( "target region not matched, skip proposing: {:?} != {:?}", r, target_region )); } } None => { return Err(box_err!( "target region {} doesn't exist.", target_region.get_id() )); } } } if !util::is_sibling_regions(target_region, region) { return Err(box_err!( "{:?} and {:?} are not sibling, skip proposing.", target_region, region )); } if !region_on_same_stores(target_region, region) { return Err(box_err!( "peers doesn't match {:?} != {:?}, reject merge", region.get_peers(), target_region.get_peers() )); } let id = self.region_id(); let version = msg.get_header().get_region_epoch().get_version(); let target_id = target_region.get_id(); let target_version = target_region.get_region_epoch().get_version(); let kv = &self.ctx.global.engines.kv; let check_result = kv.check_merge(id, version, target_id, target_version)?; info!("check_merge_proposal, {:?}", check_result); let CheckMergeResult { source_overbound, target_overbound, source_require_empty, target_require_empty, inconsistent_encryption_key, inconsistent_storage_class_spec: inconsistent_storage_class, } = check_result; // If the two regions belongs to different keyspaces, // `inconsistent_encryption_key` will always be false. if inconsistent_encryption_key { return Err(kvengine::Error::CheckMerge(format!( "shards have inconsistent encryption meta/key, source:{:?}, target:{:?}", IdVer::new(id, version), IdVer::new(target_id, target_version) )) .into()); } if inconsistent_storage_class { return Err(kvengine::Error::CheckMerge(format!( "shards have inconsistent storage classes, source:{:?}, target:{:?}", IdVer::new(id, version), IdVer::new(target_id, target_version) )) .into()); } if source_overbound || target_overbound { let parameter = TrimOverBoundParameter { source_shard: if source_overbound { Some(IdVer::new(id, version)) } else { None }, target_shard: if target_overbound { Some(IdVer::new(target_id, target_version)) } else { None }, }; self.trigger_trim_over_bound(version, parameter); return Err(kvengine::Error::CheckMerge(format!( "shards have over bound data, source:{source_overbound}, target:{target_overbound}" )) .into()); } // Shards are required to be empty when they are from different keyspaces. // As SSTs have no prefix, merge SSTs from different keyspaces will violate // correctness. // Also note that the empty status of shard is not reliable, so on commit merge, // the data will be cleared in this scene. // The checking here is just for more safety. if source_require_empty || target_require_empty { return Err(kvengine::Error::CheckMerge(format!( "shards from different keyspaces require to be empty, source:{:?}: {:?}, target:{:?}: {:?}", IdVer::new(id, version), region, IdVer::new(target_id, target_version), target_region, )) .into()); } if let Some(source_meta) = self.fsm.get_peer().get_store().shard_meta.as_ref() { // NOTE: source shard meta maybe stale, so we should also check unconverted_l0s // after propose. if !source_meta.unconverted_l0s.is_empty() { return Err(box_err!( "failed to merge source region with {} unconverted l0s", source_meta.unconverted_l0s.len() )); } } } else if admin_req.has_commit_merge() { let source_region = msg.get_admin_request().get_commit_merge().get_source(); if !util::is_sibling_regions(source_region, region) { return Err(box_err!( "{:?} and {:?} should be sibling", source_region, region )); } if !region_on_same_stores(source_region, region) { return Err(box_err!( "peers not matched: {:?} {:?}", source_region, region )); } let source_meta = msg.get_admin_request().get_commit_merge().get_source_meta(); let mut source_meta_cs = kvenginepb::ChangeSet::default(); if source_meta_cs.merge_from_bytes(source_meta).is_err() { return Err(box_err!("failed to get merge source change set from meta")); } let source_fts_indexes = tracked_fts_indexes_from_snapshot(source_meta_cs.get_snapshot()); let Some(target_meta) = self.fsm.get_peer().get_store().shard_meta.as_ref() else { return Ok(()); }; let target_fts_indexes = tracked_fts_indexes_from_shard_meta(target_meta); let source_snap = source_meta_cs.get_snapshot(); let source_data_bound = DataBound::new( InnerKey::from_outer_key(source_snap.get_outer_start()), InnerKey::from_outer_end_key(source_snap.get_outer_end()), true, ); let target_data_bound = DataBound::new( InnerKey::from_outer_key(target_meta.range.outer_start.as_ref()), InnerKey::from_outer_end_key(target_meta.range.outer_end.as_ref()), true, ); if !has_consistent_tracked_fts_indexes( &source_fts_indexes, &target_fts_indexes, source_data_bound, target_data_bound, ) { return Err(box_err!("source and target have inconsistent FTS indexes")); } } Ok(()) } pub(crate) fn pre_propose_raft_command( &mut self, msg: &RaftCmdRequest, ) -> Result> { // Check store_id, make sure that the msg is dispatched to the right place. if let Err(e) = _util::check_store_id(msg, self.store_id()) { self.ctx .raft_metrics .invalid_proposal .mismatch_store_id .inc(); return Err(e); } if msg.has_status_request() { // For status commands, we handle it here directly. let resp = self.execute_status_command(msg)?; return Ok(Some(resp)); } // Check whether the store has the right peer to handle the request. let region_id = self.region_id(); let leader_id = self.fsm.peer.leader_id(); let request = msg.get_requests(); // ReadIndex can be processed on the replicas. let is_read_index_request = request.len() == 1 && request[0].get_cmd_type() == CmdType::ReadIndex; let read_only = is_read_only_raft_command(msg); let allow_replica_read = read_only && msg.get_header().get_replica_read(); let flags = WriteBatchFlags::from_bits_check(msg.get_header().get_flags()); let allow_stale_read = read_only && flags.contains(WriteBatchFlags::STALE_READ); let reject_not_leader = !self.fsm.peer.is_leader() && !is_read_index_request && !allow_replica_read && !allow_stale_read; let get_or_snap = !request.is_empty() && request .iter() .all(|req| matches!(req.get_cmd_type(), CmdType::Get | CmdType::Snap)); let recover_cold_follower_read = !self.fsm.peer.is_leader() && self.fsm.peer.is_cold() && !allow_stale_read && (is_read_index_request || allow_replica_read || get_or_snap); if reject_not_leader && !recover_cold_follower_read { self.ctx.raft_metrics.invalid_proposal.not_leader.inc(); let leader = self.fsm.peer.get_peer_from_cache(leader_id); return Err(Error::NotLeader(region_id, leader)); } // peer_id must be the same as peer's. if let Err(e) = _util::check_peer_id(msg, self.fsm.peer.peer_id()) { self.ctx .raft_metrics .invalid_proposal .mismatch_peer_id .inc(); return Err(e); } // check whether the peer is initialized. if !self.fsm.peer.is_initialized() { self.ctx .raft_metrics .invalid_proposal .region_not_initialized .inc(); return Err(Error::RegionNotInitialized(region_id)); } // If the peer is applying snapshot, it may drop some sending messages, that // could make clients wait for response until timeout. if self.fsm.peer.is_applying_snapshot() { self.ctx .raft_metrics .invalid_proposal .is_applying_snapshot .inc(); // TODO: replace to a more suitable error. return Err(Error::Other(box_err!( "{} peer is applying snapshot", self.fsm.peer.tag() ))); } // Check whether the term is stale. if let Err(e) = _util::check_term(msg, self.fsm.peer.term()) { self.ctx.raft_metrics.invalid_proposal.stale_command.inc(); return Err(e); } let epoch_check = if Peer::is_encryption_meta_change_request(msg) { if !msg.get_header().has_region_epoch() { Err(Error::Other(box_err!("missing epoch!"))) } else { _util::compare_region_epoch( msg.get_header().get_region_epoch(), self.fsm.peer.region(), true, true, true, ) } } else { _util::check_region_epoch(msg, self.fsm.peer.region(), true) }; match epoch_check { Err(Error::EpochNotMatch(m, new_regions)) => { // Attach the region which might be split from the current region. But it // doesn't matter if the region is not split from the current // region. If the region meta received by the TiKV driver is // newer than the meta cached in the driver, the meta is // updated. // TODO(x) add sibling region. self.ctx.raft_metrics.invalid_proposal.epoch_not_match.inc(); Err(Error::EpochNotMatch(m, new_regions)) } Err(e) => Err(e), Ok(()) if recover_cold_follower_read && reject_not_leader => { self.ctx.raft_metrics.invalid_proposal.not_leader.inc(); let leader = self.fsm.peer.get_peer_from_cache(leader_id); self.start_cold_request_recovery(); Err(Error::NotLeader(region_id, leader)) } Ok(()) if recover_cold_follower_read && leader_id == raft::INVALID_ID => { if is_read_index_request || allow_replica_read { if let Err(e) = self.fsm.peer.pre_read_index() { self.ctx.raft_metrics.propose.unsafe_read_index.inc(); return Err(e); } self.ctx .raft_metrics .invalid_proposal .read_index_no_leader .inc(); } self.start_cold_request_recovery(); Err(Error::NotLeader(region_id, None)) } Ok(()) => Ok(None), } } pub(crate) fn propose_raft_command( &mut self, msg: RaftCmdRequest, trace_ctx: TraceContext, cb: Callback, store_meta: Option<&mut StoreMeta>, ) { if self.fsm.peer.pending_remove { notify_req_region_removed(self.region_id(), cb); return; } if self.fsm.peer.cold_state() != ColdPeerState::Regular && !is_read_only_raft_command(&msg) { self.fsm .set_cold_state(self.ctx, ColdPeerState::Regular, "proposal"); } self.peer.last_active_time = tikv_util::time::Instant::now_coarse(); if let Callback::Write { .. } = &cb { let tag = self.peer.tag(); let peer_id = self.fsm.peer_id(); txn_debug!(trace_ctx: trace_ctx.clone(), trace_event::types::Category::WriteDetails, "PeerMsgHandler::propose_raft_command called"; "tag" => tag, "peer_id" => peer_id, "custom_req" => msg.custom_request .as_ref() .and_then(|r| { if r.data.is_empty() { None } else { Some(format!("{}", CustomRaftLog::new_from_data(&r.data))) } }).unwrap_or_default() ); } // Waterfall metrics: record when this raft command finishes waiting and is // ready to be proposed. if self.ctx.raft_metrics.waterfall_metrics { let now = std::time::Instant::now(); for tracker in cb.write_trackers() { tracker.observe(now, &self.ctx.raft_metrics.wf_batch_wait, |t| { &mut t.metrics.wf_batch_wait_nanos }); } } match self.pre_propose_raft_command(&msg) { Ok(Some(resp)) => { cb.invoke_with_response(resp); return; } Err(e) => { debug!( "failed to propose"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "message" => ?msg, "err" => %e, ); cb.invoke_with_response(new_error(e)); return; } _ => (), } if let Err(e) = self.check_merge_proposal(&msg, store_meta) { warn!( "failed to propose merge"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "message" => ?msg, "err" => %e, ); cb.invoke_with_response(new_error(e)); return; } if !msg.get_requests().is_empty() && msg.get_requests()[0].has_ingest_sst() { self.propose_ingest_sst(msg, trace_ctx.clone(), cb); return; } // Note: // The peer that is being checked is a leader. It might step down to be a // follower later. It doesn't matter whether the peer is a leader or // not. If it's not a leader, the proposing command log entry can't be // committed. let mut resp = RaftCmdResponse::default(); let term = self.fsm.peer.term(); bind_term(&mut resp, term); self.fsm.peer.propose(self.ctx, trace_ctx, cb, msg, resp); // TODO: add timeout, if the command is not applied after timeout, // we will call the callback with timeout error. } fn propose_ingest_sst(&mut self, msg: RaftCmdRequest, trace_ctx: TraceContext, cb: Callback) { // This is a ingest sst request, we need to redirect to worker thread and // convert it to cloud engine format. let importer = self.ctx.global.importer.clone(); let router = self.ctx.global.router.clone(); let kv = self.ctx.global.engines.kv.clone(); let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap().clone(); spawn_anonymous_thread_with!(move || { STORE_INGEST_CONVERT_TASK_STATUS .with_label_values(&["running"]) .inc(); tikv_util::set_current_region(shard_meta.id); match convert_sst(kv, importer, &msg, shard_meta) { Ok(cs) => { // Make ingest command. let mut cmd = RaftCmdRequest::default(); cmd.set_header(msg.get_header().clone()); let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&cs); cmd.set_custom_request(custom_builder.build()); router.send_command(cmd, trace_ctx, cb); } Err(e) => { cb.invoke_with_response(new_error(e)); } } STORE_INGEST_CONVERT_TASK_STATUS .with_label_values(&["running"]) .dec(); }); } pub(crate) fn on_schedule_half_split_region( &mut self, region_epoch: RegionEpoch, policy: CheckPolicy, source: SplitSource, ) { info!( "on half split"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "policy" => ?policy, "source" => %source, ); if !self.fsm.peer.is_leader() { // region on this store is no longer leader, skipped. warn!( "not leader, skip"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); return; } let region = self.fsm.peer.region(); if util::is_epoch_stale(®ion_epoch, region.get_region_epoch()) { warn!( "receive a stale halfsplit message"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); return; } let split_key = self .fsm .peer .buckets .as_mut() .map(|buckets| { let length = buckets.meta.keys.len(); if length > 2 { buckets.meta.keys[length / 2].clone() } else { Vec::new() } }) .unwrap_or_default(); if !split_key.is_empty() { info!( "schedule ask split"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "split_key" => log_wrappers::hex_encode(split_key.clone()), ); // The split key comes from the bucket key, which is the encoded key. self.schedule_ask_split(vec![split_key], "half_split", SplitReason::Admin); } else { warn!( "no need to schedule, split key not found"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); } } pub(crate) fn on_schedule_half_split_key_range( &mut self, region_epoch: RegionEpoch, key_range: KeyRange, ) { let region_id = self.region_id(); info!("on half split key range"; "region_id" => region_id); if !self.fsm.peer.is_leader() { // region on this store is no longer leader, skipped. warn!("not leader, skip half split key range"; "region_id" => region_id); LOAD_BASE_SPLIT_EVENT.fallback_dispatch_failed.inc(); return; } let region = self.fsm.peer.region(); if util::is_epoch_stale(®ion_epoch, region.get_region_epoch()) { warn!("receive a stale half split key range message"; "region_id" => region_id); LOAD_BASE_SPLIT_EVENT.fallback_dispatch_failed.inc(); return; } let decision = decide_bucket_split( region, self.fsm .peer .buckets .as_ref() .map(|buckets| buckets.meta.as_ref()), &key_range, ); match decision { BucketSplitDecision::SplitKey(split_key) => { LOAD_BASE_SPLIT_EVENT.fallback_bucket.inc(); self.on_prepare_split_region( region_epoch, vec![split_key], Callback::None, SplitSource::Load, ); } BucketSplitDecision::CheckSst { start_key, end_key } => { CHECK_SPLIT_COUNTER.all.inc(); let task = SplitCheckTask::ApproximateHalfSplit { region_id, region_epoch, start_key, end_key, }; match self.ctx.global.split_check_scheduler.schedule(task) { Ok(SplitCheckScheduleOutcome::Scheduled) => {} Ok(SplitCheckScheduleOutcome::Busy) => { debug!("split check worker is busy"; "region_id" => region_id); CHECK_SPLIT_COUNTER.ignore.inc(); LOAD_BASE_SPLIT_EVENT.fallback_dispatch_failed.inc(); } Err(err) => { warn!( "failed to schedule half split key range"; "region_id" => region_id, "err" => ?err, ); CHECK_SPLIT_COUNTER.fail.inc(); LOAD_BASE_SPLIT_EVENT.fallback_dispatch_failed.inc(); } } } BucketSplitDecision::NoCandidate => { LOAD_BASE_SPLIT_EVENT.fallback_no_candidate.inc(); } } } fn on_split_region_check_tick(&mut self) -> bool /* should_split */ { // Tests use this hook to observe ticker rescheduling independent of region // state. fail_point!("on_split_region_check_tick"); self.ticker.schedule(PEER_TICK_SPLIT_CHECK); if let Some(shard) = self.ctx.global.engines.kv.get_shard(self.region_id()) { let estimated_size = shard.get_estimated_size(); let estimated_entries = shard.get_estimated_entries(); let estimated_kv_size = shard.get_estimated_kv_size(); let estimated_ia_kv_size = shard.get_estimated_ia_kv_size(); let estimated_columnar_size = shard.get_estimated_columnar_size(); let estimated_columnar_kv_size = shard.get_estimated_columnar_kv_size(); // use 1 for empty size as 0 is for unknown size in PD. self.peer.peer_stat.approximate_size = cmp::max(estimated_size, 1); self.peer.peer_stat.approximate_keys = estimated_entries; self.peer.peer_stat.approximate_kv_size = estimated_kv_size; self.peer.peer_stat.approximate_ia_kv_size = estimated_ia_kv_size; self.peer.peer_stat.approximate_columnar_size = estimated_columnar_size; self.peer.peer_stat.approximate_columnar_kv_size = estimated_columnar_kv_size; self.adjust_peer_stat_for_storage_class(shard.as_ref()); if self .fsm .update_cold_state_with_shard(self.ctx, shard.as_ref(), estimated_size) { return false; } if !shard.get_initial_flushed() { return false; } let is_leader = self.fsm.peer.is_leader(); self.check_schema(&shard, is_leader); if !is_leader { return false; } // When Lightning or BR is importing data to TiKV, their ingest-request may fail // because of region-epoch not matched. So we hope TiKV do not check // region size and split region during importing. if self.ctx.global.importer.get_mode() == SwitchMode::Import { return false; } if self.ctx.cfg.region_split_keys < 10 { // For some tests to be effective, we need to split tiny regions. // But tiny region only has mem-table, get_suggest_split_key will return None. // So we handle it specially. self.split_by_iterate(shard); return true; } if self.fsm.peer.last_bucket_update_meta_sequence != shard.get_meta_sequence() { // In case the region is under heavy write, we need to update the bucket more // frequently than pd heartbeat. // If the bucket is updated, we need to heartbeat pd to update region size in // pd, otherwise. if self.fsm.peer.update_buckets(self.ctx) { self.fsm.peer.heartbeat_pd(self.ctx, false); } } let mut region_max_size = self.ctx.cfg.region_split_size.0 * 3 / 2; let mut region_max_entries = self.ctx.cfg.region_split_keys * 3 / 2; if let Some(keyspace_config) = self .ctx .global .engines .kv .get_keyspace_config(shard.keyspace_id) { region_max_size = (region_max_size as f64 * keyspace_config.split_size_factor) as u64; region_max_entries = (region_max_entries as f64 * keyspace_config.split_size_factor) as u64; } raftstore::coprocessor::metrics::REGION_SIZE_HISTOGRAM.observe(estimated_size as f64); raftstore::coprocessor::metrics::REGION_KEYS_HISTOGRAM .observe(estimated_entries as f64); let table_count = shard.get_columnar_table_ids().len(); let max_tables_threshold = fail::eval("max_columnar_tables_threshold", |t| { t.and_then(|s: String| s.parse::().ok()) }) .flatten() .unwrap_or(LARGE_NUM_COLUMNAR_TABLES_IN_SHARD * 3); if estimated_size >= region_max_size || estimated_entries >= region_max_entries || table_count > max_tables_threshold { if let Some(k) = shard.get_suggest_split_key(self.ctx.cfg.region_bucket_size.0) { info!( "region {} split, estimated size {}, estimated entries {}, table count {}, max_size {}", self.peer.tag(), estimated_size, estimated_entries, table_count, region_max_size, ); self.schedule_ask_split( vec![Key::from_raw(k.chunk()).into_encoded()], "auto_split", SplitReason::Size, ); return true; } } } false } fn adjust_peer_stat_for_storage_class(&mut self, shard: &Shard) { if shard.get_storage_class_spec().can_be_ia() { let region_split_size = self.ctx.cfg.region_split_size.0; let peer_stat = &mut self.peer.peer_stat; // Adjust region size to prevent merge when the region is on the boundary of // table. if is_table_boundary_key(shard.inner_start()) || is_table_boundary_key(shard.inner_end()) { peer_stat.approximate_size = cmp::max(peer_stat.approximate_size, region_split_size); } } } fn schedule_ask_split( &mut self, split_keys: Vec>, split_source: &'static str, reason: SplitReason, ) { if let Some(shard_meta) = self.peer.get_store().shard_meta.as_ref() { if shard_meta.has_txn_file_locks() { record_txn_file_split_denied( "schedule_ask_split", shard_meta.txn_file_locks().len(), ); warn!("{} skip ask split because txn file locks exist", self.peer.tag(); "source" => split_source, "txn_file_locks" => %shard_meta.txn_file_locks(), ); return; } } let task = PdTask::AskBatchSplit { region: self.region().clone(), split_keys, peer: self.peer.peer.clone(), right_derive: true, callback: Callback::None, reason, }; if let Err(e) = self.ctx.global.pd_scheduler.schedule(task) { error!( "failed to schedule ask batch split"; "tag" => self.peer.tag(), "err" => ?e, ); } } // TODO: support async (or split by block keys ?) fn split_by_iterate(&mut self, shard: Arc) { let split_keys = self.ctx.cfg.region_split_keys; let split_max_keys = split_keys * 3 / 2; let snap = match shard.try_new_snap_access() { Ok(snap) => snap, Err(err) => { warn!( "{} skip split by iterate because snapshot access is unavailable", self.peer.tag(); "err" => ?err, ); return; } }; let mut iter = snap.new_iterator(0, false, false, None, true); iter.rewind(); let mut i = 0; let mut keys = vec![]; while iter.valid() { if i > 0 && i % split_keys == 0 { keys.push(Key::from_raw(iter.key()).into_encoded()); } i += 1; iter.next(); } if i <= split_max_keys { return; } info!("region {} split by iterate", self.peer.tag()); self.schedule_ask_split(keys, "auto_split", SplitReason::Size); } // For idle regions, we need tick to trigger switch mem-table and shrink // lock_cache to reduce memory consumption. Only the applier knows how large // the mem-table is and when the mem-table has been switched, so we don't // need to check anything, just send a message. fn on_maintenance_tick(&mut self) { fail_point!("on_maintenance_tick"); self.ticker.schedule(PEER_TICK_MAINTENANCE); self.maybe_schedule_txn_file_lock_active_resolve(); // We also send the message to follower to let follower do periodic maintenance // task. let region_id = self.region_id(); self.ctx .apply_msgs .msgs .push(ApplyMsg::Maintenance { region_id }); self.check_gc_compaction(); } fn on_prepare_split_region( &mut self, region_epoch: metapb::RegionEpoch, split_keys: Vec>, cb: Callback, source: SplitSource, ) { let is_load_split = source == SplitSource::Load; let mut validation_result = self.validate_split_region(®ion_epoch, &split_keys); if validation_result.is_ok() && is_load_split { validation_result = split_keys.iter().try_for_each(|key| { _util::check_key_in_region_exclusive(key, self.fsm.peer.region()) }); } if let Err(e) = validation_result { info!( "prepare split error"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "split_keys" => %KeysInfoFormatter(split_keys.iter()), "source" => %source, "error" => ?e, ); if is_load_split { LOAD_BASE_SPLIT_EVENT.split_failed.inc(); } let (resp, key_errs) = new_with_key_error(e); cb.invoke_with_response_ext(resp, key_errs); return; } info!( "on split"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "split_keys" => %KeysInfoFormatter(split_keys.iter()), "source" => %source, ); // Only load-based split should be tagged as Load. Other explicit split // requests are admin splits and must not be blocked by PD's auto-split // disable signal, which applies to Size/Load automatic split only. let reason = match &source { SplitSource::Load => SplitReason::Load, SplitSource::Schema | SplitSource::Pd(_) | SplitSource::Peer(_) => SplitReason::Admin, }; let task = PdTask::AskBatchSplit { region: self.region().clone(), split_keys, peer: self.peer.peer.clone(), right_derive: true, callback: cb, reason, }; if let Err(e) = self.ctx.global.pd_scheduler.schedule(task) { error!( "failed to schedule ask batch split"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "err" => ?e, ); let err = e.to_string(); if let PdTask::AskBatchSplit { callback, .. } = e.into_inner() { if is_load_split { LOAD_BASE_SPLIT_EVENT.split_failed.inc(); } let schedule_err: Box = box_err!("failed to schedule ask batch split: {}", err); callback.invoke_with_response(message_error(schedule_err)); } } } pub(crate) fn validate_split_region( &mut self, epoch: &metapb::RegionEpoch, split_keys: &[Vec], ) -> Result<()> { let tag = self.peer.tag(); if split_keys.is_empty() { error!( "no split key is specified."; "tag" => tag, "peer_id" => self.fsm.peer_id(), ); return Err(box_err!("{} no split key is specified.", tag)); } if split_keys.len() > self.ctx.cfg.split_region_max_keys { error!( "number of split keys exceeds limit"; "tag" => tag, ); return Err(box_err!("{} number of split keys exceeds limit", tag)); } for key in split_keys { if key.is_empty() { error!( "split key should not be empty!!!"; "tag" => tag, "peer_id" => self.fsm.peer_id(), ); return Err(box_err!("{} split key should not be empty", tag)); } if let Err(err) = decode_bytes(&mut key.as_slice(), false) { error!( "split key decode error"; "tag" => tag, "peer_id" => self.fsm.peer_id(), "err" => ?err, ); return Err(box_err!("{} split key decode failed", tag)); } } if !self.fsm.peer.is_leader() { // region on this store is no longer leader, skipped. info!( "not leader, skip."; "tag" => tag, "peer_id" => self.fsm.peer_id(), ); return Err(Error::NotLeader( self.region_id(), self.fsm.peer.get_peer_from_cache(self.fsm.peer.leader_id()), )); } let region = self.fsm.peer.region(); let latest_epoch = region.get_region_epoch(); // This is a little difference for `check_region_epoch` in region split case. // Here we just need to check `version` because `conf_ver` will be update // to the latest value of the peer, and then send to PD. if latest_epoch.get_version() != epoch.get_version() { info!( "epoch changed, retry later"; "tag" => tag, "peer_id" => self.fsm.peer_id(), "prev_epoch" => ?region.get_region_epoch(), "epoch" => ?epoch, ); return Err(Error::EpochNotMatch( format!( "{} epoch changed {:?} != {:?}, retry later", tag, latest_epoch, epoch ), vec![region.to_owned()], )); } if !self.peer.get_store().initial_flushed() { return Err(Error::Transport(DiscardReason::Full)); } if let Some(shard_meta) = self.peer.get_store().shard_meta.as_ref() { if shard_meta.has_txn_file_locks() { record_txn_file_split_denied( "validate_split_region", shard_meta.txn_file_locks().len(), ); warn!("{} prepare split: txn file locks exist, retry later", tag; "txn_file_locks" => %shard_meta.txn_file_locks(), ); let raw_key = decode_bytes(&mut split_keys.first().unwrap().as_slice(), false).unwrap(); let lock_infos = shard_meta .txn_file_locks() .get_lock_infos(&raw_key) .map_err(|err| { error!("{} validate_split_region: get txn file key errors failed", tag; "txn_file_locks" => ?shard_meta.txn_file_locks(), "err" => ?err); Error::Other(box_err!("get txn file key errors failed")) })?; return Err(Error::KeyErrors(lock_infos_to_key_errors(lock_infos))); } } Ok(()) } pub(crate) fn on_delete_prefix( &mut self, region_version: u64, prefix: Vec, callback: Callback, ) { if !self.peer.is_leader() { // As delete prefix requests are sent to all stores, not leader error can be // ignored. callback.invoke_with_response(RaftCmdResponse::default()); return; } let tag = self.peer.tag(); let id_ver = tag.id_ver; if region_version != id_ver.ver() { warn!("{} on_delete_prefix: version not match", tag); callback.invoke_with_response(new_error(Error::EpochNotMatch( "on_delete_prefix: version not match".to_owned(), vec![self.fsm.peer.region().clone()], ))); return; } let mut cmd = self.new_raft_cmd_request(); let mut cs = kvengine::new_change_set(id_ver.id(), id_ver.ver()); cs.set_property_key(DEL_PREFIXES_KEY.to_string()); cs.set_property_value(prefix); cs.set_property_merge(true); let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&cs); cmd.set_custom_request(custom_builder.build()); self.propose_raft_command(cmd, TraceContext::default(), callback, None); } pub(crate) fn on_manual_major_compact( &mut self, major_compact: bool, columnar: bool, callback: Callback, ) { if !self.peer.is_leader() { callback.invoke_with_response(RaftCmdResponse::default()); return; } let id_ver = self.peer.tag().id_ver; let mut cmd = self.new_raft_cmd_request(); let mut cs = kvengine::new_change_set(id_ver.id(), id_ver.ver()); cs.set_property_key(MANUAL_MAJOR_COMPACTION.to_string()); if major_compact { if columnar { cs.set_property_value(MANUAL_MAJOR_COMPACTION_ENABLE_COLUMNAR.to_vec()); } else { cs.set_property_value(MANUAL_MAJOR_COMPACTION_ENABLE.to_vec()); } } else { cs.set_property_value(MANUAL_MAJOR_COMPACTION_DISABLE.to_vec()); } let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&cs); cmd.set_custom_request(custom_builder.build()); self.propose_raft_command(cmd, TraceContext::default(), callback, None); } fn on_force_switch_mem_table(&mut self, current_size: u64, callback: Callback) { if !self.peer.is_leader() { callback.invoke_with_response(RaftCmdResponse::default()); return; } // Propose a command with TYPE_SWITCH_MEM_TABLE let mut req = self.new_raft_cmd_request(); let mut builder = CustomBuilder::new(); builder.set_switch_mem_table(current_size); req.set_custom_request(builder.build()); self.propose_raft_command(req, TraceContext::default(), callback, None); } pub(crate) fn on_update_schema_file(&mut self, schema_file: SchemaFile) { if !self.peer.is_leader() { return; } let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap(); let overlap = schema_file.overlap( &shard_meta.range.outer_start, &shard_meta.range.outer_end, shard_meta.range.keyspace_id, ); let tag = self.peer.tag(); let mut change_set = kvengine::new_change_set(shard_meta.id, shard_meta.ver); let update_schema_meta = change_set.mut_update_schema_meta(); update_schema_meta.set_file_id(schema_file.get_file_id()); update_schema_meta.set_version(schema_file.get_version()); if shard_meta.schema.restore_ver() != schema_file.get_restore_version() { info!( "{} skip stale schema file id {}, restore_version not match, expected {}, actual {}", tag, schema_file.get_file_id(), shard_meta.schema.restore_ver(), schema_file.get_restore_version() ); return; } update_schema_meta.set_restore_version(schema_file.get_restore_version()); info!( "{} on_update_schema_file {:?}, file_ver: {}", tag, shard_meta.schema, schema_file.get_version() ); let cur_ver = shard_meta.schema.file_ver(); let new_ver = schema_file.get_version(); let valid = shard_meta.schema.is_valid(); // If the schema file is valid, we should skip if the new_ver is not newer. // In `clear_columnar` case, the schema file is invalid and we need to update // the schema file with the same version. let should_skip = if valid { cur_ver >= new_ver } else { cur_ver > new_ver }; if should_skip { info!( "{} skip stale schema file (cur_ver={}, new_ver={}, valid={})", tag, cur_ver, new_ver, valid ); return; } if !overlap { if shard_meta.schema.is_valid() { info!("{} update schema file, remove non overlap", tag); update_schema_meta.set_file_id(0); } else { return; } } let kv = self.ctx.global.engines.kv.clone(); // Ensure the shard has been initial flushed. If not initial flushed, reject the // proposal. Update schema file will be proposed again by schema manager. if let Some(shard) = kv.get_shard(shard_meta.id) { if !shard.get_initial_flushed() { warn!( "{} shard not initial flushed, reject propose update schema file {}", tag, schema_file.get_file_id() ); return; } } info!( "{} propose update schema file {:?}", tag, update_schema_meta ); self.propose_change_set(change_set, Callback::None); } fn check_schema(&mut self, shard: &Arc, is_leader: bool) { let tag = self.peer.tag(); let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap(); let schema_meta = &shard_meta.schema; let schema_file = shard.get_schema_file(); let schema_file_id = schema_file.as_ref().map(|x| x.get_file_id()); debug!("{} check schema", tag; "schema_file_meta" => ?schema_meta, "meta.spec" => ?shard_meta.get_storage_class_spec(), "shard.spec" => ?shard.get_storage_class_spec(), "checked_schema_ver" => shard.get_checked_schema_ver(), "schema_file" => ?schema_file_id, ); if !schema_file_is_matched_with_meta(schema_file.as_ref(), schema_meta) { // Wait for schema file to be updated. debug!("{} check schema: skip, schema file is stale", tag; "schema_meta" => ?schema_meta, "schema_file" => ?schema_file_id); return; } if is_leader && !shard_is_matched_with_meta(shard.as_ref(), shard_meta) { let task = SchemaTask::StorageClass { region: self.region().clone(), schema_meta: schema_meta.clone(), }; if let Err(e) = self.ctx.global.schema_scheduler.schedule(task) { info!("{} check schema: schedule update sc task failed", tag; "err" => ?e, "schema_meta" => ?schema_meta); } } else if shard.should_try_storage_class_transition() { let tables_transited_to_ia = shard.try_transit_storage_class(); shard.set_last_transit_storage_class_instant_to_now(); if tables_transited_to_ia > 0 { shard.refresh_estimated_size_and_entries(); report_transitions(tables_transited_to_ia); info!("{} check schema: transit storage class", tag; "tables_transited_to_ia" => tables_transited_to_ia); } } else { debug!("{} check schema: skip, shard is up-to-date", tag); }; } pub(crate) fn on_check_leader(&mut self, shard_ver: u64, callback: Callback) { let mut resp = RaftCmdResponse::default(); // If the peer is not leader or not applied to current term, return not_leader // error. The client should sleep and retry. let is_leader = fail::eval("check_leader_peer_is_leader", |t| { t.and_then(|s: String| s.parse::().ok()) }) .flatten() .unwrap_or(self.peer.is_leader()); if !is_leader || !self.peer.has_applied_to_current_term() { self.start_cold_request_recovery(); let header = resp.mut_header(); let error = header.mut_error(); let not_leader = error.mut_not_leader(); not_leader.set_region_id(self.region_id()); let leader_id = self.peer.leader_id(); if leader_id != 0 { if let Some(p) = self.region().get_peers().iter().find(|p| p.id == leader_id) { not_leader.set_leader(p.clone()); } } } else if shard_ver != 0 && self.region().get_region_epoch().version != shard_ver { let header = resp.mut_header(); let error = header.mut_error(); let epoch_not_match = error.mut_epoch_not_match(); epoch_not_match .mut_current_regions() .push(self.region().clone()); } callback.invoke_with_response(resp); } fn on_clear_columnar(&mut self, restore_version: Option) { if !self.peer.is_leader() { return; } let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap(); if !shard_meta.schema.has_value() { return; } let mut change_set = kvengine::new_change_set(shard_meta.id, shard_meta.ver); let clear_columnar = change_set.mut_clear_columnar_with_restore_version(); if let Some(restore_version) = restore_version { clear_columnar.set_restore_version(restore_version); } else { clear_columnar.set_restore_version(shard_meta.schema.restore_ver()); } info!("{} propose clear_columnar", self.peer.tag()); self.propose_change_set(change_set, Callback::None); } fn on_clear_fts_index(&mut self) { if !self.peer.is_leader() { return; } let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap(); if !shard_meta.schema.has_value() { return; } let mut change_set = kvengine::new_change_set(shard_meta.id, shard_meta.ver); let fts_update = change_set.mut_fts_update(); fts_update.set_clear_all(true); info!("{} propose clear_fts_index", self.peer.tag()); self.propose_change_set(change_set, Callback::None); } pub(crate) fn on_ingest_files(&mut self, mut cs: kvenginepb::ChangeSet, callback: Callback) { if !self.peer.is_leader() { self.start_cold_request_recovery(); let mut resp = RaftCmdResponse::default(); let header = resp.mut_header(); let error = header.mut_error(); let not_leader = error.mut_not_leader(); not_leader.set_region_id(self.region_id()); let leader_id = self.peer.leader_id(); if leader_id != 0 { if let Some(p) = self.region().get_peers().iter().find(|p| p.id == leader_id) { not_leader.set_leader(p.clone()); } } callback.invoke_with_response(resp); return; } if !self.get_peer().get_store().initial_flushed() { let mut resp = RaftCmdResponse::default(); let not_initialized = resp.mut_header().mut_error().mut_region_not_initialized(); not_initialized.set_region_id(self.region_id()); callback.invoke_with_response(resp); return; } if cs.get_shard_ver() != self.region().get_region_epoch().get_version() { let mut resp = RaftCmdResponse::default(); let regions = resp .mut_header() .mut_error() .mut_epoch_not_match() .mut_current_regions(); regions.push(self.region().clone()); callback.invoke_with_response(resp); return; } let shard_meta = self.peer.get_store().shard_meta.as_ref().unwrap(); let ingest_files = cs.mut_ingest_files(); // Adjust the ingest level to avoid overlap with existing files. The // level here uses a pre-proposal `shard_meta` snapshot and may be // stale; apply-time revalidation by // `ShardMeta::check_overlap_for_ingest` in `preprocess_change_set` // rejects with `IngestOverlap` if it would create same-level overlap. for table_create in ingest_files.mut_table_creates().iter_mut() { debug_assert!(table_create.get_level() == WRITE_CF_BOTTOM_LEVEL); let table_bound = DataBound::new( InnerKey::from_inner_buf(table_create.get_smallest()), InnerKey::from_inner_buf(table_create.get_biggest()), true, ); let ingest_level = shard_meta.get_ingest_level(table_bound); // L0 uses a different file format incompatible with regular SST ingestion. // Trigger compaction to move data to L3, then client can retry. // // If any file would go to L0, trigger compaction and return early. // Other files in the same ChangeSet are not processed in this request. // The client will retry the entire batch after compaction completes. if ingest_level == 0 { self.handle_l0_ingest_overlap( shard_meta.id, shard_meta.ver, table_create.get_id(), callback, ); // Early return is ok because compaction changes the entire // shard's file state. All files' ingest_levels need to be recalculated // based on the new state after compaction completes, so we must let // the client retry the entire batch rather than processing remaining // files with stale state. return; } table_create.set_level(ingest_level); } let mut cmd = self.new_raft_cmd_request(); let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&cs); cmd.set_custom_request(custom_builder.build()); self.propose_raft_command(cmd, TraceContext::default(), callback, None); } /// Handle ingest that would go to L0 by triggering manual major compaction. /// /// When `ingest_level == 0` the file cannot be ingested directly because L0 /// uses a different file format. We propose a `MANUAL_MAJOR_COMPACTION` /// property change and return `ServerIsBusy` so the client retries after /// compaction. /// /// We do not gate this propose on whether a major compaction is already in /// flight: `CompactionRunner.compact` dedups per shard via /// `is_compacting = running ∪ notified`, so a burst of concurrent ingest /// proposes always collapses into exactly one major compaction. Gating on /// `MANUAL_MAJOR_COMPACTION` property or `shard.compacting` instead would /// couple this path to the compaction state machine and risk a stuck state /// when a terminal compaction failure leaves the flag set. fn handle_l0_ingest_overlap( &mut self, shard_id: u64, shard_ver: u64, file_id: u64, callback: Callback, ) { info!( "ingest triggered manual major compaction due to L0 overlap"; "region_id" => self.region_id(), "file_id" => file_id ); let mut compact_cs = kvengine::new_change_set(shard_id, shard_ver); compact_cs.set_property_key(MANUAL_MAJOR_COMPACTION.to_string()); compact_cs.set_property_value(MANUAL_MAJOR_COMPACTION_ENABLE.to_vec()); let mut cmd = self.new_raft_cmd_request(); let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&compact_cs); cmd.set_custom_request(custom_builder.build()); let region_id = self.region_id(); let tag = self.peer.tag(); self.propose_raft_command( cmd, TraceContext::default(), Callback::write(Box::new(move |resp| { if resp.response.get_header().has_error() { warn!( "failed to propose manual major compaction for ingest"; "tag" => %tag, "region_id" => region_id, "error" => ?resp.response.get_header().get_error() ); INGEST_COMPACTION_PROPOSAL_FAILED_COUNTER.inc(); } })), None, ); INGEST_TRIGGER_COMPACTION_COUNTER.inc(); let mut resp = RaftCmdResponse::default(); resp.mut_header() .mut_error() .set_server_is_busy(new_server_is_busy( ServerIsBusyType::RaftstoreIngest, "compaction in progress for L0 ingest overlap", )); callback.invoke_with_response(resp); } fn trigger_trim_over_bound(&mut self, shard_ver: u64, parameter: TrimOverBoundParameter) { if !self.peer.is_leader() { return; } let tag = self.peer.tag(); let id_ver = tag.id_ver; if shard_ver != id_ver.ver() { warn!("{} trigger_trim_over_bound version not match", tag); return; } let mut cmd = self.new_raft_cmd_request(); let mut custom_builder = CustomBuilder::new(); custom_builder.set_trigger_trim_over_bound(¶meter); cmd.set_custom_request(custom_builder.build()); let cb = Callback::write(Box::new(move |resp| { if resp.response.get_header().has_error() { let err_msg = resp.response.get_header().get_error().get_message(); warn!("{} failed to trigger trim_over_bound: {:?}", tag, err_msg); } else { info!( "{} trigger trim_over_bound successfully: {:?}", tag, parameter ); } })); self.propose_raft_command(cmd, TraceContext::default(), cb, None); } fn on_trigger_refresh_shard_states(&mut self) { if !self.peer.is_leader() { return; } self.ctx .apply_msgs .msgs .push(ApplyMsg::TriggerRefreshShardStates); } fn on_pd_heartbeat_tick(&mut self) { fail_point!("on_pd_heartbeat_tick"); self.ticker.schedule(PEER_TICK_PD_HEARTBEAT); self.fsm.peer.check_peers(); self.update_max_lag_metrics(); if !self.fsm.peer.is_leader() { return; } self.fsm.peer.clear_region_meta_heartbeat_pending(); self.fsm.peer.heartbeat_pd(self.ctx, true); } fn on_generate_engine_change_set(&mut self, cs: kvenginepb::ChangeSet, cb: Callback) { let tag = self.peer.tag(); info!("generate meta change event {:?}", &cs; "region" => tag); self.propose_change_set(cs, cb); } fn propose_change_set(&mut self, cs: kvenginepb::ChangeSet, cb: Callback) { let tag = self.peer.tag(); let mut req = self.new_raft_cmd_request(); let mut builder = CustomBuilder::new(); builder.set_change_set(&cs); let custom_req = builder.build(); req.set_custom_request(custom_req); let router = self.ctx.global.router.clone(); let kv = self.ctx.global.engines.kv.clone(); let propose_cb = Callback::write(Box::new(move |resp| { if resp.response.get_header().has_error() { let err_msg = resp.response.get_header().get_error().get_message(); warn!("{} failed to propose engine change set", tag; "err" => ?err_msg, "cs" => ?cs); if err_msg.contains("raft: proposal dropped") { // Proposal may dropped due to leader transfer in progress. router.send_store_engine_change_set(cs, cb); } else { // Reset the state of flush worker or compactor. kv.meta_committed(&cs, true); cb.invoke_with_response(resp.response); } } else { info!("{} proposed meta change event", tag); cb.invoke_with_response(resp.response); } })); self.propose_raft_command(req, TraceContext::default(), propose_cb, None); self.ctx.raft_metrics.propose.change_set.inc(); } fn new_raft_cmd_request(&self) -> RaftCmdRequest { let mut req = RaftCmdRequest::default(); let mut header = RaftRequestHeader::default(); header.set_region_id(self.region_id()); header.set_peer(self.peer.peer.clone()); header.set_region_epoch(self.region().get_region_epoch().clone()); header.set_term(self.peer.term()); req.set_header(header); req } fn on_apply_snapshot_result(&mut self, change: kvenginepb::ChangeSet) { let tag = self.peer.tag(); debug_assert!(change.has_snapshot(), "{}", tag); assert_eq!( change.shard_ver, self.peer .get_preprocessed_region() .get_region_epoch() .get_version(), "{} snapshot change set version not match, region {:?}, is_applying_snapshot {}, cs {:?}", tag, self.peer.get_preprocessed_region(), self.peer.get_store().is_applying_snapshot(), change ); if self.peer.mut_store().is_applying_snapshot() { self.peer.mut_store().snap_state = SnapState::Relax; // After applying a snapshot, merge is rollbacked implicitly. self.peer.clear_merge_in_mem_data(); let apply_state = RaftApplyState::from_snapshot(change.get_snapshot()); let apply_result = MsgApplyResult { peer_id: self.peer.peer_id(), results: VecDeque::new(), apply_state, metrics: ApplyMetrics::default(), bucket_stat: None, encryption_paused_index: None, }; info!( "{} apply snapshot finished, snapshot index {}", tag, apply_state.applied_index ); self.fsm.peer.post_apply(self.ctx, &apply_result); let on_apply_snapshot_msgs = std::mem::take(&mut self.peer.mut_store().on_apply_snapshot_msgs); for msg in on_apply_snapshot_msgs { if let Err(err) = self.ctx.global.trans.send(msg) { error!("failed to send on apply snapshot msg {:?}", err); } } } } fn on_persisted(&mut self, ready: PersistReady) { if ready.peer_id != self.fsm.peer_id() { error!( "peer id not match"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "persisted_peer_id" => ready.peer_id, "persisted_number" => ready.ready_number, ); return; } self.fsm.peer.on_persist_ready(self.ctx, ready.ready_number); } pub(crate) fn on_prepared_change_set( &mut self, res: kvengine::Result, peer_id: u64, ) { let change_set = match res { Ok(change_set) => change_set, Err(err) => { // TODO(x): properly handle this error. panic!("{} failed to prepare change set {:?}", self.peer.tag(), err); } }; // If peer_msg is from a stale peer, ignore it. prepare_change_set in kvengine // will be return empty change set if old peer has be destroyed. if peer_id != self.peer_id() { warn!( "{} peer id not match {} != {}, skip on_prepared_change_set", self.peer.tag(), peer_id, self.peer_id() ); return; } self.ctx .apply_msgs .msgs .push(ApplyMsg::ApplyChangeSet(change_set)); } pub(crate) fn on_prepared_commit_merge( &mut self, res: kvengine::Result, commit_index: u64, ) { let change_set = match res { Ok(change_set) => change_set, Err(err) => { // TODO(x): properly handle this error. panic!("{} failed to prepare change set {:?}", self.peer.tag(), err); } }; self.ctx.apply_msgs.msgs.push(ApplyMsg::ResumeCommitMerge { source: change_set, commit_index, }); } pub(crate) fn on_prepared_txn_file( &mut self, entry_index: u64, peer_id: u64, apply_generation: u64, txn_file: kvengine::table::TxnFile, ) { // Ignore peer_msg from a stale peer. if peer_id != self.peer_id() { warn!( "{} peer id not match {} != {}, skip on_prepared_txn_file, entry_index {}", self.peer.tag(), peer_id, self.peer_id(), entry_index, ); return; } self.ctx.apply_msgs.msgs.push(ApplyMsg::ResumeTxnFile { commit_index: entry_index, apply_generation, txn_file, }); } /// Handle completion of async raft log fetch, similar to raftstore's /// implementation pub(crate) fn on_raft_log_fetched( &mut self, context: GetEntriesContext, res: Box, ) { let low = res.low; // If the peer is not the leader anymore or is being destroyed, ignore the // result. if !self.peer.is_leader() || self.peer.pending_remove { self.peer.mut_store().clean_async_fetch_res(low); return; } if self.peer.term() != res.term { // term has changed, the result may be not correct. self.peer.mut_store().clean_async_fetch_res(low); return; } // Update async fetch result self.peer.mut_store().update_async_fetch_res(low, Some(res)); // Notify the raft group that entries have been fetched self.peer.raft_group.on_entries_fetched(context); // Clean the async fetch result immediately after use to free memory self.peer.mut_store().clean_async_fetch_res(low); info!( "{} async raft log fetch completed and processed", self.peer.tag(); "low" => low, ); } pub(crate) fn update_max_lag_metrics(&mut self) { if !self.fsm.peer.is_leader() { return; } let last_idx = self.peer.get_store().last_index(); let replicated_idx = self .peer .raft_group .raft .prs() .iter() .map(|(_, pr)| pr.matched) .min() .unwrap_or_default(); // When an election happened or a new peer is added, replicated_idx can be 0. if replicated_idx > 0 { assert!( last_idx >= replicated_idx, "expect last index {} >= replicated index {}", last_idx, replicated_idx ); raftstore::store::metrics::REGION_MAX_LOG_LAG .observe((last_idx - replicated_idx) as f64); } } fn on_raft_log_gc_tick(&mut self) { fail_point!("on_raft_log_gc_tick", |_| ()); self.ticker.schedule(PEER_TICK_RAFT_LOG_GC); let region_id = self.region_id(); let engines = &self.ctx.global.engines; if !self.peer.is_initialized() || self.peer.is_applying_snapshot() || engines.raft.has_dependents(region_id) || self.peer.pending_merge_state.is_some() { return; } let last_idx = self.peer.get_store().last_index(); let replicated_idx = if self.peer.is_leader() { self.peer .raft_group .raft .prs() .iter() .filter_map(|(peer_id, pr)| { // Don't keep raft logs for down peer. It may be too long(default 10mins). (!self.peer.down_peer_ids.contains(peer_id)).then_some(pr.matched) }) .min() .unwrap_or(last_idx) } else { // Followers can't know the progress of other replicas, so they truncate logs as // many as possible, but it may result in snapshot transport when // becomes leader. One solution is that leader propagates truncated // index to followers. last_idx }; let peer_id = self.peer_id(); let size_limit_index = engines.raft.index_to_truncate_to_size( peer_id, self.ctx.cfg.raft_log_gc_size_limit.unwrap().0 as usize, ); // If raft logs occupies too much memory, truncates them regardless of lagged // replicas to avoid OOM. let mut to_truncate_idx = cmp::max(size_limit_index, replicated_idx); // Shouldn't truncate unapplied logs. let applied_idx = self.peer.get_store().applied_index(); to_truncate_idx = cmp::min(to_truncate_idx, applied_idx); // Shouldn't truncate unpersisted logs. let mut persisted_log_idx = self.peer.get_store().data_persisted_log_index().unwrap(); if persisted_log_idx < to_truncate_idx && applied_idx == last_idx { // It's possible that the shard mem-table is empty but the data sequence is // smaller. All the entries between persisted_log_idx and last_idx are no kv // entries, there is no chance to trigger flush, the data sequence fall behind. let Some(shard) = self.ctx.global.engines.kv.get_shard(self.region_id()) else { return; }; if shard.data_all_persisted() { // All raft logs are applied and mem-table is empty, we can safely truncate to // 'to_truncate_idx'. But we don't want to pay the cost of // rewriting shard meta on each leader change. So we tolerate // config.raft_log_gc_no_kv_count to reduce the frequency of shard meta // rewriting. let truncate_no_kv_logs = persisted_log_idx + self.ctx.cfg.raft_log_gc_no_kv_count < to_truncate_idx; // change set raft logs are less frequent, we can always truncate it. let truncate_change_set_logs = shard.get_write_sequence() > persisted_log_idx; if truncate_no_kv_logs || truncate_change_set_logs { // Advance the data sequence to to_truncate_idx. let store = self.peer.mut_store(); let term = store.term(to_truncate_idx).unwrap(); // workaround for borrow checker. let mut shard_meta = store.shard_meta.take().unwrap(); shard_meta.data_sequence = to_truncate_idx; shard_meta.set_property(TERM_KEY, &term.to_le_bytes()); if self.ctx.cfg.enable_kv_engine_meta_diff { write_engine_meta_diff( &self.ctx.global.engines.raft, &mut self.ctx.raft_wb, peer_id, &shard_meta, None, self.ctx.cfg.kv_engine_meta_diff_rewrite_percent, ); } else { write_engine_meta(&mut self.ctx.raft_wb, peer_id, &shard_meta); } self.peer.mut_store().shard_meta = Some(shard_meta); persisted_log_idx = to_truncate_idx; info!( "{} shard meta advanced data sequence to {}", self.peer.tag(), to_truncate_idx, ); } } } to_truncate_idx = cmp::min(to_truncate_idx, persisted_log_idx); // Check if we need to handle pending_truncate. if let Some((_, idx)) = self.peer.pending_truncate { if idx <= applied_idx { to_truncate_idx = idx; self.peer.pending_truncate = None; } } let truncated_idx = self.peer.get_store().truncated_index(); if to_truncate_idx > truncated_idx { let to_truncate_term = self.peer.get_store().term(to_truncate_idx).unwrap(); // mem::take and mem::replace is a workaround for borrow checker. let mut raft_wb = std::mem::take(&mut self.ctx.raft_wb); self.peer.mut_store().truncate_raft_log( &mut raft_wb, to_truncate_idx, to_truncate_term, ); let _ = std::mem::replace(&mut self.ctx.raft_wb, raft_wb); info!( "{} truncate raft logs to {}, term {}", self.peer.tag(), to_truncate_idx, to_truncate_term, ); } } fn on_check_peer_stale_state(&mut self, report_leader_missing: bool) { if self.fsm.peer.is_applying_snapshot() || self.fsm.peer.has_pending_snapshot() { return; } // If this peer detects the leader is missing for a long long time, // it should consider itself as a stale peer which is removed from // the original cluster. // This most likely happens in the following scenario: // At first, there are three peer A, B, C in the cluster, and A is leader. // Peer B gets down. And then A adds D, E, F into the cluster. // Peer D becomes leader of the new cluster, and then removes peer A, B, C. // After all these peer in and out, now the cluster has peer D, E, F. // If peer B goes up at this moment, it still thinks it is one of the cluster // and has peers A, C. However, it could not reach A, C since they are removed // from the cluster or probably destroyed. // Meantime, D, E, F would not reach B, since it's not in the cluster anymore. // In this case, peer B would notice that the leader is missing for a long time, // and it would check with pd to confirm whether it's still a member of the // cluster. If not, it destroys itself as a stale peer which is removed // out already. let state = self.fsm.peer.check_stale_state(&self.ctx.cfg); fail_point!("peer_check_stale_state", state != StaleState::Valid, |_| {}); match state { StaleState::Valid => (), StaleState::LeaderMissing if report_leader_missing => { warn!( "leader missing longer than abnormal_leader_missing_duration"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "expect" => %self.ctx.cfg.abnormal_leader_missing_duration, ); self.ctx .raft_metrics .leader_missing .lock() .unwrap() .insert(self.region_id()); } StaleState::LeaderMissing => {} StaleState::ToValidate => { // for peer B in case 1 above warn!( "leader missing longer than max_leader_missing_duration. \ To check with pd and other peers whether it's still valid"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "expect" => %self.ctx.cfg.max_leader_missing_duration, ); self.fsm.peer.bcast_check_stale_peer_message(self.ctx); self.schedule_validate_peer(); } } } fn schedule_validate_peer(&self) { let task = PdTask::ValidatePeer { peer: self.fsm.peer.peer.clone(), region: self.fsm.peer.region().clone(), }; if let Err(e) = self.ctx.global.pd_scheduler.schedule(task) { error!( "failed to notify pd"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "err" => %e, ) } } fn on_check_long_tick(&mut self) { fail_point!("on_check_long_tick"); if self.fsm.peer.pending_remove { return; } self.ticker.schedule(PEER_TICK_CHECK_LONG); self.on_check_peer_stale_state(true); self.on_check_failed_compaction(); self.gc_obsolete_state_versions(); } fn gc_obsolete_state_versions(&mut self) { let peer = &self.fsm.peer; let region_id = peer.region().get_id(); if !peer.is_initialized() || peer.is_applying_snapshot() || peer.has_pending_snapshot() || peer.pending_merge_state.is_some() // Restart recovery still needs the parent version's states until // the current shard completes its initial flush. || !peer.get_store().initial_flushed() || self.ctx.global.engines.raft.has_dependents(region_id) { return; } let Some(shard_meta) = peer.get_store().shard_meta.as_ref() else { return; }; let peer_id = peer.peer_id(); let current_version = shard_meta.ver; let keyspace_id = shard_meta.range.keyspace_id; let versions = self .ctx .global .engines .raft .collect_obsolete_state_versions( peer_id, current_version, OBSOLETE_VERSION_GC_BATCH_SIZE, ); if versions.is_empty() { return; } for &version in &versions { for key in [ rfengine::raft_state_key(version), rfengine::region_state_key(version), ] { self.ctx.raft_wb.set_state_bytes( peer_id, region_id, keyspace_id, key, Bytes::new(), ); } } info!( "gc obsolete rfengine state versions"; "tag" => peer.tag(), "peer_id" => peer_id, "current_version" => current_version, "first_version" => versions.first().unwrap(), "last_version" => versions.last().unwrap(), "version_count" => versions.len(), ); } fn on_check_failed_compaction(&mut self) { if !self.peer.is_leader() { return; } let kv = &self.ctx.global.engines.kv; let Some(shard) = kv.get_shard(self.region_id()) else { return; }; if shard.has_failed_compaction() { info!("{} re-trigger failed compaction", self.peer.tag()); kv.trigger_compact(shard.id_ver()); } } /// Check if merge target region is staler than the local one in kv engine. /// It should be called when target region is not in region map in memory. /// If everything is ok, the answer should always be true because PD should /// ensure all target peers exist. So if not, error log will be printed /// and return false. pub(crate) fn is_merge_target_region_stale( &self, target_region: &metapb::Region, ) -> Result { let target_peer_id = find_peer(target_region, self.ctx.store_id()) .unwrap() .get_id(); if let Some(target_state) = load_last_peer_state(&self.ctx.global.engines.raft, target_peer_id) { let state_epoch = target_state.get_region().get_region_epoch(); if util::is_epoch_stale(target_region.get_region_epoch(), state_epoch) { return Ok(true); } // The local target region epoch is staler than target region's. // In the case where the peer is destroyed by receiving gc msg rather than // applying conf change, the epoch may staler but it's legal, so // check peer id to assure that. if let Some(local_target_peer_id) = find_peer(target_state.get_region(), self.ctx.store_id()).map(|r| r.get_id()) { match local_target_peer_id.cmp(&target_peer_id) { cmp::Ordering::Equal => { if target_state.get_state() == PeerState::Tombstone { // The local target peer has already been destroyed. return Ok(true); } error!( "the local target peer state is not tombstone in kv engine"; "target_peer_id" => target_peer_id, "target_peer_state" => ?target_state.get_state(), "target_region" => ?target_region, "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); } cmp::Ordering::Greater => { if state_epoch.get_version() == 0 && state_epoch.get_conf_ver() == 0 { // There is a new peer and it's destroyed without being initialised. return Ok(true); } // The local target peer id is greater than the one in target region, but // its epoch is staler than target_region's. That is // contradictory. panic!("{} local target peer id {} is greater than the one in target region {}, but its epoch is staler, local target region {:?}, target region {:?}", self.fsm.peer.tag(), local_target_peer_id, target_peer_id, target_state.get_region(), target_region); } cmp::Ordering::Less => { error!( "the local target peer id in kv engine is less than the one in target region"; "local_target_peer_id" => local_target_peer_id, "target_peer_id" => target_peer_id, "target_region" => ?target_region, "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); } } } else { // Can't get local target peer id probably because this target peer is removed // by applying conf change error!( "the local target peer does not exist in target region state"; "target_region" => ?target_region, "local_target" => ?target_state.get_region(), "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); } } else { error!( "failed to load target peer's RegionLocalState from kv engine"; "target_peer_id" => target_peer_id, "target_region" => ?target_region, "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), ); } Ok(false) } fn validate_merge_peer( &self, target_region: &metapb::Region, store_meta: &mut StoreMeta, ) -> Result { let target_region_id = target_region.get_id(); let exist_region = { store_meta.region_map.get(target_region_id).cloned() }; if let Some(r) = exist_region { let exist_epoch = r.get_region_epoch(); let expect_epoch = target_region.get_region_epoch(); // exist_epoch > expect_epoch if util::is_epoch_stale(expect_epoch, exist_epoch) { return Err(box_err!( "target region changed {:?} -> {:?}", target_region, r )); } // exist_epoch < expect_epoch if util::is_epoch_stale(exist_epoch, expect_epoch) { info!( "target region still not catch up, skip."; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "target_region" => ?target_region, "exist_region" => ?r, ); return Ok(false); } return Ok(true); } // All of the target peers must exist before merging which is guaranteed by PD. // Now the target peer is not in region map. match self.is_merge_target_region_stale(target_region) { Err(e) => { error!(%e; "failed to load region state, ignore"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "target_region_id" => target_region_id, ); Ok(false) } Ok(true) => Err(box_err!("region {} is destroyed", target_region_id)), Ok(false) => { error!( "something is wrong, maybe PD do not ensure all target peers exist before merging" ); Ok(false) } } } fn schedule_merge(&mut self, store_meta: &mut StoreMeta) -> Result<()> { fail_point!("on_schedule_merge_error", |_| { // Return error to trigger rollback merge. Err(box_err!("on_schedule_merge_error failpoint")) }); let (request, target_id) = { let state = self.fsm.peer.pending_merge_state.as_ref().unwrap(); let expect_region = state.get_target(); // NOTE: Should NOT check the validation of merge with target region. // The state of target region would not be consistent between peers, and lead to // the merge be rollback & commit at the same time. // See https://github.com/tidbcloud/cloud-storage-engine/issues/2503. if let Some(source_meta) = self.fsm.peer.get_store().shard_meta.as_ref() { if source_meta.has_txn_file_locks() { let tag = self.peer.tag(); info!("{} fail to schedule merge: source region has txn file locks", tag; "target" => ?expect_region, "txn_file_locks" => %source_meta.txn_file_locks(), ); return Err(box_err!( "{}: {}", tag, MERGE_REGION_WITH_TXN_FILE_LOCKS_ERR_MSG )); } // If source region has unconverted l0s, wait for retry instead of returning // error. These l0s may be compacted during merge. This avoids inconsistent // rollback/commit decisions across peers. if !source_meta.unconverted_l0s.is_empty() { let tag = self.peer.tag(); info!("{} fail to schedule merge: source region has unconverted l0s, retry", tag; "target" => ?expect_region, "unconverted_l0s" => ?source_meta.unconverted_l0s, ); return Ok(()); } } if !self.validate_merge_peer(expect_region, store_meta)? { // Wait till next round. return Ok(()); } let target_id = expect_region.get_id(); let sibling_region = expect_region; let sibling_peer = find_peer(sibling_region, self.store_id()).unwrap(); let mut request = new_admin_request(sibling_region.get_id(), sibling_peer.clone()); request .mut_header() .set_region_epoch(sibling_region.get_region_epoch().clone()); let mut admin = AdminRequest::default(); admin.set_cmd_type(AdminCmdType::CommitMerge); admin .mut_commit_merge() .set_source(self.fsm.peer.get_preprocessed_region().clone()); admin.mut_commit_merge().set_commit(state.get_commit()); // Fetch the prepare_merge entry for learner. let entries = self .peer .get_store() .entries( state.commit, state.commit + 1, None, GetEntriesContext::empty(false), ) .unwrap(); admin.mut_commit_merge().set_entries(entries.into()); let source_meta = self .fsm .peer .get_store() .shard_meta .as_ref() .unwrap() .to_change_set(); admin .mut_commit_merge() .set_source_meta(source_meta.write_to_bytes().unwrap()); request.set_admin_request(admin); (request, target_id) }; // Please note that, here assumes that the unit of network isolation is store // rather than peer. So a quorum stores of source region should also be // the quorum stores of target region. Otherwise we need to enable // proposal forwarding. self.ctx.global.router.send( target_id, PeerMsg::RaftCommand(RaftCommand::new( request, TraceContext::default(), Callback::None, )), ); Ok(()) } fn rollback_merge(&mut self) { let req = { let state = self.fsm.peer.pending_merge_state.as_ref().unwrap(); let mut request = new_admin_request(self.fsm.peer.region().get_id(), self.fsm.peer.peer.clone()); request .mut_header() .set_region_epoch(self.fsm.peer.region().get_region_epoch().clone()); let mut admin = AdminRequest::default(); admin.set_cmd_type(AdminCmdType::RollbackMerge); admin.mut_rollback_merge().set_commit(state.get_commit()); request.set_admin_request(admin); request }; self.propose_raft_command(req, TraceContext::default(), Callback::None, None); } pub(crate) fn on_check_merge(&mut self, store_meta: &mut StoreMeta) { if self.fsm.stopped || self.fsm.peer.pending_remove || self.fsm.peer.pending_merge_state.is_none() { return; } fail_point!( "on_check_merge_not_1001", self.fsm.peer_id() != 1001, |_| {} ); if let Err(e) = self.schedule_merge(store_meta) { if self.fsm.peer.is_leader() { self.fsm .peer .add_want_rollback_merge_peer(self.fsm.peer_id()); if self .fsm .peer .raft_group .raft .prs() .has_quorum(&self.fsm.peer.want_rollback_merge_peers) { info!( "failed to schedule merge, rollback"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "err" => %e, "error_code" => %e.error_code(), ); self.rollback_merge(); } } else if !is_learner(&self.fsm.peer.peer) { info!( "want to rollback merge"; "tag" => self.peer.tag(), "peer_id" => self.fsm.peer_id(), "leader_id" => self.fsm.peer.leader_id(), "err" => %e, "error_code" => %e.error_code(), ); if self.fsm.peer.leader_id() != raft::INVALID_ID { self.fsm.peer.send_want_rollback_merge( self.fsm .peer .pending_merge_state .as_ref() .unwrap() .get_commit(), self.ctx, ); } } } } pub(crate) fn on_restore_shard( &mut self, mut cs: kvenginepb::ChangeSet, expected_conf_ver: Option, callback: Callback, ) { let tag = self.peer.tag(); let region = self.peer.get_preprocessed_region(); // Check request. if !cs.has_restore_shard() { callback.invoke_with_response(message_error("invalid changeset")); return; } // The start key of first region is b"", but `encode_bytes(b""") == // 0000000000000000F7`. let encoded_start_key = encode_bytes_maybe_empty(cs.get_restore_shard().get_outer_start()); let encoded_end_key = encode_bytes(cs.get_restore_shard().get_outer_end()); if encoded_start_key != region.get_start_key() || encoded_end_key != region.get_end_key() { let err_msg = format!( "invalid snapshot range: [{:?},{:?}), expect: [{:?},{:?})", encoded_start_key, encoded_end_key, region.get_start_key(), region.get_end_key() ); callback.invoke_with_response(message_error(err_msg)); return; } // Check peer. if !self.peer.is_leader() { self.start_cold_request_recovery(); callback.invoke_with_response(new_error(Error::NotLeader( self.peer.region_id, self.fsm.peer.get_peer_from_cache(self.fsm.peer.leader_id()), ))); return; } if cs.shard_ver != region.get_region_epoch().get_version() { callback.invoke_with_response(new_error(Error::EpochNotMatch( "restore_shard".to_owned(), vec![region.clone()], ))); return; } if let Some(expected_conf_ver) = expected_conf_ver { let actual_conf_ver = region.get_region_epoch().get_conf_ver(); if actual_conf_ver != expected_conf_ver || self.fsm.peer.raft_group.raft.has_pending_conf() { callback.invoke_with_response(new_error(Error::EpochNotMatch( "restore_shard conf version fence".to_owned(), vec![region.clone()], ))); return; } } let shard = match self.ctx.global.engines.kv.get_shard(self.region_id()) { // initial flushed is necessary ? Some(shard) if !shard.get_initial_flushed() => { let err_msg = format!("{tag} not initial flushed, try again"); callback.invoke_with_response(message_error(err_msg)); return; } None => { let err_msg = format!("{tag} shard not found"); error!("{}", err_msg); callback.invoke_with_response(message_error(err_msg)); return; } Some(shard) => shard, }; debug!("{} on_restore_shard, original changeset: {:?}", tag, cs); // Adjust base version. let snap = cs.mut_restore_shard(); let write_seq = shard.get_write_sequence(); let old_mem_tbl_version = shard.get_base_version() + write_seq; let new_mem_tbl_version = cmp::max(snap.base_version + snap.data_sequence, old_mem_tbl_version); snap.set_base_version(new_mem_tbl_version - write_seq); snap.set_data_sequence(write_seq); // not necessary but just keep fields consistency. info!("{} on_restore_shard, adjusted changeset: {:?}", tag, cs); // Set `learner_skip_idx` to skip "restore_shard" replicating to learner. // NOTE: Next `MsgAppend` use the `last_index` other than `next_proposal_index`. let new_idx = self.fsm.peer.raft_group.raft.raft_log.last_index(); let old_idx = mem::replace(&mut self.fsm.peer.learner_skip_idx, new_idx); info!( "{} on_restore_shard, set peer.learner_skip_idx {}, old idx {}", tag, new_idx, old_idx ); let mut cmd = self.new_raft_cmd_request(); let mut custom_builder = CustomBuilder::new(); custom_builder.set_change_set(&cs); cmd.set_custom_request(custom_builder.build()); self.propose_raft_command(cmd, TraceContext::default(), callback, None); } fn check_gc_compaction(&self) { if !self.peer.is_leader() { return; } let kv = &self.ctx.global.engines.kv; if kv.is_critical_low_space() { debug!("{} skip gc compaction because kvengine disk space is critically low", self.fsm.peer.tag(); "available" => kv.available_space()); return; } if let Some(shard) = kv.get_shard(self.region_id()) { let gc_safe_point = kv.get_gc_safe_point(shard.keyspace_id).into_inner(); if shard.check_need_gc_compaction(gc_safe_point, kv.opts.gc_lock_extra_cf) { kv.trigger_compact(shard.id_ver()); } } } fn on_file_gc_compaction_check_tick(&mut self) { self.ticker.schedule(PEER_TICK_FILE_GC_COMPACTION_CHECK); let kv = &self.ctx.global.engines.kv; // An online file-GC knob change leaves the cached wakeup hint stale, so // recompute it on the staggered slot. Do it before the leader gate: a // follower must also refresh in the window so it carries a fresh hint into // a later leadership -- the leader-only path below would never rebuild it. let changed_at = kv.opts.file_gc_compaction_config_changed_at(); if changed_at != 0 { if let Some(shard) = kv.get_shard(self.region_id()) { let now = tikv_util::time::timespec_to_ns(tikv_util::time::monotonic_raw_now()); let tick_ns = self .ctx .cfg .file_gc_compaction_check_tick_interval .0 .as_nanos() as u64; if should_refresh_file_gc_hint(shard.id, now, changed_at, tick_ns) { shard.refresh_candidate_hint(); } } } // Only the leader can propose ChangeSet to raft. Not counted as a // skip: a follower tick is not a trigger check, and counting every one // would swamp the counter with no diagnostic value. if !self.peer.is_leader() { return; } // Skip when disk space is critically low: a reclaim compaction writes // new SSTs before deleting the inputs, so triggering one here would // worsen the pressure. Mirrors `check_gc_compaction`. Counted because it // means file-GC is held off node-wide while reclaimable data piles up. if kv.is_critical_low_space() { kvengine::metrics::ENGINE_FILE_GC_COMPACTION_TRIGGER_SKIP_COUNTER .with_label_values(&["critical_low_space"]) .inc(); return; } let Some(shard) = kv.get_shard(self.region_id()) else { return; }; // `Engine::compact` bails on inactive / not-yet-flushed shards and // clobbers any priority we set. if !shard.ready_to_compact() { return; } // Yield to in-flight compaction, existing priority, and pending // higher-priority ops (mirrors `Shard::check_need_gc_compaction`). Not // counted: these are expected, per-tick-persistent yields whose counts // would swamp the skip counter with no diagnostic value. if shard.is_compacting() || shard.has_compaction_priority() || shard.has_pending_manual_major_compaction() || !shard.get_del_prefixes().is_empty() || shard.get_truncate_ts().is_some() || shard.get_trim_over_bound() { return; } let gc_safe_point = kv.get_gc_safe_point(shard.keyspace_id).into_inner(); self.try_trigger_file_gc(&shard, gc_safe_point); } /// Try to trigger a file-GC compaction on `shard`. The fast path checks the /// per-shard wakeup hint so an idle shard skips the WRITE_CF metadata walk /// with an O(1) compare. On a real scan it locates a candidate (L1+ /// deepest-first, else L0), then claims the per-node rate limit and the /// shard priority slot only when it will actually trigger. Only the /// `rate_limited` outcome bumps the skip counter here; the other /// "no work / yielded" outcomes (including a stale hint that parks) return /// silently to keep the counter a clean blocked-work signal. fn try_trigger_file_gc(&self, shard: &kvengine::Shard, gc_safe_point: u64) { let bump_skip = |reason: &str| { kvengine::metrics::ENGINE_FILE_GC_COMPACTION_TRIGGER_SKIP_COUNTER .with_label_values(&[reason]) .inc(); }; let kv = &self.ctx.global.engines.kv; // The wakeup hint is the min `max_ts` among files passing every static // gate `find_file_gc_candidate_level` and `find_file_gc_candidate_l0` // apply. `u64::MAX` means either no file qualifies, or the last // scan parked the hint there after a stale-hint race; both mean nothing // to trigger. O(1) compare, no metadata walk. A refresh recomputes it // from the live layout. let hint = shard.file_gc_candidate_min_max_ts(); if hint == u64::MAX { return; } // A static-eligible file exists, but its data is still newer than the // safe point, so nothing is reclaimable yet. O(1) compare, no walk. if gc_safe_point < hint { return; } // Skip the candidate scan entirely when within the per-node rate-limit // window: a trigger could not run anyway. Peek only -- do NOT park the // hint, so the scan still runs once the window opens. if kv.is_file_gc_compaction_trigger_rate_limited() { bump_skip("rate_limited"); return; } let candidate = if let Some((level, ratio)) = shard.find_file_gc_candidate_level(gc_safe_point) { Some((FileGcCandidate::L1Plus(level), ratio)) } else { shard .find_file_gc_candidate_l0(gc_safe_point) .map(|(candidate_id, ratio)| (FileGcCandidate::L0(candidate_id), ratio)) }; let Some((candidate, ratio)) = candidate else { // safe_point >= hint yet the scan found nothing: a stale hint. The // hint shares every static gate with the scan, so this is only // transient -- an online knob tightened a gate, or a compaction/flush // changed the layout between the hint read and the scan. Both self-heal // on the next refresh; park the hint to skip redundant walks until then // (park_file_gc_candidate_hint CASes so a fresh value is left intact). shard.park_file_gc_candidate_hint(hint); return; }; // Re-check the priority slot before claiming the rate-limit window, so a // window is consumed only when we will actually trigger. A concurrent op // may have taken the slot since the tick-start check. if shard.has_compaction_priority() { return; } // Per-node rate limit claim: at most one file-GC trigger per interval. if !kv.try_trigger_file_gc_compaction() { bump_skip("rate_limited"); return; } let set = match candidate { FileGcCandidate::L1Plus(level) => { shard.try_set_file_gc_triggered_l1_plus_priority(level, ratio) } FileGcCandidate::L0(candidate_id) => { shard.try_set_file_gc_triggered_l0_priority(candidate_id, ratio) } }; if !set { // Lost the priority slot in the narrow gap since the re-check. return; } kvengine::metrics::ENGINE_FILE_GC_COMPACTION_RATIO_HISTOGRAM.observe(ratio); // The trigger counter is bumped at Engine::compact dispatch; bumping // here would double-count. kv.trigger_compact(shard.id_ver()); } fn maybe_destroy(&mut self) { if self.fsm.peer.maybe_destroy(self.ctx) { // Destroy the apply fsm first, wait for the reply msg from apply fsm self.ctx.apply_msgs.msgs.push(ApplyMsg::UnsafeDestroy { region_id: self.region_id(), }); } else { self.ctx .raft_metrics .message_dropped .region_tombstone_peer .inc(); } } /// Check whether the peer should send a request to fetch the committed /// index from the leader. fn try_to_fetch_committed_index(&mut self) { // Already completed, skip. if !self.fsm.peer.needs_update_last_leader_committed_idx() || self.fsm.peer.is_leader() { return; } let retry_interval = self.ctx.cfg.raft_base_tick_interval.0 * self.ctx.cfg.raft_election_timeout_ticks as u32; if !self .fsm .peer .can_fetch_last_leader_committed_idx(retry_interval) { return; } // Construct a MsgReadIndex message and send it to the leader to // fetch the latest committed index of this raft group. let leader_id = self.fsm.peer.leader_id(); if leader_id == raft::INVALID_ID { // The leader is unknown, so we can't fetch the committed index. return; } let rctx = ReadIndexContext { id: uuid::Uuid::new_v4(), request: None, locked: None, }; self.fsm.peer.raft_group.read_index(rctx.to_bytes()); self.fsm.peer.mark_fetch_last_leader_committed_idx(); debug!( "try to fetch committed index from leader"; "region_id" => self.region_id(), "peer_id" => self.fsm.peer_id() ); } /// Check whether the peer is pending on applying raft logs. /// /// If busy, the peer will be recorded, until the pending logs are /// applied. And after it completes applying, it will be removed from /// the recording list. fn on_check_peer_complete_apply_logs(&mut self) { // Already completed, skip. if self.fsm.peer.busy_on_apply.is_none() { return; } let peer_id = self.fsm.peer.peer_id(); // No need to check the applying state if the peer is leader. if self.fsm.peer.is_leader() { self.fsm.peer.busy_on_apply = None; self.fsm.peer.reset_last_leader_committed_idx_tracking(); // Clear it from recoding list and update the counter, to avoid // missing it when the peer is changed to leader. let mut meta = self.ctx.global.busy_apply_status.lock().unwrap(); meta.mark_peer_apply_completed(peer_id); return; } let applied_idx = self.fsm.peer.get_store().applied_index(); let mut last_idx = self.fsm.peer.get_store().last_index(); // If the peer is newly added or created, no need to check the apply status. if last_idx <= RAFT_INIT_LOG_INDEX { self.fsm.peer.busy_on_apply = None; self.fsm.peer.reset_last_leader_committed_idx_tracking(); // And it should be recorded in the `completed_apply_peers_count`. let mut meta = self.ctx.global.busy_apply_status.lock().unwrap(); meta.mark_peer_apply_completed(peer_id); debug!( "no need to check uninitialized peer"; "last_commit_idx" => last_idx, "last_applied_idx" => applied_idx, "region_id" => self.fsm.region_id(), "peer_id" => peer_id, ); return; } assert!(self.fsm.peer.busy_on_apply.is_some()); // This peer is restarted and the last leader commit index is not set, so // it use `u64::MAX` as the last commit index to make it wait for the update // of the `last_leader_committed_idx` until the `last_leader_committed_idx` has // been updated. last_idx = self.fsm.peer.last_leader_committed_idx.unwrap_or(u64::MAX); // If the peer has large unapplied logs, this peer should be recorded until // the lag is less than the given threshold. if last_idx >= applied_idx + self.ctx.cfg.leader_transfer_max_log_lag { if !self.fsm.peer.busy_on_apply.unwrap() { let mut meta = self.ctx.global.busy_apply_status.lock().unwrap(); meta.insert_busy_peer(peer_id); } self.fsm.peer.busy_on_apply = Some(true); debug!( "peer is busy on applying logs"; "last_commit_idx" => last_idx, "last_applied_idx" => applied_idx, "region_id" => self.fsm.region_id(), "peer_id" => peer_id, ); } else { // Already finish apply, remove it from recording list. let mut completed_count = 0; { let mut meta = self.ctx.global.busy_apply_status.lock().unwrap(); meta.mark_peer_apply_completed(peer_id); if let Some(count) = meta.completed_apply_peers_count.as_mut() { completed_count = *count; } } debug!( "peer completes applying logs"; "last_commit_idx" => last_idx, "last_applied_idx" => applied_idx, "region_id" => self.fsm.region_id(), "peer_id" => peer_id, "completed_count" => completed_count, ); self.fsm.peer.busy_on_apply = None; self.fsm.peer.reset_last_leader_committed_idx_tracking(); } } } pub fn new_admin_request(region_id: u64, peer: metapb::Peer) -> RaftCmdRequest { let mut request = RaftCmdRequest::default(); request.mut_header().set_region_id(region_id); request.mut_header().set_peer(peer); request } /// For status command. impl PeerMsgHandler<'_> { // Handle status commands here, separate the logic, maybe we can move it // to another file later. // Unlike other commands (write or admin), status commands only show current // store status, so no need to handle it in raft group. fn execute_status_command(&mut self, request: &RaftCmdRequest) -> Result { let cmd_type = request.get_status_request().get_cmd_type(); let mut response = match cmd_type { StatusCmdType::RegionLeader => self.execute_region_leader(), StatusCmdType::RegionDetail => self.execute_region_detail(request), StatusCmdType::InvalidStatus => { Err(box_err!("{} invalid status command!", self.fsm.peer.tag())) } }?; response.set_cmd_type(cmd_type); let mut resp = RaftCmdResponse::default(); resp.set_status_response(response); // Bind peer current term here. bind_term(&mut resp, self.fsm.peer.term()); Ok(resp) } fn execute_region_leader(&mut self) -> Result { let mut resp = StatusResponse::default(); if let Some(leader) = self.fsm.peer.get_peer_from_cache(self.fsm.peer.leader_id()) { resp.mut_region_leader().set_leader(leader); } Ok(resp) } fn execute_region_detail(&mut self, request: &RaftCmdRequest) -> Result { if !self.fsm.peer.get_store().is_initialized() { let region_id = request.get_header().get_region_id(); return Err(Error::RegionNotInitialized(region_id)); } let mut resp = StatusResponse::default(); resp.mut_region_detail() .set_region(self.fsm.peer.region().clone()); if let Some(leader) = self.fsm.peer.get_peer_from_cache(self.fsm.peer.leader_id()) { resp.mut_region_detail().set_leader(leader); } Ok(resp) } } pub struct MsgDebug<'a>(pub &'a RaftMessage); impl std::fmt::Display for MsgDebug<'_> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let msg = self.0.get_message(); write!( f, "{:?}, to: {}, from: {}, term: {}, log_term: {}, index: {}, commit: {}, reject: {}, num_entries: {}", msg.msg_type, msg.to, msg.from, msg.term, msg.log_term, msg.index, msg.commit, msg.reject, msg.get_entries().len(), ) } } // For active resolve txn file locks. impl PeerMsgHandler<'_> { #[inline] fn record_txn_file_active_resolve_schedule(result: &'static str) { TXN_FILE_ACTIVE_RESOLVE_SCHEDULE_TOTAL .with_label_values(&[result]) .inc(); } #[inline] fn txn_file_lock_age_ms(lock_version: u64, physical_now: u64) -> u64 { physical_now.saturating_sub(TimeStamp::new(lock_version).physical()) } #[inline] fn txn_file_lock_active_resolve_threshold_ms(&self) -> Option { let cfg = &self.ctx.cfg.txn_file_lock_active_resolve; let threshold_ms = cfg.threshold.as_millis(); (threshold_ms != 0).then_some(threshold_ms) } fn txn_file_lock_active_resolve_ready(&self) -> bool { let tag = self.peer.tag(); let region_id = self.region_id(); match self .ctx .global .txn_file_lock_resolve_scheduler .check(region_id) { ResolveScheduleCheck::Ready => true, ResolveScheduleCheck::Inflight => { Self::record_txn_file_active_resolve_schedule("deduped"); debug!("{} active resolve txn file locks already in flight", tag); false } } } fn schedule_txn_file_lock_resolve_task(&self, locks: Vec) { debug_assert!(!locks.is_empty()); let tag = self.peer.tag(); let region_id = self.region_id(); let lock_count = locks.len(); let physical_now = TimeStamp::physical_now(); let mut min_age_ms = u64::MAX; let mut max_age_ms = 0; for lock in &locks { let age_ms = Self::txn_file_lock_age_ms(lock.get_lock_version(), physical_now); min_age_ms = min_age_ms.min(age_ms); max_age_ms = max_age_ms.max(age_ms); } let task = ResolveTxnFileLocksTask { tag, region_id, locks, }; match self .ctx .global .txn_file_lock_resolve_scheduler .schedule(task) { Ok(ResolveScheduleOutcome::Scheduled) => { Self::record_txn_file_active_resolve_schedule("scheduled"); info!("{} active resolve txn file locks scheduled", tag; "eligible_locks" => lock_count, "min_age_ms" => min_age_ms, "max_age_ms" => max_age_ms, ); } Ok(ResolveScheduleOutcome::Deduped) => { Self::record_txn_file_active_resolve_schedule("deduped"); debug!("{} active resolve txn file locks deduped", tag; "eligible_locks" => lock_count, ); } Ok(ResolveScheduleOutcome::Empty) => { debug_assert!(false, "txn file active resolve scheduled with empty locks"); } Err(ScheduleError::Full(_task)) => { Self::record_txn_file_active_resolve_schedule("full"); warn!("{} active resolve txn file locks worker is full", tag; "eligible_locks" => lock_count, ); } Err(ScheduleError::Stopped(_task)) => { Self::record_txn_file_active_resolve_schedule("stopped"); warn!("{} active resolve txn file locks worker is stopped", tag; "eligible_locks" => lock_count, ); } } } fn maybe_schedule_txn_file_lock_active_resolve(&self) { if !self.fsm.peer.is_leader() { return; } if self .ctx .global .engines .kv .get_shard(self.region_id()) .is_none() { return; } let Some(shard_meta) = self.peer.get_store().shard_meta.as_ref() else { return; }; let Some(threshold_ms) = self.txn_file_lock_active_resolve_threshold_ms() else { return; }; let txn_file_locks = shard_meta.txn_file_locks(); if txn_file_locks.is_empty() { return; } let tag = self.peer.tag(); let region_id = self.region_id(); if !self.txn_file_lock_active_resolve_ready() { return; } let physical_now = TimeStamp::physical_now(); let mut eligible_start_tss = Vec::with_capacity(txn_file_locks.len()); for start_ts in txn_file_locks.get_locks().keys() { if Self::txn_file_lock_age_ms(*start_ts, physical_now) >= threshold_ms { eligible_start_tss.push(*start_ts); } } if eligible_start_tss.is_empty() { Self::record_txn_file_active_resolve_schedule("no_eligible_lock"); debug!("{} active resolve txn file locks found no eligible locks", tag; "region_id" => region_id, "threshold_ms" => threshold_ms, "current_locks" => txn_file_locks.len(), ); return; } let raw_probe_key = _util::raw_start_key(self.peer.get_preprocessed_region()); let eligible_locks = match txn_file_locks.get_lock_infos_by_start_tss(&raw_probe_key, eligible_start_tss) { Ok(lock_infos) => lock_infos, Err(err) => { Self::record_txn_file_active_resolve_schedule("parse_error"); error!("{} active resolve txn file locks failed to reconstruct lock infos", tag; "region_id" => region_id, "txn_file_locks" => ?txn_file_locks, "err" => ?err, ); return; } }; self.schedule_txn_file_lock_resolve_task(eligible_locks); } } fn lock_infos_to_key_errors(lock_infos: Vec) -> Vec { let mut key_errors = Vec::with_capacity(lock_infos.len()); for lock_info in lock_infos { let mut key_error = kvrpcpb::KeyError::default(); key_error.set_locked(lock_info); key_errors.push(key_error); } key_errors } /// File-GC check ticks over which replicas stagger their candidate-hint refresh /// after an online config change, so the refreshes do not all land on one tick. const FILE_GC_HINT_REFRESH_TICK_WINDOW: u64 = 20; /// Whether a shard refreshes its file-GC candidate hint on this tick. /// After an online knob change at `changed_at_ns`, each shard refreshes once /// within the `FILE_GC_HINT_REFRESH_TICK_WINDOW`-tick window, on a tick chosen /// by `shard_id`. Staggering refreshes only a fraction of the shards per tick, /// which smooths the work and keeps a config change's performance impact small. fn should_refresh_file_gc_hint( shard_id: u64, now_ns: u64, changed_at_ns: u64, tick_ns: u64, ) -> bool { if changed_at_ns == 0 || tick_ns == 0 { return false; } let elapsed_ticks = now_ns.saturating_sub(changed_at_ns) / tick_ns; elapsed_ticks < FILE_GC_HINT_REFRESH_TICK_WINDOW && shard_id % FILE_GC_HINT_REFRESH_TICK_WINDOW == elapsed_ticks } #[cfg(test)] mod tests { use api_version::ApiV2; use kvengine::table::fts::new_tracked_index_meta; use kvenginepb::fts::{FtsFieldLayout, FullTextIndexFieldDef}; use super::*; fn region_with_buckets(raw_keys: &[Vec]) -> (Region, BucketMeta) { let mut region = Region::default(); region.set_id(42); region.set_start_key(encode_bytes_maybe_empty(&raw_keys[0])); region.set_end_key(encode_bytes_maybe_empty(raw_keys.last().unwrap())); region.mut_region_epoch().set_version(7); region.mut_region_epoch().set_conf_ver(3); let bucket_meta = BucketMeta { region_id: region.get_id(), region_epoch: region.get_region_epoch().clone(), keys: raw_keys .iter() .map(|key| encode_bytes_maybe_empty(key)) .collect(), ..Default::default() }; (region, bucket_meta) } fn raw_range(start: &[u8], end: &[u8]) -> KeyRange { let mut range = KeyRange::default(); range.set_start_key(start.to_vec()); range.set_end_key(end.to_vec()); range } #[test] fn test_decide_bucket_split_uses_encoded_outer_key_range() { let prefix = ApiV2::get_txn_keyspace_prefix(9); let outer = |suffix: &[u8]| [prefix.as_slice(), suffix].concat(); let keys = [b"b", b"d", b"f", b"h", b"z"].map(|suffix| outer(suffix)); let (region, buckets) = region_with_buckets(&keys); let decision = decide_bucket_split( ®ion, Some(&buckets), &raw_range(&outer(b"c"), &outer(b"i")), ); assert_eq!( decision, BucketSplitDecision::SplitKey(encode_bytes(&outer(b"f"))) ); let decision = decide_bucket_split( ®ion, Some(&buckets), &raw_range(&outer(b"d"), &outer(b"h")), ); assert_eq!( decision, BucketSplitDecision::SplitKey(encode_bytes(&outer(b"f"))) ); } #[test] fn test_decide_bucket_split_handles_unbounded_region() { let keys = [b"".as_slice(), b"d", b"f", b""].map(|key| key.to_vec()); let (region, buckets) = region_with_buckets(&keys); let decision = decide_bucket_split(®ion, Some(&buckets), &raw_range(b"c", b"")); assert_eq!(decision, BucketSplitDecision::SplitKey(encode_bytes(b"f"))); } #[test] fn test_decide_bucket_split_falls_back_for_narrow_range() { let keys = [b"b", b"d", b"f", b"h", b"z"].map(|key| key.to_vec()); let (region, buckets) = region_with_buckets(&keys); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"d", b"e")), BucketSplitDecision::SplitKey(encode_bytes(b"f")) ); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"d", b"f")), BucketSplitDecision::SplitKey(encode_bytes(b"f")) ); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"z", b"zz")), BucketSplitDecision::NoCandidate ); } #[test] fn test_decide_bucket_split_rejects_point_range() { let keys = [b"b", b"d", b"f", b"h", b"z"].map(|key| key.to_vec()); let (region, buckets) = region_with_buckets(&keys); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"e", b"e")), BucketSplitDecision::NoCandidate ); assert_eq!( decide_bucket_split(®ion, None, &raw_range(b"e", b"e")), BucketSplitDecision::NoCandidate ); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"z", b"z")), BucketSplitDecision::NoCandidate ); } #[test] fn test_decide_bucket_split_checks_whole_region_ssts_without_usable_bucket() { let keys = [b"b", b"d", b"f", b"h", b"z"].map(|key| key.to_vec()); let (region, mut buckets) = region_with_buckets(&keys); let expected = BucketSplitDecision::CheckSst { start_key: b"b".to_vec(), end_key: b"z".to_vec(), }; let (bounds_only_region, bounds_only) = region_with_buckets(&[b"b".to_vec(), b"z".to_vec()]); assert_eq!( decide_bucket_split( &bounds_only_region, Some(&bounds_only), &raw_range(b"d", b"e"), ), expected ); assert_eq!( decide_bucket_split(®ion, None, &raw_range(b"d", b"e")), expected ); buckets.region_epoch.set_version(6); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"c", b"i")), expected ); buckets.region_epoch.set_version(7); buckets.keys.swap(1, 2); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"c", b"i")), expected ); assert_eq!( decide_bucket_split(®ion, None, &raw_range(b"z", b"zz")), BucketSplitDecision::NoCandidate ); let unbounded = [b"".as_slice(), b""].map(|key| key.to_vec()); let (region, buckets) = region_with_buckets(&unbounded); assert_eq!( decide_bucket_split(®ion, Some(&buckets), &raw_range(b"d", b"e")), BucketSplitDecision::CheckSst { start_key: Vec::new(), end_key: Vec::new(), } ); } #[test] fn test_is_read_only_raft_command() { let mut read = RaftCmdRequest::default(); let mut get = Request::default(); get.set_cmd_type(CmdType::Get); read.mut_requests().push(get); let mut snap = Request::default(); snap.set_cmd_type(CmdType::Snap); read.mut_requests().push(snap); assert!(is_read_only_raft_command(&read)); let mut read_index = RaftCmdRequest::default(); let mut req = Request::default(); req.set_cmd_type(CmdType::ReadIndex); read_index.mut_requests().push(req); assert!(is_read_only_raft_command(&read_index)); let mut write = RaftCmdRequest::default(); let mut put = Request::default(); put.set_cmd_type(CmdType::Put); write.mut_requests().push(put); assert!(!is_read_only_raft_command(&write)); assert!(!is_read_only_raft_command(&RaftCmdRequest::default())); } #[test] fn test_cold_data_candidate_prefilters_active_regions() { // ShardRange stores raw outer keys. Region boundaries use a different, // memcomparable-encoded representation and must not be used here. let whole_keyspace = ShardRange::new(&[b'x', 0, 0, 10], &[b'x', 0, 0, 11]); let partial_keyspace = ShardRange::new(&[b'x', 0, 0, 10, b'a'], &[b'x', 0, 0, 11]); let multi_keyspace = ShardRange::new(&[b'x', 0, 0, 10, b'a'], &[b'x', 0, 0, 11, b'a']); let bootstrap_region = ShardRange::default(); let now_physical_ms = 10_000; let threshold_ms = 1_000; let old_max_ts = TimeStamp::compose(8_999, 0).into_inner(); let recent_max_ts = TimeStamp::compose(9_000, 0).into_inner(); assert!(!is_cold_data_candidate( &partial_keyspace, 0, 0, threshold_ms, now_physical_ms, )); assert!(is_cold_data_candidate( &whole_keyspace, 0, 0, threshold_ms, now_physical_ms, )); assert!(is_cold_data_candidate( &multi_keyspace, 0, 0, threshold_ms, now_physical_ms, )); assert!(!is_cold_data_candidate( &bootstrap_region, 0, 0, threshold_ms, now_physical_ms, )); assert!(!is_cold_data_candidate( &partial_keyspace, old_max_ts, 1, threshold_ms, now_physical_ms, )); assert!(!is_cold_data_candidate( &whole_keyspace, recent_max_ts, 1, threshold_ms, now_physical_ms, )); assert!(is_cold_data_candidate( &whole_keyspace, old_max_ts, 1, threshold_ms, now_physical_ms, )); } fn tracked_field(col_id: i64) -> FullTextIndexFieldDef { let mut field = FullTextIndexFieldDef::new(); field.set_col_id(col_id); field } // Across the whole window each shard refreshes exactly once, on the tick // selected by `shard_id % WINDOW`. That selection is the stagger: which // tick a shard lands on is spread out by its id. #[test] fn test_file_gc_hint_refresh_fires_once_per_shard_staggered() { let tick_ns = 1_000; let changed_at = tick_ns; // non-zero so the decision is enabled for shard_id in 0..(2 * FILE_GC_HINT_REFRESH_TICK_WINDOW) { let mut fired_on = Vec::new(); for cycle in 0..FILE_GC_HINT_REFRESH_TICK_WINDOW { // Sample mid-tick so integer division lands inside `cycle`. let now = changed_at + cycle * tick_ns + tick_ns / 2; if should_refresh_file_gc_hint(shard_id, now, changed_at, tick_ns) { fired_on.push(cycle); } } assert_eq!( fired_on, vec![shard_id % FILE_GC_HINT_REFRESH_TICK_WINDOW], "shard {} must refresh once, on its staggered tick", shard_id ); } } // On any single tick only the shards whose id selects that tick refresh, so // the per-tick burst is ~1/WINDOW of the shards, not all at once. #[test] fn test_file_gc_hint_refresh_spreads_load_across_ticks() { let tick_ns = 1_000; let changed_at = tick_ns; let shard_count = 100u64; for cycle in 0..FILE_GC_HINT_REFRESH_TICK_WINDOW { let now = changed_at + cycle * tick_ns + tick_ns / 2; let refreshing = (0..shard_count) .filter(|&id| should_refresh_file_gc_hint(id, now, changed_at, tick_ns)) .count() as u64; assert_eq!(refreshing, shard_count / FILE_GC_HINT_REFRESH_TICK_WINDOW); } } // The decision is off with no recorded change, with ticks disabled, and // once the window has elapsed, so a stale stamp cannot refresh forever. #[test] fn test_file_gc_hint_refresh_off_on_guards_and_after_window() { let tick_ns = 1_000; let changed_at = tick_ns; // No recorded change. assert!(!should_refresh_file_gc_hint(0, changed_at, 0, tick_ns)); // Ticks disabled. assert!(!should_refresh_file_gc_hint(0, changed_at, changed_at, 0)); // Past the window: no shard refreshes again (self-expiring). let after = changed_at + FILE_GC_HINT_REFRESH_TICK_WINDOW * tick_ns + tick_ns / 2; for shard_id in 0..(2 * FILE_GC_HINT_REFRESH_TICK_WINDOW) { assert!(!should_refresh_file_gc_hint( shard_id, after, changed_at, tick_ns )); } } #[test] fn test_tracked_fts_indexes_from_shard_meta_preserves_explicit_meta() { let mut shard_meta = kvengine::ShardMeta::default(); shard_meta.fts_indexes.insert((7, 9)); shard_meta.fts_index_metas.entry(7).or_default().insert( 9, new_tracked_index_meta( 7, 9, vec![tracked_field(101)], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ), ); let tracked_indexes = tracked_fts_indexes_from_shard_meta(&shard_meta); let tracked_meta = tracked_indexes .get(&7) .and_then(|indexes| indexes.get(&9)) .unwrap(); assert_eq!(tracked_meta.get_index_fields()[0].get_col_id(), 101); assert_eq!( tracked_meta.get_field_layout(), FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString ); } #[test] fn test_tracked_fts_indexes_from_shard_meta_backfills_legacy_meta() { let mut shard_meta = kvengine::ShardMeta::default(); shard_meta.fts_indexes.insert((7, 9)); let tracked_indexes = tracked_fts_indexes_from_shard_meta(&shard_meta); let tracked_meta = tracked_indexes .get(&7) .and_then(|indexes| indexes.get(&9)) .unwrap(); assert!(tracked_meta.get_index_fields().is_empty()); assert_eq!( tracked_meta.get_field_layout(), FtsFieldLayout::FtsFieldLayoutLegacyBody ); } }