[manager] harden EventReport GC lifecycle and accounting

Author: shaohuaxi

docs/design/event_report_background_gc.md

@@ -507,16 +507,38 @@ bool LRUCacheShard::ApplyToEntryNoTouch(
         return false;
     }
     assert(e->InCache());
+    const bool in_lru = !e->HasRefs();
     const ssize_t delta = callback(e->value, e->total_charge, e->helper);
     if (delta > 0) {
-        e->total_charge += static_cast<size_t>(delta);
-        usage_ += static_cast<size_t>(delta);
+        const size_t increase = static_cast<size_t>(delta);
+        e->total_charge += increase;
+        usage_ += increase;
+        if (in_lru) {
+            lru_usage_ += increase;
+            if (e->InHighPriPool()) {
+                high_pri_pool_usage_ += increase;
+            } else if (e->InLowPriPool()) {
+                low_pri_pool_usage_ += increase;
+            }
+            MaintainPoolSize();
+        }
     } else if (delta < 0) {
         size_t decrease = static_cast<size_t>(-delta);
         decrease = std::min(decrease, e->total_charge);
         decrease = std::min(decrease, usage_);
         e->total_charge -= decrease;
         usage_ -= decrease;
+        if (in_lru) {
+            assert(lru_usage_ >= decrease);
+            lru_usage_ -= decrease;
+            if (e->InHighPriPool()) {
+                assert(high_pri_pool_usage_ >= decrease);
+                high_pri_pool_usage_ -= decrease;
+            } else if (e->InLowPriPool()) {
+                assert(low_pri_pool_usage_ >= decrease);
+                low_pri_pool_usage_ -= decrease;
+            }
+        }
     }
     return true;
 }

kv_cache_manager/common/cache/lru_cache.cc

@@ -204,6 +204,33 @@ TEST_F(LRUCacheTest, BatchReleaseFreesEntryErasedWhilePinned) {
     ValidateLRUList({"b"}, 0, 1);
 }
 
+TEST_F(LRUCacheTest, ApplyToEntryNoTouchKeepsUnpinnedPoolChargeConsistent) {
+    NewCache(100, /* high_pri_pool_ratio */ 0.50, /* low_pri_pool_ratio */ 0.50);
+    Insert("a", Cache::Priority::HIGH, 20);
+    ASSERT_EQ(20u, cache_->GetUsage());
+    ASSERT_EQ(0u, cache_->GetPinnedUsage());
+    ValidateLRUList({"a"}, 1, 0, 0);
+
+    ASSERT_TRUE(cache_->ApplyToEntryNoTouch(
+        "a", 0, [](Cache::ObjectPtr, size_t, const Cache::CacheItemHelper *) { return -10; }));
+    EXPECT_EQ(10u, cache_->GetUsage());
+    EXPECT_EQ(0u, cache_->GetPinnedUsage());
+    ValidateLRUList({"a"}, 1, 0, 0);
+
+    ASSERT_TRUE(cache_->ApplyToEntryNoTouch(
+        "a", 0, [](Cache::ObjectPtr, size_t, const Cache::CacheItemHelper *) { return 20; }));
+    EXPECT_EQ(30u, cache_->GetUsage());
+    EXPECT_EQ(0u, cache_->GetPinnedUsage());
+    ValidateLRUList({"a"}, 1, 0, 0);
+
+    // A later insertion exercises MaintainPoolSize and LRU_Remove using the
+    // adjusted charge; stale pool counters would corrupt these transitions.
+    Insert("b", Cache::Priority::HIGH, 30);
+    EXPECT_EQ(60u, cache_->GetUsage());
+    EXPECT_EQ(0u, cache_->GetPinnedUsage());
+    ValidateLRUList({"a", "b"}, 1, 1, 0);
+}
+
 TEST_F(LRUCacheTest, LowPriorityMidpointInsertion) {
     // Allocate 2 cache entries to high-pri pool and 3 to low-pri pool.
     NewCache(5, /* high_pri_pool_ratio */ 0.40, /* low_pri_pool_ratio */ 0.60);

kv_cache_manager/common/test/lru_cache_test.cc

@@ -1250,10 +1250,9 @@ EventReportBackend::CleanupLeaseAcquireResult EventReportBackend::AcquireDownLif
     if (lifecycle_fence->generation != expected_generation) {
         return CleanupLeaseAcquireResult::kStale;
     }
-    // The liveness loop publishes a cleanup intent before performing its
-    // generation-checked unregister so a concurrent heartbeat can still win.
-    // An active reporter in that same generation is therefore transient, not
-    // stale; retry after the liveness callback finishes the unregister.
+    // Producers publish DownHost only after generation-checked unregister.
+    // Treat an unexpected same-generation active state fail-closed as busy:
+    // retain the intent and never authorize deletion until down is observed.
     if (lifecycle_fence->registered) {
         return CleanupLeaseAcquireResult::kBusy;
     }

kv_cache_manager/data_storage/event_report_backend.cc

@@ -399,7 +399,8 @@ bool CacheGarbageCollector::RegisterHostCleanupIntent(const std::string &instanc
 
 void CacheGarbageCollector::CancelHostCleanupIntent(const std::string &instance_id,
                                                     DataStorageType storage_type,
-                                                    const std::string &host_ip_port) noexcept {
+                                                    const std::string &host_ip_port,
+                                                    uint64_t active_lifecycle_generation) noexcept {
     try {
         bool canceled = false;
         EventReportIntentType canceled_type = EventReportIntentType::kDownHost;
@@ -408,7 +409,11 @@ void CacheGarbageCollector::CancelHostCleanupIntent(const std::string &instance_
             const EventReportIntentKey key{instance_id, storage_type, host_ip_port};
             recovery_observations_.erase(key);
             const auto it = event_report_intents_.find(key);
-            if (it != event_report_intents_.end() && it->second.type != EventReportIntentType::kStaleSnapshot) {
+            // Cancellation is authorized only by a strictly newer active
+            // lifecycle. A same-generation notification does not prove that
+            // the captured down lifecycle has been superseded.
+            if (it != event_report_intents_.end() && it->second.type != EventReportIntentType::kStaleSnapshot &&
+                it->second.lifecycle_generation < active_lifecycle_generation) {
                 canceled_type = it->second.type;
                 event_report_intents_.erase(it);
                 canceled = true;
@@ -824,6 +829,14 @@ bool CacheGarbageCollector::RunEventReportTick() {
     for (const auto &[instance_id, _] : event_report_retry_batches_) {
         active_instances.insert(instance_id);
     }
+    // A pass owns its frozen intent snapshot until its cursor reaches base.
+    // Keep scheduling it even if every live intent is canceled mid-pass;
+    // otherwise a later intent would inherit the stale cursor and barrier.
+    for (const auto &[instance_id, state] : event_report_scan_states_) {
+        if (state.context) {
+            active_instances.insert(instance_id);
+        }
+    }
     if (active_instances.empty()) {
         UpdateEventReportMetrics();
         return false;
@@ -936,9 +949,8 @@ bool CacheGarbageCollector::BeginEventReportPass(const std::string &instance_id,
             return false;
         }
         if (intent.type == EventReportIntentType::kDownHost) {
-            // Liveness publishes the intent before its generation-checked unregister. Do not put the SyncAll
-            // barrier ahead of an old-lifecycle mutation that is still holding a shared lease: first prove that
-            // unregister has completed, release this short preflight lease, and only then flush the mutation queues.
+            // Host intents are published after generation-checked unregister. Revalidate the captured down
+            // lifecycle before SyncAll so a concurrent REGISTER cannot let an old intent cross into a new lifecycle.
             EventReportBackend::LifecycleMutationLease preflight_lease;
             const auto lease_result = backend->AcquireDownLifecycleCleanupLease(
                 {key.instance_id, key.host_ip_port}, intent.lifecycle_generation, preflight_lease);
@@ -965,6 +977,11 @@ bool CacheGarbageCollector::BeginEventReportPass(const std::string &instance_id,
         return false;
     }
 
+    auto indexer = meta_indexer_manager_->GetMetaIndexer(instance_id);
+    if (!indexer || !indexer->IsMaintenanceDeleteReady()) {
+        FailEventReportPass(instance_id, state, indexer ? "meta_recovering" : "indexer_missing");
+        return false;
+    }
     MetaSearcher *meta_searcher = meta_searcher_manager_->GetMetaSearcher(instance_id);
     if (!meta_searcher || !meta_searcher->SyncAllForMaintenance()) {
         FailEventReportPass(instance_id, state, meta_searcher ? "sync_all" : "searcher_missing");
@@ -1094,7 +1111,8 @@ std::vector<CacheGarbageCollector::EventReportDeleteTarget> CacheGarbageCollecto
                     if (backend->ParseLocationId(location_id, medium, host)) {
                         const EventReportIntentKey key{instance_id, location->type(), host};
                         if (backend->IsNodeRegistered(instance_id, host)) {
-                            CancelHostCleanupIntent(instance_id, location->type(), host);
+                            CancelHostCleanupIntent(
+                                instance_id, location->type(), host, backend->GetNodeGeneration(instance_id, host));
                         } else {
                             std::lock_guard<std::mutex> lock(event_report_intent_mutex_);
                             recovery_observations_.try_emplace(
@@ -1297,6 +1315,8 @@ bool CacheGarbageCollector::ExecuteEventReportDeleteBatch(const std::string &ins
             static_cast<double>(std::max<int64_t>(0, action_duration_ms));
         metrics_registry_->GetGauge("cache_gc.event_report_last_action_shard_lock_wait_us") =
             static_cast<double>(std::max<int64_t>(0, result.shard_lock_wait_time_us));
+        metrics_registry_->GetGauge("cache_gc.event_report_last_action_shard_lock_hold_us") =
+            static_cast<double>(std::max<int64_t>(0, result.shard_lock_hold_time_us));
     } catch (...) { KVCM_LOG_ERROR("cache gc failed to record event report action metrics"); }
 
     if (!result.sync_succeeded || (result.ec != EC_OK && result.ec != EC_MISMATCH)) {

kv_cache_manager/manager/cache_garbage_collector.cc

@@ -109,7 +109,8 @@ class CacheGarbageCollector {
                                    const std::shared_ptr<EventReportBackend> &backend) noexcept;
     void CancelHostCleanupIntent(const std::string &instance_id,
                                  DataStorageType storage_type,
-                                 const std::string &host_ip_port) noexcept;
+                                 const std::string &host_ip_port,
+                                 uint64_t active_lifecycle_generation) noexcept;
 
 private:
     using Clock = std::chrono::steady_clock;

kv_cache_manager/manager/cache_garbage_collector.h

@@ -412,10 +412,17 @@ CacheManager::CacheManager(std::shared_ptr<MetricsRegistry> metrics_registry,
 
 CacheManager::~CacheManager() {
     if (cache_garbage_collector_) {
-        cache_garbage_collector_->Stop();
-        cache_garbage_collector_.reset();
+        // Close intent admission before detaching callbacks. A liveness thread
+        // may already have copied a callback, so the callback itself also uses
+        // a weak GC reference and will either observe stopped admission or an
+        // expired collector.
+        cache_garbage_collector_->RequestStop();
     }
     ClearEventCleanupCallbacks();
+    if (cache_garbage_collector_) {
+        cache_garbage_collector_->Join();
+        cache_garbage_collector_.reset();
+    }
     StopRecoverRetryLoop();
     DeactivateEventCleanupCallbacks();
     if (write_location_manager_) {
@@ -2584,54 +2591,6 @@ class ValidatedEventLocationSpecs {
     std::vector<ValidatedEventLocationSpec> many_;
 };
 
-bool IsSnapshotLocationStale(const EventReportBackend *event_backend,
-                             const std::string &instance_id,
-                             const CacheLocation &location,
-                             bool preserve_in_flight = false) {
-    if (!event_backend) {
-        return false;
-    }
-
-    std::string medium;
-    std::string reporter_host;
-    if (!event_backend->ParseLocationId(location.id(), medium, reporter_host)) {
-        return false;
-    }
-
-    const ReporterSnapshotKey reporter_key{instance_id, reporter_host};
-    std::string committed_version;
-    std::string in_flight_version;
-    event_backend->GetSnapshotVersionTokens(reporter_key, committed_version, in_flight_version);
-    if (location.location_specs().empty()) {
-        return true;
-    }
-    bool contains_committed = false;
-    bool contains_in_flight = false;
-    for (const auto &spec : location.location_specs()) {
-        const size_t version_param_count =
-            SnapshotUriUtils::CountUriParam(spec.uri(), SnapshotUriUtils::kSnapshotVersionParam);
-        if (version_param_count == 0) {
-            // Legacy metadata is a stale reconciliation component, but a
-            // current delta spec in the same stable location still protects
-            // the location from coarse-grained cleanup.
-            continue;
-        }
-        SnapshotUriInfo info;
-        if (version_param_count != 1 || !SnapshotUriUtils::ParseSnapshotUriInfo(spec.uri(), info)) {
-            return true;
-        }
-        contains_committed = contains_committed || (!committed_version.empty() && info.version == committed_version);
-        contains_in_flight = contains_in_flight ||
-                             (preserve_in_flight && !in_flight_version.empty() && info.version == in_flight_version);
-    }
-    // Delta merge is spec-granular and can temporarily leave multiple
-    // generations in one stable location. Cleanup is location-granular, so it
-    // must preserve the whole location when any current/in-flight spec is
-    // present; deleting stale sibling specs is deferred to a later complete
-    // snapshot rather than risking a false negative for a successful delta.
-    return !contains_committed && !contains_in_flight;
-}
-
 bool IsEventReportLocationReadable(const CacheLocation &location,
                                    bool strict_query_visibility,
                                    const std::string &committed_version) {
@@ -2777,7 +2736,7 @@ ErrorCode CacheManager::ReportEvent(RequestContext *request_context,
                                                   const std::string &cleanup_instance,
                                                   const std::string &down_host,
                                                   uint64_t generation) {
-                if (auto backend = weak_backend.lock()) {
+                if (auto backend = weak_backend.lock(); backend) {
                     auto gc = weak_gc.lock();
                     if (gc && !gc->RegisterHostCleanupIntent(
                                   cleanup_instance, requested_type, down_host, generation, backend)) {
@@ -3671,7 +3630,8 @@ ErrorCode CacheManager::ReportEvent(RequestContext *request_context,
     }
     if (!has_host_down && event_backend->IsNodeRegistered(instance_id, host_ip_port) && cache_garbage_collector_ &&
         cache_garbage_collector_->IsEventReportCleanupEnabled()) {
-        cache_garbage_collector_->CancelHostCleanupIntent(instance_id, requested_type, host_ip_port);
+        cache_garbage_collector_->CancelHostCleanupIntent(
+            instance_id, requested_type, host_ip_port, event_backend->GetNodeGeneration(instance_id, host_ip_port));
     }
 
     // MetaSearcher calls this once after the fused target-location read and
@@ -3906,6 +3866,14 @@ ErrorCode CacheManager::ReportEvent(RequestContext *request_context,
             if (cache_garbage_collector_ && cache_garbage_collector_->IsEventReportCleanupEnabled()) {
                 cleanup_dispatched = cache_garbage_collector_->RegisterHostCleanupIntent(
                     instance_id, requested_type, host_ip_port, gen_at_trigger, event_backend);
+                if (!cleanup_dispatched) {
+                    KVCM_LOG_WARN("trace_id [%s] | HOST_DOWN: failed to register GC cleanup intent for host [%s], "
+                                  "instance [%s], gen=%" PRIu64,
+                                  trace_id.c_str(),
+                                  host_ip_port.c_str(),
+                                  instance_id.c_str(),
+                                  gen_at_trigger);
+                }
             } else {
                 const auto cleanup_state = event_cleanup_callback_state_;
                 uint64_t cleanup_epoch = 0;

kv_cache_manager/manager/cache_manager.cc

@@ -3529,6 +3529,7 @@ MetaSearcher::BatchDeleteLocationsForMaintenance(RequestContext *request_context
         request_context, keys, location_ids_per_key, expected_location_values);
     out.ec = result.ec;
     out.shard_lock_wait_time_us = result.shard_lock_wait_time_us;
+    out.shard_lock_hold_time_us = result.shard_lock_hold_time_us;
     out.per_location_results = std::move(result.per_location_results);
     if (out.per_location_results.size() != keys.size()) {
         out.ec = EC_ERROR;
@@ -3553,18 +3554,15 @@ MetaSearcher::BatchDeleteLocationsForMaintenance(RequestContext *request_context
     for (size_t i = 0; i < out.per_location_results.size(); ++i) {
         for (size_t j = 0; j < out.per_location_results[i].size(); ++j) {
             auto &current = out.per_location_results[i][j];
-            if (current.ec == EC_OK || current.ec == EC_NOENT) {
-                if (prior_shape_valid) {
-                    const auto &prior = (*prior_results)[i][j];
-                    current.removed_from_hot = current.removed_from_hot || prior.removed_from_hot;
-                    current.removed_from_persistent = current.removed_from_persistent || prior.removed_from_persistent;
-                    current.reclaimed_hot_key = current.reclaimed_hot_key || prior.reclaimed_hot_key;
-                    current.reclaimed_persistent_key =
-                        current.reclaimed_persistent_key || prior.reclaimed_persistent_key;
-                }
-                needs_sync = needs_sync || current.removed_from_hot || current.removed_from_persistent ||
-                             current.reclaimed_hot_key || current.reclaimed_persistent_key;
-            }
+            if (prior_shape_valid) {
+                const auto &prior = (*prior_results)[i][j];
+                current.removed_from_hot = current.removed_from_hot || prior.removed_from_hot;
+                current.removed_from_persistent = current.removed_from_persistent || prior.removed_from_persistent;
+                current.reclaimed_hot_key = current.reclaimed_hot_key || prior.reclaimed_hot_key;
+                current.reclaimed_persistent_key = current.reclaimed_persistent_key || prior.reclaimed_persistent_key;
+            }
+            needs_sync = needs_sync || current.removed_from_hot || current.removed_from_persistent ||
+                         current.reclaimed_hot_key || current.reclaimed_persistent_key;
         }
     }
 

kv_cache_manager/manager/meta_searcher.cc

@@ -347,6 +347,7 @@ class MetaSearcher {
         MaintenanceLocationDeleteResults per_location_results;
         bool sync_succeeded = false;
         int64_t shard_lock_wait_time_us = 0;
+        int64_t shard_lock_hold_time_us = 0;
     };
     // Exact-value metadata-only maintenance deletion. The optional prior
     // result carries mutations from a previous Sync failure so accounting is