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