qdrant-edge 0.8.0

use std::ops::Range;
use std::path::{Path, PathBuf};

use crate::common::generic_consts::{AccessPattern, Random};
use crate::common::mmap::{Advice, AdviceSetting};
use crate::common::universal_io::{
    CachedReadFs, IsNotFound, OpenOptions, Populate, ReadPipeline, ReadRange, UniversalAppend,
    UniversalIoError, UniversalRead, UniversalReadFs, UniversalWriteFileOps, UserData,
};

use crate::blobstore::Result;
use crate::blobstore::blobstore::Flusher;
use crate::blobstore::error::BlobstoreError;
use crate::blobstore::tracker::{OptionalPointer, PointOffset, ValuePointer};

/// File name of the append-only tracker file
///
/// Deliberately different from the mutable mode tracker file name, so that one mode never attempts to
/// load the incompatible file format of the other.
const FILE_NAME: &str = "log_tracker.dat";

/// Size in bytes of a single mapping entry in the tracker file
const ENTRY_SIZE: u64 = size_of::<OptionalPointer>() as u64;

/// Append-only tracker of value pointers for the append-only storage mode.
///
/// Stores a plain array of [`OptionalPointer`] entries in a single file, without any header. The
/// entry at index `i` is the mapping for point offset `i`, and the number of mappings is defined
/// by the exact file length. The file starts empty and only ever grows by appending; existing
/// bytes are never rewritten.
///
/// Mappings must be set in monotonically increasing point offset order. Skipped point offsets are
/// materialized as zeroed entries, which decode as `None`. New mappings are buffered in memory,
/// and appended to the file as a single atomic append when flushing.
///
/// The file is read and written through the universal IO backend `S`.
///
/// A write may be torn. If the file length is not a multiple of the entry size, the trailing
/// partial entry is ignored when reading, and truncated away when opening writable.
#[derive(Debug)]
pub(crate) struct AppendOnlyTracker<S> {
    /// Path to the tracker file
    path: PathBuf,
    /// Open handle to the tracker file
    file: S,
    /// Number of mappings persisted in the file
    persisted_count: PointOffset,
    /// Mappings that haven't been written to the file yet
    ///
    /// Entry `i` holds the mapping for point offset `persisted_count + i`, with gaps kept as
    /// `None` entries. This is byte for byte the data of the next append.
    pending: Vec<OptionalPointer>,
}

impl<S: UniversalRead> AppendOnlyTracker<S> {
    fn tracker_file_name(dir: &Path) -> PathBuf {
        dir.join(FILE_NAME)
    }

    /// Universal IO open options for the tracker file.
    fn open_options(populate: Populate, writeable: bool) -> OpenOptions {
        OpenOptions {
            writeable,
            need_sequential: false,
            populate,
            advice: AdviceSetting::Advice(Advice::Random),
        }
    }

    /// Schedule a prefetch of the tracker file, so a subsequent open is served from the prefetch
    /// pool.
    pub fn preopen<Fs: CachedReadFs<File = S>>(
        fs: &Fs,
        dir: &Path,
        populate: Populate,
    ) -> Result<()> {
        fs.schedule_prefetch(
            &Self::tracker_file_name(dir),
            Some(Self::open_options(populate, false)),
            None,
        )?;
        Ok(())
    }

    /// Open the tracker file handle, mapping a missing file to a service error.
    fn open_file<Fs: UniversalReadFs<File = S>>(
        fs: &Fs,
        path: &Path,
        populate: Populate,
        writeable: bool,
    ) -> Result<S> {
        fs.open(
            path,
            Self::open_options(populate, writeable),
            Default::default(),
        )
        .map_err(|err| {
            if err.is_not_found() {
                // If config exists and this file doesn't, it should be treated as
                // inconsistent storage rather than a missing one
                BlobstoreError::service_error(format!(
                    "Append-only tracker file does not exist: {}",
                    path.display(),
                ))
            } else {
                BlobstoreError::from(err)
            }
        })
    }

    /// Open an existing tracker in the given directory, read-only.
    ///
    /// If the file does not exist, return an error.
    ///
    /// A trailing partial entry due to a torn write is ignored, the file is left untouched.
    pub fn open_read_only<Fs: UniversalReadFs<File = S>>(
        fs: &Fs,
        dir: &Path,
        populate: Populate,
    ) -> Result<Self> {
        let path = Self::tracker_file_name(dir);
        let file = Self::open_file(fs, &path, populate, false)?;
        let len = file.len::<u8>()?;

        Ok(Self {
            path,
            file,
            persisted_count: count_from_len(len)?,
            pending: Vec::new(),
        })
    }

    pub fn files(&self) -> Vec<PathBuf> {
        vec![self.path.clone()]
    }

    /// Populate the tracker file into the RAM cache.
    pub fn populate(&self) -> Result<()> {
        self.file.populate().map_err(Into::into)
    }

    /// Ask to evict the tracker file from the RAM cache.
    pub fn clear_cache(&self) -> Result<()> {
        self.file.clear_ram_cache().map_err(Into::into)
    }

    /// Number of mappings, including pending ones.
    ///
    /// This is one past the highest point offset that was ever set, which makes it the next point
    /// offset that is allowed to be set.
    pub fn pointer_count(&self) -> PointOffset {
        self.persisted_count + self.pending.len() as PointOffset
    }

    /// Get the mapping at the given point offset.
    ///
    /// Point offsets that were skipped or that are past the highest set mapping yield `None`.
    pub fn get<P: AccessPattern>(&self, point_offset: PointOffset) -> Result<Option<ValuePointer>> {
        if point_offset >= self.pointer_count() {
            return Ok(None);
        }

        if point_offset >= self.persisted_count {
            let pending_index = (point_offset - self.persisted_count) as usize;
            return Ok(self.pending[pending_index].to_option());
        }

        let range = ReadRange::one(u64::from(point_offset) * ENTRY_SIZE);
        let pointer = self.file.read::<_, OptionalPointer>(range, P::default())?[0];
        Ok(pointer.to_option())
    }

    /// Get the mappings for a contiguous range of point offsets.
    ///
    /// The persisted part of the range is read with a single read. Point offsets that were
    /// skipped or that are past the highest set mapping yield `None`.
    ///
    /// The result holds one entry per requested point offset, so callers should bound the range
    /// they ask for.
    pub fn get_range<P: AccessPattern>(
        &self,
        point_offsets: Range<PointOffset>,
    ) -> Result<Vec<Option<ValuePointer>>> {
        let mut pointers = Vec::with_capacity(point_offsets.len());
        let start = point_offsets.start;
        let end = point_offsets.end.min(self.pointer_count());

        // Persisted part of the range, read in a single read
        let persisted_end = end.min(self.persisted_count);
        if start < persisted_end {
            let range = ReadRange {
                byte_offset: u64::from(start) * ENTRY_SIZE,
                length: u64::from(persisted_end - start),
            };
            let entries = self.file.read::<_, OptionalPointer>(range, P::default())?;
            pointers.extend(entries.iter().map(|entry| entry.to_option()));
        }

        // Pending part of the range
        for point_offset in start.max(self.persisted_count)..end {
            let pending_index = (point_offset - self.persisted_count) as usize;
            pointers.push(self.pending[pending_index].to_option());
        }

        // Point offsets past the highest set mapping
        pointers.resize(point_offsets.len(), None);

        Ok(pointers)
    }

    /// Iterate the mappings for the given point offsets.
    ///
    /// Issues batched reads against the tracker file through the backend's read pipeline, so
    /// async backends can fetch entries in parallel. Pending mappings and point offsets past the
    /// highest set mapping are yielded directly, without touching the file.
    ///
    /// Yields one item per requested point offset, possibly in a different order.
    pub fn iter<U, I>(&self, point_offsets: I) -> Result<Iter<'_, U, I, S>>
    where
        U: UserData,
        I: Iterator<Item = (U, PointOffset)>,
    {
        Ok(Iter {
            point_offsets,
            tracker: self,
            pipeline: S::ReadPipeline::new()?,
        })
    }

    /// Reopen the tracker file and read the number of mappings it holds, *without* making the
    /// new mappings visible to reads.
    ///
    /// Observing and publishing are separate steps so that the caller can reload the pages in
    /// between: a writer persists value data before the mappings that reference it, so a count
    /// observed here is backed by page data that a page reload started afterwards is guaranteed
    /// to see. Publishing that count only once the page reload succeeded, through
    /// [`commit_reload`](Self::commit_reload), keeps a failing reload from exposing mappings
    /// that point into pages which were never loaded.
    ///
    /// Reopening without committing is harmless on its own: reads stay bounded by the unchanged
    /// `persisted_count`, and the bytes below it never change.
    ///
    /// Important assumptions:
    ///
    /// - Should only be called on read-only instances of the tracker.
    /// - Mappings are append-only, existing entries never change.
    /// - Partial writes are possible, but ignored: a trailing partial entry is not counted.
    pub fn reload_count(&mut self) -> Result<PendingReload> {
        debug_assert!(
            self.pending.is_empty(),
            "live reload must only be used on read-only instances",
        );

        self.file.reopen()?;
        let len = self.file.len::<u8>()?;
        let new_count = count_from_len(len)?;

        if new_count < self.persisted_count {
            return Err(BlobstoreError::service_error(format!(
                "live reload cannot decrease mapping count, possible data loss: old count {}, new count {new_count}",
                self.persisted_count,
            )));
        }

        Ok(PendingReload { count: new_count })
    }

    /// Make the mappings observed by [`reload_count`](Self::reload_count) visible to reads.
    ///
    /// Only call this once the pages holding the value data those mappings reference have been
    /// reloaded, see [`reload_count`](Self::reload_count).
    pub fn commit_reload(&mut self, reload: PendingReload) {
        let PendingReload { count } = reload;
        debug_assert!(
            count >= self.persisted_count,
            "a committed reload must not decrease the mapping count",
        );
        self.persisted_count = count;
    }
}

/// A mapping count read from the tracker file that is not visible to reads yet.
///
/// Produced by [`AppendOnlyTracker::reload_count`] and published by
/// [`AppendOnlyTracker::commit_reload`], so that the pages backing the new mappings can be
/// reloaded in between the two.
#[must_use = "an observed reload only becomes visible once committed"]
#[derive(Debug, Copy, Clone)]
pub(crate) struct PendingReload {
    /// Number of mappings the file held at the time of the observation
    count: PointOffset,
}

#[cfg(test)]
impl PendingReload {
    /// Number of mappings the file held at the time of the observation.
    pub fn count(self) -> PointOffset {
        self.count
    }
}

impl<S: UniversalAppend> AppendOnlyTracker<S> {
    /// Create a new empty tracker in the given directory, truncating the file if it already
    /// exists.
    ///
    /// The directory must exist already.
    pub fn new(fs: &S::Fs, dir: &Path) -> Result<Self> {
        let path = Self::tracker_file_name(dir);
        fs.create(&path, 0)?;
        let file = fs.open(
            &path,
            Self::open_options(Populate::No, true),
            Default::default(),
        )?;
        Ok(Self {
            path,
            file,
            persisted_count: 0,
            pending: Vec::new(),
        })
    }

    /// Open an existing tracker in the given directory, writable.
    ///
    /// If the file does not exist, return an error.
    ///
    /// A trailing partial entry due to a torn write is truncated away, so that appends always
    /// start at a whole entry offset.
    pub fn open_writable(fs: &S::Fs, dir: &Path, populate: Populate) -> Result<Self> {
        let path = Self::tracker_file_name(dir);
        let mut file = Self::open_file(fs, &path, populate, true)?;

        let len = file.len::<u8>()?;
        let aligned_len = len - (len % ENTRY_SIZE);
        if aligned_len != len {
            // Recover from a torn write by truncating the partial trailing entry. The handle is
            // opened from scratch afterwards, shrinking is not supported through an open handle.
            drop(file);
            fs.create(&path, aligned_len as usize)?;
            file = Self::open_file(fs, &path, populate, true)?;
        }

        Ok(Self {
            path,
            file,
            persisted_count: count_from_len(aligned_len)?,
            pending: Vec::new(),
        })
    }

    /// Set the mapping for the given point offset, buffering it in memory until flushed.
    ///
    /// Point offsets must be set in monotonically increasing order: each offset must be larger
    /// than every offset set before it. Skipped offsets are backfilled as `None` entries.
    pub fn set(&mut self, point_offset: PointOffset, pointer: ValuePointer) -> Result<()> {
        // Defensive re-check: the storage validates this before appending any value data, see
        // Logstore::put_value
        let next = self.pointer_count();
        if point_offset < next {
            return Err(BlobstoreError::unsupported_operation(format!(
                "cannot set mapping for point offset {point_offset}, the tracker is append-only \
                 and requires monotonically increasing point offsets, the next allowed point \
                 offset is {next}",
            )));
        }

        // Materialize skipped point offsets as None entries
        let pending_index = (point_offset - self.persisted_count) as usize;
        self.pending.resize(pending_index, OptionalPointer::none());
        self.pending.push(OptionalPointer::some(pointer));

        Ok(())
    }

    /// Append pending mappings for point offsets up to, but excluding, `target` to the file.
    ///
    /// All appended mappings land with a single atomic append at the end of the file: one
    /// grow+write syscall on local backends, one RPC on object stores. The `target` is captured
    /// through [`pointer_count`](Self::pointer_count) when a flusher is created, so that mappings
    /// set while a flush is in progress stay pending.
    ///
    /// A stale flush, with a `target` at or below what a more recent flush already persisted, is
    /// a no-op: bytes that were appended before must never be written again.
    ///
    /// The append offset doubles as a compare-and-swap token: if a previous flush appended these
    /// mappings but its acknowledgement was lost, the retry conflicts instead of appending twice,
    /// and the mappings are adopted as persisted when the file ends exactly where this append
    /// would have ended.
    ///
    /// This does not sync the file to disk, invoke a [`flusher`](Self::flusher) afterwards.
    pub fn write_pending(&mut self, target: PointOffset) -> Result<()> {
        if target <= self.persisted_count {
            return Ok(());
        }

        let count = (target - self.persisted_count) as usize;
        debug_assert!(
            count <= self.pending.len(),
            "flush target exceeds pending mappings",
        );
        let count = count.min(self.pending.len());

        let offset = u64::from(self.persisted_count) * ENTRY_SIZE;
        let end = offset + count as u64 * ENTRY_SIZE;

        match self.file.append(offset, &self.pending[..count]) {
            Ok(()) => {}
            // A retried append after a lost acknowledgement conflicts instead of appending
            // twice. If the file ends exactly where this append would have ended, the mappings
            // landed before; adopt them as persisted. Any other length means the file was
            // modified outside this writer.
            Err(UniversalIoError::AppendOffsetConflict { .. }) => {
                self.file.reopen()?;
                let len = self.file.len::<u8>()?;
                if len != end {
                    return Err(BlobstoreError::service_error(format!(
                        "append-only tracker file {} was modified outside this writer: it ends \
                         at byte {len}, expected {offset} before or {end} after the append",
                        self.path.display(),
                    )));
                }
            }
            Err(err) => return Err(err.into()),
        }

        self.pending.drain(..count);
        self.persisted_count += count as PointOffset;

        Ok(())
    }

    /// Create a closure that syncs all written mappings in the tracker file to disk.
    pub fn flusher(&self) -> Flusher {
        let flusher = self.file.flusher();
        Box::new(move || flusher().map_err(BlobstoreError::from))
    }
}

/// Batched mapping lookup, see [`AppendOnlyTracker::iter`].
pub(crate) struct Iter<'a, U, I, S>
where
    U: UserData,
    I: Iterator<Item = (U, PointOffset)>,
    S: UniversalRead,
{
    point_offsets: I,
    tracker: &'a AppendOnlyTracker<S>,
    pipeline: S::ReadPipeline<'a, U>,
}

impl<'a, U, I, S> Iterator for Iter<'a, U, I, S>
where
    U: UserData,
    I: Iterator<Item = (U, PointOffset)>,
    S: UniversalRead,
{
    type Item = Result<(U, Option<ValuePointer>)>;

    fn next(&mut self) -> Option<Self::Item> {
        while self.pipeline.can_schedule()
            && let Some((user_data, point_offset)) = self.point_offsets.next()
        {
            // Pending mappings and point offsets past the highest set mapping are yielded
            // directly, only persisted mappings are read from the file
            if point_offset >= self.tracker.persisted_count {
                let pending_index = (point_offset - self.tracker.persisted_count) as usize;
                let pointer = self
                    .tracker
                    .pending
                    .get(pending_index)
                    .and_then(|entry| entry.to_option());
                return Some(Ok((user_data, pointer)));
            }

            let start = u64::from(point_offset) * ENTRY_SIZE;
            let result = self.pipeline.schedule::<Random>(
                user_data,
                &self.tracker.file,
                start..start + ENTRY_SIZE,
                align_of::<OptionalPointer>(),
            );

            if let Err(err) = result {
                return Some(Err(err.into()));
            }
        }

        let result = self.pipeline.wait_bytemuck::<OptionalPointer>();

        let (user_data, entry) = match result {
            Ok(entry) => entry?,
            Err(err) => return Some(Err(err.into())),
        };

        let &[entry] = entry.as_ref() else {
            unreachable!();
        };

        Some(Ok((user_data, entry.to_option())))
    }
}

/// Number of whole mapping entries in a tracker file of the given length.
///
/// A trailing partial entry due to a torn write is not counted.
fn count_from_len(len: u64) -> Result<PointOffset> {
    PointOffset::try_from(len / ENTRY_SIZE).map_err(|_| {
        BlobstoreError::service_error(format!(
            "append-only tracker file of {len} bytes holds more mappings than supported",
        ))
    })
}

#[cfg(test)]
mod tests {
    use std::io::Write as _;

    use crate::common::generic_consts::Random;
    use crate::common::universal_io::{MmapFile, MmapFs};
    use fs_err as fs;
    use tempfile::TempDir;

    use super::*;

    fn empty_tracker() -> (TempDir, AppendOnlyTracker<MmapFile>) {
        let dir = TempDir::new().unwrap();
        let tracker = AppendOnlyTracker::new(&MmapFs, dir.path()).unwrap();
        (dir, tracker)
    }

    fn pointer(n: u32) -> ValuePointer {
        ValuePointer::new(0, n * 2, n * 3 + 1)
    }

    fn file_len(tracker: &AppendOnlyTracker<MmapFile>) -> u64 {
        fs::metadata(&tracker.path).unwrap().len()
    }

    #[test]
    fn test_new_tracker_is_empty() {
        let (_dir, tracker) = empty_tracker();
        assert_eq!(tracker.pointer_count(), 0);
        assert_eq!(tracker.get::<Random>(0).unwrap(), None);
        assert_eq!(file_len(&tracker), 0);
    }

    #[test]
    fn test_open_missing_tracker_fails() {
        let dir = TempDir::new().unwrap();
        assert!(
            AppendOnlyTracker::<MmapFile>::open_writable(&MmapFs, dir.path(), Populate::No)
                .is_err()
        );
        assert!(
            AppendOnlyTracker::<MmapFile>::open_read_only(&MmapFs, dir.path(), Populate::No)
                .is_err()
        );
    }

    #[test]
    fn test_set_and_get_pending() {
        let (_dir, mut tracker) = empty_tracker();

        for n in 0..5 {
            tracker.set(n, pointer(n)).unwrap();
        }

        assert_eq!(tracker.pointer_count(), 5);
        for n in 0..5 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), Some(pointer(n)));
        }
        assert_eq!(tracker.get::<Random>(5).unwrap(), None);

        // Nothing is written to the file until flushed
        assert_eq!(file_len(&tracker), 0);
    }

    #[test]
    fn test_set_rejects_non_monotonic_point_offsets() {
        let (_dir, mut tracker) = empty_tracker();

        tracker.set(0, pointer(0)).unwrap();
        // Setting the same point offset twice is rejected
        assert!(tracker.set(0, pointer(0)).is_err());

        tracker.set(5, pointer(5)).unwrap();
        // Setting a lower point offset is rejected, even if it was never set
        assert!(tracker.set(3, pointer(3)).is_err());
        assert!(tracker.set(5, pointer(5)).is_err());

        tracker.set(6, pointer(6)).unwrap();
        assert_eq!(tracker.pointer_count(), 7);
    }

    #[test]
    fn test_skipped_point_offsets_read_as_none() {
        let (_dir, mut tracker) = empty_tracker();

        tracker.set(0, pointer(0)).unwrap();
        tracker.set(4, pointer(4)).unwrap();

        assert_eq!(tracker.pointer_count(), 5);
        assert_eq!(tracker.get::<Random>(0).unwrap(), Some(pointer(0)));
        for n in 1..4 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), None);
        }
        assert_eq!(tracker.get::<Random>(4).unwrap(), Some(pointer(4)));

        // Gaps are persisted as zeroed entries
        tracker.write_pending(tracker.pointer_count()).unwrap();
        assert_eq!(file_len(&tracker), 5 * ENTRY_SIZE);
        assert_eq!(tracker.get::<Random>(2).unwrap(), None);
    }

    #[test]
    fn test_write_pending_and_reopen() {
        let dir = TempDir::new().unwrap();

        let mut tracker = AppendOnlyTracker::<MmapFile>::new(&MmapFs, dir.path()).unwrap();
        for n in 0..5 {
            tracker.set(n, pointer(n)).unwrap();
        }
        tracker.write_pending(tracker.pointer_count()).unwrap();
        tracker.flusher()().unwrap();

        // The file length matches the exact number of mappings
        assert_eq!(file_len(&tracker), 5 * ENTRY_SIZE);
        drop(tracker);

        let tracker =
            AppendOnlyTracker::<MmapFile>::open_writable(&MmapFs, dir.path(), Populate::No)
                .unwrap();
        assert_eq!(tracker.pointer_count(), 5);
        for n in 0..5 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), Some(pointer(n)));
        }
        assert_eq!(
            tracker.get_range::<Random>(0..7).unwrap(),
            (0..5)
                .map(|n| Some(pointer(n)))
                .chain([None, None])
                .collect::<Vec<_>>(),
        );
    }

    #[test]
    fn test_partial_write_pending() {
        let (_dir, mut tracker) = empty_tracker();

        for n in 0..5 {
            tracker.set(n, pointer(n)).unwrap();
        }

        // Flush only the first three mappings
        tracker.write_pending(3).unwrap();
        assert_eq!(file_len(&tracker), 3 * ENTRY_SIZE);
        assert_eq!(tracker.pointer_count(), 5);
        for n in 0..5 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), Some(pointer(n)));
        }

        // Then the rest
        tracker.write_pending(5).unwrap();
        assert_eq!(file_len(&tracker), 5 * ENTRY_SIZE);
        for n in 0..5 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), Some(pointer(n)));
        }
    }

    #[test]
    fn test_stale_flush_is_noop() {
        let (_dir, mut tracker) = empty_tracker();

        for n in 0..3 {
            tracker.set(n, pointer(n)).unwrap();
        }
        tracker.write_pending(3).unwrap();
        assert_eq!(file_len(&tracker), 3 * ENTRY_SIZE);

        // A stale flush with an old target must not write anything again
        tracker.write_pending(2).unwrap();
        tracker.write_pending(3).unwrap();
        assert_eq!(file_len(&tracker), 3 * ENTRY_SIZE);
        assert_eq!(tracker.pointer_count(), 3);
        for n in 0..3 {
            assert_eq!(tracker.get::<Random>(n).unwrap(), Some(pointer(n)));
        }
    }

    #[test]
    fn test_torn_write_is_ignored_and_truncated() {
        let dir = TempDir::new().unwrap();

        let mut tracker = AppendOnlyTracker::<MmapFile>::new(&MmapFs, dir.path()).unwrap();
        for n in 0..5 {
            tracker.set(n, pointer(n)).unwrap();
        }
        tracker.write_pending(tracker.pointer_count()).unwrap();
        let path = tracker.path.clone();
        drop(tracker);

        // Simulate a torn write by appending a partial entry
        let mut file = fs::OpenOptions::new().append(true).open(&path).unwrap();
        file.write_all(&[0xAA; 7]).unwrap();
        drop(file);
        assert_eq!(fs::metadata(&path).unwrap().len(), 5 * ENTRY_SIZE + 7);

        // A read-only open ignores the partial entry, but leaves the file untouched
        let tracker =
            AppendOnlyTracker::<MmapFile>::open_read_only(&MmapFs, dir.path(), Populate::No)
                .unwrap();
        assert_eq!(tracker.pointer_count(), 5);
        assert_eq!(tracker.get::<Random>(4).unwrap(), Some(pointer(4)));
        assert_eq!(file_len(&tracker), 5 * ENTRY_SIZE + 7);
        drop(tracker);

        // A writable open truncates the partial entry away
        let tracker =
            AppendOnlyTracker::<MmapFile>::open_writable(&MmapFs, dir.path(), Populate::No)
                .unwrap();
        assert_eq!(tracker.pointer_count(), 5);
        assert_eq!(tracker.get::<Random>(4).unwrap(), Some(pointer(4)));
        assert_eq!(file_len(&tracker), 5 * ENTRY_SIZE);
    }

    /// A conflicting append whose mappings already landed — a retried flush after a lost
    /// acknowledgement — is adopted as persisted instead of appending twice.
    #[test]
    fn test_write_pending_adopts_lost_append_on_conflict() {
        let (_dir, mut tracker) = empty_tracker();

        tracker.set(0, pointer(0)).unwrap();

        // Simulate a flush that landed but was never acknowledged: the entry is in the
        // (previously empty) file, but the tracker still counts it as pending
        let entry = OptionalPointer::some(pointer(0));
        fs::write(&tracker.path, bytemuck::bytes_of(&entry)).unwrap();

        tracker.write_pending(1).unwrap();
        assert_eq!(tracker.persisted_count, 1);
        assert!(tracker.pending.is_empty());
        assert_eq!(
            file_len(&tracker),
            ENTRY_SIZE,
            "the mapping must not be appended twice",
        );
        assert_eq!(tracker.get::<Random>(0).unwrap(), Some(pointer(0)));
    }

    /// A conflicting append against a file that does not end where the append would have ended
    /// fails: the file was modified outside this writer.
    #[test]
    fn test_write_pending_rejects_foreign_growth() {
        let (_dir, mut tracker) = empty_tracker();

        tracker.set(0, pointer(0)).unwrap();

        // Something else grew the (previously empty) file to an unexpected length
        fs::write(&tracker.path, [9; 7]).unwrap();

        let err = tracker.write_pending(1).unwrap_err();
        assert!(matches!(err, BlobstoreError::ServiceError { .. }));

        // Nothing was adopted, the pending mapping is kept for a later flush
        assert_eq!(tracker.persisted_count, 0);
        assert_eq!(tracker.pending.len(), 1);
    }

    #[test]
    fn test_live_reload() {
        let dir = TempDir::new().unwrap();

        let mut writer = AppendOnlyTracker::<MmapFile>::new(&MmapFs, dir.path()).unwrap();
        for n in 0..3 {
            writer.set(n, pointer(n)).unwrap();
        }
        writer.write_pending(writer.pointer_count()).unwrap();

        let mut reader =
            AppendOnlyTracker::<MmapFile>::open_read_only(&MmapFs, dir.path(), Populate::No)
                .unwrap();
        assert_eq!(reader.pointer_count(), 3);
        let reload = reader.reload_count().unwrap();
        assert_eq!(reload.count(), 3);
        reader.commit_reload(reload);

        // The reader picks up newly appended mappings
        for n in 3..6 {
            writer.set(n, pointer(n)).unwrap();
        }
        writer.write_pending(writer.pointer_count()).unwrap();

        let reload = reader.reload_count().unwrap();
        assert_eq!(reload.count(), 6);

        // Observing alone does not expose them, only committing does
        assert_eq!(reader.pointer_count(), 3);
        assert_eq!(reader.get::<Random>(3).unwrap(), None);

        reader.commit_reload(reload);
        assert_eq!(reader.pointer_count(), 6);
        for n in 0..6 {
            assert_eq!(reader.get::<Random>(n).unwrap(), Some(pointer(n)));
        }

        let reload = reader.reload_count().unwrap();
        assert_eq!(reload.count(), 6);
        reader.commit_reload(reload);
    }

    /// The batched lookup resolves persisted, pending, skipped and out of range point offsets,
    /// possibly yielding them in a different order.
    #[test]
    fn test_iter_spans_persisted_and_pending() {
        let (_dir, mut tracker) = empty_tracker();

        for n in 0..3 {
            tracker.set(n, pointer(n)).unwrap();
        }
        tracker.write_pending(3).unwrap();
        tracker.set(3, pointer(3)).unwrap();
        tracker.set(6, pointer(6)).unwrap();

        let requested = [2, 0, 6, 4, 3, 9];
        let mut collected = tracker
            .iter(requested.iter().map(|&offset| (offset, offset)))
            .unwrap()
            .map(|result| result.unwrap())
            .collect::<Vec<_>>();
        collected.sort_by_key(|(point_offset, _)| *point_offset);

        assert_eq!(
            collected,
            vec![
                (0, Some(pointer(0))),
                (2, Some(pointer(2))),
                (3, Some(pointer(3))),
                (4, None),
                (6, Some(pointer(6))),
                (9, None),
            ],
        );
    }

    #[test]
    fn test_get_range_spans_persisted_and_pending() {
        let (_dir, mut tracker) = empty_tracker();

        for n in 0..3 {
            tracker.set(n, pointer(n)).unwrap();
        }
        tracker.write_pending(3).unwrap();
        tracker.set(3, pointer(3)).unwrap();
        tracker.set(6, pointer(6)).unwrap();

        assert_eq!(
            tracker.get_range::<Random>(0..9).unwrap(),
            vec![
                Some(pointer(0)),
                Some(pointer(1)),
                Some(pointer(2)),
                Some(pointer(3)),
                None,
                None,
                Some(pointer(6)),
                None,
                None,
            ],
        );
        assert_eq!(
            tracker.get_range::<Random>(2..4).unwrap(),
            vec![Some(pointer(2)), Some(pointer(3)),]
        );
        assert_eq!(tracker.get_range::<Random>(7..9).unwrap(), vec![None, None]);
        #[allow(clippy::reversed_empty_ranges)]
        let empty = tracker.get_range::<Random>(3..3).unwrap();
        assert!(empty.is_empty());
    }
}