fix(storage): #541 review follow-ups — GETEX overflow guard, stale-pair discipline, latch-at-promotion

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

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(

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

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)
     }
 

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;
         }
     }
 

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();