Merge pull request #12250 from rabbitmq/delivery-limit-disable

QQ: a delivery-limit of -1 disables the delivery limit.

Author: michaelklishin

deps/rabbit/src/rabbit_fifo.erl

@@ -195,7 +195,14 @@ update_config(Conf, State) ->
     Overflow = maps:get(overflow_strategy, Conf, drop_head),
     MaxLength = maps:get(max_length, Conf, undefined),
     MaxBytes = maps:get(max_bytes, Conf, undefined),
-    DeliveryLimit = maps:get(delivery_limit, Conf, undefined),
+    DeliveryLimit = case maps:get(delivery_limit, Conf, undefined) of
+                        DL when is_number(DL) andalso
+                                DL < 0 ->
+                            undefined;
+                        DL ->
+                            DL
+                    end,
+
     Expires = maps:get(expires, Conf, undefined),
     MsgTTL = maps:get(msg_ttl, Conf, undefined),
     ConsumerStrategy = case maps:get(single_active_consumer_on, Conf, false) of
@@ -615,24 +622,25 @@ apply(Meta, {nodeup, Node}, #?STATE{consumers = Cons0,
     checkout(Meta, State0, State, Effects);
 apply(_, {nodedown, _Node}, State) ->
     {State, ok};
-apply(#{index := _Idx} = Meta, #purge_nodes{nodes = Nodes}, State0) ->
+apply(Meta, #purge_nodes{nodes = Nodes}, State0) ->
     {State, Effects} = lists:foldl(fun(Node, {S, E}) ->
                                            purge_node(Meta, Node, S, E)
                                    end, {State0, []}, Nodes),
     {State, ok, Effects};
-apply(#{index := _Idx} = Meta,
-      #update_config{config = #{dead_letter_handler := NewDLH} = Conf},
+apply(Meta,
+      #update_config{config = #{} = Conf},
       #?STATE{cfg = #cfg{dead_letter_handler = OldDLH,
                          resource = QRes},
               dlx = DlxState0} = State0) ->
+    NewDLH = maps:get(dead_letter_handler, Conf, OldDLH),
     {DlxState, Effects0} = rabbit_fifo_dlx:update_config(OldDLH, NewDLH, QRes,
                                                          DlxState0),
     State1 = update_config(Conf, State0#?STATE{dlx = DlxState}),
     checkout(Meta, State0, State1, Effects0);
 apply(Meta, {machine_version, FromVersion, ToVersion}, V0State) ->
     State = convert(Meta, FromVersion, ToVersion, V0State),
     {State, ok, [{aux, {dlx, setup}}]};
-apply(#{index := _IncomingRaftIdx} = Meta, {dlx, _} = Cmd,
+apply(Meta, {dlx, _} = Cmd,
       #?STATE{cfg = #cfg{dead_letter_handler = DLH},
                dlx = DlxState0} = State0) ->
     {DlxState, Effects0} = rabbit_fifo_dlx:apply(Meta, Cmd, DLH, DlxState0),

deps/rabbit/src/rabbit_fifo.hrl

@@ -222,7 +222,7 @@
                     max_bytes => non_neg_integer(),
                     overflow_strategy => drop_head | reject_publish,
                     single_active_consumer_on => boolean(),
-                    delivery_limit => non_neg_integer(),
+                    delivery_limit => non_neg_integer() | -1,
                     expires => non_neg_integer(),
                     msg_ttl => non_neg_integer(),
                     created => non_neg_integer()

deps/rabbit/src/rabbit_policies.erl

@@ -165,7 +165,7 @@ validate_policy0(<<"overflow">>, Value) ->
     {error, "~tp is not a valid overflow value", [Value]};
 
 validate_policy0(<<"delivery-limit">>, Value)
-  when is_integer(Value), Value >= 0 ->
+  when is_integer(Value) ->
     ok;
 validate_policy0(<<"delivery-limit">>, Value) ->
     {error, "~tp is not a valid delivery limit", [Value]};
@@ -208,14 +208,35 @@ validate_policy0(<<"stream-filter-size-bytes">>, Value)
 validate_policy0(<<"stream-filter-size-bytes">>, Value) ->
     {error, "~tp is not a valid filter size. Valid range is 16-255", [Value]}.
 
-merge_policy_value(<<"message-ttl">>, Val, OpVal)      -> min(Val, OpVal);
-merge_policy_value(<<"max-length">>, Val, OpVal)       -> min(Val, OpVal);
-merge_policy_value(<<"max-length-bytes">>, Val, OpVal) -> min(Val, OpVal);
-merge_policy_value(<<"max-in-memory-length">>, Val, OpVal) -> min(Val, OpVal);
-merge_policy_value(<<"max-in-memory-bytes">>, Val, OpVal) -> min(Val, OpVal);
-merge_policy_value(<<"expires">>, Val, OpVal)          -> min(Val, OpVal);
-merge_policy_value(<<"delivery-limit">>, Val, OpVal)   -> min(Val, OpVal);
-merge_policy_value(<<"queue-version">>, _Val, OpVal)   -> OpVal;
-merge_policy_value(<<"overflow">>, _Val, OpVal)   -> OpVal;
-%% use operator policy value for booleans
-merge_policy_value(_Key, Val, OpVal) when is_boolean(Val) andalso is_boolean(OpVal) -> OpVal.
+merge_policy_value(<<"message-ttl">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"max-length">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"max-length-bytes">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"max-in-memory-length">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"max-in-memory-bytes">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"expires">>, Val, OpVal) ->
+    min(Val, OpVal);
+merge_policy_value(<<"delivery-limit">>, Val, OpVal) ->
+    case (is_integer(Val) andalso Val < 0) orelse
+         (is_integer(OpVal) andalso OpVal < 0) of
+        true ->
+            %% one of the policies define an unlimited delivery-limit (negative value)
+            %% choose the more conservative value
+            max(Val, OpVal);
+        false ->
+            %% else choose the lower value
+            min(Val, OpVal)
+    end;
+merge_policy_value(<<"queue-version">>, _Val, OpVal) ->
+    OpVal;
+merge_policy_value(<<"overflow">>, _Val, OpVal) ->
+    OpVal;
+merge_policy_value(_Key, Val, OpVal)
+  when is_boolean(Val) andalso
+       is_boolean(OpVal) ->
+    %% use operator policy value for booleans
+    OpVal.

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -82,7 +82,8 @@
          file_handle_release_reservation/0]).
 
 -ifdef(TEST).
--export([filter_promotable/2]).
+-export([filter_promotable/2,
+         ra_machine_config/1]).
 -endif.
 
 -import(rabbit_queue_type_util, [args_policy_lookup/3,
@@ -322,7 +323,8 @@ ra_machine_config(Q) when ?is_amqqueue(Q) ->
     OverflowBin = args_policy_lookup(<<"overflow">>, fun policy_has_precedence/2, Q),
     Overflow = overflow(OverflowBin, drop_head, QName),
     MaxBytes = args_policy_lookup(<<"max-length-bytes">>, fun min/2, Q),
-    DeliveryLimit = case args_policy_lookup(<<"delivery-limit">>, fun min/2, Q) of
+    DeliveryLimit = case args_policy_lookup(<<"delivery-limit">>,
+                                            fun resolve_delivery_limit/2, Q) of
                         undefined ->
                             rabbit_log:info("~ts: delivery_limit not set, defaulting to ~b",
                                              [rabbit_misc:rs(QName), ?DEFAULT_DELIVERY_LIMIT]),
@@ -346,6 +348,12 @@ ra_machine_config(Q) when ?is_amqqueue(Q) ->
       msg_ttl => MsgTTL
      }.
 
+resolve_delivery_limit(PolVal, ArgVal)
+  when PolVal < 0 orelse ArgVal < 0 ->
+    max(PolVal, ArgVal);
+resolve_delivery_limit(PolVal, ArgVal) ->
+    min(PolVal, ArgVal).
+
 policy_has_precedence(Policy, _QueueArg) ->
     Policy.
 
@@ -1898,8 +1906,6 @@ make_mutable_config(Q) ->
     #{tick_timeout => TickTimeout,
       ra_event_formatter => Formatter}.
 
-
-
 get_nodes(Q) when ?is_amqqueue(Q) ->
     #{nodes := Nodes} = amqqueue:get_type_state(Q),
     Nodes.

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -152,6 +152,7 @@ all_tests() ->
      queue_length_limit_drop_head,
      queue_length_limit_reject_publish,
      subscribe_redelivery_limit,
+     subscribe_redelivery_limit_disable,
      subscribe_redelivery_limit_many,
      subscribe_redelivery_policy,
      subscribe_redelivery_limit_with_dead_letter,
@@ -2495,8 +2496,8 @@ subscribe_redelivery_count(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H0}}} ->
             ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     after 5000 ->
               exit(basic_deliver_timeout)
     end,
@@ -2508,8 +2509,8 @@ subscribe_redelivery_count(Config) ->
             ct:pal("H1 ~p", [H1]),
             ?assertMatch({DCHeader, _, 1}, rabbit_basic:header(DCHeader, H1)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag1,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     after 5000 ->
               flush(1),
               exit(basic_deliver_timeout_2)
@@ -2521,7 +2522,7 @@ subscribe_redelivery_count(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H2}}} ->
             ?assertMatch({DCHeader, _, 2}, rabbit_basic:header(DCHeader, H2)),
             amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag2,
-                                               multiple     = false}),
+                                               multiple = false}),
             ct:pal("wait_for_messages_ready", []),
             wait_for_messages_ready(Servers, RaName, 0),
             ct:pal("wait_for_messages_pending_ack", []),
@@ -2551,8 +2552,8 @@ subscribe_redelivery_limit(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H0}}} ->
             ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"1">>, <<"0">>, <<"1">>]]),
@@ -2562,8 +2563,8 @@ subscribe_redelivery_limit(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H1}}} ->
             ?assertMatch({DCHeader, _, 1}, rabbit_basic:header(DCHeader, H1)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag1,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"0">>, <<"0">>, <<"0">>]]),
@@ -2574,6 +2575,51 @@ subscribe_redelivery_limit(Config) ->
             ok
     end.
 
+subscribe_redelivery_limit_disable(Config) ->
+    [Server | _] = 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">>},
+                                  {<<"x-delivery-limit">>, long, -1}])),
+    publish(Ch, QQ),
+    wait_for_messages(Config, [[QQ, <<"1">>, <<"1">>, <<"0">>]]),
+    subscribe(Ch, QQ, false),
+
+    DCHeader = <<"x-delivery-count">>,
+    receive
+        {#'basic.deliver'{delivery_tag = DeliveryTag,
+                          redelivered  = false},
+         #amqp_msg{props = #'P_basic'{headers = H0}}} ->
+            ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
+            amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
+                                                multiple = false,
+                                                requeue = true})
+    end,
+
+    wait_for_messages(Config, [[QQ, <<"1">>, <<"0">>, <<"1">>]]),
+    %% set an operator policy, this should always win
+    ok = rabbit_ct_broker_helpers:set_operator_policy(
+           Config, 0, <<"delivery-limit">>, QQ, <<"queues">>,
+           [{<<"delivery-limit">>, 0}]),
+
+    receive
+        {#'basic.deliver'{delivery_tag = DeliveryTag2,
+                          redelivered = true},
+         #amqp_msg{props = #'P_basic'{}}} ->
+            % ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
+            amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag2,
+                                                multiple = false,
+                                                requeue = true})
+    after 5000 ->
+              flush(1),
+              ct:fail("message did not arrive as expected")
+    end,
+    wait_for_messages(Config, [[QQ, <<"0">>, <<"0">>, <<"0">>]]),
+    ok = rabbit_ct_broker_helpers:clear_operator_policy(Config, 0, <<"delivery-limit">>),
+    ok.
+
 %% Test that consumer credit is increased correctly.
 subscribe_redelivery_limit_many(Config) ->
     [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -2637,8 +2683,8 @@ subscribe_redelivery_policy(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H0}}} ->
             ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"1">>, <<"0">>, <<"1">>]]),
@@ -2648,8 +2694,8 @@ subscribe_redelivery_policy(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H1}}} ->
             ?assertMatch({DCHeader, _, 1}, rabbit_basic:header(DCHeader, H1)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag1,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"0">>, <<"0">>, <<"0">>]]),
@@ -2687,8 +2733,8 @@ subscribe_redelivery_limit_with_dead_letter(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H0}}} ->
             ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"1">>, <<"0">>, <<"1">>]]),
@@ -2698,8 +2744,8 @@ subscribe_redelivery_limit_with_dead_letter(Config) ->
          #amqp_msg{props = #'P_basic'{headers = H1}}} ->
             ?assertMatch({DCHeader, _, 1}, rabbit_basic:header(DCHeader, H1)),
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag1,
-                                                multiple     = false,
-                                                requeue      = true})
+                                                multiple = false,
+                                                requeue = true})
     end,
 
     wait_for_messages(Config, [[QQ, <<"0">>, <<"0">>, <<"0">>]]),
@@ -2726,8 +2772,8 @@ consume_redelivery_count(Config) ->
                                            no_ack = false}),
     ?assertMatch(undefined, rabbit_basic:header(DCHeader, H0)),
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                        multiple     = false,
-                                        requeue      = true}),
+                                        multiple = false,
+                                        requeue = true}),
     %% wait for requeuing
     {#'basic.get_ok'{delivery_tag = DeliveryTag1,
                      redelivered = true},
@@ -2736,8 +2782,8 @@ consume_redelivery_count(Config) ->
 
     ?assertMatch({DCHeader, _, 1}, rabbit_basic:header(DCHeader, H1)),
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag1,
-                                        multiple     = false,
-                                        requeue      = true}),
+                                        multiple = false,
+                                        requeue = true}),
 
     {#'basic.get_ok'{delivery_tag = DeliveryTag2,
                      redelivered = true},
@@ -2746,8 +2792,8 @@ consume_redelivery_count(Config) ->
                                            no_ack = false}),
     ?assertMatch({DCHeader, _, 2}, rabbit_basic:header(DCHeader, H2)),
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag2,
-                                        multiple     = false,
-                                        requeue      = true}),
+                                        multiple = false,
+                                        requeue = true}),
     ok.
 
 message_bytes_metrics(Config) ->
@@ -2784,8 +2830,8 @@ message_bytes_metrics(Config) ->
         {#'basic.deliver'{delivery_tag = DeliveryTag,
                           redelivered  = false}, _} ->
             amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DeliveryTag,
-                                                multiple     = false,
-                                                requeue      = false}),
+                                                multiple = false,
+                                                requeue = false}),
             wait_for_messages_ready(Servers, RaName, 0),
             wait_for_messages_pending_ack(Servers, RaName, 0),
             rabbit_ct_helpers:await_condition(

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -2137,7 +2137,6 @@ reject_publish_applied_after_limit_test(Config) ->
              queue_resource => QName,
              max_length => 2,
              overflow_strategy => reject_publish,
-             max_in_memory_length => 0,
              dead_letter_handler => undefined
             },
     {State5, ok, Efx1} = apply(meta(Config, 5), rabbit_fifo:make_update_config(Conf), State4),
@@ -2148,6 +2147,31 @@ reject_publish_applied_after_limit_test(Config) ->
         apply(meta(Config, 1), make_register_enqueuer(Pid2), State5),
     ok.
 
+update_config_delivery_limit_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)),
+
+    %% A delivery limit of -1 (or any negative value) turns the delivery_limit
+    %% off
+    Conf = #{name => ?FUNCTION_NAME,
+             queue_resource => QName,
+             delivery_limit => -1,
+             dead_letter_handler => undefined
+            },
+    {State1, ok, _} = apply(meta(Config, ?LINE),
+                            rabbit_fifo:make_update_config(Conf), State0),
+
+    ?assertMatch(#{config := #{delivery_limit := undefined}},
+                 rabbit_fifo:overview(State1)),
+
+    ok.
+
 purge_nodes_test(Config) ->
     Node = purged@node,
     ThisNode = node(),