Function create_segment

crates/commitlog/src/stream/writer.rs:442–486

Create a new segment at offset `last_written_tx_range.end`. If the segment file already exists but has a size equal to or smaller than a segment header, the file is truncated. Otherwise, an already existing segment is an error.

(
    repo: R,
    last_written_tx_range: Option<Range<u64>>,
    commitlog_options: Options,
    header: segment::Header,
)

/// a segment header, the file is truncated. Otherwise, an already existing
/// segment is an error.
fn create_segment<R: Repo>(
    repo: R,
    last_written_tx_range: Option<Range<u64>>,
    commitlog_options: Options,
    header: segment::Header,
) -> io::Result<(R::SegmentWriter, Option<OffsetIndexWriter>)> {
    let segment_offset = last_written_tx_range
        .as_ref()
        .map(|range| range.end)
        .unwrap_or_default();
    let mut segment = loop {
        match repo.create_segment(segment_offset, header) {
            Ok(segment) => break segment,
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
                trace!("segment already exists");
                let mut s = repo.open_segment_writer(segment_offset)?;
                let len = s.segment_len()?;
                trace!("segment len: {len}");
                if len <= segment::Header::LEN as _ {
                    trace!("overwriting existing segment");
                    repo.remove_segment(segment_offset)?;
                    continue;
                }

                return Err(io::Error::new(
                    e.kind(),
                    format!(
                        "repo {}: segment {} already exists and is non-empty: {}",
                        repo, segment_offset, e
                    ),
                ));
            }
            Err(e) => return Err(e),
        }
    };
    fallocate(&mut segment, &commitlog_options)?;

    let index_writer = repo
        .create_offset_index(segment_offset, commitlog_options.offset_index_len())
        .inspect_err(|e| warn!("unable to create offset index segment={segment_offset} err={e:?}"))
        .map(|index| OffsetIndexWriter::new(index, commitlog_options))
        .ok();

    Ok((segment, index_writer))
}