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 \ diff --git a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp index 95d4cd05a9..e696c45368 100644 --- a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp +++ b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp @@ -69,110 +69,44 @@ 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(); } + // 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_); - 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; - } - 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) { - LOG(ERROR) << "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; + 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_; + } + return status == GetResult::kFound; } // Distinguishes a nil reply (key absent — kNotFound) from a connection @@ -181,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_ << ": " @@ -215,17 +149,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 +174,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 +190,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..072abdc05b --- /dev/null +++ b/mooncake-transfer-engine/tests/transfer_metadata_redis_plugin_test.cpp @@ -0,0 +1,579 @@ +// 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); +} + +// #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\"}"); + 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(); +}