perf(storage): read eligible disk GETs on the connection worker

CMakeLists.txt

@@ -1685,10 +1685,12 @@ if (BUILD_TESTING)
       NAME cluster_integration.replication_manager_control_api
       COMMAND lavik_cluster_replication_manager_integration_test
     )
+    # Allow repeated server recovery across the full suite on Debug CI runners.
+    # Individual operation waits remain bounded inside each test.
     set_tests_properties(
       cluster_integration.replication_manager_control_api PROPERTIES
       LABELS cluster-integration
-      TIMEOUT 90
+      TIMEOUT 180
       RUN_SERIAL TRUE
     )
 
@@ -1767,6 +1769,7 @@ if (BUILD_TESTING)
     tests/rdb_collection_test.cpp
     tests/record_location_test.cpp
     tests/load_value_awaitable_test.cpp
+    tests/remote_get_awaitable_test.cpp
     tests/request_gate_test.cpp
     tests/replication_command_test.cpp
     tests/replication_group_test.cpp
@@ -2149,6 +2152,14 @@ endforeach()
     )
     set_tests_properties(lavik_extent_recovery_e2e PROPERTIES TIMEOUT 600)
 
+    add_test(
+      NAME lavik_remote_get_e2e
+      COMMAND ${Python3_EXECUTABLE}
+              ${CMAKE_CURRENT_SOURCE_DIR}/tests/remote_get_e2e.py
+              $<TARGET_FILE:lavik>
+    )
+    set_tests_properties(lavik_remote_get_e2e PROPERTIES TIMEOUT 180)
+
     add_test(
       NAME lavik_defrag_payload_e2e
       COMMAND ${Python3_EXECUTABLE}
@@ -2235,8 +2246,10 @@ endforeach()
       NAME lavik_grouped_ordered_write_e2e
       COMMAND lavik_grouped_ordered_write_e2e_test $<TARGET_FILE:lavik>
     )
+    # This aggregate suite repeatedly starts and recovers storage, including
+    # large RDB and crash-recovery cases on Debug CI runners.
     set_tests_properties(lavik_grouped_ordered_write_e2e PROPERTIES
-      TIMEOUT 1200 RUN_SERIAL TRUE
+      TIMEOUT 1800 RUN_SERIAL TRUE
     )
 
     add_executable(lavik_list_e2e_test

docs/architecture/02-request-serving.md

@@ -51,7 +51,9 @@ worker-local client record separately tracks data exposed by
 `CLIENT`, including RESP version, client library name/version, subscription
 counts, blocking state, and replication-session identity. All disconnect paths
 return through one cleanup point which unregisters client, monitor, Pub/Sub, and
-WATCH state.
+WATCH state. Peer EOF remains latched across asynchronous command work;
+registering a blocking waiter after EOF immediately cancels it, including
+waiters created by commands already buffered when the peer disconnected.
 
 Meta-managed serving boundaries retire ordinary client connections, including
 idle, blocking and Pub/Sub sessions. After authority admission closes, each
@@ -198,6 +200,9 @@ owner routing.
 The selected database is request data, never ambient worker state. Storage
 chooses an owner from the key's Redis hash-slot partition. Commands with one
 owner can execute locally or through one Bycorf cross-worker submission.
+Eligible standalone disk GETs first try an admitted index snapshot and physical
+read on the connection worker; contention or unsupported storage representations
+fall back to owner dispatch. Transaction callbacks retain owner routing.
 Commands needing atomic access to several keys build a `tx::Transaction` and
 execute one or more shard callbacks. Global commands explicitly collect from or
 coordinate all workers.

docs/architecture/03-transaction-coordination.md

@@ -74,6 +74,15 @@ has two counter layers:
 - holds cover callbacks that are currently executing or suspended and tell the
   queue when a conflicting predecessor has actually drained.
 
+Storage binds admission hooks to each shard. Recording a key intent closes its
+index reader stripe before hold acquisition, and releasing or cancelling that
+intent releases the closure. Closures nest across keys sharing a stripe and
+structural index writers. Existing remote index readers must drain before the
+callback takes its holds; the same ordered and bypass queues handle this wait.
+The last reader of a closed stripe sends a worker notification that repolls
+eligibility rather than resuming all transaction frames. An already pinned disk
+read holds no index reader registration, so it does not delay a later writer.
+
 The queue contains both plain acquisition waiters and persistent transaction
 entries. It is sorted by scheduling id. Plain waiters are removed before their
 coroutine resumes; transaction entries retain their position until a releasing
@@ -86,9 +95,9 @@ reaches the front.
    allocating a scheduling id because only one owner participates.
 2. `Execute` submits to that owner and acquires the deduplicated key set through
    `TxShard::AcquireKeys` if the transaction does not already retain a guard.
-3. Acquisition records every intent. If all intents are immediately compatible,
+3. Acquisition records every intent. If all intents are immediately compatible and index readers have drained,
    it acquires holds and continues without touching the global id counter or
-   shard queue. If any intent conflicts, the awaiter keeps all recorded intents,
+   shard queue. If an intent conflicts or readers remain, the awaiter keeps all recorded intents,
    allocates an id, inserts a plain waiter, and suspends until it is the
    hold-compatible queue head.
 4. The shard entry hook runs once, then the callback runs. A releasing execution

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

@@ -80,20 +80,20 @@ Storage separates three ownership domains:
 - Each device has one allocator owner, which serializes that device's ready and
   cold-free pools, scan bitmap, fixed metadata page generations, epoch mirrors,
   and allocation-epoch counter. io_uring initially assigns this role as
-  `device_index % worker_count`. SPDK selects it from the controller's qpair
-  owners, so metadata and allocation I/O never route to a worker that cannot
-  open the namespace.
+  `device_index % worker_count`. SPDK selects it from the controller's home
+  workers, all of which hold a qpair, so metadata and allocation I/O never
+  route to a worker that cannot open the namespace.
 
 Every worker has one ordinary and one transaction append stream. Physical
 streams are per worker, not per logical partition, so active 8 MiB staging
 buffers scale with workers rather than with 16,384 slots. A worker also owns the
 `BlockState` objects assigned to it, including committed and live byte counts,
-pins, staging identity, flush state, and defrag state. The runtime owner and
-allocation epoch form immutable identity after publication and are the only
-fields a key-index owner may read directly. The owner is published atomically
-after epoch initialization, so an acquire owner read makes the immutable epoch
-visible without an extra worker hop. Other state remains owner-local and other
-workers use Bycorf cross-core submissions to read or mutate it.
+staging identity, flush state, and defrag state. Runtime owner and allocation
+epoch form immutable identity after publication. A shared atomic pin word
+protects physical lifetime and closes admission to connection-worker reads
+while maintenance owns the block. The owner is published atomically after
+epoch initialization. Other mutable block fields remain owner-local; remote
+workers submit accounting and maintenance work to that owner.
 
 Append-stream creation and rollover are single-flight within each stream. A
 writer may release owner-local store state while physical allocation waits,
@@ -122,6 +122,11 @@ best-effort subset registration, prioritizing read slots. Each lease retains
 its slot identity and immutable registration status; unregistered slots use
 the same aligned memory with plain asynchronous I/O. On SPDK, a
 worker opens only namespaces whose physical controller assigned it a qpair.
+Weighted home-device assignment controls write placement independently of
+qpair access. Each controller reserves qpairs for its home workers, then
+assigns its remaining capacity to other workers, up to one qpair per worker.
+A controller with enough qpairs is therefore directly accessible from every
+connection worker. Namespaces on that controller share each worker's qpair.
 SPDK also requires DMA buffer registration and fails initialization when that
 registration or controller-qpair coverage is insufficient.
 
@@ -368,10 +373,10 @@ zeroes stale checkpoint bitmap state when no checkpoint is usable.
 `InitializeWorker` creates the worker's aligned buffer pool, registers the
 fixed-file table, and opens the paths accessible on that backend. Before worker
 startup, io_uring assigns weighted home devices for foreground allocation.
-SPDK groups namespaces by physical controller, weights controllers by usable
-foreground blocks, and distributes available controller qpairs across workers;
-startup fails unless every controller and every worker can be covered. Worker
-barriers then coordinate one parallel recovery rather than independent
+SPDK groups namespaces by physical controller, assigns weighted home devices,
+and expands qpair access with the remaining controller capacity; startup fails
+unless every controller and every worker can be covered. Worker barriers then
+coordinate one parallel recovery rather than independent
 worker-local boots. Request readiness follows successful completion of the
 complete recovery and allocator-cleanup sequence; the exact moment a listening
 socket exists is not the readiness boundary.
@@ -519,11 +524,39 @@ seed and top-level digests appear only in the one-use checkpoint, never in
 authoritative records or replication protocols.
 
 All partition and logical-database indexes on a worker allocate entries from a
-shared worker-local arena, so sparse indexes share capacity instead of
-stranding it at partition boundaries. A population detached by `FLUSHDB`, a
-replica reset, or a replica abort retains ownership of that arena until
-reclamation finishes. Replica promotion and abort reuse the same worker-local
-detached-index drain as synchronous database flushes, so a completed lifecycle
+shared arena, so sparse indexes share capacity instead of stranding it at
+partition boundaries. Startup allocates its complete page directory, charging
+16 MiB of descriptor storage to each worker before recovery; entry pages remain
+demand-allocated. The directory stays at one address and publishes page layout
+atomically. Handle-to-address resolution can run concurrently with owner-side
+allocation when the caller protects the resolved entry and arena lifetimes;
+it does not pin entries or make the hash tables safe for concurrent access.
+Allocator mutation remains owner-local. Private auxiliary arenas retain
+incremental directory allocation and owner-only access.
+Each primary key index reserves 32 cache-line-isolated atomic read-admission
+states, including while empty. Its bucket arrays remain demand-allocated,
+start at 32 buckets, and never shrink below 32 while nonempty; checkpoint
+recovery applies the same floor. Reader admission is explicitly enabled only
+for a published map with a fixed arena directory and an externally protected
+map lifetime. A remote reader registers atomically in the low-hash stripe,
+performs only const lookup, and releases its guard before suspending. Writers
+close the affected stripe and suspend their owner coroutine until existing
+readers drain, then resolve entry identities again before mutation. The last
+reader of a closed stripe notifies the owner; transaction arbitration and
+structural writers recheck readiness there.
+Shared overflow-directory growth and population replacement close all stripes.
+Resizing closes admission before changing table layout, defers behind existing
+readers, and stays closed through the final table switch and old-table release.
+Owner lookups and periodic maintenance retry deferred resizing; nesting keeps
+resize completion from reopening another writer's scope. Recovery and detached
+populations remain private, and grouped-object and auxiliary maps retain their
+owner-only layout and bucket sizing. Transaction intents retain stripe closure
+through queued and executing work, so a remote index snapshot cannot observe
+an intermediate multi-key mutation.
+A population detached by `FLUSHDB`, a replica reset, or a replica abort retains
+ownership of that arena until reclamation finishes. Replica promotion and
+abort reuse the same worker-local detached-index drain as synchronous database
+flushes, so a completed lifecycle
 cannot leave old index arenas or record-block live-byte accounting queued for a
 later rebuild. The drain covers the worker FIFO; it is not partitioned by
 replication attempt.
@@ -634,7 +667,32 @@ need a staging buffer to commit.
 
 ### Reads and pins
 
-An ordinary single-key GET may resolve its complete inline-key index entry
+A standalone GET can read directly on its connection worker after ordinary
+request, database and serving-generation admission. It takes a short index
+reader guard on the key owner, snapshots a complete-key plain String location,
+then pins the block and validates its allocation identity before dropping the
+index guard. Only durable, non-grouped, non-external records whose block owner
+is also their key owner qualify, and the connection worker must have device
+I/O access. Closed index/block admission, unsupported representations, expired
+entries, concurrent admission contention, and an exhausted connection-worker
+read-buffer pool return to the normal owner route. An admitted fast read submits
+I/O directly from an awaitable in the requesting coroutine; completion resumes
+that same coroutine for record and population validation, without a child read
+coroutine. Index misses use the same transaction and population admission checks.
+
+The index snapshot orders the GET before a subsequently admitted writer. The
+block pin survives I/O and cancellation cleanup, so writes can append and retire
+old live-byte accounting while the read finishes. Defrag closes remote pin
+admission before inspecting the count; the last pin on a pending block asks its
+owner to reconsider reclamation. Flush uses the same close-before-count order
+when retiring a sealed staging buffer: either it releases the buffer immediately
+or the last pin notifies the owner to finish the release. Local fallback reads
+retain their existing pin rules. Database/replication epoch and population-generation
+changes invalidate an in-flight fast read. Worker shutdown drains I/O and destroys
+suspended frames on all workers before destroying any remotely referenced index,
+arena or block state.
+
+An owner-local ordinary single-key GET may resolve its complete inline-key index entry
 without registering a shared transaction lock when that logical database's
 lock table is empty. The empty-table observation and index lookup do not
 suspend, so the lookup is the read's linearization point. A multi-shard writer
@@ -770,9 +828,10 @@ The persistent allocation bitmap is the recovery authority, not merely an
 allocation hint. Device allocators activate ready IDs in batches and preserve
 eight allocatable blocks per device for defrag. Each worker first balances
 foreground allocations across its weighted home devices. io_uring can then
-fall back to every other configured device; SPDK cannot leave the namespaces of
-controllers for which that worker owns qpairs. Foreground allocation may wait
-while an active flush, extent reclaim, or runnable defrag can still return
+fall back to every other configured device; SPDK allocates only from its home
+devices, even when spare qpairs give it read access to other controllers.
+Foreground allocation may wait while an active flush, extent reclaim, or
+runnable defrag can still return
 space; it reports exhaustion only after a stable observation with no progress
 in flight. A queued defrag while defrag is paused is deliberately not counted
 as runnable progress, so a full-device write reports exhaustion instead of
@@ -1156,9 +1215,9 @@ current source code are authoritative for present storage behavior.
 |---|---|
 | Public lifecycle, routing, typed operations, locked transaction contract, snapshots, epochs, maintenance, and durability interfaces | `include/lavik/storage/engine.h` |
 | Worker, partition, block, append-stream, allocator, recovery, and background-maintenance state | `src/storage/engine/impl.h` |
-| Runtime index representation, shared entry arena, runtime key digests, and asynchronous entry-identity validation | `include/lavik/storage/scan_hash_map.h`, `include/lavik/storage/format.h`, `src/storage/format.cpp`, `src/storage/engine/impl.h`, `src/storage/engine/write.cpp`, `src/storage/engine/flush.cpp` |
+| Runtime index representation, shared entry arena, runtime key digests, and asynchronous entry-identity validation | `include/lavik/storage/scan_hash_map.h`, `include/lavik/storage/format.h`, `src/storage/format.cpp`, `src/storage/engine/impl.h`, `src/storage/engine/init.cpp`, `src/storage/engine/write.cpp`, `src/storage/engine/flush.cpp` |
 | Compact and grouped Hash/Set serving, group identity and object side index | `src/storage/engine/hash_tree.cpp`, `src/storage/engine/hash_codec.cpp`, `src/storage/engine/grouped/hash.cpp`, `src/storage/engine/grouped/object_index.cpp`, [Grouped collections](09-grouped-collections.md) |
-| Compact physical index representation shared by user-key and group-location indexes | `include/lavik/storage/detail/record_index.h` |
+| Compact physical index representation shared by user-key and group-location indexes | `include/lavik/storage/detail/record_location.h`, `include/lavik/storage/detail/record_index.h` |
 | Persistent constants, device and block IDs, A/B metadata pages, record and extent layouts, and checksums | `include/lavik/storage/format.h`, `src/storage/format.cpp` |
 | Checkpoint serialization, bitmap validation, generation publication and consumption, fallback, and block retirement | `src/storage/engine/checkpoint.cpp`, `src/storage/engine/flush.cpp`, `src/storage/engine/init.cpp`, `src/storage/engine/recovery.cpp` |
 | System-state manifest, catalog COW extents, full-sync fence, population token, promotion base, and one-use shutdown certificate | `src/storage/engine/system_state.cpp`, `include/lavik/storage/engine.h` |

docs/operations/multi-device-storage.md

@@ -180,7 +180,19 @@ stream allocates on demand and does not reserve standby capacity.
 - With more usable devices than workers, devices are greedily grouped to keep
   aggregate group weights balanced. A worker chooses within its group by the
   lowest allocated-blocks/weight ratio.
-- Allocation falls back to other devices when the preferred device is full.
+- On io_uring, allocation falls back to other devices when the preferred device
+  is full. SPDK allocation stays within the worker's home devices.
+
+SPDK groups namespaces by physical controller. Home assignments follow the
+foreground capacity weights, bounded by each controller's available qpair
+count. Qpair access is broader than write affinity: after covering home
+workers, each controller assigns spare qpairs to other workers, preferring
+those with fewer qpairs, up to one per worker. For six controllers with 16
+available qpairs each and 16 workers, every worker opens all six controllers
+(96 qpairs total), while retaining its weighted home write placement. Multiple
+namespaces on one controller share a worker's qpair. Startup logs show
+`home-workers` and `qpair-workers` separately. If a controller has fewer qpairs
+than workers, connection-worker GET falls back when its worker lacks access.
 
 The standby request starts immediately after an ordinary active block is
 installed, and starts again after rollover consumes that standby. Appends do

include/lavik/command.h

@@ -665,6 +665,8 @@ void SetClientName(std::uint64_t id, std::string name) noexcept;
 void SetClientRespVersion(std::uint64_t id, RespVersion version) noexcept;
 void SetClientPubSubCounts(std::uint64_t id, std::size_t subscriptions,
                            std::size_t pattern_subscriptions) noexcept;
+// Updates CLIENT state. After registering a waiter, setting blocked also
+// cancels it if EOF or connection retirement already occurred.
 void SetClientBlocked(std::uint64_t id, bool blocked) noexcept;
 void UnregisterClientConnection(std::uint64_t id) noexcept;