feat(umbp): byte-sized eviction rounds; demote asynchronously instead of inside EvictKey

src/umbp/distributed/master/evict_strategy.cpp

@@ -26,7 +26,7 @@
 
 namespace mori::umbp {
 
-std::unordered_map<std::string, std::vector<std::string>> LruMasterEvictStrategy::SelectVictims(
+std::unordered_map<std::string, std::vector<EvictionVictim>> LruMasterEvictStrategy::SelectVictims(
     std::vector<EvictionCandidate> candidates,
     std::unordered_map<std::string, std::map<TierType, int64_t>> bytes_to_free) {
   // Oldest-access first (LRU); no tiebreak (peers don't ship depth in KvEvent).
@@ -36,16 +36,17 @@ std::unordered_map<std::string, std::vector<std::string>> LruMasterEvictStrategy
             });
 
   // Walk oldest-first, taking a candidate while its (node, tier) has budget
-  // left.  Group by node so each peer gets a single EvictKey keys[].
-  std::unordered_map<std::string, std::vector<std::string>> per_node_keys;
+  // left.  Group by node so each peer gets a single EvictKey.  The victim keeps
+  // the tier it was charged to: that is the medium the peer must free.
+  std::unordered_map<std::string, std::vector<EvictionVictim>> per_node_victims;
   for (const auto& c : candidates) {
     auto& tier_budget = bytes_to_free[c.location.node_id];
     auto it = tier_budget.find(c.location.tier);
     if (it == tier_budget.end() || it->second <= 0) continue;
-    per_node_keys[c.location.node_id].push_back(c.key);
+    per_node_victims[c.location.node_id].push_back(EvictionVictim{c.key, c.location.tier});
     it->second -= static_cast<int64_t>(c.size);
   }
-  return per_node_keys;
+  return per_node_victims;
 }
 
 }  // namespace mori::umbp

src/umbp/distributed/master/eviction_manager.cpp

@@ -21,6 +21,7 @@
 // SOFTWARE.
 #include "umbp/distributed/master/eviction_manager.h"
 
+#include <algorithm>
 #include <chrono>
 #include <cstdint>
 #include <iterator>
@@ -53,8 +54,10 @@ void EvictionManager::Start() {
   if (running_.load(std::memory_order_relaxed)) return;
   running_.store(true, std::memory_order_relaxed);
   thread_ = std::thread(&EvictionManager::EvictionLoop, this);
-  MORI_UMBP_INFO("[EvictionManager] Started (interval={}s, high={}, low={})",
-                 config_.check_interval.count(), config_.high_watermark, config_.low_watermark);
+  MORI_UMBP_INFO(
+      "[EvictionManager] Started (interval={}s, high={}, low={}, max_bytes_per_round={})",
+      config_.check_interval.count(), config_.high_watermark, config_.low_watermark,
+      config_.max_evict_bytes_per_round);
 }
 
 void EvictionManager::Stop() {
@@ -118,28 +121,45 @@ void EvictionManager::RunOnce() {
   }
 
   if (bytes_to_free.empty()) return;
-  MORI_UMBP_INFO("[EvictionManager] {} overloaded node-tiers detected", bytes_to_free.size());
+
+  // A round's size is a byte budget: the overage itself, optionally capped by
+  // max_evict_bytes_per_round.  It used to be a row count (32 candidates per
+  // bucket), which freed a few MB per round however far over the tier was --
+  // a 512 GiB tier's 0.9 -> 0.7 band took hours to drain -- while every round
+  // still paid the store's full scan.
+  if (config_.max_evict_bytes_per_round > 0) {
+    for (auto& [ntk, bytes] : bytes_to_free) {
+      bytes = std::min(bytes, config_.max_evict_bytes_per_round);
+    }
+  }
+  uint64_t budget_total = 0;
+  for (const auto& [ntk, bytes] : bytes_to_free) budget_total += bytes;
+  MORI_UMBP_INFO("[EvictionManager] {} overloaded node-tiers detected, {} bytes to free",
+                 bytes_to_free.size(), budget_total);
 
   // Ask the store for eviction-eligible candidates in the overloaded buckets.
   // The store is policy-neutral: it returns rows ordered by the hint (LRU here,
   // so an indexed backend can push the ordering down) but makes no eviction
-  // decision and never sees the byte budget. Limit each bucket to the configured
-  // round size; a still-overloaded bucket is revisited on the next interval.
+  // decision and never sees the byte budget -- the strategy trims to it.  No
+  // row cap: both stores read every eligible row of a bucket before capping,
+  // so a cap would only discard candidates the budget may need.
   std::vector<NodeTierKey> buckets;
   buckets.reserve(bytes_to_free.size());
   for (const auto& [ntk, _bytes] : bytes_to_free) buckets.push_back(ntk);
 
-  auto candidates_by_bucket = store_.EnumerateEvictionCandidates(
-      buckets, EvictionOrder::kLeastRecentlyAccessed, config_.evict_batch_size,
-      std::chrono::system_clock::now());
+  auto candidates_by_bucket =
+      store_.EnumerateEvictionCandidates(buckets, EvictionOrder::kLeastRecentlyAccessed,
+                                         /*max_per_bucket=*/0, std::chrono::system_clock::now());
   if (candidates_by_bucket.empty()) {
     MORI_UMBP_DEBUG("[EvictionManager] No eviction candidates found");
     return;
   }
 
   // Flatten the candidates and translate the per-(node, tier) byte budget into
   // the strategy's shape, then let the strategy pick victims (default LRU),
-  // grouped by node so each peer gets a single EvictKey keys[].
+  // grouped by node so each peer gets a single EvictKey.  Each victim keeps
+  // the tier it was charged to, so the peer frees exactly the copy that is
+  // over budget.
   std::vector<EvictionCandidate> candidates;
   for (auto& [bucket, rows] : candidates_by_bucket) {
     for (auto& c : rows) candidates.push_back(std::move(c));
@@ -149,32 +169,33 @@ void EvictionManager::RunOnce() {
     strategy_budget[ntk.node_id][ntk.tier] = static_cast<int64_t>(bytes);
   }
 
-  auto per_node_keys = strategy_->SelectVictims(std::move(candidates), std::move(strategy_budget));
+  auto per_node_victims =
+      strategy_->SelectVictims(std::move(candidates), std::move(strategy_budget));
 
   size_t selected = 0;
-  for (const auto& [node_id, keys] : per_node_keys) selected += keys.size();
+  for (const auto& [node_id, victims] : per_node_victims) selected += victims.size();
 
   if (selected == 0) return;
 
   MORI_UMBP_INFO("[EvictionManager] Selected {} victims across {} nodes", selected,
-                 per_node_keys.size());
+                 per_node_victims.size());
 
   // Look up peer addresses once per dispatch round.  ClientRegistry
   // owns the (node_id -> peer_address) mapping; we can't ship an
   // EvictKey to a node that has dropped out.
   std::unordered_map<std::string, std::string> node_to_peer;
   for (const auto& client : clients) node_to_peer[client.node_id] = client.peer_address;
 
-  for (auto& [node_id, keys] : per_node_keys) {
+  for (auto& [node_id, victims] : per_node_victims) {
     auto it = node_to_peer.find(node_id);
     if (it == node_to_peer.end() || it->second.empty()) {
       MORI_UMBP_WARN("[EvictionManager] no peer_address for node={} — skipping {} keys", node_id,
-                     keys.size());
+                     victims.size());
       continue;
     }
     if (dispatcher_ == nullptr) {
       MORI_UMBP_DEBUG("[EvictionManager] dispatcher unset; would EvictKey on node={} ({} keys)",
-                      node_id, keys.size());
+                      node_id, victims.size());
       continue;
     }
     // Chunked to the shared gRPC message limit.  An eviction round frees a
@@ -189,19 +210,27 @@ void EvictionManager::RunOnce() {
     // next heartbeat are what shrink the index), and a chunk that fails is
     // retried next round.  So N chunks reach the same state as one message
     // would have, and a partial round still makes progress.
+    //
+    // Sized on the keys alone: the parallel tier field is a packed enum of one
+    // byte per key, inside the per-item overhead GrpcMaxItemsPerBatch already
+    // reserves.  Nothing else bounds a chunk: the peer queues any demotion and
+    // returns, so a call's length does not grow with the bytes it names.
+    std::vector<std::string> keys;
+    keys.reserve(victims.size());
+    for (const auto& victim : victims) keys.push_back(victim.key);
     size_t sent = 0;
     size_t chunks = 0;
-    while (sent < keys.size()) {
+    while (sent < victims.size()) {
       const size_t take = GrpcMaxItemsPerBatch(keys, sent);
-      std::vector<std::string> chunk(std::make_move_iterator(keys.begin() + sent),
-                                     std::make_move_iterator(keys.begin() + sent + take));
+      std::vector<EvictionVictim> chunk(std::make_move_iterator(victims.begin() + sent),
+                                        std::make_move_iterator(victims.begin() + sent + take));
       sent += take;
       ++chunks;
       dispatcher_->DispatchEvictKey(node_id, it->second, std::move(chunk));
     }
     if (chunks > 1) {
       MORI_UMBP_DEBUG("[EvictionManager] node={} {} keys dispatched in {} chunks", node_id,
-                      keys.size(), chunks);
+                      victims.size(), chunks);
     }
   }
 }

src/umbp/distributed/master/master_server.cpp

@@ -128,14 +128,21 @@ int EvictKeyDeadlineMs() {
 class MasterPeerStubPool : public EvictKeyDispatcher {
  public:
   void DispatchEvictKey(const std::string& node_id, const std::string& peer_address,
-                        std::vector<std::string> keys) override {
-    if (keys.empty() || peer_address.empty()) return;
+                        std::vector<EvictionVictim> victims) override {
+    if (victims.empty() || peer_address.empty()) return;
 
     auto stub = GetOrCreateStub(node_id, peer_address);
     if (stub == nullptr) return;
 
+    // keys and tiers are parallel, so a peer that predates `tiers` still reads
+    // a complete key list.
     ::umbp::EvictKeyRequest req;
-    for (auto& k : keys) req.add_keys(std::move(k));
+    req.mutable_keys()->Reserve(static_cast<int>(victims.size()));
+    req.mutable_tiers()->Reserve(static_cast<int>(victims.size()));
+    for (auto& victim : victims) {
+      req.add_keys(std::move(victim.key));
+      req.add_tiers(static_cast<::umbp::TierType>(victim.tier));
+    }
     ::umbp::EvictKeyResponse resp;
     grpc::ClientContext ctx;
     ctx.set_deadline(std::chrono::system_clock::now() +