{"record":{"id":"73cfaf10cf96d335","repo":"risingwavelabs/risingwave","slug":"exchange-executor-should-not-have-children","errorCode":null,"errorMessage":"Exchange executor should not have children!","messagePattern":"Exchange executor should not have children!","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/generic_exchange.rs","lineNumber":154,"sourceCode":"                .inspect_err(|e| {\n                    if matches!(e, BatchError::RpcError(_)) {\n                        mask_failed_serving_worker()\n                    }\n                })?,\n            ))\n        }\n    }\n}\n\npub struct GenericExchangeExecutorBuilder {}\n\nimpl BoxedExecutorBuilder for GenericExchangeExecutorBuilder {\n    async fn new_boxed_executor(\n        source: &ExecutorBuilder<'_>,\n        inputs: Vec<BoxedExecutor>,\n    ) -> Result<BoxedExecutor> {\n        ensure!(\n            inputs.is_empty(),\n            \"Exchange executor should not have children!\"\n        );\n        let node = try_match_expand!(\n            source.plan_node().get_node_body().unwrap(),\n            NodeBody::Exchange\n        )?;\n\n        let sequential = node.get_sequential();\n\n        ensure!(!node.get_sources().is_empty());\n        let proto_sources: Vec<PbExchangeSource> = node.get_sources().clone();\n        let source_creators =\n            vec![DefaultCreateSource::new(source.context().client_pool()); proto_sources.len()];\n\n        let input_schema: Vec<NodeField> = node.get_input_schema().clone();\n        let fields = input_schema.iter().map(Field::from).collect::<Vec<Field>>();\n        Ok(Box::new(ExchangeExecutor {\n            proto_sources,","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/generic_exchange.rs#L136-L172","documentation":"Same monotonicity rule as the buffered-path truncation, but for the historical-data path: when the requested truncation offset is a `TruncateOffset::Barrier { epoch }` covering pre-current-epoch data, the reader rejects truncating at or before the previously recorded truncate offset. Historical truncation is applied at barrier/epoch granularity via `rx.truncate_historical(epoch)`.","triggerScenarios":"Calling `KvLogStoreReader::truncate(TruncateOffset::Barrier { epoch })` while `offset <= self.truncate_offset` and `offset.epoch() < first_write_epoch` (historical region). Thrown at reader.rs:543.","commonSituations":"Restarted stream actor replays an old barrier epoch; epoch watermark regression after meta failover; mismatch between the epoch the executor believes it has consumed and what the reader last truncated.","solutions":["Ensure the caller tracks the last truncated barrier epoch and skips repeats or regressions.","Persist/restore truncate progress correctly across restarts so the first post-recovery truncate is newer.","Audit the epoch source (barrier manager / Hummock watermark) for out-of-order barrier delivery.","If the regression is expected (e.g. testing), reset or recreate the reader instead of re-truncating."],"exampleFix":"// before\nreader.truncate(TruncateOffset::Barrier { epoch });\n// after\nif !matches!(reader.last_truncate_offset(), Some(prev) if TruncateOffset::Barrier { epoch } <= prev) {\n    reader.truncate(TruncateOffset::Barrier { epoch });\n}","handlingStrategy":"validation","validationCode":"// rust\nfn safe_truncate_barrier(reader: &mut KvLogStoreReader, epoch: u64) -> anyhow::Result<()> {\n    let offset = TruncateOffset::Barrier { epoch };\n    if let Some(prev) = reader.last_truncate_offset() {\n        anyhow::ensure!(offset > prev, \"skip historical truncate {:?} (prev {:?})\", offset, prev);\n    }\n    reader.truncate(offset)\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Only issue historical truncations for strictly newer barrier epochs.","Persist the last truncated barrier epoch with actor state.","Monitor barrier epoch ordering from the barrier manager."],"tags":["log-store","streaming","monotonic-offset","barrier"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}