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