{"record":{"id":"9af69a632409ef79","repo":"risingwavelabs/risingwave","slug":"when-any-of-the-stream-reaches-the-end-it-should","errorCode":null,"errorMessage":"when any of the stream reaches the end, it should be right after emitting an barrier. Current state: {:?}","messagePattern":"when any of the stream reaches the end, it should be right after emitting an barrier\\. Current state: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/kv_log_store/serde.rs","lineNumber":1115,"sourceCode":"                                aligned_vnodes.set(vnode.to_index(), true);\n                                *other = StreamState::BarrierAligning {\n                                    aligned_vnodes,\n                                    read_size: size,\n                                    curr_epoch: decoded_epoch,\n                                    is_checkpoint,\n                                };\n                            }\n                        }\n                        continue;\n                    }\n                }\n            }\n        }\n        // End of stream\n        match &self.stream_state {\n            StreamState::BarrierEmitted { .. } => {}\n            s => {\n                return Err(anyhow!(\n                    \"when any of the stream reaches the end, it should be right after emitting an barrier. Current state: {:?}\",\n                    s\n                ));\n            }\n        }\n        assert!(\n            self.barrier_streams.is_empty(),\n            \"should not have any pending barrier received stream after barrier emit\"\n        );\n        if !self.not_started_streams.is_empty() {\n            return Err(anyhow!(\n                \"a stream has reached the end but some other stream has not started yet\"\n            ));\n        }\n        if cfg!(debug_assertions) {\n            while let Some((opt, _stream)) = self.row_streams.next().await {\n                if let Some(result) = opt {\n                    return Err(anyhow!(","sourceCodeStart":1097,"sourceCodeEnd":1133,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/serde.rs#L1097-L1133","documentation":"This error is raised by the KV log store's read path (`KvLogStoreReadVec`/stream reader) when a row stream reaches its end while the reader's per-stream state machine is not in the `BarrierEmitted` state. The log store replays epoch-ordered rows and barriers for multiple upstream streams, and by design any stream's end must occur immediately after a barrier was emitted; ending at any other point means the persisted log is incomplete or corrupted. It guards against silently producing a truncated replay of the stream's history.","triggerScenarios":"Iterating a KV log store reader to completion while `stream_state` is anything other than `StreamState::BarrierEmitted` — e.g. the last persisted record was a row (not a barrier), or stream state tracking diverged after a partial barrier write.","commonSituations":"Corrupted or truncated Hummock log data (e.g. recovery from a crash that cut off the barrier write), reading a log store range that stops mid-epoch, or bugs in barrier serialization in kv_log_store serde.","solutions":["Check the underlying KV (Hummock) data for the affected stream and verify the log ends with a serialized barrier; re-snapshot or re-ingest the affected range if data is truncated.","Verify the epoch range passed to the log store reader fully covers epochs written by the source; adjust so the final epoch containing the barrier is included.","Confirm all writers use the same kv_log_store serde version — mismatched versions can drop or misinterpret barrier records; align versions across nodes.","If reproducible on a healthy cluster, report as a bug with the log-store contents; this is an invariant violation in `src/stream/src/common/log_store_impl/kv_log_store/serde.rs`."],"exampleFix":"// before: reading with an epoch range that excludes the final barrier epoch\nlet reader = log_store.reader(start_epoch, end_epoch_excl_barrier);\n// after: include the epoch containing the last emitted barrier\nlet reader = log_store.reader(start_epoch, end_epoch_incl_barrier);","handlingStrategy":"try-catch","validationCode":"// Before replaying, verify the log range ends at a barrier epoch\nassert!(log_store.range_ends_with_barrier(start_epoch, end_epoch), \"log range must end right after a barrier\");","typeGuard":"// Rust: match on state before trusting the replay result\nfn replay_finished_cleanly(state: &StreamState) -> bool {\n    matches!(state, StreamState::BarrierEmitted { .. })\n}","tryCatchPattern":"// Rust\nmatch reader.next().await {\n    Ok(item) => process(item),\n    Err(e) if e.to_string().contains(\"right after emitting an barrier\") => {\n        // mark log segment as corrupt, trigger re-snapshot / alert\n        alert_corrupt_log_segment(&e);\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Always include the epoch containing the last barrier in log-store read ranges","Monitor for crash-recovery gaps between row writes and barrier writes in Hummock","Keep kv_log_store serde versions identical across all compute nodes","Add periodic integrity checks that each stream's log segment terminates with a barrier"],"tags":["streaming","log-store","state-machine","data-corruption"],"backgroundTag":"internal-invariant-violation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}