[fix](cloud) Prevent old clients from decoding unknown MetaServiceCode as OK

be/src/cloud/cloud_meta_mgr.cpp

@@ -153,9 +153,20 @@ Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concu
     return Status::OK();
 }
 
+MetaServiceCode get_response_code(const MetaServiceResponseStatus& status) {
+    if (status.has_actual_code() && MetaServiceCode_IsValid(status.actual_code())) {
+        return static_cast<MetaServiceCode>(status.actual_code());
+    }
+    return status.code();
+}
+
 namespace {
 constexpr int kBrpcRetryTimes = 3;
 
+void restore_actual_code(MetaServiceResponseStatus* status) {
+    status->set_code(get_response_code(*status));
+}
+
 bvar::LatencyRecorder _get_rowset_latency("doris_cloud_meta_mgr_get_rowset");
 bvar::LatencyRecorder g_cloud_commit_txn_resp_redirect_latency("cloud_table_stats_report_latency");
 bvar::Adder<uint64_t> g_cloud_meta_mgr_rpc_timeout_count("cloud_meta_mgr_rpc_timeout_count");
@@ -418,6 +429,16 @@ using MetaServiceMethod = void (MetaService_Stub::*)(::google::protobuf::RpcCont
                                                      const Request*, Response*,
                                                      ::google::protobuf::Closure*);
 
+template <typename Request, typename Response>
+void call_ms(MetaService_Stub* stub, MetaServiceMethod<Request, Response> method,
+             brpc::Controller* cntl, const Request& req, Response* res) {
+    (stub->*method)(cntl, &req, res, nullptr);
+    if (!cntl->Failed()) {
+        // Meta Service may downgrade code for wire compatibility; restore the exact value.
+        restore_actual_code(res->mutable_status());
+    }
+}
+
 // Rate limiting context for retry_rpc
 struct RpcRateLimitCtx {
     HostLevelMSRpcRateLimiters* host_limiters {nullptr};
@@ -503,7 +524,7 @@ Status retry_rpc(MetaServiceRPC rpc, const Request& req, Response* res,
         cntl.set_max_retry(kBrpcRetryTimes);
         res->Clear();
         int error_code = 0;
-        (stub.get()->*method)(&cntl, &req, res, nullptr);
+        call_ms(stub.get(), method, &cntl, req, res);
 
         // Record QPS statistics for all RPCs sent to MS (success or failure)
         record_rpc_qps(rpc, rate_limit_ctx);
@@ -729,7 +750,7 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
         }
 
         auto start = std::chrono::steady_clock::now();
-        stub->get_rowset(&cntl, &req, &resp, nullptr);
+        call_ms(stub.get(), &MetaService_Stub::get_rowset, &cntl, req, &resp);
         auto end = std::chrono::steady_clock::now();
         int64_t latency = cntl.latency_us();
         _get_rowset_latency << latency;

be/src/cloud/cloud_meta_mgr.h

@@ -64,6 +64,10 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
 Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concurrency,
                          std::future<Status>* fut);
 
+// Returns the exact actual_code when recognized, otherwise the legacy-compatible code.
+// Exposed for unit tests.
+MetaServiceCode get_response_code(const MetaServiceResponseStatus& status);
+
 class CloudMetaMgr {
 public:
     CloudMetaMgr() = default;

be/test/cloud/cloud_meta_mgr_test.cpp

@@ -21,6 +21,8 @@
 #include <gtest/gtest.h>
 
 #include <chrono>
+#include <cstdint>
+#include <limits>
 #include <memory>
 #include <random>
 #include <set>
@@ -44,6 +46,27 @@ class CloudMetaMgrTest : public testing::Test {
     void TearDown() override {}
 };
 
+TEST_F(CloudMetaMgrTest, response_status_uses_actual_code_when_valid) {
+    MetaServiceResponseStatus status;
+    status.set_code(MetaServiceCode::KV_TXN_CONFLICT);
+    status.set_actual_code(static_cast<int32_t>(MetaServiceCode::MS_TOO_BUSY));
+    EXPECT_EQ(get_response_code(status), MetaServiceCode::MS_TOO_BUSY);
+
+    status.set_code(MetaServiceCode::KV_TXN_CONFLICT);
+    status.set_actual_code(static_cast<int32_t>(MetaServiceCode::KV_TXN_CONFLICT));
+    EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT);
+
+    status.clear_actual_code();
+    EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT);
+}
+
+TEST_F(CloudMetaMgrTest, response_status_falls_back_for_invalid_actual_code) {
+    MetaServiceResponseStatus status;
+    status.set_code(MetaServiceCode::KV_TXN_CONFLICT);
+    status.set_actual_code(std::numeric_limits<int32_t>::max());
+    EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT);
+}
+
 static AbortTxnRequest get_abort_txn_request(CloudMetaMgr* meta_mgr, const StreamLoadContext& ctx) {
     auto* sp = SyncPoint::get_instance();
     sp->clear_all_call_backs();

cloud/src/meta-service/meta_service.h

@@ -28,9 +28,11 @@
 #include <type_traits>
 
 #include "common/config.h"
+#include "common/defer.h"
 #include "common/stats.h"
 #include "cpp/sync_point.h"
 #include "meta-service/delete_bitmap_lock_white_list.h"
+#include "meta-service/meta_service_helper.h"
 #include "meta-service/txn_lazy_committer.h"
 #include "meta-store/txn_kv.h"
 #include "rate-limiter/rate_limiter.h"
@@ -1035,6 +1037,11 @@ class MetaServiceProxy final : public MetaService {
         using namespace std::chrono;
         brpc::ClosureGuard done_guard(done);
 
+        DORIS_CLOUD_DEFER {
+            auto* status = resp->mutable_status();
+            set_response_code(status, status->code(), status->msg());
+        };
+
         // life span of this defer MUST be longer than `done`
         std::unique_ptr<int, std::function<void(int*)>> defer_injection(
                 (int*)(0x01), [&, this](int*) { idempotent_injection(method, req, resp); });

cloud/src/meta-service/meta_service_helper.h

@@ -25,6 +25,7 @@
 #include <memory>
 #include <string>
 #include <string_view>
+#include <utility>
 
 #include "common/bvars.h"
 #include "common/config.h"
@@ -41,6 +42,24 @@
 #include "resource-manager/resource_manager.h"
 
 namespace doris::cloud {
+inline MetaServiceCode get_legacy_code(MetaServiceCode code) {
+    switch (code) {
+    // MS_TOO_BUSY is a overload signal. Map it to KV_TXN_CONFLICT so the BE's existing
+    // conflict-retry path can retry the request.
+    case MetaServiceCode::MS_TOO_BUSY:
+        return MetaServiceCode::KV_TXN_CONFLICT;
+    default:
+        return code;
+    }
+}
+
+inline void set_response_code(MetaServiceResponseStatus* status, MetaServiceCode code,
+                              std::string msg) {
+    status->set_actual_code(static_cast<int32_t>(code));
+    status->set_code(get_legacy_code(code));
+    status->set_msg(std::move(msg));
+}
+
 inline std::string md5(const std::string& str) {
     unsigned char digest[MD5_DIGEST_LENGTH];
     MD5_CTX context;
@@ -315,17 +334,16 @@ inline MetaServiceCode cast_as(TxnErrorCode code) {
     [[maybe_unused]] MsStressDecision ms_stress_decision;                                     \
     if (config::enable_ms_rate_limit || config::enable_ms_rate_limit_injection) {             \
         ms_stress_decision = get_ms_stress_decision();                                        \
-    }                                                                                         \
-    if ((config::enable_ms_rate_limit || config::enable_ms_rate_limit_injection) &&           \
-        RpcRateLimitWhitelist::instance().should_rate_limit(#func_name) &&                    \
-        ms_stress_decision.under_great_stress()) {                                            \
-        drop_request = true;                                                                  \
-        code = MetaServiceCode::MS_TOO_BUSY;                                                  \
-        msg = ms_stress_decision.debug_string();                                              \
-        response->mutable_status()->set_code(code);                                           \
-        response->mutable_status()->set_msg(msg);                                             \
-        finish_rpc(#func_name, ctrl, request, response);                                      \
-        return;                                                                               \
+        if (RpcRateLimitWhitelist::instance().should_rate_limit(#func_name) &&                \
+            ms_stress_decision.under_great_stress()) {                                        \
+            drop_request = true;                                                              \
+            msg = ms_stress_decision.debug_string();                                          \
+            code = MetaServiceCode::MS_TOO_BUSY;                                              \
+            response->mutable_status()->set_code(code);                                       \
+            response->mutable_status()->set_msg(msg);                                         \
+            finish_rpc(#func_name, ctrl, request, response);                                  \
+            return;                                                                           \
+        }                                                                                     \
     }                                                                                         \
     DORIS_CLOUD_DEFER {                                                                       \
         response->mutable_status()->set_code(code);                                           \