Use list instead of array for x-cc message annotation

ansd · ansd · commit ca1367571b3f · 2024-10-23T15:23:58.000+02:00
diff --git a/deps/rabbit/src/mc_amqpl.erl b/deps/rabbit/src/mc_amqpl.erl
@@ -708,23 +708,10 @@ from_091(binary, V) -> {binary, V};
 from_091(timestamp, V) -> {timestamp, V * 1000};
 from_091(byte, V) -> {byte, V};
 from_091(void, _V) -> null;
+from_091(array, L) ->
+    {list, [from_091(T, V) || {T, V} <- L]};
 from_091(table, L) ->
-    {map, [{wrap(symbol, K), from_091(T, V)} || {K, T, V} <- L]};
-from_091(array, []) ->
-    {list, []};
-from_091(array, L0 = [{T0, _} | _]) ->
-    {L = [{T1, _} | _], {Monomorphic, _}} =
-    lists:mapfoldl(fun({T, V}, {Mono0, PrevType}) ->
-                           Mono = case Mono0 of
-                                      false -> false;
-                                      true -> T =:= PrevType
-                                  end,
-                           {from_091(T, V), {Mono, T}}
-                   end, {true, T0}, L0),
-    case Monomorphic of
-        true -> {array, T1, L};
-        false -> {list, L}
-    end.
+    {map, [{wrap(symbol, K), from_091(T, V)} || {K, T, V} <- L]}.
 
 map_add(_T, _Key, _Type, undefined, Acc) ->
     Acc;
@@ -887,24 +874,3 @@ amqp10_section_header(Header, Headers) ->
 
 amqp_encoded_binary(Section) ->
     iolist_to_binary(amqp10_framing:encode_bin(Section)).
-
--ifdef(TEST).
--include_lib("eunit/include/eunit.hrl").
-
-from_091_array_test() ->
-    {list, []} = from_091(array, []),
-    {array, utf8, [{utf8, <<"e1">>}]} = from_091(array, [{longstr, <<"e1">>}]),
-    {array, utf8, [{utf8, <<"e1">>},
-                   {utf8, <<"e2">>}]} = from_091(array, [{longstr, <<"e1">>},
-                                                         {longstr, <<"e2">>}]),
-    {list, [{utf8, <<"e1">>},
-            {binary, <<"e2">>}]} = from_091(array, [{longstr, <<"e1">>},
-                                                    {binary, <<"e2">>}]),
-    {list, [{utf8, <<"e1">>},
-            {binary, <<"e2">>},
-            {utf8, <<"e3">>},
-            {utf8, <<"e4">>}]} = from_091(array, [{longstr, <<"e1">>},
-                                                  {binary, <<"e2">>},
-                                                  {longstr, <<"e3">>},
-                                                  {longstr, <<"e4">>}]).
--endif.
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -2495,21 +2495,27 @@ lookup_routing_key(X = #exchange{name = #resource{name = XNameBin}},
             RKeys = [RKey],
             Mc = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
             {ok, X, RKeys, Mc, PermCache};
-        {array, utf8, CCs0} ->
-            CCs = lists:map(fun({utf8, CC}) -> CC end, CCs0),
-            RKeys = [RKey | CCs],
-            Mc = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
-            {ok, X, RKeys, Mc, PermCache};
+        {list, CCs0} = L ->
+            try lists:map(fun({utf8, CC}) -> CC end, CCs0) of
+                CCs ->
+                    RKeys = [RKey | CCs],
+                    Mc = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
+                    {ok, X, RKeys, Mc, PermCache}
+            catch error:function_clause ->
+                      {error, {anonymous_terminus, AnonTerm}, bad_x_cc(L)}
+            end;
         BadValue ->
-            Desc = unicode:characters_to_binary(
-                     lists:flatten(
-                       io_lib:format(
-                         "bad value for 'x-cc' message-annotation: ~tp", [BadValue]))),
-            {error, {anonymous_terminus, AnonTerm},
-             #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
-                           description = {utf8, Desc}}}
+            {error, {anonymous_terminus, AnonTerm}, bad_x_cc(BadValue)}
     end.
 
+bad_x_cc(Value) ->
+    Desc = unicode:characters_to_binary(
+             lists:flatten(
+               io_lib:format(
+                 "bad value for 'x-cc' message-annotation: ~tp", [Value]))),
+    #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
+                  description = {utf8, Desc}}.
+
 process_routing_confirm([], _SenderSettles = true, _, U) ->
     rabbit_global_counters:messages_unroutable_dropped(?PROTOCOL, 1),
     {U, []};
diff --git a/deps/rabbit/src/rabbit_stream_queue.erl b/deps/rabbit/src/rabbit_stream_queue.erl
@@ -1283,7 +1283,7 @@ entry_to_msg(Entry, Offset, #resource{kind = queue, name = QName}, Name, LocalPi
                        <<>>
                end,
     RKeys0 = case XHeaders of
-                 #{<<"x-cc">> := {array, utf8, CCs}} ->
+                 #{<<"x-cc">> := {list, CCs}} ->
                      [CC || {utf8, CC} <- CCs];
                  _ ->
                      []
diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl
@@ -4449,7 +4449,7 @@ trace(Q, QType, Config) ->
     ok = wait_for_credit(Sender),
     Msg0 = amqp10_msg:new(<<"tag 1">>, Payload, true),
     Msg = amqp10_msg:set_message_annotations(
-            #{<<"x-cc">> => {array, utf8, [{utf8, <<"my CC key">>}]}},
+            #{<<"x-cc">> => {list, [{utf8, <<"my CC key">>}]}},
             amqp10_msg:set_properties(#{correlation_id => CorrelationId}, Msg0)),
     ok = amqp10_client:send_msg(Sender, Msg),
     {ok, _} = amqp10_client:get_msg(Receiver),
@@ -5993,7 +5993,7 @@ x_cc_annotation_exchange(Config) ->
 
     Payload = <<"my message">>,
     ok = amqp10_client:send_msg(Sender, amqp10_msg:set_message_annotations(
-                                          #{<<"x-cc">> => {array, utf8, [{utf8, <<"key 2">>}]}},
+                                          #{<<"x-cc">> => {list, [{utf8, <<"key 2">>}]}},
                                           amqp10_msg:new(<<"tag">>, Payload))),
     ok = wait_for_accepted(<<"tag">>),
     ok = amqp10_client:detach_link(Sender),
@@ -6028,8 +6028,8 @@ x_cc_annotation_exchange_routing_key_empty(Config) ->
 
     Payload = <<"my message">>,
     ok = amqp10_client:send_msg(Sender, amqp10_msg:set_message_annotations(
-                                          #{<<"x-cc">> => {array, utf8, [{utf8, <<"key 1">>},
-                                                                         {utf8, <<"key 2">>}]}},
+                                          #{<<"x-cc">> => {list, [{utf8, <<"key 1">>},
+                                                                  {utf8, <<"key 2">>}]}},
                                           amqp10_msg:new(<<"tag">>, Payload))),
     ok = wait_for_accepted(<<"tag">>),
     ok = amqp10_client:detach_link(Sender),
@@ -6063,7 +6063,7 @@ x_cc_annotation_queue(Config) ->
 
     Payload = <<"my message">>,
     ok = amqp10_client:send_msg(Sender, amqp10_msg:set_message_annotations(
-                                          #{<<"x-cc">> => {array, utf8, [{utf8, QName2}]}},
+                                          #{<<"x-cc">> => {list, [{utf8, QName2}]}},
                                           amqp10_msg:new(<<"tag">>, Payload))),
     ok = wait_for_accepted(<<"tag">>),
     ok = amqp10_client:detach_link(Sender),
@@ -6097,8 +6097,8 @@ x_cc_annotation_null(Config) ->
     {ok, Receiver2} = amqp10_client:attach_receiver_link(Session, <<"receiver 2">>, QAddress2, settled),
 
     Msg1 = amqp10_msg:set_message_annotations(
-             #{<<"x-cc">> => {array, utf8, [{utf8, <<"key-1">>},
-                                            {utf8, <<"key-3">>}]}},
+             #{<<"x-cc">> => {list, [{utf8, <<"key-1">>},
+                                     {utf8, <<"key-3">>}]}},
              amqp10_msg:set_properties(
                #{to => rabbitmq_amqp_address:exchange(<<"amq.direct">>, <<"🗝️-2"/utf8>>)},
                amqp10_msg:new(<<"t1">>, <<"m1">>))),
@@ -6110,8 +6110,8 @@ x_cc_annotation_null(Config) ->
     ?assertEqual([<<"m1">>], amqp10_msg:body(R2M1)),
 
     Msg2 = amqp10_msg:set_message_annotations(
-             #{<<"x-cc">> => {array, utf8, [{utf8, <<"🗝️-2"/utf8>>},
-                                            {utf8, <<"key-1">>}]}},
+             #{<<"x-cc">> => {list, [{utf8, <<"🗝️-2"/utf8>>},
+                                     {utf8, <<"key-1">>}]}},
              amqp10_msg:set_properties(
                #{to => rabbitmq_amqp_address:exchange(<<"amq.direct">>)},
                amqp10_msg:new(<<"t2">>, <<"m2">>))),
@@ -6123,7 +6123,7 @@ x_cc_annotation_null(Config) ->
     ?assertEqual([<<"m2">>], amqp10_msg:body(R2M2)),
 
     Msg3 = amqp10_msg:set_message_annotations(
-             #{<<"x-cc">> => {array, utf8, [{utf8, QName1}]}},
+             #{<<"x-cc">> => {list, [{utf8, QName1}]}},
              amqp10_msg:set_properties(
                #{to => rabbitmq_amqp_address:queue(QName2)},
                amqp10_msg:new(<<"t3">>, <<"m3">>))),
@@ -6135,8 +6135,8 @@ x_cc_annotation_null(Config) ->
     ?assertEqual([<<"m3">>], amqp10_msg:body(R2M3)),
 
     Msg4 = amqp10_msg:set_message_annotations(
-             %% We send a symbol array instead of utf8 array.
-             #{<<"x-cc">> => {array, symbol, [{symbol, QName1}]}},
+             %% We send a symbol instead of utf8..
+             #{<<"x-cc">> => {list, [{symbol, QName1}]}},
              amqp10_msg:set_properties(
                #{to => rabbitmq_amqp_address:queue(QName2)},
                amqp10_msg:new(<<"t4">>, <<"m4">>))),
@@ -6169,23 +6169,41 @@ bad_x_cc_annotation_exchange(Config) ->
     {ok, Session} = amqp10_client:begin_session(Connection),
 
     Address = rabbitmq_amqp_address:exchange(<<"amq.direct">>, <<"key-1">>),
-    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address),
-    ok = wait_for_credit(Sender),
-
+    {ok, Sender1} = amqp10_client:attach_sender_link(Session, <<"sender 1">>, Address),
+    ok = wait_for_credit(Sender1),
     ok = amqp10_client:send_msg(
-           Sender,
+           Sender1,
            amqp10_msg:set_message_annotations(
-             %% We send a list instead of an array.
-             #{<<"x-cc">> => {list, [{utf8, <<"🗝️-2"/utf8>>}]}},
-             amqp10_msg:new(<<"tag">>, <<"msg">>))),
-    ok = wait_for_settlement(<<"tag">>, released),
-    receive {amqp10_event, {link, Sender, {detached, Error}}} ->
+             %% We send an array instead of a list.
+             #{<<"x-cc">> => {array, utf8, [{utf8, <<"🗝️-2"/utf8>>}]}},
+             amqp10_msg:new(<<"t1">>, <<"m1">>))),
+    ok = wait_for_settlement(<<"t1">>, released),
+    receive {amqp10_event, {link, Sender1, {detached, Error1}}} ->
                 ?assertMatch(
                    #'v1_0.error'{
                       condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
                       description = {utf8, <<"bad value for 'x-cc' message-annotation: "
-                                             "{list,[{utf8,<<\"🗝️-2"/utf8, _Rest/binary>>}},
-                   Error)
+                                             "{array,utf8,[{utf8,<<\"🗝️-2"/utf8, _Rest/binary>>}},
+                   Error1)
+    after 5000 -> ct:fail({missing_event, ?LINE})
+    end,
+
+    {ok, Sender2} = amqp10_client:attach_sender_link(Session, <<"sender 2">>, Address),
+    ok = wait_for_credit(Sender2),
+    ok = amqp10_client:send_msg(
+           Sender2,
+           amqp10_msg:set_message_annotations(
+             %% We include a non-utf8 type in the list.
+             #{<<"x-cc">> => {list, [{symbol, <<"key-3">>}]}},
+             amqp10_msg:new(<<"t2">>, <<"m2">>))),
+    ok = wait_for_settlement(<<"t2">>, released),
+    receive {amqp10_event, {link, Sender2, {detached, Error2}}} ->
+                ?assertEqual(
+                   #'v1_0.error'{
+                      condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
+                      description = {utf8, <<"bad value for 'x-cc' message-annotation: "
+                                             "{list,[{symbol,<<\"key-3\">>}]}">>}},
+                   Error2)
     after 5000 -> ct:fail({missing_event, ?LINE})
     end,
 
diff --git a/deps/rabbit/test/dead_lettering_SUITE.erl b/deps/rabbit/test/dead_lettering_SUITE.erl
@@ -174,22 +174,14 @@ end_per_group(Group, Config) ->
             Config
     end.
 
-init_per_testcase(T = dead_letter_reject_expire_expire, Config) ->
-    %% With feature flag message_containers_deaths_v2 disabled, this test is known to fail due to
-    %% https://github.com/rabbitmq/rabbitmq-server/issues/11159
+init_per_testcase(T, Config)
+  when T =:= dead_letter_reject_expire_expire orelse
+       T =:= stream ->
+    %% With feature flag message_containers_deaths_v2 disabled, test case:
+    %% * dead_letter_reject_expire_expire is known to fail due to https://github.com/rabbitmq/rabbitmq-server/issues/11159
+    %% * stream is known to fail due to https://github.com/rabbitmq/rabbitmq-server/issues/11173
     ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, message_containers_deaths_v2),
     init_per_testcase0(T, Config);
-init_per_testcase(T = stream, Config) ->
-    %% With feature flag message_containers_deaths_v2 disabled, this test is known to fail due to
-    %% https://github.com/rabbitmq/rabbitmq-server/issues/11173
-    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, message_containers_deaths_v2),
-    case rabbit_ct_helpers:is_mixed_versions() of
-        true ->
-            {skip, "TODO unskip this test when the lower version got bumped to v4.0.3 "
-             "which includes https://github.com/rabbitmq/rabbitmq-server/pull/12571"};
-        false ->
-            init_per_testcase0(T, Config)
-    end;
 init_per_testcase(Testcase, Config) ->
     init_per_testcase0(Testcase, Config).
 
diff --git a/deps/rabbit/test/mc_unit_SUITE.erl b/deps/rabbit/test/mc_unit_SUITE.erl
@@ -189,11 +189,14 @@ amqpl_table_x_header_array_of_tbls(_Config) ->
     Content = #content{properties = Props,
                        payload_fragments_rev = Payload},
     Msg = mc:init(mc_amqpl, Content, annotations()),
-    ?assertMatch({array, map,
-                  [{map, [{{symbol, <<"type">>}, {utf8, <<"apple">>}},
-                          {{symbol, <<"count">>}, {long, 99}}]},
-                   {map, [{{symbol, <<"type">>}, {utf8, <<"orange">>}},
-                          {{symbol, <<"count">>}, {long, 45}}]}]},
+    ?assertMatch({list,
+                  [{map,
+                    [{{symbol, <<"type">>}, {utf8, <<"apple">>}},
+                     {{symbol, <<"count">>}, {long, 99}}]},
+                   {map,
+                    [{{symbol, <<"type">>}, {utf8, <<"orange">>}},
+                     {{symbol, <<"count">>}, {long, 45}}]}
+                  ]},
                  mc:x_header(<<"x-fruit">>, Msg)).
 
 amqpl_death_v1_records(_Config) ->
@@ -447,8 +450,8 @@ amqpl_cc_amqp_bin_amqpl(_Config) ->
     Msg10 = mc:init(mc_amqp, Sections, #{}),
     ?assertMatch(#{<<"x-exchange">> := {utf8, <<"exch">>},
                    <<"x-routing-key">> := {utf8, <<"apple">>},
-                   <<"x-cc">> := {array, utf8, [{utf8, <<"q1">>},
-                                                {utf8, <<"q2">>}]}},
+                   <<"x-cc">> := {list, [{utf8, <<"q1">>},
+                                         {utf8, <<"q2">>}]}},
                  mc:x_headers(Msg10)),
 
     %% Here, we simulate what rabbit_stream_queue does:
diff --git a/deps/rabbit/test/topic_permission_SUITE.erl b/deps/rabbit/test/topic_permission_SUITE.erl
@@ -82,7 +82,7 @@ amqp_x_cc_annotation(Config) ->
     %% We have permissions to send to both topics.
     %% Therefore, m1 should be sent to both queues.
     ok = amqp10_client:send_msg(Sender1, amqp10_msg:set_message_annotations(
-                                           #{<<"x-cc">> => {array, utf8, [{utf8, <<"a.2">>}]}},
+                                           #{<<"x-cc">> => {list, [{utf8, <<"a.2">>}]}},
                                            amqp10_msg:new(<<"t1">>, <<"m1">>, true))),
     {ok, Msg1} = amqp10_client:get_msg(Receiver1),
     {ok, Msg2} = amqp10_client:get_msg(Receiver2),
@@ -99,7 +99,7 @@ amqp_x_cc_annotation(Config) ->
                       rabbitmq_amqp_address:exchange(<<"amq.topic">>, <<"x.1">>)),
     ok = amqp_utils:wait_for_credit(Sender2),
     ok = amqp10_client:send_msg(Sender2, amqp10_msg:set_message_annotations(
-                                           #{<<"x-cc">> => {array, utf8, [{utf8, <<"a.2">>}]}},
+                                           #{<<"x-cc">> => {list, [{utf8, <<"a.2">>}]}},
                                            amqp10_msg:new(<<"t2">>, <<"m2">>, true))),
     receive
         {amqp10_event,
@@ -122,7 +122,7 @@ amqp_x_cc_annotation(Config) ->
                       rabbitmq_amqp_address:exchange(<<"amq.topic">>, <<"a.1">>)),
     ok = amqp_utils:wait_for_credit(Sender3),
     ok = amqp10_client:send_msg(Sender3, amqp10_msg:set_message_annotations(
-                                           #{<<"x-cc">> => {array, utf8, [{utf8, <<"x.2">>}]}},
+                                           #{<<"x-cc">> => {list, [{utf8, <<"x.2">>}]}},
                                            amqp10_msg:new(<<"t3">>, <<"m3">>, true))),
     receive
         {amqp10_event,
diff --git a/release-notes/4.1.0.md b/release-notes/4.1.0.md
@@ -22,7 +22,7 @@ Each metric is labelled by protocol (AMQP 1.0, AMQP 0.9.1, MQTT 5.0, MQTT 3.1.1,
 
 ### Support for Multiple Routing Keys in AMQP 1.0 via `x-cc` Message Annotation
 [PR #12559](https://github.com/rabbitmq/rabbitmq-server/pull/12559) enables AMQP 1.0 publishers to set multiple routing keys by using the `x-cc` message annotation.
-This annotation allows publishers to specify an [array](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-types-v1.0-os.html#type-array) of routing keys ([strings](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-types-v1.0-os.html#type-string)) for more flexible message distribution, similar to the [CC](https://www.rabbitmq.com/docs/sender-selected) header in AMQP 0.9.1.
+This annotation allows publishers to specify a [list](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-types-v1.0-os.html#type-list) of routing keys ([strings](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-types-v1.0-os.html#type-string)) for more flexible message distribution, similar to the [CC](https://www.rabbitmq.com/docs/sender-selected) header in AMQP 0.9.1.
 
 ## Potential incompatibilities
 
