Provide queue name and reason to publisher for rejected messages

 ## What?
1. This commit adds the name of the queue that rejected a message as well
as the reason why the message was rejected in the Rejected outcome for AMQP 1.0.
2. For MQTT 5.0 publishers, the reason is provided in the PUBACK packet.

 ## Why?
It may be helpful for publishers to know which queue out of multilple
target queues rejected the message. One such use case is described in
#1443
It may also be helpful to know for publishers whether the given target
queue rejected the message because its maximum queue length was reached
or because the target queue happens to be unavailable.

 ## How?
RabbitMQ will include the UTF-8 encoded queue name (rather than an AMQP
address) as well as the reason in the `info` field of the Rejected outcome's `error` field:
* `queue: <queue name>`
* `reason: maxlen | unavailable`

For MQTT 5.0, if the queue length limit is exceeded, RabbitMQ will
return the reason code for `Quota exceeded` to the publisher.

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -373,7 +373,10 @@
 
           %% Queue actions that we will process later such that we can confirm and reject
           %% delivery IDs in ranges to reduce the number of DISPOSITION frames sent to the client.
-          stashed_rejected = [] :: [{rejected, rabbit_amqqueue:name(), [delivery_number(),...]}],
+          stashed_rejected = [] :: [{rejected,
+                                     rabbit_amqqueue:name(),
+                                     rabbit_queue_type:reject_reason(),
+                                     [delivery_number(),...]}],
           stashed_settled = [] :: [{settled, rabbit_amqqueue:name(), [delivery_number(),...]}],
           %% Classic queues that are down.
           stashed_down = []:: [rabbit_amqqueue:name()],
@@ -692,7 +695,17 @@ send_delivery_state_changes(State0 = #state{cfg = #cfg{writer_pid = Writer,
     %% Order is important:
     %% 1. Process queue rejections.
     {RejectedIds, GrantCredits0, State1} = handle_stashed_rejected(State0),
-    send_dispositions(RejectedIds, #'v1_0.rejected'{}, Writer, ChannelNum),
+    maps:foreach(
+      fun({QNameBin, Reason}, Ids) ->
+              Info = {map,
+                      [{{symbol, <<"queue">>}, {utf8, QNameBin}},
+                       {{symbol, <<"reason">>}, {symbol, reject_reason_to_binary(Reason)}}]},
+              Rej = #'v1_0.rejected'{
+                       error = #'v1_0.error'{
+                                  condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+                                  info = Info}},
+              send_dispositions(Ids, Rej, Writer, ChannelNum)
+      end, RejectedIds),
     %% 2. Process queue confirmations.
     {AcceptedIds0, GrantCredits1, State2} = handle_stashed_settled(GrantCredits0, State1),
     %% 3. Process unavailable classic queues.
@@ -716,22 +729,28 @@ send_delivery_state_changes(State0 = #state{cfg = #cfg{writer_pid = Writer,
     State.
 
 handle_stashed_rejected(#state{stashed_rejected = []} = State) ->
-    {[], #{}, State};
+    {#{}, #{}, State};
 handle_stashed_rejected(#state{cfg = #cfg{max_link_credit = MaxLinkCredit},
                                stashed_rejected = Actions,
                                incoming_links = Links} = State0) ->
     {Ids, GrantCredits, Ls} =
     lists:foldl(
-      fun({rejected, _QName, Correlations}, Accum) ->
+      fun({rejected, #resource{name = QNameBin}, Reason, Correlations}, Accum) ->
               lists:foldl(
                 fun({HandleInt, DeliveryId}, {Ids0, GrantCreds0, Links0} = Acc) ->
                         case Links0 of
                             #{HandleInt := Link0 = #incoming_link{incoming_unconfirmed_map = U0}} ->
                                 case maps:take(DeliveryId, U0) of
                                     {{_, Settled, _}, U} ->
                                         Ids1 = case Settled of
-                                                   true -> Ids0;
-                                                   false -> [DeliveryId | Ids0]
+                                                   true ->
+                                                       Ids0;
+                                                   false ->
+                                                       maps:update_with(
+                                                         {QNameBin, Reason},
+                                                         fun(L) -> [DeliveryId | L] end,
+                                                         [DeliveryId],
+                                                         Ids0)
                                                end,
                                         Link1 = Link0#incoming_link{incoming_unconfirmed_map = U},
                                         {Link, GrantCreds} = maybe_grant_link_credit(
@@ -745,7 +764,7 @@ handle_stashed_rejected(#state{cfg = #cfg{max_link_credit = MaxLinkCredit},
                                 Acc
                         end
                 end, Accum, Correlations)
-      end, {[], #{}, Links}, Actions),
+      end, {#{}, #{}, Links}, Actions),
 
     State = State0#state{stashed_rejected = [],
                          incoming_links = Ls},
@@ -2116,7 +2135,7 @@ handle_queue_actions(Actions, State) ->
     lists:foldl(
       fun ({settled, _QName, _DelIds} = Action, #state{stashed_settled = As} = S) ->
               S#state{stashed_settled = [Action | As]};
-          ({rejected, _QName, _DelIds} = Action, #state{stashed_rejected = As} = S) ->
+          ({rejected, _QName, _Reason, _DelIds} = Action, #state{stashed_rejected = As} = S) ->
               S#state{stashed_rejected = [Action | As]};
           ({deliver, CTag, AckRequired, Msgs}, S0) ->
               lists:foldl(fun(Msg, S) ->
@@ -2584,6 +2603,11 @@ rejected(DeliveryId, Error) ->
                         settled = true,
                         state = #'v1_0.rejected'{error = Error}}.
 
+reject_reason_to_binary(maxlen) ->
+    <<"maxlen">>;
+reject_reason_to_binary(down) ->
+    <<"unavailable">>.
+
 maybe_grant_link_credit(Credit, MaxLinkCredit, DeliveryCount, NumUnconfirmed, Handle) ->
     case grant_link_credit(Credit, MaxLinkCredit, NumUnconfirmed) of
         true ->

deps/rabbit/src/rabbit_channel.erl

@@ -2787,7 +2787,7 @@ handle_queue_actions(Actions, State) ->
     lists:foldl(
       fun({settled, QRef, MsgSeqNos}, S0) ->
               confirm(MsgSeqNos, QRef, S0);
-         ({rejected, _QRef, MsgSeqNos}, S0) ->
+         ({rejected, _QRef, _Reason, MsgSeqNos}, S0) ->
               {U, Rej} =
               lists:foldr(
                 fun(SeqNo, {U1, Acc}) ->

deps/rabbit/src/rabbit_classic_queue.erl

@@ -390,7 +390,7 @@ handle_event(QName, {reject_publish, SeqNo, _QPid},
     %% It does not matter which queue rejected the message,
     %% if any queue did, it should not be confirmed.
     {U, Rejected} = reject_seq_no(SeqNo, U0),
-    Actions = [{rejected, QName, Rejected}],
+    Actions = [{rejected, QName, maxlen, Rejected}],
     {ok, State#?STATE{unconfirmed = U}, Actions};
 handle_event(QName, {down, Pid, Info}, #?STATE{monitored = Monitored,
                                                unconfirmed = U0} = State0) ->
@@ -405,11 +405,19 @@ handle_event(QName, {down, Pid, Info}, #?STATE{monitored = Monitored,
                           maps:filter(fun (_, #msg_status{pending = Pids}) ->
                                               lists:member(Pid, Pids)
                                       end, U0)),
-            {Unconfirmed, Settled, Rejected} =
-                settle_seq_nos(MsgSeqNos, Pid, U0, down),
-            Actions = settlement_action(
-                        settled, QName, Settled,
-                        settlement_action(rejected, QName, Rejected, Actions0)),
+            {Unconfirmed, Settled, Rejected} = settle_seq_nos(MsgSeqNos, Pid, U0, down),
+            Actions1 = case Rejected of
+                           [] ->
+                               Actions0;
+                           _ ->
+                               [{rejected, QName, down, Rejected} | Actions0]
+                       end,
+            Actions = case Settled of
+                          [] ->
+                              Actions1;
+                          _ ->
+                              [{settled, QName, Settled} | Actions1]
+                      end,
             {ok, State#?STATE{unconfirmed = Unconfirmed}, Actions};
         true ->
             %% any abnormal exit should be considered a full reject of the
@@ -425,17 +433,12 @@ handle_event(QName, {down, Pid, Info}, #?STATE{monitored = Monitored,
                           end, [], U0),
             U = maps:without(MsgIds, U0),
             {ok, State#?STATE{unconfirmed = U},
-             [{rejected, QName, MsgIds} | Actions0]}
+             [{rejected, QName, down, MsgIds} | Actions0]}
     end;
 handle_event(_QName, Action, State)
   when element(1, Action) =:= credit_reply ->
     {ok, State, [Action]}.
 
-settlement_action(_Type, _QRef, [], Acc) ->
-    Acc;
-settlement_action(Type, QRef, MsgSeqs, Acc) ->
-    [{Type, QRef, MsgSeqs} | Acc].
-
 supports_stateful_delivery() -> true.
 
 -spec deliver([{amqqueue:target(), state()}],

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -232,7 +232,8 @@ wait_for_queue_deleted(QRef, N) ->
     end.
 
 -spec lookup_topology(state()) -> state().
-lookup_topology(#state{queue_ref = {resource, Vhost, queue, _} = QRef} = State) ->
+lookup_topology(#state{queue_ref = #resource{virtual_host = Vhost,
+                                             kind = queue} = QRef} = State) ->
     {ok, Q} = rabbit_amqqueue:lookup(QRef),
     DLRKey = rabbit_queue_type_util:args_policy_lookup(<<"dead-letter-routing-key">>,
                                                        fun(_Pol, QArg) -> QArg end, Q),
@@ -253,7 +254,7 @@ handle_queue_actions(Actions, State0) ->
               S1 = handle_settled(QRef, MsgSeqs, S0),
               S2 = ack(S1),
               maybe_cancel_timer(S2);
-          ({rejected, QRef, MsgSeqs}, S0) ->
+          ({rejected, QRef, _Reason, MsgSeqs}, S0) ->
               handle_rejected(QRef, MsgSeqs, S0);
           ({queue_down, _QRef}, S0) ->
               %% target classic queue is down, but not deleted
@@ -291,8 +292,8 @@ rejected(SeqNo, Qs, Pendings)
                              Pendings);
         false ->
             ?LOG_DEBUG("Ignoring rejection for unknown sequence number ~b "
-                             "from target dead letter queues ~tp",
-                             [SeqNo, Qs]),
+                       "from target dead letter queues ~tp",
+                       [SeqNo, Qs]),
             Pendings
     end.
 

deps/rabbit/src/rabbit_queue_type.erl

@@ -81,6 +81,7 @@
 
 -type queue_name() :: rabbit_amqqueue:name().
 -type queue_state() :: term().
+-type reject_reason() :: maxlen | down.
 %% sequence number typically
 -type correlation() :: term().
 -type arguments() :: queue_arguments | consumer_arguments.
@@ -101,6 +102,7 @@
     %% indicate to the queue type module that a message has been delivered
     %% fully to the queue
     {settled, queue_name(), [correlation()]} |
+    {rejected, queue_name(), reject_reason(), [correlation()]} |
     {deliver, rabbit_types:ctag(), boolean(), [rabbit_amqqueue:qmsg()]} |
     {block | unblock, QueueName :: term()} |
     credit_reply_action().
@@ -158,6 +160,7 @@
               credit_reply_action/0,
               action/0,
               actions/0,
+              reject_reason/0,
               settle_op/0,
               queue_type/0,
               credit/0,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1125,7 +1125,7 @@ deliver(QSs, Msg0, Options) ->
               case deliver0(QName, Correlation, Msg, S0) of
                   {reject_publish, S} ->
                       {[{Q, S} | Qs],
-                       [{rejected, QName, [Correlation]} | Actions]};
+                       [{rejected, QName, maxlen, [Correlation]} | Actions]};
                   {ok, S, As} ->
                       {[{Q, S} | Qs], As ++ Actions}
               end

deps/rabbit/test/amqp_client_SUITE.erl

@@ -1053,7 +1053,7 @@ invalid_transfer_settled_flag(Config) ->
 
 quorum_queue_rejects(Config) ->
     {_, Session, LinkPair} = Init = init(Config),
-    QName = atom_to_binary(?FUNCTION_NAME),
+    QName = <<"🎄"/utf8>>,
     QProps = #{arguments => #{<<"x-queue-type">> => {utf8, <<"quorum">>},
                               <<"x-max-length">> => {ulong, 1},
                               <<"x-overflow">> => {utf8, <<"reject-publish">>}}},
@@ -1068,12 +1068,11 @@ quorum_queue_rejects(Config) ->
     %% Therefore, we expect the first 2 messages to be accepted.
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag a">>, <<>>, false)),
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag b">>, <<>>, false)),
-    [receive {amqp10_disposition, {accepted, DTag}} -> ok
-     after 30000 -> ct:fail({missing_accepted, DTag})
-     end || DTag <- [<<"tag a">>, <<"tag b">>]],
+    ok = wait_for_accepted(<<"tag a">>),
+    ok = wait_for_accepted(<<"tag b">>),
 
     %% From now on the quorum queue should reject our publishes.
-    %% Send many messages aync.
+    %% Send many messages async.
     NumMsgs = 20,
     DTags = [begin
                  DTag = integer_to_binary(N),
@@ -1086,8 +1085,13 @@ quorum_queue_rejects(Config) ->
     %% and the final one as unsettled again
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag d">>, <<>>, false)),
 
-    [receive {amqp10_disposition, {rejected, DTag}} -> ok
-     after 30000 -> ct:fail({missing_rejected, DTag})
+    [receive {amqp10_disposition, {{rejected, Error}, DTag}} ->
+                 ?assertMatch(
+                    #'v1_0.error'{
+                       info = {map, [{{symbol, <<"queue">>}, {utf8, QName}},
+                                     {{symbol, <<"reason">>}, {symbol, <<"maxlen">>}}]}},
+                    Error)
+     after 9000 -> ct:fail({missing_rejected, DTag})
      end || DTag <- DTags ++ [<<"tag d">>]],
 
     ok = amqp10_client:detach_link(Sender),
@@ -3659,7 +3663,11 @@ target_classic_queue_down(Config) ->
     %% Instead, the server should reject messages that are sent to classic queues that are down.
     DTag2 = <<"t2">>,
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag2, <<"m2">>, false)),
-    ok = wait_for_settlement(DTag2, rejected),
+    ExpectedErr = #'v1_0.error'{
+                     condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+                     info = {map, [{{symbol, <<"queue">>}, {utf8, QName}},
+                                   {{symbol, <<"reason">>}, {symbol, <<"unavailable">>}}]}},
+    ok = wait_for_settlement(DTag2, {rejected, ExpectedErr}),
 
     ok = rabbit_ct_broker_helpers:start_node(Config, ClassicQueueNode),
     %% Now that the classic queue is up again, we should be able to attach a new receiver

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -2053,15 +2053,20 @@ handle_queue_actions(Actions, #state{} = State0) ->
               {ConfirmPktIds, U} = rabbit_mqtt_confirms:confirm(PktIds, QName, U0),
               send_puback(ConfirmPktIds, ?RC_SUCCESS, S),
               S#state{unacked_client_pubs = U};
-          ({rejected, _QName, PktIds}, S0 = #state{unacked_client_pubs = U0,
-                                                   cfg = #cfg{proto_ver = ProtoVer}}) ->
+          ({rejected, _QName, Reason, PktIds},
+           #state{unacked_client_pubs = U0,
+                  cfg = #cfg{proto_ver = ProtoVer}} = S0) ->
               {RejectPktIds, U} = rabbit_mqtt_confirms:reject(PktIds, U0),
               S = S0#state{unacked_client_pubs = U},
               %% Negative acks are supported only in MQTT v5. In MQTT v3 and v4 we ignore
               %% rejected messages since we can only (but must not) send a positive ack.
               case ProtoVer of
                   ?MQTT_PROTO_V5 ->
-                      send_puback(RejectPktIds, ?RC_IMPLEMENTATION_SPECIFIC_ERROR, S);
+                      RC = case Reason of
+                               maxlen -> ?RC_QUOTA_EXCEEDED;
+                               _ -> ?RC_IMPLEMENTATION_SPECIFIC_ERROR
+                           end,
+                      send_puback(RejectPktIds, RC, S);
                   _ ->
                       ok
               end,

deps/rabbitmq_mqtt/test/mqtt_shared_SUITE.erl

@@ -384,11 +384,13 @@ quorum_queue_rejects(Config) ->
     %% The queue will reject m3.
     V = ?config(mqtt_version, Config),
     if V =:= v3 orelse V =:= v4 ->
-           %% v3 and v4 do not support NACKs. Therefore, the server should drop the message.
+           %% v3 and v4 do not support NACKs.
+           %% Therefore, the server should drop the message.
            ?assertEqual(puback_timeout, util:publish_qos1_timeout(C, Name, <<"m3">>, 700));
        V =:= v5 ->
-           %% v5 supports NACKs. Therefore, the server should send us a NACK.
-           ?assertMatch({ok, #{reason_code_name := implementation_specific_error}},
+           %% v5 supports NACKs.
+           %% Therefore, the server should send us a NACK with the correct reason.
+           ?assertMatch({ok, #{reason_code_name := quota_exceeded}},
                         emqtt:publish(C, Name, <<"m3">>, qos1))
     end,
 

deps/rabbitmq_shovel/src/rabbit_local_shovel.erl

@@ -656,7 +656,8 @@ handle_dest_queue_actions(Actions, State) ->
     lists:foldl(
       fun({settled, QName, MsgSeqNos}, S0) ->
               confirm(MsgSeqNos, QName, S0);
-         ({rejected, _QName, MsgSeqNos}, #{dest := Dst = #{unconfirmed := U0}} = S0) ->
+         ({rejected, _QName, _Reason, MsgSeqNos},
+          #{dest := Dst = #{unconfirmed := U0}} = S0) ->
               {U, Rej} =
               lists:foldr(
                 fun(SeqNo, {U1, Acc}) ->