Trait Commands 

pub trait Commands:
    Clone
    + Send
    + 'static {
    // Required methods
    fn route<A: ArgvView + ?Sized>(&self, args: &A) -> Route;
    fn dispatch<A: ArgvView + ?Sized>(
        &self,
        store: &mut Store,
        args: &A,
    ) -> Vec<u8> ;
    fn is_quit<A: ArgvView + ?Sized>(&self, args: &A) -> bool;
    fn is_write<A: ArgvView + ?Sized>(&self, args: &A) -> bool;
    fn txn_kind<A: ArgvView + ?Sized>(&self, args: &A) -> TxnKind;

    // Provided methods
    fn dispatch_into<A: ArgvView + ?Sized>(
        &self,
        store: &mut Store,
        args: &A,
        out: &mut Vec<u8>,
    ) { ... }
    fn dispatch_into_resp3<A: ArgvView + ?Sized>(
        &self,
        store: &mut Store,
        args: &A,
        out: &mut Vec<u8>,
    ) { ... }
    fn notify_class<A: ArgvView + ?Sized>(
        &self,
        _args: &A,
    ) -> Option<NotifyClass> { ... }
    fn hello_reply<A: ArgvView + ?Sized>(
        &self,
        _args: &A,
        current_proto: RespVersion,
    ) -> (RespVersion, Vec<u8>) { ... }
    fn on_shard_init(&self, _store: &mut Store) { ... }
    fn on_shard_start(&self, _shard: usize) { ... }
    fn on_data_dir(&self, _dir: &Path) { ... }
    fn on_persist_stats(&self, _in_flight: bool, _aof_rewrites_total: u64) { ... }
    fn on_tick_gap(&self, _excess_us: u64) { ... }
    fn on_query_buffer_exceeded(&self) { ... }
    fn on_aof_format(&self, _format: u8) { ... }
    fn on_replay_report(&self, _dropped_bytes: u64, _corrupt: bool) { ... }
    fn on_conn_gauge(&self, _live: u64) { ... }
    fn on_replication_view(
        &self,
        _master_repl_offset: u64,
        _replicas: Vec<ReplicaViewRow>,
    ) { ... }
    fn on_shard_tick(&self, _store: &mut Store) { ... }
    fn shutdown_save_requested(&self) -> bool { ... }
    fn extension_op(&self, _store: &mut Store, _argv: &[Vec<u8>]) -> Vec<u8>  { ... }
    fn geo_search(&self, _store: &mut Store, _argv: &[Vec<u8>]) -> GeoHits { ... }
    fn write_denied(&self) -> Option<Vec<u8>> { ... }
    fn read_denied<A: ArgvView + ?Sized>(&self, _args: &A) -> Option<Vec<u8>> { ... }
    fn extension_reduce(
        &self,
        _argv: &[Vec<u8>],
        _chunks: Vec<Vec<u8>>,
        _proto: RespVersion,
    ) -> ExtensionReduced { ... }
    fn on_write(&self, _store: &mut Store, _key: &[u8]) { ... }
    fn on_flush(&self, _store: &mut Store) { ... }
    fn on_command(&self) { ... }
    fn on_connection(&self) { ... }
    fn shard_tick_interval_ms(&self) -> u64 { ... }
    fn live_runtime_config(&self) -> LiveRuntimeConfig { ... }
    fn block_hint<A: ArgvView + ?Sized>(&self, _args: &A) -> BlockHint { ... }
    fn resolve_block_argv<A: ArgvView + ?Sized>(
        &self,
        _store: &mut Store,
        args: &A,
        _kind: BlockKind,
    ) -> Argv { ... }
    fn block_serve_argv<A: ArgvView + ?Sized>(
        &self,
        args: &A,
        _kind: BlockKind,
        _key: &[u8],
    ) -> Argv { ... }
    fn block_restore_argv(
        &self,
        _store: &mut Store,
        _kind: BlockKind,
        _key: &[u8],
    ) -> Option<Argv> { ... }
    fn block_ready<A: ArgvView + ?Sized>(
        &self,
        _store: &mut Store,
        _serve_argv: &A,
        _kind: BlockKind,
    ) -> bool { ... }
    fn queue_error<A: ArgvView + ?Sized>(&self, _args: &A) -> Option<Vec<u8>> { ... }
    fn resolve<A: ArgvView + ?Sized>(&self, args: &A) -> ResolvedCmd { ... }
}

Command-set semantics injected into the runtime. Cloned to every core, so it must be cheap/stateless to clone.

Required Methods

fn route<A: ArgvView + ?Sized>(&self, args: &A) -> Route

Classify how a command is routed across shards.

fn dispatch<A: ArgvView + ?Sized>(&self, store: &mut Store, args: &A) -> Vec<u8>

Execute a full command against one shard’s store, returning RESP bytes.

fn is_quit<A: ArgvView + ?Sized>(&self, args: &A) -> bool

Whether this command should close the connection (QUIT).

fn is_write<A: ArgvView + ?Sized>(&self, args: &A) -> bool

Whether this command mutates the keyspace (so it must be logged to the AOF).

fn txn_kind<A: ArgvView + ?Sized>(&self, args: &A) -> TxnKind

Transaction-control classification (MULTI/EXEC/DISCARD vs anything else).

Provided Methods

fn dispatch_into<A: ArgvView + ?Sized>( &self, store: &mut Store, args: &A, out: &mut Vec<u8>, )

Execute a command, appending the RESP reply to out. The in-order local fast path uses this to write straight into the connection’s output buffer (no per-command reply Vec). Default: delegate to dispatch.

fn dispatch_into_resp3<A: ArgvView + ?Sized>( &self, store: &mut Store, args: &A, out: &mut Vec<u8>, )

RESP3 variant of Self::dispatch_into — called when the connection has negotiated HELLO 3. Default: delegate to the RESP2 path (so a server that hasn’t migrated any replies still works correctly with a RESP3 client, per spec). Override per command to emit RESP3 shapes (Map / Set / Double / …).

fn notify_class<A: ArgvView + ?Sized>(&self, _args: &A) -> Option<NotifyClass>

Classify a command for keyspace notifications. Returns Some for write commands that should fire a notification when the corresponding flag is enabled; None for read-only / no-op / not-yet-classified commands (those never publish). Default None so non-kevy embedders pay nothing.

fn hello_reply<A: ArgvView + ?Sized>( &self, _args: &A, current_proto: RespVersion, ) -> (RespVersion, Vec<u8>)

Handle HELLO — return the new connection protocol version + the reply bytes. The runtime applies the new version to the conn before scheduling the reply, so a HELLO 3 ack itself comes out shaped as a RESP3 Map (the new protocol is in effect for its own reply).

Default: ignore the args, keep current_proto, emit a minimal RESP2 +OK so embedders that don’t care still see a sane reply. kevy’s own impl in kevy::KevyCommands parses the optional protover and emits the full server-info shape.

fn on_shard_init(&self, _store: &mut Store)

Called once per shard, immediately after Store::new, before the reactor enters its event loop. Implementations install per-shard configuration that the runtime doesn’t know about — currently the maxmemory + eviction-policy pair, which kevy ships via its own process-wide config snapshot. Default: no-op so non-kevy embedders aren’t forced to override.

fn on_shard_start(&self, _shard: usize)

Called once on the shard’s own thread, first thing in the reactor entry (both reactors), before restore/replay. Implementations that need per-shard identity at dispatch time (e.g. kevy’s CLUSTER MYID / CLUSTER NODES myself flag) stash shard in a thread-local here — in a thread-per-core runtime the current thread is the shard. Default: no-op.

fn on_data_dir(&self, _dir: &Path)

The directory this runtime snapshots to and loads from — the one Runtime::builder().with_data_dir() set.

A Commands implementation carries its own configuration, and nothing told it about this. A server built programmatically therefore answered CONFIG GET dir from that configuration while writing somewhere else entirely: one face reporting what the other face is not doing. kevy::serve builds both from one Config and never saw the gap, which is why it went unnoticed until a test used CONFIG GET dir to identify its own server and was handed ..

Called once per shard, on the shard’s thread, beside Self::on_shard_start. Default: no-op, so an implementor that has no configuration to correct is unaffected:

use kevy_rt::{ArgvView, Commands, Route, Store, TxnKind};
use std::path::Path;

#[derive(Clone)]
struct Minimal;
impl Commands for Minimal {
    fn route<A: ArgvView + ?Sized>(&self, _a: &A) -> Route { Route::Local }
    fn dispatch<A: ArgvView + ?Sized>(&self, _s: &mut Store, _a: &A) -> Vec<u8> {
        b"+OK\r\n".to_vec()
    }
    fn is_quit<A: ArgvView + ?Sized>(&self, _a: &A) -> bool { false }
    fn is_write<A: ArgvView + ?Sized>(&self, _a: &A) -> bool { false }
    fn txn_kind<A: ArgvView + ?Sized>(&self, _a: &A) -> TxnKind { TxnKind::Other }
}

Minimal.on_data_dir(Path::new("/var/lib/kevy"));

An implementor that answers CONFIG GET dir overrides it and points that answer here; kevy’s own does exactly that.

fn on_persist_stats(&self, _in_flight: bool, _aof_rewrites_total: u64)

Per-tick persistence-stats publication: whether this shard has a background save/rewrite in flight and how many AOF rewrites have completed since open. Command layers that serve INFO persistence stash these in a thread-local (thread-per-core: the answering thread is the shard, same pattern as Self::on_shard_start). Default: no-op.

fn on_tick_gap(&self, _excess_us: u64)

The shard tick fired excess_us microseconds later than its interval asked — the reactor’s own stall gauge (a long-blocking iteration delays the tick by exactly its overrun). Called at tick cadence (10 Hz), so implementations may do real work.

fn on_query_buffer_exceeded(&self)

A connection was closed because its accumulated unparsed input crossed the query-buffer cap.

The enforcement path printed a line and marked the conn closing, and there was nothing a test or an operator could ask about it — so an intermittent “the server did not close” could not be told from “the server decided and the close had not landed yet”. Those are different defects. Redis exposes the same count as client_query_buffer_limit_disconnections.

Called on the closing decision, not on the close completing. That is the whole point of the distinction: a decision that has not reached the client yet is a different thing from a cap that was never noticed, and only a counter taken here can tell them apart.

The default does nothing, so an existing implementor gains the hook without changing:

use kevy_rt::{ArgvView, Commands, Route, Store, TxnKind};

#[derive(Clone)]
struct Minimal;
impl Commands for Minimal {
    fn route<A: ArgvView + ?Sized>(&self, _a: &A) -> Route { Route::Local }
    fn dispatch<A: ArgvView + ?Sized>(&self, _s: &mut Store, _a: &A) -> Vec<u8> {
        b"+OK\r\n".to_vec()
    }
    fn is_quit<A: ArgvView + ?Sized>(&self, _a: &A) -> bool { false }
    fn is_write<A: ArgvView + ?Sized>(&self, _a: &A) -> bool { false }
    fn txn_kind<A: ArgvView + ?Sized>(&self, _a: &A) -> TxnKind { TxnKind::Other }
}

// The hook is optional; the default is a no-op.
Minimal.on_query_buffer_exceeded();

An implementor that wants the number overrides it and counts; kevy’s own does exactly that, and INFO stats reports the total as client_query_buffer_limit_disconnections.

fn on_aof_format(&self, _format: u8)

Per-tick AOF on-disk format gauge (the embedder ask’s server twin): 0 = AOF off, 1 = a pre-4.0 v1 file still being appended (a 3.x binary swap-back still works), 2 = v2. Follows Self::on_persist_stats’s shard-gauge pattern.

fn on_replay_report(&self, _dropped_bytes: u64, _corrupt: bool)

One-shot boot-replay verdict for this shard: bytes dropped past the last replayable AOF frame (quarantined + truncated by the repair) and whether the stop was a corrupt frame. Fires once, after the shard’s startup replay, before the listener accepts. Non-zero drops mean the shard recovered less than its file held — command layers surface it via INFO persistence so operators can alert on it. Default: no-op.

fn on_conn_gauge(&self, _live: u64)

Per-tick live-connection gauge: how many client conns this shard currently holds (cluster-bus links excluded). Command layers publish it to their cross-shard stats slots so INFO connected_clients sums a real instance-wide value. Default: no-op.

fn on_replication_view( &self, _master_repl_offset: u64, _replicas: Vec<ReplicaViewRow>, )

Per-tick replication-view publication: the answering shard’s current master_repl_offset (== ReplicationSource::next_offset()) plus a ReplicaViewRow for every handshake-complete replica conn (in AckSent, Streaming, or SnapshotShipping); the row’s ack is None until the replica’s first REPLCONF ACK. Only called when this shard has a ReplicationSource installed (i.e. Runtime::with_replication(true, ...) was requested); standalone setups pay nothing. Command layers that serve ROLE / INFO replication stash the values in a thread-local (thread-per-core: the answering thread is the shard, same pattern as Self::on_persist_stats) and may additionally publish them to a shared slot for cross-shard aggregation. Default no-op.

fn on_shard_tick(&self, _store: &mut Store)

Periodic shard housekeeping (the equivalent of Redis’s serverCron). kevy uses this to run Store::tick_expire at the configured [expiry].hz. Default no-op so non-kevy embedders / runtimes can ignore it.

fn shutdown_save_requested(&self) -> bool

Polled once per shard as it leaves the reactor loop: true when the operator requested a final snapshot before exit (SHUTDOWN SAVE). The shard then runs one background save and drains it before the process exits. Default false — plain stops (SIGTERM, bare SHUTDOWN) drain in-flight persistence but don’t force a new snapshot.

fn extension_op(&self, _store: &mut Store, _argv: &[Vec<u8>]) -> Vec<u8>

Per-shard half of an extension fan-out command (IDX.* / future VIEW.* / FT.*): compute this shard’s raw chunk for argv. The payload encoding is the embedder’s own — the runtime treats it as opaque bytes and hands all chunks to Commands::extension_reduce at the origin.

fn geo_search(&self, _store: &mut Store, _argv: &[Vec<u8>]) -> GeoHits

Search half of a geo *STORE (GEOSEARCHSTORE / GEORADIUS…STORE), run on the SOURCE key’s shard: match argv’s query against the source zset and return the (member, score) pairs to write — the scores already in their final form (geohash, or the STOREDIST distance in the unit the command asked for). The runtime writes them at the destination’s own shard; see [crate::exec_geostore]. A command set that doesn’t route Route::GeoStore never sees this call.

fn write_denied(&self) -> Option<Vec<u8>>

Pre-dispatch write gate. Some(err_bytes) rejects every data-write client command with that RESP error before any routing (replication apply does NOT pass through here, so a read-only replica keeps applying its feed). Admin verbs (REPLICAOF / CONFIG) are not classified as writes and stay available as the operator escape hatch. Default: writes always allowed.

fn read_denied<A: ArgvView + ?Sized>(&self, _args: &A) -> Option<Vec<u8>>

Read-availability gate: called before READ verbs; return Some(error_bytes) to refuse the read (a replica whose feed is staler than the configured bound answers -STALE; one mid-way through a full-resync snapshot load answers -LOADING). args lets implementations exempt health-check verbs (PING) from the refusal. Default: reads always allowed.

fn extension_reduce( &self, _argv: &[Vec<u8>], _chunks: Vec<Vec<u8>>, _proto: RespVersion, ) -> ExtensionReduced

Origin-side reduce of an extension fan-out — merge every shard’s chunk (produced by Self::extension_op) into either the final RESP reply or a follow-up fan-out argv (see ExtensionReduced). proto is the requesting connection’s negotiated RESP version so proto-aware reduces can shape the reply (Map vs pair-array).

fn on_write(&self, _store: &mut Store, _key: &[u8])

Called after every applied write with the written key (when the resolver knew one). Default no-op; kevy uses it for synchronous secondary-index maintenance (derived-by- construction). Runs on the shard thread with store access — implementations must be cheap when their feature is off.

fn on_flush(&self, _store: &mut Store)

Keyspace-wide invalidation hook: called after FLUSHALL/FLUSHDB has emptied this shard’s store (both the client path and the replica apply path execute the same op). Synchronous index maintenance resets its derived structures here — a flushed keyspace must not keep answering from stale index entries.

fn on_command(&self)

Called once per client command at dispatch entry (before routing / fan-out, so a multi-key command counts once). kevy uses it for INFO stats: total_commands_processed. Hot path — keep it to a single thread-local bump. Default no-op so non-kevy embedders pay nothing.

fn on_connection(&self)

Called once per accepted client connection. kevy uses it for INFO stats: total_connections_received. Default no-op.

fn shard_tick_interval_ms(&self) -> u64

Interval between Self::on_shard_tick calls. Default 100 ms (matching Redis’s hz = 10). 0 disables ticking entirely.

fn live_runtime_config(&self) -> LiveRuntimeConfig

Snapshot of the runtime-owned knobs that can be hot-modified (the kevy server wires this to CONFIG SET). Called once per shard tick — each Some value is applied to the shard’s live state; each None keeps the existing setting untouched.

Default returns all-None so embedders that never hot-swap config pay nothing beyond one struct-build per tick. The cost lives in the impl’s read of its own config source.

fn block_hint<A: ArgvView + ?Sized>(&self, _args: &A) -> BlockHint

Classify a command for blocking semantics. BlockHint::None (default) is the zero-cost answer for every non-blocking verb; the dispatcher only registers a waiter when this returns BlockHint::Block and the command’s dispatch_into produced no reply (i.e. it could not satisfy itself immediately — e.g. BLPOP on an empty list). Concrete impls should fold this into their override of Self::resolve so the verb-table lookup happens once per command.

fn resolve_block_argv<A: ArgvView + ?Sized>( &self, _store: &mut Store, args: &A, _kind: BlockKind, ) -> Argv

Rewrite args into the owned Argv that the dispatcher will store as the parked waiter’s command and replay on wake. Lets a command set normalise positional ID / cursor arguments that would otherwise re-resolve to a different value on retry — most notably XREAD BLOCK ... STREAMS k $, where leaving $ literal in the retried argv causes a fresh re-resolve to the post-XADD last_id and zero matching entries (the wake hangs).

Default: just materialise the argv unchanged. Concrete impls only need to override when a registered command carries an arg whose meaning depends on store state at park time (XREAD $, the classic case).

For the cross-shard arbiter this runs on the target shard (the one that owns the key) when the waiter is armed, so $ snapshots the target’s real last_id — not the origin shard’s (which may not hold the stream at all).

fn block_serve_argv<A: ArgvView + ?Sized>( &self, args: &A, _kind: BlockKind, _key: &[u8], ) -> Argv

Build the single-key command the dispatcher will replay to satisfy one watched key of a (possibly multi-key) blocking command. args is the original command; key is one of its watched keys. Returns an Argv that, when dispatched, pops / reads only key — e.g. BLPOP k1 k2 0 watching k2 yields BLPOP k2 0; XREAD … STREAMS s1 s2 id1 id2 watching s2 yields XREAD … STREAMS s2 id2.

Any state-dependent positional arg ($) is left literal here — it’s frozen later by Self::resolve_block_argv on the key’s owning shard. No store access needed (pure argv slicing). Default: the unchanged argv (single-key blocking commands need no rewrite).

fn block_restore_argv( &self, _store: &mut Store, _kind: BlockKind, _key: &[u8], ) -> Option<Argv>

The command that would put back whatever replaying serve_argv is about to consume — read from the store before the serve runs.

A cross-shard serve pops on the target and ships the reply to the origin. If the origin’s client disconnected in that window the reply has nowhere to go, and the element would be lost: taken from the list, delivered to nobody. The origin cannot put it back (it holds a RESP frame whose shape differs per kind and per negotiated protocol), so the target captures the undo first and holds it until the origin confirms delivery.

Read, not parse: the peek runs on the owning shard immediately before the pop with nothing interleaved, so what it saw is what the pop takes, in RESP2 and RESP3 alike.

None = nothing to undo. That is the honest answer for kinds that consume nothing (XREAD is non-destructive) and the safe default for an embedder that has not implemented it.

fn block_ready<A: ArgvView + ?Sized>( &self, _store: &mut Store, _serve_argv: &A, _kind: BlockKind, ) -> bool

Non-destructive readiness peek for a parked waiter: would replaying serve_argv (built by Self::block_serve_argv, $ already frozen) produce a reply right now? Runs on the key’s owning shard when arming and is the gate for emitting a cross-shard wake. Must NOT mutate the store (no pop / no group-cursor advance). Default false so non-blocking embedders never spuriously wake.

fn queue_error<A: ArgvView + ?Sized>(&self, _args: &A) -> Option<Vec<u8>>

Validate a command being queued inside MULTI. Returns an error reply (already RESP-encoded, e.g. -ERR unknown command …) when the command cannot be queued — an unknown verb or an arity mismatch — in which case the caller answers with it instead of +QUEUED and marks the transaction dirty so EXEC aborts with -EXECABORT. None means “queue it”. Default None keeps embedders that don’t model a verb table permissive.

fn resolve<A: ArgvView + ?Sized>(&self, args: &A) -> ResolvedCmd

Resolve all verb-dependent attributes in one verb-table lookup. The default implementation calls the per-attribute methods above (five upper_verb scans + matches); concrete impls SHOULD override this with a single match so the reactor’s hot path pays the verb- resolution cost only once per command.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors