From 158072f1d205fc86004f31515d4919fb163cbe77 Mon Sep 17 00:00:00 2001 From: zhangqiang30 Date: Thu, 10 Sep 2026 20:08:32 +0800 Subject: [PATCH 1/3] [TransferEngine] Recover redis metadata connection after disconnects hiredis latches I/O errors into the redisContext, so a single disconnect made every later metadata operation fail for the rest of the process lifetime. In production a managed Redis behind a TCP load balancer reclaimed the idle connection after ~4000s and the node never recovered. Rebuild the connection and retry once on connection-level failure, move AUTH/SELECT into the connect path so they are replayed on reconnect, and add connect/command timeouts plus TCP keepalive. Add unit tests driven by an in-process RESP fake that can drop connections on demand. Fixes #4003 --- .../src/transfer_metadata_plugin.cpp | 249 +++++--- mooncake-transfer-engine/tests/CMakeLists.txt | 11 + .../transfer_metadata_redis_plugin_test.cpp | 533 ++++++++++++++++++ 3 files changed, 702 insertions(+), 91 deletions(-) create mode 100644 mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp diff --git a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp index 95d4cd05a9..b8cd277ffe 100644 --- a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp +++ b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp @@ -69,87 +69,37 @@ static bool parseJsonString(const std::string &json_str, Json::Value &value, namespace mooncake { #ifdef USE_REDIS struct RedisStoragePlugin : public MetadataStoragePlugin { + // Match the timeouts already used by the HA leader coordinator's redis + // client (mooncake-store/src/ha/common/redis/redis_connection.cpp:20-21). + static constexpr long kConnectTimeoutSec = 3; + static constexpr long kCommandTimeoutSec = 3; + RedisStoragePlugin(const std::string &metadata_uri) - : client_(nullptr), metadata_uri_(metadata_uri) { - auto hostname_port = parseHostNameWithPort(metadata_uri); - client_ = - redisConnect(hostname_port.first.c_str(), hostname_port.second); - if (!client_) { - LOG(ERROR) << "RedisStoragePlugin: unable to connect " - << metadata_uri_; - return; - } - if (client_->err) { - LOG(ERROR) << "RedisStoragePlugin: unable to connect " - << metadata_uri_ << ": " << client_->errstr; - redisFree(client_); - client_ = nullptr; - return; - } - } + : RedisStoragePlugin(metadata_uri, "", "", 0) {} RedisStoragePlugin(const std::string &metadata_uri, const std::string &username, const std::string &password, const uint8_t &db_index) - : RedisStoragePlugin(metadata_uri) { - if (!client_) { - return; - } - - if (!password.empty()) { - redisReply *reply = nullptr; - if (!username.empty()) { - reply = static_cast(redisCommand( - client_, "AUTH %b %b", username.data(), username.size(), - password.data(), password.size())); - } else { - reply = static_cast(redisCommand( - client_, "AUTH %b", password.data(), password.size())); - } - if (!reply || reply->type == REDIS_REPLY_ERROR) { - LOG(ERROR) << "RedisStoragePlugin: authentication failed for " - << metadata_uri_; - freeReplyObject(reply); - redisFree(client_); - client_ = nullptr; - return; - } - freeReplyObject(reply); - } - - if (db_index != 0) { - auto *reply = static_cast( - redisCommand(client_, "SELECT %d", db_index)); - if (!reply || reply->type == REDIS_REPLY_ERROR) { - LOG(ERROR) << "RedisStoragePlugin: failed to select database " - << (int)db_index << " for " << metadata_uri_; - freeReplyObject(reply); - redisFree(client_); - client_ = nullptr; - return; - } - freeReplyObject(reply); - } + : client_(nullptr), + metadata_uri_(metadata_uri), + username_(username), + password_(password), + db_index_(db_index) { + // A failure here is not fatal: every operation re-attempts the + // connection, so a redis that is briefly unavailable at startup no + // longer permanently disables metadata publishing. + connectLocked(); } - virtual ~RedisStoragePlugin() { - if (client_) { - redisFree(client_); - client_ = nullptr; - } - } + virtual ~RedisStoragePlugin() { disconnectLocked(); } virtual bool get(const std::string &key, Json::Value &value) { std::lock_guard lock(access_client_mutex_); - if (!client_) return false; - - redisReply *resp = - (redisReply *)redisCommand(client_, "GET %s", key.c_str()); - if (!resp) { - LOG(ERROR) << "RedisStoragePlugin: unable to get " << key - << " from " << metadata_uri_; - return false; - } + redisReply *resp = execLocked("GET", key, [&key](redisContext *ctx) { + return static_cast( + redisCommand(ctx, "GET %s", key.c_str())); + }); + if (!resp) return false; if (resp->type == REDIS_REPLY_ERROR) { LOG(ERROR) << "RedisStoragePlugin: get " << key << " rejected by " << metadata_uri_ << ": " @@ -158,8 +108,11 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { return false; } if (!resp->str) { - LOG(ERROR) << "RedisStoragePlugin: unable to get " << key - << " from " << metadata_uri_; + // A nil reply means the key is absent, which is an expected result + // when looking up a segment that is not registered (yet). Not an + // error, so it must not be logged as one. + LOG(WARNING) << "RedisStoragePlugin: unable to get " << key + << " from " << metadata_uri_; freeReplyObject(resp); return false; } @@ -215,17 +168,14 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { virtual bool set(const std::string &key, const Json::Value &value) { std::lock_guard lock(access_client_mutex_); - if (!client_) return false; - Json::FastWriter writer; const std::string json_file = writer.write(value); - redisReply *resp = (redisReply *)redisCommand( - client_, "SET %s %s", key.c_str(), json_file.c_str()); - if (!resp) { - LOG(ERROR) << "RedisStoragePlugin: unable to put " << key - << " from " << metadata_uri_; - return false; - } + redisReply *resp = + execLocked("SET", key, [&key, &json_file](redisContext *ctx) { + return static_cast(redisCommand( + ctx, "SET %s %s", key.c_str(), json_file.c_str())); + }); + if (!resp) return false; if (resp->type == REDIS_REPLY_ERROR) { // The server received and rejected the write (READONLY after a // failover, OOM at maxmemory, MISCONF, NOAUTH). Reporting success @@ -243,15 +193,11 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { virtual bool remove(const std::string &key) { std::lock_guard lock(access_client_mutex_); - if (!client_) return false; - - redisReply *resp = - (redisReply *)redisCommand(client_, "DEL %s", key.c_str()); - if (!resp) { - LOG(ERROR) << "RedisStoragePlugin: unable to remove " << key - << " from " << metadata_uri_; - return false; - } + redisReply *resp = execLocked("DEL", key, [&key](redisContext *ctx) { + return static_cast( + redisCommand(ctx, "DEL %s", key.c_str())); + }); + if (!resp) return false; if (resp->type == REDIS_REPLY_ERROR) { LOG(ERROR) << "RedisStoragePlugin: remove " << key << " rejected by " << metadata_uri_ << ": " @@ -263,8 +209,129 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { return true; } + private: + // The *Locked helpers below require access_client_mutex_ to be held, which + // get()/set()/remove() do. The constructor and destructor call them without + // the lock because no other thread can reach the object at that point. + void disconnectLocked() { + if (client_) { + redisFree(client_); + client_ = nullptr; + } + } + + bool connectLocked() { + auto hostname_port = parseHostNameWithPort(metadata_uri_); + + timeval connect_timeout{}; + connect_timeout.tv_sec = kConnectTimeoutSec; + client_ = redisConnectWithTimeout( + hostname_port.first.c_str(), hostname_port.second, connect_timeout); + if (!client_) { + LOG(ERROR) << "RedisStoragePlugin: unable to connect " + << metadata_uri_; + return false; + } + if (client_->err) { + LOG(ERROR) << "RedisStoragePlugin: unable to connect " + << metadata_uri_ << ": " << client_->errstr; + disconnectLocked(); + return false; + } + + timeval command_timeout{}; + command_timeout.tv_sec = kCommandTimeoutSec; + if (redisSetTimeout(client_, command_timeout) != REDIS_OK) { + LOG(ERROR) << "RedisStoragePlugin: unable to set command timeout " + "for " + << metadata_uri_; + disconnectLocked(); + return false; + } + + // This connection is only touched on segment (un)registration and on + // the first lookup of a given remote segment, so it can sit completely + // idle for hours. Without SO_KEEPALIVE it sends nothing during that + // time and a server-side or middlebox idle timeout closes it silently. + if (redisEnableKeepAlive(client_) != REDIS_OK) { + LOG(WARNING) << "RedisStoragePlugin: unable to enable TCP " + "keepalive for " + << metadata_uri_; + } + + if (!authenticateLocked()) { + disconnectLocked(); + return false; + } + return true; + } + + bool authenticateLocked() { + if (!password_.empty()) { + redisReply *reply = nullptr; + if (!username_.empty()) { + reply = static_cast(redisCommand( + client_, "AUTH %b %b", username_.data(), username_.size(), + password_.data(), password_.size())); + } else { + reply = static_cast(redisCommand( + client_, "AUTH %b", password_.data(), password_.size())); + } + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LOG(ERROR) << "RedisStoragePlugin: authentication failed for " + << metadata_uri_; + freeReplyObject(reply); + return false; + } + freeReplyObject(reply); + } + + if (db_index_ != 0) { + auto *reply = static_cast( + redisCommand(client_, "SELECT %d", db_index_)); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LOG(ERROR) << "RedisStoragePlugin: failed to select database " + << (int)db_index_ << " for " << metadata_uri_; + freeReplyObject(reply); + return false; + } + freeReplyObject(reply); + } + return true; + } + + // Issues a command, reconnecting and retrying once on connection-level + // failure. hiredis latches the error into the redisContext and rejects + // every subsequent command, so a dropped connection has to be rebuilt -- + // otherwise a single idle timeout permanently breaks metadata publishing + // for the lifetime of the process. + // + // Only null replies are retried. A REDIS_REPLY_ERROR means the server + // answered and rejected the command, and retrying that cannot help. + // GET/SET/DEL are all idempotent, so the retry is safe. + template + redisReply *execLocked(const char *op, const std::string &key, Fn &&issue) { + for (int attempt = 0; attempt < 2; ++attempt) { + if (!client_ && !connectLocked()) return nullptr; + + redisReply *resp = issue(client_); + if (resp) return resp; + + LOG(WARNING) << "RedisStoragePlugin: " << op << " " << key + << " failed on " << metadata_uri_ << ": " + << (client_ ? client_->errstr : "not connected") + << (attempt == 0 ? "; reconnecting and retrying once" + : "; giving up"); + disconnectLocked(); + } + return nullptr; + } + redisContext *client_; const std::string metadata_uri_; + const std::string username_; + const std::string password_; + const uint8_t db_index_; std::mutex access_client_mutex_; }; #endif // USE_REDIS diff --git a/mooncake-transfer-engine/tests/CMakeLists.txt b/mooncake-transfer-engine/tests/CMakeLists.txt index 10f47c2cf0..6c57036265 100644 --- a/mooncake-transfer-engine/tests/CMakeLists.txt +++ b/mooncake-transfer-engine/tests/CMakeLists.txt @@ -384,6 +384,17 @@ target_link_libraries(transfer_metadata_test PUBLIC transfer_engine gtest gtest_main) add_test(NAME transfer_metadata_test COMMAND transfer_metadata_test) +# Connection-lifecycle tests for the redis metadata storage plugin. Uses an +# in-process RESP fake, so no redis server is required. +if(USE_REDIS) + add_executable(transfer_metadata_redis_plugin_test + ${WORKSPACE}/transfer_metadata_redis_plugin_test.cpp) + target_link_libraries(transfer_metadata_redis_plugin_test + PUBLIC transfer_engine gtest glog::glog) + add_test(NAME transfer_metadata_redis_plugin_test + COMMAND transfer_metadata_redis_plugin_test) +endif() + add_executable(config_test ${WORKSPACE}/config_test.cpp) target_link_libraries(config_test PUBLIC transfer_engine gtest gtest_main) add_test(NAME config_test COMMAND config_test) diff --git a/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp b/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp new file mode 100644 index 0000000000..18b3361a8d --- /dev/null +++ b/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp @@ -0,0 +1,533 @@ +// Copyright 2024 KVCache.AI +// +// 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. + +// Tests for RedisStoragePlugin's connection lifecycle. The plugin is private to +// transfer_metadata_plugin.cpp, so it is exercised through the public +// MetadataStoragePlugin::Create("redis://...") factory. +// +// These tests talk to an in-process fake that speaks just enough RESP to serve +// GET/SET/DEL/AUTH/SELECT. A real redis cannot be scripted to drop a connection +// at an exact moment, which is precisely the condition being tested here. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "transfer_metadata_plugin.h" + +namespace mooncake { +namespace { + +// Binds 127.0.0.1 on an ephemeral port and returns the socket plus the port. +int BindLoopbackSocket(uint16_t &port, uint16_t requested_port = 0) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) return -1; + + int reuse = 1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(requested_port); + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + close(fd); + return -1; + } + + socklen_t addr_len = sizeof(addr); + if (getsockname(fd, reinterpret_cast(&addr), &addr_len) < 0) { + close(fd); + return -1; + } + port = ntohs(addr.sin_port); + return fd; +} + +// Reserves an ephemeral port and releases it, so a test can point the plugin at +// an address where nothing is listening. +uint16_t ReserveUnusedPort() { + uint16_t port = 0; + int fd = BindLoopbackSocket(port); + EXPECT_GE(fd, 0); + if (fd >= 0) close(fd); + return port; +} + +// A minimal RESP server: parses inline command arrays and keeps values in a +// map. Only the subset of commands the plugin issues is implemented. +class FakeRedisServer { + public: + explicit FakeRedisServer(uint16_t requested_port = 0) { + // gtest's ASSERT_* macros cannot be used in a constructor, and a + // harness that fails to bind cannot produce meaningful results anyway. + listen_fd_ = BindLoopbackSocket(port_, requested_port); + CHECK_GE(listen_fd_, 0) << "failed to bind fake redis server"; + CHECK_EQ(listen(listen_fd_, 8), 0); + worker_running_.store(true); + worker_ = std::thread([this] { Loop(); }); + } + + ~FakeRedisServer() { Stop(); } + + uint16_t port() const { return port_; } + + std::string endpoint() const { + return "redis://127.0.0.1:" + std::to_string(port_); + } + + // Closes every established connection, emulating an idle timeout on the + // server side or a middlebox reaping the socket. Blocks until the worker + // thread has actually closed them, so the next plugin call is guaranteed to + // hit a dead socket. + void DropConnections() { + drop_generation_.fetch_add(1); + uint64_t target = drop_generation_.load(); + while (worker_running_.load() && drop_applied_.load() < target) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + void Stop() { + if (stopped_.exchange(true)) return; + if (worker_.joinable()) worker_.join(); + if (listen_fd_ >= 0) { + close(listen_fd_); + listen_fd_ = -1; + } + } + + std::vector commands() { + std::lock_guard lock(mutex_); + return commands_; + } + + // Number of times a client completed a TCP connect. + int accept_count() const { return accept_count_.load(); } + + void SetValue(const std::string &key, const std::string &value) { + std::lock_guard lock(mutex_); + store_[key] = value; + } + + // Makes the next N data-plane replies (GET/SET/DEL) be RESP errors, e.g. + // "-READONLY You can't write against a read only replica.\r\n". AUTH and + // SELECT are left alone so the plugin can still connect. + void ForceErrorReplies(int count, + const std::string &message = "READONLY forced") { + std::lock_guard lock(mutex_); + error_replies_remaining_ = count; + error_message_ = message; + } + + private: + void Loop() { + std::vector clients; + std::unordered_map buffers; + + while (!stopped_.load()) { + uint64_t generation = drop_generation_.load(); + if (drop_applied_.load() < generation) { + for (int fd : clients) { + close(fd); + buffers.erase(fd); + } + clients.clear(); + drop_applied_.store(generation); + } + + std::vector fds; + fds.push_back(pollfd{listen_fd_, POLLIN, 0}); + for (int fd : clients) fds.push_back(pollfd{fd, POLLIN, 0}); + + int ready = poll(fds.data(), static_cast(fds.size()), 20); + if (ready < 0) { + if (errno == EINTR) continue; + break; + } + + if (fds[0].revents & POLLIN) { + int client = accept(listen_fd_, nullptr, nullptr); + if (client >= 0) { + clients.push_back(client); + accept_count_.fetch_add(1); + } + } + + for (size_t i = 1; i < fds.size(); ++i) { + if (!(fds[i].revents & (POLLIN | POLLHUP | POLLERR))) continue; + int fd = fds[i].fd; + if (!ServeClient(fd, buffers[fd])) { + close(fd); + buffers.erase(fd); + clients.erase( + std::remove(clients.begin(), clients.end(), fd), + clients.end()); + } + } + } + + for (int fd : clients) close(fd); + worker_running_.store(false); + } + + // Returns false when the client should be dropped (peer closed or error). + bool ServeClient(int fd, std::string &buffer) { + char chunk[4096]; + ssize_t bytes = recv(fd, chunk, sizeof(chunk), 0); + if (bytes <= 0) return false; + buffer.append(chunk, bytes); + + std::vector argv; + while (TryParseCommand(buffer, argv)) { + std::string reply = Dispatch(argv); + if (!SendAll(fd, reply)) return false; + argv.clear(); + } + return true; + } + + static bool SendAll(int fd, const std::string &data) { + size_t sent = 0; + while (sent < data.size()) { + ssize_t n = send(fd, data.data() + sent, data.size() - sent, 0); + if (n <= 0) return false; + sent += static_cast(n); + } + return true; + } + + // Consumes one complete RESP array from buffer. Leaves buffer untouched and + // returns false when the command has not fully arrived yet. + static bool TryParseCommand(std::string &buffer, + std::vector &argv) { + if (buffer.empty() || buffer[0] != '*') return false; + + size_t cursor = 0; + auto read_line = [&](std::string &line) { + size_t end = buffer.find("\r\n", cursor); + if (end == std::string::npos) return false; + line = buffer.substr(cursor, end - cursor); + cursor = end + 2; + return true; + }; + + std::string header; + if (!read_line(header)) return false; + int count = std::atoi(header.c_str() + 1); + if (count <= 0) return false; + + std::vector parsed; + for (int i = 0; i < count; ++i) { + std::string size_line; + if (!read_line(size_line)) return false; + if (size_line.empty() || size_line[0] != '$') return false; + int length = std::atoi(size_line.c_str() + 1); + if (length < 0) return false; + if (buffer.size() < cursor + static_cast(length) + 2) + return false; + parsed.emplace_back(buffer.substr(cursor, length)); + cursor += static_cast(length) + 2; + } + + buffer.erase(0, cursor); + argv = std::move(parsed); + return true; + } + + std::string Dispatch(const std::vector &argv) { + std::string command = argv[0]; + for (auto &c : command) c = std::toupper(c); + + { + std::lock_guard lock(mutex_); + std::string logged = command; + for (size_t i = 1; i < argv.size(); ++i) logged += " " + argv[i]; + commands_.push_back(logged); + } + + std::lock_guard lock(mutex_); + if (command == "GET" || command == "SET" || command == "DEL") { + if (error_replies_remaining_ > 0) { + --error_replies_remaining_; + return "-" + error_message_ + "\r\n"; + } + } + if (command == "GET") { + auto it = store_.find(argv[1]); + if (it == store_.end()) return "$-1\r\n"; + return "$" + std::to_string(it->second.size()) + "\r\n" + + it->second + "\r\n"; + } + if (command == "SET") { + store_[argv[1]] = argv[2]; + return "+OK\r\n"; + } + if (command == "DEL") { + return store_.erase(argv[1]) ? ":1\r\n" : ":0\r\n"; + } + // AUTH / SELECT and anything else the plugin may send. + return "+OK\r\n"; + } + + int listen_fd_ = -1; + uint16_t port_ = 0; + std::thread worker_; + std::atomic stopped_{false}; + std::atomic worker_running_{false}; + std::atomic drop_generation_{0}; + std::atomic drop_applied_{0}; + std::atomic accept_count_{0}; + + std::mutex mutex_; + std::unordered_map store_; + std::vector commands_; + int error_replies_remaining_ = 0; + std::string error_message_; +}; + +int CountCommands(const std::vector &commands, + const std::string &prefix) { + int count = 0; + for (const auto &command : commands) { + if (command.rfind(prefix, 0) == 0) ++count; + } + return count; +} + +class RedisStoragePluginTest : public ::testing::Test { + protected: + void SetUp() override { + // hiredis does not mask SIGPIPE on Linux; writing to a connection the + // server has closed would otherwise abort the test binary. + signal(SIGPIPE, SIG_IGN); + unsetenv("MC_REDIS_USERNAME"); + unsetenv("MC_REDIS_PASSWORD"); + unsetenv("MC_REDIS_DB_INDEX"); + } + + void TearDown() override { + unsetenv("MC_REDIS_USERNAME"); + unsetenv("MC_REDIS_PASSWORD"); + unsetenv("MC_REDIS_DB_INDEX"); + } +}; + +TEST_F(RedisStoragePluginTest, SetGetRemoveRoundTrip) { + FakeRedisServer server; + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value written; + written["segment"] = "node-1"; + ASSERT_TRUE(plugin->set("mooncake/node-1", written)); + + Json::Value read; + ASSERT_TRUE(plugin->get("mooncake/node-1", read)); + EXPECT_EQ(read["segment"].asString(), "node-1"); + + ASSERT_TRUE(plugin->remove("mooncake/node-1")); + EXPECT_FALSE(plugin->get("mooncake/node-1", read)); +} + +// The regression this patch fixes: hiredis latches the error into the +// redisContext, so before the fix every call after a dropped connection failed +// for the remaining lifetime of the process. +TEST_F(RedisStoragePluginTest, SetSucceedsAfterServerDropsConnection) { + FakeRedisServer server; + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value value; + value["segment"] = "node-1"; + ASSERT_TRUE(plugin->set("mooncake/node-1", value)); + ASSERT_EQ(server.accept_count(), 1); + + server.DropConnections(); + + value["segment"] = "node-1-again"; + EXPECT_TRUE(plugin->set("mooncake/node-1", value)); + EXPECT_EQ(server.accept_count(), 2) << "plugin did not reconnect"; + + Json::Value read; + ASSERT_TRUE(plugin->get("mooncake/node-1", read)); + EXPECT_EQ(read["segment"].asString(), "node-1-again"); +} + +TEST_F(RedisStoragePluginTest, GetSucceedsAfterServerDropsConnection) { + FakeRedisServer server; + server.SetValue("mooncake/node-2", "{\"segment\":\"node-2\"}"); + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value read; + ASSERT_TRUE(plugin->get("mooncake/node-2", read)); + + server.DropConnections(); + + read = Json::Value(); + EXPECT_TRUE(plugin->get("mooncake/node-2", read)); + EXPECT_EQ(read["segment"].asString(), "node-2"); + EXPECT_EQ(server.accept_count(), 2); +} + +TEST_F(RedisStoragePluginTest, RemoveSucceedsAfterServerDropsConnection) { + FakeRedisServer server; + server.SetValue("mooncake/node-3", "{\"segment\":\"node-3\"}"); + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value read; + ASSERT_TRUE(plugin->get("mooncake/node-3", read)); + + server.DropConnections(); + + EXPECT_TRUE(plugin->remove("mooncake/node-3")); + EXPECT_EQ(server.accept_count(), 2); +} + +// Before the fix a redis that was down at construction time left client_ null +// forever, permanently disabling metadata publishing for the process. +TEST_F(RedisStoragePluginTest, RecoversWhenRedisIsDownAtConstruction) { + uint16_t port = ReserveUnusedPort(); + ASSERT_NE(port, 0); + + auto plugin = MetadataStoragePlugin::Create("redis://127.0.0.1:" + + std::to_string(port)); + ASSERT_NE(plugin, nullptr); + + Json::Value value; + value["segment"] = "node-4"; + ASSERT_FALSE(plugin->set("mooncake/node-4", value)); + + FakeRedisServer server(port); + ASSERT_EQ(server.port(), port); + + EXPECT_TRUE(plugin->set("mooncake/node-4", value)); + + Json::Value read; + ASSERT_TRUE(plugin->get("mooncake/node-4", read)); + EXPECT_EQ(read["segment"].asString(), "node-4"); +} + +// AUTH/SELECT moved into the connect path, so they must be replayed on every +// reconnect. A reconnected-but-unauthenticated connection would fail on the +// server in production even though the socket is healthy. +TEST_F(RedisStoragePluginTest, AuthAndSelectAreReplayedOnReconnect) { + setenv("MC_REDIS_USERNAME", "mooncake", 1); + setenv("MC_REDIS_PASSWORD", "secret", 1); + setenv("MC_REDIS_DB_INDEX", "3", 1); + + FakeRedisServer server; + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value value; + value["segment"] = "node-5"; + ASSERT_TRUE(plugin->set("mooncake/node-5", value)); + + auto commands = server.commands(); + EXPECT_EQ(CountCommands(commands, "AUTH mooncake secret"), 1); + EXPECT_EQ(CountCommands(commands, "SELECT 3"), 1); + + server.DropConnections(); + ASSERT_TRUE(plugin->set("mooncake/node-5", value)); + + commands = server.commands(); + EXPECT_EQ(CountCommands(commands, "AUTH mooncake secret"), 2); + EXPECT_EQ(CountCommands(commands, "SELECT 3"), 2); +} + +// #4107: a REDIS_REPLY_ERROR means the server answered and rejected the +// command (READONLY after failover, OOM, MISCONF, NOAUTH). That must surface as +// failure — and must not be retried, because reconnecting cannot fix a +// logical rejection from a healthy socket. +TEST_F(RedisStoragePluginTest, ErrorReplyIsReportedAsFailure) { + FakeRedisServer server; + server.SetValue("mooncake/node-9", "{\"segment\":\"node-9\"}"); + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value read; + ASSERT_EQ(plugin->getWithStatus("mooncake/node-9", read), + GetResult::kFound); + ASSERT_EQ(server.accept_count(), 1); + + server.ForceErrorReplies(4, + "READONLY You can't write against a read " + "only replica"); + + Json::Value value; + value["segment"] = "node-9-write"; + EXPECT_FALSE(plugin->set("mooncake/node-9", value)); + EXPECT_FALSE(plugin->get("mooncake/node-9", read)); + EXPECT_EQ(plugin->getWithStatus("mooncake/node-9", read), + GetResult::kUnavailable); + EXPECT_FALSE(plugin->remove("mooncake/node-9")); + + // A logical error reply must not tear down and rebuild the connection. + EXPECT_EQ(server.accept_count(), 1); + + // Once the forced errors are exhausted the same connection works again. + EXPECT_TRUE(plugin->set("mooncake/node-9", value)); + EXPECT_EQ(server.accept_count(), 1); +} + +// A permanently dead redis must still fail (and fail fast) rather than hang or +// spin on reconnect attempts. +TEST_F(RedisStoragePluginTest, FailsWhenRedisStaysDown) { + FakeRedisServer server; + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value value; + value["segment"] = "node-6"; + ASSERT_TRUE(plugin->set("mooncake/node-6", value)); + + server.Stop(); + + EXPECT_FALSE(plugin->set("mooncake/node-6", value)); + + Json::Value read; + EXPECT_FALSE(plugin->get("mooncake/node-6", read)); + EXPECT_FALSE(plugin->remove("mooncake/node-6")); +} + +} // namespace +} // namespace mooncake + +int main(int argc, char **argv) { + google::InitGoogleLogging(argv[0]); + FLAGS_logtostderr = true; + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From e1e12b1285e02635ab37266e1bd3bf88cc2e25bd Mon Sep 17 00:00:00 2001 From: zhangqiang30 Date: Mon, 14 Sep 2026 15:12:01 +0800 Subject: [PATCH 2/3] [TransferEngine] Route getWithStatus through the reconnect path #3072 added RedisStoragePlugin::getWithStatus() so syncSegmentCache() can tell an absent key from a backend outage. That override issued its own redisCommand() with no reconnect, so #4003 remained reproducible through it: once hiredis latched an I/O error into the context, every later getWithStatus() returned kUnavailable for the rest of the process lifetime, and syncSegmentCache() calls it periodically. Route it through the same execLocked() retry path and let get() delegate to it, which also removes the duplicated GET handling. A null reply after execLocked() means the backend is unreachable, so it maps to kUnavailable and never to kNotFound, preserving the invariant #3072 added. Add GetWithStatusRecoversAfterServerDropsConnection and GetWithStatusDistinguishesAbsentKeyFromOutage, which pins both directions: an absent key stays kNotFound while the server is up, and a stopped server reports kUnavailable rather than kNotFound. --- .../src/transfer_metadata_plugin.cpp | 51 ++++++------------- .../transfer_metadata_redis_plugin_test.cpp | 46 +++++++++++++++++ 2 files changed, 62 insertions(+), 35 deletions(-) diff --git a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp index b8cd277ffe..e696c45368 100644 --- a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp +++ b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp @@ -93,39 +93,20 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { virtual ~RedisStoragePlugin() { disconnectLocked(); } + // Delegates to getWithStatus() so both lookup entry points share the same + // reconnect-and-retry path. Deliberately does not take + // access_client_mutex_: getWithStatus() takes it, and the mutex is not + // recursive. virtual bool get(const std::string &key, Json::Value &value) { - std::lock_guard lock(access_client_mutex_); - redisReply *resp = execLocked("GET", key, [&key](redisContext *ctx) { - return static_cast( - redisCommand(ctx, "GET %s", key.c_str())); - }); - if (!resp) return false; - if (resp->type == REDIS_REPLY_ERROR) { - LOG(ERROR) << "RedisStoragePlugin: get " << key << " rejected by " - << metadata_uri_ << ": " - << (resp->str ? resp->str : "unknown error"); - freeReplyObject(resp); - return false; - } - if (!resp->str) { + const GetResult status = getWithStatus(key, value); + if (status == GetResult::kNotFound) { // A nil reply means the key is absent, which is an expected result // when looking up a segment that is not registered (yet). Not an // error, so it must not be logged as one. LOG(WARNING) << "RedisStoragePlugin: unable to get " << key << " from " << metadata_uri_; - freeReplyObject(resp); - return false; - } - - auto json_file = std::string(resp->str); - freeReplyObject(resp); - - std::string errs; - if (!parseJsonString(json_file, value, &errs)) { - LOG(ERROR) << "RedisStoragePlugin: JSON parse error: " << errs; - return false; } - return true; + return status == GetResult::kFound; } // Distinguishes a nil reply (key absent — kNotFound) from a connection @@ -134,15 +115,15 @@ struct RedisStoragePlugin : public MetadataStoragePlugin { GetResult getWithStatus(const std::string &key, Json::Value &value) override { std::lock_guard lock(access_client_mutex_); - if (!client_) return GetResult::kUnavailable; - - redisReply *resp = - (redisReply *)redisCommand(client_, "GET %s", key.c_str()); - if (!resp) { - LOG(ERROR) << "RedisStoragePlugin: unable to get " << key - << " from " << metadata_uri_; - return GetResult::kUnavailable; - } + redisReply *resp = execLocked("GET", key, [&key](redisContext *ctx) { + return static_cast( + redisCommand(ctx, "GET %s", key.c_str())); + }); + // execLocked() has already rebuilt the connection and retried once, so + // a null reply here means the backend is unreachable, never that the + // key is absent. Reporting kNotFound in that case would let + // syncSegmentCache() evict live entries during an outage. + if (!resp) return GetResult::kUnavailable; if (resp->type == REDIS_REPLY_ERROR) { LOG(ERROR) << "RedisStoragePlugin: get " << key << " rejected by " << metadata_uri_ << ": " diff --git a/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp b/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp index 18b3361a8d..072abdc05b 100644 --- a/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp +++ b/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp @@ -400,6 +400,52 @@ TEST_F(RedisStoragePluginTest, GetSucceedsAfterServerDropsConnection) { EXPECT_EQ(server.accept_count(), 2); } +// #3072 added getWithStatus() so syncSegmentCache() can tell an absent key from +// a backend outage. It is a second entry point into the same connection, so it +// has to recover from a dropped connection exactly like get() does. +TEST_F(RedisStoragePluginTest, + GetWithStatusRecoversAfterServerDropsConnection) { + FakeRedisServer server; + server.SetValue("mooncake/node-7", "{\"segment\":\"node-7\"}"); + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value read; + ASSERT_EQ(plugin->getWithStatus("mooncake/node-7", read), + GetResult::kFound); + + server.DropConnections(); + + read = Json::Value(); + EXPECT_EQ(plugin->getWithStatus("mooncake/node-7", read), + GetResult::kFound); + EXPECT_EQ(read["segment"].asString(), "node-7"); + EXPECT_EQ(server.accept_count(), 2) << "plugin did not reconnect"; +} + +// The invariant #3072 protects: an outage must never be reported as kNotFound, +// because syncSegmentCache() would then evict still-live cache entries. An +// absent key must still be reported as kNotFound so genuine unmounts are seen. +TEST_F(RedisStoragePluginTest, GetWithStatusDistinguishesAbsentKeyFromOutage) { + FakeRedisServer server; + server.SetValue("mooncake/node-8", "{\"segment\":\"node-8\"}"); + auto plugin = MetadataStoragePlugin::Create(server.endpoint()); + ASSERT_NE(plugin, nullptr); + + Json::Value read; + ASSERT_EQ(plugin->getWithStatus("mooncake/node-8", read), + GetResult::kFound); + EXPECT_EQ(plugin->getWithStatus("mooncake/absent", read), + GetResult::kNotFound); + + server.Stop(); + + EXPECT_EQ(plugin->getWithStatus("mooncake/node-8", read), + GetResult::kUnavailable); + EXPECT_EQ(plugin->getWithStatus("mooncake/absent", read), + GetResult::kUnavailable); +} + TEST_F(RedisStoragePluginTest, RemoveSucceedsAfterServerDropsConnection) { FakeRedisServer server; server.SetValue("mooncake/node-3", "{\"segment\":\"node-3\"}"); From 9ed89a42848ba66196456edc1c4041098e107726 Mon Sep 17 00:00:00 2001 From: zhangqiang30 Date: Mon, 14 Sep 2026 15:12:03 +0800 Subject: [PATCH 3/3] [CI] Run the redis metadata plugin test with USE_REDIS=ON The job that runs ctest configured the tree without -DUSE_REDIS=ON, so transfer_metadata_redis_plugin_test was never built there, and the job that does enable USE_REDIS only builds without running ctest. The redis regression tests therefore had compile coverage but were never executed. dependencies.sh already installs libhiredis-dev on this runner and the top-level CMakeLists calls enable_testing(), so enabling the flag is enough for ctest to pick the target up. The tests need no redis server: they drive an in-process RESP fake. --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ab00940bd3..e70063c149 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -220,6 +220,7 @@ jobs: -DUSE_CXL=ON \ -DUSE_UB=ON \ -DUSE_ETCD=ON \ + -DUSE_REDIS=ON \ -DUSE_CUDA=ON \ -DWITH_P2P_STORE=ON \ -DWITH_CONDUCTOR=ON \