kevy_embedded/
ops_feed.rs

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

kevy_embedded/
ops_feed.rs
