diff --git a/include/lavik/storage/detail/grouped/collection.h b/include/lavik/storage/detail/grouped/collection.h index 5e4ee88d5..2bacc55cf 100644 --- a/include/lavik/storage/detail/grouped/collection.h +++ b/include/lavik/storage/detail/grouped/collection.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -90,6 +91,34 @@ struct OrderedCollectionEntryView { double score_ = 0; }; +struct OrderedGroupSnapshot; +struct OrderedGroupSplit; + +// An owned, immutable member sequence certified by the checked Sorted Set +// splitter. Moving the original vector preserves its allocations. Only const +// access escapes: changing an after-image requires a new mutable snapshot and +// full validation, not a flag that can outlive edits to public entries_. +class CheckedSortedSetEntries { + public: + CheckedSortedSetEntries(const CheckedSortedSetEntries&) = default; + CheckedSortedSetEntries& operator=(const CheckedSortedSetEntries&) = default; + CheckedSortedSetEntries(CheckedSortedSetEntries&&) noexcept = default; + CheckedSortedSetEntries& operator=(CheckedSortedSetEntries&&) noexcept = + default; + std::span entries() const noexcept { + return entries_; + } + + private: + friend absl::StatusOr SplitCheckedSortedSetGroup( + OrderedGroupSnapshot group, std::uint64_t next_group_id, + std::size_t target_bytes); + explicit CheckedSortedSetEntries(std::vector entries) + : entries_(std::move(entries)) {} + + std::vector entries_; +}; + struct OrderedCollectionRoot { OrderedCollectionKind kind_ = OrderedCollectionKind::kList; std::uint64_t incarnation_ = 0; @@ -133,8 +162,20 @@ struct OrderedGroupSnapshot { // page writes, including suspension; decoding always produces owned entries. // Routing directories and asynchronous commit receipts never retain it. std::optional borrowed_string_ = std::nullopt; + // Final Sorted Set after-images may own checked immutable entries instead + // of entries_. Link/identity changes still undergo ordinary encoder checks; + // member edits and rebalance must finish before this representation exists. + std::optional checked_sorted_set_ = std::nullopt; }; +// Read either mutable-plan entries or a final checked Sorted Set after-image. +// The returned span borrows the snapshot and cannot modify its members. +inline std::span OrderedGroupEntries( + const OrderedGroupSnapshot& group) noexcept { + return group.checked_sorted_set_ ? group.checked_sorted_set_->entries() + : std::span(group.entries_); +} + // Returns a String segment's raw bytes from its owned or borrowed form. The // caller must keep that form's owner alive and unchanged while using the view. inline std::string_view OrderedStringBytes( @@ -151,7 +192,7 @@ inline std::size_t OrderedGroupSize( const OrderedGroupSnapshot& group) noexcept { return group.kind_ == OrderedCollectionKind::kString ? OrderedStringBytes(group).size() - : group.entries_.size(); + : OrderedGroupEntries(group).size(); } inline constexpr std::size_t kOrderedGroupHeaderBytes = 64; @@ -599,6 +640,16 @@ absl::StatusOr SplitOrderedGroup( OrderedGroupSnapshot group, std::uint64_t next_group_id, std::size_t target_bytes = kCollectionGroupTargetBytes); +// Fully validates a mutable Sorted Set after-image before partitioning, then +// freezes each resulting member sequence. Subsequent encoders can reuse that +// uniqueness proof without another member hash set; identity, framing, score +// and ordering checks remain. No member edits or rebalance may follow, though +// callers may finish page links before publication. Other kinds, already +// frozen pages and all default codec entry points keep their ordinary paths. +absl::StatusOr SplitCheckedSortedSetGroup( + OrderedGroupSnapshot group, std::uint64_t next_group_id, + std::size_t target_bytes = kCollectionGroupTargetBytes); + struct LoadedOrderedGroup { std::uint64_t sequence_ = 0; OrderedGroupSnapshot snapshot_; diff --git a/src/storage/engine/grouped/collection.cpp b/src/storage/engine/grouped/collection.cpp index 834b43a2b..2dc3f1603 100644 --- a/src/storage/engine/grouped/collection.cpp +++ b/src/storage/engine/grouped/collection.cpp @@ -143,19 +143,30 @@ absl::StatusOr ValidateGroup( SortedSetMemberValidation validation = SortedSetMemberValidation::kVerifyUnique) { const bool borrowed_string = group.borrowed_string_.has_value(); + if (group.checked_sorted_set_ && + (group.kind_ != OrderedCollectionKind::kSortedSet || + !group.entries_.empty() || borrowed_string)) + return absl::InvalidArgumentError("invalid checked Sorted Set entries"); + const auto entries = OrderedGroupEntries(group); if (borrowed_string && (group.kind_ != OrderedCollectionKind::kString || !group.entries_.empty())) return absl::InvalidArgumentError("invalid borrowed String segment"); if (group.incarnation_ == 0 || group.id_ == 0 || group.previous_ == group.id_ || group.next_ == group.id_ || (group.previous_ != 0 && group.previous_ == group.next_) || - group.entries_.size() > std::numeric_limits::max() || - (group.retired_ ? (!group.entries_.empty() || borrowed_string || + entries.size() > std::numeric_limits::max() || + (group.retired_ ? (!entries.empty() || borrowed_string || group.previous_ != 0 || group.next_ != 0) - : (group.entries_.empty() && !borrowed_string))) { + : (entries.empty() && !borrowed_string))) { return absl::InvalidArgumentError("invalid ordered page identity or state"); } - auto status = ValidateEntries(group.kind_, group.entries_, validation); + // Only the checked splitter can create this immutable owner, after full + // after-image validation. It cannot become stale through a member edit. + // All structural and order checks still run after callers finish links. + auto status = ValidateEntries(group.kind_, entries, + group.checked_sorted_set_ + ? SortedSetMemberValidation::kTrustStored + : validation); if (!status.ok()) return status; if (group.kind_ == OrderedCollectionKind::kString) { const auto size = OrderedGroupSize(group); @@ -168,7 +179,7 @@ absl::StatusOr ValidateGroup( return kOrderedGroupHeaderBytes + size; } std::size_t bytes = kOrderedGroupHeaderBytes; - for (const auto& entry : group.entries_) { + for (const auto& entry : entries) { if (bytes > kMaxRecordPayloadBytes - kEntryHeaderBytes || entry.value_.size() > kMaxRecordPayloadBytes - kEntryHeaderBytes - bytes) { @@ -417,8 +428,9 @@ std::optional OrderedGroupEncoder::Next() noexcept { ++entry_; return OrderedStringBytes(*group_); } - if (entry_ == group_->entries_.size()) return std::nullopt; - const auto& entry = group_->entries_[entry_]; + const auto entries = OrderedGroupEntries(*group_); + if (entry_ == entries.size()) return std::nullopt; + const auto& entry = entries[entry_]; if (phase_ == 1) { phase_ = 2; Store(entry_header_, 0, entry.value_.size(), 4); @@ -726,8 +738,9 @@ absl::Status ValidateOrderedGroupBoundary(const OrderedGroupSnapshot& left, return absl::DataLossError("inconsistent ordered page neighbours"); } if (left.kind_ == OrderedCollectionKind::kString) return absl::OkStatus(); - return ValidateOrderedEntryBoundary(left.kind_, left.entries_.back(), - right.entries_.front()); + return ValidateOrderedEntryBoundary(left.kind_, + OrderedGroupEntries(left).back(), + OrderedGroupEntries(right).front()); } absl::StatusOr OrderedGroupDirectory::Recover( @@ -1631,6 +1644,7 @@ absl::StatusOr RebalanceSortedSetGroupPair(OrderedGroupSnapshot& left, std::size_t target_bytes) { if (left.kind_ != OrderedCollectionKind::kSortedSet || right.kind_ != left.kind_ || left.incarnation_ == 0 || + left.checked_sorted_set_ || right.checked_sorted_set_ || left.incarnation_ != right.incarnation_ || left.id_ == 0 || right.id_ == 0 || left.id_ == right.id_ || left.retired_ || right.retired_ || left.next_ != right.id_ || @@ -1720,7 +1734,7 @@ absl::StatusOr SplitOrderedGroupImpl( if (target_bytes <= kOrderedGroupHeaderBytes || target_bytes > kMaxRecordPayloadBytes || next_group_id <= std::max({group.id_, group.previous_, group.next_}) || - group.retired_ || group.entries_.empty()) { + group.retired_ || group.checked_sorted_set_ || group.entries_.empty()) { return absl::InvalidArgumentError("invalid ordered split input"); } // A promotion can exceed one record's aggregate limit. Validate individual @@ -1816,6 +1830,26 @@ absl::StatusOr SplitOrderedGroup(OrderedGroupSnapshot group, false); } +absl::StatusOr SplitCheckedSortedSetGroup( + OrderedGroupSnapshot group, std::uint64_t next_group_id, + std::size_t target_bytes) { + if (group.kind_ != OrderedCollectionKind::kSortedSet || + group.borrowed_string_) + return absl::InvalidArgumentError("checked split requires a Sorted Set"); + auto split = SplitOrderedGroupImpl(std::move(group), next_group_id, + target_bytes, false); + if (!split.ok()) return split.status(); + for (auto& page : split->groups_) { + // Splitting only partitions the fully validated input. Freezing the + // ownership here also covers duplicates that would land in separate + // outputs: validation happened before choosing page boundaries. + page.checked_sorted_set_ = + CheckedSortedSetEntries(std::move(page.entries_)); + page.entries_.clear(); + } + return split; +} + absl::StatusOr PlanOrderedCollectionSplice( const OrderedGroupDirectory& directory, std::vector loaded_groups, std::uint64_t rank, @@ -1854,8 +1888,8 @@ absl::StatusOr PlanOrderedCollectionSplice( const auto* found = directory.Find(page.id_); if (found == nullptr || candidate.sequence_ != found->sequence_ || page.incarnation_ != root.incarnation_ || page.kind_ != root.kind_ || - page.retired_ || page.previous_ != found->previous_ || - page.next_ != found->next_ || + page.retired_ || page.checked_sorted_set_ || + page.previous_ != found->previous_ || page.next_ != found->next_ || page.entries_.size() != found->item_count_) { return absl::AbortedError("ordered splice loaded a stale page"); } diff --git a/src/storage/engine/grouped/hash.cpp b/src/storage/engine/grouped/hash.cpp index 504e78cf8..c7dedf0ed 100644 --- a/src/storage/engine/grouped/hash.cpp +++ b/src/storage/engine/grouped/hash.cpp @@ -371,9 +371,10 @@ absl::StatusOr> HashGroupPointProbe::Create( if (payload.size() - kGroupHeaderBytes > kCollectionGroupTargetBytes || metadata->field_count_ == 0) return std::nullopt; - // An absent member takes the checked decoded fallback. Locate first without - // hashing or allocating a duplicate set, so insertion does not validate the - // same leaf twice. A hit still checks every field before certifying bytes. + // An absent member leaves uniqueness validation to the caller's checked + // splice or decoded fallback. Locate first without hashing or allocating a + // duplicate set, so insertion does not repeat that validation. A hit still + // checks every field before certifying bytes. auto reader = HashValueReader::Open(payload.substr(kGroupHeaderBytes)); if (!reader.ok()) return absl::DataLossError(reader.status().message()); if (reader->size() != metadata->field_count_) diff --git a/src/storage/engine/grouped/ordered_mutation.cpp b/src/storage/engine/grouped/ordered_mutation.cpp index fbc45915f..3dc9b8ad2 100644 --- a/src/storage/engine/grouped/ordered_mutation.cpp +++ b/src/storage/engine/grouped/ordered_mutation.cpp @@ -298,9 +298,10 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( if (page.kind_ == OrderedCollectionKind::kString) compact.push_back( {.value_ = std::string(OrderedStringBytes(page))}); - else - compact.insert(compact.end(), page.entries_.begin(), - page.entries_.end()); + else { + const auto entries = OrderedGroupEntries(page); + compact.insert(compact.end(), entries.begin(), entries.end()); + } id = page.next_; } else { const auto* route = previous->ordered_directory().Find(id); @@ -450,6 +451,7 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( continue; } const auto& page = plan.writes_[i]; + const auto entries = OrderedGroupEntries(page); RecoveredOrderedGroup candidate{ {.incarnation_ = plan.root_.incarnation_, .id_ = page.id_, @@ -461,8 +463,8 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( .encoded_bytes_ = ordered_sizes[i], .record_token_ = page.id_, .retired_ = group.retired_, - .min_score_ = page.entries_.empty() ? 0 : page.entries_.front().score_, - .max_score_ = page.entries_.empty() ? 0 : page.entries_.back().score_}, + .min_score_ = entries.empty() ? 0 : entries.front().score_, + .max_score_ = entries.empty() ? 0 : entries.back().score_}, tx->txid_, command_batch}; if (plan.root_.kind_ == OrderedCollectionKind::kStream && !page.retired_) { diff --git a/src/storage/engine/grouped/zset.cpp b/src/storage/engine/grouped/zset.cpp index 9722a880d..e4566d110 100644 --- a/src/storage/engine/grouped/zset.cpp +++ b/src/storage/engine/grouped/zset.cpp @@ -1706,6 +1706,45 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( HashGroupSnapshot{}, std::move(*point)); continue; } + if (operation.kind_ == SortedSetOperationKind::kAdd && + operation.entries_.size() == 1) { + const auto& entry = operation.entries_.front(); + const auto old_bytes = payload.size() - kHashGroupHeaderBytes; + const auto base_bytes = std::max(old_bytes, kHashValueHeaderBytes); + // A missing field does not invalidate the checked byte-splice + // editor. Keep the ordinary decoder for a split or large member, + // and never turn repeated request operands into a point insertion. + constexpr std::size_t kEntryFrameAndScoreBytes = 8 + sizeof(double); + if (entry.member_.size() <= + kCollectionGroupTargetBytes - kEntryFrameAndScoreBytes && + base_bytes <= kCollectionGroupTargetBytes - + kEntryFrameAndScoreBytes - + entry.member_.size()) { + GroupedScratchBudget incoming; + status = incoming.AddBytes(entry.member_.size() + 256); + if (!status.ok()) co_return status; + auto incoming_admission = incoming.Reserve(1); + if (!incoming_admission.ok()) co_return incoming_admission.status(); + const auto score = EncodeSortedSetMemberScore(entry.score_); + const HashEntryView edit{entry.member_, score}; + auto inserted = ApplyHashGroupEdits( + payload, object->directory().root().seed_, + HashGroupEditKind::kSetIfAbsent, {&edit, 1}); + if (!inserted.ok()) co_return inserted.status(); + if (!inserted->changed_ || inserted->added_ != 1 || + inserted->removed_ != 0 || inserted->leaves_.size() != 1 || + !inserted->leaves_.front().prepared_) + co_return absl::DataLossError( + "invalid member-index insertion probe"); + member_probe.emplace( + std::move(*admission), object, HashGroupSnapshot{}, + std::nullopt, + SortedSetMemberInsertion{std::move(*incoming_admission), + entry.member_, entry.score_, + std::move(inserted->leaves_.front())}); + continue; + } + } LAVIK_FAULT_INJECT( if (LAVIK_FAULT_MATCHES("LAVIK_FAIL_ZSET_MEMBER_DECODE_KEY", key)) co_return absl:: @@ -1996,6 +2035,17 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( destination.begin(), destination.end(), moved, OrderedEntryLess); destination.insert(position, std::move(moved)); } + } else if (operation.kind_ == SortedSetOperationKind::kAdd && + operation.entries_.size() == 1 && pending.size() == 1) { + // The untouched destination is already ordered. Insert a new member at + // its exact score/member position instead of sorting all existing entries. + // The member-index lookup and ordinary page validation remain mandatory. + const auto& [member, state] = pending.front(); + auto& destination = loaded.at(state->destination_).entries_; + OrderedCollectionEntry entry{std::string(member), *state->after_}; + const auto position = std::lower_bound( + destination.begin(), destination.end(), entry, OrderedEntryLess); + destination.insert(position, std::move(entry)); } else { for (const auto i : modified) { auto& entries = loaded.at(i).entries_; @@ -2113,7 +2163,7 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( plan.writes_.push_back(std::move(page)); continue; } - auto split = SplitOrderedGroup(std::move(page), next_id); + auto split = SplitCheckedSortedSetGroup(std::move(page), next_id); if (!split.ok()) co_return split.status(); next_id = split->next_group_id_; for (auto& part : split->groups_) { diff --git a/src/storage/engine/grouped/zset_members.cpp b/src/storage/engine/grouped/zset_members.cpp index b3dce89d4..6cf29ef47 100644 --- a/src/storage/engine/grouped/zset_members.cpp +++ b/src/storage/engine/grouped/zset_members.cpp @@ -232,9 +232,11 @@ StorageEngine::Impl::PrepareSortedSetMembers( status = incoming.AddBytes(member.size() + 256); if (!status.ok()) co_return status; } - auto scratch = incoming.Reserve(1); - if (!scratch.ok()) co_return scratch.status(); - result.scratch_ = std::move(*scratch); + if (probe == nullptr || !probe->insertion_) { + auto scratch = incoming.Reserve(1); + if (!scratch.ok()) co_return scratch.status(); + result.scratch_ = std::move(*scratch); + } auto& plan = result.plan_; plan.root_ = previous->directory().root(); if (changes.empty()) co_return result; @@ -248,10 +250,38 @@ StorageEngine::Impl::PrepareSortedSetMembers( if (probe != nullptr && (!checked_changes || probe->source_.get() != previous.get() || leaves.size() != 1 || - leaves.begin()->first != - (probe->encoded_ ? probe->encoded_->id() : probe->snapshot_.id_) || + leaves.begin()->first != (probe->encoded_ ? probe->encoded_->id() + : probe->insertion_ + ? probe->insertion_->replacement_.id_ + : probe->snapshot_.id_) || probe->snapshot_.retired_)) co_return absl::DataLossError("invalid Sorted Set member probe"); + if (probe != nullptr && probe->insertion_) { + auto& insertion = *probe->insertion_; + const auto& replacement = insertion.replacement_; + if (probe->encoded_ || changes.size() != 1 || + changes.begin()->first != insertion.member_ || + changes.begin()->second.before_ || !changes.begin()->second.after_ || + std::bit_cast(*changes.begin()->second.after_) != + std::bit_cast(insertion.score_) || + replacement.retired_ || !replacement.prepared_ || + !replacement.value_.entries_.empty() || + replacement.incarnation_ != plan.root_.incarnation_) + co_return absl::DataLossError( + "invalid encoded Sorted Set member insertion"); + // The lookup certified absence plus complete Hash framing, uniqueness + // and routing. The typed delta now certifies the same new member/score + // in the ordered graph; consume the prepared bytes without re-decoding. + // Transfer incoming admission with its after-image only after validation, + // so rejection cannot release credit while the probe still owns bytes. + result.scratch_ = std::move(insertion.admission_); + result.leaves_ = std::move(probe->admission_); + plan.root_.revision_ = ordered.root_.revision_; + plan.root_.field_count_ = ordered.root_.item_count_; + plan.changed_ = true; + plan.writes_.push_back(std::move(insertion.replacement_)); + co_return result; + } if (probe != nullptr && probe->encoded_) { // The point lookup validated every field and retained exactly this old // value. Reconcile the semantic delta before consuming its private bytes; diff --git a/src/storage/engine/impl.h b/src/storage/engine/impl.h index 6d601d515..61af42956 100644 --- a/src/storage/engine/impl.h +++ b/src/storage/engine/impl.h @@ -2506,6 +2506,15 @@ class StorageEngine::Impl { std::optional before_; std::optional after_; }; + // A bounded absent-member probe can splice its checked after-image while + // the original read lease is live. The request member outlives preparation; + // its score and identity must match the eventual typed delta before use. + struct SortedSetMemberInsertion { + MemoryReservation admission_; + std::string_view member_; + double score_; + HashGroupSnapshot replacement_; + }; // A single-member writer may hand its checked lookup leaf to index // preparation. Ownership is command-local, and the original admission // follows its bounded checked bytes or decoded fallback until the prepared @@ -2515,6 +2524,7 @@ class StorageEngine::Impl { GroupedObject::Handle source_; HashGroupSnapshot snapshot_; std::optional encoded_; + std::optional insertion_ = std::nullopt; }; // Full-image writers derive changes from complete ordered before/after // pages. Typed writers may supply exact changes after checking old members diff --git a/tests/grouped/collection_test.cpp b/tests/grouped/collection_test.cpp index 405e23648..d0bdb6a80 100644 --- a/tests/grouped/collection_test.cpp +++ b/tests/grouped/collection_test.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include "gtest/gtest.h" @@ -1050,6 +1051,148 @@ TEST(GroupedCollectionTest, UnsplitPagePreservesStorageAndIdentity) { } } +TEST(GroupedCollectionTest, CheckedSortedSetSplitOwnsImmutableEntries) { + static_assert(!std::is_constructible_v>); + static_assert( + std::is_same_v< + decltype(std::declval().entries()), + std::span>); + auto page = Page(7, 184, OrderedCollectionKind::kSortedSet); + page.previous_ = 3; + page.next_ = 9; + for (auto& entry : page.entries_) entry.value_.resize(32, 'x'); + const auto expected = page.entries_; + const auto* entries_storage = page.entries_.data(); + const auto* member_storage = page.entries_.front().value_.data(); + auto before = EncodeOrderedGroup(page); + ASSERT_TRUE(before.ok()) << before.status(); + + auto split = SplitCheckedSortedSetGroup(std::move(page), 20); + ASSERT_TRUE(split.ok()) << split.status(); + ASSERT_EQ(split->groups_.size(), 1); + auto kept = std::move(split->groups_.front()); + ASSERT_TRUE(kept.checked_sorted_set_.has_value()); + EXPECT_TRUE(kept.entries_.empty()); + const auto entries = OrderedGroupEntries(kept); + ASSERT_EQ(entries.size(), expected.size()); + EXPECT_EQ(entries.data(), entries_storage); + EXPECT_EQ(entries.front().value_.data(), member_storage); + EXPECT_TRUE(std::equal(entries.begin(), entries.end(), expected.begin())); + EXPECT_EQ(OrderedGroupSize(kept), expected.size()); + auto encoded = EncodeOrderedGroup(kept); + ASSERT_TRUE(encoded.ok()) << encoded.status(); + EXPECT_EQ(*encoded, *before); + // Link publication follows the split, but the immutable member sequence + // and its proof survive ownership moves without another payload copy. + kept.previous_ = 2; + kept.next_ = 12; + encoded = EncodeOrderedGroup(kept); + ASSERT_TRUE(encoded.ok()) << encoded.status(); + auto decoded = DecodeOrderedGroup(*encoded); + ASSERT_TRUE(decoded.ok()) << decoded.status(); + EXPECT_FALSE(decoded->checked_sorted_set_.has_value()); + EXPECT_EQ(decoded->entries_, expected); + EXPECT_EQ(decoded->previous_, 2); + EXPECT_EQ(decoded->next_, 12); +} + +TEST(GroupedCollectionTest, CheckedSortedSetSplitMatchesOrdinarySplit) { + auto page = Page(7, 30, OrderedCollectionKind::kSortedSet); + page.previous_ = 3; + page.next_ = 9; + for (auto& entry : page.entries_) entry.value_.resize(32, 'x'); + // Include an indivisible oversized member between ordinary pages. + page.entries_[15].value_.resize(4096, 'x'); + auto ordinary = SplitOrderedGroup(page, 20, 256); + auto checked = SplitCheckedSortedSetGroup(std::move(page), 20, 256); + ASSERT_TRUE(ordinary.ok()) << ordinary.status(); + ASSERT_TRUE(checked.ok()) << checked.status(); + ASSERT_EQ(checked->groups_.size(), ordinary->groups_.size()); + EXPECT_EQ(checked->next_group_id_, ordinary->next_group_id_); + for (std::size_t i = 0; i < ordinary->groups_.size(); ++i) { + ASSERT_TRUE(checked->groups_[i].checked_sorted_set_.has_value()); + auto expected = EncodeOrderedGroup(ordinary->groups_[i]); + auto encoded = EncodeOrderedGroup(checked->groups_[i]); + ASSERT_TRUE(expected.ok()) << expected.status(); + ASSERT_TRUE(encoded.ok()) << encoded.status(); + EXPECT_EQ(*encoded, *expected) << i; + if (i != 0) { + EXPECT_TRUE(ValidateOrderedGroupBoundary(checked->groups_[i - 1], + checked->groups_[i]) + .ok()); + EXPECT_TRUE(ValidateOrderedGroupBoundary(ordinary->groups_[i - 1], + checked->groups_[i]) + .ok()); + EXPECT_TRUE(ValidateOrderedGroupBoundary(checked->groups_[i - 1], + ordinary->groups_[i]) + .ok()); + } + } +} + +TEST(GroupedCollectionTest, CheckedSortedSetBoundariesRejectOverlap) { + auto left = SplitCheckedSortedSetGroup( + Page(7, 3, OrderedCollectionKind::kSortedSet), 20); + auto right = SplitCheckedSortedSetGroup( + Page(8, 3, OrderedCollectionKind::kSortedSet), 20); + ASSERT_TRUE(left.ok()) << left.status(); + ASSERT_TRUE(right.ok()) << right.status(); + left->groups_.front().next_ = right->groups_.front().id_; + right->groups_.front().previous_ = left->groups_.front().id_; + EXPECT_FALSE(ValidateOrderedGroupBoundary(left->groups_.front(), + right->groups_.front()) + .ok()); +} + +TEST(GroupedCollectionTest, CheckedSortedSetSplitRejectsUnverifiedAfterimages) { + const auto valid = Page(7, 20, OrderedCollectionKind::kSortedSet); + const auto rejected = [&](auto change) { + auto page = valid; + change(page); + EXPECT_FALSE(SplitCheckedSortedSetGroup(std::move(page), 20, 128).ok()); + }; + // These duplicate names would fall in different output pages. Checking + // only each final encoder would miss the invalid input sequence. + rejected([](auto& page) { + page.entries_.back().value_ = page.entries_.front().value_; + }); + rejected([](auto& page) { + page.entries_.front().score_ = std::numeric_limits::quiet_NaN(); + }); + rejected([](auto& page) { std::swap(page.entries_[0], page.entries_[1]); }); + rejected([](auto& page) { page.incarnation_ = 0; }); + rejected([](auto& page) { page.previous_ = page.id_; }); + rejected([](auto& page) { page.borrowed_string_ = "invalid"; }); + EXPECT_FALSE(SplitCheckedSortedSetGroup(Page(), 20).ok()); +} + +TEST(GroupedCollectionTest, CheckedSortedSetEntriesKeepEncoderGuards) { + auto split = SplitCheckedSortedSetGroup( + Page(7, 3, OrderedCollectionKind::kSortedSet), 20); + ASSERT_TRUE(split.ok()) << split.status(); + const auto& valid = split->groups_.front(); + const auto rejected = [&](auto change) { + auto page = valid; + change(page); + EXPECT_FALSE(OrderedGroupEncoder::Create(page).ok()); + }; + rejected([](auto& page) { page.entries_.push_back({"new", 3}); }); + rejected([](auto& page) { page.kind_ = OrderedCollectionKind::kList; }); + rejected([](auto& page) { page.retired_ = true; }); + rejected([](auto& page) { page.id_ = 0; }); + rejected([](auto& page) { page.next_ = page.id_; }); + rejected([](auto& page) { page.borrowed_string_ = "invalid"; }); + EXPECT_FALSE(SplitOrderedGroup(valid, 20).ok()); + EXPECT_FALSE(SplitCheckedSortedSetGroup(valid, 20).ok()); + auto left = valid; + auto right = Page(8, 3, OrderedCollectionKind::kSortedSet); + left.next_ = right.id_; + right.previous_ = left.id_; + for (auto& entry : right.entries_) entry.score_ += 3; + EXPECT_FALSE(RebalanceSortedSetGroupPair(left, right).ok()); +} + TEST(GroupedCollectionTest, UnsplitOversizedMemberPreservesStorageAndIdentity) { auto page = Page(7, 1, OrderedCollectionKind::kSortedSet); page.previous_ = 3; diff --git a/tests/grouped/zset_write_e2e_test.cpp b/tests/grouped/zset_write_e2e_test.cpp index 5e6c25d03..4ee09a9dd 100644 --- a/tests/grouped/zset_write_e2e_test.cpp +++ b/tests/grouped/zset_write_e2e_test.cpp @@ -581,6 +581,226 @@ TEST(GroupedSortedSetWriteE2e, EXPECT_EQ(client.Command({"GET", "unrelated"}).text_, "after failure"); } +struct ZSetMemberLayout { + GroupedHashRoot root_; + std::map leaves_; +}; + +ZSetMemberLayout InspectZSetMembers(const PrivateDisk& disk, + std::string_view key) { + ZSetMemberLayout result; + std::uint64_t root_sequence = 0; + std::vector> records; + disk.VisitRecords([&](const BlockHeader&, const RecordHeader& record, + std::string_view name, std::string_view payload) { + if (name != key) return; + if (record.grouped_ && !record.auxiliary_group_ && + record.mutation_sequence_ >= root_sequence) { + const auto root = DecodeOrderedCollectionRoot(payload); + Check(root.ok() && root->member_index_, "invalid fixture ZSet root"); + result.root_ = *root->member_index_; + root_sequence = record.mutation_sequence_; + } else if (record.auxiliary_group_ && + (record.group_prefix_bits_ != 0 || record.group_prefix_ == 0)) { + Check(!record.external_, "fixture member page must be inline"); + records.emplace_back(record, payload); + } + }); + Check(root_sequence != 0, "fixture has no grouped ZSet root"); + std::map sequences; + for (const auto& [record, payload] : records) { + if (record.group_incarnation_ != result.root_.incarnation_ || + record.mutation_sequence_ > result.root_.revision_) + continue; + const GroupedRecordId id{record.group_prefix_, record.group_prefix_bits_}; + if (record.mutation_sequence_ < sequences[id]) continue; + sequences[id] = record.mutation_sequence_; + if (record.group_retired_) { + result.leaves_.erase(id); + } else { + auto leaf = DecodeHashGroup(payload); + Check(leaf.ok(), "invalid fixture member page"); + result.leaves_[id] = std::move(*leaf); + } + } + Check(result.leaves_.size() == result.root_.group_count_, + "incomplete fixture member directory"); + return result; +} + +TEST(GroupedSortedSetWriteE2e, + EncodedMemberInsertionsKeepFlagsBinaryOrderAndRecovery) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires member-index owned decode failure injection"; +#endif + PrivateDisk disk; + const std::string key = "encoded-member-insert"; + std::map expected; + { + Server server(disk, 1, {}, {}, false, 2, "128M"); + Client client(server.port()); + ASSERT_EQ(client.Command(DenseZSetSeed(key, &expected)).text_, "1024"); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + // Choose deletions from the actual persisted routing leaves. Every leaf + // then has room for all new operands, independently of the random seed. + // The remaining aggregate still exceeds one page and stays grouped. + const auto layout = InspectZSetMembers(disk, key); + std::vector trim{"ZREM", key}; + for (const auto& [id, leaf] : layout.leaves_) { + for (std::size_t i = 32; i < leaf.value_.entries_.size(); ++i) { + trim.push_back(leaf.value_.entries_[i].field_); + expected.erase(trim.back()); + } + } + ASSERT_GT(trim.size(), 2); + { + Server server(disk, 2, {}, {}, false, 2, "128M"); + Client client(server.port()); + ASSERT_EQ(client.Command(trim).text_, std::to_string(trim.size() - 2)); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + ASSERT_EQ(InspectZSetMembers(disk, key).root_.group_count_, + layout.root_.group_count_); + { + ScopedEnvironment fault("LAVIK_FAIL_ZSET_MEMBER_DECODE_KEY", key.c_str()); + Server server(disk, 3, {}, {}, false, 2, "128M"); + Client client(server.port()); + const std::string binary("\0member\xff\0", 9); + const std::string long_member(1024, '\xff'); + ASSERT_EQ(client.Command({"ZADD", key, "XX", "512", "missing"}).text_, "0"); + ASSERT_EQ(client.Command({"ZADD", key, "XX", "INCR", "1", "missing"}).text_, + "-1"); + ASSERT_EQ(client.Command({"ZADD", key, "NX", "512", ""}).text_, "1"); + expected[""] = 512; + ASSERT_EQ(client.Command({"ZADD", key, "NX", "999", ""}).text_, "0"); + ASSERT_EQ(client.Command({"ZADD", key, "CH", "512", binary}).text_, "1"); + expected[binary] = 512; + ASSERT_EQ(client.Command({"ZADD", key, "LT", "512", long_member}).text_, + "1"); + expected[long_member] = 512; + ASSERT_EQ(client.Command({"ZADD", key, "GT", "512", "gt-new"}).text_, "1"); + expected["gt-new"] = 512; + ASSERT_EQ( + client.Command({"ZADD", key, "INCR", "2.5", "increment-new"}).text_, + "2.5"); + expected["increment-new"] = 2.5; + ASSERT_EQ(client.Command({"ZINCRBY", key, "-2.5", "zincr-new"}).text_, + "-2.5"); + expected["zincr-new"] = -2.5; + // Repeated operands are a batch even when they name one absent member. + // The failure must not publish the first operand's private insertion. + const auto repeated = + client.Command({"ZADD", key, "NX", "1", "repeat", "2", "repeat"}); + EXPECT_EQ(repeated.kind_, '-'); + EXPECT_NE(repeated.text_.find("owned decode failure"), std::string::npos); + EXPECT_EQ(client.Command({"ZSCORE", key, "repeat"}).text_, "-1"); + const auto batch = + client.Command({"ZADD", key, "1", "batch-a", "2", "batch-b"}); + EXPECT_EQ(batch.kind_, '-'); + EXPECT_NE(batch.text_.find("owned decode failure"), std::string::npos); + ExpectDenseZSet(client, key, expected); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + Server recovered(disk, 2, {}, {}, false, 2, "128M"); + Client client(recovered.port()); + ExpectDenseZSet(client, key, expected); +} + +TEST(GroupedSortedSetWriteE2e, + EncodedMemberInsertionOverflowUsesDecodedSplitAndRecovers) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires member-index owned decode failure injection"; +#endif + PrivateDisk disk; + const std::string key = "encoded-member-overflow"; + std::map expected; + { + Server server(disk, 1, {}, {}, false, 2, "128M"); + Client client(server.port()); + ASSERT_EQ(client.Command(DenseZSetSeed(key, &expected)).text_, "1024"); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + const auto layout = InspectZSetMembers(disk, key); + const auto& [id, leaf] = *std::min_element( + layout.leaves_.begin(), layout.leaves_.end(), + [](const auto& a, const auto& b) { + return a.second.field_count() < b.second.field_count(); + }); + const auto encoded = EncodeHashGroup(leaf); + ASSERT_TRUE(encoded.ok()) << encoded.status(); + const auto old_bytes = encoded->size() - kHashGroupHeaderBytes; + ASSERT_LE(old_bytes + 16 + 32, kCollectionGroupTargetBytes); + auto routed_member = [&](std::size_t size, std::string prefix) { + for (unsigned i = 0; i < 100000; ++i) { + auto member = prefix + std::to_string(i); + Check(member.size() <= size, "fixture routed member too short"); + member.resize(size, '\0'); + if (id.ContainsHash(ComputeDigest(member, layout.root_.seed_).value_) && + !expected.contains(member)) + return member; + } + throw std::runtime_error("cannot find fixture member in selected prefix"); + }; + const auto filler = + routed_member(kCollectionGroupTargetBytes - old_bytes - 16, "fill-"); + const auto overflow = routed_member(32, "overflow-"); + const std::string oversized(16384, 'z'); + { + Server server(disk, 2, {}, {}, false, 2, "128M"); + Client client(server.port()); + ASSERT_EQ(client.Command({"ZADD", key, "511", filler}).text_, "1"); + expected[filler] = 511; + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + const auto full = InspectZSetMembers(disk, key); + ASSERT_EQ(full.root_.group_count_, layout.root_.group_count_); + const auto full_payload = EncodeHashGroup(full.leaves_.at(id)); + ASSERT_TRUE(full_payload.ok()) << full_payload.status(); + ASSERT_EQ(full_payload->size() - kHashGroupHeaderBytes, + kCollectionGroupTargetBytes); + { + ScopedEnvironment fault("LAVIK_FAIL_ZSET_MEMBER_DECODE_KEY", key.c_str()); + Server server(disk, 3, {}, {}, false, 2, "128M"); + Client client(server.port()); + for (const auto& member : {overflow, oversized}) { + const auto failed = client.Command({"ZADD", key, "511", member}); + EXPECT_EQ(failed.kind_, '-'); + EXPECT_NE(failed.text_.find("owned decode failure"), std::string::npos); + EXPECT_EQ(client.Command({"ZSCORE", key, member}).text_, "-1"); + } + // A full leaf still supports an equal-width existing-member update. + ASSERT_EQ(client.Command({"ZADD", key, "CH", "512", filler}).text_, "1"); + expected[filler] = 512; + ExpectDenseZSet(client, key, expected); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + { + Server server(disk, 2, {}, {}, false, 2, "128M"); + Client client(server.port()); + for (const auto& member : {overflow, oversized}) { + ASSERT_EQ(client.Command({"ZADD", key, "511", member}).text_, "1"); + expected[member] = 511; + } + ExpectDenseZSet(client, key, expected); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + // The short insertion splits an ordinary full prefix; the oversized entry + // also retains its existing bounded owned-encoder path. + ASSERT_GT(InspectZSetMembers(disk, key).root_.group_count_, + layout.root_.group_count_); + Server recovered(disk, 1, {}, {}, false, 2, "128M"); + Client client(recovered.port()); + ExpectDenseZSet(client, key, expected); +} + TEST(GroupedSortedSetWriteE2e, DuplicateMemberInputsKeepSequentialConditions) { PrivateDisk disk; const std::string key = "duplicate-zadd";