kevy_client/
feed.rs

1//! Change-feed (CDC): `FEED.SHARDS` / `FEED.TAIL` / `FEED.READ`
2//! — the network face of kevy-embedded's `changes_since`.
3//!
4//! Both backends serve the same cursor contract: read from
5//! `(generation, offset)`, get frames plus the next cursor; an
6//! unservable cursor (stale generation / evicted offsets) surfaces as
7//! an error whose message starts with `FEEDRESYNC <gen> <tail>` —
8//! rebuild from a scan, then resume from that cursor. The embedded
9//! backend is single-shard (shard `0`) and requires the store to be
10//! opened with `Config::with_feed`; without it, calls answer
11//! `Unsupported`.
12
13use crate::{KevyError, KevyResult};
14
15use kevy_embedded::FeedError;
16use kevy_resp::Reply;
17use kevy_resp_client::RespClient;
18
19use crate::{Connection, string, unexpected};
20
21/// One change frame from [`Connection::feed_read`].
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct FeedFrame {
24    /// Stream offset (monotonic within a generation).
25    pub offset: u64,
26    /// The applied effect's argv (the same frame the AOF / a replica
27    /// sees), e.g. `["SET", "k", "v"]`.
28    pub argv: Vec<Vec<u8>>,
29}
30
31/// A batch of change frames plus the cursor to resume from.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct FeedBatch {
34    /// The stream's current generation.
35    pub generation: u64,
36    /// Offset to pass to the next [`Connection::feed_read`].
37    pub next_offset: u64,
38    /// Delivered frames, offset order. May be empty (caught up).
39    pub frames: Vec<FeedFrame>,
40}
41
42impl Connection {
43    /// `FEED.SHARDS` — number of change-feed shards (embedded: always 1).
44    pub fn feed_shards(&mut self) -> KevyResult<usize> {
45        match self {
46            Self::Embedded(s) => Ok(s.feed_shards()),
47            Self::Remote(c) => match c.request_borrowed(&[b"FEED.SHARDS"])? {
48                Reply::Int(n) if n >= 0 => Ok(n as usize),
49                Reply::Error(e) => Err(KevyError::Protocol(string(e))),
50                other => Err(unexpected(other)),
51            },
52        }
53    }
54
55    /// `FEED.TAIL shard` — the shard's current `(generation,
56    /// next_offset)` cursor: where a consumer starting fresh (or
57    /// resuming after a rebuild) begins.
58    pub fn feed_tail(&mut self, shard: usize) -> KevyResult<(u64, u64)> {
59        match self {
60            Self::Embedded(s) => {
61                check_embedded_shard(shard)?;
62                s.changes_tail().map_err(feed_err)
63            }
64            Self::Remote(c) => {
65                let sh = shard.to_string();
66                match c.request_borrowed(&[b"FEED.TAIL", sh.as_bytes()])? {
67                    Reply::Array(items) if items.len() == 2 => {
68                        let mut it = items.into_iter();
69                        match (it.next().unwrap(), it.next().unwrap()) {
70                            (Reply::Int(g), Reply::Int(o)) => Ok((g as u64, o as u64)),
71                            (a, _) => Err(unexpected(a)),
72                        }
73                    }
74                    Reply::Error(e) => Err(KevyError::Protocol(string(e))),
75                    other => Err(unexpected(other)),
76                }
77            }
78        }
79    }
80
81    /// `FEED.READ shard generation offset [COUNT n] [PREFIX p …]` —
82    /// deliver up to `count` frames (server default 256) past the
83    /// cursor, optionally key-prefix-filtered (fail-open: frames whose
84    /// key layout the filter can't cheaply determine are always
85    /// delivered). Resume from `(batch.generation, batch.next_offset)`.
86    pub fn feed_read(
87        &mut self,
88        shard: usize,
89        generation: u64,
90        offset: u64,
91        count: Option<usize>,
92        prefixes: &[&[u8]],
93    ) -> KevyResult<FeedBatch> {
94        match self {
95            Self::Embedded(s) => {
96                check_embedded_shard(shard)?;
97                let batch = s
98                    .changes_since(generation, offset, count.unwrap_or(256), prefixes)
99                    .map_err(feed_err)?;
100                Ok(FeedBatch {
101                    generation: batch.next.0,
102                    next_offset: batch.next.1,
103                    frames: batch
104                        .changes
105                        .into_iter()
106                        .map(|ch| FeedFrame { offset: ch.offset, argv: ch.argv })
107                        .collect(),
108                })
109            }
110            Self::Remote(c) => {
111                parse_batch(feed_read_request(c, shard, generation, offset, count, prefixes)?)
112            }
113        }
114    }
115}
116
117/// Embedded feed is single-shard; any other index is a caller bug.
118fn check_embedded_shard(shard: usize) -> KevyResult<()> {
119    if shard != 0 {
120        return Err(KevyError::InvalidInput("embedded feed is single-shard: shard must be 0".into()));
121    }
122    Ok(())
123}
124
125/// Map [`FeedError`] onto the same error text the wire produces, so
126/// resync handling code is backend-agnostic.
127fn feed_err(e: FeedError) -> KevyError {
128    match e {
129        FeedError::Resync { generation, tail } => {
130            KevyError::Protocol(format!("FEEDRESYNC {generation} {tail}"))
131        }
132        FeedError::Future => KevyError::Protocol("ERR feed cursor ahead of stream".into()),
133        FeedError::Disabled => KevyError::Unsupported("feed disabled: open the embedded store with Config::with_feed".into()),
134    }
135}
136
137fn feed_read_request(
138    c: &mut RespClient,
139    shard: usize,
140    generation: u64,
141    offset: u64,
142    count: Option<usize>,
143    prefixes: &[&[u8]],
144) -> KevyResult<Reply> {
145    let mut args: Vec<Vec<u8>> = vec![
146        b"FEED.READ".to_vec(),
147        shard.to_string().into_bytes(),
148        generation.to_string().into_bytes(),
149        offset.to_string().into_bytes(),
150    ];
151    if let Some(n) = count {
152        args.push(b"COUNT".to_vec());
153        args.push(n.to_string().into_bytes());
154    }
155    for p in prefixes {
156        args.push(b"PREFIX".to_vec());
157        args.push(p.to_vec());
158    }
159    Ok(c.request(&args)?)
160}
161
162/// `*3 [:generation, :next_offset, *N frames]`, each frame
163/// `*2 [:offset, *M argv]`.
164fn parse_batch(reply: Reply) -> KevyResult<FeedBatch> {
165    let Reply::Array(items) = reply else {
166        return match reply {
167            Reply::Error(e) => Err(KevyError::Protocol(string(e))),
168            other => Err(unexpected(other)),
169        };
170    };
171    if items.len() != 3 {
172        return Err(KevyError::Protocol("FEED.READ: expected [gen, next, frames]".into()));
173    }
174    let mut it = items.into_iter();
175    let (Reply::Int(g), Reply::Int(next)) = (it.next().unwrap(), it.next().unwrap()) else {
176        return Err(KevyError::Protocol("FEED.READ: non-integer cursor".into()));
177    };
178    let Reply::Array(raw_frames) = it.next().unwrap() else {
179        return Err(KevyError::Protocol("FEED.READ: frames not an array".into()));
180    };
181    let frames = raw_frames
182        .into_iter()
183        .map(parse_frame)
184        .collect::<KevyResult<_>>()?;
185    Ok(FeedBatch { generation: g as u64, next_offset: next as u64, frames })
186}
187
188fn parse_frame(frame: Reply) -> KevyResult<FeedFrame> {
189    let Reply::Array(cells) = frame else {
190        return Err(KevyError::Protocol("FEED.READ: frame not an array".into()));
191    };
192    let mut it = cells.into_iter();
193    let (Some(Reply::Int(off)), Some(Reply::Array(argv_raw))) = (it.next(), it.next()) else {
194        return Err(KevyError::Protocol("FEED.READ: frame shape != [offset, argv]".into()));
195    };
196    let argv = argv_raw
197        .into_iter()
198        .map(|a| match a {
199            Reply::Bulk(b) | Reply::Simple(b) => Ok(b),
200            other => Err(unexpected(other)),
201        })
202        .collect::<KevyResult<_>>()?;
203    Ok(FeedFrame { offset: off as u64, argv })
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    #[test]
211    fn embedded_without_feed_config_is_unsupported() {
212        // mem:// opens the store without Config::with_feed.
213        let mut c = Connection::connect("mem://").unwrap();
214        assert_eq!(c.feed_shards().unwrap(), 1);
215        let err = c.feed_tail(0).unwrap_err();
216        assert!(matches!(err, KevyError::Unsupported(_)));
217        let err = c.feed_read(0, 1, 0, None, &[]).unwrap_err();
218        assert!(matches!(err, KevyError::Unsupported(_)));
219    }
220
221    #[test]
222    fn embedded_nonzero_shard_rejected() {
223        let mut c = Connection::connect("mem://").unwrap();
224        let err = c.feed_tail(1).unwrap_err();
225        assert!(matches!(err, KevyError::InvalidInput(_)));
226    }
227
228    #[test]
229    fn batch_parser_maps_frames() {
230        let reply = Reply::Array(vec![
231            Reply::Int(1),
232            Reply::Int(42),
233            Reply::Array(vec![Reply::Array(vec![
234                Reply::Int(41),
235                Reply::Array(vec![
236                    Reply::Bulk(b"SET".to_vec()),
237                    Reply::Bulk(b"k".to_vec()),
238                    Reply::Bulk(b"v".to_vec()),
239                ]),
240            ])]),
241        ]);
242        let batch = parse_batch(reply).unwrap();
243        assert_eq!(batch.generation, 1);
244        assert_eq!(batch.next_offset, 42);
245        assert_eq!(batch.frames.len(), 1);
246        assert_eq!(batch.frames[0].offset, 41);
247        assert_eq!(batch.frames[0].argv[0], b"SET");
248    }
249}
kevy_client/feed.rs

kevy_client/
feed.rs
