You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2018/04/24 21:19:50 UTC
[couchdb] 10/14: Add clustered purge API to fabric
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor-2
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit e9f7d65cff3c8f049794e823f989bce43302308a
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Tue Apr 24 12:27:43 2018 -0500
Add clustered purge API to fabric
---
src/fabric/rebar.config | 4 +-
src/fabric/src/fabric.erl | 27 ++-
src/fabric/src/fabric_db_info.erl | 29 +--
src/fabric/src/fabric_db_meta.erl | 26 ++-
src/fabric/src/fabric_doc_open.erl | 42 ++--
src/fabric/src/fabric_doc_purge.erl | 413 ++++++++++++++++++++++++++++++++++++
src/fabric/src/fabric_rpc.erl | 98 ++++++++-
7 files changed, 600 insertions(+), 39 deletions(-)
diff --git a/src/fabric/rebar.config b/src/fabric/rebar.config
index 362c878..3f51af3 100644
--- a/src/fabric/rebar.config
+++ b/src/fabric/rebar.config
@@ -10,5 +10,5 @@
% License for the specific language governing permissions and limitations under
% the License.
-{cover_enabled, true}.
-{cover_print_enabled, true}.
+%{cover_enabled, true}.
+%{cover_print_enabled, true}.
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 4a07271..b2e71dc 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -21,12 +21,13 @@
delete_db/2, get_db_info/1, get_doc_count/1, set_revs_limit/3,
set_security/2, set_security/3, get_revs_limit/1, get_security/1,
get_security/2, get_all_security/1, get_all_security/2,
+ get_purge_infos_limit/1, set_purge_infos_limit/3,
compact/1, compact/2]).
% Documents
-export([open_doc/3, open_revs/4, get_doc_info/3, get_full_doc_info/3,
get_missing_revs/2, get_missing_revs/3, update_doc/3, update_docs/3,
- purge_docs/2, att_receiver/2]).
+ purge_docs/3, att_receiver/2]).
% Views
-export([all_docs/4, all_docs/5, changes/4, query_view/3, query_view/4,
@@ -137,6 +138,18 @@ set_security(DbName, SecObj) ->
set_security(DbName, SecObj, Options) ->
fabric_db_meta:set_security(dbname(DbName), SecObj, opts(Options)).
+%% @doc sets the upper bound for the number of stored purge requests
+-spec set_purge_infos_limit(dbname(), pos_integer(), [option()]) -> ok.
+set_purge_infos_limit(DbName, Limit, Options)
+ when is_integer(Limit), Limit > 0 ->
+ fabric_db_meta:set_purge_infos_limit(dbname(DbName), Limit, opts(Options)).
+
+%% @doc retrieves the upper bound for the number of stored purge requests
+-spec get_purge_infos_limit(dbname()) -> pos_integer() | no_return().
+get_purge_infos_limit(DbName) ->
+ {ok, Db} = fabric_util:get_db(dbname(DbName), [?ADMIN_CTX]),
+ try couch_db:get_purge_infos_limit(Db) after catch couch_db:close(Db) end.
+
get_security(DbName) ->
get_security(DbName, [?ADMIN_CTX]).
@@ -267,8 +280,16 @@ update_docs(DbName, Docs, Options) ->
{aborted, PreCommitFailures}
end.
-purge_docs(_DbName, _IdsRevs) ->
- not_implemented.
+
+%% @doc purge revisions for a list '{Id, Revs}'
+%% returns {ok, {PurgeSeq, Results}}
+-spec purge_docs(dbname(), [{docid(), [revision()]}], [option()]) ->
+ {ok, [{Health, [revision()]}] | {error, any()}} when
+ Health :: ok | accepted.
+purge_docs(DbName, IdsRevs, Options) when is_list(IdsRevs) ->
+ IdsRevs2 = [idrevs(IdRs) || IdRs <- IdsRevs],
+ fabric_doc_purge:go(dbname(DbName), IdsRevs2, opts(Options)).
+
%% @doc spawns a process to upload attachment data and
%% returns a function that shards can use to communicate
diff --git a/src/fabric/src/fabric_db_info.erl b/src/fabric/src/fabric_db_info.erl
index 98e8e52..97a31c2 100644
--- a/src/fabric/src/fabric_db_info.erl
+++ b/src/fabric/src/fabric_db_info.erl
@@ -23,10 +23,12 @@ go(DbName) ->
RexiMon = fabric_util:create_monitors(Shards),
Fun = fun handle_message/3,
{ok, ClusterInfo} = get_cluster_info(Shards),
- Acc0 = {fabric_dict:init(Workers, nil), [{cluster, ClusterInfo}]},
+ Acc0 = {fabric_dict:init(Workers, nil), [], [{cluster, ClusterInfo}]},
try
case fabric_util:recv(Workers, #shard.ref, Fun, Acc0) of
- {ok, Acc} -> {ok, Acc};
+
+ {ok, Acc} ->
+ {ok, Acc};
{timeout, {WorkersDict, _}} ->
DefunctWorkers = fabric_util:remove_done_workers(
WorkersDict,
@@ -37,44 +39,49 @@ go(DbName) ->
"get_db_info"
),
{error, timeout};
- {error, Error} -> throw(Error)
+ {error, Error} ->
+ throw(Error)
end
after
rexi_monitor:stop(RexiMon)
end.
-handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Shard, {Counters, Acc}) ->
+handle_message({rexi_DOWN,
+ _, {_,NodeRef},_}, _Shard, {Counters, PseqAcc, Acc}) ->
case fabric_util:remove_down_workers(Counters, NodeRef) of
{ok, NewCounters} ->
- {ok, {NewCounters, Acc}};
+ {ok, {NewCounters, PseqAcc, Acc}};
error ->
{error, {nodedown, <<"progress not possible">>}}
end;
-handle_message({rexi_EXIT, Reason}, Shard, {Counters, Acc}) ->
+handle_message({rexi_EXIT, Reason}, Shard, {Counters, PseqAcc, Acc}) ->
NewCounters = fabric_dict:erase(Shard, Counters),
case fabric_view:is_progress_possible(NewCounters) of
true ->
- {ok, {NewCounters, Acc}};
+ {ok, {NewCounters, PseqAcc, Acc}};
false ->
{error, Reason}
end;
-handle_message({ok, Info}, #shard{dbname=Name} = Shard, {Counters, Acc}) ->
+handle_message({ok, Info}, #shard{dbname=Name} = Shard, {Counters, PseqAcc, Acc}) ->
case fabric_dict:lookup_element(Shard, Counters) of
undefined ->
% already heard from someone else in this range
- {ok, {Counters, Acc}};
+ {ok, {Counters, PseqAcc, Acc}};
nil ->
Seq = couch_util:get_value(update_seq, Info),
C1 = fabric_dict:store(Shard, Seq, Counters),
C2 = fabric_view:remove_overlapping_shards(Shard, C1),
+ PSeq = couch_util:get_value(purge_seq, Info),
+ NewPseqAcc = [{Shard, PSeq}|PseqAcc],
case fabric_dict:any(nil, C2) of
true ->
- {ok, {C2, [Info|Acc]}};
+ {ok, {C2, NewPseqAcc, [Info|Acc]}};
false ->
{stop, [
{db_name,Name},
+ {purge_seq, fabric_view_changes:pack_seqs(NewPseqAcc)},
{update_seq, fabric_view_changes:pack_seqs(C2)} |
merge_results(lists:flatten([Info|Acc]))
]}
@@ -91,8 +98,6 @@ merge_results(Info) ->
[{doc_count, lists:sum(X)} | Acc];
(doc_del_count, X, Acc) ->
[{doc_del_count, lists:sum(X)} | Acc];
- (purge_seq, X, Acc) ->
- [{purge_seq, lists:sum(X)} | Acc];
(compact_running, X, Acc) ->
[{compact_running, lists:member(true, X)} | Acc];
(disk_size, X, Acc) -> % legacy
diff --git a/src/fabric/src/fabric_db_meta.erl b/src/fabric/src/fabric_db_meta.erl
index 367ef06..26e1b37 100644
--- a/src/fabric/src/fabric_db_meta.erl
+++ b/src/fabric/src/fabric_db_meta.erl
@@ -12,7 +12,8 @@
-module(fabric_db_meta).
--export([set_revs_limit/3, set_security/3, get_all_security/2]).
+-export([set_revs_limit/3, set_security/3, get_all_security/2,
+ set_purge_infos_limit/3]).
-include_lib("fabric/include/fabric.hrl").
-include_lib("mem3/include/mem3.hrl").
@@ -48,6 +49,29 @@ handle_revs_message(Error, _, _Acc) ->
{error, Error}.
+set_purge_infos_limit(DbName, Limit, Options) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, set_purge_infos_limit, [Limit, Options]),
+ Handler = fun handle_purge_message/3,
+ Acc0 = {Workers, length(Workers) - 1},
+ case fabric_util:recv(Workers, #shard.ref, Handler, Acc0) of
+ {ok, ok} ->
+ ok;
+ {timeout, {DefunctWorkers, _}} ->
+ fabric_util:log_timeout(DefunctWorkers, "set_purged_docs_limit"),
+ {error, timeout};
+ Error ->
+ Error
+ end.
+
+handle_purge_message(ok, _, {_Workers, 0}) ->
+ {stop, ok};
+handle_purge_message(ok, Worker, {Workers, Waiting}) ->
+ {ok, {lists:delete(Worker, Workers), Waiting - 1}};
+handle_purge_message(Error, _, _Acc) ->
+ {error, Error}.
+
+
set_security(DbName, SecObj, Options) ->
Shards = mem3:shards(DbName),
RexiMon = fabric_util:create_monitors(Shards),
diff --git a/src/fabric/src/fabric_doc_open.erl b/src/fabric/src/fabric_doc_open.erl
index 9c45bd9..b974880 100644
--- a/src/fabric/src/fabric_doc_open.erl
+++ b/src/fabric/src/fabric_doc_open.erl
@@ -25,7 +25,8 @@
r,
state,
replies,
- q_reply
+ q_reply,
+ replies_by_node=[] %[{Node, Reply}] used for checking if a doc is purged
}).
@@ -83,7 +84,8 @@ handle_message({rexi_EXIT, _Reason}, Worker, Acc) ->
end;
handle_message(Reply, Worker, Acc) ->
NewReplies = fabric_util:update_counter(Reply, 1, Acc#acc.replies),
- NewAcc = Acc#acc{replies = NewReplies},
+ NewNReplies = [{Worker#shard.node, Reply}|Acc#acc.replies_by_node],
+ NewAcc = Acc#acc{replies = NewReplies, replies_by_node = NewNReplies},
case is_r_met(Acc#acc.workers, NewReplies, Acc#acc.r) of
{true, QuorumReply} ->
fabric_util:cleanup(lists:delete(Worker, Acc#acc.workers)),
@@ -122,14 +124,15 @@ is_r_met(Workers, Replies, R) ->
no_more_workers
end.
-read_repair(#acc{dbname=DbName, replies=Replies}) ->
+read_repair(#acc{dbname=DbName, replies=Replies, replies_by_node=NReplies0}) ->
Docs = [Doc || {_, {{ok, #doc{}=Doc}, _}} <- Replies],
+ NReplies = [{Node, Doc} || {Node, {ok, #doc{}=Doc}} <- NReplies0],
case Docs of
% omit local docs from read repair
[#doc{id = <<?LOCAL_DOC_PREFIX, _/binary>>} | _] ->
choose_reply(Docs);
[#doc{id=Id} | _] ->
- Opts = [replicated_changes, ?ADMIN_CTX],
+ Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NReplies}],
Res = fabric:update_docs(DbName, Docs, Opts),
case Res of
{ok, []} ->
@@ -319,7 +322,8 @@ handle_message_reply_test() ->
?assertEqual(
{ok, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]
}},
handle_message(foo, Worker2, Acc0)
),
@@ -327,7 +331,8 @@ handle_message_reply_test() ->
?assertEqual(
{ok, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, bar}]
}},
handle_message(bar, Worker2, Acc0#acc{
replies=[fabric_util:kv(foo,1)]
@@ -339,18 +344,21 @@ handle_message_reply_test() ->
% is returned. Bit subtle on the assertions here.
?assertEqual(
- {stop, Acc0#acc{workers=[],replies=[fabric_util:kv(foo,1)]}},
+ {stop, Acc0#acc{workers=[],replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]}},
handle_message(foo, Worker0, Acc0#acc{workers=[Worker0]})
),
?assertEqual(
{stop, Acc0#acc{
workers=[],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, bar}, {undefined, foo}]
}},
handle_message(bar, Worker0, Acc0#acc{
workers=[Worker0],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]
})
),
@@ -362,11 +370,13 @@ handle_message_reply_test() ->
workers=[],
replies=[fabric_util:kv(foo,2)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}, {undefined, foo}]
}},
handle_message(foo, Worker1, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, foo}]
})
),
@@ -376,7 +386,8 @@ handle_message_reply_test() ->
r=1,
replies=[fabric_util:kv(foo,1)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}]
}},
handle_message(foo, Worker0, Acc0#acc{r=1})
),
@@ -386,11 +397,14 @@ handle_message_reply_test() ->
workers=[],
replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,2)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}, {undefined, foo},
+ {undefined, bar}]
}},
handle_message(foo, Worker0, Acc0#acc{
workers=[Worker0],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, foo}, {undefined, bar}]
})
),
diff --git a/src/fabric/src/fabric_doc_purge.erl b/src/fabric/src/fabric_doc_purge.erl
new file mode 100644
index 0000000..b51f28e
--- /dev/null
+++ b/src/fabric/src/fabric_doc_purge.erl
@@ -0,0 +1,413 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(fabric_doc_purge).
+
+
+-export([
+ go/3
+]).
+
+
+-include_lib("fabric/include/fabric.hrl").
+-include_lib("mem3/include/mem3.hrl").
+
+
+-record(acc, {
+ worker_uuids,
+ req_count,
+ resps,
+ w
+}).
+
+
+go(_, [], _) ->
+ {ok, []};
+go(DbName, IdsRevs, Options) ->
+ % Generate our purge requests of {UUID, DocId, Revs}
+ {UUIDs, Reqs, Count} = create_reqs(IdsRevs, [], [], 0),
+
+ % Fire off rexi workers for each shard.
+ {Workers, WorkerUUIDs} = dict:fold(fun(Shard, ShardReqs, {Ws, WUUIDs}) ->
+ #shard{name = ShardDbName, node = Node} = Shard,
+ Args = [ShardDbName, ShardReqs, Options],
+ Ref = rexi:cast(Node, {fabric_rpc, purge_docs, Args}),
+ Worker = Shard#shard{ref=Ref},
+ ShardUUIDs = [UUID || {UUID, _Id, _Revs} <- ShardReqs],
+ {[Worker | Ws], [{Worker, ShardUUIDs} | WUUIDs]}
+ end, {[], []}, group_reqs_by_shard(DbName, Reqs)),
+
+ RexiMon = fabric_util:create_monitors(Workers),
+ Timeout = fabric_util:request_timeout(),
+ Acc0 = #acc{
+ worker_uuids = WorkerUUIDs,
+ req_count = Count,
+ resps = dict:from_list([{UUID, []} || UUID <- UUIDs]),
+ w = w(DbName, Options)
+ },
+ Acc2 = try rexi_utils:recv(Workers, #shard.ref,
+ fun handle_message/3, Acc0, infinity, Timeout) of
+ {ok, Acc1} ->
+ Acc1;
+ {timeout, Acc1} ->
+ #acc{
+ worker_uuids = WorkerUUIDs,
+ resps = Resps
+ } = Acc1,
+ DefunctWorkers = [Worker || {Worker, _} <- WorkerUUIDs],
+ fabric_util:log_timeout(DefunctWorkers, "purge_docs"),
+ NewResps = append_errors(timeout, WorkerUUIDs, Resps),
+ Acc1#acc{worker_uuids = [], resps = NewResps};
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end,
+
+ {ok, format_resps(UUIDs, Acc2)}.
+
+
+handle_message({rexi_DOWN, _, {_, Node}, _}, _Worker, Acc) ->
+ #acc{
+ worker_uuids = WorkerUUIDs,
+ resps = Resps
+ } = Acc,
+ Pred = fun({#shard{node = N}, _}) -> N == Node end,
+ {Failed, Rest} = lists:partition(Pred, WorkerUUIDs),
+ NewResps = append_errors(internal_server_error, Failed, Resps),
+ maybe_stop(Acc#acc{worker_uuids = Rest, resps = NewResps});
+
+handle_message({rexi_EXIT, _}, Worker, Acc) ->
+ #acc{
+ worker_uuids = WorkerUUIDs,
+ resps = Resps
+ } = Acc,
+ {value, WorkerPair, Rest} = lists:keytake(Worker, 1, WorkerUUIDs),
+ NewResps = append_errors(internal_server_error, [WorkerPair], Resps),
+ maybe_stop(Acc#acc{worker_uuids = Rest, resps = NewResps});
+
+handle_message({ok, Replies}, Worker, Acc) ->
+ #acc{
+ worker_uuids = WorkerUUIDs,
+ resps = Resps
+ } = Acc,
+ {value, {_W, UUIDs}, Rest} = lists:keytake(Worker, 1, WorkerUUIDs),
+ couch_log:error("XKCD: ~p ~p :: ~p ~p", [length(UUIDs), length(Replies), UUIDs, Replies]),
+ NewResps = append_resps(UUIDs, Replies, Resps),
+ maybe_stop(Acc#acc{worker_uuids = Rest, resps = NewResps});
+
+handle_message({bad_request, Msg}, _, _) ->
+ throw({bad_request, Msg}).
+
+
+create_reqs([], UUIDs, Reqs, Count) ->
+ {lists:reverse(UUIDs), lists:reverse(Reqs), Count};
+
+create_reqs([{Id, Revs} | RestIdsRevs], UUIDs, Reqs, Count) ->
+ UUID = couch_uuids:new(),
+ NewUUIDs = [UUID | UUIDs],
+ NewReqs = [{UUID, Id, Revs} | Reqs],
+ create_reqs(RestIdsRevs, NewUUIDs, NewReqs, Count + 1).
+
+
+group_reqs_by_shard(DbName, Reqs) ->
+ lists:foldl(fun({_UUID, Id, _Revs} = Req, D0) ->
+ lists:foldl(fun(Shard, D1) ->
+ dict:append(Shard, Req, D1)
+ end, D0, mem3:shards(DbName, Id))
+ end, dict:new(), Reqs).
+
+
+w(DbName, Options) ->
+ try
+ list_to_integer(couch_util:get_value(w, Options))
+ catch _:_ ->
+ mem3:quorum(DbName)
+ end.
+
+
+append_errors(Type, WorkerUUIDs, Resps) ->
+ lists:foldl(fun({_Worker, UUIDs}, RespAcc) ->
+ Errors = [{error, Type} || _UUID <- UUIDs],
+ append_resps(UUIDs, Errors, RespAcc)
+ end, Resps, WorkerUUIDs).
+
+
+append_resps([], [], Resps) ->
+ Resps;
+append_resps([UUID | RestUUIDs], [Reply | RestReplies], Resps) ->
+ NewResps = dict:append(UUID, Reply, Resps),
+ append_resps(RestUUIDs, RestReplies, NewResps).
+
+
+maybe_stop(#acc{worker_uuids = []} = Acc) ->
+ {stop, Acc};
+maybe_stop(#acc{resps = Resps, w = W} = Acc) ->
+ try
+ dict:fold(fun(_UUID, UUIDResps, _) ->
+ couch_log:error("XKCD: ~p ~p", [UUIDResps, W]),
+ case has_quorum(UUIDResps, W) of
+ true -> ok;
+ false -> throw(keep_going)
+ end
+ end, nil, Resps),
+ {stop, Acc}
+ catch throw:keep_going ->
+ {ok, Acc}
+ end.
+
+
+format_resps(UUIDs, #acc{} = Acc) ->
+ #acc{
+ resps = Resps,
+ w = W
+ } = Acc,
+ FoldFun = fun(UUID, Replies, ReplyAcc) ->
+ OkReplies = [Reply || {ok, Reply} <- Replies],
+ case OkReplies of
+ [] ->
+ [Error | _] = lists:usort(Replies),
+ [{UUID, Error} | ReplyAcc];
+ _ ->
+ AllRevs = lists:usort(lists:flatten(OkReplies)),
+ Health = if length(OkReplies) >= W -> ok; true -> accepted end,
+ [{UUID, {Health, AllRevs}} | ReplyAcc]
+ end
+ end,
+ FinalReplies = dict:fold(FoldFun, {ok, []}, Resps),
+ couch_util:reorder_results(UUIDs, FinalReplies);
+
+format_resps(_UUIDs, Else) ->
+ Else.
+
+
+has_quorum([], W) when W > 0 ->
+ false;
+has_quorum(_, W) when W =< 0 ->
+ true;
+has_quorum([{ok, _} | Rest], W) when W > 0 ->
+ has_quorum(Rest, W - 1).
+
+
+%% % eunits
+%% doc_purge_ok_test() ->
+%% meck:new(couch_log),
+%% meck:expect(couch_log, warning, fun(_,_) -> ok end),
+%% meck:expect(couch_log, notice, fun(_,_) -> ok end),
+%%
+%% Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+%% UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+%% Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+%% UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+%% UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+%% Shards =
+%% mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+%% Counters = dict:to_list(
+%% group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+%% DocsDict = dict:new(),
+%%
+%% % ***test for W = 2
+%% AccW2 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCountW2_1,_,_,_,_} = AccW2_1} =
+%% handle_message({ok,[{ok, Revs1}, {ok, Revs2}]}, hd(Shards), AccW2),
+%% ?assertEqual(2, WaitingCountW2_1),
+%% {stop, FinalReplyW2 } =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+%% lists:nth(2,Shards), AccW2_1),
+%% ?assertEqual(
+%% {ok, [{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+%% FinalReplyW2
+%% ),
+%%
+%% % ***test for W = 3
+%% AccW3 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCountW3_1,_,_,_,_} = AccW3_1} =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), AccW3),
+%% ?assertEqual(2, WaitingCountW3_1),
+%% {ok, {WaitingCountW3_2,_,_,_,_} = AccW3_2} =
+%% handle_message({ok,[{ok, Revs1}, {ok, Revs2}]},
+%% lists:nth(2,Shards), AccW3_1),
+%% ?assertEqual(1, WaitingCountW3_2),
+%% {stop, FinalReplyW3 } =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+%% lists:nth(3,Shards), AccW3_2),
+%% ?assertEqual(
+%% {ok, [{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+%% FinalReplyW3
+%% ),
+%%
+%% % *** test rexi_exit on 1 node
+%% Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), Acc0),
+%% ?assertEqual(2, WaitingCount1),
+%% {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+%% handle_message({rexi_EXIT, nil}, lists:nth(2,Shards), Acc1),
+%% ?assertEqual(1, WaitingCount2),
+%% {stop, Reply} =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+%% lists:nth(3,Shards), Acc2),
+%% ?assertEqual(
+%% {ok,[{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+%% Reply
+%% ),
+%%
+%% % *** test {error, purge_during_compaction_exceeded_limit} on all nodes
+%% % *** still should return ok reply for the request
+%% ErrPDCEL = {error, purge_during_compaction_exceeded_limit},
+%% Acc20 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCount21,_,_,_,_} = Acc21} =
+%% handle_message({ok, [ErrPDCEL, ErrPDCEL]}, hd(Shards), Acc20),
+%% ?assertEqual(2, WaitingCount21),
+%% {ok, {WaitingCount22,_,_,_,_} = Acc22} =
+%% handle_message({ok, [ErrPDCEL, ErrPDCEL]}, lists:nth(2,Shards), Acc21),
+%% ?assertEqual(1, WaitingCount22),
+%% {stop, Reply2 } =
+%% handle_message({ok, [ErrPDCEL, ErrPDCEL]}, lists:nth(3,Shards), Acc22),
+%% ?assertEqual(
+%% {ok, [{UUID1, ErrPDCEL}, {UUID2, ErrPDCEL}]},
+%% Reply2
+%% ),
+%%
+%% % *** test {error, purged_docs_limit_exceeded} on all nodes
+%% % *** still should return ok reply for the request
+%% ErrPDLE = {error, purged_docs_limit_exceeded},
+%% Acc30 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCount31,_,_,_,_} = Acc31} =
+%% handle_message({ok, [ErrPDLE, ErrPDLE]}, hd(Shards), Acc30),
+%% ?assertEqual(2, WaitingCount31),
+%% {ok, {WaitingCount32,_,_,_,_} = Acc32} =
+%% handle_message({ok, [ErrPDLE, ErrPDLE]}, lists:nth(2,Shards), Acc31),
+%% ?assertEqual(1, WaitingCount32),
+%% {stop, Reply3 } =
+%% handle_message({ok, [ErrPDLE, ErrPDLE]},lists:nth(3,Shards), Acc32),
+%% ?assertEqual(
+%% {ok, [{UUID1, ErrPDLE}, {UUID2, ErrPDLE}]},
+%% Reply3
+%% ),
+%% meck:unload(couch_log).
+%%
+%%
+%% doc_purge_accepted_test() ->
+%% meck:new(couch_log),
+%% meck:expect(couch_log, warning, fun(_,_) -> ok end),
+%% meck:expect(couch_log, notice, fun(_,_) -> ok end),
+%%
+%% Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+%% UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+%% Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+%% UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+%% UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+%% Shards =
+%% mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+%% Counters = dict:to_list(
+%% group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+%% DocsDict = dict:new(),
+%%
+%% % *** test rexi_exit on 2 nodes
+%% Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+%% handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), Acc0),
+%% ?assertEqual(2, WaitingCount1),
+%% {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+%% handle_message({rexi_EXIT, nil}, lists:nth(2, Shards), Acc1),
+%% ?assertEqual(1, WaitingCount2),
+%% {stop, Reply} =
+%% handle_message({rexi_EXIT, nil}, lists:nth(3, Shards), Acc2),
+%% ?assertEqual(
+%% {accepted, [{UUID1, {accepted, Revs1}}, {UUID2, {accepted, Revs2}}]},
+%% Reply
+%% ),
+%% meck:unload(couch_log).
+%%
+%%
+%% doc_purge_error_test() ->
+%% meck:new(couch_log),
+%% meck:expect(couch_log, warning, fun(_,_) -> ok end),
+%% meck:expect(couch_log, notice, fun(_,_) -> ok end),
+%%
+%% Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+%% UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+%% Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+%% UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+%% UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+%% Shards =
+%% mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+%% Counters = dict:to_list(
+%% group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+%% DocsDict = dict:new(),
+%%
+%% % *** test rexi_exit on all 3 nodes
+%% Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+%% Counters, DocsDict},
+%% {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+%% handle_message({rexi_EXIT, nil}, hd(Shards), Acc0),
+%% ?assertEqual(2, WaitingCount1),
+%% {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+%% handle_message({rexi_EXIT, nil}, lists:nth(2,Shards), Acc1),
+%% ?assertEqual(1, WaitingCount2),
+%% {stop, Reply} =
+%% handle_message({rexi_EXIT, nil}, lists:nth(3,Shards), Acc2),
+%% ?assertEqual(
+%% {error, [{UUID1, {error, internal_server_error}},
+%% {UUID2, {error, internal_server_error}}]},
+%% Reply
+%% ),
+%%
+%% % ***test w quorum > # shards, which should fail immediately
+%% Shards2 = mem3_util:create_partition_map("foo",1,1,["node1"]),
+%% Counters2 = dict:to_list(
+%% group_idrevs_by_shard_hack(<<"foo">>, Shards2, UUIDsIDdsRevs)),
+%% AccW4 = {length(Shards2), length(UUIDsIDdsRevs), list_to_integer("2"),
+%% Counters2, DocsDict},
+%% Bool =
+%% case handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+%% hd(Shards), AccW4) of
+%% {stop, _Reply} ->
+%% true;
+%% _ -> false
+%% end,
+%% ?assertEqual(true, Bool),
+%%
+%% % *** test Docs with no replies should end up as {error, internal_server_error}
+%% SA1 = #shard{node = a, range = [1]},
+%% SA2 = #shard{node = a, range = [2]},
+%% SB1 = #shard{node = b, range = [1]},
+%% SB2 = #shard{node = b, range = [2]},
+%% Counters3 = [{SA1,[UUID1]}, {SB1,[UUID1]},
+%% {SA2,[UUID2]}, {SB2,[UUID2]}],
+%% Acc30 = {length(Counters3), length(UUIDsIDdsRevs), 2, Counters3, DocsDict},
+%% {ok, Acc31} = handle_message({ok, [{ok, Revs1}]}, SA1, Acc30),
+%% {ok, Acc32} = handle_message({rexi_EXIT, nil}, SB1, Acc31),
+%% {ok, Acc33} = handle_message({rexi_EXIT, nil}, SA2, Acc32),
+%% {stop, Acc34} = handle_message({rexi_EXIT, nil}, SB2, Acc33),
+%% ?assertEqual(
+%% {error, [{UUID1, {accepted, Revs1}},
+%% {UUID2, {error, internal_server_error}}]},
+%% Acc34
+%% ),
+%% meck:unload(couch_log).
+%%
+%%
+%% % needed for testing to avoid having to start the mem3 application
+%% group_idrevs_by_shard_hack(_DbName, Shards, UUIDsIdsRevs) ->
+%% lists:foldl(fun({UUID, _Id, _Revs}, Dict0) ->
+%% lists:foldl(fun(Shard, Dict1) ->
+%% dict:append(Shard, UUID, Dict1)
+%% end, Dict0, Shards)
+%% end, dict:new(), UUIDsIdsRevs).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 4a69e7e..a64d546 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -21,6 +21,7 @@
delete_shard_db_doc/2]).
-export([get_all_security/2, open_shard/2]).
-export([compact/1, compact/2]).
+-export([get_purge_seq/2, purge_docs/3, set_purge_infos_limit/3]).
-export([get_db_info/2, get_doc_count/2, get_update_seq/2,
changes/4, map_view/5, reduce_view/5, group_info/3, update_mrview/4]).
@@ -201,6 +202,9 @@ get_all_security(DbName, Options) ->
set_revs_limit(DbName, Limit, Options) ->
with_db(DbName, Options, {couch_db, set_revs_limit, [Limit]}).
+set_purge_infos_limit(DbName, Limit, Options) ->
+ with_db(DbName, Options, {couch_db, set_purge_infos_limit, [Limit]}).
+
open_doc(DbName, DocId, Options) ->
with_db(DbName, Options, {couch_db, open_doc, [DocId, Options]}).
@@ -236,14 +240,25 @@ get_missing_revs(DbName, IdRevsList, Options) ->
end).
update_docs(DbName, Docs0, Options) ->
- case proplists:get_value(replicated_changes, Options) of
- true ->
- X = replicated_changes;
- _ ->
- X = interactive_edit
+ X = case proplists:get_value(replicated_changes, Options) of
+ true -> replicated_changes;
+ _ -> interactive_edit
end,
- Docs = make_att_readers(Docs0),
- with_db(DbName, Options, {couch_db, update_docs, [Docs, Options, X]}).
+ DocsByNode = couch_util:get_value(read_repair, Options),
+ case {X, DocsByNode} of
+ {_, undefined} ->
+ Docs = make_att_readers(Docs0),
+ with_db(DbName, Options,
+ {couch_db, update_docs, [Docs, Options, X]});
+ {replicated_changes, _} ->
+ update_docs_read_repair(DbName, DocsByNode, Options)
+ end.
+
+get_purge_seq(DbName, Options) ->
+ with_db(DbName, Options, {couch_db, get_purge_seq, []}).
+
+purge_docs(DbName, UUIdsIdsRevs, Options) ->
+ with_db(DbName, Options, {couch_db, purge_docs, [UUIdsIdsRevs, Options]}).
%% @equiv group_info(DbName, DDocId, [])
group_info(DbName, DDocId) ->
@@ -298,6 +313,75 @@ with_db(DbName, Options, {M,F,A}) ->
rexi:reply(Error)
end.
+
+update_docs_read_repair(DbName, DocsByNode, Options) ->
+ set_io_priority(DbName, Options),
+ case get_or_create_db(DbName, Options) of
+ {ok, Db} ->
+ % omit Revisions that have been purged
+ Docs = filter_purged_revs(Db, DocsByNode),
+ Docs2 = make_att_readers(Docs),
+ {M,F,A} = {couch_db, update_docs, [Docs2, Options, replicated_changes]},
+ rexi:reply(try
+ apply(M, F, [Db | A])
+ catch Exception ->
+ Exception;
+ error:Reason ->
+ couch_log:error("rpc ~p:~p/~p ~p ~p", [M, F, length(A)+1, Reason,
+ clean_stack()]),
+ {error, Reason}
+ end);
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
+% given [{Node, Doc}] diff revs of the same DocID from diff nodes
+% returns [Doc] filtering out purged docs.
+% This is done for read-repair from fabric_doc_open,
+% so that not to recreate Docs that have been purged before
+% on this node() from Nodes that are out of sync.
+filter_purged_revs(Db, DocsByNode) ->
+ AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
+ {ok, DbPSeq} = couch_db:get_purge_seq(Db),
+ PurgeFoldFun = fun({_P,_U, Id, Revs}, Acc) -> [{Id, Revs}|Acc] end,
+ V = "v" ++ config:get("purge", "version", "1") ++ "-",
+ StartKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem3-"),
+ EndKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem31"),
+ Opts = [{start_key, StartKey}, {end_key_gt, EndKey}],
+ % go through _local/purge-mem3-.. docs
+ % find Node that this LDoc corresponds to
+ % check if update from Node has not been recently purged on current node
+ LDocsFoldFun = fun(#doc{body={Props}}, Acc) ->
+ {VOps} = couch_util:get_value(<<"verify_options">>, Props),
+ Node = couch_util:get_value(<<"node">>, VOps),
+ Result = lists:keyfind(Node, 1, DocsByNode),
+ NewAcc = if not Result -> Acc; true ->
+ {Node, Doc} = Result,
+ NodePSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ if NodePSeq == DbPSeq ->
+ [Doc|Acc];
+ (NodePSeq+AllowedPSeqLag) < DbPSeq ->
+ % Node is very out of sync, ignore updates from it
+ Acc;
+ true -> %(NodePSeq+ClientAllowedPSeqLag) >= DbPSeq
+ % if Doc has been purged recently, than ignore it
+ {ok, PurgedIdsRevs} = couch_db:fold_purged_docs(Db,
+ NodePSeq, PurgeFoldFun, [], []),
+ {Start, [FirstRevId|_]} = Doc#doc.revs,
+ DocIdRevs = {Doc#doc.id, [{Start, FirstRevId}]},
+ case lists:member(DocIdRevs, PurgedIdsRevs) of
+ true -> Acc;
+ false -> [Doc|Acc]
+ end
+ end
+ end,
+ {ok, NewAcc}
+ end,
+ {ok, Docs} = couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
+ Docs.
+
+
get_or_create_db(DbName, Options) ->
couch_db:open_int(DbName, [{create_if_missing, true} | Options]).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.