databendlabs / openraft / 19132305149

88%

Build Type

push

github

Committed by drmingdrmer

Commit Message

change: Migrate `RaftStateMachine::apply()` from Iterator to Stream

Update the `apply()` method signature to use `Stream` instead of `IntoIterator`
for better error handling and async streaming support:

- Change `IntoIterator<Item = EntryResponder<C>>` to
  `Stream<Item = Result<EntryResponder<C>, io::Error>>`
- All implementations now use `try_next().await?` for cleaner error handling
- Update all 10+ implementations across examples and benchmarks

Add leader-bounded streaming to `RaftLogReader`:
- New `leader_bounded_stream()` method that reads entries conditional on
  vote state, ensuring leader hasn't changed during iteration
- New `entries_stream()` method for unconditional streaming
- Add `LeaderChanged` and `LeaderBoundedStreamError` error types
- Add comprehensive test coverage for vote consistency checks

Benefits:
- Simpler error propagation with `try_next().await?` pattern
- Better support for async iteration and backpressure
- Safer log reading with leader change detection

- Fix: #1472

Upgrade tip:

For `RaftStateMachine::apply()` implementations:

```rust
// Before:
async fn apply<I>(&mut self, entries: I) -> Result<(), io::Error>
where
    I: IntoIterator<Item = EntryResponder<TypeConfig>> + Send,
    I::IntoIter: Send,
{
    for (entry, responder) in entries {
        // process entry
    }
    Ok(())
}

// After:
async fn apply<Strm>(&mut self, mut entries: Strm) -> Result<(), io::Error>
where Strm: Stream<Item = Result<EntryResponder<TypeConfig>, io::Error>> + Unpin + OptionalSend
{
    use futures::TryStreamExt;

    while let Some((entry, responder)) = entries.try_next().await? {
        // process entry
    }
    Ok(())
}
```

Run Details

147 of 157 new or added lines in 7 files covered. (93.63%)

2 existing lines in 2 files now uncovered.

12771 of 13974 relevant lines covered (91.39%)

133078.07 hits per line