Updates internal state based on incoming export commands. This does _not_ send commands to replicas, we only record the export in state and potentially update scheduling decisions.
(&mut self, export: RunSinkCommand)
| 411 | /// This does _not_ send commands to replicas, we only record the export |
| 412 | /// in state and potentially update scheduling decisions. |
| 413 | fn absorb_export(&mut self, export: RunSinkCommand) { |
| 414 | let existing_export_state = self.active_exports.get_mut(&export.id); |
| 415 | |
| 416 | if let Some(export_state) = existing_export_state { |
| 417 | // It's an update for an existing export. We don't need to |
| 418 | // change anything about our scheduling decisions, no need to |
| 419 | // update active_exports. |
| 420 | |
| 421 | tracing::debug!( |
| 422 | export_id = %export.id, |
| 423 | active_replicas = %export_state.active_replicas.iter().map(|id| id.to_string()).join(", "), |
| 424 | "updating export" |
| 425 | ); |
| 426 | } else { |
| 427 | // We create a new export state for this export. |
| 428 | let export_state = ActiveExport { |
| 429 | active_replicas: BTreeSet::new(), |
| 430 | }; |
| 431 | self.active_exports.insert(export.id, export_state); |
| 432 | |
| 433 | // Maybe update scheduling decisions. |
| 434 | self.update_scheduling(false); |
| 435 | } |
| 436 | } |
| 437 | |
| 438 | /// Update scheduling decisions, that is what replicas should be running a |
| 439 | /// given object, if needed. |
no test coverage detected