kevy_embedded/ops_p3.rs
1//! Multi-key string operations, keyspace scan, atomic `getex`, set
2//! algebra (`sinter` / `sunion` / `sdiff`), and absolute-time TTL
3//! variants (`expireat` / `pexpire`).
4//!
5//! The set algebra is implemented at the embedded layer (compose
6//! `smembers` per key + Rust set operations) instead of touching
7//! `kevy_store::Store` — over N small sets that is faster than
8//! serialising N RESP arrays.
9
10use crate::{InsertPosition, KevyResult};
11use std::collections::BTreeSet;
12use std::time::Duration;
13
14use crate::store::ensure_writable;
15use crate::store::{Store, commit_write, store_err};
16
17impl Store {
18 // ---- multi-key string ops ---------------------------------------
19
20 /// `MSET key value [key value ...]` — set every pair. The pairs of one
21 /// shard are set under one lock and logged as one `MSET` frame, so a
22 /// crash leaves each shard's share whole or absent; across shards there
23 /// is no such guarantee (Redis Cluster's semantics). A key given twice
24 /// takes its last value.
25 pub fn mset(&self, pairs: &[(&[u8], &[u8])]) -> KevyResult<()> {
26 ensure_writable(self)?;
27 let n = self.shards.len();
28 if n == 1 {
29 return self.mset_shard(0, pairs);
30 }
31 let mut by_shard: Vec<Vec<(&[u8], &[u8])>> = vec![Vec::new(); n];
32 for &(k, v) in pairs {
33 by_shard[crate::shard::shard_idx(k, n)].push((k, v));
34 }
35 for (i, group) in by_shard.iter().enumerate().filter(|(_, g)| !g.is_empty()) {
36 self.mset_shard(i, group)?;
37 }
38 Ok(())
39 }
40
41 fn mset_shard(&self, i: usize, pairs: &[(&[u8], &[u8])]) -> KevyResult<()> {
42 let mut g = crate::store::lock_write(&self.shards[i]);
43 let mut frame: Vec<&[u8]> = Vec::with_capacity(1 + 2 * pairs.len());
44 frame.push(b"MSET");
45 for &(k, v) in pairs {
46 g.store.set(k, v.to_vec(), None, kevy_store::SetCondition::Always);
47 frame.extend([k, v]);
48 }
49 commit_write(&mut g, &frame)
50 }
51
52 /// `MGET key [key ...]` — return `Some(value)` per requested key
53 /// that's present, `None` per absent / wrong-type.
54 ///
55 /// The wrong-type half of that sentence was prose only: the read
56 /// propagated the store's `WrongType` with `?`, so one list among
57 /// the keys turned the whole call into an error. Redis returns nil
58 /// for a key that does not hold a string and never errors here —
59 /// which is what the server's gather already did, and how the two
60 /// surfaces came to disagree in `differential_wire_vs_embedded`.
61 pub fn mget(&self, keys: &[&[u8]]) -> KevyResult<Vec<Option<Vec<u8>>>> {
62 let mut out = Vec::with_capacity(keys.len());
63 self.mget_with(keys.iter().copied(), |v| out.push(v.map(<[u8]>::to_vec)))?;
64 Ok(out)
65 }
66
67 /// [`Store::mget`] that lends each value to `f`, in key order, instead
68 /// of copying it out — for a caller that packs the values somewhere of
69 /// its own (a binding's reply buffer) and would otherwise copy twice.
70 ///
71 /// ```
72 /// let s = kevy_embedded::Store::open(kevy_embedded::Config::default()).unwrap();
73 /// s.set(b"a", b"1").unwrap();
74 /// let mut seen = Vec::new();
75 /// s.mget_with([&b"a"[..], b"nope"], |v| seen.push(v.map(<[u8]>::len))).unwrap();
76 /// assert_eq!(seen, [Some(1), None]);
77 /// ```
78 pub fn mget_with<'k>(
79 &self,
80 keys: impl IntoIterator<Item = &'k [u8]>,
81 mut f: impl FnMut(Option<&[u8]>),
82 ) -> KevyResult<()> {
83 for k in keys {
84 match self.wshard(k).store.get(k) {
85 Ok(v) => f(v.as_deref()),
86 Err(kevy_store::StoreError::WrongType) => f(None),
87 // Not reachable through a hot keyspace, and it stays:
88 // `Store::get` propagates the tiering path's errors
89 // too, and a cold-tier read that failed must not be
90 // answered as "this key holds nothing".
91 Err(e) => return Err(store_err(e)),
92 }
93 }
94 Ok(())
95 }
96
97 // ---- keyspace introspection -------------------------------------
98
99 /// `KEYS pattern` — glob-match every key in the keyspace
100 /// (across all shards). `pattern = None` matches everything.
101 /// `limit = None` is unbounded; otherwise bounds the TOTAL
102 /// returned across shards. Glob syntax matches Redis (`*` /
103 /// `?` / `[abc]` / escape).
104 pub fn keys(&self, pattern: Option<&[u8]>, limit: Option<usize>) -> Vec<Vec<u8>> {
105 self.collect_keys(pattern, limit)
106 }
107
108 // ---- atomic get + TTL -------------------------------------------
109
110 /// `GETEX key TTL` — get the value and update the TTL atomically
111 /// (single lock cycle on the owning shard). Returns the value;
112 /// `None` when absent. AOF-logged as an absolute `PEXPIREAT` — a
113 /// relative frame re-anchors on every replay, handing the key its
114 /// full TTL back at each restart (the incident class the absolute
115 /// form exists to prevent; this was its last relative holdout).
116 pub fn getex(&self, key: &[u8], ttl: Duration) -> KevyResult<Option<Vec<u8>>> {
117 ensure_writable(self)?;
118 let mut g = self.wshard(key);
119 let val = g.store.get(key).map_err(store_err)?.as_deref().map(<[u8]>::to_vec);
120 if val.is_some() {
121 g.store.expire(key, ttl);
122 crate::store_glue::commit_deadline(&mut g, key)?;
123 }
124 Ok(val)
125 }
126
127 // ---- set algebra (compose-side, not Store-side) ------------------
128
129 /// `SINTER key [key ...]` — set intersection. Reads each key's
130 /// members, computes the intersection in BTreeSet order
131 /// (sorted, no duplicates).
132 pub fn sinter(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
133 if keys.is_empty() {
134 return Ok(Vec::new());
135 }
136 let first: BTreeSet<Vec<u8>> = self.smembers(keys[0])?.into_iter().collect();
137 let mut acc = first;
138 for k in &keys[1..] {
139 if acc.is_empty() {
140 break;
141 }
142 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
143 acc.retain(|m| next.contains(m));
144 }
145 Ok(acc.into_iter().collect())
146 }
147
148 /// `SUNION key [key ...]` — set union over N sets.
149 pub fn sunion(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
150 let mut acc: BTreeSet<Vec<u8>> = BTreeSet::new();
151 for k in keys {
152 for m in self.smembers(k)? {
153 acc.insert(m);
154 }
155 }
156 Ok(acc.into_iter().collect())
157 }
158
159 /// `SDIFF key [key ...]` — `keys[0]` minus the union of every
160 /// subsequent set.
161 pub fn sdiff(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
162 if keys.is_empty() {
163 return Ok(Vec::new());
164 }
165 let mut acc: BTreeSet<Vec<u8>> = self.smembers(keys[0])?.into_iter().collect();
166 for k in &keys[1..] {
167 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
168 acc.retain(|m| !next.contains(m));
169 }
170 Ok(acc.into_iter().collect())
171 }
172
173 // ---- absolute-time TTL variants ----------------------------------
174
175 /// `EXPIREAT key unix_secs` — schedule expiry for the given
176 /// absolute UNIX wall-clock time. Returns `true` when the key
177 /// existed and the deadline was set; `false` when absent.
178 pub fn expireat(&self, key: &[u8], unix_secs: u64) -> KevyResult<bool> {
179 ensure_writable(self)?;
180 let mut g = self.wshard(key);
181 let unix_ms = unix_secs.saturating_mul(1000);
182 let ok = g.store.expire_at_unix_ms(key, unix_ms);
183 if ok {
184 let ts_str = format!("{unix_ms}");
185 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
186 }
187 Ok(ok)
188 }
189
190 /// `PEXPIREAT key unix_ms` — same as `expireat` but in
191 /// milliseconds.
192 pub fn pexpireat(&self, key: &[u8], unix_ms: u64) -> KevyResult<bool> {
193 ensure_writable(self)?;
194 let mut g = self.wshard(key);
195 let ok = g.store.expire_at_unix_ms(key, unix_ms);
196 if ok {
197 let ts_str = format!("{unix_ms}");
198 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
199 }
200 Ok(ok)
201 }
202
203 /// `PEXPIRE key ms` — relative TTL in milliseconds. (`expire`
204 /// takes `Duration`; this is the integer-ms variant matching
205 /// the Redis wire command.)
206 pub fn pexpire(&self, key: &[u8], ms: u64) -> KevyResult<bool> {
207 self.expire(key, Duration::from_millis(ms))
208 }
209
210 // ---- hash float increment ----------------------------------------
211
212 /// `HINCRBYFLOAT key field delta` — atomic float increment of a
213 /// hash field. Returns the post-increment value. Errors on
214 /// `NotFloat` when the field is present but not parseable.
215 pub fn hincrbyfloat(&self, key: &[u8], field: &[u8], delta: f64) -> KevyResult<f64> {
216 ensure_writable(self)?;
217 let mut g = self.wshard(key);
218 let new_val = g.store.hincrbyfloat(key, field, delta).map_err(store_err)?;
219 let delta_str = format!("{delta}");
220 commit_write(&mut g, &[b"HINCRBYFLOAT", key, field, delta_str.as_bytes()])?;
221 Ok(new_val)
222 }
223
224 // ---- list positional insert --------------------------------------
225
226 /// `LINSERT key BEFORE|AFTER pivot value` — insert `value` before
227 /// or after the first occurrence of `pivot` in the list. Returns:
228 /// - `Ok(new_len)` on success (`>= 1`);
229 /// - `Ok(0)` when `key` does not exist;
230 /// - `Ok(-1)` when `pivot` was not found in the list.
231 ///
232 /// ```
233 /// use kevy_embedded::{Config, InsertPosition, Store};
234 ///
235 /// let s = Store::open(Config::default())?;
236 /// s.rpush(b"l", &[b"a", b"c"])?;
237 /// assert_eq!(s.linsert(b"l", InsertPosition::Before, b"c", b"b")?, 3);
238 /// # Ok::<(), kevy_embedded::KevyError>(())
239 /// ```
240 pub fn linsert(
241 &self,
242 key: &[u8],
243 position: InsertPosition,
244 pivot: &[u8],
245 value: &[u8],
246 ) -> KevyResult<i64> {
247 ensure_writable(self)?;
248 let mut g = self.wshard(key);
249 let new_len = g.store.linsert(key, position, pivot, value).map_err(store_err)?;
250 if new_len > 0 {
251 let dir = match position {
252 InsertPosition::Before => b"BEFORE".as_slice(),
253 InsertPosition::After => b"AFTER".as_slice(),
254 };
255 commit_write(&mut g, &[b"LINSERT", key, dir, pivot, value])?;
256 }
257 Ok(new_len)
258 }
259
260 // ---- observability ----------------------------------------------
261
262 /// `Store::ping_us()` — return the round-trip duration of a
263 /// shard-0 read-lock acquire + release in **nanoseconds**, for
264 /// perfgate observability. Always returns immediately; the
265 /// duration reflects current shard-0 contention (= shorter when
266 /// idle, longer when many readers/writers compete).
267 pub fn ping_ns(&self) -> u128 {
268 let t = std::time::Instant::now();
269 let _g = self.lock();
270 t.elapsed().as_nanos()
271 }
272}