Fix bug

Fixes following regression:
```
make run-broker TEST_TMPDIR="$HOME/scratch/rabbit/test" PLUGINS="rabbitmq_management” FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 3"
deps/rabbitmq_management/bin/rabbitmqadmin declare queue name=s1 queue_type=stream durable=true
quiver //host.docker.internal//queue/s1 --durable -d 30s --credit 260 --body-size 12
```
This command just stopped receiving after a few messages when executed the 2nd time
because only 256 messages were sent by the writer proc although 260
delivery action were emitted by rabbit_stream_queue.
4 messages in the outgoing_pending queue were wrongly overwritten.

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1380,22 +1380,22 @@ send_pending(#state{remote_incoming_window = RemoteIncomingWindow,
         {{value, #'v1_0.flow'{} = Flow0}, Buf} ->
             #cfg{writer_pid = WriterPid,
                  channel_num = Ch} = State#state.cfg,
-            Flow = session_flow_fields(Flow0, State),
+            State1 = State#state{outgoing_pending = Buf},
+            Flow = session_flow_fields(Flow0, State1),
             rabbit_amqp_writer:send_command(WriterPid, Ch, Flow),
-            send_pending(State#state{outgoing_pending = Buf});
-        {{value, Delivery}, Buf1} ->
+            send_pending(State1);
+        {{value, Delivery}, Buf} ->
             case RemoteIncomingWindow =:= 0 orelse
                  credit_flow:blocked() of
                 true ->
                     State;
                 false ->
-                    {NewRemoteIncomingWindow, Buf, State1} =
-                    send_pending_delivery(Delivery, Buf1, State),
+                    {NewRemoteIncomingWindow, State1} =
+                    send_pending_delivery(Delivery, Buf, State),
                     NumTransfersSent = RemoteIncomingWindow - NewRemoteIncomingWindow,
                     State2 = session_flow_control_sent_transfers(NumTransfersSent, State1),
-                    State3 = State2#state{outgoing_pending = Buf},
                     %% Recurse to possibly send FLOW frames.
-                    send_pending(State3)
+                    send_pending(State2)
             end
     end.
 
@@ -1605,13 +1605,13 @@ send_pending_delivery(#pending_delivery{
                          outgoing_unsettled = #outgoing_unsettled{consumer_tag = Ctag,
                                                                   queue_name = QName}
                         } = Pending,
-                      Buf,
+                      Buf0,
                       #state{remote_incoming_window = Space,
                              outgoing_links = OutgoingLinks,
                              queue_states = QStates,
                              cfg = #cfg{writer_pid = WriterPid,
                                         channel_num = Ch}
-                            } = State) ->
+                            } = State0) ->
     Handle = ctag_to_handle(Ctag),
     case is_map_key(Handle, OutgoingLinks) of
         true ->
@@ -1634,37 +1634,37 @@ send_pending_delivery(#pending_delivery{
                       end,
             case send_frames(SendFun, Frames, Space) of
                 {sent_all, SpaceLeft} ->
-                    {SpaceLeft,
-                     Buf,
-                     sent_pending_delivery(Pending, Handle, State)};
+                    State1 = State0#state{outgoing_pending = Buf0},
+                    State = sent_pending_delivery(Pending, Handle, State1),
+                    {SpaceLeft, State};
                 {sent_some, SpaceLeft, Rest} ->
-                    {SpaceLeft,
-                     queue:in_r(Pending#pending_delivery{frames = Rest}, Buf),
-                     State}
+                    Buf = queue:in_r(Pending#pending_delivery{frames = Rest}, Buf0),
+                    State = State0#state{outgoing_pending = Buf},
+                    {SpaceLeft, State}
             end;
         false ->
             %% Link got detached. Either the client closed the link in which case the queue
             %% already requeued all checked out messages or the queue doesn't exist anymore
             %% in which case there is no point in requeuing this message.
             %% Therefore, ignore (drop) this delivery.
-            {Space, Buf, State}
+            State = State0#state{outgoing_pending = Buf0},
+            {Space, State}
     end;
 send_pending_delivery(#pending_management_delivery{frames = Frames} = Pending,
-                      Buf,
+                      Buf0,
                       #state{remote_incoming_window = Space,
                              cfg = #cfg{writer_pid = WriterPid,
                                         channel_num = Ch}
-                            } = State) ->
+                            } = State0) ->
     SendFun = send_fun(WriterPid, Ch),
     case send_frames(SendFun, Frames, Space) of
         {sent_all, SpaceLeft} ->
-            {SpaceLeft,
-             Buf,
-             State};
+            State = State0#state{outgoing_pending = Buf0},
+            {SpaceLeft, State};
         {sent_some, SpaceLeft, Rest} ->
-            {SpaceLeft,
-             queue:in_r(Pending#pending_management_delivery{frames = Rest}, Buf),
-             State}
+            Buf = queue:in_r(Pending#pending_management_delivery{frames = Rest}, Buf0),
+            State = State0#state{outgoing_pending = Buf},
+            {SpaceLeft, State}
     end.
 
 send_frames(_, [], SpaceLeft) ->