car_sync/

journal.rs

//! Durable JSONL persistence for the oplog — the `car-eventlog` journal
//! idiom: append-only, one record per line, torn-line tolerant on load.
//!
//! A crash mid-write leaves at most one torn (unparseable) trailing line;
//! [`OplogJournal::load`] skips it (and any blank lines) instead of failing,
//! exactly like `EventLog::load`. Nothing is lost from the sync's point of
//! view: an op that never fully reached the journal was never acked, and the
//! oplog's convergence is defined over the op-*set* — re-appending it (or
//! re-pulling it from a peer/relay in B3) folds to the same state.
//! Integrity/order checking is deliberately NOT done here — it is the
//! explicit, separate [`crate::oplog::verify_log`] pass.
//!
//! **Single writer, enforced.** A journal is one device's log: two writers
//! on one path would fork the `seq` chain and can interleave bytes
//! mid-record. [`OplogJournal::open`] therefore takes an exclusive advisory
//! lock on `<path>.lock` (held for the journal's lifetime; the OS releases
//! it on drop) and fails with `WouldBlock` if another holder exists — the
//! same protocol `car-registry`'s supervisor uses for `agents.json.lock`.
//! [`OplogJournal::load`] is read-only and takes no lock.
//!
//! **Truncated journals are marked and fenced (B4).**
//! [`OplogJournal::truncate_to`] stamps a [`TruncationMarker`] as the new
//! file's first line — atomic with the truncation itself (same rename, no
//! crash window in between). The marker names the covering checkpoint's
//! content address, and it exists to make a permanent-fork hazard a
//! **runtime error instead of a documentation footnote**: a device whose
//! ops were ALL below the frontier leaves no trace of itself in the
//! retained tail, so `DeviceLog::resume` over that tail would silently
//! restart it at `seq 0` — an unrecoverable duplicate-seq chain fork.
//! Therefore [`OplogJournal::load`] **refuses** a marked journal (use
//! [`OplogJournal::load_with_marker`], fetch the named checkpoint, and go
//! through [`crate::checkpoint::resume_anchored`] /
//! [`crate::checkpoint::verify_anchored`]), and `DeviceLog::resume` itself
//! rejects an own-chain non-zero start as a second fence.

use crate::oplog::OpRecord;
use serde::{Deserialize, Serialize};
use std::fs::{self, File, OpenOptions, TryLockError};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};

/// The first line of a truncated journal: names the checkpoint (by
/// whole-record content address) that accounts for everything the
/// truncation dropped. Written atomically WITH the truncation by
/// [`OplogJournal::truncate_to`]; surfaced by
/// [`OplogJournal::load_with_marker`]; fences [`OplogJournal::load`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TruncationMarker {
    /// [`crate::checkpoint::Checkpoint::checkpoint_hash`] of the covering
    /// checkpoint — fetch it and resume via `resume_anchored`.
    pub checkpoint_hash: String,
}

/// The on-disk line wrapper — a shape no [`OpRecord`] can collide with.
#[derive(Debug, Serialize, Deserialize)]
struct MarkerLine {
    truncation_marker: TruncationMarker,
}

/// Append-only JSONL journal for [`OpRecord`]s. Holds an exclusive advisory
/// lock on `<path>.lock` for its lifetime — one writer per journal path.
#[derive(Debug)]
pub struct OplogJournal {
    path: PathBuf,
    writer: BufWriter<File>,
    /// Advisory lock handle: its existence + the exclusive lock are the
    /// entire protocol (never written; intentionally not unlinked on drop —
    /// unlink-on-drop races a new acquirer creating the file first).
    _lock: File,
    /// Test-only fault seam: when `Some(n)`, the `n`-th subsequent
    /// [`OplogJournal::append`] fails (as a real ENOSPC-class error would),
    /// exercising the writer-recreate + caller-rollback recovery paths that
    /// no portable API can trigger deterministically.
    #[cfg(test)]
    pub(crate) fail_append_after: Option<usize>,
}

impl OplogJournal {
    /// Open (creating parents and the file if needed) for appending.
    /// Existing content is preserved — append mode, never truncate.
    ///
    /// If the file's last line is torn (a crash mid-write left it without a
    /// terminating newline), a newline is written first so the next append
    /// starts a fresh line instead of gluing itself onto the garbage —
    /// otherwise the first post-crash append would be lost with the tail.
    pub fn open(path: &Path) -> std::io::Result<Self> {
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                fs::create_dir_all(parent)?;
            }
        }
        // Exclusive advisory lock BEFORE touching the journal — a second
        // writer on this path would fork the seq chain and interleave bytes
        // mid-record (the car-registry supervisor lock pattern).
        let lock_path = {
            let mut s = path.as_os_str().to_owned();
            s.push(".lock");
            PathBuf::from(s)
        };
        let lock = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&lock_path)?;
        match lock.try_lock() {
            Ok(()) => {}
            Err(TryLockError::WouldBlock) => {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::WouldBlock,
                    format!(
                        "oplog journal already open by another writer (advisory lock held on {})",
                        lock_path.display()
                    ),
                ));
            }
            Err(TryLockError::Error(e)) => return Err(e),
        }
        let needs_newline = match File::open(path) {
            Ok(mut existing) => {
                use std::io::{Read, Seek, SeekFrom};
                if existing.metadata()?.len() == 0 {
                    false
                } else {
                    existing.seek(SeekFrom::End(-1))?;
                    let mut last = [0u8; 1];
                    existing.read_exact(&mut last)?;
                    last[0] != b'\n'
                }
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
            Err(e) => return Err(e),
        };
        let file = OpenOptions::new().create(true).append(true).open(path)?;
        let mut writer = BufWriter::new(file);
        if needs_newline {
            writer.write_all(b"\n")?;
            writer.flush()?;
        }
        Ok(Self {
            path: path.to_path_buf(),
            writer,
            _lock: lock,
            #[cfg(test)]
            fail_append_after: None,
        })
    }

    /// Append one op as a single JSON line and flush it to the OS page
    /// cache, so a subsequent crash can tear at most the *next* record.
    ///
    /// **Not a durability barrier** — `flush` clears the `BufWriter` buffer
    /// to the OS but does not `fsync`. A caller with a
    /// journal-durable-before-transmit / ack-asserts-durable contract
    /// (the sync session) MUST call [`OplogJournal::sync`] before it
    /// transmits or acks; per-op fsync would be needless latency (one batch
    /// barrier at the contract point is enough).
    ///
    /// **Failure recovers the writer.** On a write/flush error the
    /// `BufWriter` would otherwise *retain* the unflushed line; a later
    /// append would then emit that rolled-back line beside the caller's
    /// re-minted same-`seq` op — a permanent chain fork. So a failed append
    /// **recreates the writer** on a fresh append-mode handle, discarding
    /// the poisoned buffer, before returning the error (the caller rolls
    /// its in-memory chain back to the durable ops). Bytes that already
    /// reached the file are at most one torn trailing line, which
    /// [`OplogJournal::load`] tolerates.
    pub fn append(&mut self, op: &OpRecord) -> std::io::Result<()> {
        #[cfg(test)]
        if let Some(n) = self.fail_append_after {
            if n == 0 {
                self.fail_append_after = None;
                self.reopen_writer()?; // same recovery a real failure takes
                return Err(std::io::Error::other("injected append failure"));
            }
            self.fail_append_after = Some(n - 1);
        }
        let line = serde_json::to_string(op).map_err(std::io::Error::other)?;
        match self
            .writer
            .write_all(line.as_bytes())
            .and_then(|()| self.writer.write_all(b"\n"))
            .and_then(|()| self.writer.flush())
        {
            Ok(()) => Ok(()),
            Err(e) => {
                // Discard the poisoned buffer; surface the reopen error only
                // if even that fails (then the journal is truly unusable).
                match self.reopen_writer() {
                    Ok(()) => Err(e),
                    Err(reopen_err) => Err(reopen_err),
                }
            }
        }
    }

    /// Durability barrier: flush the buffer and `fsync` the journal file to
    /// stable storage. The sync session crosses this before it transmits an
    /// op (journal-durable-before-transmit, B1 MUST) and before it acks a
    /// fold frontier (ack-asserts-durable, B4 MUST) — one batch call, not
    /// per-op.
    pub fn sync(&mut self) -> std::io::Result<()> {
        self.writer.flush()?;
        self.writer.get_ref().sync_all()
    }

    /// Recreate the buffered writer on a fresh append-mode handle, dropping
    /// any bytes buffered (but not yet flushed) in the current one.
    fn reopen_writer(&mut self) -> std::io::Result<()> {
        let file = OpenOptions::new().create(true).append(true).open(&self.path)?;
        self.writer = BufWriter::new(file);
        Ok(())
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Rewrite the journal to exactly `ops`, stamped with a
    /// [`TruncationMarker`] naming `checkpoint_hash` as its first line —
    /// B4's truncation-below-a-frontier step, executed under the advisory
    /// lock this open journal already holds (no second writer can
    /// interleave). The marker travels IN the rewritten file, so it is
    /// atomic with the truncation: there is no crash window in which the
    /// journal is truncated but unmarked (or marked but untruncated).
    ///
    /// Durability: the retained ops are written to a sibling temp file,
    /// fsync'd, and atomically renamed over the journal, so a crash at any
    /// point leaves either the old complete journal or the new complete
    /// tail — never a partial rewrite. **Crash-ordering invariant (callers
    /// MUST honor it; [`crate::compact::compact_and_truncate`] does by
    /// construction): the covering checkpoint is durable BEFORE this runs.**
    /// Truncation makes the dropped ops unrecoverable from the journal; the
    /// checkpoint named by the marker is what still accounts for them.
    pub fn truncate_to(&mut self, ops: &[OpRecord], checkpoint_hash: &str) -> std::io::Result<()> {
        let tmp_path = {
            let mut s = self.path.as_os_str().to_owned();
            s.push(".compact.tmp");
            PathBuf::from(s)
        };
        {
            let mut tmp = BufWriter::new(File::create(&tmp_path)?);
            let marker = serde_json::to_string(&MarkerLine {
                truncation_marker: TruncationMarker {
                    checkpoint_hash: checkpoint_hash.to_string(),
                },
            })
            .map_err(std::io::Error::other)?;
            tmp.write_all(marker.as_bytes())?;
            tmp.write_all(b"\n")?;
            for op in ops {
                let line = serde_json::to_string(op).map_err(std::io::Error::other)?;
                tmp.write_all(line.as_bytes())?;
                tmp.write_all(b"\n")?;
            }
            tmp.flush()?;
            tmp.get_ref().sync_all()?;
        }
        fs::rename(&tmp_path, &self.path)?;
        // The old writer handle points at the renamed-over inode; reopen on
        // the new file so subsequent appends land in the truncated journal.
        let file = OpenOptions::new().create(true).append(true).open(&self.path)?;
        self.writer = BufWriter::new(file);
        // Make the rename durable where the platform allows it.
        #[cfg(unix)]
        if let Some(parent) = self.path.parent() {
            if !parent.as_os_str().is_empty() {
                let _ = File::open(parent).and_then(|d| d.sync_all());
            }
        }
        Ok(())
    }

    /// Load a **full** (never-truncated) journal: every parseable op, in
    /// file order. Blank and unparseable (torn) lines are skipped; a
    /// missing file is an empty log (a fresh device bootstraps from
    /// nothing).
    ///
    /// A journal carrying a [`TruncationMarker`] is **refused** with a
    /// runtime error: its ops are only the retained tail, and treating them
    /// as the whole log silently re-mints truncated seqs on
    /// `DeviceLog::resume` — the permanent chain fork. Use
    /// [`OplogJournal::load_with_marker`] + the named checkpoint +
    /// [`crate::checkpoint::resume_anchored`] instead.
    pub fn load(path: &Path) -> std::io::Result<Vec<OpRecord>> {
        let (marker, ops) = Self::load_with_marker(path)?;
        if let Some(marker) = marker {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!(
                    "journal {} was truncated below checkpoint {} — its ops are a retained \
                     tail, not the full log; load_with_marker + resume_anchored required",
                    path.display(),
                    marker.checkpoint_hash
                ),
            ));
        }
        Ok(ops)
    }

    /// Load a journal that may have been truncated: the
    /// [`TruncationMarker`] (if any) plus every parseable op, in file
    /// order. Blank and unparseable (torn) lines are skipped; a missing
    /// file is `(None, [])`. When the marker is `Some`, the ops are a
    /// retained tail — anchor them on the named checkpoint
    /// ([`crate::checkpoint::verify_anchored`]) and resume via
    /// [`crate::checkpoint::resume_anchored`], never `DeviceLog::resume`.
    pub fn load_with_marker(
        path: &Path,
    ) -> std::io::Result<(Option<TruncationMarker>, Vec<OpRecord>)> {
        let file = match File::open(path) {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((None, Vec::new())),
            Err(e) => return Err(e),
        };
        let reader = BufReader::new(file);
        let mut marker = None;
        let mut ops = Vec::new();
        for line in reader.lines() {
            let line = line?;
            let line = line.trim();
            if line.is_empty() {
                continue;
            }
            if let Ok(op) = serde_json::from_str::<OpRecord>(line) {
                ops.push(op);
            } else if let Ok(found) = serde_json::from_str::<MarkerLine>(line) {
                marker.get_or_insert(found.truncation_marker);
            }
        }
        Ok((marker, ops))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fold::{fold, state_hash};
    use crate::oplog::{verify_log, DeviceLog, Scope, Surface};
    use serde_json::json;

    fn sample_ops(n: usize) -> Vec<OpRecord> {
        let mut log = DeviceLog::new("d1");
        (0..n)
            .map(|i| {
                log.append(
                    Scope::Personal,
                    Surface::Knowledge,
                    json!({"id": format!("f{i}"), "n": i}),
                )
            })
            .collect()
    }

    #[test]
    fn append_then_load_round_trips() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested").join("oplog.jsonl");
        let ops = sample_ops(3);
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            for op in &ops {
                journal.append(op).unwrap();
            }
        }
        let loaded = OplogJournal::load(&path).unwrap();
        assert_eq!(loaded, ops);
        verify_log(&loaded).unwrap();
    }

    #[test]
    fn reopen_appends_without_truncating() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let ops = sample_ops(4);
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            journal.append(&ops[0]).unwrap();
            journal.append(&ops[1]).unwrap();
        }
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            journal.append(&ops[2]).unwrap();
            journal.append(&ops[3]).unwrap();
        }
        assert_eq!(OplogJournal::load(&path).unwrap(), ops);
    }

    #[test]
    fn torn_tail_and_blank_lines_are_tolerated() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let ops = sample_ops(3);
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            for op in &ops {
                journal.append(op).unwrap();
            }
        }
        // Simulate a crash mid-append: a torn, unterminated final line,
        // plus a stray blank line.
        let mut raw = fs::read_to_string(&path).unwrap();
        raw.push('\n');
        raw.push_str(r#"{"op_id":"op-torn","hlc":{"wall_ms":9,"count"#);
        fs::write(&path, raw).unwrap();

        let loaded = OplogJournal::load(&path).unwrap();
        assert_eq!(loaded, ops, "torn tail is skipped, intact records survive");
        verify_log(&loaded).unwrap();
        assert_eq!(state_hash(&fold(&loaded)), state_hash(&fold(&ops)));

        // The device resumes its chain from the loaded log and re-appends
        // the lost write — convergence is over the op-set, nothing breaks.
        let mut resumed = DeviceLog::resume("d1", &loaded).unwrap();
        let recovered = resumed.append(Scope::Personal, Surface::Knowledge, json!({"id": "f-re"}));
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            journal.append(&recovered).unwrap();
        }
        let reloaded = OplogJournal::load(&path).unwrap();
        assert_eq!(reloaded.len(), 4);
        verify_log(&reloaded).unwrap();
    }

    #[test]
    fn second_writer_on_same_path_is_rejected_until_first_drops() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let first = OplogJournal::open(&path).unwrap();
        let second = OplogJournal::open(&path);
        assert!(second.is_err(), "advisory lock must reject a second writer");
        assert_eq!(second.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);
        drop(first);
        // Lock released on drop — reopening succeeds.
        OplogJournal::open(&path).unwrap();
    }

    #[test]
    fn truncate_to_rewrites_atomically_marks_and_fences_resume() {
        use crate::checkpoint::{resume_anchored, Checkpoint};

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let ops = sample_ops(5);
        let mut journal = OplogJournal::open(&path).unwrap();
        for op in &ops {
            journal.append(op).unwrap();
        }
        // Truncate to the tail (ops 3..5) while the journal stays open,
        // stamping the covering checkpoint's address into the marker.
        let ckpt = Checkpoint::from_ops(&ops[..3]).unwrap();
        journal.truncate_to(&ops[3..], &ckpt.checkpoint_hash).unwrap();
        assert!(!dir.path().join("oplog.jsonl.compact.tmp").exists(), "no temp residue");

        // The fences (kernel-review item): a truncated journal is a RUNTIME
        // error on the naive path, not a documentation footnote.
        let err = OplogJournal::load(&path).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        let (marker, tail) = OplogJournal::load_with_marker(&path).unwrap();
        assert_eq!(marker.unwrap().checkpoint_hash, ckpt.checkpoint_hash);
        assert_eq!(tail, ops[3..].to_vec());
        // …and DeviceLog::resume refuses the tail even if handed the ops
        // directly (own chain starts past seq 0).
        assert!(matches!(
            DeviceLog::resume("d1", &tail),
            Err(crate::oplog::ChainError::TruncatedChain { first_seq: 3, .. })
        ));

        // The sanctioned path: resume anchored on the checkpoint. Appends
        // after truncation land in the NEW file (writer reopened) and the
        // marker survives them.
        let mut resumed = resume_anchored("d1", &ckpt, &tail).unwrap();
        let next = resumed.append(Scope::Personal, Surface::Knowledge, json!({"id": "f-post"}));
        journal.append(&next).unwrap();
        let (marker, reloaded) = OplogJournal::load_with_marker(&path).unwrap();
        assert!(marker.is_some(), "marker survives post-truncation appends");
        assert_eq!(reloaded.len(), 3);
        assert_eq!(reloaded[2], next);
        verify_log(&reloaded).expect("truncated chain + new append verifies (non-zero seq start)");
    }

    #[test]
    fn missing_file_loads_empty() {
        let dir = tempfile::tempdir().unwrap();
        let loaded = OplogJournal::load(&dir.path().join("absent.jsonl")).unwrap();
        assert!(loaded.is_empty());
    }

    #[test]
    fn interleaved_multi_device_appends_fold_identically_to_memory() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let mut journal = OplogJournal::open(&path).unwrap();
        let mut ops = Vec::new();
        for i in 0..3 {
            let oa = a.append(Scope::Personal, Surface::Knowledge, json!({"id": format!("a{i}")}));
            b.observe(&oa.hlc);
            let ob = b.append(
                Scope::Personal,
                Surface::Declagent,
                json!({"id": "shared", "turn": i}),
            );
            a.observe(&ob.hlc);
            journal.append(&oa).unwrap();
            journal.append(&ob).unwrap();
            ops.push(oa);
            ops.push(ob);
        }
        let loaded = OplogJournal::load(&path).unwrap();
        verify_log(&loaded).unwrap();
        assert_eq!(fold(&loaded), fold(&ops));
        // The LWW registry converged to the last turn.
        let state = fold(&loaded);
        assert_eq!(
            state.registries[&Surface::Declagent.tag()]["id:shared"].payload["turn"],
            json!(2)
        );
    }
}