Fix XTRIM or XADD with LIMIT may delete more entries than Count. (red…

…is#9048)

The decision to stop trimming due to LIMIT in XADD and XTRIM was after the limit was reached.
i.e. the code was deleting **at least** that count of records (from the LIMIT argument's perspective, not the MAXLEN),
instead of **up to** that count of records.
see redis#9046

src/t_stream.c

@@ -702,16 +702,16 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
 
     int64_t deleted = 0;
     while (raxNext(&ri)) {
-        /* Check if we exceeded the amount of work we could do */
-        if (limit && deleted >= limit)
-            break;
-
         if (trim_strategy == TRIM_STRATEGY_MAXLEN && s->length <= maxlen)
             break;
 
         unsigned char *lp = ri.data, *p = lpFirst(lp);
         int64_t entries = lpGetInteger(p);
 
+        /* Check if we exceeded the amount of work we could do */
+        if (limit && (deleted + entries) > limit)
+            break;
+
         /* Check if we can remove the whole node. */
         int remove_node;
         streamID master_id = {0}; /* For MINID */

tests/unit/type/stream.tcl

@@ -199,6 +199,15 @@ start_server {
         assert {[r EXISTS otherstream] == 0}
     }
 
+    test {XADD with LIMIT delete entries no more than limit} {
+        r del yourstream
+        for {set j 0} {$j < 3} {incr j} {
+            r XADD yourstream * xitem v
+        }
+        r XADD yourstream MAXLEN ~ 0 limit 1 * xitem v
+        assert {[r XLEN yourstream] == 4}
+    }
+
     test {XRANGE COUNT works as expected} {
         assert {[llength [r xrange mystream - + COUNT 10]] == 10}
     }
@@ -525,6 +534,16 @@ start_server {
         }
         assert_error ERR* {r XTRIM mystream MAXLEN 1 LIMIT 30}
     }
+
+    test {XTRIM with LIMIT delete entries no more than limit} {
+        r del mystream
+        r config set stream-node-max-entries 2
+        for {set j 0} {$j < 3} {incr j} {
+            r XADD mystream * xitem v
+        }
+        assert {[r XTRIM mystream MAXLEN ~ 0 LIMIT 1] == 0}
+        assert {[r XTRIM mystream MAXLEN ~ 0 LIMIT 2] == 2}
+    }
 }
 
 start_server {tags {"stream"} overrides {appendonly yes}} {