Repository: risingwavelabs/risingwave
File: src/stream/src/common/log_store_impl/kv_log_store/buffer.rs (Rust)

Code before:
        while let Some((epoch, _)) = inner.truncation_list.front()
            && *epoch < curr_epoch
        {
            ret = inner.truncation_list.pop_front();
        }
        ret
    }

    pub(crate) async fn await_barrier_truncation(
        &self,
        curr_epoch: u64,
    ) -> ReaderTruncationOffsetType {
        let mut ret = None;
        loop {
            {
                let mut inner = self.buffer.inner();
                while let Some((epoch, _)) = inner.truncation_list.front()
                    && *epoch <= curr_epoch
                {
                    ret = inner.truncation_list.pop_front();
                }
            }
            if let Some((epoch, None)) = ret
                && epoch == curr_epoch
            {
                return (epoch, None);
            }

            let notified = self.truncate_notify.notified();
            notified
                .instrument_await("Wait For Barrier Truncation")
                .await;
        }
    }

    pub(crate) fn flush_all_unflushed(
        &mut self,
        mut flush_fn: impl FnMut(&StreamChunk, u64, SeqId, SeqId) -> LogStoreResult<()>,
    ) -> LogStoreResult<()> {
        let mut inner_guard = self.buffer.inner();
        let inner = inner_guard.deref_mut();
        for (epoch, item) in inner
            .unconsumed_queue
            .iter_mut()
            .chain(inner.consumed_queue.iter_mut())
        {
            if let LogStoreBufferItem::StreamChunk {
                chunk,
                start_seq_id,
                end_seq_id,
                flushed,

Review comment (suggestion):
The `await_barrier_truncation` method may loop indefinitely if a truncation with `Some(seq_id)` (non-None) is received for the same epoch. The condition at line 339-342 only returns when `ret` is `Some((epoch, None))` and `epoch == curr_epoch`, but if a truncation with a specific `seq_id` arrives for `curr_epoch`, the loop will continue waiting indefinitely.

**Recommendation**: Consider whether truncations with non-None `seq_id` should also satisfy the condition, or add timeout/error handling to prevent indefinite waiting.
```suggestion
            if let Some((epoch, seq_id_opt)) = ret
                && epoch == curr_epoch
            {
                return (epoch, seq_id_opt);
```

Code after:
        while let Some((epoch, _)) = inner.truncation_list.front()
            && *epoch < curr_epoch
        {
            ret = inner.truncation_list.pop_front();
        }
        ret
    }

    pub(crate) async fn wait_for_barrier_truncation(
        &self,
        curr_epoch: u64,
    ) -> LogStoreResult<ReaderTruncationOffsetType> {
        loop {
            let notified = self.truncate_notify.notified();

            {
                let mut inner = self.buffer.inner();
                while let Some((epoch, seq_id)) = inner.truncation_list.pop_front() {
                    if epoch > curr_epoch {
                        // TODO: should panic, after we confirm the correctness
                        return Err(anyhow::anyhow!(
                            "truncation epoch {} should not be larger than current epoch {}",
                            epoch,
                            curr_epoch
                        ));
                    }
                    if epoch == curr_epoch && seq_id.is_none() {
                        return Ok((epoch, seq_id));
                    }
                }
            }

            notified
                .instrument_await("Wait For Barrier Truncation")
                .await;
        }
    }

    pub(crate) fn flush_all_unflushed(
        &mut self,
        mut flush_fn: impl FnMut(&StreamChunk, u64, SeqId, SeqId) -> LogStoreResult<()>,
    ) -> LogStoreResult<()> {
        let mut inner_guard = self.buffer.inner();
        let inner = inner_guard.deref_mut();
        for (epoch, item) in inner
            .unconsumed_queue
            .iter_mut()
            .chain(inner.consumed_queue.iter_mut())
        {
            if let LogStoreBufferItem::StreamChunk {
                chunk,