refactor(storage): assert grouped locking and publication contracts

docs/architecture/09-grouped-collections.md

@@ -229,9 +229,9 @@ 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
+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. 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

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

@@ -787,10 +787,11 @@ 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
-// publication. FailedPrecondition means the plan requires whole-key
-// serialization; Aborted means the current root is incompatible with the base.
+// Their logical versions must stay unchanged, and the private plan's counts
+// 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<OrderedCollectionRoot> MergeSortedSetPublication(
     const OrderedGroupDirectory& base, const OrderedGroupDirectory& current,

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.
 // 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<GroupedHashRoot> MergeHashPublication(

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. 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;

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,

src/storage/engine/grouped/collection.cpp

@@ -1521,9 +1521,10 @@ absl::StatusOr<OrderedCollectionRoot> MergeSortedSetPublication(
     std::span<const HashGroupSnapshot> 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<OrderedCollectionRoot> 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;
 }

src/storage/engine/grouped/hash.cpp

@@ -1000,14 +1000,18 @@ absl::StatusOr<GroupedHashRoot> 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<absl::int128>(page.field_count()) - old->field_count_;
   }
-  if (static_cast<absl::int128>(before.field_count_) + delta !=
-      planned.field_count_)
-    return absl::InvalidArgumentError("Hash plan count disagrees with pages");
+  assert(static_cast<absl::int128>(before.field_count_) + delta ==
+             planned.field_count_ &&
+         "Hash plan count disagrees with pages");
   const auto count =
       static_cast<absl::int128>(current.root().field_count_) + delta;
   if (count <= 0 || count > UINT32_MAX)

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);

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() {

src/storage/engine/grouped/mutation.cpp

@@ -147,8 +147,8 @@ Task<absl::Status> 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<std::uint32_t>::max() ||
@@ -165,12 +165,15 @@ Task<absl::Status> 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<absl::Status> 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;

src/storage/engine/grouped/ordered_mutation.cpp

@@ -201,12 +201,15 @@ Task<absl::Status> 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<absl::Status> 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;