1#[cfg(not(feature = "std"))]
25use crate::nostd_prelude::*;
26use crate::{SmallBytes, Store, StoreError, Value, now_unix_ms};
27
28pub type HExpireCode = i8;
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
35pub enum HExpireCond {
36 #[default]
38 Always,
39 Nx,
41 Xx,
43 Gt,
46 Lt,
48}
49
50impl Store {
51 fn hash_has_field(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError> {
53 match self.tier_serve(key, crate::value::COLD_TAG_HASH)? {
54 None => Ok(false),
55 Some(e) => match &e.value {
56 Value::Hash(h) => Ok(h.get(field).is_some()),
57 Value::SegHash(h) => Ok(h.get(field).is_some()),
58 Value::SmallHashInline(h) => Ok(h.get(field).is_some()),
59 Value::PackedRow(r) => Ok(r.has_named(field)),
60 _ => Err(StoreError::WrongType),
61 },
62 }
63 }
64
65 pub fn hexpire_at(
69 &mut self,
70 key: &[u8],
71 fields: &[&[u8]],
72 deadline_ms: u64,
73 cond: HExpireCond,
74 ) -> Result<Vec<HExpireCode>, StoreError> {
75 self.purge_hash_ttl(key);
76 let mut codes = Vec::with_capacity(fields.len());
77 let now = now_unix_ms();
78 for f in fields {
79 if !self.hash_has_field(key, f)? {
80 codes.push(-2);
81 continue;
82 }
83 let current = self.hfttl.get(key).and_then(|m| m.get(*f)).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 self.hdel(key, &[f])?;
100 codes.push(2);
101 continue;
102 }
103 hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(f), deadline_ms);
104 codes.push(1);
105 }
106 self.prune_hfttl_key(key);
107 Ok(codes)
108 }
109
110 pub fn hpttl(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<Vec<i64>, StoreError> {
113 self.purge_hash_ttl(key);
114 let now = now_unix_ms();
115 let mut out = Vec::with_capacity(fields.len());
116 for f in fields {
117 if !self.hash_has_field(key, f)? {
118 out.push(-2);
119 continue;
120 }
121 match self.hfttl.get(key).and_then(|m| m.get(*f)) {
122 Some(&d) => out.push(d.saturating_sub(now) as i64),
123 None => out.push(-1),
124 }
125 }
126 Ok(out)
127 }
128
129 pub fn hpersist(
131 &mut self,
132 key: &[u8],
133 fields: &[&[u8]],
134 ) -> 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.hfttl.get_mut(key).and_then(|m| m.remove(*f)).is_some();
143 out.push(if had { 1 } else { -1 });
144 }
145 self.prune_hfttl_key(key);
146 Ok(out)
147 }
148
149 pub(crate) fn purge_hash_ttl(&mut self, key: &[u8]) {
153 if self.hfttl.is_empty() {
154 return;
155 }
156 let now = now_unix_ms();
157 let due: Vec<Vec<u8>> = match self.hfttl.get(key) {
158 None => return,
159 Some(m) => m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect(),
160 };
161 if due.is_empty() {
162 return;
163 }
164 if let Some(m) = self.hfttl.get_mut(key) {
167 for f in &due {
168 m.remove(f.as_slice());
169 }
170 }
171 self.prune_hfttl_key(key);
172 let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
173 let _ = self.hdel(key, &due_refs);
174 }
175
176 pub(crate) fn clear_hash_field_ttls(&mut self, key: &[u8], fields: &[&[u8]]) {
178 if self.hfttl.is_empty() {
179 return;
180 }
181 if let Some(m) = self.hfttl.get_mut(key) {
182 for f in fields {
183 m.remove(*f);
184 }
185 }
186 self.prune_hfttl_key(key);
187 }
188
189 pub(crate) fn clear_hash_key_ttls(&mut self, key: &[u8]) {
191 if self.hfttl.is_empty() {
192 return;
193 }
194 self.hfttl.remove(key);
195 }
196
197 fn prune_hfttl_key(&mut self, key: &[u8]) {
198 if self.hfttl.get(key).is_some_and(kevy_map_is_empty) {
199 self.hfttl.remove(key);
200 }
201 }
202
203 pub fn tick_hash_ttl(&mut self, max_keys: usize) -> Vec<(Vec<u8>, Vec<Vec<u8>>)> {
206 if self.hfttl.is_empty() {
207 return Vec::new();
208 }
209 let now = now_unix_ms();
210 let candidates: Vec<Vec<u8>> = self
211 .hfttl
212 .iter()
213 .filter(|(_, m)| m.iter().any(|(_, d)| *d <= now))
214 .take(max_keys)
215 .map(|(k, _)| k.to_vec())
216 .collect();
217 let mut out = Vec::with_capacity(candidates.len());
218 for k in candidates {
219 let due: Vec<Vec<u8>> = self
220 .hfttl
221 .get(k.as_slice())
222 .map(|m| m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect())
223 .unwrap_or_default();
224 if due.is_empty() {
225 continue;
226 }
227 if let Some(m) = self.hfttl.get_mut(k.as_slice()) {
228 for f in &due {
229 m.remove(f.as_slice());
230 }
231 }
232 self.prune_hfttl_key(&k);
233 let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
234 let _ = self.hdel(&k, &due_refs);
235 out.push((k, due));
236 }
237 out
238 }
239
240 pub fn load_hash_field_ttl(&mut self, key: &[u8], field: &[u8], deadline_ms: u64) {
243 hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(field), deadline_ms);
244 }
245
246 pub fn hash_ttl_each<F: FnMut(&[u8], &[u8], u64)>(&self, mut f: F) {
248 for (k, m) in self.hfttl.iter() {
249 for (field, &d) in m.iter() {
250 f(k.as_slice(), field.as_slice(), d);
251 }
252 }
253 }
254}
255
256fn kevy_map_is_empty(m: &crate::KevyMap<SmallBytes, u64>) -> bool {
257 m.iter().next().is_none()
258}
259
260fn hfttl_slot<'a>(
264 hfttl: &'a mut crate::SideMap<SmallBytes, kevy_map::KevyMap<SmallBytes, u64>>,
265 key: &[u8],
266) -> &'a mut kevy_map::KevyMap<SmallBytes, u64> {
267 #[cfg(feature = "std")]
268 {
269 hfttl.entry(SmallBytes::from_slice(key)).or_default()
270 }
271 #[cfg(not(feature = "std"))]
272 {
273 if hfttl.get(key).is_none() {
274 hfttl.insert(SmallBytes::from_slice(key), kevy_map::KevyMap::default());
275 }
276 hfttl.get_mut(key).expect("inserted above")
277 }
278}
279
280#[cfg(test)]
281mod tests {
282 use super::*;
283
284 fn h(s: &mut Store) {
285 s.hset(b"h", &[(b"a".as_slice(), b"1".as_slice()), (b"b".as_slice(), b"2".as_slice())])
286 .unwrap();
287 }
288
289 #[test]
290 fn hexpire_httl_hpersist_codes() {
291 let mut s = Store::new();
292 h(&mut s);
293 let far = now_unix_ms() + 100_000;
294 let codes = s.hexpire_at(b"h", &[b"a", b"nope"], far, HExpireCond::Always).unwrap();
296 assert_eq!(codes, vec![1, -2]);
297 let ttls = s.hpttl(b"h", &[b"a", b"b", b"nope"]).unwrap();
298 assert!(ttls[0] > 90_000 && ttls[0] <= 100_000);
299 assert_eq!(&ttls[1..], &[-1, -2]);
300 assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 1, HExpireCond::Nx).unwrap(), vec![0]);
302 assert_eq!(s.hexpire_at(b"h", &[b"b"], far, HExpireCond::Xx).unwrap(), vec![0]);
303 assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 500, HExpireCond::Gt).unwrap(), vec![1]);
305 assert_eq!(s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Gt).unwrap(), vec![0]);
306 assert_eq!(s.hpersist(b"h", &[b"a", b"b", b"nope"]).unwrap(), vec![1, -1, -2]);
308 assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
309 }
310
311 #[test]
312 fn past_deadline_deletes_and_lazy_purge_enforces() {
313 let mut s = Store::new();
314 h(&mut s);
315 assert_eq!(s.hexpire_at(b"h", &[b"a"], 1, HExpireCond::Always).unwrap(), vec![2]);
317 assert!(!s.hexists(b"h", b"a").unwrap());
318 let soon = now_unix_ms() + 30;
320 s.hexpire_at(b"h", &[b"b"], soon, HExpireCond::Always).unwrap();
321 std::thread::sleep(core::time::Duration::from_millis(50));
322 assert!(!s.hexists(b"h", b"b").unwrap(), "lazy purge on access");
323 assert_eq!(s.hlen(b"h").unwrap(), 0);
325 assert!(s.hfttl.is_empty());
326 }
327
328 #[test]
329 fn overwrite_clears_ttl_and_reaper_reports() {
330 let mut s = Store::new();
331 h(&mut s);
332 let soon = now_unix_ms() + 20;
333 s.hexpire_at(b"h", &[b"a", b"b"], soon, HExpireCond::Always).unwrap();
334 s.hset(b"h", &[(b"a".as_slice(), b"new".as_slice())]).unwrap();
336 assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
337 std::thread::sleep(core::time::Duration::from_millis(40));
338 let swept = s.tick_hash_ttl(100);
340 assert_eq!(swept, vec![(b"h".to_vec(), vec![b"b".to_vec()])]);
341 assert!(s.hexists(b"h", b"a").unwrap(), "overwritten field survived");
342 let far = now_unix_ms() + 100_000;
344 s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Always).unwrap();
345 s.del(&[b"h".as_slice()]);
346 assert!(s.hfttl.is_empty());
347 }
348}