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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
//! v2.3 CDC consumer surface — embedded half (RFC 2026-07-04, LOCKED,
//! D7). One stream per store: the embedded write path already
//! serializes every shard's mutations through `commit_write`, so the
//! feed is a single `(generation, offset)` stream and
//! [`Store::feed_shards`] reports 1 (server-parity consumer loops work
//! unchanged; they just see one shard).
//!
//! Persistence: with a `data_dir`, the generation contract rides the
//! same `feed-0.gen` / `feed-0.meta` sidecars the server uses (clean
//! close keeps the cursor, crash or FLUSHALL bumps). Without
//! persistence the store's data dies with the process anyway — each
//! open starts a fresh generation-1 stream, which is exactly what the
//! (empty) restored state implies.
use std::io;
use std::sync::{Arc, Mutex};
use kevy_replicate::feed::{FeedRead, FeedSource};
use crate::store::Store;
/// One mutation delivered by [`Store::changes_since`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Change {
/// Stream offset (monotonic within a generation).
pub offset: u64,
/// The applied effect's argv (same frames the AOF / a replica sees).
pub argv: Vec<Vec<u8>>,
}
/// A batch of changes plus the cursor to resume from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChangeBatch {
/// Delivered changes, offset order.
pub changes: Vec<Change>,
/// `(generation, offset)` to pass to the next `changes_since`.
pub next: (u64, u64),
}
/// Why a feed read could not be served.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeedError {
/// Cursor unservable (stale generation / evicted offsets): rebuild
/// from a scan, then resume from `tail`.
Resync {
/// Current generation.
generation: u64,
/// Resume offset.
tail: u64,
},
/// Cursor is ahead of the stream — caller bug.
Future,
/// The store was opened without `Config::with_feed`.
Disabled,
}
/// Multi-key / keyless verbs the fail-open prefix filter never drops
/// (their key layout isn't argv[1], or they touch everything).
const FILTER_DENYLIST: &[&[u8]] = &[
b"DEL", b"UNLINK", b"MSET", b"COPY", b"RENAME", b"FLUSHALL", b"BITOP",
b"SINTERSTORE", b"SUNIONSTORE", b"SDIFFSTORE",
b"ZINTERSTORE", b"ZUNIONSTORE", b"ZDIFFSTORE",
];
fn matches_prefixes(argv: &[Vec<u8>], prefixes: &[&[u8]]) -> bool {
if prefixes.is_empty() {
return true;
}
let Some(verb) = argv.first() else { return true };
if FILTER_DENYLIST.iter().any(|d| verb.eq_ignore_ascii_case(d)) {
return true; // fail-open: over-delivery is free, drops are not
}
match argv.get(1) {
Some(key) => prefixes.iter().any(|p| key.starts_with(p)),
None => true,
}
}
/// Per-prefix keyspace stats from [`Store::info_prefix`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PrefixInfo {
/// Live keys under the prefix.
pub keys: u64,
/// How many of them carry a TTL.
pub expires: u64,
}
impl Store {
/// v2.3 `info_prefix`: count live keys (and TTL'd keys) under a
/// byte prefix, across all shards. O(keyspace) — an ops/stats
/// call, not a hot-path primitive.
pub fn info_prefix(&self, prefix: &[u8]) -> PrefixInfo {
let mut keys = 0u64;
let mut expires = 0u64;
for shard in self.shards.iter() {
let g = crate::store::lock_read(shard);
let (k, e) = g.store.prefix_stats(prefix);
keys += k;
expires += e;
}
PrefixInfo { keys, expires }
}
/// Number of independent change streams this store exposes (the
/// embedded write path serializes all shards: always 1).
pub fn feed_shards(&self) -> usize {
1
}
/// The current `(generation, next_offset)` cursor — where a
/// consumer starting fresh (or resuming after a rebuild) begins.
pub fn changes_tail(&self) -> Result<(u64, u64), FeedError> {
let feed = self.feed_handle().ok_or(FeedError::Disabled)?;
let g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
Ok(g.tail())
}
/// Deliver up to `limit` changes at cursor `(generation, offset)`,
/// optionally prefix-filtered (fail-open on multi-key verbs; the
/// filter never affects the returned cursor). At-least-once: after
/// a `Resync` rebuild, frames already applied may be seen again.
pub fn changes_since(
&self,
generation: u64,
offset: u64,
limit: usize,
prefixes: &[&[u8]],
) -> Result<ChangeBatch, FeedError> {
let feed = self.feed_handle().ok_or(FeedError::Disabled)?;
let g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let frames = match g.read(generation, offset, limit.clamp(1, 65536)) {
Ok(v) => v,
Err(FeedRead::Resync { generation, tail }) => {
return Err(FeedError::Resync { generation, tail });
}
Err(FeedRead::Future) => return Err(FeedError::Future),
};
let next_off = frames.last().map_or(offset, |f| f.offset + 1);
let mut changes = Vec::with_capacity(frames.len());
for f in &frames {
let Ok((foff, argv, _)) = kevy_replicate::wire::decode_frame(f.bytes) else {
continue;
};
let owned: Vec<Vec<u8>> = (0..argv.len()).map(|i| argv[i].to_vec()).collect();
if !matches_prefixes(&owned, prefixes) {
continue;
}
changes.push(Change { offset: foff, argv: owned });
}
Ok(ChangeBatch { changes, next: (g.generation(), next_off) })
}
/// v2.3 feed hooks used by `commit_write` / `flushall` / close —
/// `None` unless the store was opened with feed enabled.
pub(crate) fn feed_handle(&self) -> Option<&Arc<Mutex<FeedSource>>> {
self.feed.as_ref()
}
/// Break stream continuity on FLUSHALL: bump + persist the
/// generation high-water (mirrors the server's exec_op Flush arm).
pub(crate) fn feed_bump_on_flush(&self) {
let Some(feed) = self.feed_handle() else { return };
let mut g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
g.bump_generation();
if let Some(dir) = &self.config.data_dir
&& let Err(e) = kevy_persist::feed_meta::write_feed_gen(dir, 0, g.generation())
{
eprintln!("kevy-embedded: feed gen write failed: {e}");
}
}
/// Clean-close half of the continuity contract (called from the
/// DropGuard after the AOF flush).
pub(crate) fn feed_write_close_marker(shards_feed: &Arc<Mutex<FeedSource>>, dir: &std::path::Path) {
let g = shards_feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let (generation, next) = g.tail();
if let Err(e) = kevy_persist::feed_meta::write_feed_meta(dir, 0, generation, next) {
eprintln!("kevy-embedded: feed marker write failed: {e}");
}
}
/// Push one applied effect into the feed (called from
/// `commit_write` alongside the AOF append).
pub(crate) fn feed_push(feed: &Arc<Mutex<FeedSource>>, parts: &[&[u8]]) {
let mut g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mut argv = kevy_resp::Argv::default();
for p in parts {
argv.push(p);
}
let _ = g.source_mut().push_mutation(&argv);
}
/// Feed boot half for `Store::open` — resolve the cursor via the
/// sidecar decision table when persistent, else a fresh gen-1.
pub(crate) fn feed_open(
config: &crate::config::Config,
) -> io::Result<Option<Arc<Mutex<FeedSource>>>> {
if !config.feed_enabled {
return Ok(None);
}
let budget = usize::try_from(config.feed_buffer_size).unwrap_or(usize::MAX);
let (generation, next_offset) = match &config.data_dir {
Some(dir) => {
let b = kevy_persist::feed_meta::load_feed_boot(dir, 0)?;
(b.generation, b.next_offset)
}
None => (1, 0),
};
let mut src = kevy_replicate::source::ReplicationSource::new(budget);
src.set_next_offset(next_offset);
Ok(Some(Arc::new(Mutex::new(FeedSource::new(generation, src)))))
}
}