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//! v2.0.21's SPOP→SREM propagation fix.
11
12use std::io;
13
14use kevy_store::{ZaddFlags, ZaddReport};
15
16#[cfg(not(target_arch = "wasm32"))]
17use crate::replica_glue::ensure_writable;
18use crate::store::{Store, commit_write, store_err};
19
20#[cfg(target_arch = "wasm32")]
21fn ensure_writable(_s: &Store) -> io::Result<()> {
22    Ok(())
23}
24
25fn reject_invalid(flags: ZaddFlags) -> io::Result<()> {
26    if flags.valid() {
27        Ok(())
28    } else {
29        Err(io::Error::new(
30            io::ErrorKind::InvalidInput,
31            "GT, LT, and/or NX options at the same time are not compatible",
32        ))
33    }
34}
35
36impl Store {
37    /// Flags-aware `ZADD`. See [`ZaddFlags`]; read
38    /// [`ZaddReport::changed`] for the `CH` reply shape. The
39    /// monotonic-heal idiom is `zadd_flags(k, pairs, ZaddFlags { gt:
40    /// true, ..Default::default() })`.
41    pub fn zadd_flags(
42        &self,
43        key: &[u8],
44        pairs: &[(f64, &[u8])],
45        flags: ZaddFlags,
46    ) -> io::Result<ZaddReport> {
47        reject_invalid(flags)?;
48        ensure_writable(self)?;
49        let mut g = self.wshard(key);
50        let rep = g
51            .store
52            .zadd_flags_borrowed(key, pairs, flags)
53            .map_err(store_err)?;
54        if !rep.applied.is_empty() {
55            let score_strs: Vec<Vec<u8>> = rep
56                .applied
57                .iter()
58                .map(|(s, _)| format!("{s}").into_bytes())
59                .collect();
60            let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + rep.applied.len() * 2);
61            parts.push(b"ZADD");
62            parts.push(key);
63            for (i, (_, m)) in rep.applied.iter().enumerate() {
64                parts.push(&score_strs[i]);
65                parts.push(m);
66            }
67            commit_write(&mut g, &parts)?;
68        }
69        Ok(rep)
70    }
71
72    /// `ZADD … INCR` — a conditional `ZINCRBY`; `None` when the flags
73    /// veto the operation.
74    pub fn zadd_incr(
75        &self,
76        key: &[u8],
77        delta: f64,
78        member: &[u8],
79        flags: ZaddFlags,
80    ) -> io::Result<Option<f64>> {
81        reject_invalid(flags)?;
82        ensure_writable(self)?;
83        let mut g = self.wshard(key);
84        let next = g
85            .store
86            .zadd_incr(key, delta, member, flags)
87            .map_err(store_err)?;
88        if let Some(n) = next {
89            let s = format!("{n}");
90            commit_write(&mut g, &[b"ZADD", key, s.as_bytes(), member])?;
91        }
92        Ok(next)
93    }
94}
kevy_embedded/ops_zset_flags.rs

kevy_embedded/
ops_zset_flags.rs
