Merge pull request #12314 from rabbitmq/mergify/bp/v4.0.x/pr-12310

QQ: unblock publishers when queue length policy is deleted (backport #12310)

Author: michaelklishin

deps/rabbit/src/rabbit_fifo.erl

@@ -1913,11 +1913,21 @@ checkout0(_Meta, {_Activity, ExpiredMsg, State0, Effects0}, SendAcc) ->
     Effects = add_delivery_effects(Effects0, SendAcc, State0),
     {State0, ExpiredMsg, lists:reverse(Effects)}.
 
-evaluate_limit(_Index, Result, _BeforeState,
+evaluate_limit(_Index, Result,
+               #?STATE{cfg = #cfg{max_length = undefined,
+                                  max_bytes = undefined}},
                #?STATE{cfg = #cfg{max_length = undefined,
                                   max_bytes = undefined}} = State,
                Effects) ->
     {State, Result, Effects};
+evaluate_limit(_Index, Result, _BeforeState,
+               #?STATE{cfg = #cfg{max_length = undefined,
+                                  max_bytes = undefined},
+                       enqueuers = Enqs0} = State0,
+               Effects0) ->
+    %% max_length and/or max_bytes policies have just been deleted
+    {Enqs, Effects} = unblock_enqueuers(Enqs0, Effects0),
+    {State0#?STATE{enqueuers = Enqs}, Result, Effects};
 evaluate_limit(Index, Result, BeforeState,
                #?STATE{cfg = #cfg{overflow_strategy = Strategy},
                        enqueuers = Enqs0} = State0,
@@ -1947,16 +1957,7 @@ evaluate_limit(Index, Result, BeforeState,
             case {Before, is_below_soft_limit(State0)} of
                 {false, true} ->
                     %% we have moved below the lower limit
-                    {Enqs, Effects} =
-                        maps:fold(
-                          fun (P, #enqueuer{} = E0, {Enqs, Acc})  ->
-                                  E = E0#enqueuer{blocked = undefined},
-                                  {Enqs#{P => E},
-                                   [{send_msg, P, {queue_status, go}, [ra_event]}
-                                    | Acc]};
-                              (_P, _E, Acc) ->
-                                  Acc
-                          end, {Enqs0, Effects0}, Enqs0),
+                    {Enqs, Effects} = unblock_enqueuers(Enqs0, Effects0),
                     {State0#?STATE{enqueuers = Enqs}, Result, Effects};
                 _ ->
                     {State0, Result, Effects0}
@@ -1965,6 +1966,16 @@ evaluate_limit(Index, Result, BeforeState,
             {State0, Result, Effects0}
     end.
 
+unblock_enqueuers(Enqs0, Effects0) ->
+    maps:fold(
+      fun (P, #enqueuer{} = E0, {Enqs, Acc})  ->
+              E = E0#enqueuer{blocked = undefined},
+              {Enqs#{P => E},
+               [{send_msg, P, {queue_status, go}, [ra_event]}
+               | Acc]};
+          (_P, _E, Acc) ->
+              Acc
+      end, {Enqs0, Effects0}, Enqs0).
 
 %% [6,5,4,3,2,1] -> [[1,2],[3,4],[5,6]]
 chunk_disk_msgs([], _Bytes, [[] | Chunks]) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -151,6 +151,7 @@ all_tests() ->
      message_bytes_metrics,
      queue_length_limit_drop_head,
      queue_length_limit_reject_publish,
+     queue_length_limit_policy_cleared,
      subscribe_redelivery_limit,
      subscribe_redelivery_limit_disable,
      subscribe_redelivery_limit_many,
@@ -2973,6 +2974,36 @@ queue_length_limit_reject_publish(Config) ->
     ok = publish_confirm(Ch, QQ),
     ok.
 
+queue_length_limit_policy_cleared(Config) ->
+    [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    ok = rabbit_ct_broker_helpers:set_policy(
+           Config, 0, <<"max-length">>, QQ, <<"queues">>,
+           [{<<"max-length">>, 2},
+            {<<"overflow">>, <<"reject-publish">>}]),
+    timer:sleep(1000),
+    RaName = ra_name(QQ),
+    QueryFun = fun rabbit_fifo:overview/1,
+    ?awaitMatch({ok, {_, #{config := #{max_length := 2}}}, _},
+                rpc:call(Server, ra, local_query, [RaName, QueryFun]),
+                ?DEFAULT_AWAIT),
+    #'confirm.select_ok'{} = amqp_channel:call(Ch, #'confirm.select'{}),
+    ok = publish_confirm(Ch, QQ),
+    ok = publish_confirm(Ch, QQ),
+    ok = publish_confirm(Ch, QQ), %% QQs allow one message above the limit
+    wait_for_messages_ready(Servers, RaName, 3),
+    fail = publish_confirm(Ch, QQ),
+    ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"max-length">>),
+    ?awaitMatch({ok, {_, #{config := #{max_length := undefined}}}, _},
+                rpc:call(Server, ra, local_query, [RaName, QueryFun]),
+                ?DEFAULT_AWAIT),
+    ok = publish_confirm(Ch, QQ),
+    wait_for_messages_ready(Servers, RaName, 4).
+
 purge(Config) ->
     [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),