lib.rs
raw
//! 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(())
}