#[cfg(not(feature = "std"))]
use crate::nostd_prelude::*;
use crate::{SmallBytes, Store, StoreError, Value, now_unix_ms};
pub type HExpireCode = i8;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum HExpireCond {
#[default]
Always,
Nx,
Xx,
Gt,
Lt,
}
impl Store {
fn hash_has_field(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError> {
match self.tier_serve(key, crate::value::COLD_TAG_HASH)? {
None => Ok(false),
Some(e) => match &e.value {
Value::Hash(h) => Ok(h.get(field).is_some()),
Value::SegHash(h) => Ok(h.get(field).is_some()),
Value::SmallHashInline(h) => Ok(h.get(field).is_some()),
Value::PackedRow(r) => Ok(r.has_named(field)),
_ => Err(StoreError::WrongType),
},
}
}
pub fn hexpire_at(
&mut self,
key: &[u8],
fields: &[&[u8]],
deadline_ms: u64,
cond: HExpireCond,
) -> Result<Vec<HExpireCode>, StoreError> {
self.purge_hash_ttl(key);
let mut codes = Vec::with_capacity(fields.len());
let now = now_unix_ms();
for f in fields {
if !self.hash_has_field(key, f)? {
codes.push(-2);
continue;
}
let current = self.hfttl.get(key).and_then(|m| m.get(*f)).copied();
let pass = match cond {
HExpireCond::Always => true,
HExpireCond::Nx => current.is_none(),
HExpireCond::Xx => current.is_some(),
HExpireCond::Gt => current.is_some_and(|c| deadline_ms > c),
HExpireCond::Lt => current.is_none_or(|c| deadline_ms < c),
};
if !pass {
codes.push(0);
continue;
}
if deadline_ms <= now {
if let Some(m) = self.hfttl.get_mut(key) {
m.remove(*f);
}
self.hdel(key, &[f])?;
codes.push(2);
continue;
}
hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(f), deadline_ms);
codes.push(1);
}
self.prune_hfttl_key(key);
Ok(codes)
}
pub fn hpttl(&mut self, key: &[u8], fields: &[&[u8]]) -> Result<Vec<i64>, StoreError> {
self.purge_hash_ttl(key);
let now = now_unix_ms();
let mut out = Vec::with_capacity(fields.len());
for f in fields {
if !self.hash_has_field(key, f)? {
out.push(-2);
continue;
}
match self.hfttl.get(key).and_then(|m| m.get(*f)) {
Some(&d) => out.push(d.saturating_sub(now) as i64),
None => out.push(-1),
}
}
Ok(out)
}
pub fn hpersist(
&mut self,
key: &[u8],
fields: &[&[u8]],
) -> Result<Vec<HExpireCode>, StoreError> {
self.purge_hash_ttl(key);
let mut out = Vec::with_capacity(fields.len());
for f in fields {
if !self.hash_has_field(key, f)? {
out.push(-2);
continue;
}
let had = self.hfttl.get_mut(key).and_then(|m| m.remove(*f)).is_some();
out.push(if had { 1 } else { -1 });
}
self.prune_hfttl_key(key);
Ok(out)
}
pub(crate) fn purge_hash_ttl(&mut self, key: &[u8]) {
if self.hfttl.is_empty() {
return;
}
let now = now_unix_ms();
let due: Vec<Vec<u8>> = match self.hfttl.get(key) {
None => return,
Some(m) => m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect(),
};
if due.is_empty() {
return;
}
if let Some(m) = self.hfttl.get_mut(key) {
for f in &due {
m.remove(f.as_slice());
}
}
self.prune_hfttl_key(key);
let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
let _ = self.hdel(key, &due_refs);
}
pub(crate) fn clear_hash_field_ttls(&mut self, key: &[u8], fields: &[&[u8]]) {
if self.hfttl.is_empty() {
return;
}
if let Some(m) = self.hfttl.get_mut(key) {
for f in fields {
m.remove(*f);
}
}
self.prune_hfttl_key(key);
}
pub(crate) fn clear_hash_key_ttls(&mut self, key: &[u8]) {
if self.hfttl.is_empty() {
return;
}
self.hfttl.remove(key);
}
fn prune_hfttl_key(&mut self, key: &[u8]) {
if self.hfttl.get(key).is_some_and(kevy_map_is_empty) {
self.hfttl.remove(key);
}
}
pub fn tick_hash_ttl(&mut self, max_keys: usize) -> Vec<(Vec<u8>, Vec<Vec<u8>>)> {
if self.hfttl.is_empty() {
return Vec::new();
}
let now = now_unix_ms();
let candidates: Vec<Vec<u8>> = self
.hfttl
.iter()
.filter(|(_, m)| m.iter().any(|(_, d)| *d <= now))
.take(max_keys)
.map(|(k, _)| k.to_vec())
.collect();
let mut out = Vec::with_capacity(candidates.len());
for k in candidates {
let due: Vec<Vec<u8>> = self
.hfttl
.get(k.as_slice())
.map(|m| m.iter().filter(|(_, d)| **d <= now).map(|(f, _)| f.to_vec()).collect())
.unwrap_or_default();
if due.is_empty() {
continue;
}
if let Some(m) = self.hfttl.get_mut(k.as_slice()) {
for f in &due {
m.remove(f.as_slice());
}
}
self.prune_hfttl_key(&k);
let due_refs: Vec<&[u8]> = due.iter().map(Vec::as_slice).collect();
let _ = self.hdel(&k, &due_refs);
out.push((k, due));
}
out
}
pub fn load_hash_field_ttl(&mut self, key: &[u8], field: &[u8], deadline_ms: u64) {
hfttl_slot(&mut self.hfttl, key).insert(SmallBytes::from_slice(field), deadline_ms);
}
pub fn hash_ttl_each<F: FnMut(&[u8], &[u8], u64)>(&self, mut f: F) {
for (k, m) in self.hfttl.iter() {
for (field, &d) in m.iter() {
f(k.as_slice(), field.as_slice(), d);
}
}
}
}
fn kevy_map_is_empty(m: &crate::KevyMap<SmallBytes, u64>) -> bool {
m.iter().next().is_none()
}
fn hfttl_slot<'a>(
hfttl: &'a mut crate::SideMap<SmallBytes, kevy_map::KevyMap<SmallBytes, u64>>,
key: &[u8],
) -> &'a mut kevy_map::KevyMap<SmallBytes, u64> {
#[cfg(feature = "std")]
{
hfttl.entry(SmallBytes::from_slice(key)).or_default()
}
#[cfg(not(feature = "std"))]
{
if hfttl.get(key).is_none() {
hfttl.insert(SmallBytes::from_slice(key), kevy_map::KevyMap::default());
}
hfttl.get_mut(key).expect("inserted above")
}
}
#[cfg(test)]
mod tests {
use super::*;
fn h(s: &mut Store) {
s.hset(b"h", &[(b"a".as_slice(), b"1".as_slice()), (b"b".as_slice(), b"2".as_slice())])
.unwrap();
}
#[test]
fn hexpire_httl_hpersist_codes() {
let mut s = Store::new();
h(&mut s);
let far = now_unix_ms() + 100_000;
let codes = s.hexpire_at(b"h", &[b"a", b"nope"], far, HExpireCond::Always).unwrap();
assert_eq!(codes, vec![1, -2]);
let ttls = s.hpttl(b"h", &[b"a", b"b", b"nope"]).unwrap();
assert!(ttls[0] > 90_000 && ttls[0] <= 100_000);
assert_eq!(&ttls[1..], &[-1, -2]);
assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 1, HExpireCond::Nx).unwrap(), vec![0]);
assert_eq!(s.hexpire_at(b"h", &[b"b"], far, HExpireCond::Xx).unwrap(), vec![0]);
assert_eq!(s.hexpire_at(b"h", &[b"a"], far + 500, HExpireCond::Gt).unwrap(), vec![1]);
assert_eq!(s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Gt).unwrap(), vec![0]);
assert_eq!(s.hpersist(b"h", &[b"a", b"b", b"nope"]).unwrap(), vec![1, -1, -2]);
assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
}
#[test]
fn past_deadline_deletes_and_lazy_purge_enforces() {
let mut s = Store::new();
h(&mut s);
assert_eq!(s.hexpire_at(b"h", &[b"a"], 1, HExpireCond::Always).unwrap(), vec![2]);
assert!(!s.hexists(b"h", b"a").unwrap());
let soon = now_unix_ms() + 30;
s.hexpire_at(b"h", &[b"b"], soon, HExpireCond::Always).unwrap();
std::thread::sleep(core::time::Duration::from_millis(50));
assert!(!s.hexists(b"h", b"b").unwrap(), "lazy purge on access");
assert_eq!(s.hlen(b"h").unwrap(), 0);
assert!(s.hfttl.is_empty());
}
#[test]
fn overwrite_clears_ttl_and_reaper_reports() {
let mut s = Store::new();
h(&mut s);
let soon = now_unix_ms() + 20;
s.hexpire_at(b"h", &[b"a", b"b"], soon, HExpireCond::Always).unwrap();
s.hset(b"h", &[(b"a".as_slice(), b"new".as_slice())]).unwrap();
assert_eq!(s.hpttl(b"h", &[b"a"]).unwrap(), vec![-1]);
std::thread::sleep(core::time::Duration::from_millis(40));
let swept = s.tick_hash_ttl(100);
assert_eq!(swept, vec![(b"h".to_vec(), vec![b"b".to_vec()])]);
assert!(s.hexists(b"h", b"a").unwrap(), "overwritten field survived");
let far = now_unix_ms() + 100_000;
s.hexpire_at(b"h", &[b"a"], far, HExpireCond::Always).unwrap();
s.del(&[b"h".as_slice()]);
assert!(s.hfttl.is_empty());
}
}