Fix SPOP/RESTORE propagation when doing lazy free (#12320)

In SPOP, when COUNT is greater than or equal to set's size,
we will remove the set. In dbDelete, we will do DEL or UNLINK
according to the lazy flag. This is also required for propagate.

In RESTORE, we won't store expired keys into the db, see #7472.
When used together with REPLACE, it should emit a DEL or UNLINK
according to the lazy flag.

This PR also adds tests to cover the propagation. The RESTORE
test will also cover #7472.

src/cluster.c

@@ -6439,7 +6439,8 @@ void restoreCommand(client *c) {
     if (ttl && !absttl) ttl+=commandTimeSnapshot();
     if (ttl && checkAlreadyExpired(ttl)) {
         if (deleted) {
-            rewriteClientCommandVector(c,2,shared.del,key);
+            robj *aux = server.lazyfree_lazy_server_del ? shared.unlink : shared.del;
+            rewriteClientCommandVector(c, 2, aux, key);
             signalModifiedKey(c,c->db,key);
             notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
             server.dirty++;

src/t_set.c

@@ -798,8 +798,9 @@ void spopWithCountCommand(client *c) {
 
         /* todo: Move the spop notification to be executed after the command logic. */
 
-        /* Propagate this command as a DEL operation */
-        rewriteClientCommandVector(c,2,shared.del,c->argv[1]);
+        /* Propagate this command as a DEL or UNLINK operation */
+        robj *aux = server.lazyfree_lazy_server_del ? shared.unlink : shared.del;
+        rewriteClientCommandVector(c, 2, aux, c->argv[1]);
         signalModifiedKey(c,c->db,c->argv[1]);
         return;
     }

tests/unit/dump.tcl

@@ -96,6 +96,34 @@ start_server {tags {"dump"}} {
         set e
     } {*syntax*}
 
+    test {RESTORE should not store key that are already expired, with REPLACE will propagate it as DEL or UNLINK} {
+        r del key1{t} key2{t}
+        r set key1{t} value2
+        r lpush key2{t} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65
+
+        r set key{t} value
+        set encoded [r dump key{t}]
+        set now [clock milliseconds]
+
+        set repl [attach_to_replication_stream]
+
+        # Keys that have expired will not be stored.
+        r config set lazyfree-lazy-server-del no
+        assert_equal {OK} [r restore key1{t} [expr $now-5000] $encoded replace absttl]
+        r config set lazyfree-lazy-server-del yes
+        assert_equal {OK} [r restore key2{t} [expr $now-5000] $encoded replace absttl]
+        assert_equal {0} [r exists key1{t} key2{t}]
+
+        # Verify the propagate of DEL and UNLINK.
+        assert_replication_stream $repl {
+            {select *}
+            {del key1{t}}
+            {unlink key2{t}}
+        }
+
+        close_replication_stream $repl
+    } {} {needs:repl}
+
     test {DUMP of non existing key returns nil} {
         r dump nonexisting_key
     } {}

tests/unit/type/set.tcl

@@ -765,6 +765,7 @@ foreach type {single multiple single_multiple} {
         assert_encoding $type myset
         set res [r spop myset 30]
         assert {[lsort $content] eq [lsort $res]}
+        assert_equal {0} [r exists myset]
     }
 
     test "SPOP new implementation: code path #2 $type" {
@@ -788,6 +789,29 @@ foreach type {single multiple single_multiple} {
     }
     }
 
+    test "SPOP new implementation: code path #1 propagate as DEL or UNLINK" {
+        r del myset1{t} myset2{t}
+        r sadd myset1{t} 1 2 3 4 5
+        r sadd myset2{t} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65
+
+        set repl [attach_to_replication_stream]
+
+        r config set lazyfree-lazy-server-del no
+        r spop myset1{t} [r scard myset1{t}]
+        r config set lazyfree-lazy-server-del yes
+        r spop myset2{t} [r scard myset2{t}]
+        assert_equal {0} [r exists myset1{t} myset2{t}]
+
+        # Verify the propagate of DEL and UNLINK.
+        assert_replication_stream $repl {
+            {select *}
+            {del myset1{t}}
+            {unlink myset2{t}}
+        }
+
+        close_replication_stream $repl
+    } {} {needs:repl}
+
     test "SRANDMEMBER count of 0 is handled correctly" {
         r srandmember myset 0
     } {}