hayson-engine 0.25.0

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,
    /// `None` where the host reports no inode: identity then rests on size alone, which detects
    /// growth and truncation but not a same-size replacement of the file behind the path.
    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())
}

// NOTE: std's wasi MetadataExt is unstable; the raw `fd_filestat_get` carries the same `ino`.
// NOTE: a wasi host that fills no inode reports 0 for every file — that is no identity at all.
#[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
}

/// Append-only log file for one table (or one partition of one), matching the TS `LogFile`
/// contract: every read and every append names its own absolute offset, so no caller can be
/// disturbed by another's position in the file. The descriptor itself does carry a cursor —
/// stable Rust has no portable positional I/O — so the seek and the transfer that follows it are
/// held together under one mutex, which is what makes the offsets behave as if they were.
pub struct LogFile {
    path: PathBuf,
    file: Mutex<File>,
    size: u64,
    inode: Option<u64>,
    readonly: bool,
    fsync: bool,
    /// How many `fsync`s this handle actually issued. Test-only: the `sync: "never"` contract is
    /// "no `fsync` reaches the disk", and no portable observable outside the process can show the
    /// absence of one — `/proc/self/io` counts the whole process, not this handle.
    #[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),
        })
    }

    /// Whether [`LogFile::sync`] reaches the disk at all. On (the default) is the catalog's
    /// `sync: "commit"`; off is `sync: "never"`, which hands durability to the kernel and makes
    /// every `sync` a no-op. `table::Table::set_sync` translates the catalog mode into this flag —
    /// `storage/` sits below `catalog/` and does not name its types.
    pub fn set_fsync(&mut self, fsync: bool) {
        self.fsync = fsync;
    }

    // NOTE: a poisoned guard is recovered, not propagated — a panic mid-read leaves no torn state
    // here, only a cursor the next seek overwrites anyway.
    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
    }

    /// Whether the file `probe` found at this handle's path is no longer the one it holds open: a
    /// different inode (an atomic rewrite — how `compaction::compact_log` lands — or a file swapped
    /// in from outside, which is what a `git checkout` of a tracked data directory does), or bytes
    /// gone from a file this engine only ever appends to. Without an inode on either side only the
    /// shrink is visible, so a same-size or longer replacement passes as "nothing happened" there.
    pub fn replaced_by(&self, probe: &LogProbe) -> bool {
        match (self.inode, probe.inode) {
            (Some(held), Some(found)) => held != found,
            _ => probe.size < self.size,
        }
    }

    /// An append onto a file that was unlinked or replaced under this handle would land in the
    /// orphaned inode and vanish with it, while every read from memory kept showing the record —
    /// silent data loss. So every append first checks that the path still leads to the open file.
    /// A snapshot and a compaction ask the same question before they start, for the same reason:
    /// the one would describe the orphaned inode, the other would overwrite the file now at the
    /// path with that inode's contents.
    pub(crate) 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(())
    }

    /// Adopts a size this handle did not write itself — the tracked size another process's append
    /// left behind, or the end of the last complete line after a scan met a half-written tail.
    /// Unlike [`LogFile::truncate`] it touches nothing on disk: it only moves where this handle
    /// believes the file ends, which is where its next scan starts and its next append lands.
    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))
    }

    // NOTE: raw bytes, not read_slice — an arbitrary byte window can split a multi-byte char.
    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)
    }

    /// Scans from `from_offset` to the current tracked size, yielding one entry per full line.
    /// An unterminated tail is reported via `truncated_bytes`, never parsed.
    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))?;
                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(())
    }

    /// Gets rid of the half-written tail `scan` met past its last complete line — the bytes a
    /// killed writer or a full disk left behind. Nothing to do when the scan found none.
    ///
    /// A writable handle cuts them off the file itself: without that, `size` stays at the physical
    /// end and the next [`LogFile::append`] lands straight after an unterminated line, gluing the
    /// two into a single line no later scan can parse — a rebuild without a snapshot then loses the
    /// whole table, not just the write that was interrupted.
    ///
    /// A readonly handle owns no right to change the file, so it only moves its own end marker
    /// ([`LogFile::adopt_size`]); the tail stays on disk until a writable handle opens the log.
    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>()
    }

    /// Same bytes, new inode — what `git checkout` does to a tracked file. Only the inode can tell,
    /// so the test needs a host that reports one.
    #[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);
    }

    /// One complete line plus the unterminated start of a second, and where the complete one ends.
    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 {
                        // NOTE: each lane walks the slots in its own order, so a shared cursor
                        // leaking between threads would show up as a mismatched line here.
                        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);
    }
}