feat(storage): overlap grouped hash operations with IS/IX and leaf locks

docs/architecture/03-transaction-coordination.md

@@ -84,10 +84,12 @@ intent-exclusive (IX) modes. IS is compatible with IS, S and IX; S with S;
 IX with IX. All other pairs conflict. IS/IX permit partial readers and writers
 to overlap; callers must separately arbitrate their sub-key conflicts. Grouped
 storage provides an independent [sub-key lock API](09-grouped-collections.md#sub-key-arbitration).
-Standalone bounded grouped Sorted Set point writes request IX and arbitrate
-their member and ordered groups separately. Eligible standalone member-score
-projections use IS and shared member-leaf locks. Whole-key readers retain S;
-structural writes, batch commands and EXEC/Lua retain X. IS/IX are distinct from the
+Standalone bounded grouped Hash and Sorted Set writes request IX and arbitrate
+their affected groups separately. Partial readers and resident cardinality
+queries use IS. Whole-key readers retain S; structural writes and externally
+locked multi-key/EXEC/Lua commands retain their whole-key protection. Eligible
+multi-operand writes acquire every affected group before preparing their atomic
+publication. IS/IX are distinct from the
 scheduling-intent counter layer: all modes register intents, and conflicting
 queued intents prevent later arrivals from barging. In particular, a queued
 S or X blocks new IX arrivals, while only a queued X blocks new IS arrivals.

docs/architecture/09-grouped-collections.md

@@ -243,6 +243,20 @@ Once a coroutine has entered the worker's ready queue, it must resume before
 its frame can be destroyed. Group lock state has no durable representation and
 does not allocate transaction IDs or revisions.
 
+Standalone grouped Hash point reads and writes use IS/IX with shared/exclusive
+holds on their requested prefix leaves. HSCAN protects its cursor leaf; HLEN
+reads the atomically published cardinality under IS without payload holds.
+Multi-field commands acquire all leaves in identity order before capturing one
+logical view. Holds span IO, private preparation and publication. Physical
+loads validate protected leaf revisions while allowing disjoint root updates.
+Bounded topology-preserving writes merge changed leaves and their count delta
+into the latest root after append resource waits, then allocate a revision and
+stage the complete batch without yielding. The current predecessor decision
+remains a recovery dependency. Splits, demotion, final-field deletion and plans
+exceeding one append range release their groups and IX before one key-X retry.
+Compact values, full projections, random operations and externally locked
+commands retain their whole-key paths.
+
 Standalone ZADD, ZREM and ZINCRBY inspect owner-local key state
 without suspension to choose IX for an existing grouped Sorted Set, then
 recheck its representation after acquisition. The worker lazily owns one group
@@ -381,19 +395,20 @@ This includes both Sorted Set graphs and retained EXEC/Lua key holds, but not
 native candidate ingestion. Plans retain their page admission through commit.
 Reacquisition checks population and the protected logical state even for
 no-ops and last-element deletion; physical relocation alone is allowed.
-Whole-key writers validate root identity, while concurrent Sorted Set point
+Whole-key writers validate root identity, while concurrent Hash and Sorted Set
 writers validate their locked groups. The writer refreshes GC coordinates
 under store state.
 
-Existing standalone Sorted Set updates that preserve page topology and fit
+Existing standalone Hash and Sorted Set updates that preserve page topology and fit
 all changed inline groups plus their root in one append range use a prepared
 batch. Page preparation and append capacity, index-reader and retirement-pin
 waits finish before publication order is assigned. The caller refreshes its
-base after locking and protects affected member pages, ordered pages and their
-routing neighbours through publication, keeping their logical versions stable.
+base after locking and protects affected prefix leaves and, for Sorted Sets,
+ordered pages and their routing neighbours through publication, keeping their
+logical versions stable.
 At the final owner-local boundary, the writer merges count changes into the
 current root and retains the current predecessor decision. It then assigns
-revision and command sequence and stages both graphs and their root without
+revision and command sequence and stages changed pages and their root without
 yielding. Physical GC moves are allowed. A failed merge or encoding leaves the
 append boundary and logical graph unchanged. Replication and full-sync effects
 carry the sequence and cardinality of that final publication.
@@ -403,8 +418,9 @@ retain the multi-append path, with X ordering revisions across its resource
 and extent waits. IX point writers require bounded prepared publication;
 ineligible plans fall back before staging any records. Sorted Set writers bound
 the minimum remaining primary payload using locked page contents and the
-headers of untouched pages, whose retirement requires key X. Distant concurrent
-removals cannot force a compact conversion during the append wait.
+headers of untouched pages, whose retirement requires key X. Hash batches
+recheck their merged byte total after resource waits and request key X if
+disjoint removals make demotion or final-field deletion necessary.
 Full-image, promotion and random Set-pop adapters retain
 their separate preparation paths and share the same publication boundary.
 

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

@@ -364,7 +364,8 @@ struct LoadedHashGroup {
 
 struct HashGroupMutationPlan {
   GroupedHashRoot root_;
-  // Publication must revalidate this version under the exclusive key lock.
+  // Publication revalidates this version under key X, or merges a bounded
+  // plan after validating its protected leaves under IX.
   // It must also preserve the preceding root's transaction dependency until
   // that decision is durable; an uncommitted structural write is not a base
   // on which an independently durable update may be built.
@@ -375,6 +376,15 @@ struct HashGroupMutationPlan {
   std::vector<HashGroupSnapshot> writes_;
 };
 
+// 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.
+// 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(
+    const HashGroupDirectory& base, const HashGroupDirectory& current,
+    const GroupedHashRoot& planned, std::span<const HashGroupSnapshot> pages);
+
 // Plans a command using only its affected, validated group snapshots. Inputs
 // are scratch owned by this call; failure cannot mutate the current directory.
 // All returned writes and the small root after-image need one atomic storage

include/lavik/storage/engine.h

@@ -1879,6 +1879,9 @@ class StorageEngine {
       const SortedSetOperation& operation, TxShardWrites* tx = nullptr,
       ReplicationCommandAppend* replication = nullptr,
       const MutationPrecondition* mutation_precondition = nullptr);
+  // Standalone grouped Hash point/scan operations use parent IS/IX and hold
+  // all selected groups through IO. Structural writes retry under whole-key X;
+  // compact and externally locked operations retain their existing paths.
   bycorf::Task<absl::StatusOr<HashResult>> ExecuteHash(
       std::uint8_t db_id, std::string_view key, const HashOperation& operation,
       ReplicationCommandAppend* replication = nullptr,

src/storage/engine/grouped/hash.cpp

@@ -23,6 +23,7 @@
 
 #include "absl/container/flat_hash_map.h"
 #include "absl/container/inlined_vector.h"
+#include "absl/numeric/int128.h"
 
 namespace lavik::storage {
 namespace {
@@ -980,6 +981,43 @@ absl::StatusOr<HashGroupDirectory> HashGroupDirectory::Apply(
   return next;
 }
 
+absl::StatusOr<GroupedHashRoot> MergeHashPublication(
+    const HashGroupDirectory& base, const HashGroupDirectory& current,
+    const GroupedHashRoot& planned, std::span<const HashGroupSnapshot> pages) {
+  const auto& before = base.root();
+  const auto same_topology = [&](const GroupedHashRoot& root) {
+    return root.incarnation_ == before.incarnation_ &&
+           root.seed_ == before.seed_ &&
+           root.group_count_ == before.group_count_;
+  };
+  if (!same_topology(planned) || pages.empty())
+    return absl::FailedPreconditionError("Hash plan needs key X");
+  if (!same_topology(current.root()) || current.sequence() < base.sequence())
+    return absl::AbortedError("Hash topology changed during preparation");
+  absl::int128 delta = 0;
+  for (const auto& page : pages) {
+    const auto* old = base.groups().Get(page.id_.prefix_);
+    if (!old || old->id_ != page.id_ || page.retired_ ||
+        page.incarnation_ != before.incarnation_)
+      return absl::FailedPreconditionError("Hash page topology needs key X");
+    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");
+    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");
+  const auto count =
+      static_cast<absl::int128>(current.root().field_count_) + delta;
+  if (count <= 0 || count > UINT32_MAX)
+    return absl::FailedPreconditionError("Hash cardinality needs key X");
+  auto merged = current.root();
+  merged.field_count_ = static_cast<std::uint64_t>(count);
+  merged.revision_ = 0;
+  return merged;
+}
+
 absl::StatusOr<HashGroupMutationPlan> PlanHashGroupMutation(
     const HashGroupDirectory& directory,
     std::vector<LoadedHashGroup> loaded_groups, HashGroupMutationKind kind,

src/storage/engine/grouped/hash_random.cpp

@@ -209,7 +209,7 @@ StorageEngine::Impl::ExecuteGroupedHashRandomLocked(
         store, partition, db_id, key, digest, object, std::move(remaining),
         std::move(changed), result.length_, value_type,
         object->version().root_.expire_at_ms_, tx, replication,
-        mutation_precondition, nullptr, operation.diagnostics_);
+        mutation_precondition, nullptr, nullptr, operation.diagnostics_);
   } else {
     auto detail = MakeSetTrace(operation.diagnostics_);
     status = co_await AppendLocked(