diff --git a/lib/collection/src/collection/collection_ops.rs b/lib/collection/src/collection/collection_ops.rs index fea7506632..a4d00604e4 100644 --- a/lib/collection/src/collection/collection_ops.rs +++ b/lib/collection/src/collection/collection_ops.rs @@ -222,7 +222,7 @@ impl Collection { // Check that we are not removing the *last* replica or the last *active* replica // // `is_last_active_replica` counts both `Active` and `ReshardingScaleDown` replicas! - if peers.len() == 1 || replica_set.is_last_active_replica(peer_id) { + if peers.len() == 1 || replica_set.is_last_source_of_truth_replica(peer_id) { return Err(CollectionError::BadRequest { description: format!( "Shard {shard_id} must have at least one active replica after removing {peer_id}", diff --git a/lib/collection/src/collection/mod.rs b/lib/collection/src/collection/mod.rs index 7e3976b35d..21eda1dcb6 100644 --- a/lib/collection/src/collection/mod.rs +++ b/lib/collection/src/collection/mod.rs @@ -432,7 +432,7 @@ impl Collection { // 3. Do not deactivate the last active replica // // `is_last_active_replica` counts both `Active` and `ReshardingScaleDown` replicas! - if replica_set.is_last_active_replica(peer_id) && !new_state.is_active() { + if replica_set.is_last_source_of_truth_replica(peer_id) && !new_state.is_active() { return Err(CollectionError::bad_input(format!( "Cannot deactivate the last active replica {peer_id} of shard {shard_id}" ))); diff --git a/lib/collection/src/shards/replica_set/mod.rs b/lib/collection/src/shards/replica_set/mod.rs index 3a892a77ca..4c2f67a937 100644 --- a/lib/collection/src/shards/replica_set/mod.rs +++ b/lib/collection/src/shards/replica_set/mod.rs @@ -411,9 +411,40 @@ impl ShardReplicaSet { self.replica_state.read().peers() } - pub fn is_last_active_replica(&self, peer_id: PeerId) -> bool { + /// Checks if the current replica contains a unique source of truth and should never + /// be deactivated or removed. + /// If current replica is the only "alive" replica, it is considered the last source of truth. + /// + /// If our replica is `Initializing`, we consider it to be the last source of truth if there is + /// no other active replicas. If we would deactivate it, it will be impossible to recover the + /// replica later. This may happen if we got killed or crashed during collection creation. + /// + /// Same logic applies to `Listener` replicas, as they are not recoverable if there are no + /// other active replicas. + /// + /// Examples: + /// Active(this), Initializing(other), Initializing(other) -> true + /// Active(this), Active(other) -> false + /// Initializing(this) -> true + /// Initializing(this), Initializing(other) -> true + /// Initializing(this), Dead(other) -> true + /// Initializing(this), Active(other) -> false + /// Active(this), Initializing(other) -> true + /// + pub fn is_last_source_of_truth_replica(&self, peer_id: PeerId) -> bool { // This includes `Active` and `ReshardingScaleDown` replicas! let active_peers = self.replica_state.read().active_peers(); + if active_peers.is_empty() { + if let Some(peer_state) = self.peer_state(peer_id) { + // If there are no other active peers, deactivating those replicas + // is not recoverable, so it is considered the last source of truth, + // even though it is not technically active. + return matches!( + peer_state, + ReplicaState::Initializing | ReplicaState::Listener + ); + } + } active_peers.len() == 1 && active_peers.contains(&peer_id) } @@ -421,17 +452,6 @@ impl ShardReplicaSet { self.replica_state.read().get_peer_state(peer_id) } - /// List the peer IDs on which this shard is active, both the local and remote peers. - pub fn active_shards(&self) -> Vec { - let replica_state = self.replica_state.read(); - replica_state - // This is a part of deprecated built-in resharding implementation, so we don't care - .active_peers() - .into_iter() - .filter(|&peer_id| !self.is_locally_disabled(peer_id)) - .collect() - } - /// List the remote peer IDs on which this shard is active, excludes the local peer ID. pub fn active_remote_shards(&self) -> Vec { let replica_state = self.replica_state.read();