Struct Store 

pub struct Store { /* private fields */ }

The embedded keyspace.

Store is Clone (since v1.1.0). A clone is a cheap Arc bump: every clone reaches the same underlying shards + AOF + reaper + pub/sub bus. The reaper thread is joined and each shard’s AOF is flushed exactly once, when the last clone is dropped.

use kevy_embedded::{Config, Store};

let s = Store::open(Config::default().with_ttl_reaper_manual())?;
let s2 = s.clone();
std::thread::spawn(move || {
    s2.set(b"from-thread", b"v").unwrap();
}).join().unwrap();
assert_eq!(s.get(b"from-thread")?, Some(b"v".to_vec()));

Every method takes &self. Sharding (see Config::with_shards) lets a multi-threaded consumer scale across cores; pub/sub is process-wide (handled on shard 0).

Implementations

impl Store

pub fn info(&self) -> KevyInfo

One-shot snapshot of the store’s introspection counters. See KevyInfo. Takes the embedded mutex once; safe to call from a health endpoint.

pub fn expire_pending_count(&self) -> usize

Number of live keys that currently carry a TTL (the expire-set size, summed across shards).

pub fn ttl(&self, key: &[u8]) -> Option<Duration>

Remaining TTL for key as a Duration, or None when the key is absent or has no TTL (persistent). For the raw Redis PTTL sentinels (-2 no key, -1 no TTL) use Store::ttl_ms.

impl Store

pub fn set(&self, key: &[u8], value: &[u8]) -> Result<bool>

SET key value (no TTL, no NX/XX). Returns true always under the embedded API (Redis semantics: SET overwrites; NX/XX vetoes would return false but we don’t expose those here — use Store::with for the full surface).

Examples found in repository

examples/embedded.rs (line 9)

fn main() -> std::io::Result<()> {
    let s = Store::open(Config::default())?;
    s.set(b"greeting", b"hello, kevy")?;
    let v = s.get(b"greeting")?;
    println!("greeting = {}", String::from_utf8_lossy(v.as_deref().unwrap_or(b"<missing>")));
    println!("dbsize = {}", s.dbsize());
    Ok(())
}

examples/bench_embed.rs (line 19)

fn bench(label: &str, store: &Store, n: usize, keys: &[Vec<u8>]) {
    // Warm the keyspace so GET is all hits and allocations are amortized.
    for k in keys {
        store.set(k, VAL).unwrap();
    }

    let t = Instant::now();
    for i in 0..n {
        store.set(&keys[i % KEYS], VAL).unwrap();
    }
    let set_s = t.elapsed().as_secs_f64();

    let t = Instant::now();
    let mut hits = 0usize;
    for i in 0..n {
        if store.get(&keys[i % KEYS]).unwrap().is_some() {
            hits += 1;
        }
    }
    let get_s = t.elapsed().as_secs_f64();
    std::hint::black_box(hits);

    println!(
        "[{label:<13}] SET {:>10.0} ops/s   GET {:>10.0} ops/s",
        n as f64 / set_s,
        n as f64 / get_s
    );
}

examples/embedded-cache.rs (line 19)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

examples/bench_embed_mt.rs (line 30)

fn run(label: &str, store: &Store, threads: usize, n_per: usize, keys: &[Vec<u8>], write: bool) {
    let start = Instant::now();
    let handles: Vec<_> = (0..threads)
        .map(|tid| {
            let s = store.clone(); // Arc bump → same inner / same lock
            let keys = keys.to_vec();
            thread::spawn(move || {
                let mut acc = 0usize;
                for i in 0..n_per {
                    // De-correlate threads' key streams so they don't all hit
                    // the same key/bucket in lockstep.
                    let k = &keys[(i.wrapping_mul(31).wrapping_add(tid * 7)) % KEYS];
                    if write {
                        s.set(k, VAL).unwrap();
                        acc += 1;
                    } else if s.get(k).unwrap().is_some() {
                        acc += 1;
                    }
                }
                acc
            })
        })
        .collect();
    for h in handles {
        h.join().unwrap();
    }
    let secs = start.elapsed().as_secs_f64();
    let total = (threads * n_per) as f64;
    println!(
        "[{label:<10}] threads={threads:2}  {:>10.0} ops/s  ({:>5.1}M)",
        total / secs,
        total / secs / 1e6
    );
}

fn main() {
    let n_per: usize = std::env::var("KEVY_BENCH_N")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(2_000_000);
    let keys: Vec<Vec<u8>> = (0..KEYS).map(|i| format!("k{i}").into_bytes()).collect();
    let shards: usize = std::env::var("KEVY_SHARDS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(1);

    // Shared in-memory store (no AOF) — isolate the lock/keyspace from disk.
    let store = Store::open(Config::default().with_shards(shards).with_ttl_reaper_manual()).unwrap();
    for k in &keys {
        store.set(k, VAL).unwrap();
    }

    println!(
        "kevy-embedded MULTI-THREAD throughput — in-memory, shards={shards}, {KEYS} keys, {}B val, n={n_per}/thread",
        VAL.len()
    );
    for &t in &[1usize, 2, 4, 8, 10] {
        run("GET", &store, t, n_per, &keys, false);
    }
    for &t in &[1usize, 2, 4, 8, 10] {
        run("SET", &store, t, n_per, &keys, true);
    }
}

pub fn set_with_ttl( &self, key: &[u8], value: &[u8], ttl: Duration, ) -> Result<bool>

SET key value PX ms — overwrites + sets TTL. The AOF records an absolute PEXPIREAT deadline (not the relative ttl) so the key expires at the same wall-clock instant after a restart — a relative PEXPIRE would be re-anchored to replay-time, resetting the TTL to a fresh full duration on every restart (INC-2026-06-09).

Examples found in repository

examples/bench_embed.rs (line 106)

fn bench_ttl_get(label: &str, manual_reaper: bool, n: usize, keys: &[Vec<u8>]) {
    let cfg = if manual_reaper {
        Config::default().with_ttl_reaper_manual()
    } else {
        Config::default() // background reaper (default) → cached clock trusted
    };
    let store = Store::open(cfg).unwrap();
    let ttl = std::time::Duration::from_secs(3600); // never expires during the run
    for k in keys {
        store.set_with_ttl(k, VAL, ttl).unwrap();
    }
    let t = Instant::now();
    let mut hits = 0usize;
    for i in 0..n {
        if store.get(&keys[i % KEYS]).unwrap().is_some() {
            hits += 1;
        }
    }
    std::hint::black_box(hits);
    println!("[{label}] GET {:>10.0} ops/s", n as f64 / t.elapsed().as_secs_f64());
}

pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>>

GET key — Some(bytes) on hit, None on miss or expired.

With eviction off (maxmemory == 0, the default) this takes the read lock and a non-mutating store lookup, so concurrent readers scale across cores — the path a read-heavy embed cache lives on. With eviction on it falls back to the exclusive lock + mutating get so each access still stamps the LRU clock.

Examples found in repository

examples/embedded.rs (line 10)

fn main() -> std::io::Result<()> {
    let s = Store::open(Config::default())?;
    s.set(b"greeting", b"hello, kevy")?;
    let v = s.get(b"greeting")?;
    println!("greeting = {}", String::from_utf8_lossy(v.as_deref().unwrap_or(b"<missing>")));
    println!("dbsize = {}", s.dbsize());
    Ok(())
}

examples/bench_embed.rs (line 31)

fn bench(label: &str, store: &Store, n: usize, keys: &[Vec<u8>]) {
    // Warm the keyspace so GET is all hits and allocations are amortized.
    for k in keys {
        store.set(k, VAL).unwrap();
    }

    let t = Instant::now();
    for i in 0..n {
        store.set(&keys[i % KEYS], VAL).unwrap();
    }
    let set_s = t.elapsed().as_secs_f64();

    let t = Instant::now();
    let mut hits = 0usize;
    for i in 0..n {
        if store.get(&keys[i % KEYS]).unwrap().is_some() {
            hits += 1;
        }
    }
    let get_s = t.elapsed().as_secs_f64();
    std::hint::black_box(hits);

    println!(
        "[{label:<13}] SET {:>10.0} ops/s   GET {:>10.0} ops/s",
        n as f64 / set_s,
        n as f64 / get_s
    );
}

fn main() {
    let n: usize = std::env::var("KEVY_BENCH_N")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(2_000_000);
    // Keys precomputed outside the timed loop so `format!`/alloc cost isn't
    // attributed to kevy.
    let keys: Vec<Vec<u8>> = (0..KEYS).map(|i| format!("k{i}").into_bytes()).collect();

    println!("kevy-embedded in-process throughput — single thread, n={n}, {KEYS} keys, {}B val", VAL.len());

    let s1 = Store::open(Config::default().with_ttl_reaper_manual()).unwrap();
    bench("in-memory", &s1, n, &keys);

    let dir2 = std::env::temp_dir().join("kevy_embed_bench_everysec");
    let _ = std::fs::remove_dir_all(&dir2);
    let s2 = Store::open(
        Config::default()
            .with_persist(&dir2)
            .with_ttl_reaper_manual()
            .with_appendfsync(AppendFsync::EverySec),
    )
    .unwrap();
    bench("aof-everysec", &s2, n, &keys);

    let dir3 = std::env::temp_dir().join("kevy_embed_bench_always");
    let _ = std::fs::remove_dir_all(&dir3);
    let s3 = Store::open(
        Config::default()
            .with_persist(&dir3)
            .with_ttl_reaper_manual()
            .with_appendfsync(AppendFsync::Always),
    )
    .unwrap();
    // Always-fsync is one fdatasync per write (no group commit on the embedded
    // single-op path) — fsync-rate-bound, so run far fewer ops to stay bounded.
    bench("aof-always", &s3, (n / 20).max(50_000), &keys);

    // TTL'd-key GET: the mailrs path (every cache key has a TTL). With the
    // background reaper the cached clock is trusted (no per-get Instant::now);
    // manual mode reads a fresh clock per get — the gap is the cached-clock win.
    bench_ttl_get("ttl GET (cached clk)", false, n, &keys); // background reaper
    bench_ttl_get("ttl GET (fresh clk) ", true, n, &keys); // manual reaper

    drop((s1, s2, s3));
    let _ = std::fs::remove_dir_all(&dir2);
    let _ = std::fs::remove_dir_all(&dir3);
}

/// GET throughput over keys that all carry a (long) TTL — the mailrs cache
/// shape. `manual_reaper` toggles whether the store trusts the cached clock
/// (background) or reads a fresh clock per get (manual).
fn bench_ttl_get(label: &str, manual_reaper: bool, n: usize, keys: &[Vec<u8>]) {
    let cfg = if manual_reaper {
        Config::default().with_ttl_reaper_manual()
    } else {
        Config::default() // background reaper (default) → cached clock trusted
    };
    let store = Store::open(cfg).unwrap();
    let ttl = std::time::Duration::from_secs(3600); // never expires during the run
    for k in keys {
        store.set_with_ttl(k, VAL, ttl).unwrap();
    }
    let t = Instant::now();
    let mut hits = 0usize;
    for i in 0..n {
        if store.get(&keys[i % KEYS]).unwrap().is_some() {
            hits += 1;
        }
    }
    std::hint::black_box(hits);
    println!("[{label}] GET {:>10.0} ops/s", n as f64 / t.elapsed().as_secs_f64());
}

examples/embedded-cache.rs (line 30)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

examples/bench_embed_mt.rs (line 32)

fn run(label: &str, store: &Store, threads: usize, n_per: usize, keys: &[Vec<u8>], write: bool) {
    let start = Instant::now();
    let handles: Vec<_> = (0..threads)
        .map(|tid| {
            let s = store.clone(); // Arc bump → same inner / same lock
            let keys = keys.to_vec();
            thread::spawn(move || {
                let mut acc = 0usize;
                for i in 0..n_per {
                    // De-correlate threads' key streams so they don't all hit
                    // the same key/bucket in lockstep.
                    let k = &keys[(i.wrapping_mul(31).wrapping_add(tid * 7)) % KEYS];
                    if write {
                        s.set(k, VAL).unwrap();
                        acc += 1;
                    } else if s.get(k).unwrap().is_some() {
                        acc += 1;
                    }
                }
                acc
            })
        })
        .collect();
    for h in handles {
        h.join().unwrap();
    }
    let secs = start.elapsed().as_secs_f64();
    let total = (threads * n_per) as f64;
    println!(
        "[{label:<10}] threads={threads:2}  {:>10.0} ops/s  ({:>5.1}M)",
        total / secs,
        total / secs / 1e6
    );
}

pub fn del(&self, keys: &[&[u8]]) -> Result<usize>

DEL key1 [key2 ...]. Returns the count of keys actually removed. Keys fan out to their owning shards.

pub fn exists(&self, keys: &[&[u8]]) -> Result<usize>

EXISTS key1 [key2 ...]. Count of existing keys (duplicates counted multiple times, matching Redis).

pub fn incr(&self, key: &[u8]) -> Result<i64>

INCR key. Returns the post-increment value.

pub fn incr_by(&self, key: &[u8], delta: i64) -> Result<i64>

INCRBY key delta. Negative delta does DECR-style work.

pub fn expire(&self, key: &[u8], ttl: Duration) -> Result<bool>

EXPIRE key seconds. Returns true if a key was touched. The AOF records an absolute PEXPIREAT deadline (see Self::set_with_ttl) so the TTL survives a restart unchanged.

pub fn persist(&self, key: &[u8]) -> Result<bool>

PERSIST key. Returns true if a TTL was actually cleared.

pub fn ttl_ms(&self, key: &[u8]) -> i64

Remaining TTL in ms (or Redis-style -1/-2 for no-TTL/no-key).

pub fn type_of(&self, key: &[u8]) -> &'static str

TYPE key — "string", "hash", "list", "set", "zset", or "none".

pub fn dbsize(&self) -> usize

DBSIZE — total live keys across all shards.

Examples found in repository

examples/embedded.rs (line 12)

fn main() -> std::io::Result<()> {
    let s = Store::open(Config::default())?;
    s.set(b"greeting", b"hello, kevy")?;
    let v = s.get(b"greeting")?;
    println!("greeting = {}", String::from_utf8_lossy(v.as_deref().unwrap_or(b"<missing>")));
    println!("dbsize = {}", s.dbsize());
    Ok(())
}

examples/embedded-cache.rs (line 22)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

examples/replay_real_aof.rs (line 40)

fn main() {
    let path = std::env::args()
        .nth(1)
        .expect("usage: replay_real_aof <aof-path>");
    let src = PathBuf::from(&path);
    let bytes = std::fs::metadata(&src)
        .map(|m| m.len())
        .unwrap_or(0);
    println!("reproducer: staging {} bytes from {}", bytes, src.display());

    let dir = std::env::temp_dir().join(format!(
        "kevy-aof-reproducer-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).expect("create temp dir");
    let staged = dir.join("aof-0.aof");
    std::fs::copy(&src, &staged).expect("copy AOF into staging dir");

    println!("reproducer: opening Store from {}", dir.display());
    let store = Store::open(
        Config::default()
            .with_persist(&dir)
            .with_aof_filename("aof-0.aof"),
    )
    .expect("Store::open");

    println!("reproducer: opened OK; dbsize = {}", store.dbsize());
}

pub fn flush(&self) -> Result<()>

FLUSHALL — empty every shard (each logs FLUSHALL so a replay reaches the same empty state).

pub fn key_bytes(&self, key: &[u8]) -> Option<u64>

MEMORY USAGE for one key — Some(bytes) or None if absent.

pub fn used_memory(&self) -> u64

Live used_memory estimate (summed across shards).

Examples found in repository

examples/embedded-cache.rs (line 23)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

pub fn evictions_total(&self) -> u64

INFO-style counter: total keys evicted by maxmemory (all shards).

Examples found in repository

examples/embedded-cache.rs (line 24)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

pub fn expired_keys_total(&self) -> u64

INFO-style counter: total keys expired (lazy + active reaper, all shards).

pub fn hset(&self, key: &[u8], pairs: &[(&[u8], &[u8])]) -> Result<usize>

HSET key field value [field value ...]. Returns count newly added.

pub fn hget(&self, key: &[u8], field: &[u8]) -> Result<Option<Vec<u8>>>

HGET key field. None if absent.

pub fn hdel(&self, key: &[u8], fields: &[&[u8]]) -> Result<usize>

HDEL key field [field ...]. Returns count actually removed.

pub fn lpush(&self, key: &[u8], values: &[&[u8]]) -> Result<usize>

LPUSH key value [value ...]. Returns the new list length.

pub fn rpush(&self, key: &[u8], values: &[&[u8]]) -> Result<usize>

RPUSH key value [value ...]. Returns the new list length.

pub fn lpop(&self, key: &[u8], count: usize) -> Result<Vec<Vec<u8>>>

LPOP key count. Returns popped values from the head.

pub fn rpop(&self, key: &[u8], count: usize) -> Result<Vec<Vec<u8>>>

RPOP key count. Symmetric to LPOP from the tail.

pub fn llen(&self, key: &[u8]) -> Result<usize>

LLEN key. Length of the list at key; 0 if absent.

pub fn sadd(&self, key: &[u8], members: &[&[u8]]) -> Result<usize>

SADD key member [member ...]. Returns count newly added.

pub fn srem(&self, key: &[u8], members: &[&[u8]]) -> Result<usize>

SREM key member [member ...]. Returns count actually removed.

pub fn smembers(&self, key: &[u8]) -> Result<Vec<Vec<u8>>>

SMEMBERS key. Order implementation-defined; empty if absent.

pub fn scard(&self, key: &[u8]) -> Result<usize>

SCARD key. Member count; 0 if absent.

pub fn zadd(&self, key: &[u8], pairs: &[(f64, &[u8])]) -> Result<usize>

ZADD key score member [score member ...]. Returns count newly added.

pub fn zrem(&self, key: &[u8], members: &[&[u8]]) -> Result<usize>

ZREM key member [member ...]. Returns count actually removed.

pub fn zscore(&self, key: &[u8], member: &[u8]) -> Result<Option<f64>>

ZSCORE key member. Some(score) if present.

pub fn zcard(&self, key: &[u8]) -> Result<usize>

ZCARD key. Member count; 0 if absent.

pub fn publish(&self, channel: &[u8], payload: &[u8]) -> usize

PUBLISH channel payload. Delivers payload to every subscriber on channel (direct + pattern matches) inside this process. Returns the count of receivers the message reached.

pub fn subscribe(&self, channels: &[&[u8]]) -> Subscription

Open a Subscription subscribed to channels. Drop the handle to unsubscribe from everything atomically. Pass &[] to start with no subscriptions and add some later via Subscription::subscribe / Subscription::psubscribe.

pub fn psubscribe(&self, patterns: &[&[u8]]) -> Subscription

Convenience: open a Subscription starting on pattern subscriptions.

impl Store

pub fn open(config: Config) -> Result<Self>

Open an embedded keyspace per config.

• Pure in-memory when config.data_dir is None.
• With persistence: each shard loads its snapshot then replays its AOF (config.shards > 1 re-shards a legacy single AOF on first open).
• Spawns a background TTL reaper thread when config.ttl_reaper == Background (the default).

Examples found in repository

examples/embedded.rs (line 8)

fn main() -> std::io::Result<()> {
    let s = Store::open(Config::default())?;
    s.set(b"greeting", b"hello, kevy")?;
    let v = s.get(b"greeting")?;
    println!("greeting = {}", String::from_utf8_lossy(v.as_deref().unwrap_or(b"<missing>")));
    println!("dbsize = {}", s.dbsize());
    Ok(())
}

examples/embedded-cache.rs (lines 10-14)

fn main() -> std::io::Result<()> {
    let s = Store::open(
        Config::default()
            .with_max_memory(200 * 1024)
            .with_eviction(EvictionPolicy::AllKeysLru),
    )?;

    for i in 0..10_000 {
        let key = format!("user:{i:05}");
        let val = format!("user-payload-{i}");
        s.set(key.as_bytes(), val.as_bytes())?;
    }

    println!("dbsize after insert flood: {}", s.dbsize());
    println!("used_memory: {} bytes (limit 200 KiB)", s.used_memory());
    println!("evictions_total: {}", s.evictions_total());

    // Touch a recent key — it should still be live.
    let recent = format!("user:0{}", 9999);
    println!(
        "user:09999 → {:?}",
        s.get(recent.as_bytes())?.as_deref().map(String::from_utf8_lossy)
    );

    Ok(())
}

examples/replay_real_aof.rs (lines 33-37)

fn main() {
    let path = std::env::args()
        .nth(1)
        .expect("usage: replay_real_aof <aof-path>");
    let src = PathBuf::from(&path);
    let bytes = std::fs::metadata(&src)
        .map(|m| m.len())
        .unwrap_or(0);
    println!("reproducer: staging {} bytes from {}", bytes, src.display());

    let dir = std::env::temp_dir().join(format!(
        "kevy-aof-reproducer-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).expect("create temp dir");
    let staged = dir.join("aof-0.aof");
    std::fs::copy(&src, &staged).expect("copy AOF into staging dir");

    println!("reproducer: opening Store from {}", dir.display());
    let store = Store::open(
        Config::default()
            .with_persist(&dir)
            .with_aof_filename("aof-0.aof"),
    )
    .expect("Store::open");

    println!("reproducer: opened OK; dbsize = {}", store.dbsize());
}

examples/bench_embed_mt.rs (line 64)

fn main() {
    let n_per: usize = std::env::var("KEVY_BENCH_N")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(2_000_000);
    let keys: Vec<Vec<u8>> = (0..KEYS).map(|i| format!("k{i}").into_bytes()).collect();
    let shards: usize = std::env::var("KEVY_SHARDS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(1);

    // Shared in-memory store (no AOF) — isolate the lock/keyspace from disk.
    let store = Store::open(Config::default().with_shards(shards).with_ttl_reaper_manual()).unwrap();
    for k in &keys {
        store.set(k, VAL).unwrap();
    }

    println!(
        "kevy-embedded MULTI-THREAD throughput — in-memory, shards={shards}, {KEYS} keys, {}B val, n={n_per}/thread",
        VAL.len()
    );
    for &t in &[1usize, 2, 4, 8, 10] {
        run("GET", &store, t, n_per, &keys, false);
    }
    for &t in &[1usize, 2, 4, 8, 10] {
        run("SET", &store, t, n_per, &keys, true);
    }
}

examples/bench_embed.rs (line 56)

fn main() {
    let n: usize = std::env::var("KEVY_BENCH_N")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(2_000_000);
    // Keys precomputed outside the timed loop so `format!`/alloc cost isn't
    // attributed to kevy.
    let keys: Vec<Vec<u8>> = (0..KEYS).map(|i| format!("k{i}").into_bytes()).collect();

    println!("kevy-embedded in-process throughput — single thread, n={n}, {KEYS} keys, {}B val", VAL.len());

    let s1 = Store::open(Config::default().with_ttl_reaper_manual()).unwrap();
    bench("in-memory", &s1, n, &keys);

    let dir2 = std::env::temp_dir().join("kevy_embed_bench_everysec");
    let _ = std::fs::remove_dir_all(&dir2);
    let s2 = Store::open(
        Config::default()
            .with_persist(&dir2)
            .with_ttl_reaper_manual()
            .with_appendfsync(AppendFsync::EverySec),
    )
    .unwrap();
    bench("aof-everysec", &s2, n, &keys);

    let dir3 = std::env::temp_dir().join("kevy_embed_bench_always");
    let _ = std::fs::remove_dir_all(&dir3);
    let s3 = Store::open(
        Config::default()
            .with_persist(&dir3)
            .with_ttl_reaper_manual()
            .with_appendfsync(AppendFsync::Always),
    )
    .unwrap();
    // Always-fsync is one fdatasync per write (no group commit on the embedded
    // single-op path) — fsync-rate-bound, so run far fewer ops to stay bounded.
    bench("aof-always", &s3, (n / 20).max(50_000), &keys);

    // TTL'd-key GET: the mailrs path (every cache key has a TTL). With the
    // background reaper the cached clock is trusted (no per-get Instant::now);
    // manual mode reads a fresh clock per get — the gap is the cached-clock win.
    bench_ttl_get("ttl GET (cached clk)", false, n, &keys); // background reaper
    bench_ttl_get("ttl GET (fresh clk) ", true, n, &keys); // manual reaper

    drop((s1, s2, s3));
    let _ = std::fs::remove_dir_all(&dir2);
    let _ = std::fs::remove_dir_all(&dir3);
}

/// GET throughput over keys that all carry a (long) TTL — the mailrs cache
/// shape. `manual_reaper` toggles whether the store trusts the cached clock
/// (background) or reads a fresh clock per get (manual).
fn bench_ttl_get(label: &str, manual_reaper: bool, n: usize, keys: &[Vec<u8>]) {
    let cfg = if manual_reaper {
        Config::default().with_ttl_reaper_manual()
    } else {
        Config::default() // background reaper (default) → cached clock trusted
    };
    let store = Store::open(cfg).unwrap();
    let ttl = std::time::Duration::from_secs(3600); // never expires during the run
    for k in keys {
        store.set_with_ttl(k, VAL, ttl).unwrap();
    }
    let t = Instant::now();
    let mut hits = 0usize;
    for i in 0..n {
        if store.get(&keys[i % KEYS]).unwrap().is_some() {
            hits += 1;
        }
    }
    std::hint::black_box(hits);
    println!("[{label}] GET {:>10.0} ops/s", n as f64 / t.elapsed().as_secs_f64());
}

pub fn downgrade(&self) -> WeakStore

Get a weak handle that does not keep the keyspace alive.

pub fn config(&self) -> &Config

The active config (a clone — modifying it has no effect on the running store). Useful for introspection / INFO-style telemetry.

pub fn with<F, R>(&self, f: F) -> R

where F: FnOnce(&mut Store) -> R,

Run f against the underlying kevy_store::Store under its lock. Use for direct access to methods this crate hasn’t wrapped. The closure can mutate, but does not auto-log to the AOF — call Self::log yourself if the mutation must survive a crash.

Sharded stores: this targets shard 0 only. Use Self::with_key to reach the shard owning a specific key.

pub fn with_key<F, R>(&self, key: &[u8], f: F) -> R

where F: FnOnce(&mut Store) -> R,

Like Self::with but targets the shard that owns key.

pub fn collect_keys( &self, pattern: Option<&[u8]>, limit: Option<usize>, ) -> Vec<Vec<u8>>

KEYS / SCAN-glob across every shard — the cross-shard replacement for with(|s| s.collect_keys(pat, lim)), which only sees shard 0 once sharding is on. Behaves identically to with(...) when shard_count() == 1. limit bounds the total returned across shards. Takes a read lock per shard (concurrent-safe).

pub fn for_each_shard<F: FnMut(&mut Store)>(&self, f: F)

Run f against each shard’s underlying kevy_store::Store (in shard-index order) — the cross-shard escape hatch. The caller assembles the merged result. Pairs with Self::shard_count. For a single key, prefer Self::with_key; for a glob scan, prefer Self::collect_keys.

pub fn shard_count(&self) -> usize

Number of keyspace shards (== Config::shards).

pub fn log(&self, parts: &[&[u8]]) -> Result<()>

Append a raw RESP-frame argument list to the shard owning its key’s AOF. No-op when persistence is disabled.

pub fn tick(&self) -> ExpireStats

Run one TTL-reaper tick across every shard. Required call cadence in Manual mode (~10×/s to match Redis hz=10). Returns the summed stats.

pub fn rewrite_aof(&self) -> Result<Option<RewriteStats>>

BGREWRITEAOF: rebuild every shard’s AOF from current state. Synchronous. Returns the summed stats (None if persistence is off / no shard rewrote).

pub fn save_snapshot(&self) -> Result<bool>

Snapshot every shard to its dump-{i}.rdb (single shard: the configured name), atomically. Ok(false) when persistence is disabled.

Trait Implementations

impl Clone for Store

fn clone(&self) -> Store

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.