| 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 |
| 136 | // (createTransform). Streaming a transform-only session still needs this path, |
| 137 | // so the early-out must consider transform_recipes_ too, not just recipes_. |
| 138 | if (changed_inputs.empty() || (recipes_.empty() && transform_recipes_.empty())) { |
| 139 | return {}; |
| 140 | } |
| 141 | // Mark dependent nodes dirty for the just-committed inputs, then run them. |
| 142 | // scheduleAll only re-runs dirty nodes (incremental watermark), so a stateful |
| 143 | // node folds just the new tail and its accumulator persists across commits. |
| 144 | derived_->onSourceCommitted(changed_inputs); |
| 145 | // Best-effort: a processor/script failure must not abort ingest. The node |
| 146 | // surfaces its own failed state; we still notify whatever advanced. |
| 147 | (void)derived_->scheduleAll(); |
| 148 | |
| 149 | // Re-notify every filter AND transform output (covers chained nodes too); plots re-read. |
| 150 | std::vector<TopicId> outputs; |
| 151 | outputs.reserve(recipes_.size() + transform_recipes_.size()); |
| 152 | for (const auto& [out_tid, recipe] : recipes_) { |
| 153 | (void)recipe; |
| 154 | outputs.push_back(out_tid); |
| 155 | } |
| 156 | for (const auto& [key, recipe] : transform_recipes_) { |
| 157 | (void)key; |
| 158 | for (const TopicId out_tid : recipe.output_topic_ids) { |
| 159 | outputs.push_back(out_tid); |
| 160 | } |
| 161 | } |
| 162 | return outputs; |
| 163 | } |
| 164 | |
| 165 | std::vector<TopicId> DataProcessorService::recomputeForReplacedSources(const std::vector<TopicId>& replaced_inputs) { |
| 166 | std::vector<TopicId> affected_outputs; |