Merge pull request #1950 from rabbitmq/rabbit-fifo-bugfixes

Rabbit fifo bug fixes

3 files changed, 545 insertions, 165 deletions

diff --git a/src/rabbit_fifo.erl b/src/rabbit_fifo.erl
index e43e94a385..fb6f9ce770 100644
--- a/src/rabbit_fifo.erl
+++ b/src/rabbit_fifo.erl
@@ -184,11 +184,9 @@ apply(Meta, #discard{msg_ids = MsgIds, consumer_id = ConsumerId},
 apply(Meta, #return{msg_ids = MsgIds, consumer_id = ConsumerId},
       #?MODULE{consumers = Cons0} = State) ->
     case Cons0 of
-        #{ConsumerId := Con0 = #consumer{checked_out = Checked0}} ->
-            Checked = maps:without(MsgIds, Checked0),
+        #{ConsumerId := #consumer{checked_out = Checked0}} ->
             Returned = maps:with(MsgIds, Checked0),
-            MsgNumMsgs = maps:values(Returned),
-            return(Meta, ConsumerId, MsgNumMsgs, Con0, Checked, [], State);
+            return(Meta, ConsumerId, Returned, [], State);
         _ ->
             {State, ok}
     end;
@@ -310,7 +308,7 @@ apply(#{index := RaftIdx}, #purge{},
     %% reverse the effects ourselves
     {State, {purge, Total},
      lists:reverse([garbage_collection | Effects])};
-apply(_, {down, Pid, noconnection},
+apply(Meta, {down, Pid, noconnection},
       #?MODULE{consumers = Cons0,
                cfg = #cfg{consumer_strategy = single_active},
                waiting_consumers = Waiting0,
@@ -318,17 +316,30 @@ apply(_, {down, Pid, noconnection},
     Node = node(Pid),
     %% if the pid refers to the active consumer, mark it as suspected and return
     %% it to the waiting queue
-    {State1, Effects0} = case maps:to_list(Cons0) of
-                 [{{_, P} = Cid, C}] when node(P) =:= Node ->
-                     %% the consumer should be returned to waiting
-                     %%
-                     Effs = consumer_update_active_effects(
-                             State0, Cid, C, false, suspected_down, []),
-                     {State0#?MODULE{consumers = #{},
-                                    waiting_consumers = Waiting0 ++ [{Cid, C}]},
-                      Effs};
-                 _ -> {State0, []}
-             end,
+    {State1, Effects0} =
+        case maps:to_list(Cons0) of
+            [{{_, P} = Cid, C0}] when node(P) =:= Node ->
+                %% the consumer should be returned to waiting
+                %% and checked out messages should be returned
+                Effs = consumer_update_active_effects(
+                         State0, Cid, C0, false, suspected_down, []),
+                Checked = C0#consumer.checked_out,
+                Credit = increase_credit(C0, maps:size(Checked)),
+                {St, Effs1} = return_all(State0, Effs,
+                                         Cid, C0#consumer{credit = Credit}),
+                %% if the consumer was cancelled there is a chance it got
+                %% removed when returning hence we need to be defensive here
+                Waiting = case St#?MODULE.consumers of
+                              #{Cid := C} ->
+                                  Waiting0 ++ [{Cid, C}];
+                              _ ->
+                                  Waiting0
+                          end,
+                {St#?MODULE{consumers = #{},
+                            waiting_consumers = Waiting},
+                 Effs1};
+            _ -> {State0, []}
+        end,
     WaitingConsumers = update_waiting_consumer_status(Node, State1,
                                                       suspected_down),
 
@@ -342,8 +353,8 @@ apply(_, {down, Pid, noconnection},
                        (_, E) -> E
                     end, Enqs0),
     Effects = [{monitor, node, Node} | Effects1],
-    {State#?MODULE{enqueuers = Enqs}, ok, Effects};
-apply(_, {down, Pid, noconnection},
+    checkout(Meta, State#?MODULE{enqueuers = Enqs}, Effects);
+apply(Meta, {down, Pid, noconnection},
       #?MODULE{consumers = Cons0,
                enqueuers = Enqs0} = State0) ->
     %% A node has been disconnected. This doesn't necessarily mean that
@@ -355,23 +366,22 @@ apply(_, {down, Pid, noconnection},
     %% all pids for the disconnected node will be marked as suspected not just
     %% the one we got the `down' command for
     Node = node(Pid),
-    ConsumerUpdateActiveFun = consumer_active_flag_update_function(State0),
 
-    {Cons, State, Effects1} =
-        maps:fold(fun({_, P} = K, #consumer{checked_out = Checked0,
-                                            status = up} = C,
-                      {Co, St0, Eff}) when node(P) =:= Node ->
-                          {St, Eff0} = return_all(St0, Checked0, Eff, K, C),
-                          Credit = increase_credit(C, maps:size(Checked0)),
-                          Eff1 = ConsumerUpdateActiveFun(St, K, C, false,
-                                                         suspected_down, Eff0),
-                          {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),
+    {State, Effects1} =
+        maps:fold(
+          fun({_, P} = Cid, #consumer{checked_out = Checked0,
+                                      status = up} = C0,
+              {St0, Eff}) when node(P) =:= Node ->
+                  Credit = increase_credit(C0, map_size(Checked0)),
+                  C = C0#consumer{status = suspected_down,
+                                  credit = Credit},
+                  {St, Eff0} = return_all(St0, Eff, Cid, C),
+                  Eff1 = consumer_update_active_effects(St, Cid, C, false,
+                                                        suspected_down, Eff0),
+                  {St, Eff1};
+             (_, _, {St, Eff}) ->
+                  {St, Eff}
+          end, {State0, []}, Cons0),
     Enqs = maps:map(fun(P, E) when node(P) =:= Node ->
                             E#enqueuer{status = suspected_down};
                        (_, E) -> E
@@ -380,14 +390,14 @@ apply(_, {down, Pid, noconnection},
     % Monitor the node so that we can "unsuspect" these processes when the node
     % comes back, then re-issue all monitors and discover the final fate of
     % these processes
-    Effects2 = case maps:size(Cons) of
-                   0 ->
-                       [{aux, inactive}, {monitor, node, Node}];
-                   _ ->
-                       [{monitor, node, Node}]
-               end ++ Effects1,
+    Effects = case maps:size(State#?MODULE.consumers) of
+                  0 ->
+                      [{aux, inactive}, {monitor, node, Node}];
+                  _ ->
+                      [{monitor, node, Node}]
+              end ++ Effects1,
     %% TODO: should we run a checkout here?
-    {State#?MODULE{consumers = Cons, enqueuers = Enqs}, ok, Effects2};
+    checkout(Meta, State#?MODULE{enqueuers = Enqs}, Effects);
 apply(Meta, {down, Pid, _Info}, #?MODULE{consumers = Cons0,
                                          enqueuers = Enqs0} = State0) ->
     % Remove any enqueuer for the same pid and enqueue any pending messages
@@ -795,9 +805,9 @@ consumer_update_active_effects(#?MODULE{cfg = #cfg{resource = QName}},
       | Effects].
 
 cancel_consumer0(ConsumerId, #?MODULE{consumers = C0} = S0, Effects0, Reason) ->
-    case maps:take(ConsumerId, C0) of
-        {Consumer, Cons1} ->
-            {S, Effects2} = maybe_return_all(ConsumerId, Consumer, Cons1, S0,
+    case C0 of
+        #{ConsumerId := Consumer} ->
+            {S, Effects2} = maybe_return_all(ConsumerId, Consumer, S0,
                                              Effects0, Reason),
             %% The effects are emitted before the consumer is actually removed
             %% if the consumer has unacked messages. This is a bit weird but
@@ -810,7 +820,7 @@ cancel_consumer0(ConsumerId, #?MODULE{consumers = C0} = S0, Effects0, Reason) ->
                 _ ->
                     {S, Effects}
             end;
-        error ->
+        _ ->
             %% already removed: do nothing
             {S0, Effects0}
     end.
@@ -847,9 +857,9 @@ activate_next_consumer(#?MODULE{consumers = Cons,
 
 
 
-maybe_return_all(ConsumerId, #consumer{checked_out = Checked0} = Consumer,
-                 Cons1, #?MODULE{consumers = C0,
-                                 service_queue = SQ0} = S0,
+maybe_return_all(ConsumerId, Consumer,
+                 #?MODULE{consumers = C0,
+                          service_queue = SQ0} = S0,
                  Effects0, Reason) ->
     case Reason of
         consumer_cancel ->
@@ -859,11 +869,12 @@ maybe_return_all(ConsumerId, #consumer{checked_out = Checked0} = Consumer,
                                                        credit = 0,
                                                        status = cancelled},
                                      C0, SQ0, Effects0),
-            {S0#?MODULE{consumers = Cons, service_queue = SQ}, Effects1};
+            {S0#?MODULE{consumers = Cons,
+                        service_queue = SQ}, Effects1};
         down ->
-            {S1, Effects1} = return_all(S0, Checked0, Effects0, ConsumerId,
-                                        Consumer),
-            {S1#?MODULE{consumers = Cons1}, Effects1}
+            {S1, Effects1} = return_all(S0, Effects0, ConsumerId, Consumer),
+            {S1#?MODULE{consumers = maps:remove(ConsumerId, S1#?MODULE.consumers)},
+             Effects1}
     end.
 
 apply_enqueue(#{index := RaftIdx} = Meta, From, Seq, RawMsg, State0) ->
@@ -980,38 +991,43 @@ maybe_enqueue(RaftIdx, From, MsgSeqNo, RawMsg, Effects0,
 snd(T) ->
     element(2, T).
 
-return(Meta, ConsumerId, MsgNumMsgs, Con0, Checked,
-       Effects0, #?MODULE{consumers = Cons0, service_queue = SQ0} = State0) ->
-    Con = Con0#consumer{checked_out = Checked,
-                        credit = increase_credit(Con0, length(MsgNumMsgs))},
-    {Cons, SQ, Effects1} = update_or_remove_sub(ConsumerId, Con, Cons0,
-                                               SQ0, Effects0),
-    {State1, Effects2} = lists:foldl(
-                           fun({'$prefix_msg', _} = Msg, {S0, E0}) ->
-                                   return_one(0, Msg, S0, E0,
-                                              ConsumerId, Con);
-                              ({MsgNum, Msg}, {S0, E0}) ->
-                                   return_one(MsgNum, Msg, S0, E0,
-                                              ConsumerId, Con)
-                           end, {State0, Effects1}, MsgNumMsgs),
-    checkout(Meta, State1#?MODULE{consumers = Cons,
-                                  service_queue = SQ},
-             Effects2).
+return(Meta, ConsumerId, Returned,
+       Effects0, #?MODULE{service_queue = SQ0} = State0) ->
+    {State1, Effects1} = maps:fold(
+                          fun(MsgId, {'$prefix_msg', _} = Msg, {S0, E0}) ->
+                                  return_one(MsgId, 0, Msg, S0, E0, ConsumerId);
+                             (MsgId, {MsgNum, Msg}, {S0, E0}) ->
+                                  return_one(MsgId, MsgNum, Msg, S0, E0,
+                                             ConsumerId)
+                          end, {State0, Effects0}, Returned),
+    #{ConsumerId := Con0} = Cons0 = State1#?MODULE.consumers,
+    Con = Con0#consumer{credit = increase_credit(Con0, map_size(Returned))},
+    {Cons, SQ, Effects2} = update_or_remove_sub(ConsumerId, Con, Cons0,
+                                                SQ0, Effects1),
+    State = State1#?MODULE{consumers = Cons,
+                           service_queue = SQ},
+    checkout(Meta, State, Effects2).
 
 % used to processes messages that are finished
-complete(ConsumerId, MsgRaftIdxs, NumDiscarded,
-         Con0, Checked, Effects0,
+complete(ConsumerId, Discarded,
+         #consumer{checked_out = Checked} = Con0, Effects0,
          #?MODULE{consumers = Cons0, service_queue = SQ0,
                   ra_indexes = Indexes0} = State0) ->
+    MsgRaftIdxs = [RIdx || {_, {RIdx, _}} <- maps:values(Discarded)],
     %% credit_mode = simple_prefetch should automatically top-up credit
     %% as messages are simple_prefetch or otherwise returned
-    Con = Con0#consumer{checked_out = Checked,
-                        credit = increase_credit(Con0, NumDiscarded)},
+    Con = Con0#consumer{checked_out = maps:without(maps:keys(Discarded), Checked),
+                        credit = increase_credit(Con0, maps:size(Discarded))},
     {Cons, SQ, Effects} = update_or_remove_sub(ConsumerId, Con, Cons0,
                                                SQ0, Effects0),
     Indexes = lists:foldl(fun rabbit_fifo_index:delete/2, Indexes0,
                           MsgRaftIdxs),
-    {State0#?MODULE{consumers = Cons,
+    State1 = lists:foldl(fun({_, {_, {_, RawMsg}}}, Acc) ->
+                                 add_bytes_settle(RawMsg, Acc);
+                            ({'$prefix_msg', _} = M, Acc) ->
+                                 add_bytes_settle(M, Acc)
+                         end, State0, maps:values(Discarded)),
+    {State1#?MODULE{consumers = Cons,
                     ra_indexes = Indexes,
                     service_queue = SQ}, Effects}.
 
@@ -1030,19 +1046,9 @@ increase_credit(#consumer{credit = Current}, Credit) ->
 complete_and_checkout(#{index := IncomingRaftIdx} = Meta, MsgIds, ConsumerId,
                       #consumer{checked_out = Checked0} = Con0,
                       Effects0, State0) ->
-    Checked = maps:without(MsgIds, Checked0),
     Discarded = maps:with(MsgIds, Checked0),
-    MsgRaftIdxs = [RIdx || {_, {RIdx, _}} <- maps:values(Discarded)],
-    State1 = lists:foldl(fun({_, {_, {_, RawMsg}}}, Acc) ->
-                                 add_bytes_settle(RawMsg, Acc);
-                            ({'$prefix_msg', _} = M, Acc) ->
-                                 add_bytes_settle(M, Acc)
-                         end, State0, maps:values(Discarded)),
-    %% need to pass the length of discarded as $prefix_msgs would be filtered
-    %% by the above list comprehension
-    {State2, Effects1} = complete(ConsumerId, MsgRaftIdxs,
-                                  maps:size(Discarded),
-                                  Con0, Checked, Effects0, State1),
+    {State2, Effects1} = complete(ConsumerId, Discarded, Con0,
+                                  Effects0, State0),
     {State, ok, Effects} = checkout(Meta, State2, Effects1),
     % settle metrics are incremented separately
     update_smallest_raft_index(IncomingRaftIdx, State, Effects).
@@ -1102,69 +1108,70 @@ find_next_cursor(Smallest, Cursors0, Potential) ->
             {Potential, Cursors0}
     end.
 
-return_one(0, {'$prefix_msg', Header0},
+return_one(MsgId, 0, {'$prefix_msg', Header0},
            #?MODULE{returns = Returns,
+                    consumers = Consumers,
                     cfg = #cfg{delivery_limit = DeliveryLimit}} = State0,
-           Effects0, ConsumerId, Con) ->
-    Header = maps:update_with(delivery_count,
-                              fun (C) -> C+1 end,
+           Effects0, ConsumerId) ->
+    #consumer{checked_out = Checked} = Con0 = maps:get(ConsumerId, Consumers),
+    Header = maps:update_with(delivery_count, fun (C) -> C+1 end,
                               1, Header0),
     Msg = {'$prefix_msg', Header},
     case maps:get(delivery_count, Header) of
         DeliveryCount when DeliveryCount > DeliveryLimit ->
-            Checked = Con#consumer.checked_out,
-            {State1, Effects} = complete(ConsumerId, [], 1, Con, Checked,
-                                          Effects0, State0),
-            {add_bytes_settle(Msg, State1), Effects};
+            complete(ConsumerId, #{MsgId => Msg}, Con0, Effects0, State0);
         _ ->
             %% this should not affect the release cursor in any way
-            {add_bytes_return(Msg,
-                              State0#?MODULE{returns = lqueue:in(Msg, Returns)}),
+            Con = Con0#consumer{checked_out = maps:remove(MsgId, Checked)},
+            {add_bytes_return(
+               Msg,
+               State0#?MODULE{consumers = Consumers#{ConsumerId => Con},
+                              returns = lqueue:in(Msg, Returns)}),
              Effects0}
     end;
-return_one(MsgNum, {RaftId, {Header0, RawMsg}},
+return_one(MsgId, MsgNum, {RaftId, {Header0, RawMsg}},
            #?MODULE{returns = Returns,
+                    consumers = Consumers,
                     cfg = #cfg{delivery_limit = DeliveryLimit}} = State0,
-           Effects0, ConsumerId, Con) ->
-    Header = maps:update_with(delivery_count,
-                              fun (C) -> C+1 end,
+           Effects0, ConsumerId) ->
+    #consumer{checked_out = Checked} = Con0 = maps:get(ConsumerId, Consumers),
+    Header = maps:update_with(delivery_count, fun (C) -> C+1 end,
                               1, Header0),
     Msg = {RaftId, {Header, RawMsg}},
     case maps:get(delivery_count, Header) of
         DeliveryCount when DeliveryCount > DeliveryLimit ->
             DlMsg = {MsgNum, Msg},
-            Effects = dead_letter_effects(delivery_limit,
-                                          #{none => DlMsg},
+            Effects = dead_letter_effects(delivery_limit, #{none => DlMsg},
                                           State0, Effects0),
-            Checked = Con#consumer.checked_out,
-            {State1, Effects1} = complete(ConsumerId, [RaftId], 1, Con, Checked,
-                                          Effects, State0),
-            {add_bytes_settle(RawMsg, State1), Effects1};
+            complete(ConsumerId, #{MsgId => DlMsg}, Con0, Effects, State0);
         _ ->
+            Con = Con0#consumer{checked_out = maps:remove(MsgId, Checked)},
             %% this should not affect the release cursor in any way
-            {add_bytes_return(RawMsg,
-                              State0#?MODULE{returns =
-                                             lqueue:in({MsgNum, Msg}, Returns)}),
+            {add_bytes_return(
+               RawMsg,
+               State0#?MODULE{consumers = Consumers#{ConsumerId => Con},
+                              returns = lqueue:in({MsgNum, Msg}, Returns)}),
              Effects0}
     end.
 
-return_all(State0, Checked0, Effects0, ConsumerId, Consumer) ->
+return_all(#?MODULE{consumers = Cons} = State0, Effects0, ConsumerId,
+           #consumer{checked_out = Checked0} = Con) ->
     %% need to sort the list so that we return messages in the order
     %% they were checked out
     Checked = lists:sort(maps:to_list(Checked0)),
-    lists:foldl(fun ({_, {'$prefix_msg', _} = Msg}, {S, E}) ->
-                        return_one(0, Msg, S, E, ConsumerId, Consumer);
-                    ({_, {MsgNum, Msg}}, {S, E}) ->
-                        return_one(MsgNum, Msg, S, E, ConsumerId, Consumer)
-                end, {State0, Effects0}, Checked).
+    State = State0#?MODULE{consumers = Cons#{ConsumerId => Con}},
+    lists:foldl(fun ({MsgId, {'$prefix_msg', _} = Msg}, {S, E}) ->
+                        return_one(MsgId, 0, Msg, S, E, ConsumerId);
+                    ({MsgId, {MsgNum, Msg}}, {S, E}) ->
+                        return_one(MsgId, MsgNum, Msg, S, E, ConsumerId)
+                end, {State, Effects0}, Checked).
 
 %% checkout new messages to consumers
 %% reverses the effects list
 checkout(#{index := Index}, State0, Effects0) ->
     {State1, _Result, Effects1} = checkout0(checkout_one(State0),
                                             Effects0, #{}),
-    case evaluate_limit(State0#?MODULE.ra_indexes, false,
-                        State1, Effects1) of
+    case evaluate_limit(false, State1, Effects1) of
         {State, true, Effects} ->
             update_smallest_raft_index(Index, State, Effects);
         {State, false, Effects} ->
@@ -1187,17 +1194,16 @@ checkout0({Activity, State0}, Effects0, Acc) ->
                end,
     {State0, ok, lists:reverse(Effects1)}.
 
-evaluate_limit(_OldIndexes, Result,
+evaluate_limit(Result,
                #?MODULE{cfg = #cfg{max_length = undefined,
                                    max_bytes = undefined}} = State,
                Effects) ->
     {State, Result, Effects};
-evaluate_limit(OldIndexes, Result,
-               State0, Effects0) ->
+evaluate_limit(Result, State0, Effects0) ->
     case is_over_limit(State0) of
         true ->
             {State, Effects} = drop_head(State0, Effects0),
-            evaluate_limit(OldIndexes, true, State, Effects);
+            evaluate_limit(true, State, Effects);
         false ->
             {State0, Result, Effects0}
     end.
diff --git a/test/rabbit_fifo_SUITE.erl b/test/rabbit_fifo_SUITE.erl
index cf14a68d9d..6582104708 100644
--- a/test/rabbit_fifo_SUITE.erl
+++ b/test/rabbit_fifo_SUITE.erl
@@ -283,6 +283,19 @@ duplicate_enqueue_test(_) ->
     ?assertNoEffect({send_msg, _, {delivery, _, [{_, {_, first}}]}, _}, Effects3),
     ok.
 
+return_test(_) ->
+    Cid = {<<"cid">>, self()},
+    Cid2 = {<<"cid2">>, self()},
+    {State0, _} = enq(1, 1, msg, test_init(test)),
+    {State1, _} = check_auto(Cid, 2, State0),
+    {State2, _} = check_auto(Cid2, 3, State1),
+    {State3, _, _} = apply(meta(4), rabbit_fifo:make_return(Cid, [0]), State2),
+    ?assertMatch(#{Cid := #consumer{checked_out = C}} when map_size(C) == 0,
+                                                           State3#rabbit_fifo.consumers),
+    ?assertMatch(#{Cid2 := #consumer{checked_out = C2}} when map_size(C2) == 1,
+                                                           State3#rabbit_fifo.consumers),
+    ok.
+
 return_non_existent_test(_) ->
     Cid = {<<"cid">>, self()},
     {State0, [_, _Inactive]} = enq(1, 1, second, test_init(test)),
@@ -382,14 +395,18 @@ down_with_noconnection_marks_suspect_and_node_is_monitored_test(_) ->
     % monitor both enqueuer and consumer
     % because we received a noconnection we now need to monitor the node
     {State2a, _, _} = apply(meta(3), {down, Pid, noconnection}, State1),
-    #consumer{credit = 1} = maps:get(Cid, State2a#rabbit_fifo.consumers),
+    #consumer{credit = 1,
+              checked_out = Ch,
+              status = suspected_down} = maps:get(Cid, State2a#rabbit_fifo.consumers),
+    ?assertEqual(#{}, Ch),
     %% validate consumer has credit
     {State2, _, Effects2} = apply(meta(3), {down, Self, noconnection}, State2a),
     ?ASSERT_EFF({monitor, node, _}, Effects2),
     ?assertNoEffect({demonitor, process, _}, Effects2),
     % when the node comes up we need to retry the process monitors for the
     % disconnected processes
-    {_State3, _, Effects3} = apply(meta(3), {nodeup, Node}, State2),
+    {State3, _, Effects3} = apply(meta(3), {nodeup, Node}, State2),
+    #consumer{status = up} = maps:get(Cid, State3#rabbit_fifo.consumers),
     % try to re-monitor the suspect processes
     ?ASSERT_EFF({monitor, process, P}, P =:= Pid, Effects3),
     ?ASSERT_EFF({monitor, process, P}, P =:= Self, Effects3),
@@ -407,6 +424,10 @@ down_with_noconnection_returns_unack_test(_) ->
     {State2a, _, _} = apply(meta(3), {down, Pid, noconnection}, State1),
     ?assertEqual(0, maps:size(State2a#rabbit_fifo.messages)),
     ?assertEqual(1, lqueue:len(State2a#rabbit_fifo.returns)),
+    ?assertMatch(#consumer{checked_out = Ch,
+                           status = suspected_down}
+                   when map_size(Ch) == 0,
+                        maps:get(Cid, State2a#rabbit_fifo.consumers)),
     ok.
 
 down_with_noproc_enqueuer_is_cleaned_up_test(_) ->
@@ -426,7 +447,8 @@ discarded_message_without_dead_letter_handler_is_removed_test(_) ->
     ?ASSERT_EFF({send_msg, _,
                  {delivery, _, [{0, {#{}, first}}]}, _},
                 Effects1),
-    {_State2, _, Effects2} = apply(meta(1), rabbit_fifo:make_discard(Cid, [0]), State1),
+    {_State2, _, Effects2} = apply(meta(1),
+                                   rabbit_fifo:make_discard(Cid, [0]), State1),
     ?assertNoEffect({send_msg, _,
                      {delivery, _, [{0, {#{}, first}}]}, _},
                     Effects2),
@@ -556,7 +578,7 @@ purge_with_checkout_test(_) ->
     ?assertEqual(1, maps:size(Checked)),
     ok.
 
-down_returns_checked_out_in_order_test(_) ->
+down_noproc_returns_checked_out_in_order_test(_) ->
     S0 = test_init(?FUNCTION_NAME),
     %% enqueue 100
     S1 = lists:foldl(fun (Num, FS0) ->
@@ -572,6 +594,30 @@ down_returns_checked_out_in_order_test(_) ->
     {S, _, _} = apply(meta(102), {down, self(), noproc}, S2),
     Returns = lqueue:to_list(S#rabbit_fifo.returns),
     ?assertEqual(100, length(Returns)),
+    ?assertEqual(0, maps:size(S#rabbit_fifo.consumers)),
+    %% validate returns are in order
+    ?assertEqual(lists:sort(Returns), Returns),
+    ok.
+
+down_noconnection_returns_checked_out_test(_) ->
+    S0 = test_init(?FUNCTION_NAME),
+    NumMsgs = 20,
+    S1 = lists:foldl(fun (Num, FS0) ->
+                         {FS, _} = enq(Num, Num, Num, FS0),
+                         FS
+                     end, S0, lists:seq(1, NumMsgs)),
+    ?assertEqual(NumMsgs, maps:size(S1#rabbit_fifo.messages)),
+    Cid = {<<"cid">>, self()},
+    {S2, _} = check(Cid, 101, 1000, S1),
+    #consumer{checked_out = Checked} = maps:get(Cid, S2#rabbit_fifo.consumers),
+    ?assertEqual(NumMsgs, maps:size(Checked)),
+    %% simulate down
+    {S, _, _} = apply(meta(102), {down, self(), noconnection}, S2),
+    Returns = lqueue:to_list(S#rabbit_fifo.returns),
+    ?assertEqual(NumMsgs, length(Returns)),
+    ?assertMatch(#consumer{checked_out = Ch}
+                   when map_size(Ch) == 0,
+                        maps:get(Cid, S#rabbit_fifo.consumers)),
     %% validate returns are in order
     ?assertEqual(lists:sort(Returns), Returns),
     ok.
@@ -736,6 +782,41 @@ single_active_consumer_cancel_consumer_when_channel_is_down_test(_) ->
 
     ok.
 
+single_active_returns_messages_on_noconnection_test(_) ->
+    R = rabbit_misc:r("/", queue, atom_to_binary(?FUNCTION_NAME, utf8)),
+    State0 = init(#{name => ?FUNCTION_NAME,
+                    queue_resource => R,
+                    release_cursor_interval => 0,
+                    single_active_consumer_on => true}),
+    Meta = #{index => 1},
+    Nodes = [n1],
+    ConsumerIds = [{_, DownPid}] =
+        [begin
+             B = atom_to_binary(N, utf8),
+             {<<"ctag_", B/binary>>,
+              test_util:fake_pid(N)}
+         end || N <- Nodes],
+    % adding some consumers
+    State1 = lists:foldl(
+               fun(CId, Acc0) ->
+                       {Acc, _, _} =
+                           apply(Meta,
+                                 make_checkout(CId,
+                                               {once, 1, simple_prefetch}, #{}),
+                                 Acc0),
+                       Acc
+               end, State0, ConsumerIds),
+    {State2, _} = enq(4, 1, msg1, State1),
+    % simulate node goes down
+    {State3, _, _} = apply(meta(5), {down, DownPid, noconnection}, State2),
+    %% assert the consumer is up
+    ?assertMatch([_], lqueue:to_list(State3#rabbit_fifo.returns)),
+    ?assertMatch([{_, #consumer{checked_out = Checked}}]
+                 when map_size(Checked) == 0,
+                      State3#rabbit_fifo.waiting_consumers),
+
+    ok.
+
 single_active_consumer_replaces_consumer_when_down_noconnection_test(_) ->
     R = rabbit_misc:r("/", queue, atom_to_binary(?FUNCTION_NAME, utf8)),
     State0 = init(#{name => ?FUNCTION_NAME,
@@ -751,26 +832,30 @@ single_active_consumer_replaces_consumer_when_down_noconnection_test(_) ->
           test_util:fake_pid(N)}
      end || N <- Nodes],
     % adding some consumers
-    State1 = lists:foldl(
-               fun(CId, Acc0) ->
-                       {Acc, _, _} =
-                       apply(Meta,
-                             make_checkout(CId,
-                                           {once, 1, simple_prefetch}, #{}),
-                             Acc0),
-                       Acc
-               end, State0, ConsumerIds),
+    State1a = lists:foldl(
+                fun(CId, Acc0) ->
+                        {Acc, _, _} =
+                        apply(Meta,
+                              make_checkout(CId,
+                                            {once, 1, simple_prefetch}, #{}),
+                              Acc0),
+                        Acc
+                end, State0, ConsumerIds),
 
     %% assert the consumer is up
     ?assertMatch(#{C1 := #consumer{status = up}},
-                 State1#rabbit_fifo.consumers),
+                 State1a#rabbit_fifo.consumers),
+
+    {State1, _} = enq(10, 1, msg, State1a),
 
     % simulate node goes down
     {State2, _, _} = apply(meta(5), {down, DownPid, noconnection}, State1),
 
     %% assert a new consumer is in place and it is up
-    ?assertMatch([{C2, #consumer{status = up}}],
-                 maps:to_list(State2#rabbit_fifo.consumers)),
+    ?assertMatch([{C2, #consumer{status = up,
+                                 checked_out = Ch}}]
+                   when map_size(Ch) == 1,
+                        maps:to_list(State2#rabbit_fifo.consumers)),
 
     %% the disconnected consumer has been returned to waiting
     ?assert(lists:any(fun ({C,_}) -> C =:= C1 end,
@@ -990,11 +1075,11 @@ active_flag_updated_when_consumer_suspected_unsuspected_test(_) ->
     State1 = lists:foldl(AddConsumer, State0,
         [{<<"ctag1">>, Pid1}, {<<"ctag2">>, Pid2}, {<<"ctag3">>, Pid2}, {<<"ctag4">>, Pid3}]),
 
-    {State2, _, Effects2} = apply(#{}, {down, Pid1, noconnection}, State1),
+    {State2, _, Effects2} = apply(#{index => 3}, {down, Pid1, noconnection}, State1),
     % 1 effect to update the metrics of each consumer (they belong to the same node), 1 more effect to monitor the node
     ?assertEqual(4 + 1, length(Effects2)),
 
-    {_, _, Effects3} = apply(#{index => 1}, {nodeup, node(self())}, State2),
+    {_, _, Effects3} = apply(#{index => 4}, {nodeup, node(self())}, State2),
     % for each consumer: 1 effect to update the metrics, 1 effect to monitor the consumer PID
     ?assertEqual(4 + 4, length(Effects3)).
 
@@ -1023,11 +1108,11 @@ active_flag_not_updated_when_consumer_suspected_unsuspected_and_single_active_co
                          [{<<"ctag1">>, Pid1}, {<<"ctag2">>, Pid2},
                           {<<"ctag3">>, Pid2}, {<<"ctag4">>, Pid3}]),
 
-    {State2, _, Effects2} = apply(#{}, {down, Pid1, noconnection}, State1),
+    {State2, _, Effects2} = apply(#{index => 2}, {down, Pid1, noconnection}, State1),
     % one monitor and one consumer status update (deactivated)
     ?assertEqual(3, length(Effects2)),
 
-    {_, _, Effects3} = apply(#{index => 1}, {nodeup, node(self())}, State2),
+    {_, _, Effects3} = apply(#{index => 3}, {nodeup, node(self())}, State2),
     % for each consumer: 1 effect to monitor the consumer PID
     ?assertEqual(5, length(Effects3)).
 
diff --git a/test/rabbit_fifo_prop_SUITE.erl b/test/rabbit_fifo_prop_SUITE.erl
index 76b8d7e715..cd6598fe33 100644
--- a/test/rabbit_fifo_prop_SUITE.erl
+++ b/test/rabbit_fifo_prop_SUITE.erl
@@ -8,6 +8,8 @@
 -include_lib("proper/include/proper.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("ra/include/ra.hrl").
+-include("src/rabbit_fifo.hrl").
 
 %%%===================================================================
 %%% Common Test callbacks
@@ -21,6 +23,7 @@ all() ->
 
 all_tests() ->
     [
+     test_run_log,
      snapshots,
      scenario1,
      scenario2,
@@ -38,7 +41,14 @@ all_tests() ->
      scenario14,
      scenario15,
      scenario16,
-     scenario17
+     scenario17,
+     single_active,
+     single_active_01,
+     single_active_02,
+     single_active_03,
+     single_active_ordering,
+     single_active_ordering_01
+     % single_active_ordering_02
     ].
 
 groups() ->
@@ -285,7 +295,9 @@ scenario17(_Config) ->
                 make_checkout(C2, cancel),
                 make_enqueue(E,2,<<"two">>),
                 {nodeup,rabbit@fake_node1},
+                %% this has no effect as was returned
                 make_settle(C1, [0]),
+                %% this should settle "one"
                 make_settle(C1, [1])
                 ],
     run_snapshot_test(#{name => ?FUNCTION_NAME,
@@ -293,6 +305,82 @@ scenario17(_Config) ->
                        }, Commands),
     ok.
 
+single_active_01(_Config) ->
+    C1Pid = test_util:fake_pid(rabbit@fake_node1),
+    C1 = {<<0>>, C1Pid},
+    C2Pid = test_util:fake_pid(rabbit@fake_node2),
+    C2 = {<<>>, C2Pid},
+    E = test_util:fake_pid(rabbit@fake_node2),
+    Commands = [
+                make_checkout(C1, {auto,1,simple_prefetch}),
+                make_enqueue(E,1,<<"one">>),
+                make_checkout(C2, {auto,1,simple_prefetch}),
+                make_checkout(C1, cancel),
+                {nodeup,rabbit@fake_node1}
+                ],
+    ?assert(
+       single_active_prop(#{name => ?FUNCTION_NAME,
+                            single_active_consumer_on => true
+                       }, Commands, false)),
+    ok.
+
+single_active_02(_Config) ->
+    C1Pid = test_util:fake_pid(node()),
+    C1 = {<<0>>, C1Pid},
+    C2Pid = test_util:fake_pid(node()),
+    C2 = {<<>>, C2Pid},
+    E = test_util:fake_pid(node()),
+    Commands = [
+                make_checkout(C1, {auto,1,simple_prefetch}),
+                make_enqueue(E,1,<<"one">>),
+                {down,E,noconnection},
+                make_checkout(C2, {auto,1,simple_prefetch}),
+                make_checkout(C2, cancel),
+                {down,E,noconnection}
+                ],
+    Conf = config(?FUNCTION_NAME, undefined, undefined, true, 1),
+    ?assert(single_active_prop(Conf, Commands, false)),
+    ok.
+
+single_active_03(_Config) ->
+    C1Pid = test_util:fake_pid(node()),
+    C1 = {<<0>>, C1Pid},
+    % C2Pid = test_util:fake_pid(rabbit@fake_node2),
+    % C2 = {<<>>, C2Pid},
+    Pid = test_util:fake_pid(node()),
+    E = test_util:fake_pid(rabbit@fake_node2),
+    Commands = [
+                make_checkout(C1, {auto,2,simple_prefetch}),
+                make_enqueue(E, 1, 0),
+                make_enqueue(E, 2, 1),
+                {down, Pid, noconnection},
+                {nodeup, node()}
+                ],
+    Conf = config(?FUNCTION_NAME, 0, 0, true, 0),
+    ?assert(single_active_prop(Conf, Commands, true)),
+    ok.
+
+test_run_log(_Config) ->
+    Fun = {-1, fun ({Prev, _}) -> {Prev + 1, Prev + 1} end},
+    run_proper(
+      fun () ->
+              ?FORALL({Length, Bytes, SingleActiveConsumer, DeliveryLimit},
+                      frequency([{10, {0, 0, false, 0}},
+                                 {5, {oneof([range(1, 10), undefined]),
+                                      oneof([range(1, 1000), undefined]),
+                                      boolean(),
+                                      oneof([range(1, 3), undefined])
+                                     }}]),
+                      ?FORALL(O, ?LET(Ops, log_gen(100), expand(Ops, Fun)),
+                              collect({log_size, length(O)},
+                                      dump_generated(
+                                        config(?FUNCTION_NAME,
+                                               Length,
+                                               Bytes,
+                                               SingleActiveConsumer,
+                                               DeliveryLimit), O))))
+      end, [], 10).
+
 snapshots(_Config) ->
     run_proper(
       fun () ->
@@ -313,6 +401,84 @@ snapshots(_Config) ->
                                                DeliveryLimit), O))))
       end, [], 2500).
 
+single_active(_Config) ->
+    Size = 2000,
+    run_proper(
+      fun () ->
+              ?FORALL({Length, Bytes, DeliveryLimit},
+                      frequency([{10, {0, 0, 0}},
+                                 {5, {oneof([range(1, 10), undefined]),
+                                      oneof([range(1, 1000), undefined]),
+                                      oneof([range(1, 3), undefined])
+                                     }}]),
+                      ?FORALL(O, ?LET(Ops, log_gen(Size), expand(Ops)),
+                              collect({log_size, length(O)},
+                                      single_active_prop(
+                                        config(?FUNCTION_NAME,
+                                               Length,
+                                               Bytes,
+                                               true,
+                                               DeliveryLimit), O,
+                                        false))))
+      end, [], Size).
+
+single_active_ordering(_Config) ->
+    Size = 2000,
+    Fun = {-1, fun ({Prev, _}) -> {Prev + 1, Prev + 1} end},
+    run_proper(
+      fun () ->
+              ?FORALL(O, ?LET(Ops, log_gen_ordered(Size), expand(Ops, Fun)),
+                      collect({log_size, length(O)},
+                              single_active_prop(config(?FUNCTION_NAME,
+                                                        undefined,
+                                                        undefined,
+                                                        true,
+                                                        undefined), O,
+                                                 true)))
+      end, [], Size).
+
+single_active_ordering_01(_Config) ->
+% [{enqueue,<0.145.0>,1,0},
+%  {enqueue,<0.145.0>,1,1},
+%  {checkout,{<<>>,<0.148.0>},{auto,1,simple_prefetch},#{ack => true,args => [],prefetch => 1,username => <<117,115,101,114>>}}
+%  {enqueue,<0.140.0>,1,2},
+%  {settle,{<<>>,<0.148.0>},[0]}]
+    C1Pid = test_util:fake_pid(node()),
+    C1 = {<<0>>, C1Pid},
+    E = test_util:fake_pid(rabbit@fake_node2),
+    E2 = test_util:fake_pid(rabbit@fake_node2),
+    Commands = [
+                make_enqueue(E, 1, 0),
+                make_enqueue(E, 2, 1),
+                make_checkout(C1, {auto,2,simple_prefetch}),
+                make_enqueue(E2, 1, 2),
+                make_settle(C1, [0])
+                ],
+    Conf = config(?FUNCTION_NAME, 0, 0, true, 0),
+    ?assert(single_active_prop(Conf, Commands, true)),
+    ok.
+
+single_active_ordering_02(_Config) ->
+    %% this results in the pending enqueue being enqueued and violating
+    %% ordering
+% [{checkout, %   {<<>>,<0.177.0>}, %   {auto,1,simple_prefetch},
+%  {enqueue,<0.172.0>,2,1},
+%  {down,<0.172.0>,noproc},
+%  {settle,{<<>>,<0.177.0>},[0]}]
+    C1Pid = test_util:fake_pid(node()),
+    C1 = {<<0>>, C1Pid},
+    E = test_util:fake_pid(node()),
+    Commands = [
+                make_checkout(C1, {auto,1,simple_prefetch}),
+                make_enqueue(E, 2, 1),
+                %% CANNOT HAPPEN
+                {down,E,noproc},
+                make_settle(C1, [0])
+                ],
+    Conf = config(?FUNCTION_NAME, 0, 0, true, 0),
+    ?assert(single_active_prop(Conf, Commands, true)),
+    ok.
+
 config(Name, Length, Bytes, SingleActive, DeliveryLimit) ->
     #{name => Name,
       max_length => map_max(Length),
@@ -323,6 +489,65 @@ config(Name, Length, Bytes, SingleActive, DeliveryLimit) ->
 map_max(0) -> undefined;
 map_max(N) -> N.
 
+single_active_prop(Conf0, Commands, ValidateOrder) ->
+    Conf = Conf0#{release_cursor_interval => 100},
+    Indexes = lists:seq(1, length(Commands)),
+    Entries = lists:zip(Indexes, Commands),
+    %% invariant: there can only be one active consumer at any one time
+    %% there can however be multiple cancelled consumers
+    Invariant = fun (#rabbit_fifo{consumers = Consumers}) ->
+                        Up = maps:filter(fun (_, #consumer{status = S}) ->
+                                                 S == up
+                                         end, Consumers),
+                        map_size(Up) =< 1
+                end,
+    try run_log(test_init(Conf), Entries, Invariant) of
+        {_State, Effects} when ValidateOrder ->
+            %% validate message ordering
+            lists:foldl(fun ({send_msg, Pid, {delivery, Tag, Msgs}, ra_event},
+                             Acc) ->
+                                validate_msg_order({Tag, Pid}, Msgs, Acc);
+                            (_, Acc) ->
+                                Acc
+                        end, -1, Effects),
+            true;
+        _ ->
+            true
+    catch
+        Err ->
+            ct:pal("Commands: ~p~nConf~p~n", [Commands, Conf]),
+            ct:pal("Err: ~p~n", [Err]),
+            false
+    end.
+
+%% single active consumer ordering invariant:
+%% only redelivered messages can go backwards
+validate_msg_order(_, [], S) ->
+    S;
+validate_msg_order(Cid, [{_, {H, Num}} | Rem], PrevMax) ->
+    Redelivered = maps:is_key(delivery_count, H),
+    case undefined of
+        _ when Num == PrevMax + 1 ->
+            %% forwards case
+            validate_msg_order(Cid, Rem, Num);
+        _ when Redelivered andalso Num =< PrevMax ->
+            %% the seq is lower but this is a redelivery
+            %% when the consumer changed and the next messages has been redelivered
+            %% we may go backwards but keep the highest seen
+            validate_msg_order(Cid, Rem, PrevMax);
+        _ ->
+            ct:pal("out of order ~w Prev ~w Curr ~w Redel ~w",
+                   [Cid, PrevMax, Num, Redelivered]),
+            throw({outoforder, Cid, PrevMax, Num})
+    end.
+
+
+
+
+dump_generated(Conf, Commands) ->
+    ct:pal("Commands: ~p~nConf~p~n", [Commands, Conf]),
+    true.
+
 snapshots_prop(Conf, Commands) ->
     try run_snapshot_test(Conf, Commands) of
         _ -> true
@@ -334,6 +559,9 @@ snapshots_prop(Conf, Commands) ->
     end.
 
 log_gen(Size) ->
+    log_gen(Size, binary()).
+
+log_gen(Size, _Body) ->
     Nodes = [node(),
              fakenode@fake,
              fakenode@fake2
@@ -356,6 +584,31 @@ log_gen(Size) ->
                           {1, purge}
                          ]))))).
 
+log_gen_ordered(Size) ->
+    Nodes = [node(),
+             fakenode@fake,
+             fakenode@fake2
+            ],
+    ?LET(EPids, vector(1, pid_gen(Nodes)),
+         ?LET(CPids, vector(5, pid_gen(Nodes)),
+              resize(Size,
+                     list(
+                       frequency(
+                         [{20, enqueue_gen(oneof(EPids), 10, 0)},
+                          {40, {input_event,
+                                frequency([{10, settle},
+                                           {2, return},
+                                           {1, discard},
+                                           {1, requeue}])}},
+                          {2, checkout_gen(oneof(CPids))},
+                          {1, checkout_cancel_gen(oneof(CPids))},
+                          {1, down_gen(oneof(EPids ++ CPids))},
+                          {1, nodeup_gen(Nodes)}
+                         ]))))).
+
+monotonic_gen() ->
+    ?LET(_, integer(), erlang:unique_integer([positive, monotonic])).
+
 pid_gen(Nodes) ->
     ?LET(Node, oneof(Nodes),
          test_util:fake_pid(atom_to_binary(Node, utf8))).
@@ -367,9 +620,12 @@ nodeup_gen(Nodes) ->
     {nodeup, oneof(Nodes)}.
 
 enqueue_gen(Pid) ->
+    enqueue_gen(Pid, 10, 1).
+
+enqueue_gen(Pid, Enq, Del) ->
     ?LET(E, {enqueue, Pid,
-             frequency([{10, enqueue},
-                        {1, delay}]),
+             frequency([{Enq, enqueue},
+                        {Del, delay}]),
              binary()}, E).
 
 checkout_cancel_gen(Pid) ->
@@ -388,16 +644,21 @@ checkout_gen(Pid) ->
             enqueuers = #{} :: #{pid() => term()},
             consumers = #{} :: #{{binary(), pid()} => term()},
             effects = queue:new() :: queue:queue(),
+            %% to transform the body
+            enq_body_fun = {0, fun ra_lib:id/1},
             log = [] :: list(),
             down = #{} :: #{pid() => noproc | noconnection}
            }).
 
 expand(Ops) ->
+    expand(Ops, {undefined, fun ra_lib:id/1}).
+
+expand(Ops, EnqFun) ->
     %% execute each command against a rabbit_fifo state and capture all relevant
     %% effects
-    T = #t{},
+    T = #t{enq_body_fun = EnqFun},
     #t{effects = Effs} = T1 = lists:foldl(fun handle_op/2, T, Ops),
-    %% process the remaining effects
+    %% process the remaining effect
     #t{log = Log} = lists:foldl(fun do_apply/2,
                                 T1#t{effects = queue:new()},
                                 queue:to_list(Effs)),
@@ -407,6 +668,7 @@ expand(Ops) ->
 
 handle_op({enqueue, Pid, When, Data},
           #t{enqueuers = Enqs0,
+             enq_body_fun = {EnqSt0, Fun},
              down = Down,
              effects = Effs} = T) ->
     case Down of
@@ -417,13 +679,17 @@ handle_op({enqueue, Pid, When, Data},
         _ ->
             Enqs = maps:update_with(Pid, fun (Seq) -> Seq + 1 end, 1, Enqs0),
             MsgSeq = maps:get(Pid, Enqs),
-            Cmd = rabbit_fifo:make_enqueue(Pid, MsgSeq, Data),
+            {EnqSt, Msg} = Fun({EnqSt0, Data}),
+            Cmd = rabbit_fifo:make_enqueue(Pid, MsgSeq, Msg),
             case When of
                 enqueue ->
-                    do_apply(Cmd, T#t{enqueuers = Enqs});
+                    do_apply(Cmd, T#t{enqueuers = Enqs,
+                                      enq_body_fun = {EnqSt, Fun}});
                 delay ->
                     %% just put the command on the effects queue
-                    T#t{effects = queue:in(Cmd, Effs)}
+                    T#t{effects = queue:in(Cmd, Effs),
+                        enqueuers = Enqs,
+                        enq_body_fun = {EnqSt, Fun}}
             end
     end;
 handle_op({checkout, Pid, cancel}, #t{consumers  = Cons0} = T) ->
@@ -483,8 +749,8 @@ handle_op({input_event, Settlement}, #t{effects = Effs,
                       discard -> rabbit_fifo:make_discard(CId, MsgIds)
                   end,
             do_apply(Cmd, T#t{effects = Q});
-        {{value, Cmd}, Q} when element(1, Cmd) =:= enqueue ->
-            case maps:is_key(element(2, Cmd), Down) of
+        {{value, {enqueue, Pid, _, _} = Cmd}, Q} ->
+            case maps:is_key(Pid, Down) of
                 true ->
                     %% enqueues cannot arrive after down for the same process
                     %% drop message
@@ -498,21 +764,30 @@ handle_op({input_event, Settlement}, #t{effects = Effs,
 handle_op(purge, T) ->
     do_apply(rabbit_fifo:make_purge(), T).
 
-do_apply(Cmd, #t{effects = Effs, index = Index, state = S0,
+
+do_apply(Cmd, #t{effects = Effs,
+                 index = Index, state = S0,
+                 down = Down,
                  log = Log} = T) ->
-    {St, Effects} = case rabbit_fifo:apply(#{index => Index}, Cmd, S0) of
-                       {S, _, E} when is_list(E) ->
-                           {S, E};
-                       {S, _, E} ->
-                           {S, [E]};
-                       {S, _} ->
-                           {S, []}
-                   end,
-
-    T#t{state = St,
-        index = Index + 1,
-        effects = enq_effs(Effects, Effs),
-        log = [Cmd | Log]}.
+    case Cmd of
+        {enqueue, Pid, _, _} when is_map_key(Pid, Down) ->
+            %% down
+            T;
+        _ ->
+            {St, Effects} = case rabbit_fifo:apply(#{index => Index}, Cmd, S0) of
+                                {S, _, E} when is_list(E) ->
+                                    {S, E};
+                                {S, _, E} ->
+                                    {S, [E]};
+                                {S, _} ->
+                                    {S, []}
+                            end,
+
+            T#t{state = St,
+                index = Index + 1,
+                effects = enq_effs(Effects, Effs),
+                log = [Cmd | Log]}
+    end.
 
 enq_effs([], Q) -> Q;
 enq_effs([{send_msg, P, {delivery, CTag, Msgs}, ra_event} | Rem], Q) ->
@@ -580,13 +855,27 @@ prefixes(Source, N, Acc) ->
     prefixes(Source, N+1, [X | Acc]).
 
 run_log(InitState, Entries) ->
+    run_log(InitState, Entries, fun(_) -> true end).
+
+run_log(InitState, Entries, InvariantFun) ->
+    Invariant = fun(E, S) ->
+                       case InvariantFun(S) of
+                           true -> ok;
+                           false ->
+                               throw({invariant, E, S})
+                       end
+                end,
+
     lists:foldl(fun ({Idx, E}, {Acc0, Efx0}) ->
                         case rabbit_fifo:apply(meta(Idx), E, Acc0) of
                             {Acc, _, Efx} when is_list(Efx) ->
+                                Invariant(E, Acc),
                                 {Acc, Efx0 ++ Efx};
                             {Acc, _, Efx}  ->
+                                Invariant(E, Acc),
                                 {Acc, Efx0 ++ [Efx]};
                             {Acc, _}  ->
+                                Invariant(E, Acc),
                                 {Acc, Efx0}
                         end
                 end, {InitState, []}, Entries).