Updates internal state based on incoming compaction commands.
(&mut self, id: GlobalId, frontier: Antichain<Timestamp>)
| 643 | |
| 644 | /// Updates internal state based on incoming compaction commands. |
| 645 | fn absorb_compaction(&mut self, id: GlobalId, frontier: Antichain<Timestamp>) { |
| 646 | tracing::debug!(?self.active_ingestions, ?id, ?frontier, "allow_compaction"); |
| 647 | |
| 648 | if frontier.is_empty() { |
| 649 | self.active_ingestions.remove(&id); |
| 650 | self.ingestion_exports.remove(&id); |
| 651 | self.active_exports.remove(&id); |
| 652 | } |
| 653 | } |
| 654 | |
| 655 | /// Returns the replicas that are actively running the given object (ingestion or export). |
| 656 | fn active_replicas(&mut self, id: &GlobalId) -> Box<dyn Iterator<Item = &mut Replica> + '_> { |