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
64                        || (flags.gt && *score <= old)
65                        || (flags.lt && *score >= old)
66                    {
67                        continue;
68                    }
69                    if *score != old {
70                        self.zadd(key, &[(*score, m)])?;
71                        rep.changed += 1;
72                        rep.applied.push((*score, m.to_vec()));
73                    }
74                }
75                None => {
76                    if flags.xx {
77                        continue;
78                    }
79                    self.zadd(key, &[(*score, m)])?;
80                    rep.added += 1;
81                    rep.changed += 1;
82                    rep.applied.push((*score, m.to_vec()));
83                }
84            }
85        }
86        Ok(rep)
87    }
88
89    /// `ZADD … INCR` — a conditional `ZINCRBY`: returns the new score,
90    /// or `None` when the flags veto the operation (Redis replies nil).
91    pub fn zadd_incr(
92        &mut self,
93        key: &[u8],
94        delta: f64,
95        member: &[u8],
96        flags: ZaddFlags,
97    ) -> Result<Option<f64>, StoreError> {
98        debug_assert!(flags.valid(), "caller must reject invalid flag combos");
99        match self.zscore(key, member)? {
100            Some(old) => {
101                if flags.nx {
102                    return Ok(None);
103                }
104                let next = old + delta;
105                if !next.is_finite() {
106                    return Err(StoreError::NotFloat);
107                }
108                if (flags.gt && next <= old) || (flags.lt && next >= old) {
109                    return Ok(None);
110                }
111                self.zadd(key, &[(next, member)])?;
112                Ok(Some(next))
113            }
114            None => {
115                if flags.xx {
116                    return Ok(None);
117                }
118                if !delta.is_finite() {
119                    return Err(StoreError::NotFloat);
120                }
121                self.zadd(key, &[(delta, member)])?;
122                Ok(Some(delta))
123            }
124        }
125    }
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131
132    fn zf() -> ZaddFlags {
133        ZaddFlags::default()
134    }
135
136    #[test]
137    fn validity_matrix() {
138        assert!(zf().valid());
139        assert!(ZaddFlags { gt: true, ch: true, ..zf() }.valid());
140        assert!(ZaddFlags { xx: true, gt: true, ..zf() }.valid());
141        assert!(!ZaddFlags { nx: true, xx: true, ..zf() }.valid());
142        assert!(!ZaddFlags { nx: true, gt: true, ..zf() }.valid());
143        assert!(!ZaddFlags { gt: true, lt: true, ..zf() }.valid());
144    }
145
146    #[test]
147    fn nx_only_adds() {
148        let mut s = Store::new();
149        s.zadd(b"z", &[(1.0, b"m".as_slice())]).unwrap();
150        let r = s
151            .zadd_flags(b"z", &[(9.0, b"m"), (2.0, b"n")], ZaddFlags { nx: true, ..zf() })
152            .unwrap();
153        assert_eq!((r.added, r.changed), (1, 1));
154        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(1.0)); // untouched
155        assert_eq!(s.zscore(b"z", b"n").unwrap(), Some(2.0));
156    }
157
158    #[test]
159    fn xx_only_updates() {
160        let mut s = Store::new();
161        s.zadd(b"z", &[(1.0, b"m".as_slice())]).unwrap();
162        let r = s
163            .zadd_flags(b"z", &[(9.0, b"m"), (2.0, b"n")], ZaddFlags { xx: true, ..zf() })
164            .unwrap();
165        assert_eq!((r.added, r.changed), (0, 1));
166        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(9.0));
167        assert_eq!(s.zscore(b"z", b"n").unwrap(), None); // not added
168    }
169
170    #[test]
171    fn gt_is_monotonic_heal() {
172        let mut s = Store::new();
173        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
174        let gt = ZaddFlags { gt: true, ..zf() };
175        // Stale (lower) score: vetoed.
176        let r = s.zadd_flags(b"z", &[(3.0, b"m")], gt).unwrap();
177        assert_eq!(r.changed, 0);
178        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(5.0));
179        // Newer (higher) score: applied.
180        let r = s.zadd_flags(b"z", &[(7.0, b"m")], gt).unwrap();
181        assert_eq!(r.changed, 1);
182        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(7.0));
183        // GT still ADDS missing members (only XX suppresses adds).
184        let r = s.zadd_flags(b"z", &[(1.0, b"new")], gt).unwrap();
185        assert_eq!(r.added, 1);
186    }
187
188    #[test]
189    fn lt_mirror() {
190        let mut s = Store::new();
191        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
192        let lt = ZaddFlags { lt: true, ..zf() };
193        assert_eq!(s.zadd_flags(b"z", &[(7.0, b"m")], lt).unwrap().changed, 0);
194        assert_eq!(s.zadd_flags(b"z", &[(3.0, b"m")], lt).unwrap().changed, 1);
195        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(3.0));
196    }
197
198    #[test]
199    fn applied_reflects_effect_only() {
200        let mut s = Store::new();
201        s.zadd(b"z", &[(5.0, b"a".as_slice()), (5.0, b"b".as_slice())]).unwrap();
202        let r = s
203            .zadd_flags(
204                b"z",
205                &[(9.0, b"a"), (1.0, b"b"), (5.0, b"c")],
206                ZaddFlags { gt: true, ..zf() },
207            )
208            .unwrap();
209        // a updated, b vetoed, c added.
210        assert_eq!(
211            r.applied,
212            vec![(9.0, b"a".to_vec()), (5.0, b"c".to_vec())]
213        );
214    }
215
216    #[test]
217    fn incr_form_vetoes_to_none() {
218        let mut s = Store::new();
219        s.zadd(b"z", &[(5.0, b"m".as_slice())]).unwrap();
220        let gt = ZaddFlags { gt: true, ..zf() };
221        // Negative delta under GT: next < old → nil, score untouched.
222        assert_eq!(s.zadd_incr(b"z", -2.0, b"m", gt).unwrap(), None);
223        assert_eq!(s.zscore(b"z", b"m").unwrap(), Some(5.0));
224        assert_eq!(s.zadd_incr(b"z", 2.0, b"m", gt).unwrap(), Some(7.0));
225        // XX on a missing member → nil.
226        let xx = ZaddFlags { xx: true, ..zf() };
227        assert_eq!(s.zadd_incr(b"z", 1.0, b"nope", xx).unwrap(), None);
228        // NX on an existing member → nil.
229        let nx = ZaddFlags { nx: true, ..zf() };
230        assert_eq!(s.zadd_incr(b"z", 1.0, b"m", nx).unwrap(), None);
231    }
232
233    #[test]
234    fn wrongtype_propagates() {
235        let mut s = Store::new();
236        s.set(b"str", b"v".to_vec(), None, false, false);
237        assert!(s.zadd_flags(b"str", &[(1.0, b"m")], zf()).is_err());
238        assert!(s.zadd_incr(b"str", 1.0, b"m", zf()).is_err());
239    }
240}
kevy_store/zset_flags.rs

kevy_store/
zset_flags.rs
