{"record":{"id":"22d659514f71dfdd","repo":"risingwavelabs/risingwave","slug":"a-stream-has-reached-the-end-but-some-other-stream","errorCode":null,"errorMessage":"a stream has reached the end but some other stream has not started yet","messagePattern":"a stream has reached the end but some other stream has not started yet","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/kv_log_store/serde.rs","lineNumber":1126,"sourceCode":"                }\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!(\n                        \"when any of the stream reaches the end, other stream should also reaches the end, but poll result: {:?}\",\n                        result\n                    ));\n                }\n            }\n        }\n        Ok(None)\n    }\n}\n\n#[cfg(test)]","sourceCodeStart":1108,"sourceCodeEnd":1144,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/serde.rs#L1108-L1144","documentation":"After all row streams reach their end (and each end followed a barrier), the KV log store reader asserts that every registered stream actually started. If `not_started_streams` is non-empty at the point one stream finished, the multi-stream replay is inconsistent: some streams produced no rows at all while others completed. The reader throws this error rather than silently returning a partial or empty replay.","triggerScenarios":"Finishing the read of a log-store range where one upstream stream contributed zero rows/barriers (never registered as started) while at least one other stream reached the end of its stream.","commonSituations":"A source/backfill failed to initialize and wrote nothing to the log store while sibling streams progressed; misconfigured epoch range; table fragment where one stream was dropped or recreated mid-replay.","solutions":["Inspect the source/actor for the stream that never started — check actor logs and ensure the upstream stream is running and writing barriers.","Verify the read epoch range covers the lifetime of all streams; extend the range so the not-started stream's initial rows/barriers fall inside it.","Recover the stream from a consistent snapshot/backfill so all streams restart from the same epoch.","If the stream was intentionally removed, recreate the reader with the correct stream set so it does not expect the removed stream."],"exampleFix":"// before: reader expecting a stream set that includes a stopped stream\nlet reader = log_store.reader(vec![stream_a, stream_b_removed]);\n// after: only track streams that are actually registered upstream\nlet reader = log_store.reader(vec![stream_a, stream_b]);","handlingStrategy":"validation","validationCode":"// Before building the reader, ensure every expected stream has produced at least one record\nlet all_started = expected_streams.iter().all(|s| log_store.stream_started(s, start_epoch));\nassert!(all_started, \"all streams must have started before replay\");","typeGuard":"// Rust\nfn streams_consistent(not_started: &[StreamId], ended: &[StreamId]) -> bool {\n    not_started.is_empty() || ended.is_empty()\n}","tryCatchPattern":"// Rust\nif let Err(e) = reader.collect_all().await {\n    if e.to_string().contains(\"has not started yet\") {\n        // restart the missing upstream stream, then re-run replay\n        restart_missing_stream_and_retry().await?;\n    } else { return Err(e); }\n}","preventionTips":["Ensure backfill/source initialization writes an initial record for every stream before normal processing","Keep the reader's expected stream set in sync with actually-registered upstream streams","Recover all streams from a common snapshot epoch","Alert when an actor writes no rows/barriers while its siblings progress"],"tags":["streaming","log-store","recovery","inconsistent-state"],"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"}