sprawling-storage 0.0.8

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
// Copyright (c) 2026 2youg1 and the sprawling contributors

//! Ledger appends: waves, reads, and the kernel Ledger face.

use std::path::PathBuf;

use kernel::ledger::chain_hash;
use kernel::{
    AxCode, AxError, EventDraft, EventKind, EventRecord, EventRef, Payload, RunId, Seq, TimeMs,
};

use crate::error::{StorageError, io_err};

use super::barrier::Barrier;
use super::ledger::{JsonlLedger, WriteObserver, complete_lines, is_segment, segment_file_name};

impl JsonlLedger {
    pub(crate) fn append_log_truncated(
        &mut self,
        now: TimeMs,
        dropped: u64,
    ) -> Result<(), StorageError> {
        let data = Payload::of(&kernel::event::record::LogTruncated {
            dropped_bytes: dropped,
        })
        .map_err(|source| StorageError::Draft { source })?;
        let draft = EventDraft {
            run: RunId::CITY,
            t: now,
            who: kernel::event::Who::City.to_string(),
            addr: None,
            kind: EventKind::LogTruncated,
            data,
            ig: false,
        };
        self.append_all(vec![draft]).map(|_| ())
    }

    /// Group commit: one durability barrier for the whole wave
    /// (`crates/storage/Spec.lean` §3, item 1: the batch is what the wave delivered).
    pub fn append_all(&mut self, drafts: Vec<EventDraft>) -> Result<Vec<EventRef>, StorageError> {
        // A restore a failed wave left owed runs first: it is what mends
        // the barrier, and it may remove the segment this wave would
        // append to.
        self.finish_unwind()?;
        self.barrier.admit(&self.dir, self.next_seq)?;
        self.halt.admit()?;
        if drafts.is_empty() {
            return Ok(Vec::new());
        }
        let mut seq = self.next_seq;
        let mut prev = self.prev;
        let mut cur_path = self.seg_path.clone();
        let mut cur_len = self.seg_len;
        let mut writes: Vec<(PathBuf, Vec<u8>)> = Vec::new();
        let mut created: Vec<PathBuf> = Vec::new();
        let mut records = Vec::with_capacity(drafts.len());

        if cur_len == 0 && !self.vfs.exists(&cur_path) {
            created.push(cur_path.clone());
        }

        for draft in drafts {
            let record = EventRecord::from_draft(draft, seq, prev);
            let line = record
                .canonical_line()
                .map_err(|source| StorageError::Draft { source })?;
            let line_len = u64::try_from(line.len())
                .unwrap_or(u64::MAX)
                .saturating_add(1);
            if cur_len > 0 && cur_len.saturating_add(line_len) > self.roll_bytes {
                cur_path = self.dir.join(segment_file_name(seq));
                cur_len = 0;
                created.push(cur_path.clone());
            }
            prev = chain_hash(&line);
            seq = seq
                .next()
                .map_err(|source| StorageError::Draft { source })?;
            records.push(record);
            let mut terminated = line;
            terminated.push(b'\n');
            match writes.last_mut() {
                Some((path, buffer)) if *path == cur_path => buffer.extend_from_slice(&terminated),
                _ => writes.push((cur_path.clone(), terminated)),
            }
            cur_len = cur_len.saturating_add(line_len);
        }

        self.barrier = Barrier::Broken;
        self.write_wave(&writes, created)?;

        self.barrier = Barrier::Whole;
        self.seg_path = cur_path;
        self.seg_len = cur_len;
        self.next_seq = seq;
        self.prev = prev;

        // The per-room projection is laid down only now, after the wave
        // is durable: the bytes it copies exist before it runs. A
        // refusal is reported and skipped rather than returned - the
        // history already has the record, and a disposable artifact
        // must never fail history's caller (`crates/storage/spec/Sessions.lean` §8-24).
        if let Some(sessions) = self.sessions.as_mut() {
            for record in &records {
                if let Err(error) = sessions.absorb(record) {
                    eprintln!("a session slice was refused and skipped: {error}");
                }
            }
        }

        // Only now, with the bytes synced, does anyone else hear about
        // them: an observer that saw an event the disk never got would be
        // telling the interface something the history does not contain.
        if let Some(observer) = self.observer.as_mut() {
            for record in &records {
                observer(record);
            }
        }
        Ok(records.iter().map(EventRecord::to_ref).collect())
    }

    /// Hands this ledger's session slices to whoever will file them from
    /// now on; this writer files none after it. `None` for a ledger that
    /// is not a city's, and for one that already handed them off
    /// (`crates/storage/spec/Sessions.lean` §8-24).
    pub fn hand_off_session_slices(&mut self) -> Option<crate::sessions::Sessions> {
        self.sessions.take()
    }

    /// The position a record written now would take.
    ///
    /// A position, never content: this is what anchors a diagnostic log
    /// line to the only history (`docs/logging.md` section 4), and a
    /// reader accessor that returned records would invite decision logic
    /// to consult the ledger it is in the middle of writing.
    #[must_use]
    pub fn position(&self) -> Seq {
        self.next_seq
    }

    /// Installs the write-path observer, replacing any earlier one. The
    /// sink runs on the appending thread after durability, so it must not
    /// block; a bounded send that drops on lag is the shape this expects.
    pub fn observe(&mut self, sink: WriteObserver) {
        self.observer = Some(sink);
    }

    /// The read face for replay, fixtures and inspection: every canonical
    /// line (no terminators), segments flattened in order.
    pub fn read_raw_lines(&self) -> Result<Vec<Vec<u8>>, StorageError> {
        let mut out = Vec::new();
        let segments: Vec<PathBuf> = self
            .vfs
            .list(&self.dir)
            .map_err(io_err("list ledger dir", &self.dir))?
            .into_iter()
            .filter(|p| is_segment(p))
            .collect();
        for seg in segments {
            let bytes = self.vfs.read(&seg).map_err(io_err("read segment", &seg))?;
            let (lines, _) = complete_lines(&bytes);
            for line in lines {
                if !line.is_empty() {
                    out.push(line.to_vec());
                }
            }
        }
        Ok(out)
    }

    #[cfg(test)]
    pub(crate) fn set_roll_bytes_for_test(&mut self, roll_bytes: u64) {
        self.roll_bytes = roll_bytes;
    }
}

impl kernel::Ledger for JsonlLedger {
    fn append(&mut self, draft: EventDraft) -> Result<EventRef, AxError> {
        let refs = self
            .append_all(vec![draft])
            .map_err(StorageError::into_ax)?;
        refs.into_iter().next().ok_or_else(|| {
            AxError::failure(AxCode::InvalidArgs, "append event", "empty wave echo").with_recovery(
                "report this against storage::jsonl::append with the city path: \
                         append_all owes one echo per draft and returned none",
            )
        })
    }

    /// One buffer, one write and one barrier for the whole wave.
    fn append_all(&mut self, drafts: Vec<EventDraft>) -> Result<Vec<EventRef>, AxError> {
        JsonlLedger::append_all(self, drafts).map_err(StorageError::into_ax)
    }
}

#[cfg(feature = "conformance")]
impl kernel::ledger::conformance::LedgerInspect for JsonlLedger {
    fn raw_lines(&self) -> Result<Vec<Vec<u8>>, AxError> {
        self.read_raw_lines().map_err(StorageError::into_ax)
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::indexing_slicing,
    reason = "test code"
)]
mod tests {
    use super::*;
    use kernel::GENESIS_PREV;
    use std::fs;
    use std::path::Path;
    fn draft(kind: EventKind, t: u64) -> EventDraft {
        EventDraft {
            run: RunId::CITY,
            t: TimeMs::new(t),
            who: "city".to_string(),
            addr: None,
            kind,
            data: Payload::empty(),
            ig: false,
        }
    }
    fn verify_chain(lines: &[Vec<u8>]) {
        let mut prev = GENESIS_PREV;
        for (i, line) in lines.iter().enumerate() {
            let record = EventRecord::parse_line(line).unwrap();
            assert_eq!(record.prev(), prev, "prev broken at line {i}");
            assert_eq!(record.seq(), Seq::new(u64::try_from(i).unwrap()));
            prev = chain_hash(line);
        }
    }

    #[cfg(feature = "conformance")]
    #[test]
    fn passes_the_kernel_conformance_suite() {
        use kernel::ledger::conformance::assert_ledger_conformance;
        let keep: std::cell::RefCell<Vec<tempfile::TempDir>> = std::cell::RefCell::new(Vec::new());
        assert_ledger_conformance(|| {
            let dir = tempfile::tempdir().unwrap();
            let (ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
            keep.borrow_mut().push(dir);
            ledger
        });
    }

    #[test]
    fn rolls_segments_without_breaking_seq_or_chain() {
        let dir = tempfile::tempdir().unwrap();
        let (mut ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
        ledger.set_roll_bytes_for_test(1);
        for t in 0..4 {
            ledger
                .append_all(vec![draft(EventKind::GateChecked, t)])
                .unwrap();
        }
        let segments = fs::read_dir(dir.path()).unwrap().count();
        assert!(segments >= 4, "tiny roll budget must produce many segments");
        let lines = ledger.read_raw_lines().unwrap();
        assert_eq!(lines.len(), 4);
        verify_chain(&lines);

        drop(ledger);
        let (mut reopened, report) = JsonlLedger::open(dir.path(), TimeMs::new(9)).unwrap();
        assert!(report.recovered.is_none());
        reopened
            .append_all(vec![draft(EventKind::RunFrozen, 9)])
            .unwrap();
        verify_chain(&reopened.read_raw_lines().unwrap());
    }

    /// What one production append costs the disk, counted rather than
    /// timed: `kernel::Ledger::append` is a batch of one, so every
    /// record pays a whole durability barrier. The op count is the
    /// budget reading; the wall clock belongs to the machine.
    #[test]
    fn one_append_costs_a_fixed_number_of_disk_operations() {
        let dir = tempfile::tempdir().unwrap();
        let fs = crate::fault_fs::FaultFs::new(crate::fault_fs::FaultPlan {
            cut_at_op: None,
            cut_on_write: None,
            torn_tail: crate::fault_fs::TornTail::None,
        });
        let (mut ledger, _) =
            JsonlLedger::open_faulty(fs.clone(), dir.path(), TimeMs::new(0)).unwrap();
        let opened = fs.op_count();
        kernel::Ledger::append(&mut ledger, draft(EventKind::GateChecked, 1)).unwrap();
        let first = fs.op_count().saturating_sub(opened);
        kernel::Ledger::append(&mut ledger, draft(EventKind::GateChecked, 2)).unwrap();
        let second = fs.op_count().saturating_sub(opened).saturating_sub(first);
        let bytes: u64 = ledger
            .read_raw_lines()
            .unwrap()
            .iter()
            .map(|line| u64::try_from(line.len().saturating_add(1)).unwrap())
            .sum();
        eprintln!(
            "ledger_append_ops: {first} disk ops for the first record, {second} for \n             every one after it, {bytes} B on disk for two records"
        );
        assert!(
            first > second,
            "the first record also creates the segment and syncs its directory entry"
        );
    }

    fn only_segment(dir: &Path) -> std::path::PathBuf {
        let mut files: Vec<_> = fs::read_dir(dir)
            .unwrap()
            .map(|e| e.unwrap().path())
            .collect();
        files.sort();
        assert_eq!(files.len(), 1);
        files.remove(0)
    }

    proptest::proptest! {
        #[test]
        fn any_tail_damage_recovers_to_a_valid_chain(
            n in 1usize..5,
            cut_back in 1usize..40,
            garbage in proptest::collection::vec(1u8..=255, 0..24),
        ) {
            let dir = tempfile::tempdir().unwrap();
            let (mut ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
            let drafts: Vec<_> = (0..n)
                .map(|i| draft(EventKind::GateChecked, u64::try_from(i).unwrap()))
                .collect();
            ledger.append_all(drafts).unwrap();
            let baseline = ledger.read_raw_lines().unwrap();
            let seg = only_segment(dir.path());
            drop(ledger);

            let mut bytes = fs::read(&seg).unwrap();
            let keep = bytes.len().saturating_sub(cut_back);
            bytes.truncate(keep);
            bytes.extend_from_slice(&garbage);
            fs::write(&seg, &bytes).unwrap();

            let (reopened, _) = JsonlLedger::open(dir.path(), TimeMs::new(999)).unwrap();
            let lines = reopened.read_raw_lines().unwrap();
            verify_chain(&lines);
            // Every surviving pre-damage line is a byte-exact prefix entry.
            let survivors = lines
                .iter()
                .filter(|l| {
                    EventRecord::parse_line(l).unwrap().kind() != EventKind::LogTruncated
                })
                .count();
            proptest::prop_assert!(survivors <= baseline.len());
            for (mine, original) in lines.iter().take(survivors).zip(baseline.iter()) {
                proptest::prop_assert_eq!(mine, original);
            }
        }
    }
}