Do not add a key into db if it has a past expiration time (#14784)

For **RESTORE** and **SET** command, if the expiration time is already
elapsed, we skip adding it to the DB. However, we still increment the
`expiredkeys` counter. From stats perspective we behave as if we
inserted a new key (possibly an overwrite) and later expired it, but
from the per-key KSN observability, we reflect what we've actually done
in the db (deletion of old key, and no insertion of new one), so we
don't confuse modules.

**Changes**:
- **RESTORE**: Increment the `expiredkeys` counter if the key has an
expiration time in the past.
- **SET**: If the key has an expiration time in the past, do not add it
to the main DB, increment the `expiredkeys` counter instead. And we also
delete the old key if it exists.

For reference, **EXPIREAT** with TTL in the past, which implicitly
deletes the key and return success. Now **SET** command has the same
behavior.

src/cluster.c

@@ -284,6 +284,8 @@ void restoreCommand(client *c) {
             notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
             server.dirty++;
         }
+        /* Update the stats, see setGenericCommand for details. */
+        server.stat_expiredkeys++;
         keyMetaSpecCleanup(&keymeta);
         decrRefCount(obj);
         addReply(c, shared.ok);

src/t_string.c

@@ -147,6 +147,30 @@ void setGenericCommand(client *c, int flags, robj *key, robj **valref, robj *exp
         }
     }
 
+    /* If the expire time is already elapsed, we don't need to add the key,
+     * but we still need to update the stats, and we also need to delete the
+     * key if it exists.
+     *
+     * From stats perspective, we behave as if we inserted a new key (possibly
+     * an overwrite) and later expired it, but from the per-key KSN observability,
+     * we reflect what we've actually done in the db (deletion of old key, and
+     * no insertion of new one), so we don't confuse modules. */
+    if (expire && checkAlreadyExpired(milliseconds)) {
+        if (found) {
+            dbDelete(c->db, key);
+            robj *aux = server.lazyfree_lazy_server_del ? shared.unlink : shared.del;
+            rewriteClientCommandVector(c, 2, aux, key);
+            keyModified(c, c->db, key, NULL, 1);
+            notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, c->db->id);
+            server.dirty++;
+        }
+        server.stat_expiredkeys++;
+        if (!(flags & OBJ_SET_GET)) {
+            addReply(c, ok_reply ? ok_reply : shared.ok);
+        }
+        return;
+    }
+
     /* When expire is not NULL, we avoid deleting the TTL so it can be updated later instead of being deleted and then created again. */
     setkey_flags |= ((flags & OBJ_KEEPTTL) || expire) ? SETKEY_KEEPTTL : 0;
     setkey_flags |= found ? SETKEY_ALREADY_EXIST : SETKEY_DOESNT_EXIST;

tests/unit/dump.tcl

@@ -42,9 +42,13 @@ start_server {tags {"dump"}} {
         set encoded [r dump foo]
         set now [clock milliseconds]
         r debug set-active-expire 0
+        set expiredkeys [s expired_keys]
         r restore foo [expr $now-3000] $encoded absttl REPLACE
         catch {r debug object foo} e
         r debug set-active-expire 1
+        # Verify that expired_keys was incremented, even though
+        # the key was not added to the DB actually.
+        assert_equal [expr $expiredkeys + 1] [s expired_keys]
         set e
     } {ERR no such key} {needs:debug}
 

tests/unit/type/string.tcl

@@ -665,6 +665,21 @@ if {[string match {*jemalloc*} [s mem_allocator]]} {
         list $old_value $new_value
     } {bar bar}
 
+    test {Extended SET GET option with a past expiration time and no previous value} {
+        r del foo
+        r debug set-active-expire 0
+        set now [clock milliseconds]
+        set expiredkeys [s expired_keys]
+        set old_value [r set foo baz GET PXAT [expr $now-3000]]
+        assert_equal $old_value {}
+        # Verify that expired_keys was incremented, even though
+        # the key was not added to the DB actually.
+        assert_equal [expr $expiredkeys + 1] [s expired_keys]
+        catch {r debug object foo} e
+        r debug set-active-expire 1
+        set e
+    } {ERR no such key} {needs:debug}
+
     test {Extended SET GET with incorrect type should result in wrong type error} {
       r del foo
       r rpush foo waffle
@@ -698,6 +713,43 @@ if {[string match {*jemalloc*} [s mem_allocator]]} {
         r set foo bar pxat [expr [clock milliseconds] + 10000]
         assert_range [r ttl foo] 5 10
     }
+
+    test {Extended SET PXAT option with a past expiration time} {
+        r set foo bar
+        r debug set-active-expire 0
+        set now [clock milliseconds]
+        set expiredkeys [s expired_keys]
+        r set foo baz PXAT [expr $now-3000]
+        # Verify that expired_keys was incremented, even though
+        # the key was not added to the DB actually.
+        assert_equal [expr $expiredkeys + 1] [s expired_keys]
+        catch {r debug object foo} e
+        r debug set-active-expire 1
+        set e
+    } {ERR no such key} {needs:debug}
+
+    test {SET PXAT with a past expiration time will propagate it as DEL or UNLINK} {
+        r flushall
+        r set foo foo
+        r set bar bar
+        set repl [attach_to_replication_stream]
+
+        # Keys that have expired timestamp will be deleted immediately
+        set now [clock milliseconds]
+        r config set lazyfree-lazy-server-del no
+        assert_equal {OK} [r set foo foo PXAT [expr $now-3000]]
+        r config set lazyfree-lazy-server-del yes
+        assert_equal {OK} [r set bar bar PXAT [expr $now-3000]]
+
+        # Verify the propagate of DEL and UNLINK.
+        assert_replication_stream $repl {
+            {select *}
+            {del foo}
+            {unlink bar}
+        }
+        close_replication_stream $repl
+    } {} {needs:repl}
+
     test {Extended SET using multiple options at once} {
         r set foo val
         assert {[r set foo bar xx px 10000] eq {OK}}
@@ -1215,7 +1267,6 @@ if {[string match {*jemalloc*} [s mem_allocator]]} {
     test {Extended SET with IFDEQ - key exists and digest matches} {
         r set mykey "hello"
         set digest [r digest mykey]
-        puts $digest
         assert_equal "OK" [r set mykey "world" IFDEQ $digest]
         assert_equal "world" [r get mykey]
     }