Function reset_to_internal

crates/commitlog/src/commitlog.rs:705–759

(repo: &impl Repo, segments: &[u64], offset: u64)

}

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,