Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 52 additions & 1 deletion include/lavik/storage/detail/grouped/collection.h
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
#include <span>
#include <string>
#include <string_view>
#include <utility>
#include <variant>
#include <vector>

Expand Down Expand Up @@ -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<const OrderedCollectionEntry> entries() const noexcept {
return entries_;
}

private:
friend absl::StatusOr<OrderedGroupSplit> SplitCheckedSortedSetGroup(
OrderedGroupSnapshot group, std::uint64_t next_group_id,
std::size_t target_bytes);
explicit CheckedSortedSetEntries(std::vector<OrderedCollectionEntry> entries)
: entries_(std::move(entries)) {}

std::vector<OrderedCollectionEntry> entries_;
};

struct OrderedCollectionRoot {
OrderedCollectionKind kind_ = OrderedCollectionKind::kList;
std::uint64_t incarnation_ = 0;
Expand Down Expand Up @@ -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<std::string_view> 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<CheckedSortedSetEntries> 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<const OrderedCollectionEntry> 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(
Expand All @@ -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;
Expand Down Expand Up @@ -599,6 +640,16 @@ absl::StatusOr<OrderedGroupSplit> 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<OrderedGroupSplit> SplitCheckedSortedSetGroup(
OrderedGroupSnapshot group, std::uint64_t next_group_id,
std::size_t target_bytes = kCollectionGroupTargetBytes);

struct LoadedOrderedGroup {
std::uint64_t sequence_ = 0;
OrderedGroupSnapshot snapshot_;
Expand Down
58 changes: 46 additions & 12 deletions src/storage/engine/grouped/collection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -143,19 +143,30 @@ absl::StatusOr<std::size_t> 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<std::uint32_t>::max() ||
(group.retired_ ? (!group.entries_.empty() || borrowed_string ||
entries.size() > std::numeric_limits<std::uint32_t>::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);
Expand All @@ -168,7 +179,7 @@ absl::StatusOr<std::size_t> 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) {
Expand Down Expand Up @@ -417,8 +428,9 @@ std::optional<std::string_view> 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);
Expand Down Expand Up @@ -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> OrderedGroupDirectory::Recover(
Expand Down Expand Up @@ -1631,6 +1644,7 @@ absl::StatusOr<bool> 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_ ||
Expand Down Expand Up @@ -1720,7 +1734,7 @@ 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.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
Expand Down Expand Up @@ -1816,6 +1830,26 @@ absl::StatusOr<OrderedGroupSplit> SplitOrderedGroup(OrderedGroupSnapshot group,
false);
}

absl::StatusOr<OrderedGroupSplit> 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<OrderedCollectionMutationPlan> PlanOrderedCollectionSplice(
const OrderedGroupDirectory& directory,
std::vector<LoadedOrderedGroup> loaded_groups, std::uint64_t rank,
Expand Down Expand Up @@ -1854,8 +1888,8 @@ 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.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");
}
Expand Down
7 changes: 4 additions & 3 deletions src/storage/engine/grouped/hash.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -371,9 +371,10 @@ absl::StatusOr<std::optional<HashGroupPointProbe>> 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_)
Expand Down
12 changes: 7 additions & 5 deletions src/storage/engine/grouped/ordered_mutation.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -298,9 +298,10 @@ Task<absl::Status> 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);
Expand Down Expand Up @@ -450,6 +451,7 @@ Task<absl::Status> StorageEngine::Impl::CommitGroupedOrderedMutationLocked(
continue;
}
const auto& page = plan.writes_[i];
const auto entries = OrderedGroupEntries(page);
RecoveredOrderedGroup candidate{
{.incarnation_ = plan.root_.incarnation_,
.id_ = page.id_,
Expand All @@ -461,8 +463,8 @@ Task<absl::Status> 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_) {
Expand Down
52 changes: 51 additions & 1 deletion src/storage/engine/grouped/zset.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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::
Expand Down Expand Up @@ -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_;
Expand Down Expand Up @@ -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_) {
Expand Down
Loading