chore: check fail for empty listpacks in streams (#5704)

Add check fails when we trim or delete list packs in stream code. Push check in low level functions of lp/stream code

---------

Signed-off-by: kostas <[email protected]>

Author: kostasrim

src/redis/rax.c

@@ -890,11 +890,12 @@ int raxGenericInsert(rax *rax, unsigned char *s, size_t len, void *data, void **
      * do that only if the node is a terminal node, otherwise if the OOM
      * happened reallocating a node in the middle, we don't need to free
      * anything. */
+    fprintf(stderr, "OOM during raxGenericInsert");
     if (h->size == 0) {
         h->isnull = 1;
         h->iskey = 1;
         rax->numele++; /* Compensate the next remove. */
-        assert(raxRemove(rax,s,i,NULL) != 0);
+        checkedRaxRemove(rax, s, i, NULL);
     }
     errno = ENOMEM;
     return 0;
@@ -1941,3 +1942,13 @@ unsigned long raxTouch(raxNode *n) {
     }
     return sum;
 }
+
+int checkedRaxRemove(rax *rax, unsigned char *s, size_t len, void **old) {
+  int res = raxRemove(rax, s, len, old);
+  if(res == 0) {
+    // lp freed but node not removed!
+    fprintf(stderr, "Error: corrupted listpack found.");
+    abort();
+  }
+  return res;
+}

src/redis/rax.h

@@ -209,6 +209,8 @@ uint64_t raxSize(rax *rax);
 unsigned long raxTouch(raxNode *n);
 void raxSetDebugMsg(int onoff);
 
+int checkedRaxRemove(rax *rax, unsigned char *s, size_t len, void **old);
+
 /* Internal API. May be used by the node callback in order to access rax nodes
  * in a low level way, so this function is exported as well. */
 void raxSetData(raxNode *n, void *data);

src/redis/t_stream.c

@@ -28,6 +28,7 @@
  */
 
 #include <errno.h>
+#include <stdio.h>
 #include <string.h>
 
 #include "endianconv.h"
@@ -277,6 +278,13 @@ void streamGetEdgeID(stream *s, int first, int skip_tombstones, streamID *edge_i
     streamIteratorStop(&si);
 }
 
+void checkListPackNotEmpty(unsigned char* lp) {
+  if(lpBytes(lp) == 0) {
+   fprintf(stderr, "Error: corrupted listpack found.");
+   abort();
+  }
+}
+
 /* Trim the stream 's' according to args->trim_strategy, and return the
  * number of elements removed from the stream. The 'approx' option, if non-zero,
  * specifies that the trimming must be performed in a approximated way in
@@ -345,7 +353,7 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
 
         if (remove_node) {
             lpFree(lp);
-            raxRemove(s->rax_tree,ri.key,ri.key_len,NULL);
+            checkedRaxRemove(s->rax_tree, ri.key, ri.key_len, NULL);
             raxSeek(&ri,">=",ri.key,ri.key_len);
             s->length -= entries;
             deleted += entries;
@@ -444,6 +452,7 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
 
         /* Update the listpack with the new pointer. */
         raxInsert(s->rax_tree,ri.key,ri.key_len,lp,NULL);
+        checkListPackNotEmpty(lp);
 
         break; /* If we are here, there was enough to delete in the current
                   node, so no need to go to the next node. */
@@ -739,7 +748,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
         /* If this is the last element in the listpack, we can remove the whole
          * node. */
         lpFree(lp);
-        raxRemove(si->stream->rax_tree,si->ri.key,si->ri.key_len,NULL);
+        checkedRaxRemove(si->stream->rax_tree, si->ri.key, si->ri.key_len, NULL);
     } else {
         /* In the base case we alter the counters of valid/deleted entries. */
         lp = lpReplaceInteger(lp,&p,aux-1);
@@ -750,6 +759,8 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
         /* Update the listpack with the new pointer. */
         if (si->lp != lp)
             raxInsert(si->stream->rax_tree,si->ri.key,si->ri.key_len,lp,NULL);
+
+        checkListPackNotEmpty(lp);
     }
 
     /* Update the number of entries counter. */

src/server/stream_family.cc

@@ -526,6 +526,8 @@ int StreamAppendItem(stream* s, CmdArgList fields, uint64_t now_ms, streamID* ad
     }
     lp = lpAppendInteger(lp, 0); /* Master entry zero terminator. */
     raxInsert(s->rax_tree, (unsigned char*)&rax_key, sizeof(rax_key), lp, NULL);
+    // TODO remove this check
+    CHECK_GT(lpBytes(lp), 0U);
     /* The first entry we insert, has obviously the same fields of the
      * master entry. */
     flags |= STREAM_ITEM_FLAG_SAMEFIELDS;
@@ -610,8 +612,11 @@ int StreamAppendItem(stream* s, CmdArgList fields, uint64_t now_ms, streamID* ad
   lp = lpAppendInteger(lp, lp_count);
 
   /* Insert back into the tree in order to update the listpack pointer. */
-  if (ri.data != lp)
+  if (ri.data != lp) {
     raxInsert(s->rax_tree, (unsigned char*)&rax_key, sizeof(rax_key), lp, NULL);
+    // TODO remove this
+    CHECK_GT(lpBytes(lp), 0U);
+  }
   s->length++;
   s->entries_added++;
   s->last_id = id;

tests/dragonfly/cluster_test.py

@@ -3065,7 +3065,7 @@ async def test_cluster_sharded_pub_sub(df_factory: DflyInstanceFactory):
     await c_nodes[0].execute_command("SPUBLISH kostas hello")
     # We need to sleep cause we use DispatchBrief internally. Otherwise we can't really gurantee
     # that the client received the message
-    await asyncio.sleep(1)
+    await asyncio.sleep(2)
 
     # Consume subscription message result from above
     message = consumer.get_sharded_message(target_node=node_a)
@@ -3075,7 +3075,7 @@ async def test_cluster_sharded_pub_sub(df_factory: DflyInstanceFactory):
     assert message == {"type": "message", "pattern": None, "channel": b"kostas", "data": b"hello"}
 
     consumer.sunsubscribe("kostas")
-    await asyncio.sleep(1)
+    await asyncio.sleep(2)
     await c_nodes[0].execute_command("SPUBLISH kostas new_message")
     message = consumer.get_sharded_message(target_node=node_a)
     assert message == {"type": "unsubscribe", "pattern": None, "channel": b"kostas", "data": 0}