a stream has reached the end but some other stream has not s

Error message

a stream has reached the end but some other stream has not started yet

What it means

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.

Source

Thrown at src/stream/src/common/log_store_impl/kv_log_store/serde.rs:1126

                }
            }
        }
        // End of stream
        match &self.stream_state {
            StreamState::BarrierEmitted { .. } => {}
            s => {
                return Err(anyhow!(
                    "when any of the stream reaches the end, it should be right after emitting an barrier. Current state: {:?}",
                    s
                ));
            }
        }
        assert!(
            self.barrier_streams.is_empty(),
            "should not have any pending barrier received stream after barrier emit"
        );
        if !self.not_started_streams.is_empty() {
            return Err(anyhow!(
                "a stream has reached the end but some other stream has not started yet"
            ));
        }
        if cfg!(debug_assertions) {
            while let Some((opt, _stream)) = self.row_streams.next().await {
                if let Some(result) = opt {
                    return Err(anyhow!(
                        "when any of the stream reaches the end, other stream should also reaches the end, but poll result: {:?}",
                        result
                    ));
                }
            }
        }
        Ok(None)
    }
}

#[cfg(test)]

View on GitHub (pinned to 6469eb736d)

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.

Example fix

// before: reader expecting a stream set that includes a stopped stream
let reader = log_store.reader(vec![stream_a, stream_b_removed]);
// after: only track streams that are actually registered upstream
let reader = log_store.reader(vec![stream_a, stream_b]);

Defensive patterns

Strategy: validation

Validate before calling

// Before building the reader, ensure every expected stream has produced at least one record
let all_started = expected_streams.iter().all(|s| log_store.stream_started(s, start_epoch));
assert!(all_started, "all streams must have started before replay");

Type guard

// Rust
fn streams_consistent(not_started: &[StreamId], ended: &[StreamId]) -> bool {
    not_started.is_empty() || ended.is_empty()
}

Try / catch

// Rust
if let Err(e) = reader.collect_all().await {
    if e.to_string().contains("has not started yet") {
        // restart the missing upstream stream, then re-run replay
        restart_missing_stream_and_retry().await?;
    } else { return Err(e); }
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/22d659514f71dfdd. Report an issue: GitHub.