use std::fs::{File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::{Mutex, MutexGuard, PoisonError};
use super::record::{parse_entry, LogEntry};
use crate::errors::HaysonLogReplacedError;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Slot {
pub offset: u64,
pub length: u64,
}
pub struct ScannedEntry {
pub entry: LogEntry,
pub slot: Slot,
}
pub struct ScanResult {
pub entries: Vec<ScannedEntry>,
pub end_offset: u64,
pub truncated_bytes: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LogProbe {
pub size: u64,
pub inode: Option<u64>,
}
const NEWLINE: u8 = b'\n';
fn open_at(path: &Path, readonly: bool) -> io::Result<File> {
if readonly {
OpenOptions::new().read(true).open(path)
} else {
OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
}
}
#[cfg(unix)]
fn inode_of_file(file: &File) -> Option<u64> {
use std::os::unix::fs::MetadataExt;
file.metadata().ok().map(|metadata| metadata.ino())
}
#[cfg(unix)]
fn inode_at(path: &Path) -> Option<u64> {
use std::os::unix::fs::MetadataExt;
std::fs::metadata(path).ok().map(|metadata| metadata.ino())
}
#[cfg(target_os = "wasi")]
fn inode_of_file(file: &File) -> Option<u64> {
use std::os::fd::AsRawFd;
let stat = unsafe { wasi::fd_filestat_get(file.as_raw_fd() as wasi::Fd) }.ok()?;
Some(stat.ino).filter(|&ino| ino != 0)
}
#[cfg(target_os = "wasi")]
fn inode_at(path: &Path) -> Option<u64> {
File::open(path).ok().and_then(|file| inode_of_file(&file))
}
#[cfg(not(any(unix, target_os = "wasi")))]
fn inode_of_file(_file: &File) -> Option<u64> {
None
}
#[cfg(not(any(unix, target_os = "wasi")))]
fn inode_at(_path: &Path) -> Option<u64> {
None
}
pub struct LogFile {
path: PathBuf,
file: Mutex<File>,
size: u64,
inode: Option<u64>,
readonly: bool,
fsync: bool,
#[cfg(test)]
fsyncs: std::sync::atomic::AtomicU64,
}
impl LogFile {
pub fn open(path: impl AsRef<Path>, readonly: bool) -> io::Result<Self> {
let path = path.as_ref().to_path_buf();
let file = open_at(&path, readonly)?;
let metadata = file.metadata()?;
Ok(LogFile {
size: metadata.len(),
inode: inode_of_file(&file),
path,
file: Mutex::new(file),
readonly,
fsync: true,
#[cfg(test)]
fsyncs: std::sync::atomic::AtomicU64::new(0),
})
}
pub fn set_fsync(&mut self, fsync: bool) {
self.fsync = fsync;
}
fn handle(&self) -> MutexGuard<'_, File> {
self.file.lock().unwrap_or_else(PoisonError::into_inner)
}
fn read_exact_at(&self, buffer: &mut [u8], offset: u64) -> io::Result<()> {
let mut file = self.handle();
file.seek(SeekFrom::Start(offset))?;
file.read_exact(buffer)
}
pub fn size(&self) -> u64 {
self.size
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn replaced_by(&self, probe: &LogProbe) -> bool {
match (self.inode, probe.inode) {
(Some(held), Some(found)) => held != found,
_ => probe.size < self.size,
}
}
fn refuse_if_replaced(&self) -> io::Result<()> {
let replaced = match std::fs::metadata(&self.path) {
Ok(info) => self.replaced_by(&LogProbe {
size: info.len(),
inode: inode_at(&self.path),
}),
Err(error) if error.kind() == io::ErrorKind::NotFound => true,
Err(error) => return Err(error),
};
if replaced {
return Err(io::Error::other(HaysonLogReplacedError {
path: self.path.display().to_string(),
}));
}
Ok(())
}
pub fn adopt_size(&mut self, size: u64) {
self.size = size;
}
pub fn probe(&self) -> Option<LogProbe> {
let info = std::fs::metadata(&self.path).ok()?;
Some(LogProbe {
size: info.len(),
inode: inode_at(&self.path),
})
}
pub fn append(&mut self, lines: &[String]) -> io::Result<Vec<Slot>> {
if self.readonly {
return Err(io::Error::other("append on a readonly log"));
}
if lines.is_empty() {
return Ok(Vec::new());
}
self.refuse_if_replaced()?;
let payload = format!("{}\n", lines.join("\n"));
{
let mut file = self.handle();
file.seek(SeekFrom::Start(self.size))?;
file.write_all(payload.as_bytes())?;
}
let mut slots = Vec::with_capacity(lines.len());
let mut cursor = self.size;
for line in lines {
let length = line.len() as u64;
slots.push(Slot { offset: cursor, length });
cursor += length + 1;
}
self.size = cursor;
Ok(slots)
}
pub fn read_slice(&self, slot: &Slot) -> io::Result<String> {
let mut buffer = vec![0u8; slot.length as usize];
self.read_exact_at(&mut buffer, slot.offset)?;
String::from_utf8(buffer).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
}
pub fn read_bytes(&self, offset: u64, length: u64) -> io::Result<Vec<u8>> {
let mut buffer = vec![0u8; length as usize];
self.read_exact_at(&mut buffer, offset)?;
Ok(buffer)
}
pub fn scan(&self, from_offset: u64) -> io::Result<ScanResult> {
let mut entries = Vec::new();
if from_offset >= self.size {
return Ok(ScanResult {
entries,
end_offset: from_offset,
truncated_bytes: 0,
});
}
let mut buffer = vec![0u8; (self.size - from_offset) as usize];
self.read_exact_at(&mut buffer, from_offset)?;
let mut line_start = from_offset;
let mut search_from = 0usize;
while let Some(rel) = memchr(NEWLINE, &buffer[search_from..]) {
let break_at = search_from + rel;
let raw = &buffer[search_from..break_at];
let length = raw.len() as u64;
if length > 0 {
let line = std::str::from_utf8(raw).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
let entry = parse_entry(line, self.path.to_string_lossy().as_ref(), line_start)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;
entries.push(ScannedEntry {
entry,
slot: Slot {
offset: line_start,
length,
},
});
}
line_start += length + 1;
search_from = break_at + 1;
}
Ok(ScanResult {
entries,
end_offset: line_start,
truncated_bytes: (buffer.len() - search_from) as u64,
})
}
pub fn truncate(&mut self, size: u64) -> io::Result<()> {
self.handle().set_len(size)?;
self.size = size;
Ok(())
}
pub fn discard_partial_tail(&mut self, scan: &ScanResult) -> io::Result<()> {
if scan.truncated_bytes == 0 {
return Ok(());
}
if self.readonly {
self.adopt_size(scan.end_offset);
return Ok(());
}
self.truncate(scan.end_offset)?;
self.sync()
}
pub fn sync(&self) -> io::Result<()> {
if self.readonly || !self.fsync {
return Ok(());
}
self.handle().sync_data()?;
#[cfg(test)]
self.fsyncs.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
#[cfg(test)]
pub fn fsyncs(&self) -> u64 {
self.fsyncs.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn reopen(&mut self) -> io::Result<()> {
let file = open_at(&self.path, self.readonly)?;
self.size = file.metadata()?.len();
self.inode = inode_of_file(&file);
*self.handle() = file;
Ok(())
}
}
fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
haystack.iter().position(|&b| b == needle)
}
#[cfg(test)]
mod tests {
use std::fs;
use std::io::Write;
use super::super::record::{serialize_entry, LogMeta, LogOp};
use super::*;
use crate::schema::HaysonRecord;
fn line(key: &str, seq: u64, name: &str) -> String {
let mut data = HaysonRecord::new();
data.insert("name".into(), serde_json::Value::String(name.into()));
serialize_entry(
&LogMeta {
key: key.into(),
op: LogOp::Put,
seq,
ts: 1,
},
Some(&data),
)
}
fn log_replaced_of(error: &io::Error) -> Option<&HaysonLogReplacedError> {
error.get_ref()?.downcast_ref::<HaysonLogReplacedError>()
}
#[cfg(unix)]
#[test]
fn refuses_to_append_once_the_file_at_its_path_is_a_different_one() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania")]).unwrap();
let swapped = dir.path().join("users.jsonl.swap");
fs::copy(&path, &swapped).unwrap();
fs::rename(&swapped, &path).unwrap();
let error = log.append(&[line("b", 2, "Piotr")]).unwrap_err();
let replaced = log_replaced_of(&error).expect("a typed refusal, not a plain io error");
assert_eq!(replaced.path, path.display().to_string());
assert_eq!(
fs::read_to_string(&path).unwrap().lines().count(),
1,
"nothing landed in the orphaned inode either"
);
log.reopen().unwrap();
log.append(&[line("b", 2, "Piotr")]).unwrap();
assert_eq!(fs::read_to_string(&path).unwrap().lines().count(), 2);
}
#[test]
fn refuses_to_append_once_the_file_is_gone_from_its_path() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania")]).unwrap();
fs::remove_file(&path).unwrap();
let error = log.append(&[line("b", 2, "Piotr")]).unwrap_err();
assert!(log_replaced_of(&error).is_some(), "{error}");
assert!(!path.exists(), "the refusal did not recreate the file");
}
#[test]
fn refuses_to_append_once_the_file_at_its_path_is_shorter_than_what_it_wrote() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania"), line("b", 2, "Piotr")]).unwrap();
let swapped = dir.path().join("users.jsonl.swap");
fs::write(&swapped, format!("{}\n", line("a", 1, "Ania"))).unwrap();
fs::rename(&swapped, &path).unwrap();
let error = log.append(&[line("c", 3, "Zosia")]).unwrap_err();
assert!(log_replaced_of(&error).is_some(), "{error}");
}
#[test]
fn returns_byte_offsets_that_survive_multibyte_text() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
let first = line("a", 1, "Łódź");
let second = line("b", 2, "Ania");
let slots = log.append(&[first.clone(), second.clone()]).unwrap();
assert_eq!(slots[0].offset, 0);
assert_eq!(slots[0].length, first.len() as u64);
assert_eq!(slots[1].offset, first.len() as u64 + 1);
assert_eq!(log.read_slice(&slots[0]).unwrap(), first);
assert_eq!(log.read_slice(&slots[1]).unwrap(), second);
}
#[test]
fn reconstructs_state_after_reopening_the_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
{
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania"), line("b", 2, "Piotr")]).unwrap();
log.sync().unwrap();
}
let mut reopened = LogFile::open(&path, false).unwrap();
assert_eq!(reopened.size(), fs::metadata(&path).unwrap().len());
let scan = reopened.scan(0).unwrap();
assert_eq!(scan.entries.len(), 2);
assert_eq!(scan.truncated_bytes, 0);
assert_eq!(scan.entries[0].entry.meta.key, "a");
assert_eq!(scan.entries[1].entry.meta.key, "b");
assert_eq!(
scan.entries[0].entry.data.as_ref().unwrap().get("name"),
Some(&serde_json::Value::String("Ania".into()))
);
reopened.append(&[line("c", 3, "Zosia")]).unwrap();
reopened.reopen().unwrap();
let after_reopen = reopened.scan(0).unwrap();
assert_eq!(after_reopen.entries.len(), 3);
assert_eq!(after_reopen.entries[2].entry.meta.key, "c");
}
#[test]
fn reads_to_the_last_full_line_when_the_tail_is_truncated_mid_write() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
{
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania")]).unwrap();
}
{
let mut file = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
file.write_all(br#"{"_k":"b","_op":"put","_seq":2,"_ts"#).unwrap();
}
let log = LogFile::open(&path, false).unwrap();
let scan = log.scan(0).unwrap();
assert_eq!(scan.entries.len(), 1);
assert_eq!(scan.entries[0].entry.meta.key, "a");
assert!(scan.truncated_bytes > 0);
assert_eq!(scan.end_offset, scan.entries[0].slot.length + 1);
}
fn torn_log(path: &Path) -> u64 {
let mut log = LogFile::open(path, false).unwrap();
let healthy = log.append(&[line("a", 1, "Ania")]).unwrap()[0].length + 1;
let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap();
file.write_all(br#"{"_k":"b","_op":"put","_seq":2,"_ts"#).unwrap();
healthy
}
#[test]
fn a_writable_handle_cuts_the_torn_tail_so_the_next_append_lands_on_its_own_line() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let healthy = torn_log(&path);
let mut log = LogFile::open(&path, false).unwrap();
let scan = log.scan(0).unwrap();
log.discard_partial_tail(&scan).unwrap();
assert_eq!(log.size(), healthy);
assert_eq!(fs::metadata(&path).unwrap().len(), healthy);
log.append(&[line("c", 3, "Zosia")]).unwrap();
let after = LogFile::open(&path, false).unwrap().scan(0).unwrap();
assert_eq!(after.truncated_bytes, 0);
assert_eq!(after.entries.len(), 2);
assert_eq!(after.entries[1].entry.meta.key, "c");
}
#[test]
fn a_readonly_handle_forgets_the_torn_tail_without_touching_the_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let healthy = torn_log(&path);
let torn_size = fs::metadata(&path).unwrap().len();
let mut log = LogFile::open(&path, true).unwrap();
let scan = log.scan(0).unwrap();
log.discard_partial_tail(&scan).unwrap();
assert_eq!(log.size(), healthy);
assert_eq!(fs::metadata(&path).unwrap().len(), torn_size);
}
#[test]
fn sync_never_issues_no_fsync_while_the_default_commit_still_does() {
let dir = tempfile::tempdir().unwrap();
let payload = [line("a", 1, "Ania"), line("b", 2, "Piotr")];
let quiet_path = dir.path().join("quiet.jsonl");
let mut quiet = LogFile::open(&quiet_path, false).unwrap();
quiet.set_fsync(false);
quiet.append(&payload).unwrap();
quiet.sync().unwrap();
quiet.sync().unwrap();
let loud_path = dir.path().join("loud.jsonl");
let mut loud = LogFile::open(&loud_path, false).unwrap();
loud.append(&payload).unwrap();
loud.sync().unwrap();
loud.sync().unwrap();
assert_eq!(quiet.fsyncs(), 0, "sync on a never-fsync log must not reach the disk");
assert_eq!(loud.fsyncs(), 2, "the default log must still fsync once per sync");
assert_eq!(fs::read(&quiet_path).unwrap(), fs::read(&loud_path).unwrap());
}
#[test]
fn a_never_fsync_log_still_reads_back_everything_it_was_given() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.set_fsync(false);
let first = line("a", 1, "Ania");
let slots = log.append(&[first.clone(), line("b", 2, "Piotr")]).unwrap();
log.sync().unwrap();
assert_eq!(log.read_slice(&slots[0]).unwrap(), first);
assert_eq!(log.size(), fs::metadata(&path).unwrap().len());
let scan = LogFile::open(&path, false).unwrap().scan(0).unwrap();
assert_eq!(scan.entries.len(), 2);
assert_eq!(scan.truncated_bytes, 0);
}
#[test]
fn cutting_a_torn_tail_off_a_never_fsync_log_still_cuts_it_without_an_fsync() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let healthy = torn_log(&path);
let mut log = LogFile::open(&path, false).unwrap();
log.set_fsync(false);
let scan = log.scan(0).unwrap();
log.discard_partial_tail(&scan).unwrap();
assert_eq!(
fs::metadata(&path).unwrap().len(),
healthy,
"the tail is cut whatever sync says"
);
assert_eq!(log.fsyncs(), 0);
}
#[test]
fn reopening_keeps_the_durability_the_handle_was_configured_with() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.set_fsync(false);
log.append(&[line("a", 1, "Ania")]).unwrap();
log.reopen().unwrap();
log.append(&[line("b", 2, "Piotr")]).unwrap();
log.sync().unwrap();
assert_eq!(log.fsyncs(), 0, "reopen must not silently rearm fsync");
}
#[test]
fn probe_reflects_size_and_inode_from_the_path_not_the_handle() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
log.append(&[line("a", 1, "Ania")]).unwrap();
let probe = log.probe().unwrap();
assert_eq!(probe.size, log.size());
#[cfg(unix)]
assert!(probe.inode.is_some());
}
#[test]
fn concurrent_readers_sharing_one_handle_each_get_their_own_offset() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
let written: Vec<String> = (0..64)
.map(|i| line(&format!("k{i}"), i, &format!("name-{i}")))
.collect();
let slots = log.append(&written).unwrap();
let log = &log;
let slots = &slots;
let written = &written;
std::thread::scope(|scope| {
for lane in 0..8usize {
scope.spawn(move || {
for round in 0..50 {
let i = (lane * 7 + round * 13) % slots.len();
assert_eq!(log.read_slice(&slots[i]).unwrap(), written[i]);
let raw = log.read_bytes(slots[i].offset, slots[i].length).unwrap();
assert_eq!(raw, written[i].as_bytes());
}
});
}
});
}
#[test]
fn a_read_between_two_appends_does_not_disturb_where_the_next_append_lands() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
let first = line("a", 1, "Ania");
let first_slots = log.append(std::slice::from_ref(&first)).unwrap();
assert_eq!(log.read_slice(&first_slots[0]).unwrap(), first);
let second = line("b", 2, "Piotr");
let second_slots = log.append(std::slice::from_ref(&second)).unwrap();
assert_eq!(second_slots[0].offset, first.len() as u64 + 1);
assert_eq!(log.read_slice(&second_slots[0]).unwrap(), second);
assert_eq!(log.size(), fs::metadata(&path).unwrap().len());
let scan = log.scan(0).unwrap();
assert_eq!(scan.entries.len(), 2);
assert_eq!(scan.truncated_bytes, 0);
}
#[test]
fn a_scan_after_a_truncate_reads_only_what_is_left() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut log = LogFile::open(&path, false).unwrap();
let first = line("a", 1, "Ania");
log.append(&[first.clone(), line("b", 2, "Piotr")]).unwrap();
log.truncate(first.len() as u64 + 1).unwrap();
let scan = log.scan(0).unwrap();
assert_eq!(scan.entries.len(), 1);
assert_eq!(scan.entries[0].entry.meta.key, "a");
let third = line("c", 3, "Zosia");
let slots = log.append(std::slice::from_ref(&third)).unwrap();
assert_eq!(slots[0].offset, first.len() as u64 + 1);
assert_eq!(log.read_slice(&slots[0]).unwrap(), third);
}
}