openraft 0.10.0-alpha.35

use std::fmt::Debug;
use std::io;
use std::ops::RangeBounds;
use std::ops::RangeInclusive;

use futures_util::Stream;
use openraft_macros::add_async_trait;
use openraft_macros::since;

use crate::OptionalSend;
use crate::OptionalSync;
use crate::RaftTypeConfig;
use crate::base::BoxStream;
use crate::engine::LogIdList;
use crate::errors::LeaderChanged;
use crate::type_config::alias::CommittedLeaderIdOf;
use crate::type_config::alias::EntryOf;
use crate::type_config::alias::LeaderIdOf;
use crate::type_config::alias::LogIdOf;
use crate::type_config::alias::VoteOf;
use crate::vote::RaftVote;

/// Error that can occur when reading log entries from a leader-bounded stream.
///
/// This error is returned by [`RaftLogReader::leader_bounded_stream`] when either:
/// - The leader has changed ([`LeaderChanged`])
/// - An I/O error occurred during log reading
#[derive(Debug, thiserror::Error)]
pub enum LeaderBoundedStreamError<C>
where C: RaftTypeConfig
{
    /// The leader has changed, making the stream invalid.
    #[error(transparent)]
    LeaderChanged(#[from] LeaderChanged<C>),

    /// An I/O error occurred while reading log entries.
    #[error(transparent)]
    IoError(#[from] io::Error),
}

/// Result type for [`RaftLogReader::leader_bounded_stream`] stream items.
pub type LeaderBoundedStreamResult<C> = Result<EntryOf<C>, LeaderBoundedStreamError<C>>;

/// Result type for [`RaftLogReader::entries_stream`] stream items.
pub type EntriesStreamResult<C> = Result<EntryOf<C>, io::Error>;

/// A trait defining the interface for a Raft log subsystem.
///
/// This interface is accessed read-only by replication sub-task: `ReplicationCore`.
///
/// A log reader must also be able to read the last saved vote by [`RaftLogStorage::save_vote`],
/// See: [log-stream](`crate::docs::protocol::replication::log_stream`).
///
/// Typically, the log reader implementation as such will be hidden behind an `Arc<T>` and
/// this interface implemented on the `Arc<T>`. It can be co-implemented with [`RaftLogStorage`]
/// interface on the same cloneable object if the underlying state machine is anyway synchronized.
///
/// [`RaftLogStorage`]: crate::storage::RaftLogStorage
/// [`RaftLogStorage::save_vote`]: crate::storage::RaftLogStorage::save_vote
#[add_async_trait]
pub trait RaftLogReader<C>: OptionalSend + OptionalSync + 'static
where C: RaftTypeConfig
{
    /// Read log entries as a stream, conditional on vote state.
    ///
    /// Returns log entries within the specified range as a stream, but only if:
    /// - The current vote belongs to the given `leader`
    /// - The vote is committed (i.e., [`Self::read_vote`]`.await?.is_committed()`)
    ///
    /// # Important
    ///
    /// The stream must stop yielding entries immediately and return [`LeaderChanged`] error if the
    /// leader changes during iteration, as log truncation may occur and entries may no longer
    /// be consecutive.
    ///
    /// # Arguments
    ///
    /// * `leader` - The expected leader ID. Entries are only returned if the vote belongs to this
    ///   leader.
    /// * `range` - The range of log indices to read.
    ///
    /// # Returns
    ///
    /// Returns a stream of log entries wrapped in `Result`. If the vote doesn't match the leader,
    /// isn't committed, or changes during the read, returns a stream yielding a single
    /// [`LeaderChanged`] error.
    ///
    /// # Default Implementation
    ///
    /// The default implementation is a very simple wrapper of [`Self::try_get_log_entries`], and
    /// application should implement it for better performance.
    ///
    /// The default impl:
    /// 1. Checks if the stored vote matches the given leader and is committed
    /// 2. If conditions are met, calls [`Self::try_get_log_entries`] to get all entries
    /// 3. Verifies the vote hasn't changed after reading entries
    /// 4. Returns stream with [`LeaderChanged`] error if vote doesn't match or changed, otherwise
    ///    returns entries as stream
    #[since(version = "0.10.0")]
    async fn leader_bounded_stream<RB>(
        &mut self,
        leader: LeaderIdOf<C>,
        range: RB,
    ) -> impl Stream<Item = LeaderBoundedStreamResult<C>> + OptionalSend
    where
        RB: RangeBounds<u64> + Clone + Debug + OptionalSend,
    {
        use futures_util::stream;

        let changed_err = |leader, vote| {
            let err = LeaderChanged::new(leader, vote);
            LeaderBoundedStreamError::LeaderChanged(err)
        };

        let fu = async move {
            let vote = self.read_vote().await?;

            let Some(vote) = vote else {
                return Ok::<BoxStream<_>, _>(Box::pin(stream::iter([Err(changed_err(leader, None))])));
            };

            if vote.leader_id() == &leader && vote.is_committed() {
                // valid vote, continue
            } else {
                return Ok::<BoxStream<_>, _>(Box::pin(stream::iter([Err(changed_err(leader, Some(vote)))])));
            }

            let entries = self.try_get_log_entries(range).await?;

            let current_vote = self.read_vote().await?;

            if current_vote.as_ref() != Some(&vote) {
                // Vote has changed, return empty stream
                return Ok::<BoxStream<_>, _>(Box::pin(stream::iter([Err(changed_err(leader, current_vote))])));
            }

            // Vote unchanged
            let stream = stream::iter(entries.into_iter().map(Ok));
            Ok::<BoxStream<_>, io::Error>(Box::pin(stream))
        };

        match fu.await {
            Ok(strm) => strm,
            Err(io_err) => Box::pin(stream::iter([Err(LeaderBoundedStreamError::IoError(io_err))])),
        }
    }

    /// Read log entries as a stream without leader validation.
    ///
    /// Returns log entries within the specified range as a stream. Unlike
    /// [`leader_bounded_stream`](Self::leader_bounded_stream), this method does not check vote
    /// state or leader validity.
    ///
    /// This method is primarily used for reading committed log entries that will no longer be
    /// changed, such as when applying logs to the state machine. It can also be used during
    /// initialization or when the leader context is not relevant.
    ///
    /// # Arguments
    ///
    /// * `range` - The range of log indices to read.
    ///
    /// # Returns
    ///
    /// Returns a stream of log entries wrapped in `Result`. I/O errors during reading are
    /// propagated through the stream.
    ///
    /// # Default Implementation
    ///
    /// The default implementation calls [`Self::try_get_log_entries`] to get all entries and
    /// converts them to a stream. Applications may implement this for better performance, such as
    /// streaming entries incrementally.
    #[since(version = "0.10.0")]
    async fn entries_stream<RB>(&mut self, range: RB) -> impl Stream<Item = EntriesStreamResult<C>> + OptionalSend
    where RB: RangeBounds<u64> + Clone + Debug + OptionalSend {
        use futures_util::stream;

        let fu = async move {
            let entries = self.try_get_log_entries(range).await?;

            let stream = stream::iter(entries.into_iter().map(Ok));
            Ok::<BoxStream<_>, io::Error>(Box::pin(stream))
        };

        match fu.await {
            Ok(strm) => strm,
            Err(io_err) => Box::pin(stream::iter([Err(io_err)])),
        }
    }

    /// Get a series of log entries from storage.
    ///
    /// This method retrieves log entries within the specified range. The range is defined by
    /// any type that implements `RangeBounds<u64>`, such as `start..end` or `start..=end`.
    ///
    /// ### Correctness requirements
    ///
    /// - Every entry in the range that the log **contains** must be returned. Unlike
    ///   [`limited_get_log_entries()`](Self::limited_get_log_entries), which may return only the
    ///   first several log entries, this method must not stop early.
    ///
    /// - The absence of an entry is tolerated **only at the beginning or end of the range, and only
    ///   for entries the log genuinely does not contain**:
    ///   - at the beginning: entries already removed by [`RaftLogStorage::purge`];
    ///   - at the end: entries not yet stored by [`RaftLogStorage::append`].
    ///
    ///   This tolerance does **not** permit omitting an entry that has already been appended but is
    ///   not yet visible to this reader handle (for example, a read-after-write or
    ///   read-after-flush visibility lag in the storage implementation).
    ///   [`RaftLogStorage::append`] requires appended entries to be readable the moment it returns
    ///   — even before its flush callback fires — so once the log "contains" an entry, every reader
    ///   handle must return it. Omitting such an entry is a storage contract violation, not a
    ///   tolerated boundary short read, and openraft may treat it as a fatal error (committed
    ///   entries are read back through this method on the apply path).
    ///
    ///   Missing entries *within* the range (i.e., holes) are never permitted and must result in an
    ///   error.
    ///
    /// - The read operation must be atomic. That is, it should not reflect any state changes that
    ///   occur after the read operation has started.
    ///
    /// [`RaftLogStorage::append`]: crate::storage::RaftLogStorage::append
    /// [`RaftLogStorage::purge`]: crate::storage::RaftLogStorage::purge
    async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug + OptionalSend>(
        &mut self,
        range: RB,
    ) -> Result<Vec<C::Entry>, io::Error>;

    /// Return the last saved vote by [`RaftLogStorage::save_vote`].
    ///
    /// A log reader must also be able to read the last saved vote by [`RaftLogStorage::save_vote`],
    /// See: [log-stream](`crate::docs::protocol::replication::log_stream`)
    ///
    /// [`RaftLogStorage::save_vote`]: crate::storage::RaftLogStorage::save_vote
    async fn read_vote(&mut self) -> Result<Option<VoteOf<C>>, io::Error>;

    /// Returns log entries within range `[start, end)`, `end` is exclusive,
    /// potentially limited by implementation-defined constraints.
    ///
    /// If the specified range is too large, the implementation may return only the first few log
    /// entries to ensure the result is not excessively large.
    ///
    /// It must not return an empty result if the input range is not empty.
    ///
    /// The default implementation just returns the full range of log entries.
    #[since(version = "0.10.0")]
    async fn limited_get_log_entries(&mut self, start: u64, end: u64) -> Result<Vec<C::Entry>, io::Error> {
        self.try_get_log_entries(start..end).await
    }

    /// Retrieves a list of key log ids that mark the beginning of each Leader.
    ///
    /// This method returns log entries that represent leadership transitions in the log history,
    /// including
    /// - The first log entry in the storage (regardless of Leader);
    /// - The first log entry from each Leader;
    /// - The last log entry in the storage (regardless of Leader);
    ///
    /// # Example
    ///
    /// Given:
    /// Log entries: `[(2,2), (2,3), (5,4), (5,5)]` (format: `(term, index)`)
    ///
    /// Returns: `[(2,2), (5,4), (5,5)]`
    ///
    /// # Usage
    ///
    /// This method is called only during node startup to build an initial log index.
    ///
    /// # Implementation Notes
    ///
    /// - Optional method: If your [`RaftLogStorage`] implementation doesn't maintain this
    ///   information, do not implement it and use the default implementation.
    /// - Default implementation: Uses a binary search algorithm to find key log entries
    /// - Time complexity: `O(k * log(n))` where:
    ///   - `k` = average number of unique Leaders
    ///   - `n` = average number of logs per Leader
    ///
    /// # Arguments
    ///
    /// - `range`: range of the log id to return, inclusive. Such as `(1, 10)..=(2, 20)`.
    ///
    /// # Returns
    ///
    /// Returns a vector of log entries marking leadership transitions and boundaries.
    ///
    /// [`RaftLogStorage`]: crate::storage::RaftLogStorage
    #[since(version = "0.10.0")]
    async fn get_key_log_ids(&mut self, range: RangeInclusive<LogIdOf<C>>) -> Result<Vec<LogIdOf<C>>, io::Error> {
        LogIdList::<CommittedLeaderIdOf<C>>::get_key_log_ids(range, self)
            .await
            .map_err(|e| io::Error::other(e.to_string()))
    }
}

#[cfg(test)]
mod tests {
    use std::collections::VecDeque;

    use futures_util::StreamExt;

    use super::LeaderBoundedStreamError;
    use super::RaftLogReader;
    use crate::engine::testing::UTConfig;
    use crate::engine::testing::log_id;
    use crate::entry::RaftEntry;
    use crate::errors::LeaderChanged;
    use crate::type_config::TypeConfigExt;
    use crate::type_config::alias::EntryOf;
    use crate::type_config::alias::VoteOf;

    struct ChangingVoteReader {
        votes: VecDeque<Option<VoteOf<UTConfig>>>,
        entries: Vec<EntryOf<UTConfig>>,
        calls: Vec<&'static str>,
    }

    impl ChangingVoteReader {
        fn new(votes: [Option<VoteOf<UTConfig>>; 2], entries: Vec<EntryOf<UTConfig>>) -> Self {
            Self {
                votes: VecDeque::from(votes),
                entries,
                calls: vec![],
            }
        }
    }

    impl RaftLogReader<UTConfig> for ChangingVoteReader {
        async fn try_get_log_entries<RB>(&mut self, _range: RB) -> Result<Vec<EntryOf<UTConfig>>, std::io::Error>
        where RB: std::ops::RangeBounds<u64> + Clone + std::fmt::Debug + crate::OptionalSend {
            self.calls.push("read_entries");
            Ok(self.entries.clone())
        }

        async fn read_vote(&mut self) -> Result<Option<VoteOf<UTConfig>>, std::io::Error> {
            self.calls.push("read_vote");
            Ok(self.votes.pop_front().unwrap_or(None))
        }
    }

    #[test]
    fn leader_bounded_stream_stops_when_vote_changes_between_reads() {
        UTConfig::<()>::run(async {
            let expected = VoteOf::<UTConfig>::new_committed(3, 1);
            let changed = VoteOf::<UTConfig>::new(4, 2);
            let leader = *expected.leader_id();
            let entry = EntryOf::<UTConfig>::new_blank(log_id(3, 1, 7));

            let mut reader = ChangingVoteReader::new([Some(expected), Some(changed)], vec![entry]);

            let stream = reader.leader_bounded_stream(leader, 7..8).await;
            let items: Vec<_> = stream.collect().await;

            let [Err(LeaderBoundedStreamError::LeaderChanged(actual))] = items.as_slice() else {
                panic!("expected one LeaderChanged item, got: {items:?}");
            };

            assert_eq!(&LeaderChanged::new(leader, Some(changed)), actual);
            assert_eq!(vec!["read_vote", "read_entries", "read_vote"], reader.calls);
        })
    }
}