Struct Store 

pub struct Store { /* private fields */ }

A single-database keyspace.

The keyspace map is a KevyMap — a pure-Rust open-addressing Swiss table tuned for kevy’s per-shard, single-trust-domain keyspace. The hasher is kevy_hash::KevyHash (one-call inlinable; no DoS hardening since the shard is single-threaded with no cross-trust keys). Owning the table also exposes bucket addresses for software prefetch on the batch driver.

Implementations

impl Store

pub fn prefetch_for_key(&self, key: &[u8])

Hint the CPU to fetch the bucket cache line for key into L1. Called by the reactor’s parse loop on command N+1 while command N is still being dispatched — by the time N+1 actually probes the table, the metadata line is hot. No-op when the table is empty. Cheap when not.

impl Store

pub fn set_bio_drop_sender(&mut self, sender: Sender<Vec<Value>>)

Install the runtime’s bio-drop channel. Called once from kevy-rt::Runtime::run per shard before the reactor loop starts. After install, Self::maybe_offload_drop (invoked from the SET overwrite fast path) accumulates oversize Values into a per-shard batch; the reactor calls Self::flush_pending_drops at the end of every iter to ship the batch in one mpsc send. Bounds the 10 KB-SET p999/max latency blow-up that synchronous Box::<[u8]>::drop of an allocator large-class slot causes (see kevy_rt::bio).

pub fn flush_pending_drops(&mut self)

Ship the per-shard bio-drop batch buffer to the bio thread in one mpsc send. Called from kevy-rt’s reactor loop at the end of every iteration (both the epoll Shard::run and the io_uring Shard::run_uring paths, just before the AOF fsync window so a pending fsync stall doesn’t pin a batch-ful of heavy values in per-shard memory).

Empty-buffer fast path: zero work, predictable not-taken branch. Reactor calls this unconditionally per iter; the steady- state cost for a no-SET-overwrite iter is one length check.

SendError here means the bio thread has exited (shutdown territory — Runtime::run has dropped its sender AFTER the shard threads joined). Drop the batch inline; the SendError payload carries the Vec back so its Box<Value>s run their Drop here, preserving correctness.

impl Store

pub fn getbit(&mut self, key: &[u8], offset: u64) -> Result<u8, StoreError>

GETBIT key offset — read the bit at offset (MSB-first within each byte, matching Redis). Returns 0 for missing key or offset past the end. Errors on wrong type.

pub fn setbit( &mut self, key: &[u8], offset: u64, value: u8, ) -> Result<u8, StoreError>

SETBIT key offset value — set the bit at offset to value (0 or 1). Extends the underlying string with zero-padding if offset / 8 >= current_len. Returns the PREVIOUS bit value. Errors on wrong type or value > 1.

pub fn bitcount( &mut self, key: &[u8], range: Option<(i64, i64)>, ) -> Result<u64, StoreError>

BITCOUNT key [start end [BYTE|BIT]] — count set bits. start/end are byte offsets (inclusive, negative-from-tail like Redis). None for both = whole string.

pub fn bitpos( &mut self, key: &[u8], bit: u8, range: Option<(i64, i64)>, ) -> Result<Option<u64>, StoreError>

BITPOS key bit [start [end]] — return the position (bit index, MSB-first) of the first bit equal to bit (0 or 1) in the byte range [start, end] (inclusive, Redis-style negative indexing). Returns None (Redis -1) when not found. Errors with OutOfRange if bit > 1.

pub fn getrange( &mut self, key: &[u8], start: i64, end: i64, ) -> Result<Vec<u8>, StoreError>

GETRANGE key start end — substring with Redis-style negative indexing; [start, end] inclusive. Returns empty Vec when key absent or range out of bounds.

pub fn setrange( &mut self, key: &[u8], offset: u64, value: &[u8], ) -> Result<usize, StoreError>

SETRANGE key offset value — overwrite bytes at offset with value. Extends the string with zero padding if offset > len. Returns the new total length. Preserves any existing TTL.

impl Store

pub fn tick_expire( &mut self, samples_per_round: usize, max_rounds: u32, ) -> ExpireStats

Run up to max_rounds of active-expiry sampling against this shard.

Per round: sample samples_per_round TTL-bearing keys at random and drop any whose deadline has passed. Stop early as soon as the in-batch expire-rate drops below 25 % (Redis’s activeExpireCycle continuation threshold) — that’s the signal the keyspace doesn’t have a “thick band” of expired keys to clean up right now.

Cost when there are no TTL-bearing keys at all: one map-emptiness check + a single bucket-iter probe per round. Designed so the active reaper is never a tax on TTL-free workloads.

pub fn expired_keys_total(&self) -> u64

Total keys expired (by lazy reap OR active reaper). Surfaced via INFO keyspace and MEMORY STATS once those grow the field.

impl Store

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

HSET — returns the count of newly-added fields.

pub fn hsetnx( &mut self, key: &[u8], field: &[u8], val: &[u8], ) -> Result<bool, StoreError>

HSETNX — set only if the field is absent; returns whether set.

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

HDEL — returns count removed; deletes the key if emptied.

pub fn hincrbyfloat( &mut self, key: &[u8], field: &[u8], delta: f64, ) -> Result<f64, StoreError>

HINCRBYFLOAT — atomic float increment of a hash field.

pub fn hincrby( &mut self, key: &[u8], field: &[u8], delta: i64, ) -> Result<i64, StoreError>

HINCRBY — preserves TTL; errors if the field isn’t an integer.

impl Store

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

pub fn hexists(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError>

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

pub fn hmget( &mut self, key: &[u8], fields: &[&[u8]], ) -> Result<Vec<Option<Vec<u8>>>, StoreError>

HMGET — one Option per requested field, in input order.

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

HGETALL — flat [field, value, field, value, ...].

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

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

impl Store

pub fn hexpire_at( &mut self, key: &[u8], fields: &[&[u8]], deadline_ms: u64, cond: HExpireCond, ) -> Result<Vec<i8>, StoreError>

Set per-field deadlines (absolute unix-ms). One code per field, request order. Due-or-past deadlines delete the field immediately (code 2, Redis semantics).

pub fn hpttl( &mut self, key: &[u8], fields: &[&[u8]], ) -> Result<Vec<i64>, StoreError>

Remaining TTL per field: -2 key/field missing, -1 no TTL, else remaining ms.

pub fn hpersist( &mut self, key: &[u8], fields: &[&[u8]], ) -> Result<Vec<i8>, StoreError>

Clear per-field TTLs: -2 missing, -1 had no TTL, 1 cleared.

pub fn tick_hash_ttl(&mut self, max_keys: usize) -> Vec<(Vec<u8>, Vec<Vec<u8>>)>

Reaper sweep: remove every due field store-wide; returns (key, removed fields) pairs so the caller logs HDEL effects.

pub fn load_hash_field_ttl( &mut self, key: &[u8], field: &[u8], deadline_ms: u64, )

Snapshot loader hook: restore one field TTL (deadlines already absolute unix-ms; past deadlines simply purge on first access).

pub fn hash_ttl_each<F>(&self, f: F)

where F: FnMut(&[u8], &[u8], u64),

Snapshot support: visit every live (key, field, deadline_ms).

impl Store

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

DEL — returns the count of keys actually removed.

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

EXISTS — count of live keys (duplicates count per occurrence).

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

pub fn expire_at_unix_ms(&mut self, key: &[u8], deadline_ms: u64) -> bool

EXPIREAT/PEXPIREAT semantics: set an absolute wall-clock deadline (Unix epoch millis). This is the persistence-safe form — a deadline survives restart unchanged, unlike the relative Self::expire (whose duration is re-anchored to “now”). A deadline already in the past deletes the key immediately (Redis behaviour). Returns true iff the key existed (and was either re-dated or deleted). The wall-clock → monotonic-Instant conversion happens here so callers persist absolute time but the hot path keeps its cheap monotonic deadline.

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

Cross-shard RENAME step 1: atomically remove the entry at key (if any), returning the (value, ttl_ms_remaining). The orchestrator on the origin shard ships the result into a follow-up Self::put_with_ttl on the destination shard. Lazy-reaps an expired entry before the take (so an expired key is observed as None, not silently rehomed).

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

Clone key’s whole entry — value plus remaining TTL — without removing it. The read half of a transaction snapshot: pair it with Self::put_with_ttl to restore, or with a delete when this returns None (the key did not exist).

Unlike Self::take_with_ttl this leaves the entry in place, so a transaction can record the prior state on first touch and still let the closure read its own writes afterwards.

pub fn put_with_ttl(&mut self, key: Vec<u8>, value: Value, ttl_ms: Option<u64>)

Cross-shard RENAME step 2: write value at key on this shard, overwriting any prior entry. ttl_ms is set as a TTL relative to now (i.e. the orchestrator should have computed the remaining TTL on the source shard via take_with_ttl and is shipping that exact remaining value here).

pub fn key_exists(&mut self, key: &[u8]) -> bool

Whether a live (non-expired) entry exists at key. Reaps an expired entry as a side effect. Used by the cross-shard RENAME orchestrator’s nx pre-check.

pub fn rename(&mut self, src: &[u8], dst: &[u8], nx: bool) -> RenameOutcome

RENAME (or RENAMENX if nx). Atomic on this shard. Returns the outcome so the dispatch layer can emit the right RESP frame (RENAME: +OK or -ERR no such key; RENAMENX: :1/:0/error).

Cross-shard rename is the runtime’s job — by the time this is called, both src and dst are guaranteed to live on the same shard. See kevy-rt::start_rename for the cross-shard split.

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

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

Remaining TTL in ms: -2 no key, -1 no expiry, else >= 0.

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

pub fn dbsize(&self) -> usize

pub fn random_key(&mut self) -> Option<Vec<u8>>

One arbitrary live key, drawn by probing a random slot and walking forward (wrapping once) to the first occupied, unexpired one.

This used to be collect_keys(None, Some(1)) — the first key in hash-bucket order, i.e. the same key every call until it was deleted. O(1) expected, same slight run-length bias as Redis’s dictGetRandomKey, and the contract is “arbitrary”, not “uniform”.

pub fn rand_draw(&mut self) -> u64

One raw draw from the store’s random stream, for callers that need randomness OUTSIDE the store — the RANDOMKEY reducer’s weighted reservoir runs on the origin shard, which must not have to invent its own entropy source to pick between candidates.

pub fn flushall(&mut self)

Wipe every key in this shard’s keyspace (the FLUSHALL/FLUSHDB primitive). Resets used_memory; used_memory_peak is lifetime-cumulative and intentionally not reset.

Named flushall — not flush — to avoid colliding with Write::flush’s “sync buffered writes to disk” meaning. This method DESTROYS data; it does not persist it.

pub fn ttl_pending_count(&self) -> usize

Count live (non-expired) keys that carry a TTL — the size of the “expire set” Redis tracks. Useful as an introspection signal for confirming the TTL subsystem actually registered keys. O(n) over the keyspace; call it for diagnostics, not on the hot path.

pub fn snapshot_each<F>(&self, f: F)

where F: FnMut(&[u8], &Value, Option<u64>),

Visit every live entry as (key, &value, ttl_ms) for snapshotting.

pub fn load_str(&mut self, key: Vec<u8>, value: Vec<u8>, ttl_ms: Option<u64>)

pub fn load_hash( &mut self, key: Vec<u8>, fields: Vec<(Vec<u8>, Vec<u8>)>, ttl_ms: Option<u64>, )

pub fn load_list( &mut self, key: Vec<u8>, items: Vec<Vec<u8>>, ttl_ms: Option<u64>, )

pub fn load_set( &mut self, key: Vec<u8>, members: Vec<Vec<u8>>, ttl_ms: Option<u64>, )

pub fn prefix_stats(&self, prefix: &[u8]) -> (u64, u64)

Count live keys under a byte prefix and how many of them carry a TTL. O(keyspace) — a stats/ops call, not a hot-path primitive.

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

Collect live keys (optionally matching a glob pattern, up to limit). Used by KEYS/SCAN/RANDOMKEY. Treats expired keys as absent (no removal).

pub fn load_zset( &mut self, key: Vec<u8>, pairs: Vec<(Vec<u8>, f64)>, ttl_ms: Option<u64>, )

impl Store

pub fn load_value(&mut self, key: &[u8], value: &Value, ttl_ms: Option<u64>)

Insert one already-typed (key, value, ttl) triple, e.g. straight out of another store’s Self::snapshot_each — the redistribution step both reshard paths (embedded shards bring-up, server routing migration) use to re-home keys after a layout change.

pub fn load_stream( &mut self, key: Vec<u8>, entries: Vec<(u64, u64, Vec<(Vec<u8>, Vec<u8>)>)>, last_id: (u64, u64), max_deleted_id: (u64, u64), entries_added: u64, groups: Vec<LoadedGroup>, ttl_ms: Option<u64>, )

Snapshot-load a stream: every entry plus the per-stream scalar state (last_id, max_deleted_id, entries_added) and the consumer groups are restored verbatim. Caller passes already-decoded primitive tuples; this fn does the SmallBytes / crate::StreamData conversion.

impl Store

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

LPUSH — prepend each value in turn; returns the new length.

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

RPUSH — append each value; returns the new length.

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

LPOP — pop up to count from the head (deleting emptied key).

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

RPOP — pop up to count from the tail.

pub fn lset( &mut self, key: &[u8], idx: i64, val: &[u8], ) -> Result<(), StoreError>

LSET — errors with NoSuchKey / OutOfRange like Redis.

pub fn linsert( &mut self, key: &[u8], before: bool, pivot: &[u8], val: &[u8], ) -> Result<i64, StoreError>

LINSERT key BEFORE|AFTER pivot value — insert value before/after the first occurrence of pivot in the list at key. Returns:

• new list length on success (>= 1);
• 0 when key does not exist;
• -1 when pivot was not found in the list.

Matches Redis semantics.

pub fn lrem( &mut self, key: &[u8], count: i64, val: &[u8], ) -> Result<usize, StoreError>

LREM — remove count occurrences (>0 head, <0 tail, 0 all).

pub fn ltrim( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<(), StoreError>

LTRIM — keep only [start, stop] (deleting emptied key).

impl Store

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

pub fn lindex( &mut self, key: &[u8], idx: i64, ) -> Result<Option<Vec<u8>>, StoreError>

pub fn lrange( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<Vec<Vec<u8>>, StoreError>

impl Store

pub fn set_notify_capture( &mut self, new_key: bool, expired: bool, evicted: bool, )

Choose which store-origin event kinds to capture. The serving layer mirrors its notify-keyspace-events flags here; all-off (the default) reduces every capture hook to one byte test.

pub fn has_notify_events(&self) -> bool

Whether any events are waiting to be drained (one length read).

pub fn has_expired_keys(&self) -> bool

Whether any key has expired since the last drain.

pub fn take_expired_keys(&mut self) -> Vec<Vec<u8>>

Take the keys dropped by expiry since the last drain.

pub fn take_notify_events(&mut self) -> Vec<(KeyspaceEvent, Vec<u8>)>

Take every captured event, in capture order.

impl Store

pub fn scan_page( &self, cursor: u64, count: usize, pattern: Option<&[u8]>, type_filter: Option<&[u8]>, ) -> (u64, Vec<Vec<u8>>, usize)

One SCAN page over this shard: walk from cursor, visiting roughly count buckets (COUNT is a work bound, not a result-size promise — Redis semantics), collecting live keys that pass the optional MATCH glob and TYPE filter.

Returns (next_cursor, keys, buckets_visited). next_cursor == 0 means this shard’s sweep is complete. Expired-but-unreaped keys are treated as absent (no removal — same as Store::collect_keys). type_filter compares case-insensitively against the value’s type name; an unknown name simply never matches (Redis behaviour).

impl Store

pub fn rpoplpush( &mut self, src: &[u8], dst: &[u8], ) -> Result<Option<Vec<u8>>, StoreError>

RPOPLPUSH source destination — atomically pop one element from the tail of src and push it onto the head of dst. Returns the moved element, or None if src was empty / absent.

When src == dst Redis defines the result as a rotation (tail → head of the same list), which falls out of this code naturally because the pop sees the pre-rotation tail.

pub fn lmove( &mut self, src: &[u8], dst: &[u8], from_left: bool, to_left: bool, ) -> Result<Option<Vec<u8>>, StoreError>

LMOVE source destination LEFT|RIGHT LEFT|RIGHT — generalised RPOPLPUSH. from_left=true pops from the head, otherwise the tail; to_left=true pushes to the head, otherwise the tail.

pub fn lpos( &mut self, key: &[u8], element: &[u8], rank: i64, count: Option<i64>, maxlen: usize, ) -> Result<Vec<i64>, StoreError>

LPOS key element [RANK n] [COUNT n] [MAXLEN n] — find the zero-based position(s) of element in the list.

• rank > 0 — scan head→tail, skipping the first rank-1 matches. rank == 1 (default) returns the first match.
• rank < 0 — scan tail→head, returning matches as absolute (head-relative) indices.
• count — None returns the first match as a 1-element vec (caller emits an integer / nil); Some(0) returns all matches; Some(n) caps to n.
• maxlen — 0 means unlimited; otherwise stop after scanning that many elements (in the chosen direction).

Returns the matched indices in scan order. An empty result with count == None is the caller’s signal to emit RESP nil.

impl Store

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

SADD — returns the count of newly-added members.

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

SREM — returns the count removed (deleting an emptied key).

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

SPOP key count — remove and return up to count arbitrary members. Each draw starts at a random slot and takes the first occupied one — O(1) expected, Redis’s dictGetRandomKey shape (sharded sets weight the bucket pick by length first).

impl Store

pub fn sismember( &mut self, key: &[u8], member: &[u8], ) -> Result<bool, StoreError>

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

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

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

SRANDMEMBER key count — up to count DISTINCT arbitrary members, not removed.

Two regimes, as Redis has: when count is a small fraction of the set, probe random slots and reject duplicates — O(count) expected. When it is most of the set, rejection would thrash, so copy the members out and shuffle a prefix instead.

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

SRANDMEMBER key -count — exactly count members, WITH repetition.

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

Snapshot of a set’s members for cross-shard algebra (SINTER/etc.).

impl Store

pub fn collect_snapshot(&self) -> SnapshotView

Freeze a point-in-time SnapshotView of every live entry.

O(n) shallow: per entry one key clone + one Value clone (string bytes copied, collections refcount-bumped) + the TTL resolved to remaining millis. Expired-but-unreaped entries are skipped, matching Store::snapshot_each.

impl Store

pub fn stream_view( &mut self, key: &[u8], ) -> Result<Option<&StreamData>, StoreError>

Read-only access to a stream’s StreamData, used by XINFO to inspect entries / groups / consumers without going through the wrapper layer. Returns Ok(None) for a missing key, WrongType for a non-stream value at key.

pub fn xadd( &mut self, key: &[u8], spec: XAddIdSpec, fields: Vec<(Vec<u8>, Vec<u8>)>, nomkstream: bool, now_ms: u64, ) -> Result<Option<StreamId>, StoreError>

XADD key <spec> field value [field value ...]. Returns the assigned ID. nomkstream matches Redis’s NOMKSTREAM flag — suppress key creation, returning Ok(None). now_ms is the wall-clock used for XAddIdSpec::AutoAll.

pub fn xlen(&mut self, key: &[u8]) -> Result<u64, StoreError>

XLEN key. Returns 0 for a missing key.

pub fn xrange( &mut self, key: &[u8], start: StreamId, end: StreamId, count: Option<usize>, ) -> Result<Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, StoreError>

XRANGE key start end [COUNT n].

pub fn xrevrange( &mut self, key: &[u8], start: StreamId, end: StreamId, count: Option<usize>, ) -> Result<Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, StoreError>

XREVRANGE key end start [COUNT n].

pub fn xread( &mut self, key: &[u8], last_seen: StreamId, count: Option<usize>, ) -> Result<Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, StoreError>

XREAD ... STREAMS key last_seen [...] — per-key part.

pub fn xread_dollar_last_id( &mut self, key: &[u8], ) -> Result<StreamId, StoreError>

Resolve $ as XREAD’s “last-seen” to the stream’s current last ID. Returns MIN for a missing key.

pub fn xdel(&mut self, key: &[u8], ids: &[StreamId]) -> Result<u64, StoreError>

XDEL key id [...]. Returns count actually removed.

pub fn xtrim_maxlen( &mut self, key: &[u8], maxlen: u64, ) -> Result<u64, StoreError>

XTRIM key MAXLEN n. Returns number removed.

pub fn xtrim_minid( &mut self, key: &[u8], minid: StreamId, ) -> Result<u64, StoreError>

XTRIM key MINID id. Returns number removed.

pub fn xsetid( &mut self, key: &[u8], last_id: StreamId, entries_added: Option<u64>, max_deleted_id: Option<StreamId>, ) -> Result<(), StoreError>

XSETID key last-id [ENTRIESADDED n] [MAXDELETEDID id]. Returns NoSuchKey for a missing key (dispatch maps it to Redis’s “requires the key to exist” wording), OutOfRange when last_id is below the stream’s top entry.

pub fn xgroup_create( &mut self, key: &[u8], group: &[u8], mode: GroupCreateMode, mkstream: bool, ) -> Result<bool, StoreError>

XGROUP CREATE key group <id|$> [MKSTREAM]. Returns Ok(true) when a fresh group was added; Ok(false) if the group already existed (caller emits -BUSYGROUP). mkstream matches Redis: auto-create the stream key when missing.

pub fn xgroup_destroy( &mut self, key: &[u8], group: &[u8], ) -> Result<bool, StoreError>

XGROUP DESTROY key group. Returns true if a group was dropped.

pub fn xgroup_setid( &mut self, key: &[u8], group: &[u8], mode: GroupCreateMode, ) -> Result<bool, StoreError>

XGROUP SETID key group <id|$>.

pub fn xgroup_create_consumer( &mut self, key: &[u8], group: &[u8], consumer: &[u8], now_ms: u64, ) -> Result<bool, StoreError>

XGROUP CREATECONSUMER key group consumer.

pub fn xgroup_del_consumer( &mut self, key: &[u8], group: &[u8], consumer: &[u8], ) -> Result<u64, StoreError>

XGROUP DELCONSUMER key group consumer. Returns dropped PEL count.

pub fn xreadgroup( &mut self, key: &[u8], group: &[u8], consumer: &[u8], last_seen: ReadGroupId, count: Option<usize>, noack: bool, now_ms: u64, ) -> Result<Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, StoreError>

XREADGROUP GROUP g c [COUNT n] [NOACK] STREAMS key id.

pub fn xreadgroup_has_new( &mut self, key: &[u8], group: &[u8], ) -> Result<bool, StoreError>

Non-destructive: would XREADGROUP … STREAMS key > yield new entries for group right now? True iff the stream’s last id is past the group’s last-delivered id. Used by the cross-shard BLOCK arbiter’s readiness peek — never advances the group cursor. False for a missing key / group.

pub fn xack( &mut self, key: &[u8], group: &[u8], ids: &[StreamId], ) -> Result<u64, StoreError>

XACK key group id [id ...]. Returns count of PEL removals.

pub fn xpending_summary( &mut self, key: &[u8], group: &[u8], ) -> Result<Option<PendingSummary>, StoreError>

XPENDING key group — summary form.

pub fn xpending_extended( &mut self, key: &[u8], group: &[u8], idle_min_ms: Option<u64>, start: StreamId, end: StreamId, count: usize, consumer_filter: Option<&[u8]>, now_ms: u64, ) -> Result<Option<PendingExtended>, StoreError>

XPENDING key group [IDLE ms] start end count [consumer] — extended form.

pub fn xclaim( &mut self, key: &[u8], group: &[u8], new_owner: &[u8], ids: &[StreamId], opts: &XClaimOpts, now_ms: u64, ) -> Result<Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, StoreError>

XCLAIM key group consumer min-idle-ms id [id ...] [...]. Returns the (id, field-value) pairs successfully claimed — dispatcher trims to ID-only when JUSTID is set.

pub fn xautoclaim( &mut self, key: &[u8], group: &[u8], new_owner: &[u8], min_idle_ms: u64, start: StreamId, count: usize, justid: bool, now_ms: u64, ) -> Result<(StreamId, Vec<(StreamId, Vec<(Vec<u8>, Vec<u8>)>)>, Vec<StreamId>), StoreError>

XAUTOCLAIM key group consumer min-idle-ms start [COUNT n] [JUSTID]. Returns the cursor + claimed payloads + deleted IDs.

impl Store

pub fn get_for_reply( &mut self, key: &[u8], ) -> Result<Option<GetReply<'_>>, StoreError>

GET variant that exposes the underlying encoding so the reactor’s reply path can choose zero-copy (Value::ArcBulk → push the Arc to the conn’s output_arcs for a writev iovec) vs memcpy (Value::Str / Value::Int → encode bytes into the conn’s output Vec). ONE keyspace lookup; the variant tag chooses the encoding without a second probe.

pub fn get_shared_owned( &self, key: &[u8], ) -> Result<Option<GetShared>, StoreError>

Owned GET for the FFI scalar shared lane (kevy_get_shared). Bulk values (Value::ArcBulk) return an Arc::clone — no byte copy; the FFI holds the Arc alive and hands JS a buffer that views it directly (mirrors MMKV’s zero-copy mmap-page view, the thing that made kevy lose large GET). Small values (Str/Int) allocate a fresh Arc<Box<[u8]>> — the same one copy the Vec lane already pays — so the caller’s free path is uniform. Read-only (&self, like Self::get_shared) so the FFI can take it under a SHARED shard lock — no LRU stamp, matching the maxmemory == 0 fast path the mobile door runs on. Wrong type errors like Self::get_for_reply.

pub fn get_into_output( &mut self, key: &[u8], output: &mut Vec<u8>, output_arcs: &mut Vec<(usize, Arc<Box<[u8]>>)>, ) -> Result<bool, StoreError>

Fused GET-into-output. Skips the GetReply enum tag round-trip + caller match arm by writing the RESP frame directly into output (header + bytes + CRLF for Str/Int) or pushing the Arc into output_arcs at the right offset (ArcBulk zero-copy via writev). Returns the same outcomes as Self::get_for_reply: Ok(true) if the key was found and emitted, Ok(false) if absent (the caller emits the $-1 null bulk — preserves the existing inline-null semantics on the reactor side), Err for WRONGTYPE.

pub fn get(&mut self, key: &[u8]) -> Result<Option<Cow<'_, [u8]>>, StoreError>

GET — returns a Cow<[u8]> so Value::Int callers can format the integer to ASCII without storing it. Value::Str returns Cow::Borrowed (zero copy); Value::Int formats to a small owned Vec<u8> (up to 20 bytes for i64::MIN).

pub fn get_shared( &self, key: &[u8], ) -> Result<Option<Cow<'_, [u8]>>, StoreError>

Read-only GET: &self, so concurrent readers can run under a shared lock (embedded mode’s RwLock read path). Expiry is checked against the coarse cached clock but an expired key is not removed here (no &mut) — the reaper / next write reclaims it; a reader just sees None. LRU is not touched, so this path is only used when eviction is off (maxmemory == 0); with eviction, the caller takes the mutating Self::get under an exclusive lock so access still stamps the LRU.

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

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

INCRBY family; preserves any TTL.

Following valkey’s OBJ_ENCODING_INT approach: the hot path matches Value::Int(n) and does the increment in place — no parse, no format, no allocation. The Value::Str arm parses, increments, and promotes to Value::Int(next) so subsequent INCRs land on the fast path. Insert-new path also lands as Int.

impl Store

pub fn append(&mut self, key: &[u8], data: &[u8]) -> Result<usize, StoreError>

pub fn getset( &mut self, key: &[u8], val: Vec<u8>, ) -> Result<Option<Vec<u8>>, StoreError>

GETSET — set to val, return the previous string (WRONGTYPE if the old value isn’t a string). Clears any TTL, like SET.

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

GETDEL — get then delete (WRONGTYPE if non-string).

pub fn incr_by_float( &mut self, key: &[u8], delta: f64, ) -> Result<Vec<u8>, StoreError>

INCRBYFLOAT — returns the new value formatted as Redis would. Preserves TTL.

impl Store

pub fn set( &mut self, key: &[u8], value: Vec<u8>, expire: Option<Duration>, nx: bool, xx: bool, ) -> bool

SET — overwrites any existing value/type. NX/XX guards; clears TTL. Takes an owned Vec so a >22 B value’s allocation is adopted as-is (no copy). For callers holding a borrowed slice, prefer Self::set_slice — it skips the to_vec entirely for values that inline.

pub fn set_slice( &mut self, key: &[u8], value: &[u8], expire: Option<Duration>, nx: bool, xx: bool, ) -> bool

Self::set for a borrowed value. Values ≤ 22 B store inline in the entry — zero allocator traffic, where set(key, value.to_vec(), …) paid a malloc for the Vec and a free when the inline copy dropped it (the dominant overwrite-SET pattern). Larger values pay the same single allocation either way.

impl Store

pub fn enable_seg_rows(&mut self, dir: &Path) -> Result<(), String>

Turn row segments on for this shard, rooted at dir, loading every manifest-registered row segment from the previous run — they are truth, and the AOF’s SEGMENTED frames (or a stub snapshot) will reference them by seq. Idempotent.

pub fn sweep_orphan_row_segs(&mut self)

After replay: rebuild each loaded segment’s live count from the stubs that actually reference it, and sweep the segments nothing references — a crash between sealing and the SEGMENTED frame leaves exactly such an orphan (its rows replayed hot).

pub fn seal_rows_to_seg( &mut self, table: &[u8], keys: &[Vec<u8>], ) -> Result<Option<SealedRows>, String>

The two-phase producer face: seal the batch (durable half) and return (seq, file) for the caller to log a SEGMENTED frame BEFORE Store::commit_row_eviction phase-changes the rows — the R2c ordering (frame after the durable copy, before the hot deletion) that makes every crash gap recoverable.

pub fn commit_row_eviction(&mut self, sealed: &SealedRows) -> u64

Phase-change the sealed batch after its SEGMENTED frame is logged.

pub fn load_row_stub(&mut self, key: Vec<u8>, seq: u32, value_weight: u32)

Load one snapshot stub record: the row’s identity re-enters the map as a seg-backed stub (the segment directory, loaded before the snapshot, holds its data). TTL-free by the eviction filter.

pub fn row_seg_files(&self) -> Vec<(u32, String)>

The live row segments’ (seq, file) identities — the rewrite’s trailing SEGMENTED frames name these.

impl Store

pub fn enable_tiering(&mut self, dir: &Path, budget: u64) -> Result<(), Error>

Turn tiering on: open (wiping) the vlog under dir and set the RAM budget the demotion watermark works against. Callers own the dir choice (<data>/tier/ by convention).

pub fn set_tier_budget(&mut self, bytes: u64)

Live-update the tiering budget (auto/percent re-resolution on the shard tick, CONFIG SET — the maxmemory reapply precedent). Touches nothing but the number: the vlog, the stubs and every counter stay as they are. No-op when tiering is off.

pub fn set_tier_max_spill(&mut self, bytes: u64)

Cap the largest spillable value (0 = unlimited). Embedded sets 256 KiB by default (RFC §7) to bound cold-read lock-hold time; the server leaves it unlimited. No-op when tiering is off.

pub fn set_tier_reserved(&mut self, bytes: u64)

Feed the index/view memory floor (Σ segment approx_bytes on this shard) into the unified watermark. Called per shard tick by the serving layer. No-op when tiering is off.

pub fn tier_index_floor_blocked(&self, extra: u64) -> bool

Whether the index/view floor (reserved_bytes + extra) already exhausts the tier’s demotable headroom — the IDX.CREATE refusal predicate (RFC §4 row 16). false when tiering is off.

pub fn tier_enabled(&self) -> bool

Whether tiering is on for this shard.

pub fn tier_stats(&self) -> TierStats

Tiering gauges — zeros when tiering is off.

pub fn tier_pins(&self) -> Vec<Arc<VlogFile>>

Pin every current vlog file (view pinning): a snapshot view / rewrite plan captured from a tiered store carries these so its frozen ColdRefs stay readable on the serializer thread across compaction — a retired file is unlinked only when the last pin drops. Empty when tiering is off.

pub fn materialize_cold(&self, key: &[u8], v: &Value) -> Option<Value>

Serialization-side cold materialization: decode v’s record into a fresh owned hot value WITHOUT installing, promoting, or setting the probation mark — persistence is a bulk path and never promotes. None when v is hot.

impl Store

pub fn try_demote_after_write(&mut self) -> usize

The demotion twin of Store::try_evict_after_write, called beside it from the write-commit sites. No-op unless tiering is on AND used_memory is past the unified target (the plain watermark minus the index/view floor and the stub floor); then spills at most one batch (a single write never funds an unbounded spill storm — continuation rides Store::demote_step on the tick). Returns keys demoted.

pub fn demote_step(&mut self) -> usize

Tick continuation of Store::try_demote_after_write: one more budgeted batch per shard tick while over the watermark — with backoff. A tick whose batch moves nothing while over target (every spillable value already cold, or the floor alone exceeds the budget so effective_target == 0) doubles the tick’s skip up to [BACKOFF_CEILING_TICKS]; any demotion — here or on the write path — resets it. During a backoff window this is one decrement: the sampler does not run. “Idempotent is not convergent”: before this, an over-target store with nothing left to spill re-walked the sample window every tick forever.

pub fn demote_to_watermark(&mut self) -> usize

Bulk-load drain: demote batch after batch until the store is back under the watermark or candidates run dry. Replay / snapshot-load / reshard call this every K applied frames — those paths are single-threaded, so draining more than one write-path batch per check is safe (there is no reactor to stall). The sampler runs UNBOUNDED here: a fixed visit window keeps a stale start position between calls (the access clock does not advance mid-drain), so a window that has gone all-cold would end the drain while still over the watermark — ending under the watermark is this path’s hard contract. Returns total keys demoted.

pub fn tier_compact_tick(&mut self) -> usize

Reactor-tick compaction: one bounded step while a sealed file is below the live threshold. Cheap no-op (an O(files) scan) when there is nothing to compact. Returns records processed.

impl Store

pub fn peek_scope<R>(&mut self, f: impl FnOnce(&mut Store) -> R) -> R

Run f in bulk-read (no-promote peek) mode: every cold materializing read inside serves via pread WITHOUT setting the probation mark and WITHOUT promoting. The whole-value peek for digest / scope-move / export sweeps — a bulk sweep must never thrash the hot tier.

pub fn peek_hash_fields( &mut self, key: &[u8], fields: &[&[u8]], ) -> Result<Option<Vec<Option<Vec<u8>>>>, StoreError>

Row peek: fields of the hash at key, without promotion and without advancing the touched gate. A hot row reads as hmget does; a COLD hash stub costs ONE record read + ONE decode for all fields. Ok(None) = missing key; Err(WrongType) = non-hash (zero preads when cold — the stage-1 tag answers).

pub fn peek_hash_rows( &mut self, keys: &[&[u8]], fields: &[&[u8]], reader: &mut dyn ColdBatchReader, ) -> Vec<Result<Option<Vec<Option<Vec<u8>>>>, StoreError>>

Page peek: keys × fields with every cold row coalesced — sorted by (file_id, offset) — into ONE ColdBatchReader batch, decoded once per row, results in input order. Per-key result mirrors Store::peek_hash_fields. No promotion, no gate advancement; hot rows read exactly as hmget does.

impl Store

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

ZADD — returns the count of newly-added members. Borrowed argv: no per-member allocation; routes through the encoding-switch path.

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

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

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

ZREM — returns the count of members removed.

pub fn zrank( &mut self, key: &[u8], member: &[u8], ) -> Result<Option<usize>, StoreError>

ZRANK — 0-based position in ascending order. O(log N): a hash lookup for the score, then one order-statistic tree descent.

pub fn zincrby( &mut self, key: &[u8], incr: f64, member: &[u8], ) -> Result<f64, StoreError>

ZINCRBY — add incr to a member’s score; returns the new score.

impl Store

pub fn zset_or_set_members( &mut self, key: &[u8], ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

Extract one source key’s scored members for the algebra ops: zsets as-is, sets with score 1.0 (Redis semantics), absent key = empty, any other type = WrongType.

pub fn zstore_result(&mut self, dst: &[u8], pairs: &[(Vec<u8>, f64)]) -> usize

Materialize an algebra result at dst: existing value (any type) is dropped, result written as a zset — Redis *STORE overwrite semantics. Empty result deletes dst (Redis drops the destination rather than storing an empty zset). Returns the stored cardinality.

impl Store

pub fn zrange( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

ZRANGE key start stop by rank.

pub fn zrange_by_score( &mut self, key: &[u8], min: ScoreBound, max: ScoreBound, ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

ZRANGEBYSCORE.

pub fn zcount( &mut self, key: &[u8], min: ScoreBound, max: ScoreBound, ) -> Result<usize, StoreError>

ZCOUNT.

pub fn zpopmin( &mut self, key: &[u8], count: usize, ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

ZPOPMIN key [count] — pop and return the count lowest-scored members (ascending by (score, member)). Returns (member, score) pairs in pop order; empty when the key is absent / empty.

pub fn zpopmin_below( &mut self, key: &[u8], below: f64, count: usize, ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

zpopmin_below — pop up to count lowest-scored members whose score is < below (strictly). The delayed-job primitive: “pop everything that is due” in one atomic call (score = due time, below = now). Absent key = empty; wrong type errors.

pub fn zrem_range_by_rank( &mut self, key: &[u8], start: i64, stop: i64, ) -> Result<usize, StoreError>

ZREMRANGEBYRANK key start stop — remove members in the rank range [start, stop] (inclusive, negative indices count from the tail). Returns the number of members removed.

pub fn zrem_range_by_score( &mut self, key: &[u8], min: ScoreBound, max: ScoreBound, ) -> Result<usize, StoreError>

ZREMRANGEBYSCORE key min max — remove every member whose score satisfies min ≤ score ≤ max (with ( for exclusive bounds via ScoreBound). Returns the number removed.

pub fn zrev_range_by_score( &mut self, key: &[u8], min: ScoreBound, max: ScoreBound, ) -> Result<Vec<(Vec<u8>, f64)>, StoreError>

ZREVRANGEBYSCORE — zrange_by_score reversed. Bounds are passed in the (min, max) order already (the caller is responsible for swapping the user-facing max first, min second at the dispatch layer).

impl Store

pub fn zadd_flags( &mut self, key: &[u8], pairs: &[(f64, &[u8])], flags: ZaddFlags, ) -> Result<ZaddReport, StoreError>

Flags-aware ZADD. Caller validates ZaddFlags::valid at its input boundary (RESP parse / typed API) — invalid combos here are a caller bug.

pub fn zadd_incr( &mut self, key: &[u8], delta: f64, member: &[u8], flags: ZaddFlags, ) -> Result<Option<f64>, StoreError>

ZADD … INCR — a conditional ZINCRBY: returns the new score, or None when the flags veto the operation (Redis replies nil).

impl Store

pub fn new() -> Store

pub fn refresh_clock(&mut self)

Refresh the coarse cached clock (Self::cached_ns) from a single Instant::now(). Call once per reactor-loop batch / reaper tick; the per-access read path then skips its own clock read. Lazy expiry is coarse to this cadence (a key expires ≤ one refresh-interval late, never early — writes stamp deadlines from a fresh clock).

pub fn set_cached_clock(&mut self, on: bool)

Enable/disable trusting the cached clock for lazy expiry (see Self::cached_ns). Call with true only when something refreshes the clock regularly (the server reactor per batch, the embedded background reaper per tick); leave false for manual-reaper mode. Seeds the cache when enabling so the first access is accurate.

pub fn set_max_memory(&mut self, maxmemory: u64, policy: EvictionPolicy)

Install (or clear, with maxmemory == 0) the eviction limit and policy. Cheap; safe to call repeatedly (e.g. on CONFIG SET).

pub fn used_memory(&self) -> u64

Live byte estimate (see field doc).

pub fn used_memory_peak(&self) -> u64

used_memory high-water mark since startup.

pub fn maxmemory(&self) -> u64

Configured maxmemory (0 = unlimited).

pub fn eviction_policy(&self) -> EvictionPolicy

Configured eviction policy.

pub fn evictions_total(&self) -> u64

Total keys evicted since startup.

pub fn expires_count(&self) -> usize

Live keys carrying a TTL (INFO keyspace’s expires=). O(1) — reads the maintained counter, not an O(n) scan (cf. Self::ttl_pending_count).

pub fn record_watch(&mut self, key: &[u8]) -> u64

WATCH — record this key in the version tracker and return its current version. Subsequent writes on this shard bump the version via Self::bump_if_watched. Caller (the conn’s origin shard) stores the returned version; EXEC later asks every owning shard “is the version still N?” via Self::key_version.

Keys that have never been written stay at version 0 — the first write after a WATCH bumps to 1, which is what makes the “dirty” comparison work (stored 0 ≠ current 1 ⇒ abort EXEC).

pub fn key_version(&self, key: &[u8]) -> u64

Read-only version lookup used by EXEC’s pre-execution check. Returns 0 for keys never WATCH-ed (matches the initial value record_watch would have inserted, so a WATCH → no-write → EXEC sequence sees the stored 0 == current 0 and proceeds).

pub fn bump_if_watched(&mut self, key: &[u8])

Bump the version of key if (and only if) it has been WATCH-ed at least once. Write-side call after every mutation. The empty check runs BEFORE the key is hashed — the common nothing-watched case pays one branch, not a guaranteed-miss probe.

pub fn bump_all_watched(&mut self)

Invalidate every watched key in one shot. Called from FLUSHDB / FLUSHALL execution paths — every WATCH against this shard must invalidate so a pending EXEC aborts.

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

Cached weight of key (dynamic part + ENTRY_OVERHEAD). Returns None when the key is absent or expired (no implicit reap).

pub fn precheck_for_write(&self) -> Result<(), StoreError>

O(1) precondition check the dispatch layer calls before every write command. Returns Err(OutOfMemory) only when maxmemory > 0, the budget is already over, AND the policy is NoEviction (Redis behaviour). All other policies let the write proceed and recover via Self::try_evict_after_write.

pub fn try_evict_after_write(&mut self) -> usize

Run after every write command. No-op when disabled or under budget; otherwise samples per Self::eviction_policy and removes keys until back under maxmemory or no eligible candidate remains. Returns the number of keys evicted (0 on the common fast path).

Trait Implementations

impl Default for Store

fn default() -> Store

Returns the “default value” for a type.

impl SnapshotSource for Store

fn for_each_entry(&self, f: impl FnMut(&[u8], &Value, Option<u64>))

Visit every live entry as (key, &value, remaining_ttl_ms).

fn for_each_hash_ttl(&self, f: impl FnMut(&[u8], &[u8], u64))

Visit every live hash field TTL as (key, field, absolute_unix_ms). Default = none (sources without the feature).

fn row_seg_files(&self) -> Vec<(u32, String)>

The live row segments’ (seq, file) identities — the AOF rewrite’s trailing SEGMENTED frames and the snapshot writer’s version choice read these. Default = none.