Repository: rabbitmq/rabbitmq-server (language: javascript)

Issue:
Quorum queues: memory spike when applying a `max-length` policy retroactively to a long queue
### Describe the bug

Given a long quorum queue, if I apply a policy to limit the queue's length (in a real-world scenario, likely with the intention of preventing further queue growth and running out of memory), a significant memory spike occurs to drop the messages above the new threshold. This can easily cause the opposite effect than intended: I run out of memory because I was trying to prevent running out of memory...

In my particular case, it was even "funnier": I had a cluster on Kubernetes, applied a policy, the leader was OOMkilled, a new leader was elected and tried to apply the policy, so it was OOMkilled. The remaining node survived because a leader could not be elected, but as soon as one of the nodes restarted, a leader was elected and OOMkilled. A policy that was meant to limit memory usage, cause an OOMkill-loop. :)

### Reproduction steps

1. `make run-broker` (tested on `main`)
2. Publish a significant number of messages: `perf-test -qq -u qq -x 4 -y 0 -c 100 -s 5000 -ms -C 1250000`
3. Apply a policy that sets the limit to a low value: `rabbitmqctl set_policy max qq '{"max-length": 1234}'`
4. Observe memory usage

<img width="862" alt="Screenshot 2024-10-29 at 12 35 47" src="https://github.com/user-attachments/assets/23ff4dc3-2282-4268-8a68-ad85062777b3">

![set_policy-main](https://github.com/user-attachments/assets/54678b8f-2845-4323-a30b-6626bc83f7e0)


### Expected behavior

Ideally there should be no significant spike when dropping messages.

### Additional context

_No response_


Discussion / hints:
We create an effect for each dropped message here: https://github.com/rabbitmq/rabbitmq-server/blob/5f4715979c5b0f0e09cae923d6f4c2386ec03e58/deps/rabbit/src/rabbit_fifo_dlx.erl#L151-L154

That's probably what causes the memory growth combined with the use of `++` to concatenate the resulting effects.

Fix patch:
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
index 6a61d1d2e87f..bed02ecda30b 100644
--- a/deps/rabbit/src/rabbit_fifo.erl
+++ b/deps/rabbit/src/rabbit_fifo.erl
@@ -1596,11 +1596,30 @@ drop_head(#?STATE{ra_indexes = Indexes0} = State0, Effects) ->
             #?STATE{cfg = #cfg{dead_letter_handler = DLH},
                     dlx = DlxState} = State = State3,
             {_, DlxEffects} = rabbit_fifo_dlx:discard([Msg], maxlen, DLH, DlxState),
-            {State, DlxEffects ++ Effects};
+            {State, combine_effects(DlxEffects, Effects)};
         empty ->
             {State0, Effects}
     end.
 
+%% combine global counter update effects to avoid bulding a huge list of
+%% effects if many messages are dropped at the same time as could happen
+%% when the `max_length' is changed via a configuration update.
+combine_effects([{mod_call,
+                  rabbit_global_counters,
+                  messages_dead_lettered,
+                  [Reason, rabbit_quorum_queue, Type, NewLen]}],
+                [{mod_call,
+                  rabbit_global_counters,
+                  messages_dead_lettered,
+                  [Reason, rabbit_quorum_queue, Type, PrevLen]} | Rem]) ->
+    [{mod_call,
+      rabbit_global_counters,
+      messages_dead_lettered,
+      [Reason, rabbit_quorum_queue, Type, PrevLen + NewLen]} | Rem];
+combine_effects(New, Old) ->
+    New ++ Old.
+
+
 maybe_set_msg_ttl(Msg, RaCmdTs, Header,
                   #?STATE{cfg = #cfg{msg_ttl = MsgTTL}}) ->
     case mc:is(Msg) of


Test patch:
diff --git a/deps/rabbit/test/rabbit_fifo_SUITE.erl b/deps/rabbit/test/rabbit_fifo_SUITE.erl
index e14b9406eee8..45f3f2cd12cd 100644
--- a/deps/rabbit/test/rabbit_fifo_SUITE.erl
+++ b/deps/rabbit/test/rabbit_fifo_SUITE.erl
@@ -2185,6 +2185,34 @@ update_config_delivery_limit_test(Config) ->
 
     ok.
 
+update_config_max_length_test(Config) ->
+    QName = rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
+    InitConf = #{name => ?FUNCTION_NAME,
+                 queue_resource => QName,
+                 delivery_limit => 20
+                },
+    State0 = init(InitConf),
+    ?assertMatch(#{config := #{delivery_limit := 20}},
+                 rabbit_fifo:overview(State0)),
+
+    State1 = lists:foldl(fun (Num, FS0) ->
+                         {FS, _} = enq(Config, Num, Num, Num, FS0),
+                         FS
+                     end, State0, lists:seq(1, 100)),
+    Conf = #{name => ?FUNCTION_NAME,
+             queue_resource => QName,
+             max_length => 2,
+             dead_letter_handler => undefined},
+    %% assert only one global counter effect is generated rather than 1 per
+    %% dropped message
+    {State, ok, Effects} = apply(meta(Config, ?LINE),
+                                 rabbit_fifo:make_update_config(Conf), State1),
+    ?assertMatch([{mod_call, rabbit_global_counters, messages_dead_lettered,
+                   [maxlen, rabbit_quorum_queue,disabled, 98]}], Effects),
+    ?assertMatch(#{config := #{max_length := 2},
+                   num_ready_messages := 2}, rabbit_fifo:overview(State)),
+    ok.
+
 purge_nodes_test(Config) ->
     Node = purged@node,
     ThisNode = node(),