kevy_store/
hash_ttl.rs

1//! v2.4 — 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
24use crate::{SmallBytes, Store, StoreError, Value, now_unix_ms};
25
26/// Per-field reply codes for `HEXPIRE`-family calls (Redis 7.4):
27/// `-2` key or field missing, `0` condition (NX/XX/GT/LT) not met,
28/// `1` deadline set, `2` field deleted (deadline already due).
29pub type HExpireCode = i8;
30
31/// Condition flags for `HEXPIRE` (`NX`/`XX`/`GT`/`LT`; at most one).
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
33pub enum HExpireCond {
34    /// Unconditional.
35    #[default]
36    Always,
37    /// Only when the field has no TTL.
38    Nx,
39    /// Only when the field already has a TTL.
40    Xx,
41    /// Only when the new deadline is later than the current one
42    /// (no TTL counts as infinitely late — GT never replaces it).
43    Gt,
44    /// Only when the new deadline is earlier (no TTL = always).
45    Lt,
46}
47
48impl Store {
49    /// Does this hash field exist (ignoring TTL state)?
50    fn hash_has_field(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError> {
51        match self.live_entry(key) {
52            None => Ok(false),
53            Some(e) => match &e.value {
54                Value::Hash(h) => Ok(h.get(field).is_some()),
55                Value::SmallHashInline(h) => Ok(h.get(field).is_some()),
56                _ => Err(StoreError::WrongType),
57            },
58        }
59    }
60
61    /// Set per-field deadlines (absolute unix-ms). One code per field,
62    /// request order. Due-or-past deadlines delete the field
63    /// immediately (code `2`, Redis semantics).
64    pub fn hexpire_at(
65        &mut self,
66        key: &[u8],
67        fields: &[&[u8]],
68        deadline_ms: u64,
69        cond: HExpireCond,
70    ) -> Result<Vec<HExpireCode>, StoreError> {
71        self.purge_hash_ttl(key);
72        let mut codes = Vec::with_capacity(fields.len());
73        let now = now_unix_ms();
74        for f in fields {
75            if !self.hash_has_field(key, f)? {
76                codes.push(-2);
77                continue;
78            }
79            let current = self
80                .hfttl
81                .get(key)
82                .and_then(|m| m.get(*f))
83                .copied();
84            let pass = match cond {
85                HExpireCond::Always => true,
86                HExpireCond::Nx => current.is_none(),
87                HExpireCond::Xx => current.is_some(),
88                HExpireCond::Gt => current.is_some_and(|c| deadline_ms > c),
89                HExpireCond::Lt => current.is_none_or(|c| deadline_ms < c),
90            };
91            if !pass {
92                codes.push(0);
93                continue;
94            }
95            if deadline_ms <= now {
96                if let Some(m) = self.hfttl.get_mut(key) {
97                    m.remove(*f);
98                }
99                let owned = [f.to_vec()];
100                self.hdel(key, &owned)?;
101                codes.push(2);
102                continue;
103            }
104            self.hfttl
105                .entry(SmallBytes::from_slice(key))
106                .or_default()
107                .insert(SmallBytes::from_slice(f), deadline_ms);
108            codes.push(1);
109        }
110        self.prune_hfttl_key(key);
111        Ok(codes)
112    }
113
114    /// Remaining TTL per field: `-2` key/field missing, `-1` no TTL,
115    /// else remaining ms.
116    pub fn httl(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<Vec<i64>, StoreError> {
117        self.purge_hash_ttl(key);
118        let now = now_unix_ms();
119        let mut out = Vec::with_capacity(fields.len());
120        for f in fields {
121            if !self.hash_has_field(key, f)? {
122                out.push(-2);
123                continue;
124            }
125            match self.hfttl.get(key).and_then(|m| m.get(*f)) {
126                Some(&d) => out.push(d.saturating_sub(now) as i64),
127                None => out.push(-1),
128            }
129        }
130        Ok(out)
131    }
132
133    /// Clear per-field TTLs: `-2` missing, `-1` had no TTL, `1` cleared.
134    pub fn hpersist(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<Vec<HExpireCode>, StoreError> {
135        self.purge_hash_ttl(key);
136        let mut out = Vec::with_capacity(fields.len());
137        for f in fields {
138            if !self.hash_has_field(key, f)? {
139                out.push(-2);
140                continue;
141            }
142            let had = self
143                .hfttl
144                .get_mut(key)
145                .and_then(|m| m.remove(*f))
146                .is_some();
147            out.push(if had { 1 } else { -1 });
148        }
149        self.prune_hfttl_key(key);
150        Ok(out)
151    }
152
153    /// Lazy enforcement hook — call at the top of every hash op.
154    /// Removes expired fields from the hash + sidecar. One `is_empty`
155    /// branch when the feature is unused.
156    pub(crate) fn purge_hash_ttl(&mut self, key: &[u8]) {
157        if self.hfttl.is_empty() {
158            return;
159        }
160        let now = now_unix_ms();
161        let due: Vec<Vec<u8>> = match self.hfttl.get(key) {
162            None => return,
163            Some(m) => m
164                .iter()
165                .filter(|(_, d)| **d <= now)
166                .map(|(f, _)| f.to_vec())
167                .collect(),
168        };
169        if due.is_empty() {
170            return;
171        }
172        // Sidecar first: `hdel` re-enters this fn, and a clean sidecar
173        // makes the re-entry a no-op (no recursion).
174        if let Some(m) = self.hfttl.get_mut(key) {
175            for f in &due {
176                m.remove(f.as_slice());
177            }
178        }
179        self.prune_hfttl_key(key);
180        let _ = self.hdel(key, &due);
181    }
182
183    /// Overwrite hook — `HSET`/`HINCRBY*` on a field discards its TTL.
184    pub(crate) fn clear_hash_field_ttls(&mut self, key: &[u8], fields: &[&[u8]]) {
185        if self.hfttl.is_empty() {
186            return;
187        }
188        if let Some(m) = self.hfttl.get_mut(key) {
189            for f in fields {
190                m.remove(*f);
191            }
192        }
193        self.prune_hfttl_key(key);
194    }
195
196    /// Whole-key hook — key removed/overwritten wholesale.
197    pub(crate) fn clear_hash_key_ttls(&mut self, key: &[u8]) {
198        if self.hfttl.is_empty() {
199            return;
200        }
201        self.hfttl.remove(key);
202    }
203
204    fn prune_hfttl_key(&mut self, key: &[u8]) {
205        if self.hfttl.get(key).is_some_and(kevy_map_is_empty) {
206            self.hfttl.remove(key);
207        }
208    }
209
210    /// Reaper sweep: remove every due field store-wide; returns
211    /// `(key, removed fields)` pairs so the caller logs `HDEL` effects.
212    pub fn tick_hash_ttl(&mut self, max_keys: usize) -> Vec<(Vec<u8>, Vec<Vec<u8>>)> {
213        if self.hfttl.is_empty() {
214            return Vec::new();
215        }
216        let now = now_unix_ms();
217        let candidates: Vec<Vec<u8>> = self
218            .hfttl
219            .iter()
220            .filter(|(_, m)| m.iter().any(|(_, d)| *d <= now))
221            .take(max_keys)
222            .map(|(k, _)| k.to_vec())
223            .collect();
224        let mut out = Vec::with_capacity(candidates.len());
225        for k in candidates {
226            let due: Vec<Vec<u8>> = self
227                .hfttl
228                .get(k.as_slice())
229                .map(|m| {
230                    m.iter()
231                        .filter(|(_, d)| **d <= now)
232                        .map(|(f, _)| f.to_vec())
233                        .collect()
234                })
235                .unwrap_or_default();
236            if due.is_empty() {
237                continue;
238            }
239            if let Some(m) = self.hfttl.get_mut(k.as_slice()) {
240                for f in &due {
241                    m.remove(f.as_slice());
242                }
243            }
244            self.prune_hfttl_key(&k);
245            let _ = self.hdel(&k, &due);
246            out.push((k, due));
247        }
248        out
249    }
250
251    /// Snapshot loader hook: restore one field TTL (deadlines already
252    /// absolute unix-ms; past deadlines simply purge on first access).
253    pub fn load_hash_field_ttl(&mut self, key: &[u8], field: &[u8], deadline_ms: u64) {
254        self.hfttl
255            .entry(SmallBytes::from_slice(key))
256            .or_default()
257            .insert(SmallBytes::from_slice(field), deadline_ms);
258    }
259
260    /// Snapshot support: visit every live (key, field, deadline_ms).
261    pub fn hash_ttl_each<F: FnMut(&[u8], &[u8], u64)>(&self, mut f: F) {
262        for (k, m) in self.hfttl.iter() {
263            for (field, &d) in m.iter() {
264                f(k.as_slice(), field.as_slice(), d);
265            }
266        }
267    }
268}
269
270fn kevy_map_is_empty(m: &crate::KevyMap<SmallBytes, u64>) -> bool {
271    m.iter().next().is_none()
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    fn h(s: &mut Store) {
279        s.hset(
280            b"h",
281            &[(b"a".to_vec(), b"1".to_vec()), (b"b".to_vec(), b"2".to_vec())],
282        )
283        .unwrap();
284    }
285
286    #[test]
287    fn hexpire_httl_hpersist_codes() {
288        let mut s = Store::new();
289        h(&mut s);
290        let far = now_unix_ms() + 100_000;
291        // set on a + missing field
292        let codes = s
293            .hexpire_at(b"h", &[b"a", b"nope"], far, HExpireCond::Always)
294            .unwrap();
295        assert_eq!(codes, vec![1, -2]);
296        let ttls = s.httl(b"h", &[b"a", b"b", b"nope"]).unwrap();
297        assert!(ttls[0] > 90_000 && ttls[0] <= 100_000);
298        assert_eq!(&ttls[1..], &[-1, -2]);
299        // NX refuses existing, XX refuses missing
300        assert_eq!(
301            s.hexpire_at(b"h", &[b"a"], far + 1, HExpireCond::Nx).unwrap(),
302            vec![0]
303        );
304        assert_eq!(
305            s.hexpire_at(b"h", &[b"b"], far, HExpireCond::Xx).unwrap(),
306            vec![0]
307        );
308        // GT/LT
309        assert_eq!(
310            s.hexpire_at(b"h", &[b"a"], far + 500, HExpireCond::Gt).unwrap(),
311            vec![1]
312        );
313        assert_eq!(
314            s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Gt).unwrap(),
315            vec![0]
316        );
317        // persist
318        assert_eq!(s.hpersist(b"h", &[b"a", b"b", b"nope"]).unwrap(), vec![1, -1, -2]);
319        assert_eq!(s.httl(b"h", &[b"a"]).unwrap(), vec![-1]);
320    }
321
322    #[test]
323    fn past_deadline_deletes_and_lazy_purge_enforces() {
324        let mut s = Store::new();
325        h(&mut s);
326        // past deadline → immediate delete, code 2
327        assert_eq!(
328            s.hexpire_at(b"h", &[b"a"], 1, HExpireCond::Always).unwrap(),
329            vec![2]
330        );
331        assert!(!s.hexists(b"h", b"a").unwrap());
332        // near-future deadline → lazily gone after it passes
333        let soon = now_unix_ms() + 30;
334        s.hexpire_at(b"h", &[b"b"], soon, HExpireCond::Always).unwrap();
335        std::thread::sleep(std::time::Duration::from_millis(50));
336        assert!(!s.hexists(b"h", b"b").unwrap(), "lazy purge on access");
337        // hash is now empty → hlen 0, sidecar pruned
338        assert_eq!(s.hlen(b"h").unwrap(), 0);
339        assert!(s.hfttl.is_empty());
340    }
341
342    #[test]
343    fn overwrite_clears_ttl_and_reaper_reports() {
344        let mut s = Store::new();
345        h(&mut s);
346        let soon = now_unix_ms() + 20;
347        s.hexpire_at(b"h", &[b"a", b"b"], soon, HExpireCond::Always).unwrap();
348        // overwrite a → its TTL is discarded (Redis 7.4)
349        s.hset(b"h", &[(b"a".to_vec(), b"new".to_vec())]).unwrap();
350        assert_eq!(s.httl(b"h", &[b"a"]).unwrap(), vec![-1]);
351        std::thread::sleep(std::time::Duration::from_millis(40));
352        // reaper sweeps b, reports the removal for effect logging
353        let swept = s.tick_hash_ttl(100);
354        assert_eq!(swept, vec![(b"h".to_vec(), vec![b"b".to_vec()])]);
355        assert!(s.hexists(b"h", b"a").unwrap(), "overwritten field survived");
356        // whole-key delete drops the sidecar
357        let far = now_unix_ms() + 100_000;
358        s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Always).unwrap();
359        s.del(&[b"h".to_vec()]);
360        assert!(s.hfttl.is_empty());
361    }
362}
kevy_store/hash_ttl.rs

kevy_store/
hash_ttl.rs
