diff --git a/docs/architecture/09-grouped-collections.md b/docs/architecture/09-grouped-collections.md index 9aea44e89..85eb0bc8d 100644 --- a/docs/architecture/09-grouped-collections.md +++ b/docs/architecture/09-grouped-collections.md @@ -327,6 +327,21 @@ cleaner disabling or the allocator's authoritative capacity checks. ## Command access paths +Compact Hash/Set point edits, List end edits, single-member Sorted Set writes +and bounded String range edits can prepare borrowed after-images. Their read +leases and command operands remain immutable through append allocation waits; +only framing and changed command data need separate ownership. The append +funnel consumes each encoder after its final suspension, then publishes the +same complete checked record format. Published records and commit receipts +retain no borrowed payload. +Remote read leases and local read-pool exhaustion use owned output so a queued +writer does not retain the last read slot needed by an earlier state-lock owner. +Ordinary existing compact String writes retain key intent and database +admission while releasing worker store state for the old-value read and private +preparation; reacquisition validates logical identity and population, allowing +physical GC relocation. Transactional and external-value preparation retain +their existing locked paths. Promotion uses the owned collection builders. + Whole-Hash replacement (`LAVIK.HREPLACE`) checks existing key/type/expiry and the preceding grouped decision, but does not load old field payloads. Its after-image comes entirely from admitted request data. A large replacement diff --git a/include/lavik/storage/detail/compact_payload.h b/include/lavik/storage/detail/compact_payload.h new file mode 100644 index 000000000..181da04a4 --- /dev/null +++ b/include/lavik/storage/detail/compact_payload.h @@ -0,0 +1,75 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include + +#include "absl/container/inlined_vector.h" +#include "lavik/storage/detail/record_payload_cursor.h" + +namespace lavik::storage { + +// A preflighted compact after-image: framing is owned, payload fragments borrow +// immutable read leases or command operands. Keep those owners alive through +// append's allocation waits. The plan may move before encoding, because owned +// framing uses no self-referential views. Encoding consumes it once, only after +// the writer's final suspension; it allocates and publishes nothing itself. +class CompactPayload { + public: + void AppendBytes(std::string_view bytes) { + if (bytes.empty()) return; + assert(bytes.size() <= SIZE_MAX - bytes_); + bytes_ += bytes.size(); + parts_.push_back(Part{.borrowed = bytes}); + } + + // Encodes fixed-width little-endian framing, independent of host endianness. + void AppendUnsigned(std::uint64_t value, unsigned width) { + assert(width <= 8 && width > 0 && bytes_ <= SIZE_MAX - width); + if (parts_.empty() || parts_.back().owned_size == 0 || + parts_.back().owned_size + width > parts_.back().owned.size()) + parts_.push_back(Part{}); + auto& part = parts_.back(); + for (unsigned i = 0; i < width; ++i) + part.owned[part.owned_size++] = static_cast(value >> (8 * i)); + bytes_ += width; + } + + std::size_t encoded_bytes() const noexcept { return bytes_; } + bool empty() const noexcept { return bytes_ == 0; } + std::optional Next() noexcept { + if (next_ == parts_.size()) return std::nullopt; + const auto& part = parts_[next_++]; + return part.owned_size + ? std::string_view(part.owned.data(), part.owned_size) + : part.borrowed; + } + + private: + struct Part { + std::string_view borrowed{}; + std::array owned{}; + unsigned owned_size = 0; + }; + absl::InlinedVector parts_; + std::size_t bytes_ = 0; + std::size_t next_ = 0; +}; + +} // namespace lavik::storage diff --git a/include/lavik/storage/detail/hash_codec.h b/include/lavik/storage/detail/hash_codec.h index 1e7e99397..bf27040b3 100644 --- a/include/lavik/storage/detail/hash_codec.h +++ b/include/lavik/storage/detail/hash_codec.h @@ -24,6 +24,7 @@ #include #include "absl/status/statusor.h" +#include "lavik/storage/detail/compact_payload.h" #include "lavik/storage/format.h" namespace lavik::storage { @@ -78,6 +79,27 @@ class HashValueReader { absl::StatusOr DecodeHashValue(std::string_view payload); absl::StatusOr EncodeHashValue(const HashValue& value); +enum class CompactHashEditKind { kSet, kDelete }; + +struct CompactHashEditPlan { + bool changed_ = false; + std::size_t added_ = 0; + std::size_t removed_ = 0; + std::size_t count_ = 0; + CompactPayload payload_; +}; + +// Validates the complete old image and builds a bounded splice plan. Sources +// must remain immutable through encoding, including append allocation waits. +// Duplicate operands retain sequential command semantics. Delete accepts no +// values. Nullopt selects promotion without allocating an oversized image. +// Callers admit O(fields.size()) planning space before entering this helper. +absl::StatusOr> PlanCompactHashEdit( + std::string_view payload, std::size_t expected_count, + std::span fields, + std::span values, std::size_t max_bytes, + CompactHashEditKind kind = CompactHashEditKind::kSet); + // A bounded HSET after-image. No-op results have no payload; a changed result // owns its bytes independently of both the old payload and request operands. struct CompactHashEdit { diff --git a/include/lavik/storage/detail/ordered_compact_codec.h b/include/lavik/storage/detail/ordered_compact_codec.h index f155e1119..4369df18a 100644 --- a/include/lavik/storage/detail/ordered_compact_codec.h +++ b/include/lavik/storage/detail/ordered_compact_codec.h @@ -20,6 +20,33 @@ namespace lavik::storage { +struct OrderedCompactEntryView { + std::string_view value_; + double score_ = 0; + std::string_view encoded_; +}; + +// Checked traversal without per-element ownership. Consume exactly size() +// entries to validate the tail. Returned views borrow the immutable payload; +// copying a reader preserves its position for a second planning pass. +class OrderedCompactReader { + public: + static absl::StatusOr Open( + OrderedCollectionKind kind, std::string_view encoded, + std::uint64_t expected_count); + std::size_t size() const noexcept { return size_; } + absl::StatusOr Next(); + + private: + OrderedCompactReader(std::string_view bytes, std::size_t count, bool sorted) + : bytes_(bytes), size_(count), remaining_(count), sorted_(sorted) {} + std::string_view bytes_; + std::size_t size_; + std::size_t remaining_; + std::size_t offset_ = 8; + bool sorted_; +}; + // Logical full-image bridge used by existing callback/RDB integrations. // A collection can span many physical pages, so aggregate bytes have no // String or record-payload limit here. Each item remains limited to 512 MiB diff --git a/include/lavik/storage/engine.h b/include/lavik/storage/engine.h index 3384e612d..2e6f0d0db 100644 --- a/include/lavik/storage/engine.h +++ b/include/lavik/storage/engine.h @@ -56,6 +56,8 @@ class ReplicationHistory; namespace lavik::storage { +class InlineRecordPayload; + struct StorageEngineOptions { // Runtime databases [0, database_count_). Cluster servers use only DB0; // durable and replication epoch metadata retain all 16 database slots. @@ -995,6 +997,11 @@ struct CompactValueUpdate { // DELCONSUMER receives its removed PEL cardinality after storage has scanned // and planned the selected consumer's removals, before publishing the root. std::function stream_pending_removed_; + // Optional after-image for an existing compact Sorted Set write, exclusive + // with encoded_/reuse_encoded_. The callback owner retains the encoder and + // its borrowed sources until the awaited ExecuteCompact call returns. It + // must stay below promotion size. + const InlineRecordPayload* inline_payload_ = nullptr; }; // Inclusive/exclusive 128-bit ID bounds for a read-only Stream callback. The diff --git a/src/storage/engine/compact_api.cpp b/src/storage/engine/compact_api.cpp index d6eec8614..dd206d62b 100644 --- a/src/storage/engine/compact_api.cpp +++ b/src/storage/engine/compact_api.cpp @@ -366,6 +366,30 @@ Task StorageEngine::Impl::ExecuteCompactLocked( auto update = callback(view); if (!update.ok()) co_return update.status(); + if (update->inline_payload_ && + (grouped || !exists || !loaded || !update->changed_ || update->erase_ || + update->reuse_encoded_ || !update->encoded_.empty() || + value_type != ValueType::kSortedSet || + update->inline_payload_->encoded_bytes() >= kCollectionPromotionBytes)) + co_return absl::InternalError("invalid compact callback encoder"); + if (update->inline_payload_ && loaded->lease_.buffer_id() != 0 && + (loaded->lease_.owner_worker() != store.worker_->id() || + store.buffers_.available_read_buffers() == 0)) { + // A queued writer must not pin the last read slot needed by a state-lock + // owner ahead of it. Materialize under the callback's existing admission + // before reacquiring state; remote pools cannot be queried on this worker. + update->encoded_.resize(update->inline_payload_->encoded_bytes()); + auto encoded = update->inline_payload_->WriteTo( + std::as_writable_bytes(std::span(update->encoded_))); + if (!encoded.ok()) co_return encoded; + update->inline_payload_ = nullptr; + } + if (!grouped && !update->inline_payload_ && !update->reuse_encoded_) { + // Owned callback output (or a no-op) no longer needs the read slot while + // waiting for publication state. Grouped/reuse paths still borrow it. + loaded.reset(); + view.reset(); + } // A missing key may be created directly as a large graph. Keep both // indexes private/admitted until absence and population are revalidated; // the commit adapter stamps their placeholder incarnations together. @@ -541,7 +565,8 @@ Task StorageEngine::Impl::ExecuteCompactLocked( store, partition, db_id, key, digest, update->erase_ ? std::string_view{} : encoded, kind, published_type, expire_at_ms, tx, update->erase_ ? 0 : logical_size, nullptr, nullptr, - replication, true, nullptr, mutation_precondition); + replication, true, nullptr, mutation_precondition, nullptr, + update->inline_payload_); co_return status; } diff --git a/src/storage/engine/grouped/string.cpp b/src/storage/engine/grouped/string.cpp index ba75b98d2..1afcf3463 100644 --- a/src/storage/engine/grouped/string.cpp +++ b/src/storage/engine/grouped/string.cpp @@ -234,25 +234,70 @@ StorageEngine::Impl::ExecuteStringSegmentLocked( } }; if (!grouped) { - std::string bytes; + const auto extents = exists ? ExtentsFor(store, found) : ExtentManifest{}; + const bool prepare_unlocked = + !read_only && exists && + CanPrepareCompactWriteUnlocked(store, partition, found, location, tx); + const auto snapshot = CaptureCompactWriteSnapshot(store, partition, db_id); + if (prepare_unlocked) { + // Key intent and DB admission survive this interval. Owned read leases + // protect borrowed bytes; logical identity, not GC coordinates, is + // revalidated before publication, just like compact Hash/List writes. + found = nullptr; + unlock.Unlock(); + LAVIK_FAULT_INJECT({ + auto paused = co_await PauseCompactWriteForTest(*store.worker_, key); + if (!paused.ok()) co_return paused; + }); + } + std::optional loaded; + std::string_view before; if (exists) { - auto loaded = co_await LoadValue(store, partition, db_id, key, digest, - location, ExtentsFor(store, found)); - if (!loaded.ok()) co_return loaded.status(); - auto data = loaded->value(); - bytes.assign(reinterpret_cast(data.data()), data.size()); + auto value = co_await LoadValue(store, partition, db_id, key, digest, + location, extents); + if (!value.ok()) co_return value.status(); + loaded.emplace(std::move(*value)); + const auto data = loaded->value(); + before = {reinterpret_cast(data.data()), data.size()}; + } + std::string bytes; + CompactPayload plan; + std::optional inline_payload; + // Waiting for state while holding the last read slot could block an + // earlier lock owner. Remote pool state is likewise not ours to inspect. + const bool borrow_source = + !loaded || loaded->lease_.buffer_id() == 0 || + (loaded->lease_.owner_worker() == store.worker_->id() && + store.buffers_.available_read_buffers() != 0); + if (!read_only && !bit && start <= before.size() && + new_size < kCollectionPromotionBytes && borrow_source) { + plan.AppendBytes(before.substr(0, start)); + plan.AppendBytes(operation.value_); + plan.AppendBytes( + before.substr(std::min(start + count, before.size()))); + inline_payload.emplace(plan); + } else { + bytes.assign(before); + if (!read_only) bytes.resize(new_size, '\0'); + apply(bytes, 0); + loaded.reset(); + } + if (prepare_unlocked) { + co_await store.store_state_mutex_.Lock(); + unlock.Adopt(); + const auto valid = ValidateCompactWriteSnapshot( + store, partition, db_id, key, digest, location, snapshot); + if (!valid.ok()) co_return valid; } - if (!read_only) bytes.resize(new_size, '\0'); - apply(bytes, 0); if (operation.kind_ == Kind::kSetBit && start < old_size && result.bit_ == operation.bit_) co_return result; if (!read_only) { const auto status = co_await AppendLocked( store, partition, db_id, key, digest, bytes, RecordKind::kValue, - ValueType::kString, exists ? location.expire_at_ms_ : 0, tx, - bytes.size(), nullptr, nullptr, replication, true, nullptr, - mutation_precondition); + ValueType::kString, exists ? location.expire_at_ms_ : 0, tx, new_size, + nullptr, nullptr, replication, true, nullptr, mutation_precondition, + nullptr, inline_payload ? &*inline_payload : nullptr); if (!status.ok()) co_return status; } } else { diff --git a/src/storage/engine/grouped/zset.cpp b/src/storage/engine/grouped/zset.cpp index 9722a880d..36e9ed733 100644 --- a/src/storage/engine/grouped/zset.cpp +++ b/src/storage/engine/grouped/zset.cpp @@ -634,10 +634,96 @@ absl::StatusOr ReserveInputs( return budget.Reserve(1); } +absl::StatusOr> PlanCompactSortedSetPoint( + std::optional value, const SortedSetOperation& operation, + Members& members, SortedSetResult& result, CompactPayload& output) { + const bool add = operation.kind_ == SortedSetOperationKind::kAdd; + if (!value || (add ? operation.entries_.size() != 1 + : operation.kind_ != SortedSetOperationKind::kRemove || + operation.members_.size() != 1)) + return std::nullopt; + const auto member = + add ? operation.entries_.front().member_ : operation.members_.front(); + // Conservatively leave near-threshold additions to the owned promotion + // path, before applying input semantics or populating the reply. + if (value->encoded_.size() >= kCollectionPromotionBytes || + (add && member.size() + 12 >= + kCollectionPromotionBytes - value->encoded_.size())) + return std::nullopt; + auto reader = OrderedCompactReader::Open( + OrderedCollectionKind::kSortedSet, value->encoded_, value->logical_size_); + if (!reader.ok()) return reader.status(); + auto scan = *reader; + auto& state = members.at(member); + std::optional previous; + for (std::size_t i = 0; i < scan.size(); ++i) { + auto entry = scan.Next(); + if (!entry.ok()) return entry.status(); + if (previous && (previous->score_ > entry->score_ || + (previous->score_ == entry->score_ && + previous->value_ > entry->value_))) + return absl::DataLossError("unsorted compact Sorted Set"); + previous = *entry; + if (entry->value_ == member) { + if (state.before_) + return absl::DataLossError("duplicate persisted Sorted Set member"); + state.before_ = state.after_ = entry->score_; + } + } + result.key_exists_ = true; + result.length_ = value->logical_size_; + auto applied = ApplyInputs(operation, &members, &result); + if (!applied.ok()) return applied; + CompactValueUpdate update; + if (result.changed_ == 0) return update; + update.changed_ = true; + update.logical_size_ = result.length_; + if (result.length_ == 0) { + update.erase_ = true; + return update; + } + output.AppendBytes("LZS1"); + output.AppendUnsigned(result.length_, 4); + auto append = [&] { + output.AppendUnsigned(std::bit_cast(*state.after_), 8); + output.AppendUnsigned(member.size(), 4); + output.AppendBytes(member); + }; + bool inserted = !state.after_.has_value(); + // Only the removed and inserted positions split the borrowed old image. + // Keep whole intervals rather than one descriptor per unchanged member. + std::size_t kept_begin = 8; + for (std::size_t i = 0; i < reader->size(); ++i) { + auto entry = reader->Next(); + if (!entry.ok()) return entry.status(); + const auto entry_begin = static_cast(entry->encoded_.data() - + value->encoded_.data()); + if (entry->value_ == member) { + output.AppendBytes( + value->encoded_.substr(kept_begin, entry_begin - kept_begin)); + kept_begin = entry_begin + entry->encoded_.size(); + continue; + } + if (!inserted && + (*state.after_ < entry->score_ || + (*state.after_ == entry->score_ && member < entry->value_))) { + output.AppendBytes( + value->encoded_.substr(kept_begin, entry_begin - kept_begin)); + append(); + kept_begin = entry_begin; + inserted = true; + } + } + output.AppendBytes(value->encoded_.substr(kept_begin)); + if (!inserted) append(); + return update; +} + absl::StatusOr ApplyCompactSortedSet( std::optional value, const SortedSetOperation& operation, Members& members, SortedSetResult& result, - std::optional& compact_admission) { + std::optional& compact_admission, + CompactPayload* point_payload = nullptr) { // ExecuteCompactLocked owns decode/re-encode admission, including the // bounded inline workspace that lets shrinking commands run at // maxmemory. Range removal adds no input members or retained reply; @@ -661,6 +747,12 @@ absl::StatusOr ApplyCompactSortedSet( if (!admission.ok()) return admission.status(); compact_admission.emplace(std::move(*admission)); } + if (point_payload) { + auto point = PlanCompactSortedSetPoint(value, operation, members, result, + *point_payload); + if (!point.ok()) return point.status(); + if (*point) return std::move(**point); + } std::vector entries; if (value) { auto decoded = @@ -951,10 +1043,17 @@ StorageEngine::Impl::ExecuteSortedSetLocked( // The encoded callback result survives until the compact writer returns. // Keep its admission outside the synchronous callback's stack frame. std::optional compact_admission; + CompactPayload point_payload; + std::optional inline_payload; auto callback = [&](std::optional value) -> absl::StatusOr { - return ApplyCompactSortedSet(value, operation, members, result, - compact_admission); + auto update = ApplyCompactSortedSet(value, operation, members, result, + compact_admission, &point_payload); + if (update.ok() && !point_payload.empty()) { + inline_payload.emplace(point_payload); + update->inline_payload_ = &*inline_payload; + } + return update; }; // kAdd also implements ZINCRBY and GEOADD, so the operation kind alone is // insufficient: only explicit single-key command admission opts in. diff --git a/src/storage/engine/hash_codec.cpp b/src/storage/engine/hash_codec.cpp index f875d42f0..1f05a4a49 100644 --- a/src/storage/engine/hash_codec.cpp +++ b/src/storage/engine/hash_codec.cpp @@ -158,11 +158,13 @@ absl::StatusOr DecodeHashValue(std::string_view payload) { return value; } -absl::StatusOr> EditCompactHash( +absl::StatusOr> PlanCompactHashEdit( std::string_view payload, std::size_t expected_count, std::span fields, - std::span values, std::size_t max_bytes) { - if (fields.size() != values.size()) + std::span values, std::size_t max_bytes, + CompactHashEditKind kind) { + const bool erase = kind == CompactHashEditKind::kDelete; + if ((!erase && fields.size() != values.size()) || (erase && !values.empty())) return absl::InvalidArgumentError("Hash field/value mismatch"); auto reader = HashValueReader::Open(payload); if (!reader.ok()) return reader.status(); @@ -190,21 +192,22 @@ absl::StatusOr> EditCompactHash( edits.reserve(fields.size()); if (fields.size() > 1) positions.reserve(fields.size()); for (std::size_t i = 0; i < fields.size(); ++i) { + const auto value = erase ? std::string_view{} : values[i]; auto size = AppendHashEntrySize(kHashValueHeaderBytes, fields[i].size(), - values[i].size(), std::string{}.max_size()); + value.size(), std::string{}.max_size()); if (!size.ok()) return size.status(); if (fields.size() > 1) { const auto [it, inserted] = positions.emplace(fields[i], edits.size()); if (!inserted) { auto& edit = edits[it->second]; - edit.varied |= edit.value != values[i]; - edit.value = values[i]; + edit.varied |= edit.value != value; + edit.value = value; continue; } } - edits.push_back({fields[i], values[i]}); + edits.push_back({fields[i], value}); } - CompactHashEdit result{.count_ = expected_count, .payload_ = {}}; + CompactHashEditPlan result{.count_ = expected_count, .payload_ = {}}; std::size_t bytes = payload.size(); for (std::size_t i = 0; i < reader->size(); ++i) { auto entry = reader->Next(); @@ -221,20 +224,25 @@ absl::StatusOr> EditCompactHash( if (edit.found) return absl::DataLossError("duplicate requested Hash field"); edit.found = true; - result.changed_ |= edit.varied || edit.value != entry->value_; - if (edit.value == entry->value_) continue; + result.changed_ |= erase || edit.varied || edit.value != entry->value_; + if (!erase && edit.value == entry->value_) continue; const auto begin = static_cast(entry->field_.data() - payload.data()) - 8; const auto end = begin + 8 + entry->field_.size() + entry->value_.size(); - auto size = - AppendHashEntrySize(bytes - (end - begin), edit.field.size(), - edit.value.size(), std::string{}.max_size()); - if (!size.ok()) return size.status(); - bytes = *size; + bytes -= end - begin; + if (erase) { + ++result.removed_; + } else { + auto size = + AppendHashEntrySize(bytes, edit.field.size(), edit.value.size(), + std::string{}.max_size()); + if (!size.ok()) return size.status(); + bytes = *size; + } splices.push_back({begin, end, index}); } for (const auto& edit : edits) { - if (edit.found) continue; + if (erase || edit.found) continue; auto size = AppendHashEntrySize(bytes, edit.field.size(), edit.value.size(), std::string{}.max_size()); if (!size.ok()) return size.status(); @@ -244,36 +252,53 @@ absl::StatusOr> EditCompactHash( if (result.added_ > UINT32_MAX - result.count_) return absl::OutOfRangeError("Hash element count overflow"); result.count_ += result.added_; + result.count_ -= result.removed_; result.changed_ |= result.added_ != 0; if (bytes > max_bytes) return std::nullopt; - if (!result.changed_) return result; + if (!result.changed_ || result.count_ == 0) return result; auto& output = result.payload_; - output.reserve(bytes); - AppendU64(&output, kHashValueMagic); - AppendU32(&output, kStorageFormatVersion); - AppendU32(&output, kHashValueHeaderBytes); - AppendU32(&output, result.count_); - AppendU32(&output, 0); - AppendU64(&output, bytes); + output.AppendUnsigned(kHashValueMagic, 8); + output.AppendUnsigned(kStorageFormatVersion, 4); + output.AppendUnsigned(kHashValueHeaderBytes, 4); + output.AppendUnsigned(result.count_, 4); + output.AppendUnsigned(0, 4); + output.AppendUnsigned(bytes, 8); auto append = [&](const Edit& edit) { - AppendU32(&output, edit.field.size()); - AppendU32(&output, edit.value.size()); - output.append(edit.field); - output.append(edit.value); + output.AppendUnsigned(edit.field.size(), 4); + output.AppendUnsigned(edit.value.size(), 4); + output.AppendBytes(edit.field); + output.AppendBytes(edit.value); }; std::size_t offset = kHashValueHeaderBytes; for (const auto& splice : splices) { - output.append(payload.substr(offset, splice.begin - offset)); - append(edits[splice.edit]); + output.AppendBytes(payload.substr(offset, splice.begin - offset)); + if (!erase) append(edits[splice.edit]); offset = splice.end; } - output.append(payload.substr(offset)); + output.AppendBytes(payload.substr(offset)); for (const auto& edit : edits) - if (!edit.found) append(edit); + if (!erase && !edit.found) append(edit); return result; } +absl::StatusOr> EditCompactHash( + std::string_view payload, std::size_t expected_count, + std::span fields, + std::span values, std::size_t max_bytes) { + auto plan = + PlanCompactHashEdit(payload, expected_count, fields, values, max_bytes); + if (!plan.ok()) return plan.status(); + if (!*plan) return std::nullopt; + auto& edit = **plan; + auto encoded = EncodeInlineRecordPayload(edit.payload_); + if (!encoded.ok()) return encoded.status(); + return CompactHashEdit{.changed_ = edit.changed_, + .added_ = edit.added_, + .count_ = edit.count_, + .payload_ = std::move(*encoded)}; +} + absl::StatusOr EncodeHashValue(const HashValue& value) { if (value.entries_.empty() || value.entries_.size() > std::numeric_limits::max()) { diff --git a/src/storage/engine/hash_tree.cpp b/src/storage/engine/hash_tree.cpp index a97390b37..3c6c0a80a 100644 --- a/src/storage/engine/hash_tree.cpp +++ b/src/storage/engine/hash_tree.cpp @@ -1297,7 +1297,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( } HashValue compact; std::optional compact_edit_scratch; - std::optional compact_edit; + std::optional compact_edit; + std::optional compact_source; std::optional grouped_plan; std::uint64_t edited_length = location.logical_size_; if (edit_leaves) { @@ -1388,11 +1389,22 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (read_only) co_return ReadEncodedHash(payload, location.logical_size_, operation, value_type); - if (value_type == ValueType::kHash && - operation.kind_ == HashOperationKind::kSet) { - // Only operand views and a bounded after-image survive preparation; the - // old fields remain borrowed from this read lease. Publication still - // uses the existing key intent, snapshot validation and append boundary. + // A prepared write may wait behind a state-lock owner that needs a read + // slot. Do not retain the pool's last slot across that wait. Remote pool + // availability cannot be inspected here, so those leases use owned edits. + const bool borrow_source = + loaded->lease_.buffer_id() == 0 || + (loaded->lease_.owner_worker() == store.worker_->id() && + store.buffers_.available_read_buffers() != 0); + if (borrow_source && ((value_type == ValueType::kHash && + operation.kind_ == HashOperationKind::kSet) || + (value_type == ValueType::kSet && + (operation.kind_ == HashOperationKind::kSet || + operation.kind_ == HashOperationKind::kDelete)))) { + // Only operand views and framing survive preparation; unchanged bytes + // remain borrowed from compact_source through the awaited append. + // Publication still uses the existing key intent, snapshot validation and + // append boundary. GroupedScratchBudget budget; if (operation.fields_.size() > SIZE_MAX / 256) co_return absl::ResourceExhaustedError( @@ -1402,20 +1414,31 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( added = budget.AddBytes(kCollectionPromotionBytes * 2); if (!added.ok()) co_return added; auto admitted = budget.Reserve(1); - if (!admitted.ok()) co_return admitted.status(); - compact_edit_scratch.emplace(std::move(*admitted)); - auto edited = - EditCompactHash(payload, location.logical_size_, operation.fields_, - operation.values_, kCollectionPromotionBytes - 1); - if (!edited.ok()) co_return edited.status(); - compact_edit = std::move(*edited); - if (compact_edit) { - result.changed_ = compact_edit->changed_; - result.integer_ = compact_edit->added_; - } else { - // Promotion keeps the existing owned builder and its admission. No - // oversized output was allocated by the bounded editor. - compact_edit_scratch.reset(); + // Preserve the bounded legacy workspace for shrinking at maxmemory. + // A denied splice plan falls back before it borrows or changes data. + if (!admitted.ok() && operation.kind_ != HashOperationKind::kDelete) + co_return admitted.status(); + if (admitted.ok()) { + compact_edit_scratch.emplace(std::move(*admitted)); + auto edited = PlanCompactHashEdit( + payload, location.logical_size_, operation.fields_, + operation.values_, kCollectionPromotionBytes - 1, + operation.kind_ == HashOperationKind::kDelete + ? CompactHashEditKind::kDelete + : CompactHashEditKind::kSet); + if (!edited.ok()) co_return edited.status(); + compact_edit = std::move(*edited); + if (compact_edit) { + result.changed_ = compact_edit->changed_; + result.integer_ = operation.kind_ == HashOperationKind::kDelete + ? compact_edit->removed_ + : compact_edit->added_; + compact_source.emplace(std::move(*loaded)); + } else { + // Promotion keeps the existing owned builder and its admission. No + // oversized output was allocated by the bounded editor. + compact_edit_scratch.reset(); + } } } if (!compact_edit) { @@ -1631,7 +1654,7 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( break; } case HashOperationKind::kDelete: - if (edit_leaves) break; + if (edit_leaves || compact_edit) break; for (std::string_view field : operation.fields_) { auto entry = find_entry(field); if (entry != compact.entries_.end()) { @@ -1712,18 +1735,16 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (!valid.ok()) co_return valid; } std::optional prepared_compact_payload; - if (compact_edit && compact_edit->changed_) - prepared_compact_payload.emplace(std::move(compact_edit->payload_)); bool compact_write_promotes = false; if (unlocked_compact_write || unlocked_create) { if (result.changed_ && result.key_exists_) { - compact_write_promotes = NeedsGroupedHash(compact); + compact_write_promotes = !compact_edit && NeedsGroupedHash(compact); if (unlocked_create && compact_write_promotes) { auto prepared = PrepareGroupedHashMutation(nullptr, std::move(compact), {}, result.length_, 1); if (!prepared.ok()) co_return prepared.status(); grouped_plan.emplace(std::move(*prepared)); - } else if (!compact_write_promotes && !prepared_compact_payload) { + } else if (!compact_write_promotes && !compact_edit) { auto encoded = EncodeHashValue(compact); if (!encoded.ok()) co_return encoded.status(); prepared_compact_payload.emplace(std::move(*encoded)); @@ -1810,6 +1831,8 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (!result.key_exists_) { kind = RecordKind::kTombstone; published_type = ValueType::kNone; + } else if (compact_edit) { + // The borrowed plan is consumed directly into the unpublished record. } else if (prepared_compact_payload.has_value()) { payload = std::move(*prepared_compact_payload); } else { @@ -1817,11 +1840,15 @@ Task> StorageEngine::Impl::ExecuteHashLikeLocked( if (!encoded.ok()) co_return encoded.status(); payload = std::move(*encoded); } + std::optional inline_payload; + if (compact_edit && result.key_exists_) + inline_payload.emplace(compact_edit->payload_); absl::Status written = co_await AppendLocked( store, partition, db_id, key, digest, payload, kind, published_type, kind == RecordKind::kValue ? expire_at_ms : 0, tx, kind == RecordKind::kValue ? result.length_ : 0, nullptr, nullptr, - replication, true, nullptr, mutation_precondition); + replication, true, nullptr, mutation_precondition, nullptr, + inline_payload ? &*inline_payload : nullptr); if (!written.ok()) co_return written; co_return result; } diff --git a/src/storage/engine/impl.h b/src/storage/engine/impl.h index 6d601d515..7c34d7034 100644 --- a/src/storage/engine/impl.h +++ b/src/storage/engine/impl.h @@ -4221,7 +4221,8 @@ class StorageEngine::Impl { ReplicationCommandAppend* replication = nullptr, bool capture_fullsync = true, TxUndoLog* replacement_undo = nullptr, const MutationPrecondition* mutation_precondition = nullptr, - GroupMutationWrite* grouped = nullptr + GroupMutationWrite* grouped = nullptr, + const InlineRecordPayload* inline_payload = nullptr #if LAVIK_ENABLE_TRACE , SetLatencyTrace* trace = nullptr @@ -4478,7 +4479,7 @@ class StorageEngine::Impl { struct RecordWriteRequest { std::string_view key_{}; std::string_view value_{}; - // Optional preflighted inline auxiliary encoder, mutually exclusive with + // Optional preflighted compact/auxiliary encoder, mutually exclusive with // value_ and external_. Borrowed through allocation waits, then consumed // synchronously into the unpublished staging tail after final validation. const InlineRecordPayload* inline_payload_ = nullptr; diff --git a/src/storage/engine/list_tree.cpp b/src/storage/engine/list_tree.cpp index 182f0c608..a1fa350e6 100644 --- a/src/storage/engine/list_tree.cpp +++ b/src/storage/engine/list_tree.cpp @@ -102,6 +102,88 @@ absl::StatusOr EncodeList(std::span elements) { return output; } +struct CompactListEdit { + std::optional admission_; + CompactPayload payload_; +}; + +// Only end edits need this borrowed plan. General List operations and +// promotion keep their owned builder. Pop owns just the returned elements. +absl::StatusOr> PlanCompactListEdit( + std::string_view payload, std::uint64_t count, + const ListOperation& operation, ListResult& result) { + using Kind = ListOperationKind; + const bool push = operation.kind_ == Kind::kPushLeft || + operation.kind_ == Kind::kPushRight || + operation.kind_ == Kind::kPushLeftIfExists || + operation.kind_ == Kind::kPushRightIfExists; + const bool pop = + operation.kind_ == Kind::kPopLeft || operation.kind_ == Kind::kPopRight; + if ((!push && !pop) || payload.size() >= kCollectionPromotionBytes) + return std::nullopt; + const bool left = operation.kind_ == Kind::kPushLeft || + operation.kind_ == Kind::kPushLeftIfExists || + operation.kind_ == Kind::kPopLeft; + auto reader = + OrderedCompactReader::Open(OrderedCollectionKind::kList, payload, count); + if (!reader.ok()) return reader.status(); + std::size_t size = payload.size(); + if (push) { + if (operation.values_.size() > UINT32_MAX - count) + return absl::OutOfRangeError("invalid List element count"); + for (const auto value : operation.values_) { + auto next = + AppendOrderedEntrySize(OrderedCollectionKind::kList, size, + value.size(), std::string{}.max_size()); + if (!next.ok()) return next.status(); + size = *next; + } + if (size >= kCollectionPromotionBytes) return std::nullopt; + } + const auto removed = pop ? std::min(operation.count_, count) : 0; + const auto length = push ? count + operation.values_.size() : count - removed; + CompactListEdit edit; + if (push && operation.values_.size() > 1) { + edit.admission_ = TryReserveMemory(operation.values_.size() * 256); + if (!edit.admission_) return std::nullopt; + } + auto& output = edit.payload_; + if (length != 0) { + output.AppendBytes(kListMagic); + output.AppendUnsigned(length, 4); + } + auto append_inputs = [&] { + for (std::size_t i = 0; i < operation.values_.size(); ++i) { + const auto value = + operation.values_[left ? operation.values_.size() - 1 - i : i]; + output.AppendUnsigned(value.size(), 4); + output.AppendBytes(value); + } + }; + if (push && left) append_inputs(); + std::size_t kept_begin = 0; + std::size_t kept_bytes = 0; + for (std::size_t i = 0; i < count; ++i) { + auto entry = reader->Next(); + if (!entry.ok()) return entry.status(); + if (pop && (left ? i < removed : i >= count - removed)) { + result.values_.emplace_back(entry->value_); + } else { + if (kept_bytes == 0) kept_begin = entry->encoded_.data() - payload.data(); + kept_bytes += entry->encoded_.size(); + } + } + // End edits retain one contiguous interval, regardless of cardinality. + output.AppendBytes(payload.substr(kept_begin, kept_bytes)); + if (push && !left) append_inputs(); + if (pop && !left) std::reverse(result.values_.begin(), result.values_.end()); + result.changed_ = push ? !operation.values_.empty() : removed != 0; + result.length_ = length; + // key_exists_ describes the input for POP's nil-vs-array reply, including + // a counted pop that consumes the last element and publishes a tombstone. + return edit; +} + std::optional NormalizeIndex(std::int64_t index, std::uint64_t size) { if (index < 0) { @@ -545,6 +627,8 @@ Task> StorageEngine::Impl::ExecuteListLocked( create_admission.emplace(std::move(*admitted)); } std::vector elements; + std::optional compact_edit; + std::optional compact_source; if (exists) { auto loaded = co_await LoadValue(store, partition, db_id, key, digest, location, extents); @@ -558,149 +642,173 @@ Task> StorageEngine::Impl::ExecuteListLocked( if (read_only && operation.kind_ == ListOperationKind::kRange && operation.packed_output_) co_return ReadPackedListRange(payload, location.logical_size_, operation); - auto decoded = DecodeList(payload, location.logical_size_); - if (!decoded.ok()) co_return decoded.status(); - elements = std::move(*decoded); + // Leave a local read slot for operations ahead of this writer in the + // state-lock queue. Remote leases and pool pressure use the owned path. + const bool borrow_source = + loaded->lease_.buffer_id() == 0 || + (loaded->lease_.owner_worker() == store.worker_->id() && + store.buffers_.available_read_buffers() != 0); + if (!read_only && borrow_source) { + auto planned = PlanCompactListEdit(payload, location.logical_size_, + operation, result); + if (!planned.ok()) co_return planned.status(); + compact_edit = std::move(*planned); + } + if (compact_edit) { + compact_source.emplace(std::move(*loaded)); + } else { + auto decoded = DecodeList(payload, location.logical_size_); + if (!decoded.ok()) co_return decoded.status(); + elements = std::move(*decoded); + } } if (read_only) co_return ReadListSnapshot(elements, std::move(result), operation); - switch (operation.kind_) { - case ListOperationKind::kPushLeft: - case ListOperationKind::kPushRight: - case ListOperationKind::kPushLeftIfExists: - case ListOperationKind::kPushRightIfExists: { - const bool only_if_exists = - operation.kind_ == ListOperationKind::kPushLeftIfExists || - operation.kind_ == ListOperationKind::kPushRightIfExists; - if (!exists && only_if_exists) co_return result; - const bool left = operation.kind_ == ListOperationKind::kPushLeft || - operation.kind_ == ListOperationKind::kPushLeftIfExists; - for (std::string_view value : operation.values_) { - if (unlocked_create && !elements.empty() && elements.size() % 256 == 0) - co_await bycorf::Yield(*store.worker_); - if (value.size() > kMaxStringBytes) { - co_return absl::OutOfRangeError( - "List element exceeds Redis-compatible 512 MiB limit"); + if (!compact_edit) { + switch (operation.kind_) { + case ListOperationKind::kPushLeft: + case ListOperationKind::kPushRight: + case ListOperationKind::kPushLeftIfExists: + case ListOperationKind::kPushRightIfExists: { + const bool only_if_exists = + operation.kind_ == ListOperationKind::kPushLeftIfExists || + operation.kind_ == ListOperationKind::kPushRightIfExists; + if (!exists && only_if_exists) co_return result; + const bool left = + operation.kind_ == ListOperationKind::kPushLeft || + operation.kind_ == ListOperationKind::kPushLeftIfExists; + for (std::string_view value : operation.values_) { + if (unlocked_create && !elements.empty() && + elements.size() % 256 == 0) + co_await bycorf::Yield(*store.worker_); + if (value.size() > kMaxStringBytes) { + co_return absl::OutOfRangeError( + "List element exceeds Redis-compatible 512 MiB limit"); + } + if (left) + elements.insert(elements.begin(), std::string(value)); + else + elements.emplace_back(value); } - if (left) - elements.insert(elements.begin(), std::string(value)); - else - elements.emplace_back(value); + result.changed_ = !operation.values_.empty(); + result.key_exists_ = true; + break; } - result.changed_ = !operation.values_.empty(); - result.key_exists_ = true; - break; - } - case ListOperationKind::kPopLeft: - case ListOperationKind::kPopRight: { - const std::size_t count = static_cast( - std::min(operation.count_, elements.size())); - const bool left = operation.kind_ == ListOperationKind::kPopLeft; - for (std::size_t i = 0; i < count; ++i) { - if (left) { - result.values_.push_back(std::move(elements.front())); - elements.erase(elements.begin()); - } else { - result.values_.push_back(std::move(elements.back())); - elements.pop_back(); + case ListOperationKind::kPopLeft: + case ListOperationKind::kPopRight: { + const std::size_t count = static_cast( + std::min(operation.count_, elements.size())); + const bool left = operation.kind_ == ListOperationKind::kPopLeft; + for (std::size_t i = 0; i < count; ++i) { + if (left) { + result.values_.push_back(std::move(elements.front())); + elements.erase(elements.begin()); + } else { + result.values_.push_back(std::move(elements.back())); + elements.pop_back(); + } } + result.changed_ = count != 0; + break; } - result.changed_ = count != 0; - break; - } - case ListOperationKind::kSet: { - if (!exists) co_return absl::NotFoundError("no such key"); - if (operation.value_.size() > kMaxStringBytes) - co_return absl::OutOfRangeError("List element exceeds storage limits"); - auto position = NormalizeIndex(operation.first_, elements.size()); - if (!position) co_return absl::OutOfRangeError("index out of range"); - elements[*position] = operation.value_; - result.changed_ = true; - break; - } - case ListOperationKind::kInsertBefore: - case ListOperationKind::kInsertAfter: { - if (!exists) co_return result; - if (operation.value_.size() > kMaxStringBytes) - co_return absl::OutOfRangeError("List element exceeds storage limits"); - auto it = std::find(elements.begin(), elements.end(), operation.pivot_); - if (it == elements.end()) { - result.integer_ = -1; - // A successful no-op still has to validate the captured population - // after an unlocked read, just like a replacement or final-element - // pop. + case ListOperationKind::kSet: { + if (!exists) co_return absl::NotFoundError("no such key"); + if (operation.value_.size() > kMaxStringBytes) + co_return absl::OutOfRangeError( + "List element exceeds storage limits"); + auto position = NormalizeIndex(operation.first_, elements.size()); + if (!position) co_return absl::OutOfRangeError("index out of range"); + elements[*position] = operation.value_; + result.changed_ = true; break; } - if (operation.kind_ == ListOperationKind::kInsertAfter) ++it; - elements.insert(it, std::string(operation.value_)); - result.changed_ = true; - result.integer_ = elements.size(); - break; - } - case ListOperationKind::kRemove: { - const ListRemovePlan plan = NormalizeListRemoveLimit(operation.first_); - std::uint64_t removed = 0; - if (plan.forward_) { - for (auto it = elements.begin(); - it != elements.end() && removed < plan.limit_;) { - if (*it == operation.value_) { - it = elements.erase(it); - ++removed; - } else { - ++it; - } + case ListOperationKind::kInsertBefore: + case ListOperationKind::kInsertAfter: { + if (!exists) co_return result; + if (operation.value_.size() > kMaxStringBytes) + co_return absl::OutOfRangeError( + "List element exceeds storage limits"); + auto it = std::find(elements.begin(), elements.end(), operation.pivot_); + if (it == elements.end()) { + result.integer_ = -1; + // A successful no-op still has to validate the captured population + // after an unlocked read, just like a replacement or final-element + // pop. + break; } - } else { - for (std::size_t i = elements.size(); i > 0 && removed < plan.limit_;) { - --i; - if (elements[i] == operation.value_) { - elements.erase(elements.begin() + i); - ++removed; + if (operation.kind_ == ListOperationKind::kInsertAfter) ++it; + elements.insert(it, std::string(operation.value_)); + result.changed_ = true; + result.integer_ = elements.size(); + break; + } + case ListOperationKind::kRemove: { + const ListRemovePlan plan = NormalizeListRemoveLimit(operation.first_); + std::uint64_t removed = 0; + if (plan.forward_) { + for (auto it = elements.begin(); + it != elements.end() && removed < plan.limit_;) { + if (*it == operation.value_) { + it = elements.erase(it); + ++removed; + } else { + ++it; + } + } + } else { + for (std::size_t i = elements.size(); + i > 0 && removed < plan.limit_;) { + --i; + if (elements[i] == operation.value_) { + elements.erase(elements.begin() + i); + ++removed; + } } } + result.integer_ = removed; + result.changed_ = removed != 0; + break; } - result.integer_ = removed; - result.changed_ = removed != 0; - break; - } - case ListOperationKind::kTrim: { - const auto [begin, end] = - NormalizeRange(operation.first_, operation.second_, elements.size()); - std::vector kept; - kept.reserve(end - begin); - for (std::uint64_t i = begin; i < end; ++i) - kept.push_back(std::move(elements[i])); - result.changed_ = kept.size() != elements.size(); - elements = std::move(kept); - break; - } - case ListOperationKind::kMoveWithin: { - if (elements.empty()) co_return result; - const bool source_left = operation.first_ != 0; - const bool destination_left = operation.second_ != 0; - std::string moved = source_left ? std::move(elements.front()) - : std::move(elements.back()); - if (source_left) - elements.erase(elements.begin()); - else - elements.pop_back(); - if (destination_left) - elements.insert(elements.begin(), moved); - else - elements.push_back(moved); - result.values_.push_back(std::move(moved)); - result.changed_ = source_left != destination_left && elements.size() > 1; - break; + case ListOperationKind::kTrim: { + const auto [begin, end] = NormalizeRange( + operation.first_, operation.second_, elements.size()); + std::vector kept; + kept.reserve(end - begin); + for (std::uint64_t i = begin; i < end; ++i) + kept.push_back(std::move(elements[i])); + result.changed_ = kept.size() != elements.size(); + elements = std::move(kept); + break; + } + case ListOperationKind::kMoveWithin: { + if (elements.empty()) co_return result; + const bool source_left = operation.first_ != 0; + const bool destination_left = operation.second_ != 0; + std::string moved = source_left ? std::move(elements.front()) + : std::move(elements.back()); + if (source_left) + elements.erase(elements.begin()); + else + elements.pop_back(); + if (destination_left) + elements.insert(elements.begin(), moved); + else + elements.push_back(moved); + result.values_.push_back(std::move(moved)); + result.changed_ = + source_left != destination_left && elements.size() > 1; + break; + } + case ListOperationKind::kLength: + co_return result; + default: + co_return absl::InternalError("read operation reached List mutation"); } - case ListOperationKind::kLength: - co_return result; - default: - co_return absl::InternalError("read operation reached List mutation"); - } - result.length_ = elements.size(); + result.length_ = elements.size(); + } std::optional prepared_compact_payload; std::optional prepared_groups; bool compact_write_promotes = false; @@ -765,9 +873,11 @@ Task> StorageEngine::Impl::ExecuteListLocked( RecordKind kind = RecordKind::kValue; ValueType type = ValueType::kList; std::string payload; - if (elements.empty()) { + if (result.length_ == 0) { kind = RecordKind::kTombstone; type = ValueType::kNone; + } else if (compact_edit) { + // Encoding waits until the append tail is exclusively owned. } else if (prepared_compact_payload.has_value()) { payload = std::move(*prepared_compact_payload); } else { @@ -775,11 +885,15 @@ Task> StorageEngine::Impl::ExecuteListLocked( if (!encoded.ok()) co_return encoded.status(); payload = std::move(*encoded); } + std::optional inline_payload; + if (compact_edit && result.length_ != 0) + inline_payload.emplace(compact_edit->payload_); absl::Status written = co_await AppendLocked( store, partition, db_id, key, digest, payload, kind, type, kind == RecordKind::kValue ? expire_at_ms : 0, tx, - kind == RecordKind::kValue ? elements.size() : 0, nullptr, nullptr, - replication, true, nullptr, mutation_precondition); + kind == RecordKind::kValue ? result.length_ : 0, nullptr, nullptr, + replication, true, nullptr, mutation_precondition, nullptr, + inline_payload ? &*inline_payload : nullptr); if (!written.ok()) co_return written; co_return result; } diff --git a/src/storage/engine/ordered_compact_codec.cpp b/src/storage/engine/ordered_compact_codec.cpp index e444f38c3..3566c6aff 100644 --- a/src/storage/engine/ordered_compact_codec.cpp +++ b/src/storage/engine/ordered_compact_codec.cpp @@ -47,6 +47,41 @@ std::uint64_t Get(std::string_view input, std::size_t offset, unsigned width) { } // namespace +absl::StatusOr OrderedCompactReader::Open( + OrderedCollectionKind kind, std::string_view encoded, + std::uint64_t expected_count) { + const bool sorted = kind == OrderedCollectionKind::kSortedSet; + const std::size_t framing = sorted ? 12 : 4; + if (!ValidKind(kind) || expected_count == 0 || expected_count > UINT32_MAX || + encoded.size() < 8 || !encoded.starts_with(sorted ? "LZS1" : "LVL1") || + Get(encoded, 4, 4) != expected_count || + expected_count > (encoded.size() - 8) / framing) + return absl::DataLossError("invalid ordered full-image framing"); + return OrderedCompactReader(encoded, expected_count, sorted); +} + +absl::StatusOr OrderedCompactReader::Next() { + const std::size_t framing = sorted_ ? 12 : 4; + if (remaining_ == 0 || framing > bytes_.size() - offset_) + return absl::DataLossError("truncated ordered full-image entry"); + const auto score = + sorted_ ? std::bit_cast(Get(bytes_, offset_, 8)) : 0; + const auto size = Get(bytes_, offset_ + (sorted_ ? 8 : 0), 4); + if (std::isnan(score) || size > kMaxStringBytes || + size > bytes_.size() - offset_ - framing) + return absl::DataLossError("invalid ordered full-image entry"); + const auto end = offset_ + framing + size; + if (remaining_ == 1 && end != bytes_.size()) + return absl::DataLossError("trailing ordered full-image bytes"); + OrderedCompactEntryView result{ + .value_ = bytes_.substr(offset_ + framing, size), + .score_ = score, + .encoded_ = bytes_.substr(offset_, framing + size)}; + offset_ = end; + --remaining_; + return result; +} + absl::StatusOr AppendOrderedEntrySize(OrderedCollectionKind kind, std::size_t encoded_bytes, std::size_t item_bytes, diff --git a/src/storage/engine/write.cpp b/src/storage/engine/write.cpp index 7a5ba7b6f..9d8b2f733 100644 --- a/src/storage/engine/write.cpp +++ b/src/storage/engine/write.cpp @@ -240,7 +240,7 @@ Task> StorageEngine::Impl::SetWithLockState( true, nullptr, mutation_precondition #if LAVIK_ENABLE_TRACE , - /*grouped=*/nullptr, trace + /*grouped=*/nullptr, /*inline_payload=*/nullptr, trace #endif ); if (trace != nullptr) trace->append_done_ns_ = SetTraceNowNanos(); @@ -1977,7 +1977,7 @@ Task StorageEngine::Impl::AppendLocked( std::uint64_t* committed_sequence, ReplicationCommandAppend* replication, bool capture_fullsync, TxUndoLog* replacement_undo, const MutationPrecondition* mutation_precondition, - GroupMutationWrite* grouped + GroupMutationWrite* grouped, const InlineRecordPayload* inline_payload #if LAVIK_ENABLE_TRACE , SetLatencyTrace* trace @@ -1987,8 +1987,16 @@ Task StorageEngine::Impl::AppendLocked( co_return absl::FailedPreconditionError( "storage writer is stopped after an IO failure"); } + const auto value_bytes = + inline_payload ? inline_payload->encoded_bytes() : value.size(); + // Compact editors borrow read leases through allocation waits. Oversized + // results must use the existing promotion/extent path before reaching here. + if (inline_payload && + (!value.empty() || grouped || replacement_undo || + kind != RecordKind::kValue || value_bytes >= kCollectionPromotionBytes)) + co_return absl::InvalidArgumentError("invalid compact encoder write"); if (logical_size == std::numeric_limits::max()) { - logical_size = value.size(); + logical_size = value_bytes; } if (!ValidRecordKeySize(key.size())) { co_return absl::OutOfRangeError("record key exceeds storage limit"); @@ -2081,7 +2089,7 @@ Task StorageEngine::Impl::AppendLocked( absl::Status status = absl::OkStatus(); const bool key_indirect = key.size() > kInlineKeyMaxBytes; const std::uint64_t logical_payload_bytes = - static_cast(value.size()); + static_cast(value_bytes); const std::size_t inline_bytes = AlignRecord(RecordHeaderBytes(key.size(), key_indirect, tx != nullptr, expire_at_ms != 0) + @@ -2131,6 +2139,7 @@ Task StorageEngine::Impl::AppendLocked( const RecordWriteRequest record_write{ .key_ = key, .value_ = value, + .inline_payload_ = inline_payload, .digest_ = digest, .expire_at_ms_ = expire_at_ms, .mutation_sequence_ = mutation_sequence, @@ -2677,8 +2686,12 @@ Task StorageEngine::Impl::WriteRecordLocked( const bool grouped_root = request.group_ != nullptr && !request.group_->auxiliary_; if (request.inline_payload_ != nullptr && - (!auxiliary || request.external_ || !request.value_.empty())) - co_return absl::InvalidArgumentError("invalid inline group encoder write"); + (request.external_ || !request.value_.empty() || + (!auxiliary && + (request.group_ != nullptr || request.kind_ != RecordKind::kValue || + payload_bytes >= kCollectionPromotionBytes)))) + co_return absl::InvalidArgumentError( + "invalid inline payload encoder write"); struct FailIncompleteGroupedRoot { WorkerStore& store_; TxShardWrites* tx_; @@ -2740,7 +2753,7 @@ Task StorageEngine::Impl::WriteRecordLocked( } } else if (request.kind_ == RecordKind::kValue && request.value_type_ == ValueType::kString && !request.group_ && - request.value_.size() != logical_size) { + payload_bytes != logical_size) { co_return absl::Status(absl::StatusCode::kInvalidArgument, "inline string length mismatch"); } @@ -3418,13 +3431,17 @@ Task StorageEngine::Impl::WriteRecordLocked( // tail, and neither an index identity nor an indirect-key reference exists. auto encoded = request.inline_payload_->WriteTo(std::span( staging.data_ + record_offset + record_header_bytes, payload_bytes)); - LAVIK_FAULT_INJECT(if (LAVIK_FAULT_MATCHES( - "LAVIK_FAIL_INLINE_GROUP_ENCODE_KEY", - request.key_)) { - static std::atomic injected{false}; - if (!injected.exchange(true, std::memory_order_relaxed)) - encoded = absl::DataLossError("injected inline group encoding fault"); - }); + LAVIK_FAULT_INJECT( + if (LAVIK_FAULT_MATCHES(auxiliary + ? "LAVIK_FAIL_INLINE_GROUP_ENCODE_KEY" + : "LAVIK_FAIL_INLINE_COMPACT_ENCODE_KEY", + request.key_)) { + static std::atomic injected{false}; + if (!injected.exchange(true, std::memory_order_relaxed)) + encoded = absl::DataLossError( + auxiliary ? "injected inline group encoding fault" + : "injected inline compact encoding fault"); + }); if (!encoded.ok()) co_return encoded; } // Admit exceptional reference-directory allocations before publishing any diff --git a/tests/compact_collection_write_e2e_test.cpp b/tests/compact_collection_write_e2e_test.cpp index 6cda47ec8..9a8b7416c 100644 --- a/tests/compact_collection_write_e2e_test.cpp +++ b/tests/compact_collection_write_e2e_test.cpp @@ -14,6 +14,8 @@ * limitations under the License. */ +#include +#include #include #include @@ -236,6 +238,259 @@ void ExpectWatchOutcome(Client& watcher, Collection type, EXPECT_EQ(result.items_.size(), 1); } +TEST(CompactStringWriteE2e, ColdRangeWriteYieldsStateAndRetainsKey) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires compact write pause hook"; +#endif + PrivateDisk disk(kPrivateDiskBytes); + std::string expected(8192, 'a'); + { + Server server(disk, 1); + Client client(server.port()); + ASSERT_EQ(client.Command({"SET", "target", expected, "EX", "600"}).text_, + "OK"); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + { + ScopedFault pause_key("LAVIK_COMPACT_WRITE_PAUSE_KEY", "target"); + WritePauseGate gate(disk, "LAVIK_COMPACT_WRITE_HOLD_FILE"); + Server server(disk, 1); + Client client(server.port()); + PendingWrite first(server.port()), second(server.port()); + gate.Arm(); + first.Send({"SETRANGE", "target", "2048", std::string(256, 'b')}); + ASSERT_TRUE(AwaitMarker(server, gate.marker())) << server.Log(); + second.Send({"SETRANGE", "target", "4096", std::string(256, 'c')}); + EXPECT_EQ(client.Command({"SET", "unrelated", "progress"}).text_, "OK"); + EXPECT_EQ(client.Command({"HSET", "hash", "f", "progress"}).text_, "1"); + EXPECT_FALSE(first.HasReply()); + EXPECT_FALSE(second.HasReply()); + gate.Release(); + EXPECT_EQ(first.Read().text_, "8192"); + EXPECT_EQ(second.Read().text_, "8192"); + expected.replace(2048, 256, 256, 'b'); + expected.replace(4096, 256, 256, 'c'); + EXPECT_EQ(client.Command({"GET", "target"}).text_, expected); + EXPECT_GT(std::stoll(client.Command({"PTTL", "target"}).text_), 0); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + Server recovered(disk, 2); + Client client(recovered.port()); + EXPECT_EQ(client.Command({"GET", "target"}).text_, expected); + EXPECT_GT(std::stoll(client.Command({"PTTL", "target"}).text_), 0); + ASSERT_EQ(recovered.Wait(true), 0) << recovered.Log(); +} + +TEST(CompactPointWriteE2e, MixedEightKiBEditsMatchModelsAndRecover) { + PrivateDisk disk; + std::deque list; + std::set set; + std::map zset; + std::string value(8192, 's'); + auto member = [](unsigned i) { + auto text = std::to_string(i); + text.resize(32, static_cast(i)); + return text; + }; + auto verify = [&](Client& client) { + const auto actual_list = client.Command({"LRANGE", "list", "0", "-1"}); + ASSERT_EQ(actual_list.items_.size(), list.size()); + for (std::size_t i = 0; i < list.size(); ++i) + EXPECT_EQ(actual_list.items_[i].text_, list[i]); + const auto actual_set = client.Command({"SMEMBERS", "set"}); + std::set members; + for (const auto& item : actual_set.items_) members.insert(item.text_); + EXPECT_EQ(members, set); + std::vector> ordered; + for (const auto& [name, score] : zset) ordered.emplace_back(score, name); + std::sort(ordered.begin(), ordered.end()); + const auto actual_zset = + client.Command({"ZRANGE", "zset", "0", "-1", "WITHSCORES"}); + ASSERT_EQ(actual_zset.items_.size(), ordered.size() * 2); + for (std::size_t i = 0; i < ordered.size(); ++i) { + EXPECT_EQ(actual_zset.items_[2 * i].text_, ordered[i].second); + EXPECT_EQ(std::stod(actual_zset.items_[2 * i + 1].text_), + ordered[i].first); + } + EXPECT_EQ(client.Command({"GET", "string"}).text_, value); + }; + { + Server server(disk, 2); + Client client(server.port()); + std::vector push{"RPUSH", "list"}; + for (unsigned i = 0; i < 32; ++i) { + list.emplace_back(256, static_cast(i)); + push.push_back(list.back()); + } + ASSERT_EQ(client.Command(push).text_, "32"); + std::vector sadd{"SADD", "set"}, zadd{"ZADD", "zset"}; + for (unsigned i = 0; i < 256; ++i) { + const auto name = member(i); + set.insert(name); + zset[name] = i % 8; + sadd.push_back(name); + zadd.push_back(std::to_string(i % 8)); + zadd.push_back(name); + } + ASSERT_EQ(client.Command(sadd).text_, "256"); + ASSERT_EQ(client.Command(zadd).text_, "256"); + ASSERT_EQ(client.Command({"SET", "string", value}).text_, "OK"); + for (unsigned i = 0; i < 128; ++i) { + const std::string a(256, static_cast(i)), + b(256, static_cast(i + 1)); + const bool left = i % 2 == 0; + ASSERT_EQ(client.Command({left ? "LPUSH" : "RPUSH", "list", a, b}).text_, + "34"); + if (left) { + list.push_front(a); + list.push_front(b); + } else { + list.push_back(a); + list.push_back(b); + } + const auto popped = client.Command({left ? "RPOP" : "LPOP", "list", "2"}); + ASSERT_EQ(popped.items_.size(), 2); + for (unsigned j = 0; j < 2; ++j) { + EXPECT_EQ(popped.items_[j].text_, left ? list.back() : list.front()); + if (left) + list.pop_back(); + else + list.pop_front(); + } + const auto name = member((i * 37) % 256); + const int delta = i % 2 == 0 ? 32768 : -32768; + const auto incremented = + client.Command({"ZINCRBY", "zset", std::to_string(delta), name}); + ASSERT_EQ(incremented.kind_, '$'); + zset[name] += delta; + EXPECT_EQ(std::stod(incremented.text_), zset[name]); + if (i % 7 == 0) { + ASSERT_EQ(client.Command({"ZREM", "zset", name}).text_, "1"); + ASSERT_EQ(client.Command({"ZADD", "zset", "3", name}).text_, "1"); + zset[name] = 3; + } + ASSERT_EQ(client.Command({"SREM", "set", name, name}).text_, "1"); + ASSERT_EQ(client.Command({"SADD", "set", name, name}).text_, "1"); + const auto offset = (i * 317) % 8192; + value.resize(std::max(value.size(), offset + a.size()), '\0'); + value.replace(offset, a.size(), a); + EXPECT_EQ( + client.Command({"SETRANGE", "string", std::to_string(offset), a}) + .text_, + std::to_string(value.size())); + } + verify(client); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + Server recovered(disk, 3); + Client client(recovered.port()); + verify(client); + ASSERT_EQ(recovered.Wait(true), 0) << recovered.Log(); +} + +TEST(CompactPointWriteE2e, FailedInlineEncodingKeepsValueAndWatch) { +#if !LAVIK_TEST_FAULTS_AVAILABLE + GTEST_SKIP() << "requires inline encoder fault hook"; +#endif + const std::vector< + std::pair, std::vector>> + cases{{{"SET", "target", std::string(8192, 'a')}, + {"SETRANGE", "target", "3", "changed"}}, + {{"HSET", "target", "a", "before", "b", "keep"}, + {"HSET", "target", "a", "after"}}, + {{"SADD", "target", "a", "b"}, {"SREM", "target", "a"}}, + {{"RPUSH", "target", "a", "b"}, {"LPUSH", "target", "after"}}, + {{"ZADD", "target", "1", "a", "2", "b"}, + {"ZINCRBY", "target", "9", "a"}}}; + for (const auto& [seed, mutation] : cases) { + for (const bool in_exec : {false, true}) { + SCOPED_TRACE(seed.front()); + SCOPED_TRACE(in_exec); + PrivateDisk disk(kPrivateDiskBytes); + std::string after; + { + ScopedFault fault("LAVIK_FAIL_INLINE_COMPACT_ENCODE_KEY", "target"); + Server server(disk, 1); + Client client(server.port()), watcher(server.port()); + ASSERT_NE(client.Command(seed).kind_, '-'); + const auto before = client.Command({"DUMP", "target"}); + ASSERT_EQ(before.kind_, '$'); + ASSERT_EQ(watcher.Command({"WATCH", "target"}).text_, "OK"); + Reply failed; + if (in_exec) { + ASSERT_EQ(client.Command({"MULTI"}).text_, "OK"); + ASSERT_EQ(client.Command(mutation).text_, "QUEUED"); + ASSERT_EQ( + client.Command({"SET", "exec-successor", "committed"}).text_, + "QUEUED"); + auto executed = client.Command({"EXEC"}); + ASSERT_EQ(executed.items_.size(), 2); + failed = std::move(executed.items_[0]); + EXPECT_EQ(executed.items_[1].text_, "OK"); + } else { + failed = client.Command(mutation); + } + ASSERT_EQ(failed.kind_, '-') << failed.text_; + EXPECT_EQ(client.Command({"DUMP", "target"}).text_, before.text_); + ASSERT_EQ(watcher.Command({"MULTI"}).text_, "OK"); + ASSERT_EQ(watcher.Command({"EXISTS", "target"}).text_, "QUEUED"); + ASSERT_EQ(watcher.Command({"EXEC"}).items_.size(), 1); + ASSERT_NE(client.Command(mutation).kind_, '-'); + after = client.Command({"DUMP", "target"}).text_; + EXPECT_NE(after, before.text_); + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); + } + Server recovered(disk, 2); + Client client(recovered.port()); + EXPECT_EQ(client.Command({"DUMP", "target"}).text_, after); + if (in_exec) + EXPECT_EQ(client.Command({"GET", "exec-successor"}).text_, "committed"); + ASSERT_EQ(recovered.Wait(true), 0) << recovered.Log(); + } + } +} + +TEST(CompactPointWriteE2e, SingleReadSlotUsesOwnedFallback) { + PrivateDisk disk(kPrivateDiskBytes); + const std::array options{ + "--registered-buffer-mb-per-worker", "33", "--storage-read-buffer-kb", + "1024"}; + // Four default write slots leave exactly one read slot. Indirect-key + // publication needs to read its key while the old value is being replaced. + Server server(disk, 1, {}, {}, false, 2, "1G", {}, {}, 0, 20, options); + Client client(server.port()); + for (const auto& key : {std::string("target"), std::string(3072, 'k')}) { + ASSERT_EQ(client.Command({"SET", key, std::string(8192, 'a')}).text_, "OK"); + EXPECT_EQ(client.Command({"SETRANGE", key, "0", "b"}).text_, "8192"); + EXPECT_EQ(client.Command({"GETRANGE", key, "0", "1"}).text_, "ba"); + ASSERT_EQ(client.Command({"DEL", key}).text_, "1"); + ASSERT_EQ(client.Command({"HSET", key, "a", "old", "b", "keep"}).text_, + "2"); + EXPECT_EQ(client.Command({"HSET", key, "a", "new"}).text_, "0"); + EXPECT_EQ(client.Command({"HGET", key, "a"}).text_, "new"); + ASSERT_EQ(client.Command({"DEL", key}).text_, "1"); + ASSERT_EQ(client.Command({"SADD", key, "a", "b"}).text_, "2"); + EXPECT_EQ(client.Command({"SREM", key, "a"}).text_, "1"); + EXPECT_EQ(client.Command({"SADD", key, "c"}).text_, "1"); + EXPECT_EQ(client.Command({"SCARD", key}).text_, "2"); + ASSERT_EQ(client.Command({"DEL", key}).text_, "1"); + ASSERT_EQ(client.Command({"RPUSH", key, "a", "b"}).text_, "2"); + EXPECT_EQ(client.Command({"LPUSH", key, "c"}).text_, "3"); + EXPECT_EQ(client.Command({"RPOP", key}).text_, "b"); + ASSERT_EQ(client.Command({"DEL", key}).text_, "1"); + ASSERT_EQ(client.Command({"ZADD", key, "1", "a", "2", "b"}).text_, "2"); + EXPECT_EQ(client.Command({"ZINCRBY", key, "3", "a"}).text_, "4"); + EXPECT_EQ(client.Command({"ZREM", key, "b"}).text_, "1"); + EXPECT_EQ(client.Command({"ZSCORE", key, "a"}).text_, "4"); + } + client.Durable(); + ASSERT_EQ(server.Wait(true), 0) << server.Log(); +} + class CompactCollectionWriteE2e : public testing::TestWithParam {}; TEST_P(CompactCollectionWriteE2e, ColdPauseReleasesStateButSerializesSameKey) { @@ -367,7 +622,10 @@ TEST_P(CompactCollectionWriteE2e, ConditionsErrorsTtlWatchAndDeleteRecover) { EXPECT_EQ(client.Command({"LPOP", key}).text_, "left"); EXPECT_EQ(client.Command({"RPOP", key}).text_, "right"); EXPECT_EQ(client.Command({"LSET", key, "-1", "B"}).text_, "OK"); - EXPECT_EQ(client.Command({"LPOP", "last"}).text_, "a"); + const auto last = client.Command({"LPOP", "last", "99"}); + ASSERT_EQ(last.kind_, '*'); + ASSERT_EQ(last.items_.size(), 1); + EXPECT_EQ(last.items_[0].text_, "a"); EXPECT_EQ(client.Command({"LSET", "last", "0", "missing"}).kind_, '-'); } else { EXPECT_EQ(client.Command({"ZADD", key, "NX", "9", "a"}).text_, "0"); diff --git a/tests/grouped/hash_test.cpp b/tests/grouped/hash_test.cpp index 907a00189..9c4efe31c 100644 --- a/tests/grouped/hash_test.cpp +++ b/tests/grouped/hash_test.cpp @@ -282,6 +282,77 @@ TEST(CompactHashEdit, ValidatesPastMatchesAndChecksCounts) { EXPECT_FALSE(EditCompactHash(broken, 2, fields, values, 16383).ok()); } +TEST(CompactHashEdit, DeletePlanMatchesSequentialSetRemoval) { + std::mt19937 random(91); + for (unsigned trial = 0; trial < 200; ++trial) { + auto original = Value(1 + random() % 256, 0); + original.entries_.push_back({.field_ = "", .value_ = ""}); + original.entries_.push_back( + {.field_ = std::string("a\0b", 3), .value_ = ""}); + auto bytes = EncodeHashValue(original); + ASSERT_TRUE(bytes.ok()); + std::vector storage; + for (unsigned i = 0; i < 32; ++i) + storage.push_back("field-" + std::to_string(random() % 256)); + std::vector fields(storage.begin(), storage.end()); + fields.insert(fields.end(), {"", "", std::string_view("a\0b", 3)}); + auto expected = original; + std::size_t removed = 0; + for (auto field : fields) { + auto it = std::find_if( + expected.entries_.begin(), expected.entries_.end(), + [&](const auto& entry) { return entry.field_ == field; }); + if (it != expected.entries_.end()) { + expected.entries_.erase(it); + ++removed; + } + } + auto planned = PlanCompactHashEdit(*bytes, original.entries_.size(), fields, + {}, 16383, CompactHashEditKind::kDelete); + ASSERT_TRUE(planned.ok()) << planned.status(); + ASSERT_TRUE(planned->has_value()); + // Moving a plan must retain its framing even when its parts spill inline + // capacity; the old image and request storage remain the byte owners. + auto plan = std::move(**planned); + EXPECT_EQ(plan.removed_, removed); + EXPECT_EQ(plan.count_, expected.entries_.size()); + auto encoded = EncodeInlineRecordPayload(plan.payload_); + ASSERT_TRUE(encoded.ok()); + if (expected.entries_.empty()) { + EXPECT_TRUE(encoded->empty()); + } else { + auto reference = EncodeHashValue(expected); + ASSERT_TRUE(reference.ok()); + EXPECT_EQ(*encoded, *reference); + } + } +} + +TEST(CompactHashEdit, DeleteValidatesNoopTailAndLastMember) { + HashValue value{.entries_ = {{.field_ = "a", .value_ = ""}, + {.field_ = "b", .value_ = ""}}}; + auto bytes = EncodeHashValue(value); + ASSERT_TRUE(bytes.ok()); + const std::array missing{"absent"}; + auto no_op = PlanCompactHashEdit(*bytes, 2, missing, {}, 16383, + CompactHashEditKind::kDelete); + ASSERT_TRUE(no_op.ok()); + ASSERT_TRUE(no_op->has_value()); + EXPECT_FALSE((**no_op).changed_); + EXPECT_TRUE((**no_op).payload_.empty()); + EXPECT_FALSE(PlanCompactHashEdit(*bytes + "x", 2, missing, {}, 16383, + CompactHashEditKind::kDelete) + .ok()); + const std::array all{"b", "a", "a"}; + auto empty = PlanCompactHashEdit(*bytes, 2, all, {}, 16383, + CompactHashEditKind::kDelete); + ASSERT_TRUE(empty.ok()); + ASSERT_TRUE(empty->has_value()); + EXPECT_EQ((**empty).removed_, 2); + EXPECT_EQ((**empty).count_, 0); + EXPECT_TRUE((**empty).payload_.empty()); +} + TEST(HashGroupEdits, OrderedDuplicatesOwnTheirBytesAndKeepWireFormat) { HashGroupSnapshot group{.incarnation_ = 17, .value_ = Value(8, 64)}; auto payload = EncodeHashGroup(group); diff --git a/tests/grouped/write_e2e_support.h b/tests/grouped/write_e2e_support.h index f5614d4a1..e2a94b7e1 100644 --- a/tests/grouped/write_e2e_support.h +++ b/tests/grouped/write_e2e_support.h @@ -36,6 +36,7 @@ #include #include #include +#include #include #include #include @@ -342,7 +343,8 @@ class Server { bool pause_handoff = false, unsigned fail_aux_nth = 2, std::string_view max_memory = "1G", std::string_view load_rdb = {}, std::string_view max_memory_clients = {}, unsigned flush_pause_ms = 0, - unsigned flush_max_ms = 20) { + unsigned flush_max_ms = 20, + std::span extra_args = {}) { const int socket = ::socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); Check(socket >= 0, "port socket failed"); sockaddr_in address{.sin_family = AF_INET, @@ -415,6 +417,7 @@ class Server { args.emplace_back("--maxmemory-clients"); args.emplace_back(max_memory_clients); } + for (const auto arg : extra_args) args.emplace_back(arg); std::vector argv; for (auto& arg : args) argv.push_back(arg.data()); argv.push_back(nullptr); diff --git a/tests/ordered_compact_codec_test.cpp b/tests/ordered_compact_codec_test.cpp index 9e5169b17..2742f9229 100644 --- a/tests/ordered_compact_codec_test.cpp +++ b/tests/ordered_compact_codec_test.cpp @@ -82,6 +82,52 @@ TEST(OrderedCompactCodecTest, BoundsAndFramingFailClosed) { .ok()); } +TEST(OrderedCompactCodecTest, BorrowedReaderChecksEveryTruncatedBoundary) { + const std::vector entries{ + {.value_ = "", .score_ = 0}, + {.value_ = std::string("a\0b", 3), .score_ = 0}, + {.value_ = std::string(32, 'm'), .score_ = 0}}; + for (auto kind : + {OrderedCollectionKind::kList, OrderedCollectionKind::kSortedSet}) { + auto bytes = EncodeOrderedCompactValue(kind, entries); + ASSERT_TRUE(bytes.ok()); + for (std::size_t length = 0; length <= bytes->size(); ++length) { + auto reader = OrderedCompactReader::Open( + kind, std::string_view(*bytes).substr(0, length), entries.size()); + bool valid = reader.ok(); + if (valid) { + for (std::size_t i = 0; i < entries.size(); ++i) { + auto entry = reader->Next(); + if (!entry.ok()) { + valid = false; + break; + } + EXPECT_EQ(entry->value_, entries[i].value_); + EXPECT_EQ(entry->score_, entries[i].score_); + } + } + EXPECT_EQ(valid, length == bytes->size()); + } + auto reader = OrderedCompactReader::Open(kind, *bytes, entries.size()); + ASSERT_TRUE(reader.ok()); + auto copied = *reader; + for (std::size_t i = 0; i < entries.size(); ++i) { + auto a = reader->Next(); + auto b = copied.Next(); + ASSERT_TRUE(a.ok()); + ASSERT_TRUE(b.ok()); + EXPECT_EQ(a->encoded_, b->encoded_); + } + EXPECT_FALSE(reader->Next().ok()); + const auto trailing = *bytes + "x"; + auto broken = OrderedCompactReader::Open(kind, trailing, entries.size()); + ASSERT_TRUE(broken.ok()); + ASSERT_TRUE(broken->Next().ok()); + ASSERT_TRUE(broken->Next().ok()); + EXPECT_FALSE(broken->Next().ok()); + } +} + TEST(OrderedCompactCodecTest, LargeItemAndInvalidScore) { std::vector entries{ {.value_ = std::string(9 * 1024 * 1024, 'x')}};