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

kevy_store/
hash_ttl.rs
