[manager] address asynchronous Copy review findings

Author: shaohuaxi

kv_cache_manager/config/registry_manager.cc

@@ -176,10 +176,30 @@ ErrorCode RegistryManager::RemoveStorage(RequestContext *request_context, const
     // Async Copy recovery needs that config to recreate the PACE client; the
     // old order could lose the only recovery route when Close returned busy or
     // failed with in-flight operations.
+    const auto backend = data_storage_manager_->GetDataStorageBackend(global_unique_name);
+    if (!backend) {
+        RETURN_IF_EC_NOT_OK_WITH_LOG_S(WARN, EC_NOENT, "remove storage failed: runtime backend not found");
+    }
+    const StorageConfig storage_config = backend->GetStorageConfig();
     auto ec = data_storage_manager_->UnRegisterStorage(global_unique_name);
     RETURN_IF_EC_NOT_OK_WITH_LOG_S(WARN, ec, "remove storage failed");
     ec = LoadAndDelete(kRegistryStorageKey, global_unique_name);
-    RETURN_IF_EC_NOT_OK_WITH_LOG_S(WARN, ec, "load and delete storage failed after backend close");
+    if (ec != EC_OK) {
+        // The durable registry still contains this storage. Restore the runtime
+        // backend so the current leader matches the state that the next leader
+        // will recover, and so a retry can make progress instead of failing
+        // permanently with EC_NOENT.
+        const auto rollback_ec =
+            data_storage_manager_->RegisterStorage(request_context, global_unique_name, storage_config);
+        if (rollback_ec != EC_OK) {
+            PREFIX_LOG_S(ERROR,
+                         "load and delete storage failed after backend close, and runtime rollback failed, "
+                         "delete ec[%d] rollback ec[%d]",
+                         static_cast<int>(ec),
+                         static_cast<int>(rollback_ec));
+        }
+        RETURN_IF_EC_NOT_OK_WITH_LOG_S(WARN, ec, "load and delete storage failed after backend close");
+    }
     PREFIX_LOG_S(INFO, "remove storage OK");
     return ec;
 }

kv_cache_manager/manager/cache_manager.cc

@@ -644,12 +644,22 @@ ErrorCode CacheManager::RemoveInstance(RequestContext *request_context,
             }
             if (!migration_manager_->GetActiveBlockKeysForInstance(instance_id).empty()) {
                 KVCM_LOG_WARN("[%s] RemoveInstance drain timeout (%dms) for instance %s, "
-                              "proceeding with trim; residual WRITING targets will be orphan-cleaned",
+                              "refusing removal while asynchronous Copy state is still owned",
                               trace_id.c_str(),
                               kDrainTimeoutMs,
                               instance_id.c_str());
             }
         }
+        // A quarantined guard is intentionally invisible to the active task
+        // table, and an accepted physical cleanup may outlive task completion.
+        // Removing the registry entry here would make those durable guards
+        // undiscoverable on the next leader and could close their backend while
+        // it is still being used. Keep the instance registered until every
+        // active/quarantined/cleanup reference has drained.
+        if (migration_manager_->HasAsyncCopyInstanceReference(instance_id)) {
+            PREFIX_LOG(WARN, "remove instance rejected: asynchronous Copy reference remains");
+            return EC_EXIST;
+        }
     }
 
     auto ec = registry_manager_->RemoveInstance(request_context, instance_group, instance_id);

kv_cache_manager/manager/migration_manager.cc

@@ -144,6 +144,7 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
                                          const std::string &operator_name,
                                          const std::string &external_fencing_evidence);
     bool HasAsyncCopyStorageReference(const std::string &storage_name) const;
+    bool HasAsyncCopyInstanceReference(const std::string &instance_id) const;
 
     // ---- Copy 路径 ----
     // 同步完成"建目标 location + 提交 copy 任务"，copy 字节复制异步执行；
@@ -240,6 +241,7 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
         kTargetWritingExists,   // 目标 storage 上存在 WRITING 副本（可能为其他迁移半成品）
         kSourceServingNotFound, // 源 storage 上没有 SERVING 副本，无可复制源
         kSourceRetrySuppressed, // 尚未被目标覆盖的源均处于失败退避，当前不应再次提交 Copy
+        kSourcePinnedByGuard,   // 源被另一条持久异步 Copy guard 钉住，不能并行派生新迁移
     };
 
     struct CopyAdmission {
@@ -455,6 +457,7 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
     struct PendingCopy {
         std::string instance_id;
         int64_t block_key = 0;
+        size_t expected_items = 0;
         std::future<PlanExecuteResult> future;
         bool native_async = false;
         std::future<AsyncCopyRemoteSubmitResult> remote_submit_future;
@@ -463,6 +466,16 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
         std::chrono::steady_clock::time_point completion_retry_after{};
     };
 
+    // An async Copy is not fully cleaned up when its target/source metadata is
+    // merely queued for deletion. Keep a reference until the executor finishes
+    // the physical delete so RemoveStorage cannot close the backend underneath
+    // the cleanup task.
+    struct PendingLocationCleanup {
+        std::string instance_id;
+        std::string storage_name;
+        std::future<PlanExecuteResult> future;
+    };
+
     struct ExpiringMark {
         int64_t deadline_ms = 0;
         std::string instance_id;
@@ -517,10 +530,15 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
     bool ResumePendingGuardFinalization(const CopyTaskContext &ctx);
     bool PersistRemoteAsyncCopyAcceptance(const std::string &instance_id,
                                           int64_t block_key,
+                                          size_t expected_items,
                                           const AsyncCopyRemoteSubmitResult &remote_result,
                                           CopyTaskContext &out_ctx,
                                           bool &out_cancel_requested);
     void SubmitPreparedTargetLocationDelete(const CopyTaskContext &ctx);
+    void TrackLocationCleanup(const std::string &instance_id,
+                              const std::string &storage_name,
+                              std::future<PlanExecuteResult> future);
+    void ProcessCompletedLocationCleanups();
     bool ReserveAsyncCopyCredit(const MigrationRequest &request, uint64_t total_bytes);
     void ReleaseAsyncCopyCredit(const CopyTaskContext &ctx);
     void MoveAsyncCopyCreditToQuarantine(const CopyTaskContext &ctx, const std::string &reason);
@@ -682,6 +700,8 @@ class MigrationManager : public std::enable_shared_from_this<MigrationManager> {
     std::mutex pending_mutex_;
     std::condition_variable pending_cv_;
     std::deque<PendingCopy> pending_copies_;
+    mutable std::mutex pending_cleanup_mutex_;
+    std::deque<PendingLocationCleanup> pending_location_cleanups_;
 
     std::mutex mark_expiry_mutex_;
     std::priority_queue<ExpiringMark, std::vector<ExpiringMark>, ExpiringMarkGreater> mark_expiry_queue_;

kv_cache_manager/manager/migration_manager.h

@@ -732,13 +732,12 @@ SchedulePlanExecutor::PrepareDeleteTaskImpl(const std::string &instance_id,
         if (admission_result.actual_task.block_keys.empty()) {
             return admission_result;
         }
-        if (!indexer->Sync(admission_result.actual_task.block_keys)) {
-            admission_result.result = MakeErrorResult(
-                EC_ERROR,
-                StringUtil::FormatString(
-                    "Sync failed or timed out for prepared location delete, instance[%s]", instance_id.c_str()));
-            return admission_result;
-        }
+        // prepared_deleting is used only after the caller has durably changed
+        // the exact location to CLS_DELETING and cleared its Copy guard.  There
+        // is no metadata mutation to flush here.  A second Sync could fail after
+        // the authoritative transition already succeeded and strand a
+        // guard-free CLS_DELETING location without ever scheduling its physical
+        // cleanup.
         admission_result.needs_physical_delete = true;
         return admission_result;
     }

kv_cache_manager/manager/schedule_plan_executor.cc

@@ -146,6 +146,8 @@ class SchedulePlanExecutor {
     ErrorCode RequestCancelAsyncCopy(const std::string &storage_name, const std::string &operation_id);
     AsyncDeleteSubmitResult SubmitAsync(const CacheMetaDelRequest &task);
     AsyncDeleteSubmitResult SubmitAsync(const CacheLocationDelRequest &task);
+    std::future<PlanExecuteResult> SubmitLocationDelete(const CacheLocationDelRequest &task,
+                                                        ScheduleTaskClass task_class);
 
     bool SubmitNonBlocking(const CacheMetaDelRequest &req, ScheduleTaskClass task_class = ScheduleTaskClass::kSystem);
     bool SubmitNonBlocking(const CacheLocationDelRequest &req,
@@ -211,8 +213,6 @@ class SchedulePlanExecutor {
     AsyncDeleteSubmitResult SubmitDeleteTaskAsync(std::chrono::microseconds delay,
                                                   std::function<LocationDelAdmissionResult()> prepare);
     std::future<PlanExecuteResult> SubmitMetaDelete(const CacheMetaDelRequest &task, ScheduleTaskClass task_class);
-    std::future<PlanExecuteResult> SubmitLocationDelete(const CacheLocationDelRequest &task,
-                                                        ScheduleTaskClass task_class);
     PlanExecuteResult DoLocationDelTask(const CacheLocationDelRequest &task);
     void DoCopyTask(const std::shared_ptr<std::promise<PlanExecuteResult>> &promise,
                     const CacheLocationCopyRequest &task);