Support Direct Reply-To for AMQP 1.0

 # What?
* Support Direct Reply-To for AMQP 1.0
* Compared to AMQP 0.9.1, this PR allows for multiple volatile queues on a single
  AMQP 1.0 session. Use case: JMS clients can create multiple temporary queues on
  the same JMS/AMQP session:
  * https://jakarta.ee/specifications/messaging/3.1/apidocs/jakarta.messaging/jakarta/jms/session#createTemporaryQueue()
  * https://jakarta.ee/specifications/messaging/3.1/apidocs/jakarta.messaging/jakarta/jms/jmscontext#createTemporaryQueue()
* Huge speed up for Direct Reply-To in AMQP 0.9.1 because prior to this
  PR sending each RPC reply was bottlenecked by the slowest network link
  between the RabbitMQ node the responder connected to and all other
  RabbitMQ nodes.
* Fix missing metrics in for Direct Reply-To in AMQP 0.9.1, e.g.
  `messages_delivered_total`
* Fix missing metrics (even without using Direct Reply-To ) in AMQP 0.9.1:
  If stats level is not `fine`, global metrics `rabbitmq_global_messages_delivered_*` should still be incremented.

 # Why?
* Allow for scalable at-most-once RPC reply delivery
  Example use case: thousands of requesters connect, send a single
  request, wait for a single reply, and disconnect.
  This PR won't create any queue and won't write to the metadata store.
* Feature parity with AMQP 0.9.1

 # How?
This PR extracts the previously channel specific Direct Reply-To code
into a new queue type: `rabbit_volatile_queue`.
"Volatile" describes the semantics, not a use-case. It signals non-durable,
zero-buffer, at-most-once, may-drop, and "not stored in Khepri."

This new queue type is then used for AMQP 1.0 and AMQP 0.9.1.

Sending to the volatile queue is stateless like previously with Direct Reply-To in AMQP 0.9.1 and like done
for the MQTT QoS 0 queue.
This allows for use cases where a single responder replies to e.g. 100k different requesters.

RabbitMQ will automatically auto grant new link-credit to the responder because the new queue type confirms immediately.

The key gets implicitly checked by the channel/session:
If the queue name (including the key) doesn’t exist, the `handle_event` callback for this queue isn’t invoked and therefore
no delivery will be sent to the responder.

This commit supports Direct Reply-To across AMQP 1.0 and 0.9.1. In other
words, the requester can be an AMQP 1.0 client while the responder is an
AMQP 0.9.1 client or vice versa.
RabbitMQ will internally convert between AMQP 0.9.1 `reply_to` and AMQP
1.0 `/queues/<queue>` address. The AMQP 0.9.1 `reply_to` property is
expected to contain a queue name. That's in line with the AMQP 0.9.1
spec:
> One of the standard message properties is Reply-To, which is designed
specifically for carrying the name of reply queues.

Compared to AMQP 0.9.1 where the requester sets the `reply_to` property
to `amq.rabbitmq.reply-to` and RabbitMQ modifies this field when
forwarding the message to the request queue, in AMQP 1.0 the requester
learns about the queue name from the broker at link attachment time.
The requester has to set the reply-to property to the server generated
queue name. That's because the server isn't allowed to modify the bare
message.

During link attachment time, the client has to set certain fields.
These fields are exected to be set by the RabbitMQ client libraries.
Here is an Erlang example:
```erl
Source = #{address => undefined,
           durable => none,
           expiry_policy => <<"link-detach">>,
           dynamic => true,
           capabilities => [<<"rabbitmq:volatile-queue">>]},
AttachArgs = #{name => <<"receiver">>,
               role => {receiver, Source, self()},
               snd_settle_mode => settled,
               rcv_settle_mode => first},
{ok, Receiver} = amqp10_client:attach_link(Session, AttachArgs),
AddressReplyQ = receive {amqp10_event, {link, Receiver, {attached, Attach}}} ->
                  #'v1_0.attach'{source = #'v1_0.source'{address = {utf8, Addr}}} = Attach,
                  Addr
end,
```

The client then sends the message by setting the reply-to address as
follows:
```erl
amqp10_client:send_msg(
  SenderRequester,
  amqp10_msg:set_properties(
    #{message_id => <<"my ID">>,
      reply_to => AddressReplyQ},
    amqp10_msg:new(<<"tag">>, <<"request">>))),
```

If the responder attaches to the queue target in the reply-to field,
RabbitMQ will check if the requester link is still attached. If the
requester detached, the link will be refused.

The responder can also attach to the anonymous null target and set the
`to` field to the `reply-to` address.

It's the requester's reponsiblity to grant sufficient link credit to
volatile queue. RabbitMQ will drop replies if the requester ran out of
link credit. The following Prometheus metric will be incremented:
```
rabbitmq_global_messages_dead_lettered_maxlen_total{queue_type="rabbit_volatile_queue",dead_letter_strategy="disabled"} 0.0
```
That's in line with the MQTT QoS 0 queue type.

The main difference between the volatile queue and the MQTT QoS 0 queue
is that the former isn't written to the metadata store.

Author: ansd

deps/amqp10_client/src/amqp10_client_session.erl

@@ -92,8 +92,8 @@
 -type max_message_size() :: undefined | non_neg_integer().
 -type footer_opt() :: crc32 | adler32.
 
--type attach_args() :: #{name => binary(),
-                         role => attach_role(),
+-type attach_args() :: #{name := binary(),
+                         role := attach_role(),
                          snd_settle_mode => snd_settle_mode(),
                          rcv_settle_mode => rcv_settle_mode(),
                          filter => filter(),
@@ -739,13 +739,19 @@ build_frames(Channel, Trf, Payload, MaxPayloadSize, Acc) ->
 
 make_source(#{role := {sender, _}}) ->
     #'v1_0.source'{};
-make_source(#{role := {receiver, Source, _Pid},
-              filter := Filter}) ->
+make_source(#{role := {receiver, Source, _Pid}} = AttachArgs) ->
     Durable = translate_terminus_durability(maps:get(durable, Source, none)),
+    ExpiryPolicy = case Source of
+                       #{expiry_policy := Policy} when is_binary(Policy) ->
+                           {symbol, Policy};
+                       _ ->
+                           undefined
+                   end,
     Dynamic = maps:get(dynamic, Source, false),
-    TranslatedFilter = translate_filters(Filter),
+    TranslatedFilter = translate_filters(maps:get(filter, AttachArgs, #{})),
     #'v1_0.source'{address = make_address(Source),
                    durable = {uint, Durable},
+                   expiry_policy = ExpiryPolicy,
                    dynamic = Dynamic,
                    filter = TranslatedFilter,
                    capabilities = make_capabilities(Source)}.

deps/rabbit/Makefile

@@ -258,7 +258,7 @@ endef
 
 PARALLEL_CT_SET_1_A = unit_rabbit_ssl unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking
 PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_credit_api_v2 amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet amqp_jms signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
-PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack amqpl_direct_reply_to backing_queue bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange rabbit_direct_reply_to_prop cluster_limit cluster_minority term_to_binary_compat_prop topic_permission transactions unicode unit_access_control
+PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack direct_reply_to_amqp direct_reply_to_amqpl backing_queue bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange rabbit_direct_reply_to_prop cluster_limit cluster_minority term_to_binary_compat_prop topic_permission transactions unicode unit_access_control
 PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit
 
 PARALLEL_CT_SET_2_A = cluster confirms_rejects consumer_timeout rabbit_access_control rabbit_confirms rabbit_core_metrics_gc rabbit_cuttlefish rabbit_db_binding rabbit_db_exchange

deps/rabbit/ct.test.spec

@@ -21,7 +21,8 @@
 ,   amqp_dotnet_SUITE
 ,   amqp_jms_SUITE
 ,   amqpl_consumer_ack_SUITE
-,   amqpl_direct_reply_to_SUITE
+,   direct_reply_to_amqpl_SUITE
+,   direct_reply_to_amqp_SUITE
 ,   amqqueue_backward_compatibility_SUITE
 ,   backing_queue_SUITE
 ,   bindings_SUITE

deps/rabbit/src/mc_amqpl.erl

@@ -228,7 +228,14 @@ convert_from(mc_amqp, Sections, Env) ->
                      %% drop it, what else can we do?
                      undefined
              end,
-
+    ReplyTo = case unwrap_shortstr(ReplyTo0) of
+                  <<"/queues/", Queue/binary>> ->
+                      try cow_uri:urldecode(Queue)
+                      catch error:_ -> undefined
+                      end;
+                  Other ->
+                      Other
+              end,
     BP = #'P_basic'{message_id = MsgId091,
                     delivery_mode = DelMode,
                     expiration = Expiration,
@@ -237,7 +244,7 @@ convert_from(mc_amqp, Sections, Env) ->
                                   [] -> undefined;
                                   AllHeaders -> AllHeaders
                               end,
-                    reply_to = unwrap_shortstr(ReplyTo0),
+                    reply_to = ReplyTo,
                     type = Type,
                     app_id = unwrap_shortstr(GroupId),
                     priority = Priority,
@@ -349,7 +356,7 @@ convert_to(mc_amqp, #content{payload_fragments_rev = PFR} = Content, Env) ->
                delivery_mode = DelMode,
                headers = Headers0,
                user_id = UserId,
-               reply_to = ReplyTo,
+               reply_to = ReplyTo0,
                type = Type,
                priority = Priority,
                app_id = AppId,
@@ -382,25 +389,32 @@ convert_to(mc_amqp, #content{payload_fragments_rev = PFR} = Content, Env) ->
                        ttl = wrap(uint, Ttl),
                        %% TODO: check Priority is a ubyte?
                        priority = wrap(ubyte, Priority)},
+    ReplyTo = case ReplyTo0 of
+                  undefined ->
+                      undefined;
+                  _ ->
+                      Queue = uri_string:quote(ReplyTo0),
+                      {utf8, <<"/queues/", Queue/binary>>}
+              end,
     CorrId = case mc_util:urn_string_to_uuid(CorrId0) of
                  {ok, CorrUUID} ->
                      {uuid, CorrUUID};
                  _ ->
                      wrap(utf8, CorrId0)
              end,
     MsgId = case mc_util:urn_string_to_uuid(MsgId0) of
-                 {ok, MsgUUID} ->
-                     {uuid, MsgUUID};
-                 _ ->
-                     wrap(utf8, MsgId0)
-             end,
+                {ok, MsgUUID} ->
+                    {uuid, MsgUUID};
+                _ ->
+                    wrap(utf8, MsgId0)
+            end,
     P = case amqp10_section_header(?AMQP10_PROPERTIES_HEADER, Headers) of
             undefined ->
                 #'v1_0.properties'{message_id = MsgId,
                                    user_id = wrap(binary, UserId),
                                    to = undefined,
                                    % subject = wrap(utf8, RKey),
-                                   reply_to = wrap(utf8, ReplyTo),
+                                   reply_to = ReplyTo,
                                    correlation_id = CorrId,
                                    content_type = wrap(symbol, ContentType),
                                    content_encoding = wrap(symbol, ContentEncoding),

deps/rabbit/src/pid_recomposition.erl

@@ -1002,6 +1002,7 @@ start(normal, []) ->
         end,
 
         persist_static_configuration(),
+        persistent_term:put(cp_dot, binary:compile_pattern(<<".">>)),
 
         ?LOG_DEBUG(""),
         ?LOG_DEBUG("== Boot steps =="),
@@ -1157,7 +1158,6 @@ pg_local_amqp_connection() ->
 pg_local_scope(Prefix) ->
     list_to_atom(io_lib:format("~s_~s", [Prefix, node()])).
 
-
 -spec update_cluster_tags() -> 'ok'.
 
 update_cluster_tags() ->