kevy_embedded/

store_persist.rs

//! Durability methods on [`Store`] — `BGREWRITEAOF` and `SAVE` — plus
//! their per-shard helpers. Extracted from `store.rs` to keep that file
//! under the 500-LOC project ceiling. Behaviour is unchanged from the
//! pre-split layout; this module hosts long-running disk paths
//! separately from the hot lock/dispatch surface in `store.rs`.

// Temporary files, on the failure path of the write that created them.
// The real error is the one being returned; a temp file that will not
// delete is a stray byte range the next rewrite overwrites, and
// reporting it here would replace the cause with its cleanup.
#![expect(
    clippy::let_underscore_must_use,
    reason = "removing a temp file cannot outrank the error that stranded it"
)]

use crate::KevyResult;
use std::io;
use std::sync::RwLock;
use std::time::Instant;

use kevy_persist::{Argv, RewriteStats};

use crate::metric::KevyMetric;
use crate::store::{Inner, Store, lock_write};

impl Store {
    /// Whether a 3.x binary could still open this data directory —
    /// the downgrade window `UPGRADING.md` describes, as a readable
    /// state instead of an assumption.
    ///
    /// `Some(true)`: every shard's AOF still speaks `KEVYAOF1`, so
    /// downgrading is a binary swap back. `Some(false)`: at least one
    /// shard has upgraded to v2 (a rewrite ran), so a downgrade needs a
    /// keyspace export through a client. `None`: no AOF is configured —
    /// the question has no meaning for a memory-only store.
    ///
    /// From an embedder's dogfood report: their `doctor`
    /// command exists to tell users "where you are and what to do
    /// next", and this state was `pub(crate)` — so their CHANGELOG had
    /// to declare the window closed unconditionally when it was open,
    /// observable, and per-directory.
    #[cfg(all(feature = "persist", not(target_arch = "wasm32")))]
    pub fn downgradeable_to_v3(&self) -> Option<bool> {
        let mut any = false;
        let mut all_v1 = true;
        for shard in self.shards.iter() {
            let g = crate::store::lock_write(shard);
            if let Some(aof) = g.aof.as_ref() {
                any = true;
                if aof.format() != kevy_persist::AofFormat::V1 {
                    all_v1 = false;
                }
            }
        }
        any.then_some(all_v1)
    }

    /// Durability barrier: flush + `fdatasync` every shard's
    /// AOF now, regardless of the configured `appendfsync` policy. On
    /// `Ok(())`, every write acknowledged before this call is on
    /// stable storage. The `EverySec` serving-store idiom:
    ///
    /// ```
    /// # use kevy_embedded::{AppendFsync, Config, Store};
    /// # let dir = kevy_tmpdir::TmpDir::new("fsync-aof-doc");
    /// # let store = Store::open(Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::EverySec))?;
    /// store.atomic(|c| c.incr_by(b"orders", 1))?; // the critical write
    /// store.fsync_aof()?; // durable-on-ack for THIS block only
    /// # Ok::<(), kevy_embedded::KevyError>(())
    /// ```
    ///
    /// Cost: one `fdatasync` per dirty shard; a no-op on clean shards.
    /// Under `appendfsync = always` it is a no-op (already durable).
    pub fn fsync_aof(&self) -> KevyResult<()> {
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            if let Some(aof) = &mut g.aof {
                aof.sync_now()?;
            }
        }
        Ok(())
    }

    /// `BGREWRITEAOF`: rebuild every shard's AOF from current state.
    /// Synchronous (despite the RESP name). Returns the summed stats
    /// (`None` if persistence is off / no shard rewrote). Shards mid
    /// rewrite are skipped. Emits [`KevyMetric::Rewrite`] per shard.
    pub fn rewrite_aof(&self) -> KevyResult<Option<RewriteStats>> {
        let mut agg: Option<RewriteStats> = None;
        for shard in self.shards.iter() {
            if let Some(stats) = self.rewrite_one_shard(shard)? {
                let acc = agg.get_or_insert_with(RewriteStats::default);
                acc.keys += stats.keys;
                acc.bytes += stats.bytes;
            }
        }
        Ok(agg)
    }

    /// One shard's synchronous three-phase rewrite. `Ok(None)` =
    /// skipped (persistence off / already mid-rewrite).
    fn rewrite_one_shard(&self, shard: &RwLock<Inner>) -> KevyResult<Option<RewriteStats>> {
        let start = Instant::now();
        // Phase 1 (locked): freeze the COW view + start the tee —
        // O(n)-shallow, no serialization under the lock.
        let (view, aux, tmp, before_bytes) = {
            let mut g = lock_write(shard);
            let aux = crate::shard_restore::catalog_aux(&g);
            let Inner { store, aof, .. } = &mut *g;
            let Some(aof) = aof else { return Ok(None) };
            if aof.is_rewriting() {
                return Ok(None);
            }
            let before = aof.size_bytes();
            let view = store.collect_snapshot();
            (view, aux, aof.begin_view_rewrite()?, before)
        };
        // Phase 2 (unlocked): serialize + fsync the compacted log.
        let image = kevy_persist::WithAux::new(&view, aux.as_ref());
        let keys = match kevy_persist::dump_aof(&tmp, &image) {
            Ok(stats) => stats.keys,
            Err(e) => {
                let mut g = lock_write(shard);
                if let Some(aof) = &mut g.aof {
                    aof.abort_concurrent_rewrite();
                }
                let _ = std::fs::remove_file(&tmp);
                return Err(e.into());
            }
        };
        // Phase 3 (locked): append the tee'd diff and swap.
        let mut g = lock_write(shard);
        let Some(aof) = &mut g.aof else { return Ok(None) };
        let stats = match aof.finish_concurrent_rewrite(&tmp, keys) {
            Ok(s) => s,
            Err(e) => {
                aof.abort_concurrent_rewrite();
                let _ = std::fs::remove_file(&tmp);
                return Err(e.into());
            }
        };
        if let Some(sink) = &self.config.metric_sink {
            sink.emit(KevyMetric::Rewrite {
                keys: stats.keys,
                before_bytes,
                after_bytes: stats.bytes,
                elapsed_ms: start.elapsed().as_millis() as u64,
            });
        }
        Ok(Some(stats))
    }

    /// Apply one AOF-format command frame directly to the keyspace — the
    /// programmatic face of AOF replay, for hosts that store the log
    /// themselves (targets without a filesystem, where the embedding
    /// host reads the log back and feeds it in frame by frame on open).
    ///
    /// Speaks the same verb set `Store::open` replays from an on-disk
    /// AOF; unknown verbs are skipped (forward compatibility with logs
    /// written by a newer kevy). Keyed verbs route to the owning shard;
    /// `FLUSHALL`/`FLUSHDB` reach every shard. A catalog frame (the
    /// declared indexes, views and tables) replaces the store's catalog
    /// when it is newer, or when the store has recorded none of its own,
    /// and the indexes rebuild from the keyspace. The frame is **not**
    /// re-appended to any AOF — this is the read-back half of the pump,
    /// so re-logging would double-apply on the next replay.
    pub fn apply_frame(&self, args: &Argv) {
        let Some(verb) = args.first() else { return };
        // the index, view and table catalog is store-wide state, not a
        // key: it is taken whole when newer, as a native open takes it
        #[cfg(feature = "index")]
        if crate::shard_restore::is_catalog(args) {
            self.guard.catalog.adopt_fed(args);
            return;
        }
        if verb.eq_ignore_ascii_case(b"FLUSHALL") || verb.eq_ignore_ascii_case(b"FLUSHDB") {
            for shard in self.shards.iter() {
                crate::replay::apply(&mut lock_write(shard).store, args);
            }
            return;
        }
        let mut buf = [0u8; 32];
        let up = kevy_verbs::args::upper_verb(verb, &mut buf);
        if let Some(key) = crate::verb_keys::shard_key(up, args) {
            crate::replay::apply(&mut self.wshard(key).store, args);
        }
    }

    /// Serialize the whole keyspace into an in-memory compacted AOF image
    /// (magic header + one command stream per key — the same bytes an
    /// AOF rewrite puts on disk). The write half of host-mediated
    /// persistence: hosts without a filesystem hand this buffer to their
    /// own storage, replacing the accumulated append log, and feed it
    /// back through [`Self::apply_frame`] on the next open. Once the store
    /// has recorded an index, view or table catalog, the image ends the
    /// first shard's commands with the catalog frame.
    ///
    /// Each shard is frozen copy-on-write and serialized off-lock, so
    /// concurrent readers and writers on other shards are not blocked
    /// for the duration of the dump.
    pub fn dump_aof_buf(&self) -> Vec<u8> {
        let mut out = Vec::new();
        // the catalog, once one was recorded, rides the first shard's
        // image, the way a native rewrite keeps it beside the keyspace
        #[cfg(feature = "index")]
        let aux = self.guard.catalog.fed_aux();
        #[cfg(not(feature = "index"))]
        let aux: Option<Argv> = None;
        for (i, shard) in self.shards.iter().enumerate() {
            let view = lock_write(shard).store.collect_snapshot();
            let image = kevy_persist::WithAux::new(&view, if i == 0 { aux.as_ref() } else { None });
            // V2, like every other rewrite output: the wasm door's
            // host-mediated pump replays both formats, and its dump is
            // the log's upgrade point (mirroring the native
            // first-rewrite upgrade).
            let (buf, _keys) = kevy_persist::dump_store_to_buf(&image, kevy_persist::AofFormat::V2);
            if i == 0 {
                out = buf;
            } else {
                // One magic header per image, not per shard.
                out.extend_from_slice(&buf[kevy_persist::AOF2_MAGIC.len()..]);
            }
        }
        out
    }

    /// Snapshot every shard to its `dump-{i}.rdb`, atomically. `Ok(false)`
    /// when persistence is disabled.
    pub fn save_snapshot(&self) -> KevyResult<bool> {
        let Some(dir) = self.config.data_dir.as_ref() else {
            return Ok(false);
        };
        for (i, shard) in self.shards.iter().enumerate() {
            save_shard_snapshot(shard, &kevy_persist::layout::snapshot_path(dir, i))?;
        }
        Ok(true)
    }
}

/// Save one shard's snapshot with the snapshot+log contract intact:
/// after a successful save the AOF holds **only post-collect writes**,
/// so a restart replays them over the snapshot without double-applying
/// history (non-idempotent commands like RPUSH duplicated before this).
///
/// Phase 1 (write lock): freeze the COW view + start the AOF tee — no
/// write may land between the two (the tee atomicity contract). Phase 2
/// (unlocked): serialize the view to the snapshot's durable tmp.
/// Phase 3 (write lock): commit — snapshot rename and tee'd AOF reset
/// adjacent, so the snapshot/log commit window stays microseconds.
/// A shard's frozen view, the catalog frame beside it, and the log reset
/// the save started.
type Frozen = (kevy_store::SnapshotView, Option<kevy_persist::Argv>, Option<std::path::PathBuf>);

pub(crate) fn save_shard_snapshot(shard: &RwLock<Inner>, path: &std::path::Path) -> KevyResult<()> {
    let (view, aux, reset_tmp) = freeze_for_save(shard)?;
    let image = kevy_persist::WithAux::new(&view, aux.as_ref());
    let tmp = match kevy_persist::write_snapshot_tmp(&image, path) {
        Ok(t) => t,
        Err(e) => {
            if reset_tmp.is_some()
                && let Some(aof) = &mut lock_write(shard).aof
            {
                aof.abort_concurrent_rewrite();
            }
            return Err(e.into());
        }
    };
    let mut g = lock_write(shard);
    match (reset_tmp, &mut g.aof) {
        (Some(reset), Some(aof)) => {
            if let Err(e) = aof.commit_snapshot(&tmp, path, &reset) {
                let _ = std::fs::remove_file(&tmp);
                return Err(e.into());
            }
        }
        _ => std::fs::rename(&tmp, path)?,
    }
    Ok(())
}

/// Phase-1 helper: collect the view and the catalog frame beside it, and
/// start the tee, under one write lock. A racing background auto-rewrite owns the tee; it runs its
/// slow half off-lock and finishes in milliseconds, so wait it out
/// (bounded) rather than saving a snapshot whose log would double-
/// apply on replay.
fn freeze_for_save(shard: &RwLock<Inner>) -> KevyResult<Frozen> {
    for _ in 0..2000 {
        {
            let mut g = lock_write(shard);
            let aux = crate::shard_restore::catalog_aux(&g);
            let Inner { store, aof, .. } = &mut *g;
            match aof {
                Some(a) if a.is_rewriting() => {} // racing rewrite — retry
                Some(a) => {
                    let view = store.collect_snapshot();
                    return Ok((view, aux, Some(a.begin_view_rewrite()?)));
                }
                None => return Ok((store.collect_snapshot(), aux, None)),
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(5));
    }
    Err(io::Error::new(
        io::ErrorKind::TimedOut,
        "kevy-embedded: AOF rewrite still in flight after 10s; snapshot aborted",
    )
    .into())
}