yo/keyspace.rs
1//! The Redis string keyspace, from the embedded side.
2//!
3//! This is Y23 where you can see it. A program that calls [`Strings::incr`]
4//! reaches the same `yo_kv::Keyspace::incr` that `INCR` off a socket reaches,
5//! not a second implementation that agrees with it today. The difference
6//! between the two callers is a socket, a parser and a reply, and this side
7//! pays none of them: an embedded `INCR` is a probe, an add and a store.
8//!
9//! It is also where inline execution mode (`15` section 7) actually happens.
10//! The calling thread is the shard, so there is no queue, no message and no
11//! wakeup, which is what makes the number in `bench/00` a number about the
12//! store rather than about a channel.
13//!
14//! # About the clock
15//!
16//! `04` section 5 says the clock is read once per turn of the shard loop and
17//! never on the data path, because a clock read is tens of nanoseconds against
18//! a budget of a hundred and fifty. Inline mode has no loop, so one call is one
19//! turn, and reading the clock per call would double the cost of a `GET`.
20//!
21//! So the clock is read only when its answer can be observed, which is when
22//! some key in the keyspace has a deadline. A database that has never been
23//! given one cannot have an expired key, and its clock never moves and is never
24//! read. The first call that sets a deadline turns the reads on, and from then
25//! on the cost is the same one a server pays per batch.
26
27use std::time::Duration;
28
29use yo_common::{Code, Error, Result};
30use yo_kv::{Expire, SetOptions, Str};
31
32use crate::db::Handle;
33
34/// The keyspace every Redis string command works on.
35///
36/// Cheap to clone, and every clone is the same keyspace. Keys are byte strings
37/// the way Redis's are, so anything that is bytes will do: `"hits"`, a
38/// `String`, a `&[u8]` or a `Vec<u8>`.
39///
40/// ```
41/// let db = yo::open(yo::MEMORY)?;
42/// let keys = db.strings();
43///
44/// keys.set("greeting", "hello")?;
45/// assert_eq!(keys.get("greeting")?.as_deref(), Some(&b"hello"[..]));
46///
47/// assert_eq!(keys.incr("hits")?, 1);
48/// assert_eq!(keys.incr_by("hits", 9)?, 10);
49/// # Ok::<(), yo::Error>(())
50/// ```
51#[derive(Clone)]
52pub struct Strings {
53 pub(crate) db: Handle,
54}
55
56impl core::fmt::Debug for Strings {
57 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
58 f.debug_struct("Strings")
59 .field("keys", &self.len().ok())
60 .finish()
61 }
62}
63
64impl Strings {
65 /// Read a value.
66 ///
67 /// Owned, because most callers want the bytes to outlive the call.
68 /// [`Strings::with`] is the same read without the copy.
69 ///
70 /// # Errors
71 ///
72 /// [`Code::Invalid`] if called from inside a callback that is already
73 /// holding this database.
74 pub fn get(&self, key: impl AsRef<[u8]>) -> Result<Option<Vec<u8>>> {
75 self.with(key, |v| v.to_vec())
76 }
77
78 /// Read a value without copying it, by handing what is in the record to
79 /// `f`.
80 ///
81 /// The view is [`Str`], which is either the bytes where they lie or the
82 /// integer an int encoded value holds. This is the read the G6 budget is
83 /// about, and it allocates nothing at all.
84 ///
85 /// ```
86 /// let db = yo::open(yo::MEMORY)?;
87 /// let keys = db.strings();
88 /// keys.set("greeting", "hello")?;
89 ///
90 /// assert_eq!(keys.with("greeting", |v| v.len())?, Some(5));
91 /// # Ok::<(), yo::Error>(())
92 /// ```
93 ///
94 /// # Errors
95 ///
96 /// [`Code::Invalid`] if called from inside a callback that is already
97 /// holding this database.
98 pub fn with<R>(
99 &self,
100 key: impl AsRef<[u8]>,
101 f: impl FnOnce(Str<'_>) -> R,
102 ) -> Result<Option<R>> {
103 self.db
104 .run(|inner| Ok(inner.strings.get(key.as_ref())?.map(f)))
105 }
106
107 /// Whether a key is there and has not expired.
108 ///
109 /// # Errors
110 ///
111 /// As [`Strings::get`].
112 pub fn exists(&self, key: impl AsRef<[u8]>) -> Result<bool> {
113 self.db.run(|inner| Ok(inner.strings.exists(key.as_ref())))
114 }
115
116 /// The length of a value in bytes, which is zero for a key that is not
117 /// there. `STRLEN`.
118 ///
119 /// # Errors
120 ///
121 /// As [`Strings::get`].
122 pub fn len_of(&self, key: impl AsRef<[u8]>) -> Result<usize> {
123 self.db.run(|inner| inner.strings.strlen(key.as_ref()))
124 }
125
126 /// Store a value, clearing any deadline the key had. Plain `SET`.
127 ///
128 /// # Errors
129 ///
130 /// [`Code::Full`] for a value past [`yo_kv::STRING_MAX`].
131 pub fn set(&self, key: impl AsRef<[u8]>, value: impl AsRef<[u8]>) -> Result<()> {
132 self.db
133 .run(|inner| inner.strings.set_plain(key.as_ref(), value.as_ref()))
134 }
135
136 /// Store a value only if the key is missing, and say whether it was.
137 /// `SET NX`.
138 ///
139 /// # Errors
140 ///
141 /// As [`Strings::set`].
142 pub fn set_if_missing(&self, key: impl AsRef<[u8]>, value: impl AsRef<[u8]>) -> Result<bool> {
143 self.db
144 .run(|inner| inner.strings.setnx(key.as_ref(), value.as_ref()))
145 }
146
147 /// Store a value that expires after `ttl`. `SET PX`.
148 ///
149 /// This is the call that turns the clock on for this database, because it
150 /// is the first moment a deadline can be observed.
151 ///
152 /// # Errors
153 ///
154 /// As [`Strings::set`], and [`Code::Invalid`] for a `ttl` past what fits in
155 /// a millisecond deadline.
156 pub fn set_for(
157 &self,
158 key: impl AsRef<[u8]>,
159 value: impl AsRef<[u8]>,
160 ttl: Duration,
161 ) -> Result<()> {
162 let ms = u64::try_from(ttl.as_millis()).map_err(|_| too_far())?;
163 self.db.deadlines(|inner| {
164 let at = inner
165 .strings
166 .clock()
167 .now_ms()
168 .checked_add(ms)
169 .ok_or_else(too_far)?;
170 inner
171 .strings
172 .set(
173 key.as_ref(),
174 value.as_ref(),
175 SetOptions::PLAIN.expiring(Expire::At(at)),
176 )
177 .map(|_| ())
178 })
179 }
180
181 /// How long a key has left, or `None` if it has no deadline or is not
182 /// there. `PTTL`.
183 ///
184 /// [`Keys::ttl`](crate::Keys::ttl) is the one that tells those two apart,
185 /// and it works on a key of any type rather than only on a string.
186 ///
187 /// # Errors
188 ///
189 /// As [`Strings::get`].
190 pub fn ttl(&self, key: impl AsRef<[u8]>) -> Result<Option<Duration>> {
191 self.db.run(|inner| {
192 let now = inner.strings.clock().now_ms();
193 Ok(inner
194 .strings
195 .expire_at(key.as_ref())
196 .map(|at| Duration::from_millis(at.saturating_sub(now))))
197 })
198 }
199
200 /// Store a value and hand back what was there. `GETSET`.
201 ///
202 /// # Errors
203 ///
204 /// As [`Strings::set`].
205 pub fn replace(
206 &self,
207 key: impl AsRef<[u8]>,
208 value: impl AsRef<[u8]>,
209 ) -> Result<Option<Vec<u8>>> {
210 self.db
211 .run(|inner| inner.strings.getset(key.as_ref(), value.as_ref()))
212 }
213
214 /// Remove a key and hand back what it held. `GETDEL`.
215 ///
216 /// # Errors
217 ///
218 /// As [`Strings::get`].
219 pub fn take(&self, key: impl AsRef<[u8]>) -> Result<Option<Vec<u8>>> {
220 self.db.run(|inner| inner.strings.getdel(key.as_ref()))
221 }
222
223 /// Remove a key, and say whether it was there. `DEL`.
224 ///
225 /// # Errors
226 ///
227 /// As [`Strings::get`].
228 pub fn del(&self, key: impl AsRef<[u8]>) -> Result<bool> {
229 self.db.run(|inner| Ok(inner.strings.del(key.as_ref())))
230 }
231
232 /// Store several values, all of them or none. `MSET`.
233 ///
234 /// The pairs reach the store as an iterator rather than a `Vec`, which is
235 /// the same thing the wire layer does and for the same reason: `MSET` is on
236 /// the gate list and an API that forces an allocation to call it is the
237 /// wrong API.
238 ///
239 /// ```
240 /// let db = yo::open(yo::MEMORY)?;
241 /// let keys = db.strings();
242 ///
243 /// keys.set_many(&[("a", "1"), ("b", "2")])?;
244 /// assert_eq!(keys.get("b")?.as_deref(), Some(&b"2"[..]));
245 /// # Ok::<(), yo::Error>(())
246 /// ```
247 ///
248 /// # Errors
249 ///
250 /// As [`Strings::set`], and nothing is written if any pair fails.
251 pub fn set_many<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, pairs: &[(K, V)]) -> Result<()> {
252 self.db.run(|inner| {
253 inner
254 .strings
255 .mset(pairs.iter().map(|(k, v)| (k.as_ref(), v.as_ref())))
256 })
257 }
258
259 /// Read several values in one call. `MGET`.
260 ///
261 /// # Errors
262 ///
263 /// As [`Strings::get`].
264 pub fn get_many<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<Option<Vec<u8>>>> {
265 self.db.run(|inner| {
266 Ok(keys
267 .iter()
268 .map(|k| inner.strings.mget_one(k.as_ref()).map(|v| v.to_vec()))
269 .collect())
270 })
271 }
272
273 /// Add one and hand back the result. `INCR`.
274 ///
275 /// A key that is not there counts as zero, which is Redis's rule and not a
276 /// convenience: it is what makes a counter usable without a create step.
277 ///
278 /// # Errors
279 ///
280 /// [`Code::Invalid`] when the value is not an integer, or when adding would
281 /// leave the range of an `i64`.
282 pub fn incr(&self, key: impl AsRef<[u8]>) -> Result<i64> {
283 self.db.run(|inner| inner.strings.incr(key.as_ref()))
284 }
285
286 /// Add `by` and hand back the result. `INCRBY`, and `DECRBY` for a negative
287 /// `by`.
288 ///
289 /// # Errors
290 ///
291 /// As [`Strings::incr`].
292 pub fn incr_by(&self, key: impl AsRef<[u8]>, by: i64) -> Result<i64> {
293 self.db.run(|inner| inner.strings.incrby(key.as_ref(), by))
294 }
295
296 /// Subtract one and hand back the result. `DECR`.
297 ///
298 /// # Errors
299 ///
300 /// As [`Strings::incr`].
301 pub fn decr(&self, key: impl AsRef<[u8]>) -> Result<i64> {
302 self.db.run(|inner| inner.strings.decr(key.as_ref()))
303 }
304
305 /// Add `by` to a float counter and hand back the result. `INCRBYFLOAT`.
306 ///
307 /// # Errors
308 ///
309 /// [`Code::Invalid`] when the value is not a float, or when the result
310 /// would be infinite or not a number.
311 pub fn incr_by_float(&self, key: impl AsRef<[u8]>, by: f64) -> Result<f64> {
312 self.db
313 .run(|inner| inner.strings.incrbyfloat(key.as_ref(), by))
314 }
315
316 /// Append to a value and hand back its new length. `APPEND`.
317 ///
318 /// # Errors
319 ///
320 /// As [`Strings::set`].
321 pub fn append(&self, key: impl AsRef<[u8]>, tail: impl AsRef<[u8]>) -> Result<usize> {
322 self.db
323 .run(|inner| inner.strings.append(key.as_ref(), tail.as_ref()))
324 }
325
326 /// How many keys the keyspace holds, expired ones that nothing has touched
327 /// yet included. `DBSIZE`.
328 ///
329 /// # Errors
330 ///
331 /// As [`Strings::get`].
332 pub fn len(&self) -> Result<usize> {
333 self.db.run(|inner| Ok(inner.strings.len()))
334 }
335
336 /// Whether the keyspace is empty.
337 ///
338 /// # Errors
339 ///
340 /// As [`Strings::get`].
341 pub fn is_empty(&self) -> Result<bool> {
342 self.db.run(|inner| Ok(inner.strings.is_empty()))
343 }
344
345 /// Keys reclaimed by running into them after their deadline.
346 ///
347 /// # Errors
348 ///
349 /// As [`Strings::get`].
350 pub fn expired_keys(&self) -> Result<u64> {
351 self.db.run(|inner| Ok(inner.strings.expired_keys()))
352 }
353}
354
355fn too_far() -> Error {
356 Error::new(
357 Code::Invalid,
358 "that deadline is further away than a millisecond timestamp reaches",
359 )
360}
361
362#[cfg(test)]
363mod tests {
364 use super::*;
365 use crate::{MEMORY, open};
366
367 #[test]
368 fn the_string_commands_are_the_ones_a_redis_client_would_send() {
369 let db = open(MEMORY).unwrap();
370 let keys = db.strings();
371
372 keys.set("greeting", "hello").unwrap();
373 assert_eq!(
374 keys.get("greeting").unwrap().as_deref(),
375 Some(&b"hello"[..])
376 );
377 assert_eq!(keys.len_of("greeting").unwrap(), 5);
378 assert_eq!(keys.append("greeting", " there").unwrap(), 11);
379 assert!(keys.exists("greeting").unwrap());
380
381 assert!(!keys.set_if_missing("greeting", "other").unwrap());
382 assert!(keys.set_if_missing("fresh", "yes").unwrap());
383
384 assert_eq!(
385 keys.replace("greeting", "hi").unwrap().as_deref(),
386 Some(&b"hello there"[..])
387 );
388 assert_eq!(keys.take("greeting").unwrap().as_deref(), Some(&b"hi"[..]));
389 assert!(!keys.exists("greeting").unwrap());
390 assert!(!keys.del("greeting").unwrap());
391 }
392
393 #[test]
394 fn a_counter_starts_at_zero_without_being_created() {
395 let db = open(MEMORY).unwrap();
396 let keys = db.strings();
397
398 assert_eq!(keys.incr("hits").unwrap(), 1);
399 assert_eq!(keys.incr_by("hits", 9).unwrap(), 10);
400 assert_eq!(keys.decr("hits").unwrap(), 9);
401 assert_eq!(keys.incr_by_float("ratio", 1.5).unwrap(), 1.5);
402
403 keys.set("word", "nope").unwrap();
404 let e = keys.incr("word").expect_err("that is not a number");
405 assert_eq!(e.code(), Code::Invalid);
406 assert_eq!(e.message(), "value is not an integer or out of range");
407 }
408
409 #[test]
410 fn many_keys_at_once_go_in_and_come_out_together() {
411 let db = open(MEMORY).unwrap();
412 let keys = db.strings();
413
414 keys.set_many(&[("a", "1"), ("b", "2"), ("c", "3")])
415 .unwrap();
416 let got = keys.get_many(&["a", "c", "missing"]).unwrap();
417 assert_eq!(got[0].as_deref(), Some(&b"1"[..]));
418 assert_eq!(got[1].as_deref(), Some(&b"3"[..]));
419 assert_eq!(got[2], None);
420 assert_eq!(keys.len().unwrap(), 3);
421 assert!(!keys.is_empty().unwrap());
422 }
423
424 /// The clock policy: a keyspace with no deadline in it never reads the
425 /// clock, and the first `set_for` is what turns the reads on.
426 #[test]
427 fn a_deadline_is_what_makes_time_start_moving() {
428 let db = open(MEMORY).unwrap();
429 let keys = db.strings();
430
431 keys.set("plain", "v").unwrap();
432 assert_eq!(keys.ttl("plain").unwrap(), None);
433 assert!(!db.reads_the_clock());
434
435 keys.set_for("short", "v", Duration::from_millis(50))
436 .unwrap();
437 assert!(db.reads_the_clock());
438 assert!(keys.ttl("short").unwrap().unwrap() <= Duration::from_millis(50));
439
440 std::thread::sleep(Duration::from_millis(60));
441 assert_eq!(keys.get("short").unwrap(), None);
442 assert_eq!(keys.expired_keys().unwrap(), 1);
443 // The key with no deadline is not affected by any of that.
444 assert_eq!(keys.get("plain").unwrap().as_deref(), Some(&b"v"[..]));
445 }
446
447 #[test]
448 fn a_key_is_bytes_and_not_only_text() {
449 let db = open(MEMORY).unwrap();
450 let keys = db.strings();
451
452 keys.set(b"\x00\xff", vec![1u8, 2, 3]).unwrap();
453 assert_eq!(
454 keys.get(b"\x00\xff").unwrap().as_deref(),
455 Some(&[1u8, 2, 3][..])
456 );
457 assert_eq!(keys.with(b"\x00\xff", |v| v.len()).unwrap(), Some(3));
458 assert!(format!("{keys:?}").contains("keys"));
459 }
460}