kevy_persist/

lib.rs

//! kevy-persist — durability for a [`kevy_store::Store`].
//!
//! Two mechanisms, both zero-dependency pure Rust over `std::fs`:
//!
//! - **Snapshot (RDB-style):** [`save_snapshot`] dumps a whole store to a temp
//!   file then atomically renames it (fsync before rename); [`load_snapshot`]
//!   restores it. A compact, type-tagged binary format.
//! - **AOF:** an [`Aof`] append-only command log with a configurable fsync
//!   policy; [`replay_aof`] re-applies it on startup, tolerating a truncated
//!   trailing frame from a crash mid-write.
//!
//! In a shared-nothing runtime each shard persists its own store to its own
//! file, so there is no cross-core coordination. Part of the [kevy] server.
//!
//! [kevy]: https://crates.io/crates/kevy
//!
//! # Example (AOF)
//!
//! ```
//! use kevy_persist::{Aof, Argv, Fsync, replay_aof};
//!
//! # fn main() -> std::io::Result<()> {
//! let path = std::env::temp_dir().join("kevy-persist-doctest.aof");
//! # let _ = std::fs::remove_file(&path);
//! {
//!     let mut aof = Aof::open(&path, Fsync::No)?;
//!     aof.append(&Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]))?;
//! } // flushed on drop
//!
//! let mut replayed: Vec<Argv> = Vec::new();
//! replay_aof(&path, |args| replayed.push(args))?;
//! assert_eq!(replayed, vec![vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]]);
//! # std::fs::remove_file(&path).ok();
//! # Ok(())
//! # }
//! ```
#![forbid(unsafe_code)]

mod aof;
pub mod feed_meta;
pub mod layout;
mod replay;
pub mod reshard;
mod rewrite_fmt;
mod shards_meta;
mod snapshot_payload;

pub use aof::{Aof, Fsync, RewritePlan, RewriteStats, write_aof_base};
pub use replay::replay_aof;
pub use shards_meta::{Routing, ShardsMeta, read_shards_meta, write_shards_meta};
pub use kevy_resp::{Argv, ArgvView};
pub use rewrite_fmt::dump_aof;
pub(crate) use rewrite_fmt::{dump_store_to_buf, estimate_multibulk_bytes, write_multibulk};
use kevy_store::Store;
use kevy_store::Value;
// ZSet snapshot iterates ordered (member, score) pairs via `Value::ZSet`.
use std::fs::File;
use std::io::{self, BufReader, BufWriter, Read, Write};
use std::path::Path;

/// File magic + format version. Bump `VERSION` on any layout change.
///
/// v2 stored each entry's TTL as **remaining millis** (relative), so a load
/// re-anchored the deadline to load-time — a restart reset every key to a
/// fresh full TTL (INC-2026-06-09). v3 stores the **absolute** Unix-ms
/// deadline, so a load reconstructs the original instant. v4 appends a
/// consumer-group section to each `OP_STREAM` payload (groups + consumers
/// plus PEL) — before that, SAVE/reshard silently dropped group state. The
/// loader still accepts v2 (relative TTL) and v3 (no group section).
const MAGIC: &[u8; 8] = b"KEVYSNAP";
const VERSION: u8 = 4;
/// v2.3: version 5 carries a 16-byte feed cursor (`gen u64 LE` +
/// `offset u64 LE`) right after the version byte — the snapshot half
/// of the recovery-point contract (docs/cdc.md): snapshot S + feed
/// frames from S's cursor = exact restore. Writers emit v5 only when
/// a cursor is supplied; cursor-less writes stay at v4 so every
/// existing path is byte-identical.
const VERSION_FEED_CURSOR: u8 = 5;
/// v2.4: version 6 additionally carries `OP_HFTTL` hash field-TTL
/// records after the entry stream. Written only when field TTLs
/// exist; the header still carries the (possibly zero) feed cursor.
const VERSION_HASH_TTL: u8 = 6;
const VERSION_RELATIVE_TTL: u8 = 2;
const VERSION_ABSOLUTE_TTL: u8 = 3;

// Record opcodes (one per value type). Each record is:
//   [op][ttl: u8 flag + optional u64][key][type payload]
const OP_EOF: u8 = 0;
const OP_STR: u8 = 1;
const OP_HASH: u8 = 2;
const OP_LIST: u8 = 3;
const OP_SET: u8 = 4;
const OP_ZSET: u8 = 5;
const OP_STREAM: u8 = 6;
/// v2.4 hash field TTL record: `[key][field][deadline_ms: u64 LE]`.
/// Appears only in format v6+ snapshots, after the entry stream's
/// records (before OP_EOF).
const OP_HFTTL: u8 = 7;

/// BufWriter capacity for bulk snapshot / AOF-rewrite writes. The 8 KiB
/// default made SAVE ~12 % of disk bandwidth (tens of thousands of small
/// `write(2)`s); 1 MiB amortizes the syscalls toward disk speed.
pub(crate) const SNAPSHOT_BUF_CAP: usize = 1 << 20;

/// Anything that can enumerate `(key, &Value, ttl_ms)` triples for
/// serialization: a live [`Store`] (its `snapshot_each`, the synchronous
/// paths) or a frozen [`kevy_store::SnapshotView`] (the COW paths — collect
/// on the owning thread, serialize on a background one).
pub trait SnapshotSource {
    /// Visit every live entry as `(key, &value, remaining_ttl_ms)`.
    fn for_each_entry(&self, f: impl FnMut(&[u8], &Value, Option<u64>));

    /// v2.4: visit every live hash field TTL as `(key, field,
    /// absolute_unix_ms)`. Default = none (sources predating the
    /// feature).
    fn for_each_hash_ttl(&self, _f: impl FnMut(&[u8], &[u8], u64)) {}
}

impl SnapshotSource for Store {
    fn for_each_entry(&self, f: impl FnMut(&[u8], &Value, Option<u64>)) {
        self.snapshot_each(f);
    }
    fn for_each_hash_ttl(&self, f: impl FnMut(&[u8], &[u8], u64)) {
        self.hash_ttl_each(f);
    }
}

impl SnapshotSource for kevy_store::SnapshotView {
    fn for_each_entry(&self, f: impl FnMut(&[u8], &Value, Option<u64>)) {
        self.each(f);
    }
    fn for_each_hash_ttl(&self, f: impl FnMut(&[u8], &[u8], u64)) {
        self.each_hash_ttl(f);
    }
}

/// Write a point-in-time snapshot of `src` (a live [`Store`] or a frozen
/// [`kevy_store::SnapshotView`]) to `path`, atomically: data is written to
/// `<path>.tmp`, fsynced, then renamed over `path`.
pub fn save_snapshot<S: SnapshotSource>(src: &S, path: &Path) -> io::Result<()> {
    let tmp = write_snapshot_tmp(src, path)?;
    std::fs::rename(&tmp, path)
}

/// Serialize a point-in-time snapshot of `src` into any `Write` sink.
/// Used by both [`write_snapshot_tmp`] (sink = `BufWriter<File>` +
/// extra fsync after) and the v3-cluster replication path (sink =
/// `&mut Vec<u8>` for in-memory snapshot ship, see
/// `kevy-replicate/docs/snapshot.md`).
///
/// On-disk bytes are identical regardless of sink — the same magic +
/// version header, same entry stream, same `OP_EOF` trailer. Callers
/// that need durability (disk) wrap in `BufWriter<File>` and call
/// `sync_all` themselves; callers that need bytes (network ship)
/// pass a `Vec<u8>`.
pub fn write_snapshot_to<S: SnapshotSource, W: Write>(src: &S, sink: &mut W) -> io::Result<()> {
    write_snapshot_to_with_cursor(src, sink, None)
}

/// [`write_snapshot_to`] with the v2.3 recovery-point header: when
/// `cursor = Some((generation, offset))` the snapshot records the feed
/// position it was taken at (format v5); `None` writes the legacy v4
/// stream unchanged.
pub fn write_snapshot_to_with_cursor<S: SnapshotSource, W: Write>(
    src: &S,
    sink: &mut W,
    cursor: Option<(u64, u64)>,
) -> io::Result<()> {
    // v2.4: field-TTL records force format v6; collect them first so
    // the header version is known before anything is written.
    let mut fttl: Vec<(Vec<u8>, Vec<u8>, u64)> = Vec::new();
    src.for_each_hash_ttl(|k, f, d| fttl.push((k.to_vec(), f.to_vec(), d)));
    let mut w = BufWriter::with_capacity(SNAPSHOT_BUF_CAP, sink);
    w.write_all(MAGIC)?;
    let version = if !fttl.is_empty() {
        VERSION_HASH_TTL
    } else if cursor.is_some() {
        VERSION_FEED_CURSOR
    } else {
        VERSION
    };
    w.write_all(&[version])?;
    if version >= VERSION_FEED_CURSOR {
        let (generation, offset) = cursor.unwrap_or((0, 0));
        w.write_all(&generation.to_le_bytes())?;
        w.write_all(&offset.to_le_bytes())?;
    }
    // The source yields *remaining* ms; v3 persists the absolute
    // Unix-ms deadline (now + remaining) so the TTL survives a restart.
    let now = kevy_store::now_unix_ms();
    // Enumeration is infallible; capture the first write error to surface.
    let mut err: Option<io::Error> = None;
    src.for_each_entry(|key, value, ttl| {
        let deadline = ttl.map(|ms| now.saturating_add(ms));
        if err.is_none()
            && let Err(e) = write_entry(&mut w, key, value, deadline)
        {
            err = Some(e);
        }
    });
    if let Some(e) = err {
        return Err(e);
    }
    for (k, f, d) in &fttl {
        w.write_all(&[OP_HFTTL])?;
        write_bytes(&mut w, k)?;
        write_bytes(&mut w, f)?;
        w.write_all(&d.to_le_bytes())?;
    }
    w.write_all(&[OP_EOF])?;
    w.flush()?;
    Ok(())
}

/// The write half of [`save_snapshot`]: produce the durable (fsynced)
/// `<path>.tmp` and return its path **without** the final rename. For the
/// COW background-save flow: the serializer thread writes the temp file at
/// leisure, then the store-owning thread renames it in the same critical
/// section that resets the AOF — keeping the snapshot/AOF commit adjacent
/// instead of seconds apart.
pub fn write_snapshot_tmp<S: SnapshotSource>(src: &S, path: &Path) -> io::Result<std::path::PathBuf> {
    let tmp = tmp_path(path);
    {
        let mut file = File::create(&tmp)?;
        write_snapshot_to(src, &mut file)?;
        file.sync_all()?; // durably on disk before the rename
    }
    Ok(tmp)
}

/// Read the v2.3 recovery-point cursor from a snapshot's header:
/// `Some((generation, offset))` for format v5+, `None` for older
/// (cursor-less) snapshots. Does not load entries.
pub fn read_snapshot_cursor(path: &Path) -> io::Result<Option<(u64, u64)>> {
    let mut r = BufReader::new(File::open(path)?);
    let mut magic = [0u8; 8];
    r.read_exact(&mut magic)?;
    if &magic != MAGIC {
        return Err(io::Error::new(io::ErrorKind::InvalidData, "kevy snapshot: bad magic"));
    }
    let version = read_u8(&mut r)?;
    if version < VERSION_FEED_CURSOR {
        return Ok(None);
    }
    let mut cur = [0u8; 16];
    r.read_exact(&mut cur)?;
    let generation = u64::from_le_bytes(cur[..8].try_into().expect("8 bytes"));
    let offset = u64::from_le_bytes(cur[8..].try_into().expect("8 bytes"));
    Ok(Some((generation, offset)))
}

/// Load a snapshot from `path` into `store` (entries are inserted, not cleared
/// first — call on a fresh store). Errors on a bad magic/version or truncation.
pub fn load_snapshot(store: &mut Store, path: &Path) -> io::Result<()> {
    let r = BufReader::new(File::open(path)?);
    load_snapshot_from(store, r)
}

/// Load a snapshot from any [`std::io::Read`] sink into `store` —
/// symmetric to [`write_snapshot_to`]. Used by the v3-cluster
/// replication path (sink = `&[u8]` wrapped in `std::io::Cursor`) to
/// apply a primary-shipped snapshot to a fresh local store without
/// touching disk. Entries are inserted, not cleared first — call on
/// a fresh store. Errors on bad magic/version or truncation.
pub fn load_snapshot_from<R: Read>(store: &mut Store, r: R) -> io::Result<()> {
    load_snapshot_filtered(store, r, |_| true)
}

/// v3.2 — [`load_snapshot_from`] with a key predicate: only records
/// whose key satisfies `keep` are loaded (skipped records are still
/// parsed to stay in frame). The single-source replica path broadcasts
/// one snapshot payload to every shard and each loads its own hash
/// slice — no intermediate store, no re-serialization.
pub fn load_snapshot_filtered<R: Read>(
    store: &mut Store,
    mut r: R,
    keep: impl Fn(&[u8]) -> bool,
) -> io::Result<()> {
    let mut magic = [0u8; 8];
    r.read_exact(&mut magic)?;
    if &magic != MAGIC {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "kevy snapshot: bad magic",
        ));
    }
    let version = read_u8(&mut r)?;
    if !(VERSION_RELATIVE_TTL..=VERSION_HASH_TTL).contains(&version) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "kevy snapshot: bad version",
        ));
    }
    if version >= VERSION_FEED_CURSOR {
        // Loader-side the cursor is advisory (read via
        // [`read_snapshot_cursor`] by restore tooling); skip it here.
        let mut cur = [0u8; 16];
        r.read_exact(&mut cur)?;
    }
    // v3+ stores absolute Unix-ms deadlines; convert each to remaining ms
    // against one `now` read so the load is internally consistent. A deadline
    // already past becomes `Some(0)` → loaded then immediately reaped (lazy
    // get / active reaper), matching "expired key is gone". v2 ttls are
    // already remaining, so pass them through.
    let absolute_ttl = version >= VERSION_ABSOLUTE_TTL;
    let now = kevy_store::now_unix_ms();

    loop {
        let op = read_u8(&mut r)?;
        if op == OP_EOF {
            return Ok(());
        }
        // v2.4 field-TTL records: no ttl/value framing of their own.
        if op == OP_HFTTL {
            let key = read_bytes(&mut r)?;
            let field = read_bytes(&mut r)?;
            let mut d = [0u8; 8];
            r.read_exact(&mut d)?;
            if keep(&key) {
                store.load_hash_field_ttl(&key, &field, u64::from_le_bytes(d));
            }
            continue;
        }
        let raw_ttl = read_ttl(&mut r)?;
        let ttl = if absolute_ttl {
            raw_ttl.map(|deadline| deadline.saturating_sub(now))
        } else {
            raw_ttl
        };
        let key = read_bytes(&mut r)?;
        match op {
            OP_STR => {
                let val = read_bytes(&mut r)?;
                if keep(&key) {
                    store.load_str(key, val, ttl);
                }
            }
            OP_HASH => {
                let n = read_u32(&mut r)? as usize;
                let mut fields = Vec::with_capacity(n);
                for _ in 0..n {
                    let f = read_bytes(&mut r)?;
                    let v = read_bytes(&mut r)?;
                    fields.push((f, v));
                }
                if keep(&key) {
                    store.load_hash(key, fields, ttl);
                }
            }
            OP_LIST => {
                let n = read_u32(&mut r)? as usize;
                let mut items = Vec::with_capacity(n);
                for _ in 0..n {
                    items.push(read_bytes(&mut r)?);
                }
                if keep(&key) {
                    store.load_list(key, items, ttl);
                }
            }
            OP_SET => {
                let n = read_u32(&mut r)? as usize;
                let mut members = Vec::with_capacity(n);
                for _ in 0..n {
                    members.push(read_bytes(&mut r)?);
                }
                if keep(&key) {
                    store.load_set(key, members, ttl);
                }
            }
            OP_ZSET => {
                let n = read_u32(&mut r)? as usize;
                let mut pairs = Vec::with_capacity(n);
                for _ in 0..n {
                    let m = read_bytes(&mut r)?;
                    let score = f64::from_bits(read_u64(&mut r)?);
                    pairs.push((m, score));
                }
                if keep(&key) {
                    store.load_zset(key, pairs, ttl);
                }
            }
            OP_STREAM => {
                let last_ms = read_u64(&mut r)?;
                let last_seq = read_u64(&mut r)?;
                let mxd_ms = read_u64(&mut r)?;
                let mxd_seq = read_u64(&mut r)?;
                let entries_added = read_u64(&mut r)?;
                let n = read_u32(&mut r)? as usize;
                let mut entries = Vec::with_capacity(n);
                for _ in 0..n {
                    let ms = read_u64(&mut r)?;
                    let seq = read_u64(&mut r)?;
                    let nf = read_u32(&mut r)? as usize;
                    let mut fv = Vec::with_capacity(nf);
                    for _ in 0..nf {
                        let f = read_bytes(&mut r)?;
                        let v = read_bytes(&mut r)?;
                        fv.push((f, v));
                    }
                    entries.push((ms, seq, fv));
                }
                // v4 appends the consumer-group section; v2/v3 files
                // predate groups-in-snapshot, so they load with none.
                let groups = if version >= VERSION {
                    read_stream_groups(&mut r)?
                } else {
                    Vec::new()
                };
                if keep(&key) {
                    store.load_stream(
                        key,
                        entries,
                        (last_ms, last_seq),
                        (mxd_ms, mxd_seq),
                        entries_added,
                        groups,
                        ttl,
                    );
                }
            }
            other => {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    format!("kevy snapshot: unknown opcode {other}"),
                ));
            }
        }
    }
}

/// Serialize one entry: `[op][ttl][key][payload]`.
fn write_entry<W: Write>(w: &mut W, key: &[u8], value: &Value, ttl: Option<u64>) -> io::Result<()> {
    let op = match value {
        Value::Str(_) | Value::Int(_) | Value::ArcBulk(_) => OP_STR, // L1/L2: all reuse OP_STR.

        Value::Hash(_) | Value::SmallHashInline(_) => OP_HASH,
        Value::List(_) | Value::SmallListInline(_) => OP_LIST,
        // A.7 O5: both Set encodings share the OP_SET wire format —
        // payload is `[len: u32 LE][bulk: len-prefixed bytes]*`, agnostic
        // of whether the in-memory representation is `SmallSetInline` or
        // `Arc<KevySet>`.
        Value::Set(_) | Value::SmallSetInline(_) => OP_SET,
        Value::ZSet(_) | Value::SmallZSetInline(_) => OP_ZSET,
        Value::Stream(_) => OP_STREAM,
    };
    w.write_all(&[op])?;
    write_ttl(w, ttl)?;
    write_bytes(w, key)?;
    match value {
        Value::Str(v) => write_bytes(w, v.as_slice()),
        Value::Int(n) => write_bytes(w, n.to_string().as_bytes()),
        Value::ArcBulk(a) => write_bytes(w, a.as_ref()),
        Value::Hash(h) => snapshot_payload::write_hash_payload(w, h),
        Value::SmallHashInline(h) => snapshot_payload::write_small_hash_payload(w, h),
        Value::List(l) => snapshot_payload::write_list_payload(w, l),
        Value::SmallListInline(l) => snapshot_payload::write_small_list_payload(w, l),
        Value::Set(set) => snapshot_payload::write_set_payload(w, set),
        Value::SmallSetInline(s) => snapshot_payload::write_small_set_payload(w, s),
        Value::ZSet(z) => snapshot_payload::write_zset_payload(w, z),
        Value::SmallZSetInline(z) => snapshot_payload::write_small_zset_payload(w, z),
        Value::Stream(s) => snapshot_payload::write_stream_payload(w, s),
    }
}

/// v4 consumer-group section: `[n_groups][per group: name, last_delivered,
/// consumers (name + last_seen_ms), PEL rows]`. Tombstone PEL rows are kept
/// — the snapshot path is the full-fidelity one (the AOF rewrite can't
/// re-create them via XCLAIM, see `rewrite_fmt`).
pub(crate) fn write_stream_groups<W: Write>(w: &mut W, groups: &[kevy_store::LoadedGroup]) -> io::Result<()> {
    w.write_all(&(groups.len() as u32).to_le_bytes())?;
    for g in groups {
        write_bytes(w, &g.name)?;
        w.write_all(&g.last_delivered.0.to_le_bytes())?;
        w.write_all(&g.last_delivered.1.to_le_bytes())?;
        w.write_all(&(g.consumers.len() as u32).to_le_bytes())?;
        for (name, last_seen_ms) in &g.consumers {
            write_bytes(w, name)?;
            w.write_all(&last_seen_ms.to_le_bytes())?;
        }
        w.write_all(&(g.pel.len() as u32).to_le_bytes())?;
        for (ms, seq, consumer, delivery_time_ms, delivery_count) in &g.pel {
            w.write_all(&ms.to_le_bytes())?;
            w.write_all(&seq.to_le_bytes())?;
            write_bytes(w, consumer)?;
            w.write_all(&delivery_time_ms.to_le_bytes())?;
            w.write_all(&delivery_count.to_le_bytes())?;
        }
    }
    Ok(())
}

/// Loader-side twin of [`write_stream_groups`].
fn read_stream_groups<R: Read>(r: &mut R) -> io::Result<Vec<kevy_store::LoadedGroup>> {
    let n = read_u32(r)? as usize;
    let mut groups = Vec::with_capacity(n);
    for _ in 0..n {
        let name = read_bytes(r)?;
        let last_delivered = (read_u64(r)?, read_u64(r)?);
        let nc = read_u32(r)? as usize;
        let mut consumers = Vec::with_capacity(nc);
        for _ in 0..nc {
            let cname = read_bytes(r)?;
            consumers.push((cname, read_u64(r)?));
        }
        let np = read_u32(r)? as usize;
        let mut pel = Vec::with_capacity(np);
        for _ in 0..np {
            let ms = read_u64(r)?;
            let seq = read_u64(r)?;
            let consumer = read_bytes(r)?;
            let delivery_time_ms = read_u64(r)?;
            let delivery_count = read_u32(r)?;
            pel.push((ms, seq, consumer, delivery_time_ms, delivery_count));
        }
        groups.push(kevy_store::LoadedGroup { name, last_delivered, consumers, pel });
    }
    Ok(groups)
}

fn write_ttl<W: Write>(w: &mut W, ttl: Option<u64>) -> io::Result<()> {
    match ttl {
        Some(ms) => {
            w.write_all(&[1u8])?;
            w.write_all(&ms.to_le_bytes())?;
        }
        None => w.write_all(&[0u8])?,
    }
    Ok(())
}

fn read_ttl<R: Read>(r: &mut R) -> io::Result<Option<u64>> {
    if read_u8(r)? == 1 {
        Ok(Some(read_u64(r)?))
    } else {
        Ok(None)
    }
}

fn tmp_path(path: &Path) -> std::path::PathBuf {
    let mut s = path.as_os_str().to_owned();
    s.push(".tmp");
    s.into()
}

pub(crate) fn write_bytes<W: Write>(w: &mut W, b: &[u8]) -> io::Result<()> {
    w.write_all(&(b.len() as u32).to_le_bytes())?;
    w.write_all(b)
}

fn read_bytes<R: Read>(r: &mut R) -> io::Result<Vec<u8>> {
    let len = read_u32(r)? as usize;
    let mut buf = vec![0u8; len];
    r.read_exact(&mut buf)?;
    Ok(buf)
}

fn read_u8<R: Read>(r: &mut R) -> io::Result<u8> {
    let mut b = [0u8; 1];
    r.read_exact(&mut b)?;
    Ok(b[0])
}

fn read_u32<R: Read>(r: &mut R) -> io::Result<u32> {
    let mut b = [0u8; 4];
    r.read_exact(&mut b)?;
    Ok(u32::from_le_bytes(b))
}

fn read_u64<R: Read>(r: &mut R) -> io::Result<u64> {
    let mut b = [0u8; 8];
    r.read_exact(&mut b)?;
    Ok(u64::from_le_bytes(b))
}

#[cfg(test)]
mod tests;
#[cfg(test)]
mod tests_aof;
#[cfg(test)]
mod tests_rewrite;