//! A file-backed FIFO queue with crash-tolerant, at-least-once delivery.
//!
//! A [`PersistentQueue`] stores CBOR-serialized values in a single append-mostly
//! file. It is designed to be the durable backbone of a work pipeline: the
//! process owning the queue can be suspended, killed, or lose power at any
//! moment, and on reopening the queue every value that was durably enqueued
//! and not durably acknowledged is delivered again.
//!
//! # Delivery model
//!
//! [`PersistentQueue::get`] does not remove a value; it returns a [`Lease`].
//! Dropping the lease returns the value to the queue, while [`Lease::ack`]
//! consumes it and marks the record as dead. Because a leased record stays
//! live on disk until its acknowledgement is synced, a crash while values are
//! in flight simply redelivers them later: delivery is *at least once*, and
//! consumers must be prepared to see a value twice (idempotent processing is
//! the intended usage).
//!
//! # Durability model
//!
//! Enqueued values are written to the file immediately but not synced;
//! acknowledgements are buffered in memory. [`PersistentQueue::sync`] writes
//! the buffered acknowledgements and flushes everything to stable storage.
//! After a crash, the queue rolls back to a state no older than the last
//! `sync`: un-synced puts may be lost, and un-synced acks are redelivered.
//!
//! Because acknowledgements never reach the disk before `sync` is called,
//! a pipeline that moves values from one queue to another preserves
//! at-least-once delivery end to end by syncing the downstream queue before
//! the upstream one.
//!
//! # Exclusivity
//!
//! A queue file may be opened by only one process at a time, enforced with an
//! OS-level lock on a `.lock` sidecar file. Within a process, the queue is
//! [`Send`] and [`Sync`] and all operations take `&self`.
//!
//! # Example
//!
//! ```
//! use persistent_queue::PersistentQueue;
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let dir = tempfile::tempdir()?;
//! let queue: PersistentQueue<String> = PersistentQueue::open(dir.path().join("queue"))?;
//!
//! queue.put(&String::from("hello"))?;
//!
//! if let Some(lease) = queue.get()? {
//!     assert_eq!(&*lease, "hello");
//!     let value = lease.ack();
//!     assert_eq!(value, "hello");
//! }
//!
//! queue.sync()?;
//! # Ok(())
//! # }
//! ```

mod filelock;

#[cfg(test)]
mod tests;

use std::{
    collections::{HashMap, HashSet, VecDeque},
    fs::File,
    io::{BufReader, Read, Seek, SeekFrom, Write},
    marker::PhantomData,
    ops::{Deref, DerefMut},
    path::{Path, PathBuf},
    sync::Mutex,
};

use filelock::FileLock;

const MAGIC: [u8; 8] = *b"persistq";
const VERSION: u16 = 1;
const FILE_HEADER_BYTES: u64 = 18;
const RECORD_HEADER_BYTES: u64 = 13;

const STATE_LIVE: u8 = 0;
const STATE_DEAD: u8 = 1;

/// Compaction runs during [`PersistentQueue::sync`] once at least this many
/// bytes are reclaimable and they make up at least half of the record data.
const COMPACT_MIN_RECLAIMABLE: u64 = 1 << 20;

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("attempted to open an incorrectly structured file as a PersistentQueue")]
    InvalidFile,
    #[error("the file contains an unrecognized file structure version")]
    UnknownVersion,
    #[error("the queue is already open in another process")]
    QueueLocked,
    #[error(transparent)]
    CiboriumEncode(#[from] ciborium::ser::Error<std::io::Error>),
    #[error(transparent)]
    CiboriumDecode(#[from] ciborium::de::Error<std::io::Error>),
    #[error(transparent)]
    IO(#[from] std::io::Error),
}

/// A durable FIFO queue of values of type `T`.
///
/// See the crate-level documentation for the delivery and durability model.
pub struct PersistentQueue<T> {
    path: PathBuf,
    inner: Mutex<Inner>,
    _lock: FileLock,
    // No T is ever stored in the queue itself; the phantom only ties the
    // on-disk records to a single element type.
    _phantom: PhantomData<fn() -> T>,
}

impl<T> std::fmt::Debug for PersistentQueue<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "PersistentQueue {{ path: {:?}, .. }}", self.path)
    }
}

impl<T> PersistentQueue<T> {
    /// Opens the queue at `path`, creating the file if it does not exist.
    ///
    /// Opening recovers from any earlier crash: incomplete or corrupt
    /// records at the tail of the file are discarded, and previously
    /// leased-but-unacknowledged values become deliverable again.
    ///
    /// Fails with [`Error::QueueLocked`] if another process has the queue
    /// open, using a `.lock` sidecar file next to the data file.
    pub fn open(path: impl Into<PathBuf>) -> Result<PersistentQueue<T>, Error> {
        let path = path.into();

        let lock = FileLock::acquire(&sibling_path(&path, ".lock"))?;

        // Clean up the temporary file left behind if a compaction was
        // interrupted by a crash before its atomic rename.
        match std::fs::remove_file(sibling_path(&path, ".compact")) {
            Ok(()) => {}
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
            Err(err) => return Err(err.into()),
        }

        let file = File::options()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)?;
        let len = file.metadata()?.len();

        let inner = if len < FILE_HEADER_BYTES {
            Inner::create(file, len)?
        } else {
            Inner::recover(file, len)?
        };

        Ok(PersistentQueue {
            path,
            inner: Mutex::new(inner),
            _lock: lock,
            _phantom: PhantomData,
        })
    }

    /// Makes all completed operations durable.
    ///
    /// Buffered acknowledgements are written out, the file header is
    /// updated, and the file is synced to stable storage. If enough of the
    /// file consists of dead records afterwards, it is compacted by
    /// rewriting the live records to a temporary file and atomically
    /// renaming it into place.
    pub fn sync(&self) -> Result<(), Error> {
        self.inner.lock().unwrap().sync(&self.path)
    }

    /// True if the queue holds no values, counting leased values as held.
    pub fn is_empty(&self) -> bool {
        self.inner.lock().unwrap().live == 0
    }

    /// The number of values available for delivery.
    pub fn pending(&self) -> usize {
        let inner = self.inner.lock().unwrap();
        (inner.live as usize) - inner.leased.len()
    }

    /// The number of values currently out on leases.
    pub fn in_flight(&self) -> usize {
        self.inner.lock().unwrap().leased.len()
    }

    fn ack_lease(&self, id: u64) {
        let mut inner = self.inner.lock().unwrap();

        let location = inner
            .leased
            .remove(&id)
            .expect("lease ids are issued once and consumed once");
        inner.acked.push(location);
        inner.live -= 1;
        inner.dirty = true;
    }

    fn release_lease(&self, id: u64) {
        // Called from Lease::drop, so this must not panic during a panic
        // unwind; if the mutex is poisoned the record is simply left for
        // crash recovery to redeliver.
        let Ok(mut inner) = self.inner.lock() else {
            return;
        };

        if let Some(location) = inner.leased.remove(&id) {
            inner.requeued.push_back(location);
        }
    }
}

impl<T> PersistentQueue<T>
where
    T: serde::Serialize + for<'a> serde::Deserialize<'a>,
{
    /// Appends a value to the back of the queue.
    ///
    /// The value is written to the file immediately, but is only guaranteed
    /// to survive a crash after the next [`sync`](PersistentQueue::sync).
    pub fn put(&self, value: &T) -> Result<(), Error> {
        let mut payload = Vec::new();
        ciborium::into_writer(value, &mut payload)?;

        self.inner.lock().unwrap().append_record(&payload)
    }

    /// Takes the value at the front of the queue, if any, on a lease.
    ///
    /// Values returned to the queue by dropped leases are delivered before
    /// values that have never been delivered.
    pub fn get(&self) -> Result<Option<Lease<'_, T>>, Error> {
        let mut inner = self.inner.lock().unwrap();

        let location = match inner.requeued.pop_front() {
            Some(location) => location,
            None => match inner.scan_next()? {
                Some(location) => location,
                None => return Ok(None),
            },
        };

        let value = match Self::decode(&mut inner, location) {
            Ok(value) => value,
            Err(err) => {
                // The record was already claimed; put it back so that a
                // transient read failure does not lose it.
                inner.requeued.push_front(location);
                return Err(err);
            }
        };

        let id = inner.next_lease_id;
        inner.next_lease_id += 1;
        inner.leased.insert(id, location);

        Ok(Some(Lease {
            queue: self,
            id,
            value: Some(value),
        }))
    }

    /// The values [`get`](PersistentQueue::get) would deliver next, in
    /// delivery order, up to `limit` of them.
    ///
    /// Nothing is leased, acknowledged, or consumed: this is a look at
    /// the front of the queue for reporting, and by the time it returns
    /// a concurrent `get` may already have taken the values it names.
    pub fn peek(&self, limit: usize) -> Result<Vec<T>, Error> {
        let mut inner = self.inner.lock().unwrap();
        let locations = inner.peek_locations(limit)?;

        locations
            .into_iter()
            .map(|location| Self::decode(&mut inner, location))
            .collect()
    }

    /// Walks the values the queue holds, in delivery order, without
    /// taking any of them.
    ///
    /// Like [`peek`](PersistentQueue::peek) but streaming, for a caller
    /// that wants to look at everything and has no use for a copy of it
    /// all at once.
    pub fn inspect(&self, mut visit: impl FnMut(&T)) -> Result<(), Error> {
        let mut inner = self.inner.lock().unwrap();
        let locations = inner.peek_locations(usize::MAX)?;

        for location in locations {
            visit(&Self::decode(&mut inner, location)?);
        }

        Ok(())
    }

    /// Drops every value `keep` rejects, leaving the rest in the order
    /// they were in, and reports how many went.
    ///
    /// The queue is read to its end and the survivors are appended
    /// behind them, so nothing is examined twice and nothing new is
    /// read; the caller must have the queue to itself, which having a
    /// `&PersistentQueue` at all nearly assures.
    ///
    /// Crash-safe in the queue's usual direction: none of it is durable
    /// until the next [`sync`](PersistentQueue::sync), so an interrupted
    /// pass leaves the queue as it was, or — if some of the appends
    /// reached the file — holding a few values twice. It never drops
    /// what `keep` accepted.
    pub fn retain(&self, mut keep: impl FnMut(&T) -> bool) -> Result<usize, Error> {
        let mut dropped = 0;

        for _ in 0..self.pending() {
            let Some(lease) = self.get()? else {
                break;
            };

            let value = lease.ack();

            match keep(&value) {
                true => self.put(&value)?,
                false => dropped += 1,
            }
        }

        Ok(dropped)
    }

    fn decode(inner: &mut Inner, location: RecordLocation) -> Result<T, Error> {
        let payload = inner.read_payload(location)?;

        Ok(ciborium::from_reader(payload.as_slice())?)
    }
}

impl<T> Drop for PersistentQueue<T> {
    fn drop(&mut self) {
        // Best-effort final sync; anything that fails to reach the disk
        // here is redelivered by crash recovery on the next open.
        let Ok(mut inner) = self.inner.lock() else {
            return;
        };

        if let Err(err) = inner.sync(&self.path) {
            eprintln!("persistent-queue: error while syncing queue during drop: {err}");
        }
    }
}

/// A value delivered from a [`PersistentQueue`], pending acknowledgement.
///
/// The lease dereferences to the value. Call [`ack`](Lease::ack) once the
/// value has been fully processed; dropping the lease instead returns the
/// value to the queue for redelivery. Mutations made through [`DerefMut`]
/// affect only this in-memory copy, never the queued record.
pub struct Lease<'queue, T> {
    queue: &'queue PersistentQueue<T>,
    id: u64,
    value: Option<T>,
}

impl<T> Lease<'_, T> {
    /// Acknowledges the value, removing it from the queue, and returns it.
    ///
    /// The removal becomes durable at the queue's next
    /// [`sync`](PersistentQueue::sync); until then a crash would cause the
    /// value to be delivered again.
    pub fn ack(mut self) -> T {
        let value = self
            .value
            .take()
            .expect("a lease holds its value until consumed");
        self.queue.ack_lease(self.id);

        value
    }

    /// Returns the value to the queue for redelivery.
    ///
    /// This is the same as dropping the lease; it exists to make the
    /// intent explicit at call sites.
    pub fn release(self) {}
}

impl<T> Deref for Lease<'_, T> {
    type Target = T;

    fn deref(&self) -> &T {
        self.value
            .as_ref()
            .expect("a lease holds its value until consumed")
    }
}

impl<T> DerefMut for Lease<'_, T> {
    fn deref_mut(&mut self) -> &mut T {
        self.value
            .as_mut()
            .expect("a lease holds its value until consumed")
    }
}

impl<T> std::fmt::Debug for Lease<'_, T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Lease {{ id: {}, .. }}", self.id)
    }
}

impl<T> Drop for Lease<'_, T> {
    fn drop(&mut self) {
        if self.value.is_some() {
            self.queue.release_lease(self.id);
        }
    }
}

struct FileHeader {
    front: u64,
}

impl Default for FileHeader {
    fn default() -> Self {
        Self {
            front: FILE_HEADER_BYTES,
        }
    }
}

impl FileHeader {
    fn load(source: &mut impl Read) -> Result<FileHeader, Error> {
        let mut magic = [0u8; 8];
        let mut version = [0u8; 2];
        let mut front = [0u8; 8];

        source.read_exact(&mut magic)?;
        source.read_exact(&mut version)?;
        source.read_exact(&mut front)?;

        if magic != MAGIC {
            return Err(Error::InvalidFile);
        }

        // If there are later versions of the file structure, we may
        // need to perform migrations here.
        if u16::from_be_bytes(version) != VERSION {
            return Err(Error::UnknownVersion);
        }

        Ok(FileHeader {
            front: u64::from_be_bytes(front),
        })
    }

    fn store(&self, target: &mut impl Write) -> Result<(), Error> {
        target.write_all(&MAGIC)?;
        target.write_all(&VERSION.to_be_bytes())?;
        target.write_all(&self.front.to_be_bytes())?;

        Ok(())
    }
}

struct RecordHeader {
    state: u8,
    length: u64,
    crc: u32,
}

impl RecordHeader {
    fn new(payload: &[u8]) -> RecordHeader {
        RecordHeader {
            state: STATE_LIVE,
            length: payload.len() as u64,
            crc: crc32fast::hash(payload),
        }
    }

    fn load(source: &mut impl Read) -> Result<RecordHeader, Error> {
        let mut state = [0u8; 1];
        let mut length = [0u8; 8];
        let mut crc = [0u8; 4];

        source.read_exact(&mut state)?;
        source.read_exact(&mut length)?;
        source.read_exact(&mut crc)?;

        Ok(RecordHeader {
            state: state[0],
            length: u64::from_be_bytes(length),
            crc: u32::from_be_bytes(crc),
        })
    }

    fn store(&self, target: &mut impl Write) -> Result<(), Error> {
        target.write_all(&[self.state])?;
        target.write_all(&self.length.to_be_bytes())?;
        target.write_all(&self.crc.to_be_bytes())?;

        Ok(())
    }

    fn total_len(&self) -> u64 {
        RECORD_HEADER_BYTES + self.length
    }
}

#[derive(Debug, Clone, Copy)]
struct RecordLocation {
    offset: u64,
    payload_len: u64,
}

impl RecordLocation {
    fn total_len(&self) -> u64 {
        RECORD_HEADER_BYTES + self.payload_len
    }

    fn end(&self) -> u64 {
        self.offset + self.total_len()
    }
}

/// The result of rewriting the live records into a fresh file.
struct Compacted {
    file: File,
    end: u64,
    read_cursor: u64,
    remap: HashMap<u64, u64>,
}

struct Inner {
    file: File,
    /// Offset one past the last valid record; appends land here.
    end: u64,
    /// Offset of the first record not known-dead on disk; persisted in
    /// the file header at sync so recovery can skip the dead prefix.
    front: u64,
    /// Offset where the scan for the next never-delivered record resumes.
    read_cursor: u64,
    /// Count of records that are neither acknowledged nor dead: this
    /// includes pending, requeued, and leased values.
    live: u64,
    next_lease_id: u64,
    /// Locations of records currently out on leases, by lease id.
    leased: HashMap<u64, RecordLocation>,
    /// Acknowledged records not yet marked dead on disk; materialized
    /// during sync so that acks never persist before a sync point.
    acked: Vec<RecordLocation>,
    /// Records whose leases were dropped, awaiting redelivery.
    requeued: VecDeque<RecordLocation>,
    /// Bytes of dead records located at or after `front`.
    dead_bytes: u64,
    /// True when the file contains writes that a sync has not yet covered.
    dirty: bool,
}

impl Inner {
    fn from_parts(file: File, end: u64, front: u64, live: u64, dead_bytes: u64) -> Inner {
        Inner {
            file,
            end,
            front,
            read_cursor: front,
            live,
            next_lease_id: 0,
            leased: HashMap::new(),
            acked: Vec::new(),
            requeued: VecDeque::new(),
            dead_bytes,
            dirty: false,
        }
    }

    /// Initializes a fresh queue file, tolerating a partial header left by
    /// a crash during an earlier initialization.
    fn create(mut file: File, existing_len: u64) -> Result<Inner, Error> {
        let mut expected = Vec::with_capacity(FILE_HEADER_BYTES as usize);
        FileHeader::default().store(&mut expected)?;

        if existing_len > 0 {
            // A file shorter than the header can only be ours if it is a
            // prefix of a freshly initialized header, meaning the process
            // died while creating it. Anything else is not a queue file.
            let mut present = vec![0u8; existing_len as usize];
            file.seek(SeekFrom::Start(0))?;
            file.read_exact(&mut present)?;

            if present != expected[..existing_len as usize] {
                return Err(Error::InvalidFile);
            }
        }

        file.seek(SeekFrom::Start(0))?;
        file.write_all(&expected)?;
        file.sync_all()?;

        Ok(Inner::from_parts(
            file,
            FILE_HEADER_BYTES,
            FILE_HEADER_BYTES,
            0,
            0,
        ))
    }

    /// Loads an existing queue file, discarding any invalid data at the
    /// tail left behind by a crash.
    fn recover(mut file: File, len: u64) -> Result<Inner, Error> {
        file.seek(SeekFrom::Start(0))?;
        let header = FileHeader::load(&mut file)?;

        if header.front < FILE_HEADER_BYTES {
            return Err(Error::InvalidFile);
        }

        // A front pointer past the end of the file can only mean the tail
        // it pointed into was lost in a crash before it fully persisted;
        // the queue content it referred to is gone, which recovery below
        // treats the same as an empty tail.
        let start = header.front.min(len);

        let mut end = len;
        let mut live = 0u64;
        let mut first_live = None;
        let mut dead_bytes = 0u64;
        let mut offset = start;
        let mut truncate = false;

        file.seek(SeekFrom::Start(start))?;

        {
            let mut reader = BufReader::new(&file);
            let mut payload = Vec::new();

            while offset < end {
                let Some(header) = Self::validate_record(&mut reader, offset, end, &mut payload)?
                else {
                    truncate = true;
                    break;
                };

                if header.state == STATE_LIVE {
                    live += 1;

                    if first_live.is_none() {
                        first_live = Some(offset);
                    }
                } else if first_live.is_some() {
                    dead_bytes += header.total_len();
                }

                offset += header.total_len();
            }
        }

        if truncate {
            file.set_len(offset)?;
            end = offset;
        }

        let front = first_live.unwrap_or(end);

        Ok(Inner::from_parts(file, end, front, live, dead_bytes))
    }

    /// Reads one record and checks that it is complete and well-formed,
    /// returning `None` if the data at `offset` cannot be trusted. Writes
    /// are sequential, so nothing after the first bad record is reliable.
    fn validate_record(
        reader: &mut impl Read,
        offset: u64,
        end: u64,
        payload: &mut Vec<u8>,
    ) -> Result<Option<RecordHeader>, Error> {
        let Some(header_end) = offset.checked_add(RECORD_HEADER_BYTES) else {
            return Ok(None);
        };

        if header_end > end {
            return Ok(None);
        }

        let header = RecordHeader::load(reader)?;

        if header.state != STATE_LIVE && header.state != STATE_DEAD {
            return Ok(None);
        }

        let Some(record_end) = header_end.checked_add(header.length) else {
            return Ok(None);
        };

        if record_end > end {
            return Ok(None);
        }

        payload.resize(header.length as usize, 0);
        reader.read_exact(payload)?;

        if crc32fast::hash(payload) != header.crc {
            return Ok(None);
        }

        Ok(Some(header))
    }

    fn append_record(&mut self, payload: &[u8]) -> Result<(), Error> {
        let header = RecordHeader::new(payload);

        // Seeking to the tracked end rather than SeekFrom::End means a
        // partial record left by an earlier failed append is overwritten.
        self.file.seek(SeekFrom::Start(self.end))?;
        header.store(&mut self.file)?;
        self.file.write_all(payload)?;

        self.end += header.total_len();
        self.live += 1;
        self.dirty = true;

        Ok(())
    }

    /// Finds the next never-delivered live record, skipping records that
    /// recovery loaded as already dead.
    fn scan_next(&mut self) -> Result<Option<RecordLocation>, Error> {
        while self.read_cursor < self.end {
            let header = self.read_record_header(self.read_cursor)?;
            let location = RecordLocation {
                offset: self.read_cursor,
                payload_len: header.length,
            };

            self.read_cursor = location.end();

            if header.state == STATE_LIVE {
                return Ok(Some(location));
            }
        }

        Ok(None)
    }

    /// The locations the next `limit` deliveries would come from,
    /// without taking any of them: redelivered records first, as `get`
    /// takes them, then the never-delivered ones ahead of the read
    /// cursor, which is left where it was.
    fn peek_locations(&mut self, limit: usize) -> Result<Vec<RecordLocation>, Error> {
        let mut found: Vec<RecordLocation> = self.requeued.iter().take(limit).copied().collect();
        let mut cursor = self.read_cursor;

        while found.len() < limit && cursor < self.end {
            let header = self.read_record_header(cursor)?;
            let location = RecordLocation {
                offset: cursor,
                payload_len: header.length,
            };

            cursor = location.end();

            if header.state == STATE_LIVE {
                found.push(location);
            }
        }

        Ok(found)
    }

    fn read_record_header(&mut self, offset: u64) -> Result<RecordHeader, Error> {
        self.file.seek(SeekFrom::Start(offset))?;

        RecordHeader::load(&mut self.file)
    }

    fn read_payload(&mut self, location: RecordLocation) -> Result<Vec<u8>, Error> {
        let mut payload = vec![0u8; location.payload_len as usize];

        self.file
            .seek(SeekFrom::Start(location.offset + RECORD_HEADER_BYTES))?;
        self.file.read_exact(&mut payload)?;

        Ok(payload)
    }

    fn mark_dead(&mut self, location: RecordLocation) -> Result<(), Error> {
        self.file.seek(SeekFrom::Start(location.offset))?;
        self.file.write_all(&[STATE_DEAD])?;

        Ok(())
    }

    /// Advances `front` past the dead records at the head of the file.
    fn advance_front(&mut self) -> Result<(), Error> {
        while self.front < self.end {
            let header = self.read_record_header(self.front)?;

            if header.state == STATE_LIVE {
                break;
            }

            self.front += header.total_len();
            self.dead_bytes = self.dead_bytes.saturating_sub(header.total_len());
        }

        Ok(())
    }

    fn write_header(&mut self) -> Result<(), Error> {
        self.file.seek(SeekFrom::Start(0))?;

        FileHeader { front: self.front }.store(&mut self.file)
    }

    fn sync(&mut self, path: &Path) -> Result<(), Error> {
        if !self.dirty {
            return Ok(());
        }

        // Materialize the buffered acknowledgements. Marking and draining
        // are separate passes so that an I/O error leaves the buffer
        // intact for an idempotent retry.
        for index in 0..self.acked.len() {
            let location = self.acked[index];
            self.mark_dead(location)?;
        }

        for location in self.acked.drain(..) {
            self.dead_bytes += location.total_len();
        }

        if self.live == 0 {
            // The queue is completely drained (live counts leased and
            // requeued values), so the file can simply be reset.
            self.file.set_len(FILE_HEADER_BYTES)?;
            self.front = FILE_HEADER_BYTES;
            self.read_cursor = FILE_HEADER_BYTES;
            self.end = FILE_HEADER_BYTES;
            self.dead_bytes = 0;
        } else {
            self.advance_front()?;
        }

        self.write_header()?;
        self.file.sync_all()?;
        self.dirty = false;

        self.compact_if_needed(path)
    }

    fn reclaimable(&self) -> u64 {
        (self.front - FILE_HEADER_BYTES) + self.dead_bytes
    }

    fn compact_if_needed(&mut self, path: &Path) -> Result<(), Error> {
        let used = self.end - FILE_HEADER_BYTES;
        let reclaimable = self.reclaimable();

        if reclaimable >= COMPACT_MIN_RECLAIMABLE && reclaimable * 2 >= used {
            self.compact(path)?;
        }

        Ok(())
    }

    /// Rewrites the live records into a fresh file and atomically renames
    /// it over the data file. Outstanding leases and requeued records are
    /// remapped to their new offsets.
    ///
    /// Only called from `sync` after acknowledgements are materialized, so
    /// dropping the dead records here never persists an ack early.
    fn compact(&mut self, path: &Path) -> Result<(), Error> {
        let tmp_path = sibling_path(path, ".compact");

        let compacted = match self.write_compacted(&tmp_path) {
            Ok(compacted) => compacted,
            Err(err) => {
                let _ = std::fs::remove_file(&tmp_path);
                return Err(err);
            }
        };

        // Renaming over the data file while we hold an open handle to it
        // is fine on POSIX systems; supporting Windows would require
        // closing the handles around the rename, confined to this spot.
        if let Err(err) = std::fs::rename(&tmp_path, path) {
            let _ = std::fs::remove_file(&tmp_path);
            return Err(err.into());
        }

        // Point of no return: the swap on disk has happened, so the
        // in-memory state must follow unconditionally.
        self.file = compacted.file;
        self.end = compacted.end;
        self.front = FILE_HEADER_BYTES;
        self.read_cursor = compacted.read_cursor;
        self.dead_bytes = 0;

        for location in self.leased.values_mut() {
            location.offset = *compacted
                .remap
                .get(&location.offset)
                .expect("leased records are live and survive compaction");
        }

        for location in self.requeued.iter_mut() {
            location.offset = *compacted
                .remap
                .get(&location.offset)
                .expect("requeued records are live and survive compaction");
        }

        sync_parent_dir(path)
    }

    fn write_compacted(&mut self, tmp_path: &Path) -> Result<Compacted, Error> {
        let mut tmp = File::options()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(tmp_path)?;

        FileHeader {
            front: FILE_HEADER_BYTES,
        }
        .store(&mut tmp)?;

        let referenced: HashSet<u64> = self
            .leased
            .values()
            .chain(self.requeued.iter())
            .map(|location| location.offset)
            .collect();

        let mut remap = HashMap::new();
        let mut new_position = FILE_HEADER_BYTES;
        // If every record before the read cursor is dead, `front` may have
        // passed it; the scan below starts at `front` and would never see
        // it, but it maps to the start of the new file.
        let mut new_read_cursor = if self.read_cursor <= self.front {
            Some(FILE_HEADER_BYTES)
        } else {
            None
        };
        let mut offset = self.front;
        let mut payload = Vec::new();

        while offset < self.end {
            if offset == self.read_cursor {
                new_read_cursor = Some(new_position);
            }

            let header = self.read_record_header(offset)?;

            if header.state == STATE_LIVE {
                payload.resize(header.length as usize, 0);
                // read_record_header leaves the file positioned at the
                // start of the payload.
                self.file.read_exact(&mut payload)?;

                if referenced.contains(&offset) {
                    remap.insert(offset, new_position);
                }

                header.store(&mut tmp)?;
                tmp.write_all(&payload)?;
                new_position += header.total_len();
            }

            offset += header.total_len();
        }

        tmp.sync_all()?;

        Ok(Compacted {
            file: tmp,
            end: new_position,
            read_cursor: new_read_cursor.unwrap_or(new_position),
            remap,
        })
    }
}

fn sibling_path(path: &Path, suffix: &str) -> PathBuf {
    let mut name = path.as_os_str().to_os_string();
    name.push(suffix);

    PathBuf::from(name)
}

#[cfg(unix)]
fn sync_parent_dir(path: &Path) -> Result<(), Error> {
    let parent = match path.parent() {
        Some(parent) if !parent.as_os_str().is_empty() => parent,
        _ => Path::new("."),
    };

    File::open(parent)?.sync_all()?;

    Ok(())
}

#[cfg(not(unix))]
fn sync_parent_dir(_path: &Path) -> Result<(), Error> {
    Ok(())
}