kevy_client/
feed.rs

1//! Change-feed (CDC): `FEED.SHARDS` / `FEED.TAIL` / `FEED.READ`
2//! (v1.14.0) — 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 std::io;
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) -> io::Result<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(io::Error::other(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) -> io::Result<(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(io::Error::other(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    ) -> io::Result<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) -> io::Result<()> {
119    if shard != 0 {
120        return Err(io::Error::new(
121            io::ErrorKind::InvalidInput,
122            "embedded feed is single-shard: shard must be 0",
123        ));
124    }
125    Ok(())
126}
127
128/// Map [`FeedError`] onto the same error text the wire produces, so
129/// resync handling code is backend-agnostic.
130fn feed_err(e: FeedError) -> io::Error {
131    match e {
132        FeedError::Resync { generation, tail } => {
133            io::Error::other(format!("FEEDRESYNC {generation} {tail}"))
134        }
135        FeedError::Future => io::Error::other("ERR feed cursor ahead of stream"),
136        FeedError::Disabled => io::Error::new(
137            io::ErrorKind::Unsupported,
138            "feed disabled: open the embedded store with Config::with_feed",
139        ),
140    }
141}
142
143fn feed_read_request(
144    c: &mut RespClient,
145    shard: usize,
146    generation: u64,
147    offset: u64,
148    count: Option<usize>,
149    prefixes: &[&[u8]],
150) -> io::Result<Reply> {
151    let mut args: Vec<Vec<u8>> = vec![
152        b"FEED.READ".to_vec(),
153        shard.to_string().into_bytes(),
154        generation.to_string().into_bytes(),
155        offset.to_string().into_bytes(),
156    ];
157    if let Some(n) = count {
158        args.push(b"COUNT".to_vec());
159        args.push(n.to_string().into_bytes());
160    }
161    for p in prefixes {
162        args.push(b"PREFIX".to_vec());
163        args.push(p.to_vec());
164    }
165    c.request(&args)
166}
167
168/// `*3 [:generation, :next_offset, *N frames]`, each frame
169/// `*2 [:offset, *M argv]`.
170fn parse_batch(reply: Reply) -> io::Result<FeedBatch> {
171    let Reply::Array(items) = reply else {
172        return match reply {
173            Reply::Error(e) => Err(io::Error::other(string(e))),
174            other => Err(unexpected(other)),
175        };
176    };
177    if items.len() != 3 {
178        return Err(io::Error::other("FEED.READ: expected [gen, next, frames]"));
179    }
180    let mut it = items.into_iter();
181    let (Reply::Int(g), Reply::Int(next)) = (it.next().unwrap(), it.next().unwrap()) else {
182        return Err(io::Error::other("FEED.READ: non-integer cursor"));
183    };
184    let Reply::Array(raw_frames) = it.next().unwrap() else {
185        return Err(io::Error::other("FEED.READ: frames not an array"));
186    };
187    let frames = raw_frames
188        .into_iter()
189        .map(parse_frame)
190        .collect::<io::Result<_>>()?;
191    Ok(FeedBatch { generation: g as u64, next_offset: next as u64, frames })
192}
193
194fn parse_frame(frame: Reply) -> io::Result<FeedFrame> {
195    let Reply::Array(cells) = frame else {
196        return Err(io::Error::other("FEED.READ: frame not an array"));
197    };
198    let mut it = cells.into_iter();
199    let (Some(Reply::Int(off)), Some(Reply::Array(argv_raw))) = (it.next(), it.next()) else {
200        return Err(io::Error::other("FEED.READ: frame shape != [offset, argv]"));
201    };
202    let argv = argv_raw
203        .into_iter()
204        .map(|a| match a {
205            Reply::Bulk(b) | Reply::Simple(b) => Ok(b),
206            other => Err(unexpected(other)),
207        })
208        .collect::<io::Result<_>>()?;
209    Ok(FeedFrame { offset: off as u64, argv })
210}
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215
216    #[test]
217    fn embedded_without_feed_config_is_unsupported() {
218        // mem:// opens the store without Config::with_feed.
219        let mut c = Connection::open("mem://").unwrap();
220        assert_eq!(c.feed_shards().unwrap(), 1);
221        let err = c.feed_tail(0).unwrap_err();
222        assert_eq!(err.kind(), io::ErrorKind::Unsupported);
223        let err = c.feed_read(0, 1, 0, None, &[]).unwrap_err();
224        assert_eq!(err.kind(), io::ErrorKind::Unsupported);
225    }
226
227    #[test]
228    fn embedded_nonzero_shard_rejected() {
229        let mut c = Connection::open("mem://").unwrap();
230        let err = c.feed_tail(1).unwrap_err();
231        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
232    }
233
234    #[test]
235    fn batch_parser_maps_frames() {
236        let reply = Reply::Array(vec![
237            Reply::Int(1),
238            Reply::Int(42),
239            Reply::Array(vec![Reply::Array(vec![
240                Reply::Int(41),
241                Reply::Array(vec![
242                    Reply::Bulk(b"SET".to_vec()),
243                    Reply::Bulk(b"k".to_vec()),
244                    Reply::Bulk(b"v".to_vec()),
245                ]),
246            ])]),
247        ]);
248        let batch = parse_batch(reply).unwrap();
249        assert_eq!(batch.generation, 1);
250        assert_eq!(batch.next_offset, 42);
251        assert_eq!(batch.frames.len(), 1);
252        assert_eq!(batch.frames[0].offset, 41);
253        assert_eq!(batch.frames[0].argv[0], b"SET");
254    }
255}
kevy_client/feed.rs

kevy_client/
feed.rs
