| 1280 | } |
| 1281 | |
| 1282 | PJ::Status DerivedEngine::recomputeBatchLocked(PJ::NodeId node_id) { |
| 1283 | auto it = impl_->nodes.find(node_id); |
| 1284 | if (it == impl_->nodes.end()) { |
| 1285 | return PJ::unexpected(fmt::format("recompute_batch: node {} not found", node_id)); |
| 1286 | } |
| 1287 | |
| 1288 | // Recompute the target node itself first. If it fails, its downstream inputs are bad — |
| 1289 | // stop and surface the error (don't cascade onto an empty/failed upstream). |
| 1290 | PJ::Status self = recomputeNodeSelfOnly(*impl_, engine_, it.value()); |
| 1291 | if (!self.has_value()) { |
| 1292 | return self; |
| 1293 | } |
| 1294 | |
| 1295 | // Cascade: a rewritten output must propagate to every chained filter downstream (a filter |
| 1296 | // of a filter), else editing/reloading the upstream leaves the downstream stale. Collect |
| 1297 | // the transitive downstream set (BFS over downstream_of), then recompute each in |
| 1298 | // topological order via a FULL reset+replay — never the incremental path, whose watermark |
| 1299 | // would skip the rewritten upstream output. Mirror scheduleActive's resilience: keep |
| 1300 | // cascading the rest of the set and return only the first error. |
| 1301 | tsl::robin_set<PJ::NodeId> reachable; |
| 1302 | std::queue<PJ::NodeId> bfs; |
| 1303 | bfs.push(node_id); |
| 1304 | while (!bfs.empty()) { |
| 1305 | PJ::NodeId curr = bfs.front(); |
| 1306 | bfs.pop(); |
| 1307 | auto dit = impl_->downstream_of.find(curr); |
| 1308 | if (dit == impl_->downstream_of.end()) { |
| 1309 | continue; |
| 1310 | } |
| 1311 | for (PJ::NodeId down : dit->second) { |
| 1312 | if (impl_->nodes.contains(down) && reachable.insert(down).second) { |
| 1313 | bfs.push(down); |
| 1314 | } |
| 1315 | } |
| 1316 | } |
| 1317 | if (reachable.empty()) { |
| 1318 | return PJ::okStatus(); |
| 1319 | } |
| 1320 | |
| 1321 | PJ::Status first_error = PJ::okStatus(); |
| 1322 | for (PJ::NodeId nid : topologicalOrder()) { |
| 1323 | if (!reachable.contains(nid)) { |
| 1324 | continue; |
| 1325 | } |
| 1326 | PJ::Status s = recomputeNodeSelfOnly(*impl_, engine_, impl_->nodes.at(nid)); |
| 1327 | if (!s.has_value() && first_error.has_value()) { |
| 1328 | first_error = std::move(s); // remember only the first downstream error |
| 1329 | } |
| 1330 | } |
| 1331 | return first_error; |
| 1332 | } |
| 1333 | |
| 1334 | // --------------------------------------------------------------------------- |
| 1335 | // replace_siso_transform |