graph_r/

writer.rs

// Author: spicyzhug@gmail.com
// Copyright: lituus-io, all rights reserved
//
//! The single writer. Mutations are staged as WAL ops, made durable on
//! [`Writer::commit`] (encode → append → fsync per policy), then published
//! into the current generation's overlay where new snapshots see them
//! immediately. Freshness math runs *here*, at touch time, and the computed
//! state travels inside the op — replay never re-derives policy.

use crate::error::{Error, Result};
use crate::format::base::{nflags, sflags};
use crate::format::wal::{encode_frame, Op, OwnedSeg};
use crate::key::{EdgeType, UrlKey};
use crate::store::{wal_path, Durability, Store};
use crate::ttl::Outcome;
use compact_str::CompactString;
use std::io::Write as _;
use std::sync::atomic::Ordering;
use std::sync::MutexGuard;

/// Maximum label length persisted (bytes of UTF-8).
pub const MAX_LABEL: usize = 2048;
/// Maximum segments per document.
pub const MAX_SEGS: usize = 128;
/// Maximum outbound edges per node (wide joins and conditions included).
pub use crate::format::wal::MAX_EDGES;
/// Maximum attributes per node, and the longest attribute value.
pub use crate::format::wal::{MAX_ATTRS, MAX_ATTR_LEN};
/// Most value bytes one node's attribute set may hold in total: what fits a
/// single log frame (the op's fixed fields and per-value headers included).
pub const MAX_ATTR_BYTES: usize = crate::format::wal::MAX_FRAME - (1 + 8 + 1) - MAX_ATTRS * (1 + 4);
/// Longest attribute name accepted by [`Writer::set_named_attrs`].
pub const MAX_ATTR_NAME: usize = 255;

/// A document to upsert — borrowed, produced by the caller or the link-r
/// bridge. Everything here is compact lookup metadata; bodies and vectors
/// are deliberately not representable.
#[derive(Clone, Copy, Debug)]
pub struct DocRecord<'a> {
    /// Canonical URL.
    pub url: &'a str,
    /// Durable key (xxh3 of `url`).
    pub url_key: UrlKey,
    /// xxh3 of the fetched body.
    pub content_hash: u64,
    /// Fetch stamp, epoch ms.
    pub fetched_at_ms: u64,
    /// Title, if extracted.
    pub title: Option<&'a str>,
    /// Distilled snippet (≤ ~180 chars by convention).
    pub snippet: Option<&'a str>,
    /// Entity tag from the fetch, for conditional revalidation.
    pub etag: Option<&'a str>,
    /// Exempt from eviction sweeps.
    pub pinned: bool,
}

/// A sub-document segment to record — an anchor reference, never content.
#[derive(Clone, Copy, Debug)]
pub struct SegmentRecord<'a> {
    /// Heading-path label (e.g. `"Install > Linux"`).
    pub label: &'a str,
    /// Byte range in the source document, when known.
    pub byte_range: Option<(u32, u32)>,
    /// Heading depth (1 = H1 …).
    pub depth: u8,
    /// Importance in 1/65535 units (deterministic prior or enrichment score).
    pub importance: u16,
}

/// A revalidation observation to record against a node.
#[derive(Clone, Copy, Debug)]
pub struct Touch<'a> {
    /// When the check happened, epoch ms.
    pub checked_at_ms: u64,
    /// What was observed.
    pub outcome: Outcome,
    /// New content hash when `outcome == Changed`.
    pub content_hash: Option<u64>,
    /// New entity tag, if the server sent one.
    pub etag: Option<&'a str>,
}

pub(crate) struct WriteState {
    next_seq: u64,
    staged: Vec<Op>,
    pub(crate) ops_since_compact: usize,
    pub(crate) wal_bytes: u64,
    commits_since_sync: u32,
    wal: Option<std::fs::File>,
    /// Generation of the newest base written to disk. A compaction whose
    /// in-memory install was refused (every ring slot pinned) still leaves a
    /// durable base behind; numbering from this, not from the served
    /// snapshot, means a generation number is never reused.
    pub(crate) disk_generation: u64,
}

#[allow(clippy::missing_fields_in_debug)] // file handle and op buffer are noise
impl std::fmt::Debug for WriteState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WriteState")
            .field("next_seq", &self.next_seq)
            .field("staged", &self.staged.len())
            .finish()
    }
}

impl WriteState {
    pub(crate) fn new(
        next_seq: u64,
        replayed_ops: usize,
        wal_bytes: u64,
        disk_generation: u64,
    ) -> Self {
        Self {
            next_seq,
            staged: Vec::new(),
            ops_since_compact: replayed_ops,
            wal_bytes,
            commits_since_sync: 0,
            wal: None,
            disk_generation,
        }
    }

    pub(crate) fn reset_after_compact(&mut self, next_seq: u64) {
        self.next_seq = next_seq;
        self.ops_since_compact = 0;
        self.wal_bytes = u64::try_from(crate::format::wal::HEADER_LEN).unwrap_or(64);
        self.commits_since_sync = 0;
        self.wal = None; // reopen against the fresh log on next commit
    }

    pub(crate) fn staged_is_empty(&self) -> bool {
        self.staged.is_empty()
    }

    pub(crate) fn last_committed_seq(&self) -> u64 {
        self.next_seq.saturating_sub(1)
    }
}

/// The exclusive write handle; holds the store's writer mutex for its
/// lifetime. Stage mutations, then [`Writer::commit`] to make them durable
/// and visible.
#[derive(Debug)]
pub struct Writer<'s> {
    store: &'s Store,
    state: MutexGuard<'s, WriteState>,
    /// Committed-state view, cached for the duration of one staging batch.
    ///
    /// Several staging methods read committed state (segment identity for
    /// enrichment carry, the other tier's edges, freshness counters, node
    /// existence). Taking a fresh snapshot per staged op folds the overlay
    /// arena every time — during a bulk ingest that made staging O(batch ×
    /// pending-ops). Nothing can commit while this writer holds the mutex, so
    /// one snapshot is exactly as current as a fresh one for the whole batch;
    /// [`Writer::commit`] drops it, both because the world changes there and so
    /// the pinned generation slot is free before any auto-compaction installs.
    snap: Option<crate::snapshot::Snapshot<'s>>,
}

fn check_label(name: &str, s: &str) -> Result<()> {
    if s.len() > MAX_LABEL {
        return Err(Error::format(format!("{name} label exceeds {MAX_LABEL} bytes")));
    }
    Ok(())
}

impl<'s> Writer<'s> {
    pub(crate) fn new(store: &'s Store, state: MutexGuard<'s, WriteState>) -> Self {
        Self { store, state, snap: None }
    }

    /// The cached committed-state view (created on first use per batch).
    fn view(&mut self) -> &crate::snapshot::Snapshot<'s> {
        if self.snap.is_none() {
            self.snap = Some(self.store.snapshot());
        }
        self.snap.as_ref().expect("just filled")
    }

    /// Stage a node create/replace.
    pub fn upsert_node(&mut self, doc: &DocRecord<'_>) -> Result<()> {
        check_label("url", doc.url)?;
        if doc.url.is_empty() {
            return Err(Error::format("url must not be empty"));
        }
        for (name, v) in [("title", doc.title), ("snippet", doc.snippet), ("etag", doc.etag)] {
            if let Some(v) = v {
                check_label(name, v)?;
            }
        }
        let flags = if doc.pinned { nflags::PINNED } else { 0 };
        self.state.staged.push(Op::UpsertNode {
            key: doc.url_key,
            content_hash: doc.content_hash,
            fetched_at_ms: doc.fetched_at_ms,
            flags,
            url: doc.url.into(),
            title: doc.title.map(Into::into),
            snippet: doc.snippet.map(Into::into),
            etag: doc.etag.map(Into::into),
        });
        Ok(())
    }

    /// Stage a whole-set segment replacement for `key`.
    pub fn set_segments(&mut self, key: UrlKey, segs: &[SegmentRecord<'_>]) -> Result<()> {
        if segs.len() > MAX_SEGS {
            return Err(Error::format(format!("more than {MAX_SEGS} segments")));
        }
        let mut owned = Vec::with_capacity(segs.len());
        for s in segs {
            check_label("segment", s.label)?;
            let (importance, extra_flags) = (s.importance, 0);
            let (byte_start, byte_len, range_flag) = match s.byte_range {
                Some((start, len)) => (start, len, 0),
                None => (0, 0, sflags::NO_RANGE),
            };
            owned.push(OwnedSeg {
                label: s.label.into(),
                byte_start,
                byte_len,
                depth: s.depth.max(1),
                importance,
                flags: range_flag | extra_flags,
            });
        }
        self.state.staged.push(Op::SetSegments { key, segs: owned });
        Ok(())
    }

    /// Stage a whole-set outbound-edge replacement for `key`. Edges are
    /// deduplicated by destination (first occurrence wins — document order is
    /// salience order) and stored sorted by destination key.
    pub fn set_edges(&mut self, key: UrlKey, edges: &[(UrlKey, EdgeType, u16)]) {
        self.stage_edges(key, edges.iter().map(|&(d, t, w)| (d, t, w, 0)));
    }

    /// [`Writer::set_edges`] with per-edge flags; only caller-settable ones
    /// are kept (`eflags::EXACT`: an unchanged copy along `Derives`).
    pub fn set_edges_with_flags(&mut self, key: UrlKey, edges: &[(UrlKey, EdgeType, u16, u8)]) {
        let mask = crate::format::base::eflags::EXACT;
        self.stage_edges(key, edges.iter().map(|&(d, t, w, f)| (d, t, w, f & mask)));
    }

    fn stage_edges(
        &mut self,
        key: UrlKey,
        edges: impl Iterator<Item = (UrlKey, EdgeType, u16, u8)>,
    ) {
        let mut seen = std::collections::HashSet::new();
        let mut owned: Vec<(UrlKey, u8, u8, u16)> = Vec::new();
        for (dst, etype, weight, mut flags) in edges {
            if dst == key || !seen.insert(dst.0) {
                continue;
            }
            if matches!(etype, EdgeType::Related) {
                flags |= crate::format::base::eflags::INFERRED;
            }
            owned.push((dst, etype.as_tag(), flags, weight));
            if owned.len() == MAX_EDGES {
                break;
            }
        }
        owned.sort_unstable_by_key(|(dst, _, _, _)| dst.0);
        self.state.staged.push(Op::SetEdges { key, edges: owned });
    }

    /// Stage a whole-set replacement of `key`'s attributes: small keyed
    /// values beside the node (a column's type, a table's producing SQL, a
    /// description's provenance). Keys are the caller's; a later key wins
    /// over an earlier duplicate. Values longer than [`MAX_ATTR_LEN`], more
    /// than [`MAX_ATTRS`] keys, or a set whose encoding would not fit one log
    /// frame ([`MAX_ATTR_BYTES`] in total) are refused.
    pub fn set_attrs(&mut self, key: UrlKey, attrs: &[(u8, &str)]) -> Result<()> {
        if attrs.len() > MAX_ATTRS {
            return Err(Error::format("too many attributes"));
        }
        let mut owned: Vec<(u8, CompactString)> = Vec::with_capacity(attrs.len());
        for &(k, v) in attrs.iter().rev() {
            if v.len() > MAX_ATTR_LEN {
                return Err(Error::format("attribute too long"));
            }
            if !owned.iter().any(|(o, _)| *o == k) {
                owned.push((k, CompactString::from(v)));
            }
        }
        // Replay reads an oversized frame as a torn tail and a writable open
        // truncates there, so an op that does not fit one frame would lose
        // itself and every later commit on reopen. Bound the whole op.
        if owned.iter().map(|(_, v)| v.len()).sum::<usize>() > MAX_ATTR_BYTES {
            return Err(Error::format(format!(
                "attributes exceed {MAX_ATTR_BYTES} bytes in total"
            )));
        }
        owned.sort_unstable_by_key(|(k, _)| *k);
        self.state.staged.push(Op::SetAttrs { key, attrs: owned });
        Ok(())
    }

    /// [`Writer::set_attrs`] with string names: a whole-set replacement of
    /// `key`'s attributes as `name → value`. Each record is stored as
    /// `name NUL value` under keys `0..n` in name order, so
    /// [`crate::Snapshot::named_attrs`] reads names and values back as
    /// borrowed slices and any producer (Rust or a binding) agrees on the
    /// encoding. A later duplicate name wins. Names must be non-empty, at
    /// most [`MAX_ATTR_NAME`] bytes and free of NUL; each `name NUL value`
    /// must fit [`MAX_ATTR_LEN`], and the whole set [`MAX_ATTR_BYTES`].
    pub fn set_named_attrs(&mut self, key: UrlKey, attrs: &[(&str, &str)]) -> Result<()> {
        if attrs.len() > MAX_ATTRS {
            return Err(Error::format(format!("more than {MAX_ATTRS} attributes")));
        }
        let mut named: Vec<(&str, &str)> = Vec::with_capacity(attrs.len());
        for &(name, value) in attrs.iter().rev() {
            if name.is_empty() || name.len() > MAX_ATTR_NAME || name.contains('\0') {
                return Err(Error::format(format!(
                    "attribute name must be 1..={MAX_ATTR_NAME} bytes without NUL"
                )));
            }
            if name.len() + 1 + value.len() > MAX_ATTR_LEN {
                return Err(Error::format(format!(
                    "attribute {name:?} exceeds {MAX_ATTR_LEN} bytes"
                )));
            }
            if !named.iter().any(|(n, _)| *n == name) {
                named.push((name, value));
            }
        }
        named.sort_unstable_by_key(|(n, _)| *n);
        let mut owned: Vec<(u8, CompactString)> = Vec::with_capacity(named.len());
        let mut total = 0usize;
        for (k, (name, value)) in named.into_iter().enumerate() {
            let mut rec = CompactString::with_capacity(name.len() + 1 + value.len());
            rec.push_str(name);
            rec.push('\0');
            rec.push_str(value);
            total += rec.len();
            owned.push((k as u8, rec));
        }
        if total > MAX_ATTR_BYTES {
            return Err(Error::format(format!(
                "attributes exceed {MAX_ATTR_BYTES} bytes in total"
            )));
        }
        self.state.staged.push(Op::SetAttrs { key, attrs: owned });
        Ok(())
    }

    /// Record a revalidation outcome. Freshness policy (interval growth/cut,
    /// counters) is computed here from the node's current state and the store's
    /// [`crate::TtlConfig`], and the resulting op carries the node's *full*
    /// effective state so replay never re-derives policy.
    ///
    /// Returns `false`, staging nothing, when `key` is not a **committed** node.
    /// That is the important word: the state is read through a snapshot, which
    /// sees committed ops only, so a node staged earlier in this same batch is
    /// not yet visible. Commit the upsert before touching it — the link-r bridge
    /// already does.
    ///
    /// The same rule means two touches of one key inside a single uncommitted
    /// batch do not accumulate: both are computed from the pre-batch state and
    /// the second wins under last-wins overlay folding. Commit between
    /// observations of the same document. Pinned by `touch_requires_a_committed_node`
    /// in `tests/regression.rs`.
    pub fn touch(&mut self, key: UrlKey, t: Touch<'_>) -> Result<bool> {
        if let Some(e) = t.etag {
            check_label("etag", e)?;
        }
        let staged = self.state.staged.iter().rev().find_map(|op| match op {
            Op::UpsertNode { key: k, content_hash, etag, .. } if *k == key => {
                Some((*content_hash, etag.clone()))
            }
            _ => None,
        });
        let (interval_s, checks, changes, last_change_ms, content_hash, etag) = {
            let snap = self.view();
            let Some(node) = snap.node(key) else { return Ok(false) };
            let cfg = snap.ttl();
            let next = cfg.next_interval(node.interval_s.max(1), t.outcome);
            let checks = node.checks.saturating_add(1);
            let changes = if t.outcome == Outcome::Changed {
                node.changes.saturating_add(1)
            } else {
                node.changes
            };
            let last_change =
                if t.outcome == Outcome::Changed { t.checked_at_ms } else { node.last_change_ms };
            // The op carries the node's *full* effective state, not a delta:
            // touches fold last-wins in the overlay, so an unchanged check
            // must not erase the hash/etag a prior change observed — nor the
            // ones an upsert staged earlier in this same batch.
            let (base_hash, base_etag) = match staged {
                Some((h, e)) => (h, e),
                None => (node.content_hash, node.etag.map(Into::into)),
            };
            let hash = t.content_hash.unwrap_or(base_hash);
            let etag: Option<compact_str::CompactString> = t.etag.map(Into::into).or(base_etag);
            (next, checks, changes, last_change, hash, etag)
        };
        self.state.staged.push(Op::Touch {
            key,
            checked_at_ms: t.checked_at_ms,
            outcome: t.outcome.as_tag(),
            content_hash: Some(content_hash),
            etag,
            interval_s,
            checks,
            changes,
            last_change_ms,
            tombstone: t.outcome == Outcome::Gone,
        });
        Ok(true)
    }

    /// Stage a pin/unpin. Returns `false` (staging nothing) when `key` is
    /// neither committed nor upserted earlier in this same batch — a pin on a
    /// document that does not exist must not lie in wait for a future upsert.
    pub fn set_pinned(&mut self, key: UrlKey, pinned: bool) -> Result<bool> {
        let exists = self
            .state
            .staged
            .iter()
            .any(|op| matches!(op, Op::UpsertNode { key: k, .. } if *k == key))
            || self.view().node(key).is_some();
        if !exists {
            return Ok(false);
        }
        self.state.staged.push(Op::SetPinned { key, pinned });
        Ok(true)
    }

    /// Stage a hard delete.
    pub fn remove(&mut self, key: UrlKey) -> Result<()> {
        self.state.staged.push(Op::Remove { key });
        Ok(())
    }

    /// Make staged ops durable and visible: encode frames, append to the
    /// WAL, fsync per the configured [`Durability`], publish to the overlay,
    /// then auto-compact past thresholds. Returns the last committed
    /// sequence number.
    pub fn commit(&mut self) -> Result<u64> {
        // The cached view is about to go stale, and its pinned generation slot
        // must be free before any auto-compaction below installs a new one.
        self.snap = None;
        if self.state.staged.is_empty() {
            return Ok(self.state.next_seq.saturating_sub(1));
        }
        let staged = std::mem::take(&mut self.state.staged);
        let count = staged.len() as u64;
        let first_seq = self.state.next_seq;
        let mut buf = Vec::new();
        for (i, op) in staged.iter().enumerate() {
            buf.extend_from_slice(&encode_frame(first_seq + i as u64, op));
        }
        // Append + flush + fsync-per-policy.
        if self.state.wal.is_none() {
            self.state.wal =
                Some(std::fs::OpenOptions::new().append(true).open(wal_path(&self.store.dir))?);
        }
        {
            let state = &mut *self.state;
            let f = state.wal.as_mut().expect("opened above");
            f.write_all(&buf)?;
            f.flush()?;
            let sync = match self.store.cfg.durability {
                Durability::Always => true,
                Durability::Batch(n) => {
                    state.commits_since_sync += 1;
                    state.commits_since_sync >= n.max(1)
                }
                Durability::Never => false,
            };
            if sync {
                f.sync_all()?;
                state.commits_since_sync = 0;
            }
        }
        // Publish into the current generation (stable while we hold the
        // writer mutex — only the writer compacts).
        let cur = self.store.current.load(Ordering::Acquire);
        {
            let guard = self.store.gens[cur]
                .inner
                .try_read()
                .map_err(|_| Error::busy("current slot mid-install"))?;
            let inner = guard.as_ref().ok_or_else(|| Error::corrupt("current slot empty"))?;
            for (i, op) in staged.into_iter().enumerate() {
                if !inner.arena.publish(first_seq + i as u64, op) {
                    return Err(Error::busy("overlay arena full"));
                }
            }
        }
        self.state.wal_bytes += buf.len() as u64;
        self.state.next_seq += count;
        self.state.ops_since_compact += count as usize;
        let last = self.state.next_seq - 1;
        if self.state.ops_since_compact >= self.store.cfg.compact_after_ops
            || self.state.wal_bytes >= self.store.cfg.compact_after_wal_bytes
        {
            if self.store.compact_inline.load(Ordering::Relaxed) {
                crate::compact::compact_locked(self.store, &mut self.state)?;
            } else {
                // Off the hot path: the owner compacts when it chooses.
                self.store.due.store(true, Ordering::Release);
            }
        }
        Ok(last)
    }
}