diff --git a/bycorf b/bycorf index 86d686724..587a9e776 160000 --- a/bycorf +++ b/bycorf @@ -1 +1 @@ -Subproject commit 86d686724ddd1c5617fb1b62f8131e05571e8423 +Subproject commit 587a9e776f9f64729456d6772596d53597f4a983 diff --git a/docs/architecture/09-grouped-collections.md b/docs/architecture/09-grouped-collections.md index 85eb0bc8d..b6db79385 100644 --- a/docs/architecture/09-grouped-collections.md +++ b/docs/architecture/09-grouped-collections.md @@ -480,10 +480,12 @@ Rank/score page readers copy borrowed members into page blocks before releasing the read lease, then transfer selected slices and blocks into the result. Reverse order, bounds and LIMIT select the same logical members in either output representation. Replies retain their admitted buffers through encoding, -including EXEC and Lua. Mutation source and boundary probes, and rewrite -planning, keep the default owned-member snapshots. Command-local probes retain admitted source -pages containing requested members within a fixed scratch bound and reuse -them in planning. Unrelated or excess search pages are released immediately; +including EXEC and Lua. Single-member mutation probes and after-images may +retain validated encoded pages, owning their bytes after the physical read +lease ends. Plans that change page topology materialize mutable entries; +batch and pop planning use owned-member snapshots. Command-local probes retain +admitted source pages containing requested members within a fixed scratch bound +and reuse them in planning. Unrelated or excess search pages are released immediately; large members retain the ordinary read fallback. Pop selection passes its owned pages into deletion under the same exclusive key intent. Reuse checks the logical population and preserves member-index/ordered-index agreement; physical GC diff --git a/include/lavik/storage/detail/grouped/collection.h b/include/lavik/storage/detail/grouped/collection.h index 2bacc55cf..47f0cf0a5 100644 --- a/include/lavik/storage/detail/grouped/collection.h +++ b/include/lavik/storage/detail/grouped/collection.h @@ -17,6 +17,7 @@ #pragma once #include +#include #include #include #include @@ -32,6 +33,7 @@ #include "lavik/containers/cow_array.h" #include "lavik/containers/cow_map.h" #include "lavik/containers/fenwick_tree.h" +#include "lavik/local_shared_ptr.h" #include "lavik/storage/detail/collection_limits.h" #include "lavik/storage/detail/grouped/hash.h" #include "lavik/storage/format.h" @@ -119,6 +121,36 @@ class CheckedSortedSetEntries { std::vector entries_; }; +// A fully checked page held in one immutable byte block. Views share that +// block across snapshot copies/moves, so no physical read lease survives +// decoding and no member acquires its own string allocation. The owner and +// all snapshots stay on the key owner worker, including across suspension and +// destruction; only the checked decoder and topology-preserving point editor +// can construct this owner. +class EncodedSortedSetEntries { + public: + std::span entries() const noexcept { + return entries_; + } + std::string_view bytes() const noexcept { return *bytes_; } + + private: + friend absl::StatusOr DecodeEncodedSortedSetGroup( + std::string_view bytes); + friend absl::StatusOr> + EditEncodedSortedSetGroup(const OrderedGroupSnapshot& page, + std::string_view member, + std::optional before, + std::optional after, + std::size_t target_bytes); + EncodedSortedSetEntries(LocalSharedPtr bytes, + std::vector entries) + : bytes_(std::move(bytes)), entries_(std::move(entries)) {} + // Declare the owner first so every borrowing view dies before its bytes. + LocalSharedPtr bytes_; + std::vector entries_; +}; + struct OrderedCollectionRoot { OrderedCollectionKind kind_ = OrderedCollectionKind::kList; std::uint64_t incarnation_ = 0; @@ -166,12 +198,18 @@ struct OrderedGroupSnapshot { // 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; + // Ordinary point edits keep the complete, checked entry framing in one + // owned block. Topology changes materialize mutable entries before planning. + std::optional encoded_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. +// Encoded pages expose views through encoded_sorted_set_->entries() instead; +// a mutable planner must materialize them before using this owned-entry API. inline std::span OrderedGroupEntries( const OrderedGroupSnapshot& group) noexcept { + assert(!group.encoded_sorted_set_); return group.checked_sorted_set_ ? group.checked_sorted_set_->entries() : std::span(group.entries_); } @@ -192,6 +230,8 @@ inline std::size_t OrderedGroupSize( const OrderedGroupSnapshot& group) noexcept { return group.kind_ == OrderedCollectionKind::kString ? OrderedStringBytes(group).size() + : group.encoded_sorted_set_ + ? group.encoded_sorted_set_->entries().size() : OrderedGroupEntries(group).size(); } @@ -289,6 +329,26 @@ absl::StatusOr DecodeOrderedGroup( std::string_view bytes, SortedSetMemberValidation validation = SortedSetMemberValidation::kVerifyUnique); +// Fully validates a live Sorted Set page, including member uniqueness, and +// retains its encoded entries in one owned block after the physical lease ends. +absl::StatusOr DecodeEncodedSortedSetGroup( + std::string_view bytes); + +// Edits one member in an encoded page without changing its ID or links. before +// and after describe this page only (a cross-page move calls it twice). Returns +// nullopt for an empty/oversized after-image, leaving topology changes to the +// ordinary planner. The input is immutable even on failure; score mismatches, +// duplicate additions and invalid inputs are errors, never silent fallbacks. +absl::StatusOr> EditEncodedSortedSetGroup( + const OrderedGroupSnapshot& page, std::string_view member, + std::optional before, std::optional after, + std::size_t target_bytes = kCollectionGroupTargetBytes); + +// Converts a private checked/encoded page into mutable, independently owned +// entries before split/rebalance planning. Existing mutable pages are +// unchanged. +void MaterializeOrderedGroupEntries(OrderedGroupSnapshot& page); + // Validates a complete live List page, including entries outside the requested // interval, but allocates/copies only [first, first + count). The interval must // fit the page. Returned strings own their bytes independently of the payload. diff --git a/src/storage/engine/grouped/collection.cpp b/src/storage/engine/grouped/collection.cpp index 2dc3f1603..d0d8eb148 100644 --- a/src/storage/engine/grouped/collection.cpp +++ b/src/storage/engine/grouped/collection.cpp @@ -143,23 +143,36 @@ absl::StatusOr ValidateGroup( SortedSetMemberValidation validation = SortedSetMemberValidation::kVerifyUnique) { const bool borrowed_string = group.borrowed_string_.has_value(); + if (group.encoded_sorted_set_ && + (group.kind_ != OrderedCollectionKind::kSortedSet || + !group.entries_.empty() || borrowed_string || + group.checked_sorted_set_ || group.retired_)) + return absl::InvalidArgumentError("invalid encoded Sorted Set entries"); 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); + const auto entries = group.encoded_sorted_set_ + ? std::span{} + : OrderedGroupEntries(group); + const auto count = OrderedGroupSize(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_) || - entries.size() > std::numeric_limits::max() || + count > std::numeric_limits::max() || (group.retired_ ? (!entries.empty() || borrowed_string || group.previous_ != 0 || group.next_ != 0) - : (entries.empty() && !borrowed_string))) { + : (count == 0 && !borrowed_string))) { return absl::InvalidArgumentError("invalid ordered page identity or state"); } + // The encoded owner is immutable and was fully checked on decode. Its + // point editor preserves framing/order and rejects duplicates; link/identity + // checks above still apply to the independently mutable envelope. + if (group.encoded_sorted_set_) + return kOrderedGroupHeaderBytes + group.encoded_sorted_set_->bytes().size(); // 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. @@ -428,6 +441,11 @@ std::optional OrderedGroupEncoder::Next() noexcept { ++entry_; return OrderedStringBytes(*group_); } + if (group_->encoded_sorted_set_) { + if (entry_ != 0) return std::nullopt; + ++entry_; + return group_->encoded_sorted_set_->bytes(); + } const auto entries = OrderedGroupEntries(*group_); if (entry_ == entries.size()) return std::nullopt; const auto& entry = entries[entry_]; @@ -503,6 +521,129 @@ absl::StatusOr DecodeOrderedGroup( return group; } +absl::StatusOr DecodeEncodedSortedSetGroup( + std::string_view bytes) { + auto entries = DecodeSortedSetGroupViews(bytes); + if (!entries.ok()) return entries.status(); + auto metadata = DecodeOrderedGroupMetadata(bytes, bytes.size()); + if (!metadata.ok()) return metadata.status(); + auto owned = MakeLocalShared( + bytes.substr(kOrderedGroupHeaderBytes)); + for (auto& entry : *entries) { + const auto offset = + entry.value_.data() - bytes.data() - kOrderedGroupHeaderBytes; + entry.value_ = std::string_view(*owned).substr(offset, entry.value_.size()); + } + OrderedGroupSnapshot page{.kind_ = metadata->kind_, + .incarnation_ = metadata->incarnation_, + .id_ = metadata->id_, + .previous_ = metadata->previous_, + .next_ = metadata->next_, + .entries_ = {}}; + page.encoded_sorted_set_ = + EncodedSortedSetEntries(std::move(owned), std::move(*entries)); + return page; +} + +absl::StatusOr> EditEncodedSortedSetGroup( + const OrderedGroupSnapshot& page, std::string_view member, + std::optional before, std::optional after, + std::size_t target_bytes) { + if ((!before && !after) || member.size() > kMaxStringBytes || + (before && std::isnan(*before)) || (after && std::isnan(*after))) + return absl::InvalidArgumentError("invalid encoded Sorted Set point edit"); + if (!page.encoded_sorted_set_) return std::nullopt; + const auto valid = ValidateGroup(page); + if (!valid.ok()) return valid.status(); + const auto entries = page.encoded_sorted_set_->entries(); + const auto bytes = page.encoded_sorted_set_->bytes(); + const auto missing = entries.size(); + std::size_t removed = missing; + for (std::size_t i = 0; i < entries.size(); ++i) { + if (entries[i].value_ != member) continue; + if (!before || entries[i].score_ != *before) + return absl::DataLossError("encoded Sorted Set member score mismatch"); + removed = i; + } + if (before && removed == missing) + return absl::DataLossError("encoded Sorted Set point source missing"); + const auto count = entries.size() - bool(before) + bool(after); + target_bytes = std::min(target_bytes, kMaxRecordPayloadBytes); + const auto frame_bytes = kEntryHeaderBytes + member.size(); + const auto retained_bytes = bytes.size() - (before ? frame_bytes : 0); + if (count == 0 || target_bytes < kOrderedGroupHeaderBytes || + retained_bytes > target_bytes - kOrderedGroupHeaderBytes || + (after && + frame_bytes > target_bytes - kOrderedGroupHeaderBytes - retained_bytes)) + return std::nullopt; + const auto size = retained_bytes + (after ? frame_bytes : 0); + auto output = MakeLocalShared(); + output->reserve(size); + auto append_member = [&] { + std::array header{}; + Store(header, 0, member.size(), 4); + Store(header, 4, std::bit_cast(*after), 8); + output->append(header.data(), header.size()); + output->append(member); + }; + bool inserted = !after; + std::size_t start = 0; + // Every unchanged framed byte is copied once, in at most three runs split + // by the old and new positions. Numeric order treats -0/+0 as equal and + // binary member order breaks ties, exactly as the ordinary page planner. + for (std::size_t i = 0; i < entries.size(); ++i) { + const auto offset = + entries[i].value_.data() - bytes.data() - kEntryHeaderBytes; + if (i == removed) { + output->append(bytes.substr(start, offset - start)); + start = offset + frame_bytes; + continue; + } + if (!inserted && + EntryLess(OrderedCollectionEntryView{member, *after}, entries[i])) { + output->append(bytes.substr(start, offset - start)); + start = offset; + append_member(); + inserted = true; + } + } + output->append(bytes.substr(start)); + if (!inserted) append_member(); + std::vector views; + views.reserve(count); + std::string_view encoded(*output); + for (std::size_t offset = 0; offset != encoded.size();) { + const auto length = Load(encoded, offset, 4); + const auto score = std::bit_cast(Load(encoded, offset + 4, 8)); + offset += kEntryHeaderBytes; + views.push_back({encoded.substr(offset, length), score}); + offset += length; + } + OrderedGroupSnapshot result{.kind_ = page.kind_, + .incarnation_ = page.incarnation_, + .id_ = page.id_, + .previous_ = page.previous_, + .next_ = page.next_, + .entries_ = {}}; + result.encoded_sorted_set_ = + EncodedSortedSetEntries(std::move(output), std::move(views)); + return result; +} + +void MaterializeOrderedGroupEntries(OrderedGroupSnapshot& page) { + if (page.encoded_sorted_set_) { + const auto entries = page.encoded_sorted_set_->entries(); + page.entries_.reserve(entries.size()); + for (const auto& entry : entries) + page.entries_.push_back({std::string(entry.value_), entry.score_}); + page.encoded_sorted_set_.reset(); + } else if (page.checked_sorted_set_) { + const auto entries = page.checked_sorted_set_->entries(); + page.entries_.assign(entries.begin(), entries.end()); + page.checked_sorted_set_.reset(); + } +} + absl::StatusOr> DecodeOrderedListRange( std::string_view bytes, std::size_t first, std::size_t count) { auto metadata = DecodeOrderedGroupMetadata(bytes, bytes.size()); @@ -738,6 +879,22 @@ absl::Status ValidateOrderedGroupBoundary(const OrderedGroupSnapshot& left, return absl::DataLossError("inconsistent ordered page neighbours"); } if (left.kind_ == OrderedCollectionKind::kString) return absl::OkStatus(); + if (left.encoded_sorted_set_ || right.encoded_sorted_set_) { + const auto last = left.encoded_sorted_set_ + ? left.encoded_sorted_set_->entries().back() + : OrderedCollectionEntryView{ + OrderedGroupEntries(left).back().value_, + OrderedGroupEntries(left).back().score_}; + const auto first = right.encoded_sorted_set_ + ? right.encoded_sorted_set_->entries().front() + : OrderedCollectionEntryView{ + OrderedGroupEntries(right).front().value_, + OrderedGroupEntries(right).front().score_}; + return EntryLess(last, first) + ? absl::OkStatus() + : absl::DataLossError( + "Sorted Set pages overlap or are unordered"); + } return ValidateOrderedEntryBoundary(left.kind_, OrderedGroupEntries(left).back(), OrderedGroupEntries(right).front()); @@ -1645,6 +1802,7 @@ absl::StatusOr RebalanceSortedSetGroupPair(OrderedGroupSnapshot& left, if (left.kind_ != OrderedCollectionKind::kSortedSet || right.kind_ != left.kind_ || left.incarnation_ == 0 || left.checked_sorted_set_ || right.checked_sorted_set_ || + left.encoded_sorted_set_ || right.encoded_sorted_set_ || left.incarnation_ != right.incarnation_ || left.id_ == 0 || right.id_ == 0 || left.id_ == right.id_ || left.retired_ || right.retired_ || left.next_ != right.id_ || @@ -1734,7 +1892,8 @@ absl::StatusOr SplitOrderedGroupImpl( if (target_bytes <= kOrderedGroupHeaderBytes || target_bytes > kMaxRecordPayloadBytes || next_group_id <= std::max({group.id_, group.previous_, group.next_}) || - group.retired_ || group.checked_sorted_set_ || group.entries_.empty()) { + group.retired_ || group.checked_sorted_set_ || + group.encoded_sorted_set_ || group.entries_.empty()) { return absl::InvalidArgumentError("invalid ordered split input"); } // A promotion can exceed one record's aggregate limit. Validate individual @@ -1888,7 +2047,7 @@ 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.checked_sorted_set_ || + page.retired_ || page.checked_sorted_set_ || page.encoded_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/ordered_io.cpp b/src/storage/engine/grouped/ordered_io.cpp index ec2cb17c9..e4edf6f03 100644 --- a/src/storage/engine/grouped/ordered_io.cpp +++ b/src/storage/engine/grouped/ordered_io.cpp @@ -147,18 +147,30 @@ StorageEngine::Impl::LoadOrderedGroupSnapshot( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::uint64_t page_id, bool pinned, - GroupReadLimiter* limiter, SortedSetMemberValidation validation) { + GroupReadLimiter* limiter, SortedSetMemberValidation validation, + bool encoded_sorted_set) { return LoadOrderedGroup( store, partition, db_id, key, digest, std::move(object), page_id, pinned, - [validation](std::string_view payload, std::uint64_t sequence, - const OrderedGroupEntry& route, - LoadedValue&&) -> absl::StatusOr { - auto decoded = DecodeOrderedGroup(payload, validation); + [validation, encoded_sorted_set]( + std::string_view payload, std::uint64_t sequence, + const OrderedGroupEntry& route, + LoadedValue&&) -> absl::StatusOr { + auto decoded = encoded_sorted_set + ? DecodeEncodedSortedSetGroup(payload) + : DecodeOrderedGroup(payload, validation); if (!decoded.ok()) return decoded.status(); - if (decoded->kind_ == OrderedCollectionKind::kSortedSet && - (decoded->entries_.front().score_ != route.min_score_ || - decoded->entries_.back().score_ != route.max_score_)) - return absl::DataLossError("ordered page score bounds mismatch"); + if (decoded->kind_ == OrderedCollectionKind::kSortedSet) { + const auto min_score = + decoded->encoded_sorted_set_ + ? decoded->encoded_sorted_set_->entries().front().score_ + : decoded->entries_.front().score_; + const auto max_score = + decoded->encoded_sorted_set_ + ? decoded->encoded_sorted_set_->entries().back().score_ + : decoded->entries_.back().score_; + if (min_score != route.min_score_ || max_score != route.max_score_) + return absl::DataLossError("ordered page score bounds mismatch"); + } return LoadedOrderedGroup{.sequence_ = sequence, .snapshot_ = std::move(*decoded)}; }, diff --git a/src/storage/engine/grouped/ordered_mutation.cpp b/src/storage/engine/grouped/ordered_mutation.cpp index 3dc9b8ad2..2951f3bfc 100644 --- a/src/storage/engine/grouped/ordered_mutation.cpp +++ b/src/storage/engine/grouped/ordered_mutation.cpp @@ -298,7 +298,10 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( if (page.kind_ == OrderedCollectionKind::kString) compact.push_back( {.value_ = std::string(OrderedStringBytes(page))}); - else { + else if (page.encoded_sorted_set_) { + for (const auto& entry : page.encoded_sorted_set_->entries()) + compact.push_back({std::string(entry.value_), entry.score_}); + } else { const auto entries = OrderedGroupEntries(page); compact.insert(compact.end(), entries.begin(), entries.end()); } @@ -451,7 +454,9 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( continue; } const auto& page = plan.writes_[i]; - const auto entries = OrderedGroupEntries(page); + const auto entries = page.encoded_sorted_set_ + ? std::span{} + : OrderedGroupEntries(page); RecoveredOrderedGroup candidate{ {.incarnation_ = plan.root_.incarnation_, .id_ = page.id_, @@ -463,8 +468,14 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( .encoded_bytes_ = ordered_sizes[i], .record_token_ = page.id_, .retired_ = group.retired_, - .min_score_ = entries.empty() ? 0 : entries.front().score_, - .max_score_ = entries.empty() ? 0 : entries.back().score_}, + .min_score_ = page.encoded_sorted_set_ + ? page.encoded_sorted_set_->entries().front().score_ + : entries.empty() ? 0 + : entries.front().score_, + .max_score_ = page.encoded_sorted_set_ + ? page.encoded_sorted_set_->entries().back().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 e75e27b05..860cd3cef 100644 --- a/src/storage/engine/grouped/zset.cpp +++ b/src/storage/engine/grouped/zset.cpp @@ -1254,7 +1254,7 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( else return LoadOrderedGroupSnapshot(store, partition, db_id, key, digest, object, metadata[i].id_, false, limiter, - validation); + validation, point_write); }; auto page = co_await load(); if (!page.ok()) co_return page.status(); @@ -1263,7 +1263,7 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( auto read_page = [&](std::size_t i) { return read_page_impl.template operator()(i); }; - // Mutation probes own their entries, releasing physical leases before + // Mutation probes own their payloads, releasing physical leases before // routing. Both wrappers return the loader task without another coroutine. auto read_owned_page = [&](std::size_t i) { return read_page_impl.template operator()(i); @@ -1938,7 +1938,11 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( GroupedScratchBudget budget; status = add_page_budget(&budget, i); if (!status.ok()) co_return status; - auto located = locate_sources(i, page.page_.snapshot_.entries_); + const auto& snapshot = page.page_.snapshot_; + auto located = + snapshot.encoded_sorted_set_ + ? locate_sources(i, snapshot.encoded_sorted_set_->entries()) + : locate_sources(i, snapshot.entries_); if (!located.ok()) co_return located.status(); } // The two resident doubles bound old-score candidates without reading @@ -1963,7 +1967,11 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( if (probed_pages.contains(i)) continue; auto page = co_await read_owned_page(i); if (!page.ok()) co_return page.status(); - auto located = locate_sources(i, page->page_.snapshot_.entries_); + const auto& snapshot = page->page_.snapshot_; + auto located = + snapshot.encoded_sorted_set_ + ? locate_sources(i, snapshot.encoded_sorted_set_->entries()) + : locate_sources(i, snapshot.entries_); if (!located.ok()) co_return located.status(); // Equal-score scans may touch unrelated pages. Retain only pages with // requested members, so a point update never caches the whole set. @@ -2029,7 +2037,12 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( if (!page.ok()) co_return page.status(); boundary.emplace(std::move(*page)); } - if (member > boundary->page_.snapshot_.entries_.back().value_) break; + const auto& snapshot = boundary->page_.snapshot_; + const std::string_view last_member = + snapshot.encoded_sorted_set_ + ? snapshot.encoded_sorted_set_->entries().back().value_ + : std::string_view(snapshot.entries_.back().value_); + if (member > last_member) break; } } state->destination_ = i; @@ -2083,7 +2096,7 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( auto page = co_await read_owned_page(i); if (!page.ok()) co_return page.status(); loaded.emplace(i, std::move(page->page_.snapshot_)); - // working_admission now covers the transferred strings. + // working_admission now covers the transferred owned payload. } else { status = check_plan_read(i); if (!status.ok()) co_return status; @@ -2095,220 +2108,257 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( "Sorted Set plan cancelled by shutdown"); } auto page = co_await LoadOrderedGroupSnapshot( - store, partition, db_id, key, digest, object, metadata[i].id_); + store, partition, db_id, key, digest, object, metadata[i].id_, false, + nullptr, SortedSetMemberValidation::kVerifyUnique, point_write); if (!page.ok()) co_return page.status(); loaded.emplace(i, std::move(page->snapshot_)); } } - if (operation.kind_ == SortedSetOperationKind::kAdd && - operation.entries_.size() == 1 && pending.size() == 1 && - pending.front().second->before_) { - // A point score update owns the existing member already. Keep that - // allocation and the page's ordered subsequences instead of deleting the - // member, copying the request, and sorting the entire destination. Batch - // inputs retain the general path because repeated members apply in order. - const auto& [member, state] = pending.front(); - auto& source = loaded.at(state->source_).entries_; - const auto entry = - std::find_if(source.begin(), source.end(), - [&](const auto& at) { return at.value_ == member; }); - if (entry == source.end()) - co_return absl::DataLossError("missing Sorted Set point-update source"); - entry->score_ = *state->after_; - if (state->source_ == state->destination_) { - if (entry != source.begin() && OrderedEntryLess(*entry, *(entry - 1))) { - const auto position = - std::lower_bound(source.begin(), entry, *entry, OrderedEntryLess); - std::rotate(position, entry, entry + 1); - } else if (entry + 1 != source.end() && - OrderedEntryLess(*(entry + 1), *entry)) { - const auto position = - std::lower_bound(entry + 1, source.end(), *entry, OrderedEntryLess); - std::rotate(entry, entry + 1, position); + bool encoded_point = point_write; + if (encoded_point) { + const auto& [member, state] = *members.begin(); + for (const auto i : modified) { + auto edited = EditEncodedSortedSetGroup( + loaded.at(i), member, + i == state.source_ ? state.before_ : std::nullopt, + i == state.destination_ ? state.after_ : std::nullopt); + if (!edited.ok()) co_return edited.status(); + if (!*edited) { + encoded_point = false; + plan.writes_.clear(); + break; } - } else { - auto moved = std::move(*entry); - source.erase(entry); + plan.writes_.push_back(std::move(**edited)); + } + } + if (!encoded_point) { + LAVIK_FAULT_INJECT( + if (point_write && + LAVIK_FAULT_MATCHES("LAVIK_FAIL_ZSET_MUTABLE_PLAN_KEY", + key)) co_return absl:: + UnavailableError( + "injected mutable Sorted Set point plan failure");); + // Empty/overflowing pages need the ordinary retirement, redistribution and + // split planner. Materialize only on this uncommon path; no encoded page + // or physical lease escapes into the mutable topology machinery. + for (auto& [index, page] : loaded) MaterializeOrderedGroupEntries(page); + if (operation.kind_ == SortedSetOperationKind::kAdd && + operation.entries_.size() == 1 && pending.size() == 1 && + pending.front().second->before_) { + // A point score update owns the existing member already. Keep that + // allocation and the page's ordered subsequences instead of deleting the + // member, copying the request, and sorting the entire destination. Batch + // inputs retain the general path because repeated members apply in order. + const auto& [member, state] = pending.front(); + auto& source = loaded.at(state->source_).entries_; + const auto entry = + std::find_if(source.begin(), source.end(), + [&](const auto& at) { return at.value_ == member; }); + if (entry == source.end()) + co_return absl::DataLossError("missing Sorted Set point-update source"); + entry->score_ = *state->after_; + if (state->source_ == state->destination_) { + if (entry != source.begin() && OrderedEntryLess(*entry, *(entry - 1))) { + const auto position = + std::lower_bound(source.begin(), entry, *entry, OrderedEntryLess); + std::rotate(position, entry, entry + 1); + } else if (entry + 1 != source.end() && + OrderedEntryLess(*(entry + 1), *entry)) { + const auto position = std::lower_bound(entry + 1, source.end(), + *entry, OrderedEntryLess); + std::rotate(entry, entry + 1, position); + } + } else { + auto moved = std::move(*entry); + source.erase(entry); + auto& destination = loaded.at(state->destination_).entries_; + const auto position = std::lower_bound( + 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(), 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_; - std::erase_if(entries, [&](const auto& entry) { - const auto member = members.find(entry.value_); - return member != members.end() && member->second.touched_; - }); - } - for (const auto& [member, state] : pending) - loaded.at(state->destination_) - .entries_.push_back({std::string(member), *state->after_}); - - // Removal preserves the order of every surviving entry. - if (!pending.empty()) { + destination.begin(), destination.end(), entry, OrderedEntryLess); + destination.insert(position, std::move(entry)); + } else { for (const auto i : modified) { auto& entries = loaded.at(i).entries_; - std::sort(entries.begin(), entries.end(), OrderedEntryLess); + std::erase_if(entries, [&](const auto& entry) { + const auto member = members.find(entry.value_); + return member != members.end() && member->second.touched_; + }); } - } - } - // A score move can overflow its destination while freeing the adjacent - // source. Reuse both already admitted pages when they still fit together; - // allocating an extra id would force full-directory reconstruction. - for (const auto i : modified) { - if (!modified.contains(i + 1)) continue; - auto balanced = RebalanceSortedSetGroupPair(loaded.at(i), loaded.at(i + 1)); - if (!balanced.ok()) co_return balanced.status(); - } + for (const auto& [member, state] : pending) + loaded.at(state->destination_) + .entries_.push_back({std::string(member), *state->after_}); - // Repeated inserts before a full page otherwise split off one tiny successor - // per command. Reuse at most one unchanged neighbour before allocating an - // identity, using resident sizes to avoid speculative reads of full pages. - // selected already admits both immediate neighbours. A successful pair keeps - // its links and IDs, so it needs neither a wider interval nor more admission. - constexpr auto kPagePayloadBytes = - kCollectionGroupTargetBytes - kOrderedGroupHeaderBytes; - constexpr auto kPairPayloadBytes = 2 * kPagePayloadBytes; - auto bounded_payload_bytes = [](const OrderedGroupSnapshot& page) { - std::size_t bytes = 0; - for (const auto& entry : page.entries_) { - const auto entry_bytes = kOrderedEntryHeaderBytes + entry.value_.size(); - if (entry_bytes > kPairPayloadBytes - bytes) return kPairPayloadBytes + 1; - bytes += entry_bytes; - } - return bytes; - }; - absl::InlinedVector, 4> overflowing; - for (const auto i : modified) { - const auto bytes = bounded_payload_bytes(loaded.at(i)); - if (bytes > kPagePayloadBytes && bytes <= kPairPayloadBytes) - overflowing.emplace_back(i, bytes); - } - // Iterate only original overflows: adopted neighbours must fit the target - // after redistribution and cannot start an outward chain of extra reads. - for (const auto& [i, bytes] : overflowing) { - std::size_t neighbor = kNoPage; - for (const auto candidate : {i == 0 ? kNoPage : i - 1, - i + 1 == metadata.size() ? kNoPage : i + 1}) { - if (candidate == kNoPage || modified.contains(candidate)) continue; - const auto encoded = metadata[candidate].encoded_bytes_; - if (encoded < kOrderedGroupHeaderBytes || - encoded - kOrderedGroupHeaderBytes > kPairPayloadBytes - bytes) - continue; - if (neighbor == kNoPage || encoded < metadata[neighbor].encoded_bytes_) - neighbor = candidate; + // Removal preserves the order of every surviving entry. + if (!pending.empty()) { + for (const auto i : modified) { + auto& entries = loaded.at(i).entries_; + std::sort(entries.begin(), entries.end(), OrderedEntryLess); + } + } } - if (neighbor == kNoPage) continue; - if (!loaded.contains(neighbor)) { - status = check_plan_read(neighbor); - if (!status.ok()) co_return status; - auto page = co_await LoadOrderedGroupSnapshot( - store, partition, db_id, key, digest, object, metadata[neighbor].id_); - if (!page.ok()) co_return page.status(); - loaded.emplace(neighbor, std::move(page->snapshot_)); + // A score move can overflow its destination while freeing the adjacent + // source. Reuse both already admitted pages when they still fit together; + // allocating an extra id would force full-directory reconstruction. + for (const auto i : modified) { + if (!modified.contains(i + 1)) continue; + auto balanced = + RebalanceSortedSetGroupPair(loaded.at(i), loaded.at(i + 1)); + if (!balanced.ok()) co_return balanced.status(); } - auto balanced = - neighbor < i - ? RebalanceSortedSetGroupPair(loaded.at(neighbor), loaded.at(i)) - : RebalanceSortedSetGroupPair(loaded.at(i), loaded.at(neighbor)); - if (!balanced.ok()) co_return balanced.status(); - // Indivisible variable-sized entries can still prevent a legal cut. Keep - // that unchanged snapshot for a possible link rewrite after the split. - if (*balanced) modified.insert(neighbor); - } - struct Route { - std::uint64_t id_; - std::size_t source_; - std::size_t write_; - }; - std::vector route; - route.reserve(selected.size()); - auto next_id = directory.root().next_group_id_; - std::uint64_t page_count = metadata.size(); - // Original modified pages have their immediate neighbours in selected; - // redistributed neighbours retain their outer links. Process contiguous - // intervals independently: gaps need neither routing nor payload rebuilds. - for (auto selected_it = selected.begin(); selected_it != selected.end();) { - const auto begin = *selected_it++; - auto end = begin + 1; - while (selected_it != selected.end() && *selected_it == end) { - ++selected_it; - ++end; - } - route.clear(); - for (std::size_t i = begin; i < end; ++i) { - if (!modified.contains(i)) { - route.push_back({metadata[i].id_, i, kNoPage}); - continue; + // Repeated inserts before a full page otherwise split off one tiny + // successor per command. Reuse at most one unchanged neighbour before + // allocating an identity, using resident sizes to avoid speculative reads + // of full pages. selected already admits both immediate neighbours. A + // successful pair keeps its links and IDs, so it needs neither a wider + // interval nor more admission. + constexpr auto kPagePayloadBytes = + kCollectionGroupTargetBytes - kOrderedGroupHeaderBytes; + constexpr auto kPairPayloadBytes = 2 * kPagePayloadBytes; + auto bounded_payload_bytes = [](const OrderedGroupSnapshot& page) { + std::size_t bytes = 0; + for (const auto& entry : page.entries_) { + const auto entry_bytes = kOrderedEntryHeaderBytes + entry.value_.size(); + if (entry_bytes > kPairPayloadBytes - bytes) + return kPairPayloadBytes + 1; + bytes += entry_bytes; } - auto page = std::move(loaded.at(i)); - if (page.entries_.empty()) { - page.retired_ = true; - page.previous_ = page.next_ = 0; - plan.writes_.push_back(std::move(page)); - continue; + return bytes; + }; + absl::InlinedVector, 4> overflowing; + for (const auto i : modified) { + const auto bytes = bounded_payload_bytes(loaded.at(i)); + if (bytes > kPagePayloadBytes && bytes <= kPairPayloadBytes) + overflowing.emplace_back(i, bytes); + } + // Iterate only original overflows: adopted neighbours must fit the target + // after redistribution and cannot start an outward chain of extra reads. + for (const auto& [i, bytes] : overflowing) { + std::size_t neighbor = kNoPage; + for (const auto candidate : + {i == 0 ? kNoPage : i - 1, + i + 1 == metadata.size() ? kNoPage : i + 1}) { + if (candidate == kNoPage || modified.contains(candidate)) continue; + const auto encoded = metadata[candidate].encoded_bytes_; + if (encoded < kOrderedGroupHeaderBytes || + encoded - kOrderedGroupHeaderBytes > kPairPayloadBytes - bytes) + continue; + if (neighbor == kNoPage || encoded < metadata[neighbor].encoded_bytes_) + neighbor = candidate; } - 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_) { - route.push_back({part.id_, i, plan.writes_.size()}); - plan.writes_.push_back(std::move(part)); + if (neighbor == kNoPage) continue; + if (!loaded.contains(neighbor)) { + status = check_plan_read(neighbor); + if (!status.ok()) co_return status; + auto page = co_await LoadOrderedGroupSnapshot(store, partition, db_id, + key, digest, object, + metadata[neighbor].id_); + if (!page.ok()) co_return page.status(); + loaded.emplace(neighbor, std::move(page->snapshot_)); } + auto balanced = + neighbor < i + ? RebalanceSortedSetGroupPair(loaded.at(neighbor), loaded.at(i)) + : RebalanceSortedSetGroupPair(loaded.at(i), loaded.at(neighbor)); + if (!balanced.ok()) co_return balanced.status(); + // Indivisible variable-sized entries can still prevent a legal cut. Keep + // that unchanged snapshot for a possible link rewrite after the split. + if (*balanced) modified.insert(neighbor); } - // A non-global interval includes an unchanged boundary neighbour. The - // global empty result was handled above by deleting the complete key. - if (route.empty()) - co_return absl::DataLossError("empty Sorted Set selected interval"); - page_count = page_count - (end - begin) + route.size(); - for (std::size_t i = 0; i < route.size(); ++i) { - auto& at = route[i]; - const auto previous = - i == 0 ? metadata[begin].previous_ : route[i - 1].id_; - const auto following = - i + 1 == route.size() ? metadata[end - 1].next_ : route[i + 1].id_; - if (at.write_ == kNoPage) { - const auto& old = metadata[at.source_]; - if (old.previous_ == previous && old.next_ == following) continue; - at.write_ = plan.writes_.size(); - if (const auto cached = loaded.find(at.source_); - cached != loaded.end()) { - plan.writes_.push_back(std::move(cached->second)); - } else { - status = check_plan_read(at.source_); - if (!status.ok()) co_return status; - // Revalidate the captured logical view and current physical - // location through the ordinary loader, including GC retry checks. - auto neighbor = co_await LoadOrderedGroupSnapshot( - store, partition, db_id, key, digest, object, old.id_); - if (!neighbor.ok()) co_return neighbor.status(); - plan.writes_.push_back(std::move(neighbor->snapshot_)); + + struct Route { + std::uint64_t id_; + std::size_t source_; + std::size_t write_; + }; + std::vector route; + route.reserve(selected.size()); + auto next_id = directory.root().next_group_id_; + std::uint64_t page_count = metadata.size(); + // Original modified pages have their immediate neighbours in selected; + // redistributed neighbours retain their outer links. Process contiguous + // intervals independently: gaps need neither routing nor payload rebuilds. + for (auto selected_it = selected.begin(); selected_it != selected.end();) { + const auto begin = *selected_it++; + auto end = begin + 1; + while (selected_it != selected.end() && *selected_it == end) { + ++selected_it; + ++end; + } + route.clear(); + for (std::size_t i = begin; i < end; ++i) { + if (!modified.contains(i)) { + route.push_back({metadata[i].id_, i, kNoPage}); + continue; + } + auto page = std::move(loaded.at(i)); + if (page.entries_.empty()) { + page.retired_ = true; + page.previous_ = page.next_ = 0; + plan.writes_.push_back(std::move(page)); + continue; + } + 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_) { + route.push_back({part.id_, i, plan.writes_.size()}); + plan.writes_.push_back(std::move(part)); } } - plan.writes_[at.write_].previous_ = previous; - plan.writes_[at.write_].next_ = following; - } - if (begin == 0) plan.root_.first_group_ = route.front().id_; - if (end == metadata.size()) plan.root_.last_group_ = route.back().id_; + // A non-global interval includes an unchanged boundary neighbour. The + // global empty result was handled above by deleting the complete key. + if (route.empty()) + co_return absl::DataLossError("empty Sorted Set selected interval"); + page_count = page_count - (end - begin) + route.size(); + for (std::size_t i = 0; i < route.size(); ++i) { + auto& at = route[i]; + const auto previous = + i == 0 ? metadata[begin].previous_ : route[i - 1].id_; + const auto following = + i + 1 == route.size() ? metadata[end - 1].next_ : route[i + 1].id_; + if (at.write_ == kNoPage) { + const auto& old = metadata[at.source_]; + if (old.previous_ == previous && old.next_ == following) continue; + at.write_ = plan.writes_.size(); + if (const auto cached = loaded.find(at.source_); + cached != loaded.end()) { + plan.writes_.push_back(std::move(cached->second)); + } else { + status = check_plan_read(at.source_); + if (!status.ok()) co_return status; + // Revalidate the captured logical view and current physical + // location through the ordinary loader, including GC retry checks. + auto neighbor = co_await LoadOrderedGroupSnapshot( + store, partition, db_id, key, digest, object, old.id_); + if (!neighbor.ok()) co_return neighbor.status(); + plan.writes_.push_back(std::move(neighbor->snapshot_)); + } + } + plan.writes_[at.write_].previous_ = previous; + plan.writes_[at.write_].next_ = following; + } + if (begin == 0) plan.root_.first_group_ = route.front().id_; + if (end == metadata.size()) plan.root_.last_group_ = route.back().id_; + } + if (page_count == 0 || + page_count > std::numeric_limits::max()) + co_return absl::DataLossError("invalid Sorted Set resulting page count"); + plan.root_.group_count_ = page_count; + plan.root_.next_group_id_ = next_id; } - if (page_count == 0 || page_count > std::numeric_limits::max()) - co_return absl::DataLossError("invalid Sorted Set resulting page count"); - plan.root_.group_count_ = page_count; - plan.root_.next_group_id_ = next_id; // A failed redistribution may have inspected a neighbour whose link stayed // unchanged. Keep only the owned after-images through member preparation and // publication; no later phase needs those speculative decoded bytes. diff --git a/src/storage/engine/impl.h b/src/storage/engine/impl.h index 85bcc6c9e..792ca088b 100644 --- a/src/storage/engine/impl.h +++ b/src/storage/engine/impl.h @@ -2572,7 +2572,7 @@ class StorageEngine::Impl { const MutationPrecondition* mutation_precondition = nullptr, PreparedOrderedMutation* prepared = nullptr); // Command-local checked ordered pages. Retain only pages containing requested - // members (or a destination boundary), with admission beside owned strings. + // members (or a destination boundary), with admission beside owned payloads. // Pop selection passes this cache into deletion under the same key intent; // it is never stored in the resident object or reused by another command. struct SortedSetOrderedProbe { @@ -4440,13 +4440,16 @@ class StorageEngine::Impl { // caller-owned limiter accounts against the refreshed physical owner until // decoding finishes; it must outlive this task, including cancellation. // Only read-only consumers may trust stored Sorted Set member uniqueness. + // Point writers may retain a checked encoded block instead of member strings; + // that representation always checks uniqueness and also releases the lease. Task> LoadOrderedGroupSnapshot( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::uint64_t id, bool pinned = false, GroupReadLimiter* limiter = nullptr, SortedSetMemberValidation validation = - SortedSetMemberValidation::kVerifyUnique); + SortedSetMemberValidation::kVerifyUnique, + bool encoded_sorted_set = false); Task> LoadSortedSetPage( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, diff --git a/tests/grouped/collection_test.cpp b/tests/grouped/collection_test.cpp index d0bdb6a80..07b7ceb83 100644 --- a/tests/grouped/collection_test.cpp +++ b/tests/grouped/collection_test.cpp @@ -284,6 +284,144 @@ TEST(GroupedCollectionTest, SortedSetViewsMatchOwnedDecoderCorruptionChecks) { EXPECT_FALSE(DecodeSortedSetGroupViews(*list).ok()); } +TEST(GroupedCollectionTest, EncodedSortedSetPageOwnsBytesAcrossCopyAndMove) { + auto original = Page(7, 40, OrderedCollectionKind::kSortedSet); + original.previous_ = 3; + original.next_ = 9; + for (auto& entry : original.entries_) entry.value_.append(32, '\0'); + auto wire = EncodeOrderedGroup(original); + ASSERT_TRUE(wire.ok()); + auto decoded = DecodeEncodedSortedSetGroup(*wire); + ASSERT_TRUE(decoded.ok()) << decoded.status(); + EXPECT_TRUE(decoded->entries_.empty()); + ASSERT_TRUE(decoded->encoded_sorted_set_); + auto copied = *decoded; + auto moved = std::move(copied); + decoded = absl::DataLossError("release first owner"); + wire->assign(wire->size(), 'x'); + EXPECT_EQ(OrderedGroupSize(moved), original.entries_.size()); + auto encoder = OrderedGroupEncoder::Create(moved); + ASSERT_TRUE(encoder.ok()); + ASSERT_TRUE(encoder->Next()); // The regenerated envelope. + auto body = encoder->Next(); + ASSERT_TRUE(body); + EXPECT_EQ(body->data(), moved.encoded_sorted_set_->bytes().data()); + EXPECT_FALSE(encoder->Next()); // The complete body is one borrowed span. + auto roundtrip = EncodeOrderedGroup(moved); + ASSERT_TRUE(roundtrip.ok()); + EXPECT_EQ(*roundtrip, *EncodeOrderedGroup(original)); + MaterializeOrderedGroupEntries(moved); + EXPECT_FALSE(moved.encoded_sorted_set_); + EXPECT_EQ(moved.entries_, original.entries_); + auto ambiguous = *DecodeEncodedSortedSetGroup(*roundtrip); + ambiguous.entries_.push_back({"unvalidated", 0}); + EXPECT_FALSE(OrderedGroupEncoder::Create(ambiguous).ok()); +} + +TEST(GroupedCollectionTest, EncodedSortedSetPointEditsMatchOwnedImages) { + auto original = Page(7, 40, OrderedCollectionKind::kSortedSet); + original.previous_ = 3; + original.next_ = 9; + for (auto& entry : original.entries_) entry.value_.append(32, '\0'); + const auto wire = EncodeOrderedGroup(original); + ASSERT_TRUE(wire.ok()); + const auto decoded = DecodeEncodedSortedSetGroup(*wire); + ASSERT_TRUE(decoded.ok()); + for (std::size_t index = 0; index < original.entries_.size(); ++index) { + for (const double score : + {-std::numeric_limits::infinity(), -0.0, 0.0, 4.0, 19.25, 39.0, + std::numeric_limits::infinity()}) { + auto expected = original; + const auto before = expected.entries_[index].score_; + const auto member = expected.entries_[index].value_; + expected.entries_[index].score_ = score; + std::sort(expected.entries_.begin(), expected.entries_.end(), + OrderedEntryLess); + auto edited = EditEncodedSortedSetGroup(*decoded, member, before, score); + ASSERT_TRUE(edited.ok()) << edited.status(); + ASSERT_TRUE(*edited); + EXPECT_TRUE((*edited)->entries_.empty()); + EXPECT_EQ(*EncodeOrderedGroup(**edited), *EncodeOrderedGroup(expected)); + } + auto expected = original; + const auto member = expected.entries_[index].value_; + const auto score = expected.entries_[index].score_; + expected.entries_.erase(expected.entries_.begin() + index); + auto erased = + EditEncodedSortedSetGroup(*decoded, member, score, std::nullopt); + ASSERT_TRUE(erased.ok()); + ASSERT_TRUE(*erased); + EXPECT_EQ(*EncodeOrderedGroup(**erased), *EncodeOrderedGroup(expected)); + auto restored = + EditEncodedSortedSetGroup(**erased, member, std::nullopt, score); + ASSERT_TRUE(restored.ok()); + ASSERT_TRUE(*restored); + EXPECT_EQ(*EncodeOrderedGroup(**restored), *wire); + } + // The immutable predecessor must survive every candidate edit unchanged. + EXPECT_EQ(*EncodeOrderedGroup(*decoded), *wire); +} + +TEST(GroupedCollectionTest, EncodedSortedSetPointRejectsBadEditsAndFallsBack) { + const auto wire = + EncodeOrderedGroup(Page(1, 3, OrderedCollectionKind::kSortedSet)); + ASSERT_TRUE(wire.ok()); + const auto page = DecodeEncodedSortedSetGroup(*wire); + ASSERT_TRUE(page.ok()); + EXPECT_FALSE( + EditEncodedSortedSetGroup(*page, "item-0", std::nullopt, 10).ok()); + EXPECT_FALSE(EditEncodedSortedSetGroup(*page, "item-0", 9, 10).ok()); + EXPECT_FALSE(EditEncodedSortedSetGroup(*page, "missing", 0, 10).ok()); + EXPECT_FALSE( + EditEncodedSortedSetGroup(*page, "missing", std::nullopt, std::nullopt) + .ok()); + EXPECT_FALSE(EditEncodedSortedSetGroup( + *page, "item-0", 0, std::numeric_limits::quiet_NaN()) + .ok()); + auto overflow = + EditEncodedSortedSetGroup(*page, "new", std::nullopt, 10, wire->size()); + ASSERT_TRUE(overflow.ok()); + EXPECT_FALSE(*overflow); + auto single_wire = + EncodeOrderedGroup(Page(1, 1, OrderedCollectionKind::kSortedSet)); + ASSERT_TRUE(single_wire.ok()); + auto single = DecodeEncodedSortedSetGroup(*single_wire); + ASSERT_TRUE(single.ok()); + auto empty = EditEncodedSortedSetGroup(*single, "item-0", 0, std::nullopt); + ASSERT_TRUE(empty.ok()); + EXPECT_FALSE(*empty); + auto invalid = *page; + invalid.retired_ = true; + EXPECT_FALSE(OrderedGroupEncoder::Create(invalid).ok()); + invalid = *page; + invalid.kind_ = OrderedCollectionKind::kList; + EXPECT_FALSE(OrderedGroupEncoder::Create(invalid).ok()); + invalid = *page; + invalid.next_ = invalid.id_; + EXPECT_FALSE(OrderedGroupEncoder::Create(invalid).ok()); + EXPECT_EQ(*EncodeOrderedGroup(*page), *wire); +} + +TEST(GroupedCollectionTest, EncodedSortedSetDecoderKeepsFullValidation) { + auto wire = EncodeOrderedGroup(Page(1, 3, OrderedCollectionKind::kSortedSet)); + ASSERT_TRUE(wire.ok()); + for (std::size_t offset = 0; offset < wire->size(); ++offset) { + for (const unsigned mask : {1u, 0x80u, 0xffu}) { + auto bad = *wire; + bad[offset] ^= static_cast(mask); + EXPECT_EQ(DecodeOrderedGroup(bad).ok(), + DecodeEncodedSortedSetGroup(bad).ok()) + << offset; + } + EXPECT_FALSE(DecodeEncodedSortedSetGroup(wire->substr(0, offset)).ok()); + } + wire->replace(wire->find("item-2"), 6, "item-0"); + EXPECT_TRUE(absl::IsDataLoss(DecodeEncodedSortedSetGroup(*wire).status())); + const auto list = EncodeOrderedGroup(Page()); + ASSERT_TRUE(list.ok()); + EXPECT_FALSE(DecodeEncodedSortedSetGroup(*list).ok()); +} + TEST(GroupedCollectionTest, SortedSetStoredReadsTrustMemberUniqueness) { auto page = Page(1, 3, OrderedCollectionKind::kSortedSet); auto encoded = EncodeOrderedGroup(page); diff --git a/tests/grouped/zset_write_e2e_test.cpp b/tests/grouped/zset_write_e2e_test.cpp index 4ee09a9dd..fbbc1782b 100644 --- a/tests/grouped/zset_write_e2e_test.cpp +++ b/tests/grouped/zset_write_e2e_test.cpp @@ -239,6 +239,58 @@ std::optional FirstDenseOrderedPage( return first; } +TEST(GroupedSortedSetWriteE2e, EncodedPointEditsPreserveExpiryAndDemote) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires mutable point-plan failure injection"; +#endif + PrivateDisk disk; + const std::string key = "encoded-point-demotion"; + std::map expected; + { + Server server(disk, 1); + Client client(server.port()); + ASSERT_EQ(client.Command(DenseZSetSeed(key, &expected)).text_, "1024"); + ASSERT_EQ(client.Command({"PEXPIRE", key, "3600000"}).text_, "1"); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + ASSERT_EQ(disk.LatestRootGrouped(key), true); + { + ScopedEnvironment fault("LAVIK_FAIL_ZSET_MUTABLE_PLAN_KEY", key.c_str()); + Server server(disk, 3); + Client client(server.port()); + const auto low = DenseZSetMember(10); + ASSERT_EQ(client.Command({"ZINCRBY", key, "0.25", low}).text_, "10.25"); + expected[low] = 10.25; + // The last page has spare space: this move edits two encoded pages while + // retaining every page ID/link, and must not enter the mutable planner. + ASSERT_EQ(client.Command({"ZADD", key, "CH", "2000", low}).text_, "1"); + expected[low] = 2000; + const auto erased = DenseZSetMember(12); + ASSERT_EQ(client.Command({"ZREM", key, erased}).text_, "1"); + expected.erase(erased); + ASSERT_EQ(client.Command({"ZADD", key, "12", erased}).text_, "1"); + expected[erased] = 12; + // Walk a coprime permutation so no ordered page becomes empty. The final + // point deletion crosses the aggregate demotion threshold with encoded + // after-images, exercising the publication path's compact materialization. + for (unsigned i = 0; i < 854; ++i) { + const auto member = DenseZSetMember((i * 17) % 1024); + ASSERT_EQ(client.Command({"ZREM", key, member}).text_, "1") << i; + expected.erase(member); + } + ExpectDenseZSet(client, key, expected); + EXPECT_GT(std::stoll(client.Command({"PTTL", key}).text_), 0); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + ASSERT_EQ(disk.LatestRootGrouped(key), false); + Server recovered(disk, 2); + Client client(recovered.port()); + ExpectDenseZSet(client, key, expected); + EXPECT_GT(std::stoll(client.Command({"PTTL", key}).text_), 0); +} + TEST(GroupedSortedSetWriteE2e, DensePointSourceAvoidsSecondPlanRead) { #if !LAVIK_TEST_FAULTS_AVAILABLE GTEST_SKIP() << "requires ordered plan read failure injection";