paxakos 0.13.0

//! Defines the applied log entry [`Buffer`] trait.
//!
//! When a node is temporarily shut down or becomes unreachable, it needs to be
//! catch back up with the rest of the cluster. One way to do that is to poll
//! other nodes for entries that have been committed in the meantime. These
//! other nodes will in turn query their applied entry buffer for these entries.
//!
//! A node may fall so far behind that no other node can catch it up through its
//! `Buffer`. In such a case the node must ask another node to create a snapshot
//! and catch up that way. It should be noted that the cluster will likely
//! progress while the snapshot is taken, transferred and installed. That is to
//! say that `Buffer`s are indispensable and should have a reasonable minimal
//! capacity.
//!
//! The default implementation is [`InMemoryBuffer`].
use std::collections::VecDeque;
use std::convert::Infallible;
use std::sync::Arc;

use crate::CoordNum;
use crate::LogEntry;
use crate::RoundNum;

/// A buffer of entries that were applied to the local state.
// TODO provide an (optional) persistent implementation
pub trait Buffer: Send + 'static {
    /// The round number type.
    type RoundNum: RoundNum;
    /// The coordination number type.
    type CoordNum: CoordNum;
    /// The log entry type.
    type Entry: LogEntry;
    /// The type of error this buffer may raise.
    type Error: std::error::Error + Send + Sync + 'static;

    /// Insert an applied entry into the buffer.
    fn insert(
        &mut self,
        round_num: Self::RoundNum,
        coord_num: Self::CoordNum,
        entry: Arc<Self::Entry>,
    );

    /// Retrieve an entry from the buffer.
    ///
    /// May be called for round numbers that were not previously inserted.
    #[allow(clippy::type_complexity)]
    fn get(
        &self,
        round_num: Self::RoundNum,
    ) -> Result<Option<(Self::CoordNum, Arc<Self::Entry>)>, Self::Error>;
}

/// An in-memory buffer.
pub struct InMemoryBuffer<R, C, E: LogEntry> {
    capacity: usize,
    buffer: VecDeque<(R, C, Arc<E>)>,
}

impl<R, C, E: LogEntry> InMemoryBuffer<R, C, E> {
    /// Creates a new in-memory buffer with the given capacity.
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0);

        Self {
            capacity,
            buffer: VecDeque::new(),
        }
    }
}

impl<R, C, E> Buffer for InMemoryBuffer<R, C, E>
where
    R: RoundNum,
    C: CoordNum,
    E: LogEntry,
{
    type RoundNum = R;
    type CoordNum = C;
    type Entry = E;
    type Error = Infallible;

    fn insert(
        &mut self,
        round_num: Self::RoundNum,
        coord_num: Self::CoordNum,
        entry: Arc<Self::Entry>,
    ) {
        if self.capacity <= self.buffer.len() {
            self.buffer.pop_front();
        }

        self.buffer.push_back((round_num, coord_num, entry));
    }

    #[allow(clippy::type_complexity)]
    fn get(
        &self,
        round_num: Self::RoundNum,
    ) -> Result<Option<(Self::CoordNum, Arc<Self::Entry>)>, Self::Error> {
        // hot path
        if let Some((r, _, _)) = self.buffer.get(0) {
            if round_num >= *r {
                if let Some(delta) = crate::util::try_usize_delta(round_num, *r) {
                    if let Some((r, c, e)) = self.buffer.get(delta) {
                        if *r == round_num {
                            return Ok(Some((*c, Arc::clone(e))));
                        }
                    }
                }
            }
        }

        // cold path
        Ok(self
            .buffer
            .iter()
            .find(|(r, _, _)| *r == round_num)
            .map(|(_, c, e)| (*c, Arc::clone(e))))
    }
}