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