feat(eviction): chunked, throttled, pressure-aware eviction (v1.7.0 T4 — Zoe fix)

Replaces the one-shot 'evict everything down to target' loop with a
steady pressure-driven drip:

  - At most evictionChunkSize (1000) docs per pass
  - evictionChunkPauseMs (50) between chunks
  - maxEvictionPassesPerTrigger (20) safety cap
  - hotWriteFloorMs (30000) protects in-flight writes
  - Per-collection budget = (share x 1.5) of chunk — one big collection
    can't monopolise the drop
  - Pressure-aware: SOFT = 1 chunk/tick trickle, HARD = half-cap
    aggressive, EMERGENCY = full-cap aggressive
  - Priority-aware sort: Low-memoryPriority collections evicted first,
    then Normal, then High; coldest LRU within each tier

The Zoe incident (137k docs evicted in 3s) would now spread to ~20
passes over 1s+ of wall time with 50ms pauses, giving concurrent
shadowman writes time to land without hitting their 5s gRPC deadline.

Reuses the existing evictDocument() helper (stub creation, memory
counter, expiration index) so WAL-fallback recovery semantics are
unchanged. evictDocuments() remains for explicit one-shot callers.

Quiesce integration (T6) still pending — hot-write floor covers the
bulk of what quiesce would, but an in-flight patch mid-eviction race
is still possible until T6 lands.

service/src/memory_store.cpp

@@ -2355,48 +2355,222 @@ void MemoryStore::setDocumentLoadCallback(DocumentLoadCallback callback) {
     documentLoadCallback_ = std::move(callback);
 }
 
-void MemoryStore::evictionLoop() {
-    uint64_t checkCount = 0;
+void MemoryStore::logMemoryCheck() {
+    auto stats = getStats();
+    spdlog::info("Memory check: {} MB estimated ({}%, pressure={}), {} docs, {} evicted",
+                stats.estimatedMemoryBytes / (1024 * 1024),
+                pressurePercent(),
+                memoryPressureToString(pressure()),
+                stats.totalDocuments,
+                stats.evictedCount);
+}
 
-    while (running_.load()) {
+void MemoryStore::evictionLoop() {
+    // v1.7.0 T4: chunked, throttled, pressure-aware eviction.
+    //
+    // Replaces the pre-1.7 "one shot from 80% down to 60%" loop which, for
+    // large caps, could hold per-collection write locks for seconds and
+    // starve concurrent writers (the Zoe incident). Each tick now evicts
+    // at most a bounded handful of chunks and yields between them so
+    // in-flight writes have time to land.
+    while (running_.load(std::memory_order_acquire)) {
         std::this_thread::sleep_for(
             std::chrono::milliseconds(config_.evictionCheckIntervalMs)
         );
+        if (!running_.load(std::memory_order_acquire)) break;
 
-        if (!running_.load()) break;
+        evictionCheckCount_++;
 
-        checkCount++;
+        // Log memory status every 12 checks (~60s with default 5s interval)
+        if (evictionCheckCount_ % 12 == 1) {
+            logMemoryCheck();
+        }
 
-        // Check memory usage
-        auto stats = getStats();
-        uint64_t threshold = config_.maxMemoryBytes * config_.evictionThresholdPercent / 100;
+        MemoryPressure p = pressure();
+        if (p == MemoryPressure::Normal) {
+            continue;  // nothing to do
+        }
 
-        // Log memory status every 12 checks (~60 seconds with 5s interval)
-        if (checkCount % 12 == 1) {
-            spdlog::info("Memory check: {} MB estimated ({}%, pressure={}), {} docs, {} evicted",
-                        stats.estimatedMemoryBytes / (1024 * 1024),
-                        pressurePercent(),
-                        memoryPressureToString(pressure()),
-                        stats.totalDocuments,
-                        stats.evictedCount);
+        // Pick how aggressive we're willing to be this tick. SOFT trickles
+        // one chunk at a time so steady-state background noise doesn't
+        // thrash; HARD uses half the cap; EMERGENCY uses the full cap.
+        uint32_t maxPassesThisTick;
+        if (p == MemoryPressure::Emergency) {
+            maxPassesThisTick = config_.maxEvictionPassesPerTrigger;
+        } else if (p == MemoryPressure::Hard) {
+            maxPassesThisTick = std::max<uint32_t>(
+                1, config_.maxEvictionPassesPerTrigger / 2);
+        } else {  // Soft
+            maxPassesThisTick = 1;
         }
 
-        if (stats.estimatedMemoryBytes > threshold) {
-            uint64_t target = config_.maxMemoryBytes * config_.evictionTargetPercent / 100;
-            uint64_t bytesToFree = stats.estimatedMemoryBytes - target;
+        uint64_t targetBytes = static_cast<uint64_t>(config_.maxMemoryBytes)
+                             * config_.evictionTargetPercent / 100;
+
+        uint64_t totalFreed = 0;
+        uint64_t totalDocs = 0;
+        uint32_t passesRun = 0;
+        for (uint32_t pass = 0; pass < maxPassesThisTick; ++pass) {
+            if (!running_.load(std::memory_order_acquire)) break;
+
+            uint64_t current = estimatedMemoryBytes_.load(std::memory_order_relaxed);
+            if (current <= targetBytes) break;
 
-            spdlog::info("Memory at {} MB ({}%), starting eviction to free {} MB",
-                        stats.estimatedMemoryBytes / (1024 * 1024),
-                        stats.estimatedMemoryBytes * 100 / config_.maxMemoryBytes,
-                        bytesToFree / (1024 * 1024));
+            uint64_t freed = evictOneChunk();
+            passesRun++;
+            if (freed == 0) {
+                // Nothing evictable (all pinned / hot-write / empty). Don't
+                // spin — wait for the next tick.
+                spdlog::warn("Eviction: nothing evictable this pass, backing off (pressure={})",
+                            memoryPressureToString(p));
+                break;
+            }
+            totalFreed += freed;
 
-            uint64_t evicted = evictDocuments();
+            // Sleep between chunks (but not after the last chunk of the tick)
+            if (pass + 1 < maxPassesThisTick) {
+                std::this_thread::sleep_for(
+                    std::chrono::milliseconds(config_.evictionChunkPauseMs));
+            }
+        }
 
-            spdlog::info("Eviction complete: {} documents evicted", evicted);
+        if (passesRun > 0) {
+            spdlog::info("Eviction tick complete: {} passes, {} MB freed (pressure={})",
+                        passesRun, totalFreed / (1024 * 1024),
+                        memoryPressureToString(pressure()));
+            (void)totalDocs;  // chunk log lines carry the per-pass doc count
         }
     }
 }
 
+uint64_t MemoryStore::evictOneChunk() {
+    // Phase 1 — snapshot candidates under shared global + per-coll locks.
+    // Apply hot-write floor & pinned filters here so Phase 2's decision set
+    // is already clean.
+    struct Candidate {
+        std::string collection;
+        std::string id;
+        uint64_t lastAccessedAt;
+        uint64_t estimatedSize;
+        MemoryPriority priority;
+    };
+    std::vector<Candidate> candidates;
+
+    const uint64_t now = currentTimeMs();
+    // hotWriteFloorMs may exceed `now` during the first seconds of a run; cap
+    // the subtraction so we don't underflow to a huge uint64_t.
+    const uint64_t hotWriteFloor = (config_.hotWriteFloorMs < now)
+        ? (now - config_.hotWriteFloorMs)
+        : 0;
+
+    {
+        std::shared_lock<std::shared_mutex> glock(globalMutex_);
+        for (const auto& [collName, collPtr] : collections_) {
+            if (collPtr->options.pinned) continue;  // never evict pinned collections
+
+            std::shared_lock<std::shared_mutex> clock(collPtr->mutex);
+            candidates.reserve(candidates.size() + collPtr->documents.size());
+            for (const auto& [docId, doc] : collPtr->documents) {
+                // Hot-write protection: a doc updated within the floor window
+                // is probably still being worked on by a writer. Skip it so
+                // we don't race with a patch()/update() in flight.
+                if (doc.updatedAt > hotWriteFloor) continue;
+
+                Candidate c;
+                c.collection = collName;
+                c.id = docId;
+                c.lastAccessedAt = doc.lastAccessedAt > 0 ? doc.lastAccessedAt
+                                 : (doc.updatedAt > 0 ? doc.updatedAt : doc.createdAt);
+                c.estimatedSize = estimateDocumentSize(doc);
+                c.priority = collPtr->options.memoryPriority;
+                candidates.push_back(std::move(c));
+            }
+        }
+    }
+
+    if (candidates.empty()) return 0;
+
+    // Phase 2 — rank. Evict Low-priority first, then Normal, then High;
+    // within a priority tier evict coldest (smallest lastAccessedAt) first.
+    auto priorityWeight = [](MemoryPriority pr) -> uint32_t {
+        switch (pr) {
+            case MemoryPriority::Low:    return 0;
+            case MemoryPriority::Normal: return 1;
+            case MemoryPriority::High:   return 2;
+        }
+        return 1;
+    };
+    std::sort(candidates.begin(), candidates.end(),
+              [&](const Candidate& a, const Candidate& b) {
+        uint32_t wa = priorityWeight(a.priority);
+        uint32_t wb = priorityWeight(b.priority);
+        if (wa != wb) return wa < wb;
+        return a.lastAccessedAt < b.lastAccessedAt;
+    });
+
+    // Per-collection budget: no single collection may contribute more than
+    // (share * 1.5) docs to this chunk, so a 90%-of-RAM collection can't
+    // monopolise the chunk and starve others. Always allow at least 1 per
+    // collection so small collections still drain over multiple ticks.
+    std::unordered_map<std::string, uint32_t> collectionTotals;
+    for (const auto& c : candidates) collectionTotals[c.collection]++;
+
+    const uint64_t totalCandidates = candidates.size();
+    std::unordered_map<std::string, uint32_t> collectionCap;
+    for (const auto& [coll, total] : collectionTotals) {
+        uint64_t cap = (uint64_t)config_.evictionChunkSize * total * 3
+                     / (2 * totalCandidates);
+        if (cap < 1) cap = 1;
+        collectionCap[coll] = static_cast<uint32_t>(cap);
+    }
+
+    // Phase 3 — take up to chunkSize, respecting the per-collection cap.
+    std::unordered_map<std::string, uint32_t> collectionCounts;
+    std::vector<std::pair<std::string, std::string>> toEvict;
+    toEvict.reserve(config_.evictionChunkSize);
+    for (const auto& c : candidates) {
+        if (toEvict.size() >= config_.evictionChunkSize) break;
+        uint32_t& count = collectionCounts[c.collection];
+        if (count >= collectionCap[c.collection]) continue;
+        toEvict.emplace_back(c.collection, c.id);
+        count++;
+    }
+
+    if (toEvict.empty()) return 0;
+
+    // Phase 4 — actually evict. Reuse evictDocument() which handles the
+    // stub creation, expiration-index cleanup, stats, and atomic memory
+    // counter update. It takes the per-collection write lock for the
+    // duration of a single doc — small, per-doc windows keep latency
+    // impact on concurrent writers bounded (the whole point of chunking).
+    uint64_t bytesFreedBefore = estimatedMemoryBytes_.load(std::memory_order_relaxed);
+    uint64_t evictedCount = 0;
+    for (const auto& [coll, id] : toEvict) {
+        uint64_t walSequence = 0;
+        {
+            std::shared_lock<std::shared_mutex> glock(globalMutex_);
+            auto it = collections_.find(coll);
+            if (it == collections_.end()) continue;
+            std::shared_lock<std::shared_mutex> clock(it->second->mutex);
+            auto dIt = it->second->documents.find(id);
+            if (dIt == it->second->documents.end()) continue;
+            walSequence = dIt->second.version;
+        }
+        if (evictDocument(coll, id, walSequence)) {
+            evictedCount++;
+        }
+    }
+
+    uint64_t bytesFreedAfter = estimatedMemoryBytes_.load(std::memory_order_relaxed);
+    uint64_t actuallyFreed = (bytesFreedBefore > bytesFreedAfter)
+        ? (bytesFreedBefore - bytesFreedAfter) : 0;
+
+    spdlog::info("Eviction chunk: {} docs evicted, {} bytes freed, pressure={}",
+                 evictedCount, actuallyFreed, memoryPressureToString(pressure()));
+
+    return actuallyFreed;
+}
+
 uint64_t MemoryStore::evictDocuments() {
     auto stats = getStats();
     uint64_t target = config_.maxMemoryBytes * config_.evictionTargetPercent / 100;

service/src/memory_store.hpp

@@ -664,6 +664,27 @@ private:
     std::vector<std::pair<std::string, std::string>> selectDocumentsForEviction(uint64_t bytesToFree);
     bool evictDocument(const std::string& collection, const std::string& id, uint64_t walSequence);
     std::optional<Document> loadEvictedDocument(const std::string& collection, const std::string& id);
+
+    /**
+     * Evict up to evictionChunkSize documents in a single pass, respecting
+     * pinned collections, the hot-write floor, per-collection budget, and
+     * memoryPriority sort order. Returns bytes freed (0 when no evictable
+     * candidates exist — caller should back off).
+     *
+     * Introduced in v1.7.0 T4 as the core fix for the Zoe incident
+     * (137k-doc one-shot eviction starving concurrent writers).
+     */
+    uint64_t evictOneChunk();
+
+    /**
+     * Emit the periodic "Memory check" observability log line.
+     * Extracted so both the chunked eviction loop and future callers can
+     * reuse the same formatting.
+     */
+    void logMemoryCheck();
+
+    // Counter for logMemoryCheck() throttling (every 12th tick ~60s).
+    uint64_t evictionCheckCount_{0};
 };
 
 } // namespace smartbotic::database