QQ: resend pending commands when new leader detected on applied notification.

When a leader changes all enqueuer and consumer processes are notified
from the `state_enter(leader,` callback. However a new leader may not
yet have applied all commands that the old leader had. If any of those
commands is a checkout or a register_enqueuer command these processes
will not be notified of the new leader and thus may never resend their
pending commands.

The new leader will however send an applied notification when it does
apply these entries and these are always sent from the leader process
so can also be used to trigger pending resends. This commit implements
that.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo_client.erl

@@ -586,34 +586,49 @@ update_machine_state(Server, Conf) ->
                       ra_server_proc:ra_event_body(), state()) ->
     {internal, Correlators :: [term()], rabbit_queue_type:actions(), state()} |
     {rabbit_fifo:client_msg(), state()} | {eol, rabbit_queue_type:actions()}.
-handle_ra_event(QName, From, {applied, Seqs},
-                #state{cfg = #cfg{soft_limit = SftLmt}} = State0) ->
+handle_ra_event(QName, Leader, {applied, Seqs},
+                #state{leader = OldLeader,
+                       cfg = #cfg{soft_limit = SftLmt}} = State0) ->
 
     {Corrs, ActionsRev, State1} = lists:foldl(fun seq_applied/2,
-                                              {[], [], State0#state{leader = From}},
+                                              {[], [], State0#state{leader = Leader}},
                                               Seqs),
+
+    %% if the leader has changed we need to resend any pending commands remaining
+    %% after the applied processing
+    State2 = if OldLeader =/= Leader ->
+                    rabbit_log:debug("~ts: Detected QQ leader change (applied) "
+                                     "from ~w to ~w, "
+                                     "resending ~b pending commands",
+                     [?MODULE, OldLeader, Leader,
+                      maps:size(State1#state.pending)]),
+                    resend_all_pending(State1);
+                true ->
+                    State1
+             end,
+
     Actions0 = lists:reverse(ActionsRev),
     Actions = case Corrs of
                   [] ->
                       Actions0;
                   _ ->
-                      %%TODO consider using lists:foldr/3 above because
+                      %%TODO: consider using lists:foldr/3 above because
                       %% Corrs is returned in the wrong order here.
                       %% The wrong order does not matter much because the channel sorts the
                       %% sequence numbers before confirming to the client. But rabbit_fifo_client
                       %% is sequence numer agnostic: it handles any correlation terms.
                       [{settled, QName, Corrs} | Actions0]
               end,
-    case map_size(State1#state.pending) < SftLmt of
-        true when State1#state.slow == true ->
+    case map_size(State2#state.pending) < SftLmt of
+        true when State2#state.slow == true ->
             % we have exited soft limit state
             % send any unsent commands and cancel the time as
             % TODO: really the timer should only be cancelled when the channel
             % exits flow state (which depends on the state of all queues the
             % channel is interacting with)
             % but the fact the queue has just applied suggests
             % it's ok to cancel here anyway
-            State2 = cancel_timer(State1#state{slow = false,
+            State3 = cancel_timer(State2#state{slow = false,
                                                unsent_commands = #{}}),
             % build up a list of commands to issue
             Commands = maps:fold(
@@ -622,16 +637,16 @@ handle_ra_event(QName, From, {applied, Seqs},
                                              add_command(Cid, return, Returns,
                                                          add_command(Cid, discard,
                                                                      Discards, Acc)))
-                         end, [], State1#state.unsent_commands),
-            ServerId = pick_server(State2),
+                         end, [], State2#state.unsent_commands),
+            ServerId = pick_server(State3),
             %% send all the settlements and returns
             State = lists:foldl(fun (C, S0) ->
                                         send_command(ServerId, undefined, C,
                                                      normal, S0)
-                                end, State2, Commands),
+                                end, State3, Commands),
             {ok, State, [{unblock, cluster_name(State)} | Actions]};
         _ ->
-            {ok, State1, Actions}
+            {ok, State2, Actions}
     end;
 handle_ra_event(QName, From, {machine, {delivery, _ConsumerTag, _} = Del}, State0) ->
     handle_delivery(QName, From, Del, State0);