Fix crashes when list-compress-depth is used. (#9779)

Recently we started using list-compress-depth in tests (was completely untested till now).
Turns this triggered test failures with the external mode, since the tests left the setting enabled
and then it was used in other tests (specifically the fuzzer named "Stress tester for #3343-alike bugs").

This PR fixes the issue of the `recompress` flag being left set by mistake, which caused the code to
later to compress the head or tail nodes (which should never be compressed)

The solution is to reset the recompress flag when it should have been (when it was decided not to compress).

Additionally we're adding some assertions and improve the tests so in order to catch other similar bugs.

3 files changed, 51 insertions, 4 deletions

diff --git a/src/quicklist.c b/src/quicklist.c
index 22bd9b516..98ede7d90 100644
--- a/src/quicklist.c
+++ b/src/quicklist.c
@@ -199,6 +199,11 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
     node->attempted_compress = 1;
 #endif
 
+    /* validate that the node is neither
+     * tail nor head (it has prev and next)*/
+    assert(node->prev && node->next);
+
+    node->recompress = 0;
     /* Don't bother compressing small values */
     if (node->sz < MIN_COMPRESS_BYTES)
         return 0;
@@ -217,7 +222,6 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
     zfree(node->entry);
     node->entry = (unsigned char *)lzf;
     node->encoding = QUICKLIST_NODE_ENCODING_LZF;
-    node->recompress = 0;
     return 1;
 }
 
@@ -235,6 +239,7 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
 #ifdef REDIS_TEST
     node->attempted_compress = 0;
 #endif
+    node->recompress = 0;
 
     void *decompressed = zmalloc(node->sz);
     quicklistLZF *lzf = (quicklistLZF *)node->entry;
@@ -313,6 +318,11 @@ void quicklistRepr(unsigned char *ql, int full) {
  * If compress depth is larger than the entire list, we return immediately. */
 REDIS_STATIC void __quicklistCompress(const quicklist *quicklist,
                                       quicklistNode *node) {
+    /* The head and tail should never be compressed (we should not attempt to recompress them)
+     * This needs to be an assertion in the future */
+    if (quicklist->head) quicklist->head->recompress = 0;
+    if (quicklist->tail) quicklist->tail->recompress = 0;
+
     /* If length is less than our compress depth (from both sides),
      * we can't compress anything. */
     if (!quicklistAllowsCompression(quicklist) ||
@@ -1565,6 +1575,9 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
         return 0;
     }
 
+    /* The head and tail should never be compressed */
+    assert(node->encoding != QUICKLIST_NODE_ENCODING_LZF);
+
     if (unlikely(QL_NODE_IS_PLAIN(node))) {
         if (data)
             *data = saver(node->entry, node->sz);
@@ -1628,6 +1641,12 @@ int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
 /* Wrapper to allow argument-based switching between HEAD/TAIL pop */
 void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
                    int where) {
+    /* The head and tail should never be compressed (we don't attempt to decompress them) */
+    if (quicklist->head)
+        assert(quicklist->head->encoding != QUICKLIST_NODE_ENCODING_LZF);
+    if (quicklist->tail)
+        assert(quicklist->tail->encoding != QUICKLIST_NODE_ENCODING_LZF);
+
     if (where == QUICKLIST_HEAD) {
         quicklistPushHead(quicklist, value, sz);
     } else if (where == QUICKLIST_TAIL) {
diff --git a/tests/unit/type/list-3.tcl b/tests/unit/type/list-3.tcl
index 94cedbccc..3f3ea6cfd 100644
--- a/tests/unit/type/list-3.tcl
+++ b/tests/unit/type/list-3.tcl
@@ -26,10 +26,29 @@ start_server {
         r ping ; # It's enough if the server is still alive
     } {PONG}
 
-    test {Stress tester for #3343-alike bugs} {
+    test {Check compression with recompress} {
         r del key
-        for {set j 0} {$j < 10000} {incr j} {
-            set op [randomInt 6]
+        config_set list-compress-depth 1
+        config_set list-max-ziplist-size 16
+        r rpush key a
+        r rpush key [string repeat b 50000]
+        r rpush key c
+        r lset key 1 d
+        r rpop key
+        r rpush key [string repeat e 5000]
+        r linsert key before f 1
+        r rpush key g
+   }
+
+foreach comp {2 1 0} {
+    set cycles 1000
+    if {$::accurate} { set cycles 10000 }
+    config_set list-compress-depth $comp
+    
+    test "Stress tester for #3343-alike bugs comp: $comp" {
+        r del key
+        for {set j 0} {$j < $cycles} {incr j} {
+            set op [randomInt 7]
             set small_signed_count [expr 5-[randomInt 10]]
             if {[randomInt 2] == 0} {
                 set ele [randomInt 1000]
@@ -53,9 +72,15 @@ start_server {
                     }
                     r linsert key $where $otherele $ele
                 }
+                6 {
+                    set index [randomInt [r llen key]]
+                    set otherele [r lindex key $index]
+                    r lrem key 1 $otherele
+                }
             }
         }
     }
+} ;# foreach comp
 
     tags {slow} {
         test {ziplist implementation: value encoding and backlink} {
diff --git a/tests/unit/type/list.tcl b/tests/unit/type/list.tcl
index 4b1dc156d..44f201276 100644
--- a/tests/unit/type/list.tcl
+++ b/tests/unit/type/list.tcl
@@ -122,6 +122,9 @@ start_server [list overrides [list save ""] ] {
         r LSET list6 0 [string repeat d 500]
         assert_equal [string repeat d 500] [r lindex list6 0]
     } {} {needs:debug}
+
+    # revert config for external mode tests.
+    r config set list-compress-depth 0
 }
 
 # check functionality of plain nodes using low packed-threshold