kevy_embedded/
ops_zset_flags.rs

1//! `ZADD` condition-flag surfaces (Redis 6.2 `NX`/`XX`/`GT`/`LT`/`CH`
2//! + the `INCR` form). The plain `zadd` hot path is untouched.
3//!
4//! AOF form: the **effect** is logged, never the condition — applied
5//! pairs go into the log as plain unconditional `ZADD` (and the
6//! `INCR` form as `ZADD key <new-absolute-score> member`). A
7//! conditional verb replayed against divergent state (a replica
8//! applying frames onto snapshot-loaded state) could veto differently
9//! and diverge; the absolute form is deterministic. Same lesson as
10//! the SPOP→SREM propagation fix (log the effect, not the verb).
11
12use crate::{KevyError, KevyResult};
13
14use kevy_store::{ZaddFlags, ZaddReport};
15
16use crate::store::ensure_writable;
17use crate::store::{Store, commit_write, store_err};
18
19fn reject_invalid(flags: ZaddFlags) -> KevyResult<()> {
20    if flags.valid() {
21        Ok(())
22    } else {
23        Err(KevyError::InvalidInput("GT, LT, and/or NX options at the same time are not compatible".into()))
24    }
25}
26
27impl Store {
28    /// Flags-aware `ZADD`. See [`ZaddFlags`]; read
29    /// [`ZaddReport::changed`] for the `CH` reply shape. The
30    /// monotonic-heal idiom is `zadd_flags(k, pairs, ZaddFlags { gt:
31    /// true, ..Default::default() })`.
32    pub fn zadd_flags(
33        &self,
34        key: &[u8],
35        pairs: &[(f64, &[u8])],
36        flags: ZaddFlags,
37    ) -> KevyResult<ZaddReport> {
38        reject_invalid(flags)?;
39        ensure_writable(self)?;
40        let mut g = self.wshard(key);
41        let rep = g
42            .store
43            .zadd_flags(key, pairs, flags)
44            .map_err(store_err)?;
45        if !rep.applied.is_empty() {
46            let score_strs: Vec<Vec<u8>> = rep
47                .applied
48                .iter()
49                .map(|(s, _)| format!("{s}").into_bytes())
50                .collect();
51            let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + rep.applied.len() * 2);
52            parts.push(b"ZADD");
53            parts.push(key);
54            for (i, (_, m)) in rep.applied.iter().enumerate() {
55                parts.push(&score_strs[i]);
56                parts.push(m);
57            }
58            commit_write(&mut g, &parts)?;
59        }
60        Ok(rep)
61    }
62
63    /// `ZADD … INCR` — a conditional `ZINCRBY`; `None` when the flags
64    /// veto the operation.
65    pub fn zadd_incr(
66        &self,
67        key: &[u8],
68        delta: f64,
69        member: &[u8],
70        flags: ZaddFlags,
71    ) -> KevyResult<Option<f64>> {
72        reject_invalid(flags)?;
73        ensure_writable(self)?;
74        let mut g = self.wshard(key);
75        let next = g
76            .store
77            .zadd_incr(key, delta, member, flags)
78            .map_err(store_err)?;
79        if let Some(n) = next {
80            let s = format!("{n}");
81            commit_write(&mut g, &[b"ZADD", key, s.as_bytes(), member])?;
82        }
83        Ok(next)
84    }
85}
kevy_embedded/ops_zset_flags.rs

kevy_embedded/
ops_zset_flags.rs
