(&mut self, id: ReplicaId)
| 1281 | /// Remove an existing instance replica, by ID. |
| 1282 | #[mz_ore::instrument(level = "debug")] |
| 1283 | pub fn remove_replica(&mut self, id: ReplicaId) -> Result<(), ReplicaMissing> { |
| 1284 | let replica = self.replicas.remove(&id).ok_or(ReplicaMissing(id))?; |
| 1285 | |
| 1286 | // Before dropping the replica state (and the contained input read holds), log read holds |
| 1287 | // that are the last line of defense against compaction of a dataflow's storage inputs. If |
| 1288 | // the corresponding global read hold has already been released, dropping the per-replica |
| 1289 | // read hold will allow compaction, which can cause the replica to panic trying to install |
| 1290 | // the dataflow. |
| 1291 | // |
| 1292 | // This exists primarily to help diagnose incidents-and-escalations#39. |
| 1293 | for (collection_id, replica_collection) in &replica.collections { |
| 1294 | let collection = self.collections.get(collection_id); |
| 1295 | for replica_hold in &replica_collection.input_read_holds { |
| 1296 | let input_id = replica_hold.id(); |
| 1297 | let global_hold = collection.and_then(|c| c.storage_dependencies.get(&input_id)); |
| 1298 | let unprotected = global_hold |
| 1299 | .is_none_or(|h| PartialOrder::less_than(replica_hold.since(), h.since())); |
| 1300 | if unprotected { |
| 1301 | tracing::warn!( |
| 1302 | replica_id = %id, |
| 1303 | %collection_id, |
| 1304 | %input_id, |
| 1305 | replica_hold_since = ?replica_hold.since(), |
| 1306 | global_hold_since = ?global_hold.map(|h| h.since()), |
| 1307 | "dropping per-replica read hold without equivalent global read hold", |
| 1308 | ); |
| 1309 | } |
| 1310 | } |
| 1311 | } |
| 1312 | drop(replica); |
| 1313 | |
| 1314 | // Subscribes targeting this replica either won't be served anymore (if the replica is |
| 1315 | // dropped) or might produce inconsistent output (if the target collection is an |
| 1316 | // introspection index). We produce an error to inform upstream. |
| 1317 | let to_drop: Vec<_> = self.subscribes_targeting(id).collect(); |
| 1318 | for subscribe_id in to_drop { |
| 1319 | let subscribe = self.subscribes.remove(&subscribe_id).unwrap(); |
| 1320 | let response = ComputeControllerResponse::SubscribeResponse( |
| 1321 | subscribe_id, |
| 1322 | SubscribeBatch { |
| 1323 | lower: subscribe.frontier.clone(), |
| 1324 | upper: subscribe.frontier, |
| 1325 | updates: Err(ERROR_TARGET_REPLICA_FAILED.into()), |
| 1326 | }, |
| 1327 | ); |
| 1328 | self.deliver_response(response); |
| 1329 | } |
| 1330 | |
| 1331 | // Peeks targeting this replica might not be served anymore (if the replica is dropped). |
| 1332 | // If the replica has failed it might come back and respond to the peek later, but it still |
| 1333 | // seems like a good idea to cancel the peek to inform the caller about the failure. This |
| 1334 | // is consistent with how we handle targeted subscribes above. |
| 1335 | let mut peek_responses = Vec::new(); |
| 1336 | let mut to_drop = Vec::new(); |
| 1337 | for (uuid, peek) in self.peeks_targeting(id) { |
| 1338 | peek_responses.push(ComputeControllerResponse::PeekNotification( |
| 1339 | uuid, |
| 1340 | PeekNotification::Error(ERROR_TARGET_REPLICA_FAILED.into()), |
no test coverage detected