From 4cf376ed448acd4e775233f811d3e26ba2fb2ae9 Mon Sep 17 00:00:00 2001 From: eric_song Date: Fri, 17 Jul 2026 20:32:28 +0800 Subject: [PATCH 1/3] [Store] Prevent file-per-key path collisions --- mooncake-store/include/utils.h | 6 + mooncake-store/src/storage_backend.cpp | 73 +++++++++- mooncake-store/src/utils.cpp | 27 +++- mooncake-store/tests/storage_backend_test.cpp | 129 ++++++++++++++++++ mooncake-store/tests/utils_test.cpp | 15 ++ 5 files changed, 243 insertions(+), 7 deletions(-) diff --git a/mooncake-store/include/utils.h b/mooncake-store/include/utils.h index 0b1fe210fa..073fad9183 100644 --- a/mooncake-store/include/utils.h +++ b/mooncake-store/include/utils.h @@ -540,4 +540,10 @@ std::string ResolveMooncakeHostId(const std::string& local_hostname); std::string ResolvePathFromKey(const std::string& key, const std::string& root_dir, const std::string& fsdir); + +// Resolves the pre-digest file-per-key layout. Kept for reading cache files +// written before ResolvePathFromKey switched to stable digest filenames. +std::string ResolveLegacyPathFromKey(const std::string& key, + const std::string& root_dir, + const std::string& fsdir); } // namespace mooncake diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index 3b666fade2..b6e59d13b6 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -1421,10 +1421,40 @@ StorageBackendAdaptor::EvictAboveDiskWatermark( tl::expected StorageBackendAdaptor::IsExist( const std::string& key) { - auto path = ResolvePathFromKey(key, file_storage_config_.storage_filepath, - file_per_key_config_.fsdir); namespace fs = std::filesystem; - return fs::exists(path); + const auto path = ResolvePathFromKey( + key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); + if (fs::exists(path)) { + return true; + } + + const auto legacy_path = ResolveLegacyPathFromKey( + key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); + if (!fs::exists(legacy_path)) { + return false; + } + + std::error_code ec; + const auto size = fs::file_size(legacy_path, ec); + if (ec) { + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + std::string kv_buf; + auto load_result = storage_backend_->LoadObject(legacy_path, kv_buf, + static_cast(size)); + if (!load_result) { + return tl::make_unexpected(load_result.error()); + } + + try { + KVEntry kv; + struct_pb::from_pb(kv, kv_buf); + return kv.key == key; + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to decode legacy file for key: " << key + << ", error: " << e.what(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } } tl::expected StorageBackendAdaptor::BatchLoad( @@ -1435,6 +1465,14 @@ tl::expected StorageBackendAdaptor::BatchLoad( auto path = ResolvePathFromKey(kv.key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); + if (!std::filesystem::exists(path)) { + const auto legacy_path = ResolveLegacyPathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + if (std::filesystem::exists(legacy_path)) { + path = legacy_path; + } + } kv.value.resize(slice.size); @@ -1447,7 +1485,25 @@ tl::expected StorageBackendAdaptor::BatchLoad( return tl::make_unexpected(r.error()); } - struct_pb::from_pb(kv, kv_buf); + try { + struct_pb::from_pb(kv, kv_buf); + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to decode file for key: " << key + << ", error: " << e.what(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + if (kv.key != key) { + LOG(ERROR) << "Stored key does not match requested key"; + return tl::make_unexpected(ErrorCode::INVALID_KEY); + } + if (kv.value.size() != slice.size) { + LOG(ERROR) + << "Stored value size does not match destination for key: " + << key << ", expected: " << slice.size + << ", got: " << kv.value.size(); + return tl::make_unexpected(ErrorCode::INVALID_PARAMS); + } if (!kv.value.empty()) { std::memcpy(slice.ptr, kv.value.data(), kv.value.size()); @@ -1536,6 +1592,15 @@ tl::expected StorageBackendAdaptor::ScanMeta( KVEntry kv; struct_pb::from_pb(kv, buf); + + const auto canonical_path = ResolvePathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + if (p.lexically_normal() != + fs::path(canonical_path).lexically_normal() && + fs::exists(canonical_path)) { + continue; + } storage_backend_->UpdateFileRecordKey(p.string(), kv.key); total_keys++; diff --git a/mooncake-store/src/utils.cpp b/mooncake-store/src/utils.cpp index 3f923b1b80..5548722341 100644 --- a/mooncake-store/src/utils.cpp +++ b/mooncake-store/src/utils.cpp @@ -14,11 +14,14 @@ #include #include #include +#include #include +#include #include #include #include +#include #include #include #include @@ -766,9 +769,9 @@ static std::string SanitizeKey(const std::string &key) { return sanitized_key; } -std::string ResolvePathFromKey(const std::string &key, - const std::string &root_dir, - const std::string &fsdir) { +std::string ResolveLegacyPathFromKey(const std::string &key, + const std::string &root_dir, + const std::string &fsdir) { // Compute hash of the key size_t hash = std::hash{}(key); @@ -788,4 +791,22 @@ std::string ResolvePathFromKey(const std::string &key, return full_path.lexically_normal().string(); } + +std::string ResolvePathFromKey(const std::string &key, + const std::string &root_dir, + const std::string &fsdir) { + const XXH128_hash_t hash = XXH3_128bits(key.data(), key.size()); + std::array digest{}; + std::snprintf(digest.data(), digest.size(), "%016llx%016llx", + static_cast(hash.high64), + static_cast(hash.low64)); + + const std::string digest_string(digest.data()); + namespace fs = std::filesystem; + const fs::path dir_path = + fs::path(digest_string.substr(0, 2)) / digest_string.substr(2, 2); + return (fs::path(root_dir) / fsdir / dir_path / digest_string) + .lexically_normal() + .string(); +} } // namespace mooncake diff --git a/mooncake-store/tests/storage_backend_test.cpp b/mooncake-store/tests/storage_backend_test.cpp index b87272358a..72f89f09be 100644 --- a/mooncake-store/tests/storage_backend_test.cpp +++ b/mooncake-store/tests/storage_backend_test.cpp @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -23,6 +24,28 @@ namespace fs = std::filesystem; namespace mooncake::test { +static std::pair FindLegacyPathCollision( + const std::string& root_dir, const std::string& fsdir) { + constexpr std::array kChars = {'_', '/', '\\', ':', '*', + '?', '"', '<', '>', '|'}; + std::unordered_map key_by_path; + const std::string tenant_prefix("default\0", 8); + for (size_t value = 0; value < 10000; ++value) { + size_t remaining = value; + std::string key = tenant_prefix; + for (size_t i = 0; i < 4; ++i) { + key.push_back(kChars[remaining % kChars.size()]); + remaining /= kChars.size(); + } + const auto path = ResolveLegacyPathFromKey(key, root_dir, fsdir); + auto [it, inserted] = key_by_path.emplace(path, key); + if (!inserted && it->second != key) { + return {it->second, key}; + } + } + return {}; +} + class StorageBackendTest : public ::testing::Test { protected: std::string data_path; @@ -708,6 +731,112 @@ TEST_F(StorageBackendTest, AdaptorBatchOffloadAndBatchLoad) { } } +TEST_F(StorageBackendTest, AdaptorSeparatesLegacyPathCollisions) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_collision"; + backend_config.enable_eviction = false; + + auto [first_key, second_key] = + FindLegacyPathCollision(cfg.storage_filepath, backend_config.fsdir); + ASSERT_FALSE(first_key.empty()); + ASSERT_EQ(ResolveLegacyPathFromKey(first_key, cfg.storage_filepath, + backend_config.fsdir), + ResolveLegacyPathFromKey(second_key, cfg.storage_filepath, + backend_config.fsdir)); + ASSERT_NE(ResolvePathFromKey(first_key, cfg.storage_filepath, + backend_config.fsdir), + ResolvePathFromKey(second_key, cfg.storage_filepath, + backend_config.fsdir)); + + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + std::string first_value(64, 'a'); + std::string second_value(64, 'b'); + std::unordered_map> batch = { + {first_key, {{first_value.data(), first_value.size()}}}, + {second_key, {{second_value.data(), second_value.size()}}}, + }; + auto offload_result = adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; }); + ASSERT_TRUE(offload_result); + EXPECT_EQ(offload_result.value(), 2); + + std::array first_output{}; + std::array second_output{}; + std::unordered_map loads = { + {first_key, {first_output.data(), first_output.size()}}, + {second_key, {second_output.data(), second_output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(loads)); + EXPECT_EQ(std::string(first_output.data(), first_output.size()), + first_value); + EXPECT_EQ(std::string(second_output.data(), second_output.size()), + second_value); +} + +TEST_F(StorageBackendTest, AdaptorValidatesLegacyFallbackKey) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_legacy_fallback"; + backend_config.enable_eviction = false; + + auto [stored_key, colliding_key] = + FindLegacyPathCollision(cfg.storage_filepath, backend_config.fsdir); + ASSERT_FALSE(stored_key.empty()); + + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + std::string value(64, 'v'); + std::unordered_map> batch = { + {stored_key, {{value.data(), value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + const auto canonical_path = ResolvePathFromKey( + stored_key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey( + stored_key, cfg.storage_filepath, backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + fs::rename(canonical_path, legacy_path); + + auto stored_exists = adaptor.IsExist(stored_key); + ASSERT_TRUE(stored_exists); + EXPECT_TRUE(stored_exists.value()); + auto colliding_exists = adaptor.IsExist(colliding_key); + ASSERT_TRUE(colliding_exists); + EXPECT_FALSE(colliding_exists.value()); + + std::array output{}; + std::unordered_map valid_load = { + {stored_key, {output.data(), output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(valid_load)); + EXPECT_EQ(std::string(output.data(), output.size()), value); + + std::unordered_map wrong_load = { + {colliding_key, {output.data(), output.size()}}, + }; + auto wrong_result = adaptor.BatchLoad(wrong_load); + ASSERT_FALSE(wrong_result); + EXPECT_EQ(wrong_result.error(), ErrorCode::INVALID_KEY); +} + TEST_F(StorageBackendTest, AdaptorBatchOffload_PartialSuccess) { FileStorageConfig cfg; cfg.storage_filepath = data_path; diff --git a/mooncake-store/tests/utils_test.cpp b/mooncake-store/tests/utils_test.cpp index b3102fd9fa..aa422f1522 100644 --- a/mooncake-store/tests/utils_test.cpp +++ b/mooncake-store/tests/utils_test.cpp @@ -6,8 +6,23 @@ #include #include +#include + using namespace mooncake; +TEST(UtilsTest, ResolvePathFromKeyUsesBoundedStableDigest) { + const std::string root = "/tmp/mooncake"; + const std::string fsdir = "file_per_key"; + const std::string key(4096, '/'); + + const auto first = ResolvePathFromKey(key, root, fsdir); + const auto second = ResolvePathFromKey(key, root, fsdir); + EXPECT_EQ(first, second); + EXPECT_EQ(std::filesystem::path(first).filename().string().size(), 32); + EXPECT_NE(first, ResolvePathFromKey(key + "x", root, fsdir)); + EXPECT_GT(ResolveLegacyPathFromKey(key, root, fsdir).size(), first.size()); +} + TEST(UtilsTest, ByteSizeToString) { EXPECT_EQ(byte_size_to_string(999), "999 B"); EXPECT_EQ(byte_size_to_string(2048), "2.00 KB"); From b63b8172e13a977cfd5e6d48ed3e1888dbff4f25 Mon Sep 17 00:00:00 2001 From: eric_song Date: Sat, 18 Jul 2026 02:01:22 +0800 Subject: [PATCH 2/3] [Store] Handle missing legacy files during lookup --- mooncake-store/src/storage_backend.cpp | 7 +++---- mooncake-store/tests/storage_backend_test.cpp | 5 +++++ 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index b6e59d13b6..2287eb0f33 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -1430,13 +1430,12 @@ tl::expected StorageBackendAdaptor::IsExist( const auto legacy_path = ResolveLegacyPathFromKey( key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); - if (!fs::exists(legacy_path)) { - return false; - } - std::error_code ec; const auto size = fs::file_size(legacy_path, ec); if (ec) { + if (ec == std::errc::no_such_file_or_directory) { + return false; + } return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); } std::string kv_buf; diff --git a/mooncake-store/tests/storage_backend_test.cpp b/mooncake-store/tests/storage_backend_test.cpp index 72f89f09be..5223c498e4 100644 --- a/mooncake-store/tests/storage_backend_test.cpp +++ b/mooncake-store/tests/storage_backend_test.cpp @@ -835,6 +835,11 @@ TEST_F(StorageBackendTest, AdaptorValidatesLegacyFallbackKey) { auto wrong_result = adaptor.BatchLoad(wrong_load); ASSERT_FALSE(wrong_result); EXPECT_EQ(wrong_result.error(), ErrorCode::INVALID_KEY); + + ASSERT_TRUE(fs::remove(legacy_path)); + auto removed_exists = adaptor.IsExist(stored_key); + ASSERT_TRUE(removed_exists); + EXPECT_FALSE(removed_exists.value()); } TEST_F(StorageBackendTest, AdaptorBatchOffload_PartialSuccess) { From 398bbbd695ea011ad2c2fca229fe4f931375d16e Mon Sep 17 00:00:00 2001 From: eric_song Date: Wed, 22 Jul 2026 01:11:36 +0800 Subject: [PATCH 3/3] [Store] Harden file-per-key legacy migration --- mooncake-store/include/storage_backend.h | 74 +- mooncake-store/include/utils.h | 11 +- mooncake-store/src/storage_backend.cpp | 968 ++++++++++++++++-- mooncake-store/src/utils.cpp | 18 +- .../tests/master_service_ssd_test.cpp | 10 +- mooncake-store/tests/storage_backend_test.cpp | 829 ++++++++++++++- mooncake-store/tests/utils_test.cpp | 20 +- 7 files changed, 1794 insertions(+), 136 deletions(-) diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index 1ccef58e64..6b77862b19 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -4,13 +4,16 @@ #include #include +#include #include #include #include #include #include +#include #include #include +#include #include #include @@ -410,9 +413,9 @@ class StorageBackend { * Idempotency: This method is idempotent; calling it multiple times has the * same effect as calling it once. * - * Existing files: If there are existing files in the storage directory, - * they will be scanned and incorporated into the internal state. No files - * are deleted or overwritten during initialization. + * Existing files: Committed files are scanned and incorporated into the + * internal state. Uncommitted files in the reserved staging directory are + * removed before quota reconstruction. * * Thread-safety: This method is not thread-safe and should not be called * concurrently from multiple threads. It is recommended to call Init() from @@ -467,6 +470,20 @@ class StorageBackend { const std::string& key = "", StorageBackendInterface::EvictionHandler eviction_handler = nullptr); + /** + * @brief Atomically stores a guard file and conditionally retires another + * file while holding both path locks + * @return true when the retired file was validated but its unlink failed; + * false when it was removed, absent, or rejected by the validator + */ + tl::expected StoreObjectAndRemoveFileIf( + const std::string& path, std::span data, + const std::string& key, const std::string& retire_path, + const std::function( + const std::string& retire_path, const std::string& guard_path)>& + validator, + StorageBackendInterface::EvictionHandler eviction_handler = nullptr); + tl::expected, ErrorCode> EvictAboveDiskWatermark( double high_watermark_ratio, double low_watermark_ratio, StorageBackendInterface::EvictionHandler eviction_handler = nullptr); @@ -500,6 +517,28 @@ class StorageBackend { */ void RemoveFile(const std::string& path); + /** + * @brief Deletes a file only when a caller-provided validator accepts it + * @param path Path to the file to inspect and remove + * @param guard_path A second path that must remain stable during validation + * @param guard_key Key assigned to guard_path after successful validation + * @param validator Validation callback invoked while both path locks are + * held + * @return true if the file was removed, false if it was absent or rejected + */ + tl::expected RemoveFileIf( + const std::string& path, const std::string& guard_path, + const std::string& guard_key, + const std::function( + const std::string& path, const std::string& guard_path)>& + validator); + + /** + * @brief Removes uncommitted atomic-write files from the reserved staging + * directory and updates eviction accounting + */ + tl::expected CleanupStagingFiles(); + /** * @brief Removes objects from the storage backend whose keys match a regex * pattern. @@ -530,6 +569,8 @@ class StorageBackend { std::unordered_map::iterator> file_queue_map_; std::unordered_set pending_eviction_paths_; + std::unordered_map eviction_protected_path_counts_; + std::condition_variable_any pending_eviction_cv_; mutable std::shared_mutex file_queue_mutex_; // Mutex to protect file queue operations static constexpr size_t kFilePathLockCount = 64; @@ -565,7 +606,13 @@ class StorageBackend { */ FileRecord EvictFile(); - FileRecord PopFileToEvictByFIFO(); + FileRecord PopFileToEvictByFIFO( + const std::unordered_set& excluded_paths = {}); + + bool ProtectPathsFromEviction(const std::unordered_set& paths); + + void UnprotectPathsFromEviction( + const std::unordered_set& paths); void RestoreFileToWriteQueueFront(const FileRecord& record); @@ -604,7 +651,8 @@ class StorageBackend { */ tl::expected, ErrorCode> EnsureDiskSpace( size_t required_size, - StorageBackendInterface::EvictionHandler eviction_handler = nullptr); + StorageBackendInterface::EvictionHandler eviction_handler = nullptr, + const std::unordered_set& excluded_paths = {}); /** * @brief Releases a specified amount of disk space and updates internal @@ -745,6 +793,9 @@ class StorageBackendAdaptor : public StorageBackendInterface { int64_t total_size GUARDED_BY(mutex_); + std::unordered_map accounted_size_by_key_ + GUARDED_BY(mutex_); + struct KVEntry { std::string key; // K tensor or its storage identifier std::string value; // V tensor or its storage block @@ -756,6 +807,19 @@ class StorageBackendAdaptor : public StorageBackendInterface { YLT_REFL(KVEntry, key, value); }; + + struct LoadedKVEntry { + KVEntry kv; + int64_t serialized_size; + }; + + tl::expected, ErrorCode> TryLoadKvEntry( + const std::string& path, bool unrepresentable_is_missing = false); + + tl::expected CleanupLegacyFile(const std::string& key); + + void UpsertKeyAccounting(const std::string& key, int64_t serialized_size) + REQUIRES(mutex_); }; class BucketStorageBackend : public StorageBackendInterface { diff --git a/mooncake-store/include/utils.h b/mooncake-store/include/utils.h index 073fad9183..c187a1d6bd 100644 --- a/mooncake-store/include/utils.h +++ b/mooncake-store/include/utils.h @@ -537,12 +537,19 @@ std::string GetEnvStringOr(const char* name, const std::string& default_value); std::string ResolveMooncakeHostId(const std::string& local_hostname); +// Resolves the long-lived path layout used by master-managed DISK replicas. +// Keep this layout stable because existing replica metadata stores these paths. std::string ResolvePathFromKey(const std::string& key, const std::string& root_dir, const std::string& fsdir); -// Resolves the pre-digest file-per-key layout. Kept for reading cache files -// written before ResolvePathFromKey switched to stable digest filenames. +// Resolves the digest-based canonical layout used by the file-per-key backend. +std::string ResolveFilePerKeyPathFromKey(const std::string& key, + const std::string& root_dir, + const std::string& fsdir); + +// Explicit alias for the pre-digest file-per-key layout, kept for reading and +// migrating cache files written before digest filenames were introduced. std::string ResolveLegacyPathFromKey(const std::string& key, const std::string& root_dir, const std::string& fsdir); diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index 2287eb0f33..8538813b6e 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -6,8 +6,10 @@ #include #include #include +#include #include +#include #include #include #include @@ -231,6 +233,23 @@ tl::expected StorageBackend::Init(uint64_t quota_bytes = 0) { return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); } } + + const fs::path staging_root = storage_root / ".staging"; + const bool staging_exists = fs::exists(staging_root, ec); + if (ec) { + LOG(ERROR) << "Init: Failed to inspect staging directory: " + << ec.message(); + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); + } + if (staging_exists) { + fs::remove_all(staging_root, ec); + if (ec) { + LOG(ERROR) << "Init: Failed to remove uncommitted staging files: " + << ec.message(); + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); + } + } + const auto space_info = fs::space(storage_root, ec); if (ec) { LOG(ERROR) << "Init: Failed to get disk space info: " << ec.message(); @@ -485,6 +504,229 @@ tl::expected, ErrorCode> StorageBackend::StoreObject( return evicted_keys; } +tl::expected StorageBackend::StoreObjectAndRemoveFileIf( + const std::string& path, std::span data, const std::string& key, + const std::string& retire_path, + const std::function( + const std::string& retire_path, const std::string& guard_path)>& + validator, + StorageBackendInterface::EvictionHandler eviction_handler) { + namespace fs = std::filesystem; + + const uint64_t file_total_size = data.size(); + uint64_t reserved_size = 0; + + auto erase_tracked_record = [&](const std::string& record_path) { + uint64_t tracked_size = 0; + std::unique_lock lock(file_queue_mutex_); + auto it = file_queue_map_.find(record_path); + if (it != file_queue_map_.end()) { + tracked_size = it->second->size; + file_write_queue_.erase(it->second); + file_queue_map_.erase(it); + } + return tracked_size; + }; + + const fs::path staging_root = + fs::path(root_dir_) / GetActualFsdir() / ".staging"; + std::string temp_path; + + auto store_locked = [&]() -> tl::expected { + if (IsEvictionEnabled()) { + std::shared_lock lock(file_queue_mutex_); + if (pending_eviction_paths_.find(path) != + pending_eviction_paths_.end() || + pending_eviction_paths_.find(retire_path) != + pending_eviction_paths_.end()) { + lock.unlock(); + ReleaseSpace(reserved_size); + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + } + + std::error_code staging_ec; + fs::create_directories(staging_root, staging_ec); + if (staging_ec) { + ReleaseSpace(reserved_size); + LOG(ERROR) << "Failed to create staging directory: " << staging_root + << ", error: " << staging_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL); + } + + std::string temp_template = + (staging_root / (fs::path(path).filename().string() + ".XXXXXX")) + .string(); + std::vector temp_buffer(temp_template.begin(), + temp_template.end()); + temp_buffer.push_back('\0'); + const int temp_fd = mkstemp(temp_buffer.data()); + if (temp_fd < 0) { + ReleaseSpace(reserved_size); + LOG(ERROR) << "Failed to create exclusive staging file in: " + << staging_root << ", error: " << strerror(errno); + return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL); + } + temp_path = temp_buffer.data(); + if (fchmod(temp_fd, 0644) != 0 || + fcntl(temp_fd, F_SETFD, FD_CLOEXEC) == -1) { + const int setup_errno = errno; + close(temp_fd); + std::error_code remove_ec; + fs::remove(temp_path, remove_ec); + const bool removed = + !remove_ec || remove_ec == std::errc::no_such_file_or_directory; + if (IsEvictionEnabled()) { + if (removed) { + ReleaseSpace(reserved_size); + } else { + AddFileToWriteQueue(temp_path, file_total_size, ""); + } + } + LOG(ERROR) << "Failed to configure staging file: " << temp_path + << ", error: " << strerror(setup_errno); + return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL); + } + + std::unique_ptr temp_file; +#ifdef USE_URING + if (use_uring_) { + temp_file = + std::make_unique(temp_path, temp_fd, 32, false); + } else { +#endif + temp_file = std::make_unique(temp_path, temp_fd); +#ifdef USE_URING + } +#endif + + auto discard_temp = + [&](ErrorCode error) -> tl::expected { + std::error_code remove_ec; + fs::remove(temp_path, remove_ec); + const bool removed = + !remove_ec || remove_ec == std::errc::no_such_file_or_directory; + if (IsEvictionEnabled()) { + if (removed) { + ReleaseSpace(reserved_size); + } else { + // Conservatively retain the full reservation until the + // uncommitted temp file is evicted. + AddFileToWriteQueue(temp_path, file_total_size, ""); + } + } + if (!removed) { + LOG(ERROR) << "Failed to remove uncommitted temp file: " + << temp_path << ", error: " << remove_ec.message(); + } + return tl::make_unexpected(error); + }; + + // Stage on the same filesystem and commit with one rename. The old + // canonical file and FIFO authority remain intact on every pre-commit + // failure. + auto write_result = WriteDataToFile(temp_file, temp_path, data, + /*reserved_size=*/0); + if (!write_result) { + temp_file.reset(); + return discard_temp(write_result.error()); + } + temp_file.reset(); + + auto validation_result = validator(retire_path, temp_path); + if (!validation_result) { + return discard_temp(validation_result.error()); + } + + std::error_code parent_ec; + fs::create_directories(fs::path(path).parent_path(), parent_ec); + if (parent_ec) { + LOG(ERROR) << "Failed to create canonical parent directory for: " + << path << ", error: " << parent_ec.message(); + return discard_temp(ErrorCode::FILE_WRITE_FAIL); + } + + std::error_code rename_ec; + fs::rename(temp_path, path, rename_ec); + if (rename_ec) { + LOG(ERROR) << "Failed to commit canonical file: " << path + << ", error: " << rename_ec.message(); + return discard_temp(ErrorCode::FILE_WRITE_FAIL); + } + + bool retire_remove_failed = false; + if (validation_result.value()) { + std::error_code remove_ec; + fs::remove(retire_path, remove_ec); + if (remove_ec && + remove_ec != std::errc::no_such_file_or_directory) { + std::unique_lock lock(file_queue_mutex_); + auto it = file_queue_map_.find(retire_path); + if (it != file_queue_map_.end()) { + it->second->key.clear(); + } + retire_remove_failed = true; + LOG(ERROR) << "Failed to retire validated file: " << retire_path + << ", error: " << remove_ec.message(); + } else if (IsEvictionEnabled()) { + const uint64_t retired_size = erase_tracked_record(retire_path); + ReleaseSpace(retired_size); + } + } + + if (IsEvictionEnabled()) { + AddFileToWriteQueue(path, file_total_size, key); + } + return retire_remove_failed; + }; + + const std::unordered_set protected_paths = {path, retire_path}; + bool paths_protected = false; + if (IsEvictionEnabled()) { + if (!initialized_.load(std::memory_order_acquire)) { + LOG(ERROR) + << "StorageBackend is not initialized. Call Init() before " + "storing objects."; + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); + } + if (!ProtectPathsFromEviction(protected_paths)) { + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + paths_protected = true; + + auto space_result = + EnsureDiskSpace(file_total_size, eviction_handler, protected_paths); + if (!space_result) { + UnprotectPathsFromEviction(protected_paths); + return tl::make_unexpected(space_result.error()); + } + reserved_size = file_total_size; + } + + auto run_store_locked = [&]() -> tl::expected { + Mutex* path_mutex = &GetFilePathMutex(path); + Mutex* retire_mutex = &GetFilePathMutex(retire_path); + if (path_mutex == retire_mutex) { + MutexLocker locker(path_mutex); + return store_locked(); + } + if (std::less{}(path_mutex, retire_mutex)) { + MutexLocker path_locker(path_mutex); + MutexLocker retire_locker(retire_mutex); + return store_locked(); + } + MutexLocker retire_locker(retire_mutex); + MutexLocker path_locker(path_mutex); + return store_locked(); + }; + + auto result = run_store_locked(); + if (paths_protected) { + UnprotectPathsFromEviction(protected_paths); + } + return result; +} + tl::expected StorageBackend::LoadObject( const std::string& path, std::vector& slices, int64_t length) { ResolvePath(path); @@ -645,6 +887,178 @@ void StorageBackend::RemoveFile(const std::string& path) { ReleaseSpace(file_size); } +tl::expected StorageBackend::CleanupStagingFiles() { + namespace fs = std::filesystem; + + const fs::path staging_root = + fs::path(root_dir_) / GetActualFsdir() / ".staging"; + std::error_code ec; + const bool staging_exists = fs::exists(staging_root, ec); + if (ec) { + LOG(ERROR) << "Failed to inspect staging directory: " << staging_root + << ", error: " << ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (!staging_exists) { + return {}; + } + + for (auto it = fs::directory_iterator(staging_root, ec); + !ec && it != fs::directory_iterator(); it.increment(ec)) { + const bool is_regular_file = it->is_regular_file(ec); + if (ec) break; + if (!is_regular_file) { + LOG(ERROR) << "Unexpected entry in staging directory: " + << it->path(); + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + + const auto staged_path = it->path().string(); + RemoveFile(staged_path); + std::error_code verify_ec; + const bool staged_exists = fs::exists(staged_path, verify_ec); + if (verify_ec) { + LOG(ERROR) << "Failed to verify staged file cleanup: " + << staged_path << ", error: " << verify_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (staged_exists) { + LOG(ERROR) << "Staged file remains after cleanup: " << staged_path; + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + } + if (ec) { + LOG(ERROR) << "Failed while scanning staging directory: " + << staging_root << ", error: " << ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + fs::remove(staging_root, ec); + if (ec) { + LOG(ERROR) << "Failed to remove staging directory: " << staging_root + << ", error: " << ec.message(); + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + return {}; +} + +tl::expected StorageBackend::RemoveFileIf( + const std::string& path, const std::string& guard_path, + const std::string& guard_key, + const std::function( + const std::string& path, const std::string& guard_path)>& validator) { + namespace fs = std::filesystem; + + bool pending_eviction = false; + auto remove_if_locked = [&]() -> tl::expected { + uint64_t tracked_size = 0; + bool tracked = false; + { + std::shared_lock lock(file_queue_mutex_); + if (pending_eviction_paths_.find(path) != + pending_eviction_paths_.end() || + pending_eviction_paths_.find(guard_path) != + pending_eviction_paths_.end()) { + pending_eviction = true; + return false; + } + auto map_it = file_queue_map_.find(path); + if (map_it != file_queue_map_.end()) { + tracked = true; + tracked_size = map_it->second->size; + } + } + + auto validation_result = validator(path, guard_path); + if (!validation_result) { + return tl::make_unexpected(validation_result.error()); + } + + auto update_file_records = [&](bool remove_retired_record, + bool disarm_retired_record) { + std::unique_lock lock(file_queue_mutex_); + auto retired_it = file_queue_map_.find(path); + if (retired_it != file_queue_map_.end()) { + if (remove_retired_record) { + file_write_queue_.erase(retired_it->second); + file_queue_map_.erase(retired_it); + } else if (disarm_retired_record) { + retired_it->second->key.clear(); + } + } + + auto guard_it = file_queue_map_.find(guard_path); + if (guard_it != file_queue_map_.end()) { + guard_it->second->key = guard_key; + } + }; + + if (!validation_result.value()) { + update_file_records(false, false); + return false; + } + + std::error_code ec; + const bool removed = fs::remove(path, ec); + if (ec && ec != std::errc::no_such_file_or_directory) { + // The canonical guard is authoritative after validation. Prevent a + // later eviction of the same-key legacy shadow from notifying the + // master while leaving physical accounting intact for retry. + update_file_records(false, true); + LOG(ERROR) << "Failed to conditionally remove file: " << path + << ", error: " << ec.message(); + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } + + update_file_records(true, false); + if (tracked) { + ReleaseSpace(tracked_size); + } + return removed || !ec || ec == std::errc::no_such_file_or_directory; + }; + + Mutex* path_mutex = &GetFilePathMutex(path); + Mutex* guard_mutex = &GetFilePathMutex(guard_path); + auto run_with_path_locks = [&]() -> tl::expected { + if (path_mutex == guard_mutex) { + MutexLocker locker(path_mutex); + return remove_if_locked(); + } + + if (std::less{}(path_mutex, guard_mutex)) { + MutexLocker path_locker(path_mutex); + MutexLocker guard_locker(guard_mutex); + return remove_if_locked(); + } + + MutexLocker guard_locker(guard_mutex); + MutexLocker path_locker(path_mutex); + return remove_if_locked(); + }; + + while (true) { + pending_eviction = false; + auto result = run_with_path_locks(); + if (!pending_eviction) { + return result; + } + + std::unique_lock lock(file_queue_mutex_); + const bool ready = + pending_eviction_cv_.wait_for(lock, std::chrono::seconds(5), [&]() { + return pending_eviction_paths_.find(path) == + pending_eviction_paths_.end() && + pending_eviction_paths_.find(guard_path) == + pending_eviction_paths_.end(); + }); + if (!ready) { + LOG(ERROR) << "Timed out waiting for file eviction before retiring " + << path; + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); + } + } +} + void StorageBackend::RemoveByRegex(const std::string& regex_pattern) { namespace fs = std::filesystem; std::regex pattern; @@ -970,34 +1384,86 @@ void StorageBackend::RemoveFileFromWriteQueue(const std::string& path) { } } -FileRecord StorageBackend::PopFileToEvictByFIFO() { +bool StorageBackend::ProtectPathsFromEviction( + const std::unordered_set& paths) { + std::unique_lock lock(file_queue_mutex_); + for (const auto& path : paths) { + if (pending_eviction_paths_.find(path) != + pending_eviction_paths_.end()) { + return false; + } + } + for (const auto& path : paths) { + ++eviction_protected_path_counts_[path]; + } + return true; +} + +void StorageBackend::UnprotectPathsFromEviction( + const std::unordered_set& paths) { + std::unique_lock lock(file_queue_mutex_); + for (const auto& path : paths) { + auto it = eviction_protected_path_counts_.find(path); + if (it == eviction_protected_path_counts_.end()) { + continue; + } + if (--it->second == 0) { + eviction_protected_path_counts_.erase(it); + } + } +} + +FileRecord StorageBackend::PopFileToEvictByFIFO( + const std::unordered_set& excluded_paths) { + auto is_excluded = [&](const FileRecord& record) { + for (const auto& excluded_path : excluded_paths) { + if (record.path == excluded_path) { + return true; + } + } + for (const auto& [protected_path, count] : + eviction_protected_path_counts_) { + if (count > 0 && record.path == protected_path) { + return true; + } + } + return false; + }; + while (true) { std::string candidate_path; { std::shared_lock lock(file_queue_mutex_); - if (file_write_queue_.empty()) { - LOG(WARNING) << "Queue is empty, cannot select file to evict"; + const auto candidate = std::find_if( + file_write_queue_.begin(), file_write_queue_.end(), + [&](const FileRecord& record) { return !is_excluded(record); }); + if (candidate == file_write_queue_.end()) { + LOG(WARNING) << "Queue has no eligible file to evict"; return {}; } - candidate_path = file_write_queue_.front().path; + candidate_path = candidate->path; } MutexLocker path_locker(&GetFilePathMutex(candidate_path)); std::unique_lock lock(file_queue_mutex_); - if (file_write_queue_.empty() || - file_write_queue_.front().path != candidate_path) { + const auto candidate = std::find_if( + file_write_queue_.begin(), file_write_queue_.end(), + [&](const FileRecord& record) { return !is_excluded(record); }); + if (candidate == file_write_queue_.end()) { + return {}; + } + if (candidate->path != candidate_path) { continue; } auto map_it = file_queue_map_.find(candidate_path); - if (map_it == file_queue_map_.end() || - map_it->second != file_write_queue_.begin()) { + if (map_it == file_queue_map_.end() || map_it->second != candidate) { continue; } - FileRecord record = file_write_queue_.front(); + FileRecord record = *candidate; file_queue_map_.erase(map_it); - file_write_queue_.pop_front(); + file_write_queue_.erase(candidate); pending_eviction_paths_.insert(record.path); return record; } @@ -1013,6 +1479,7 @@ void StorageBackend::RestoreFileToWriteQueueFront(const FileRecord& record) { MutexLocker path_locker(&GetFilePathMutex(record.path)); std::unique_lock lock(file_queue_mutex_); pending_eviction_paths_.erase(record.path); + pending_eviction_cv_.notify_all(); if (file_queue_map_.find(record.path) != file_queue_map_.end()) { was_replaced = true; } else { @@ -1043,6 +1510,7 @@ tl::expected StorageBackend::DeleteEvictedFile( // accounting without deleting the new file. was_replaced = true; pending_eviction_paths_.erase(record.path); + pending_eviction_cv_.notify_all(); } } if (was_replaced) { @@ -1057,6 +1525,7 @@ tl::expected StorageBackend::DeleteEvictedFile( { std::unique_lock queue_lock(file_queue_mutex_); pending_eviction_paths_.erase(record.path); + pending_eviction_cv_.notify_all(); } ReleaseSpace(record.size); return {}; @@ -1070,7 +1539,8 @@ tl::expected StorageBackend::DeleteEvictedFile( tl::expected, ErrorCode> StorageBackend::EnsureDiskSpace( size_t required_size, - StorageBackendInterface::EvictionHandler eviction_handler) { + StorageBackendInterface::EvictionHandler eviction_handler, + const std::unordered_set& excluded_paths) { std::vector evicted_keys; // If eviction is disabled, skip space checking and eviction if (!IsEvictionEnabled()) { @@ -1094,7 +1564,7 @@ StorageBackend::EnsureDiskSpace( std::vector pending_evictions; while (projected_available_space < required_size && attempts < kMaxEvictionAttempts) { - FileRecord pending = PopFileToEvictByFIFO(); + FileRecord pending = PopFileToEvictByFIFO(excluded_paths); if (pending.path.empty()) { LOG(ERROR) << "Failed to evict file to make space."; for (auto it = pending_evictions.rbegin(); @@ -1293,6 +1763,99 @@ StorageBackendAdaptor::StorageBackendAdaptor( total_keys(0), total_size(0) {} +tl::expected, ErrorCode> +StorageBackendAdaptor::TryLoadKvEntry(const std::string& path, + bool unrepresentable_is_missing) { + namespace fs = std::filesystem; + + std::error_code ec; + const auto size = fs::file_size(path, ec); + if (ec) { + if (ec == std::errc::no_such_file_or_directory || + (unrepresentable_is_missing && + (ec == std::errc::filename_too_long || + ec == std::errc::is_a_directory))) { + return std::nullopt; + } + LOG(ERROR) << "Failed to inspect file-per-key entry: " << path + << ", error: " << ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + if (size > static_cast(std::numeric_limits::max())) { + LOG(ERROR) << "File-per-key entry is too large to read: " << path; + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + std::string kv_buf; + auto load_result = + storage_backend_->LoadObject(path, kv_buf, static_cast(size)); + if (!load_result) { + return tl::make_unexpected(load_result.error()); + } + + try { + KVEntry kv; + struct_pb::from_pb(kv, kv_buf); + return std::optional( + LoadedKVEntry{std::move(kv), static_cast(size)}); + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to decode file-per-key entry: " << path + << ", error: " << e.what(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } +} + +tl::expected StorageBackendAdaptor::CleanupLegacyFile( + const std::string& key) { + namespace fs = std::filesystem; + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey( + key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); + if (fs::path(canonical_path).lexically_normal() == + fs::path(legacy_path).lexically_normal()) { + return false; + } + + return storage_backend_->RemoveFileIf( + legacy_path, canonical_path, key, + [this, &key]( + const std::string& candidate_path, + const std::string& guard_path) -> tl::expected { + auto canonical_entry = TryLoadKvEntry(guard_path); + if (!canonical_entry) { + return tl::make_unexpected(canonical_entry.error()); + } + if (!canonical_entry.value() || + canonical_entry.value()->kv.key != key) { + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + auto legacy_entry = + TryLoadKvEntry(candidate_path, + /*unrepresentable_is_missing=*/true); + if (!legacy_entry) { + return tl::make_unexpected(legacy_entry.error()); + } + return legacy_entry.value() && legacy_entry.value()->kv.key == key; + }); +} + +void StorageBackendAdaptor::UpsertKeyAccounting(const std::string& key, + int64_t serialized_size) { + auto [it, inserted] = accounted_size_by_key_.emplace(key, serialized_size); + if (inserted) { + total_keys++; + total_size += serialized_size; + return; + } + + total_size += serialized_size - it->second; + it->second = serialized_size; +} + tl::expected StorageBackendAdaptor::Init() { std::string storage_root = file_storage_config_.storage_filepath + file_per_key_config_.fsdir; @@ -1338,56 +1901,123 @@ tl::expected StorageBackendAdaptor::BatchOffload( return tl::make_unexpected(ErrorCode::INVALID_KEY); } - auto enable_offloading_res = IsEnableOffloading(); - if (!enable_offloading_res) { - return tl::make_unexpected(enable_offloading_res.error()); - } std::vector metadatas; std::vector keys; metadatas.reserve(batch_object.size()); keys.reserve(batch_object.size()); - // Process each key; continue on individual failures to support partial - // success - for (auto& object : batch_object) { - KVEntry kv; - kv.key = object.first; - auto value = object.second; - - // Test-only: Check if this key should fail (deterministic failure - // injection) - if (test_failure_predicate_ && test_failure_predicate_(kv.key)) { - LOG(INFO) << "[TEST] Injecting failure for key: " << kv.key - << " (test failure predicate)"; - continue; // Simulate StoreObject failure + { + // Serialize ScanMeta with the whole write/legacy-retirement window so + // active staging files can never be mistaken for crash leftovers. + MutexLocker batch_lock(&mutex_); + if (!file_per_key_config_.enable_eviction) { + if (!meta_scanned_.load(std::memory_order_acquire)) { + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); + } + if (total_keys > file_storage_config_.total_keys_limit || + total_size > file_storage_config_.total_size_limit) { + return tl::make_unexpected(ErrorCode::KEYS_ULTRA_LIMIT); + } } - auto path = - ResolvePathFromKey(kv.key, file_storage_config_.storage_filepath, - file_per_key_config_.fsdir); - kv.value = ConcatSlicesToString(value); + // Process each key; continue on individual failures to support partial + // success + for (auto& object : batch_object) { + KVEntry kv; + kv.key = object.first; + auto value = object.second; - std::string kv_buf; - struct_pb::to_pb(kv, kv_buf); - auto store_result = storage_backend_->StoreObject(path, kv_buf, kv.key, - eviction_handler); - if (!store_result) { - LOG(ERROR) << "Failed to store object for key: " << kv.key - << ", error: " << store_result.error() - << " - continuing with remaining keys"; - continue; // Continue processing other keys - } + // Test-only: Check if this key should fail (deterministic failure + // injection) + if (test_failure_predicate_ && test_failure_predicate_(kv.key)) { + LOG(INFO) << "[TEST] Injecting failure for key: " << kv.key + << " (test failure predicate)"; + continue; // Simulate StoreObject failure + } - { - MutexLocker lock(&mutex_); - total_keys++; - total_size += kv_buf.size(); - } + auto path = ResolveFilePerKeyPathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + kv.value = ConcatSlicesToString(value); - metadatas.emplace_back( - StorageObjectMetadata{-1, 0, static_cast(kv.key.size()), - static_cast(kv.value.size()), ""}); - keys.emplace_back(kv.key); + std::string kv_buf; + struct_pb::to_pb(kv, kv_buf); + int64_t accounted_size = static_cast(kv_buf.size()); + const auto legacy_path = ResolveLegacyPathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + int64_t validated_legacy_size = 0; + bool legacy_remove_failed = false; + tl::expected store_result = false; + if (std::filesystem::path(path).lexically_normal() == + std::filesystem::path(legacy_path).lexically_normal()) { + auto direct_store = storage_backend_->StoreObject( + path, kv_buf, kv.key, eviction_handler); + if (!direct_store) { + store_result = tl::make_unexpected(direct_store.error()); + } + } else { + store_result = storage_backend_->StoreObjectAndRemoveFileIf( + path, std::span(kv_buf.data(), kv_buf.size()), + kv.key, legacy_path, + [this, &kv, &validated_legacy_size]( + const std::string& candidate_path, + const std::string&) -> tl::expected { + auto legacy_entry = + TryLoadKvEntry(candidate_path, + /*unrepresentable_is_missing=*/true); + if (!legacy_entry) { + return tl::make_unexpected(legacy_entry.error()); + } + if (!legacy_entry.value() || + legacy_entry.value()->kv.key != kv.key) { + return false; + } + validated_legacy_size = + legacy_entry.value()->serialized_size; + return true; + }, + eviction_handler); + } + if (!store_result) { + LOG(ERROR) << "Failed to atomically store object for key: " + << kv.key << ", error: " << store_result.error() + << " - continuing with remaining keys"; + if (!file_per_key_config_.enable_eviction) { + meta_scanned_.store(false, std::memory_order_release); + } + continue; + } + legacy_remove_failed = store_result.value(); + if (legacy_remove_failed) { + LOG(WARNING) + << "Failed to remove validated legacy file for key: " + << kv.key + << ". Its FIFO record was disarmed and cleanup will be " + "retried " + "by ScanMeta."; + if (!file_per_key_config_.enable_eviction) { + if (validated_legacy_size < 0 || + validated_legacy_size > + std::numeric_limits::max() - + accounted_size) { + LOG(ERROR) + << "Failed to account legacy shadow for key: " + << kv.key; + meta_scanned_.store(false, std::memory_order_release); + continue; + } + accounted_size += validated_legacy_size; + } + } + + UpsertKeyAccounting(kv.key, accounted_size); + + metadatas.emplace_back(StorageObjectMetadata{ + -1, 0, static_cast(kv.key.size()), + static_cast(kv.value.size()), ""}); + keys.emplace_back(kv.key); + } } // Only report successful keys to master @@ -1422,38 +2052,31 @@ StorageBackendAdaptor::EvictAboveDiskWatermark( tl::expected StorageBackendAdaptor::IsExist( const std::string& key) { namespace fs = std::filesystem; - const auto path = ResolvePathFromKey( + const auto path = ResolveFilePerKeyPathFromKey( key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); - if (fs::exists(path)) { + std::error_code canonical_ec; + const bool canonical_exists = fs::exists(path, canonical_ec); + if (canonical_ec) { + LOG(ERROR) << "Failed to inspect canonical file-per-key entry: " << path + << ", error: " << canonical_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (canonical_exists) { return true; } const auto legacy_path = ResolveLegacyPathFromKey( key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); - std::error_code ec; - const auto size = fs::file_size(legacy_path, ec); - if (ec) { - if (ec == std::errc::no_such_file_or_directory) { - return false; - } - return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); - } - std::string kv_buf; - auto load_result = storage_backend_->LoadObject(legacy_path, kv_buf, - static_cast(size)); - if (!load_result) { - return tl::make_unexpected(load_result.error()); - } - - try { - KVEntry kv; - struct_pb::from_pb(kv, kv_buf); - return kv.key == key; - } catch (const std::exception& e) { - LOG(ERROR) << "Failed to decode legacy file for key: " << key - << ", error: " << e.what(); - return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); - } + // Legacy filenames can collide and contain no independent key metadata, so + // ownership cannot be established from existence alone. This compatibility + // path reads and decodes the full object; a prefix-only parser can optimize + // it in a follow-up without weakening the key check. + auto legacy_entry = + TryLoadKvEntry(legacy_path, /*unrepresentable_is_missing=*/true); + if (!legacy_entry) { + return tl::make_unexpected(legacy_entry.error()); + } + return legacy_entry.value() && legacy_entry.value()->kv.key == key; } tl::expected StorageBackendAdaptor::BatchLoad( @@ -1461,14 +2084,30 @@ tl::expected StorageBackendAdaptor::BatchLoad( for (const auto& [key, slice] : batched_slices) { KVEntry kv; kv.key = key; - auto path = - ResolvePathFromKey(kv.key, file_storage_config_.storage_filepath, - file_per_key_config_.fsdir); - if (!std::filesystem::exists(path)) { + auto path = ResolveFilePerKeyPathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + std::error_code canonical_ec; + const bool canonical_exists = + std::filesystem::exists(path, canonical_ec); + if (canonical_ec) { + LOG(ERROR) << "Failed to inspect canonical file for key: " << key + << ", error: " << canonical_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (!canonical_exists) { const auto legacy_path = ResolveLegacyPathFromKey( kv.key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); - if (std::filesystem::exists(legacy_path)) { + std::error_code legacy_ec; + const bool legacy_exists = + std::filesystem::exists(legacy_path, legacy_ec); + if (legacy_ec) { + LOG(ERROR) << "Failed to inspect legacy file for key: " << key + << ", error: " << legacy_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (legacy_exists) { path = legacy_path; } } @@ -1516,13 +2155,13 @@ tl::expected StorageBackendAdaptor::IsEnableOffloading() { return true; } + MutexLocker lock(&mutex_); + if (!meta_scanned_.load(std::memory_order_acquire)) { LOG(ERROR) << "Metadata has not been loaded yet"; return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); } - MutexLocker lock(&mutex_); - auto is_enable_offloading = total_keys <= file_storage_config_.total_keys_limit && total_size <= file_storage_config_.total_size_limit; @@ -1536,15 +2175,38 @@ tl::expected StorageBackendAdaptor::ScanMeta( std::vector& metadatas)>& handler) { namespace fs = std::filesystem; + MutexLocker lock(&mutex_); + + // Rebuild accounting from the current on-disk snapshot. Offloading remains + // disabled until the whole scan and any legacy cleanup succeed. + meta_scanned_.store(false, std::memory_order_release); + accounted_size_by_key_.clear(); + total_keys = 0; + total_size = 0; + fs::path root = fs::path(file_storage_config_.storage_filepath) / file_per_key_config_.fsdir; - if (!fs::exists(root)) { + std::error_code root_ec; + const bool root_exists = fs::exists(root, root_ec); + if (root_ec) { + LOG(ERROR) << "Failed to inspect file-per-key root: " << root + << ", error: " << root_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (!root_exists) { meta_scanned_.store(true, std::memory_order_release); return {}; } + auto staging_cleanup = storage_backend_->CleanupStagingFiles(); + if (!staging_cleanup) { + return tl::make_unexpected(staging_cleanup.error()); + } + std::vector keys; std::vector metas; + std::vector legacy_keys_to_cleanup; + std::unordered_set registered_keys; auto flush = [&]() -> tl::expected { if (keys.empty()) return {}; @@ -1555,13 +2217,33 @@ tl::expected StorageBackendAdaptor::ScanMeta( return {}; }; - MutexLocker lock(&mutex_); + auto register_entry = + [&](const std::string& key, int64_t value_size, int64_t serialized_size, + const std::string& path) -> tl::expected { + if (!registered_keys.emplace(key).second) { + return {}; + } + + storage_backend_->UpdateFileRecordKey(path, key); + UpsertKeyAccounting(key, serialized_size); + keys.emplace_back(key); + metas.emplace_back(StorageObjectMetadata{ + -1, 0, static_cast(key.size()), value_size, ""}); + + if (static_cast(keys.size()) >= + file_storage_config_.scanmeta_iterator_keys_limit) { + return flush(); + } + return {}; + }; std::error_code ec_root; for (auto it1 = fs::directory_iterator(root, ec_root); !ec_root && it1 != fs::directory_iterator(); it1.increment(ec_root)) { if (ec_root) break; - if (!it1->is_directory(ec_root) || ec_root) continue; + const bool is_first_level_directory = it1->is_directory(ec_root); + if (ec_root) break; + if (!is_first_level_directory) continue; const auto& d1 = it1->path(); @@ -1569,7 +2251,9 @@ tl::expected StorageBackendAdaptor::ScanMeta( for (auto it2 = fs::directory_iterator(d1, ec_d1); !ec_d1 && it2 != fs::directory_iterator(); it2.increment(ec_d1)) { if (ec_d1) break; - if (!it2->is_directory(ec_d1) || ec_d1) continue; + const bool is_second_level_directory = it2->is_directory(ec_d1); + if (ec_d1) break; + if (!is_second_level_directory) continue; const auto& leaf = it2->path(); @@ -1579,47 +2263,111 @@ tl::expected StorageBackendAdaptor::ScanMeta( it.increment(ec_leaf)) { if (ec_leaf) break; const auto& p = it->path(); - if (!it->is_regular_file(ec_leaf) || ec_leaf) continue; + const bool is_regular_file = it->is_regular_file(ec_leaf); + if (ec_leaf) break; + if (!is_regular_file) continue; uintmax_t sz = fs::file_size(p, ec_leaf); - if (ec_leaf) continue; + if (ec_leaf) break; std::string buf; auto r = storage_backend_->LoadObject(p.string(), buf, (int64_t)sz); - if (!r) continue; + if (!r) { + LOG(ERROR) << "Failed to read file-per-key metadata: " << p + << ", error: " << r.error(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } KVEntry kv; struct_pb::from_pb(kv, buf); - const auto canonical_path = ResolvePathFromKey( + const auto canonical_path = ResolveFilePerKeyPathFromKey( kv.key, file_storage_config_.storage_filepath, file_per_key_config_.fsdir); if (p.lexically_normal() != - fs::path(canonical_path).lexically_normal() && - fs::exists(canonical_path)) { - continue; + fs::path(canonical_path).lexically_normal()) { + auto canonical_entry = TryLoadKvEntry(canonical_path); + if (!canonical_entry) { + return tl::make_unexpected(canonical_entry.error()); + } + if (canonical_entry.value() && + canonical_entry.value()->kv.key == kv.key) { + auto register_result = register_entry( + canonical_entry.value()->kv.key, + static_cast( + canonical_entry.value()->kv.value.size()), + canonical_entry.value()->serialized_size, + canonical_path); + if (!register_result) { + return register_result; + } + const auto legacy_path = ResolveLegacyPathFromKey( + kv.key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + if (p.lexically_normal() == + fs::path(legacy_path).lexically_normal()) { + legacy_keys_to_cleanup.emplace_back(kv.key); + } + continue; + } } - storage_backend_->UpdateFileRecordKey(p.string(), kv.key); - - total_keys++; - total_size += buf.size(); - - keys.emplace_back(std::move(kv.key)); - metas.emplace_back(StorageObjectMetadata{ - -1, 0, (int64_t)keys.back().size(), - static_cast(kv.value.size()), ""}); - - if ((int64_t)keys.size() >= - file_storage_config_.scanmeta_iterator_keys_limit) { - auto fr = flush(); - if (!fr) return fr; + auto register_result = register_entry( + kv.key, static_cast(kv.value.size()), + static_cast(buf.size()), p.string()); + if (!register_result) { + return register_result; } } + if (ec_leaf) { + LOG(ERROR) << "Failed while scanning file-per-key directory: " + << leaf << ", error: " << ec_leaf.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + auto fr = flush(); if (!fr) return fr; } + + if (ec_d1) { + LOG(ERROR) << "Failed while scanning file-per-key directory: " << d1 + << ", error: " << ec_d1.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + } + + if (ec_root) { + LOG(ERROR) << "Failed while scanning file-per-key root: " << root + << ", error: " << ec_root.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + + for (const auto& key : legacy_keys_to_cleanup) { + auto cleanup_result = CleanupLegacyFile(key); + if (!cleanup_result) { + return tl::make_unexpected(cleanup_result.error()); + } + if (cleanup_result.value()) { + continue; + } + + const auto legacy_path = + ResolveLegacyPathFromKey(key, file_storage_config_.storage_filepath, + file_per_key_config_.fsdir); + std::error_code legacy_ec; + const bool legacy_exists = fs::exists(legacy_path, legacy_ec); + if (legacy_ec) { + LOG(ERROR) << "Failed to verify legacy cleanup for key: " << key + << ", error: " << legacy_ec.message(); + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + } + if (legacy_exists) { + LOG(ERROR) << "Legacy file remains after canonical recovery for " + "key: " + << key; + return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL); + } } meta_scanned_.store(true, std::memory_order_release); diff --git a/mooncake-store/src/utils.cpp b/mooncake-store/src/utils.cpp index 5548722341..977884921e 100644 --- a/mooncake-store/src/utils.cpp +++ b/mooncake-store/src/utils.cpp @@ -769,9 +769,9 @@ static std::string SanitizeKey(const std::string &key) { return sanitized_key; } -std::string ResolveLegacyPathFromKey(const std::string &key, - const std::string &root_dir, - const std::string &fsdir) { +std::string ResolvePathFromKey(const std::string &key, + const std::string &root_dir, + const std::string &fsdir) { // Compute hash of the key size_t hash = std::hash{}(key); @@ -792,9 +792,15 @@ std::string ResolveLegacyPathFromKey(const std::string &key, return full_path.lexically_normal().string(); } -std::string ResolvePathFromKey(const std::string &key, - const std::string &root_dir, - const std::string &fsdir) { +std::string ResolveLegacyPathFromKey(const std::string &key, + const std::string &root_dir, + const std::string &fsdir) { + return ResolvePathFromKey(key, root_dir, fsdir); +} + +std::string ResolveFilePerKeyPathFromKey(const std::string &key, + const std::string &root_dir, + const std::string &fsdir) { const XXH128_hash_t hash = XXH3_128bits(key.data(), key.size()); std::array digest{}; std::snprintf(digest.data(), digest.size(), "%016llx%016llx", diff --git a/mooncake-store/tests/master_service_ssd_test.cpp b/mooncake-store/tests/master_service_ssd_test.cpp index 6e9c42f9d0..a6bddb5f3b 100644 --- a/mooncake-store/tests/master_service_ssd_test.cpp +++ b/mooncake-store/tests/master_service_ssd_test.cpp @@ -9,6 +9,7 @@ #include "master_metric_manager.h" #include "types.h" +#include "utils.h" namespace mooncake::test { @@ -119,7 +120,14 @@ TEST_F(MasterServiceSSDTest, PutEndBothReplica) { bool has_mem = false, has_disk = false; for (const auto& r : replicas) { if (r.is_memory_replica()) has_mem = true; - if (r.is_disk_replica()) has_disk = true; + if (r.is_disk_replica()) { + has_disk = true; + const auto& file_path = r.get_disk_descriptor().file_path; + EXPECT_EQ(file_path, ResolveLegacyPathFromKey(key, "/mnt/ssd", + DEFAULT_CLUSTER_ID)); + EXPECT_NE(file_path, ResolveFilePerKeyPathFromKey( + key, "/mnt/ssd", DEFAULT_CLUSTER_ID)); + } } EXPECT_TRUE(has_mem); EXPECT_TRUE(has_disk); diff --git a/mooncake-store/tests/storage_backend_test.cpp b/mooncake-store/tests/storage_backend_test.cpp index 5223c498e4..24e6481f74 100644 --- a/mooncake-store/tests/storage_backend_test.cpp +++ b/mooncake-store/tests/storage_backend_test.cpp @@ -745,10 +745,10 @@ TEST_F(StorageBackendTest, AdaptorSeparatesLegacyPathCollisions) { backend_config.fsdir), ResolveLegacyPathFromKey(second_key, cfg.storage_filepath, backend_config.fsdir)); - ASSERT_NE(ResolvePathFromKey(first_key, cfg.storage_filepath, - backend_config.fsdir), - ResolvePathFromKey(second_key, cfg.storage_filepath, - backend_config.fsdir)); + ASSERT_NE(ResolveFilePerKeyPathFromKey(first_key, cfg.storage_filepath, + backend_config.fsdir), + ResolveFilePerKeyPathFromKey(second_key, cfg.storage_filepath, + backend_config.fsdir)); StorageBackendAdaptor adaptor(cfg, backend_config); ASSERT_TRUE(adaptor.Init()); @@ -808,7 +808,7 @@ TEST_F(StorageBackendTest, AdaptorValidatesLegacyFallbackKey) { [](const std::vector&, std::vector&) { return ErrorCode::OK; })); - const auto canonical_path = ResolvePathFromKey( + const auto canonical_path = ResolveFilePerKeyPathFromKey( stored_key, cfg.storage_filepath, backend_config.fsdir); const auto legacy_path = ResolveLegacyPathFromKey( stored_key, cfg.storage_filepath, backend_config.fsdir); @@ -836,12 +836,753 @@ TEST_F(StorageBackendTest, AdaptorValidatesLegacyFallbackKey) { ASSERT_FALSE(wrong_result); EXPECT_EQ(wrong_result.error(), ErrorCode::INVALID_KEY); + std::string colliding_value(64, 'c'); + std::unordered_map> colliding_batch = { + {colliding_key, {{colliding_value.data(), colliding_value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + colliding_batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + EXPECT_TRUE(fs::exists(legacy_path)) + << "The colliding legacy file belongs to the first key"; + + std::array stored_output{}; + std::array colliding_output{}; + std::unordered_map collision_load = { + {stored_key, {stored_output.data(), stored_output.size()}}, + {colliding_key, {colliding_output.data(), colliding_output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(collision_load)); + EXPECT_EQ(std::string(stored_output.data(), stored_output.size()), value); + EXPECT_EQ(std::string(colliding_output.data(), colliding_output.size()), + colliding_value); + ASSERT_TRUE(fs::remove(legacy_path)); auto removed_exists = adaptor.IsExist(stored_key); ASSERT_TRUE(removed_exists); EXPECT_FALSE(removed_exists.value()); } +TEST_F(StorageBackendTest, + AdaptorOffloadsLongKeyWithoutLegacyPathProbeFailure) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_long_key"; + backend_config.enable_eviction = false; + + const std::string key(4096, '/'); + std::string value(64, 'v'); + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + ASSERT_EQ(fs::path(canonical_path).filename().string().size(), 32); + ASSERT_GT(fs::path(legacy_path).filename().string().size(), 255); + fs::create_directories(fs::path(legacy_path).parent_path()); + std::error_code legacy_status_ec; + const auto legacy_size = fs::file_size(legacy_path, legacy_status_ec); + EXPECT_EQ(legacy_size, static_cast(-1)); + ASSERT_EQ(legacy_status_ec, + std::make_error_code(std::errc::filename_too_long)); + + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + std::unordered_map> batch = { + {key, {{value.data(), value.size()}}}, + }; + auto offload_result = adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; }); + ASSERT_TRUE(offload_result); + EXPECT_EQ(offload_result.value(), 1); + EXPECT_TRUE(fs::exists(canonical_path)); + + auto exists_result = adaptor.IsExist(key); + ASSERT_TRUE(exists_result); + EXPECT_TRUE(exists_result.value()); + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(load)); + EXPECT_EQ(std::string(output.data(), output.size()), value); + } + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + std::vector recovered_keys; + ASSERT_TRUE( + restart_adaptor.ScanMeta([&](const std::vector& keys, + std::vector&) { + recovered_keys.insert(recovered_keys.end(), keys.begin(), + keys.end()); + return ErrorCode::OK; + })); + EXPECT_EQ(recovered_keys, std::vector{key}); +} + +TEST_F(StorageBackendTest, + AdaptorOffloadsDotKeysWhenLegacyPathsAreDirectories) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_dot_keys"; + backend_config.enable_eviction = false; + + const std::vector keys = {".", ".."}; + std::vector values = {std::string(64, 'a'), + std::string(64, 'b')}; + for (const auto& key : keys) { + const auto legacy_path = ResolveLegacyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + fs::create_directories(legacy_path); + ASSERT_TRUE(fs::is_directory(legacy_path)); + } + + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + std::unordered_map> batch; + for (size_t i = 0; i < keys.size(); ++i) { + batch.emplace(keys[i], std::vector{ + {values[i].data(), values[i].size()}}); + } + auto offload_result = adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; }); + ASSERT_TRUE(offload_result); + EXPECT_EQ(offload_result.value(), keys.size()); + + std::array dot_output{}; + std::array dot_dot_output{}; + std::unordered_map load = { + {keys[0], {dot_output.data(), dot_output.size()}}, + {keys[1], {dot_dot_output.data(), dot_dot_output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(load)); + EXPECT_EQ(std::string(dot_output.data(), dot_output.size()), values[0]); + EXPECT_EQ(std::string(dot_dot_output.data(), dot_dot_output.size()), + values[1]); + } + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + std::vector recovered_keys; + ASSERT_TRUE( + restart_adaptor.ScanMeta([&](const std::vector& found, + std::vector&) { + recovered_keys.insert(recovered_keys.end(), found.begin(), + found.end()); + return ErrorCode::OK; + })); + std::ranges::sort(recovered_keys); + EXPECT_EQ(recovered_keys, keys); +} + +TEST_F(StorageBackendTest, + AdaptorScanMetaFailsClosedOnCanonicalDirectoryEntryError) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_scan_entry_error"; + backend_config.enable_eviction = false; + + const std::string key = "canonical-entry-error"; + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + fs::create_directories(fs::path(canonical_path).parent_path()); + std::error_code symlink_ec; + fs::create_symlink(fs::absolute(canonical_path), canonical_path, + symlink_ec); + ASSERT_FALSE(symlink_ec) << symlink_ec.message(); + + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + bool handler_called = false; + auto scan_result = + adaptor.ScanMeta([&](const std::vector&, + std::vector&) { + handler_called = true; + return ErrorCode::OK; + }); + ASSERT_FALSE(scan_result); + EXPECT_EQ(scan_result.error(), ErrorCode::FILE_READ_FAIL); + EXPECT_FALSE(handler_called); + + auto is_enabled = adaptor.IsEnableOffloading(); + ASSERT_FALSE(is_enabled); + EXPECT_EQ(is_enabled.error(), ErrorCode::INTERNAL_ERROR); + + std::error_code cleanup_ec; + fs::remove(canonical_path, cleanup_ec); + EXPECT_FALSE(cleanup_ec) << cleanup_ec.message(); +} + +TEST_F(StorageBackendTest, + AdaptorFailedLegacyValidationPreservesCanonicalAuthority) { + for (const bool enable_eviction : {false, true}) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = enable_eviction + ? "file_per_key_atomic_rollback_evict" + : "file_per_key_atomic_rollback_no_evict"; + backend_config.enable_eviction = enable_eviction; + + const std::string key = "atomic-rollback-key"; + std::string old_value(64, 'o'); + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + if (!enable_eviction) { + ASSERT_TRUE( + adaptor.ScanMeta([](const std::vector&, + std::vector&) { + return ErrorCode::OK; + })); + } + + std::unordered_map> initial = { + {key, {{old_value.data(), old_value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + initial, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + std::error_code symlink_ec; + fs::create_symlink(fs::absolute(legacy_path), legacy_path, symlink_ec); + ASSERT_FALSE(symlink_ec) << symlink_ec.message(); + + std::string new_value(64, 'n'); + std::unordered_map> replacement = { + {key, {{new_value.data(), new_value.size()}}}, + }; + bool replacement_reported = false; + auto replacement_result = adaptor.BatchOffload( + replacement, [&](const std::vector&, + std::vector&) { + replacement_reported = true; + return ErrorCode::OK; + }); + ASSERT_TRUE(replacement_result); + EXPECT_EQ(replacement_result.value(), 0); + EXPECT_FALSE(replacement_reported); + EXPECT_TRUE(fs::is_regular_file(canonical_path)); + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + ASSERT_TRUE(adaptor.BatchLoad(load)); + EXPECT_EQ(std::string(output.data(), output.size()), old_value); + + if (enable_eviction) { + std::vector notified_keys; + auto evict_result = adaptor.EvictAboveDiskWatermark( + /*high_watermark_ratio=*/0.0, + /*low_watermark_ratio=*/0.0, + [&](const std::vector& keys) + -> tl::expected { + notified_keys = keys; + return {}; + }); + ASSERT_TRUE(evict_result); + EXPECT_EQ(evict_result.value(), std::vector{key}); + EXPECT_EQ(notified_keys, std::vector{key}); + } else { + auto is_enabled = adaptor.IsEnableOffloading(); + ASSERT_FALSE(is_enabled); + EXPECT_EQ(is_enabled.error(), ErrorCode::INTERNAL_ERROR); + } + } +} + +TEST_F(StorageBackendTest, AdaptorScanMetaCleansCrashStagingFiles) { + for (const bool keep_canonical : {false, true}) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = keep_canonical + ? "file_per_key_staging_with_canonical" + : "file_per_key_staging_without_canonical"; + backend_config.enable_eviction = true; + + const std::string key = "staged-key"; + std::string value(64, 's'); + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + std::unordered_map> batch = { + {key, {{value.data(), value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, [](const std::vector&, + std::vector&) { + return ErrorCode::OK; + })); + } + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const fs::path staging_root = + fs::path(cfg.storage_filepath) / backend_config.fsdir / ".staging"; + fs::create_directories(staging_root); + const fs::path staged_path = staging_root / "crash-residue"; + if (keep_canonical) { + ASSERT_TRUE(fs::copy_file(canonical_path, staged_path)); + } else { + fs::rename(canonical_path, staged_path); + } + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + std::vector recovered_keys; + ASSERT_TRUE( + restart_adaptor.ScanMeta([&](const std::vector& keys, + std::vector&) { + recovered_keys.insert(recovered_keys.end(), keys.begin(), + keys.end()); + return ErrorCode::OK; + })); + + EXPECT_FALSE(fs::exists(staged_path)); + EXPECT_FALSE(fs::exists(staging_root)); + if (keep_canonical) { + EXPECT_EQ(recovered_keys, std::vector{key}); + std::vector notified_keys; + auto evict_result = restart_adaptor.EvictAboveDiskWatermark( + /*high_watermark_ratio=*/0.0, + /*low_watermark_ratio=*/0.0, + [&](const std::vector& keys) + -> tl::expected { + notified_keys = keys; + return {}; + }); + ASSERT_TRUE(evict_result); + EXPECT_EQ(evict_result.value(), std::vector{key}); + EXPECT_EQ(notified_keys, std::vector{key}); + } else { + EXPECT_TRUE(recovered_keys.empty()); + } + } +} + +TEST_F(StorageBackendTest, + StorageBackendInitCleansStagingBeforeQuotaReconstruction) { + const std::string fsdir = "staging_quota_reconstruction"; + const fs::path storage_root = fs::path(data_path) / fsdir; + const fs::path committed_path = storage_root / "aa" / "bb" / "committed"; + const fs::path staging_root = storage_root / ".staging"; + const fs::path staged_path = staging_root / "crash-residue"; + fs::create_directories(committed_path.parent_path()); + { + std::ofstream committed_file(committed_path, std::ios::binary); + const std::string contents(1024, 'c'); + committed_file.write(contents.data(), contents.size()); + } + + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + fs::create_directories(staging_root); + { + std::ofstream staged_file(staged_path, std::ios::binary); + const std::string contents(1024, 's'); + staged_file.write(contents.data(), contents.size()); + } + + StorageBackend backend(data_path, fsdir, /*enable_eviction=*/true); + ASSERT_TRUE(backend.Init(/*quota_bytes=*/1536)); + EXPECT_TRUE(fs::is_regular_file(committed_path)); + EXPECT_FALSE(fs::exists(staged_path)); + EXPECT_FALSE(fs::exists(staging_root)); +} + +TEST_F(StorageBackendTest, AdaptorScanMetaRemovesSupersededLegacyFile) { + for (const bool enable_eviction : {false, true}) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = enable_eviction ? "file_per_key_shadow_evict" + : "file_per_key_shadow_no_evict"; + backend_config.enable_eviction = enable_eviction; + + const std::string key = enable_eviction ? "evict-key" : "plain-key"; + std::string value(128, enable_eviction ? 'e' : 'p'); + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + if (!enable_eviction) { + ASSERT_TRUE( + adaptor.ScanMeta([](const std::vector&, + std::vector&) { + return ErrorCode::OK; + })); + } + std::unordered_map> batch = { + {key, {{value.data(), value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, [](const std::vector&, + std::vector&) { + return ErrorCode::OK; + })); + } + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + ASSERT_TRUE(fs::copy_file(canonical_path, legacy_path)); + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + std::vector recovered_keys; + ASSERT_TRUE( + restart_adaptor.ScanMeta([&](const std::vector& keys, + std::vector&) { + recovered_keys.insert(recovered_keys.end(), keys.begin(), + keys.end()); + return ErrorCode::OK; + })); + + EXPECT_EQ(recovered_keys, std::vector{key}); + EXPECT_TRUE(fs::exists(canonical_path)); + EXPECT_FALSE(fs::exists(legacy_path)); + + if (enable_eviction) { + std::vector notified_keys; + auto evict_result = restart_adaptor.EvictAboveDiskWatermark( + /*high_watermark_ratio=*/0.0, + /*low_watermark_ratio=*/0.0, + [&](const std::vector& keys) + -> tl::expected { + notified_keys = keys; + return {}; + }); + ASSERT_TRUE(evict_result); + EXPECT_EQ(evict_result.value(), std::vector{key}); + EXPECT_EQ(notified_keys, std::vector{key}); + } + } +} + +TEST_F(StorageBackendTest, AdaptorOnlineRewriteRetiresLegacyAccounting) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + cfg.total_keys_limit = 1; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_online_migration"; + backend_config.enable_eviction = false; + + const std::string key = "migrated-key"; + std::string old_value(64, 'o'); + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + std::unordered_map> batch = { + {key, {{old_value.data(), old_value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + } + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + fs::rename(canonical_path, legacy_path); + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + ASSERT_TRUE(restart_adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + ASSERT_TRUE(restart_adaptor.IsEnableOffloading().value_or(false)); + + std::string new_value(64, 'n'); + std::unordered_map> replacement = { + {key, {{new_value.data(), new_value.size()}}}, + }; + ASSERT_TRUE(restart_adaptor.BatchOffload( + replacement, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + EXPECT_TRUE(fs::exists(canonical_path)); + EXPECT_FALSE(fs::exists(legacy_path)); + EXPECT_TRUE(restart_adaptor.IsEnableOffloading().value_or(false)) + << "Rewriting a recovered key must not count it twice"; + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + ASSERT_TRUE(restart_adaptor.BatchLoad(load)); + EXPECT_EQ(std::string(output.data(), output.size()), new_value); +} + +TEST_F(StorageBackendTest, AdaptorOnlineRewriteTransfersEvictionAuthority) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_online_eviction_migration"; + backend_config.enable_eviction = true; + + const std::string key = "eviction-migrated-key"; + std::string old_value(64, 'o'); + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + std::unordered_map> batch = { + {key, {{old_value.data(), old_value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + } + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + fs::rename(canonical_path, legacy_path); + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + ASSERT_TRUE(restart_adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + std::string new_value(64, 'n'); + std::unordered_map> replacement = { + {key, {{new_value.data(), new_value.size()}}}, + }; + ASSERT_TRUE(restart_adaptor.BatchOffload( + replacement, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + EXPECT_TRUE(fs::exists(canonical_path)); + EXPECT_FALSE(fs::exists(legacy_path)); + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + ASSERT_TRUE(restart_adaptor.BatchLoad(load)); + EXPECT_EQ(std::string(output.data(), output.size()), new_value); + + std::vector notified_keys; + auto evict_result = restart_adaptor.EvictAboveDiskWatermark( + /*high_watermark_ratio=*/0.0, /*low_watermark_ratio=*/0.0, + [&](const std::vector& keys) + -> tl::expected { + notified_keys = keys; + return {}; + }); + ASSERT_TRUE(evict_result); + EXPECT_EQ(evict_result.value(), std::vector{key}); + EXPECT_EQ(notified_keys, std::vector{key}); +} + +TEST_F(StorageBackendTest, + AdaptorBatchLoadReturnsFileReadFailOnCanonicalStatusError) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_canonical_status_error"; + backend_config.enable_eviction = false; + + const std::string key = "canonical-status-key"; + std::string value(64, 'v'); + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + std::unordered_map> batch = { + {key, {{value.data(), value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + fs::rename(canonical_path, legacy_path); + + std::error_code symlink_ec; + fs::create_symlink(fs::absolute(canonical_path), canonical_path, + symlink_ec); + ASSERT_FALSE(symlink_ec) << symlink_ec.message(); + std::error_code status_ec; + EXPECT_FALSE(fs::exists(canonical_path, status_ec)); + EXPECT_EQ(status_ec, + std::make_error_code(std::errc::too_many_symbolic_link_levels)); + + auto exists_result = adaptor.IsExist(key); + EXPECT_FALSE(exists_result); + if (!exists_result) { + EXPECT_EQ(exists_result.error(), ErrorCode::FILE_READ_FAIL); + } + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + auto load_result = adaptor.BatchLoad(load); + EXPECT_FALSE(load_result); + if (!load_result) { + EXPECT_EQ(load_result.error(), ErrorCode::FILE_READ_FAIL); + } + + std::error_code cleanup_ec; + fs::remove(canonical_path, cleanup_ec); + EXPECT_FALSE(cleanup_ec) << cleanup_ec.message(); +} + +TEST_F(StorageBackendTest, + AdaptorBatchLoadReturnsFileReadFailOnLegacyStatusError) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_legacy_status_error"; + backend_config.enable_eviction = false; + + const std::string key = "legacy-status-key"; + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + std::error_code symlink_ec; + fs::create_symlink(fs::absolute(legacy_path), legacy_path, symlink_ec); + ASSERT_FALSE(symlink_ec) << symlink_ec.message(); + + std::array output{}; + std::unordered_map load = { + {key, {output.data(), output.size()}}, + }; + auto load_result = adaptor.BatchLoad(load); + EXPECT_FALSE(load_result); + if (!load_result) { + EXPECT_EQ(load_result.error(), ErrorCode::FILE_READ_FAIL); + } + + std::error_code cleanup_ec; + fs::remove(legacy_path, cleanup_ec); + EXPECT_FALSE(cleanup_ec) << cleanup_ec.message(); +} + +TEST_F(StorageBackendTest, + AdaptorScanMetaReturnsFileReadFailOnCanonicalStatusError) { + FileStorageConfig cfg; + cfg.storage_filepath = data_path; + FilePerKeyConfig backend_config; + backend_config.fsdir = "file_per_key_scan_status_error"; + backend_config.enable_eviction = false; + + const std::string key = "scan-status-key"; + std::string value(64, 's'); + { + StorageBackendAdaptor adaptor(cfg, backend_config); + ASSERT_TRUE(adaptor.Init()); + ASSERT_TRUE(adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + std::unordered_map> batch = { + {key, {{value.data(), value.size()}}}, + }; + ASSERT_TRUE(adaptor.BatchOffload( + batch, + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + } + + const auto canonical_path = ResolveFilePerKeyPathFromKey( + key, cfg.storage_filepath, backend_config.fsdir); + const auto legacy_path = ResolveLegacyPathFromKey(key, cfg.storage_filepath, + backend_config.fsdir); + fs::create_directories(fs::path(legacy_path).parent_path()); + fs::rename(canonical_path, legacy_path); + + StorageBackendAdaptor restart_adaptor(cfg, backend_config); + ASSERT_TRUE(restart_adaptor.Init()); + bool handler_called = false; + ASSERT_TRUE( + restart_adaptor.ScanMeta([&](const std::vector&, + std::vector&) { + handler_called = true; + return ErrorCode::OK; + })); + EXPECT_TRUE(handler_called); + auto initially_enabled = restart_adaptor.IsEnableOffloading(); + ASSERT_TRUE(initially_enabled); + EXPECT_TRUE(initially_enabled.value()); + + std::error_code symlink_ec; + fs::create_symlink(fs::absolute(canonical_path), canonical_path, + symlink_ec); + ASSERT_FALSE(symlink_ec) << symlink_ec.message(); + + handler_called = false; + auto scan_result = + restart_adaptor.ScanMeta([&](const std::vector&, + std::vector&) { + handler_called = true; + return ErrorCode::OK; + }); + EXPECT_FALSE(scan_result); + if (!scan_result) { + EXPECT_EQ(scan_result.error(), ErrorCode::FILE_READ_FAIL); + } + EXPECT_FALSE(handler_called); + EXPECT_TRUE(fs::is_regular_file(legacy_path)); + auto is_enabled = restart_adaptor.IsEnableOffloading(); + EXPECT_FALSE(is_enabled); + if (!is_enabled) { + EXPECT_EQ(is_enabled.error(), ErrorCode::INTERNAL_ERROR); + } + + std::error_code cleanup_ec; + fs::remove(canonical_path, cleanup_ec); + EXPECT_FALSE(cleanup_ec) << cleanup_ec.message(); +} + TEST_F(StorageBackendTest, AdaptorBatchOffload_PartialSuccess) { FileStorageConfig cfg; cfg.storage_filepath = data_path; @@ -953,8 +1694,7 @@ TEST_F(StorageBackendTest, AdaptorScanMetaAndIsEnableOffloading) { ASSERT_TRUE(enable_after_write); EXPECT_TRUE(enable_after_write.value()); - // Verify scan results via a fresh adaptor instance to avoid double-counting - // totals if ScanMeta is called again on the same adaptor. + // Verify restart recovery through a fresh adaptor instance. StorageBackendAdaptor restart_adaptor(cfg, file_per_key_config); ASSERT_TRUE(restart_adaptor.Init()); @@ -979,7 +1719,7 @@ TEST_F(StorageBackendTest, AdaptorScanMetaAndIsEnableOffloading) { FileStorageConfig strict_cfg = cfg; strict_cfg.total_keys_limit = 1; - strict_cfg.total_size_limit = 1; + strict_cfg.total_size_limit = cfg.total_size_limit; StorageBackendAdaptor strict_adaptor(strict_cfg, file_per_key_config); ASSERT_TRUE(strict_adaptor.Init()); @@ -992,6 +1732,19 @@ TEST_F(StorageBackendTest, AdaptorScanMetaAndIsEnableOffloading) { auto enable_strict = strict_adaptor.IsEnableOffloading(); ASSERT_TRUE(enable_strict); EXPECT_FALSE(enable_strict.value()); + + const auto removed_path = ResolveFilePerKeyPathFromKey( + "k1", cfg.storage_filepath, file_per_key_config.fsdir); + std::error_code remove_ec; + EXPECT_TRUE(fs::remove(removed_path, remove_ec)); + ASSERT_FALSE(remove_ec) << remove_ec.message(); + + ASSERT_TRUE(strict_adaptor.ScanMeta( + [](const std::vector&, + std::vector&) { return ErrorCode::OK; })); + auto enable_after_rescan = strict_adaptor.IsEnableOffloading(); + ASSERT_TRUE(enable_after_rescan); + EXPECT_TRUE(enable_after_rescan.value()); } TEST_F(StorageBackendTest, AdaptorScanMetaAndBatchLoadAcrossRestart) { @@ -2905,6 +3658,66 @@ TEST_F(StorageBackendTest, StoreObjectReturnsEvictedKeys) { EXPECT_EQ(evicted[0], "key_a"); } +TEST_F(StorageBackendTest, AtomicOverwriteAtQuotaPinsOldCanonicalUntilCommit) { + const std::string test_dir = data_path + "/atomic_tight_quota_test"; + std::filesystem::create_directories(test_dir); + + StorageBackend backend(test_dir, "", true); + ASSERT_TRUE(backend.Init(/*quota_bytes=*/2048)); + + const std::string canonical_path = test_dir + "/canonical"; + const std::string legacy_path = test_dir + "/legacy"; + const std::string victim_path = test_dir + "/victim"; + const std::string key = "target-key"; + const std::string old_data(1024, 'o'); + const std::string new_data(1024, 'n'); + ASSERT_TRUE(backend.StoreObject(canonical_path, old_data, key)); + ASSERT_TRUE(backend.StoreObject(victim_path, old_data, "victim-key")); + + std::vector notified_keys; + auto eviction_handler = [&](const std::vector& keys) + -> tl::expected { + notified_keys.insert(notified_keys.end(), keys.begin(), keys.end()); + return {}; + }; + auto failed_replacement = backend.StoreObjectAndRemoveFileIf( + canonical_path, std::span(new_data.data(), new_data.size()), + key, legacy_path, + [](const std::string&, + const std::string&) -> tl::expected { + return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); + }, + eviction_handler); + ASSERT_FALSE(failed_replacement); + EXPECT_EQ(failed_replacement.error(), ErrorCode::FILE_READ_FAIL); + EXPECT_EQ(notified_keys, std::vector{"victim-key"}); + EXPECT_FALSE(fs::exists(victim_path)); + + std::string loaded; + ASSERT_TRUE(backend.LoadObject(canonical_path, loaded, old_data.size())); + EXPECT_EQ(loaded, old_data); + + auto committed_replacement = backend.StoreObjectAndRemoveFileIf( + canonical_path, std::span(new_data.data(), new_data.size()), + key, legacy_path, + [](const std::string&, const std::string&) + -> tl::expected { return false; }, + eviction_handler); + ASSERT_TRUE(committed_replacement); + EXPECT_FALSE(committed_replacement.value()); + EXPECT_EQ(notified_keys, std::vector{"victim-key"}); + + ASSERT_TRUE(backend.LoadObject(canonical_path, loaded, new_data.size())); + EXPECT_EQ(loaded, new_data); + + auto eviction_result = backend.EvictAboveDiskWatermark( + /*high_watermark_ratio=*/0.0, /*low_watermark_ratio=*/0.0, + eviction_handler); + ASSERT_TRUE(eviction_result); + EXPECT_EQ(eviction_result.value(), std::vector{key}); + EXPECT_EQ(notified_keys, (std::vector{"victim-key", key})); +} + TEST_F(StorageBackendTest, StoreObjectEvictionWithEmptyKey) { std::string test_dir = data_path + "/evict_empty_key_test"; std::filesystem::create_directories(test_dir); diff --git a/mooncake-store/tests/utils_test.cpp b/mooncake-store/tests/utils_test.cpp index aa422f1522..e529e37e7b 100644 --- a/mooncake-store/tests/utils_test.cpp +++ b/mooncake-store/tests/utils_test.cpp @@ -10,19 +10,31 @@ using namespace mooncake; -TEST(UtilsTest, ResolvePathFromKeyUsesBoundedStableDigest) { +TEST(UtilsTest, ResolveFilePerKeyPathFromKeyUsesBoundedStableDigest) { const std::string root = "/tmp/mooncake"; const std::string fsdir = "file_per_key"; const std::string key(4096, '/'); - const auto first = ResolvePathFromKey(key, root, fsdir); - const auto second = ResolvePathFromKey(key, root, fsdir); + const auto first = ResolveFilePerKeyPathFromKey(key, root, fsdir); + const auto second = ResolveFilePerKeyPathFromKey(key, root, fsdir); EXPECT_EQ(first, second); EXPECT_EQ(std::filesystem::path(first).filename().string().size(), 32); - EXPECT_NE(first, ResolvePathFromKey(key + "x", root, fsdir)); + EXPECT_NE(first, ResolveFilePerKeyPathFromKey(key + "x", root, fsdir)); EXPECT_GT(ResolveLegacyPathFromKey(key, root, fsdir).size(), first.size()); } +TEST(UtilsTest, ResolvePathFromKeyPreservesLegacyLayout) { + const std::string root = "/tmp/mooncake"; + const std::string fsdir = "mooncake_cluster"; + const std::string key = "legacy/disk:key"; + + const auto path = ResolvePathFromKey(key, root, fsdir); + EXPECT_EQ(path, ResolveLegacyPathFromKey(key, root, fsdir)); + EXPECT_EQ(std::filesystem::path(path).filename().string(), + "legacy_disk_key"); + EXPECT_NE(path, ResolveFilePerKeyPathFromKey(key, root, fsdir)); +} + TEST(UtilsTest, ByteSizeToString) { EXPECT_EQ(byte_size_to_string(999), "999 B"); EXPECT_EQ(byte_size_to_string(2048), "2.00 KB");