Support Direct Reply-To for AMQP 1.0

[ansd]

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()
• 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.

This PR supersedes #5103

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:

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:

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.

In addition, the responder can check at any time whether the requester is still there by doing an HTTP GET (over AMQP) on the /queues/<queue> endpoint. If the HTTP return code is 404, the requester is gone. The motivation is similar to what our docs state for AMQP 0.9.1:

If the RPC server is going to perform some expensive computation it might wish to check if the client has gone away. To do this the server can declare the generated reply name first on a disposable channel in order to determine whether it still exists. Note that even if you declare the "queue" with passive=false there is no way to create it; the declare will just succeed (with 0 messages ready and 1 consumer) or fail.

This existence check works across both AMQP versions.

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.

Breaking Change

Prior to this PR the following documented caveat applied:

If the RPC server publishes with the mandatory flag set then amq.rabbitmq.reply-to.*
is treated as not a queue; i.e. if the server only publishes to this name then the message
will be considered "not routed"; a basic.return will be sent if the mandatory flag was set.

This PR fixes this caveat.
This PR introduces the following new behaviour:

If the RPC server publishes with the mandatory flag set, then amq.rabbitmq.reply-to.*
is treated as a queue (assuming this queue name is encoded correctly). However,
whether the requester is still there to consume the reply is not checked at routing time.
In other words, if the RPC server only publishes to this name, then the message will be
considered "routed" and RabbitMQ will therefore not send a basic.return.