feat(eviction): WAL-fallback async unlocking (v1.7.0 T8)

find() previously held the per-collection shared mutex across the
Phase 2 WAL scan of evicted docs. Concurrent finds against the same
collection serialized on disk I/O — the Zoe incident's contention
amplifier. Phase 1 (in-memory match scan) now runs under the shared
lock and releases it before Phase 2's per-evicted-doc WAL reads, so
concurrent readers can proceed against the live map while one reader
pays the disk cost.

get()'s per-collection lock was already released before the fallback
path. But loadEvictedDocument() and peekEvictedDocument() held
callbackMutex_ across the documentLoadCallback_ invocation itself,
which globally serialized evicted-doc reads regardless of collection.
Now we snapshot the callback under the short mutex, release it, and
invoke the callback with no locks held.

loadEvictedDocument() also:
 - Handles the race where a concurrent reader already hot-loaded the
   doc: re-check the live map under the unique lock, prefer the
   existing copy, skip the memory accounting bump.
 - Bumps updatedAt on hot-load so T4's hot-write floor protects the
   freshly paged-in doc from being re-evicted in the next pass.

find() keeps its existing read-path behaviour for filter evaluation
over evicted-match candidates — it still WAL-reads each evicted doc
without the collection lock, and any future optimization (page-in
cache, evicted-doc filter index) is a separate task.

service/src/memory_store.cpp

@@ -811,33 +811,39 @@ QueryResult MemoryStore::find(const std::string& collection, const Query& query)
         return {};
     }
 
-    std::shared_lock<std::shared_mutex> lock(coll->mutex);
-
     // Update query count
     {
         std::lock_guard<std::mutex> statsLock(statsMutex_);
         const_cast<MemoryStore*>(this)->stats_.queryCount++;
     }
 
-    // Phase 1: Search in-memory documents
+    // Phase 1: Search in-memory documents under shared lock. Release the
+    // lock before Phase 2's WAL reads so concurrent readers against the
+    // same collection don't serialize on disk I/O (v1.7.0 T8).
     auto memoryStartTime = std::chrono::steady_clock::now();
     std::vector<Document> matches;
     uint32_t memoryMatchCount = 0;
 
-    for (const auto& [id, doc] : coll->documents) {
-        if (doc.isExpired()) {
-            continue;
-        }
-        if (matchesFilters(doc, query.filters)) {
-            matches.push_back(doc);
-            memoryMatchCount++;
+    {
+        std::shared_lock<std::shared_mutex> lock(coll->mutex);
+        for (const auto& [id, doc] : coll->documents) {
+            if (doc.isExpired()) {
+                continue;
+            }
+            if (matchesFilters(doc, query.filters)) {
+                matches.push_back(doc);
+                memoryMatchCount++;
+            }
         }
     }
 
     auto memoryElapsed = std::chrono::duration_cast<std::chrono::microseconds>(
         std::chrono::steady_clock::now() - memoryStartTime).count();
 
-    // Phase 2: Search evicted documents from WAL (only if there are any)
+    // Phase 2: Search evicted documents from WAL (only if there are any).
+    // Runs WITHOUT the per-collection shared lock held — concurrent finds
+    // against the same collection can proceed against the live map while
+    // we page in evicted payloads.
     uint32_t walMatchCount = 0;
     uint64_t walElapsed = 0;
     bool usedWalFallback = false;
@@ -2771,13 +2777,19 @@ std::optional<Document> MemoryStore::loadEvictedDocument(const std::string& coll
         stub = stubIt->second;
     }
 
-    // Load from persistence via callback
-    std::optional<Document> doc;
+    // v1.7.0 T8: snapshot the callback under the short callbackMutex_ and
+    // release it before doing the WAL I/O. Holding callbackMutex_ across the
+    // callback call would globally serialize evicted-doc reads — the Zoe
+    // incident amplifier.
+    DocumentLoadCallback cb;
     {
         std::lock_guard<std::mutex> lock(callbackMutex_);
-        if (documentLoadCallback_) {
-            doc = documentLoadCallback_(collection, id, stub.walSequence);
-        }
+        cb = documentLoadCallback_;
+    }
+
+    std::optional<Document> doc;
+    if (cb) {
+        doc = cb(collection, id, stub.walSequence);
     }
 
     if (!doc) {
@@ -2785,48 +2797,72 @@ std::optional<Document> MemoryStore::loadEvictedDocument(const std::string& coll
         return std::nullopt;
     }
 
-    // Re-insert into memory
+    doc->collection = collection;
+
+    // Re-insert into memory (hot-load). Another reader may have raced us; if so,
+    // prefer their already-inserted copy and skip our accounting bump.
     CollectionData* coll = getOrCreateCollection(collection);
+    bool inserted = false;
+    Document returned;
     {
         std::unique_lock<std::shared_mutex> collLock(coll->mutex);
 
-        doc->collection = collection;
-        doc->lastAccessedAt = currentTimeMs();  // Mark as recently accessed
+        auto existing = coll->documents.find(id);
+        if (existing != coll->documents.end()) {
+            // Another thread hot-loaded concurrently — use their copy, don't
+            // double-count memory.
+            returned = existing->second;
+        } else {
+            const uint64_t now = currentTimeMs();
+            doc->lastAccessedAt = now;  // Mark as recently accessed
+            // v1.7.0 T8: bump updatedAt so T4's hot-write floor protects the
+            // freshly paged-in doc from being re-evicted in the next pass.
+            // This is conservative; the logical document content is unchanged.
+            if (doc->updatedAt == 0 || doc->updatedAt < now) {
+                doc->updatedAt = now;
+            }
 
-        // Add to expiration index if TTL is set
-        if (doc->expiresAt > 0) {
-            addToExpirationIndex(*coll, id, doc->expiresAt);
-        }
+            // Add to expiration index if TTL is set
+            if (doc->expiresAt > 0) {
+                addToExpirationIndex(*coll, id, doc->expiresAt);
+            }
 
-        coll->documents[id] = *doc;
+            coll->documents.emplace(id, *doc);
+            inserted = true;
+            returned = *doc;
+        }
     }
 
-    // Remove from evicted map
-    {
-        std::unique_lock<std::shared_mutex> evictLock(evictionMutex_);
-        auto collIt = evictedDocs_.find(collection);
-        if (collIt != evictedDocs_.end()) {
-            collIt->second.erase(id);
-            if (collIt->second.empty()) {
-                evictedDocs_.erase(collIt);
+    if (inserted) {
+        // Remove from evicted map
+        {
+            std::unique_lock<std::shared_mutex> evictLock(evictionMutex_);
+            auto collIt = evictedDocs_.find(collection);
+            if (collIt != evictedDocs_.end()) {
+                collIt->second.erase(id);
+                if (collIt->second.empty()) {
+                    evictedDocs_.erase(collIt);
+                }
             }
         }
-    }
 
-    // Update stats
-    {
-        std::lock_guard<std::mutex> statsLock(statsMutex_);
-        stats_.totalDocuments++;
-        stats_.evictedCount--;
-        stats_.recoveryCount++;
-    }
+        // Update stats
+        {
+            std::lock_guard<std::mutex> statsLock(statsMutex_);
+            stats_.totalDocuments++;
+            stats_.evictedCount--;
+            stats_.recoveryCount++;
+        }
 
-    // Update memory tracking atomically (add recovered document size)
-    estimatedMemoryBytes_.fetch_add(estimateDocumentSize(*doc), std::memory_order_relaxed);
+        // Update memory tracking atomically (add recovered document size)
+        estimatedMemoryBytes_.fetch_add(estimateDocumentSize(returned), std::memory_order_relaxed);
 
-    spdlog::debug("Recovered evicted document {}/{}", collection, id);
+        spdlog::debug("Recovered evicted document {}/{}", collection, id);
+    } else {
+        SPDLOG_DEBUG("Raced on hot-load of {}/{}, used concurrent reader's copy", collection, id);
+    }
 
-    return doc;
+    return returned;
 }
 
 bool MemoryStore::isDocumentEvicted(const std::string& collection, const std::string& id) const {
@@ -2892,15 +2928,19 @@ std::optional<Document> MemoryStore::peekEvictedDocument(
         stub = stubIt->second;
     }
 
-    // Load from persistence via callback (without re-inserting to memory)
-    std::optional<Document> doc;
+    // v1.7.0 T8: snapshot the callback under the short callbackMutex_ and
+    // release it before doing the WAL I/O. Holding callbackMutex_ across the
+    // callback call would globally serialize concurrent evicted-doc reads.
+    DocumentLoadCallback cb;
     {
         std::lock_guard<std::mutex> lock(callbackMutex_);
-        if (documentLoadCallback_) {
-            doc = documentLoadCallback_(collection, id, stub.walSequence);
-        }
+        cb = documentLoadCallback_;
     }
 
+    std::optional<Document> doc;
+    if (cb) {
+        doc = cb(collection, id, stub.walSequence);
+    }
     return doc;
 }