From 295e6af957b40e470315c5e613a8eeddde674658 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Thu, 5 Nov 2026 13:01:34 +0000 Subject: [PATCH 1/2] feat(storage): overlap grouped hash reads and writes with intent locks --- .../03-transaction-coordination.md | 10 +- docs/architecture/09-grouped-collections.md | 30 +- include/lavik/storage/detail/grouped/hash.h | 12 +- include/lavik/storage/engine.h | 3 + src/storage/engine/grouped/hash.cpp | 38 +++ src/storage/engine/grouped/hash_random.cpp | 2 +- src/storage/engine/grouped/mutation.cpp | 206 ++++++++++++-- src/storage/engine/grouped/read.cpp | 4 +- src/storage/engine/hash_api.cpp | 166 ++++++++++- src/storage/engine/hash_tree.cpp | 71 ++++- src/storage/engine/impl.h | 29 +- tests/grouped/hash_test.cpp | 51 ++++ tests/grouped/hash_write_e2e_test.cpp | 269 ++++++++++++++++++ 13 files changed, 828 insertions(+), 63 deletions(-) diff --git a/docs/architecture/03-transaction-coordination.md b/docs/architecture/03-transaction-coordination.md index 4ccb092b9..46698c202 100644 --- a/docs/architecture/03-transaction-coordination.md +++ b/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. diff --git a/docs/architecture/09-grouped-collections.md b/docs/architecture/09-grouped-collections.md index 3e8fc7a0b..68039339e 100644 --- a/docs/architecture/09-grouped-collections.md +++ b/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. diff --git a/include/lavik/storage/detail/grouped/hash.h b/include/lavik/storage/detail/grouped/hash.h index b9da790c4..6efcacfc9 100644 --- a/include/lavik/storage/detail/grouped/hash.h +++ b/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 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 MergeHashPublication( + const HashGroupDirectory& base, const HashGroupDirectory& current, + const GroupedHashRoot& planned, std::span 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 diff --git a/include/lavik/storage/engine.h b/include/lavik/storage/engine.h index 2fc1cf4a3..e5a7ca8cf 100644 --- a/include/lavik/storage/engine.h +++ b/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> ExecuteHash( std::uint8_t db_id, std::string_view key, const HashOperation& operation, ReplicationCommandAppend* replication = nullptr, diff --git a/src/storage/engine/grouped/hash.cpp b/src/storage/engine/grouped/hash.cpp index c7dedf0ed..679abada5 100644 --- a/src/storage/engine/grouped/hash.cpp +++ b/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::Apply( return next; } +absl::StatusOr MergeHashPublication( + const HashGroupDirectory& base, const HashGroupDirectory& current, + const GroupedHashRoot& planned, std::span 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(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"); + const auto count = + static_cast(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(count); + merged.revision_ = 0; + return merged; +} + absl::StatusOr PlanHashGroupMutation( const HashGroupDirectory& directory, std::vector loaded_groups, HashGroupMutationKind kind, diff --git a/src/storage/engine/grouped/hash_random.cpp b/src/storage/engine/grouped/hash_random.cpp index ef4f823a9..8536556d5 100644 --- a/src/storage/engine/grouped/hash_random.cpp +++ b/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( diff --git a/src/storage/engine/grouped/mutation.cpp b/src/storage/engine/grouped/mutation.cpp index 2b47445a5..3d31d0bec 100644 --- a/src/storage/engine/grouped/mutation.cpp +++ b/src/storage/engine/grouped/mutation.cpp @@ -138,7 +138,17 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( ValueType value_type, std::uint64_t expire_at_ms, TxShardWrites* tx, ReplicationCommandAppend* replication, const MutationPrecondition* mutation_precondition, - HashGroupMutationPlan* prepared, OnlineLatencyTrace* diagnostics) { + HashGroupMutationPlan* prepared, bool* retry_exclusive, + OnlineLatencyTrace* diagnostics) { + if (retry_exclusive) *retry_exclusive = false; + const bool prepared_publication = retry_exclusive != nullptr; + const auto needs_x = [&]() { + assert(retry_exclusive); + *retry_exclusive = true; + return absl::AbortedError("Hash plan requires key X"); + }; + if (prepared_publication && (!previous || !prepared || tx)) + co_return needs_x(); if (!diagnostics && tx) diagnostics = tx->diagnostics_; if (field_count == 0 || field_count > std::numeric_limits::max() || @@ -149,13 +159,48 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( if (previous && previous->version().root_.value_type() != value_type) { co_return absl::InvalidArgumentError("grouped mutation changes value type"); } + auto plan = prepared != nullptr + ? absl::StatusOr(std::move(*prepared)) + : PrepareGroupedHashMutation(previous, std::move(after_image), + 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"); + } + const auto planned_root = plan->root_; + if (prepared_publication && + !MergeHashPublication(previous->directory(), previous->directory(), + planned_root, plan->writes_) + .ok()) + co_return needs_x(); const auto db_epoch = EffectiveRecordDbEpoch(partition, db_id); const auto replication_epoch = partition.replication_epoch_; const auto index_generation = store.index_generations_[db_id]; const auto grouped_generation = partition.grouped_generations_[db_id]; auto& side = partition.grouped_objects_[db_id]; auto source_side = side.CurrentForMutation(key); - if (previous && !SameLogicalView(previous, source_side)) { + const auto source_matches = [&](const GroupedObject::Handle& current) { + if (SameLogicalView(previous, current)) return true; + if (!prepared_publication || !current || current->is_ordered()) + return false; + const auto& before = previous->version(); + const auto& now = current->version(); + return before.db_epoch_ == now.db_epoch_ && + before.replication_epoch_ == now.replication_epoch_ && + before.index_generation_ == now.index_generation_ && + before.root_.value_type() == now.root_.value_type() && + before.root_.expire_at_ms_ == now.root_.expire_at_ms_ && + MergeHashPublication(previous->directory(), current->directory(), + planned_root, plan->writes_) + .ok(); + }; + if (previous && !source_matches(source_side)) { + if (prepared_publication) co_return needs_x(); co_return absl::AbortedError("grouped mutation source changed"); } const bool outer_transaction = tx != nullptr; @@ -182,16 +227,23 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( if (!dependency.ok()) co_return dependency; if (EffectiveRecordDbEpoch(partition, db_id) != db_epoch || partition.replication_epoch_ != replication_epoch || - store.index_generations_[db_id] != index_generation || - !SameLogicalView(source_side, side.CurrentForMutation(key))) { + store.index_generations_[db_id] != index_generation) { co_return absl::AbortedError("grouped population changed before mutation"); } + if (!SameLogicalView(source_side, side.CurrentForMutation(key)) && + !source_matches(side.CurrentForMutation(key))) { + if (prepared_publication) co_return needs_x(); + co_return absl::AbortedError("grouped source changed before mutation"); + } // Admission and cross-coordinator dependency waits release the store lock. // GC may publish another physical view of this same logical root meanwhile; // the publication reservation compares handle identity, not just C/R. // Refresh only after the population/logical checks, while the lock is held. source_side = side.CurrentForMutation(key); - std::uint64_t sequence = 0; + // Append admission needs a nonzero placeholder. A prepared batch replaces + // it with publication order after the final resource wait. + std::uint64_t sequence = + prepared_publication ? source_side->command_sequence() : 0; if (replica_loading_.load(std::memory_order_acquire)) { const auto* sync = partition.replica_sync_.get(); if (!sync || !sync->command_sequence_) { @@ -199,10 +251,12 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( "grouped replay has no command sequence"); } sequence = *sync->command_sequence_; - } else { + } else if (!prepared_publication) { sequence = ++partition.mutation_sequence_; } - if (sequence == 0 || (previous && sequence < previous->command_sequence())) { + if (!prepared_publication && + (sequence == 0 || + (previous && sequence < previous->command_sequence()))) { co_return absl::FailedPreconditionError( "grouped mutation has a stale source command sequence"); } @@ -226,22 +280,11 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( const auto command_batch = tx->grouped_ingest_batch_ != nullptr ? tx->grouped_ingest_batch_->txid_ : batch.txid_; - auto plan = prepared != nullptr - ? absl::StatusOr(std::move(*prepared)) - : PrepareGroupedHashMutation(previous, std::move(after_image), - 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"); - } - const auto revision = AllocateGroupedRevision(); - if (revision == 0 || - (previous && revision <= previous->directory().sequence())) { + auto revision = + prepared_publication ? previous->revision() : AllocateGroupedRevision(); + if (!prepared_publication && + (revision == 0 || + (previous && revision <= previous->directory().sequence()))) { co_return absl::OutOfRangeError( "group revision allocation did not advance"); } @@ -260,7 +303,9 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( // unmoved and immutable through all writes, including extent IO waits. GroupedScratchBudget encoder_budget; for (std::size_t i = 0; i < plan->writes_.size(); ++i) { - auto added = encoder_budget.AddBytes(sizeof(HashGroupEncoder)); + auto added = encoder_budget.AddBytes( + sizeof(HashGroupEncoder) + + (prepared_publication ? sizeof(PreparedGroupRecord) : 0)); if (!added.ok()) co_return added; } auto encoder_admission = encoder_budget.Reserve(1); @@ -289,6 +334,34 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( encoders.push_back(std::move(*encoder)); } } + PreparedGroupBatch prepared_batch; + if (prepared_publication) { + const auto header_bytes = RecordHeaderBytes( + key.size(), key.size() > kInlineKeyMaxBytes, true, false, true); + std::size_t bytes = AlignRecord( + RecordHeaderBytes(key.size(), key.size() > kInlineKeyMaxBytes, true, + expire_at_ms != 0) + + root_payload->size()); + const auto limit = + std::min(kStorageBlockBytes - kBlockHeaderBytes, + options_.buffers_.write_buffer_bytes_); + for (const auto size : encoded_sizes) { + const auto added = AlignRecord(header_bytes + size); + if (bytes > limit || added > limit - bytes) co_return needs_x(); + bytes += added; + } + prepared_batch.records_.reserve(encoders.size()); + for (std::size_t i = 0; i < encoders.size(); ++i) { + const auto size = AlignRecord(header_bytes + encoded_sizes[i]); + prepared_batch.records_.push_back( + PreparedGroupRecord{.id_ = plan->writes_[i].id_, + .logical_size_ = plan->writes_[i].field_count(), + .payload_ = InlineRecordPayload(encoders[i]), + .disk_bytes_ = size, + .location_ = {}}); + prepared_batch.disk_bytes_ += size; + } + } // Group payloads include a per-group envelope (and a compact Hash header // for each nonempty group). Their sum is therefore an upper bound on the // one-header compact image. Decide before staging any auxiliary record. @@ -311,6 +384,7 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( } } if (total < kCollectionDemotionBytes) { + if (prepared_publication) co_return needs_x(); GroupedScratchBudget budget; for (const auto& [prefix, metadata] : previous->directory().groups()) { (void)prefix; @@ -410,6 +484,8 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( // outer commit retires them. Its existing fence must not point at a block we // recycled early; the absent batch decision makes those bytes invisible. auto abandon = [&]() -> Task { + // Prepared locations are private predictions until the root is staged. + if (prepared_publication) co_return absl::OkStatus(); for (const auto& record : written) { const auto& location = record.location_; if (outer_transaction) { @@ -435,7 +511,17 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( } co_return absl::OkStatus(); }; - for (const auto& snapshot : plan->writes_) { + if (prepared_publication) { + for (const auto& page : prepared_batch.records_) { + GroupedRecordLocation record{ + .id_ = page.id_, .location_ = {}, .extents_ = nullptr}; + record.location_.logical_size_ = page.logical_size_; + written.push_back(std::move(record)); + } + } + for (std::size_t i = 0; !prepared_publication && i < plan->writes_.size(); + ++i) { + const auto& snapshot = plan->writes_[i]; // Fail before the selected auxiliary begins. Already staged earlier // auxiliaries exercise command-local batch abort inside an outer EXEC. // The key is explicit so unrelated client/maintenance writes are untouched. @@ -475,6 +561,59 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( .encoded_bytes_ = encoded_sizes[i], .retired_ = group.retired_}); } + if (prepared_publication) { + prepared_batch.prepare_ = [&]() -> absl::Status { + const auto current = side.CurrentForMutation(key); + if (!current || current->is_ordered() || + EffectiveRecordDbEpoch(partition, db_id) != db_epoch || + partition.replication_epoch_ != replication_epoch || + store.index_generations_[db_id] != index_generation || + partition.grouped_generations_[db_id] != grouped_generation || + current->version().root_.expire_at_ms_ != expire_at_ms) + return absl::AbortedError("Hash population changed before publication"); + auto merged = + MergeHashPublication(previous->directory(), current->directory(), + planned_root, plan->writes_); + if (!merged.ok()) { + if (absl::IsFailedPrecondition(merged.status())) return needs_x(); + return merged.status(); + } + // Other groups can shrink while append resources are awaited. Recheck + // the demotion threshold before any bytes become visible or durable. + absl::int128 total = current->directory().total_group_bytes(); + for (std::size_t i = 0; i < plan->writes_.size(); ++i) { + total -= current->directory() + .groups() + .Get(plan->writes_[i].id_.prefix_) + ->encoded_bytes_; + total += encoded_sizes[i]; + } + if (total < kCollectionDemotionBytes) return needs_x(); + const auto retained = RetainGroupedDependency(store, current, tx); + if (!retained.ok()) return retained; + revision = AllocateGroupedRevision(); + if (revision == 0 || revision <= current->revision()) + return absl::OutOfRangeError("Hash revision did not advance"); + sequence = ++partition.mutation_sequence_; + if (sequence == 0 || sequence < current->command_sequence()) + return absl::OutOfRangeError("Hash command sequence did not advance"); + merged->revision_ = revision; + auto encoded = EncodeGroupedHashRoot(*merged); + 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"); + std::memcpy(root_payload->data(), encoded->data(), encoded->size()); + plan->root_ = *merged; + prepared_batch.revision_ = revision; + prepared_batch.sequence_ = sequence; + prepared_batch.logical_size_ = merged->field_count_; + for (auto& page : candidates) page.sequence_ = revision; + source_side = current; + return absl::OkStatus(); + }; + } GroupedObject::PreparedHandle builder; GroupRecordWrite root_write{ .prepared_root_ = &builder, @@ -493,6 +632,10 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( return absl::AbortedError( "grouped source changed before publication"); } + if (prepared_publication) { + for (std::size_t i = 0; i < written.size(); ++i) + written[i].location_ = prepared_batch.records_[i].location_; + } GroupedObjectVersion version = physical; version.decision_ = *decision; absl::StatusOr directory = @@ -522,8 +665,19 @@ Task StorageEngine::Impl::CommitGroupedHashMutationLocked( return absl::OkStatus(); }, .changed_groups_ = written_ids, - .root_incarnation_ = plan->root_.incarnation_}; + .root_incarnation_ = plan->root_.incarnation_, + .batch_ = prepared_publication ? &prepared_batch : nullptr}; GroupMutationWrite mutation{.sequence_ = sequence, .root_ = &root_write}; + LAVIK_FAULT_INJECT(if (prepared_publication && + LAVIK_FAULT_MATCHES("LAVIK_GROUP_PREPARE_PAUSE_KEY", + key)) { + spdlog::info("grouped publication prepared before revision key={}", key); + store.store_state_mutex_.Unlock(*store.worker_); + auto paused = co_await fault_injection::PauseWhileFileExists( + "LAVIK_GROUP_PREPARE_PAUSE_FILE"); + co_await store.store_state_mutex_.Lock(); + if (!paused.ok()) co_return paused; + }); LAVIK_MAYBE_CRASH_AT("group-batch-before-root"); const auto appended = co_await AppendLocked( store, partition, db_id, key, digest, *root_payload, RecordKind::kValue, diff --git a/src/storage/engine/grouped/read.cpp b/src/storage/engine/grouped/read.cpp index 3d63fc360..72cfae8fd 100644 --- a/src/storage/engine/grouped/read.cpp +++ b/src/storage/engine/grouped/read.cpp @@ -29,7 +29,8 @@ absl::StatusOr StorageEngine::Impl::RefreshLockedGroup( GroupedRecordId id) { const auto current = partition.grouped_objects_[db_id].CurrentForMutation(key); - if (!source || !current || !current->has_member_index() || + if (!source || !current || + (current->is_ordered() && !current->has_member_index()) || current->incarnation() != source->incarnation()) return absl::AbortedError("locked group incarnation changed"); const auto& old = source->version(); @@ -38,6 +39,7 @@ absl::StatusOr StorageEngine::Impl::RefreshLockedGroup( old.replication_epoch_ != now.replication_epoch_ || old.index_generation_ != now.index_generation_ || old.root_.expire_at_ms_ != now.root_.expire_at_ms_ || + old.root_.value_type() != now.root_.value_type() || EffectiveRecordDbEpoch(partition, db_id) != old.db_epoch_ || partition.replication_epoch_ != old.replication_epoch_ || partition.grouped_generations_[db_id] != old.index_generation_) diff --git a/src/storage/engine/hash_api.cpp b/src/storage/engine/hash_api.cpp index 1141145df..5a754f172 100644 --- a/src/storage/engine/hash_api.cpp +++ b/src/storage/engine/hash_api.cpp @@ -15,6 +15,7 @@ */ #include "impl.h" +#include "lavik/error_detail.h" namespace lavik::storage { @@ -30,23 +31,180 @@ bool IsHashLikeWrite(const HashOperation& operation) { operation.kind_ == HashOperationKind::kIncrementFloat; } +bool HasGroupLockPlan(HashOperationKind kind) { + switch (kind) { + case HashOperationKind::kSet: + case HashOperationKind::kSetIfAbsent: + case HashOperationKind::kDelete: + case HashOperationKind::kIncrementInteger: + case HashOperationKind::kIncrementFloat: + case HashOperationKind::kGet: + case HashOperationKind::kGetMany: + case HashOperationKind::kExists: + case HashOperationKind::kStringLength: + case HashOperationKind::kLength: + case HashOperationKind::kScan: + return true; + default: + return false; + } +} + } // namespace Task> StorageEngine::Impl::ExecuteHash( std::uint8_t db_id, std::string_view key, const HashOperation& operation, ReplicationCommandAppend* replication, const MutationPrecondition* mutation_precondition) { + return ExecuteHashLike(db_id, key, operation, ValueType::kHash, replication, + mutation_precondition); +} + +Task> StorageEngine::Impl::ExecuteHashLike( + std::uint8_t db_id, std::string_view key, const HashOperation& operation, + ValueType value_type, ReplicationCommandAppend* replication, + const MutationPrecondition* mutation_precondition) { assert(db_id < kLogicalDatabaseCount); const Digest digest = ComputeDigest(key); + const bool write = IsHashLikeWrite(operation); + bool concurrent = false; + if (HasGroupLockPlan(operation.kind_)) { + // Only the owner reads this hint, without suspension or retained pointers. + // Admission drains direct readers; the callee rechecks after the key wait. + auto& partition = PartitionForKey(CurrentStore(), key); + const auto* found = partition.indexes_[db_id].Find(digest, key); + concurrent = CanPrepareGroupedWriteUnlocked(partition) && found && + found->key_complete() && found->value_.grouped() && + found->value_.value_type() == value_type; + } TraceSpanScope key_wait(operation.diagnostics_, TraceStage::kKeyLock); auto key_lock = co_await tx::CurrentTxShard().AcquireKey( db_id, tx::FingerprintOf(digest), - IsHashLikeWrite(operation) ? tx::LockMode::kExclusive - : tx::LockMode::kShared, + concurrent ? (write ? tx::LockMode::kIntentExclusive + : tx::LockMode::kIntentShared) + : (write ? tx::LockMode::kExclusive : tx::LockMode::kShared), lavik::storage::RedisSlot(key)); key_wait.End(); - co_return co_await ExecuteHashLocked(db_id, key, digest, operation, nullptr, - replication, mutation_precondition); + if (concurrent) { + auto result = co_await TryConcurrentHashLike(db_id, key, digest, operation, + value_type, replication, + mutation_precondition); + if (!result.ok()) co_return result.status(); + if (*result) co_return std::move(**result); + // The nested task has released every group. Upgrading an intent while + // retaining groups could deadlock another intent holder waiting for them. + key_lock.Reset(); + TraceSpanScope fallback(operation.diagnostics_, TraceStage::kKeyLock); + key_lock = co_await tx::CurrentTxShard().AcquireKey( + db_id, tx::FingerprintOf(digest), + write ? tx::LockMode::kExclusive : tx::LockMode::kShared, + lavik::storage::RedisSlot(key)); + } + co_return co_await ExecuteHashLikeLocked(db_id, key, digest, operation, + value_type, nullptr, replication, + mutation_precondition); +} + +Task>> +StorageEngine::Impl::TryConcurrentHashLike( + std::uint8_t db_id, std::string_view key, const Digest& digest, + const HashOperation& operation, ValueType value_type, + ReplicationCommandAppend* replication, + const MutationPrecondition* mutation_precondition) { + auto& store = CurrentStore(); + auto& partition = PartitionForKey(store, key); + co_await store.store_state_mutex_.Lock(); + UnlockGuard unlock(&store.store_state_mutex_, store.worker_); + const auto* found = partition.indexes_[db_id].Find(digest, key); + const auto now = operation.now_ms_ ? operation.now_ms_ : UnixTimeMillis(); + if (!CanPrepareGroupedWriteUnlocked(partition) || !found || + !found->key_complete() || !found->value_.grouped() || + found->value_.value_type() != value_type || IsExpired(*found, now)) + co_return std::nullopt; + auto view = partition.grouped_objects_[db_id].Lookup( + key, GroupedObjectVersion{ + .root_ = MaterializeIndexLocation(*found), + .db_epoch_ = EffectiveRecordDbEpoch(partition, db_id), + .replication_epoch_ = partition.replication_epoch_, + .index_generation_ = partition.grouped_generations_[db_id]}); + if (!view.ok()) co_return view.status(); + auto object = std::move(*view); + if (!object || object->is_ordered()) co_return std::nullopt; + if (operation.kind_ == HashOperationKind::kLength) { + // No payload can affect this projection after the atomic root snapshot. + const auto readable = object->ReadStatus(); + if (!readable.ok()) co_return readable; + HashResult result; + result.key_exists_ = true; + result.length_ = object->directory().root().field_count_; + co_return std::optional(std::move(result)); + } + const bool scan = operation.kind_ == HashOperationKind::kScan; + const auto count = scan ? std::size_t{1} : operation.fields_.size(); + if (count > SIZE_MAX / sizeof(GroupedRecordId)) + co_return MemoryLimitError("OOM Hash group lock plan overflow"); + auto admission = TryReserveMemory(count * sizeof(GroupedRecordId)); + if (!admission) co_return MemoryLimitError("OOM Hash group lock plan"); + std::vector groups; + groups.reserve(count); + if (groups.capacity() > admission->bytes() / sizeof(GroupedRecordId)) + co_return MemoryLimitError("OOM Hash group lock plan"); + if (scan) { + const auto* route = object->directory().groups().Floor(operation.cursor_); + if (!route) co_return absl::DataLossError("Hash scan has no route"); + groups.push_back(route->id_); + } else { + for (const auto field : operation.fields_) { + const auto* route = object->directory().Find(field); + if (!route) co_return absl::DataLossError("Hash field has no route"); + groups.push_back(route->id_); + } + } + std::sort(groups.begin(), groups.end()); + groups.erase(std::unique(groups.begin(), groups.end()), groups.end()); + if (!store.group_locks_) + store.group_locks_ = std::make_unique(*store.worker_); + auto holds = store.group_locks_->MakeLockSet( + {db_id, tx::FingerprintOf(digest), object->incarnation()}); + unlock.Unlock(); + const bool write = IsHashLikeWrite(operation); + for (const auto id : groups) { + const auto locked = 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 + // same epoch/incarnation checks as other grouped readers. + const auto current = + partition.grouped_objects_[db_id].CurrentForMutation(key); + if (!current || current->is_ordered() || + current->incarnation() != object->incarnation() || + current->version().db_epoch_ != object->version().db_epoch_ || + current->version().replication_epoch_ != + object->version().replication_epoch_ || + current->version().index_generation_ != + object->version().index_generation_) + co_return std::nullopt; + const auto held = [&](const auto* route) { + return route && + std::binary_search(groups.begin(), groups.end(), route->id_); + }; + if (scan) { + if (!held(current->directory().groups().Floor(operation.cursor_))) + co_return std::nullopt; + } else { + for (const auto field : operation.fields_) + if (!held(current->directory().Find(field))) co_return std::nullopt; + } + bool retry = false; + auto result = co_await ExecuteHashLikeLocked(db_id, key, digest, operation, + value_type, nullptr, replication, + mutation_precondition, &retry); + if (retry) co_return std::nullopt; + if (!result.ok()) co_return result.status(); + co_return std::optional(std::move(*result)); } Task> StorageEngine::Impl::ExecuteSet( diff --git a/src/storage/engine/hash_tree.cpp b/src/storage/engine/hash_tree.cpp index 2b6c35526..e68627079 100644 --- a/src/storage/engine/hash_tree.cpp +++ b/src/storage/engine/hash_tree.cpp @@ -774,7 +774,7 @@ Task> StorageEngine::Impl::ReadGroupedHashFields( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::span fields, - OnlineLatencyTrace* diagnostics) { + bool group_locked, OnlineLatencyTrace* diagnostics) { // Keep multi-field planning in its own coroutine so single-field reads do // not pay for the plan's allocations or coroutine-frame storage. constexpr auto max_size = std::numeric_limits::max(); @@ -840,7 +840,7 @@ Task> StorageEngine::Impl::ReadGroupedHashFields( ++end; auto loaded = co_await LoadHashGroupPayload( store, partition, db_id, key, digest, object, requests[begin].group_, - false, false, diagnostics); + false, group_locked, diagnostics); if (!loaded.ok()) [[unlikely]] co_return loaded.status(); const auto bytes = loaded->loaded_.value(); @@ -875,10 +875,19 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( std::uint8_t db_id, std::string_view key, const Digest& digest, const HashOperation& operation, ValueType value_type, TxShardWrites* tx, ReplicationCommandAppend* replication, - const MutationPrecondition* mutation_precondition) { + const MutationPrecondition* mutation_precondition, bool* retry_exclusive) { // Private preparation allocations must unwind here: Bycorf terminates on // exceptions escaping a coroutine body, even when its caller has a catch. assert(db_id < kLogicalDatabaseCount); + if (retry_exclusive) { + assert(tx == nullptr); + *retry_exclusive = false; + } + const auto needs_key_lock = [&]() { + assert(retry_exclusive); + *retry_exclusive = true; + return absl::AbortedError("Hash operation requires whole-key lock"); + }; const bool replace = operation.kind_ == HashOperationKind::kReplaceOnly; if (replace && (value_type != ValueType::kHash || operation.fields_.empty() || operation.fields_.size() != operation.values_.size())) { @@ -913,6 +922,11 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( const std::uint64_t now_ms = operation.now_ms_ == 0 ? UnixTimeMillis() : operation.now_ms_; const bool exists = stored_value && !IsExpired(*found, now_ms); + if (retry_exclusive && + (!exists || !found->key_complete() || !found->value_.grouped() || + found->value_.value_type() != value_type || + !CanPrepareGroupedWriteUnlocked(partition))) + co_return needs_key_lock(); if (!exists && stored_value && found->value_.grouped()) { auto metadata = co_await ReadKeyMetadataLocked(db_id, key, digest); if (!metadata.ok()) co_return metadata.status(); @@ -1054,6 +1068,20 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( } }); + LAVIK_FAULT_INJECT({ + if (retry_exclusive && + LAVIK_FAULT_MATCHES("LAVIK_HASH_GROUP_PAUSE_KEY", key) && + (operation.fields_.empty() + ? std::getenv("LAVIK_HASH_GROUP_PAUSE_FIELD") == nullptr + : LAVIK_FAULT_MATCHES("LAVIK_HASH_GROUP_PAUSE_FIELD", + operation.fields_.front()))) { + spdlog::info("hash group locks held before page IO write={}", !read_only); + const auto paused = co_await fault_injection::PauseWhileFileExists( + "LAVIK_HASH_GROUP_PAUSE_FILE"); + if (!paused.ok()) co_return paused; + } + }); + if (grouped != nullptr && IsCompactPointRead(operation.kind_) && operation.fields_.empty()) co_return result; @@ -1070,8 +1098,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (route == nullptr) [[unlikely]] co_return absl::DataLossError("missing Hash field route"); auto loaded = co_await LoadHashGroupPayload( - store, partition, db_id, key, digest, grouped, route->id_, false, false, - operation.diagnostics_); + store, partition, db_id, key, digest, grouped, route->id_, false, + retry_exclusive != nullptr, operation.diagnostics_); if (!loaded.ok()) [[unlikely]] { if (read_epoch_changed()) co_return HashResult{}; co_return loaded.status(); @@ -1103,7 +1131,7 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( operation.kind_ == HashOperationKind::kGetMany)) { auto values = co_await ReadGroupedHashFields( store, partition, db_id, key, digest, grouped, operation.fields_, - operation.diagnostics_); + retry_exclusive != nullptr, operation.diagnostics_); if (!values.ok() && read_epoch_changed()) [[unlikely]] co_return HashResult{}; co_return values; @@ -1132,8 +1160,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (entry == nullptr) co_return absl::DataLossError("grouped scan has no physical page"); auto loaded = co_await LoadHashGroupPayload( - store, partition, db_id, key, digest, grouped, route->id_, false, false, - operation.diagnostics_); + store, partition, db_id, key, digest, grouped, route->id_, false, + retry_exclusive != nullptr, operation.diagnostics_); if (!loaded.ok()) { if (read_epoch_changed()) co_return HashResult{}; co_return loaded.status(); @@ -1330,8 +1358,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( edits.push_back(leaf_edits[begin++].view_); } while (begin < leaf_edits.size() && leaf_edits[begin].id_ == id); auto loaded = co_await LoadHashGroupPayload( - store, partition, db_id, key, digest, grouped, id, false, false, - operation.diagnostics_); + store, partition, db_id, key, digest, grouped, id, false, + retry_exclusive != nullptr, operation.diagnostics_); if (!loaded.ok()) co_return loaded.status(); const auto bytes = loaded->loaded_.value(); auto edited = ApplyHashGroupEdits( @@ -1368,7 +1396,7 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (unlocked_grouped_write) co_await bycorf::Yield(*store.worker_); auto loaded = co_await LoadHashGroupSnapshot( store, partition, db_id, key, digest, grouped, id, false, nullptr, - false, operation.diagnostics_); + retry_exclusive != nullptr, operation.diagnostics_); if (!loaded.ok()) { if (read_epoch_changed()) co_return HashResult{}; co_return loaded.status(); @@ -1727,6 +1755,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( : location.logical_size_ - selected_field_count + compact.entries_.size(); result.key_exists_ = result.length_ != 0; + if (retry_exclusive && result.changed_ && !result.key_exists_) + co_return needs_key_lock(); if (unlocked_grouped_write) { if (!grouped_plan && result.changed_ && result.key_exists_) { const std::vector changed(changed_groups.begin(), @@ -1738,8 +1768,23 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( } co_await store.store_state_mutex_.Lock(); unlock.Adopt(); + auto validated = grouped; + if (retry_exclusive) { + // Keep the original root for the plan's count delta; validate only the + // protected leaves against current state. Disjoint publications and GC + // can change the root while these page reads and preparations suspend. + for (const auto field : operation.fields_) { + const auto* route = grouped->directory().Find(field); + if (!route) + co_return absl::DataLossError("missing protected Hash route"); + auto refreshed = + RefreshLockedGroup(partition, db_id, key, grouped, route->id_); + if (!refreshed.ok()) co_return refreshed.status(); + validated = std::move(*refreshed); + } + } const auto valid = ValidateGroupedWriteSnapshot( - store, partition, db_id, key, grouped, observed_write, + store, partition, db_id, key, validated, observed_write, mutation_precondition != nullptr ? mutation_precondition : (tx != nullptr ? &tx->mutation_precondition_ : nullptr)); @@ -1833,7 +1878,7 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( changed_groups.end()), result.length_, value_type, expire_at_ms, tx, replication, mutation_precondition, grouped_plan ? &*grouped_plan : nullptr, - operation.diagnostics_); + retry_exclusive, operation.diagnostics_); if (!written.ok()) co_return written; co_return result; } diff --git a/src/storage/engine/impl.h b/src/storage/engine/impl.h index 333b6a214..f8571792e 100644 --- a/src/storage/engine/impl.h +++ b/src/storage/engine/impl.h @@ -2560,12 +2560,27 @@ class StorageEngine::Impl { const MutationPrecondition* mutation_precondition = nullptr, CompactAccessOptions access = {}); + // Standalone adapters choose IS/IX only for grouped operations whose group + // set is known. Externally locked EXEC/Lua and multi-key calls bypass this. + Task> ExecuteHashLike( + std::uint8_t db_id, std::string_view key, const HashOperation& operation, + ValueType value_type, ReplicationCommandAppend* replication, + const MutationPrecondition* mutation_precondition); + // nullopt means every group hold has been released before one S/X fallback. + Task>> TryConcurrentHashLike( + std::uint8_t db_id, std::string_view key, const Digest& digest, + const HashOperation& operation, ValueType value_type, + ReplicationCommandAppend* replication, + const MutationPrecondition* mutation_precondition); + // With retry_exclusive, the caller holds IS/IX and all selected groups; + // true certifies that no mutation was staged and permits a whole-key retry. Task> ExecuteHashLikeLocked( std::uint8_t db_id, std::string_view key, const Digest& digest, const HashOperation& operation, ValueType value_type, TxShardWrites* tx = nullptr, ReplicationCommandAppend* replication = nullptr, - const MutationPrecondition* mutation_precondition = nullptr); + const MutationPrecondition* mutation_precondition = nullptr, + bool* retry_exclusive = nullptr); // Pure private-page planning; no store lock is needed. The after-image owns // selected groups for point operations or all groups for full-image callers. @@ -2581,9 +2596,10 @@ class StorageEngine::Impl { // it after their resource waits and publish without yielding; ordinary // multi-append/extent writers retain key X through the remaining waits. static std::uint64_t AllocateGroupedRevision() noexcept; - // Caller retains exclusive key intent and store state. A prepared plan must - // name the validated predecessor or a validated creation; commit assigns - // its durable revision and stamps every page of a fresh incarnation. + // Caller retains store state and key X, or IX plus X on changed groups. + // retry_exclusive requires a prepared topology-preserving bounded batch; + // true rejects before staging so the adapter can release groups and take X. + // Such batches merge with the latest root and assign revision after waits. Task CommitGroupedHashMutationLocked( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, @@ -2593,6 +2609,7 @@ class StorageEngine::Impl { ReplicationCommandAppend* replication, const MutationPrecondition* mutation_precondition = nullptr, HashGroupMutationPlan* prepared = nullptr, + bool* retry_exclusive = nullptr, OnlineLatencyTrace* diagnostics = nullptr); // The plan contains complete changed ordered pages, not an append-only @@ -2815,7 +2832,7 @@ class StorageEngine::Impl { WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::span fields, - OnlineLatencyTrace* diagnostics = nullptr); + bool group_locked = false, OnlineLatencyTrace* diagnostics = nullptr); // Full read-only replies retain only projected output under bounded owner // windows. Pages release their leases before joins, which recheck the view. @@ -4331,7 +4348,7 @@ class StorageEngine::Impl { GroupedObject::Handle object, GroupedRecordId id, bool pinned, Decode decode, GroupReadLimiter* limiter = nullptr, bool group_locked = false, OnlineLatencyTrace* diagnostics = nullptr); - // Requires parent IX plus this group's S/X hold. Refresh physical metadata + // Requires parent IS/IX plus this group's S/X hold. Refresh physical metadata // across GC without rejecting publications to unrelated groups. No pin is // implied: each physical read still owns its usual allocation lease. absl::StatusOr RefreshLockedGroup( diff --git a/tests/grouped/hash_test.cpp b/tests/grouped/hash_test.cpp index 9c4efe31c..3c4abaf97 100644 --- a/tests/grouped/hash_test.cpp +++ b/tests/grouped/hash_test.cpp @@ -565,6 +565,57 @@ std::vector Candidates( return candidates; } +TEST(GroupedHashTest, + MergePublicationPreservesDisjointChangesAndRejectsConflicts) { + auto pages = GroupHashValue(Value(128), 17, Seed(), 1024); + ASSERT_TRUE(pages.ok()); + ASSERT_GT(pages->size(), 2); + auto records = Candidates(*pages); + auto base = Recover(Root(128, pages->size()), 1, records, {}); + ASSERT_TRUE(base.ok()) << base.status(); + auto plan_root = base->root(); + --plan_root.field_count_; + auto changed = pages->front(); + ASSERT_FALSE(changed.value_.entries_.empty()); + changed.value_.entries_.pop_back(); + const auto writes = std::span(&changed, 1); + + auto other = records.back(); + ASSERT_GT(other.field_count_, 0); + --other.field_count_; + other.sequence_ = other.lsn_ = 2; + auto current_root = base->root(); + current_root.revision_ = 2; + --current_root.field_count_; + auto current = base->Apply(current_root, 2, std::span(&other, 1)); + ASSERT_TRUE(current.ok()) << current.status(); + auto merged = MergeHashPublication(*base, *current, plan_root, writes); + ASSERT_TRUE(merged.ok()) << merged.status(); + EXPECT_EQ(merged->field_count_, 126); + 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())); +} + TEST(GroupedHashTest, PrefixBoundsIncludeZeroAndFullWidth) { const auto maximum = std::numeric_limits::max(); EXPECT_TRUE((GroupedRecordId{}).ContainsHash(0)); diff --git a/tests/grouped/hash_write_e2e_test.cpp b/tests/grouped/hash_write_e2e_test.cpp index c4987ef0c..0cb5446bd 100644 --- a/tests/grouped/hash_write_e2e_test.cpp +++ b/tests/grouped/hash_write_e2e_test.cpp @@ -19,6 +19,7 @@ #include #include +#include "absl/cleanup/cleanup.h" #include "lavik/storage/detail/grouped/hash.h" #include "write_e2e_support.h" @@ -37,6 +38,7 @@ std::vector HashCommand(std::string key, char value = 'v') { struct HashDiskLayout { GroupedHashRoot root_; bool empty_tail_ = false; + std::vector leaves_; }; TEST(CompactHashWriteE2e, ColdUpdateKeepsCompactEncodingAndRecovers) { @@ -166,6 +168,10 @@ HashDiskLayout InspectHashLayout(const PrivateDisk& disk, auto& winner = newest[{record.group_prefix_, record.group_prefix_bits_}]; if (winner.mutation_sequence_ <= record.mutation_sequence_) winner = record; } + for (const auto& [id, record] : newest) + if (!record.group_retired_) + result.leaves_.push_back( + {id.first, static_cast(id.second)}); for (auto it = newest.rbegin(); it != newest.rend(); ++it) { if (!it->second.group_retired_) { result.empty_tail_ = it->second.logical_size_ == 0; @@ -175,6 +181,269 @@ HashDiskLayout InspectHashLayout(const PrivateDisk& disk, return result; } +// A paused page operation must retain its group while other groups can publish +// a new root. Decode routes from the durable fixture, not the process seed. +void CheckHashLikeIntent(bool set, bool read) { + PrivateDisk disk; + const std::string key = set ? "set-intent" : "hash-intent"; + constexpr unsigned count = 1024; + auto field = [](unsigned i) { + return "field-" + std::to_string(i) + std::string(128, 'f'); + }; + { + Server server(disk, 1); + Client client(server.port()); + std::vector seed{set ? "SADD" : "HSET", key}; + for (unsigned i = 0; i < count; ++i) { + seed.push_back(field(i)); + if (!set) seed.push_back("0"); + } + ASSERT_EQ(client.Command(seed).text_, std::to_string(count)); + client.Durable(); + } + const auto layout = InspectHashLayout(disk, key); + ASSERT_GT(layout.leaves_.size(), 2); + const auto route = [&](const std::string& member) { + const auto hash = ComputeDigest(member, layout.root_.seed_).value_; + for (auto id : layout.leaves_) + if (id.ContainsHash(hash)) return id; + throw std::runtime_error("missing fixture Hash route"); + }; + auto a = field(read ? 0 : count); + if (!read) { + // A structural retry drops IX before acquiring key X, allowing the + // queued remover to run before this insertion. This case tests ordinary + // group-lock handoff, so choose a leaf with room under the persisted seed. + std::map entry_bytes; + for (unsigned i = 0; i < count; ++i) { + const auto member = field(i); + entry_bytes[route(member).prefix_] += 8 + member.size() + (set ? 0 : 1); + } + unsigned candidate = count; + for (; candidate < 2 * count; ++candidate) { + a = field(candidate); + if (kHashValueHeaderBytes + entry_bytes[route(a).prefix_] + 8 + a.size() + + (set ? 0 : 1) <= + kCollectionGroupTargetBytes) + break; + } + ASSERT_LT(candidate, 2 * count); + } + unsigned other = 1; + while (other < count && route(field(other)) == route(a)) ++other; + ASSERT_LT(other, count); + const auto b = field(other); + const auto expected_count = + count - 1 - (read ? 1 : 0) + (!read && !set ? 1 : 0); + { + const std::string gate = disk.path() + ".intent-gate"; + std::ofstream(gate).put('1'); + ScopedEnvironment pause_key("LAVIK_HASH_GROUP_PAUSE_KEY", key.c_str()); + ScopedEnvironment pause_field("LAVIK_HASH_GROUP_PAUSE_FIELD", a.c_str()); + ScopedEnvironment pause_file("LAVIK_HASH_GROUP_PAUSE_FILE", gate.c_str()); + const std::array crc{"--record-crc"}; + Server server(disk, 1, {}, {}, false, 2, "2G", {}, {}, 0, 20, crc); + std::future first, disjoint, conflicting; + absl::Cleanup release = [&] { ::unlink(gate.c_str()); }; + first = std::async(std::launch::async, [&] { + Client client(server.port()); + if (read) + return client.Command({set ? "SMISMEMBER" : "HMGET", key, a, a}); + if (set) return client.Command({"SADD", key, a, a}); + return client.Command({"HSET", key, a, "1", a, "2"}); + }); + bool held = false; + for (unsigned i = 0; i < 200; ++i) { + held = server.Log().find("hash group locks held before page IO") != + std::string::npos; + if (held) break; + std::this_thread::sleep_for(25ms); + } + ASSERT_TRUE(held) << server.Log(); + disjoint = std::async(std::launch::async, [&] { + Client client(server.port()); + return client.Command({set ? "SREM" : "HDEL", key, b}); + }); + ASSERT_EQ(disjoint.wait_for(3s), std::future_status::ready) << server.Log(); + ASSERT_EQ(disjoint.get().text_, "1"); + Client observer(server.port()); + EXPECT_EQ(observer.Command({set ? "SCARD" : "HLEN", key}).text_, + std::to_string(count - 1)); + // Scan a different leaf while the first request still owns its group. + const auto scan = + observer.Command({set ? "SSCAN" : "HSCAN", key, + std::to_string(route(b).prefix_), "COUNT", "1"}); + EXPECT_EQ(scan.kind_, '*') << scan.text_; + conflicting = std::async(std::launch::async, [&] { + Client client(server.port()); + if (read || set) return client.Command({set ? "SREM" : "HDEL", key, a}); + return client.Command({"HINCRBY", key, a, "1"}); + }); + EXPECT_EQ(conflicting.wait_for(100ms), std::future_status::timeout); + EXPECT_EQ(first.wait_for(0ms), std::future_status::timeout); + ::unlink(gate.c_str()); + const auto result = first.get(); + if (read) { + ASSERT_EQ(result.items_.size(), 2) << result.text_ << server.Log(); + EXPECT_EQ(result.items_[0].text_, set ? "1" : "0"); + EXPECT_EQ(result.items_[1].text_, set ? "1" : "0"); + } else { + EXPECT_EQ(result.text_, "1") << server.Log(); + } + ASSERT_EQ(conflicting.get().text_, read || set ? "1" : "3") << server.Log(); + EXPECT_EQ(observer.Command({set ? "SCARD" : "HLEN", key}).text_, + std::to_string(expected_count)); + observer.Durable(); + } + Server recovered(disk, 2); + Client client(recovered.port()); + EXPECT_EQ(client.Command({set ? "SCARD" : "HLEN", key}).text_, + std::to_string(expected_count)); + EXPECT_EQ(client.Command({set ? "SISMEMBER" : "HEXISTS", key, b}).text_, "0"); + if (!read && !set) { + EXPECT_EQ(client.Command({"HGET", key, a}).text_, "3"); + EXPECT_EQ(client.Command({"HSETNX", key, a, "9"}).text_, "0"); + EXPECT_EQ(client.Command({"HINCRBYFLOAT", key, a, "0.5"}).text_, "3.5"); + EXPECT_EQ(client.Command({"HSTRLEN", key, a}).text_, "3"); + } else { + EXPECT_EQ(client.Command({set ? "SISMEMBER" : "HEXISTS", key, a}).text_, + "0"); + } +} + +void CheckHashLikeLastDeletion(bool set) { + PrivateDisk disk; + const std::string key = "intent-final-delete"; + const std::string a(18000, 'a'), b(18000, 'b'); + { + Server server(disk, 1); + Client client(server.port()); + const auto seed = + set ? std::vector{"SADD", key, a, b} + : std::vector{"HSET", key, a, "0", b, "0"}; + ASSERT_EQ(client.Command(seed).text_, "2"); + client.Durable(); + } + ASSERT_GT(InspectHashLayout(disk, key).root_.group_count_, 1); + { + const std::string gate = disk.path() + ".delete-gate"; + std::ofstream(gate).put('1'); + ScopedEnvironment pause_key("LAVIK_HASH_GROUP_PAUSE_KEY", key.c_str()); + ScopedEnvironment pause_field("LAVIK_HASH_GROUP_PAUSE_FIELD", a.c_str()); + ScopedEnvironment pause_file("LAVIK_HASH_GROUP_PAUSE_FILE", gate.c_str()); + Server server(disk, 1); + std::future first, other; + absl::Cleanup release = [&] { ::unlink(gate.c_str()); }; + first = std::async(std::launch::async, [&] { + Client client(server.port()); + return client.Command({set ? "SREM" : "HDEL", key, a}); + }); + bool held = false; + for (unsigned i = 0; i < 200; ++i) { + held = server.Log().find("hash group locks held before page IO") != + std::string::npos; + if (held) break; + std::this_thread::sleep_for(25ms); + } + ASSERT_TRUE(held) << server.Log(); + other = std::async(std::launch::async, [&] { + Client client(server.port()); + return client.Command({set ? "SREM" : "HDEL", key, b}); + }); + ASSERT_EQ(other.wait_for(3s), std::future_status::ready) << server.Log(); + ASSERT_EQ(other.get().text_, "1"); + ::unlink(gate.c_str()); + ASSERT_EQ(first.wait_for(5s), std::future_status::ready) << server.Log(); + EXPECT_EQ(first.get().text_, "1") << server.Log(); + Client client(server.port()); + EXPECT_EQ(client.Command({"EXISTS", key}).text_, "0"); + client.Durable(); + } + Server recovered(disk, 1); + Client client(recovered.port()); + EXPECT_EQ(client.Command({"EXISTS", key}).text_, "0"); +} + +TEST(GroupedHashWriteE2e, ConcurrentFinalDeletionRetriesUnderKeyX) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires a deterministic group-lock pause"; +#endif + CheckHashLikeLastDeletion(false); +} + +TEST(GroupedHashWriteE2e, StructuralRetryReleasesGroupsBeforeTakingKeyX) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires a deterministic group-lock pause"; +#endif + PrivateDisk disk; + const std::string key = "hash-structural-retry"; + const std::string field = "field0"; + const std::string large(70000, 'x'); + { + Server server(disk, 1); + Client client(server.port()); + ASSERT_EQ(client.Command(HashCommand(key)).text_, "256"); + client.Durable(); + } + { + const std::string gate = disk.path() + ".structural-gate"; + std::ofstream(gate).put('1'); + ScopedEnvironment pause_key("LAVIK_HASH_GROUP_PAUSE_KEY", key.c_str()); + ScopedEnvironment pause_field("LAVIK_HASH_GROUP_PAUSE_FIELD", + field.c_str()); + ScopedEnvironment pause_file("LAVIK_HASH_GROUP_PAUSE_FILE", gate.c_str()); + Server server(disk, 1); + std::future first, conflict; + absl::Cleanup release = [&] { ::unlink(gate.c_str()); }; + first = std::async(std::launch::async, [&] { + Client client(server.port()); + return client.Command({"HSET", key, field, large}); + }); + bool held = false; + for (unsigned i = 0; i < 200; ++i) { + held = server.Log().find("hash group locks held before page IO") != + std::string::npos; + if (held) break; + std::this_thread::sleep_for(25ms); + } + ASSERT_TRUE(held) << server.Log(); + conflict = std::async(std::launch::async, [&] { + Client client(server.port()); + return client.Command({"HDEL", key, field}); + }); + EXPECT_EQ(conflict.wait_for(100ms), std::future_status::timeout); + ::unlink(gate.c_str()); + ASSERT_EQ(conflict.wait_for(5s), std::future_status::ready) << server.Log(); + EXPECT_EQ(conflict.get().text_, "1"); + ASSERT_EQ(first.wait_for(5s), std::future_status::ready) << server.Log(); + // The waiting IX delete runs before the released writer can get key X. + // Re-execution must observe that deletion and report an insertion. + EXPECT_EQ(first.get().text_, "1") << server.Log(); + Client client(server.port()); + EXPECT_EQ(client.Command({"HLEN", key}).text_, "256"); + EXPECT_EQ(client.Command({"HGET", key, field}).text_, large); + client.Durable(); + } + Server recovered(disk, 1); + Client client(recovered.port()); + EXPECT_EQ(client.Command({"HLEN", key}).text_, "256"); + EXPECT_EQ(client.Command({"HGET", key, field}).text_, large); +} + +TEST(GroupedHashWriteE2e, IntentWritesMergeDisjointCountsAndWakeConflicts) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires a deterministic group-lock pause"; +#endif + CheckHashLikeIntent(false, false); +} + +TEST(GroupedHashWriteE2e, IntentReadsProtectOnlyRequestedLeaves) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires a deterministic group-lock pause"; +#endif + CheckHashLikeIntent(false, true); +} + TEST(GroupedHashWriteE2e, DemotesBelowStrictGroupPayloadThresholdAfterRecovery) { PrivateDisk disk; From bd2f382ef6fe3f43b58d8d422e5d5eb7fa3a288d Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Thu, 5 Nov 2026 16:00:57 +0000 Subject: [PATCH 2/2] fix(storage): discard speculative replication effects on hash intent retry --- src/storage/engine/hash_api.cpp | 18 +++ src/storage/engine/write.cpp | 17 +++ tests/grouped/hash_write_e2e_test.cpp | 158 ++++++++++++++++++++++++++ 3 files changed, 193 insertions(+) diff --git a/src/storage/engine/hash_api.cpp b/src/storage/engine/hash_api.cpp index 5a754f172..b27219c87 100644 --- a/src/storage/engine/hash_api.cpp +++ b/src/storage/engine/hash_api.cpp @@ -86,11 +86,29 @@ Task> StorageEngine::Impl::ExecuteHashLike( lavik::storage::RedisSlot(key)); key_wait.End(); if (concurrent) { + // The speculative writer can canonicalize increments/pops and append TTL + // effects before its final batch admission requests key X. Preserve the + // original command: after IX is released, X may observe another incarnation + // with different contents and expiry. Only replicated intent writers need + // this copy; ordinary compact operations retain their existing fast path. + std::optional retry_replication; + if (write && replication) { + retry_replication.emplace(*replication); + // The snapshot owns distinct argument buffers, not the original charge. + retry_replication->retained_owner_.reset(); + AccountReplicationCommand(&*retry_replication); + } auto result = co_await TryConcurrentHashLike(db_id, key, digest, operation, value_type, replication, mutation_precondition); if (!result.ok()) co_return result.status(); if (*result) co_return std::move(**result); + if (retry_replication) { + // Discard every attempt-derived effect and publication coordinate. The + // retry accounts its newly canonicalized buffers at append admission. + retry_replication->retained_owner_.reset(); + *replication = std::move(*retry_replication); + } // The nested task has released every group. Upgrading an intent while // retaining groups could deadlock another intent holder waiting for them. key_lock.Reset(); diff --git a/src/storage/engine/write.cpp b/src/storage/engine/write.cpp index e4a3e2a08..8a02800f9 100644 --- a/src/storage/engine/write.cpp +++ b/src/storage/engine/write.cpp @@ -2281,6 +2281,23 @@ Task StorageEngine::Impl::AppendLocked( fullsync_command = std::move(owned_copy); } } + LAVIK_FAULT_INJECT({ + if (replication && grouped && grouped->root_ && grouped->root_->batch_ && + LAVIK_FAULT_MATCHES("LAVIK_GROUP_REPLICATION_PAUSE_KEY", key)) { + const char* file = std::getenv("LAVIK_GROUP_REPLICATION_PAUSE_FILE"); + // One prepared attempt pauses after canonical TTL capture. Other groups + // must still publish so its final batch admission can require key X. + if (file && ::unlink((std::string(file) + ".arm").c_str()) == 0) { + spdlog::info("grouped replication prepared before publication key={}", + key); + store.store_state_mutex_.Unlock(*store.worker_); + const auto paused = co_await fault_injection::PauseWhileFileExists( + "LAVIK_GROUP_REPLICATION_PAUSE_FILE"); + co_await store.store_state_mutex_.Lock(); + if (!paused.ok()) co_return paused; + } + } + }); absl::Status status = absl::OkStatus(); const bool key_indirect = key.size() > kInlineKeyMaxBytes; const std::uint64_t logical_payload_bytes = diff --git a/tests/grouped/hash_write_e2e_test.cpp b/tests/grouped/hash_write_e2e_test.cpp index 0cb5446bd..9c1c31189 100644 --- a/tests/grouped/hash_write_e2e_test.cpp +++ b/tests/grouped/hash_write_e2e_test.cpp @@ -181,6 +181,164 @@ HashDiskLayout InspectHashLayout(const PrivateDisk& disk, return result; } +// A late IX rejection must not carry its captured expiration into the next +// incarnation. Use a real native replica and a same-partition barrier so an +// expired stale PEXPIREAT cannot hide behind replication lag. +void CheckHashLikeRetryReplication(bool set) { + PrivateDisk primary_disk, replica_disk; + const std::string key = "collection{retry-ttl}"; + const std::string barrier = "barrier{retry-ttl}"; + const std::string a = set ? std::string(10000, 'a') : "a"; + const std::string b = set ? std::string(10000, 'b') : "b"; + { + Server seed(primary_disk, 1); + Client client(seed.port()); + const auto reply = + set ? client.Command({"SADD", key, a, b, "keep"}) + : client.Command({"HSET", key, a, std::string(10000, 'a'), b, + std::string(10000, 'b'), "keep", "old"}); + ASSERT_EQ(reply.text_, "3"); + client.Durable(); + ASSERT_EQ(seed.Wait(true), 0) << seed.Log(); + } + const auto layout = InspectHashLayout(primary_disk, key); + const auto route = [&](std::string_view field) { + const auto digest = ComputeDigest(field, layout.root_.seed_).value_; + for (const auto id : layout.leaves_) + if (id.ContainsHash(digest)) return id; + throw std::runtime_error("missing TTL fixture route"); + }; + ASSERT_NE(route(a), route(b)); + const std::string gate = primary_disk.path() + ".replication-gate"; + std::ofstream(gate).put('1'); + std::ofstream(gate + ".arm").put('1'); + std::unique_ptr primary; + { + // Only the primary inherits the fault; replica replay uses its normal path. + ScopedEnvironment pause_key("LAVIK_GROUP_REPLICATION_PAUSE_KEY", + key.c_str()); + ScopedEnvironment pause_file("LAVIK_GROUP_REPLICATION_PAUSE_FILE", + gate.c_str()); + primary = std::make_unique(primary_disk, 1); + } + Server replica(replica_disk, 1); + Client source(primary->port()), follower(replica.port()); + std::future deleting, disjoint, rebuilding; + absl::Cleanup release = [&] { + ::unlink(gate.c_str()); + ::unlink((gate + ".arm").c_str()); + }; + ASSERT_EQ(follower + .Command({"LAVIK.REPLICAOF", "127.0.0.1", + std::to_string(primary->port())}) + .text_, + "OK"); + ASSERT_EQ(follower.Command({"READONLY"}).text_, "OK"); + auto await_barrier = [&](std::string_view expected) { + for (unsigned i = 0; i < 600; ++i) { + const auto reply = follower.Command({"GET", barrier}); + if (reply.kind_ == '$' && reply.text_ == expected) return true; + std::this_thread::sleep_for(25ms); + } + return false; + }; + ASSERT_EQ(source.Command({"SET", barrier, "before"}).text_, "OK"); + ASSERT_TRUE(await_barrier("before")) << primary->Log() << replica.Log(); + const auto expires = std::chrono::system_clock::now() + 10s; + const auto expires_ms = std::chrono::duration_cast( + expires.time_since_epoch()) + .count(); + ASSERT_EQ( + source.Command({"PEXPIREAT", key, std::to_string(expires_ms)}).text_, + "1"); + deleting = std::async(std::launch::async, [&] { + Client client(primary->port()); + return client.Command({set ? "SREM" : "HDEL", key, a}); + }); + bool paused = false; + for (unsigned i = 0; i < 200; ++i) { + paused = primary->Log().find( + "grouped replication prepared before publication") != + std::string::npos; + if (paused) break; + std::this_thread::sleep_for(25ms); + } + ASSERT_TRUE(paused) << primary->Log(); + // Each deletion alone stays above demotion. Their combined image drops + // below the demotion threshold, so A must fail its final prepared admission. + disjoint = std::async(std::launch::async, [&] { + Client client(primary->port()); + return client.Command({set ? "SREM" : "HDEL", key, b}); + }); + ASSERT_EQ(disjoint.wait_for(3s), std::future_status::ready) << primary->Log(); + ASSERT_EQ(disjoint.get().text_, "1"); + ASSERT_LT(std::chrono::system_clock::now(), expires); + rebuilding = std::async(std::launch::async, [&] { + Client client(primary->port()); + Check(client.Command({"MULTI"}).text_ == "OK", "MULTI failed"); + Check(client.Command({"DEL", key}).text_ == "QUEUED", "DEL not queued"); + const auto write = + set ? client.Command({"SADD", key, "keep", a}) + : client.Command({"HSET", key, "keep", "safe", a, "remove-me"}); + Check(write.text_ == "QUEUED", "recreation not queued"); + return client.Command({"EXEC"}); + }); + EXPECT_EQ(rebuilding.wait_for(100ms), std::future_status::timeout); + // The queued key-X recreation wins when A releases IX. Make the old TTL + // already expired so a leaked effect destroys the fresh non-expiring key. + std::this_thread::sleep_until(expires + 50ms); + ::unlink(gate.c_str()); + ASSERT_EQ(rebuilding.wait_for(5s), std::future_status::ready) + << primary->Log(); + const auto rebuilt = rebuilding.get(); + ASSERT_EQ(rebuilt.items_.size(), 2) << rebuilt.text_ << primary->Log(); + EXPECT_EQ(rebuilt.items_[1].text_, "2"); + ASSERT_EQ(deleting.wait_for(5s), std::future_status::ready) << primary->Log(); + EXPECT_EQ(deleting.get().text_, "1"); + EXPECT_EQ(source.Command({set ? "SCARD" : "HLEN", key}).text_, "1"); + EXPECT_EQ(source.Command({set ? "SISMEMBER" : "HGET", key, "keep"}).text_, + set ? "1" : "safe"); + EXPECT_EQ(source.Command({"PTTL", key}).text_, "-1"); + ASSERT_EQ(source.Command({"SET", barrier, "after"}).text_, "OK"); + ASSERT_TRUE(await_barrier("after")) << primary->Log() << replica.Log(); + EXPECT_EQ(follower.Command({set ? "SCARD" : "HLEN", key}).text_, "1") + << replica.Log(); + EXPECT_EQ(follower.Command({set ? "SISMEMBER" : "HGET", key, "keep"}).text_, + set ? "1" : "safe"); + EXPECT_EQ(follower.Command({"PTTL", key}).text_, "-1"); + + // The successful IX path must still publish its current TTL. Re-promote, + // then remove one large field while the other keeps the graph grouped. + const auto expanded = + set ? source.Command({"SADD", key, a, b}) + : source.Command({"HSET", key, a, std::string(10000, 'a'), b, + std::string(10000, 'b')}); + ASSERT_EQ(expanded.text_, "2"); + const auto new_expiry = + std::chrono::duration_cast( + (std::chrono::system_clock::now() + 60s).time_since_epoch()) + .count(); + ASSERT_EQ( + source.Command({"PEXPIREAT", key, std::to_string(new_expiry)}).text_, + "1"); + ASSERT_EQ(source.Command({set ? "SREM" : "HDEL", key, b}).text_, "1"); + ASSERT_EQ(source.Command({"SET", barrier, "retained"}).text_, "OK"); + ASSERT_TRUE(await_barrier("retained")) << primary->Log() << replica.Log(); + EXPECT_EQ(source.Command({"PEXPIRETIME", key}).text_, + std::to_string(new_expiry)); + EXPECT_EQ(follower.Command({"PEXPIRETIME", key}).text_, + std::to_string(new_expiry)); + EXPECT_EQ(follower.Command({set ? "SCARD" : "HLEN", key}).text_, "2"); +} + +TEST(GroupedHashWriteE2e, + IntentRetryDiscardsPreviousIncarnationReplicationTTL) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires a deterministic prepared replication pause"; +#endif + CheckHashLikeRetryReplication(false); +} + // A paused page operation must retain its group while other groups can publish // a new root. Decode routes from the durable fixture, not the process seed. void CheckHashLikeIntent(bool set, bool read) {