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

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.

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)

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.