AMQP10 shovel: make bare message inmutable

dcorbacho · dcorbacho · commit 0110742bdb54 · 2025-07-04T11:11:48.000+02:00
diff --git a/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl b/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl
@@ -155,9 +155,9 @@ init_source(State = #{source := #{current := #{link := Link},
 init_dest(#{name := Name,
             shovel_type := Type,
             dest := #{add_forward_headers := true} = Dst} = State) ->
-    Props = #{<<"shovelled-by">> => rabbit_nodes:cluster_name(),
-              <<"shovel-type">> => rabbit_data_coercion:to_binary(Type),
-              <<"shovel-name">> => rabbit_data_coercion:to_binary(Name)},
+    Props = #{<<"x-opt-shovelled-by">> => rabbit_nodes:cluster_name(),
+              <<"x-opt-shovel-type">> => rabbit_data_coercion:to_binary(Type),
+              <<"x-opt-shovel-name">> => rabbit_data_coercion:to_binary(Name)},
     State#{dest => Dst#{cached_forward_headers => Props}};
 init_dest(State) ->
     State.
@@ -337,7 +337,7 @@ forward(Tag, Msg0,
     Msg1 = mc:protocol_state(mc:convert(mc_amqp, Msg0)),
     Records = lists:flatten([amqp10_framing:decode_bin(iolist_to_binary(S)) || S <- Msg1]),
     Msg2 = amqp10_msg:new(OutTag, Records, AckMode =/= on_confirm),
-    Msg = update_amqp10_message(Msg2, mc:exchange(Msg0), mc:routing_keys(Msg0), State),
+    Msg = add_timestamp_header(State, add_forward_headers(State, Msg2)),
     case send_msg(Link, Msg) of
         ok ->
             rabbit_shovel_behaviour:decr_remaining_unacked(
@@ -366,23 +366,13 @@ send_msg(Link, Msg) ->
             end
     end.
 
-update_amqp10_message(Msg0, Exchange, RK, #{dest := #{properties := Props,
-                                                      application_properties := AppProps0,
-                                                      message_annotations := MsgAnns}} = State) ->
-    Msg1 = amqp10_msg:set_properties(Props, Msg0),
-    Msg2 = amqp10_msg:set_message_annotations(MsgAnns, Msg1),
-    AppProps = AppProps0#{<<"exchange">> => Exchange,
-                          <<"routing_key">> => RK},
-    Msg = amqp10_msg:set_application_properties(AppProps, Msg2),
-    add_timestamp_header(State, add_forward_headers(State, Msg)).
-
 add_timestamp_header(#{dest := #{add_timestamp_header := true}}, Msg) ->
-    P =#{creation_time => os:system_time(milli_seconds)},
-    amqp10_msg:set_properties(P, Msg);
+    Anns = #{<<"x-opt-shovelled-timestamp">> => os:system_time(milli_seconds)},
+    amqp10_msg:set_message_annotations(Anns, Msg);
 add_timestamp_header(_, Msg) -> Msg.
 
-add_forward_headers(#{dest := #{cached_forward_headers := Props}}, Msg) ->
-    amqp10_msg:set_application_properties(Props, Msg);
+add_forward_headers(#{dest := #{cached_forward_headers := Anns}}, Msg) ->
+    amqp10_msg:set_message_annotations(Anns, Msg);
 add_forward_headers(_, Msg) -> Msg.
 
 gen_unique_name(Pre0, Post0) ->
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl
@@ -165,9 +165,16 @@ amqp10_dest_validation(_Def, User) ->
      {<<"dest-address">>, fun rabbit_parameter_validation:binary/2, mandatory},
      {<<"dest-add-forward-headers">>, fun rabbit_parameter_validation:boolean/2, optional},
      {<<"dest-add-timestamp-header">>, fun rabbit_parameter_validation:boolean/2, optional},
+     %% The bare message should be inmutable in the AMQP network.
+     %% Before RabbitMQ 4.2, we allowed to set application properties, message
+     %% annotations and any property. This is wrong.
+     %% From 4.2, the only message modification allowed is the optional
+     %% addition of forward headers and shovelled timestamp inside message
+     %% annotations.
+     %% To avoid breaking existing deployments, the following configuration
+     %% keys are still accepted but will be ignored.
      {<<"dest-application-properties">>, fun validate_amqp10_map/2, optional},
      {<<"dest-message-annotations">>, fun validate_amqp10_map/2, optional},
-     % TODO: restrict to allowed fields
      {<<"dest-properties">>, fun validate_amqp10_map/2, optional}
     ].
 
diff --git a/deps/rabbitmq_shovel/test/amqp10_SUITE.erl b/deps/rabbitmq_shovel/test/amqp10_SUITE.erl
@@ -8,6 +8,7 @@
 -module(amqp10_SUITE).
 
 -include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 
 -compile(export_all).
@@ -116,27 +117,36 @@ amqp10_destination(Config, AckMode) ->
                                       }},
     publish(Chan, Msg, ?EXCHANGE, ?TO_SHOVEL),
 
+    [NodeA] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Node = atom_to_binary(NodeA),
+
     receive
         {amqp10_msg, Receiver, InMsg} ->
             ct:pal("GOT ~p", [InMsg]),
             [<<42>>] = amqp10_msg:body(InMsg),
-            #{content_type := ?UNSHOVELLED,
-              content_encoding := ?UNSHOVELLED,
-              correlation_id := ?UNSHOVELLED,
-              user_id := <<"guest">>,
-              message_id := ?UNSHOVELLED,
-              reply_to := ?UNSHOVELLED
-              %% timestamp gets overwritten
-              % creation_time := Timestamp
-             } = amqp10_msg:properties(InMsg),
-            #{<<"routing_key">> := ?TO_SHOVEL,
-              <<"exchange">> := ?EXCHANGE,
-              <<"header1">> := 1,
-              <<"header2">> := <<"h2">>
-             } = amqp10_msg:application_properties(InMsg),
-            #{<<"x-basic-type">> := ?UNSHOVELLED
-             } = amqp10_msg:message_annotations(InMsg),
-            #{durable := true} = amqp10_msg:headers(InMsg),
+            Ts = Timestamp * 1000,
+            ?assertMatch(
+               #{content_type := ?UNSHOVELLED,
+                 content_encoding := ?UNSHOVELLED,
+                 correlation_id := ?UNSHOVELLED,
+                 user_id := <<"guest">>,
+                 message_id := ?UNSHOVELLED,
+                 reply_to := ?UNSHOVELLED,
+                 %% Message timestamp is no longer overwritten
+                 creation_time := Ts},
+               amqp10_msg:properties(InMsg)),
+            ?assertMatch(
+               #{<<"header1">> := 1,
+                 <<"header2">> := <<"h2">>},
+               amqp10_msg:application_properties(InMsg)),
+            ?assertMatch(
+               #{<<"x-basic-type">> := ?UNSHOVELLED,
+                 <<"x-opt-shovel-type">> := <<"static">>,
+                 <<"x-opt-shovel-name">> := <<"test_shovel">>,
+                 <<"x-opt-shovelled-by">> := Node,
+                 <<"x-opt-shovelled-timestamp">> := _},
+               amqp10_msg:message_annotations(InMsg)),
+            ?assertMatch(#{durable := true}, amqp10_msg:headers(InMsg)),
             ok
     after ?TIMEOUT ->
               throw(timeout_waiting_for_deliver1)
diff --git a/deps/rabbitmq_shovel/test/amqp10_dynamic_SUITE.erl b/deps/rabbitmq_shovel/test/amqp10_dynamic_SUITE.erl
@@ -154,18 +154,20 @@ test_amqp10_destination(Config, Src, Dest, Sess, Protocol, ProtocolSrc) ->
                                             <<"message-ann-value">>}]
                                   end}]),
     Msg = publish_expect(Sess, Src, Dest, <<"tag1">>, <<"hello">>),
-    ct:pal("GOT ~p", [Msg]),
     AppProps = amqp10_msg:application_properties(Msg),
-
-    ?assertMatch((#{user_id := <<"guest">>, creation_time := _}),
-                 (amqp10_msg:properties(Msg))),
-    ?assertMatch((#{<<"shovel-name">> := <<"test">>,
-                    <<"shovel-type">> := <<"dynamic">>, <<"shovelled-by">> := _,
-                    <<"app-prop-key">> := <<"app-prop-value">>}),
-                 (AppProps)),
+    Anns = amqp10_msg:message_annotations(Msg),
+    %% We no longer add/override properties, application properties or
+    %% message annotations. Just the forward headers and timestamp as
+    %% message annotations. The AMQP 1.0 message is inmutable
+    ?assertNot(maps:is_key(user_id, amqp10_msg:properties(Msg))),
+    ?assertNot(maps:is_key(<<"app-prop-key">>, AppProps)),
     ?assertEqual(undefined, maps:get(<<"delivery_mode">>, AppProps, undefined)),
-    ?assertMatch((#{<<"x-message-ann-key">> := <<"message-ann-value">>}),
-                 (amqp10_msg:message_annotations(Msg))).
+    ?assertNot(maps:is_key(<<"x-message-ann-key">>, Anns)),
+    ?assertMatch(#{<<"x-opt-shovel-name">> := <<"test">>,
+                   <<"x-opt-shovel-type">> := <<"dynamic">>,
+                   <<"x-opt-shovelled-by">> := _,
+                   <<"x-opt-shovelled-timestamp">> := _
+                  }, Anns).
 
 simple_amqp10_src(Config) ->
     MapConfig = ?config(map_config, Config),
