Update scheduling decisions, that is what replicas should be running a given object, if needed. An important property of this scheduling algorithm is that we never change the scheduling decision for single-replica objects unless we have to, that is unless the replica that they are running on goes away. We do this, so that we don't send a mix of "run"/"allow compaction"/"run" messages to replicas,
(&mut self, send_commands: bool)
| 454 | /// If `send_commands` is true, will send commands for newly-scheduled |
| 455 | /// single-replica objects. |
| 456 | fn update_scheduling(&mut self, send_commands: bool) { |
| 457 | #[derive(Debug)] |
| 458 | enum ObjectId { |
| 459 | Ingestion(GlobalId), |
| 460 | Export(GlobalId), |
| 461 | } |
| 462 | // We first collect scheduling preferences and then schedule below. |
| 463 | // Applying the decision needs a mutable borrow but we also need to |
| 464 | // borrow for determining `prefers_single_replica`, so we split this |
| 465 | // into two loops. |
| 466 | let mut scheduling_preferences: Vec<(ObjectId, bool)> = Vec::new(); |
| 467 | |
| 468 | for ingestion_id in self.active_ingestions.keys() { |
| 469 | let ingestion_description = self |
| 470 | .get_ingestion_description(ingestion_id) |
| 471 | .expect("missing ingestion description"); |
| 472 | |
| 473 | let prefers_single_replica = ingestion_description |
| 474 | .desc |
| 475 | .connection |
| 476 | .prefers_single_replica(); |
| 477 | |
| 478 | scheduling_preferences |
| 479 | .push((ObjectId::Ingestion(*ingestion_id), prefers_single_replica)); |
| 480 | } |
| 481 | |
| 482 | for export_id in self.active_exports.keys() { |
| 483 | // All sinks prefer single replica |
| 484 | scheduling_preferences.push((ObjectId::Export(*export_id), true)); |
| 485 | } |
| 486 | |
| 487 | // Collect all commands per replica and send them in one go. |
| 488 | let mut commands_by_replica: BTreeMap<ReplicaId, Vec<ObjectId>> = BTreeMap::new(); |
| 489 | |
| 490 | for (object_id, prefers_single_replica) in scheduling_preferences { |
| 491 | let active_replicas = match object_id { |
| 492 | ObjectId::Ingestion(ingestion_id) => { |
| 493 | &mut self |
| 494 | .active_ingestions |
| 495 | .get_mut(&ingestion_id) |
| 496 | .expect("missing ingestion state") |
| 497 | .active_replicas |
| 498 | } |
| 499 | ObjectId::Export(export_id) => { |
| 500 | &mut self |
| 501 | .active_exports |
| 502 | .get_mut(&export_id) |
| 503 | .expect("missing ingestion state") |
| 504 | .active_replicas |
| 505 | } |
| 506 | }; |
| 507 | |
| 508 | if prefers_single_replica { |
| 509 | // For single-replica ingestion, schedule only if it's not already running. |
| 510 | if active_replicas.is_empty() { |
| 511 | let target_replica = self.replicas.keys().min().copied(); |
| 512 | if let Some(first_replica_id) = target_replica { |
| 513 | tracing::info!( |
no test coverage detected