| 1141 | } |
| 1142 | |
| 1143 | PJ::Status DerivedEngine::scheduleActiveLocked(const std::unordered_set<PJ::NodeId>& active_nodes) { |
| 1144 | auto order = topologicalOrder(); |
| 1145 | |
| 1146 | // Compute the set of nodes to consider (active_nodes ∪ their transitive upstream deps). |
| 1147 | tsl::robin_set<PJ::NodeId> filter; |
| 1148 | if (!active_nodes.empty()) { |
| 1149 | std::queue<PJ::NodeId> bfs; |
| 1150 | for (PJ::NodeId n : active_nodes) { |
| 1151 | if (impl_->nodes.contains(n)) { |
| 1152 | filter.insert(n); |
| 1153 | bfs.push(n); |
| 1154 | } |
| 1155 | } |
| 1156 | while (!bfs.empty()) { |
| 1157 | PJ::NodeId curr = bfs.front(); |
| 1158 | bfs.pop(); |
| 1159 | auto nit = impl_->nodes.find(curr); |
| 1160 | if (nit == impl_->nodes.end()) { |
| 1161 | continue; |
| 1162 | } |
| 1163 | for (PJ::TopicId in_tid : nit->second.all_input_topic_ids) { |
| 1164 | auto prod_it = impl_->output_topic_to_node.find(in_tid); |
| 1165 | if (prod_it == impl_->output_topic_to_node.end()) { |
| 1166 | continue; |
| 1167 | } |
| 1168 | PJ::NodeId prod = prod_it->second; |
| 1169 | if (filter.insert(prod).second) { |
| 1170 | bfs.push(prod); |
| 1171 | } |
| 1172 | } |
| 1173 | } |
| 1174 | } |
| 1175 | |
| 1176 | // The first per-node failure is remembered and returned, but a failed node must |
| 1177 | // NOT abort the others — one bad filter must not freeze every derived series. |
| 1178 | PJ::Status first_error = PJ::okStatus(); |
| 1179 | for (PJ::NodeId node_id : order) { |
| 1180 | if (!active_nodes.empty() && !filter.contains(node_id)) { |
| 1181 | continue; |
| 1182 | } |
| 1183 | |
| 1184 | auto& node = impl_->nodes.at(node_id); |
| 1185 | if (!node.dirty) { |
| 1186 | continue; |
| 1187 | } |
| 1188 | |
| 1189 | PJ::Status s = PJ::okStatus(); |
| 1190 | if (nodeInputRegressed(engine_, node)) { |
| 1191 | // Late (out-of-order) input behind the node's watermark: reset + full |
| 1192 | // replay over the now time-merged input instead of incremental work. |
| 1193 | s = recomputeBatchLocked(node_id); // already holding engine_.lockEngine() |
| 1194 | } else if (!node.is_mimo) { |
| 1195 | s = runSisoIncremental(*impl_, engine_, node); |
| 1196 | } else { |
| 1197 | s = runMimoIncremental(*impl_, engine_, node); |
| 1198 | } |
| 1199 | |
| 1200 | node.dirty = false; // processed even on failure, so a sticky-failed node does |
nothing calls this directly
no test coverage detected