Fix direct reply to crash when tracing is enabled

This commit fixes https://github.com/rabbitmq/rabbitmq-server/discussions/11662

Prior to this commit the following crash occurred when an RPC reply message entered
RabbitMQ and tracing was enabled:
```
** Reason for termination ==
** {function_clause,
       [{rabbit_trace,'-tap_in/6-fun-0-',
            [{virtual_reply_queue,
                 <<"amq.rabbitmq.reply-to.g1h2AA5yZXBseUAyNzc5NjQyMAAAC1oAAAAAZo4bIw==.+Uvn1EmAp0ZA+oQx2yoQFA==">>}],
            [{file,"rabbit_trace.erl"},{line,62}]},
        {lists,map,2,[{file,"lists.erl"},{line,1559}]},
        {rabbit_trace,tap_in,6,[{file,"rabbit_trace.erl"},{line,62}]},
        {rabbit_channel,handle_method,3,
            [{file,"rabbit_channel.erl"},{line,1284}]},
        {rabbit_channel,handle_cast,2,
            [{file,"rabbit_channel.erl"},{line,659}]},
        {gen_server2,handle_msg,2,[{file,"gen_server2.erl"},{line,1056}]},
        {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,241}]}]}
```

(Note that no trace message is emitted for messages that are delivered to
direct reply to requesting clients (neither in 3.12, nor in 3.13, nor
after this commit). This behaviour can be added in future when a direct
reply virtual queue becomes its own queue type.)

(cherry picked from commit 5deff45)

Author: ansd

deps/rabbit/src/rabbit_trace.erl

@@ -62,6 +62,8 @@ tap_in(Msg, QNames, ConnName, ChannelNum, Username, TraceX) ->
     RoutedQs = lists:map(fun(#resource{kind = queue, name = Name}) ->
                                  {longstr, Name};
                             ({#resource{kind = queue, name = Name}, _}) ->
+                                 {longstr, Name};
+                            ({virtual_reply_queue, Name}) ->
                                  {longstr, Name}
                          end, QNames),
     trace(TraceX, Msg, <<"publish">>, XName,

deps/rabbit/test/amqpl_direct_reply_to_SUITE.erl

@@ -16,11 +16,16 @@
 
 all() ->
     [
+     {group, cluster_size_1},
      {group, cluster_size_3}
     ].
 
 groups() ->
     [
+     {cluster_size_1, [shuffle],
+      [
+       trace
+      ]},
      {cluster_size_3, [shuffle],
       [
        rpc_new_to_old_node,
@@ -39,8 +44,11 @@ init_per_suite(Config) ->
 end_per_suite(Config) ->
     Config.
 
-init_per_group(_Group, Config) ->
-    Nodes = 3,
+init_per_group(Group, Config) ->
+    Nodes = case Group of
+                cluster_size_1 -> 1;
+                cluster_size_3 -> 3
+            end,
     Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(
                 Config, [{rmq_nodes_count, Nodes},
@@ -62,6 +70,114 @@ init_per_testcase(Testcase, Config) ->
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
+%% Test case for
+%% https://github.com/rabbitmq/rabbitmq-server/discussions/11662
+trace(Config) ->
+    {ok, _} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["trace_on"]),
+
+    Node = atom_to_binary(rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename)),
+    TraceQueue = <<"my trace queue">>,
+    RequestQueue = <<"my request queue">>,
+    %% This is the pseudo queue that is specially interpreted by RabbitMQ.
+    ReplyQueue = <<"amq.rabbitmq.reply-to">>,
+    RequestPayload = <<"my request">>,
+    ReplyPayload = <<"my reply">>,
+    CorrelationId = <<"my correlation ID">>,
+    Qs = [RequestQueue, TraceQueue],
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    RequesterCh = rabbit_ct_client_helpers:open_channel(Config, 0),
+    ResponderCh = rabbit_ct_client_helpers:open_channel(Config, 0),
+
+    [#'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = Q0}) || Q0 <- Qs],
+    #'queue.bind_ok'{} = amqp_channel:call(
+                           Ch, #'queue.bind'{
+                                  queue = TraceQueue,
+                                  exchange = <<"amq.rabbitmq.trace">>,
+                                  %% We subscribe only to messages entering RabbitMQ.
+                                  routing_key = <<"publish.#">>}),
+
+    %% There is no need to declare this pseudo queue first.
+    amqp_channel:subscribe(RequesterCh,
+                           #'basic.consume'{queue = ReplyQueue,
+                                            no_ack = true},
+                           self()),
+    CTag = receive #'basic.consume_ok'{consumer_tag = CTag0} -> CTag0
+           end,
+    #'confirm.select_ok'{} = amqp_channel:call(RequesterCh, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(RequesterCh, self()),
+
+    %% Send the request.
+    amqp_channel:cast(
+      RequesterCh,
+      #'basic.publish'{routing_key = RequestQueue},
+      #amqp_msg{props = #'P_basic'{reply_to = ReplyQueue,
+                                   correlation_id = CorrelationId},
+                payload = RequestPayload}),
+    receive #'basic.ack'{} -> ok
+    after 5000 -> ct:fail(confirm_timeout)
+    end,
+
+    %% Receive the request.
+    {#'basic.get_ok'{},
+     #amqp_msg{props = #'P_basic'{reply_to = ReplyTo,
+                                  correlation_id = CorrelationId},
+               payload = RequestPayload}
+    } = amqp_channel:call(ResponderCh, #'basic.get'{queue = RequestQueue}),
+
+    %% Send the reply.
+    amqp_channel:cast(
+      ResponderCh,
+      #'basic.publish'{routing_key = ReplyTo},
+      #amqp_msg{props = #'P_basic'{correlation_id = CorrelationId},
+                payload = ReplyPayload}),
+
+    %% Receive the reply.
+    receive {#'basic.deliver'{consumer_tag = CTag},
+             #amqp_msg{payload = ReplyPayload,
+                       props = #'P_basic'{correlation_id = CorrelationId}}} ->
+                ok
+    after 5000 -> ct:fail(missing_reply)
+    end,
+
+    %% 2 messages should have entered RabbitMQ:
+    %% 1. the RPC request
+    %% 2. the RPC reply
+
+    {#'basic.get_ok'{routing_key = <<"publish.">>},
+     #amqp_msg{props = #'P_basic'{headers = RequestHeaders},
+               payload = RequestPayload}
+    } = amqp_channel:call(Ch, #'basic.get'{queue = TraceQueue}),
+    ?assertMatch(#{
+                   <<"exchange_name">> := <<>>,
+                   <<"routing_keys">> := [RequestQueue],
+                   <<"connection">> := <<"127.0.0.1:", _/binary>>,
+                   <<"node">> := Node,
+                   <<"vhost">> := <<"/">>,
+                   <<"user">> := <<"guest">>,
+                   <<"properties">> := #{<<"correlation_id">> := CorrelationId},
+                   <<"routed_queues">> := [RequestQueue]
+                  },
+                 rabbit_misc:amqp_table(RequestHeaders)),
+
+    {#'basic.get_ok'{routing_key = <<"publish.">>},
+     #amqp_msg{props = #'P_basic'{headers = ResponseHeaders},
+               payload = ReplyPayload}
+    } = amqp_channel:call(Ch, #'basic.get'{queue = TraceQueue}),
+    ?assertMatch(#{
+                   <<"exchange_name">> := <<>>,
+                   <<"routing_keys">> := [<<"amq.rabbitmq.reply-to.", _/binary>>],
+                   <<"connection">> := <<"127.0.0.1:", _/binary>>,
+                   <<"node">> := Node,
+                   <<"vhost">> := <<"/">>,
+                   <<"user">> := <<"guest">>,
+                   <<"properties">> := #{<<"correlation_id">> := CorrelationId},
+                   <<"routed_queues">> := [<<"amq.rabbitmq.reply-to.", _/binary>>]
+                  },
+                 rabbit_misc:amqp_table(ResponseHeaders)),
+
+    [#'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = Q0}) || Q0 <- Qs],
+    {ok, _} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["trace_off"]).
+
 %% "new" and "old" refers to new and old RabbitMQ versions in mixed version tests.
 rpc_new_to_old_node(Config) ->
     rpc(0, 1, Config).