kevy_embedded/
ops_feed.rs

1//! CDC consumer surface — embedded half.
2//! One stream per store: the embedded write path already
3//! serializes every shard's mutations through `commit_write`, so the
4//! feed is a single `(generation, offset)` stream and
5//! [`Store::feed_shards`] reports 1 (server-parity consumer loops work
6//! unchanged; they just see one shard).
7//!
8//! Persistence: with a `data_dir`, the generation contract rides the
9//! same `feed-0.gen` / `feed-0.meta` sidecars the server uses (clean
10//! close keeps the cursor, crash or FLUSHALL bumps). Without
11//! persistence the store's data dies with the process anyway — each
12//! open starts a fresh generation-1 stream, which is exactly what the
13//! (empty) restored state implies.
14
15use crate::KevyResult;
16use std::sync::{Arc, Mutex};
17
18use kevy_replicate::feed::{FeedRead, FeedSource};
19
20use crate::store::Store;
21
22/// One mutation delivered by [`Store::changes_since`].
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct Change {
25    /// Stream offset (monotonic within a generation).
26    pub offset: u64,
27    /// The applied effect's argv (same frames the AOF / a replica sees).
28    pub argv: Vec<Vec<u8>>,
29}
30
31/// A batch of changes plus the cursor to resume from.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct ChangeBatch {
34    /// Delivered changes, offset order.
35    pub changes: Vec<Change>,
36    /// `(generation, offset)` to pass to the next `changes_since`.
37    pub next: (u64, u64),
38}
39
40/// Why a feed read could not be served.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum FeedError {
43    /// Cursor unservable (stale generation / evicted offsets): rebuild
44    /// from a scan, then resume from `tail`.
45    Resync {
46        /// Current generation.
47        generation: u64,
48        /// Resume offset.
49        tail: u64,
50    },
51    /// Cursor is ahead of the stream — caller bug.
52    Future,
53    /// The store was opened without `Config::with_feed`.
54    Disabled,
55}
56
57/// Multi-key / keyless verbs the fail-open prefix filter never drops
58/// (their key layout isn't argv[1], or they touch everything).
59const FILTER_DENYLIST: &[&[u8]] = &[
60    b"DEL",
61    b"UNLINK",
62    b"MSET",
63    b"COPY",
64    b"RENAME",
65    b"FLUSHALL",
66    b"BITOP",
67    b"SINTERSTORE",
68    b"SUNIONSTORE",
69    b"SDIFFSTORE",
70    b"ZINTERSTORE",
71    b"ZUNIONSTORE",
72    b"ZDIFFSTORE",
73];
74
75fn matches_prefixes(argv: &[Vec<u8>], prefixes: &[&[u8]]) -> bool {
76    if prefixes.is_empty() {
77        return true;
78    }
79    let Some(verb) = argv.first() else { return true };
80    if FILTER_DENYLIST.iter().any(|d| verb.eq_ignore_ascii_case(d)) {
81        return true; // fail-open: over-delivery is free, drops are not
82    }
83    match argv.get(1) {
84        Some(key) => prefixes.iter().any(|p| key.starts_with(p)),
85        None => true,
86    }
87}
88
89/// Per-prefix keyspace stats from [`Store::info_prefix`].
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub struct PrefixInfo {
92    /// Live keys under the prefix.
93    pub keys: u64,
94    /// How many of them carry a TTL.
95    pub expires: u64,
96}
97
98impl Store {
99    /// `info_prefix`: count live keys (and TTL'd keys) under a
100    /// byte prefix, across all shards. O(keyspace) — an ops/stats
101    /// call, not a hot-path primitive.
102    pub fn info_prefix(&self, prefix: &[u8]) -> PrefixInfo {
103        let mut keys = 0u64;
104        let mut expires = 0u64;
105        for shard in self.shards.iter() {
106            let g = crate::store::lock_read(shard);
107            let (k, e) = g.store.prefix_stats(prefix);
108            keys += k;
109            expires += e;
110        }
111        PrefixInfo { keys, expires }
112    }
113
114    /// Number of independent change streams this store exposes (the
115    /// embedded write path serializes all shards: always 1).
116    pub fn feed_shards(&self) -> usize {
117        1
118    }
119
120    /// The current `(generation, next_offset)` cursor — where a
121    /// consumer starting fresh (or resuming after a rebuild) begins.
122    pub fn changes_tail(&self) -> Result<(u64, u64), FeedError> {
123        let feed = self.feed_handle().ok_or(FeedError::Disabled)?;
124        let g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
125        Ok(g.tail())
126    }
127
128    /// Deliver up to `limit` changes at cursor `(generation, offset)`,
129    /// optionally prefix-filtered (fail-open on multi-key verbs; the
130    /// filter never affects the returned cursor). At-least-once: after
131    /// a `Resync` rebuild, frames already applied may be seen again.
132    pub fn changes_since(
133        &self,
134        generation: u64,
135        offset: u64,
136        limit: usize,
137        prefixes: &[&[u8]],
138    ) -> Result<ChangeBatch, FeedError> {
139        let feed = self.feed_handle().ok_or(FeedError::Disabled)?;
140        let g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
141        let frames = match g.read(generation, offset, limit.clamp(1, 65536)) {
142            Ok(v) => v,
143            Err(FeedRead::Resync { generation, tail }) => {
144                return Err(FeedError::Resync { generation, tail });
145            }
146            Err(FeedRead::Future) => return Err(FeedError::Future),
147        };
148        let next_off = frames.last().map_or(offset, |f| f.offset + 1);
149        let mut changes = Vec::with_capacity(frames.len());
150        for f in &frames {
151            let Ok((foff, argv, _)) = kevy_replicate::wire::decode_frame(f.bytes) else {
152                continue;
153            };
154            let owned: Vec<Vec<u8>> = (0..argv.len()).map(|i| argv[i].to_vec()).collect();
155            if !matches_prefixes(&owned, prefixes) {
156                continue;
157            }
158            changes.push(Change { offset: foff, argv: owned });
159        }
160        Ok(ChangeBatch { changes, next: (g.generation(), next_off) })
161    }
162
163    /// Feed hooks used by `commit_write` / `flushall` / close —
164    /// `None` unless the store was opened with feed enabled.
165    pub(crate) fn feed_handle(&self) -> Option<&Arc<Mutex<FeedSource>>> {
166        self.feed.as_ref()
167    }
168
169    /// Break stream continuity on FLUSHALL: bump + persist the
170    /// generation high-water (mirrors the server's exec_op Flush arm).
171    pub(crate) fn feed_bump_on_flush(&self) {
172        let Some(feed) = self.feed_handle() else { return };
173        let mut g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
174        g.bump_generation();
175        if let Some(dir) = &self.config.data_dir
176            && let Err(e) = kevy_persist::feed_meta::write_feed_gen(dir, 0, g.generation())
177        {
178            eprintln!("kevy-embedded: feed gen write failed: {e}");
179        }
180    }
181
182    /// Clean-close half of the continuity contract (called from the
183    /// DropGuard after the AOF flush).
184    pub(crate) fn feed_write_close_marker(
185        shards_feed: &Arc<Mutex<FeedSource>>,
186        dir: &std::path::Path,
187    ) {
188        let g = shards_feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
189        let (generation, next) = g.tail();
190        if let Err(e) = kevy_persist::feed_meta::write_feed_meta(dir, 0, generation, next) {
191            eprintln!("kevy-embedded: feed marker write failed: {e}");
192        }
193    }
194
195    /// Push one applied effect into the feed (called from
196    /// `commit_write` alongside the AOF append).
197    pub(crate) fn feed_push(feed: &Arc<Mutex<FeedSource>>, parts: &[&[u8]]) {
198        let mut g = feed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
199        let mut argv = kevy_resp::Argv::default();
200        for p in parts {
201            argv.push(p);
202        }
203        let _ = g.source_mut().push_mutation(&argv);
204    }
205
206    /// Feed boot half for `Store::open` — resolve the cursor via the
207    /// sidecar decision table when persistent, else a fresh gen-1.
208    pub(crate) fn feed_open(
209        config: &crate::config::Config,
210    ) -> KevyResult<Option<Arc<Mutex<FeedSource>>>> {
211        if !config.feed_enabled {
212            return Ok(None);
213        }
214        let budget = usize::try_from(config.feed_buffer_size).unwrap_or(usize::MAX);
215        let (generation, next_offset) = match &config.data_dir {
216            Some(dir) => {
217                let b = kevy_persist::feed_meta::load_feed_boot(dir, 0)?;
218                (b.generation, b.next_offset)
219            }
220            None => (1, 0),
221        };
222        let mut src = kevy_replicate::source::ReplicationSource::new(budget);
223        src.set_next_offset(next_offset);
224        Ok(Some(Arc::new(Mutex::new(FeedSource::new(generation, src)))))
225    }
226}
kevy_embedded/ops_feed.rs

kevy_embedded/
ops_feed.rs
