[fix](cloud) Add predecessor instance id (#62276)

When a rollback exists, the predecessor instance is not the same as the
source instance. This change separates the two semantics into distinct
fields for clarity.

Author: w41ter

cloud/src/resource-manager/resource_manager.cpp

@@ -1414,18 +1414,19 @@ std::pair<MetaServiceCode, std::string> ResourceManager::refresh_instance(
 void ResourceManager::refresh_instance(const std::string& instance_id,
                                        const InstanceInfoPB& instance) {
     bool is_successor_instance = instance.has_original_instance_id();
-    std::string source_instance_id = is_successor_instance ? instance.source_instance_id() : "";
+    std::string predecessor_instance_id =
+            is_successor_instance ? instance.predecessor_instance_id() : "";
 
     std::lock_guard l(mtx_);
     for (auto i = node_info_.begin(); i != node_info_.end();) {
-        // erase all nodes not belong to this instance_id
-        if (i->second.instance_id != instance_id &&
-            // ... or, if is_successor_instance, erase nodes belong to source_instance_id
-            (!is_successor_instance || i->second.instance_id != source_instance_id)) {
+        // erase all nodes belong to this instance_id
+        if (i->second.instance_id == instance_id ||
+            // ... or, if is_successor_instance, erase nodes belong to predecessor_instance_id
+            (is_successor_instance && i->second.instance_id == predecessor_instance_id)) {
+            i = node_info_.erase(i);
+        } else {
             ++i;
-            continue;
         }
-        i = node_info_.erase(i);
     }
 
     // If successor_instance_id is set, it means this instance has a successor instance,

gensrc/proto/cloud.proto

@@ -129,7 +129,7 @@ message InstanceInfoPB {
     // For snapshot
     optional MultiVersionStatus multi_version_status = 110;
     optional SnapshotSwitchStatus snapshot_switch_status = 111;
-    optional string source_instance_id = 112;   // The instance cloned from.
+    optional string source_instance_id = 112;   // The instance cloned from (the snapshot instance id).
     optional string source_snapshot_id = 113;   // The snapshot cloned from.
 
     // Inherited from which instance (only used during rollback, the earliest instance id).
@@ -147,6 +147,11 @@ message InstanceInfoPB {
     optional int64 snapshot_retained_data_size = 121;
     optional int64 snapshot_billable_data_size = 122;
     optional SnapshotCompactStatus snapshot_compact_status = 123;
+
+    // The instance that is being rolled back, only used during rollback.
+    // It is not always same as source_instance_id, because the source instance may inherit from another
+    // instance during rollback, and the predecessor instance is the real instance which to execute the rollback.
+    optional string predecessor_instance_id = 124;
 }
 
 message StagePB {