Fix XADD/XTRIM MAXLEN with ACKED stopping early at a node with referenced entries (#16006)

Fixes #16004
Introduced since #14623

### Problem

`streamTrim()` walks the stream nodes from the head. A node that isn't
entirely inside the trim range is the boundary node: its entries are
trimmed one by one, and then the loop ends, assuming enough was deleted:

```
    if (node_eligible_for_remove)
        continue;
    break; /* there was enough to delete in the current node */
```

That assumption holds for `KEEPREF` and `DELREF`, which delete every
entry they visit. It doesn't hold for `ACKED`, which skips entries that
are still referenced by a consumer group. When the boundary node holds
such entries, `MAXLEN n ACKED` stops with the stream still longer than
`n`, even though later nodes hold acked entries that could be removed.

How far the trim gets therefore depends on where the node boundaries
fall (`stream-node-max-entries` / `stream-node-max-bytes`, tombstones,
AOF rewrite repacking). Exact trims are propagated verbatim, and `~`
trims are rewritten to exact ones. So a replica or an AOF replay with a
different node layout deletes a different set of entries than the
master. For example, with 6 entries where `3-0` and `4-0` are pending,
`XTRIM s MAXLEN = 3 ACKED` deletes 2 entries with
`stream-node-max-entries 2` but 3 entries with `stream-node-max-entries
100`.

### Fix

Keep the `stop` flag of the per-entry loop outside of it, and move on to
the next node whenever the loop finished a node without reaching the
trim target (`!stop`), instead of deciding based on
`node_eligible_for_remove`.

- `MAXLEN` + `ACKED` (exact or `~`): keeps going past nodes with
referenced entries, until `s->length <= maxlen` (already checked at the
top of the loop) or the end of the stream. The result no longer depends
on the node layout.
- `MINID` (any strategy): `!stop` after a node means all of its entries
are `< minid`, which is the same as `node_eligible_for_remove`. No
behavior change.
- `MAXLEN` + `KEEPREF` / `DELREF`: every visited entry is deleted, so
the boundary node always reaches `stop`.

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

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}} {