You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2019/02/11 15:54:16 UTC
[couchdb] branch shard-split updated: Add tests for when source db
gets deleted
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch shard-split
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/shard-split by this push:
new 0846f49 Add tests for when source db gets deleted
0846f49 is described below
commit 0846f49f9b86ef857111911c6e216a251129251a
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Mon Feb 11 10:53:26 2019 -0500
Add tests for when source db gets deleted
Found a bunch of bugs and realized we don't need the source monitor
---
src/couch/src/couch_db_split.erl | 99 ++++++++++++++++------------
src/mem3/src/mem3_reshard.erl | 79 +++++-----------------
src/mem3/src/mem3_reshard.hrl | 1 -
src/mem3/src/mem3_reshard_dbdoc.erl | 14 ++--
src/mem3/src/mem3_reshard_job.erl | 71 ++++++++++++++------
src/mem3/test/mem3_reshard_api_test.erl | 113 +++++++++++++++++++++++++-------
6 files changed, 225 insertions(+), 152 deletions(-)
diff --git a/src/couch/src/couch_db_split.erl b/src/couch/src/couch_db_split.erl
index adb8bb8..c459fda 100644
--- a/src/couch/src/couch_db_split.erl
+++ b/src/couch/src/couch_db_split.erl
@@ -55,58 +55,71 @@
% Public API
split(Source, #{} = Targets, PickFun) when is_function(PickFun, 3) ->
- {ok, SourceDb} = couch_db:open_int(Source, [?ADMIN_CTX]),
- Engine = get_engine(SourceDb),
- Partitioned = couch_db:is_partitioned(SourceDb),
- {_M, _F, _A} = HashFun = mem3_hash:get_hash_fun(couch_db:name(SourceDb)),
- try
- try
- split(SourceDb, Partitioned, Engine, Targets, PickFun, HashFun)
- catch
- throw:{target_create_error, DbName, Error, CreatedTargets} ->
- cleanup_targets(CreatedTargets, Engine),
- {error, {target_create_error, DbName, Error}}
- end
- after
- couch_db:close(SourceDb)
+ case couch_db:open_int(Source, [?ADMIN_CTX]) of
+ {ok, SourceDb} ->
+ Engine = get_engine(SourceDb),
+ Partitioned = couch_db:is_partitioned(SourceDb),
+ HashFun = mem3_hash:get_hash_fun(couch_db:name(SourceDb)),
+ try
+ try
+ split(SourceDb, Partitioned, Engine, Targets, PickFun,
+ HashFun)
+ catch
+ throw:{target_create_error, DbName, Error, TargetDbs} ->
+ cleanup_targets(TargetDbs, Engine),
+ {error, {target_create_error, DbName, Error}}
+ end
+ after
+ couch_db:close(SourceDb)
+ end;
+ {not_found, _} ->
+ {error, missing_source}
end.
copy_local_docs(Source, #{} = Targets0, PickFun) when
is_binary(Source), is_function(PickFun, 3) ->
- {ok, SourceDb} = couch_db:open_int(Source, [?ADMIN_CTX]),
- try
- Targets = maps:map(fun(_, DbName) ->
- {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
- #target{db = Db, uuid = couch_db:get_uuid(Db)}
- end, Targets0),
- try
- State = #state{
- source_db = SourceDb,
- source_uuid = couch_db:get_uuid(SourceDb),
- targets = Targets,
- pickfun = PickFun,
- hashfun = mem3_hash:get_hash_fun(couch_db:name(SourceDb))
- },
- copy_local_docs(State),
- ok
- after
- maps:map(fun(_, #target{db = Db} = T) ->
- couch_db:close(Db),
- T#target{db = undefined}
- end, Targets)
- end
- after
- couch_db:close(SourceDb)
+ case couch_db:open_int(Source, [?ADMIN_CTX]) of
+ {ok, SourceDb} ->
+ try
+ Targets = maps:map(fun(_, DbName) ->
+ {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
+ #target{db = Db, uuid = couch_db:get_uuid(Db)}
+ end, Targets0),
+ try
+ State = #state{
+ source_db = SourceDb,
+ source_uuid = couch_db:get_uuid(SourceDb),
+ targets = Targets,
+ pickfun = PickFun,
+ hashfun = mem3_hash:get_hash_fun(couch_db:name(SourceDb))
+ },
+ copy_local_docs(State),
+ ok
+ after
+ maps:map(fun(_, #target{db = Db} = T) ->
+ couch_db:close(Db),
+ T#target{db = undefined}
+ end, Targets)
+ end
+ after
+ couch_db:close(SourceDb)
+ end;
+ {not_found, _} ->
+ {error, missing_source}
end.
cleanup_target(Source, Target) when is_binary(Source), is_binary(Target) ->
- {ok, SourceDb} = couch_db:open_int(Source, [?ADMIN_CTX]),
- try
- delete_target(Target, get_engine(SourceDb))
- after
- couch_db:close(SourceDb)
+ case couch_db:open_int(Source, [?ADMIN_CTX]) of
+ {ok, SourceDb} ->
+ try
+ delete_target(Target, get_engine(SourceDb))
+ after
+ couch_db:close(SourceDb)
+ end;
+ {not_found, _} ->
+ {error, missing_source}
end.
diff --git a/src/mem3/src/mem3_reshard.erl b/src/mem3/src/mem3_reshard.erl
index 4ba0e6d..d5f8fea 100644
--- a/src/mem3/src/mem3_reshard.erl
+++ b/src/mem3/src/mem3_reshard.erl
@@ -28,7 +28,6 @@
stop/1,
get_state/0,
reset_state/0,
- db_monitor/1,
now_sec/0,
update_history/4,
@@ -175,8 +174,7 @@ init(_) ->
state = running,
state_info = [],
update_time = now_sec(),
- node = node(),
- db_monitor = spawn_link(?MODULE, db_monitor, [self()])
+ node = node()
},
State1 = mem3_reshard_store:init(State, ?JOB_PREFIX, state_id()),
State2 = mem3_reshard_store:load_state(State1),
@@ -186,8 +184,6 @@ init(_) ->
terminate(Reason, State) ->
couch_log:notice("~p terminate ~p ~p", [?MODULE, Reason, statefmt(State)]),
- catch unlink(State#state.db_monitor),
- catch exit(State#state.db_monitor, kill),
[kill_job_int(Job) || Job <- running_jobs()],
ok.
@@ -328,7 +324,6 @@ handle_cast({report, Job}, State) ->
report_int(Job),
{noreply, State};
-
handle_cast({checkpoint, Job}, State) ->
{noreply, checkpoint_int(Job, State)};
@@ -338,11 +333,6 @@ handle_cast(reload_jobs, State) ->
couch_log:notice("~p finished reloading jobs", [?MODULE]),
{noreply, State1};
-handle_cast({db_deleted, Jobs}, State) ->
- couch_log:notice("~p db_deleted: ~B jobs", [?MODULE, length(Jobs)]),
- fail_jobs_for_deleted_sources(Jobs, State),
- {noreply, State};
-
handle_cast(Cast, State) ->
couch_log:error("~p unexpected cast ~p", [?MODULE, Cast]),
{noreply, State}.
@@ -644,26 +634,12 @@ now_sec() ->
Mega * 1000000 + Sec.
--spec jobs_by_dbname(binary()) -> [#job{}].
-jobs_by_dbname(Name) ->
- Pat = #job{source = #shard{name = Name, _ = '_'}, _ = '_'},
- [Job || [Job] <- ets:match_object(?MODULE, Pat)].
-
-
-spec running_jobs() -> [#job{}].
running_jobs() ->
Pat = #job{job_state = running, _ = '_'},
ets:match_object(?MODULE, Pat).
--spec fail_jobs_for_deleted_sources([#job{}], #state{}) -> ok.
-fail_jobs_for_deleted_sources(Jobs, #state{} = State) ->
- JobCount = length(Jobs),
- couch_log:notice("~p fail ~B jobs, source deleted", [?MODULE, JobCount]),
- [stop_job_int(Job, failed, "Source was deleted", State) || Job <- Jobs],
- ok.
-
-
-spec info_update(atom(), any(), [tuple()]) -> [tuple()].
info_update(Key, Val, StateInfo) ->
lists:keystore(Key, 1, StateInfo, {Key, Val}).
@@ -710,46 +686,27 @@ report_int(Job) ->
end.
--spec db_monitor(pid()) -> no_return().
-db_monitor(Server) ->
- couch_log:notice("~p db monitor ~p starting", [?MODULE, self()]),
- EvtRef = erlang:monitor(process, couch_event_server),
- couch_event:register_all(self()),
- db_monitor_loop(Server, EvtRef).
-
-
--spec db_monitor_loop(pid(), reference()) -> no_return().
-db_monitor_loop(Server, EvtRef) ->
- receive
- {'$couch_event', DbName, deleted} ->
- case jobs_by_dbname(DbName) of
- [] ->
- ok;
- Jobs ->
- gen_server:cast(Server, {db_deleted, Jobs})
- end,
- db_monitor_loop(Server, EvtRef);
- {'$couch_event', _, _} ->
- db_monitor_loop(Server, EvtRef);
- {'DOWN', EvtRef, _, _, Info} ->
- couch_log:error("~p db monitor listener died ~p", [?MODULE, Info]),
- exit({db_monitor_died, Info});
- Msg ->
- couch_log:error("~p db monitor unexpected msg ~p", [?MODULE, Msg]),
- db_monitor_loop(Server, EvtRef)
- end.
-
-
+% This function is for testing and debugging only
-spec reset_state(#state{}) -> #state{}.
reset_state(#state{} = State) ->
- couch_log:warning("~p deleting state", [?MODULE]),
+ couch_log:warning("~p resetting state", [?MODULE]),
ok = mem3_reshard_store:delete_state(State),
+ couch_log:warning("~p killing all running jobs", [?MODULE]),
+ [kill_job_int(Job) || Job <- running_jobs()],
+ ets:delete_all_objects(?MODULE),
+ couch_log:warning("~p resetting all job states", [?MODULE]),
Jobs = mem3_reshard_store:get_jobs(State),
lists:foldl(fun(#job{id = Id}, StateAcc) ->
- couch_log:warning("~p deleting job ~p", [?MODULE, Id]),
+ couch_log:warning("~p resetting job state ~p", [?MODULE, Id]),
ok = mem3_reshard_store:delete_job(StateAcc, Id),
StateAcc
- end, State, Jobs).
+ end, State, Jobs),
+ couch_log:warning("~p resetting state done", [?MODULE]),
+ State#state{
+ state = running,
+ state_info = [],
+ update_time = now_sec()
+ }.
-spec update_job_history(#job{}) -> #job{}.
@@ -798,11 +755,11 @@ max_history() ->
-spec statefmt(#state{} | term()) -> string().
-statefmt(#state{state = StateName, db_monitor = Pid}) ->
+statefmt(#state{state = StateName}) ->
Total = ets:info(?MODULE, size),
Active = mem3_reshard_job_sup:count_children(),
- Msg = "#state{~s total:~B active:~B mon:~p}",
- Fmt = io_lib:format(Msg, [StateName, Total, Active, Pid]),
+ Msg = "#state{~s total:~B active:~B}",
+ Fmt = io_lib:format(Msg, [StateName, Total, Active]),
lists:flatten(Fmt);
statefmt(State) ->
diff --git a/src/mem3/src/mem3_reshard.hrl b/src/mem3/src/mem3_reshard.hrl
index 3a32f1f..652f681 100644
--- a/src/mem3/src/mem3_reshard.hrl
+++ b/src/mem3/src/mem3_reshard.hrl
@@ -69,7 +69,6 @@
update_time :: non_neg_integer(),
job_prefix :: binary(),
state_id :: binary(),
- db_monitor :: pid(),
node :: node()
}).
diff --git a/src/mem3/src/mem3_reshard_dbdoc.erl b/src/mem3/src/mem3_reshard_dbdoc.erl
index af2501a..2e93866 100644
--- a/src/mem3/src/mem3_reshard_dbdoc.erl
+++ b/src/mem3/src/mem3_reshard_dbdoc.erl
@@ -29,15 +29,17 @@ update_shard_map(#job{source = Source, targets = Targets} = Job) ->
ok ->
ok;
Error ->
- error(Error)
+ exit(Error)
end,
#shard{name = SourceName} = Source,
DocId = mem3:dbname(SourceName),
OldBody = case mem3_util:open_db_doc(DocId) of
{ok, #doc{body = DocBody}} ->
DocBody;
+ {not_found, deleted} ->
+ exit({error, missing_source});
OpenErr ->
- error({shard_doc_open_error, OpenErr})
+ exit({shard_doc_open_error, OpenErr})
end,
Node = hd(mem3_util:live_nodes()),
Body = update_shard_props(OldBody, Source, Targets),
@@ -53,12 +55,12 @@ update_shard_map(#job{source = Source, targets = Targets} = Job) ->
true ->
ok;
false ->
- error({shard_update_did_not_propagate, Source})
+ exit({shard_update_did_not_propagate, Source})
end;
UpdateErr ->
ErrArgs = [?MODULE, JobStr, Node, UpdateErr],
couch_log:error("~p : ~s node:~p error:~p", ErrArgs),
- error({shard_doc_update_error, UpdateErr})
+ exit({shard_doc_update_error, UpdateErr})
end.
@@ -152,7 +154,7 @@ remove_node_from_source(ByRange, Source) ->
true ->
ok;
false ->
- error({source_shard_missing_node, NodeKey, SourceNodes})
+ exit({source_shard_missing_node, NodeKey, SourceNodes})
end,
SourceNodes1 = SourceNodes -- [NodeKey],
case SourceNodes1 of
@@ -173,7 +175,7 @@ add_node_to_target_foldl(#shard{} = Target, ByRange) ->
false ->
ok;
true ->
- error({target_shard_already_has_node, NodeKey, Nodes})
+ exit({target_shard_already_has_node, NodeKey, Nodes})
end,
Nodes1 = lists:sort([NodeKey | Nodes]),
lists:keyreplace(TKey, 1, ByRange, {TKey, Nodes1});
diff --git a/src/mem3/src/mem3_reshard_job.erl b/src/mem3/src/mem3_reshard_job.erl
index f22720f..92d5f03 100644
--- a/src/mem3/src/mem3_reshard_job.erl
+++ b/src/mem3/src/mem3_reshard_job.erl
@@ -305,6 +305,12 @@ worker_exited(normal, #job{workers = Workers} = Job) when Workers =/= [] ->
couch_log:debug(Msg, [?MODULE, jobfmt(Job), WaitingOn]),
{noreply, Job};
+worker_exited({error, missing_source}, #job{} = Job) ->
+ Msg = "~p stopping worker due to source missing ~p",
+ couch_log:error(Msg, [?MODULE, jobfmt(Job)]),
+ [begin unlink(Pid), exit(Pid, kill) end || Pid <- Job#job.workers],
+ {stop, {error, missing_source}, Job};
+
worker_exited(Reason, #job{} = Job0) ->
couch_log:error("~p worker error ~p ~p", [?MODULE, jobfmt(Job0), Reason]),
[begin unlink(Pid), exit(Pid, kill) end || Pid <- Job0#job.workers],
@@ -333,10 +339,21 @@ initial_copy(#job{source = Source, targets = Targets0} = Job) ->
#shard{name = SourceName} = Source,
Targets = [{Range, Name} || #shard{range = Range, name = Name} <- Targets0],
TMap = maps:from_list(Targets),
- couch_log:notice("~p initial_copy start ~s", [?MODULE, shardsstr(Source, Targets0)]),
- {ok, Seq} = couch_db_split:split(SourceName, TMap, fun pickfun/3),
- couch_log:notice("~p initial_copy of ~p finished @ ~p", [?MODULE, Source, Seq]),
- create_artificial_mem3_rep_checkpoints(Job, Seq).
+ LogMsg1 = "~p initial_copy started ~s",
+ LogArgs1 = [?MODULE, shardsstr(Source, Targets0)],
+ couch_log:notice(LogMsg1, LogArgs1),
+ case couch_db_split:split(SourceName, TMap, fun pickfun/3) of
+ {ok, Seq} ->
+ LogMsg2 = "~p initial_copy of ~s finished @ seq:~p",
+ LogArgs2 = [?MODULE, shardsstr(Source, Targets0), Seq],
+ couch_log:notice(LogMsg2, LogArgs2),
+ create_artificial_mem3_rep_checkpoints(Job, Seq);
+ {error, Error} ->
+ LogMsg3 = "~p initial_copy of ~p finished @ ~p",
+ LogArgs3 = [?MODULE, shardsstr(Source, Targets0), Error],
+ couch_log:notice(LogMsg3, LogArgs3),
+ exit({error, Error})
+ end.
create_artificial_mem3_rep_checkpoints(#job{} = Job, Seq) ->
@@ -419,24 +436,34 @@ copy_local_docs(#job{source = Source, targets = Targets0}) ->
#shard{name = SourceName} = Source,
Targets = [{Range, Name} || #shard{range = Range, name = Name} <- Targets0],
TMap = maps:from_list(Targets),
- couch_log:notice("~p copy local docs start ~p -> ~p", [?MODULE, Source, Targets]),
- ok = couch_db_split:copy_local_docs(SourceName, TMap, fun pickfun/3),
- couch_log:notice("~p copy local docs finished for ~p", [?MODULE, Source]),
- ok.
+ LogArg1 = [?MODULE, shardsstr(Source, Targets)],
+ couch_log:notice("~p copy local docs start ~s", LogArg1),
+ case couch_db_split:copy_local_docs(SourceName, TMap, fun pickfun/3) of
+ ok ->
+ couch_log:notice("~p copy local docs finished for ~s", LogArg1),
+ ok;
+ {error, Error} ->
+ LogArg2 = [?MODULE, shardsstr(Source, Targets), Error],
+ couch_log:error("~p copy local docs failed for ~s ~p", LogArg2),
+ exit({error, Error})
+ end.
wait_source_close(#job{source = #shard{name = Name}}) ->
Timeout = config:get_integer("mem3_reshard",
"source_close_timeout_sec", 600),
- couch_util:with_db(Name, fun(Db) ->
- Now = mem3_reshard:now_sec(),
- case wait_source_close(Db, 5, Now + Timeout) of
- true ->
- ok;
- false ->
- exit({error, source_db_close_timeout, Name, Timeout})
- end
- end).
+ case couch_db:open_int(Name, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ Now = mem3_reshard:now_sec(),
+ case wait_source_close(Db, 5, Now + Timeout) of
+ true ->
+ ok;
+ false ->
+ exit({error, source_db_close_timeout, Name, Timeout})
+ end;
+ {not_found, _} ->
+ exit({error, missing_source})
+ end.
wait_source_close(Db, SleepSec, UntilSec) ->
@@ -460,8 +487,14 @@ wait_source_close(Db, SleepSec, UntilSec) ->
source_delete(#job{source = #shard{name = Name}}) ->
case config:get_boolean("mem3_reshard", "delete_source", true) of
true ->
- ok = couch_server:delete(Name, [?ADMIN_CTX]),
- couch_log:warning("~p : deleted source shard ~p", [?MODULE, Name]);
+ case couch_server:delete(Name, [?ADMIN_CTX]) of
+ ok ->
+ couch_log:notice("~p : deleted source shard ~p",
+ [?MODULE, Name]);
+ not_found ->
+ couch_log:warning("~p : source was already deleted ~p",
+ [?MODULE, Name])
+ end;
false ->
couch_log:notice("~p : not deleting source shard ~p", [?MODULE, Name])
end,
diff --git a/src/mem3/test/mem3_reshard_api_test.erl b/src/mem3/test/mem3_reshard_api_test.erl
index 224116a..703e167 100644
--- a/src/mem3/test/mem3_reshard_api_test.erl
+++ b/src/mem3/test/mem3_reshard_api_test.erl
@@ -45,11 +45,12 @@ setup() ->
teardown(Url) ->
- meck:unload(),
mem3_reshard:reset_state(),
- delete_db(Url ++ ?DB1),
- delete_db(Url ++ ?DB2),
- delete_db(Url ++ ?DB3),
+ meck:unload(),
+ catch delete_db(Url ++ ?DB1),
+ catch delete_db(Url ++ ?DB2),
+ catch delete_db(Url ++ ?DB3),
+ ok = config:delete("mem3_reshard", "max_jobs", _Persist=false),
ok = config:delete("admins", ?USER, _Persist=false).
@@ -74,6 +75,7 @@ mem3_reshard_api_test_() ->
fun basics/1,
fun create_job_basic/1,
fun create_two_jobs/1,
+ fun create_multiple_jobs_from_one_post/1,
fun start_stop_cluster_basic/1,
fun start_stop_cluster_with_a_job/1,
fun individual_job_start_stop/1,
@@ -81,7 +83,13 @@ mem3_reshard_api_test_() ->
fun create_job_with_invalid_arguments/1,
fun create_job_with_db/1,
fun create_job_with_shard_name/1,
- fun completed_job_handling/1
+ fun completed_job_handling/1,
+ fun handle_source_deletion_in_initial_copy/1,
+ fun handle_source_deletion_in_topoff/1,
+ fun handle_source_deletion_in_copy_local_docs/1,
+ fun handle_source_deletion_in_build_indices/1,
+ fun handle_source_deletion_in_update_shard_map/1,
+ fun handle_source_deletion_in_source_delete/1
]
}
}
@@ -176,27 +184,39 @@ create_job_basic(Top) ->
create_two_jobs(Top) ->
?_test(begin
+ Jobs = Top ++ ?JOBS,
+
?assertMatch({201, [#{?OK := true}]},
- req(post, Top ++ ?JOBS, #{type => split, db => <<?DB1>>})),
+ req(post, Jobs, #{type => split, db => <<?DB1>>})),
?assertMatch({201, [#{?OK := true}]},
- req(post, Top ++ ?JOBS, #{type => split, db => <<?DB2>>})),
+ req(post, Jobs, #{type => split, db => <<?DB2>>})),
+
+ ?assertMatch({200, #{<<"total">> := 2}}, req(get, Top ++ ?RESHARD)),
?assertMatch({200, #{
<<"jobs">> := [#{?ID := Id1}, #{?ID := Id2}],
<<"offset">> := 0,
<<"total_rows">> := 2
- }} when Id1 =/= Id2, req(get, Top ++ ?JOBS)),
- ?assertMatch({200, #{<<"total">> := 2}}, req(get, Top ++ ?RESHARD)),
- {200, #{<<"jobs">> := [#{?ID := Id1}, #{?ID := Id2}]}} =
- req(get, Top ++ ?JOBS),
+ }} when Id1 =/= Id2, req(get, Jobs)),
- {200, #{?OK := true}} = req(delete, Top ++ ?JOBS ++ ?b2l(Id1)),
+ {200, #{<<"jobs">> := [#{?ID := Id1}, #{?ID := Id2}]}} = req(get, Jobs),
+
+ {200, #{?OK := true}} = req(delete, Jobs ++ ?b2l(Id1)),
?assertMatch({200, #{<<"total">> := 1}}, req(get, Top ++ ?RESHARD)),
- {200, #{?OK := true}} = req(delete, Top ++ ?JOBS ++ ?b2l(Id2)),
+ {200, #{?OK := true}} = req(delete, Jobs ++ ?b2l(Id2)),
?assertMatch({200, #{<<"total">> := 0}}, req(get, Top ++ ?RESHARD))
end).
+create_multiple_jobs_from_one_post(Top) ->
+ ?_test(begin
+ Jobs = Top ++ ?JOBS,
+ {C1, R1} = req(post, Jobs, #{type => split, db => <<?DB3>>}),
+ ?assertMatch({201, [#{?OK := true}, #{?OK := true}]}, {C1, R1}),
+ ?assertMatch({200, #{<<"total">> := 2}}, req(get, Top ++ ?RESHARD))
+ end).
+
+
start_stop_cluster_basic(Top) ->
?_test(begin
Url = Top ++ ?STATE,
@@ -270,9 +290,7 @@ start_stop_cluster_with_a_job(Top) ->
% Job should start after resharding is started on the cluster
?assertMatch({200, _}, req(put, Url, #{state => running})),
?assertMatch({200, #{?ID := Id2, <<"job_state">> := JSt}}
- when JSt =/= <<"stopped">>, req(get, Top ++ ?JOBS ++ ?b2l(Id2))),
-
- {200, #{?OK := true}} = req(delete, Top ++ ?JOBS ++ ?b2l(Id2))
+ when JSt =/= <<"stopped">>, req(get, Top ++ ?JOBS ++ ?b2l(Id2)))
end).
@@ -308,9 +326,7 @@ individual_job_start_stop(Top) ->
?assertMatch({200, #{<<"state">> := <<"running">>}}, req(get, StUrl)),
% Let it continue running and it should complete eventually
JobPid2 ! continue,
- wait_state(StUrl, <<"completed">>),
-
- ?assertMatch({200, #{?OK := true}}, req(delete, JobUrl))
+ wait_state(StUrl, <<"completed">>)
end).
@@ -353,9 +369,8 @@ individual_job_stop_when_cluster_stopped(Top) ->
?assertMatch({200, #{<<"state">> := <<"running">>}}, req(get, StUrl)),
% Let it continue running and it should complete eventually
JobPid2 ! continue,
- wait_state(StUrl, <<"completed">>),
+ wait_state(StUrl, <<"completed">>)
- ?assertMatch({200, #{?OK := true}}, req(delete, JobUrl))
end).
@@ -504,6 +519,50 @@ completed_job_handling(Top) ->
end).
+handle_source_deletion_in_topoff(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, topoff1),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"failed">>)
+ end).
+
+
+handle_source_deletion_in_initial_copy(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, initial_copy),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"failed">>)
+ end).
+
+
+handle_source_deletion_in_copy_local_docs(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, copy_local_docs),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"failed">>)
+ end).
+
+
+handle_source_deletion_in_build_indices(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, build_indices),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"failed">>)
+ end).
+
+
+handle_source_deletion_in_update_shard_map(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, update_shardmap),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"failed">>)
+ end).
+
+
+handle_source_deletion_in_source_delete(Top) ->
+ ?_test(begin
+ JobId = delete_source_in_state(Top, ?DB1, source_delete),
+ wait_state(Top ++ ?JOBS ++ ?b2l(JobId) ++ "/state", <<"completed">>)
+ end).
+
+
+% Test help functions
+
wait_to_complete_then_cleanup(Top, Jobs) ->
JobsUrl = Top ++ ?JOBS,
lists:foreach(fun(#{?ID := Id}) ->
@@ -540,11 +599,21 @@ wait_state(Url, State) ->
test_util:wait(fun() ->
case req(get, Url) of
{200, #{<<"state">> := State}} -> ok;
- {200, #{}} -> timer:sleep(10), wait
+ {200, #{}} -> timer:sleep(100), wait
end
end).
+delete_source_in_state(Top, Db, State) when is_atom(State) ->
+ intercept_state(State),
+ Body = #{type => split, db => list_to_binary(Db)},
+ {201, [#{?ID := Id}]} = req(post, Top ++ ?JOBS, Body),
+ receive {JobPid, State} -> ok end,
+ delete_db(Top ++ Db),
+ JobPid ! continue,
+ Id.
+
+
create_db(Url) ->
{ok, Status, _, _} = test_request:put(Url, [?JSON, ?AUTH], "{}"),
?assert(Status =:= 201 orelse Status =:= 202).