when any of the stream reaches the end, it should be right…

Error message

when any of the stream reaches the end, it should be right after emitting an barrier. Current state: {:?}

What it means

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.

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`.

Example fix

// before: reading with an epoch range that excludes the final barrier epoch
let reader = log_store.reader(start_epoch, end_epoch_excl_barrier);
// after: include the epoch containing the last emitted barrier
let reader = log_store.reader(start_epoch, end_epoch_incl_barrier);

Defensive patterns

Strategy: try-catch

Validate before calling

// Before replaying, verify the log range ends at a barrier epoch
assert!(log_store.range_ends_with_barrier(start_epoch, end_epoch), "log range must end right after a barrier");

Type guard

// Rust: match on state before trusting the replay result
fn replay_finished_cleanly(state: &StreamState) -> bool {
    matches!(state, StreamState::BarrierEmitted { .. })
}

Try / catch

// Rust
match reader.next().await {
    Ok(item) => process(item),
    Err(e) if e.to_string().contains("right after emitting an barrier") => {
        // mark log segment as corrupt, trigger re-snapshot / alert
        alert_corrupt_log_segment(&e);
    }
    Err(e) => return Err(e),
}

Prevention

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

When it happens

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

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

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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

Appendix: source

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

                                aligned_vnodes.set(vnode.to_index(), true);
                                *other = StreamState::BarrierAligning {
                                    aligned_vnodes,
                                    read_size: size,
                                    curr_epoch: decoded_epoch,
                                    is_checkpoint,
                                };
                            }
                        }
                        continue;
                    }
                }
            }
        }
        // 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!(

View on GitHub (pinned to 6469eb736d)