Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions docs/architecture/09-grouped-collections.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
75 changes: 75 additions & 0 deletions include/lavik/storage/detail/compact_payload.h
Original file line number Diff line number Diff line change
@@ -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 <array>
#include <cassert>
#include <cstdint>

#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<char>(value >> (8 * i));
bytes_ += width;
}

std::size_t encoded_bytes() const noexcept { return bytes_; }
bool empty() const noexcept { return bytes_ == 0; }
std::optional<std::string_view> 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<char, 32> owned{};
unsigned owned_size = 0;
};
absl::InlinedVector<Part, 8> parts_;
std::size_t bytes_ = 0;
std::size_t next_ = 0;
};

} // namespace lavik::storage
22 changes: 22 additions & 0 deletions include/lavik/storage/detail/hash_codec.h
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <vector>

#include "absl/status/statusor.h"
#include "lavik/storage/detail/compact_payload.h"
#include "lavik/storage/format.h"

namespace lavik::storage {
Expand Down Expand Up @@ -78,6 +79,27 @@ class HashValueReader {
absl::StatusOr<HashValue> DecodeHashValue(std::string_view payload);
absl::StatusOr<std::string> 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<std::optional<CompactHashEditPlan>> PlanCompactHashEdit(
std::string_view payload, std::size_t expected_count,
std::span<const std::string_view> fields,
std::span<const std::string_view> 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 {
Expand Down
27 changes: 27 additions & 0 deletions include/lavik/storage/detail/ordered_compact_codec.h
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrderedCompactReader> Open(
OrderedCollectionKind kind, std::string_view encoded,
std::uint64_t expected_count);
std::size_t size() const noexcept { return size_; }
absl::StatusOr<OrderedCompactEntryView> 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
Expand Down
7 changes: 7 additions & 0 deletions include/lavik/storage/engine.h
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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<void(std::uint64_t)> 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
Expand Down
27 changes: 26 additions & 1 deletion src/storage/engine/compact_api.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,30 @@ Task<absl::Status> 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.
Expand Down Expand Up @@ -541,7 +565,8 @@ Task<absl::Status> 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;
}

Expand Down
67 changes: 56 additions & 11 deletions src/storage/engine/grouped/string.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<LoadedValue> 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<const char*>(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<const char*>(data.data()), data.size()};
}
std::string bytes;
CompactPayload plan;
std::optional<InlineRecordPayload> 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<std::uint64_t>(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 {
Expand Down
Loading