diff --git a/CHANGELOG.md b/CHANGELOG.md index c4863786..f8ac4c99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,7 +21,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `set_expiry` (previously a GETEX-only TTL never armed the sweep latch, so the key was invisible to active expiry forever), and EXPIRE-family commands hitting an already-expired key now hide and queue it for the emitting drain (#542 semantics) instead of deleting it silently with no - `expired` notification and no dual-plane DEL. + `expired` notification and no dual-plane DEL. Review follow-ups: GETEX `EX`/`EXAT` seconds + values that overflow the millisecond conversion now answer the range error instead of + wrapping (debug builds panicked); the sweep only drops an index pair that is provably stale + (entry gone or TTL retargeted) so a backwards wall-clock step can't discard live pairs; the + per-key budget clock read is batched to every 64 pops; and sweep 2 lowers the hash latch from + its own reap outcomes instead of a second O(N) rescan. ### Added - **An acceptance suite driven by unmodified redis-py** (`scripts/client-compat/redis_py/`), wired diff --git a/src/command/string/string_read.rs b/src/command/string/string_read.rs index c6a67af0..fa155546 100644 --- a/src/command/string/string_read.rs +++ b/src/command/string/string_read.rs @@ -262,9 +262,9 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame { if args.len() < 3 { return Frame::Error(Bytes::from_static(b"ERR syntax error")); } - match parse_positive_i64(&args[2]) { - Some(secs) => { - db.set_expiry(&key, current_time_ms() + (secs as u64) * 1000); + match parse_positive_i64(&args[2]).and_then(|secs| (secs as u64).checked_mul(1000)) { + Some(ms) => { + db.set_expiry(&key, current_time_ms().saturating_add(ms)); } None => { return Frame::Error(Bytes::from_static( @@ -290,9 +290,9 @@ pub fn getex(db: &mut Database, args: &[Frame]) -> Frame { if args.len() < 3 { return Frame::Error(Bytes::from_static(b"ERR syntax error")); } - match parse_positive_i64(&args[2]) { - Some(ts) => { - db.set_expiry(&key, (ts as u64) * 1000); + match parse_positive_i64(&args[2]).and_then(|ts| (ts as u64).checked_mul(1000)) { + Some(ms) => { + db.set_expiry(&key, ms); } None => { return Frame::Error(Bytes::from_static( diff --git a/src/server/expiration.rs b/src/server/expiration.rs index c4374f88..965d9646 100644 --- a/src/server/expiration.rs +++ b/src/server/expiration.rs @@ -145,54 +145,70 @@ fn expire_cycle(db: &mut Database, on_removed: &mut dyn FnMut(&[u8])) { // ── Sweep 1: deadline-ordered whole-key expiry (moon#541) ─────────────── let now_ms = current_time_ms(); + let mut popped = 0u32; while let Some((ts, key)) = db.peek_due_expiry(now_ms) { if db.is_key_expired(key.as_bytes()) { // `remove` unindexes the entry's CURRENT pair via `remove_hot`. db.remove(key.as_bytes()); on_removed(key.as_bytes()); - } else { - // The pair failed re-verification: the entry is gone or carries - // a different TTL than when this pair was written — a stale - // pair a writer failed to retire (writer-coverage bug; the + } else if db + .data() + .get(key.as_bytes()) + .is_none_or(|e| e.expires_at_ms() != ts) + { + // The pair is PROVABLY stale: the entry is gone or its TTL was + // retargeted since this pair was written — a pair a writer + // failed to retire (writer-coverage bug; the // debug_expiry_index_consistent oracle exists to catch those in // tests). Drop it or this loop would peek it forever. db.drop_expiry_index_pair(ts, &key); + } else { + // The pair matches the entry exactly, yet the fresh clock says + // "not expired" — the wall clock stepped backwards between the + // cycle-start peek and this re-verification. The pair is VALID, + // just not due; keep it for a later tick. The index is ordered, + // so nothing after the head is due either. + break; } - if start.elapsed() >= budget { + // Budget check every 64 pops, not per key: `Instant::elapsed` is a + // clock read, and the common tick pops far fewer than 64. At least + // one key is always processed before the first check can stop us. + popped += 1; + if popped % 64 == 0 && start.elapsed() >= budget { break; } } // ── Sweep 2: hash-field expiry (latch-gated, moon#541) ─────────────────── + // The reap outcomes double as the latch's self-reset evidence: a key + // stays eligible only while it remains `HashWithTtl` (FieldsRemoved / + // NoOp); Downgraded and KeyDeleted leave the kind. When zero eligible + // keys remain after the sweep, the latch lowers — no second O(N) scan. if db.hash_field_ttl_possible() { // Collect keys up front to avoid borrow conflicts during mutation. let hash_keys = db.hashes_with_field_expiry(); + let mut remaining = 0usize; for key in &hash_keys { - let outcome = db.reap_expired_fields_one_hash(key.as_bytes()); - if outcome == ReapOutcome::KeyDeleted { - db.remove(key.as_bytes()); + match db.reap_expired_fields_one_hash(key.as_bytes()) { + ReapOutcome::KeyDeleted => { + db.remove(key.as_bytes()); + } + ReapOutcome::Downgraded => {} + ReapOutcome::FieldsRemoved | ReapOutcome::NoOp => remaining += 1, } } + if remaining == 0 { + db.clear_hash_field_ttl_latch(); + } } // ── Flag maintenance ───────────────────────────────────────────────────── // Clear the fast-path flag only when both sweeps have nothing left. // If hash-field TTLs remain, the flag must stay set so future ticks - // continue to run sweep 2. Both checks are O(1)-or-latch-gated now: - // the whole-key side reads the index's emptiness, and the hash side - // only rescans while the latch is up (lowering it once the scan - // proves zero HashWithTtl keys remain — the self-reset gate). - let no_whole_key_expiry = db.expiry_index_is_empty(); - let no_hash_field_expiry = if db.hash_field_ttl_possible() { - let none_remain = db.hashes_with_field_expiry().is_empty(); - if none_remain { - db.clear_hash_field_ttl_latch(); - } - none_remain - } else { - true - }; - if no_whole_key_expiry && no_hash_field_expiry { + // continue to run sweep 2. Both checks are O(1) now: the whole-key + // side reads the index's emptiness, and the hash side reads the latch + // sweep 2 just maintained from its own reap outcomes. + if db.expiry_index_is_empty() && !db.hash_field_ttl_possible() { db.clear_maybe_has_expiring_keys(); } } @@ -285,13 +301,77 @@ mod tests { ); } + // Deterministic under CI preemption: a stalled runner can trip the + // 1ms budget mid-sweep, so allow a bounded number of cycles and + // assert the CUMULATIVE count. Still red on the sampling sweep: 20 + // cycles × a 20-key sample of 10_050 finds ~2 due keys, not 50. let mut removed = 0usize; - expire_cycle(&mut db, &mut |_| removed += 1); + let mut cycles = 0; + while removed < 50 && cycles < 20 { + expire_cycle(&mut db, &mut |_| removed += 1); + cycles += 1; + } - assert_eq!(removed, 50, "one cycle must remove exactly the due keys"); + assert_eq!(removed, 50, "the due keys must all be removed promptly"); assert_eq!(db.len(), 10_000, "live keys must all survive"); } + /// A pair that fails the expiry re-check but still matches its entry's + /// TTL exactly must be KEPT (the only honest explanation is a backwards + /// wall-clock step); only a provably-stale pair — entry gone or TTL + /// retargeted — is dropped, and the entry itself is never deleted. + #[test] + fn sweep_drops_only_provably_stale_pairs() { + let mut db = Database::new(); + let future_ms = current_time_ms() + 3_600_000; + db.set( + Bytes::from_static(b"k"), + Entry::new_string_with_expiry(Bytes::from_static(b"v"), future_ms), + ); + assert_eq!(db.expiry_index_len(), 1); + + // Inject a bogus DUE pair for the same key (simulating a pair a + // buggy writer failed to retire): due by the clock, but the entry's + // real TTL differs -> provably stale -> dropped, entry untouched. + db.expiry_index_insert(current_time_ms() - 1_000, b"k"); + assert_eq!(db.expiry_index_len(), 2); + + let mut removed = 0usize; + expire_cycle(&mut db, &mut |_| removed += 1); + + assert_eq!(removed, 0, "a stale pair must never delete a live entry"); + assert_eq!(db.len(), 1, "the entry survives"); + assert_eq!( + db.expiry_index_len(), + 1, + "the stale pair is dropped, the real pair is kept" + ); + assert!(db.debug_expiry_index_consistent()); + } + + /// GETEX EX/EXAT with a seconds value near i64::MAX must answer the + /// range error, not overflow the *1000 conversion (debug builds + /// panicked; release builds silently wrapped to a bogus TTL). + #[test] + fn getex_rejects_overflowing_seconds() { + use crate::protocol::Frame; + let mut db = Database::new(); + for opt in [&b"EX"[..], &b"EXAT"[..]] { + db.set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v")); + let args = [ + Frame::BulkString(Bytes::from_static(b"k")), + Frame::BulkString(Bytes::copy_from_slice(opt)), + Frame::BulkString(Bytes::from(i64::MAX.to_string())), + ]; + let reply = crate::command::string::getex(&mut db, &args); + assert!( + matches!(reply, Frame::Error(_)), + "overflowing {} must answer the range error", + String::from_utf8_lossy(opt) + ); + } + } + /// GETEX wrote its TTL through a raw `get_mut` + `set_expires_at_ms`, /// bypassing `Database::set_expiry` — so the DB-level latch never /// flipped and a key whose ONLY TTL came from GETEX was invisible to diff --git a/src/storage/db/hash_ttl.rs b/src/storage/db/hash_ttl.rs index 0e2923a6..e4bafcce 100644 --- a/src/storage/db/hash_ttl.rs +++ b/src/storage/db/hash_ttl.rs @@ -106,6 +106,12 @@ impl Database { // 4. Promote to HashWithTtl if needed. promote_to_hash_with_ttl(rv); + // moon#541: arm the latch the moment the value IS HashWithTtl — the + // NX/XX/GT/LT gate below can return early AFTER promotion (e.g. GT + // on a non-volatile field), and a HashWithTtl existing with the + // latch down breaks the latch's conservativeness invariant (the + // debug_expiry_index_consistent oracle checks exactly that). + self.hash_field_ttl_latch = true; let RedisValue::HashWithTtl { ttls, min_expiry_ms, @@ -135,10 +141,6 @@ impl Database { *min_expiry_ms = ts_ms; } self.maybe_has_expiring_keys = true; - // moon#541: arm the hash-specific latch too, so the sweep's - // HashWithTtl scan only runs for databases that actually use - // field TTLs. - self.hash_field_ttl_latch = true; Ok(1) } diff --git a/src/storage/db/kv_ops.rs b/src/storage/db/kv_ops.rs index e1caab19..295daba6 100644 --- a/src/storage/db/kv_ops.rs +++ b/src/storage/db/kv_ops.rs @@ -516,9 +516,10 @@ impl Database { self.used_memory = total; self.maybe_has_expiring_keys = any_expiring; self.expiry_index = index; - if any_hash_ttl { - self.hash_field_ttl_latch = true; - } + // Authoritative, like the index rebuild above: the scan just proved + // exactly whether any HashWithTtl exists, so a restore WITHOUT them + // also clears a previously raised latch. + self.hash_field_ttl_latch = any_hash_ttl; } /// Pre-size the internal hash table for an expected key count. @@ -540,6 +541,11 @@ impl Database { let new_table = DashTable::with_capacity(additional); self.data = new_table; self.maybe_has_expiring_keys = false; + // moon#541: the replaced table's index/latch state goes with it — + // matching `clear` (no-ops on the documented empty-db call, but + // the misuse case must not leave stale pairs behind). + self.expiry_index.clear(); + self.hash_field_ttl_latch = false; } } diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs index 9d26eb1d..34a104fd 100644 --- a/src/storage/db/mod.rs +++ b/src/storage/db/mod.rs @@ -508,6 +508,13 @@ impl Database { /// Earliest `(expires_at_ms, key)` pair that is due at `now_ms`, or /// `None` when nothing is due. O(log n). Returns a clone (inline for /// keys ≤ 23B) so the sweep can mutate the db while holding it. + /// + /// "Due" here (`ts <= now_ms`) is aligned with `Entry::is_expired_at` + /// (`now >= ttl`): with a monotone clock, a pair this returns always + /// re-verifies as expired. A FAILED re-verification is therefore NOT + /// proof of staleness on its own — the wall clock may have stepped + /// backwards after `now_ms` was captured — so the sweep only drops a + /// pair when the entry is gone or its TTL differs from `ts`. #[inline] pub fn peek_due_expiry(&self, now_ms: u64) -> Option<(u64, CompactKey)> { self.expiry_index @@ -517,9 +524,10 @@ impl Database { } /// Drop one specific index pair. The sweep calls this when a popped - /// pair fails re-verification (the entry is gone or carries a different - /// TTL — a stale pair): without dropping it, the sweep would peek the - /// same head pair forever. + /// pair is PROVABLY stale (the entry is gone or carries a different + /// TTL than `ts`): without dropping it, the sweep would peek the same + /// head pair forever. A pair that merely failed the expiry re-check is + /// not dropped — see [`Self::peek_due_expiry`]. #[inline] pub fn drop_expiry_index_pair(&mut self, ts: u64, key: &CompactKey) { self.expiry_index.remove(&(ts, key.clone())); @@ -570,7 +578,10 @@ impl Database { /// #541 property battery — O(N), never call on a hot path): the index /// must equal the scan-derived pair set exactly, and the hash latch /// must be conservative (any `HashWithTtl` present ⇒ latch raised). - pub fn debug_expiry_index_consistent(&self) -> bool { + /// Test-only: an oracle for the #541 battery, not part of the storage + /// API — compiled out of production builds entirely. + #[cfg(test)] + pub(crate) fn debug_expiry_index_consistent(&self) -> bool { let scan: std::collections::BTreeSet<(u64, CompactKey)> = self .data .iter() @@ -1230,6 +1241,29 @@ mod tests { assert!(db.hash_field_ttl_possible()); } + /// The NX/XX/GT/LT gate runs AFTER promotion: HEXPIRE GT on a + /// non-volatile field answers -2, but the value has already become + /// `HashWithTtl` — the latch must arm at promotion, not at success, or + /// a `HashWithTtl` exists with the latch down (breaking the latch's + /// conservativeness invariant the oracle checks). + #[test] + fn hash_field_ttl_latch_arms_even_when_condition_fails() { + let mut db = Database::new(); + { + let map = db.get_or_create_hash(b"h").expect("hash"); + map.insert(Bytes::from_static(b"f"), Bytes::from_static(b"v")); + } + let future_ms = db.now_ms() + 60_000; + // GT on a field with NO current TTL: condition not met. + let r = db.hash_set_field_ttl(b"h", b"f", future_ms, HashTtlCond::Gt); + assert_eq!(r, Ok(-2)); + assert!( + db.hash_field_ttl_possible(), + "promotion happened, so the latch must be up" + ); + assert!(db.debug_expiry_index_consistent()); + } + #[test] fn test_data_accessor() { let mut db = Database::new();