summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/clustering_management_SUITE.erl39
-rw-r--r--test/quorum_queue_SUITE.erl28
-rw-r--r--test/rabbit_fifo_SUITE.erl189
-rw-r--r--test/rabbit_fifo_int_SUITE.erl2
-rw-r--r--test/rabbit_fifo_prop_SUITE.erl348
5 files changed, 547 insertions, 59 deletions
diff --git a/test/clustering_management_SUITE.erl b/test/clustering_management_SUITE.erl
index c9f1565c51..40317ec604 100644
--- a/test/clustering_management_SUITE.erl
+++ b/test/clustering_management_SUITE.erl
@@ -55,7 +55,8 @@ groups() ->
force_boot,
status_with_alarm,
pid_file_and_await_node_startup,
- await_running_count
+ await_running_count,
+ start_with_invalid_schema_in_path
]},
{cluster_size_4, [], [
forget_promotes_offline_slave
@@ -121,6 +122,42 @@ end_per_testcase(Testcase, Config) ->
%% Testcases.
%% -------------------------------------------------------------------
+
+start_with_invalid_schema_in_path(Config) ->
+ [Rabbit, Hare] = cluster_members(Config),
+ stop_app(Rabbit),
+ stop_app(Hare),
+
+ create_bad_schema(Rabbit, Hare, Config),
+
+ start_app(Hare),
+ case start_app(Rabbit) of
+ ok -> ok;
+ ErrRabbit -> error({unable_to_start_with_bad_schema_in_work_dir, ErrRabbit})
+ end.
+
+create_bad_schema(Rabbit, Hare, Config) ->
+
+ {ok, RabbitMnesiaDir} = rpc:call(Rabbit, application, get_env, [mnesia, dir]),
+ {ok, HareMnesiaDir} = rpc:call(Hare, application, get_env, [mnesia, dir]),
+ %% Make sure we don't use the current dir:
+ PrivDir = ?config(priv_dir, Config),
+ ct:pal("Priv dir ~p~n", [PrivDir]),
+ ok = filelib:ensure_dir(filename:join(PrivDir, "file")),
+
+ ok = rpc:call(Rabbit, file, set_cwd, [PrivDir]),
+ ok = rpc:call(Hare, file, set_cwd, [PrivDir]),
+
+ ok = rpc:call(Rabbit, application, unset_env, [mnesia, dir]),
+ ok = rpc:call(Hare, application, unset_env, [mnesia, dir]),
+ ok = rpc:call(Rabbit, mnesia, create_schema, [[Rabbit, Hare]]),
+ ok = rpc:call(Rabbit, mnesia, start, []),
+ {atomic,ok} = rpc:call(Rabbit, mnesia, create_table,
+ [rabbit_queue, [{ram_copies, [Rabbit, Hare]}]]),
+ stopped = rpc:call(Rabbit, mnesia, stop, []),
+ ok = rpc:call(Rabbit, application, set_env, [mnesia, dir, RabbitMnesiaDir]),
+ ok = rpc:call(Hare, application, set_env, [mnesia, dir, HareMnesiaDir]).
+
join_and_part_cluster(Config) ->
[Rabbit, Hare, Bunny] = cluster_members(Config),
assert_not_clustered(Rabbit),
diff --git a/test/quorum_queue_SUITE.erl b/test/quorum_queue_SUITE.erl
index 31d6a4c21b..6b623304e8 100644
--- a/test/quorum_queue_SUITE.erl
+++ b/test/quorum_queue_SUITE.erl
@@ -54,8 +54,9 @@ groups() ->
add_member_not_found,
delete_member_not_running,
delete_member_classic,
- delete_member_not_found,
- delete_member]
+ delete_member_queue_not_found,
+ delete_member,
+ delete_member_not_a_member]
++ all_tests()},
{cluster_size_2, [], memory_tests()},
{cluster_size_3, [], [
@@ -1235,7 +1236,8 @@ add_member_already_a_member(Config) ->
QQ = ?config(queue_name, Config),
?assertEqual({'queue.declare_ok', QQ, 0, 0},
declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
- ?assertEqual({error, already_a_member},
+ %% idempotent by design
+ ?assertEqual(ok,
rpc:call(Server, rabbit_quorum_queue, add_member,
[<<"/">>, QQ, Server])).
@@ -1273,7 +1275,8 @@ delete_member_not_running(Config) ->
QQ = ?config(queue_name, Config),
?assertEqual({'queue.declare_ok', QQ, 0, 0},
declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
- ?assertEqual({error, node_not_running},
+ %% it should be possible to delete members that are not online (e.g. decomissioned)
+ ?assertEqual(ok,
rpc:call(Server, rabbit_quorum_queue, delete_member,
[<<"/">>, QQ, 'rabbit@burrow'])).
@@ -1286,7 +1289,7 @@ delete_member_classic(Config) ->
rpc:call(Server, rabbit_quorum_queue, delete_member,
[<<"/">>, CQ, Server])).
-delete_member_not_found(Config) ->
+delete_member_queue_not_found(Config) ->
[Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
QQ = ?config(queue_name, Config),
?assertEqual({error, not_found},
@@ -1302,12 +1305,23 @@ delete_member(Config) ->
timer:sleep(100),
?assertEqual(ok,
rpc:call(Server, rabbit_quorum_queue, delete_member,
+ [<<"/">>, QQ, Server])).
+
+delete_member_not_a_member(Config) ->
+ [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+ Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+ QQ = ?config(queue_name, Config),
+ ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+ declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+ timer:sleep(100),
+ ?assertEqual(ok,
+ rpc:call(Server, rabbit_quorum_queue, delete_member,
[<<"/">>, QQ, Server])),
- ?assertEqual({error, not_a_member},
+ %% idempotent by design
+ ?assertEqual(ok,
rpc:call(Server, rabbit_quorum_queue, delete_member,
[<<"/">>, QQ, Server])).
-
cleanup_data_dir(Config) ->
%% This test is slow, but also checks that we handle properly errors when
%% trying to delete a queue in minority. A case clause there had gone
diff --git a/test/rabbit_fifo_SUITE.erl b/test/rabbit_fifo_SUITE.erl
index cf14a68d9d..310553dc56 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),
@@ -458,9 +480,12 @@ tick_test(_) ->
{S3, {_, _}} = deq(4, Cid2, unsettled, S2),
{S4, _, _} = apply(meta(5), rabbit_fifo:make_return(Cid, [MsgId]), S3),
- [{mod_call, _, _,
+ [{mod_call, rabbit_quorum_queue, handle_tick,
[#resource{},
- {?FUNCTION_NAME, 1, 1, 2, 1, 3, 3}]}, {aux, emit}] = rabbit_fifo:tick(1, S4),
+ {?FUNCTION_NAME, 1, 1, 2, 1, 3, 3},
+ [_Node]
+ ]},
+ {aux, emit}] = rabbit_fifo:tick(1, S4),
ok.
@@ -556,7 +581,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 +597,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 +785,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 +835,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,
@@ -836,10 +924,11 @@ single_active_consumer_all_disconnected_test(_) ->
single_active_consumer_state_enter_leader_include_waiting_consumers_test(_) ->
State0 = init(#{name => ?FUNCTION_NAME,
- queue_resource => rabbit_misc:r("/", queue,
- atom_to_binary(?FUNCTION_NAME, utf8)),
- release_cursor_interval => 0,
- single_active_consumer_on => true}),
+ queue_resource =>
+ rabbit_misc:r("/", queue,
+ atom_to_binary(?FUNCTION_NAME, utf8)),
+ release_cursor_interval => 0,
+ single_active_consumer_on => true}),
DummyFunction = fun() -> ok end,
Pid1 = spawn(DummyFunction),
@@ -990,11 +1079,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 +1112,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)).
@@ -1118,6 +1207,54 @@ single_active_with_credited_test(_) ->
State3#rabbit_fifo.waiting_consumers),
ok.
+purge_nodes_test(_) ->
+ Node = purged@node,
+ ThisNode = node(),
+ EnqPid = test_util:fake_pid(Node),
+ EnqPid2 = test_util:fake_pid(node()),
+ ConPid = test_util:fake_pid(Node),
+ Cid = {<<"tag">>, ConPid},
+ % WaitingPid = test_util:fake_pid(Node),
+
+ State0 = init(#{name => ?FUNCTION_NAME,
+ queue_resource => rabbit_misc:r("/", queue,
+ atom_to_binary(?FUNCTION_NAME, utf8)),
+ single_active_consumer_on => false}),
+ {State1, _, _} = apply(meta(1),
+ rabbit_fifo:make_enqueue(EnqPid, 1, msg1),
+ State0),
+ {State2, _, _} = apply(meta(2),
+ rabbit_fifo:make_enqueue(EnqPid2, 1, msg2),
+ State1),
+ {State3, _} = check(Cid, 3, 1000, State2),
+ {State4, _, _} = apply(meta(4),
+ {down, EnqPid, noconnection},
+ State3),
+ ?assertMatch(
+ [{mod_call, rabbit_quorum_queue, handle_tick,
+ [#resource{}, _Metrics,
+ [ThisNode, Node]
+ ]},
+ {aux, emit}] , rabbit_fifo:tick(1, State4)),
+ %% assert there are both enqueuers and consumers
+ {State, _, _} = apply(meta(5),
+ rabbit_fifo:make_purge_nodes([Node]),
+ State4),
+
+ %% assert there are no enqueuers nor consumers
+ ?assertMatch(#rabbit_fifo{enqueuers = Enqs} when map_size(Enqs) == 1,
+ State),
+
+ ?assertMatch(#rabbit_fifo{consumers = Cons} when map_size(Cons) == 0,
+ State),
+ ?assertMatch(
+ [{mod_call, rabbit_quorum_queue, handle_tick,
+ [#resource{}, _Metrics,
+ [ThisNode]
+ ]},
+ {aux, emit}] , rabbit_fifo:tick(1, State)),
+ ok.
+
meta(Idx) ->
#{index => Idx, term => 1}.
diff --git a/test/rabbit_fifo_int_SUITE.erl b/test/rabbit_fifo_int_SUITE.erl
index f281d15795..d4ae417a78 100644
--- a/test/rabbit_fifo_int_SUITE.erl
+++ b/test/rabbit_fifo_int_SUITE.erl
@@ -54,7 +54,7 @@ end_per_group(_, Config) ->
init_per_testcase(TestCase, Config) ->
meck:new(rabbit_quorum_queue, [passthrough]),
- meck:expect(rabbit_quorum_queue, handle_tick, fun (_, _) -> ok end),
+ meck:expect(rabbit_quorum_queue, handle_tick, fun (_, _, _) -> ok end),
meck:expect(rabbit_quorum_queue, cancel_consumer_handler,
fun (_, _) -> ok end),
ra_server_sup_sup:remove_all(),
diff --git a/test/rabbit_fifo_prop_SUITE.erl b/test/rabbit_fifo_prop_SUITE.erl
index 58e6c2452c..d670f2bdd7 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,
@@ -41,7 +44,14 @@ all_tests() ->
scenario17,
scenario18,
scenario19,
- scenario20
+ scenario20,
+ single_active,
+ single_active_01,
+ single_active_02,
+ single_active_03,
+ single_active_ordering,
+ single_active_ordering_01
+ % single_active_ordering_02
].
groups() ->
@@ -288,7 +298,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,
@@ -343,6 +355,87 @@ scenario20(_Config) ->
max_in_memory_length => 1}, 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, undefined, undefined),
+ ?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, undefined, undefined),
+ ?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, InMemoryLength,
+ InMemoryBytes},
+ frequency([{10, {0, 0, false, 0, 0, 0}},
+ {5, {oneof([range(1, 10), undefined]),
+ oneof([range(1, 1000), undefined]),
+ boolean(),
+ oneof([range(1, 3), undefined]),
+ oneof([range(1, 10), undefined]),
+ oneof([range(1, 1000), 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,
+ InMemoryLength,
+ InMemoryBytes), O))))
+ end, [], 10).
+
snapshots(_Config) ->
run_proper(
fun () ->
@@ -368,6 +461,90 @@ snapshots(_Config) ->
InMemoryBytes), O))))
end, [], 2500).
+single_active(_Config) ->
+ Size = 2000,
+ run_proper(
+ fun () ->
+ ?FORALL({Length, Bytes, DeliveryLimit, InMemoryLength, InMemoryBytes},
+ frequency([{10, {0, 0, 0, 0, 0}},
+ {5, {oneof([range(1, 10), undefined]),
+ oneof([range(1, 1000), undefined]),
+ oneof([range(1, 3), undefined]),
+ oneof([range(1, 10), undefined]),
+ oneof([range(1, 1000), 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,
+ InMemoryLength,
+ InMemoryBytes), 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,
+ undefined,
+ 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, 0, 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, 0, 0),
+ ?assert(single_active_prop(Conf, Commands, true)),
+ ok.
+
config(Name, Length, Bytes, SingleActive, DeliveryLimit, InMemoryLength, InMemoryBytes) ->
#{name => Name,
max_length => map_max(Length),
@@ -380,6 +557,65 @@ config(Name, Length, Bytes, SingleActive, DeliveryLimit, InMemoryLength, InMemor
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
@@ -391,6 +627,9 @@ snapshots_prop(Conf, Commands) ->
end.
log_gen(Size) ->
+ log_gen(Size, binary()).
+
+log_gen(Size, _Body) ->
Nodes = [node(),
fakenode@fake,
fakenode@fake2
@@ -413,6 +652,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))).
@@ -424,9 +688,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) ->
@@ -445,16 +712,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)),
@@ -464,6 +736,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
@@ -474,13 +747,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) ->
@@ -540,8 +817,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
@@ -555,21 +832,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) ->
@@ -637,13 +923,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).
View Lorry Controller Status