kevy_store/stream/
mod.rs

1//! Redis-compatible Streams storage. Each stream is an append-only log
2//! of (ID, field-value-list) entries keyed by a monotonically increasing
3//! `<ms>-<seq>` ID. The entries live in a `BTreeMap<StreamId, _>` so
4//! range queries are O(log n + k) and the iterator natural order is the
5//! ID order (ascending).
6//!
7//! Sprint A scope: bare stream (no consumer groups). The `StreamData`
8//! type carries a `groups` slot reserved for sprint B; this file only
9//! implements the entry-side ops.
10
11#[cfg(not(feature = "std"))]
12use crate::nostd_prelude::*;
13use alloc::collections::BTreeMap;
14#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
15#[cfg(not(any(
16    feature = "external-clock",
17    all(target_arch = "wasm32", target_os = "unknown")
18)))]
19use std::time::{SystemTime, UNIX_EPOCH};
20
21use kevy_map::KevyMap;
22
23use crate::StoreError;
24use crate::value::{BTREE_SLOT_BYTES, SmallBytes};
25
26// ───────────── StreamId ─────────────
27
28/// A stream entry's `<ms>-<seq>` identifier. The `Ord` derivation compares
29/// `ms` first then `seq`, which is exactly the monotonic order the protocol
30/// requires; same derivation gives `Eq`, `Hash`, and the `BTreeMap` key bound.
31#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Default)]
32pub struct StreamId {
33    /// Unix milliseconds timestamp component.
34    pub ms: u64,
35    /// Per-ms sequence number, 0-based.
36    pub seq: u64,
37}
38
39impl StreamId {
40    /// The numerically smallest ID; XRANGE `-` start.
41    pub const MIN: StreamId = StreamId { ms: 0, seq: 0 };
42    /// The numerically largest representable ID; XRANGE `+` end.
43    pub const MAX: StreamId = StreamId { ms: u64::MAX, seq: u64::MAX };
44
45    /// Render as the canonical `<ms>-<seq>` wire form.
46    pub fn encode(self) -> Vec<u8> {
47        format!("{}-{}", self.ms, self.seq).into_bytes()
48    }
49
50    /// Step one ID past `self`. Saturates at [`Self::MAX`].
51    #[must_use]
52    pub fn next(self) -> Self {
53        if self.seq < u64::MAX {
54            StreamId { ms: self.ms, seq: self.seq + 1 }
55        } else if self.ms < u64::MAX {
56            StreamId { ms: self.ms + 1, seq: 0 }
57        } else {
58            StreamId::MAX
59        }
60    }
61}
62
63/// XADD's ID argument: either an explicit `<ms>-<seq>` (both parts may
64/// be `*` to auto-fill `seq` only) or fully auto-generate via `*`.
65#[derive(Clone, Copy, Debug, Eq, PartialEq)]
66pub enum XAddIdSpec {
67    /// `*` — generate both `ms` (= current wall-clock) and `seq`.
68    AutoAll,
69    /// `<ms>-*` — caller fixes `ms`, server picks the next free `seq`.
70    AutoSeq(u64),
71    /// `<ms>-<seq>` — caller fully specifies the ID.
72    Explicit(StreamId),
73}
74
75/// Parse an XADD ID argument (`*`, `ms`, `ms-*`, `ms-seq`).
76pub fn parse_xadd_id(s: &[u8]) -> Result<XAddIdSpec, StreamIdError> {
77    if s == b"*" {
78        return Ok(XAddIdSpec::AutoAll);
79    }
80    let txt = core::str::from_utf8(s).map_err(|_| StreamIdError::Invalid)?;
81    match txt.split_once('-') {
82        None => {
83            let ms = txt.parse::<u64>().map_err(|_| StreamIdError::Invalid)?;
84            Ok(XAddIdSpec::Explicit(StreamId { ms, seq: 0 }))
85        }
86        Some((ms_s, seq_s)) => {
87            let ms = ms_s.parse::<u64>().map_err(|_| StreamIdError::Invalid)?;
88            if seq_s == "*" {
89                Ok(XAddIdSpec::AutoSeq(ms))
90            } else {
91                let seq = seq_s.parse::<u64>().map_err(|_| StreamIdError::Invalid)?;
92                Ok(XAddIdSpec::Explicit(StreamId { ms, seq }))
93            }
94        }
95    }
96}
97
98/// Parse an XRANGE `start` ID. Accepts `-` (= [`StreamId::MIN`]), bare
99/// `ms` (seq=0), and full `ms-seq`.
100pub fn parse_range_start(s: &[u8]) -> Result<StreamId, StreamIdError> {
101    if s == b"-" {
102        return Ok(StreamId::MIN);
103    }
104    parse_explicit_id(s, /*end=*/ false)
105}
106
107/// Parse an XRANGE `end` ID. Accepts `+` (= [`StreamId::MAX`]), bare `ms`
108/// (seq=u64::MAX so the entire ms is included), and full `ms-seq`.
109pub fn parse_range_end(s: &[u8]) -> Result<StreamId, StreamIdError> {
110    if s == b"+" {
111        return Ok(StreamId::MAX);
112    }
113    parse_explicit_id(s, /*end=*/ true)
114}
115
116/// Parse a fully-explicit ID for XREAD's per-stream "last-seen" arg
117/// (`0`, `0-0`, `5-2`). `$` is handled by the caller (it means "the
118/// stream's current `last_id`", which only Store can resolve).
119pub fn parse_explicit_id(s: &[u8], end: bool) -> Result<StreamId, StreamIdError> {
120    let txt = core::str::from_utf8(s).map_err(|_| StreamIdError::Invalid)?;
121    let (ms_s, seq_s) = match txt.split_once('-') {
122        Some(p) => p,
123        None => (txt, if end { "" } else { "0" }),
124    };
125    let ms = ms_s.parse::<u64>().map_err(|_| StreamIdError::Invalid)?;
126    let seq = if seq_s.is_empty() {
127        u64::MAX
128    } else {
129        seq_s.parse::<u64>().map_err(|_| StreamIdError::Invalid)?
130    };
131    Ok(StreamId { ms, seq })
132}
133
134/// Errors `parse_*_id` may emit. Distinct from `StoreError::NotInteger`
135/// so callers can map to the more specific Redis wire shape (`ERR
136/// Invalid stream ID specified as stream command argument`).
137#[derive(Debug, Clone, Copy, PartialEq, Eq)]
138pub enum StreamIdError {
139    /// Couldn't parse the bytes as `<ms>[-<seq>]` / `*` / `-` / `+`.
140    Invalid,
141}
142
143// ───────────── StreamData ─────────────
144
145/// One stream's storage: every entry in `entries` plus the per-stream
146/// scalar state Redis exposes via `XINFO STREAM`, plus the consumer
147/// groups map (sprint B). An empty `groups` map costs ~8 bytes and
148/// makes the no-group fast path (sprint A XADD/XREAD) zero-overhead.
149#[derive(Default, Clone)]
150pub struct StreamData {
151    /// Sorted entries; the `BTreeMap` enforces strict-increasing IDs.
152    pub(super) entries: BTreeMap<StreamId, Vec<(SmallBytes, SmallBytes)>>,
153    /// Largest ID **ever** seen on this stream, even after the entry
154    /// has been deleted (XDEL doesn't roll the clock back).
155    pub(super) last_id: StreamId,
156    /// Largest ID that has been deleted (`max_deleted_entry_id` in
157    /// Redis XINFO). Used to detect "deletion-only" gaps for clients.
158    pub(super) max_deleted_id: StreamId,
159    /// Cumulative number of entries ever added — never decreases. Used
160    /// by XINFO STREAM's `entries-added`.
161    pub(super) entries_added: u64,
162    /// Consumer groups keyed by name (sprint B). Boxed so the
163    /// `StreamData` struct stays compact when no groups are attached.
164    pub(super) groups: KevyMap<SmallBytes, Box<group::ConsumerGroup>>,
165}
166
167impl StreamData {
168    /// Current entry count (never larger than `entries_added`).
169    pub fn length(&self) -> u64 {
170        self.entries.len() as u64
171    }
172
173    /// Last ID ever assigned. Resets to `MIN` only when the whole key
174    /// is deleted (we never down-rev a stream).
175    pub fn last_id(&self) -> StreamId {
176        self.last_id
177    }
178
179    /// XINFO STREAM helpers.
180    pub fn entries_added(&self) -> u64 {
181        self.entries_added
182    }
183
184    /// The highest id XDEL has removed. XINFO reports it, and a consumer
185    /// group uses it to tell "never existed" from "existed and was
186    /// deleted" when a pending entry cannot be found.
187    pub fn max_deleted_id(&self) -> StreamId {
188        self.max_deleted_id
189    }
190
191    /// Iterate every entry in ID-ascending order. Snapshot serializers
192    /// walk this to dump the stream.
193    pub fn iter_entries(&self) -> impl Iterator<Item = (StreamId, &[(SmallBytes, SmallBytes)])> {
194        self.entries.iter().map(|(id, fv)| (*id, fv.as_slice()))
195    }
196
197    /// First (smallest-ID) entry — `None` if empty.
198    pub fn first_entry(&self) -> Option<(StreamId, &[(SmallBytes, SmallBytes)])> {
199        self.entries.iter().next().map(|(id, fv)| (*id, fv.as_slice()))
200    }
201
202    /// Last (largest-ID) entry — `None` if empty.
203    pub fn last_entry(&self) -> Option<(StreamId, &[(SmallBytes, SmallBytes)])> {
204        self.entries.iter().next_back().map(|(id, fv)| (*id, fv.as_slice()))
205    }
206
207    /// Iterate `(group_name, group)` pairs — used by `XINFO GROUPS`.
208    pub fn groups_iter(&self) -> impl Iterator<Item = (&[u8], &group::ConsumerGroup)> {
209        self.groups.iter().map(|(k, v)| (k.as_slice(), v.as_ref()))
210    }
211
212    /// Lookup one group by name (for `XINFO CONSUMERS`).
213    pub fn group(&self, name: &[u8]) -> Option<&group::ConsumerGroup> {
214        self.groups.get(name).map(core::convert::AsRef::as_ref)
215    }
216
217    /// Group count — `XINFO STREAM`'s `groups` field.
218    pub fn group_count(&self) -> usize {
219        self.groups.len()
220    }
221
222    /// Snapshot-loader entry-point: insert a pre-existing entry without
223    /// touching scalar state. Used by `Store::load_stream`; the loader
224    /// pumps every entry then calls [`Self::set_loaded_state`] once.
225    pub fn load_entry(&mut self, id: StreamId, fields: Vec<(SmallBytes, SmallBytes)>) {
226        self.entries.insert(id, fields);
227    }
228
229    /// Snapshot-loader: restore the per-stream scalars after every
230    /// entry has been pushed via [`Self::load_entry`].
231    pub fn set_loaded_state(
232        &mut self,
233        last_id: StreamId,
234        max_deleted_id: StreamId,
235        entries_added: u64,
236    ) {
237        self.last_id = last_id;
238        self.max_deleted_id = max_deleted_id;
239        self.entries_added = entries_added;
240    }
241
242    /// Insert a pre-resolved entry. Caller is responsible for picking
243    /// the ID via [`StreamData::resolve_xadd_id`] so monotonicity holds.
244    pub(crate) fn insert(&mut self, id: StreamId, fields: Vec<(SmallBytes, SmallBytes)>) {
245        debug_assert!(id > self.last_id || (id == StreamId::MIN && self.last_id == StreamId::MIN));
246        self.entries.insert(id, fields);
247        self.last_id = id;
248        self.entries_added += 1;
249    }
250
251    /// Translate XADD's `XAddIdSpec` into a concrete `StreamId`,
252    /// rejecting any spec that would not be strictly greater than
253    /// `self.last_id`. `now_ms` is injected so tests can pin wall-clock.
254    pub fn resolve_xadd_id(&self, spec: XAddIdSpec, now_ms: u64) -> Result<StreamId, StoreError> {
255        let candidate = match spec {
256            XAddIdSpec::AutoAll => {
257                let ms = now_ms.max(self.last_id.ms);
258                if ms == self.last_id.ms {
259                    StreamId { ms, seq: self.last_id.seq + 1 }
260                } else {
261                    StreamId { ms, seq: 0 }
262                }
263            }
264            XAddIdSpec::AutoSeq(ms) => {
265                if ms < self.last_id.ms {
266                    return Err(StoreError::OutOfRange);
267                }
268                if ms == self.last_id.ms {
269                    StreamId { ms, seq: self.last_id.seq + 1 }
270                } else {
271                    StreamId { ms, seq: 0 }
272                }
273            }
274            XAddIdSpec::Explicit(id) => {
275                if id <= self.last_id {
276                    return Err(StoreError::OutOfRange);
277                }
278                if id == StreamId::MIN {
279                    return Err(StoreError::OutOfRange);
280                }
281                id
282            }
283        };
284        Ok(candidate)
285    }
286
287    /// XRANGE — inclusive `[start, end]`, optionally COUNT-bounded.
288    pub fn range(
289        &self,
290        start: StreamId,
291        end: StreamId,
292        count: Option<usize>,
293    ) -> Vec<(StreamId, &[(SmallBytes, SmallBytes)])> {
294        let iter = self.entries.range(start..=end).map(|(id, fv)| (*id, fv.as_slice()));
295        match count {
296            Some(n) => iter.take(n).collect(),
297            None => iter.collect(),
298        }
299    }
300
301    /// XREVRANGE — same `[start, end]` interval, descending order.
302    pub fn revrange(
303        &self,
304        start: StreamId,
305        end: StreamId,
306        count: Option<usize>,
307    ) -> Vec<(StreamId, &[(SmallBytes, SmallBytes)])> {
308        let iter = self.entries.range(start..=end).rev().map(|(id, fv)| (*id, fv.as_slice()));
309        match count {
310            Some(n) => iter.take(n).collect(),
311            None => iter.collect(),
312        }
313    }
314
315    /// XREAD — entries strictly after `last_seen`, optionally COUNT-bounded.
316    pub fn read_after(
317        &self,
318        last_seen: StreamId,
319        count: Option<usize>,
320    ) -> Vec<(StreamId, &[(SmallBytes, SmallBytes)])> {
321        if last_seen == StreamId::MAX {
322            return Vec::new();
323        }
324        self.range(last_seen.next(), StreamId::MAX, count)
325    }
326
327    /// XDEL — remove `ids`. Returns the count actually removed (missing
328    /// IDs silently skipped). Updates `max_deleted_id` so XINFO can
329    /// report it.
330    pub(crate) fn del_ids(&mut self, ids: &[StreamId]) -> usize {
331        let mut removed = 0usize;
332        for id in ids {
333            if self.entries.remove(id).is_some() {
334                removed += 1;
335                if *id > self.max_deleted_id {
336                    self.max_deleted_id = *id;
337                }
338            }
339        }
340        removed
341    }
342
343    /// XTRIM MAXLEN — keep the most recent `n` entries.
344    pub(crate) fn trim_maxlen(&mut self, n: usize) -> usize {
345        let len = self.entries.len();
346        if len <= n {
347            return 0;
348        }
349        let drop = len - n;
350        let mut removed = 0;
351        let drop_ids: Vec<StreamId> = self.entries.keys().copied().take(drop).collect();
352        for id in drop_ids {
353            self.entries.remove(&id);
354            if id > self.max_deleted_id {
355                self.max_deleted_id = id;
356            }
357            removed += 1;
358        }
359        removed
360    }
361
362    /// Approximate heap footprint for `Value::weight`. Walks the entry
363    /// list once; cheap relative to the size of the stream itself.
364    pub fn weight(&self) -> u64 {
365        let entry_sum: u64 = self
366            .entries
367            .values()
368            .map(|fv| {
369                24 + fv
370                    .iter()
371                    .map(|(f, v)| 48 + f.heap_bytes() as u64 + v.heap_bytes() as u64)
372                    .sum::<u64>()
373            })
374            .sum();
375        (self.entries.len() as u64).saturating_mul(BTREE_SLOT_BYTES) + entry_sum
376    }
377
378    /// XTRIM MINID — drop every entry with ID < `floor`.
379    pub(crate) fn trim_minid(&mut self, floor: StreamId) -> usize {
380        let drop_ids: Vec<StreamId> = self.entries.range(..floor).map(|(id, _)| *id).collect();
381        let removed = drop_ids.len();
382        for id in drop_ids {
383            self.entries.remove(&id);
384            if id > self.max_deleted_id {
385                self.max_deleted_id = id;
386            }
387        }
388        removed
389    }
390}
391
392mod claim;
393mod group;
394mod load;
395mod store;
396#[allow(unused_imports)]
397pub use claim::AutoclaimResult;
398#[allow(unused_imports)]
399pub use group::{
400    ConsumerGroup, ConsumerState, GroupCreateMode, PelEntry, PendingExtended, PendingExtendedRow,
401    PendingSummary, ReadGroupId, XClaimOpts,
402};
403pub use load::{LoadedGroup, LoadedPelEntry};
404pub use store::EntryBatch;
405
406/// Snapshot-loader payload: one stream entry decoded into primitive
407/// tuples `(ms, seq, [(field, value), ...])`. The persist crate emits
408/// these and `Store::load_stream` consumes them.
409pub type LoadedStreamEntry = (u64, u64, Vec<(Vec<u8>, Vec<u8>)>);
410
411// ───────────── small helpers (shared with `store.rs`) ─────────────
412
413/// Wall-clock millis. Shared with dispatchers so every XADD on a shard uses
414/// the same clock source. On native targets reads `SystemTime::now()` (falls
415/// back to 0 on a pre-UNIX-EPOCH clock — impossible on supported platforms);
416/// on `wasm32-unknown-unknown`, where `SystemTime::now()` traps, reads the
417/// host-fed wall clock (see `crate::set_wall_clock_ms`, wasm-only).
418#[cfg(not(any(feature = "external-clock", all(target_arch = "wasm32", target_os = "unknown"))))]
419pub fn now_unix_ms() -> u64 {
420    SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64)
421}
422
423/// Wall-clock milliseconds since the epoch, for stream ids.
424///
425/// The twin of the `SystemTime` version above, for builds that have no
426/// `SystemTime`: an external-clock build or wasm. Both must agree on the
427/// unit — a stream id is a millisecond and nothing downstream re-scales.
428#[cfg(any(feature = "external-clock", all(target_arch = "wasm32", target_os = "unknown")))]
429pub fn now_unix_ms() -> u64 {
430    crate::clock::wall_now_unix_ms()
431}
432
433pub(super) fn stream_entry_weight(fields: &[(SmallBytes, SmallBytes)]) -> u64 {
434    // BTreeMap slot + Vec header + each (field, value) cell + their heap.
435    BTREE_SLOT_BYTES
436        + 24
437        + fields
438            .iter()
439            .map(|(f, v)| 48 + f.heap_bytes() as u64 + v.heap_bytes() as u64)
440            .sum::<u64>()
441}
442
443pub(super) fn clone_entries(src: Vec<(StreamId, &[(SmallBytes, SmallBytes)])>) -> EntryBatch {
444    src.into_iter()
445        .map(|(id, fv)| (id, fv.iter().map(|(f, v)| (f.to_vec(), v.to_vec())).collect()))
446        .collect()
447}
kevy_store/stream/mod.rs

kevy_store/stream/
mod.rs
