Add further backing queue methods

1 files changed, 133 insertions, 54 deletions

diff --git a/src/rabbit_backing_queue_qc.erl b/src/rabbit_backing_queue_qc.erl
index f5cd7f95b1..83e7da5ec6 100644
--- a/src/rabbit_backing_queue_qc.erl
+++ b/src/rabbit_backing_queue_qc.erl
@@ -23,26 +23,50 @@
 
 -define(BQMOD, rabbit_variable_queue).
 
+-define(RECORD_INDEX(Key, Record),
+    erlang:element(2, proplists:lookup(Key, lists:zip(
+       record_info(fields, Record), lists:seq(2, record_info(size, Record)))))).
+
 -export([initial_state/0, command/1, precondition/2, postcondition/3,
          next_state/3]).
 
 -export([prop_backing_queue_test/0]).
 
-
 -record(state, {bqstate,
-                messages,
-                acks}).
+                messages,   %% queue of {msg_props, basic_msg}
+                acks,       %% list of {acktag, {message_props, basic_msg}}
+                confirms}). %% set of msgid
 
+%% Initialise model
 
 initial_state() ->
-    VQ = qc_variable_queue_init(qc_test_queue()),
-    #state{bqstate=VQ, messages = queue:new(), acks = []}.
+    #state{bqstate  = qc_variable_queue_init(qc_test_queue()),
+           messages = queue:new(),
+           acks     = [],
+           confirms = gb_sets:new()}.
+
+%% Property
 
 prop_backing_queue_test() ->
     ?FORALL(Cmds, commands(?MODULE, initial_state()),
         begin
-            {_H, #state{bqstate = VQ}, Res} = run_commands(?MODULE, Cmds),
-            rabbit_variable_queue:delete_and_terminate(shutdown, VQ),
+            {ok, FileSizeLimit} =
+                application:get_env(rabbit, msg_store_file_size_limit),
+            application:set_env(rabbit, msg_store_file_size_limit, 512,
+                                infinity),
+            {ok, MaxJournal} =
+                application:get_env(rabbit, queue_index_max_journal_entries),
+            application:set_env(rabbit, queue_index_max_journal_entries, 128,
+                                infinity),
+
+            {_H, #state{bqstate = BQ}, Res} = run_commands(?MODULE, Cmds),
+
+            application:set_env(rabbit, msg_store_file_size_limit,
+                                FileSizeLimit, infinity),
+            application:set_env(rabbit, queue_index_max_journal_entries,
+                                MaxJournal, infinity),
+
+            rabbit_variable_queue:delete_and_terminate(shutdown, BQ),
             ?WHENFAIL(
                 io:format("Result: ~p~n", [Res]),
                 aggregate(command_names(Cmds), Res =:= ok))
@@ -50,32 +74,46 @@ prop_backing_queue_test() ->
 
 %% Commands
 
-command(#state{bqstate = VQ} = S) ->
+command(S) ->
     ?SIZED(Size,
-           frequency([{Size, qc_publish(S)},
-                      {Size, qc_fetch(S)},
-                      {Size, qc_ack(S)},
-                      {Size, qc_requeue(S)},
-                      {Size, qc_ram(S)},
-                      {1,    {call, ?BQMOD, purge, [VQ]}}])).
-
-qc_publish(#state{bqstate = VQ}) ->
+        frequency([{Size, qc_publish(S)},
+                   {Size, qc_fetch(S)},
+                   {Size, qc_ack(S)},
+                   {Size, qc_requeue(S)},
+                   {Size, qc_ram(S)},
+                   {Size, qc_drain_confirmed(S)},
+                   {Size, qc_dropwhile(S)},
+                   {1,    qc_purge(S)}])).
+
+qc_publish(#state{bqstate = BQ}) ->
     {call, ?BQMOD, publish,
-      [qc_message(), #message_properties{}, self(), VQ]}.
+      [qc_message(),
+       #message_properties{needs_confirming = frequency([{1,  true},
+                                                         {20, false}]),
+                           expiry = choose(0, 10)},
+       self(), BQ]}.
 
-qc_fetch(#state{bqstate = VQ}) ->
-    {call, ?BQMOD, fetch, [boolean(), VQ]}.
+qc_fetch(#state{bqstate = BQ}) ->
+    {call, ?BQMOD, fetch, [boolean(), BQ]}.
 
-qc_ack(#state{bqstate = VQ, acks = Acks}) ->
-    {call, ?BQMOD, ack, [rand_choice(proplists:get_keys(Acks)), VQ]}.
+qc_ack(#state{bqstate = BQ, acks = Acks}) ->
+    {call, ?BQMOD, ack, [rand_choice(proplists:get_keys(Acks)), BQ]}.
 
-qc_requeue(#state{bqstate = VQ, acks = Acks}) ->
+qc_requeue(#state{bqstate = BQ, acks = Acks}) ->
     {call, ?BQMOD, requeue,
-      [rand_choice(proplists:get_keys(Acks)), fun(MsgOpts) -> MsgOpts end, VQ]}.
+      [rand_choice(proplists:get_keys(Acks)), fun(MsgOpts) -> MsgOpts end, BQ]}.
+
+qc_ram(#state{bqstate = BQ}) ->
+    {call, ?BQMOD, set_ram_duration_target, [oneof([0, infinity]), BQ]}.
+
+qc_drain_confirmed(#state{bqstate = BQ}) ->
+    {call, ?BQMOD, drain_confirmed, [BQ]}.
 
-qc_ram(#state{bqstate = VQ}) ->
-    {call, ?BQMOD, set_ram_duration_target,
-      [oneof([0, infinity]), VQ]}.
+qc_dropwhile(#state{bqstate = BQ}) ->
+    {call, ?BQMOD, dropwhile, [fun dropfun/1, BQ]}.
+
+qc_purge(#state{bqstate = BQ}) ->
+    {call, ?BQMOD, purge, [BQ]}.
 
 %% Preconditions
 
@@ -85,68 +123,92 @@ precondition(#state{acks = Acks}, {call, ?BQMOD, Fun, _Arg})
 precondition(_S, {call, ?BQMOD, _Fun, _Arg}) ->
     true.
 
-%% Next state
-
-next_state(S, VQ, {call, ?BQMOD, publish, [Msg, _MsgProps, _Pid, _VQ]}) ->
-    #state{messages = Messages} = S,
-    S#state{bqstate = VQ, messages = queue:in(Msg, Messages)};
-
-next_state(S, Res, {call, ?BQMOD, fetch, [AckReq, _VQ]}) ->
+%% Model updates
+
+next_state(S, BQ, {call, ?BQMOD, publish, [Msg, MsgProps, _Pid, _BQ]}) ->
+    #state{messages = Messages, confirms = Confirms} = S,
+    MsgId = {call, erlang, element, [?RECORD_INDEX(id, basic_message), Msg]},
+    NeedsConfirm =
+        {call, erlang, element,
+            [?RECORD_INDEX(needs_confirming, message_properties), MsgProps]},
+    Confirms1 = case eval(NeedsConfirm) of
+                    true -> gb_sets:add(MsgId, Confirms);
+                    _    -> Confirms
+                end,
+    S#state{bqstate  = BQ,
+            messages = queue:in({MsgProps, Msg}, Messages),
+            confirms = Confirms1};
+
+next_state(S, Res, {call, ?BQMOD, fetch, [AckReq, _BQ]}) ->
     #state{messages = Messages, acks = Acks} = S,
     ResultInfo = {call, erlang, element, [1, Res]},
-    VQ1        = {call, erlang, element, [2, Res]},
+    BQ1        = {call, erlang, element, [2, Res]},
     AckTag     = {call, erlang, element, [3, ResultInfo]},
-    S1         = S#state{bqstate = VQ1},
+    S1         = S#state{bqstate = BQ1},
     case queue:out(Messages) of
         {empty, _M2}       ->
             S1;
-        {{value, Msg}, M2} ->
+        {{value, MsgProp_Msg}, M2} ->
             S2 = S1#state{messages = M2},
             case AckReq of
-                true  -> S2#state{acks = Acks ++ [{AckTag, Msg}]};
+                true  -> S2#state{acks = Acks ++ [{AckTag, MsgProp_Msg}]};
                 false -> S2
            end
     end;
 
-next_state(S, Res, {call, ?BQMOD, ack, [AcksArg, _VQ]}) ->
+next_state(S, Res, {call, ?BQMOD, ack, [AcksArg, _BQ]}) ->
     #state{acks = AcksState} = S,
-    VQ1 = {call, erlang, element, [2, Res]},
-    S#state{bqstate = VQ1,
+    BQ1 = {call, erlang, element, [2, Res]},
+    S#state{bqstate = BQ1,
             acks    = propvals_by_keys(AcksState, AcksArg)};
 
 next_state(S, Res, {call, ?BQMOD, requeue, [AcksArg, _F, _V]}) ->
     #state{messages = Messages, acks = AcksState} = S,
-    VQ1 = {call, erlang, element, [2, Res]},
-    RequeueMsgs = [proplists:get_value(Key, AcksState) || Key <- AcksArg ],
-    S#state{bqstate  = VQ1,
+    BQ1 = {call, erlang, element, [2, Res]},
+    RequeueMsgs = [proplists:get_value(Key, AcksState) || Key <- AcksArg],
+    S#state{bqstate  = BQ1,
             messages = queue:join(Messages, queue:from_list(RequeueMsgs)),
             acks     = propvals_by_keys(AcksState, AcksArg)};
 
-next_state(S, VQ, {call, ?BQMOD, set_ram_duration_target, _A}) ->
-    S#state{bqstate = VQ};
+next_state(S, BQ, {call, ?BQMOD, set_ram_duration_target, _Args}) ->
+    S#state{bqstate = BQ};
 
-next_state(S, Res, {call, ?BQMOD, purge, _A}) ->
-    VQ1 = {call, erlang, element, [2, Res]},
-    S#state{bqstate = VQ1, messages = queue:new()}.
+next_state(S, Res, {call, ?BQMOD, drain_confirmed, _Args}) ->
+    BQ1 = {call, erlang, element, [2, Res]},
+    S#state{bqstate = BQ1};
+
+next_state(S, BQ1, {call, ?BQMOD, dropwhile, _Args}) ->
+    #state{messages = Messages} = S,
+    S#state{bqstate = BQ1, messages = drop_messages(Messages)};
+
+next_state(S, Res, {call, ?BQMOD, purge, _Args}) ->
+    BQ1 = {call, erlang, element, [2, Res]},
+    S#state{bqstate = BQ1, messages = queue:new()}.
 
 %% Postconditions
 
 postcondition(#state{messages = Messages}, {call, ?BQMOD, fetch, _Args}, Res) ->
     case Res of
-        {{MsgFetched, _IsDelivered, _AckTag, _Remaining_Len}, _VQ} ->
-            MsgFetched =:= queue:head(Messages);
-        {empty, _VQ} ->
+        {{MsgFetched, _IsDelivered, _AckTag, _Remaining_Len}, _BQ} ->
+            {_MsgProps, Msg} = queue:head(Messages),
+            MsgFetched =:= Msg;
+        {empty, _BQ} ->
             queue:len(Messages) =:= 0
     end;
 
 postcondition(#state{messages = Messages}, {call, ?BQMOD, purge, _Args}, Res) ->
-    {PurgeCount, _VQ} = Res,
+    {PurgeCount, _BQ} = Res,
     queue:len(Messages) =:= PurgeCount;
 
-postcondition(#state{bqstate  = VQ,
+postcondition(S, {call, ?BQMOD, drain_confirmed, _Args}, Res) ->
+    #state{confirms = Confirms} = S,
+    {ReportedConfirmed, _BQ} = Res,
+    lists:all(fun (M) -> lists:member(M, Confirms) end, ReportedConfirmed);
+
+postcondition(#state{bqstate  = BQ,
                      messages = Messages},
               {call, ?BQMOD, _Fun, _Args}, _Res) ->
-    ?BQMOD:len(VQ) =:= queue:len(Messages).
+    ?BQMOD:len(BQ) =:= queue:len(Messages).
 
 %% Helpers
 
@@ -201,3 +263,20 @@ rand_choice(List) ->
         []  -> [];
         _   -> [lists:nth(random:uniform(length(List)), List)]
     end.
+
+dropfun(Props) ->
+    Expiry = eval({call, erlang, element,
+                       [?RECORD_INDEX(expiry, message_properties), Props]}),
+    Expiry =/= 0.
+
+drop_messages(Messages) ->
+    case queue:out(Messages) of
+        {empty, _} ->
+            Messages;
+        {{value, MsgProps_Msg}, M2} ->
+            MsgProps = {call, erlang, element, [1, MsgProps_Msg]},
+            case dropfun(MsgProps) of
+                true  -> drop_messages(M2);
+                false -> Messages
+            end
+    end.