(
&mut self,
id: ReplicaId,
client: ReplicaClient,
config: ReplicaConfig,
epoch: u64,
)
| 347 | } |
| 348 | |
| 349 | fn add_replica_state( |
| 350 | &mut self, |
| 351 | id: ReplicaId, |
| 352 | client: ReplicaClient, |
| 353 | config: ReplicaConfig, |
| 354 | epoch: u64, |
| 355 | ) -> Result<(), read_holds::ReadHoldIssuerHungUp> { |
| 356 | let log_ids: BTreeSet<_> = config.logging.index_logs.values().copied().collect(); |
| 357 | |
| 358 | let metrics = self.metrics.for_replica(id); |
| 359 | let mut replica = ReplicaState::new( |
| 360 | id, |
| 361 | client, |
| 362 | config, |
| 363 | metrics, |
| 364 | self.introspection_tx.clone(), |
| 365 | epoch, |
| 366 | ); |
| 367 | |
| 368 | // Add per-replica collection state. |
| 369 | let mut shutdown_input = None; |
| 370 | for (collection_id, collection) in &self.collections { |
| 371 | // Skip log collections not maintained by this replica, |
| 372 | // and collections targeted at a different replica. |
| 373 | if (collection.log_collection && !log_ids.contains(collection_id)) |
| 374 | || collection.target_replica.is_some_and(|rid| rid != id) |
| 375 | { |
| 376 | continue; |
| 377 | } |
| 378 | |
| 379 | let as_of = if collection.log_collection { |
| 380 | // For log collections, we don't send a `CreateDataflow` command to the replica, so |
| 381 | // it doesn't know which as-of the controler chose and defaults to the minimum |
| 382 | // frontier instead. We need to initialize the controller-side tracking with the |
| 383 | // same frontier, to avoid observing regressions in the reported frontiers. |
| 384 | Antichain::from_elem(Timestamp::MIN) |
| 385 | } else { |
| 386 | collection.read_frontier().to_owned() |
| 387 | }; |
| 388 | |
| 389 | // Cloning a `ReadHold` fails when its issuer has hung up. For these holds the issuer |
| 390 | // is the `StorageCollections`, which doesn't hang up as long as the `Instance` exists, |
| 391 | // except during process shutdown, when the tokio runtime drops tasks in arbitrary |
| 392 | // order. In that case there is no way of correctly initializing the per-replica |
| 393 | // collection state, so we give up. We still add the replica itself, to keep the |
| 394 | // bookkeeping consistent with the controller's, and then signal the unrecoverable |
| 395 | // error to the caller, which shuts the instance down. |
| 396 | let mut input_read_holds = Vec::with_capacity(collection.storage_dependencies.len()); |
| 397 | let mut hung_up = Vec::new(); |
| 398 | for hold in collection.storage_dependencies.values() { |
| 399 | match hold.try_clone() { |
| 400 | Ok(hold) => input_read_holds.push(hold), |
| 401 | Err(read_holds::ReadHoldIssuerHungUp(input_id)) => hung_up.push(input_id), |
| 402 | } |
| 403 | } |
| 404 | if !hung_up.is_empty() { |
| 405 | tracing::error!( |
| 406 | replica_id = %id, |
no test coverage detected