Merge pull request #4479 from rabbitmq/mergify/bp/v3.9.x/pr-4476

Bugfix: check unsupported policies on queue.declare (backport #4463) (backport #4476)

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -362,6 +362,8 @@ policy_changed(Q1, Q2) ->
     %% mirroring-related has changed - the policy may have changed anyway.
     notify_policy_changed(Q2).
 
+is_policy_applicable(Q, Policy) when ?is_amqqueue(Q) ->
+    rabbit_queue_type:is_policy_applicable(Q, Policy);
 is_policy_applicable(QName, Policy) ->
     case lookup(QName) of
         {ok, Q} ->

deps/rabbit/src/rabbit_policy.erl

@@ -93,9 +93,14 @@ merge_operator_definitions(Policy, OpPolicy) ->
     lists:umerge(Keys, OpKeys)).
 
 set(Q0) when ?is_amqqueue(Q0) ->
-    Name = amqqueue:get_name(Q0),
-    Policy = match(Name),
-    OpPolicy = match_op(Name),
+    %% On queue.declare the queue record doesn't exist yet, so the later lookup in
+    %% `rabbit_amqqueue:is_policy_applicable` fails. Thus, let's send the whole record
+    %% through the match functions. `match_all` still needs to support resources only,
+    %% as that is used by some plugins. Maybe `match` too, difficult to figure out
+    %% what is public API and what is not, so let's support both `amqqueue` and `resource`
+    %% records as arguments.
+    Policy = match(Q0),
+    OpPolicy = match_op(Q0),
     Q1 = amqqueue:set_policy(Q0, Policy),
     Q2 = amqqueue:set_operator_policy(Q1, OpPolicy),
     Q2;
@@ -147,9 +152,15 @@ list_formatted_op(VHost, Ref, AggregatorPid) ->
     rabbit_control_misc:emitting_map(AggregatorPid, Ref,
                                      fun(P) -> P end, list_formatted_op(VHost)).
 
+match(Q) when ?is_amqqueue(Q) ->
+    #resource{virtual_host = VHost} = amqqueue:get_name(Q),
+    match(Q, list(VHost));
 match(Name = #resource{virtual_host = VHost}) ->
     match(Name, list(VHost)).
 
+match_op(Q) when ?is_amqqueue(Q) ->
+    #resource{virtual_host = VHost} = amqqueue:get_name(Q),
+    match(Q, list_op(VHost));
 match_op(Name = #resource{virtual_host = VHost}) ->
     match(Name, list_op(VHost)).
 
@@ -172,15 +183,21 @@ get(Name, EntityName = #resource{virtual_host = VHost}) ->
          match(EntityName, list(VHost)),
          match(EntityName, list_op(VHost))).
 
-match(Name, Policies) ->
-    case match_all(Name, Policies) of
+match(NameOrQueue, Policies) ->
+    case match_all(NameOrQueue, Policies) of
         []           -> undefined;
         [Policy | _] -> Policy
     end.
 
-match_all(Name, Policies) ->
-   lists:sort(fun priority_comparator/2, [P || P <- Policies, matches(Name, P)]).
+match_all(NameOrQueue, Policies) ->
+   lists:sort(fun priority_comparator/2, [P || P <- Policies, matches(NameOrQueue, P)]).
 
+matches(Q, Policy) when ?is_amqqueue(Q) ->
+    #resource{name = Name, kind = Kind, virtual_host = VHost} = amqqueue:get_name(Q),
+    matches_type(Kind, pget('apply-to', Policy)) andalso
+        is_applicable(Q, pget(definition, Policy)) andalso
+        match =:= re:run(Name, pget(pattern, Policy), [{capture, none}]) andalso
+        VHost =:= pget(vhost, Policy);
 matches(#resource{name = Name, kind = Kind, virtual_host = VHost} = Resource, Policy) ->
     matches_type(Kind, pget('apply-to', Policy)) andalso
         is_applicable(Resource, pget(definition, Policy)) andalso
@@ -527,6 +544,8 @@ matches_type(_,        _)               -> false.
 
 priority_comparator(A, B) -> pget(priority, A) >= pget(priority, B).
 
+is_applicable(Q, Policy) when ?is_amqqueue(Q) ->
+    rabbit_amqqueue:is_policy_applicable(Q, rabbit_data_coercion:to_list(Policy));
 is_applicable(#resource{kind = queue} = Resource, Policy) ->
     rabbit_amqqueue:is_policy_applicable(Resource, rabbit_data_coercion:to_list(Policy));
 is_applicable(_, _) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -133,6 +133,7 @@ all_tests() ->
      in_memory,
      consumer_metrics,
      invalid_policy,
+     pre_existing_invalid_policy,
      delete_if_empty,
      delete_if_unused,
      queue_ttl,
@@ -963,6 +964,22 @@ invalid_policy(Config) ->
     ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ha">>),
     ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ttl">>).
 
+pre_existing_invalid_policy(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),
+    ok = rabbit_ct_broker_helpers:set_policy(
+           Config, 0, <<"ha">>, <<"invalid_policy.*">>, <<"queues">>,
+           [{<<"ha-mode">>, <<"all">>}]),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    Info = rpc:call(Server, rabbit_quorum_queue, infos,
+                    [rabbit_misc:r(<<"/">>, queue, QQ)]),
+    ?assertEqual('', proplists:get_value(policy, Info)),
+    ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ha">>),
+    ok.
+
 dead_letter_to_quorum_queue(Config) ->
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),