tephra 0.4.0

//! SegmentSet: the collection of segment files and the mapping from global
//! positions to bytes.
//!
//! Turns N independent `seglog` files into one logically continuous,
//! position-addressed log. Everything about *what* an event is stays above it;
//! everything about record framing stays below it (in `seglog`).
//!
//! Positions are 1-based: the first event stored is position 1, and `Position::ZERO`
//! is reserved to mean "empty" (no events yet). The first segment's `base_position`
//! is therefore 1.
//!
//! Invariants enforced here:
//!
//! 1. Segments are position-disjoint and contiguous: segment N's `base_position`
//!    equals segment N-1's `base_position + event_count`. The first segment's
//!    `base_position` is 1.
//! 2. Exactly one segment is active (writable) at a time; the rest are sealed
//!    and immutable.
//! 3. A batch never spans segments.
//! 4. Segment files are never recycled or reused.
//! 5. At most one process writes a directory at a time, held by an advisory lock on
//!    `LOCK`. Read-only openers take no lock at all.

use std::borrow::Cow;
use std::cmp::Ordering;
use std::fmt;
use std::fs::{self, File};
use std::io;
use std::mem;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};

use thiserror::Error;

use seglog::parse::parse_record_ref;
use seglog::read::{ReadError, ReadHint, Reader, RecordKind};
use seglog::tail::{Tail, TailError};
use seglog::write::{WriteError, Writer};
use seglog::{COMMIT_MARKER_PAYLOAD, FlushedOffset, RECORD_HEAD_SIZE};

use crate::Position;
use crate::log::header::{HeaderError, SEGMENT_HEADER_SIZE, SegmentHeader};
use crate::log::lock::{self, DirLock, LockFailure};

/// Number of digits in a segment file name. Twenty digits covers `u64::MAX`, so
/// zero-padded lexicographic order equals numeric order.
const NAME_DIGITS: usize = 20;

/// The first position assigned in a fresh log. `Position::ZERO` is reserved to mean
/// "empty", so events (and the first segment's base) start at 1.
const FIRST_POSITION: u64 = 1;

/// Per-record framing overhead (length + CRC), on top of the record's own bytes.
/// Exposed so the write coordinator can budget batch sizes without importing seglog.
pub const RECORD_OVERHEAD: usize = RECORD_HEAD_SIZE;

/// Fixed overhead a batch pays once for its trailing commit marker (the marker's own
/// record frame plus its payload).
pub const BATCH_OVERHEAD: usize = RECORD_HEAD_SIZE + COMMIT_MARKER_PAYLOAD;

/// Configuration for the segments in a set. All segments in a set share it.
#[derive(Clone, Copy, Debug)]
pub struct SegmentConfig {
    /// Total size of each segment file in bytes (including the header).
    pub segment_size: usize,
    /// Largest total on-disk record length a single record may occupy.
    pub max_record_len: usize,
    /// Bytes reserved at the start of every segment for its [`SegmentHeader`].
    pub header_size: usize,
}

impl SegmentConfig {
    /// Config for the given segment size with the conventional defaults:
    /// `max_record_len = segment_size / 4` and `header_size = SEGMENT_HEADER_SIZE`.
    ///
    /// The result is not guaranteed valid for very small `segment_size`; that is
    /// checked by [`SegmentSet::open`] via [`validate`](Self::validate).
    pub fn new(segment_size: usize) -> Self {
        SegmentConfig {
            segment_size,
            max_record_len: segment_size / 4,
            header_size: SEGMENT_HEADER_SIZE,
        }
    }

    /// Rejects configs that cannot address, or cannot usefully store, records.
    ///
    /// Validated once at open rather than defended against with a panic on every
    /// append (byte offsets are stored as `u32`, so segments must stay under 4 GiB).
    pub fn validate(&self) -> Result<(), LogError> {
        let invalid = |reason: String| Err(LogError::InvalidConfig { reason });

        if self.segment_size <= self.header_size {
            return invalid(format!(
                "segment_size {} must exceed header_size {}",
                self.segment_size, self.header_size
            ));
        }
        if self.segment_size > u32::MAX as usize {
            return invalid(format!(
                "segment_size {} exceeds u32::MAX; byte offsets are stored as u32",
                self.segment_size
            ));
        }
        if self.max_record_len < RECORD_HEAD_SIZE {
            return invalid(format!(
                "max_record_len {} is smaller than a record header ({RECORD_HEAD_SIZE} bytes)",
                self.max_record_len
            ));
        }
        let usable = self.segment_size - self.header_size;
        let need = self.max_record_len + RECORD_HEAD_SIZE + COMMIT_MARKER_PAYLOAD;
        if need > usable {
            return invalid(format!(
                "a max-size record plus its commit marker ({need} bytes) does not fit a \
                 segment's usable space ({usable} bytes)"
            ));
        }
        Ok(())
    }
}

/// A single segment file: one `seglog` file, its base position, and the offset
/// sidecar mapping local position to byte offset.
///
/// Shared behind `Arc` so a reader can hold a segment across a rollover while the
/// set swaps the active segment without invalidating it.
pub struct Segment {
    base_position: Position,
    path: PathBuf,
    /// Durable extent used by readers: the active segment shares its writer's (or tail's)
    /// live offset, and a sealed segment carries the frozen offset its opening scan
    /// established.
    ///
    /// It is not enough to let a sealed segment fall back to the file length. A failed
    /// append leaves CRC-valid records past the last commit marker (`Writer::rewind_to`),
    /// and if the next batch rolls over, the segment seals with those orphans still on
    /// disk. Bounding reads by the file length would adopt them as events; bounding by
    /// the commit point discards them, exactly as recovery does for the active segment.
    /// Not an `Option`: every segment now has one, and the type says so rather than a
    /// comment.
    flushed_offset: FlushedOffset,
    /// Byte offset of each data record, indexed by `position - base_position`.
    /// Never persisted in v1: derivable by one sequential scan on open.
    offsets: RwLock<Vec<u32>>,
    /// A cached reader for random reads. Segments are immutable, so one open fd is
    /// reusable indefinitely; the `Mutex` serializes the reader's internal buffers.
    reader: Mutex<Option<Reader<0>>>,
}

impl Segment {
    /// The base (first) global position of this segment.
    pub fn base_position(&self) -> Position {
        self.base_position
    }

    /// Number of events (data records) currently in this segment.
    pub fn event_count(&self) -> u64 {
        self.offsets.read().unwrap().len() as u64
    }

    /// The byte offset of the data record at `local` (`position - base_position`), or
    /// `None` if `local` is past the events currently in this segment.
    pub(crate) fn data_offset(&self, local: usize) -> Option<u32> {
        self.offsets.read().unwrap().get(local).copied()
    }

    /// Opens a fresh reader (its own file descriptor and read-ahead buffer) over this
    /// segment. Segments are immutable once written past their flushed point, so a reader
    /// can outlive a rollover. Used by concurrent readers, which must not share the one
    /// cached reader ([`Segment::reader`]); the write coordinator's point reads reuse the
    /// cached one.
    pub(crate) fn open_reader(&self) -> Result<Reader<0>, LogError> {
        Reader::<0>::open_read_only(&self.path, Some(self.flushed_offset.clone()))
            .map_err(|source| LogError::read(&self.path, source))
    }

    /// Reads the data record at local position `local` using a caller-supplied `reader`,
    /// with the random read hint. `None` if `local` is past this segment's events. Lets a
    /// reader reuse one open fd across consecutive positions in the same segment rather than
    /// opening one per record.
    pub(crate) fn read_at_local(
        &self,
        reader: &mut Reader<0>,
        local: usize,
    ) -> Result<Option<Record>, LogError> {
        let Some(offset) = self.data_offset(local) else {
            return Ok(None);
        };
        let record = reader
            .read_record(offset as u64, ReadHint::Random)
            .map_err(|source| LogError::read(&self.path, source))?;
        Ok(Some(Record {
            position: Position::new(self.base_position.get() + local as u64),
            data: record.data.into_owned(),
        }))
    }
}

/// A source of ordered, position-disjoint segments for a [`Scan`]: either the live
/// [`SegmentSet`] (writer side) or an immutable read snapshot. Extracting this keeps the
/// zero-copy segment-rolling scan (the highest-risk logic in layer 1) as **one**
/// implementation shared by both sides, rather than a second copy over the same bytes.
///
/// Segments are addressed by a logical index: sealed segments first (`0..segment_count`),
/// then the active segment at `segment_count`.
pub trait SegmentSource {
    /// Bytes reserved at the start of every segment for its header.
    fn header_size(&self) -> u64;

    /// Number of sealed segments; the active segment sits at this index.
    fn segment_count(&self) -> usize;

    /// The segment at logical index `idx`, or `None` past the active one.
    fn segment_at(&self, idx: usize) -> Option<&Arc<Segment>>;

    /// Locates the segment owning `pos`: its logical index and a handle. `None` if `pos`
    /// is the empty sentinel or precedes the first segment. Binary search over the
    /// monotonic base positions, then the active segment.
    fn locate(&self, pos: Position) -> Option<(usize, &Arc<Segment>)> {
        if pos == Position::ZERO {
            return None;
        }
        let active_idx = self.segment_count();
        if let Some(active) = self.segment_at(active_idx)
            && pos >= active.base_position
        {
            return Some((active_idx, active));
        }
        // Binary search the sealed segments for the last base <= pos.
        let mut lo = 0usize;
        let mut hi = active_idx; // exclusive
        let mut found = None;
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            let base = self.segment_at(mid)?.base_position;
            if base <= pos {
                found = Some(mid);
                lo = mid + 1;
            } else {
                hi = mid;
            }
        }
        found.and_then(|idx| self.segment_at(idx).map(|seg| (idx, seg)))
    }
}

impl fmt::Debug for Segment {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Segment")
            .field("base_position", &self.base_position)
            .field("path", &self.path)
            .field("event_count", &self.offsets.read().unwrap().len())
            .finish_non_exhaustive()
    }
}

/// A record read back out of the log: its global position and payload.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Record {
    pub position: Position,
    pub data: Vec<u8>,
}

/// A borrowed view of a record yielded by [`Scan`], pointing directly into the
/// reader's read-ahead buffer for zero-copy sequential scans. It is valid only
/// until the next [`Scan::next`] call; use [`to_owned`](RecordRef::to_owned) to
/// keep it beyond that.
#[derive(Clone, Copy, Debug)]
pub struct RecordRef<'a> {
    pub position: Position,
    pub data: &'a [u8],
}

impl RecordRef<'_> {
    /// Copies the view into an owned [`Record`].
    pub fn to_owned(&self) -> Record {
        Record {
            position: self.position,
            data: self.data.to_vec(),
        }
    }
}

/// The inclusive range of positions assigned to an appended batch.
///
/// A batch always contains at least one record, so a range always covers at least
/// one position; there is no empty range.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PositionRange {
    pub first: Position,
    pub last: Position,
}

impl PositionRange {
    /// Number of positions in the range (always at least 1).
    pub fn count(&self) -> u64 {
        (self.last - self.first) + 1
    }
}

/// Owns the collection of segment files and the global-position addressing over them.
/// How this process relates to the one live segment.
///
/// One type rather than a `SegmentSet`/`ReadOnlySegments` split: everything above layer 1
/// (`Snapshot::capture`, `IndexSet`, `ReadCore`, the condition evaluator) takes `&SegmentSet`
/// concretely, and the open path's invariant checks (contiguity, the base-position
/// cross-check, the committed scan) are the same either way. Duplicating them is how they
/// drift, and a drift there is silently wrong data. The cost is one branch per batch, never
/// per record, against the fsync on the same path.
#[derive(Debug)]
enum ActiveEnd {
    /// This process holds the directory's write lock and appends.
    Writer(Writer<0>),
    /// Another process owns the log; this one follows its committed prefix.
    Follower(Tail),
}

#[derive(Debug)]
pub struct SegmentSet {
    dir: PathBuf,
    config: SegmentConfig,
    /// Sealed, immutable segments ordered by `base_position`.
    sealed: Vec<Arc<Segment>>,
    /// The single active (writable) segment.
    active: Arc<Segment>,
    /// How this process relates to the active segment. Kept out of `Segment` because
    /// `Segment` is shared read-only via `Arc`; only the set writes or advances.
    active_end: ActiveEnd,
    /// The next global position to assign. A fresh log starts at [`FIRST_POSITION`].
    next_position: Position,
    /// The held write lock. Dropping the set releases it, so it outlives every mutation of
    /// the directory. `None` only for a follower, which takes no lock at all.
    _lock: Option<DirLock>,
}

impl SegmentSet {
    /// Opens (or creates) the segment set rooted at `dir`.
    ///
    /// On an empty directory this creates the first segment (base position 1).
    /// Otherwise it reads every segment header, verifies the base-position chain is
    /// contiguous, and runs crash recovery on the last (active) segment, rolling
    /// back any incomplete trailing batch. Any gap, overlap, or corruption is a hard
    /// error: the set refuses to open rather than guess.
    pub fn open(dir: impl AsRef<Path>, config: SegmentConfig) -> Result<Self, LogError> {
        config.validate()?;
        let dir = dir.as_ref().to_path_buf();

        // 1. Create dir if absent, and make its directory entry durable.
        if !dir.exists() {
            fs::create_dir_all(&dir).map_err(|source| LogError::io(&dir, source))?;
            if let Some(parent) = dir.parent()
                && !parent.as_os_str().is_empty()
            {
                sync_dir(parent).map_err(|source| LogError::io(parent, source))?;
            }
        }

        // 2. Take the write lock before anything below can touch the directory: the steps
        //    that follow delete files, create segments, and recover the active segment, none
        //    of which is safe with a second writer present.
        let lock = Some(acquire_lock(&dir)?);

        // 3. Read the directory, then split off the trailing run whose creation did not
        //    finish. A writer owns the directory, so any header that is present but invalid
        //    is corruption wherever it sits, never a race.
        let SegmentSplit { valid, unready } = split_trailing_unready(list_segments(&dir)?, true)?;

        // 4. Each unfinished trailing segment holds no committed data. Delete it.
        for path in &unready {
            #[cfg(feature = "tracing")]
            tracing::warn!(
                "deleting unwritten trailing segment {path:?} (creation did not finish)"
            );
            fs::remove_file(path).map_err(|source| LogError::io(path, source))?;
            sync_dir(&dir).map_err(|source| LogError::io(&dir, source))?;
        }

        // 7. Empty directory (or only an unwritten file we just deleted): fresh log.
        if valid.is_empty() {
            let (writer, active) =
                Self::create_segment(&dir, &config, Position::new(FIRST_POSITION))?;
            #[cfg(feature = "tracing")]
            tracing::trace!("initialized empty segment set at {dir:?}");
            return Ok(SegmentSet {
                dir,
                config,
                sealed: Vec::new(),
                active,
                active_end: ActiveEnd::Writer(writer),
                next_position: Position::new(FIRST_POSITION),
                _lock: lock,
            });
        }

        // 4 & 5. Build sealed segments (scan to count events + rebuild sidecar),
        //         verifying contiguity, then recover the active segment.
        let (active_entry, sealed_entries) = valid.split_last().unwrap();
        let (sealed, expected_base) = open_sealed_chain(sealed_entries, config.header_size as u64)?;

        let (active_base, active_path) = active_entry;
        if *active_base != expected_base {
            return Err(LogError::NonContiguous {
                path: active_path.clone(),
                found: *active_base,
                expected: expected_base,
            });
        }

        // 6. Recovery: reopen the active segment for writing, rolling back any
        //    incomplete trailing batch to the last valid commit point.
        let mut writer =
            Writer::<0>::open(active_path, config.segment_size, config.header_size as u64)
                .map_err(|source| LogError::write(active_path, source))?;
        configure_writer(&mut writer, &config);

        #[cfg(feature = "tracing")]
        {
            let committed = writer.write_offset();
            if trailing_bytes_present(&writer, committed, config.segment_size) {
                tracing::warn!(
                    "segment {active_path:?} recovered with rollback, discarding bytes from offset {committed}"
                );
            } else {
                tracing::trace!("segment {active_path:?} opened cleanly at offset {committed}");
            }
        }

        let flushed = writer.flushed_offset();
        let offsets = scan_offsets(active_path, flushed.clone(), config.header_size as u64)?;
        let count = offsets.len() as u64;

        // Cross-check the recovered commit marker against the event count. Position
        // assignment is contiguous from the base, so the last marker's highest
        // position must be base + count - 1.
        if let Some(highest) = writer.last_committed_position()
            && highest + 1 != *active_base + count
        {
            return Err(LogError::PositionMismatch {
                path: active_path.clone(),
                found: Position::new(highest),
                expected: Position::new(*active_base + count - 1),
            });
        }

        let next_position = Position::new(*active_base + count);
        let active = Arc::new(Segment {
            base_position: *active_base,
            path: active_path.clone(),
            flushed_offset: flushed,
            offsets: RwLock::new(offsets),
            reader: Mutex::new(None),
        });

        Ok(SegmentSet {
            dir,
            config,
            sealed,
            active,
            active_end: ActiveEnd::Writer(writer),
            next_position,
            _lock: lock,
        })
    }

    /// Opens the segment set rooted at `dir` read-only, to follow a writer in another
    /// process.
    ///
    /// Mutates nothing: no directory is created, no unfinished segment is deleted, no
    /// segment is created, and every file is opened `O_RDONLY`. The directory may therefore
    /// be a read-only mount, and it needs no lock, since correctness against a live writer
    /// comes from the commit markers and from segments never being rewritten, not from
    /// exclusion. A follower that took even a shared lock would block the writer.
    ///
    /// `config.segment_size` is ignored: each segment's real size comes from the file, which
    /// is authoritative because segments are `fallocate`d at creation and never extended.
    /// Only `header_size` is used.
    ///
    /// A directory with no readable segment is [`LogError::Uninitialized`], not an empty
    /// set. A follower that raced the writer's very first segment would otherwise report
    /// "no events" for a store that is merely not ready, and on this layer an error must
    /// never look like end-of-stream. Callers retry.
    pub fn open_read_only(dir: impl AsRef<Path>, config: SegmentConfig) -> Result<Self, LogError> {
        config.validate()?;
        let dir = dir.as_ref().to_path_buf();

        // Tolerant classification: the trailing segment may be mid-creation right now.
        let SegmentSplit { valid, unready } = split_trailing_unready(list_segments(&dir)?, false)?;
        if valid.is_empty() {
            return Err(LogError::Uninitialized { dir });
        }
        #[cfg(feature = "tracing")]
        for path in &unready {
            tracing::debug!("segment {path:?} has no readable header yet; skipping for now");
        }
        let _ = unready;

        let (active_entry, sealed_entries) = valid.split_last().unwrap();
        let (sealed, expected_base) = open_sealed_chain(sealed_entries, config.header_size as u64)?;

        let (active_base, active_path) = active_entry;
        if *active_base != expected_base {
            return Err(LogError::NonContiguous {
                path: active_path.clone(),
                found: *active_base,
                expected: expected_base,
            });
        }

        let (tail, active, next_position) =
            Self::follow_segment(active_path, *active_base, &config)?;

        #[cfg(feature = "tracing")]
        tracing::debug!("following segment set at {dir:?} from position {next_position}");

        Ok(SegmentSet {
            dir,
            config,
            sealed,
            active,
            active_end: ActiveEnd::Follower(tail),
            next_position,
            _lock: None,
        })
    }

    /// Opens one segment as a follower and drains whatever is already committed in it.
    fn follow_segment(
        path: &Path,
        base: Position,
        config: &SegmentConfig,
    ) -> Result<(Tail, Arc<Segment>, Position), LogError> {
        let mut tail = Tail::open(path, config.header_size as u64)
            .map_err(|source| LogError::tail(path, source))?;
        let mut offsets = Vec::new();
        let progress = tail
            .poll(|offset| {
                offsets.push(
                    u32::try_from(offset)
                        .expect("segment_size <= u32::MAX enforced by SegmentConfig::validate"),
                );
            })
            .map_err(|source| LogError::tail(path, source))?;
        let count = offsets.len() as u64;
        if let Some(highest) = progress.last_position
            && highest + 1 != base.get() + count
        {
            return Err(LogError::PositionMismatch {
                path: path.to_path_buf(),
                found: Position::new(highest),
                expected: Position::new(base.get() + count - 1),
            });
        }
        let segment = Arc::new(Segment {
            base_position: base,
            path: path.to_path_buf(),
            flushed_offset: tail.flushed_offset(),
            offsets: RwLock::new(offsets),
            reader: Mutex::new(None),
        });
        Ok((tail, segment, Position::new(base.get() + count)))
    }

    /// Advances a follower to the writer's current committed prefix.
    ///
    /// Ordering matters and is the mirror of the writer's own commit seam: the active
    /// segment's sidecar grows before its flushed extent does, and a segment is sealed only
    /// once it has been drained, so a caller that publishes a watermark after this returns
    /// hands readers a snapshot that already covers it.
    ///
    /// The directory is listed on every call, because a rollover is visible nowhere else.
    /// Skipping the listing when the active tail produced records would save a `read_dir` at
    /// the cost of leaving the follower an interval behind whenever new records and a
    /// rollover land together, which is exactly when it is busiest. One listing covers every
    /// rollover since the last call, so a follower far behind catches up in one refresh.
    pub fn refresh(&mut self) -> Result<Refreshed, LogError> {
        if !self.is_read_only() {
            return Err(LogError::ReadOnly {
                dir: self.dir.clone(),
                op: "refresh",
            });
        }

        let mut sealed_added = 0;

        // 1. Take whatever the active segment has committed since the last refresh.
        self.drain_active()?;

        // 2. A rollover is visible only in the directory, so look every time. Checking only
        //    when the tail came up empty would save a `read_dir`, but it leaves `refresh` an
        //    interval behind whenever new records and a rollover land in the same interval,
        //    which is precisely when a follower is busiest.
        let SegmentSplit { valid, unready } =
            split_trailing_unready(list_segments(&self.dir)?, false)?;
        let pending_segment = unready.into_iter().next_back();

        // One listing covers every rollover the writer managed since the last refresh, so a
        // follower far behind catches up in a single call rather than one segment per call.
        for (base, path) in valid {
            if base <= self.active.base_position {
                continue;
            }
            // The writer syncs the outgoing segment before creating its successor, so once
            // the successor's directory entry exists the predecessor's committed bytes are
            // final. Drain it to that end before sealing it.
            self.drain_active()?;
            let expected =
                Position::new(self.active.base_position.get() + self.active.event_count());
            if base != expected {
                return Err(LogError::NonContiguous {
                    path,
                    found: base,
                    expected,
                });
            }

            let (tail, new_active, next_position) =
                Self::follow_segment(&path, base, &self.config)?;
            let old_active = mem::replace(&mut self.active, new_active);
            self.sealed.push(old_active);
            self.active_end = ActiveEnd::Follower(tail);
            self.next_position = next_position;
            sealed_added += 1;

            #[cfg(feature = "tracing")]
            tracing::debug!("follower adopted segment with base_position {base}");
        }

        Ok(Refreshed {
            tip: self.last_position(),
            sealed_added,
            pending_segment,
        })
    }

    /// Polls the active segment's tail, extending its sidecar with whatever is newly
    /// committed. Returns how many records that was.
    fn drain_active(&mut self) -> Result<u64, LogError> {
        let ActiveEnd::Follower(tail) = &mut self.active_end else {
            return Err(LogError::ReadOnly {
                dir: self.dir.clone(),
                op: "refresh",
            });
        };
        let mut fresh = Vec::new();
        let progress = tail
            .poll(|offset| {
                fresh.push(
                    u32::try_from(offset)
                        .expect("segment_size <= u32::MAX enforced by SegmentConfig::validate"),
                );
            })
            .map_err(|source| LogError::tail(&self.active.path, source))?;
        let added = progress.committed_records;
        if added > 0 {
            self.active
                .offsets
                .write()
                .unwrap()
                .extend_from_slice(&fresh);
            self.next_position = Position::new(self.next_position.get() + added);
        }

        let base = self.active.base_position.get();
        let count = self.active.event_count();
        if let Some(highest) = progress.last_position
            && highest + 1 != base + count
        {
            return Err(LogError::PositionMismatch {
                path: self.active.path.clone(),
                found: Position::new(highest),
                expected: Position::new(base + count - 1),
            });
        }
        Ok(added)
    }

    /// The active segment's writer, or an error if this set is following someone else's log.
    ///
    /// Defensive: [`Follower`](crate::follow::Follower) keeps its read-only set private and
    /// never offers an append, so this is not reachable through the public API.
    fn writer_mut(&mut self) -> Result<&mut Writer<0>, LogError> {
        match &mut self.active_end {
            ActiveEnd::Writer(writer) => Ok(writer),
            ActiveEnd::Follower(_) => Err(LogError::ReadOnly {
                dir: self.dir.clone(),
                op: "append",
            }),
        }
    }

    /// Whether this set follows another process's log rather than owning it.
    pub fn is_read_only(&self) -> bool {
        matches!(self.active_end, ActiveEnd::Follower(_))
    }

    /// Appends a batch of records as a single durable unit and returns the range
    /// of positions assigned. Called only by the write coordinator, single-threaded.
    ///
    /// A batch never spans segments: if it does not fit in the active segment's
    /// remaining space, the set rolls over first. A batch that cannot fit in an
    /// empty segment is rejected rather than looping.
    ///
    /// The append is all-or-nothing: if any record or the commit fails midway, the
    /// writer is rewound so no orphan records are left for the next batch to adopt.
    pub fn append_batch(&mut self, records: &[&[u8]]) -> Result<PositionRange, LogError> {
        if records.is_empty() {
            return Err(LogError::EmptyBatch);
        }

        // 1. Reject empty records (their zero-length frame collides with the
        //    zero-filled segment tail) and records over the configured maximum.
        for record in records {
            if record.is_empty() {
                return Err(LogError::EmptyRecord);
            }
            let record_len = RECORD_HEAD_SIZE + record.len();
            if record_len > self.config.max_record_len {
                return Err(LogError::RecordTooLarge {
                    size: record_len,
                    max: self.config.max_record_len,
                });
            }
        }

        // 2. Total encoded size, including the trailing commit marker.
        let records_len: usize = records.iter().map(|r| RECORD_HEAD_SIZE + r.len()).sum();
        let total_size = records_len + RECORD_HEAD_SIZE + COMMIT_MARKER_PAYLOAD;

        // A batch that can never fit even a fresh segment is a hard error.
        let capacity = self.config.segment_size - self.config.header_size;
        if total_size > capacity {
            return Err(LogError::BatchTooLarge {
                size: total_size,
                capacity,
            });
        }

        // 3. Roll over first if it does not fit in the active segment.
        if total_size as u64 > self.writer_mut()?.remaining_bytes() {
            self.rollover()?;
        }

        // 4. Append records, then a commit marker carrying the highest position,
        //    made durable together by the single sync inside `commit`. On any
        //    failure, rewind so the file matches our in-memory view.
        let first = self.next_position;
        let last = Position::new(first + records.len() as u64 - 1);
        let writer = self.writer_mut()?;
        let rewind_to = writer.write_offset();

        // The error path wants the segment's path, but naming it up front would borrow `self`
        // alongside the writer, so it is resolved below instead: the hot path never touches it.
        let mut new_offsets = Vec::with_capacity(records.len());
        let outcome = (|| -> Result<(), WriteError> {
            for record in records {
                let (offset, _len) = writer.append_data(record)?;
                new_offsets.push(
                    u32::try_from(offset)
                        .expect("segment_size <= u32::MAX enforced by SegmentConfig::validate"),
                );
            }
            writer.commit(last.get()).map(|_| ())
        })();

        if let Err(err) = outcome {
            // Discard the partial batch. If even the rewind fails the writer is
            // wedged, so surface that; otherwise surface the original error.
            let rewound = writer.rewind_to(rewind_to);
            let path = &self.active.path;
            rewound.map_err(|source| LogError::write(path, source))?;
            return Err(LogError::write(path, err));
        }

        // 5. Extend the active segment's in-memory sidecar.
        self.active
            .offsets
            .write()
            .unwrap()
            .extend_from_slice(&new_offsets);

        // 6. Advance and return.
        self.next_position = last.next();
        Ok(PositionRange { first, last })
    }

    /// Seals the active segment and installs a fresh one at `next_position`.
    fn rollover(&mut self) -> Result<(), LogError> {
        // Seal: everything is already synced (the previous batch ended in a commit),
        // but sync defensively before dropping the writer.
        if let Err(source) = self.writer_mut()?.sync() {
            return Err(LogError::write(&self.active.path, source));
        }

        let (writer, new_active) =
            Self::create_segment(&self.dir, &self.config, self.next_position)?;

        // Crash point: mid rollover. The new segment file exists and its header is fsynced,
        // but the batch that triggered the rollover has not been committed to it yet (there is
        // no separate manifest here: the filename plus header is the record of the segment).
        // Recovery must accept a trailing header-only segment with zero events.
        seglog::crash_point!("segment_created_before_commit");

        let old_active = mem::replace(&mut self.active, new_active);
        self.sealed.push(old_active);
        self.active_end = ActiveEnd::Writer(writer);

        #[cfg(feature = "tracing")]
        tracing::trace!(
            "rolled over to segment with base_position {}",
            self.next_position
        );
        Ok(())
    }

    /// Creates a new segment file: fallocate + write header + sync.
    fn create_segment(
        dir: &Path,
        config: &SegmentConfig,
        base: Position,
    ) -> Result<(Writer<0>, Arc<Segment>), LogError> {
        let path = dir.join(segment_file_name(base));

        // `create` fallocates (zero-filling) then makes the file and its directory
        // entry durable, so the file reads back as an unwritten segment until we
        // write the header. Writing the header only changes file content, not the
        // directory entry, so `sync_all` on the file is enough, no second fsync
        // of the directory is needed.
        let mut writer = Writer::<0>::create(&path, config.segment_size, config.header_size as u64)
            .map_err(|source| LogError::write(&path, source))?;
        configure_writer(&mut writer, config);

        let header = SegmentHeader::new(base);
        writer
            .file()
            .write_all_at(&header.to_bytes(), 0)
            .map_err(|source| LogError::io(&path, source))?;
        writer
            .file()
            .sync_all()
            .map_err(|source| LogError::io(&path, source))?;

        let segment = Arc::new(Segment {
            base_position: base,
            path,
            flushed_offset: writer.flushed_offset(),
            offsets: RwLock::new(Vec::new()),
            reader: Mutex::new(None),
        });
        Ok((writer, segment))
    }

    /// Reads a single record at `pos`. Optimized for a random access pattern.
    pub fn read_at(&self, pos: Position) -> Result<Record, LogError> {
        let segment = self
            .segment_for(pos)
            .ok_or(LogError::NotFound { position: pos })?;

        let local = pos.offset_from(segment.base_position) as usize;
        let offset = {
            let offsets = segment.offsets.read().unwrap();
            match offsets.get(local) {
                Some(offset) => *offset as u64,
                None => return Err(LogError::NotFound { position: pos }),
            }
        };

        // Reuse the segment's cached reader (one open fd per segment).
        let mut guard = segment.reader.lock().unwrap();
        if guard.is_none() {
            *guard = Some(segment.open_reader()?);
        }
        let record = guard
            .as_mut()
            .unwrap()
            .read_record(offset, ReadHint::Random)
            .map_err(|source| LogError::read(&segment.path, source))?;
        Ok(Record {
            position: pos,
            data: record.data.into_owned(),
        })
    }

    /// Returns a sequential scan of every record at or after `pos` (inclusive),
    /// rolling across segment boundaries and skipping control records silently.
    ///
    /// `pos` is clamped up to the first position, so `scan_from(Position::ZERO)` (the
    /// "before everything" empty sentinel) scans the whole log rather than nothing.
    /// It is a thin inclusive wrapper over [`scan_after`](Self::scan_after).
    pub fn scan_from(&self, pos: Position) -> Scan<&SegmentSet> {
        self.scan_at(pos.max(Position::new(FIRST_POSITION)))
    }

    /// Returns a sequential scan of every record strictly after `pos` (exclusive).
    ///
    /// This is the natural primitive for subscriptions, which hold "the last
    /// position I processed": `scan_after(Position::ZERO)` scans the whole log with no
    /// sentinel special case, and `scan_after(last)` resumes right after `last`.
    pub fn scan_after(&self, pos: Position) -> Scan<&SegmentSet> {
        self.scan_at(Position::new(pos.get().saturating_add(1)))
    }

    /// Returns a reverse scan of every record in `[first, upto]`, **descending**, rolling
    /// across segment boundaries and skipping control records. `upto` is clamped down to the
    /// live tip, so `scan_back(Position::new(1), Position::MAX)` walks the whole log
    /// newest-first. The counterpart to [`scan_from`](Self::scan_from).
    pub fn scan_back(&self, first: Position, upto: Position) -> ScanBack<&SegmentSet> {
        ScanBack::start(self, first, upto.min(self.last_position()))
    }

    /// Core scan constructor: emits records beginning at `first` (inclusive), up to the
    /// live tip. The writer scans its own live log, so the upper bound is
    /// [`last_position`](Self::last_position).
    fn scan_at(&self, first: Position) -> Scan<&SegmentSet> {
        Scan::start(self, first, self.last_position())
    }

    /// The highest assigned position, or `Position::ZERO` if the log is empty.
    pub fn last_position(&self) -> Position {
        // Positions are 1-based and `next_position >= FIRST_POSITION`, so this never
        // underflows; an empty log yields `Position::ZERO`, the empty sentinel.
        Position::new(self.next_position - 1)
    }

    /// The next position that will be assigned.
    pub fn next_position(&self) -> Position {
        self.next_position
    }

    /// Largest batch (records plus commit marker) that can fit an empty segment. The
    /// write coordinator budgets against this so a multi-request batch always fits.
    pub fn segment_capacity(&self) -> usize {
        self.config.segment_size - self.config.header_size
    }

    /// Largest a single record may be. A batch containing a larger record is rejected.
    pub fn max_record_len(&self) -> usize {
        self.config.max_record_len
    }

    /// Number of sealed (immutable) segments.
    pub fn sealed_len(&self) -> usize {
        self.sealed.len()
    }

    /// The sealed segments in base order. Cloned into a read snapshot so off-thread
    /// readers hold the same immutable `Arc<Segment>`s the writer sealed.
    pub fn sealed_arcs(&self) -> &[Arc<Segment>] {
        &self.sealed
    }

    /// A handle to the active segment. Its offset sidecar updates live under an existing
    /// reader, so a snapshot taken now still sees records appended before the reader's
    /// watermark.
    pub fn active_arc(&self) -> Arc<Segment> {
        Arc::clone(&self.active)
    }

    /// The directory holding the segment files. The index layer roots its own segments
    /// under this (`{dir}/index`) so it aligns to the log one-for-one.
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// The base (first) position of the active segment. The index layer rebuilds the
    /// active segment's tail index by scanning from here on open.
    pub fn active_base(&self) -> Position {
        self.active.base_position
    }

    /// Each sealed segment's `(base_position, event_count)`, in order. The index layer
    /// pairs one on-disk index segment with each of these, pruning and rebuilding by the
    /// same disjoint ranges the log uses.
    pub fn sealed_segments(&self) -> impl Iterator<Item = (Position, u64)> + '_ {
        self.sealed
            .iter()
            .map(|s| (s.base_position(), s.event_count()))
    }

    /// Resolves the segment owning `pos`, or `None` if out of range. The segment lookup
    /// itself is [`SegmentSource::locate`] (shared with the read path); this adds the
    /// upper bound, since a position at or past `next_position` has not been assigned.
    pub fn segment_for(&self, pos: Position) -> Option<&Arc<Segment>> {
        if pos >= self.next_position {
            return None;
        }
        self.locate(pos).map(|(_, segment)| segment)
    }
}

impl SegmentSource for SegmentSet {
    fn header_size(&self) -> u64 {
        self.config.header_size as u64
    }

    fn segment_count(&self) -> usize {
        self.sealed.len()
    }

    /// The segment at logical index `idx`: sealed segments first, then the active
    /// one at `sealed.len()`.
    fn segment_at(&self, idx: usize) -> Option<&Arc<Segment>> {
        match idx.cmp(&self.sealed.len()) {
            Ordering::Less => Some(&self.sealed[idx]),
            Ordering::Equal => Some(&self.active),
            Ordering::Greater => None,
        }
    }
}

/// Wrapper sources forward to their target, so the one scan serves a borrow of the live
/// [`SegmentSet`] (`Scan<&SegmentSet>`, writer side) and an owned snapshot
/// (`Scan<Arc<Snapshot>>`, reader side) alike. One macro keeps the forwards in lockstep, so
/// a new [`SegmentSource`] method is added in exactly one place.
macro_rules! forward_segment_source {
    ($wrapper:ty) => {
        impl<T: SegmentSource + ?Sized> SegmentSource for $wrapper {
            fn header_size(&self) -> u64 {
                (**self).header_size()
            }
            fn segment_count(&self) -> usize {
                (**self).segment_count()
            }
            fn segment_at(&self, idx: usize) -> Option<&Arc<Segment>> {
                (**self).segment_at(idx)
            }
        }
    };
}

forward_segment_source!(&T);
forward_segment_source!(Arc<T>);

/// Sequential scan over the log, starting from a position and rolling across
/// segment boundaries. Yields records in position order; control records are
/// skipped silently, and it never reads past the active segment's flushed point.
///
/// A failure to open a segment or read a record is surfaced as an `Err` item and
/// terminates the scan: it never looks like a clean end-of-stream.
///
/// The scan **owns** its [`SegmentSource`] (`S`), so it can be either a borrow of the live
/// [`SegmentSet`] (`Scan<&SegmentSet>`, writer side) or an owned read snapshot
/// (`Scan<Arc<Snapshot>>`, reader side) that keeps its segments alive for the scan's whole
/// lifetime with no self-referential borrow. Blanket impls of [`SegmentSource`] for `&T`
/// and `Arc<T>` make both forms work through the one implementation.
pub struct Scan<S: SegmentSource> {
    source: S,
    /// Logical index of the segment currently being read (see [`SegmentSource::segment_at`]).
    seg_idx: usize,
    /// Byte offset within the current segment of the next record to read.
    offset: u64,
    /// Global position of the next record to emit.
    position: Position,
    /// Highest position to emit (inclusive). The writer bounds to its live tip; a reader
    /// snapshot bounds to its pinned watermark, so it never reads past what was durable
    /// (and index-fed) when the scan began.
    upto: Position,
    /// The reader for the current segment. `Reader` owns its 64 KB read-ahead
    /// buffer, so keeping it here (rather than reopening per record) is what makes
    /// the scan do roughly one syscall per read-ahead window, not one per record.
    reader: Option<Reader<0>>,
    /// A setup error to surface as the first (and only) item.
    pending_err: Option<LogError>,
    done: bool,
}

impl<S: SegmentSource> Scan<S> {
    /// Starts a scan of `source` emitting records from `first` (inclusive) up to `upto`
    /// (inclusive). `first == Position::ZERO` or `first > upto` yields an empty stream (a
    /// caught-up subscription is a normal, non-error state); a `first` that no segment
    /// covers surfaces `NotFound` as the sole item.
    pub(crate) fn start(source: S, first: Position, upto: Position) -> Self {
        if first == Position::ZERO || first > upto {
            return Scan::empty(source);
        }
        let (seg_idx, offset) = match source.locate(first) {
            Some((seg_idx, segment)) => {
                let local = first.offset_from(segment.base_position) as usize;
                match segment.data_offset(local) {
                    Some(offset) => (seg_idx, offset),
                    None => return Scan::failed(source, LogError::NotFound { position: first }),
                }
            }
            None => return Scan::failed(source, LogError::NotFound { position: first }),
        };
        let reader = match source.segment_at(seg_idx).unwrap().open_reader() {
            Ok(reader) => reader,
            Err(err) => return Scan::failed(source, err),
        };
        Scan {
            source,
            seg_idx,
            offset: offset as u64,
            position: first,
            upto,
            reader: Some(reader),
            pending_err: None,
            done: false,
        }
    }

    fn empty(source: S) -> Self {
        Scan {
            source,
            seg_idx: 0,
            offset: 0,
            position: Position::ZERO,
            upto: Position::ZERO,
            reader: None,
            pending_err: None,
            done: true,
        }
    }

    fn failed(source: S, err: LogError) -> Self {
        Scan {
            pending_err: Some(err),
            done: false,
            ..Scan::empty(source)
        }
    }

    /// Moves to the next segment, opening its reader and pointing at its first
    /// record. Returns `false` when there are no more segments.
    fn advance_segment(&mut self) -> Result<bool, LogError> {
        let next_idx = self.seg_idx + 1;
        let Some(segment) = self.source.segment_at(next_idx) else {
            return Ok(false);
        };
        self.reader = Some(segment.open_reader()?);
        self.offset = self.source.header_size();
        self.seg_idx = next_idx;
        Ok(true)
    }
}

impl<S: SegmentSource> Scan<S> {
    /// Advances to the next record and returns a view borrowing the reader's
    /// read-ahead buffer.
    ///
    /// This is a *lending* iterator, so it is not `std::iter::Iterator` (which can't
    /// yield a borrow of itself). Consume it with
    /// `while let Some(item) = scan.next() { … }`; the returned [`RecordRef`] is
    /// valid only until the following `next` call.
    ///
    /// Returns `None` at the end of the log; a read failure is surfaced once as an
    /// `Err` item and then terminates the scan.
    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<Result<RecordRef<'_>, LogError>> {
        if let Some(err) = self.pending_err.take() {
            self.done = true;
            return Some(Err(err));
        }
        if self.done {
            return None;
        }
        // Stop at the upper bound: `self.position` is always the next data record to emit,
        // so a reader snapshot never yields past its pinned watermark.
        if self.position > self.upto {
            self.done = true;
            return None;
        }

        // Step 1: position the cursor on the next data record, skipping control
        // records and rolling across segments. This is header-only (`peek`), so it
        // holds no payload borrow while it swaps readers, which is what lets the
        // borrowing read in step 2 return a view without fighting the borrow checker
        // at a segment boundary.
        let total_len = match self.position_at_data() {
            Ok(Some(total_len)) => total_len,
            Ok(None) => {
                self.done = true;
                return None;
            }
            Err(err) => {
                self.done = true;
                return Some(Err(err));
            }
        };

        // Advance the cursor *before* the borrowing read, so no `self` field is
        // written while the returned view borrows the reader. `total_len` from the
        // header matches the record's framed length, so this is exact.
        let position = self.position;
        let offset = self.offset;
        self.offset = offset + total_len as u64;
        self.position = position.next();

        // Step 2: one borrowing read of the data record. Sequential reads always
        // borrow the read-ahead buffer (locked by seglog's
        // `test_sequential_read_borrows_even_large_records`), so `data` is a slice
        // into it, zero copy.
        let source = &self.source;
        let seg_idx = self.seg_idx;
        let reader = self.reader.as_mut().unwrap();
        match reader.read_record(offset, ReadHint::Sequential) {
            Ok(record) => {
                let data = match record.data {
                    Cow::Borrowed(bytes) => bytes,
                    Cow::Owned(_) => {
                        // Pinned by seglog's `test_sequential_read_borrows_even_large_records`:
                        // a `ReadHint::Sequential` read always returns `Cow::Borrowed`,
                        // even for payloads larger than the optimistic/fallback buffers.
                        unreachable!(
                            "sequential reads borrow the read-ahead buffer \
                             (seglog::test_sequential_read_borrows_even_large_records)"
                        )
                    }
                };
                Some(Ok(RecordRef { position, data }))
            }
            Err(err) => {
                self.done = true;
                Some(Err(LogError::read(scan_segment_path(source, seg_idx), err)))
            }
        }
    }

    /// Positions the cursor on the next data record, skipping control records and
    /// rolling across segment boundaries. Returns the record's framed length, or
    /// `Ok(None)` when the log is exhausted. Header-only, so it holds no payload
    /// borrow while it swaps readers.
    fn position_at_data(&mut self) -> Result<Option<usize>, LogError> {
        loop {
            if self.reader.is_none() && !self.advance_segment()? {
                return Ok(None);
            }
            let seg_idx = self.seg_idx;
            let path = scan_segment_path(&self.source, seg_idx);
            let reader = self.reader.as_mut().unwrap();
            let kind = reader
                .peek(self.offset)
                .map_err(|err| LogError::read(path, err))?;
            match kind {
                RecordKind::Data { total_len } => return Ok(Some(total_len)),
                RecordKind::Control { total_len } => self.offset += total_len as u64,
                RecordKind::End => self.reader = None, // advance on the next iteration
            }
        }
    }
}

/// The path of the segment at logical index `idx`, for error reporting.
fn scan_segment_path<S: SegmentSource>(source: &S, idx: usize) -> PathBuf {
    source
        .segment_at(idx)
        .map(|segment| segment.path.clone())
        .unwrap_or_default()
}

/// Target size of one reverse-scan window, matching `seglog`'s forward read-ahead window. A
/// [`ScanBack`] reads records in windows this large and emits them high-to-low within each,
/// so the I/O stays large-block sequential (only the emit order reverses) and the forward
/// read-ahead's throughput is preserved going backward.
const REVERSE_WINDOW_BYTES: usize = 64 * 1024;

/// Reverse counterpart to [`Scan`]: emits data records in `[first, upto]` in **descending**
/// position order, rolling backward across segment boundaries and skipping control records.
///
/// Where [`Scan`] streams forward through `seglog`'s read-ahead buffer, `ScanBack` reads each
/// segment's records in ~64 KB windows (`REVERSE_WINDOW_BYTES`) using the offset sidecar for
/// record boundaries: one positioned read per window, records parsed and yielded from the
/// window buffer high-to-low. Control records interleaved between data records are skipped for
/// free, because the sidecar only maps *data* records and `ScanBack` parses only at those
/// offsets. Like [`Scan`] it is a *lending* iterator (the yielded [`RecordRef`] borrows the
/// window buffer and is valid only until the next [`next`](ScanBack::next)) and owns its
/// [`SegmentSource`], so it works over both a borrow of the live [`SegmentSet`] and an owned
/// read snapshot.
///
/// The forward path is untouched: `ScanBack` shares only the read-only offset sidecar and the
/// borrowing parser, so supporting backwards reads cannot slow forward reads down.
pub struct ScanBack<S: SegmentSource> {
    source: S,
    /// Lowest global position to emit (inclusive).
    first: Position,
    /// Global position of the next record to emit; descends from `upto`. When it drops below
    /// `first` the scan is exhausted.
    position: Position,
    /// Target window size in bytes (tunable for tests; [`REVERSE_WINDOW_BYTES`] by default).
    window_bytes: usize,

    /// The segment currently being read, its logical index, base, event count, an open reader,
    /// and the reader's flushed byte extent. `None` until the first `next` (and after a segment
    /// is exhausted downward), forcing a re-locate.
    segment: Option<Arc<Segment>>,
    seg_idx: usize,
    seg_base: Position,
    seg_count: u64,
    reader: Option<Reader<0>>,
    flushed_len: u64,

    /// The window buffer, reused across windows. It holds the bytes `[win_base_byte,
    /// win_base_byte + window.len())`; `win_lo_local` is the lowest local it covers, or `None`
    /// when no window is loaded (a fresh segment, or the cursor dropped below the window).
    window: Vec<u8>,
    win_base_byte: u64,
    win_lo_local: Option<u64>,

    pending_err: Option<LogError>,
    done: bool,
}

impl<S: SegmentSource> ScanBack<S> {
    /// Starts a reverse scan of `source` emitting records in `[first, upto]` descending, with
    /// the default window size. `upto == Position::ZERO` or `first > upto` yields an empty
    /// scan. The caller clamps `upto` to what is durable (the pinned watermark, or the live
    /// tip); positions above that must not be requested.
    pub(crate) fn start(source: S, first: Position, upto: Position) -> Self {
        Self::start_with_window(source, first, upto, REVERSE_WINDOW_BYTES)
    }

    /// [`start`](Self::start) with an explicit window size, so tests can force many window
    /// boundaries with a tiny window.
    pub(crate) fn start_with_window(
        source: S,
        first: Position,
        upto: Position,
        window_bytes: usize,
    ) -> Self {
        // Positions are 1-based; treat the empty sentinel as "down to the first position".
        let first = Position::new(first.get().max(FIRST_POSITION));
        let done = upto == Position::ZERO || first > upto;
        ScanBack {
            source,
            first,
            position: upto,
            window_bytes: window_bytes.max(1),
            segment: None,
            seg_idx: 0,
            seg_base: Position::ZERO,
            seg_count: 0,
            reader: None,
            flushed_len: 0,
            window: Vec::new(),
            win_base_byte: 0,
            win_lo_local: None,
            pending_err: None,
            done,
        }
    }

    /// Advances to the next record (descending) and returns a view borrowing the window
    /// buffer, valid only until the following `next` call. A lending iterator, so not
    /// `std::iter::Iterator`; a read failure is surfaced once as an `Err` item and then
    /// terminates the scan.
    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<Result<RecordRef<'_>, LogError>> {
        if let Some(err) = self.pending_err.take() {
            self.done = true;
            return Some(Err(err));
        }
        if self.done || self.position.get() < self.first.get() {
            self.done = true;
            return None;
        }

        // Position the cursor on the segment, then the window, that hold `self.position`.
        if let Err(err) = self.ensure_segment() {
            self.done = true;
            return Some(Err(err));
        }
        if self.done {
            return None; // no segment holds the cursor; nothing left
        }
        if let Err(err) = self.ensure_window() {
            self.done = true;
            return Some(Err(err));
        }

        // Compute the emit and advance the cursor downward *before* the borrowing parse, so no
        // `self` field is written while the returned view borrows the window (mirrors `Scan`).
        let position = self.position;
        let local = (position.get() - self.seg_base.get()) as usize;
        let byte = match self.segment.as_ref().unwrap().data_offset(local) {
            Some(byte) => byte as u64,
            None => {
                self.done = true;
                return Some(Err(LogError::NotFound { position }));
            }
        };
        let rel = (byte - self.win_base_byte) as usize;
        let seg_idx = self.seg_idx;
        self.position = Position::new(position.get() - 1);

        match parse_record_ref::<0>(&self.window, rel) {
            Ok((data, _)) => Some(Ok(RecordRef { position, data })),
            Err(err) => {
                self.done = true;
                Some(Err(LogError::read(
                    scan_segment_path(&self.source, seg_idx),
                    err,
                )))
            }
        }
    }

    /// Ensures the loaded segment contains `self.position`, re-locating (a fresh reader, its
    /// flushed extent, and a forced window reload) when the cursor has descended below the
    /// current segment or none is loaded. Sets `done` if no segment holds the cursor.
    fn ensure_segment(&mut self) -> Result<(), LogError> {
        if self.reader.is_some() && self.position.get() >= self.seg_base.get() {
            return Ok(());
        }
        // Clone the located `Arc` out first, so the borrow of `self.source` ends before the
        // field writes below.
        let located = self
            .source
            .locate(self.position)
            .map(|(idx, seg)| (idx, Arc::clone(seg)));
        match located {
            Some((idx, seg)) => {
                let reader = seg.open_reader()?;
                self.flushed_len = reader.flushed_offset().load();
                self.seg_base = seg.base_position();
                self.seg_count = seg.event_count();
                self.seg_idx = idx;
                self.reader = Some(reader);
                self.segment = Some(seg);
                self.win_lo_local = None; // force a window load for the new segment
                Ok(())
            }
            None => {
                self.done = true;
                Ok(())
            }
        }
    }

    /// Ensures a window covering `self.position`'s local is loaded, reading a fresh ~64 KB
    /// window (topped at that local) when the cursor has dropped below the current one.
    fn ensure_window(&mut self) -> Result<(), LogError> {
        let h = self.position.get() - self.seg_base.get();
        if let Some(lo) = self.win_lo_local
            && h >= lo
        {
            return Ok(());
        }

        let seg = self.segment.clone().unwrap();
        let head_off = seg.data_offset(h as usize).ok_or(LogError::NotFound {
            position: self.position,
        })? as u64;

        // The exact byte just past record `h`. For a middle record that is the next data
        // record's start (any control record between is inside the span and harmlessly
        // unparsed); for a segment's last data record the successor offset is unknown and the
        // file tail is zero-filled fallocate padding, so read the framed length from the header
        // (`peek`, which uses the read-ahead buffer). Clamp to the flushed extent so the read
        // is always in bounds even if the sidecar momentarily leads the flushed point.
        let end = if h + 1 < self.seg_count {
            seg.data_offset((h + 1) as usize)
                .map(|off| off as u64)
                .unwrap_or(self.flushed_len)
                .min(self.flushed_len)
        } else {
            let reader = self.reader.as_mut().unwrap();
            match reader
                .peek(head_off)
                .map_err(|err| LogError::read(scan_segment_path(&self.source, self.seg_idx), err))?
            {
                RecordKind::Data { total_len } => {
                    (head_off + total_len as u64).min(self.flushed_len)
                }
                // The sidecar only ever points at data records.
                _ => {
                    return Err(LogError::NotFound {
                        position: self.position,
                    });
                }
            }
        };

        // Never read below `stop_local`: `first`'s local if `first` falls in this segment,
        // else the segment base. Positions below that either end the scan or belong to a lower
        // segment.
        let stop_local = if self.first.get() > self.seg_base.get() {
            self.first.get() - self.seg_base.get()
        } else {
            0
        };

        // Extend the window down from `h` while the span stays within budget, always covering
        // at least record `h` (a single record larger than the window is read whole).
        let mut lo = h;
        let mut lo_byte = head_off;
        while lo > stop_local {
            let Some(cand) = seg.data_offset((lo - 1) as usize) else {
                break;
            };
            let cand = cand as u64;
            if end - cand > self.window_bytes as u64 {
                break;
            }
            lo -= 1;
            lo_byte = cand;
        }

        let read_len = (end - lo_byte) as usize;
        self.window.resize(read_len, 0);
        {
            let reader = self.reader.as_ref().unwrap();
            reader
                .read_bytes(lo_byte, &mut self.window)
                .map_err(|err| {
                    LogError::read(scan_segment_path(&self.source, self.seg_idx), err)
                })?;
            // Warm the page just below this window: the kernel prefetches forward, not
            // backward, so the next (lower) window would otherwise be a cold read.
            if lo_byte > 0 {
                reader.prefetch(lo_byte - 1);
            }
        }
        self.win_base_byte = lo_byte;
        self.win_lo_local = Some(lo);
        Ok(())
    }
}

/// Errors from segment-set operations.
#[derive(Debug, Error)]
pub enum LogError {
    #[error("invalid segment config: {reason}")]
    InvalidConfig { reason: String },
    #[error("i/o error at {path:?}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: io::Error,
    },
    #[error("segment header error in {path:?}: {source}")]
    Header {
        path: PathBuf,
        #[source]
        source: HeaderError,
    },
    #[error(
        "segment {path:?}: header base_position {header} disagrees with filename position {name}"
    )]
    BasePositionMismatch {
        path: PathBuf,
        header: Position,
        name: Position,
    },
    #[error("unwritten segment {path:?} is not the last segment; refusing to open")]
    UnwrittenNonLast { path: PathBuf },
    #[error("non-contiguous segments: {path:?} has base_position {found}, expected {expected}")]
    NonContiguous {
        path: PathBuf,
        found: Position,
        expected: Position,
    },
    #[error(
        "recovered commit position {found} disagrees with event count (expected highest {expected}) in {path:?}"
    )]
    PositionMismatch {
        path: PathBuf,
        found: Position,
        expected: Position,
    },
    #[error("record of {size} bytes exceeds the maximum record length of {max} bytes")]
    RecordTooLarge { size: usize, max: usize },
    #[error("batch of {size} bytes cannot fit in a segment (capacity {capacity} bytes)")]
    BatchTooLarge { size: usize, capacity: usize },
    #[error("empty batch")]
    EmptyBatch,
    #[error("empty record")]
    EmptyRecord,
    #[error("position {position} not found")]
    NotFound { position: Position },
    #[error("write error at {path:?}: {source}")]
    Write {
        path: PathBuf,
        #[source]
        source: WriteError,
    },
    #[error("read error at {path:?}: {source}")]
    Read {
        path: PathBuf,
        #[source]
        source: ReadError,
    },
    #[error("tail error at {path:?}: {source}")]
    Tail {
        path: PathBuf,
        #[source]
        source: TailError,
    },
    #[error(
        "another handle already holds the write lock on {path:?}; a data directory takes one \
         writer at a time"
    )]
    Locked {
        path: PathBuf,
        /// The pid the holder recorded, when it could be read. Advisory and possibly stale,
        /// so it is a hint for an operator and never something to act on.
        holder: Option<u32>,
    },
    #[error("the segment set at {dir:?} is open read-only; {op} is not permitted")]
    ReadOnly { dir: PathBuf, op: &'static str },
    #[error("no readable segment in {dir:?}; a read-only open needs an initialized store")]
    Uninitialized { dir: PathBuf },
}

impl LogError {
    fn io(path: impl AsRef<Path>, source: io::Error) -> Self {
        LogError::Io {
            path: path.as_ref().to_path_buf(),
            source,
        }
    }

    fn write(path: impl AsRef<Path>, source: WriteError) -> Self {
        LogError::Write {
            path: path.as_ref().to_path_buf(),
            source,
        }
    }

    fn read(path: impl AsRef<Path>, source: ReadError) -> Self {
        LogError::Read {
            path: path.as_ref().to_path_buf(),
            source,
        }
    }

    fn tail(path: impl AsRef<Path>, source: TailError) -> Self {
        LogError::Tail {
            path: path.as_ref().to_path_buf(),
            source,
        }
    }
}

/// What one [`SegmentSet::refresh`] moved.
#[derive(Clone, Debug)]
pub struct Refreshed {
    /// The last readable position after this refresh.
    pub tip: Position,
    /// Segments sealed by this refresh, that is, rollovers the writer performed. Usually 0.
    pub sealed_added: usize,
    /// A trailing segment file whose header does not validate yet. Not an error: the writer
    /// creates and `fallocate`s a segment before writing its header, so this is normally the
    /// few microseconds of a rollover. Surfaced so a caller can alarm if it persists, which
    /// would mean a follower stuck behind real damage rather than a race.
    pub pending_segment: Option<PathBuf>,
}

/// Whether a segment file's header can be used.
enum Headerness {
    /// Validated, and its `base_position` agrees with the filename.
    Ready,
    /// Creation did not finish: the file is shorter than a header, or the header is still
    /// the all-zero `fallocate` fill. Either way it holds no committed data.
    Unwritten,
    /// A header is present but does not validate. Corruption, except on the trailing
    /// segment of a live log, where it is a torn read of a header being written right now.
    Unusable(LogError),
}

/// Validates one segment file's header against its filename.
///
/// Errors are returned as [`Headerness::Unusable`] rather than raised, so the caller can
/// decide by position: the same torn header is corruption in the middle of the chain and a
/// normal race at its end.
fn classify_header(path: &Path, name_base: Position) -> Result<Headerness, LogError> {
    let Some(buf) = read_header(path)? else {
        return Ok(Headerness::Unwritten);
    };
    match SegmentHeader::from_bytes(&buf) {
        Ok(header) if header.base_position == name_base => Ok(Headerness::Ready),
        Ok(header) => Ok(Headerness::Unusable(LogError::BasePositionMismatch {
            path: path.to_path_buf(),
            header: header.base_position,
            name: name_base,
        })),
        Err(HeaderError::Unwritten) => Ok(Headerness::Unwritten),
        Err(source) => Ok(Headerness::Unusable(LogError::Header {
            path: path.to_path_buf(),
            source,
        })),
    }
}

/// Every `{20 digits}.log` in `dir`, sorted numerically by base position.
fn list_segments(dir: &Path) -> Result<Vec<(Position, PathBuf)>, LogError> {
    let mut entries: Vec<(Position, PathBuf)> = Vec::new();
    for entry in fs::read_dir(dir).map_err(|source| LogError::io(dir, source))? {
        let entry = entry.map_err(|source| LogError::io(dir, source))?;
        if let Some(base) = parse_base_position(&entry.file_name().to_string_lossy()) {
            entries.push((base, entry.path()));
        }
    }
    entries.sort_by_key(|(base, _)| *base);
    Ok(entries)
}

/// The result of [`split_trailing_unready`]: the usable segment chain, plus the trailing files
/// whose creation has not finished.
struct SegmentSplit {
    valid: Vec<(Position, PathBuf)>,
    unready: Vec<PathBuf>,
}

/// Splits a sorted listing into the usable chain and the trailing run whose creation has not
/// finished.
///
/// An unfinished segment is legal only as a trailing run: a failed rollover can leave several
/// (ENOSPC on each retry). Anywhere else it is a real gap and a hard error.
///
/// `strict_headers` decides what a present-but-invalid header means. A writer treats it as
/// corruption wherever it sits, because nothing should be writing headers but itself. A
/// follower tolerates it on the trailing segment, where the writer creates and `fallocate`s a
/// file before writing its header, so an all-zero or half-written header there is a race, not
/// damage.
fn split_trailing_unready(
    entries: Vec<(Position, PathBuf)>,
    strict_headers: bool,
) -> Result<SegmentSplit, LogError> {
    let mut state = Vec::with_capacity(entries.len());
    for (base, path) in &entries {
        let headerness = classify_header(path, *base)?;
        if strict_headers && let Headerness::Unusable(err) = headerness {
            return Err(err);
        }
        state.push(headerness);
    }

    // The last usable segment; everything after it is the trailing run.
    let last_ready = (0..state.len())
        .rev()
        .find(|&i| matches!(state[i], Headerness::Ready));

    let mut valid = Vec::new();
    let mut unready = Vec::new();
    for (i, ((base, path), headerness)) in entries.into_iter().zip(state).enumerate() {
        if matches!(headerness, Headerness::Ready) {
            valid.push((base, path));
            continue;
        }
        if last_ready.is_none_or(|lr| i > lr) {
            unready.push(path);
            continue;
        }
        return Err(match headerness {
            Headerness::Unwritten => LogError::UnwrittenNonLast { path },
            Headerness::Unusable(err) => err,
            Headerness::Ready => unreachable!("handled above"),
        });
    }
    Ok(SegmentSplit { valid, unready })
}

/// `{base_position:020}.log`.
fn segment_file_name(base: Position) -> String {
    format!("{:0width$}.log", base.get(), width = NAME_DIGITS)
}

/// Parses `base_position` from a segment file name, or `None` if it does not match
/// the `{20 digits}.log` pattern.
fn parse_base_position(name: &str) -> Option<Position> {
    let stem = name.strip_suffix(".log")?;
    if stem.len() != NAME_DIGITS || !stem.bytes().all(|b| b.is_ascii_digit()) {
        return None;
    }
    stem.parse::<u64>().ok().map(Position::new)
}

/// Reads the first [`SEGMENT_HEADER_SIZE`] bytes of a segment file, or `None` if the file is
/// shorter than a full header. A short file is a segment whose creation did not finish (the
/// `create_new` succeeded but the `fallocate` or the header write did not, as on a crash or an
/// `ENOSPC` during rollover), which the caller treats as unwritten.
fn read_header(path: &Path) -> Result<Option<[u8; SEGMENT_HEADER_SIZE]>, LogError> {
    let file = File::open(path).map_err(|source| LogError::io(path, source))?;
    let mut buf = [0u8; SEGMENT_HEADER_SIZE];
    match file.read_exact_at(&mut buf, 0) {
        Ok(()) => Ok(Some(buf)),
        Err(source) if source.kind() == io::ErrorKind::UnexpectedEof => Ok(None),
        Err(source) => Err(LogError::io(path, source)),
    }
}

/// Scans a segment for its data-record byte offsets, indexed by local position.
/// Skips control records; stops at `flushed`, the segment's committed extent.
fn scan_offsets(
    path: &Path,
    flushed: FlushedOffset,
    header_size: u64,
) -> Result<Vec<u32>, LogError> {
    let mut reader = Reader::<0>::open_read_only(path, Some(flushed))
        .map_err(|source| LogError::read(path, source))?;
    let mut offsets = Vec::new();
    let mut iter = reader.iter(header_size);
    while let Some(record) = iter
        .next_record()
        .map_err(|source| LogError::read(path, source))?
    {
        offsets.push(
            u32::try_from(record.offset)
                .expect("segment_size <= u32::MAX enforced by SegmentConfig::validate"),
        );
    }
    Ok(offsets)
}

/// Builds the sealed half of a segment chain, verifying it is contiguous from the first
/// position and that each segment's last commit marker agrees with its event count.
///
/// Shared by both open paths. The read-write and read-only opens differ in what they may do
/// to the directory, never in what the chain must satisfy, and duplicating these checks is
/// how the two would drift into disagreeing about the same bytes.
///
/// Returns the sealed segments and the position the active segment must start at.
fn open_sealed_chain(
    sealed_entries: &[(Position, PathBuf)],
    header_size: u64,
) -> Result<(Vec<Arc<Segment>>, Position), LogError> {
    let mut sealed = Vec::with_capacity(sealed_entries.len());
    let mut expected_base = Position::new(FIRST_POSITION);
    for (base, path) in sealed_entries {
        if *base != expected_base {
            return Err(LogError::NonContiguous {
                path: path.clone(),
                found: *base,
                expected: expected_base,
            });
        }
        let scanned = scan_committed(path, header_size)?;
        let count = scanned.offsets.len() as u64;
        // Position assignment is contiguous from the base, so the last marker's highest
        // position must be base + count - 1.
        if let Some(highest) = scanned.last_position
            && highest + 1 != *base + count
        {
            return Err(LogError::PositionMismatch {
                path: path.clone(),
                found: Position::new(highest),
                expected: Position::new(*base + count - 1),
            });
        }
        expected_base = Position::new(*base + count);
        sealed.push(Arc::new(Segment {
            base_position: *base,
            path: path.clone(),
            flushed_offset: scanned.flushed,
            offsets: RwLock::new(scanned.offsets),
            reader: Mutex::new(None),
        }));
    }
    Ok((sealed, expected_base))
}

/// The committed contents of a segment: the offsets of its data records, the extent
/// readers may read to, and the highest position its last commit marker claimed.
struct Scanned {
    offsets: Vec<u32>,
    flushed: FlushedOffset,
    last_position: Option<u64>,
}

/// Scans a segment through the recovery rule, so only records belonging to a batch that
/// terminated in a valid commit marker are counted.
///
/// This is the same walk the active segment gets from `Writer::open`, applied to sealed
/// segments too. Stopping at the zero-filled tail instead would adopt the CRC-valid orphans
/// a failed append leaves behind (`Writer::rewind_to` abandons them in place), which inflates
/// the segment's event count and breaks the contiguity chain on the next open.
fn scan_committed(path: &Path, header_size: u64) -> Result<Scanned, LogError> {
    let mut tail = Tail::open(path, header_size).map_err(|source| LogError::tail(path, source))?;
    let mut offsets = Vec::new();
    let progress = tail
        .poll(|offset| {
            offsets.push(
                u32::try_from(offset)
                    .expect("segment_size <= u32::MAX enforced by SegmentConfig::validate"),
            );
        })
        .map_err(|source| LogError::tail(path, source))?;
    Ok(Scanned {
        offsets,
        flushed: tail.flushed_offset(),
        last_position: progress.last_position,
    })
}

/// Applies per-writer settings from the config to a freshly created or reopened
/// writer: currently just the max record length.
fn configure_writer(writer: &mut Writer<0>, config: &SegmentConfig) {
    writer.set_max_record(config.max_record_len);
}

/// Whether a non-zero record header sits at `offset`, i.e. recovery discarded a
/// torn trailing batch (as opposed to a clean end at the zero-filled tail).
#[cfg(feature = "tracing")]
fn trailing_bytes_present(writer: &Writer<0>, offset: u64, segment_size: usize) -> bool {
    if offset + RECORD_HEAD_SIZE as u64 > segment_size as u64 {
        return false;
    }
    let mut head = [0u8; RECORD_HEAD_SIZE];
    writer.file().read_exact_at(&mut head, offset).is_ok() && head.iter().any(|&b| b != 0)
}

fn sync_dir(dir: &Path) -> io::Result<()> {
    File::open(dir)?.sync_all()
}

/// Takes the directory's write lock for the lifetime of the returned guard.
///
/// See [`crate::log::lock`] for why this is a POSIX record lock plus an in-process registry
/// rather than a `flock`: the descriptor-held kind is inherited by a forked child, so a
/// process that spawns a subprocess can be refused its own directory afterwards.
fn acquire_lock(dir: &Path) -> Result<DirLock, LogError> {
    match lock::acquire(dir) {
        Ok(acquired) => {
            if let Some(_source) = acquired.unsupported {
                #[cfg(feature = "tracing")]
                tracing::warn!(
                    "{dir:?}: this filesystem does not implement record locking ({_source}); \
                     opening without cross-process single-writer protection"
                );
            }
            Ok(acquired.lock)
        }
        Err(LockFailure::Contended { path, holder }) => Err(LogError::Locked { path, holder }),
        Err(LockFailure::Io { path, source }) => Err(LogError::io(path, source)),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::log::lock::LOCK_FILE;
    use std::fs::OpenOptions;
    use std::process;
    use tempfile::TempDir;

    const HEADER: usize = SEGMENT_HEADER_SIZE;
    /// Framing overhead of a single record (its length + CRC head).
    const REC_OVERHEAD: usize = RECORD_HEAD_SIZE;
    /// Framing of a batch's trailing commit marker.
    const MARKER: usize = RECORD_HEAD_SIZE + COMMIT_MARKER_PAYLOAD;

    fn open(dir: &Path, segment_size: usize) -> SegmentSet {
        SegmentSet::open(dir, SegmentConfig::new(segment_size)).unwrap()
    }

    /// Appends one single-record batch and returns its position.
    fn append_one(set: &mut SegmentSet, data: &[u8]) -> Position {
        let range = set.append_batch(&[data]).unwrap();
        assert_eq!(range.first, range.last);
        range.first
    }

    /// Drains a lending [`Scan`] into owned records.
    fn drain<S: SegmentSource>(mut scan: Scan<S>) -> Vec<Record> {
        let mut out = Vec::new();
        while let Some(item) = scan.next() {
            out.push(item.unwrap().to_owned());
        }
        out
    }

    #[test]
    fn open_empty_creates_first_segment() {
        let dir = TempDir::new().unwrap();
        let set = open(dir.path(), 4096);

        // Fresh log: next position is 1, and last_position is the empty sentinel 0.
        assert_eq!(set.next_position(), Position::new(1));
        assert_eq!(set.last_position(), Position::new(0));
        assert_eq!(set.sealed_len(), 0);
        assert!(dir.path().join("00000000000000000001.log").exists());
    }

    #[test]
    fn tiny_config_rejected() {
        let dir = TempDir::new().unwrap();
        // Default max_record_len = 16, header 64, so nothing usable fits.
        let err = SegmentSet::open(dir.path(), SegmentConfig::new(64)).unwrap_err();
        assert!(matches!(err, LogError::InvalidConfig { .. }), "got {err:?}");
    }

    #[test]
    fn reopen_after_clean_shutdown_preserves_state() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            for i in 1..=5u64 {
                append_one(&mut set, format!("event-{i}").as_bytes());
            }
            assert_eq!(set.next_position(), Position::new(6));
        }

        let set = open(dir.path(), 4096);
        assert_eq!(set.next_position(), Position::new(6));
        assert_eq!(set.last_position(), Position::new(5));
        for i in 1..=5u64 {
            let record = set.read_at(Position::new(i)).unwrap();
            assert_eq!(record.position, Position::new(i));
            assert_eq!(record.data, format!("event-{i}").into_bytes());
        }
    }

    #[test]
    fn open_deletes_short_trailing_segment_and_recovers() {
        // A rollover whose `create_new` succeeded but whose `fallocate` or header write did not (a
        // crash or ENOSPC during extension) leaves a segment file shorter than a header, commonly
        // zero bytes. Opening must treat that trailing short file as unwritten: delete it and
        // recover the committed data, not refuse to open on a short header read.
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            for i in 1..=5u64 {
                append_one(&mut set, format!("event-{i}").as_bytes());
            }
        }

        // Stray 0-byte segments left by failed extensions. A retrying rollover under ENOSPC can
        // leave more than one, so cover a trailing run of two: both sort after the real segment
        // and both must be deleted.
        let stray1 = dir.path().join(segment_file_name(Position::new(6)));
        let stray2 = dir.path().join(segment_file_name(Position::new(7)));
        File::create(&stray1).unwrap();
        File::create(&stray2).unwrap();
        assert_eq!(fs::metadata(&stray1).unwrap().len(), 0);

        let set = open(dir.path(), 4096);
        assert!(
            !stray1.exists(),
            "trailing 0-byte segment 6 should be deleted on open"
        );
        assert!(
            !stray2.exists(),
            "trailing 0-byte segment 7 should be deleted on open"
        );
        assert_eq!(set.next_position(), Position::new(6));
        assert_eq!(set.last_position(), Position::new(5));
        for i in 1..=5u64 {
            assert_eq!(
                set.read_at(Position::new(i)).unwrap().data,
                format!("event-{i}").into_bytes()
            );
        }
    }

    #[test]
    fn rollover_keeps_positions_contiguous() {
        let dir = TempDir::new().unwrap();
        // Small segment so a handful of batches force rollovers.
        let mut set = open(dir.path(), 256);

        let n = 20u64;
        for i in 1..=n {
            let pos = append_one(&mut set, format!("evt{i:03}").as_bytes());
            assert_eq!(pos, Position::new(i));
        }

        assert_eq!(set.next_position(), Position::new(n + 1));
        assert!(set.sealed_len() >= 1, "expected at least one rollover");

        for i in 1..=n {
            let record = set.read_at(Position::new(i)).unwrap();
            assert_eq!(record.data, format!("evt{i:03}").into_bytes());
        }
    }

    #[test]
    fn read_at_across_boundary_for_every_position() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 200);
        let n = 30u64;
        for i in 1..=n {
            append_one(&mut set, format!("r{i:04}").as_bytes());
        }
        assert!(set.sealed_len() >= 2);
        for i in 1..=n {
            assert_eq!(
                set.read_at(Position::new(i)).unwrap().data,
                format!("r{i:04}").into_bytes()
            );
        }
        // The empty sentinel and a position past the end are both absent.
        assert!(matches!(
            set.read_at(Position::new(0)),
            Err(LogError::NotFound { .. })
        ));
        assert!(matches!(
            set.read_at(Position::new(n + 1)),
            Err(LogError::NotFound { .. })
        ));
    }

    #[test]
    fn scan_from_mid_segment_yields_expected_order() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        let n = 12u64;
        for i in 1..=n {
            append_one(&mut set, format!("s{i}").as_bytes());
        }

        let start = 5u64;
        let got = drain(set.scan_from(Position::new(start)));
        assert_eq!(got.len() as u64, n - start + 1);
        for (idx, record) in got.iter().enumerate() {
            let pos = start + idx as u64;
            assert_eq!(record.position, Position::new(pos));
            assert_eq!(record.data, format!("s{pos}").into_bytes());
        }
    }

    #[test]
    fn scan_across_segments_is_contiguous_and_ordered() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 200);
        let n = 25u64;
        for i in 1..=n {
            append_one(&mut set, format!("x{i:04}").as_bytes());
        }
        assert!(set.sealed_len() >= 2);

        let got = drain(set.scan_from(Position::new(1)));
        assert_eq!(got.len() as u64, n);
        for (idx, record) in got.iter().enumerate() {
            let pos = idx as u64 + 1;
            assert_eq!(record.position, Position::new(pos));
            assert_eq!(record.data, format!("x{pos:04}").into_bytes());
        }
    }

    #[test]
    fn scan_from_zero_clamps_to_whole_log() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        for i in 1..=3u64 {
            append_one(&mut set, format!("e{i}").as_bytes());
        }

        // scan_from(0) means "from before everything": the whole log, not nothing.
        let positions: Vec<Position> = drain(set.scan_from(Position::new(0)))
            .iter()
            .map(|r| r.position)
            .collect();
        assert_eq!(
            positions,
            vec![Position::new(1), Position::new(2), Position::new(3)]
        );

        // Past the end is a normal caught-up state: empty, not an error.
        assert!(set.scan_from(Position::new(5)).next().is_none());
    }

    #[test]
    fn scan_after_is_exclusive() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        for i in 1..=3u64 {
            append_one(&mut set, format!("e{i}").as_bytes());
        }

        // scan_after(0) scans the whole log with no sentinel special case.
        let all: Vec<Position> = drain(set.scan_after(Position::new(0)))
            .iter()
            .map(|r| r.position)
            .collect();
        assert_eq!(
            all,
            vec![Position::new(1), Position::new(2), Position::new(3)]
        );

        // scan_after(pos) is exclusive: it resumes strictly after `pos`.
        let resumed: Vec<Position> = drain(set.scan_after(Position::new(1)))
            .iter()
            .map(|r| r.position)
            .collect();
        assert_eq!(resumed, vec![Position::new(2), Position::new(3)]);

        // scan_after(last) is the caught-up state: empty, not an error.
        assert!(set.scan_after(set.last_position()).next().is_none());

        // Inclusive/exclusive agree: scan_from(n) == scan_after(n - 1).
        let from2: Vec<Position> = drain(set.scan_from(Position::new(2)))
            .iter()
            .map(|r| r.position)
            .collect();
        assert_eq!(from2, resumed);
    }

    #[test]
    fn oversized_record_rejected() {
        let dir = TempDir::new().unwrap();
        let mut config = SegmentConfig::new(4096);
        config.max_record_len = 100;
        let mut set = SegmentSet::open(dir.path(), config).unwrap();

        let big = vec![0u8; 200];
        let err = set.append_batch(&[&big]).unwrap_err();
        assert!(
            matches!(err, LogError::RecordTooLarge { .. }),
            "got {err:?}"
        );
    }

    #[test]
    fn batch_larger_than_segment_rejected() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 200);

        // Five 30-byte records plus overhead exceed capacity (segment_size - header),
        // yet each record is under max_record_len.
        let records: Vec<Vec<u8>> = (0..5).map(|_| vec![0u8; 30]).collect();
        let refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();
        let err = set.append_batch(&refs).unwrap_err();
        assert!(matches!(err, LogError::BatchTooLarge { .. }), "got {err:?}");
    }

    #[test]
    fn empty_batch_and_empty_record_rejected() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        assert!(matches!(set.append_batch(&[]), Err(LogError::EmptyBatch)));
        assert!(matches!(
            set.append_batch(&[b""]),
            Err(LogError::EmptyRecord)
        ));
        // A rejected empty record must not have advanced anything.
        assert_eq!(set.next_position(), Position::new(1));
    }

    #[test]
    fn crash_after_create_before_header_write_is_deleted() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        {
            let mut set = open(dir.path(), segment_size);
            for i in 1..=3u64 {
                append_one(&mut set, format!("e{i}").as_bytes());
            }
            assert_eq!(set.next_position(), Position::new(4));
        }

        // Simulate a crash between create and header write for the next segment: a
        // zero-filled trailing file at the next base position.
        let ghost = dir.path().join(segment_file_name(Position::new(4)));
        fs::write(&ghost, vec![0u8; segment_size]).unwrap();

        let set = open(dir.path(), segment_size);
        assert!(
            !ghost.exists(),
            "zero-filled trailing segment should be deleted"
        );
        assert_eq!(set.next_position(), Position::new(4));
        for i in 1..=3u64 {
            assert_eq!(
                set.read_at(Position::new(i)).unwrap().data,
                format!("e{i}").into_bytes()
            );
        }
    }

    #[test]
    fn missing_middle_segment_fails_open() {
        let dir = TempDir::new().unwrap();
        let segment_size = 200;
        {
            let mut set = open(dir.path(), segment_size);
            for i in 1..=15u64 {
                append_one(&mut set, format!("m{i:03}").as_bytes());
            }
            assert!(set.sealed_len() >= 3, "need several sealed segments");
        }

        let mut files: Vec<PathBuf> = fs::read_dir(dir.path())
            .unwrap()
            .map(|e| e.unwrap().path())
            .filter(|p| p.extension().is_some_and(|e| e == "log"))
            .collect();
        files.sort();
        assert!(files.len() >= 3);
        fs::remove_file(&files[1]).unwrap();

        let err = SegmentSet::open(dir.path(), SegmentConfig::new(segment_size)).unwrap_err();
        assert!(matches!(err, LogError::NonContiguous { .. }), "got {err:?}");
    }

    #[test]
    fn header_base_position_disagreeing_with_filename_fails() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"only");
        }

        // The first segment's file is named for base position 1.
        let path = dir.path().join(segment_file_name(Position::new(1)));
        let bogus = SegmentHeader::new(Position::new(7));
        let file = File::options().write(true).open(&path).unwrap();
        file.write_all_at(&bogus.to_bytes(), 0).unwrap();
        file.sync_all().unwrap();
        drop(file);

        let err = SegmentSet::open(dir.path(), SegmentConfig::new(segment_size)).unwrap_err();
        assert!(
            matches!(err, LogError::BasePositionMismatch { .. }),
            "got {err:?}"
        );
    }

    #[test]
    fn multi_record_batch_shares_positions() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        let range = set.append_batch(&[b"a", b"bb", b"ccc"]).unwrap();
        assert_eq!(range.first, Position::new(1));
        assert_eq!(range.last, Position::new(3));
        assert_eq!(range.count(), 3);
        assert_eq!(set.next_position(), Position::new(4));
        assert_eq!(set.read_at(Position::new(1)).unwrap().data, b"a");
        assert_eq!(set.read_at(Position::new(2)).unwrap().data, b"bb");
        assert_eq!(set.read_at(Position::new(3)).unwrap().data, b"ccc");
    }

    #[test]
    fn append_continues_after_recovery() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"before"); // position 1
        }
        let mut set = open(dir.path(), segment_size);
        let pos = append_one(&mut set, b"after"); // position 2
        assert_eq!(pos, Position::new(2));
        assert_eq!(set.read_at(Position::new(1)).unwrap().data, b"before");
        assert_eq!(set.read_at(Position::new(2)).unwrap().data, b"after");
    }

    /// Data payload for the 1-based `position` in the single-record-per-batch logs
    /// built by [`build_single_segment`] and the truncation tests.
    fn payload_for(position: u64, record_len: usize) -> Vec<u8> {
        // Batch index is `position - 1`; the builder tags each with index+1.
        vec![((position - 1) as u8).wrapping_add(1); record_len]
    }

    /// Builds a single-segment log with `batches` single-record batches (positions
    /// 1..=batches), then returns the raw file bytes and the byte offset just past
    /// each batch's commit marker.
    fn build_single_segment(
        dir: &Path,
        segment_size: usize,
        batches: usize,
        record_len: usize,
    ) -> (Vec<u8>, Vec<usize>) {
        let mut set = open(dir, segment_size);
        for p in 1..=batches as u64 {
            append_one(&mut set, &payload_for(p, record_len));
        }
        assert_eq!(set.sealed_len(), 0, "test assumes a single segment");
        drop(set);

        let path = dir.join(segment_file_name(Position::new(FIRST_POSITION)));
        let bytes = fs::read(&path).unwrap();

        let batch_size = REC_OVERHEAD + record_len + MARKER;
        let commit_ends: Vec<usize> = (0..batches)
            .map(|i| HEADER + (i + 1) * batch_size)
            .collect();
        (bytes, commit_ends)
    }

    #[test]
    fn truncation_mid_batch_rolls_back_to_previous_commit() {
        let segment_size = 4096;
        let batches = 8;
        let record_len = 10;

        let source = TempDir::new().unwrap();
        let (good_bytes, commit_ends) =
            build_single_segment(source.path(), segment_size, batches, record_len);
        let total_end = *commit_ends.last().unwrap();

        // Table-driven over a dense range of truncation offsets: for each cutoff,
        // corrupt the tail and assert recovery rolls back to the last commit marker
        // whose batch lies entirely before the cutoff.
        for cutoff in HEADER..=total_end {
            let dir = TempDir::new().unwrap();
            let mut corrupt = good_bytes.clone();
            for byte in corrupt.iter_mut().skip(cutoff) {
                *byte = 0xFF;
            }
            let path = dir
                .path()
                .join(segment_file_name(Position::new(FIRST_POSITION)));
            fs::write(&path, &corrupt).unwrap();

            let set = open(dir.path(), segment_size);

            // `survived` batches map to positions 1..=survived, so next is survived + 1.
            let survived = commit_ends.iter().filter(|&&end| end <= cutoff).count() as u64;
            assert_eq!(
                set.next_position(),
                Position::new(survived + 1),
                "cutoff {cutoff}: expected {survived} surviving events"
            );

            for p in 1..=survived {
                let record = set.read_at(Position::new(p)).unwrap();
                assert_eq!(
                    record.data,
                    payload_for(p, record_len),
                    "cutoff {cutoff}, position {p}"
                );
            }
        }
    }

    #[test]
    fn corrupt_record_with_intact_marker_rejects_whole_batch() {
        // The rule that matters most: a batch is committed only if *every* record
        // in it validates, not merely if its trailing marker is present.
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let rec_len = 6;
        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"aaaa"); // batch A: position 1, survives
            let recs: Vec<Vec<u8>> = (0..5).map(|i| vec![b'B' + i as u8; rec_len]).collect();
            let refs: Vec<&[u8]> = recs.iter().map(|r| r.as_slice()).collect();
            set.append_batch(&refs).unwrap(); // batch B: positions 2..=6
            assert_eq!(set.next_position(), Position::new(7));
        }

        // Flip one byte inside record 2 of batch B, leaving batch B's commit marker
        // completely intact.
        let batch_a = REC_OVERHEAD + 4 + MARKER;
        let rec_stride = REC_OVERHEAD + rec_len;
        let rec2_data = HEADER + batch_a + rec_stride + REC_OVERHEAD;
        let path = dir
            .path()
            .join(segment_file_name(Position::new(FIRST_POSITION)));
        let file = File::options().read(true).write(true).open(&path).unwrap();
        let mut byte = [0u8; 1];
        file.read_exact_at(&mut byte, rec2_data as u64).unwrap();
        byte[0] ^= 0xFF;
        file.write_all_at(&byte, rec2_data as u64).unwrap();
        file.sync_all().unwrap();
        drop(file);

        let set = open(dir.path(), segment_size);
        assert_eq!(
            set.next_position(),
            Position::new(2),
            "whole batch B must roll back"
        );
        assert_eq!(set.read_at(Position::new(1)).unwrap().data, b"aaaa");
        assert!(matches!(
            set.read_at(Position::new(2)),
            Err(LogError::NotFound { .. })
        ));
    }

    #[test]
    fn physical_truncation_mid_batch_rolls_back() {
        // A short file (real truncation) exercises different recovery paths than
        // garbage-overwrite: reads run off the physical end.
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let record_len = 10;
        let batches = 6;
        {
            let mut set = open(dir.path(), segment_size);
            for p in 1..=batches as u64 {
                append_one(&mut set, &payload_for(p, record_len));
            }
        }

        let batch_size = REC_OVERHEAD + record_len + MARKER;
        let survive = 3usize;
        let cut = HEADER + survive * batch_size + 5; // partway into batch index 3

        let path = dir
            .path()
            .join(segment_file_name(Position::new(FIRST_POSITION)));
        let file = File::options().write(true).open(&path).unwrap();
        file.set_len(cut as u64).unwrap();
        file.sync_all().unwrap();
        drop(file);

        let set = open(dir.path(), segment_size);
        assert_eq!(set.next_position(), Position::new(survive as u64 + 1));
        for p in 1..=survive as u64 {
            assert_eq!(
                set.read_at(Position::new(p)).unwrap().data,
                payload_for(p, record_len)
            );
        }
    }

    /// Drains a lending [`ScanBack`] into owned records, preserving its descending order.
    fn drain_back<S: SegmentSource>(mut scan: ScanBack<S>) -> Vec<Record> {
        let mut out = Vec::new();
        while let Some(item) = scan.next() {
            out.push(item.unwrap().to_owned());
        }
        out
    }

    /// The core property: a reverse scan yields exactly the forward scan reversed, across
    /// window sizes (single-record up to the default) and interior sub-ranges, over a log with
    /// several sealed segments and interleaved commit-marker control records. Any window-
    /// boundary, control-skip, or segment-crossing bug shows up as an inequality here.
    #[test]
    fn scan_back_is_the_reverse_of_scan_forward() {
        let dir = TempDir::new().unwrap();
        // Small segments so several seal; varied record sizes so windows pack differently and
        // a single record can exceed the tiny window.
        let mut set = open(dir.path(), 1024);
        for i in 1..=80u64 {
            let size = 8 + (i as usize % 17) * 5;
            append_one(&mut set, &vec![(i % 251) as u8; size]);
        }
        assert!(set.sealed_len() >= 2, "need several sealed segments");
        let last = set.last_position();

        let forward = drain(set.scan_from(Position::new(1)));
        assert_eq!(forward.len(), 80);

        for window in [1usize, 32, 200, REVERSE_WINDOW_BYTES] {
            let full = drain_back(ScanBack::start_with_window(
                &set,
                Position::new(1),
                last,
                window,
            ));
            let want: Vec<Record> = forward.iter().rev().cloned().collect();
            assert_eq!(full, want, "window {window}, full range");

            let (lo, hi) = (Position::new(20), Position::new(60));
            let sub = drain_back(ScanBack::start_with_window(&set, lo, hi, window));
            let want_sub: Vec<Record> = forward
                .iter()
                .filter(|record| record.position >= lo && record.position <= hi)
                .rev()
                .cloned()
                .collect();
            assert_eq!(sub, want_sub, "window {window}, sub range");
        }
    }

    /// `upto` is clamped down to the live tip, so a huge `upto` still starts at the last
    /// record; and the empty cases (`first > upto`, the `ZERO` sentinel, an empty log) yield
    /// nothing.
    #[test]
    fn scan_back_bounds_and_empty_cases() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 4096);
        for i in 1..=5u64 {
            append_one(&mut set, format!("e{i}").as_bytes());
        }

        // Position::MAX clamps to the tip: the whole log, newest first.
        let all = drain_back(set.scan_back(Position::new(1), Position::MAX));
        let positions: Vec<u64> = all.iter().map(|record| record.position.get()).collect();
        assert_eq!(positions, vec![5, 4, 3, 2, 1]);

        assert!(drain_back(set.scan_back(Position::new(4), Position::new(3))).is_empty());
        assert!(drain_back(set.scan_back(Position::new(1), Position::ZERO)).is_empty());

        let empty_dir = TempDir::new().unwrap();
        let empty = open(empty_dir.path(), 4096);
        assert!(drain_back(empty.scan_back(Position::new(1), Position::MAX)).is_empty());
    }

    /// A record larger than the whole window budget is still read and yielded intact (the
    /// window always covers at least its top record).
    #[test]
    fn scan_back_record_larger_than_window() {
        let dir = TempDir::new().unwrap();
        let mut set = open(dir.path(), 1 << 16);
        let big = vec![7u8; 4096];
        append_one(&mut set, b"small-before");
        let big_pos = append_one(&mut set, &big);
        append_one(&mut set, b"small-after");

        // A one-byte window forces each record into its own window, so the 4 KB record is read
        // whole despite far exceeding the budget.
        let got = drain_back(ScanBack::start_with_window(
            &set,
            Position::new(1),
            set.last_position(),
            1,
        ));
        assert_eq!(got[0].position, Position::new(3));
        assert_eq!(got[1].position, big_pos);
        assert_eq!(got[1].data, big);
        assert_eq!(got[2].position, Position::new(1));
    }

    /// A failed append leaves CRC-valid records past the last commit marker (`rewind_to`
    /// abandons them in place). If the next batch rolls over, the segment seals with those
    /// orphans still on disk. Counting them would inflate the segment's event count and break
    /// the contiguity chain, so a sealed segment must be scanned through the commit rule too.
    #[test]
    fn sealed_segment_orphans_are_not_counted() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let config = SegmentConfig::new(segment_size);

        {
            let mut set = open(dir.path(), segment_size);
            for i in 1..=5u64 {
                append_one(&mut set, format!("event-{i}").as_bytes());
            }
        }

        let first = dir.path().join(segment_file_name(Position::new(1)));

        // A batch that got its records out and then failed before its commit marker.
        {
            let mut writer =
                Writer::<0>::open(&first, segment_size, config.header_size as u64).unwrap();
            let rewind = writer.write_offset();
            writer.append_data(b"orphan-a").unwrap();
            writer.append_data(b"orphan-b").unwrap();
            writer.flush_writer().unwrap();
            writer.rewind_to(rewind).unwrap();
        }

        // The next batch does not fit, so the set rolls over and seals the orphan-bearing
        // segment. Position 6 is where the writer would have continued.
        SegmentSet::create_segment(dir.path(), &config, Position::new(6)).unwrap();

        let set = open(dir.path(), segment_size);
        assert_eq!(
            set.sealed_len(),
            1,
            "the orphan-bearing segment should be sealed"
        );
        assert_eq!(
            set.sealed_arcs()[0].event_count(),
            5,
            "orphans past the last commit marker are not events"
        );
        assert_eq!(set.last_position(), Position::new(5));
        assert_eq!(set.next_position(), Position::new(6));
        for i in 1..=5u64 {
            assert_eq!(
                set.read_at(Position::new(i)).unwrap().data,
                format!("event-{i}").into_bytes()
            );
        }
        assert!(
            set.read_at(Position::new(6)).is_err(),
            "an orphan must not be readable as an event"
        );
    }

    /// A sealed segment's reads stop at its commit point, not at the zero-filled tail.
    #[test]
    fn sealed_segment_flushed_offset_bounds_reads_at_the_commit_point() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let config = SegmentConfig::new(segment_size);

        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"only");
        }
        let first = dir.path().join(segment_file_name(Position::new(1)));
        let committed_end = {
            let writer =
                Writer::<0>::open(&first, segment_size, config.header_size as u64).unwrap();
            writer.write_offset()
        };
        {
            let mut writer =
                Writer::<0>::open(&first, segment_size, config.header_size as u64).unwrap();
            let rewind = writer.write_offset();
            writer.append_data(b"orphan").unwrap();
            writer.flush_writer().unwrap();
            writer.rewind_to(rewind).unwrap();
        }
        SegmentSet::create_segment(dir.path(), &config, Position::new(2)).unwrap();

        let set = open(dir.path(), segment_size);
        let sealed = &set.sealed_arcs()[0];
        assert_eq!(
            sealed.flushed_offset.load(),
            committed_end,
            "a sealed segment is bounded by its commit point"
        );
    }

    // The single-writer lock

    #[test]
    fn a_second_open_in_the_same_process_is_locked_out() {
        let dir = TempDir::new().unwrap();
        let _held = open(dir.path(), 4096);

        let err = SegmentSet::open(dir.path(), SegmentConfig::new(4096)).unwrap_err();
        let LogError::Locked { path, holder } = &err else {
            panic!("expected Locked, got {err:?}");
        };
        assert_eq!(path.file_name().unwrap(), LOCK_FILE);
        assert_eq!(
            *holder,
            Some(process::id()),
            "the holder hint should name the holding pid"
        );
    }

    #[test]
    fn the_lock_is_released_on_drop() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            append_one(&mut set, b"event");
        }
        // Dropping the set closes the descriptor, which is what releases the lock.
        let set = open(dir.path(), 4096);
        assert_eq!(set.last_position(), Position::new(1));
    }

    #[test]
    fn a_stale_lock_file_does_not_block() {
        // The lock lives on the descriptor, so a `LOCK` left behind by a killed writer holds
        // nothing. Opening must not treat its presence as contention.
        let dir = TempDir::new().unwrap();
        fs::write(dir.path().join(LOCK_FILE), b"999999\n").unwrap();

        let mut set = open(dir.path(), 4096);
        append_one(&mut set, b"event");
        assert_eq!(set.last_position(), Position::new(1));
    }

    #[test]
    fn the_lock_file_is_not_mistaken_for_a_segment() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            append_one(&mut set, b"event");
        }
        assert!(
            dir.path().join(LOCK_FILE).exists(),
            "the lock file is left on disk"
        );
        assert!(parse_base_position(LOCK_FILE).is_none());

        let set = open(dir.path(), 4096);
        assert_eq!(set.sealed_len(), 0);
        assert_eq!(set.last_position(), Position::new(1));
    }

    // Read-only (follower) opens

    fn follow(dir: &Path, segment_size: usize) -> SegmentSet {
        SegmentSet::open_read_only(dir, SegmentConfig::new(segment_size)).unwrap()
    }

    /// Every path and length under `dir`, for asserting a follower changed nothing.
    fn tree(dir: &Path) -> Vec<(PathBuf, u64)> {
        let mut out = Vec::new();
        let mut stack = vec![dir.to_path_buf()];
        while let Some(next) = stack.pop() {
            for entry in fs::read_dir(&next).unwrap() {
                let entry = entry.unwrap();
                let meta = entry.metadata().unwrap();
                if meta.is_dir() {
                    stack.push(entry.path());
                } else {
                    out.push((entry.path(), meta.len()));
                }
            }
        }
        out.sort();
        out
    }

    #[test]
    fn open_read_only_on_an_empty_dir_errors() {
        let dir = TempDir::new().unwrap();
        let err = SegmentSet::open_read_only(dir.path(), SegmentConfig::new(4096)).unwrap_err();
        assert!(matches!(err, LogError::Uninitialized { .. }), "got {err:?}");
    }

    #[test]
    fn open_read_only_on_a_missing_dir_errors() {
        let dir = TempDir::new().unwrap();
        let missing = dir.path().join("nope");
        assert!(SegmentSet::open_read_only(&missing, SegmentConfig::new(4096)).is_err());
        assert!(
            !missing.exists(),
            "a read-only open must not create the directory"
        );
    }

    #[test]
    fn open_read_only_mutates_nothing() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            for i in 1..=5u64 {
                append_one(&mut set, format!("event-{i}").as_bytes());
            }
        }
        // A trailing segment mid-creation, which a read-write open would delete.
        let stray = dir.path().join(segment_file_name(Position::new(6)));
        File::create(&stray).unwrap();

        let before = tree(dir.path());
        let mut follower = follow(dir.path(), 4096);
        follower.refresh().unwrap();
        drop(follower);
        assert_eq!(
            before,
            tree(dir.path()),
            "a follower must not touch the directory"
        );
        assert!(
            stray.exists(),
            "a follower must not delete an unfinished segment"
        );
    }

    #[test]
    fn open_read_only_takes_no_lock() {
        let dir = TempDir::new().unwrap();
        let mut writer = open(dir.path(), 4096);
        append_one(&mut writer, b"event");

        // The writer holds the exclusive lock; a follower must attach anyway.
        let follower = follow(dir.path(), 4096);
        assert_eq!(follower.last_position(), Position::new(1));
        assert!(follower.is_read_only());
        assert!(!writer.is_read_only());
    }

    #[test]
    fn open_read_only_skips_a_trailing_unwritten_segment() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            append_one(&mut set, b"event");
        }
        let stray = dir.path().join(segment_file_name(Position::new(2)));
        File::create(&stray).unwrap();

        let mut follower = follow(dir.path(), 4096);
        assert_eq!(follower.active_base(), Position::new(1));
        let refreshed = follower.refresh().unwrap();
        assert_eq!(refreshed.pending_segment.as_deref(), Some(stray.as_path()));
        assert_eq!(refreshed.sealed_added, 0);
        assert_eq!(refreshed.tip, Position::new(1));
    }

    #[test]
    fn open_read_only_skips_a_trailing_torn_header() {
        // A header is written after the file is created and fallocated, and a 64-byte write
        // is not atomic against a concurrent reader, so a follower can see a half-written
        // header. That is a race at the end of the chain, not damage.
        let dir = TempDir::new().unwrap();
        let config = SegmentConfig::new(4096);
        {
            let mut set = open(dir.path(), 4096);
            append_one(&mut set, b"event");
        }
        SegmentSet::create_segment(dir.path(), &config, Position::new(2)).unwrap();
        let second = dir.path().join(segment_file_name(Position::new(2)));
        // Corrupt the header's CRC, which is checked before any field.
        let file = OpenOptions::new().write(true).open(&second).unwrap();
        file.write_all_at(&[0xAB; 4], (SEGMENT_HEADER_SIZE - 4) as u64)
            .unwrap();
        file.sync_all().unwrap();

        let mut follower = follow(dir.path(), 4096);
        assert_eq!(follower.active_base(), Position::new(1));
        let refreshed = follower.refresh().unwrap();
        assert_eq!(refreshed.pending_segment.as_deref(), Some(second.as_path()));

        // The same file in the middle of the chain is corruption, not a race.
        SegmentSet::create_segment(dir.path(), &config, Position::new(3)).unwrap();
        let err = SegmentSet::open_read_only(dir.path(), config).unwrap_err();
        assert!(matches!(err, LogError::Header { .. }), "got {err:?}");
    }

    #[test]
    fn open_read_only_rejects_append_and_a_writer_rejects_refresh() {
        let dir = TempDir::new().unwrap();
        {
            let mut set = open(dir.path(), 4096);
            append_one(&mut set, b"event");
        }
        let mut follower = follow(dir.path(), 4096);
        let err = follower.append_batch(&[b"nope"]).unwrap_err();
        assert!(
            matches!(err, LogError::ReadOnly { op: "append", .. }),
            "got {err:?}"
        );

        let mut writer = open(dir.path(), 4096);
        let err = writer.refresh().unwrap_err();
        assert!(
            matches!(err, LogError::ReadOnly { op: "refresh", .. }),
            "got {err:?}"
        );
    }

    #[test]
    fn refresh_picks_up_committed_batches() {
        let dir = TempDir::new().unwrap();
        let mut writer = open(dir.path(), 4096);
        append_one(&mut writer, b"first");

        let mut follower = follow(dir.path(), 4096);
        assert_eq!(follower.last_position(), Position::new(1));

        for i in 2..=6u64 {
            append_one(&mut writer, format!("event-{i}").as_bytes());
            let refreshed = follower.refresh().unwrap();
            assert_eq!(refreshed.tip, Position::new(i));
            assert_eq!(
                follower.read_at(Position::new(i)).unwrap().data,
                format!("event-{i}").into_bytes()
            );
        }
    }

    #[test]
    fn refresh_is_idempotent_when_nothing_changed() {
        let dir = TempDir::new().unwrap();
        let mut writer = open(dir.path(), 4096);
        append_one(&mut writer, b"only");

        let mut follower = follow(dir.path(), 4096);
        let first = follower.refresh().unwrap();
        let second = follower.refresh().unwrap();
        assert_eq!(first.tip, second.tip);
        assert_eq!(second.sealed_added, 0);
        assert_eq!(follower.active.event_count(), 1, "no double counting");
    }

    #[test]
    fn refresh_ignores_an_uncommitted_partial_batch() {
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let config = SegmentConfig::new(segment_size);
        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"committed");
        }
        let first = dir.path().join(segment_file_name(Position::new(1)));

        let mut follower = follow(dir.path(), segment_size);
        assert_eq!(follower.last_position(), Position::new(1));

        // Records on disk with no commit marker behind them.
        let mut writer =
            Writer::<0>::open(&first, segment_size, config.header_size as u64).unwrap();
        writer.append_data(b"pending-a").unwrap();
        writer.append_data(b"pending-b").unwrap();
        writer.flush_writer().unwrap();

        let refreshed = follower.refresh().unwrap();
        assert_eq!(
            refreshed.tip,
            Position::new(1),
            "an uncommitted batch is invisible"
        );

        writer.commit(3).unwrap();
        let refreshed = follower.refresh().unwrap();
        assert_eq!(
            refreshed.tip,
            Position::new(3),
            "and appears whole once committed"
        );
        assert_eq!(
            follower.read_at(Position::new(3)).unwrap().data,
            b"pending-b".to_vec()
        );
    }

    #[test]
    fn refresh_adopts_a_new_segment_and_seals_the_old() {
        let dir = TempDir::new().unwrap();
        // Small enough that a couple of records force a rollover.
        let segment_size = HEADER + (REC_OVERHEAD + 8 + MARKER) * 2;
        let mut writer = open(dir.path(), segment_size);
        append_one(&mut writer, b"aaaaaaaa");

        let mut follower = follow(dir.path(), segment_size);
        assert_eq!(follower.sealed_len(), 0);

        append_one(&mut writer, b"bbbbbbbb");
        append_one(&mut writer, b"cccccccc");
        assert!(
            writer.sealed_len() >= 1,
            "the writer should have rolled over"
        );

        let refreshed = follower.refresh().unwrap();
        assert_eq!(refreshed.tip, Position::new(3));
        assert!(
            refreshed.sealed_added >= 1,
            "the follower should have sealed a segment"
        );
        assert_eq!(follower.sealed_len(), writer.sealed_len());
        assert_eq!(follower.active_base(), writer.active_base());
        for i in 1..=3u64 {
            assert_eq!(
                follower.read_at(Position::new(i)).unwrap().data,
                writer.read_at(Position::new(i)).unwrap().data
            );
        }
    }

    #[test]
    fn refresh_adopts_multiple_new_segments_in_one_call() {
        let dir = TempDir::new().unwrap();
        let segment_size = HEADER + (REC_OVERHEAD + 8 + MARKER) * 2;
        let mut writer = open(dir.path(), segment_size);
        append_one(&mut writer, b"aaaaaaaa");

        let mut follower = follow(dir.path(), segment_size);
        for i in 2..=12u64 {
            append_one(&mut writer, format!("{i:08}").as_bytes());
        }
        assert!(writer.sealed_len() >= 3, "several rollovers expected");

        // One refresh, not one per segment.
        let refreshed = follower.refresh().unwrap();
        assert_eq!(refreshed.tip, Position::new(12));
        assert_eq!(follower.sealed_len(), writer.sealed_len());
        assert_eq!(refreshed.sealed_added, writer.sealed_len());
        let scanned: Vec<_> = drain(follower.scan_from(Position::new(1)))
            .into_iter()
            .map(|r| r.position)
            .collect();
        assert_eq!(
            scanned.len(),
            12,
            "every position is readable across the seams"
        );
    }

    #[test]
    fn refresh_tolerates_a_header_only_trailing_segment() {
        // A rollover creates and fsyncs the successor before committing the batch that
        // triggered it, so a follower can legitimately see an empty new segment.
        let dir = TempDir::new().unwrap();
        let segment_size = 4096;
        let config = SegmentConfig::new(segment_size);
        {
            let mut set = open(dir.path(), segment_size);
            append_one(&mut set, b"event");
        }
        SegmentSet::create_segment(dir.path(), &config, Position::new(2)).unwrap();

        // The successor already exists, so the follower adopts it at open, not at refresh.
        let mut follower = follow(dir.path(), segment_size);
        assert_eq!(follower.sealed_len(), 1);
        assert_eq!(follower.active_base(), Position::new(2));
        let refreshed = follower.refresh().unwrap();
        assert_eq!(refreshed.sealed_added, 0);
        assert_eq!(refreshed.tip, Position::new(1));
        assert_eq!(follower.next_position(), Position::new(2));
    }
}