Reset lazyfreed_objects info field with RESETSTAT, test for stream lazyfree (#8934)

And also add tests to cover lazy free of streams with various types of
metadata (see #8932)

src/lazyfree.c

@@ -70,6 +70,10 @@ size_t lazyfreeGetFreedObjectsCount(void) {
     return aux;
 }
 
+void lazyfreeResetStats() {
+    atomicSet(lazyfreed_objects,0);
+}
+
 /* Return the amount of work needed in order to free an object.
  * The return value is not always the actual number of allocations the
  * object is composed of, but a number proportional to it.

src/server.c

@@ -3127,6 +3127,7 @@ void resetServerStats(void) {
     server.stat_total_error_replies = 0;
     server.stat_dump_payload_sanitizations = 0;
     server.aof_delayed_fsync = 0;
+    lazyfreeResetStats();
 }
 
 /* Make the thread killable at any time, so that kill threads functions

src/server.h

@@ -2371,6 +2371,7 @@ void emptyDbAsync(redisDb *db);
 void slotToKeyFlush(int async);
 size_t lazyfreeGetPendingObjectsCount(void);
 size_t lazyfreeGetFreedObjectsCount(void);
+void lazyfreeResetStats(void);
 void freeObjAsync(robj *key, robj *obj);
 void freeSlotsToKeysMapAsync(rax *rt);
 void freeSlotsToKeysMap(rax *rt, int async);

tests/unit/lazyfree.tcl

@@ -36,4 +36,45 @@ start_server {tags {"lazyfree"}} {
             fail "Memory is not reclaimed by FLUSHDB ASYNC"
         }
     }
+
+    test "lazy free a stream with all types of metadata" {
+        r config resetstat
+        r config set stream-node-max-entries 5
+        for {set j 0} {$j < 1000} {incr j} {
+            if {rand() < 0.9} {
+                r xadd stream * foo $j
+            } else {
+                r xadd stream * bar $j
+            }
+        }
+        r xgroup create stream mygroup 0
+        set records [r xreadgroup GROUP mygroup Alice COUNT 2 STREAMS stream >]
+        r xdel stream [lindex [lindex [lindex [lindex $records 0] 1] 1] 0]
+        r xack stream mygroup [lindex [lindex [lindex [lindex $records 0] 1] 0] 0]
+        r unlink stream
+
+        # make sure it was lazy freed
+        wait_for_condition 5 100 {
+            [s lazyfree_pending_objects] == 0
+        } else {
+            fail "lazyfree isn't done"
+        }
+        assert_equal [s lazyfreed_objects] 1
+    }
+
+    test "lazy free a stream with deleted cgroup" {
+        r config resetstat
+        r xadd s * a b
+        r xgroup create s bla $
+        r xgroup destroy s bla
+        r unlink s
+
+        # make sure it was not lazy freed
+        wait_for_condition 5 100 {
+            [s lazyfree_pending_objects] == 0
+        } else {
+            fail "lazyfree isn't done"
+        }
+        assert_equal [s lazyfreed_objects] 0
+    }
 }