use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use super::expiry_index::{DeadlineScheduler, Generation, WallClock};
use super::*;
use crate::store::MemoryStore;
use crate::sync::ballot::Stamp;
use crate::wal::{FsyncPolicy, WalRecovery};
#[derive(Default)]
struct FakeClock {
now: AtomicU64,
}
impl FakeClock {
fn at(now: u64) -> Self {
Self {
now: AtomicU64::new(now),
}
}
fn set(&self, now: u64) {
self.now.store(now, Ordering::SeqCst);
}
}
impl fmt::Debug for FakeClock {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_tuple("FakeClock")
.field(&self.now.load(Ordering::SeqCst))
.finish()
}
}
impl WallClock for FakeClock {
fn now(&self) -> u64 {
self.now.load(Ordering::SeqCst)
}
}
#[derive(Debug, Default)]
struct DeadlineLatch {
arms: Vec<(Duration, Generation)>,
}
impl DeadlineScheduler for DeadlineLatch {
fn schedule(
&mut self,
delay: Duration,
generation: Generation,
) -> Result<(), super::expiry_index::ArmError> {
self.arms.push((delay, generation));
Ok(())
}
}
fn actor_with_clock(
clock: Arc<FakeClock>,
) -> Result<(tempfile::TempDir, ShardActor, MemoryStore), Box<dyn std::error::Error>> {
let dir = tempfile::tempdir()?;
let wal = DurableWal::new(dir.path().join("expiry.wal"), FsyncPolicy::CommitOnly)?;
let actor = ShardActor::new_with_clock(wal, clock);
Ok((dir, actor, MemoryStore::new()))
}
fn arm(
actor: &mut ShardActor,
latch: &mut DeadlineLatch,
) -> Result<Generation, Box<dyn std::error::Error>> {
actor.arm_expiry(latch)?;
actor
.current_expiry_generation()
.ok_or_else(|| "expected a current expiry generation".into())
}
fn fire(
actor: &mut ShardActor,
generation: Generation,
store: &MemoryStore,
) -> Result<bool, Box<dyn std::error::Error>> {
let Some(due) = actor.begin_expiry_deadline(generation) else {
return Ok(false);
};
for (_deadline, key) in &due {
actor.inspect_expiry_key();
actor.delete_if_expired(key, store)?;
}
actor.finish_expiry_deadline();
Ok(true)
}
#[test]
fn empty_expiry_index_arms_zero_timers_and_wakes_zero_times()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(10));
let (_dir, mut actor, _store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.arm_expiry(&mut latch)?;
clock.set(u64::MAX);
let metrics = actor.expiry_metrics();
assert!(latch.arms.is_empty());
assert_eq!(metrics.current_arms, 0);
assert_eq!(metrics.physical_arms, 0);
assert_eq!(metrics.deadline_deliveries, 0);
assert_eq!(metrics.actor_wakes, 0);
assert_eq!(metrics.inspected_keys, 0);
assert_eq!(metrics.deletes, 0);
Ok(())
}
#[test]
fn no_wake_before_minimum_deadline_and_one_arm_per_minimum_change()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(1_000));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"a", b"a", Some(Duration::from_nanos(100)), &store)?;
arm(&mut actor, &mut latch)?;
actor.put_with_ttl(b"b", b"b", Some(Duration::from_nanos(200)), &store)?;
actor.arm_expiry(&mut latch)?;
actor.put_with_ttl(b"c", b"c", Some(Duration::from_nanos(50)), &store)?;
arm(&mut actor, &mut latch)?;
actor.put_with_ttl(b"d", b"d", Some(Duration::from_nanos(50)), &store)?;
actor.arm_expiry(&mut latch)?;
actor.delete(b"b", Stamp::bottom(), &store)?;
actor.arm_expiry(&mut latch)?;
actor.delete(b"c", Stamp::bottom(), &store)?;
actor.arm_expiry(&mut latch)?;
actor.delete(b"d", Stamp::bottom(), &store)?;
arm(&mut actor, &mut latch)?;
clock.set(1_099);
let metrics = actor.expiry_metrics();
assert_eq!(latch.arms.len(), 3);
assert_eq!(metrics.physical_arms, 3);
assert_eq!(metrics.deadline_deliveries, 0);
assert_eq!(metrics.actor_wakes, 0);
assert_eq!(metrics.inspected_keys, 0);
assert_eq!(metrics.deletes, 0);
Ok(())
}
#[test]
fn stale_generation_firing_moves_zero_work_counters() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(10_000));
let (_dir, mut actor, store) = actor_with_clock(clock)?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"v1", Some(Duration::from_nanos(100)), &store)?;
let stale = arm(&mut actor, &mut latch)?;
actor.put_with_ttl(b"k", b"v2", Some(Duration::from_nanos(200)), &store)?;
arm(&mut actor, &mut latch)?;
let before = actor.expiry_metrics();
assert!(!fire(&mut actor, stale, &store)?);
let after = actor.expiry_metrics();
assert_eq!(after.stale_drops, before.stale_drops + 1);
assert_eq!(after.index_mutations, before.index_mutations);
assert_eq!(after.inspected_keys, before.inspected_keys);
assert_eq!(after.delete_attempts, before.delete_attempts);
assert_eq!(after.deletes, before.deletes);
assert_eq!(after.physical_arms, before.physical_arms);
assert_eq!(after.current_arms, before.current_arms);
Ok(())
}
#[test]
fn deadline_firing_physically_deletes_expired_entry() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(100));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"victim", b"doomed", Some(Duration::from_nanos(5)), &store)?;
let generation = arm(&mut actor, &mut latch)?;
assert!(actor.get_raw(b"victim", &store)?.is_some());
clock.set(105);
assert!(fire(&mut actor, generation, &store)?);
actor.arm_expiry(&mut latch)?;
assert!(actor.get_raw(b"victim", &store)?.is_none());
let metrics = actor.expiry_metrics();
assert_eq!(metrics.deadline_deliveries, 1);
assert_eq!(metrics.accepted_deliveries, 1);
assert_eq!(metrics.inspected_keys, 1);
assert_eq!(metrics.delete_attempts, 1);
assert_eq!(metrics.deletes, 1);
assert_eq!(latch.arms.len(), 1);
Ok(())
}
#[test]
fn backward_jump_rearms_unchanged_minimum_once_per_delivery()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(1_000));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(100)), &store)?;
let generation = arm(&mut actor, &mut latch)?;
clock.set(900);
assert!(fire(&mut actor, generation, &store)?);
arm(&mut actor, &mut latch)?;
assert_eq!(latch.arms.len(), 2);
assert_eq!(latch.arms[1].0, Duration::from_nanos(200));
assert_eq!(actor.expiry_metrics().inspected_keys, 0);
assert!(actor.get_raw(b"k", &store)?.is_some());
Ok(())
}
#[test]
fn plain_put_old_value_decode_cost_is_counter_measurable() -> Result<(), Box<dyn std::error::Error>>
{
let clock = Arc::new(FakeClock::at(1_000));
let (_dir, mut actor, store) = actor_with_clock(clock)?;
actor.put(b"raw", b"abc")?;
let before_raw = actor.expiry_metrics();
actor.put_with_ttl(b"raw", b"next", None, &store)?;
let after_raw = actor.expiry_metrics();
assert_eq!(
after_raw.old_value_decodes - before_raw.old_value_decodes,
1
);
assert_eq!(
after_raw.old_value_decode_bytes - before_raw.old_value_decode_bytes,
3
);
assert_eq!(after_raw.old_value_expiring, before_raw.old_value_expiring);
actor.put_with_ttl(b"ttl", b"old", Some(Duration::from_nanos(5)), &store)?;
let before_ttl = actor.expiry_metrics();
actor.put_with_ttl(b"ttl", b"new", None, &store)?;
let after_ttl = actor.expiry_metrics();
assert_eq!(
after_ttl.old_value_decodes - before_ttl.old_value_decodes,
1
);
assert!(after_ttl.old_value_decode_bytes > before_ttl.old_value_decode_bytes);
assert_eq!(
after_ttl.old_value_expiring - before_ttl.old_value_expiring,
1
);
Ok(())
}
#[test]
fn unswept_expired_remains_durable_after_crash() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(100));
let dir = tempfile::tempdir()?;
let wal_path = dir.path().join("unswept.wal");
let mut store = MemoryStore::new();
let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
let mut actor = ShardActor::new_with_clock(wal, clock);
actor.put_with_ttl(b"expired", b"bytes", Some(Duration::ZERO), &store)?;
actor.commit(&mut store)?;
drop(actor);
let recovered = WalRecovery::recover_path(&wal_path, &store)?;
let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
let recovered = ShardActor::from_recovered(wal, recovered, &store)?;
assert!(recovered.get_raw(b"expired", &store)?.is_some());
assert_eq!(recovered.expiry_metrics().rebuild_entries, 1);
Ok(())
}
#[test]
fn staged_expiry_delete_recovers_absent() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(100));
let dir = tempfile::tempdir()?;
let wal_path = dir.path().join("staged-delete.wal");
let mut store = MemoryStore::new();
let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
let mut actor = ShardActor::new_with_clock(wal, clock.clone());
actor.put_with_ttl(b"expired", b"bytes", Some(Duration::from_nanos(1)), &store)?;
actor.commit(&mut store)?;
clock.set(101);
assert!(actor.delete_if_expired(b"expired", &store)?);
drop(actor);
let recovered = WalRecovery::recover_path(&wal_path, &store)?;
let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
let recovered = ShardActor::from_recovered(wal, recovered, &store)?;
assert!(recovered.get_raw(b"expired", &store)?.is_none());
Ok(())
}
#[test]
fn generation_exhaustion_is_a_typed_arm_refusal() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(0));
let (_dir, mut actor, store) = actor_with_clock(clock)?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(1)), &store)?;
actor.force_expiry_generation_for_test(super::expiry_index::MAX_GENERATION);
let error = match actor.arm_expiry(&mut latch) {
Ok(()) => return Err("generation exhaustion unexpectedly armed".into()),
Err(error) => error,
};
assert_eq!(error, super::expiry_index::ArmError::GenerationExhausted);
assert!(latch.arms.is_empty());
assert_eq!(actor.expiry_metrics().current_arms, 0);
Ok(())
}
#[test]
fn relative_delay_preserves_near_u64_max_nanoseconds_exactly()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(0));
let (_dir, mut actor, store) = actor_with_clock(clock)?;
let mut latch = DeadlineLatch::default();
let nanos = u64::MAX - 1;
actor.put_with_ttl(b"far", b"v", Some(Duration::from_nanos(nanos)), &store)?;
arm(&mut actor, &mut latch)?;
assert_eq!(latch.arms[0].0, Duration::from_nanos(nanos));
Ok(())
}
#[test]
fn restarted_actor_rejects_pre_restart_generation() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(10));
let first_dir = tempfile::tempdir()?;
let first_wal = DurableWal::new(first_dir.path().join("first.wal"), FsyncPolicy::CommitOnly)?;
let mut first = ShardActor::new_with_global_clock(first_wal, clock.clone());
let first_store = MemoryStore::new();
let mut first_latch = DeadlineLatch::default();
first.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(5)), &first_store)?;
let old_generation = arm(&mut first, &mut first_latch)?;
let second_dir = tempfile::tempdir()?;
let second_wal = DurableWal::new(
second_dir.path().join("second.wal"),
FsyncPolicy::CommitOnly,
)?;
let mut restarted = ShardActor::new_with_global_clock(second_wal, clock);
let second_store = MemoryStore::new();
let mut second_latch = DeadlineLatch::default();
restarted.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(5)), &second_store)?;
let new_generation = arm(&mut restarted, &mut second_latch)?;
assert_ne!(old_generation, new_generation);
assert!(restarted.begin_expiry_deadline(old_generation).is_none());
assert_eq!(restarted.expiry_metrics().stale_drops, 1);
Ok(())
}
#[test]
fn cas_raw_overwrite_removes_current_minimum_deadline() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(0));
let (_dir, mut actor, mut store) = actor_with_clock(clock)?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(
b"counter",
1_u64.to_be_bytes(),
Some(Duration::from_nanos(1)),
&store,
)?;
let stale = arm(&mut actor, &mut latch)?;
actor.cas(b"counter", None, 2, &mut store)?;
actor.arm_expiry(&mut latch)?;
assert_eq!(actor.current_expiry_generation(), None);
assert_eq!(actor.expiry_metrics().current_arms, 0);
assert_eq!(latch.arms.len(), 1);
assert!(!fire(&mut actor, stale, &store)?);
Ok(())
}
#[test]
fn expiry_index_production_cost_shape_is_measured() {
let state_bytes = std::mem::size_of::<super::expiry_index::ExpiryState>();
let metrics_bytes = std::mem::size_of::<super::expiry_index::ExpiryMetrics>();
let actor_bytes = std::mem::size_of::<ShardActor>();
eprintln!(
"expiry_state_bytes={state_bytes}; expiry_metrics_bytes={metrics_bytes}; \
shard_actor_bytes={actor_bytes}"
);
assert!(state_bytes > metrics_bytes);
assert!(actor_bytes >= state_bytes);
}
#[test]
fn equal_deadline_bucket_drains_all_keys_in_key_order() -> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(500));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
for key in [b"a".as_slice(), b"b".as_slice()] {
actor.put_with_ttl(key, key, Some(Duration::from_nanos(10)), &store)?;
}
let generation = arm(&mut actor, &mut latch)?;
assert_eq!(latch.arms.len(), 1);
clock.set(510);
assert!(fire(&mut actor, generation, &store)?);
actor.arm_expiry(&mut latch)?;
assert!(actor.get_raw(b"a", &store)?.is_none());
assert!(actor.get_raw(b"b", &store)?.is_none());
assert_eq!(actor.expiry_metrics().inspected_keys, 2);
assert_eq!(actor.expiry_metrics().deletes, 2);
assert_eq!(latch.arms.len(), 1);
Ok(())
}
#[test]
fn refresh_after_detach_is_rechecked_restored_and_rearmed_once()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(1_000));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"old", Some(Duration::from_nanos(5)), &store)?;
let generation = arm(&mut actor, &mut latch)?;
clock.set(1_005);
let due = actor
.begin_expiry_deadline(generation)
.ok_or("current delivery was treated as stale")?;
assert_eq!(due.len(), 1);
actor.put_with_ttl(b"k", b"fresh", Some(Duration::from_nanos(100)), &store)?;
actor.inspect_expiry_key();
assert!(!actor.delete_if_expired(b"k", &store)?);
actor.finish_expiry_deadline();
arm(&mut actor, &mut latch)?;
assert!(actor.get_raw(b"k", &store)?.is_some());
assert_eq!(actor.expiry_metrics().deletes, 0);
assert_eq!(latch.arms.len(), 2);
Ok(())
}
#[test]
fn shutdown_invalidates_queued_delivery_without_work_or_rearm()
-> Result<(), Box<dyn std::error::Error>> {
let clock = Arc::new(FakeClock::at(10));
let (_dir, mut actor, store) = actor_with_clock(clock)?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(1)), &store)?;
let generation = arm(&mut actor, &mut latch)?;
actor.invalidate_expiry_on_shutdown();
let before = actor.expiry_metrics();
assert!(actor.begin_expiry_deadline(generation).is_none());
actor.arm_expiry(&mut latch)?;
let after = actor.expiry_metrics();
assert_eq!(after.stale_drops, before.stale_drops + 1);
assert_eq!(after.inspected_keys, before.inspected_keys);
assert_eq!(after.deletes, before.deletes);
assert_eq!(after.physical_arms, before.physical_arms);
assert_eq!(latch.arms.len(), 1);
Ok(())
}
#[test]
fn overdue_at_and_one_nanosecond_future_convert_without_cadence() {
use super::expiry_index::relative_delay;
assert_eq!(relative_delay(99, 100), Ok(Duration::ZERO));
assert_eq!(relative_delay(100, 100), Ok(Duration::ZERO));
assert_eq!(relative_delay(101, 100), Ok(Duration::from_nanos(1)));
}
#[test]
fn forward_wall_jump_hides_then_existing_one_shot_drains_without_correction_arm()
-> Result<(), Box<dyn std::error::Error>> {
use crate::ttl::filter::{Visibility, visible_value_at};
let clock = Arc::new(FakeClock::at(100));
let (_dir, mut actor, store) = actor_with_clock(Arc::clone(&clock))?;
let mut latch = DeadlineLatch::default();
actor.put_with_ttl(b"k", b"v", Some(Duration::from_nanos(100)), &store)?;
let generation = arm(&mut actor, &mut latch)?;
let raw = actor.get_raw(b"k", &store)?.ok_or("missing raw value")?;
assert_eq!(
visible_value_at(&raw, 199)?,
Visibility::Live(b"v".to_vec())
);
clock.set(1_000);
let after_jump = actor.expiry_metrics();
assert_eq!(visible_value_at(&raw, 1_000)?, Visibility::Expired);
assert_eq!(after_jump.physical_arms, 1);
assert!(fire(&mut actor, generation, &store)?);
actor.arm_expiry(&mut latch)?;
assert!(actor.get_raw(b"k", &store)?.is_none());
assert_eq!(actor.expiry_metrics().physical_arms, 1);
assert_eq!(latch.arms.len(), 1);
Ok(())
}