Raft: Fix coprocessor ignore new added RegionReadStatus thus cause inconsistent result (pingcap#10543) (pingcap#10544)

ti-chi-bot · JaySon-Huang · web-flow · commit b50ff411fc6e · 2025-11-28T16:09:16.000Z
close pingcap#10510 * Handling new added RegionReadStatus enum values for error response of coprocessor * Do not set the `default` branch for switch (e.status) so that the compiler will throw an error if there is not covered new enum values Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io> Signed-off-by: JaySon-Huang <tshent@qq.com> Co-authored-by: JaySon <tshent@qq.com> Co-authored-by: JaySon-Huang <tshent@qq.com>
diff --git a/dbms/src/Common/TiFlashMetrics.h b/dbms/src/Common/TiFlashMetrics.h
@@ -580,6 +580,9 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       F(type_key_not_in_region, {{"type", "key_not_in_region"}}),                                                                   \
       F(type_tikv_server_issue, {{"type", "tikv_server_issue"}}),                                                                   \
       F(type_tikv_lock, {{"type", "tikv_lock"}}),                                                                                   \
+      F(type_server_is_busy, {{"type", "server_is_busy"}}),                                                                         \
+      F(type_stale_command, {{"type", "stale_command"}}),                                                                           \
+      F(type_store_not_match, {{"type", "store_not_match"}}),                                                                       \
       F(type_other, {{"type", "other"}}))                                                                                           \
     /* required by DBaaS */                                                                                                         \
     M(tiflash_server_info,                                                                                                          \
diff --git a/dbms/src/Debug/ReadIndexStressTest.cpp b/dbms/src/Debug/ReadIndexStressTest.cpp
@@ -38,6 +38,7 @@ static const std::map<std::string, ReadIndexStressTest::TestType> TestName2Type
 
 ReadIndexStressTest::ReadIndexStressTest(const TMTContext & tmt_)
     : tmt(tmt_)
+    , logger(Logger::get("ReadIndexStressTest"))
 {
     MockStressTestCfg::enable = true;
     LOG_WARNING(logger, "enable MockStressTest");
diff --git a/dbms/src/Debug/ReadIndexStressTest.h b/dbms/src/Debug/ReadIndexStressTest.h
@@ -57,7 +57,7 @@ class ReadIndexStressTest
 
 private:
     const TMTContext & tmt;
-    Poco::Logger * logger{&Poco::Logger::get("ReadIndexStressTest")};
+    LoggerPtr logger;
 };
 
 
diff --git a/dbms/src/Flash/CoprocessorHandler.cpp b/dbms/src/Flash/CoprocessorHandler.cpp
@@ -231,31 +231,8 @@ grpc::Status CoprocessorHandler<is_stream>::execute()
             response = cop_response;
         }
 
-        errorpb::Error * region_err;
-        switch (e.status)
-        {
-        case RegionException::RegionReadStatus::OTHER:
-        case RegionException::RegionReadStatus::BUCKET_EPOCH_NOT_MATCH:
-        case RegionException::RegionReadStatus::FLASHBACK:
-        case RegionException::RegionReadStatus::KEY_NOT_IN_REGION:
-        case RegionException::RegionReadStatus::TIKV_SERVER_ISSUE:
-        case RegionException::RegionReadStatus::READ_INDEX_TIMEOUT:
-        case RegionException::RegionReadStatus::NOT_LEADER:
-        case RegionException::RegionReadStatus::NOT_FOUND_TIKV:
-        case RegionException::RegionReadStatus::NOT_FOUND:
-            GET_METRIC(tiflash_coprocessor_request_error, reason_region_not_found).Increment();
-            region_err = response->mutable_region_error();
-            region_err->mutable_region_not_found()->set_region_id(cop_request->context().region_id());
-            break;
-        case RegionException::RegionReadStatus::EPOCH_NOT_MATCH:
-            GET_METRIC(tiflash_coprocessor_request_error, reason_epoch_not_match).Increment();
-            region_err = response->mutable_region_error();
-            region_err->mutable_epoch_not_match();
-            break;
-        default:
-            // should not happen
-            break;
-        }
+        setResponseByRegionException(response, e, cop_request->context().region_id());
+
         if constexpr (is_stream)
         {
             cop_writer->Write(stream_response);
diff --git a/dbms/src/Storages/KVStore/FFI/ProxyFFI.cpp b/dbms/src/Storages/KVStore/FFI/ProxyFFI.cpp
@@ -51,8 +51,6 @@ const std::string ColumnFamilyName::Lock = "lock";
 const std::string ColumnFamilyName::Default = "default";
 const std::string ColumnFamilyName::Write = "write";
 
-extern const uint64_t DEFAULT_BATCH_READ_INDEX_TIMEOUT_MS;
-
 ColumnFamilyType NameToCF(const std::string & cf)
 {
     if (cf.empty() || cf == ColumnFamilyName::Default)
diff --git a/dbms/src/Storages/KVStore/Read/LearnerReadWorker.cpp b/dbms/src/Storages/KVStore/Read/LearnerReadWorker.cpp
diff --git a/dbms/src/Storages/KVStore/Read/LearnerReadWorker.h b/dbms/src/Storages/KVStore/Read/LearnerReadWorker.h
@@ -48,57 +48,40 @@ struct UnavailableRegions
         , is_wn_disagg_read(is_wn_disagg_read_)
     {}
 
-    size_t size() const { return ids.size(); }
+    size_t size() const { return unavailable_ids.size(); }
 
-    bool empty() const { return ids.empty(); }
+    bool empty() const { return unavailable_ids.empty(); }
 
-    bool contains(RegionID region_id) const { return ids.contains(region_id); }
+    bool contains(RegionID region_id) const { return unavailable_ids.contains(region_id); }
 
-    void addStatus(RegionID id, RegionException::RegionReadStatus status_, std::string && extra_msg_)
+    void addStatus(RegionID region_id, RegionException::RegionReadStatus status_, std::string && extra_msg_)
     {
-        status = status_;
-        ids.emplace(id);
-        extra_msg = std::move(extra_msg_);
+        unavailable_ids[region_id] = UnavailableDesc{status_, std::move(extra_msg_)};
     }
 
     void addRegionLock(RegionID region_id_, LockInfoPtr && region_lock_)
     {
         region_locks.emplace_back(region_id_, std::move(region_lock_));
-        ids.emplace(region_id_);
+        unavailable_ids[region_id_] = UnavailableDesc{RegionException::RegionReadStatus::MEET_LOCK, ""};
     }
 
     void tryThrowRegionException();
 
     void addRegionWaitIndexTimeout(RegionID region_id, UInt64 index_to_wait, UInt64 current_applied_index);
 
-    String toDebugString() const
-    {
-        FmtBuffer buffer;
-        buffer.append("{ids=[");
-        buffer.joinStr(
-            ids.begin(),
-            ids.end(),
-            [](const auto & v, FmtBuffer & f) { f.fmtAppend("{}", v); },
-            "|");
-        buffer.append("] locks=");
-        buffer.append("[");
-        buffer.joinStr(
-            region_locks.begin(),
-            region_locks.end(),
-            [](const auto & v, FmtBuffer & f) { f.fmtAppend("{}({})", v.first, v.second->DebugString()); },
-            "|");
-        buffer.append("]}");
-        return buffer.toString();
-    }
+    String toDebugString(size_t num_show) const;
 
 private:
     const bool batch_cop;
     const bool is_wn_disagg_read;
 
-    RegionException::UnavailableRegions ids;
+    struct UnavailableDesc
+    {
+        RegionException::RegionReadStatus s;
+        std::string extra_msg;
+    };
+    std::unordered_map<RegionID, UnavailableDesc> unavailable_ids;
     std::vector<std::pair<RegionID, LockInfoPtr>> region_locks;
-    RegionException::RegionReadStatus status{RegionException::RegionReadStatus::NOT_FOUND};
-    std::string extra_msg;
 };
 
 using RegionsReadIndexResult = std::unordered_map<RegionID, kvrpcpb::ReadIndexResponse>;
diff --git a/dbms/src/Storages/KVStore/Read/ReadIndex.cpp b/dbms/src/Storages/KVStore/Read/ReadIndex.cpp
@@ -137,9 +137,13 @@ void WaitCheckRegionReadyImpl(
     static constexpr double BATCH_READ_INDEX_TIME_RATE = 0.2;
     auto log = Logger::get(__FUNCTION__);
 
+    UInt64 read_index_timeout = tmt.batchReadIndexTimeout();
+
     LOG_INFO(
         log,
-        "start to check regions ready, min_wait_tick={:.3f}s max_wait_tick={:.3f}s wait_region_ready_timeout={:.3f}s",
+        "start to check regions ready, read_index_timeout={} min_wait_tick={:.3f}s max_wait_tick={:.3f}s "
+        "wait_region_ready_timeout={:.3f}s",
+        read_index_timeout,
         wait_tick_time,
         max_wait_tick_time,
         get_wait_region_ready_timeout_sec);
@@ -172,7 +176,7 @@ void WaitCheckRegionReadyImpl(
         }
 
         // Record the latest commit index in TiKV
-        auto read_index_res = kvstore.batchReadIndex(batch_read_index_req, tmt.batchReadIndexTimeout());
+        auto read_index_res = kvstore.batchReadIndex(batch_read_index_req, read_index_timeout);
         for (auto && [resp, region_id] : read_index_res)
         {
             bool need_retry = resp.read_index() == 0;
diff --git a/dbms/src/Storages/KVStore/Read/ReadIndexWorker.h b/dbms/src/Storages/KVStore/Read/ReadIndexWorker.h
@@ -21,8 +21,6 @@
 #include <common/logger_useful.h>
 #include <kvproto/kvrpcpb.pb.h>
 
-#include <algorithm>
-#include <condition_variable>
 #include <memory>
 #include <thread>
 
@@ -32,6 +30,7 @@ namespace tests
 {
 class ReadIndexTest;
 } // namespace tests
+class KVStore;
 
 struct AsyncWaker
 {
@@ -95,9 +94,7 @@ class ReadIndexWorkerManager : boost::noncopyable
     void runOneRound(SteadyClock::duration min_dur, size_t id);
     void stop();
     ~ReadIndexWorkerManager();
-    BatchReadIndexRes batchReadIndex(
-        const std::vector<kvrpcpb::ReadIndexRequest> & reqs,
-        uint64_t timeout_ms = 10 * 1000);
+    BatchReadIndexRes batchReadIndex(const std::vector<kvrpcpb::ReadIndexRequest> & reqs, uint64_t timeout_ms);
 
     static std::unique_ptr<ReadIndexWorkerManager> newReadIndexWorkerManager(
         const TiFlashRaftProxyHelper & proxy_helper,
diff --git a/dbms/src/Storages/KVStore/Read/ReadIndexWorkerManager.cpp b/dbms/src/Storages/KVStore/Read/ReadIndexWorkerManager.cpp
@@ -14,6 +14,7 @@
 
 #include <Common/setThreadName.h>
 #include <Storages/KVStore/Read/ReadIndexWorkerImpl.h>
+#include <common/logger_useful.h>
 
 namespace DB
 {
@@ -244,8 +245,16 @@ BatchReadIndexRes ReadIndexWorkerManager::batchReadIndex(
             }
             else
             {
+                // The read index request might be dropped by a leader/candidate/follower.
+                // The learner will retry the read index request internally in raftstore.
+                // See https://github.com/tikv/tikv/pull/19071. The default retry interval is 4s at the time of writing.
+                // Reaching this point means the request still timed out after retries.
+                GET_METRIC(tiflash_raft_learner_read_failures_count, type_read_index_timeout).Increment();
+                // Generate a "region not found" error response for the region that still has no response after timeout
                 kvrpcpb::ReadIndexResponse tmp;
-                tmp.mutable_region_error()->mutable_region_not_found();
+                auto * e = tmp.mutable_region_error();
+                e->mutable_region_not_found()->set_region_id(it.first);
+                e->set_message("tiflash read index timeout(" + std::to_string(timeout_ms) + "ms)");
                 resps.emplace_back(std::move(tmp), it.first);
             }
             tasks.pop();
diff --git a/dbms/src/Storages/KVStore/Read/RegionException.cpp b/dbms/src/Storages/KVStore/Read/RegionException.cpp
@@ -0,0 +1,65 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <Common/TiFlashMetrics.h>
+#include <Storages/KVStore/Read/RegionException.h>
+#include <kvproto/coprocessor.pb.h>
+#include <kvproto/errorpb.pb.h>
+
+namespace DB
+{
+void setResponseByRegionException(coprocessor::Response * response, const RegionException & e, RegionID region_id)
+{
+    assert(response != nullptr);
+
+    errorpb::Error * region_err;
+    switch (e.status)
+    {
+    case RegionException::RegionReadStatus::OK:
+    case RegionException::RegionReadStatus::MEET_LOCK:
+        // Should not happen RegionException with RegionReadStatus::OK status
+        // And RegionReadStatus::MEET_LOCK should be handled in LockException
+        region_err = response->mutable_region_error();
+        region_err->set_message(
+            fmt::format("Unexpected RegionException with status {}", magic_enum::enum_name(e.status)));
+        region_err->mutable_region_not_found()->set_region_id(region_id);
+        break;
+    case RegionException::RegionReadStatus::OTHER:
+    case RegionException::RegionReadStatus::BUCKET_EPOCH_NOT_MATCH:
+    case RegionException::RegionReadStatus::FLASHBACK:
+    case RegionException::RegionReadStatus::KEY_NOT_IN_REGION:
+    case RegionException::RegionReadStatus::TIKV_SERVER_ISSUE:
+    case RegionException::RegionReadStatus::READ_INDEX_TIMEOUT:
+    case RegionException::RegionReadStatus::NOT_LEADER:
+    case RegionException::RegionReadStatus::NOT_FOUND_TIKV:
+    case RegionException::RegionReadStatus::NOT_FOUND:
+    case RegionException::RegionReadStatus::STALE_COMMAND:
+    case RegionException::RegionReadStatus::STORE_NOT_MATCH:
+        GET_METRIC(tiflash_coprocessor_request_error, reason_region_not_found).Increment();
+        region_err = response->mutable_region_error();
+        region_err->mutable_region_not_found()->set_region_id(region_id);
+        break;
+    case RegionException::RegionReadStatus::EPOCH_NOT_MATCH:
+        GET_METRIC(tiflash_coprocessor_request_error, reason_epoch_not_match).Increment();
+        region_err = response->mutable_region_error();
+        region_err->mutable_epoch_not_match();
+        // TODO: set current_regions when mutable epoch not match is hit
+        break;
+        // Notice: We need to handle all enum cases here to ensure correctly retrying.
+        // Do NOT add a default case here.
+        // default:
+        //     break;
+    }
+}
+} // namespace DB
diff --git a/dbms/src/Storages/KVStore/Read/RegionException.h b/dbms/src/Storages/KVStore/Read/RegionException.h
@@ -19,6 +19,11 @@
 
 #include <magic_enum.hpp>
 
+namespace coprocessor
+{
+class Response;
+}
+
 namespace DB
 {
 
@@ -39,6 +44,9 @@ class RegionException : public Exception
         KEY_NOT_IN_REGION,
         TIKV_SERVER_ISSUE,
         READ_INDEX_TIMEOUT,
+        STALE_COMMAND,
+        STORE_NOT_MATCH,
+        MEET_LOCK, // meet LockInfoPtr when reading
         OTHER,
     };
 
@@ -56,4 +64,6 @@ class RegionException : public Exception
     RegionReadStatus status;
 };
 
+void setResponseByRegionException(coprocessor::Response * response, const RegionException & e, RegionID region_id);
+
 } // namespace DB
diff --git a/dbms/src/Storages/KVStore/tests/gtest_learner_read.cpp b/dbms/src/Storages/KVStore/tests/gtest_learner_read.cpp
diff --git a/dbms/src/Storages/KVStore/tests/gtest_read_index_worker.cpp b/dbms/src/Storages/KVStore/tests/gtest_read_index_worker.cpp

Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@ static const std::map<std::string, ReadIndexStressTest::TestType> TestName2Type`
`38`	`38`
`39`	`39`	`ReadIndexStressTest::ReadIndexStressTest(const TMTContext & tmt_)`
`40`	`40`	`: tmt(tmt_)`
	`41`	`+ , logger(Logger::get("ReadIndexStressTest"))`
`41`	`42`	`{`
`42`	`43`	`MockStressTestCfg::enable = true;`
`43`	`44`	`LOG_WARNING(logger, "enable MockStressTest");`
Original file line number	Diff line number	Diff line change
`@@ -51,8 +51,6 @@ const std::string ColumnFamilyName::Lock = "lock";`
`51`	`51`	`const std::string ColumnFamilyName::Default = "default";`
`52`	`52`	`const std::string ColumnFamilyName::Write = "write";`
`53`	`53`
`54`		`-extern const uint64_t DEFAULT_BATCH_READ_INDEX_TIMEOUT_MS;`
`55`		`-`
`56`	`54`	`ColumnFamilyType NameToCF(const std::string & cf)`
`57`	`55`	`{`
`58`	`56`	`if (cf.empty() \|\| cf == ColumnFamilyName::Default)`
Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,7 @@`
`14`	`14`
`15`	`15`	`#include <Common/setThreadName.h>`
`16`	`16`	`#include <Storages/KVStore/Read/ReadIndexWorkerImpl.h>`
	`17`	`+#include <common/logger_useful.h>`
`17`	`18`
`18`	`19`	`namespace DB`
`19`	`20`	`{`
`@@ -244,8 +245,16 @@ BatchReadIndexRes ReadIndexWorkerManager::batchReadIndex(`
`244`	`245`	`}`
`245`	`246`	`else`
`246`	`247`	`{`
	`248`	`+ // The read index request might be dropped by a leader/candidate/follower.`
	`249`	`+ // The learner will retry the read index request internally in raftstore.`
	`250`	`+ // See https://github.com/tikv/tikv/pull/19071. The default retry interval is 4s at the time of writing.`
	`251`	`+ // Reaching this point means the request still timed out after retries.`
	`252`	`+ GET_METRIC(tiflash_raft_learner_read_failures_count, type_read_index_timeout).Increment();`
	`253`	`+ // Generate a "region not found" error response for the region that still has no response after timeout`
`247`	`254`	`kvrpcpb::ReadIndexResponse tmp;`
`248`		`- tmp.mutable_region_error()->mutable_region_not_found();`
	`255`	`+ auto * e = tmp.mutable_region_error();`
	`256`	`+ e->mutable_region_not_found()->set_region_id(it.first);`
	`257`	`+ e->set_message("tiflash read index timeout(" + std::to_string(timeout_ms) + "ms)");`
`249`	`258`	`resps.emplace_back(std::move(tmp), it.first);`
`250`	`259`	`}`
`251`	`260`	`tasks.pop();`