1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
//! The public entry point: configure and run the thread-per-core server.
use crate::Commands;
use kevy_persist::Fsync;
use std::path::PathBuf;
/// Default slots in each per-core-pair SPSC ring. A full ring spills
/// to a local backlog (see [`Shard`]), so this only bounds the
/// lock-free fast path, not capacity. Overridable via the
/// `[advanced] ring_capacity` config field threaded through
/// [`Runtime::with_advanced`].
const DEFAULT_RING_CAPACITY: usize = 1024;
/// The public entry point: configure and run the thread-per-core server.
pub struct Runtime<C: Commands> {
pub(crate) ip: [u8; 4],
pub(crate) port: u16,
pub(crate) nshards: usize,
pub(crate) commands: C,
/// Directory for per-shard snapshot files (`dump-<id>.rdb`) and AOF logs.
pub(crate) data_dir: PathBuf,
/// Whether the append-only log is enabled.
pub(crate) enable_aof: bool,
/// fsync policy for the AOF. Default `EverySec` matches Redis.
pub(crate) appendfsync: Fsync,
/// auto-trigger BGREWRITEAOF when AOF grew this many % above the size
/// at the previous rewrite. `0` disables. Default `100` (matches Redis).
pub(crate) auto_aof_rewrite_pct: u32,
/// Floor below which auto-rewrite is skipped. Default `64 MiB`.
pub(crate) auto_aof_rewrite_min_size: u64,
/// Reactor SPSC ring slot count. See [`DEFAULT_RING_CAPACITY`].
pub(crate) ring_capacity: usize,
/// Reactor busy-poll iter limit before parking. Stored as `u32`
/// for the per-shard counter; the [`Shard`] field carries it
/// forward into the loop.
pub(crate) spin_limit: u32,
/// **v1.30** — `Some(N)` = only shards `0..N` arm accept SQE. `None`
/// = every shard accepts (v1.29 byte-identical).
pub(crate) accept_shards: Option<usize>,
/// **v1.37** — total cap on active client conns. `0` = unlimited.
pub(crate) max_clients: usize,
/// Reactor blocking-wait timeout in ms when parked.
pub(crate) park_timeout_ms: u32,
/// Wall-clock-read throttle for the tick check (TTL reaper / live
/// config refresh / auto-AOF-rewrite).
pub(crate) tick_check_every: u32,
/// `[slowlog].slower_than_micros`. Default: `-1` (OFF — zero
/// hot-path cost: every command would otherwise pay an
/// `Instant::now()` pair around dispatch). Set to `10_000` to match
/// Redis's default 10 ms threshold; see [`Self::with_slowlog`] /
/// `CONFIG SET slowlog-log-slower-than 10000`.
pub(crate) slowlog_slower_than_micros: i64,
/// `[slowlog].max_len`. Per-shard cap.
pub(crate) slowlog_max_len: u32,
/// Single-node cluster mode: slot-based key routing (CRC16 `{hashtag}`
/// → contiguous ranges) + one deterministic extra listener per shard at
/// `cluster_port_base + id`. `None` = off (default, zero change).
pub(crate) cluster_port_base: Option<u16>,
/// v3-cluster replication: when `true`, each shard runs a
/// `ReplicationSource` with `replication_buffer_size` byte budget;
/// every applied mutation is pushed to the backlog. The TCP
/// listener + streaming loop arrive in subsequent tasks (T1.12+);
/// this batch only wires the producer side. Default `false`.
pub(crate) enable_replication: bool,
/// v2.3: FEED.* consumer surface. When set, every shard keeps a
/// backlog (even with no replicas) and persists the (generation,
/// offset) cursor via the feed sidecars.
pub(crate) feed_enabled: bool,
/// Per-shard backlog byte budget for the feed (`[feed]
/// feed_buffer_size`); the effective budget is
/// `max(replication_buffer_size, feed_buffer_size)` when both
/// features are on (one backlog, two readers).
pub(crate) feed_buffer_size: u64,
/// Per-shard backlog byte budget when `enable_replication` is set.
/// Fed from `[replication] replication_buffer_size`. Default
/// `256 MiB` (matches the kevy-config default).
pub(crate) replication_buffer_size: u64,
/// v3-cluster replication listener: shard `i` binds at
/// `replication_port_base + i` (mirrors cluster listener pattern;
/// per Issue Ledger I2). `None` = no listener (producer side runs
/// without a network surface, backlog accumulates and evicts —
/// useful for benchmarks). Default `None`.
pub(crate) replication_port_base: Option<u16>,
/// Per-shard SlotTable reconnect-window in ms (T1.15). After a
/// streaming replica disconnects, its `(replica_id, sent_offset)`
/// is recorded in the shard's `slots` map; slots past this age
/// are reaped on the next shard tick. Default `60_000` (60 s)
/// matches the kevy-config default.
pub(crate) replication_reconnect_window_ms: u32,
/// Per-shard replica inboxes installed by
/// [`Self::with_replica_inboxes`]. Each entry is consumed
/// (via `Option::take`) when its shard is constructed, so the
/// receiver flows from this Vec to the matching `Shard.replica_inbox`.
/// Empty when no replica mode is configured.
pub(crate) replica_inboxes: Vec<Option<crate::replica_inbox::ReplicaInboxReceiver>>,
/// v1.25 UDS: when `Some(path)`, ALSO bind a Unix-domain stream
/// listener at `path` on shard 0 (single global socket, like valkey's
/// `unixsocket` config). Lets benches/local clients skip TCP loopback
/// overhead. TCP listener stays bound regardless.
#[allow(dead_code)] // consumed during run() via take-into-Shard
pub(crate) unix_socket_path: Option<PathBuf>,
}
impl<C: Commands> Runtime<C> {
#[must_use]
pub fn new(ip: [u8; 4], port: u16, nshards: usize, commands: C) -> Self {
Runtime {
ip,
port,
nshards: nshards.max(1),
commands,
data_dir: PathBuf::from("."),
enable_aof: true,
appendfsync: Fsync::EverySec,
auto_aof_rewrite_pct: 100,
auto_aof_rewrite_min_size: 64 * 1024 * 1024,
ring_capacity: DEFAULT_RING_CAPACITY,
spin_limit: 256,
accept_shards: None,
max_clients: 10_000,
park_timeout_ms: 50,
tick_check_every: 256,
slowlog_slower_than_micros: -1,
slowlog_max_len: 128,
cluster_port_base: None,
enable_replication: false,
feed_enabled: false,
feed_buffer_size: 64 * 1024 * 1024,
replica_inboxes: Vec::new(),
replication_buffer_size: 256 * 1024 * 1024,
replication_port_base: None,
replication_reconnect_window_ms: 60_000,
unix_socket_path: None,
}
}
/// Spawn one thread per shard and run until `stop` is set.
/// v1.25 UDS: also bind a Unix-domain stream listener at `path`. Lets
/// local clients (and benchmarks) skip the TCP loopback round-trip.
/// Bound on shard 0 only (no SO_REUSEPORT for AF_UNIX, single global
/// socket like valkey's `unixsocket` config). TCP listener stays
/// bound at the configured `port` regardless.
#[must_use]
pub fn with_unix_socket(mut self, path: PathBuf) -> Self {
self.unix_socket_path = Some(path);
self
}
// `run` (and its per-stage build helpers) lives in
// [`crate::runtime_run`] — same `impl<C: Commands> Runtime<C>`,
// split out so this file stays under the 500-LOC house rule.
}