Merge pull request #8739 from rabbitmq/mergify/bp/v3.11.x/pr-8736

Settle unroutable message with released state (backport #8015) (backport #8736)

Author: michaelklishin

deps/amqp_client/src/amqp_channel.erl

@@ -886,6 +886,9 @@ flush_writer(#state{driver = direct}) ->
     ok.
 amqp_msg(none) ->
     none;
+amqp_msg({DTag, Content}) ->
+    {Props, Payload} = rabbit_basic_common:from_content(Content),
+    {DTag, #amqp_msg{props = Props, payload = Payload}};
 amqp_msg(Content) ->
     {Props, Payload} = rabbit_basic_common:from_content(Content),
     #amqp_msg{props = Props, payload = Payload}.

deps/rabbit/src/rabbit_channel.erl

@@ -113,7 +113,10 @@
           consumer_timeout,
           authz_context,
           %% defines how ofter gc will be executed
-          writer_gc_threshold
+          writer_gc_threshold,
+          %% true with AMQP 1.0 to include the publishing sequence
+          %% in the return callback, false otherwise
+          extended_return_callback
          }).
 
 -record(pending_ack, {
@@ -520,6 +523,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
     MaxMessageSize = get_max_message_size(),
     ConsumerTimeout = get_consumer_timeout(),
     OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams),
+    UseExtendedReturnCallback = use_extended_return_callback(AmqpParams),
     {ok, GCThreshold} = application:get_env(rabbit, writer_gc_threshold),
     State = #ch{cfg = #conf{state = starting,
                             protocol = Protocol,
@@ -538,7 +542,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             max_message_size = MaxMessageSize,
                             consumer_timeout = ConsumerTimeout,
                             authz_context = OptionalVariables,
-                            writer_gc_threshold = GCThreshold
+                            writer_gc_threshold = GCThreshold,
+                            extended_return_callback = UseExtendedReturnCallback
                            },
                 limiter = Limiter,
                 tx                      = none,
@@ -1101,6 +1106,15 @@ extract_variable_map_from_amqp_params([Value]) ->
 extract_variable_map_from_amqp_params(_) ->
     #{}.
 
+%% Use tuple representation of amqp_params to avoid a dependency on amqp_client.
+%% Used for AMQP 1.0
+use_extended_return_callback({amqp_params_direct,_,_,_,_,
+                              {amqp_adapter_info,_,_,_,_,_,{'AMQP',"1.0"},_},
+                              _}) ->
+    true;
+use_extended_return_callback(_) ->
+    false.
+
 check_msg_size(Content, MaxMessageSize, GCThreshold) ->
     Size = rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
     case Size of
@@ -1942,9 +1956,8 @@ binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
             ok
     end.
 
-basic_return(#basic_message{exchange_name = ExchangeName,
-                            routing_keys  = [RoutingKey | _CcRoutes],
-                            content       = Content},
+basic_return(Content, #basic_message{exchange_name = ExchangeName,
+                                     routing_keys  = [RoutingKey | _CcRoutes]},
              State = #ch{cfg = #conf{protocol = Protocol,
                                      writer_pid = WriterPid}},
              Reason) ->
@@ -2179,7 +2192,9 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                                         mandatory  = Mandatory,
                                         confirm    = Confirm,
                                         msg_seq_no = MsgSeqNo},
-                   RoutedToQueueNames = [QName]}, State0 = #ch{queue_states = QueueStates0}) -> %% optimisation when there is one queue
+                                        RoutedToQueueNames = [QName]},
+                                        State0 = #ch{cfg = #conf{extended_return_callback = ExtendedReturnCallback},
+                                                     queue_states = QueueStates0}) -> %% optimisation when there is one queue
     Qs0 = rabbit_amqqueue:lookup(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
@@ -2188,7 +2203,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
             rabbit_global_counters:messages_routed(amqp091, erlang:min(1, length(Qs))),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
-            ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
+            ok = process_routing_mandatory(ExtendedReturnCallback, Mandatory, Qs, MsgSeqNo, Message, State0),
             State1 = process_routing_confirm(Confirm, QueueNames, MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
             %% contain rejections of publishes
@@ -2216,7 +2231,9 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                                         mandatory  = Mandatory,
                                         confirm    = Confirm,
                                         msg_seq_no = MsgSeqNo},
-                   RoutedToQueueNames}, State0 = #ch{queue_states = QueueStates0}) ->
+                   RoutedToQueueNames},
+                   State0 = #ch{cfg = #conf{extended_return_callback = ExtendedReturnCallback},
+                                queue_states = QueueStates0}) ->
     Qs0 = rabbit_amqqueue:lookup(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
@@ -2225,7 +2242,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
             rabbit_global_counters:messages_routed(amqp091, length(Qs)),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
-            ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
+            ok = process_routing_mandatory(ExtendedReturnCallback, Mandatory, Qs, MsgSeqNo, Message, State0),
             State1 = process_routing_confirm(Confirm, QueueNames,
                                              MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
@@ -2247,19 +2264,32 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
               [rabbit_misc:rs(Resource)])
     end.
 
-process_routing_mandatory(_Mandatory = true,
+process_routing_mandatory(_ExtendedReturnCallback = false,
+                          _Mandatory = true,
+                          _RoutedToQs = [],
+                          _MsgSeqNo,
+                          #basic_message{content = Content} = Msg, State) ->
+    rabbit_global_counters:messages_unroutable_returned(amqp091, 1),
+    ok = basic_return(Content, Msg, State, no_route),
+    ok;
+process_routing_mandatory(_ExtendedReturnCallback = true,
+                          _Mandatory = true,
                           _RoutedToQs = [],
-                          Msg, State) ->
+                          MsgSeqNo,
+                          #basic_message{content = Content} = Msg, State) ->
     rabbit_global_counters:messages_unroutable_returned(amqp091, 1),
-    ok = basic_return(Msg, State, no_route),
+    %% providing the publishing sequence for AMQP 1.0
+    ok = basic_return({MsgSeqNo, Content}, Msg, State, no_route),
     ok;
-process_routing_mandatory(_Mandatory = false,
+process_routing_mandatory(_ExtendedReturnCallback,
+                          _Mandatory = false,
                           _RoutedToQs = [],
+                          _MsgSeqNo,
                           #basic_message{exchange_name = ExchangeName}, State) ->
     rabbit_global_counters:messages_unroutable_dropped(amqp091, 1),
     ?INCR_STATS(exchange_stats, ExchangeName, 1, drop_unroutable, State),
     ok;
-process_routing_mandatory(_, _, _, _) ->
+process_routing_mandatory(_, _, _, _, _, _) ->
     ok.
 
 process_routing_confirm(false, _, _, _, State) ->

deps/rabbit_common/src/rabbit_writer.erl

@@ -105,7 +105,10 @@
 
 -spec send_command(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command
-        (pid(), rabbit_framing:amqp_method_record(), rabbit_types:content()) ->
+        (pid(), rabbit_framing:amqp_method_record(),
+         rabbit_types:content() |
+         {integer(), rabbit_types:content()} %% publishing sequence for AMQP 1.0 return callback
+        ) ->
             'ok'.
 -spec send_command_sync(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command_sync

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_incoming_link.erl

@@ -17,7 +17,7 @@
 %% Just make these constant for the time being.
 -define(INCOMING_CREDIT, 65536).
 
--record(incoming_link, {name, exchange, routing_key,
+-record(incoming_link, {name, exchange, routing_key, mandatory,
                         delivery_id = undefined,
                         delivery_count = 0,
                         send_settle_mode = undefined,
@@ -53,6 +53,7 @@ attach(#'v1_0.attach'{name = Name,
                            SndSettleMode == ?V_1_0_SENDER_SETTLE_MODE_MIXED ->
                         amqp_channel:register_confirm_handler(BCh, self()),
                         rabbit_amqp1_0_channel:call(BCh, #'confirm.select'{}),
+                        amqp_channel:register_return_handler(BCh, self()),
                         true
                 end,
             Flow = #'v1_0.flow'{ handle = Handle,
@@ -69,7 +70,8 @@ attach(#'v1_0.attach'{name = Name,
               initial_delivery_count = undefined, % must be, I am the receiver
               role = ?RECV_ROLE}, %% server is receiver
             IncomingLink1 =
-                IncomingLink#incoming_link{recv_settle_mode = RcvSettleMode},
+                IncomingLink#incoming_link{recv_settle_mode = RcvSettleMode,
+                                           mandatory = Confirm},
             {ok, [Attach, Flow], IncomingLink1, Confirm};
         {error, Reason} ->
             %% TODO proper link establishment protocol here?
@@ -142,7 +144,8 @@ transfer(#'v1_0.transfer'{delivery_id     = DeliveryId0,
            end,
     rabbit_amqp1_0_channel:cast_flow(
       BCh, #'basic.publish'{exchange    = X,
-                            routing_key = RKey}, Msg),
+                            routing_key = RKey,
+                            mandatory = true}, Msg),
     {SendFlow, CreditUsed1} = case CreditUsed - 1 of
                                   C when C =< 0 ->
                                       {true,  ?INCOMING_CREDIT div 2};
@@ -206,6 +209,7 @@ ensure_target(Target = #'v1_0.target'{address       = Address,
                                     dest, DCh, Dest, DeclareParams,
                                     RouteState)
                           end),
+                    maybe_ensure_queue(Dest, DCh),
                     {XName, RK} = rabbit_routing_util:parse_routing(Dest),
                     {ok, Target, Link#incoming_link{
                                    route_state = RouteState1,
@@ -222,6 +226,20 @@ ensure_target(Target = #'v1_0.target'{address       = Address,
             {error, {address_not_utf8_string, Address}}
     end.
 
+maybe_ensure_queue({amqqueue, Q}, Ch) ->
+    try
+        rabbit_amqp1_0_channel:convert_error(
+          fun () ->
+                  Method = #'queue.declare'{queue = list_to_binary(Q),
+                                            passive = true},
+                  amqp_channel:call(Ch, Method)
+          end)
+    catch exit:#'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED} ->
+              ok
+    end;
+maybe_ensure_queue(_, _) ->
+    ok.
+
 incoming_flow(#incoming_link{ delivery_count = Count }, Handle) ->
     #'v1_0.flow'{handle         = Handle,
                  delivery_count = {uint, Count},

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -14,7 +14,7 @@
          incr_incoming_id/1, next_delivery_id/1, transfers_left/1,
          record_transfers/2, bump_outgoing_window/1,
          record_outgoing/4, settle/3, flow_fields/2, channel/1,
-         flow/2, ack/2, validate_attach/1]).
+         flow/2, ack/2, return/2, validate_attach/1]).
 
 -import(rabbit_amqp1_0_util, [protocol_error/3,
                               serial_add/2, serial_diff/2, serial_compare/2]).
@@ -396,3 +396,25 @@ acknowledgement(DeliveryIds, Disposition) ->
                                     last = {uint, lists:last(DeliveryIds)},
                                     settled = true,
                                     state = #'v1_0.accepted'{} }.
+
+return(DTag, Session = #session{incoming_unsettled_map = Unsettled}) ->
+    {DeliveryId,
+     Unsettled1} = case gb_trees:lookup(DTag, Unsettled) of
+                       {value, #incoming_delivery{ delivery_id = Id }} ->
+                           {Id, gb_trees:delete(DTag, Unsettled)};
+                       none ->
+                           {undefined, Unsettled}
+                   end,
+    Disposition = case DeliveryId of
+                      undefined -> undefined;
+                      _  -> release(DeliveryId,
+                                    #'v1_0.disposition'{role = ?RECV_ROLE})
+    end,
+    {Disposition,
+     Session#session{incoming_unsettled_map = Unsettled1}}.
+
+release(DeliveryId, Disposition) ->
+    Disposition#'v1_0.disposition'{ first = {uint, DeliveryId},
+                                    last = {uint, DeliveryId},
+                                    settled = true,
+                                    state = #'v1_0.released'{} }.

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_process.erl

@@ -130,6 +130,24 @@ handle_info(#'basic.ack'{} = Ack, State = #state{writer_pid = WriterPid,
         F <- rabbit_amqp1_0_session:flow_fields(Reply, Session)],
     {noreply, state(Session1, State)};
 
+handle_info({#'basic.return'{}, {DTag, _Msg}}, State = #state{writer_pid = WriterPid,
+                                                              session    = Session}) ->
+    {Reply, Session1} = rabbit_amqp1_0_session:return(DTag, Session),
+    case Reply of
+        undefined ->
+            ok;
+        _ ->
+            rabbit_amqp1_0_writer:send_command(
+              WriterPid,
+              rabbit_amqp1_0_session:flow_fields(Reply, Session)
+             )
+    end,
+    {noreply, state(Session1, State)};
+
+handle_info({#'basic.return'{}, _Msg}, State = #state{session = Session}) ->
+    rabbit_log:warning("AMQP 1.0 message return without publishing sequence"),
+    {noreply, state(Session, State)};
+
 handle_info({bump_credit, Msg}, State) ->
     credit_flow:handle_bump_msg(Msg),
     {noreply, State};

deps/rabbitmq_amqp1_0/test/amqp10_client_SUITE.erl

@@ -24,6 +24,8 @@ groups() ->
     [
      {tests, [], [
                   reliable_send_receive_with_outcomes,
+                  publishing_to_non_existing_queue_should_settle_with_released,
+                  open_link_to_non_existing_destination_should_end_session,
                   roundtrip_classic_queue_with_drain,
                   roundtrip_quorum_queue_with_drain,
                   roundtrip_stream_queue_with_drain,
@@ -163,6 +165,68 @@ reliable_send_receive(Config, Outcome) ->
 
     ok.
 
+publishing_to_non_existing_queue_should_settle_with_released(Config) ->
+    Container = atom_to_binary(?FUNCTION_NAME, utf8),
+    Suffix = <<"foo">>,
+    %% does not exist
+    QName = <<Container/binary, Suffix/binary>>,
+    Host = ?config(rmq_hostname, Config),
+    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
+    Address = <<"/exchange/amq.direct/", QName/binary>>,
+
+    OpnConf = #{address => Host,
+                port => Port,
+                container_id => Container,
+                sasl => {plain, <<"guest">>, <<"guest">>}},
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session(Connection),
+    SenderLinkName = <<"test-sender">>,
+    {ok, Sender} = amqp10_client:attach_sender_link(Session,
+                                                    SenderLinkName,
+                                                    Address),
+    ok = wait_for_credit(Sender),
+    DTag1 = <<"dtag-1">>,
+    %% create an unsettled message,
+    %% link will be in "mixed" mode by default
+    Msg1 = amqp10_msg:new(DTag1, <<"body-1">>, false),
+    ok = amqp10_client:send_msg(Sender, Msg1),
+    ok = wait_for_settlement(DTag1, released),
+
+    ok = amqp10_client:detach_link(Sender),
+    ok = amqp10_client:close_connection(Connection),
+    flush("post sender close"),
+    ok.
+
+open_link_to_non_existing_destination_should_end_session(Config) ->
+    Container = atom_to_list(?FUNCTION_NAME),
+    Name = Container ++ "foo",
+    Addresses = [
+                 "/exchange/" ++ Name ++ "/bar",
+                 "/amq/queue/" ++ Name
+                ],
+    Host = ?config(rmq_hostname, Config),
+    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
+    OpnConf = #{address => Host,
+                port => Port,
+                container_id => list_to_binary(Container),
+                sasl => {plain, <<"guest">>, <<"guest">>}},
+
+    [begin
+         {ok, Connection} = amqp10_client:open_connection(OpnConf),
+         {ok, Session} = amqp10_client:begin_session(Connection),
+         SenderLinkName = <<"test-sender">>,
+         ct:pal("Address ~p", [Address]),
+         {ok, _} = amqp10_client:attach_sender_link(Session,
+                                                    SenderLinkName,
+                                                    list_to_binary(Address)),
+
+         wait_for_session_end(Session),
+         ok = amqp10_client:close_connection(Connection),
+         flush("post sender close")
+
+     end || Address <- Addresses],
+    ok.
+
 roundtrip_classic_queue_with_drain(Config) ->
     QName  = atom_to_binary(?FUNCTION_NAME, utf8),
     roundtrip_queue_with_drain(Config, <<"classic">>, QName).
@@ -382,14 +446,27 @@ wait_for_credit(Sender) ->
               ct:fail(credited_timeout)
     end.
 
+wait_for_session_end(Session) ->
+    receive
+        {amqp10_event, {session, Session, {ended, _}}} ->
+            flush(?FUNCTION_NAME),
+            ok
+    after 5000 ->
+              flush("wait_for_session_end timed out"),
+              ct:fail(settled_timeout)
+    end.
+
 wait_for_settlement(Tag) ->
+    wait_for_settlement(Tag, accepted).
+
+wait_for_settlement(Tag, State) ->
     receive
-        {amqp10_disposition, {accepted, Tag}} ->
+        {amqp10_disposition, {State, Tag}} ->
             flush(?FUNCTION_NAME),
             ok
     after 5000 ->
               flush("wait_for_settlement timed out"),
-              ct:fail(credited_timeout)
+              ct:fail(settled_timeout)
     end.
 
 wait_for_accepts(0) -> ok;