perf(storage): use direct awaitables for value and extent reads

CMakeLists.txt

@@ -1766,6 +1766,7 @@ if (BUILD_TESTING)
     tests/rdb_test.cpp
     tests/rdb_collection_test.cpp
     tests/record_location_test.cpp
+    tests/load_value_awaitable_test.cpp
     tests/request_gate_test.cpp
     tests/replication_command_test.cpp
     tests/replication_group_test.cpp

docs/architecture/04-storage-and-recovery.md

@@ -649,7 +649,17 @@ rereads. Transaction-owned reads always use that locked path directly.
 Tombstones and expired values are invisible; an expired observation can enqueue
 a bounded active-expiration candidate. A staged location is copied or framed
 from its write buffer. A disk location is read on its physical block owner into
-an aligned lease.
+an aligned lease. `LoadValue` and its single-location `LoadValueLocal` share an
+awaitable embedded in the caller's task. Local inline reads, including staged
+copies, use that task directly. Buffer contention retains the pool's FIFO
+handoff and schedules an owner-local continuation under the caller's task class;
+it does not require a waiter coroutine. The awaitable owns the disk-read pin
+and buffer through completion and validates record identity, checksums, and
+population before returning it. Grouped reads enter their graph reader directly;
+cross-owner, external, and relocation flows use lazy helper tasks. A stale
+logical read resolves the current same-mutation location before resuming its
+caller, while a single-location read returns the stale status to its enclosing
+representation-specific retry loop.
 
 Worker-local MGET uses `BatchGetLocked` after the command has acquired its key
 locks. It classifies index entries in one coroutine and issues ordinary-size
@@ -668,9 +678,13 @@ taking the reader's key lock; a stale physical read aborts and follows the
 current same-sequence relocation, while a true logical mutation becomes a
 normal miss.
 
-External values read and validate every extent on its current owner in bounded
-parallel waves, then assemble the logical value. Every read in a wave finishes
-before its shared output buffer can be released, including on error. A move-only
+External values read and validate every extent on its current owner and
+assemble the logical value in manifest order. Whole-extent and slice reads
+share a physical-read awaitable that retains its pin across buffer waiting,
+I/O, and validation. Slices validate the entire extent before copying any
+requested bytes. Local extent routing and physical reads need no child task;
+cross-owner dispatch retains the runtime's task bridge. The output buffer
+outlives every extent read, including on error. A move-only
 `ReadBufferLease` may cross to the connection worker; destruction returns a
 registered slot to its storage owner.
 

include/lavik/storage/buffer_pool.h

@@ -32,6 +32,7 @@
 namespace bycorf {
 class CrossCore;
 class Worker;
+struct RemoteWork;
 }  // namespace bycorf
 
 namespace lavik::storage {
@@ -208,12 +209,20 @@ class RegisteredBufferPool {
     bool await_suspend(std::coroutine_handle<> awaiting);
     absl::StatusOr<ReadBufferLease> await_resume();
 
+    // Queue an owner-local continuation when a slot is handed off. This uses
+    // the worker's scheduled work queue (including its task-class budget),
+    // never runs inline in Release(), and sends no cross-worker reply. The
+    // caller supplies run_fn_/task_class_ and keeps both objects alive until
+    // that callback consumes the assigned lease through await_resume().
+    bool SuspendWith(bycorf::RemoteWork& continuation);
+
    private:
     friend class RegisteredBufferPool;
 
     RegisteredBufferPool* pool_ = nullptr;
     std::size_t minimum_payload_bytes_ = 0;
     std::coroutine_handle<> awaiting_{};
+    bycorf::RemoteWork* continuation_ = nullptr;
     std::uint16_t assigned_buffer_id_ = 0;
   };
 

include/lavik/storage/engine.h

@@ -1944,6 +1944,7 @@ class StorageEngine {
  private:
   friend class ExpirationAuthorityTestPeer;
   friend class TombRaiderTestPeer;
+  friend class LoadValueAwaitableTestPeer;
   class Impl;
   std::unique_ptr<Impl> impl_;
 };

src/storage/buffer_pool.cpp

@@ -468,16 +468,22 @@ bool RegisteredBufferPool::AcquireReadAwaiter::await_ready() const noexcept {
 
 bool RegisteredBufferPool::AcquireReadAwaiter::await_suspend(
     std::coroutine_handle<> awaiting) {
-  if (pool_ == nullptr || !pool_->initialized() ||
-      minimum_payload_bytes_ > pool_->options_.read_payload_bytes() ||
-      !pool_->free_read_buffers_.empty()) {
-    return false;
-  }
+  if (await_ready()) return false;
   awaiting_ = awaiting;
   pool_->read_waiters_.push_back(this);
   return true;
 }
 
+bool RegisteredBufferPool::AcquireReadAwaiter::SuspendWith(
+    bycorf::RemoteWork& continuation) {
+  if (await_ready()) return false;
+  continuation_ = &continuation;
+  continuation.origin_ = pool_->owner_worker_;
+  continuation.reply_deferred_ = true;
+  pool_->read_waiters_.push_back(this);
+  return true;
+}
+
 absl::StatusOr<ReadBufferLease>
 RegisteredBufferPool::AcquireReadAwaiter::await_resume() {
   if (pool_ == nullptr || !pool_->initialized()) {
@@ -717,7 +723,9 @@ void RegisteredBufferPool::ReleaseLocal(std::uint16_t buffer_id) noexcept {
     AcquireReadAwaiter* waiter = read_waiters_.front();
     read_waiters_.pop_front();
     waiter->assigned_buffer_id_ = buffer_id;
-    if (worker_ != nullptr) {
+    if (waiter->continuation_ != nullptr) {
+      bycorf::PostRequest(cross_core_, owner_worker_, waiter->continuation_);
+    } else if (worker_ != nullptr) {
       worker_->Enqueue(waiter->awaiting_);
     }
     return;

src/storage/engine/impl.h

@@ -3257,6 +3257,7 @@ class StorageEngine::Impl {
  private:
   friend class ExpirationAuthorityTestPeer;
   friend class TombRaiderTestPeer;
+  friend class LoadValueAwaitableTestPeer;
 
   struct DurableSystemState {
     std::uint64_t generation_ = 0;
@@ -3342,6 +3343,118 @@ class StorageEngine::Impl {
     }
   };
 
+  // The awaitable lives in the caller's frame. Ordinary local reads suspend
+  // that caller directly, including through pool backpressure. Only multi-step
+  // logical reads need a lazy helper Task. Local mode returns stale-location
+  // errors to its caller, while logical mode resolves same-mutation relocations
+  // before returning.
+  class LoadValueAwaitable final : public bycorf::OneShotIoAwaitable,
+                                   private bycorf::RemoteWork {
+   public:
+    using Result = absl::StatusOr<LoadedValue>;
+    LoadValueAwaitable(Impl& engine, WorkerStore& store,
+                       WorkerStore::PartitionStore& partition,
+                       std::uint8_t db_id, std::string_view key,
+                       const Digest& digest, RecordLocation location,
+                       ExtentManifest extents, ReadLatencyTrace* trace,
+                       GroupedObject::Handle grouped_snapshot);
+    LoadValueAwaitable(Impl& engine, WorkerStore& store, std::uint8_t db_id,
+                       std::string_view key, RecordLocation location,
+                       std::uint64_t replication_epoch, ReadLatencyTrace* trace,
+                       std::optional<std::uint64_t> expected_db_epoch);
+    ~LoadValueAwaitable();
+
+    bool await_ready() const noexcept { return ready_ && !slow_.valid(); }
+    std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting);
+    Result await_resume();
+    void Complete(bycorf::Worker& worker, int result, unsigned flags) override;
+
+   private:
+    friend class LoadValueAwaitableTestPeer;
+    struct Payload {
+      std::uint64_t txid_;
+      std::uint32_t offset_;
+      std::uint32_t bytes_;
+      std::uint32_t checksum_;
+    };
+
+    void PrepareLocal();
+    void FinishBufferWait();
+    static void BufferReady(bycorf::RemoteWork* work);
+    std::coroutine_handle<> SuspendRead(std::coroutine_handle<> awaiting);
+    absl::StatusOr<Payload> ParseRecord(const std::byte* data) const;
+    Result FinishRecord(const std::byte* data, Payload payload, bool staged);
+    bool PopulationChanged() const;
+    void Unpin();
+    void Retry(absl::Status failure);
+
+    Impl& engine_;
+    WorkerStore& store_;
+    WorkerStore::PartitionStore* partition_ = nullptr;
+    std::string_view key_;
+    Digest digest_{};
+    RecordLocation location_;
+    ReadLatencyTrace* trace_;
+    std::uint64_t replication_epoch_;
+    std::optional<std::uint64_t> expected_db_epoch_;
+    std::uint64_t aligned_offset_ = 0;
+    std::size_t file_id_ = 0;
+    std::size_t record_headroom_ = 0;
+    std::size_t read_bytes_ = 0;
+    std::uint8_t db_id_;
+    BlockState* pinned_ = nullptr;
+    std::optional<Payload> payload_;
+    std::optional<Result> ready_;
+    // Runtime shutdown drains I/O before destroying suspended frames. The
+    // waiter retains its assigned slot until the scheduled continuation runs.
+    ReadBufferLease lease_;
+    std::optional<RegisteredBufferPool::AcquireReadAwaiter> buffer_wait_;
+    Task<Result> slow_;
+  };
+
+  // One physical extent read, including buffer waiting and slice validation,
+  // embedded in the caller. Routing adds a runtime task bridge only when the
+  // block belongs to another worker; local reads need no child coroutine.
+  class ReadExtentAwaitable final : public bycorf::OneShotIoAwaitable,
+                                    private bycorf::RemoteWork {
+   public:
+    ReadExtentAwaitable(Impl& engine, WorkerStore& store, ExtentRef ref,
+                        std::uint32_t extent_index, std::size_t source_offset,
+                        std::span<std::byte> destination, bool route = false);
+    ~ReadExtentAwaitable();
+    bool await_ready() const noexcept { return ready_.has_value(); }
+    bool await_suspend(std::coroutine_handle<> awaiting);
+    absl::Status await_resume();
+
+   private:
+    friend class LoadValueAwaitableTestPeer;
+    struct RemoteRead {
+      Impl* engine_;
+      unsigned owner_;
+      ExtentRef ref_;
+      std::uint32_t extent_index_;
+      std::size_t source_offset_;
+      std::span<std::byte> destination_;
+      Task<absl::Status> operator()();
+    };
+    void FinishBufferWait();
+    static void BufferReady(bycorf::RemoteWork* work);
+    bool SubmitRead(std::coroutine_handle<> awaiting);
+
+    Impl& engine_;
+    WorkerStore& store_;
+    ExtentRef ref_;
+    std::uint32_t extent_index_;
+    std::size_t source_offset_;
+    std::span<std::byte> destination_;
+    std::size_t read_bytes_ = 0;
+    BlockState* pinned_ = nullptr;
+    std::optional<absl::Status> ready_;
+    ReadBufferLease lease_;
+    std::optional<RegisteredBufferPool::AcquireReadAwaiter> buffer_wait_;
+    std::optional<bycorf::SubmitTaskAwaiter<RemoteRead>> remote_;
+  };
+
   // Owns the page lease while a grouped reader inspects its checked envelope.
   // Views into loaded_.value() must not outlive this object.
   struct LoadedHashGroupPayload {
@@ -3726,13 +3839,21 @@ class StorageEngine::Impl {
       WorkerStore& store, RecordIndex& index, const Digest& digest,
       std::string_view key, const RecordLocation& verified_location);
 
-  Task<absl::StatusOr<LoadedValue>> LoadValue(
+  LoadValueAwaitable LoadValue(
       WorkerStore& key_store, WorkerStore::PartitionStore& partition,
       std::uint8_t db_id, std::string_view key, const Digest& digest,
       RecordLocation location, ExtentManifest extents,
       ReadLatencyTrace* trace = nullptr,
       GroupedObject::Handle grouped_snapshot = nullptr);
 
+  // Used only for multi-step reads or an aborted direct read. A supplied
+  // failure enters relocation resolution without reading the stale bytes again.
+  Task<absl::StatusOr<LoadedValue>> LoadValueSlow(
+      WorkerStore& key_store, WorkerStore::PartitionStore& partition,
+      std::uint8_t db_id, std::string_view key, Digest digest,
+      RecordLocation location, ExtentManifest extents, ReadLatencyTrace* trace,
+      std::optional<absl::Status> failure = std::nullopt);
+
   // The optional snapshot view must already own physical pins. Ordinary
   // reads instead retry GC relocation against the same logical root version.
   Task<absl::StatusOr<LoadedHashGroupPayload>> LoadHashGroupPayload(
@@ -3752,26 +3873,22 @@ class StorageEngine::Impl {
       std::uint8_t db_id, std::string_view key, const Digest& digest,
       RecordLocation location, GroupedObject::Handle snapshot);
 
-  // Reads one extent block's payload into `destination`. Runs on the worker
-  // that owns that block, which is not necessarily the one holding the
-  // manifest, so everything it needs is passed by value.
-  Task<absl::Status> ReadExtentInto(WorkerStore& store, ExtentRef ref,
-                                    std::uint32_t extent_index,
-                                    std::byte* destination);
-  Task<absl::Status> ReadExtentOnOwner(ExtentRef ref,
-                                       std::uint32_t extent_index,
-                                       std::byte* destination);
-  Task<absl::Status> ReadExtentSlice(WorkerStore& store, ExtentRef ref,
-                                     std::uint32_t extent_index,
-                                     std::size_t source_offset,
-                                     std::span<std::byte> destination);
+  // Reads an entire extent on its physical owner into caller-owned storage.
+  // Local routing returns the physical awaitable without a wrapper task.
+  ReadExtentAwaitable ReadExtentOnOwner(ExtentRef ref,
+                                        std::uint32_t extent_index,
+                                        std::byte* destination);
+  ReadExtentAwaitable ReadExtentSlice(WorkerStore& store, ExtentRef ref,
+                                      std::uint32_t extent_index,
+                                      std::size_t source_offset,
+                                      std::span<std::byte> destination);
 
   Task<absl::StatusOr<LoadedValue>> LoadExternalValueLocal(
       WorkerStore& store, const RecordLocation& location,
       ExtentManifest extents, ReadLatencyTrace* trace,
       bool grouped_payload = false);
 
-  Task<absl::StatusOr<LoadedValue>> LoadValueLocal(
+  LoadValueAwaitable LoadValueLocal(
       WorkerStore& store, std::uint8_t db_id, std::string_view key,
       RecordLocation location, std::uint64_t replication_epoch,
       ReadLatencyTrace* trace = nullptr,