From 3c4b21eea8dd03b6af2e54c0394d080f99b0371a Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 6 Nov 2026 10:35:08 +0000 Subject: [PATCH 1/2] refactor(storage): assert grouped locking and publication contracts Treat invalid group acquisition order, upgrades and overlapping acquisitions as caller bugs. Return void after acquiring a group and remove Status handling from every caller. Assert protected Hash leaf versions and private plan, encoding and dependency-pin invariants while retaining normal concurrent invalidation, structural fallback, storage validation and resource failures. Keep standard IS/IX mode names and distinguish them from VLL scheduling intents in the public lock-mode comments. Update the group lock contract and cover misuse with Debug death tests. Validation: Debug build; 77 focused unit tests; 27 grouped E2E tests; NDEBUG syntax checks for all 10 changed production translation units; pre-commit. --- docs/architecture/09-grouped-collections.md | 9 +-- .../lavik/storage/detail/grouped/collection.h | 5 +- include/lavik/storage/detail/grouped/hash.h | 2 + include/lavik/storage/detail/grouped/lock.h | 10 ++- include/lavik/tx/fingerprint.h | 11 +-- src/storage/engine/grouped/collection.cpp | 11 +-- src/storage/engine/grouped/hash.cpp | 14 ++-- src/storage/engine/grouped/hash_random.cpp | 3 +- src/storage/engine/grouped/lock.cpp | 38 +++++----- src/storage/engine/grouped/mutation.cpp | 23 ++++--- .../engine/grouped/ordered_mutation.cpp | 19 ++--- src/storage/engine/grouped/zset.cpp | 33 ++++----- src/storage/engine/grouped/zset_members.cpp | 5 +- src/storage/engine/hash_api.cpp | 3 +- src/storage/engine/write.cpp | 26 ++++--- tests/grouped/hash_test.cpp | 48 +++++++++---- tests/grouped/lock_test.cpp | 69 ++++++++++++++----- tests/grouped/ordered_object_test.cpp | 24 +++++++ tests/storage_write_buffer_test.cpp | 7 +- 19 files changed, 219 insertions(+), 141 deletions(-) diff --git a/docs/architecture/09-grouped-collections.md b/docs/architecture/09-grouped-collections.md index 7524e2e0f..117e8facb 100644 --- a/docs/architecture/09-grouped-collections.md +++ b/docs/architecture/09-grouped-collections.md @@ -229,10 +229,11 @@ logical database, parent-key fingerprint, incarnation, graph and group identity. Callers retain parent-key protection and database admission for the entire lock-set lifetime, including group waits and IO. Each set acquires member groups before ordered groups and uses the total order of `GroupedRecordId` within each -graph. Earlier holds remain -owned while the next group waits; descending acquisitions and shared-to-exclusive -upgrades fail without releasing existing holds. Callers revalidate routing after -waiting and release the set before replanning or acquiring a stronger key lock. +graph. Earlier holds remain owned while the next group waits. Descending +acquisitions, shared-to-exclusive upgrades and simultaneous acquisitions on one +set violate the caller's lock-order contract and are asserted in Debug. Callers +revalidate routing after waiting and release the set before replanning or +acquiring a stronger key lock. Each group has its own FIFO. These queues are independent of the transaction shard's key queue, so a waiting operation cannot block an unrelated group needed diff --git a/include/lavik/storage/detail/grouped/collection.h b/include/lavik/storage/detail/grouped/collection.h index b3fa3c20e..1f8d25b2f 100644 --- a/include/lavik/storage/detail/grouped/collection.h +++ b/include/lavik/storage/detail/grouped/collection.h @@ -787,8 +787,9 @@ struct OrderedCollectionMutationPlan { // Merge an existing Sorted Set's non-structural page replacements into the // latest directory. The caller refreshes the base after locking and protects // affected member/ordered groups and routing neighbours through publication. -// Their logical versions must stay unchanged (asserted in Debug); physical -// relocation is allowed. The owner must not yield between merge and +// Their logical versions must stay unchanged, and the private plan's counts +// must agree across both indexes (asserted in Debug). Physical relocation is +// allowed. The owner must not yield between merge and // publication. FailedPrecondition means the plan requires whole-key // serialization; Aborted means the current root is incompatible with the base. // The returned root has no publication revision yet. diff --git a/include/lavik/storage/detail/grouped/hash.h b/include/lavik/storage/detail/grouped/hash.h index 6efcacfc9..d130f94e1 100644 --- a/include/lavik/storage/detail/grouped/hash.h +++ b/include/lavik/storage/detail/grouped/hash.h @@ -379,6 +379,8 @@ struct HashGroupMutationPlan { // Merge topology-preserving leaf replacements into the latest root. The caller // holds parent IX and X on every changed leaf, or parent X. Unrelated leaves // may publish while pages are prepared; their counts and revisions survive. +// The base is captured after locking. Protected leaf versions and the plan's +// own count delta must agree; violations are asserted in Debug. // Structural edits and deletion of the final field require whole-key X. // The returned root has no revision; assign one at the publication boundary. absl::StatusOr MergeHashPublication( diff --git a/include/lavik/storage/detail/grouped/lock.h b/include/lavik/storage/detail/grouped/lock.h index cfcba3b3d..23cdf3535 100644 --- a/include/lavik/storage/detail/grouped/lock.h +++ b/include/lavik/storage/detail/grouped/lock.h @@ -24,7 +24,6 @@ #include "absl/container/inlined_vector.h" #include "absl/container/node_hash_map.h" -#include "absl/status/status.h" #include "lavik/storage/detail/grouped/hash.h" #include "lavik/tx/fingerprint.h" @@ -86,9 +85,9 @@ class GroupedLockTable { // Acquires one group, retaining all earlier groups across the wait and IO. // New identities must increase. An already-held identity is a no-op when - // its mode suffices. A descending acquisition or S-to-X upgrade returns - // FailedPrecondition without changing ownership; the caller must replan - // after releasing the set, or fall back to a whole-key X acquisition. + // its mode suffices. Descending acquisitions and S-to-X upgrades violate + // the caller's lock-order contract and are asserted in Debug. Replanning + // or whole-key fallback must release the set first. // Only one acquisition may be outstanding per set. After a wait, callers // must revalidate routing before accessing the group they intended to use. [[nodiscard]] Awaiter Acquire(GroupedLockId id, GroupedLockMode mode); @@ -127,7 +126,7 @@ class GroupedLockTable { bool await_ready(); void await_suspend(std::coroutine_handle<> handle) noexcept; - absl::Status await_resume() noexcept; + void await_resume() noexcept; private: friend class LockSet; @@ -138,7 +137,6 @@ class GroupedLockTable { LockSet& set_; GroupedLockId id_; GroupedLockMode mode_; - absl::Status status_; Entry* entry_ = nullptr; Awaiter* previous_ = nullptr; Awaiter* next_ = nullptr; diff --git a/include/lavik/tx/fingerprint.h b/include/lavik/tx/fingerprint.h index 7339f7fc3..391d40abf 100644 --- a/include/lavik/tx/fingerprint.h +++ b/include/lavik/tx/fingerprint.h @@ -39,10 +39,13 @@ struct LockFpIdentityHash { std::size_t operator()(LockFp fp) const noexcept { return fp; } }; -// S protects the complete key from writes. IS and IX allow partial readers -// and writers, whose callers must separately arbitrate sub-key conflicts. -// IS is compatible with IS, S and IX; S with S; IX with IX. X excludes every -// mode. Modes apply to queued intents and executing holds, including IO waits. +// S protects the complete key from writes. IS/IX are the standard +// multi-granularity intention modes: the parent permits shared/exclusive locks +// on sub-key groups, whose conflicts are arbitrated separately. These mode +// names are distinct from VLL scheduling intents, which every mode registers +// while queued or executing. IS is compatible with IS, S and IX; S with S; IX +// with IX. X excludes every mode. Modes apply to queued intents and executing +// holds, including IO waits. enum class LockMode : std::uint8_t { kShared, kExclusive, diff --git a/src/storage/engine/grouped/collection.cpp b/src/storage/engine/grouped/collection.cpp index 5dcc37a2c..f2196ca6e 100644 --- a/src/storage/engine/grouped/collection.cpp +++ b/src/storage/engine/grouped/collection.cpp @@ -1521,9 +1521,10 @@ absl::StatusOr MergeSortedSetPublication( std::span members) { const auto& before = base.root(); if (!SameSortedSetTopology(planned, before) || ordered.empty() || - planned.item_count_ == 0 || - planned.member_index_->field_count_ != planned.item_count_) + planned.item_count_ == 0) return absl::FailedPreconditionError("Sorted Set plan needs key X"); + assert(planned.member_index_->field_count_ == planned.item_count_ && + "Sorted Set plan counts disagree between indexes"); if (!SameSortedSetTopology(current.root(), before) || current.sequence() < base.sequence()) return absl::AbortedError("Sorted Set topology changed during preparation"); @@ -1572,9 +1573,9 @@ absl::StatusOr MergeSortedSetPublication( if (!members.empty()) { result.member_index_->field_count_ = result.item_count_; result.member_index_->revision_ = 0; - } else if (planned.item_count_ != before.item_count_) { - return absl::InvalidArgumentError( - "Sorted Set count changed without members"); + } else { + assert(planned.item_count_ == before.item_count_ && + "Sorted Set count changed without members"); } return result; } diff --git a/src/storage/engine/grouped/hash.cpp b/src/storage/engine/grouped/hash.cpp index 679abada5..fa00067b7 100644 --- a/src/storage/engine/grouped/hash.cpp +++ b/src/storage/engine/grouped/hash.cpp @@ -1000,14 +1000,18 @@ absl::StatusOr MergeHashPublication( if (!old || old->id_ != page.id_ || page.retired_ || page.incarnation_ != before.incarnation_) return absl::FailedPreconditionError("Hash page topology needs key X"); +#ifndef NDEBUG + // The base was captured after taking X on this leaf (or the whole key). + // Only unrelated leaves may publish while its private plan is prepared. const auto* now = current.groups().Get(page.id_.prefix_); - if (!now || now->id_ != old->id_ || now->sequence_ != old->sequence_) - return absl::AbortedError("protected Hash page changed"); + assert(now && now->id_ == old->id_ && now->sequence_ == old->sequence_ && + "Hash page changed while protected"); +#endif delta += static_cast(page.field_count()) - old->field_count_; } - if (static_cast(before.field_count_) + delta != - planned.field_count_) - return absl::InvalidArgumentError("Hash plan count disagrees with pages"); + assert(static_cast(before.field_count_) + delta == + planned.field_count_ && + "Hash plan count disagrees with pages"); const auto count = static_cast(current.root().field_count_) + delta; if (count <= 0 || count > UINT32_MAX) diff --git a/src/storage/engine/grouped/hash_random.cpp b/src/storage/engine/grouped/hash_random.cpp index fd1b35d9d..d827400f6 100644 --- a/src/storage/engine/grouped/hash_random.cpp +++ b/src/storage/engine/grouped/hash_random.cpp @@ -123,10 +123,9 @@ StorageEngine::Impl::ExecuteGroupedHashRandomLocked( for (const auto& selection : selected) ids.push_back(selection.id_); std::sort(ids.begin(), ids.end()); for (const auto id : ids) { - const auto locked = co_await holds->Acquire( + co_await holds->Acquire( {GroupedLockIndex::kMember, id}, pop ? GroupedLockMode::kExclusive : GroupedLockMode::kShared); - if (!locked.ok()) co_return locked; } const auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); diff --git a/src/storage/engine/grouped/lock.cpp b/src/storage/engine/grouped/lock.cpp index daf484dd5..49a9dfda6 100644 --- a/src/storage/engine/grouped/lock.cpp +++ b/src/storage/engine/grouped/lock.cpp @@ -67,31 +67,28 @@ bool GroupedLockTable::Entry::CanHold(GroupedLockMode mode) const noexcept { bool GroupedLockTable::Awaiter::await_ready() { assert(!started_); started_ = true; - if (set_.pending_ != nullptr) { - status_ = absl::FailedPreconditionError( - "a group acquisition is already pending for this lock set"); - return true; - } - if ((mode_ != GroupedLockMode::kShared && - mode_ != GroupedLockMode::kExclusive) || - (id_.index_ != GroupedLockIndex::kMember && - id_.index_ != GroupedLockIndex::kOrdered)) { - status_ = absl::InvalidArgumentError("invalid group lock mode or index"); - return true; - } + assert(set_.pending_ == nullptr && + "a group acquisition is already pending for this lock set"); + assert((mode_ == GroupedLockMode::kShared || + mode_ == GroupedLockMode::kExclusive) && + "invalid group lock mode"); + assert((id_.index_ == GroupedLockIndex::kMember || + id_.index_ == GroupedLockIndex::kOrdered) && + "invalid group lock index"); if (!set_.held_.empty() && id_ <= set_.held_.back().entry_->id_) { +#ifndef NDEBUG + // A valid request can only revisit an already-held identity here. Keep + // the diagnostic lookup out of Release's idempotent acquisition path. const auto it = std::lower_bound( set_.held_.begin(), set_.held_.end(), id_, [](const LockSet::Held& held, const GroupedLockId& id) { return held.entry_->id_ < id; }); - if (it == set_.held_.end() || it->entry_->id_ != id_) { - status_ = absl::FailedPreconditionError( - "group locks must be acquired in increasing order"); - } else if (it->mode_ != mode_ && it->mode_ != GroupedLockMode::kExclusive) { - status_ = absl::FailedPreconditionError( - "group lock upgrade requires releasing and replanning the lock set"); - } + assert(it != set_.held_.end() && it->entry_->id_ == id_ && + "group locks must be acquired in increasing order"); + assert((it->mode_ == mode_ || it->mode_ == GroupedLockMode::kExclusive) && + "group lock upgrade requires releasing and replanning the lock set"); +#endif return true; } @@ -127,13 +124,12 @@ void GroupedLockTable::Awaiter::await_suspend( queued_ = true; } -absl::Status GroupedLockTable::Awaiter::await_resume() noexcept { +void GroupedLockTable::Awaiter::await_resume() noexcept { if (entry_ != nullptr) { assert(granted_ && set_.pending_ == this); set_.pending_ = nullptr; entry_ = nullptr; } - return std::move(status_); } GroupedLockTable::Awaiter::~Awaiter() { diff --git a/src/storage/engine/grouped/mutation.cpp b/src/storage/engine/grouped/mutation.cpp index 3d31d0bec..dd69bdc4b 100644 --- a/src/storage/engine/grouped/mutation.cpp +++ b/src/storage/engine/grouped/mutation.cpp @@ -147,8 +147,8 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( *retry_exclusive = true; return absl::AbortedError("Hash plan requires key X"); }; - if (prepared_publication && (!previous || !prepared || tx)) - co_return needs_x(); + assert((!prepared_publication || (previous && prepared && !tx)) && + "group-locked publication requires a prepared standalone update"); if (!diagnostics && tx) diagnostics = tx->diagnostics_; if (field_count == 0 || field_count > std::numeric_limits::max() || @@ -165,12 +165,15 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( changed_groups, field_count); if (!plan.ok()) co_return plan.status(); if (prepared != nullptr) { - if (plan->root_.field_count_ != field_count || - (previous - ? (plan->expected_sequence_ != previous->directory().sequence() || - plan->root_.incarnation_ != previous->incarnation()) - : plan->expected_sequence_ != 0)) - co_return absl::AbortedError("prepared Hash mutation is stale"); + // This compares two command-owned inputs, not the live root that may + // change during IO. The caller must pass the predecessor it planned from. + assert(plan->root_.field_count_ == field_count && + "prepared Hash mutation count disagrees with the command"); + assert((previous ? (plan->expected_sequence_ == + previous->directory().sequence() && + plan->root_.incarnation_ == previous->incarnation()) + : plan->expected_sequence_ == 0) && + "prepared Hash mutation has the wrong predecessor"); } const auto planned_root = plan->root_; if (prepared_publication && @@ -602,8 +605,8 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( if (!encoded.ok()) return encoded.status(); // AppendLocked retains a view of this fixed-width allocation across // resource waits. Update its bytes without invalidating that view. - if (encoded->size() != root_payload->size()) - return absl::InternalError("Hash root encoded size changed"); + assert(encoded->size() == root_payload->size() && + "Hash root encoded size changed"); std::memcpy(root_payload->data(), encoded->data(), encoded->size()); plan->root_ = *merged; prepared_batch.revision_ = revision; diff --git a/src/storage/engine/grouped/ordered_mutation.cpp b/src/storage/engine/grouped/ordered_mutation.cpp index a62a4502b..d83287c29 100644 --- a/src/storage/engine/grouped/ordered_mutation.cpp +++ b/src/storage/engine/grouped/ordered_mutation.cpp @@ -201,12 +201,15 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( if (!member_mutation.ok()) co_return member_mutation.status(); auto& member_plan = member_mutation->plan_; if (value_type == ValueType::kSortedSet) { - if (!previous && member_plan.expected_sequence_ != 0) - co_return absl::AbortedError("prepared member creation is stale"); - if ((previous && - member_plan.root_.incarnation_ != plan.root_.incarnation_) || - member_plan.root_.field_count_ != plan.root_.item_count_) - co_return absl::AbortedError("prepared member index is stale"); + // Both private plans describe the same edit; publication separately + // revalidates their predecessor against current population and page state. + assert((previous || member_plan.expected_sequence_ == 0) && + "prepared member creation has a predecessor"); + assert((!previous || + member_plan.root_.incarnation_ == plan.root_.incarnation_) && + "prepared member index has the wrong incarnation"); + assert(member_plan.root_.field_count_ == plan.root_.item_count_ && + "prepared member index count disagrees with ordered plan"); } auto planned_root = plan.root_; if (value_type == ValueType::kSortedSet) @@ -668,8 +671,8 @@ Task StorageEngine::Impl::CommitGroupedOrderedMutationLocked( if (!encoded.ok()) return encoded.status(); // AppendLocked borrows this string's allocation through capacity waits. // ZSET roots have fixed width; overwrite without invalidating its view. - if (encoded->size() != root_payload->size()) - return absl::InternalError("prepared root changed encoded size"); + assert(encoded->size() == root_payload->size() && + "prepared root changed encoded size"); std::memcpy(root_payload->data(), encoded->data(), encoded->size()); plan.root_ = std::move(*merged); prepared_batch.revision_ = revision; diff --git a/src/storage/engine/grouped/zset.cpp b/src/storage/engine/grouped/zset.cpp index 8a085f8b8..177cec45c 100644 --- a/src/storage/engine/grouped/zset.cpp +++ b/src/storage/engine/grouped/zset.cpp @@ -1190,9 +1190,8 @@ StorageEngine::Impl::TryConcurrentSortedSetRead( {db_id, tx::FingerprintOf(digest), object->incarnation()}); unlock.Unlock(); for (const auto id : groups) { - const auto locked = co_await holds.Acquire({GroupedLockIndex::kMember, id}, - GroupedLockMode::kShared); - if (!locked.ok()) co_return locked; + co_await holds.Acquire({GroupedLockIndex::kMember, id}, + GroupedLockMode::kShared); } // Acquiring a later group can yield to writers. Capture the common logical // cut only after all holds exist; never combine scores read before and after @@ -1262,9 +1261,8 @@ StorageEngine::Impl::TryConcurrentSortedSetPageRead( const auto* route = object->directory().Find(operation.members_.front()); if (!route) co_return absl::DataLossError("missing member prefix route"); const auto id = route->id_; - auto locked = co_await holds.Acquire({GroupedLockIndex::kMember, id}, - GroupedLockMode::kShared); - if (!locked.ok()) co_return locked; + co_await holds.Acquire({GroupedLockIndex::kMember, id}, + GroupedLockMode::kShared); // A waiting reader must use the last holder's new leaf, not the revision // observed before acquiring S. IS keeps the prefix topology unchanged. auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); @@ -1428,8 +1426,7 @@ StorageEngine::Impl::TryConcurrentSortedSetPageRead( std::sort(groups.begin(), groups.end()); groups.erase(std::unique(groups.begin(), groups.end()), groups.end()); for (const auto id : groups) { - const auto locked = co_await holds.Acquire(id, GroupedLockMode::kShared); - if (!locked.ok()) co_return locked; + co_await holds.Acquire(id, GroupedLockMode::kShared); } auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); if (!current || current->incarnation() != object->incarnation() || @@ -1524,9 +1521,8 @@ StorageEngine::Impl::TryConcurrentSortedSetPoint( store.group_locks_ = std::make_unique(*store.worker_); auto holds = store.group_locks_->MakeLockSet( {db_id, tx::FingerprintOf(digest), object->incarnation()}); - auto status = co_await holds.Acquire({GroupedLockIndex::kMember, member_id}, - GroupedLockMode::kExclusive); - if (!status.ok()) co_return status; + co_await holds.Acquire({GroupedLockIndex::kMember, member_id}, + GroupedLockMode::kExclusive); // A preceding holder may have updated this leaf while we waited. Rebase // before reading, rather than mistaking its revision for a failed read. auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); @@ -1542,7 +1538,7 @@ StorageEngine::Impl::TryConcurrentSortedSetPoint( auto input_admission = ReserveInputs(operation); if (!input_admission.ok()) co_return input_admission.status(); Members members; - status = PrepareMembers(operation, &members); + auto status = PrepareMembers(operation, &members); if (!status.ok()) co_return status; auto& state = members.begin()->second; SortedSetResult result; @@ -1676,9 +1672,7 @@ StorageEngine::Impl::TryConcurrentSortedSetPoint( // Merge both routes before locking: a destination can also be a source's // neighbour, and S-to-X upgrades would invalidate the global lock order. for (const auto& [id, mode] : selected) { - status = - co_await holds.Acquire({GroupedLockIndex::kOrdered, {id, 0}}, mode); - if (!status.ok()) co_return status; + co_await holds.Acquire({GroupedLockIndex::kOrdered, {id, 0}}, mode); } refreshed = RefreshLockedGroup(partition, db_id, key, object, member_id); if (!refreshed.ok()) co_return refreshed.status(); @@ -1839,9 +1833,8 @@ StorageEngine::Impl::TryConcurrentSortedSetBatch( auto holds = store.group_locks_->MakeLockSet( {db_id, tx::FingerprintOf(digest), object->incarnation()}); for (const auto id : member_groups) { - status = co_await holds.Acquire({GroupedLockIndex::kMember, id}, - GroupedLockMode::kExclusive); - if (!status.ok()) co_return status; + co_await holds.Acquire({GroupedLockIndex::kMember, id}, + GroupedLockMode::kExclusive); } auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); if (!current || current->incarnation() != object->incarnation() || @@ -2000,9 +1993,7 @@ StorageEngine::Impl::TryConcurrentSortedSetBatch( } } for (const auto& [id, mode] : ordered_groups) { - status = - co_await holds.Acquire({GroupedLockIndex::kOrdered, {id, 0}}, mode); - if (!status.ok()) co_return status; + co_await holds.Acquire({GroupedLockIndex::kOrdered, {id, 0}}, mode); } for (const auto id : member_groups) { auto refreshed = RefreshLockedGroup(partition, db_id, key, object, id); diff --git a/src/storage/engine/grouped/zset_members.cpp b/src/storage/engine/grouped/zset_members.cpp index 1ac926f39..68b5dd874 100644 --- a/src/storage/engine/grouped/zset_members.cpp +++ b/src/storage/engine/grouped/zset_members.cpp @@ -34,9 +34,8 @@ StorageEngine::Impl::PrepareSortedSetMembers( std::optional> checked_changes, SortedSetMemberProbe* probe, bool group_locked) { SortedSetMemberMutation result; - if (group_locked && !checked_changes) - co_return absl::InvalidArgumentError( - "group-locked member preparation requires checked changes"); + assert((!group_locked || checked_changes.has_value()) && + "group-locked member preparation requires checked changes"); SortedSetPageWorkBudget page_work; if (ordered.root_.kind_ != OrderedCollectionKind::kSortedSet) co_return result; diff --git a/src/storage/engine/hash_api.cpp b/src/storage/engine/hash_api.cpp index e2b16a7ce..b873d394e 100644 --- a/src/storage/engine/hash_api.cpp +++ b/src/storage/engine/hash_api.cpp @@ -202,10 +202,9 @@ StorageEngine::Impl::TryConcurrentHashLike( unlock.Unlock(); const bool write = IsHashLikeWrite(operation); for (const auto id : groups) { - const auto locked = co_await holds.Acquire( + co_await holds.Acquire( {GroupedLockIndex::kMember, id}, write ? GroupedLockMode::kExclusive : GroupedLockMode::kShared); - if (!locked.ok()) co_return locked; } // All leaves must be held before the adapter captures its common read view. // IS/IX excludes topology edits, but population replacement still needs the diff --git a/src/storage/engine/write.cpp b/src/storage/engine/write.cpp index 8a02800f9..d495b9c96 100644 --- a/src/storage/engine/write.cpp +++ b/src/storage/engine/write.cpp @@ -2912,9 +2912,10 @@ Task StorageEngine::Impl::WriteRecordLocked( const bool grouped_root = request.group_ != nullptr && !request.group_->auxiliary_; auto* batch = grouped_root ? request.group_->batch_ : nullptr; - if (batch && (request.for_defrag_ || request.external_ || !request.tx_ || - request.tx_->collect_undo_ || batch->records_.empty())) - co_return absl::InvalidArgumentError("invalid prepared group batch"); + assert((!batch || (!request.for_defrag_ && !request.external_ && + request.tx_ && !request.tx_->collect_undo_ && + !batch->records_.empty() && batch->prepare_)) && + "invalid prepared group batch"); if (request.inline_payload_ != nullptr && (request.external_ || !request.value_.empty() || (!auxiliary && @@ -3628,8 +3629,9 @@ Task StorageEngine::Impl::WriteRecordLocked( // The physical root can move independently of every unchanged child, // including during a TTL-only write. Re-enter the pre-stage owner-hop // protocol instead of borrowing the new block's transaction identity. - if (batch) - co_return absl::InternalError("prepared root lost its dependency pin"); + // Prepared batches checked this pin before prepare_ and cannot suspend + // afterwards, so losing it violates their publication contract. + assert(!batch && "prepared root lost its dependency pin"); goto acquire_active_stream; } if (request.tx_ != nullptr) { @@ -3650,9 +3652,8 @@ Task StorageEngine::Impl::WriteRecordLocked( // A physical-only relocation won during pinning/index resolution. // Nothing is staged yet; re-pin the new identity before re-entering // this non-suspending root publication section. - if (batch) - co_return absl::InternalError( - "prepared page lost its dependency pin"); + // Prepared batches have already passed their final pin check. + assert(!batch && "prepared page lost its dependency pin"); goto acquire_active_stream; } } @@ -3860,9 +3861,12 @@ Task StorageEngine::Impl::WriteRecordLocked( auxiliary_header.expire_at_ms_ = 0; auxiliary_header.lsn_ = *page_lsn; auxiliary_header.payload_checksum_ = record.no_crc_ ? 0 : Crc32c(output); - if (!EncodeRecordHeader(auxiliary_header, request.key_, - {staging.data_ + offset, header_bytes})) - co_return absl::InternalError("prepared group header encoding failed"); + // The admitted batch fixes every header field and output bound. Encoding + // must still run in Release; rejection indicates a batch-builder bug. + [[maybe_unused]] const bool header_encoded = + EncodeRecordHeader(auxiliary_header, request.key_, + {staging.data_ + offset, header_bytes}); + assert(header_encoded && "prepared group header encoding failed"); std::fill_n(staging.data_ + offset + header_bytes + output.size(), page.disk_bytes_ - header_bytes - output.size(), std::byte{0}); diff --git a/tests/grouped/hash_test.cpp b/tests/grouped/hash_test.cpp index 3c4abaf97..2f96217be 100644 --- a/tests/grouped/hash_test.cpp +++ b/tests/grouped/hash_test.cpp @@ -565,8 +565,7 @@ std::vector Candidates( return candidates; } -TEST(GroupedHashTest, - MergePublicationPreservesDisjointChangesAndRejectsConflicts) { +TEST(GroupedHashTest, MergePublicationPreservesDisjointChanges) { auto pages = GroupHashValue(Value(128), 17, Seed(), 1024); ASSERT_TRUE(pages.ok()); ASSERT_GT(pages->size(), 2); @@ -595,27 +594,48 @@ TEST(GroupedHashTest, EXPECT_EQ(merged->revision_, 0); EXPECT_EQ(merged->group_count_, base->root().group_count_); - auto overlap = records.front(); - overlap.sequence_ = overlap.lsn_ = 2; - auto overlap_root = base->root(); - overlap_root.revision_ = 2; - auto conflicting = base->Apply(overlap_root, 2, std::span(&overlap, 1)); - ASSERT_TRUE(conflicting.ok()); - EXPECT_TRUE(absl::IsAborted( - MergeHashPublication(*base, *conflicting, plan_root, writes).status())); ++plan_root.group_count_; EXPECT_TRUE(absl::IsFailedPrecondition( MergeHashPublication(*base, *current, plan_root, writes).status())); --plan_root.group_count_; - ++plan_root.field_count_; - EXPECT_TRUE(absl::IsInvalidArgument( - MergeHashPublication(*base, *current, plan_root, writes).status())); - --plan_root.field_count_; changed.retired_ = true; EXPECT_TRUE(absl::IsFailedPrecondition( MergeHashPublication(*base, *current, plan_root, writes).status())); } +#ifndef NDEBUG +TEST(GroupedHashDeathTest, + MergePublicationRequiresProtectedLeavesAndValidPlan) { + auto pages = GroupHashValue(Value(128), 17, Seed(), 1024); + ASSERT_TRUE(pages.ok()); + auto records = Candidates(*pages); + auto base = Recover(Root(128, pages->size()), 1, records, {}); + ASSERT_TRUE(base.ok()) << base.status(); + auto planned = base->root(); + const auto writes = std::span(&pages->front(), 1); + auto overlap = records.front(); + overlap.sequence_ = overlap.lsn_ = 2; + auto current_root = base->root(); + current_root.revision_ = 2; + auto current = base->Apply(current_root, 2, std::span(&overlap, 1)); + ASSERT_TRUE(current.ok()) << current.status(); + EXPECT_DEATH( + { + const auto merged = + MergeHashPublication(*base, *current, planned, writes); + (void)merged; + }, + "Hash page changed while protected"); + --planned.field_count_; + EXPECT_DEATH( + { + const auto merged = MergeHashPublication(*base, *base, planned, writes); + (void)merged; + }, + "Hash plan count disagrees with pages"); +} +#endif + TEST(GroupedHashTest, PrefixBoundsIncludeZeroAndFullWidth) { const auto maximum = std::numeric_limits::max(); EXPECT_TRUE((GroupedRecordId{}).ContainsHash(0)); diff --git a/tests/grouped/lock_test.cpp b/tests/grouped/lock_test.cpp index c66e39119..fd3fce92d 100644 --- a/tests/grouped/lock_test.cpp +++ b/tests/grouped/lock_test.cpp @@ -104,18 +104,14 @@ void AcquireNow(GroupedLockTable::LockSet& locks, GroupedLockId id, GroupedLockMode mode = kX) { auto acquire = locks.Acquire(id, mode); ASSERT_TRUE(acquire.await_ready()); - EXPECT_TRUE(acquire.await_resume().ok()); + acquire.await_resume(); } Task RunWithGroups(GroupedLockTable& table, std::vector ids, GroupedLockMode mode, std::vector& events, std::string name, IoWait* io = nullptr) { auto locks = table.MakeLockSet(kObject); - for (auto id : ids) { - auto status = co_await locks.Acquire(id, mode); - EXPECT_TRUE(status.ok()) << status; - if (!status.ok()) co_return; - } + for (auto id : ids) co_await locks.Acquire(id, mode); events.push_back(name + ":start"); if (io) co_await *io; events.push_back(name + ":done"); @@ -141,7 +137,7 @@ TEST(GroupedLockTest, SameGroupCompatibilityAndUncontendedCleanup) { // The grant is installed even before the worker resumes this waiter. EXPECT_EQ(second.size(), 1); queue.RunAll(); - EXPECT_TRUE(acquire.await_resume().ok()); + acquire.await_resume(); second.Reset(); EXPECT_EQ(table.size(), 0); } @@ -165,7 +161,7 @@ TEST(GroupedLockTest, DatabaseKeyIncarnationAndGraphScopeIdentities) { EXPECT_TRUE(queue.ready_.empty()); } -TEST(GroupedLockTest, AcquisitionOrderAndUpgradesFailWithoutLosingHolds) { +TEST(GroupedLockTest, OrderedAcquisitionsCanRevisitSufficientHolds) { ResumeQueue queue; GroupedLockTable table(queue.queue()); auto locks = table.MakeLockSet(kObject); @@ -182,18 +178,56 @@ TEST(GroupedLockTest, AcquisitionOrderAndUpgradesFailWithoutLosingHolds) { AcquireNow(locks, member, kS); AcquireNow(locks, later_member, kS); EXPECT_EQ(locks.size(), 4); - for (auto id : {member, GroupedLockId{GroupedLockIndex::kMember, {9, 1}}}) { - auto acquire = locks.Acquire(id, kX); - ASSERT_TRUE(acquire.await_ready()); - EXPECT_TRUE(absl::IsFailedPrecondition(acquire.await_resume())); - EXPECT_EQ(locks.size(), 4); - EXPECT_EQ(table.size(), 4); - } locks.Reset(); EXPECT_EQ(table.size(), 0); AcquireNow(locks, member, kX); } +#ifndef NDEBUG +TEST(GroupedLockDeathTest, InvalidRequestsViolateAcquisitionContract) { + ResumeQueue queue; + GroupedLockTable table(queue.queue()); + auto locks = table.MakeLockSet(kObject); + EXPECT_DEATH( + { + auto acquire = locks.Acquire(Page(1), static_cast(2)); + acquire.await_ready(); + }, + "invalid group lock mode"); + EXPECT_DEATH( + { + auto acquire = + locks.Acquire({static_cast(2), {1, 0}}, kX); + acquire.await_ready(); + }, + "invalid group lock index"); + AcquireNow(locks, Page(2), kS); + EXPECT_DEATH( + { + auto acquire = locks.Acquire(Page(1), kX); + acquire.await_ready(); + }, + "group locks must be acquired in increasing order"); + EXPECT_DEATH( + { + auto acquire = locks.Acquire(Page(2), kX); + acquire.await_ready(); + }, + "group lock upgrade requires releasing and replanning"); + auto owner = table.MakeLockSet(kObject); + AcquireNow(owner, Page(3)); + auto pending = locks.Acquire(Page(3), kX); + ASSERT_FALSE(pending.await_ready()); + pending.await_suspend(std::noop_coroutine()); + EXPECT_DEATH( + { + auto acquire = locks.Acquire(Page(4), kX); + acquire.await_ready(); + }, + "a group acquisition is already pending"); +} +#endif + TEST(GroupedLockTest, RetainsEarlierGroupsWhileOtherGroupsKeepProgressing) { ResumeQueue queue; GroupedLockTable table(queue.queue()); @@ -327,9 +361,6 @@ TEST(GroupedLockTest, CancellingPendingAcquisitionRetainsEarlierHolds) { auto pending = waiting.Acquire(Page(2), kX); ASSERT_FALSE(pending.await_ready()); pending.await_suspend(std::noop_coroutine()); - auto concurrent = waiting.Acquire(Page(3), kX); - ASSERT_TRUE(concurrent.await_ready()); - EXPECT_TRUE(absl::IsFailedPrecondition(concurrent.await_resume())); } EXPECT_EQ(waiting.size(), 1); AcquireNow(waiting, Page(3)); @@ -410,7 +441,7 @@ TEST(GroupedLockTest, WholeKeyWaiterCannotHideAGroupNeededByAnIxHolder) { first_guard.Reset(); EXPECT_EQ(shard.queue().size(), 1); queue.RunAll(); - EXPECT_TRUE(waiting.await_resume().ok()); + waiting.await_resume(); second.Reset(); second_guard.Reset(); EXPECT_TRUE(shard.queue().Empty()); diff --git a/tests/grouped/ordered_object_test.cpp b/tests/grouped/ordered_object_test.cpp index 3c10a1c5d..97c7e6ec2 100644 --- a/tests/grouped/ordered_object_test.cpp +++ b/tests/grouped/ordered_object_test.cpp @@ -269,6 +269,30 @@ TEST(SortedSetPublicationDeathTest, }, "Sorted Set member page changed while protected"); } + +TEST(SortedSetPublicationDeathTest, PrivatePlanCountsMustAgreeAcrossIndexes) { + PublicationFixture f; + auto page = f.Page(0); + auto member = f.Member(0); + auto planned = f.root; + ++planned.item_count_; + EXPECT_DEATH( + { + const auto merged = MergeSortedSetPublication( + f.directory, f.directory, planned, std::span(&page, 1), + std::span(&member, 1)); + (void)merged; + }, + "Sorted Set plan counts disagree between indexes"); + planned.member_index_->field_count_ = planned.item_count_; + EXPECT_DEATH( + { + const auto merged = MergeSortedSetPublication( + f.directory, f.directory, planned, std::span(&page, 1), {}); + (void)merged; + }, + "Sorted Set count changed without members"); +} #endif TEST(SortedSetPublicationTest, StructuralPlansRequireWholeKeyProtection) { diff --git a/tests/storage_write_buffer_test.cpp b/tests/storage_write_buffer_test.cpp index 1ba5ff1e9..58600d78e 100644 --- a/tests/storage_write_buffer_test.cpp +++ b/tests/storage_write_buffer_test.cpp @@ -921,9 +921,8 @@ class WriteBufferTestPeer { const GroupedLockObject identity{0, tx::FingerprintOf(digest), object->incarnation()}; auto gate = store.group_locks_->MakeLockSet(identity); - auto locked = co_await gate.Acquire({GroupedLockIndex::kMember, last}, - GroupedLockMode::kExclusive); - if (!locked.ok()) co_return locked; + co_await gate.Acquire({GroupedLockIndex::kMember, last}, + GroupedLockMode::kExclusive); const std::array later{ {{members[1], scores[1] + 0.25}}}; const std::array earlier{ @@ -961,7 +960,7 @@ class WriteBufferTestPeer { GroupedLockMode::kExclusive); const bool available = acquire.await_ready(); EXPECT_FALSE(available) << "reader released its earlier leaf"; - if (available) EXPECT_TRUE(acquire.await_resume().ok()); + if (available) acquire.await_resume(); } auto earlier_write = start(storage.ExecuteSortedSet(0, key, earlier_op)); EXPECT_FALSE(earlier_write.done()); From aabf5d02d8de34042892fa6e967e132abe1c32b8 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 6 Nov 2026 11:02:18 +0000 Subject: [PATCH 2/2] test(storage): run grouped assertion tests unconditionally Remove redundant build-mode wording from the lock and publication contracts. Keep grouped death tests registered without NDEBUG guards in the assertion-enabled test configurations. Validation: Debug unit-test build; 77 focused unit tests; formatting hooks and git diff --check. --- docs/architecture/09-grouped-collections.md | 5 ++--- include/lavik/storage/detail/grouped/collection.h | 8 ++++---- include/lavik/storage/detail/grouped/hash.h | 2 +- include/lavik/storage/detail/grouped/lock.h | 4 ++-- tests/grouped/hash_test.cpp | 2 -- tests/grouped/lock_test.cpp | 2 -- tests/grouped/ordered_object_test.cpp | 2 -- 7 files changed, 9 insertions(+), 16 deletions(-) diff --git a/docs/architecture/09-grouped-collections.md b/docs/architecture/09-grouped-collections.md index 117e8facb..671689660 100644 --- a/docs/architecture/09-grouped-collections.md +++ b/docs/architecture/09-grouped-collections.md @@ -231,9 +231,8 @@ lock-set lifetime, including group waits and IO. Each set acquires member groups before ordered groups and uses the total order of `GroupedRecordId` within each graph. Earlier holds remain owned while the next group waits. Descending acquisitions, shared-to-exclusive upgrades and simultaneous acquisitions on one -set violate the caller's lock-order contract and are asserted in Debug. Callers -revalidate routing after waiting and release the set before replanning or -acquiring a stronger key lock. +set violate the caller's lock-order contract. Callers revalidate routing after +waiting and release the set before replanning or acquiring a stronger key lock. Each group has its own FIFO. These queues are independent of the transaction shard's key queue, so a waiting operation cannot block an unrelated group needed diff --git a/include/lavik/storage/detail/grouped/collection.h b/include/lavik/storage/detail/grouped/collection.h index 1f8d25b2f..7b2744793 100644 --- a/include/lavik/storage/detail/grouped/collection.h +++ b/include/lavik/storage/detail/grouped/collection.h @@ -788,10 +788,10 @@ struct OrderedCollectionMutationPlan { // latest directory. The caller refreshes the base after locking and protects // affected member/ordered groups and routing neighbours through publication. // Their logical versions must stay unchanged, and the private plan's counts -// must agree across both indexes (asserted in Debug). Physical relocation is -// allowed. The owner must not yield between merge and -// publication. FailedPrecondition means the plan requires whole-key -// serialization; Aborted means the current root is incompatible with the base. +// must agree across both indexes. Physical relocation is allowed. The owner +// must not yield between merge and publication. FailedPrecondition means the +// plan requires whole-key serialization; Aborted means the current root is +// incompatible with the base. // The returned root has no publication revision yet. absl::StatusOr MergeSortedSetPublication( const OrderedGroupDirectory& base, const OrderedGroupDirectory& current, diff --git a/include/lavik/storage/detail/grouped/hash.h b/include/lavik/storage/detail/grouped/hash.h index d130f94e1..6f296e90a 100644 --- a/include/lavik/storage/detail/grouped/hash.h +++ b/include/lavik/storage/detail/grouped/hash.h @@ -380,7 +380,7 @@ struct HashGroupMutationPlan { // holds parent IX and X on every changed leaf, or parent X. Unrelated leaves // may publish while pages are prepared; their counts and revisions survive. // The base is captured after locking. Protected leaf versions and the plan's -// own count delta must agree; violations are asserted in Debug. +// own count delta must agree. // Structural edits and deletion of the final field require whole-key X. // The returned root has no revision; assign one at the publication boundary. absl::StatusOr MergeHashPublication( diff --git a/include/lavik/storage/detail/grouped/lock.h b/include/lavik/storage/detail/grouped/lock.h index 23cdf3535..174757004 100644 --- a/include/lavik/storage/detail/grouped/lock.h +++ b/include/lavik/storage/detail/grouped/lock.h @@ -86,8 +86,8 @@ class GroupedLockTable { // Acquires one group, retaining all earlier groups across the wait and IO. // New identities must increase. An already-held identity is a no-op when // its mode suffices. Descending acquisitions and S-to-X upgrades violate - // the caller's lock-order contract and are asserted in Debug. Replanning - // or whole-key fallback must release the set first. + // the caller's lock-order contract. Replanning or whole-key fallback must + // release the set first. // Only one acquisition may be outstanding per set. After a wait, callers // must revalidate routing before accessing the group they intended to use. [[nodiscard]] Awaiter Acquire(GroupedLockId id, GroupedLockMode mode); diff --git a/tests/grouped/hash_test.cpp b/tests/grouped/hash_test.cpp index 2f96217be..df245c159 100644 --- a/tests/grouped/hash_test.cpp +++ b/tests/grouped/hash_test.cpp @@ -603,7 +603,6 @@ TEST(GroupedHashTest, MergePublicationPreservesDisjointChanges) { MergeHashPublication(*base, *current, plan_root, writes).status())); } -#ifndef NDEBUG TEST(GroupedHashDeathTest, MergePublicationRequiresProtectedLeavesAndValidPlan) { auto pages = GroupHashValue(Value(128), 17, Seed(), 1024); @@ -634,7 +633,6 @@ TEST(GroupedHashDeathTest, }, "Hash plan count disagrees with pages"); } -#endif TEST(GroupedHashTest, PrefixBoundsIncludeZeroAndFullWidth) { const auto maximum = std::numeric_limits::max(); diff --git a/tests/grouped/lock_test.cpp b/tests/grouped/lock_test.cpp index fd3fce92d..df963e05c 100644 --- a/tests/grouped/lock_test.cpp +++ b/tests/grouped/lock_test.cpp @@ -183,7 +183,6 @@ TEST(GroupedLockTest, OrderedAcquisitionsCanRevisitSufficientHolds) { AcquireNow(locks, member, kX); } -#ifndef NDEBUG TEST(GroupedLockDeathTest, InvalidRequestsViolateAcquisitionContract) { ResumeQueue queue; GroupedLockTable table(queue.queue()); @@ -226,7 +225,6 @@ TEST(GroupedLockDeathTest, InvalidRequestsViolateAcquisitionContract) { }, "a group acquisition is already pending"); } -#endif TEST(GroupedLockTest, RetainsEarlierGroupsWhileOtherGroupsKeepProgressing) { ResumeQueue queue; diff --git a/tests/grouped/ordered_object_test.cpp b/tests/grouped/ordered_object_test.cpp index 97c7e6ec2..cec04cf9d 100644 --- a/tests/grouped/ordered_object_test.cpp +++ b/tests/grouped/ordered_object_test.cpp @@ -232,7 +232,6 @@ TEST(SortedSetPublicationTest, DisjointCountChangesMergeInEitherOrder) { } } -#ifndef NDEBUG TEST(SortedSetPublicationDeathTest, ConflictingPagesAndRoutingNeighboursViolateLockContract) { PublicationFixture f; @@ -293,7 +292,6 @@ TEST(SortedSetPublicationDeathTest, PrivatePlanCountsMustAgreeAcrossIndexes) { }, "Sorted Set count changed without members"); } -#endif TEST(SortedSetPublicationTest, StructuralPlansRequireWholeKeyProtection) { PublicationFixture f;