fix(umbp): retry eviction demotions instead of dropping the only copy; count what is dropped

src/umbp/distributed/metrics/activity_summary.cpp

@@ -133,6 +133,8 @@ std::vector<std::string> ActivitySummary::Render(const ActivitySnapshot& now,
       Delta(now.transitions.offloaded_bytes, before.transitions.offloaded_bytes);
   const uint64_t moves_ok = Delta(now.transitions.succeeded, before.transitions.succeeded);
   const uint64_t moves_failed = Delta(now.transitions.failed, before.transitions.failed);
+  const uint64_t moves_retried = Delta(now.transitions.retried, before.transitions.retried);
+  const uint64_t moves_dropped = Delta(now.transitions.dropped, before.transitions.dropped);
 
   const ReCacheActivity& rc_now = now.recache;
   const ReCacheActivity& rc_old = before.recache;
@@ -157,8 +159,9 @@ std::vector<std::string> ActivitySummary::Render(const ActivitySnapshot& now,
   std::snprintf(window.data(), window.size(), "%.1fs", window_seconds);
 
   const bool anything = Moved(total) || promoted != 0 || demoted != 0 || moves_failed != 0 ||
-                        rc_admitted != 0 || rc_dropped != 0 || rc_prefetched != 0 ||
-                        rc_fragmented != 0 || rc_already != 0;
+                        moves_retried != 0 || moves_dropped != 0 || rc_admitted != 0 ||
+                        rc_dropped != 0 || rc_prefetched != 0 || rc_fragmented != 0 ||
+                        rc_already != 0;
 
   std::vector<std::string> lines;
 
@@ -205,11 +208,14 @@ std::vector<std::string> ActivitySummary::Render(const ActivitySnapshot& now,
   // they are printed as a pair even when one of them is zero: "promote 0 B"
   // next to a large demote is the shape of a tier graph that only sheds.
   head += " · promote " + FormatBytes(promoted) + " · demote " + FormatBytes(demoted);
-  if (moves_ok != 0 || moves_failed != 0) {
-    std::array<char, 80> moves{};
-    std::snprintf(moves.data(), moves.size(), " (%llu moved, %llu failed)",
-                  static_cast<unsigned long long>(moves_ok),
-                  static_cast<unsigned long long>(moves_failed));
+  if (moves_ok != 0 || moves_failed != 0 || moves_retried != 0 || moves_dropped != 0) {
+    // dropped = evicted copies lost for good; retried = demotions re-queued.
+    std::array<char, 128> moves{};
+    std::snprintf(
+        moves.data(), moves.size(), " (%llu moved, %llu failed, %llu retried, %llu dropped)",
+        static_cast<unsigned long long>(moves_ok), static_cast<unsigned long long>(moves_failed),
+        static_cast<unsigned long long>(moves_retried),
+        static_cast<unsigned long long>(moves_dropped));
     head += moves.data();
   }
   lines.push_back(std::move(head));

src/umbp/distributed/pool/peer_pool.cpp

@@ -34,9 +34,23 @@
 #include <tuple>
 #include <utility>
 
+#include "umbp/common/env_time.h"
+
 namespace mori::umbp {
+
 namespace {
 
+// How long a retryable eviction demotion keeps retrying before its copy is
+// dropped.  Time, not a count: the faults it rides out last a wall-clock
+// interval, and a short queue would burn a count in milliseconds.
+std::chrono::milliseconds EvictDemotionRetryBudget() {
+  static const auto v = GetEnvMilliseconds("UMBP_EVICT_DEMOTION_RETRY_MS",
+                                           std::chrono::milliseconds(30000), /*min_allowed=*/0);
+  return v;
+}
+// Pause before a lone retry, so waiting out pressure does not spin the worker.
+constexpr std::chrono::milliseconds kEvictRetryIdlePause{20};
+
 // One watermark scan queues at most this many keys; the worker tops the queue
 // up after each completed offload, so a tier still drains all the way down to
 // its low watermark without ever queueing its whole key set at once.
@@ -90,6 +104,7 @@ PeerPool::PeerPool(BackendRegistry* backends, std::unique_ptr<PoolPolicy> policy
         break;
       }
     }
+    evict_demotion_retry_budget_ = EvictDemotionRetryBudget();
     transition_worker_ = std::thread(&PeerPool::TransitionWorkerLoop, this);
   }
 }
@@ -117,7 +132,8 @@ void PeerPool::DropEvictedCopyLocked(const TransitionJob& job) {
   if (migrating_keys_.count(job.key) != 0) return;
   auto* source = backends_ == nullptr ? nullptr : backends_->Get(job.source_backend_id);
   if (source == nullptr) return;
-  source->Evict({job.key});
+  const auto evicted = source->Evict({job.key});
+  if (!evicted.empty() && evicted.front().bytes_freed > 0) ++transition_metrics_.dropped;
   RepairAfterDropLocked(job.key);
 }
 
@@ -324,17 +340,27 @@ void PeerPool::TransitionWorkerLoop() {
       transition_queue_.pop_front();
     }
     bool retry = false;
+    bool evict_retry = false;
     {
       std::unique_lock<std::mutex> lock(operation_mutex_);
       if (tier_graph_ != nullptr) {
         const auto& tier = tier_graph_->NodeAt(job.source_tier);
         const bool watermark_driven = WatermarkDrivenLocked(job);
         if (!watermark_driven || tier_graph_->AtOrAbove(job.source_tier, tier.low_watermark)) {
+          if (job.evict && job.first_attempt == std::chrono::steady_clock::time_point{}) {
+            job.first_attempt = std::chrono::steady_clock::now();
+          }
           const auto result = RunTransition(lock, job);
-          if (job.evict && !result.success) DropEvictedCopyLocked(job);
+          // The evicted copy is the only one: retry what can clear, and drop it
+          // only once the budget is spent or the target can never take it.
+          evict_retry =
+              job.evict && !result.success && result.retryable &&
+              std::chrono::steady_clock::now() - job.first_attempt < evict_demotion_retry_budget_;
+          if (evict_retry) ++transition_metrics_.retried;
+          if (job.evict && !result.success && !evict_retry) DropEvictedCopyLocked(job);
           const bool still_pressured =
               watermark_driven && tier_graph_->AtOrAbove(job.source_tier, tier.low_watermark);
-          retry = still_pressured && !result.success && job.attempts < 3;
+          retry = (still_pressured && !result.success && job.attempts < 3) || evict_retry;
           // Keep the drain going without waiting for the next commit batch to
           // notice the tier is still above its low watermark. The scan skips
           // whatever is still queued, so this tops the batch back up rather
@@ -346,15 +372,22 @@ void PeerPool::TransitionWorkerLoop() {
         }
       }
     }
-    if (retry) {
+    if (evict_retry) {
+      bool idle = false;
+      {
+        std::lock_guard<std::mutex> lock(transition_mutex_);
+        idle = transition_queue_.empty();
+      }
+      if (idle) std::this_thread::sleep_for(kEvictRetryIdlePause);
+    } else if (retry) {
       std::this_thread::sleep_for(std::chrono::milliseconds(uint64_t{1} << job.attempts));
     }
     {
       std::lock_guard<std::mutex> lock(transition_mutex_);
       queued_transition_keys_.erase(job.key);
       const size_t source_tier = job.source_tier;
       if (retry && !stop_transition_worker_) {
-        ++job.attempts;
+        if (job.attempts < UINT8_MAX) ++job.attempts;
         queued_transition_keys_.insert(job.key);
         transition_queue_.push_back(std::move(job));
         transition_cv_.notify_one();

src/umbp/distributed/pool/tier_transition.cpp

@@ -99,25 +99,32 @@ TierTransitionResult TierTransitionExecutor::Execute(
       if (!BuildTransfers(source, resolved, target, allocation, &transfers) ||
           !transfer_engine_->Transfer(transfers, nullptr)) {
         target->BatchAbort({allocation.slot_id});
+        result.retryable = true;
         continue;
       }
       auto committed = target->BatchCommit({CommitRequest{allocation.slot_id, key}});
       if (committed.empty() || !committed.front().success) {
         target->BatchAbort({allocation.slot_id});
+        result.retryable = true;
         continue;
       }
     } else if (allocation.outcome != AllocateOutcome::kSuccessAlreadyExists) {
+      // No space is pressure (staging full of reads, eviction catching up);
+      // a shape failure (kFailed) stays final.
+      if (allocation.outcome == AllocateOutcome::kFailedNoSpace) result.retryable = true;
       continue;
     }
 
     ResolvedEntry target_read;
     if (!target->AcquireMigrationRead(key, &target_read) || target_read.size != resolved.size) {
       if (target_read.found) target->ReleaseMigrationRead(key);
       if (allocation.outcome == AllocateOutcome::kSuccessAllocated) target->Evict({key});
+      result.retryable = true;
       continue;
     }
 
     result.success = true;
+    result.retryable = false;
     result.target_backend_id = target_backend_id;
     result.bytes_moved = resolved.size;
     source->ReleaseMigrationRead(key);

src/umbp/distributed/pool_client.cpp

@@ -5171,6 +5171,10 @@ void PoolClient::PublishComponentMetrics() {
            {{"outcome", "succeeded"}}, tiers.succeeded);
     sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP,
            {{"outcome", "failed"}}, tiers.failed);
+    sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP,
+           {{"outcome", "dropped"}}, tiers.dropped);
+    sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP,
+           {{"outcome", "retried"}}, tiers.retried);
     sample(MORI_UMBP_METRIC_CLIENT_TIER_OFFLOADED_BYTES,
            MORI_UMBP_METRIC_CLIENT_TIER_OFFLOADED_BYTES_HELP, {}, tiers.offloaded_bytes);
     sample(MORI_UMBP_METRIC_CLIENT_TIER_PROMOTED_BYTES,

src/umbp/doc/runtime-env-vars.md

@@ -61,6 +61,7 @@ Read by the **master process** (`bin/master_main.cpp` via
 | `UMBP_HEARTBEAT_INTERVAL_DIVISOR` | `2` | count | Recommended client heartbeat interval = `heartbeat_ttl / divisor`. `min_allowed=1` guards against div-by-zero. Read by the master and echoed in `RegisterClientResponse.heartbeat_interval_ms`. |
 | `UMBP_EVICTKEY_DEADLINE_MS` | `1000` | ms | Per-call gRPC deadline applied to outbound `EvictKey` RPCs from `MasterPeerStubPool`. |
 | `UMBP_EVICT_MAX_BYTES_PER_ROUND` | `0` | bytes | Most bytes one eviction round frees from one `(node, tier)`. `0` = its whole overage above the low watermark. Set it to throttle eviction, e.g. how much a demoting tier copies downstream per round. |
+| `UMBP_EVICT_DEMOTION_RETRY_MS` | `30000` | ms | How long a peer retries an eviction demotion that failed in a way that can clear (no space at the target, e.g. an SSD staging arena full of reads, or a copy/commit error) before it drops the evicted copy so the tier still frees. Counted from the first attempt. `0` = drop on the first such failure. |
 | `UMBP_HIT_INDEX_TTL_SEC` | `7200` | sec | External KV hit-count entry TTL. A hash with no counted match for longer than this is removed from the hit index. |
 | `UMBP_HIT_INDEX_GC_INTERVAL_SEC` | `60` | sec | External KV hit-count GC sweep interval. |
 | `UMBP_HIT_QUERY_MAX_BATCH` | `4096` | count | Maximum hashes accepted by one `GetExternalKvHitCounts` request. Oversized requests return gRPC `INVALID_ARGUMENT`; the server does not truncate. |

src/umbp/include/umbp/distributed/pool/peer_pool.h

@@ -156,6 +156,10 @@ class PeerPool {
   LocalEvictionMetrics LocalEvictionStats() const;
   // Test seam: one synchronous pass.  Returns keys dropped or queued.
   size_t RunLocalEvictionOnceForTest() { return LocalEvictionPass(); }
+  // UMBP_EVICT_DEMOTION_RETRY_MS override; set it before queueing work.
+  void SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds budget) {
+    evict_demotion_retry_budget_ = budget;
+  }
 
   // Victims for a backend over its watermark, coldest first, before the
   // medium's own order: (tier, logical tier, bytes) -> offers, or nullopt when
@@ -177,7 +181,9 @@ class PeerPool {
     uint32_t source_backend_id = BackendRegistry::kMaxBackends;
     LogicalTierGraph::TierIndex source_tier = 0;
     uint8_t attempts = 0;
-    bool evict = false;  // from Evict(): ignores watermarks; drops on failure
+    // An eviction demotion's retries are bounded by time from its first attempt.
+    std::chrono::steady_clock::time_point first_attempt{};
+    bool evict = false;  // from Evict(): ignores watermarks; drops once retries run out
   };
 
   // What the copy phase needs, resolved while the pool lock is still held.
@@ -296,6 +302,7 @@ class PeerPool {
   std::vector<size_t> queued_by_tier_;
   bool stop_transition_worker_ = false;
   std::thread transition_worker_;
+  std::chrono::milliseconds evict_demotion_retry_budget_{30000};  // transition worker only
 
   // Local watermark eviction.  local_evict_mutex_ guards local_evict_,
   // stop_local_evict_ and the worker's start.

src/umbp/include/umbp/distributed/pool/tier_transition.h

@@ -43,6 +43,9 @@ struct TierTransitionResult {
   uint64_t bytes_moved = 0;
   uint64_t bytes_freed = 0;
   bool source_draining = false;
+  // Failed in a way that can clear (no space, or a copy/commit/read-back
+  // error), so a later attempt may succeed.  False for a shape failure.
+  bool retryable = false;
 };
 
 // Executes one peer-local copy or move. Placement policy and scheduling remain

src/umbp/include/umbp/distributed/types.h

@@ -80,6 +80,8 @@ struct TierTransitionMetrics {
   uint64_t attempted = 0;
   uint64_t succeeded = 0;
   uint64_t failed = 0;
+  uint64_t dropped = 0;  // evicted copies whose demotion never landed (lost keys)
+  uint64_t retried = 0;  // eviction demotions re-queued after a retryable failure
   uint64_t offloaded_bytes = 0;
   uint64_t promoted_bytes = 0;
 };

tests/cpp/umbp/distributed/test_activity_summary.cpp

@@ -135,10 +135,12 @@ TEST(ActivitySummary, ReportsPromotionAndDemotionAsAPairEvenWhenOneIsZero) {
   second.transitions.offloaded_bytes = 8ULL * 1024 * 1024;
   second.transitions.succeeded = 4;
   second.transitions.failed = 1;
+  second.transitions.retried = 3;
+  second.transitions.dropped = 2;
   const std::string all = Joined(summary.Render(second, 60.0));
   EXPECT_NE(all.find("promote 0 B"), std::string::npos) << all;
   EXPECT_NE(all.find("demote 8.00 MiB"), std::string::npos) << all;
-  EXPECT_NE(all.find("1 failed"), std::string::npos) << all;
+  EXPECT_NE(all.find("1 failed, 3 retried, 2 dropped"), std::string::npos) << all;
 }
 
 TEST(ActivitySummary, SurfacesLoadMissesWhichBytesAloneHide) {

tests/cpp/umbp/distributed/test_peer_pool.cpp

@@ -176,6 +176,37 @@ class NoSpaceBackend final : public MockBackend {
   }
 };
 
+// LocalCopyEngine whose next `failures` copies fail.
+class FailingEngine final : public TransferEngine {
+ public:
+  explicit FailingEngine(int failures) : failures_(failures) {}
+  void FailNext(int n) { failures_.store(n); }
+
+  const char* Name() const override { return "failing"; }
+  TransferRef RegisterMemory(void* base, size_t size, mori::io::MemoryLocationType loc,
+                             int device) override {
+    return inner_.RegisterMemory(base, size, loc, device);
+  }
+  void Deregister(const TransferRef& ref) override { inner_.Deregister(ref); }
+  bool CanHandle(const TransferRef& src, const TransferRef& dst) const override {
+    return inner_.CanHandle(src, dst);
+  }
+  TransferPlanSet Plan(const std::vector<TransferItem>& items) const override {
+    return inner_.Plan(items);
+  }
+  std::unique_ptr<TransferHandle> Submit(std::vector<TransferPlan> plans) override {
+    if (failures_.load() > 0) {
+      failures_.fetch_sub(1);
+      return nullptr;  // the copy never happened
+    }
+    return inner_.Submit(std::move(plans));
+  }
+
+ private:
+  LocalCopyEngine inner_;
+  std::atomic<int> failures_;
+};
+
 class BusyResolveBackend final : public MockBackend {
  public:
   explicit BusyResolveBackend(TierType tier) : MockBackend(tier) {}
@@ -810,6 +841,7 @@ TEST(PeerPool, DemotionWithNoRoomDownstreamDropsTheCopy) {
   auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry);
   ASSERT_TRUE(compiled.ok()) << compiled.error;
   PeerPool pool(&registry, MakeTieredPlacementPolicy(compiled.graph), &engine);
+  pool.SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds{50});
   // Fill cold's only page directly.
   auto* cold = registry.Get("cold");
   auto filler = cold->BatchAllocate({AllocateRequest{"filler", kPageSize}}).front();
@@ -824,6 +856,65 @@ TEST(PeerPool, DemotionWithNoRoomDownstreamDropsTheCopy) {
   EXPECT_FALSE(cold->Contains("k"));
   EXPECT_TRUE(cold->Contains("filler"));
   EXPECT_FALSE(pool.PlacementBackend("k").has_value());
+  EXPECT_EQ(pool.TransitionMetrics().dropped, 1u);
+}
+
+// A demotion copy that fails is retried, since the evicted copy is the only
+// one; it is dropped only once the retry budget runs out, so the tier frees.
+TEST(PeerPool, AFailedDemotionIsRetriedUntilItsBudgetRunsOut) {
+  constexpr uint64_t kPageSize = 4096;
+  FailingEngine engine(/*failures=*/0);
+  BackendRegistry registry;
+  RegisterPageBackend(&registry, &engine, "hot", TierType::DRAM, kPageSize, 4);
+  RegisterPageBackend(&registry, &engine, "cold", TierType::SSD, kPageSize, 4);
+  auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry);
+  ASSERT_TRUE(compiled.ok()) << compiled.error;
+  PeerPool pool(&registry, MakeTieredPlacementPolicy(compiled.graph), &engine);
+  CommitKey(&pool, "a");
+  CommitKey(&pool, "b");
+
+  engine.FailNext(2);  // the first two copies fail; the third lands
+  auto evicted = pool.Evict({PoolEvictRequest{"a", TierType::DRAM}}, PoolEvictMode::kReclaim);
+  EXPECT_EQ(evicted.front().outcome, PoolEvictOutcome::kDemotionQueued);
+  ASSERT_TRUE(WaitForKeyGone(&registry, "hot", "a"));
+  EXPECT_TRUE(registry.Get("cold")->Contains("a")) << "dropped, not demoted";
+  EXPECT_EQ(pool.TransitionMetrics().dropped, 0u);
+  EXPECT_GE(pool.TransitionMetrics().failed, 2u) << "the failed attempts still count";
+  EXPECT_GE(pool.TransitionMetrics().retried, 2u);
+
+  pool.SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds{50});
+  engine.FailNext(1000000);  // every copy fails
+  pool.Evict({PoolEvictRequest{"b", TierType::DRAM}}, PoolEvictMode::kReclaim);
+  ASSERT_TRUE(WaitForKeyGone(&registry, "hot", "b"));
+  EXPECT_FALSE(registry.Get("cold")->Contains("b"));
+  EXPECT_FALSE(pool.PlacementBackend("b").has_value());
+  EXPECT_EQ(pool.TransitionMetrics().dropped, 1u);
+}
+
+// No space at the target (an SSD whose staging is full of reads) is waited
+// out, not a reason to drop the only copy.
+TEST(PeerPool, ADemotionWaitsOutATargetWithNoRoomForNow) {
+  constexpr uint64_t kPageSize = 4096;
+  LocalCopyEngine engine;
+  BackendRegistry registry;
+  RegisterPageBackend(&registry, &engine, "hot", TierType::DRAM, kPageSize, 4);
+  RegisterPageBackend(&registry, &engine, "cold", TierType::SSD, kPageSize, 1);
+  auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry);
+  ASSERT_TRUE(compiled.ok()) << compiled.error;
+  PeerPool pool(&registry, MakeTieredPlacementPolicy(compiled.graph), &engine);
+  auto* cold = registry.Get("cold");
+  auto filler = cold->BatchAllocate({AllocateRequest{"filler", kPageSize}}).front();
+  ASSERT_EQ(filler.outcome, AllocateOutcome::kSuccessAllocated);  // cold's only page, pending
+  CommitKey(&pool, "k");
+
+  pool.Evict({PoolEvictRequest{"k", TierType::DRAM}}, PoolEvictMode::kReclaim);
+  std::this_thread::sleep_for(std::chrono::milliseconds{100});
+  EXPECT_TRUE(registry.Get("hot")->Contains("k")) << "dropped while the target was only busy";
+
+  cold->BatchAbort({filler.slot_id});  // the pressure clears
+  ASSERT_TRUE(WaitForKeyGone(&registry, "hot", "k"));
+  EXPECT_TRUE(cold->Contains("k"));
+  EXPECT_EQ(pool.TransitionMetrics().dropped, 0u);
 }
 
 TEST(PeerPool, WatermarkMigratesCommittedKey) {