perf: edit grouped ZSet pages without per-member allocations

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

include/lavik/storage/detail/grouped/collection.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <array>
+#include <cassert>
 #include <cstddef>
 #include <cstdint>
 #include <optional>
@@ -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<OrderedCollectionEntry> 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<const OrderedCollectionEntryView> entries() const noexcept {
+    return entries_;
+  }
+  std::string_view bytes() const noexcept { return *bytes_; }
+
+ private:
+  friend absl::StatusOr<OrderedGroupSnapshot> DecodeEncodedSortedSetGroup(
+      std::string_view bytes);
+  friend absl::StatusOr<std::optional<OrderedGroupSnapshot>>
+  EditEncodedSortedSetGroup(const OrderedGroupSnapshot& page,
+                            std::string_view member,
+                            std::optional<double> before,
+                            std::optional<double> after,
+                            std::size_t target_bytes);
+  EncodedSortedSetEntries(LocalSharedPtr<const std::string> bytes,
+                          std::vector<OrderedCollectionEntryView> entries)
+      : bytes_(std::move(bytes)), entries_(std::move(entries)) {}
+  // Declare the owner first so every borrowing view dies before its bytes.
+  LocalSharedPtr<const std::string> bytes_;
+  std::vector<OrderedCollectionEntryView> 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<CheckedSortedSetEntries> 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<EncodedSortedSetEntries> 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<const OrderedCollectionEntry> 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<OrderedGroupSnapshot> 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<OrderedGroupSnapshot> 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<std::optional<OrderedGroupSnapshot>> EditEncodedSortedSetGroup(
+    const OrderedGroupSnapshot& page, std::string_view member,
+    std::optional<double> before, std::optional<double> 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.

src/storage/engine/grouped/collection.cpp

@@ -143,23 +143,36 @@ absl::StatusOr<std::size_t> 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<const OrderedCollectionEntry>{}
+                           : 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<std::uint32_t>::max() ||
+      count > std::numeric_limits<std::uint32_t>::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<std::string_view> 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<OrderedGroupSnapshot> DecodeOrderedGroup(
   return group;
 }
 
+absl::StatusOr<OrderedGroupSnapshot> 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<const std::string>(
+      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<std::optional<OrderedGroupSnapshot>> EditEncodedSortedSetGroup(
+    const OrderedGroupSnapshot& page, std::string_view member,
+    std::optional<double> before, std::optional<double> 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<std::size_t>(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<std::string>();
+  output->reserve(size);
+  auto append_member = [&] {
+    std::array<char, kEntryHeaderBytes> header{};
+    Store(header, 0, member.size(), 4);
+    Store(header, 4, std::bit_cast<std::uint64_t>(*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<OrderedCollectionEntryView> 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<double>(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<std::vector<std::string>> 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<bool> 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<OrderedGroupSplit> 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<OrderedCollectionMutationPlan> 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");