diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/rabbit_fifo_SUITE.erl | 125 | ||||
| -rw-r--r-- | test/rabbit_fifo_prop_SUITE.erl | 337 |
2 files changed, 418 insertions, 44 deletions
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). |