Enforce correct transfer settled flag

deps/amqp10_client/src/amqp10_client_session.erl

@@ -69,9 +69,6 @@
 %% "The remotely chosen handle is referred to as the input handle." [2.6.2]
 -type input_handle() :: link_handle().
 
--type snd_settle_mode() :: unsettled | settled | mixed.
--type rcv_settle_mode() :: first | second.
-
 -type terminus_durability() :: none | configuration | unsettled_state.
 
 -type target_def() :: #{address => link_address(),

deps/amqp10_common/include/amqp10_types.hrl

@@ -15,5 +15,10 @@
 -define(AMQP_ROLE_SENDER, false).
 -define(AMQP_ROLE_RECEIVER, true).
 
+% [2.8.2]
+-type snd_settle_mode() :: unsettled | settled | mixed.
+% [2.8.3]
+-type rcv_settle_mode() :: first | second.
+
 % [3.2.16]
 -define(MESSAGE_FORMAT, 0).

deps/rabbit/src/rabbit_amqp_session.erl

@@ -140,6 +140,7 @@
          }).
 
 -record(incoming_link, {
+          snd_settle_mode :: snd_settle_mode(),
           %% The exchange is either defined in the ATTACH frame and static for
           %% the life time of the link or dynamically provided in each message's
           %% "to" field (address v2).
@@ -1232,7 +1233,7 @@ handle_attach(#'v1_0.attach'{role = ?AMQP_ROLE_SENDER,
                              name = LinkName,
                              handle = Handle = ?UINT(HandleInt),
                              source = Source,
-                             snd_settle_mode = SndSettleMode,
+                             snd_settle_mode = MaybeSndSettleMode,
                              target = Target,
                              initial_delivery_count = DeliveryCount = ?UINT(DeliveryCountInt)
                             },
@@ -1243,8 +1244,10 @@ handle_attach(#'v1_0.attach'{role = ?AMQP_ROLE_SENDER,
                                          user = User}}) ->
     case ensure_target(Target, Vhost, User, PermCache0) of
         {ok, Exchange, RoutingKey, QNameBin, PermCache} ->
+            SndSettleMode = snd_settle_mode(MaybeSndSettleMode),
             MaxMessageSize = persistent_term:get(max_message_size),
             IncomingLink = #incoming_link{
+                              snd_settle_mode = SndSettleMode,
                               exchange = Exchange,
                               routing_key = RoutingKey,
                               queue_name_bin = QNameBin,
@@ -1256,7 +1259,7 @@ handle_attach(#'v1_0.attach'{role = ?AMQP_ROLE_SENDER,
                        name = LinkName,
                        handle = Handle,
                        source = Source,
-                       snd_settle_mode = SndSettleMode,
+                       snd_settle_mode = MaybeSndSettleMode,
                        rcv_settle_mode = ?V_1_0_RECEIVER_SETTLE_MODE_FIRST,
                        target = Target,
                        %% We are the receiver.
@@ -2304,7 +2307,8 @@ incoming_link_transfer(
                    rcv_settle_mode = RcvSettleMode,
                    handle = Handle = ?UINT(HandleInt)},
   MsgPart,
-  #incoming_link{exchange = LinkExchange,
+  #incoming_link{snd_settle_mode = SndSettleMode,
+                 exchange = LinkExchange,
                  routing_key = LinkRKey,
                  max_message_size = MaxMessageSize,
                  delivery_count = DeliveryCount0,
@@ -2335,6 +2339,7 @@ incoming_link_transfer(
             ok = validate_multi_transfer_settled(MaybeSettled, FirstSettled),
             {MsgBin0, FirstDeliveryId, FirstSettled}
     end,
+    validate_transfer_snd_settle_mode(SndSettleMode, Settled),
     validate_transfer_rcv_settle_mode(RcvSettleMode, Settled),
     PayloadSize = iolist_size(PayloadBin),
     validate_message_size(PayloadSize, MaxMessageSize),
@@ -2914,6 +2919,15 @@ credit_reply_timeout(QType, QName) ->
 default(undefined, Default) -> Default;
 default(Thing,    _Default) -> Thing.
 
+snd_settle_mode({ubyte, Val}) ->
+    case Val of
+        0 -> unsettled;
+        1 -> settled;
+        2 -> mixed
+    end;
+snd_settle_mode(undefined) ->
+    mixed.
+
 transfer_frames(Transfer, Sections, unlimited) ->
     [[Transfer, Sections]];
 transfer_frames(Transfer, Sections, MaxFrameSize) ->
@@ -3059,6 +3073,22 @@ validate_multi_transfer_settled(Other, First)
       "(interpreted) field 'settled' on first transfer (~p)",
       [Other, First]).
 
+validate_transfer_snd_settle_mode(mixed, _Settled) ->
+    ok;
+validate_transfer_snd_settle_mode(unsettled, false) ->
+    %% "If the negotiated value for snd-settle-mode at attachment is unsettled,
+    %% then this field MUST be false (or unset) on every transfer frame for a delivery" [2.7.5]
+    ok;
+validate_transfer_snd_settle_mode(settled, true) ->
+    %% "If the negotiated value for snd-settle-mode at attachment is settled,
+    %% then this field MUST be true on at least one transfer frame for a delivery" [2.7.5]
+    ok;
+validate_transfer_snd_settle_mode(SndSettleMode, Settled) ->
+    protocol_error(
+      ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+      "sender settle mode is '~s' but transfer settled flag is interpreted as being '~s'",
+      [SndSettleMode, Settled]).
+
 %% "If the message is being sent settled by the sender,
 %% the value of this field [rcv-settle-mode] is ignored." [2.7.5]
 validate_transfer_rcv_settle_mode(?V_1_0_RECEIVER_SETTLE_MODE_SECOND, _Settled = false) ->

deps/rabbit/test/amqp_client_SUITE.erl

@@ -44,6 +44,7 @@ groups() ->
        sender_settle_mode_unsettled,
        sender_settle_mode_unsettled_fanout,
        sender_settle_mode_mixed,
+       invalid_transfer_settled_flag,
        quorum_queue_rejects,
        receiver_settle_mode_first,
        publishing_to_non_existing_queue_should_settle_with_released,
@@ -757,6 +758,51 @@ sender_settle_mode_mixed(Config) ->
     ok = end_session_sync(Session),
     ok = amqp10_client:close_connection(Connection).
 
+invalid_transfer_settled_flag(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session1} = amqp10_client:begin_session(Connection),
+    {ok, Session2} = amqp10_client:begin_session(Connection),
+    TargetAddr = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
+    {ok, SenderSettled} = amqp10_client:attach_sender_link_sync(
+                            Session1, <<"link 1">>, TargetAddr, settled),
+    {ok, SenderUnsettled} = amqp10_client:attach_sender_link_sync(
+                              Session2, <<"link 2">>, TargetAddr, unsettled),
+    ok = wait_for_credit(SenderSettled),
+    ok = wait_for_credit(SenderUnsettled),
+
+    ok = amqp10_client:send_msg(SenderSettled, amqp10_msg:new(<<"tag1">>, <<"m1">>, false)),
+    receive
+        {amqp10_event,
+         {session, Session1,
+          {ended,
+           #'v1_0.error'{
+              condition = ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+              description = {utf8, Description1}}}}} ->
+            ?assertEqual(
+               <<"sender settle mode is 'settled' but transfer settled flag is interpreted as being 'false'">>,
+               Description1)
+    after 5000 -> flush(missing_ended),
+                  ct:fail({missing_event, ?LINE})
+    end,
+
+    ok = amqp10_client:send_msg(SenderUnsettled, amqp10_msg:new(<<"tag2">>, <<"m2">>, true)),
+    receive
+        {amqp10_event,
+         {session, Session2,
+          {ended,
+           #'v1_0.error'{
+              condition = ?V_1_0_CONNECTION_ERROR_FRAMING_ERROR,
+              description = {utf8, Description2}}}}} ->
+            ?assertEqual(
+               <<"sender settle mode is 'unsettled' but transfer settled flag is interpreted as being 'true'">>,
+               Description2)
+    after 5000 -> flush(missing_ended),
+                  ct:fail({missing_event, ?LINE})
+    end,
+
+    ok = amqp10_client:close_connection(Connection).
+
 quorum_queue_rejects(Config) ->
     {Connection, Session, LinkPair} = init(Config),
     QName = atom_to_binary(?FUNCTION_NAME),
@@ -4761,7 +4807,7 @@ dead_letter_reject_message_order(QType, Config) ->
     {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPair, QName2, #{}),
 
     {ok, Sender} = amqp10_client:attach_sender_link(
-                     Session, <<"sender">>, rabbitmq_amqp_address:queue(QName1), unsettled),
+                     Session, <<"sender">>, rabbitmq_amqp_address:queue(QName1), settled),
     wait_for_credit(Sender),
     {ok, Receiver1} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 1">>, rabbitmq_amqp_address:queue(QName1), unsettled),
@@ -4852,7 +4898,7 @@ dead_letter_reject_many_message_order(QType, Config) ->
     {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPair, QName2, #{}),
 
     {ok, Sender} = amqp10_client:attach_sender_link(
-                     Session, <<"sender">>, rabbitmq_amqp_address:queue(QName1), unsettled),
+                     Session, <<"sender">>, rabbitmq_amqp_address:queue(QName1), settled),
     wait_for_credit(Sender),
     {ok, Receiver1} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 1">>, rabbitmq_amqp_address:queue(QName1), unsettled),
@@ -5141,7 +5187,7 @@ footer_checksum(FooterOpt, Config) ->
     SndAttachArgs = #{name => <<"my sender">>,
                       role => {sender, #{address => Addr,
                                          durable => configuration}},
-                      snd_settle_mode => settled,
+                      snd_settle_mode => mixed,
                       rcv_settle_mode => first,
                       footer_opt => FooterOpt},
     {ok, Receiver} = amqp10_client:attach_link(Session, RecvAttachArgs),

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -74,8 +74,13 @@ connect_source(State = #{name := Name,
                          ack_mode := AckMode,
                          source := #{uris := [Uri | _],
                                      source_address := Addr} = Src}) ->
+    SndSettleMode = case AckMode of
+                        no_ack -> settled;
+                        on_publish -> unsettled;
+                        on_confirm -> unsettled
+                    end,
     AttachFun = fun amqp10_client:attach_receiver_link/5,
-    {Conn, Sess, LinkRef} = connect(Name, AckMode, Uri, "receiver", Addr, Src,
+    {Conn, Sess, LinkRef} = connect(Name, SndSettleMode, Uri, "receiver", Addr, Src,
                                     AttachFun),
     State#{source => Src#{current => #{conn => Conn,
                                        session => Sess,
@@ -87,8 +92,13 @@ connect_dest(State = #{name := Name,
                        ack_mode := AckMode,
                        dest := #{uris := [Uri | _],
                                  target_address := Addr} = Dst}) ->
+    SndSettleMode = case AckMode of
+                        no_ack -> settled;
+                        on_publish -> settled;
+                        on_confirm -> unsettled
+                    end,
     AttachFun = fun amqp10_client:attach_sender_link_sync/5,
-    {Conn, Sess, LinkRef} = connect(Name, AckMode, Uri, "sender", Addr, Dst,
+    {Conn, Sess, LinkRef} = connect(Name, SndSettleMode, Uri, "sender", Addr, Dst,
                                     AttachFun),
     %% wait for link credit here as if there are messages waiting we may try
     %% to forward before we've received credit
@@ -99,7 +109,7 @@ connect_dest(State = #{name := Name,
                                      link => LinkRef,
                                      uri => Uri}}}.
 
-connect(Name, AckMode, Uri, Postfix, Addr, Map, AttachFun) ->
+connect(Name, SndSettleMode, Uri, Postfix, Addr, Map, AttachFun) ->
     {ok, Config0} = amqp10_client:parse_uri(Uri),
     %% As done for AMQP 0.9.1, exclude AMQP 1.0 shovel connections from maintenance mode
     %% to prevent crashes and errors being logged by the shovel plugin when a node gets drained.
@@ -113,16 +123,11 @@ connect(Name, AckMode, Uri, Postfix, Addr, Map, AttachFun) ->
                    LinkName0 = gen_unique_name(Name, Postfix),
                    rabbit_data_coercion:to_binary(LinkName0)
                end,
-    % mixed settlement mode covers all the ack_modes
-    SettlementMode = case AckMode of
-                         no_ack -> settled;
-                         _ -> unsettled
-                     end,
     % needs to be sync, i.e. awaits the 'attach' event as
     % else we may try to use the link before it is ready
     Durability = maps:get(durability, Map, unsettled_state),
     {ok, LinkRef} = AttachFun(Sess, LinkName, Addr,
-                              SettlementMode,
+                              SndSettleMode,
                               Durability),
     {Conn, Sess, LinkRef}.