| 120 | } |
| 121 | |
| 122 | void DataProcessorService::clearAllFilters() { |
| 123 | for (const auto& [output_topic_id, recipe] : recipes_) { |
| 124 | (void)output_topic_id; // key == recipe.output_topic_id; iterate by recipe for clarity |
| 125 | // Drop the engine node, then retire the materialized output so the next |
| 126 | // CatalogModel::rebuildFromDatastore stops listing it. removeNode alone would |
| 127 | // leave a catalog zombie (it does not touch DataEngine::retired_topic_ids). |
| 128 | (void)derived_->removeNode(recipe.node_id); |
| 129 | engine_.retireTopic(recipe.output_topic_id); |
| 130 | } |
| 131 | recipes_.clear(); |
| 132 | } |
| 133 | |
| 134 | std::vector<TopicId> DataProcessorService::advanceOnCommit(const std::vector<TopicId>& changed_inputs) { |
| 135 | // Run whenever there is ANY eager node — per-curve filters OR named transforms |