mooncake-track
/
Mooncake
0
0
Fork 12
Code Issues Pull Requests 13 Actions Packages Projects Releases Wiki Activity

[CCF Archive] Store object type eviction policy submission #3

Closed
kancel wants to merge 382 commits from kancel:ccf-archive-pr2746 into main
pull from: kancel:ccf-archive-pr2746
merge into: mooncake-track:main
Conversation 0 Commits 382 Files Changed 654 +263 -42
4 changed files with 263 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a1495cce80 - Show all commits

39
mooncake-store/src/ha/snapshot/catalog_backed_snapshot_provider.cpp
View File

@ -131,14 +131,39 @@ DeserializeStandbyObjectMetadata(
const auto soft_pin_timestamp_ms = array[index++].as<uint64_t>();
const auto replica_count = array[index++].as<uint32_t>();
if (object.via.array.size != 7 + replica_count &&
object.via.array.size != 8 + replica_count) {
// Optional fields are decoded by type for backward/forward
// compatibility with MasterService::MetadataSerializer, which appends
// them over time:
// data_type (positive int) appears before the replicas;
// hard_pinned (bool) and group_id (str) trail them.
// v1: 7 + replica_count, no optional fields
// v2: 8 + replica_count, either data_type or hard_pinned
// v3: 9 + replica_count, data_type + hard_pinned or
// hard_pinned + group_id
// v4: 10 + replica_count, data_type + hard_pinned + group_id
// 64-bit arithmetic keeps an attacker-controlled near-UINT32_MAX
// replica_count from wrapping the bounds and slipping an out-of-bounds
// index through.
constexpr uint64_t kBaseFieldCount = 7;
constexpr uint64_t kMaxOptionalFieldCount = 3;
const uint64_t total_elements = object.via.array.size;
const uint64_t min_elements = kBaseFieldCount + replica_count;
if (total_elements < min_elements ||
total_elements > min_elements + kMaxOptionalFieldCount) {
LOG(ERROR) << "Snapshot metadata entry replica count mismatch, "
<< "replicas=" << replica_count
<< ", total_fields=" << object.via.array.size;
<< ", total_fields=" << total_elements;
return tl::make_unexpected(ErrorCode::DESERIALIZE_FAIL);
}
// Skip the optional data_type; the standby restore path does not use
// it. A leading positive integer is data_type, whereas a replica is
// serialized as an array.
if (index < total_elements &&
array[index].type == msgpack::type::POSITIVE_INTEGER) {
++index; // data_type
}
const auto lease_timeout = std::chrono::system_clock::time_point(
std::chrono::milliseconds(lease_timestamp_ms));
std::optional<std::chrono::system_clock::time_point> soft_pin_timeout;
@ -156,6 +181,14 @@ DeserializeStandbyObjectMetadata(
std::vector<Replica::Descriptor> replicas;
replicas.reserve(replica_count);
for (uint32_t i = 0; i < replica_count; ++i) {
// Defensive bound: a corrupt entry whose first post-count field
// looks like a data_type could otherwise read past the array.
if (index >= total_elements) {
LOG(ERROR) << "Snapshot metadata entry truncated, "
<< "replicas=" << replica_count
<< ", total_fields=" << total_elements;
return tl::make_unexpected(ErrorCode::DESERIALIZE_FAIL);
}
auto replica_result =
Serializer<Replica>::deserialize(array[index++], segment_view);
if (!replica_result) {

22
mooncake-store/src/master_service.cpp
View File

@ -6472,15 +6472,18 @@ MasterService::MetadataSerializer::DeserializeMetadata(
// Deserialize replicas count
uint32_t replicas_count = array[index++].as<uint32_t>();
// Format detection:
// Format detection (decode optional fields by type for back-compat):
// v1: 7 + replicas_count, no optional fields
// v2: 8 + replicas_count, either data_type or hard_pinned
// v3: 9 + replicas_count, data_type + hard_pinned or hard_pinned +
// group_id v4: 10 + replicas_count, data_type + hard_pinned + group_id
constexpr uint32_t kBaseFieldCount = 7;
constexpr uint32_t kMaxOptionalFieldCount = 3;
const uint32_t total_elements = obj.via.array.size;
const uint32_t min_elements = kBaseFieldCount + replicas_count;
// 64-bit arithmetic keeps an attacker-controlled near-UINT32_MAX
// replicas_count from wrapping the bounds and slipping an out-of-bounds
// index past the size check.
constexpr uint64_t kBaseFieldCount = 7;
constexpr uint64_t kMaxOptionalFieldCount = 3;
const uint64_t total_elements = obj.via.array.size;
const uint64_t min_elements = kBaseFieldCount + replicas_count;
if (total_elements < min_elements ||
total_elements > min_elements + kMaxOptionalFieldCount) {
return tl::unexpected(SerializationError(
@ -6499,6 +6502,15 @@ MasterService::MetadataSerializer::DeserializeMetadata(
replicas.reserve(replicas_count);
for (uint32_t i = 0; i < replicas_count; i++) {
// Defensive bound: the data_type skip above can consume a slot the
// size check counted on, so a crafted entry whose first post-count
// field looks like a data_type could otherwise read past the array.
// Mirrors the standby reader in catalog_backed_snapshot_provider.cpp.
if (index >= total_elements) {
return tl::unexpected(
SerializationError(ErrorCode::DESERIALIZE_FAIL,
"deserialize ObjectMetadata truncated"));
}
auto result = Serializer<Replica>::deserialize(
array[index++], service_->segment_manager_.getView());
if (!result) {

89
mooncake-store/tests/ha/snapshot/catalog_backed_snapshot_provider_test.cpp
View File

@ -2,7 +2,9 @@
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <cstdint>
#include <filesystem>
#include <limits>
#include <memory>
#include <optional>
#include <string>
@ -57,13 +59,42 @@ class CatalogBackedSnapshotProviderTest
}
}
void PublishSnapshotPayload() {
void PublishSnapshotPayload(
SnapshotMetadataFormat format = SnapshotMetadataFormat::kLegacy) {
auto result = mooncake::test::PublishSnapshotPayload(
*object_store_, *catalog_store_, descriptor_);
*object_store_, *catalog_store_, descriptor_, UUID{1, 2},
kDefaultTestObjectKey, kDefaultTestDiskFilePath,
kDefaultTestObjectSize, format);
ASSERT_TRUE(result.has_value()) << result.error();
snapshot_published_ = true;
}
// Loads the published snapshot and asserts the single default object
// round-trips intact, regardless of the metadata on-wire format.
void ExpectLoadsDefaultObject() {
auto provider = CreateProvider();
ASSERT_TRUE(provider.has_value()) << toString(provider.error());
auto snapshot = provider.value()->LoadLatestSnapshot(cluster_id_);
ASSERT_TRUE(snapshot.has_value()) << toString(snapshot.error());
ASSERT_TRUE(snapshot->has_value());
ASSERT_EQ(snapshot->value().metadata.size(), 1u);
const auto& [key, metadata] = snapshot->value().metadata.front();
EXPECT_EQ(key, kDefaultTestObjectKey);
EXPECT_EQ(metadata.client_id, (UUID{1, 2}));
EXPECT_EQ(metadata.size, kDefaultTestObjectSize);
ASSERT_EQ(metadata.replicas.size(), 1u);
const auto& replica = metadata.replicas.front();
EXPECT_EQ(replica.status, ReplicaStatus::COMPLETE);
ASSERT_TRUE(replica.is_disk_replica());
EXPECT_EQ(replica.get_disk_descriptor().file_path,
kDefaultTestDiskFilePath);
EXPECT_EQ(replica.get_disk_descriptor().object_size,
kDefaultTestObjectSize);
}
tl::expected<std::unique_ptr<SnapshotProvider>, ErrorCode> CreateProvider()
const {
return CreateCatalogBackedSnapshotProvider(MakeSnapshotProviderConfig(
@ -119,6 +150,60 @@ TEST_P(CatalogBackedSnapshotProviderTest, LoadLatestSnapshotRoundTrip) {
kDefaultTestObjectSize);
}
// The master snapshot writer evolved its per-object metadata layout over time
// (data_type field, trailing hard_pinned flag, trailing group_id). The standby
// restore reader must accept every shape; otherwise it rejects the snapshot and
// falls back to OpLog-only bootstrap. These tests pin each on-wire format.
TEST_P(CatalogBackedSnapshotProviderTest, LoadLatestSnapshotWithDataTypeField) {
// 8 + replica_count: data_type packed right after replica_count.
PublishSnapshotPayload(SnapshotMetadataFormat::kDataTypeOnly);
ExpectLoadsDefaultObject();
}
TEST_P(CatalogBackedSnapshotProviderTest,
LoadLatestSnapshotWithHardPinnedField) {
// 8 + replica_count: trailing hard_pinned flag, no data_type. Exercises the
// type-based disambiguation (first replica is not a positive integer).
PublishSnapshotPayload(SnapshotMetadataFormat::kHardPinnedOnly);
ExpectLoadsDefaultObject();
}
TEST_P(CatalogBackedSnapshotProviderTest,
LoadLatestSnapshotWithDataTypeAndHardPinned) {
// 9 + replica_count: data_type + trailing hard_pinned.
PublishSnapshotPayload(SnapshotMetadataFormat::kDataTypeAndHardPinned);
ExpectLoadsDefaultObject();
}
TEST_P(CatalogBackedSnapshotProviderTest, LoadLatestSnapshotWithGroupId) {
// 10 + replica_count: current writer format (data_type + hard_pinned +
// trailing group_id). Regression test for the live snapshot restore
// failure against the latest metadata layout.
PublishSnapshotPayload(SnapshotMetadataFormat::kWithGroupId);
ExpectLoadsDefaultObject();
}
TEST_P(CatalogBackedSnapshotProviderTest, RejectsOverflowingReplicaCount) {
// A near-UINT32_MAX replica_count must not wrap the format-detection
// arithmetic into a valid-looking total and slip an out-of-bounds index
// past the size check. The entry packs zero replicas (array size 7) but
// declares UINT32_MAX, so every format must be rejected, not parsed.
auto published = mooncake::test::PublishSnapshotPayloadBytes(
*object_store_, *catalog_store_, descriptor_,
BuildMetadataPayloadWithDeclaredReplicaCount(
std::numeric_limits<uint32_t>::max()));
ASSERT_TRUE(published.has_value()) << published.error();
snapshot_published_ = true;
auto provider = CreateProvider();
ASSERT_TRUE(provider.has_value()) << toString(provider.error());
auto snapshot = provider.value()->LoadLatestSnapshot(cluster_id_);
ASSERT_FALSE(snapshot.has_value());
EXPECT_EQ(snapshot.error(), ErrorCode::DESERIALIZE_FAIL);
}
TEST_P(CatalogBackedSnapshotProviderTest, RejectsClusterMismatch) {
PublishSnapshotPayload();

155
mooncake-store/tests/ha/snapshot/snapshot_test_utils.h
View File

@ -44,6 +44,24 @@ struct CatalogBackendParam {
bool requires_redis{false};
};
// On-wire shapes the master snapshot writer has emitted over time. The standby
// restore reader must tolerate all of them. See
// MasterService::MetadataSerializer in master_service.cpp.
// kLegacy: 7 + replica_count, no data_type, no trailing hard_pinned
// kDataTypeOnly: 8 + replica_count, data_type after replica_count
// kHardPinnedOnly: 8 + replica_count, trailing hard_pinned, no data_type
// kDataTypeAndHardPinned:
// 9 + replica_count, data_type plus trailing hard_pinned
// kWithGroupId: 10 + replica_count, data_type + hard_pinned + group_id
// (the current writer format)
enum class SnapshotMetadataFormat {
kLegacy,
kDataTypeOnly,
kHardPinnedOnly,
kDataTypeAndHardPinned,
kWithGroupId,
};
class ScopedEnvVar {
public:
ScopedEnvVar(std::string name, std::string value) : name_(std::move(name)) {
@ -128,30 +146,10 @@ inline std::vector<uint8_t> BuildSegmentsPayload() {
return serialized.value();
}
inline std::vector<uint8_t> BuildMetadataPayload(
const UUID& client_id, std::string_view object_key = kDefaultTestObjectKey,
std::string_view disk_file_path = kDefaultTestDiskFilePath,
uint64_t object_size = kDefaultTestObjectSize,
uint64_t put_start_time_ms = kDefaultTestPutStartTimeMs,
uint64_t lease_timeout_ms = kDefaultTestLeaseTimeoutMs) {
msgpack::sbuffer shard_buffer;
MsgpackPacker shard_packer(&shard_buffer);
shard_packer.pack_map(1);
shard_packer.pack(std::string("metadata"));
shard_packer.pack_array(1);
shard_packer.pack_array(2);
shard_packer.pack(std::string(object_key));
shard_packer.pack_array(8);
shard_packer.pack(UuidToString(client_id));
shard_packer.pack(put_start_time_ms);
shard_packer.pack(object_size);
shard_packer.pack(lease_timeout_ms);
shard_packer.pack(false);
shard_packer.pack(uint64_t{0});
shard_packer.pack(uint32_t{1});
PackDiskReplica(shard_packer, disk_file_path, object_size);
// Compresses a packed shard buffer and wraps it in the {"shards": {"0": ...}}
// root map that the snapshot metadata payload expects.
inline std::vector<uint8_t> WrapShardIntoMetadataRoot(
const msgpack::sbuffer& shard_buffer) {
auto compressed_shard =
zstd_compress(reinterpret_cast<const uint8_t*>(shard_buffer.data()),
shard_buffer.size(), 3);
@ -169,6 +167,87 @@ inline std::vector<uint8_t> BuildMetadataPayload(
return ToByteVector(root_buffer);
}
inline std::vector<uint8_t> BuildMetadataPayload(
const UUID& client_id, std::string_view object_key = kDefaultTestObjectKey,
std::string_view disk_file_path = kDefaultTestDiskFilePath,
uint64_t object_size = kDefaultTestObjectSize,
uint64_t put_start_time_ms = kDefaultTestPutStartTimeMs,
uint64_t lease_timeout_ms = kDefaultTestLeaseTimeoutMs,
SnapshotMetadataFormat format = SnapshotMetadataFormat::kLegacy) {
const bool include_data_type =
format == SnapshotMetadataFormat::kDataTypeOnly ||
format == SnapshotMetadataFormat::kDataTypeAndHardPinned ||
format == SnapshotMetadataFormat::kWithGroupId;
const bool include_hard_pinned =
format == SnapshotMetadataFormat::kHardPinnedOnly ||
format == SnapshotMetadataFormat::kDataTypeAndHardPinned ||
format == SnapshotMetadataFormat::kWithGroupId;
const bool include_group_id =
format == SnapshotMetadataFormat::kWithGroupId;
constexpr uint32_t kReplicaCount = 1;
// 7 leading fields + replicas + optional data_type/hard_pinned/group_id.
const size_t array_size = 7 + kReplicaCount + (include_data_type ? 1 : 0) +
(include_hard_pinned ? 1 : 0) +
(include_group_id ? 1 : 0);
msgpack::sbuffer shard_buffer;
MsgpackPacker shard_packer(&shard_buffer);
shard_packer.pack_map(1);
shard_packer.pack(std::string("metadata"));
shard_packer.pack_array(1);
shard_packer.pack_array(2);
shard_packer.pack(std::string(object_key));
shard_packer.pack_array(array_size);
shard_packer.pack(UuidToString(client_id));
shard_packer.pack(put_start_time_ms);
shard_packer.pack(object_size);
shard_packer.pack(lease_timeout_ms);
shard_packer.pack(false);
shard_packer.pack(uint64_t{0});
shard_packer.pack(kReplicaCount);
if (include_data_type) {
shard_packer.pack(static_cast<uint8_t>(ObjectDataType::TENSOR));
}
PackDiskReplica(shard_packer, disk_file_path, object_size);
if (include_hard_pinned) {
shard_packer.pack(true);
}
if (include_group_id) {
shard_packer.pack(std::string("test-group"));
}
return WrapShardIntoMetadataRoot(shard_buffer);
}
// Builds a metadata payload whose declared replica_count field is set to
// `declared_replica_count` while no replicas are actually packed (the entry
// array stays at the 7 leading fields). Used to verify the deserializer
// rejects a hostile count instead of overflowing into an out-of-bounds read.
inline std::vector<uint8_t> BuildMetadataPayloadWithDeclaredReplicaCount(
uint32_t declared_replica_count,
std::string_view object_key = kDefaultTestObjectKey) {
msgpack::sbuffer shard_buffer;
MsgpackPacker shard_packer(&shard_buffer);
shard_packer.pack_map(1);
shard_packer.pack(std::string("metadata"));
shard_packer.pack_array(1);
shard_packer.pack_array(2);
shard_packer.pack(std::string(object_key));
// Exactly the 7 leading fields, zero trailing replicas.
shard_packer.pack_array(7);
shard_packer.pack(UuidToString(UUID{1, 2}));
shard_packer.pack(kDefaultTestPutStartTimeMs);
shard_packer.pack(kDefaultTestObjectSize);
shard_packer.pack(kDefaultTestLeaseTimeoutMs);
shard_packer.pack(false);
shard_packer.pack(uint64_t{0});
shard_packer.pack(declared_replica_count);
return WrapShardIntoMetadataRoot(shard_buffer);
}
inline ha::SnapshotDescriptor MakeTestSnapshotDescriptor(
std::string_view snapshot_id = kDefaultTestSnapshotId,
uint64_t last_included_seq = kDefaultTestSnapshotSeq,
@ -216,13 +295,12 @@ inline std::unique_ptr<ha::SnapshotCatalogStore> CreateCatalogStoreForTest(
return nullptr;
}
inline tl::expected<void, std::string> PublishSnapshotPayload(
// Uploads the manifest/segments/metadata objects (using the caller-provided
// metadata payload) and publishes the descriptor.
inline tl::expected<void, std::string> PublishSnapshotPayloadBytes(
SnapshotObjectStore& object_store, ha::SnapshotCatalogStore& catalog_store,
const ha::SnapshotDescriptor& descriptor,
const UUID& client_id = UUID{1, 2},
std::string_view object_key = kDefaultTestObjectKey,
std::string_view disk_file_path = kDefaultTestDiskFilePath,
uint64_t object_size = kDefaultTestObjectSize) {
const std::vector<uint8_t>& metadata_payload) {
auto manifest = object_store.UploadString(descriptor.manifest_key,
"messagepack|1.0.0|standby-test");
if (!manifest) {
@ -236,9 +314,7 @@ inline tl::expected<void, std::string> PublishSnapshotPayload(
}
auto metadata = object_store.UploadBuffer(
descriptor.object_prefix + "metadata",
BuildMetadataPayload(client_id, object_key, disk_file_path,
object_size));
descriptor.object_prefix + "metadata", metadata_payload);
if (!metadata) {
return tl::make_unexpected(metadata.error());
}
@ -251,4 +327,19 @@ inline tl::expected<void, std::string> PublishSnapshotPayload(
return {};
}
inline tl::expected<void, std::string> PublishSnapshotPayload(
SnapshotObjectStore& object_store, ha::SnapshotCatalogStore& catalog_store,
const ha::SnapshotDescriptor& descriptor,
const UUID& client_id = UUID{1, 2},
std::string_view object_key = kDefaultTestObjectKey,
std::string_view disk_file_path = kDefaultTestDiskFilePath,
uint64_t object_size = kDefaultTestObjectSize,
SnapshotMetadataFormat format = SnapshotMetadataFormat::kLegacy) {
return PublishSnapshotPayloadBytes(
object_store, catalog_store, descriptor,
BuildMetadataPayload(client_id, object_key, disk_file_path, object_size,
kDefaultTestPutStartTimeMs,
kDefaultTestLeaseTimeoutMs, format));
}
} // namespace mooncake::test