commitlog.rs

use std::{
    fmt::Debug,
    io,
    marker::PhantomData,
    mem,
    ops::{Range, RangeBounds},
    vec,
};

use itertools::Itertools;
use log::{debug, error, info, trace, warn};

use crate::{
    commit::StoredCommit,
    error::{self, source_chain},
    index::IndexError,
    payload::Decoder,
    repo::{self, Repo, SegmentLen as _, TxOffsetIndex},
    segment::{self, FileLike, Transaction, Writer},
    Commit, Encode, Options, DEFAULT_LOG_FORMAT_VERSION,
};

pub use crate::segment::Committed;

/// A commitlog generic over the storage backend as well as the type of records
/// its [`Commit`]s contain.
#[derive(Debug)]
pub struct Generic<R: Repo, T> {
    /// The storage backend.
    pub(crate) repo: R,
    /// The segment currently being written to.
    ///
    /// If we squint, all segments in a log are a non-empty linked list, the
    /// head of which is the segment open for writing.
    pub(crate) head: Writer<R::SegmentWriter>,
    /// The tail of the non-empty list of segments.
    ///
    /// We only retain the min transaction offset of each, from which the
    /// segments can be opened for reading when needed.
    ///
    /// This is a `Vec`, not a linked list, so the last element is the newest
    /// segment (after `head`).
    tail: Vec<u64>,
    /// Configuration options.
    opts: Options,
    /// Type of a single record in this log's [`Commit::records`].
    _record: PhantomData<T>,
    /// Tracks panics/errors to control what happens on drop.
    ///
    /// Set to `true` before any I/O operation, and back to `false` after it
    /// succeeded. This way, we won't try to perform I/O on drop when it is
    /// unlikely to succeed, or even has a chance to panic.
    panicked: bool,
}

impl<R: Repo, T> Generic<R, T> {
    pub fn open(repo: R, opts: Options) -> io::Result<Self> {
        let mut tail = repo.existing_offsets()?;
        if !tail.is_empty() {
            debug!("segments: {tail:?}");
        }

        // Resume the last segment for writing, or
        // create a new segment starting from the last good commit + 1.
        let head = loop {
            if let Some(last) = tail.pop() {
                info!("repo {}: resuming last segment: {}", repo, last);
                match repo::resume_segment_writer(&repo, opts, last)? {
                    repo::ResumedSegment::Empty => {
                        repo.remove_segment(last)?;
                        continue;
                    }
                    repo::ResumedSegment::Resumed(writer) => break writer,
                    repo::ResumedSegment::Sealed(meta) | repo::ResumedSegment::Corrupted(meta) => {
                        tail.push(meta.tx_range.start);
                        break repo::create_segment_writer(&repo, opts, meta.max_epoch, meta.tx_range.end)?;
                    }
                }
            } else {
                info!("repo {}: starting fresh log", repo);
                break repo::create_segment_writer(&repo, opts, Commit::DEFAULT_EPOCH, 0)?;
            }
        };

        Ok(Self {
            repo,
            head,
            tail,
            opts,
            _record: PhantomData,
            panicked: false,
        })
    }

    /// Get the current epoch.
    ///
    /// See also: [`Commit::epoch`].
    pub fn epoch(&self) -> u64 {
        self.head.commit.epoch
    }

    /// Update the current epoch.
    ///
    /// Does nothing if the given `epoch` is equal to the current epoch.
    ///
    /// # Errors
    ///
    /// If `epoch` is smaller than the current epoch, an error of kind
    /// [`io::ErrorKind::InvalidInput`] is returned.
    pub fn set_epoch(&mut self, epoch: u64) -> io::Result<()> {
        if epoch < self.head.epoch() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "new epoch is smaller than current epoch",
            ));
        }
        self.head.set_epoch(epoch);
        Ok(())
    }

    /// Force the currently active segment to be flushed to storage.
    ///
    /// Using a filesystem backend, this means to call `fsync(2)`.
    ///
    /// **Note** that this does not flush the buffered data from calls to
    /// [Self::commit], it only instructs the underlying storage to flush its
    /// buffers. Call [Self::flush] prior to this method to ensure data from
    /// all previous [Self::commit] calls is flushed to the underlying storage.
    ///
    /// # Panics
    ///
    /// As an `fsync` failure leaves a file in a more of less undefined state,
    /// this method panics in this case, thereby preventing any further writes
    /// to the log and forcing the user to re-read the state from disk.
    pub fn sync(&mut self) {
        self.panicked = true;
        if let Err(e) = self.head.fsync() {
            panic!("Failed to fsync segment: {e}");
        }
        self.panicked = false;
    }

    /// Flush the buffered data from previous calls to [Self::commit] to the
    /// underlying storage.
    ///
    /// Call [Self::sync] to instruct the underlying storage to flush its
    /// buffers as well.
    pub fn flush(&mut self) -> io::Result<()> {
        self.head.flush()
    }

    /// Calls [Self::flush] and then [Self::sync].
    fn flush_and_sync(&mut self) -> io::Result<()> {
        self.flush()?;
        self.sync();
        Ok(())
    }

    /// The last transaction offset written to disk, or `None` if nothing has
    /// been written yet.
    ///
    /// Note that this does not imply durability: [`Self::sync`] may not have
    /// been called at this offset.
    pub fn max_committed_offset(&self) -> Option<u64> {
        // Naming is hard: the segment's `next_tx_offset` indicates how many
        // txs are already in the log (it's the next commit's min-tx-offset).
        // If the value is zero, however, the initial commit hasn't been
        // committed yet.
        self.head.next_tx_offset().checked_sub(1)
    }

    /// The first transaction offset written to disk, or `None` if nothing has
    /// been written yet.
    pub fn min_committed_offset(&self) -> Option<u64> {
        self.tail
            .first()
            .copied()
            .or_else(|| (!self.head.is_empty()).then(|| self.head.min_tx_offset()))
    }

    // Helper to obtain a list of the segment offsets which include transaction
    // offset `offset`.
    //
    // The returned `Vec` is sorted in **ascending** order, such that the first
    // element is the segment which contains `offset`.
    //
    // The offset of `self.head` is always included, regardless of how many
    // entries it actually contains.
    fn segment_offsets_from(&self, offset: u64) -> Vec<u64> {
        if offset >= self.head.min_tx_offset {
            vec![self.head.min_tx_offset]
        } else {
            let mut offs = Vec::with_capacity(self.tail.len() + 1);
            if let Some(pos) = self.tail.iter().rposition(|off| off <= &offset) {
                offs.extend_from_slice(&self.tail[pos..]);
                offs.push(self.head.min_tx_offset);
            }

            offs
        }
    }

    pub fn commits_from(&self, offset: u64) -> Commits<R> {
        let offsets = self.segment_offsets_from(offset);
        let segments = Segments {
            offs: offsets.into_iter(),
            repo: self.repo.clone(),
            max_log_format_version: self.opts.log_format_version,
        };
        Commits {
            inner: None,
            segments,
            last_commit: CommitInfo::Initial { next_offset: offset },
            last_error: None,
        }
    }

    pub fn reset(mut self) -> io::Result<Self> {
        info!("hard reset");

        self.panicked = true;
        self.tail.reserve(1);
        self.tail.push(self.head.min_tx_offset);
        for segment in self.tail.iter().rev() {
            debug!("removing segment {segment}");
            self.repo.remove_segment(*segment)?;
        }
        // Prevent finalizer from running by not updating self.panicked.

        Self::open(self.repo.clone(), self.opts)
    }

    pub fn reset_to(mut self, offset: u64) -> io::Result<Self> {
        info!("reset to {offset}");

        self.panicked = true;
        self.tail.reserve(1);
        self.tail.push(self.head.min_tx_offset);
        reset_to_internal(&self.repo, &self.tail, offset)?;
        // Prevent finalizer from running by not updating self.panicked.

        Self::open(self.repo.clone(), self.opts)
    }

    /// Start a new segment, preserving the current head's `Commit`.
    ///
    /// The caller must ensure that the current head is synced to disk as
    /// appropriate. It is not appropriate to sync after a write error, as that
    /// is likely to return an error as well: the `Commit` will be written to
    /// the new segment anyway.
    fn start_new_segment(&mut self) -> io::Result<&mut Writer<R::SegmentWriter>> {
        debug!(
            "starting new segment offset={} prev-offset={}",
            self.head.next_tx_offset(),
            self.head.min_tx_offset()
        );
        let new = repo::create_segment_writer(&self.repo, self.opts, self.head.epoch(), self.head.next_tx_offset())?;
        let old = mem::replace(&mut self.head, new);
        self.tail.push(old.min_tx_offset());
        self.head.commit = old.commit;

        Ok(&mut self.head)
    }
}

impl<R: Repo, T: Encode> Generic<R, T> {
    /// Write `transactions` to the log.
    ///
    /// This will store all `transactions` as a single [Commit]
    /// (note that `transactions` must not yield more than [u16::MAX] elements).
    ///
    /// Data is buffered by the underlying segment [Writer].
    /// Call [Self::flush] to force flushing to the underlying storage.
    ///
    /// If, after writing the transactions, the writer's total written bytes
    /// exceed [Options::max_segment_size], the current segment is flushed,
    /// `fsync`ed and closed, and a new segment is created.
    ///
    /// Returns `Ok(None)` if `transactions` was empty, otherwise [Committed],
    /// which contains the offset range and checksum of the commit.
    ///
    /// Note that supplying empty `transactions` may cause the current segment
    /// to be rotated.
    ///
    /// # Errors
    ///
    /// An `Err` value is returned in the following cases:
    ///
    /// - if the transaction sequence is invalid, e.g. because the transaction
    ///   offsets are not contiguous.
    ///
    ///   In this case, **none** of the `transactions` will be written.
    ///
    /// - if creating the new segment fails due to an I/O error.
    ///
    /// # Panics
    ///
    /// The method panics if:
    ///
    /// - `transactions` exceeds [u16::MAX] elements
    ///
    /// - [Self::flush] or writing to the underlying [Writer] fails
    ///
    ///   This is likely caused by some storage issue. As we cannot tell with
    ///   certainty how much data (if any) has been written, the internal state
    ///   becomes invalid and thus a panic is raised.
    ///
    /// - [Self::sync] panics (called when rotating segments)
    pub fn commit<U: Into<Transaction<T>>>(
        &mut self,
        transactions: impl IntoIterator<Item = U>,
    ) -> io::Result<Option<Committed>> {
        self.panicked = true;
        let writer = &mut self.head;
        let committed = writer.commit(transactions)?;
        if writer.len() >= self.opts.max_segment_size {
            self.flush().expect("failed to flush segment upon rotation");
            self.sync();
            self.start_new_segment()?;
        }
        self.panicked = false;

        Ok(committed)
    }

    pub fn transactions_from<'a, D>(
        &self,
        offset: u64,
        decoder: &'a D,
    ) -> impl Iterator<Item = Result<Transaction<T>, D::Error>> + 'a + use<'a, D, R, T>
    where
        D: Decoder<Record = T>,
        D::Error: From<error::Traversal>,
        R: 'a,
        T: 'a,
    {
        transactions_from_internal(self.commits_from(offset).with_log_format_version(), offset, decoder)
    }

    pub fn fold_transactions_from<D>(&self, offset: u64, decoder: D) -> Result<(), D::Error>
    where
        D: Decoder,
        D::Error: From<error::Traversal>,
    {
        fold_transactions_internal(self.commits_from(offset).with_log_format_version(), decoder, offset..)
    }

    pub fn fold_transaction_range<D>(&self, range: impl RangeBounds<u64>, decoder: D) -> Result<(), D::Error>
    where
        D: Decoder,
        D::Error: From<error::Traversal>,
    {
        use std::ops::Bound::*;

        let start = match range.start_bound() {
            Included(x) => *x,
            Excluded(x) => x + 1,
            Unbounded => 0,
        };
        fold_transactions_internal(self.commits_from(start).with_log_format_version(), decoder, range)
    }
}

impl<R: Repo, T> Drop for Generic<R, T> {
    fn drop(&mut self) {
        if !self.panicked
            && let Err(e) = self.flush_and_sync()
        {
            error!("failed to commit on drop: {e}");
        }
    }
}

/// The most recent non empty segment in repo `R`.
///
/// Created by [open_newest_non_empty_segment].
struct MostRecentNonEmptySegment<R> {
    /// Number of empty segments that were ignored.
    empty_segments: usize,
    /// Offset of the non-empty segment.
    segment_offset: u64,
    /// [Repo::SegmentReader] for the non-empty segment.
    segment_reader: R,
}

/// Open the most recent segment in `repo` that is larger than
/// [segment::Header::LEN].
///
/// Note that there should be at most one empty segment in the log. We may,
/// however, want to be lenient on this read-only path, so the number of
/// empty segments is tracked in the returned type rather than returning an
/// error.
fn open_newest_non_empty_segment<R: Repo>(repo: R) -> io::Result<Option<MostRecentNonEmptySegment<R::SegmentReader>>> {
    let mut segments = repo.existing_offsets()?;

    let mut empty_segments = 0;
    let mut segment_offset;
    let mut segment_reader;
    loop {
        let Some(last) = segments.pop() else {
            return Ok(None);
        };
        segment_offset = last;
        segment_reader = repo.open_segment_reader(segment_offset)?;
        if segment_reader.segment_len()? > segment::Header::LEN as u64 {
            break;
        } else {
            empty_segments += 1;
        }
    }

    Ok(Some(MostRecentNonEmptySegment {
        empty_segments,
        segment_offset,
        segment_reader,
    }))
}

/// The most recently written [segment::Metadata] for a given [Repo].
///
/// The type preserves the error information in case the most recent segment
/// contains corrupted data at the end (typically due to a torn write).
///
/// Created by [committed_meta].
pub enum CommittedMeta {
    /// The most recent segment could not be traversed successfully until the
    /// end, i.e. there is trailing garbage in the segment.
    ///
    /// This variant is also returned in case [open_newest_non_empty_segment]
    /// finds more than a single empty segment at the end of the log.
    Prefix {
        /// The metadata of the prefix that could be traversed successfully.
        ///
        /// It is guaranteed that the metadata spans at least one commit.
        metadata: segment::Metadata,
        /// The error encountered.
        error: io::Error,
    },
    /// The most recent segment could be traversed successfully until the end.
    Complete {
        /// The segment metadata.
        ///
        /// It is guaranteed that the metadata spans at least one commit.
        metadata: segment::Metadata,
    },
}

impl CommittedMeta {
    pub fn metadata(&self) -> &segment::Metadata {
        let (Self::Prefix { metadata, .. } | Self::Complete { metadata }) = self;
        metadata
    }

    fn extract(repo: impl Repo) -> io::Result<Option<Self>> {
        let Some(MostRecentNonEmptySegment {
            empty_segments,
            segment_offset,
            mut segment_reader,
        }) = open_newest_non_empty_segment(&repo)?
        else {
            return Ok(None);
        };
        let offset_index = repo.get_offset_index(segment_offset).ok();
        match segment::Metadata::extract(segment_offset, &mut segment_reader, offset_index.as_ref()) {
            // Segment is intact.
            Ok(metadata) if empty_segments <= 1 => {
                assert!(
                    !metadata.tx_range.is_empty(),
                    "segment was promised to be non-empty but contains zero transactions"
                );
                Ok(Some(CommittedMeta::Complete { metadata }))
            }
            // Segment is good, but there are too many empty segments.
            Ok(metadata) => Ok(Some(CommittedMeta::Prefix {
                metadata,
                error: io::Error::new(
                    io::ErrorKind::InvalidData,
                    format!("repo {}: too many empty segments: {}", repo, empty_segments),
                ),
            })),
            // Segment is non-empty, but first commit is corrupt.
            Err(error::SegmentMetadata::InvalidCommit { sofar, source }) if sofar.tx_range.is_empty() => {
                Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    format!(
                        "repo {}: first commit in the most recent segment is corrupt: {}",
                        repo, source
                    ),
                ))
            }
            // Some prefix of the segment is good.
            Err(error::SegmentMetadata::InvalidCommit { sofar, source }) => Ok(Some(CommittedMeta::Prefix {
                metadata: sofar,
                error: source,
            })),
            // Something went wrong, including out-of-order errors and such.
            Err(error::SegmentMetadata::Io(e)) => Err(e),
        }
    }
}

impl From<CommittedMeta> for segment::Metadata {
    fn from(meta: CommittedMeta) -> Self {
        let (CommittedMeta::Prefix { metadata, .. } | CommittedMeta::Complete { metadata }) = meta;
        metadata
    }
}

/// Extract the most recently written [CommittedMeta] from the commitlog
/// in `repo`.
///
/// Returns `None` if the commitlog is empty.
///
/// Note that this function validates the most recent segment, which entails
/// traversing it from the start.
///
/// The function can be used instead of the pattern:
///
/// ```ignore
/// let log = Commitlog::open(..)?;
/// let max_offset = log.max_committed_offset();
/// ```
///
/// like so:
///
/// ```ignore
/// let max_offset = committed_meta(..)?.map(|meta| meta.metadata().tx_range.end);
/// ```
///
/// Unlike `open`, no segment will be created in an empty `repo`.
pub fn committed_meta(repo: impl Repo) -> io::Result<Option<CommittedMeta>> {
    CommittedMeta::extract(repo)
}

pub fn commits_from<R: Repo>(repo: R, max_log_format_version: u8, offset: u64) -> io::Result<Commits<R>> {
    let mut offsets = repo.existing_offsets()?;
    if let Some(pos) = offsets.iter().rposition(|&off| off <= offset) {
        offsets = offsets.split_off(pos);
    }
    let segments = Segments {
        offs: offsets.into_iter(),
        repo,
        max_log_format_version,
    };
    Ok(Commits {
        inner: None,
        segments,
        last_commit: CommitInfo::Initial { next_offset: offset },
        last_error: None,
    })
}

pub fn transactions_from<'a, R, D, T>(
    repo: R,
    max_log_format_version: u8,
    offset: u64,
    de: &'a D,
) -> io::Result<impl Iterator<Item = Result<Transaction<T>, D::Error>> + 'a>
where
    R: Repo + 'a,
    D: Decoder<Record = T>,
    D::Error: From<error::Traversal>,
    T: 'a,
{
    commits_from(repo, max_log_format_version, offset)
        .map(|commits| transactions_from_internal(commits.with_log_format_version(), offset, de))
}

pub fn fold_transactions_from<R, D>(repo: R, max_log_format_version: u8, offset: u64, de: D) -> Result<(), D::Error>
where
    R: Repo,
    D: Decoder,
    D::Error: From<error::Traversal> + From<io::Error>,
{
    fold_transaction_range(repo, max_log_format_version, offset.., de)
}

pub fn fold_transaction_range<R, D>(
    repo: R,
    max_log_format_version: u8,
    range: impl RangeBounds<u64>,
    de: D,
) -> Result<(), D::Error>
where
    R: Repo,
    D: Decoder,
    D::Error: From<error::Traversal> + From<io::Error>,
{
    use std::ops::Bound::*;

    let start = match range.start_bound() {
        Included(x) => *x,
        Excluded(x) => x + 1,
        Unbounded => 0,
    };
    let commits = commits_from(repo, max_log_format_version, start)?;
    fold_transactions_internal(commits.with_log_format_version(), de, range)
}

fn transactions_from_internal<'a, R, D, T>(
    commits: CommitsWithVersion<R>,
    offset: u64,
    de: &'a D,
) -> impl Iterator<Item = Result<Transaction<T>, D::Error>> + 'a
where
    R: Repo + 'a,
    D: Decoder<Record = T>,
    D::Error: From<error::Traversal>,
    T: 'a,
{
    commits
        .map(|x| x.map_err(D::Error::from))
        .map_ok(move |(version, commit)| commit.into_transactions(version, offset, de))
        .flatten_ok()
        .map(|x| x.and_then(|y| y))
}

fn fold_transactions_internal<R, D>(
    mut commits: CommitsWithVersion<R>,
    de: D,
    range: impl RangeBounds<u64>,
) -> Result<(), D::Error>
where
    R: Repo,
    D: Decoder,
    D::Error: From<error::Traversal>,
{
    use std::ops::Bound::*;

    // Avoid reading the first commit if it wouldn't be in the range anyway.
    if range_is_empty(&range) {
        return Ok(());
    }

    // `true` if `offset` is outside `range`, s.t. it is smaller than the start
    // bound.
    let before_start = |offset: &u64| match range.start_bound() {
        Included(x) => offset < x,
        Excluded(x) => offset <= x,
        Unbounded => false,
    };
    // `true` if `offset` is outside `range`, s.t. it is greater than the end
    // bound.
    let past_end = |offset: &u64| match range.end_bound() {
        Included(x) => offset > x,
        Excluded(x) => offset >= x,
        Unbounded => false,
    };

    while let Some(commit) = commits.next() {
        let (version, commit) = match commit {
            Ok(version_and_commit) => version_and_commit,
            Err(e) => {
                // Ignore it if the very last commit in the log is broken.
                // The next `append` will fix the log, but the `decoder`
                // has no way to tell whether we're at the end or not.
                // This is unlike the consumer of an iterator, which can
                // perform below check itself.
                if commits.next().is_none() {
                    return Ok(());
                }

                return Err(e.into());
            }
        };
        trace!("commit {} n={} version={}", commit.min_tx_offset, commit.n, version);

        let max_tx_offset = commit.min_tx_offset + commit.n as u64;
        // Skip if no transaction in the commit is in range.
        if before_start(&max_tx_offset) {
            continue;
        }

        let records = &mut commit.records.as_slice();
        for n in 0..commit.n {
            let tx_offset = commit.min_tx_offset + n as u64;
            if before_start(&tx_offset) {
                de.skip_record(version, tx_offset, records)?;
            } else if past_end(&tx_offset) {
                return Ok(());
            } else {
                de.consume_record(version, tx_offset, records)?;
            }
        }
    }

    Ok(())
}

/// Remove all data past the given transaction `offset`.
///
/// The function deletes log segments starting from the newest. As multiple
/// segments cannot be deleted atomically, the log may be left longer than
/// `offset` if the function does not return successfully.
///
/// If the function returns successfully, the most recent [`Commit`] in the
/// log will contain the transaction at `offset`.
///
/// The log must be re-opened if it is to be used after calling this function.
pub fn reset_to(repo: &impl Repo, offset: u64) -> io::Result<()> {
    let segments = repo.existing_offsets()?;
    reset_to_internal(repo, &segments, offset)
}

fn reset_to_internal(repo: &impl Repo, segments: &[u64], offset: u64) -> io::Result<()> {
    for segment in segments.iter().copied().rev() {
        if segment > offset {
            // Segment is outside the offset, so remove it wholesale.
            debug!("removing segment {segment}");
            repo.remove_segment(segment)?;
        } else {
            // Read commit-wise until we find the byte offset.
            let mut reader = repo::open_segment_reader(repo, DEFAULT_LOG_FORMAT_VERSION, segment)?;

            let (index_file, mut byte_offset) = try_seek_using_offset_index(repo, &mut reader, offset)
                .map(|(index_file, byte_offset)| (Some(index_file), byte_offset))
                .unwrap_or((None, segment::Header::LEN as u64));

            let commits = reader.commits();

            for commit in commits {
                let commit = commit?;
                if commit.min_tx_offset > offset {
                    break;
                }
                byte_offset += Commit::from(commit).encoded_len() as u64;
            }

            if byte_offset == segment::Header::LEN as u64 {
                // Segment is empty, just remove it.
                repo.remove_segment(segment)?;
            } else {
                debug!("truncating segment {segment} to {offset} at {byte_offset}");
                let mut file = repo.open_segment_writer(segment)?;

                if let Some(mut index_file) = index_file {
                    let index_file = index_file.as_mut();
                    // Note: The offset index truncates equal or greater,
                    // inclusive. We'd like to retain `offset` in the index, as
                    // the commit is also retained in the log.
                    index_file.ftruncate(offset + 1, byte_offset).map_err(|e| {
                        io::Error::new(
                            io::ErrorKind::InvalidData,
                            format!("Failed to truncate offset index: {e}"),
                        )
                    })?;
                    index_file.async_flush()?;
                }

                file.ftruncate(offset, byte_offset)?;
                // Some filesystems require fsync after ftruncate.
                file.fsync()?;
                break;
            }
        }
    }

    Ok(())
}

pub struct Segments<R> {
    repo: R,
    offs: vec::IntoIter<u64>,
    max_log_format_version: u8,
}

impl<R: Repo> Iterator for Segments<R> {
    type Item = io::Result<segment::Reader<R::SegmentReader>>;

    fn next(&mut self) -> Option<Self::Item> {
        let off = self.offs.next()?;
        debug!("iter segment {off}");
        Some(repo::open_segment_reader(&self.repo, self.max_log_format_version, off))
    }
}

/// Helper for the [`Commits`] iterator.
enum CommitInfo {
    /// Constructed in [`Generic::commits_from`], specifying the offset the next
    /// commit should have.
    Initial { next_offset: u64 },
    /// The last commit seen by the iterator.
    ///
    /// Stores the range of transaction offsets, where `tx_range.end` is the
    /// offset the next commit is expected to have. Also retains the checksum
    /// needed to detect duplicate commits.
    LastSeen { tx_range: Range<u64>, checksum: u32 },
}

impl CommitInfo {
    /// `true` if the last seen commit in self and the provided one have the
    /// same `min_tx_offset`.
    fn same_offset_as(&self, commit: &StoredCommit) -> bool {
        let Self::LastSeen { tx_range, .. } = self else {
            return false;
        };
        tx_range.start == commit.min_tx_offset
    }

    /// `true` if the last seen commit in self and the provided one have the
    /// same `checksum`.
    fn same_checksum_as(&self, commit: &StoredCommit) -> bool {
        let Some(checksum) = self.checksum() else { return false };
        checksum == &commit.checksum
    }

    fn checksum(&self) -> Option<&u32> {
        match self {
            Self::Initial { .. } => None,
            Self::LastSeen { checksum, .. } => Some(checksum),
        }
    }

    fn expected_offset(&self) -> &u64 {
        match self {
            Self::Initial { next_offset } => next_offset,
            Self::LastSeen { tx_range, .. } => &tx_range.end,
        }
    }

    // If initial offset falls within a commit, adjust it to the commit boundary.
    //
    // Returns `true` if the initial offset is past `commit`.
    // Returns `false` if `self` isn't `Self::Initial`,
    // or the initial offset has been adjusted to the starting offset of `commit`.
    //
    // For iteration, `true` means to skip the commit, `false` to yield it.
    fn adjust_initial_offset(&mut self, commit: &StoredCommit) -> bool {
        if let Self::Initial { next_offset } = self {
            let last_tx_offset = commit.min_tx_offset + commit.n as u64 - 1;
            if *next_offset > last_tx_offset {
                return true;
            } else {
                *next_offset = commit.min_tx_offset;
            }
        }

        false
    }
}

pub struct Commits<R: Repo> {
    inner: Option<segment::Commits<R::SegmentReader>>,
    segments: Segments<R>,
    last_commit: CommitInfo,
    last_error: Option<error::Traversal>,
}

impl<R: Repo> Commits<R> {
    fn current_segment_header(&self) -> Option<&segment::Header> {
        self.inner.as_ref().map(|segment| &segment.header)
    }

    /// Turn `self` into an iterator which pairs the log format version of the
    /// current segment with the [`Commit`].
    pub fn with_log_format_version(self) -> CommitsWithVersion<R> {
        CommitsWithVersion { inner: self }
    }

    /// Advance the current-segment iterator to yield the next commit.
    ///
    /// Checks that the offset sequence is contiguous, and may skip commits
    /// until the requested offset.
    ///
    /// Returns `None` if the segment iterator is exhausted or returns an error.
    fn next_commit(&mut self) -> Option<Result<StoredCommit, error::Traversal>> {
        loop {
            match self.inner.as_mut()?.next()? {
                Ok(commit) => {
                    // Pop the last error. Either we'll return it below, or it's no longer
                    // interesting.
                    let prev_error = self.last_error.take();

                    // Skip entries before the initial commit.
                    if self.last_commit.adjust_initial_offset(&commit) {
                        trace!("adjust initial offset");
                        continue;
                    // Same offset: ignore if duplicate (same crc), else report a "fork".
                    } else if self.last_commit.same_offset_as(&commit) {
                        if !self.last_commit.same_checksum_as(&commit) {
                            warn!(
                                "forked: commit={:?} last-error={:?} last-crc={:?}",
                                commit,
                                prev_error,
                                self.last_commit.checksum()
                            );
                            return Some(Err(error::Traversal::Forked {
                                offset: commit.min_tx_offset,
                            }));
                        } else {
                            trace!("ignore duplicate");
                            continue;
                        }
                    // Not the expected offset: report out-of-order.
                    } else if self.last_commit.expected_offset() != &commit.min_tx_offset {
                        warn!("out-of-order: commit={commit:?} last-error={prev_error:?}");
                        return Some(Err(error::Traversal::OutOfOrder {
                            expected_offset: *self.last_commit.expected_offset(),
                            actual_offset: commit.min_tx_offset,
                            prev_error: prev_error.map(Box::new),
                        }));
                    // Seems legit, record info.
                    } else {
                        self.last_commit = CommitInfo::LastSeen {
                            tx_range: commit.tx_range(),
                            checksum: commit.checksum,
                        };

                        return Some(Ok(commit));
                    }
                }

                Err(e) => {
                    warn!("error reading next commit: {e}");
                    // Stop traversing this segment here.
                    //
                    // If this is just a partial write at the end of the segment,
                    // we may be able to obtain a commit with right offset from
                    // the next segment.
                    //
                    // If we don't, the error here is likely more helpful, but
                    // would be clobbered by `OutOfOrder`. Therefore we store it
                    // here.
                    self.set_last_error(e);

                    return None;
                }
            }
        }
    }

    /// Store `e` has the last error for delayed reporting.
    fn set_last_error(&mut self, e: io::Error) {
        // Recover a checksum mismatch.
        let last_error = if e.kind() == io::ErrorKind::InvalidData && e.get_ref().is_some() {
            e.into_inner()
                .unwrap()
                .downcast::<error::ChecksumMismatch>()
                .map(|source| error::Traversal::Checksum {
                    offset: *self.last_commit.expected_offset(),
                    source: *source,
                })
                .unwrap_or_else(|e| io::Error::new(io::ErrorKind::InvalidData, e).into())
        } else {
            error::Traversal::from(e)
        };
        self.last_error = Some(last_error);
    }

    /// If we're still looking for the initial commit, try to use the offset
    /// index to advance the segment reader.
    fn try_seek_to_initial_offset(&self, segment: &mut segment::Reader<R::SegmentReader>) {
        if let CommitInfo::Initial { next_offset } = &self.last_commit {
            try_seek_using_offset_index(&self.segments.repo, segment, *next_offset);
        }
    }
}

impl<R: Repo> Iterator for Commits<R> {
    type Item = Result<StoredCommit, error::Traversal>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(item) = self.next_commit() {
            return Some(item);
        }

        match self.segments.next() {
            // When there is no more data, the last commit being bad is an error
            None => self.last_error.take().map(Err),
            Some(segment) => segment.map_or_else(
                |e| Some(Err(e.into())),
                |mut segment| {
                    self.try_seek_to_initial_offset(&mut segment);
                    self.inner = Some(segment.commits());
                    self.next()
                },
            ),
        }
    }
}

pub struct CommitsWithVersion<R: Repo> {
    inner: Commits<R>,
}

impl<R: Repo> Iterator for CommitsWithVersion<R> {
    type Item = Result<(u8, Commit), error::Traversal>;

    fn next(&mut self) -> Option<Self::Item> {
        let next = self.inner.next()?;
        match next {
            Ok(commit) => {
                let version = self
                    .inner
                    .current_segment_header()
                    .map(|hdr| hdr.log_format_version)
                    .expect("segment header none even though segment yielded a commit");
                Some(Ok((version, commit.into())))
            }
            Err(e) => Some(Err(e)),
        }
    }
}

/// Try to advance `reader` to `offset` using the offset index.
///
/// If successful, returns the offset index and the byte position of `reader`.
/// `None` if the position of `reader` is unchanged.
fn try_seek_using_offset_index<R: Repo>(
    repo: &R,
    reader: &mut segment::Reader<R::SegmentReader>,
    offset: u64,
) -> Option<(TxOffsetIndex, u64)> {
    let segment_offset = reader.min_tx_offset;
    let index = repo
        .get_offset_index(segment_offset)
        .inspect_err(|e| {
            if e.kind() == io::ErrorKind::NotFound {
                debug!("offset index does not exist segment={segment_offset}");
            } else {
                warn!(
                    "error opening offset index segment={segment_offset}: {e} {}",
                    source_chain(&e)
                );
            }
        })
        .ok()?;

    reader
        .seek_to_offset(&index, offset)
        .inspect_err(|e| match e {
            // Can happen if the segment is empty or small, so don't spam the logs.
            IndexError::KeyNotFound => {
                debug!("offset not found segment={segment_offset} offset={offset}");
            }
            e => {
                warn!(
                    "error reading index segment={segment_offset} offset={offset}: {e} {}",
                    source_chain(&e)
                );
            }
        })
        .ok()
        .map(|pos| (index, pos))
}

// `range_bounds_is_empty` https://github.com/rust-lang/rust/issues/137300
//
// This is correct for integers, but unsound for arbitrary `T`, so unlikely to
// be stabilized.
fn range_is_empty(range: &impl RangeBounds<u64>) -> bool {
    use std::ops::Bound::*;

    #[rustfmt::skip]
    let not_empty = match (range.start_bound(), range.end_bound()) {
        (Unbounded, _) | (_, Unbounded) => true,
        (Included(start), Excluded(end))
        | (Excluded(start), Included(end))
        | (Excluded(start), Excluded(end)) => start < end,
        (Included(start), Included(end)) => start <= end,
    };

    !not_empty
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, iter::repeat};

    use pretty_assertions::assert_matches;

    use super::*;
    use crate::{
        payload::{ArrayDecodeError, ArrayDecoder},
        tests::helpers::{enable_logging, fill_log, mem_log},
    };

    #[test]
    fn rotate_segments_simple() {
        let mut log = mem_log::<[u8; 32]>(128);
        for i in 0..4 {
            log.commit([(i, [0; 32])]).unwrap();
        }
        log.flush_and_sync().unwrap();

        let offsets = log.repo.existing_offsets().unwrap();
        assert_eq!(&offsets[..offsets.len() - 1], &log.tail);
        // TODO: We overshoot the max segment size.
        assert_eq!(&offsets, &[0, 3]);
    }

    #[test]
    fn huge_commit() {
        let mut log = mem_log::<[u8; 32]>(32);

        log.commit([(0, [0; 32]), (1, [1; 32])]).unwrap();
        log.flush_and_sync().unwrap();
        // First segment got rotated out.
        assert_eq!(&log.tail, &[0]);

        log.commit([(2, [2; 32])]).unwrap();
        log.flush_and_sync().unwrap();

        // Second segment got rotated out and segment 3 is created.
        assert_eq!(&log.repo.existing_offsets().unwrap(), &[0, 2, 3]);
    }

    #[test]
    fn traverse_commits() {
        let mut log = mem_log::<[u8; 32]>(32);
        fill_log(&mut log, 10, repeat(1));

        for (i, commit) in (0..10).zip(log.commits_from(0)) {
            assert_eq!(i, commit.unwrap().min_tx_offset);
        }
    }

    #[test]
    fn traverse_commits_with_offset() {
        let mut log = mem_log::<[u8; 32]>(32);
        fill_log(&mut log, 10, repeat(1));

        for offset in 0..10 {
            for commit in log.commits_from(offset) {
                let commit = commit.unwrap();
                assert!(commit.min_tx_offset >= offset);
            }
        }
        assert_eq!(0, log.commits_from(10).count());
    }

    #[test]
    fn fold_transactions_with_offset() {
        let mut log = mem_log::<[u8; 32]>(32);
        fill_log(&mut log, 10, repeat(1));

        /// A [`Decoder`] which counts the number of records decoded,
        /// and asserts that the `tx_offset` is as expected.
        struct CountDecoder {
            count: Cell<u64>,
            next_tx_offset: Cell<u64>,
        }

        impl Decoder for &CountDecoder {
            type Record = [u8; 32];
            type Error = ArrayDecodeError;

            fn decode_record<'a, R: spacetimedb_sats::buffer::BufReader<'a>>(
                &self,
                _version: u8,
                _tx_offset: u64,
                _reader: &mut R,
            ) -> Result<Self::Record, Self::Error> {
                unreachable!("Folding never calls `decode_record`")
            }

            fn consume_record<'a, R: spacetimedb_sats::buffer::BufReader<'a>>(
                &self,
                version: u8,
                tx_offset: u64,
                reader: &mut R,
            ) -> Result<(), Self::Error> {
                let decoder = ArrayDecoder::<32>;
                decoder.consume_record(version, tx_offset, reader)?;
                self.count.set(self.count.get() + 1);
                let expected_tx_offset = self.next_tx_offset.get();
                assert_eq!(expected_tx_offset, tx_offset);
                self.next_tx_offset.set(expected_tx_offset + 1);
                Ok(())
            }

            fn skip_record<'a, R: spacetimedb_sats::buffer::BufReader<'a>>(
                &self,
                version: u8,
                tx_offset: u64,
                reader: &mut R,
            ) -> Result<(), Self::Error> {
                let decoder = ArrayDecoder::<32>;
                decoder.consume_record(version, tx_offset, reader)?;
                Ok(())
            }
        }

        for offset in 0..10 {
            let decoder = CountDecoder {
                count: Cell::new(0),
                next_tx_offset: Cell::new(offset),
            };

            log.fold_transactions_from(offset, &decoder).unwrap();

            assert_eq!(decoder.count.get(), 10 - offset);
            assert_eq!(decoder.next_tx_offset.get(), 10);
        }
    }

    #[test]
    fn traverse_commits_ignores_duplicates() {
        let mut log = mem_log::<[u8; 32]>(1024);

        let tx1 = [42u8; 32];
        let tx2 = [43u8; 32];

        log.commit([(0, tx1)]).unwrap();
        let commit1 = Commit {
            min_tx_offset: 0,
            n: 1,
            records: tx1.to_vec(),
            ..log.head.commit.clone()
        };

        // Reset the commit offset, so we can write the same commit twice.
        log.head.commit.min_tx_offset = 0;
        log.commit([(0, tx1)]).unwrap();

        // Write another one.
        log.commit([(1, tx2)]).unwrap();
        let commit2 = Commit {
            min_tx_offset: 1,
            n: 1,
            records: tx2.to_vec(),
            ..log.head.commit.clone()
        };

        log.flush_and_sync().unwrap();

        assert_eq!(
            [commit1, commit2].as_slice(),
            &log.commits_from(0)
                .map_ok(Commit::from)
                .collect::<Result<Vec<_>, _>>()
                .unwrap()
        );
    }

    #[test]
    fn traverse_commits_errors_when_forked() {
        let mut log = mem_log::<[u8; 32]>(1024);

        log.commit([(0, [42; 32])]).unwrap();
        // Reset the commit offset,
        // and write a different commit at the same offset.
        // This is considered a fork.
        log.head.commit.min_tx_offset = 0;
        log.commit([(0, [43; 32])]).unwrap();

        log.flush_and_sync().unwrap();

        let res = log.commits_from(0).collect::<Result<Vec<_>, _>>();
        assert!(
            matches!(res, Err(error::Traversal::Forked { offset: 0 })),
            "expected fork error: {res:?}"
        )
    }

    #[test]
    fn traverse_commits_errors_when_offset_not_contiguous() {
        let mut log = mem_log::<[u8; 32]>(1024);

        log.commit([(0, [42; 32])]).unwrap();
        log.head.commit.min_tx_offset = 18;
        log.commit([(18, [42; 32])]).unwrap();

        log.flush_and_sync().unwrap();

        let res = log.commits_from(0).collect::<Result<Vec<_>, _>>();
        assert!(
            matches!(
                res,
                Err(error::Traversal::OutOfOrder {
                    expected_offset: 1,
                    actual_offset: 18,
                    prev_error: None
                })
            ),
            "expected out-of-order error: {res:?}"
        )
    }

    #[test]
    fn traverse_transactions() {
        let mut log = mem_log::<[u8; 32]>(32);
        let total_txs = fill_log(&mut log, 10, (1..=3).cycle()) as u64;

        for (i, tx) in (0..total_txs).zip(log.transactions_from(0, &ArrayDecoder)) {
            assert_eq!(i, tx.unwrap().offset);
        }
    }

    #[test]
    fn traverse_transactions_with_offset() {
        let mut log = mem_log::<[u8; 32]>(32);
        let total_txs = fill_log(&mut log, 10, (1..=3).cycle()) as u64;

        for offset in 0..total_txs {
            let mut iter = log.transactions_from(offset, &ArrayDecoder);
            assert_eq!(offset, iter.next().expect("at least one tx expected").unwrap().offset);
            for tx in iter {
                assert!(tx.unwrap().offset >= offset);
            }
        }
        assert_eq!(0, log.transactions_from(total_txs, &ArrayDecoder).count());
    }

    #[test]
    fn traverse_empty() {
        let log = mem_log::<[u8; 32]>(32);

        assert_eq!(0, log.commits_from(0).count());
        assert_eq!(0, log.commits_from(42).count());
        assert_eq!(0, log.transactions_from(0, &ArrayDecoder).count());
        assert_eq!(0, log.transactions_from(42, &ArrayDecoder).count());
    }

    #[test]
    fn reset_hard() {
        let mut log = mem_log::<[u8; 32]>(128);
        fill_log(&mut log, 50, (1..=10).cycle());

        log = log.reset().unwrap();
        assert_eq!(0, log.transactions_from(0, &ArrayDecoder).count());
    }

    #[test]
    fn reset_to_offset() {
        enable_logging();

        let mut log = mem_log::<[u8; 32]>(128);
        let total_txs = fill_log(&mut log, 50, repeat(1)) as u64;

        for offset in (0..total_txs).rev() {
            log = log.reset_to(offset).unwrap();
            assert_eq!(
                offset,
                log.transactions_from(0, &ArrayDecoder)
                    .map(Result::unwrap)
                    .last()
                    .unwrap()
                    .offset
            );
            // We're counting from zero, so offset + 1 is the # of txs.
            assert_eq!(
                offset + 1,
                log.transactions_from(0, &ArrayDecoder).map(Result::unwrap).count() as u64
            );
        }
    }

    #[test]
    fn reset_to_offset_many_txs_per_commit() {
        let mut log = mem_log::<[u8; 32]>(128);
        let total_txs = fill_log(&mut log, 50, (1..=10).cycle()) as u64;

        // No op.
        log = log.reset_to(total_txs).unwrap();
        assert_eq!(total_txs, log.transactions_from(0, &ArrayDecoder).count() as u64);

        let middle_commit = log.commits_from(0).nth(25).unwrap().unwrap();

        // Both fall into the middle commit, which should be retained.
        log = log.reset_to(middle_commit.min_tx_offset + 1).unwrap();
        assert_eq!(
            middle_commit.tx_range().end,
            log.transactions_from(0, &ArrayDecoder).count() as u64
        );
        log = log.reset_to(middle_commit.min_tx_offset).unwrap();
        assert_eq!(
            middle_commit.tx_range().end,
            log.transactions_from(0, &ArrayDecoder).count() as u64
        );

        // Offset falls into 2nd commit.
        // 1st commit (1 tx) + 2nd commit (2 txs) = 3
        log = log.reset_to(1).unwrap();
        assert_eq!(3, log.transactions_from(0, &ArrayDecoder).count() as u64);

        // Offset falls into 1st commit.
        // 1st commit (1 tx) = 1
        log = log.reset_to(0).unwrap();
        assert_eq!(1, log.transactions_from(0, &ArrayDecoder).count() as u64);
    }

    #[test]
    fn reopen() {
        let mut log = mem_log::<[u8; 32]>(1024);
        let total_txs = fill_log(&mut log, 100, (1..=10).cycle());
        assert_eq!(
            total_txs,
            log.transactions_from(0, &ArrayDecoder).map(Result::unwrap).count()
        );

        let mut log = Generic::<_, [u8; 32]>::open(
            log.repo.clone(),
            Options {
                max_segment_size: 1024,
                ..Options::default()
            },
        )
        .unwrap();
        let total_txs = fill_log(&mut log, 100, (1..=10).cycle());

        assert_eq!(
            total_txs,
            log.transactions_from(0, &ArrayDecoder).map(Result::unwrap).count()
        );
    }

    #[test]
    fn set_new_epoch() {
        let mut log = Generic::<_, [u8; 32]>::open(repo::Memory::unlimited(), <_>::default()).unwrap();
        assert_eq!(log.epoch(), Commit::DEFAULT_EPOCH);
        log.commit([(0, [12; 32])]).unwrap();
        log.set_epoch(42).unwrap();
        assert_eq!(log.epoch(), 42);
        log.commit([(1, [13; 32])]).unwrap();

        log.flush_and_sync().unwrap();

        let epochs = log
            .commits_from(0)
            .map(Result::unwrap)
            .map(|commit| commit.epoch)
            .collect::<Vec<_>>();
        assert_eq!(&[Commit::DEFAULT_EPOCH, 42], epochs.as_slice());
    }

    #[test]
    fn set_lower_epoch_returns_error() {
        let mut log = Generic::<_, [u8; 32]>::open(repo::Memory::unlimited(), <_>::default()).unwrap();
        log.set_epoch(42).unwrap();
        assert_eq!(log.epoch(), 42);
        assert_matches!(log.set_epoch(7), Err(e) if e.kind() == io::ErrorKind::InvalidInput)
    }
}