rabbif_fifo: refactor and fix tests

Rename rabbit_fifo consumer suspected_down field to status.

1 files changed, 84 insertions, 80 deletions

diff --git a/src/rabbit_fifo.erl b/src/rabbit_fifo.erl
index 2580c1bfc0..4f548568be 100644
--- a/src/rabbit_fifo.erl
+++ b/src/rabbit_fifo.erl
@@ -184,7 +184,7 @@
          %% command: `{consumer_credit, ReceiverDeliveryCount, Credit}'
          credit_mode = simple_prefetch :: credit_mode(), % part of snapshot data
          lifetime = once :: once | auto,
-         suspected_down = false :: 'cancel' | boolean()
+         status = up :: up | suspected_down | cancelled
         }).
 
 -type consumer() :: #consumer{}.
@@ -193,7 +193,7 @@
         {next_seqno = 1 :: msg_seqno(),
          % out of order enqueues - sorted list
          pending = [] :: [{msg_seqno(), ra_index(), raw_msg()}],
-         suspected_down = false :: boolean()
+         status = up :: up | suspected_down
         }).
 
 -record(state,
@@ -463,27 +463,30 @@ apply(_, {down, ConsumerPid, noconnection},
     % mark all consumers and enqueuers as suspected down
     % and monitor the node so that we can find out the final state of the
     % process at some later point
-    {Cons, State, Effects1} = maps:fold(
-                                  fun({_, P} = K,
-                                      #consumer{checked_out = Checked0} = C,
-                                      {Co, St0, Eff}) when (node(P) =:= Node) and
-                                                           (C#consumer.suspected_down =/= cancel)->
-                                          St = return_all(St0, Checked0),
-                                          Credit = increase_credit(C, maps:size(Checked0)),
-                                          Eff1 = ConsumerUpdateActiveFun(St, K, C, false, suspected_down, Eff),
-                                          {maps:put(K, C#consumer{suspected_down = true,
-                                                                  credit = Credit,
-                                                                  checked_out = #{}}, Co),
-                                           St, Eff1};
-                                     (K, C, {Co, St, Eff}) ->
-                                          {maps:put(K, C, Co), St, Eff}
-                                  end, {#{}, State0, []}, Cons0),
+    {Cons, State, Effects1} =
+        maps:fold(fun({_, P} = K,
+                      #consumer{checked_out = Checked0} = C,
+                      {Co, St0, Eff}) when (node(P) =:= Node) and
+                                           (C#consumer.status =/= cancelled)->
+                          St = return_all(St0, Checked0),
+                          Credit = increase_credit(C, maps:size(Checked0)),
+                          Eff1 = ConsumerUpdateActiveFun(St, K, C, false,
+                                                         suspected_down, Eff),
+                          {maps:put(K,
+                                    C#consumer{status = suspected_down,
+                                               credit = Credit,
+                                               checked_out = #{}}, Co),
+                           St, Eff1};
+                     (K, C, {Co, St, Eff}) ->
+                          {maps:put(K, C, Co), St, Eff}
+                  end, {#{}, State0, []}, Cons0),
     Enqs = maps:map(fun(P, E) when node(P) =:= Node ->
-                            E#enqueuer{suspected_down = true};
+                            E#enqueuer{status = suspected_down};
                        (_, E) -> E
                     end, Enqs0),
     % mark waiting consumers as suspected if necessary
-    WaitingConsumers = maybe_mark_suspect_waiting_consumers(Node, State0, true),
+    WaitingConsumers = update_waiting_consumer_status(Node, State0,
+                                                            suspected_down),
 
     Effects2 = case maps:size(Cons) of
                    0 ->
@@ -516,8 +519,8 @@ apply(Meta, {down, Pid, _Info}, #state{consumers = Cons0,
                                    end, {State2, Effects1}, DownConsumers),
     checkout(Meta, State, Effects);
 apply(Meta, {nodeup, Node}, #state{consumers = Cons0,
-                                enqueuers = Enqs0,
-                                service_queue = SQ0} = State0) ->
+                                   enqueuers = Enqs0,
+                                   service_queue = SQ0} = State0) ->
     %% A node we are monitoring has come back.
     %% If we have suspected any processes of being
     %% down we should now re-issue the monitors for them to detect if they're
@@ -526,20 +529,20 @@ apply(Meta, {nodeup, Node}, #state{consumers = Cons0,
                 || P <- suspected_pids_for(Node, State0)],
 
     % un-suspect waiting consumers when necessary
-    WaitingConsumers = maybe_mark_suspect_waiting_consumers(Node, State0,
-                                                            false),
+    WaitingConsumers = update_waiting_consumer_status(Node, State0, up),
 
     Enqs1 = maps:map(fun(P, E) when node(P) =:= Node ->
-                             E#enqueuer{suspected_down = false};
+                             E#enqueuer{status = up};
                         (_, E) -> E
                      end, Enqs0),
     ConsumerUpdateActiveFun = consumer_active_flag_update_function(State0),
     {Cons1, SQ, Effects} =
         maps:fold(fun({_, P} = ConsumerId, C, {CAcc, SQAcc, EAcc})
-                        when (node(P) =:= Node) and (C#consumer.suspected_down =/= cancel) ->
+                        when (node(P) =:= Node) and
+                             (C#consumer.status =/= cancelled) ->
                           EAcc1 = ConsumerUpdateActiveFun(State0, ConsumerId, C, true, up, EAcc),
                           update_or_remove_sub(
-                            ConsumerId, C#consumer{suspected_down = false},
+                            ConsumerId, C#consumer{status = up},
                             CAcc, SQAcc, EAcc1);
                      (_, _, Acc) ->
                           Acc
@@ -585,27 +588,27 @@ handle_waiting_consumer_down(Pid,
     State = State0#state{waiting_consumers = StillUp},
     {Effects, State}.
 
-maybe_mark_suspect_waiting_consumers(_Node, #state{consumer_strategy = default},
-                                     _Suspected) ->
+update_waiting_consumer_status(_Node, #state{consumer_strategy = default},
+                               _Status) ->
     [];
-maybe_mark_suspect_waiting_consumers(_Node,
-                                     #state{consumer_strategy = single_active,
-                                            waiting_consumers = []},
-                                     _Suspected) ->
+update_waiting_consumer_status(_Node,
+                               #state{consumer_strategy = single_active,
+                                      waiting_consumers = []},
+                               _Status) ->
     [];
-maybe_mark_suspect_waiting_consumers(Node,
-                                     #state{consumer_strategy = single_active,
-                                            waiting_consumers = WaitingConsumers},
-                                     Suspected) ->
+update_waiting_consumer_status(Node,
+                               #state{consumer_strategy = single_active,
+                                      waiting_consumers = WaitingConsumers},
+                               Status) ->
     [begin
          case node(P) of
              Node ->
-                 {ConsumerId, Consumer#consumer{suspected_down = Suspected}};
+                 {ConsumerId, Consumer#consumer{status = Status}};
              _ ->
                  {ConsumerId, Consumer}
          end
      end || {{_, P} = ConsumerId, Consumer} <- WaitingConsumers,
-            Consumer#consumer.suspected_down =/= cancel].
+            Consumer#consumer.status =/= cancelled].
 
 -spec state_enter(ra_server:ra_state(), state()) -> ra_machine:effects().
 state_enter(leader, #state{consumers = Cons,
@@ -738,12 +741,13 @@ query_consumers(#state{consumers = Consumers,
                        consumer_strategy = ConsumerStrategy } = State) ->
     ActiveActivityStatusFun = case ConsumerStrategy of
                                   default ->
-                                      fun(_ConsumerId, #consumer{suspected_down = SuspectedDown}) ->
-                                          case SuspectedDown of
-                                              true ->
-                                                  {false, suspected_down};
-                                              false ->
-                                                  {true, up}
+                                      fun(_ConsumerId,
+                                          #consumer{status = Status}) ->
+                                          case Status of
+                                              suspected_down  ->
+                                                  {false, Status};
+                                              _ ->
+                                                  {true, Status}
                                           end
                                       end;
                                   single_active ->
@@ -757,7 +761,7 @@ query_consumers(#state{consumers = Consumers,
                                           end
                                       end
                               end,
-    FromConsumers = maps:fold(fun (_, #consumer{suspected_down = cancel}, Acc) ->
+    FromConsumers = maps:fold(fun (_, #consumer{status = cancelled}, Acc) ->
                                       Acc;
                                    ({Tag, Pid}, #consumer{meta = Meta} = Consumer, Acc) ->
                                       {Active, ActivityStatus} = ActiveActivityStatusFun({Tag, Pid}, Consumer),
@@ -772,7 +776,7 @@ query_consumers(#state{consumers = Consumers,
                                                Acc)
                               end, #{}, Consumers),
     FromWaitingConsumers =
-        lists:foldl(fun ({_, #consumer{suspected_down = cancel}}, Acc) ->
+        lists:foldl(fun ({_, #consumer{status = cancelled}}, Acc) ->
                                       Acc;
                         ({{Tag, Pid}, #consumer{meta = Meta} = Consumer}, Acc) ->
                             {Active, ActivityStatus} = ActiveActivityStatusFun({Tag, Pid}, Consumer),
@@ -944,7 +948,7 @@ maybe_return_all(ConsumerId, #consumer{checked_out = Checked0} = Consumer, Cons1
                 update_or_remove_sub(ConsumerId,
                                      Consumer#consumer{lifetime = once,
                                                        credit = 0,
-                                                       suspected_down = cancel},
+                                                       status = cancelled},
                                      C0, SQ0, Effects0),
             {S0#state{consumers = Cons, service_queue = SQ}, Effects1};
         down ->
@@ -1324,9 +1328,9 @@ checkout_one(#state{service_queue = SQ0,
                             %% can happen when draining
                             %% recurse without consumer on queue
                             checkout_one(InitState#state{service_queue = SQ1});
-                        {ok, #consumer{suspected_down = cancel}} ->
+                        {ok, #consumer{status = cancelled}} ->
                             checkout_one(InitState#state{service_queue = SQ1});
-                        {ok, #consumer{suspected_down = true}} ->
+                        {ok, #consumer{status = suspected_down}} ->
                             checkout_one(InitState#state{service_queue = SQ1});
                         {ok, #consumer{checked_out = Checked0,
                                        next_msg_id = Next,
@@ -1381,8 +1385,9 @@ update_or_remove_sub(ConsumerId, #consumer{lifetime = once,
     case maps:size(Checked)  of
         0 ->
             % we're done with this consumer
-            {maps:remove(ConsumerId, Cons), ServiceQueue,
-             [{demonitor, process, ConsumerId} | Effects]};
+            % TODO: demonitor consumer pid but _only_ if there are no other
+            % monitors for this pid
+            {maps:remove(ConsumerId, Cons), ServiceQueue, Effects};
         _ ->
             % there are unsettled items so need to keep around
             {maps:put(ConsumerId, Con, Cons), ServiceQueue, Effects}
@@ -1402,7 +1407,6 @@ uniq_queue_in(Key, Queue) ->
             queue:in(Key, Queue)
     end.
 
-
 update_consumer(ConsumerId, Meta, Spec,
                 #state{consumer_strategy = default} = State0) ->
     %% general case, single active consumer off
@@ -1576,18 +1580,18 @@ message_size(Msg) ->
 suspected_pids_for(Node, #state{consumers = Cons0,
                                 enqueuers = Enqs0,
                                 waiting_consumers = WaitingConsumers0}) ->
-    Cons = maps:fold(fun({_, P}, #consumer{suspected_down = true}, Acc)
+    Cons = maps:fold(fun({_, P}, #consumer{status = suspected_down}, Acc)
                            when node(P) =:= Node ->
                              [P | Acc];
                         (_, _, Acc) -> Acc
                      end, [], Cons0),
-    Enqs = maps:fold(fun(P, #enqueuer{suspected_down = true}, Acc)
+    Enqs = maps:fold(fun(P, #enqueuer{status = suspected_down}, Acc)
                            when node(P) =:= Node ->
                              [P | Acc];
                         (_, _, Acc) -> Acc
                      end, Cons, Enqs0),
     lists:foldl(fun({{_, P},
-                     #consumer{suspected_down = true}}, Acc)
+                     #consumer{status = suspected_down}}, Acc)
                       when node(P) =:= Node ->
                         [P | Acc];
                    (_, Acc) -> Acc
@@ -1837,10 +1841,11 @@ return_checked_out_test() ->
     {State0, [_, _]} = enq(1, 1, first, test_init(test)),
     {State1, [_Monitor,
               {send_msg, _, {delivery, _, [{MsgId, _}]}, ra_event},
-              {aux, active} | _
-             ]} = check(Cid, 2, State0),
-    % return
-    {_State2, _, [_]} = apply(meta(3), make_return(Cid, [MsgId]), State1),
+              {aux, active} | _ ]} = check_auto(Cid, 2, State0),
+    % returning immediately checks out the same message again
+    {_, ok, [{send_msg, _, {delivery, _, [{_, _}]}, ra_event},
+             {aux, active}]} =
+        apply(meta(3), make_return(Cid, [MsgId]), State1),
     ok.
 
 return_auto_checked_out_test() ->
@@ -1867,15 +1872,19 @@ cancelled_checkout_out_test() ->
     {State00, [_, _]} = enq(1, 1, first, test_init(test)),
     {State0, [_]} = enq(2, 2, second, State00),
     {State1, _} = check_auto(Cid, 2, State0),
-    % cancelled checkout should return all pending messages to queue
+    % cancelled checkout should not return pending messages to queue
     {State2, _, _} = apply(meta(3), make_checkout(Cid, cancel, #{}), State1),
     ?assertEqual(1, maps:size(State2#state.messages)),
-    ?assertEqual(1, lqueue:len(State2#state.returns)),
+    ?assertEqual(0, lqueue:len(State2#state.returns)),
 
-    {State3, {dequeue, {0, {_, first}}, _}, _} =
+    {State3, {dequeue, empty}} =
         apply(meta(3), make_checkout(Cid, {dequeue, settled}, #{}), State2),
+    %% settle
+    {State4, ok, _} =
+        apply(meta(4), make_settle(Cid, [0]), State3),
+
     {_State, {dequeue, {_, {_, second}}, _}, _} =
-        apply(meta(4), make_checkout(Cid, {dequeue, settled}, #{}), State3),
+        apply(meta(5), make_checkout(Cid, {dequeue, settled}, #{}), State4),
     ok.
 
 down_with_noproc_consumer_returns_unsettled_test() ->
@@ -1937,16 +1946,6 @@ down_with_noproc_enqueuer_is_cleaned_up_test() ->
     ?assert(0 =:= maps:size(State1#state.enqueuers)),
     ok.
 
-completed_consumer_yields_demonitor_effect_test() ->
-    Cid = {<<"completed_consumer_yields_demonitor_effect_test">>, self()},
-    {State0, [_, _]} = enq(1, 1, second, test_init(test)),
-    {State1, [{monitor, process, _} |  _]} = check(Cid, 2, State0),
-    {_, Effects} = settle(Cid, 3, 0, State1),
-    ?ASSERT_EFF({demonitor, _, _}, Effects),
-    % release cursor for empty queue
-    ?ASSERT_EFF({release_cursor, 3, _}, Effects),
-    ok.
-
 discarded_message_without_dead_letter_handler_is_removed_test() ->
     Cid = {<<"completed_consumer_yields_demonitor_effect_test">>, self()},
     {State0, [_, _]} = enq(1, 1, first, test_init(test)),
@@ -2322,7 +2321,10 @@ single_active_consumer_test() ->
     ?assertEqual(1, length(Effects1)),
 
     % cancelling the active consumer
-    {State3, _, Effects2} = apply(meta(3), #checkout{spec = cancel, consumer_id = {<<"ctag1">>, self()}}, State2),
+    {State3, _, Effects2} = apply(meta(3),
+                                  #checkout{spec = cancel,
+                                            consumer_id = {<<"ctag1">>, self()}},
+                                  State2),
     % the second registered consumer is now the active one
     ?assertEqual(1, map_size(State3#state.consumers)),
     ?assert(maps:is_key({<<"ctag2">>, self()}, State3#state.consumers)),
@@ -2425,15 +2427,16 @@ single_active_consumer_mark_waiting_consumers_as_suspected_when_down_noconnnecti
             State),
         NewState
                   end,
-    State1 = lists:foldl(AddConsumer, State0, [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
+    State1 = lists:foldl(AddConsumer, State0,
+                         [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
 
     % simulate node goes down
     {State2, _, _} = apply(#{}, {down, self(), noconnection}, State1),
 
     % all the waiting consumers should be suspected down
     ?assertEqual(3, length(State2#state.waiting_consumers)),
-    lists:foreach(fun({_, #consumer{suspected_down = SuspectedDown}}) ->
-                      ?assert(SuspectedDown)
+    lists:foreach(fun({_, #consumer{status = Status}}) ->
+                      ?assert(Status == suspected_down)
                   end, State2#state.waiting_consumers),
 
     % simulate node goes back up
@@ -2441,8 +2444,8 @@ single_active_consumer_mark_waiting_consumers_as_suspected_when_down_noconnnecti
 
     % all the waiting consumers should be un-suspected
     ?assertEqual(3, length(State3#state.waiting_consumers)),
-    lists:foreach(fun({_, #consumer{suspected_down = SuspectedDown}}) ->
-                      ?assertNot(SuspectedDown)
+    lists:foreach(fun({_, #consumer{status = Status}}) ->
+                      ?assert(Status /= suspected_down)
                   end, State3#state.waiting_consumers),
 
     ok.
@@ -2527,7 +2530,8 @@ query_consumers_test() ->
     State1 = lists:foldl(AddConsumer, State0, [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
     Consumers0 = State1#state.consumers,
     Consumer = maps:get({<<"ctag2">>, self()}, Consumers0),
-    Consumers1 = maps:put({<<"ctag2">>, self()}, Consumer#consumer{suspected_down = true}, Consumers0),
+    Consumers1 = maps:put({<<"ctag2">>, self()},
+                          Consumer#consumer{status = suspected_down}, Consumers0),
     State2 = State1#state{consumers = Consumers1},
 
     ?assertEqual(4, query_consumer_count(State2)),