kevy_store/
zset_flags.rs

1//! `ZADD` condition flags (Redis 6.2): `NX` / `XX` / `GT` / `LT` /
2//! `CH` / `INCR`. Split from `zset.rs` (500-LOC rule). The no-flags
3//! hot path stays `zadd` / `zadd` — nothing here taxes it.
4
5#[cfg(not(feature = "std"))]
6use crate::nostd_prelude::*;
7use crate::{Store, StoreError};
8
9/// Parsed `ZADD` condition flags. `CH` only changes the *reply*
10/// (changed count instead of added count) — callers read
11/// [`ZaddReport::changed`] when set; the engine behavior is identical.
12#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
13pub struct ZaddFlags {
14    /// Only add new members; never update existing ones.
15    pub nx: bool,
16    /// Only update existing members; never add new ones.
17    pub xx: bool,
18    /// Only update when the new score is greater than the current.
19    pub gt: bool,
20    /// Only update when the new score is less than the current.
21    pub lt: bool,
22    /// Reply with changed (added + updated) instead of added.
23    pub ch: bool,
24}
25
26impl ZaddFlags {
27    /// Redis 6.2 rule: `GT`, `LT` and `NX` are mutually exclusive
28    /// (and `GT`+`LT` together are, too). `XX`+`NX` likewise.
29    pub fn valid(self) -> bool {
30        !(self.nx && (self.xx || self.gt || self.lt)) && !(self.gt && self.lt)
31    }
32}
33
34/// Outcome of a flags-aware `ZADD`.
35pub struct ZaddReport {
36    /// Members newly added.
37    pub added: usize,
38    /// Members added or whose score actually changed (`CH` reply).
39    pub changed: usize,
40    /// The `(score, member)` pairs actually applied, in input order —
41    /// vetoed pairs are absent. Lets an AOF writer log the *effect*
42    /// as a plain unconditional `ZADD` (deterministic on replay; a
43    /// conditional replayed against divergent state could veto
44    /// differently).
45    pub applied: Vec<(f64, Vec<u8>)>,
46}
47
48impl Store {
49    /// Flags-aware `ZADD`. Caller validates [`ZaddFlags::valid`] at
50    /// its input boundary (RESP parse / typed API) — invalid combos
51    /// here are a caller bug.
52    pub fn zadd_flags(
53        &mut self,
54        key: &[u8],
55        pairs: &[(f64, &[u8])],
56        flags: ZaddFlags,
57    ) -> Result<ZaddReport, StoreError> {
58        debug_assert!(flags.valid(), "caller must reject invalid flag combos");
59        let mut rep = ZaddReport { added: 0, changed: 0, applied: Vec::new() };
60        for (score, m) in pairs {
61            match self.zscore(key, m)? {
62                Some(old) => {
63                    if flags.nx || (flags.gt && *score <= old) || (flags.lt && *score >= old) {
64                        continue;
65                    }
66                    if *score != old {
67                        self.zadd(key, &[(*score, m)])?;
68                        rep.changed += 1;
69                        rep.applied.push((*score, m.to_vec()));
70                    }
71                }
72                None => {
73                    if flags.xx {
74                        continue;
75                    }
76                    self.zadd(key, &[(*score, m)])?;
77                    rep.added += 1;
78                    rep.changed += 1;
79                    rep.applied.push((*score, m.to_vec()));
80                }
81            }
82        }
83        Ok(rep)
84    }
85
86    /// `ZADD … INCR` — a conditional `ZINCRBY`: returns the new score,
87    /// or `None` when the flags veto the operation (Redis replies nil).
88    pub fn zadd_incr(
89        &mut self,
90        key: &[u8],
91        delta: f64,
92        member: &[u8],
93        flags: ZaddFlags,
94    ) -> Result<Option<f64>, StoreError> {
95        debug_assert!(flags.valid(), "caller must reject invalid flag combos");
96        match self.zscore(key, member)? {
97            Some(old) => {
98                if flags.nx {
99                    return Ok(None);
100                }
101                let next = old + delta;
102                if !next.is_finite() {
103                    return Err(StoreError::NotFloat);
104                }
105                if (flags.gt && next <= old) || (flags.lt && next >= old) {
106                    return Ok(None);
107                }
108                self.zadd(key, &[(next, member)])?;
109                Ok(Some(next))
110            }
111            None => {
112                if flags.xx {
113                    return Ok(None);
114                }
115                if !delta.is_finite() {
116                    return Err(StoreError::NotFloat);
117                }
118                self.zadd(key, &[(delta, member)])?;
119                Ok(Some(delta))
120            }
121        }
122    }
123}
124
125#[cfg(test)]
126mod tests {
127    use super::*;
128
129    fn zf() -> ZaddFlags {
130        ZaddFlags::default()
131    }
132
133    #[test]
134    fn validity_matrix() {
135        assert!(zf().valid());
136        assert!(ZaddFlags { gt: true, ch: true, ..zf() }.valid());
137        assert!(ZaddFlags { xx: true, gt: true, ..zf() }.valid());
138        assert!(!ZaddFlags { nx: true, xx: true, ..zf() }.valid());
139        assert!(!ZaddFlags { nx: true, gt: true, ..zf() }.valid());
140        assert!(!ZaddFlags { gt: true, lt: true, ..zf() }.valid());
141    }
142
143    #[test]
144    fn nx_only_adds() {
145        let mut s = Store::new();
146        s.zadd(b"z", &[(1.0, b"m".as_slice())]).unwrap();
147        let r = s
148            .zadd_flags(b"z", &[(9.0, b"m"), (2.0, b"n")], ZaddFlags { nx: true, ..zf() })
149            .unwrap();
150        assert_eq!((r.added, r.changed), (1, 1));
151        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(1.0)); // untouched
152        assert_eq!(s.zscore(b"z", b"n").unwrap(), Some(2.0));
153    }
154
155    #[test]
156    fn xx_only_updates() {
157        let mut s = Store::new();
158        s.zadd(b"z", &[(1.0, b"m".as_slice())]).unwrap();
159        let r = s
160            .zadd_flags(b"z", &[(9.0, b"m"), (2.0, b"n")], ZaddFlags { xx: true, ..zf() })
161            .unwrap();
162        assert_eq!((r.added, r.changed), (0, 1));
163        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(9.0));
164        assert_eq!(s.zscore(b"z", b"n").unwrap(), None); // not added
165    }
166
167    #[test]
168    fn gt_is_monotonic_heal() {
169        let mut s = Store::new();
170        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
171        let gt = ZaddFlags { gt: true, ..zf() };
172        // Stale (lower) score: vetoed.
173        let r = s.zadd_flags(b"z", &[(3.0, b"m")], gt).unwrap();
174        assert_eq!(r.changed, 0);
175        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(5.0));
176        // Newer (higher) score: applied.
177        let r = s.zadd_flags(b"z", &[(7.0, b"m")], gt).unwrap();
178        assert_eq!(r.changed, 1);
179        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(7.0));
180        // GT still ADDS missing members (only XX suppresses adds).
181        let r = s.zadd_flags(b"z", &[(1.0, b"new")], gt).unwrap();
182        assert_eq!(r.added, 1);
183    }
184
185    #[test]
186    fn lt_mirror() {
187        let mut s = Store::new();
188        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
189        let lt = ZaddFlags { lt: true, ..zf() };
190        assert_eq!(s.zadd_flags(b"z", &[(7.0, b"m")], lt).unwrap().changed, 0);
191        assert_eq!(s.zadd_flags(b"z", &[(3.0, b"m")], lt).unwrap().changed, 1);
192        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(3.0));
193    }
194
195    #[test]
196    fn applied_reflects_effect_only() {
197        let mut s = Store::new();
198        s.zadd(b"z", &[(5.0, b"a".as_slice()), (5.0, b"b".as_slice())]).unwrap();
199        let r = s
200            .zadd_flags(
201                b"z",
202                &[(9.0, b"a"), (1.0, b"b"), (5.0, b"c")],
203                ZaddFlags { gt: true, ..zf() },
204            )
205            .unwrap();
206        // a updated, b vetoed, c added.
207        assert_eq!(r.applied, vec![(9.0, b"a".to_vec()), (5.0, b"c".to_vec())]);
208    }
209
210    #[test]
211    fn incr_form_vetoes_to_none() {
212        let mut s = Store::new();
213        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
214        let gt = ZaddFlags { gt: true, ..zf() };
215        // Negative delta under GT: next < old → nil, score untouched.
216        assert_eq!(s.zadd_incr(b"z", -2.0, b"m", gt).unwrap(), None);
217        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(5.0));
218        assert_eq!(s.zadd_incr(b"z", 2.0, b"m", gt).unwrap(), Some(7.0));
219        // XX on a missing member → nil.
220        let xx = ZaddFlags { xx: true, ..zf() };
221        assert_eq!(s.zadd_incr(b"z", 1.0, b"nope", xx).unwrap(), None);
222        // NX on an existing member → nil.
223        let nx = ZaddFlags { nx: true, ..zf() };
224        assert_eq!(s.zadd_incr(b"z", 1.0, b"m", nx).unwrap(), None);
225    }
226
227    #[test]
228    fn wrongtype_propagates() {
229        let mut s = Store::new();
230        s.set(b"str", b"v".to_vec(), None, false, false);
231        assert!(s.zadd_flags(b"str", &[(1.0, b"m")], zf()).is_err());
232        assert!(s.zadd_incr(b"str", 1.0, b"m", zf()).is_err());
233    }
234}
kevy_store/zset_flags.rs

kevy_store/
zset_flags.rs
