Merge pull request #12523 from rabbitmq/annotation-key-error

Provide clear error message for reserved annotation keys

Author: ansd

.github/workflows/test-make-target.yaml

@@ -45,7 +45,7 @@ jobs:
       uses: actions/setup-dotnet@v4
       if: inputs.plugin == 'rabbit'
       with:
-        dotnet-version: '3.1.x'
+        dotnet-version: '8.0'
 
     - name: SETUP SLAPD (rabbitmq_auth_backend_ldap)
       if: inputs.plugin == 'rabbitmq_auth_backend_ldap'

.github/workflows/test-plugin.yaml

@@ -75,7 +75,7 @@ jobs:
     - uses: actions/setup-dotnet@v4
       if: inputs.plugin == 'rabbit'
       with:
-        dotnet-version: '3.1.x'
+        dotnet-version: '8.0'
     - name: deps/amqp10_client SETUP
       if: inputs.plugin == 'amqp10_client'
       run: |

deps/amqp10_common/src/amqp10_framing.erl

@@ -153,10 +153,12 @@ decode_map(Fields) ->
 %% or of type ulong. All ulong keys, and all symbolic keys except those beginning
 %% with "x-" are reserved." [3.2.10]
 %% Since we already parse annotations here and neither the client nor server uses
-%% reserved keys, we perform strict validation and crash if any reserved keys are used.
+%% reserved keys, we perform strict validation and throw if any reserved keys are used.
 decode_annotations(Fields) ->
     lists:map(fun({{symbol, <<"x-", _/binary>>} = K, V}) ->
-                      {K, decode(V)}
+                      {K, decode(V)};
+                 ({ReservedKey, _V}) ->
+                      throw({reserved_annotation_key, ReservedKey})
               end, Fields).
 
 -spec encode_described(list | map | binary | annotations | '*',

deps/rabbit/src/rabbit_amqp_session.erl

@@ -514,12 +514,16 @@ handle_cast({frame_body, FrameBody},
             noreply(State);
         {stop, _, _} = Stop ->
             Stop
-    catch exit:#'v1_0.error'{} = Error ->
-              log_error_and_close_session(Error, State0);
-          exit:normal ->
+    catch exit:normal ->
               {stop, normal, State0};
+          exit:#'v1_0.error'{} = Error ->
+              log_error_and_close_session(Error, State0);
           _:Reason:Stacktrace ->
-              {stop, {Reason, Stacktrace}, State0}
+              Description = unicode:characters_to_binary(
+                              lists:flatten(io_lib:format("~tp~n~tp", [Reason, Stacktrace]))),
+              Err = #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                                  description = {utf8, Description}},
+              log_error_and_close_session(Err, State0)
     end;
 handle_cast({queue_event, _, _} = QEvent, State0) ->
     try handle_queue_event(QEvent, State0) of

deps/rabbit/test/amqp_client_SUITE.erl

@@ -155,7 +155,8 @@ groups() ->
        tcp_back_pressure_rabbitmq_internal_flow_classic_queue,
        tcp_back_pressure_rabbitmq_internal_flow_quorum_queue,
        session_max_per_connection,
-       link_max_per_session
+       link_max_per_session,
+       reserved_annotation
       ]},
 
      {cluster_size_3, [shuffle],
@@ -5917,6 +5918,32 @@ link_max_per_session(Config) ->
     flush(test_succeeded),
     ok = rpc(Config, application, set_env, [App, Par, Default]).
 
+reserved_annotation(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session(Connection),
+    TargetAddr = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
+    {ok, Sender} = amqp10_client:attach_sender_link_sync(
+                     Session, <<"sender">>, TargetAddr, settled),
+    ok = wait_for_credit(Sender),
+
+    Msg = amqp10_msg:set_message_annotations(
+            #{<<"reserved-key">> => 1},
+            amqp10_msg:new(<<"tag">>, <<"payload">>, true)),
+    ok = amqp10_client:send_msg(Sender, Msg),
+    receive
+        {amqp10_event,
+         {session, Session,
+          {ended,
+           #'v1_0.error'{description = {utf8, Description}}}}} ->
+            ?assertMatch(
+               <<"{reserved_annotation_key,{symbol,<<\"reserved-key\">>}}", _/binary>>,
+               Description)
+    after 5000 -> flush(missing_ended),
+                  ct:fail({missing_event, ?LINE})
+    end,
+    ok = close_connection_sync(Connection).
+
 %% internal
 %%
 

deps/rabbit/test/amqp_system_SUITE_data/fsharp-tests/global.json

@@ -1,5 +1,5 @@
 {
   "sdk": {
-    "version": "3.1"
+    "version": "8.0"
   }
 }