*: warn linger import sst files

Cargo.toml

@@ -144,7 +144,7 @@ paste = "1.0"
 pd_client = { workspace = true }
 pin-project = "1.0"
 pnet_datalink = "0.23"
-pprof = { version = "0.11", default-features = false, features = [
+pprof = { version = "0.13", default-features = false, features = [
   "flamegraph",
   "protobuf-codec",
 ] }
@@ -249,6 +249,8 @@ procinfo = { git = "https://github.com/tikv/procinfo-rs", rev = "7693954bd1dd86e
 # After the PR to kvproto is merged, remember to comment this out and run `cargo update -p kvproto`.
 # [patch.'https://github.com/pingcap/kvproto']
 # kvproto = { git = "https://github.com/your_github_id/kvproto", branch = "your_branch" }
+[patch.'https://github.com/pingcap/kvproto']
+kvproto = { git = "https://github.com/overvenus/kvproto", branch = "resolved-ts-compat" }
 #
 # After the PR to rust-rocksdb is merged, remember to comment this out and run `cargo update -p rocksdb`.
 # [patch.'https://github.com/tikv/rust-rocksdb']

components/backup-stream/Cargo.toml

@@ -64,6 +64,7 @@ resolved_ts = { workspace = true }
 security = { path = "../security" }
 slog = { workspace = true }
 slog-global = { workspace = true }
+sst_importer = { workspace = true }
 thiserror = "1"
 tidb_query_datatype = { workspace = true }
 tikv = { workspace = true }

components/cdc/Cargo.toml

@@ -58,6 +58,7 @@ security = { workspace = true }
 semver = "1.0"
 slog = { workspace = true }
 slog-global = { workspace = true }
+sst_importer = { workspace = true }
 thiserror = "1.0"
 tikv = { workspace = true }
 tikv_kv = { workspace = true }

components/error_code/src/sst_importer.rs

@@ -22,10 +22,12 @@ define_error_codes!(
     TTL_LEN_NOT_EQUALS_TO_PAIRS => ("TtlLenNotEqualsToPairs", "", ""),
     INCOMPATIBLE_API_VERSION => ("IncompatibleApiVersion", "", ""),
     INVALID_KEY_MODE => ("InvalidKeyMode", "", ""),
-    RESOURCE_NOT_ENOUTH => ("ResourceNotEnough", "", ""),
+    RESOURCE_NOT_ENOUGH => ("ResourceNotEnough", "", ""),
     SUSPENDED => ("Suspended",
         "this request has been suspended.",
         "Probably there are some export tools don't support exporting data inserted by `ingest`(say, snapshot backup). Check the user manual and stop them."),
     REQUEST_TOO_NEW => ("RequestTooNew", "", ""),
-    REQUEST_TOO_OLD => ("RequestTooOld", "", "")
+    REQUEST_TOO_OLD => ("RequestTooOld", "", ""),
+    LEASE_EXPIRED => ("LeaseExpired", "", ""),
+    INVALID_LEASE => ("InvalidLease", "", "")
 );

components/raftstore-v2/src/operation/mod.rs

@@ -42,16 +42,28 @@ pub mod test_util {
     use kvproto::{kvrpcpb::ApiVersion, metapb::RegionEpoch, raft_cmdpb::RaftRequestHeader};
     use raft::prelude::{Entry, EntryType};
     use raftstore::store::simple_write::SimpleWriteEncoder;
-    use sst_importer::SstImporter;
+    use sst_importer::{IngestMediator, IngestObserver, Mediator, SstImporter};
     use tempfile::TempDir;
 
     use super::{CatchUpLogs, SimpleWriteReqEncoder};
     use crate::{fsm::ApplyResReporter, router::ApplyRes};
 
     pub fn create_tmp_importer<E: KvEngine>() -> (TempDir, Arc<SstImporter<E>>) {
+        let mut ingest_mediator = IngestMediator::default();
+        let ingest_observer = Arc::new(IngestObserver::default());
+        ingest_mediator.register(ingest_observer.clone());
         let dir = TempDir::new().unwrap();
         let importer = Arc::new(
-            SstImporter::new(&Default::default(), dir.path(), None, ApiVersion::V1, true).unwrap(),
+            SstImporter::new(
+                &Default::default(),
+                dir.path(),
+                None,
+                ApiVersion::V1,
+                true,
+                Arc::new(ingest_mediator),
+                ingest_observer,
+            )
+            .unwrap(),
         );
         (dir, importer)
     }

components/raftstore-v2/tests/integrations/cluster.rs

@@ -48,7 +48,7 @@ use resource_control::{ResourceController, ResourceGroupManager};
 use resource_metering::CollectorRegHandle;
 use service::service_manager::GrpcServiceManager;
 use slog::{debug, o, Logger};
-use sst_importer::SstImporter;
+use sst_importer::{IngestMediator, IngestObserver, Mediator, SstImporter};
 use tempfile::TempDir;
 use test_pd::mocker::Service;
 use tikv_util::{
@@ -326,13 +326,18 @@ impl RunningState {
         .unwrap();
         let coprocessor_host =
             CoprocessorHost::new(router.store_router().clone(), cop_cfg.value().clone());
+        let mut ingest_mediator = IngestMediator::default();
+        let ingest_observer = Arc::new(IngestObserver::default());
+        ingest_mediator.register(ingest_observer.clone());
         let importer = Arc::new(
             SstImporter::new(
                 &Default::default(),
                 path.join("importer"),
                 key_manager.clone(),
                 ApiVersion::V1,
                 true,
+                Arc::new(ingest_mediator),
+                ingest_observer,
             )
             .unwrap(),
         );

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

@@ -5037,7 +5037,7 @@ mod tests {
     };
     use protobuf::Message;
     use raft::eraftpb::{ConfChange, ConfChangeV2};
-    use sst_importer::Config as ImportConfig;
+    use sst_importer::{Config as ImportConfig, IngestMediator, IngestObserver, Mediator};
     use tempfile::{Builder, TempDir};
     use test_sst_importer::*;
     use tikv_util::{
@@ -5074,6 +5074,9 @@ mod tests {
     }
 
     pub fn create_tmp_importer(path: &str) -> (TempDir, Arc<SstImporter<KvTestEngine>>) {
+        let mut ingest_mediator = IngestMediator::default();
+        let ingest_observer = Arc::new(IngestObserver::default());
+        ingest_mediator.register(ingest_observer.clone());
         let dir = Builder::new().prefix(path).tempdir().unwrap();
         let importer = Arc::new(
             SstImporter::new(
@@ -5082,6 +5085,8 @@ mod tests {
                 None,
                 ApiVersion::V1,
                 false,
+                Arc::new(ingest_mediator),
+                ingest_observer,
             )
             .unwrap(),
         );