kevy_embedded/ops_keyspace.rs
1//! Cross-key operations: `copy`, `randomkey`, `unlink`, `touch`.
2//!
3//! These compose existing `kevy_store::Store` primitives at the
4//! embedded layer:
5//!
6//! - `copy` is `get` + (optional) read TTL + `set` on dst + `expire`
7//! on dst.
8//! - `randomkey` collects matching keys and picks one by index.
9//! - `unlink` is an alias for `del`; kevy has no async deletion, so
10//! sync delete is the unblocking semantic.
11//! - `touch` counts existing keys and reads bump LRU/LFU bookkeeping
12//! as a side effect.
13
14use std::io;
15
16#[cfg(not(target_arch = "wasm32"))]
17use crate::replica_glue::ensure_writable;
18use crate::store::{Store, commit_write};
19
20#[cfg(target_arch = "wasm32")]
21fn ensure_writable(_s: &Store) -> io::Result<()> { Ok(()) }
22
23impl Store {
24 /// `COPY src dst [REPLACE]` — copy `src`'s value (and TTL if any)
25 /// to `dst`. Returns `true` when the copy happened.
26 ///
27 /// Semantics:
28 /// - `false` if `src` doesn't exist.
29 /// - `false` if `dst` exists and `replace = false`.
30 /// - Preserves source TTL on the destination via `pexpireat`.
31 pub fn copy(&self, src: &[u8], dst: &[u8], replace: bool) -> io::Result<bool> {
32 ensure_writable(self)?;
33 // Read source under its own shard lock.
34 let src_val = match self.get(src)? {
35 Some(v) => v,
36 None => return Ok(false),
37 };
38 // Sample the source's TTL (ms since UNIX epoch) BEFORE the
39 // write — captures the deadline that should survive the copy.
40 let src_ttl_ms = self.ttl_ms(src);
41 // Veto if dst exists and replace is false.
42 if !replace {
43 // Use a fresh wshard on dst so this works cross-shard.
44 let mut g = self.wshard(dst);
45 if g.store.key_exists(dst) {
46 return Ok(false);
47 }
48 // AOF-log first (SET dst <value>), then write dst — both
49 // under dst's shard lock. Log-before-apply avoids cloning
50 // the value; an AOF error leaves memory untouched.
51 commit_write(&mut g, &[b"SET", dst, &src_val])?;
52 g.store.set(dst, src_val, None, false, false);
53 } else {
54 let mut g = self.wshard(dst);
55 commit_write(&mut g, &[b"SET", dst, &src_val])?;
56 g.store.set(dst, src_val, None, false, false);
57 }
58 // Re-attach absolute deadline if the source had one.
59 if src_ttl_ms > 0 {
60 let unix_ms = std::time::SystemTime::now()
61 .duration_since(std::time::UNIX_EPOCH)
62 .map(|d| d.as_millis() as u64)
63 .unwrap_or(0)
64 .saturating_add(src_ttl_ms as u64);
65 self.pexpireat(dst, unix_ms)?;
66 }
67 // The dst SET is AOF-logged above under dst's shard lock; the
68 // TTL re-attach goes through the `pexpireat` facade which logs
69 // its own PEXPIREAT. (v1.15.1: before this, the dst value was
70 // written to memory only and vanished on reopen.)
71 Ok(true)
72 }
73
74 /// `RANDOMKEY` — return a randomly-chosen existing key, or
75 /// `None` when the keyspace is empty.
76 ///
77 /// Implementation: snapshot all keys via `collect_keys`, then
78 /// pick a uniform index. For large keyspaces this is O(N); a
79 /// future ship can add a `key_at(rank)` Store method for O(1)
80 /// random pick.
81 pub fn randomkey(&self) -> Option<Vec<u8>> {
82 let keys = self.collect_keys(None, None);
83 if keys.is_empty() {
84 return None;
85 }
86 // Cheap PRNG via nanosecond clock — embedded in-process so
87 // this just needs decent distribution, not crypto strength.
88 let idx = std::time::SystemTime::now()
89 .duration_since(std::time::UNIX_EPOCH)
90 .map(|d| d.subsec_nanos() as usize)
91 .unwrap_or(0)
92 % keys.len();
93 Some(keys[idx].clone())
94 }
95
96 /// `UNLINK key [key ...]` — alias for [`Self::del`]. In Redis
97 /// this is the async (non-blocking) variant; kevy is in-process
98 /// so the sync `del` IS the unblocking semantic. Returns count
99 /// actually removed.
100 pub fn unlink(&self, keys: &[&[u8]]) -> io::Result<usize> {
101 self.del(keys)
102 }
103
104 /// `TOUCH key [key ...]` — count keys that exist. Side effect:
105 /// the existence check refreshes LRU/LFU bookkeeping on the
106 /// touched shards, matching Redis semantics.
107 pub fn touch(&self, keys: &[&[u8]]) -> io::Result<usize> {
108 self.exists(keys)
109 }
110}
111
112impl crate::Store {
113 /// v2.10 — order-insensitive prefix checksum for migration
114 /// verification: `(row_count, xor_of_row_digests)`. Matches the
115 /// server's `PREFIX.DIGEST` bit for bit (same canonicalization).
116 pub fn prefix_digest(&self, prefix: &[u8]) -> (u64, u64) {
117 let mut pat = prefix.to_vec();
118 pat.push(b'*');
119 let keys = self.keys(Some(&pat), None);
120 let mut xor = 0u64;
121 for key in &keys {
122 xor ^= self.row_digest_embedded(key);
123 }
124 (keys.len() as u64, xor)
125 }
126
127 fn row_digest_embedded(&self, key: &[u8]) -> u64 {
128 let mut h = FNV_OFFSET;
129 fnv(&mut h, key);
130 let ty = self.type_of(key);
131 fnv(&mut h, ty.as_bytes());
132 self.digest_row_body(key, ty, &mut h);
133 h
134 }
135
136 /// Fold one row's canonicalized value into the FNV state, per type
137 /// (hash fields and set members sort first; zset folds score bits
138 /// then member, rank order).
139 fn digest_row_body(&self, key: &[u8], ty: &str, h: &mut u64) {
140 match ty {
141 "string" => {
142 if let Ok(Some(v)) = self.get(key) {
143 fnv(h, &v);
144 }
145 }
146 "hash" => {
147 if let Ok(mut pairs) = self.hgetall(key) {
148 pairs.sort();
149 for (f, v) in pairs {
150 fnv(h, &f);
151 fnv(h, &v);
152 }
153 }
154 }
155 "list" => {
156 if let Ok(items) = self.lrange(key, 0, -1) {
157 for i in items {
158 fnv(h, &i);
159 }
160 }
161 }
162 "set" => {
163 if let Ok(mut ms) = self.smembers(key) {
164 ms.sort();
165 for m in ms {
166 fnv(h, &m);
167 }
168 }
169 }
170 "zset" => {
171 if let Ok(items) = self.zrange(key, 0, -1) {
172 for (member, score) in items {
173 fnv(h, &score.to_bits().to_le_bytes());
174 fnv(h, &member);
175 }
176 }
177 }
178 _ => {}
179 }
180 }
181}
182
183const FNV_OFFSET: u64 = 0xCBF2_9CE4_8422_2325;
184const FNV_PRIME: u64 = 0x0000_0100_0000_01B3;
185
186fn fnv(h: &mut u64, bytes: &[u8]) {
187 for &b in bytes {
188 *h ^= u64::from(b);
189 *h = h.wrapping_mul(FNV_PRIME);
190 }
191}