kevy_store/hash_ttl.rs
1//! Per-field hash TTLs (`HEXPIRE` / `HPEXPIRE` / `HPEXPIREAT` /
2//! `HTTL` / `HPERSIST`, Redis 7.4 semantics).
3//!
4//! Storage: a store-level sidecar `hfttl: key → (field → absolute
5//! unix-ms deadline)` holding ONLY keys that have at least one
6//! field TTL — a store that never uses the feature pays one
7//! `is_empty()` branch per hash access and nothing else.
8//!
9//! Enforcement follows the key-TTL discipline exactly:
10//! - **lazy on access**: every hash op calls [`Store::purge_hash_ttl`]
11//! first, which removes expired fields from the hash (and the
12//! sidecar) — mutating-on-read like `live_entry`. No AOF frames are
13//! written for lazy purges: the `HPEXPIREAT` frame that created the
14//! deadline is already in the log, so a replay reconstructs the
15//! sidecar and purges identically (deterministic).
16//! - **actively by the reaper**: [`Store::tick_hash_ttl`] sweeps due
17//! fields and reports what it removed so the caller can log the
18//! `HDEL` effect (server tick / embedded reaper).
19//! - **cleared on overwrite**: `HSET`/`HINCRBY*` on a field discards
20//! that field's TTL (Redis 7.4 behavior) via
21//! [`Store::clear_hash_field_ttls`]; whole-key removal drops the
22//! sidecar entry in `remove_entry`.
23
24// The discarded value is the operation's own count — how many fields
25// went, how many members landed — and the caller returns its own.
26#![expect(
27 clippy::let_underscore_must_use,
28 reason = "the discarded value is a count, not an error report"
29)]
30
31#[cfg(not(feature = "std"))]
32use crate::nostd_prelude::*;
33use crate::{SmallBytes, Store, StoreError, Value, now_unix_ms};
34
35/// Per-field reply codes for `HEXPIRE`-family calls (Redis 7.4):
36/// `-2` key or field missing, `0` condition (NX/XX/GT/LT) not met,
37/// `1` deadline set, `2` field deleted (deadline already due).
38///
39/// ```
40/// use kevy_store::{HExpireCode, HExpireCond, Store};
41/// let mut s = Store::new();
42/// s.hset(b"h", &[(b"a".as_slice(), b"1".as_slice()), (b"b", b"2")])?;
43/// let far = kevy_store::now_unix_ms() + 60_000;
44/// let codes: Vec<HExpireCode> =
45/// s.hexpire_at(b"h", &[b"a".as_slice(), b"zz"], far, HExpireCond::Always)?;
46/// assert_eq!(codes, [1, -2]);
47/// assert_eq!(s.hexpire_at(b"h", &[b"b".as_slice()], 1, HExpireCond::Always)?, [2]);
48/// # Ok::<(), kevy_store::StoreError>(())
49/// ```
50pub type HExpireCode = i8;
51
52/// Condition flags for `HEXPIRE` (`NX`/`XX`/`GT`/`LT`; at most one).
53///
54/// ```
55/// assert_eq!(kevy_store::HExpireCond::default(), kevy_store::HExpireCond::Always);
56/// ```
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
58#[non_exhaustive]
59pub enum HExpireCond {
60 /// Unconditional.
61 ///
62 /// ```
63 /// use kevy_store::{HExpireCond, Store};
64 /// let mut s = Store::new();
65 /// s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())])?;
66 /// let t = kevy_store::now_unix_ms() + 60_000;
67 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Always)?, [1]);
68 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t + 1, HExpireCond::Always)?, [1]);
69 /// # Ok::<(), kevy_store::StoreError>(())
70 /// ```
71 #[default]
72 Always,
73 /// Only when the field has no TTL.
74 ///
75 /// ```
76 /// use kevy_store::{HExpireCond, Store};
77 /// let mut s = Store::new();
78 /// s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())])?;
79 /// let t = kevy_store::now_unix_ms() + 60_000;
80 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Nx)?, [1]);
81 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Nx)?, [0]);
82 /// # Ok::<(), kevy_store::StoreError>(())
83 /// ```
84 Nx,
85 /// Only when the field already has a TTL.
86 ///
87 /// ```
88 /// use kevy_store::{HExpireCond, Store};
89 /// let mut s = Store::new();
90 /// s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())])?;
91 /// let t = kevy_store::now_unix_ms() + 60_000;
92 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Xx)?, [0]);
93 /// s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Always)?;
94 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Xx)?, [1]);
95 /// # Ok::<(), kevy_store::StoreError>(())
96 /// ```
97 Xx,
98 /// Only when the new deadline is later than the current one
99 /// (no TTL counts as infinitely late — GT never replaces it).
100 ///
101 /// ```
102 /// use kevy_store::{HExpireCond, Store};
103 /// let mut s = Store::new();
104 /// s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())])?;
105 /// let t = kevy_store::now_unix_ms() + 60_000;
106 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Gt)?, [0]);
107 /// s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Always)?;
108 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t + 1, HExpireCond::Gt)?, [1]);
109 /// # Ok::<(), kevy_store::StoreError>(())
110 /// ```
111 Gt,
112 /// Only when the new deadline is earlier (no TTL = always).
113 ///
114 /// ```
115 /// use kevy_store::{HExpireCond, Store};
116 /// let mut s = Store::new();
117 /// s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())])?;
118 /// let t = kevy_store::now_unix_ms() + 60_000;
119 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t, HExpireCond::Lt)?, [1]);
120 /// assert_eq!(s.hexpire_at(b"h", &[b"f".as_slice()], t + 1, HExpireCond::Lt)?, [0]);
121 /// # Ok::<(), kevy_store::StoreError>(())
122 /// ```
123 Lt,
124}
125
126impl HExpireCond {
127 /// The command keyword for this condition (`NX` / `XX` / `GT` /
128 /// `LT`), or `None` for the unconditional form, which has none.
129 ///
130 /// ```
131 /// use kevy_store::HExpireCond;
132 /// assert_eq!(HExpireCond::Gt.keyword(), Some("GT"));
133 /// assert_eq!(HExpireCond::Always.keyword(), None);
134 /// ```
135 pub fn keyword(self) -> Option<&'static str> {
136 match self {
137 Self::Always => None,
138 Self::Nx => Some("NX"),
139 Self::Xx => Some("XX"),
140 Self::Gt => Some("GT"),
141 Self::Lt => Some("LT"),
142 }
143 }
144}
145
146impl Store {
147 /// Does this hash field exist (ignoring TTL state)?
148 fn hash_has_field(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError> {
149 match self.tier_serve(key, crate::value::COLD_TAG_HASH)? {
150 None => Ok(false),
151 Some(e) => match &e.value {
152 Value::Hash(h) => Ok(h.get(field).is_some()),
153 Value::SegHash(h) => Ok(h.get(field).is_some()),
154 Value::SmallHashInline(h) => Ok(h.get(field).is_some()),
155 Value::PackedRow(r) => Ok(r.has_named(field)),
156 _ => Err(StoreError::WrongType),
157 },
158 }
159 }
160
161 /// Set per-field deadlines (absolute unix-ms). One code per field,
162 /// request order. Due-or-past deadlines delete the field
163 /// immediately (code `2`, Redis semantics).
164 pub fn hexpire_at(
165 &mut self,
166 key: &[u8],
167 fields: &[&[u8]],
168 deadline_ms: u64,
169 cond: HExpireCond,
170 ) -> Result<Vec<HExpireCode>, StoreError> {
171 self.purge_hash_ttl(key);
172 let mut codes = Vec::with_capacity(fields.len());
173 let now = now_unix_ms();
174 for f in fields {
175 if !self.hash_has_field(key, f)? {
176 codes.push(-2);
177 continue;
178 }
179 let current = self.hfttl.get(key).and_then(|m| m.get(*f)).copied();
180 let pass = match cond {
181 HExpireCond::Always => true,
182 HExpireCond::Nx => current.is_none(),
183 HExpireCond::Xx => current.is_some(),
184 HExpireCond::Gt => current.is_some_and(|c| deadline_ms > c),
185 HExpireCond::Lt => current.is_none_or(|c| deadline_ms < c),
186 };
187 if !pass {
188 codes.push(0);
189 continue;
190 }
191 if deadline_ms <= now {
192 if let Some(m) = self.hfttl.get_mut(key) {
193 m.remove(*f);
194 }
195 self.hdel(key, &[f])?;
196 codes.push(2);
197 continue;
198 }
199 hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(f), deadline_ms);
200 codes.push(1);
201 }
202 self.prune_hfttl_key(key);
203 Ok(codes)
204 }
205
206 /// Each field's absolute deadline (unix ms) as stored, `None` for a
207 /// field with no TTL. A read with no side effects: unlike [`Self::hpttl`]
208 /// it purges nothing, so a deadline already passed is still reported,
209 /// and removing that field stays the expiry sweep's job.
210 ///
211 /// ```
212 /// let mut s = kevy_store::Store::new();
213 /// s.hset(b"h", &[(b"f", b"v"), (b"g", b"w")]).unwrap();
214 /// let at = kevy_store::now_unix_ms() + 60_000;
215 /// s.hexpire_at(b"h", &[b"f"], at, kevy_store::HExpireCond::Always).unwrap();
216 /// assert_eq!(s.hash_field_deadlines(b"h", &[b"f", b"g"]), [Some(at), None]);
217 /// ```
218 pub fn hash_field_deadlines(&self, key: &[u8], fields: &[&[u8]]) -> Vec<Option<u64>> {
219 let per_key = self.hfttl.get(key);
220 fields.iter().map(|f| per_key.and_then(|m| m.get(*f)).copied()).collect()
221 }
222
223 /// Remaining TTL per field: `-2` key/field missing, `-1` no TTL,
224 /// else remaining ms.
225 pub fn hpttl(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<Vec<i64>, StoreError> {
226 self.purge_hash_ttl(key);
227 let now = now_unix_ms();
228 let mut out = Vec::with_capacity(fields.len());
229 for f in fields {
230 if !self.hash_has_field(key, f)? {
231 out.push(-2);
232 continue;
233 }
234 match self.hfttl.get(key).and_then(|m| m.get(*f)) {
235 Some(&d) => out.push(d.saturating_sub(now) as i64),
236 None => out.push(-1),
237 }
238 }
239 Ok(out)
240 }
241
242 /// Clear per-field TTLs: `-2` missing, `-1` had no TTL, `1` cleared.
243 pub fn hpersist(
244 &mut self,
245 key: &[u8],
246 fields: &[&[u8]],
247 ) -> Result<Vec<HExpireCode>, StoreError> {
248 self.purge_hash_ttl(key);
249 let mut out = Vec::with_capacity(fields.len());
250 for f in fields {
251 if !self.hash_has_field(key, f)? {
252 out.push(-2);
253 continue;
254 }
255 let had = self.hfttl.get_mut(key).and_then(|m| m.remove(*f)).is_some();
256 out.push(if had { 1 } else { -1 });
257 }
258 self.prune_hfttl_key(key);
259 Ok(out)
260 }
261
262 /// Lazy enforcement hook — call at the top of every hash op.
263 /// Removes expired fields from the hash + sidecar. One `is_empty`
264 /// branch when the feature is unused.
265 pub(crate) fn purge_hash_ttl(&mut self, key: &[u8]) {
266 if self.hfttl.is_empty() {
267 return;
268 }
269 let now = now_unix_ms();
270 let due: Vec<Vec<u8>> = match self.hfttl.get(key) {
271 None => return,
272 Some(m) => m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect(),
273 };
274 if due.is_empty() {
275 return;
276 }
277 // Sidecar first: `hdel` re-enters this fn, and a clean sidecar
278 // makes the re-entry a no-op (no recursion).
279 if let Some(m) = self.hfttl.get_mut(key) {
280 for f in &due {
281 m.remove(f.as_slice());
282 }
283 }
284 self.prune_hfttl_key(key);
285 let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
286 let _ = self.hdel(key, &due_refs);
287 }
288
289 /// Overwrite hook — `HSET`/`HINCRBY*` on a field discards its TTL.
290 pub(crate) fn clear_hash_field_ttls(&mut self, key: &[u8], fields: &[&[u8]]) {
291 if self.hfttl.is_empty() {
292 return;
293 }
294 if let Some(m) = self.hfttl.get_mut(key) {
295 for f in fields {
296 m.remove(*f);
297 }
298 }
299 self.prune_hfttl_key(key);
300 }
301
302 /// Whole-key hook — key removed/overwritten wholesale.
303 pub(crate) fn clear_hash_key_ttls(&mut self, key: &[u8]) {
304 if self.hfttl.is_empty() {
305 return;
306 }
307 self.hfttl.remove(key);
308 }
309
310 fn prune_hfttl_key(&mut self, key: &[u8]) {
311 if self.hfttl.get(key).is_some_and(kevy_map_is_empty) {
312 self.hfttl.remove(key);
313 }
314 }
315
316 /// Reaper sweep: remove every due field store-wide; returns
317 /// `(key, removed fields)` pairs so the caller logs `HDEL` effects.
318 pub fn tick_hash_ttl(&mut self, max_keys: usize) -> Vec<(Vec<u8>, Vec<Vec<u8>>)> {
319 if self.hfttl.is_empty() {
320 return Vec::new();
321 }
322 let now = now_unix_ms();
323 let candidates: Vec<Vec<u8>> = self
324 .hfttl
325 .iter()
326 .filter(|(_, m)| m.iter().any(|(_, d)| *d <= now))
327 .take(max_keys)
328 .map(|(k, _)| k.to_vec())
329 .collect();
330 let mut out = Vec::with_capacity(candidates.len());
331 for k in candidates {
332 let due: Vec<Vec<u8>> = self
333 .hfttl
334 .get(k.as_slice())
335 .map(|m| m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect())
336 .unwrap_or_default();
337 if due.is_empty() {
338 continue;
339 }
340 if let Some(m) = self.hfttl.get_mut(k.as_slice()) {
341 for f in &due {
342 m.remove(f.as_slice());
343 }
344 }
345 self.prune_hfttl_key(&k);
346 let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
347 let _ = self.hdel(&k, &due_refs);
348 out.push((k, due));
349 }
350 out
351 }
352
353 /// Snapshot loader hook: restore one field TTL (deadlines already
354 /// absolute unix-ms; past deadlines simply purge on first access).
355 pub fn load_hash_field_ttl(&mut self, key: &[u8], field: &[u8], deadline_ms: u64) {
356 hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(field), deadline_ms);
357 }
358
359 /// Snapshot support: visit every live (key, field, deadline_ms).
360 pub fn hash_ttl_each<F: FnMut(&[u8], &[u8], u64)>(&self, mut f: F) {
361 for (k, m) in self.hfttl.iter() {
362 for (field, &d) in m.iter() {
363 f(k.as_slice(), field.as_slice(), d);
364 }
365 }
366 }
367}
368
369fn kevy_map_is_empty(m: &crate::KevyMap<SmallBytes, u64>) -> bool {
370 m.iter().next().is_none()
371}
372
373/// `entry().or_default()` over both side-map backends — `KevyMap` (the
374/// `no_std` arm) has no entry API, so that arm inserts-if-absent and
375/// re-probes.
376fn hfttl_slot<'a>(
377 hfttl: &'a mut crate::SideMap<SmallBytes, kevy_map::KevyMap<SmallBytes, u64>>,
378 key: &[u8],
379) -> &'a mut kevy_map::KevyMap<SmallBytes, u64> {
380 #[cfg(feature = "std")]
381 {
382 hfttl.entry(SmallBytes::from_slice(key)).or_default()
383 }
384 #[cfg(not(feature = "std"))]
385 {
386 if hfttl.get(key).is_none() {
387 hfttl.insert(SmallBytes::from_slice(key), kevy_map::KevyMap::default());
388 }
389 hfttl.get_mut(key).expect("inserted above")
390 }
391}
392
393#[cfg(test)]
394mod tests {
395 use super::*;
396
397 fn h(s: &mut Store) {
398 s.hset(b"h", &[(b"a".as_slice(), b"1".as_slice()), (b"b".as_slice(), b"2".as_slice())])
399 .unwrap();
400 }
401
402 #[test]
403 fn hexpire_httl_hpersist_codes() {
404 let mut s = Store::new();
405 h(&mut s);
406 let far = now_unix_ms() + 100_000;
407 // set on a + missing field
408 let codes = s.hexpire_at(b"h", &[b"a", b"nope"], far, HExpireCond::Always).unwrap();
409 assert_eq!(codes, vec![1, -2]);
410 let ttls = s.hpttl(b"h", &[b"a", b"b", b"nope"]).unwrap();
411 assert!(ttls[0] > 90_000 && ttls[0] <= 100_000);
412 assert_eq!(&ttls[1..], &[-1, -2]);
413 // NX refuses existing, XX refuses missing
414 assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 1, HExpireCond::Nx).unwrap(), vec![0]);
415 assert_eq!(s.hexpire_at(b"h", &[b"b"], far, HExpireCond::Xx).unwrap(), vec![0]);
416 // GT/LT
417 assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 500, HExpireCond::Gt).unwrap(), vec![1]);
418 assert_eq!(s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Gt).unwrap(), vec![0]);
419 // persist
420 assert_eq!(s.hpersist(b"h", &[b"a", b"b", b"nope"]).unwrap(), vec![1, -1, -2]);
421 assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
422 }
423
424 #[test]
425 fn past_deadline_deletes_and_lazy_purge_enforces() {
426 let mut s = Store::new();
427 h(&mut s);
428 // past deadline → immediate delete, code 2
429 assert_eq!(s.hexpire_at(b"h", &[b"a"], 1, HExpireCond::Always).unwrap(), vec![2]);
430 assert!(!s.hexists(b"h", b"a").unwrap());
431 // near-future deadline → lazily gone after it passes
432 let soon = now_unix_ms() + 30;
433 s.hexpire_at(b"h", &[b"b"], soon, HExpireCond::Always).unwrap();
434 std::thread::sleep(core::time::Duration::from_millis(50));
435 assert!(!s.hexists(b"h", b"b").unwrap(), "lazy purge on access");
436 // hash is now empty → hlen 0, sidecar pruned
437 assert_eq!(s.hlen(b"h").unwrap(), 0);
438 assert!(s.hfttl.is_empty());
439 }
440
441 #[test]
442 fn overwrite_clears_ttl_and_reaper_reports() {
443 let mut s = Store::new();
444 h(&mut s);
445 let soon = now_unix_ms() + 20;
446 s.hexpire_at(b"h", &[b"a", b"b"], soon, HExpireCond::Always).unwrap();
447 // overwrite a → its TTL is discarded (Redis 7.4)
448 s.hset(b"h", &[(b"a".as_slice(), b"new".as_slice())]).unwrap();
449 assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
450 std::thread::sleep(core::time::Duration::from_millis(40));
451 // reaper sweeps b, reports the removal for effect logging
452 let swept = s.tick_hash_ttl(100);
453 assert_eq!(swept, vec![(b"h".to_vec(), vec![b"b".to_vec()])]);
454 assert!(s.hexists(b"h", b"a").unwrap(), "overwritten field survived");
455 // whole-key delete drops the sidecar
456 let far = now_unix_ms() + 100_000;
457 s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Always).unwrap();
458 s.del(&[b"h".as_slice()]);
459 assert!(s.hfttl.is_empty());
460 }
461}