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::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 atomically
21 /// per-key. Each pair is logged independently to its shard's
22 /// AOF (no cross-shard atomic guarantee — a crash mid-call may
23 /// leave a prefix applied; matches Redis Cluster semantics).
24 pub fn mset(&self, pairs: &[(&[u8], &[u8])]) -> KevyResult<()> {
25 ensure_writable(self)?;
26 for (k, v) in pairs {
27 let mut g = self.wshard(k);
28 g.store.set(k, v.to_vec(), None, false, false);
29 commit_write(&mut g, &[b"SET", k, v])?;
30 }
31 Ok(())
32 }
33
34 /// `MGET key [key ...]` — return `Some(value)` per requested key
35 /// that's present, `None` per absent / wrong-type.
36 pub fn mget(&self, keys: &[&[u8]]) -> KevyResult<Vec<Option<Vec<u8>>>> {
37 let mut out = Vec::with_capacity(keys.len());
38 for k in keys {
39 out.push(
40 self.wshard(k)
41 .store
42 .get(k)
43 .map_err(store_err)?
44 .as_deref()
45 .map(<[u8]>::to_vec),
46 );
47 }
48 Ok(out)
49 }
50
51 // ---- keyspace introspection -------------------------------------
52
53 /// `KEYS pattern` — glob-match every key in the keyspace
54 /// (across all shards). `pattern = None` matches everything.
55 /// `limit = None` is unbounded; otherwise bounds the TOTAL
56 /// returned across shards. Glob syntax matches Redis (`*` /
57 /// `?` / `[abc]` / escape).
58 pub fn keys(&self, pattern: Option<&[u8]>, limit: Option<usize>) -> Vec<Vec<u8>> {
59 self.collect_keys(pattern, limit)
60 }
61
62 // ---- atomic get + TTL -------------------------------------------
63
64 /// `GETEX key TTL` — get the value and update the TTL atomically
65 /// (single lock cycle on the owning shard). Returns the value;
66 /// `None` when absent. AOF-logged as an absolute `PEXPIREAT` — a
67 /// relative frame re-anchors on every replay, handing the key its
68 /// full TTL back at each restart (the incident class the absolute
69 /// form exists to prevent; this was its last relative holdout).
70 pub fn getex(&self, key: &[u8], ttl: Duration) -> KevyResult<Option<Vec<u8>>> {
71 ensure_writable(self)?;
72 let mut g = self.wshard(key);
73 let val = g.store.get(key).map_err(store_err)?.as_deref().map(<[u8]>::to_vec);
74 if val.is_some() {
75 g.store.expire(key, ttl);
76 let ms = ttl.as_millis().min(u128::from(u64::MAX)) as u64;
77 let deadline = kevy_store::now_unix_ms().saturating_add(ms);
78 commit_write(&mut g, &[b"PEXPIREAT", key, deadline.to_string().as_bytes()])?;
79 }
80 Ok(val)
81 }
82
83 // ---- set algebra (compose-side, not Store-side) ------------------
84
85 /// `SINTER key [key ...]` — set intersection. Reads each key's
86 /// members, computes the intersection in BTreeSet order
87 /// (sorted, no duplicates).
88 pub fn sinter(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
89 if keys.is_empty() {
90 return Ok(Vec::new());
91 }
92 let first: BTreeSet<Vec<u8>> = self
93 .smembers(keys[0])?
94 .into_iter()
95 .collect();
96 let mut acc = first;
97 for k in &keys[1..] {
98 if acc.is_empty() {
99 break;
100 }
101 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
102 acc.retain(|m| next.contains(m));
103 }
104 Ok(acc.into_iter().collect())
105 }
106
107 /// `SUNION key [key ...]` — set union over N sets.
108 pub fn sunion(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
109 let mut acc: BTreeSet<Vec<u8>> = BTreeSet::new();
110 for k in keys {
111 for m in self.smembers(k)? {
112 acc.insert(m);
113 }
114 }
115 Ok(acc.into_iter().collect())
116 }
117
118 /// `SDIFF key [key ...]` — `keys[0]` minus the union of every
119 /// subsequent set.
120 pub fn sdiff(&self, keys: &[&[u8]]) -> KevyResult<Vec<Vec<u8>>> {
121 if keys.is_empty() {
122 return Ok(Vec::new());
123 }
124 let mut acc: BTreeSet<Vec<u8>> = self
125 .smembers(keys[0])?
126 .into_iter()
127 .collect();
128 for k in &keys[1..] {
129 let next: BTreeSet<Vec<u8>> = self.smembers(k)?.into_iter().collect();
130 acc.retain(|m| !next.contains(m));
131 }
132 Ok(acc.into_iter().collect())
133 }
134
135 // ---- absolute-time TTL variants ----------------------------------
136
137 /// `EXPIREAT key unix_secs` — schedule expiry for the given
138 /// absolute UNIX wall-clock time. Returns `true` when the key
139 /// existed and the deadline was set; `false` when absent.
140 pub fn expireat(&self, key: &[u8], unix_secs: u64) -> KevyResult<bool> {
141 ensure_writable(self)?;
142 let mut g = self.wshard(key);
143 let unix_ms = unix_secs.saturating_mul(1000);
144 let ok = g.store.expire_at_unix_ms(key, unix_ms);
145 if ok {
146 let ts_str = format!("{unix_ms}");
147 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
148 }
149 Ok(ok)
150 }
151
152 /// `PEXPIREAT key unix_ms` — same as `expireat` but in
153 /// milliseconds.
154 pub fn pexpireat(&self, key: &[u8], unix_ms: u64) -> KevyResult<bool> {
155 ensure_writable(self)?;
156 let mut g = self.wshard(key);
157 let ok = g.store.expire_at_unix_ms(key, unix_ms);
158 if ok {
159 let ts_str = format!("{unix_ms}");
160 commit_write(&mut g, &[b"PEXPIREAT", key, ts_str.as_bytes()])?;
161 }
162 Ok(ok)
163 }
164
165 /// `PEXPIRE key ms` — relative TTL in milliseconds. (`expire`
166 /// takes `Duration`; this is the integer-ms variant matching
167 /// the Redis wire command.)
168 pub fn pexpire(&self, key: &[u8], ms: u64) -> KevyResult<bool> {
169 self.expire(key, Duration::from_millis(ms))
170 }
171
172 // ---- hash float increment ----------------------------------------
173
174 /// `HINCRBYFLOAT key field delta` — atomic float increment of a
175 /// hash field. Returns the post-increment value. Errors on
176 /// `NotFloat` when the field is present but not parseable.
177 pub fn hincrbyfloat(
178 &self,
179 key: &[u8],
180 field: &[u8],
181 delta: f64,
182 ) -> KevyResult<f64> {
183 ensure_writable(self)?;
184 let mut g = self.wshard(key);
185 let new_val = g
186 .store
187 .hincrbyfloat(key, field, delta)
188 .map_err(store_err)?;
189 let delta_str = format!("{delta}");
190 commit_write(&mut g, &[b"HINCRBYFLOAT", key, field, delta_str.as_bytes()])?;
191 Ok(new_val)
192 }
193
194 // ---- list positional insert --------------------------------------
195
196 /// `LINSERT key BEFORE|AFTER pivot value` — insert `value` before
197 /// or after the first occurrence of `pivot` in the list. Returns:
198 /// - `Ok(new_len)` on success (`>= 1`);
199 /// - `Ok(0)` when `key` does not exist;
200 /// - `Ok(-1)` when `pivot` was not found in the list.
201 ///
202 /// `before = true` matches Redis `LINSERT … BEFORE`, `false`
203 /// matches `LINSERT … AFTER`.
204 pub fn linsert(
205 &self,
206 key: &[u8],
207 before: bool,
208 pivot: &[u8],
209 value: &[u8],
210 ) -> KevyResult<i64> {
211 ensure_writable(self)?;
212 let mut g = self.wshard(key);
213 let new_len = g
214 .store
215 .linsert(key, before, pivot, value)
216 .map_err(store_err)?;
217 if new_len > 0 {
218 let dir = if before { b"BEFORE".as_slice() } else { b"AFTER".as_slice() };
219 commit_write(&mut g, &[b"LINSERT", key, dir, pivot, value])?;
220 }
221 Ok(new_len)
222 }
223
224 // ---- observability ----------------------------------------------
225
226 /// `Store::ping_us()` — return the round-trip duration of a
227 /// shard-0 read-lock acquire + release in **nanoseconds**, for
228 /// perfgate observability. Always returns immediately; the
229 /// duration reflects current shard-0 contention (= shorter when
230 /// idle, longer when many readers/writers compete).
231 pub fn ping_ns(&self) -> u128 {
232 let t = std::time::Instant::now();
233 let _g = self.lock();
234 t.elapsed().as_nanos()
235 }
236}