diff --git a/src/t_stream.c b/src/t_stream.c index 911306c5f4..288f27505a 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -1124,6 +1124,7 @@ int64_t streamTrim(redisDb *db, stream *s, streamAddTrimArgs *args, streamID *ma /* 'p' is now pointing to the first entry inside the listpack. * We have to run entry after entry, marking entries as deleted * if they are already not deleted. */ + int stop = 0; while (p) { /* We keep a copy of p (which point to flags part) in order to * update it after (and if) we actually remove the entry */ @@ -1142,7 +1143,6 @@ int64_t streamTrim(redisDb *db, stream *s, streamAddTrimArgs *args, streamID *ma currid.ms = master_id.ms + ms_delta; currid.seq = master_id.seq + seq_delta; - int stop; if (trim_strategy == TRIM_STRATEGY_MAXLEN) { stop = s->length <= maxlen; } else { @@ -1218,10 +1218,10 @@ int64_t streamTrim(redisDb *db, stream *s, streamAddTrimArgs *args, streamID *ma /* Update the node with the new pointer. */ raxIteratorSetData(&ri,lp); - /* If the node is eligible for removal but we couldn't remove it due to delete strategy - * constraints (we need to check each entry individually), continue to the next node - * instead of stopping here. */ - if (node_eligible_for_remove) + /* If we didn't reach the trim limit within this node (it was eligible for removal, + * or some entries were skipped because they are still referenced), continue to the + * next node instead of stopping here. */ + if (!stop) continue; break; /* If we are here, there was enough to delete in the current diff --git a/tests/unit/type/stream.tcl b/tests/unit/type/stream.tcl index 69935bc26d..ceea4d035f 100644 --- a/tests/unit/type/stream.tcl +++ b/tests/unit/type/stream.tcl @@ -3541,6 +3541,28 @@ start_server { r config set stream-node-max-entries $origin_max_entries } + + test {XTRIM with exact MAXLEN and ACKED continues past nodes with unacked entries} { + r DEL mystream + set origin_max_entries [config_get_set stream-node-max-entries 2] + + # Create 6 entries in 3 nodes (2 entries per node) + for {set i 1} {$i <= 6} {incr i} { + r XADD mystream $i-0 f v + } + + r XGROUP CREATE mystream mygroup 0 + r XREADGROUP GROUP mygroup consumer1 STREAMS mystream > + + # 3-0 and 4-0 (second node) stay pending, the rest are acked + r XACK mystream mygroup 1-0 2-0 5-0 6-0 + + # The trim must not stop at the second node, 5-0 is acked and can go + assert_equal 3 [r XTRIM mystream MAXLEN = 3 ACKED] + assert_equal {{3-0 {f v}} {4-0 {f v}} {6-0 {f v}}} [r XRANGE mystream - +] + + r config set stream-node-max-entries $origin_max_entries + } } start_server {tags {"stream needs:debug"} overrides {appendonly yes}} {