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/03/21 18:26:18 UTC
[couchdb] branch COUCHDB-3326-clustered-purge-davisp-refactor
updated (1a46337 -> 79e3d22)
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a change to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
discard 1a46337 WIP - Internal replication
discard 45e5fdb Temporarily disable should_compare_compression_methods/1
discard 4b1f80a Fix tests to work with clustered purge
discard e878d66 Add metrics for clustered purge
discard da04896 Implement clustered purge HTTP endpoint
discard ca2867a Update fabric_doc_open_revs to handle purges
discard 1495b4b Implement clustered purge API
discard 195fb5a WIP - Add internal replication of purges
new 1b2eeeb WIP - Add internal replication of purges
new 509548f Implement clustered purge API
new 57719ec Update fabric_doc_open_revs to handle purges
new 90ac409 Implement clustered purge HTTP endpoint
new 633d838 Add metrics for clustered purge
new d3a2b59 Fix tests to work with clustered purge
new 79e3d22 Temporarily disable should_compare_compression_methods/1
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1a46337)
\
N -- N -- N refs/heads/COUCHDB-3326-clustered-purge-davisp-refactor (79e3d22)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 01/07: WIP - Add internal replication of purges
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 1b2eeeb93bddf8c889ec362ce47cf43ee0c7669b
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 21 12:26:11 2018 -0500
WIP - Add internal replication of purges
---
src/mem3/src/mem3_rep.erl | 193 +++++++++++++++++++++++++++++++++++++++++++---
src/mem3/src/mem3_rpc.erl | 72 ++++++++++++++++-
2 files changed, 252 insertions(+), 13 deletions(-)
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
index 6d38af7..338d20e 100644
--- a/src/mem3/src/mem3_rep.erl
+++ b/src/mem3/src/mem3_rep.erl
@@ -17,6 +17,8 @@
go/2,
go/3,
make_local_id/2,
+ make_purge_id/2,
+ verify_purge_checkpoint/3,
find_source_seq/4
]).
@@ -35,6 +37,7 @@
infos = [],
seq = 0,
localid,
+ purgeid,
source,
target,
filter,
@@ -118,6 +121,34 @@ make_local_id(SourceThing, TargetThing, Filter) ->
<<"_local/shard-sync-", S/binary, "-", T/binary, F/binary>>.
+make_purge_id(SourceUUID, TargetUUID) ->
+ <<"_local/purge-mem3-", SourceUUID/binary, "-", TargetUUID/binary>>.
+
+
+verify_purge_checkpoint(_Db, _DocId, Props) ->
+ DbName = couch_util:get_value(<<"dbname">>, Props),
+ SourceBin = couch_util:get_value(<<"source">>, Props),
+ TargetBin = couch_util:get_value(<<"target">>, Props),
+ Range = couch_util:get_value(<<"range">>, Props),
+
+ Source = binary_to_existing_atom(SourceBin, latin1),
+ Target = binary_to_existing_atom(TargetBin, latin1),
+
+ try
+ Shards = mem3:shards(DbName),
+ Nodes = lists:foldl(fun(Shard, Acc) ->
+ case Shard#shard.range == Range of
+ true -> [Shard#shard.node | Acc];
+ false -> Acc
+ end
+ end, [], mem3:shards(DbName)),
+ lists:member(Source, Nodes) andalso lists:member(Target, Nodes)
+ catch
+ error:database_does_not_exist ->
+ false
+ end.
+
+
%% @doc Find and return the largest update_seq in SourceDb
%% that the client has seen from TargetNode.
%%
@@ -169,20 +200,122 @@ find_source_seq_int(#doc{body={Props}}, SrcNode0, TgtNode0, TgtUUID, TgtSeq) ->
end.
-repl(#acc{db = Db} = Acc0) ->
- erlang:put(io_priority, {internal_repl, couch_db:name(Db)}),
- #acc{seq=Seq} = Acc1 = calculate_start_seq(Acc0),
- case Seq >= couch_db:get_update_seq(Db) of
- true ->
- {ok, 0};
- false ->
- Fun = fun ?MODULE:changes_enumerator/2,
- {ok, Acc2} = couch_db:fold_changes(Db, Seq, Fun, Acc1),
- {ok, #acc{seq = LastSeq}} = replicate_batch(Acc2),
- {ok, couch_db:count_changes_since(Db, LastSeq)}
+repl(Acc0#acc{db = Db0}) ->
+ erlang:put(io_priority, {internal_repl, couch_db:name(Db0)}),
+ #acc{seq = Seq} = Acc1 = calculate_start_seq(Acc0),
+ try
+ Acc3 = case config:get_boolean("mem3", "replicate_purges", false) of
+ true ->
+ Acc2 = pull_purges(Acc1),
+ push_purges(Acc2);
+ false ->
+ Acc1
+ end,
+ push_changes(Acc3)
+ catch
+ throw:{finished, Count} ->
+ {ok, Count}
end.
+pull_purges(#acc{} = Acc0) ->
+ #acc{
+ batch_size = Count,
+ seq = UpdateSeq,
+ target = Target
+ } = Acc0,
+ #shard{
+ node = TgtNode,
+ name = TgtDbName
+ } = Target,
+
+ {Acc2, RemToPull} = with_src_db(Acc0, fun(Db) ->
+ SrcUUID = couch_db:get_uuid(Db),
+ {ok, LocalPurgeId, Infos, ThroughSeq, Remaining} =
+ mem3_rpc:load_purge_infos(TNode, DbName, SrcUUID, Count),
+
+ if Infos == [] -> ok; true ->
+ {ok, _} = couch_db:purge_docs(Db, Infos, [replicated_edits]),
+ Body = purge_cp_body(Acc0, ThroughSeq),
+ ok = mem3_rpc:save_purge_checkpoint(
+ TgtNode, TgtDbName, LocalPurgeId, Body)
+ end,
+
+ if Remaining =< 0 -> ok; true ->
+ OldestPurgeSeq = couch_db:get_oldest_purge_seq(Db),
+ PurgesToPush = couch_db:get_purge_seq(Db) - OldestPurgeSeq,
+ Changes = couch_db:count_changes_since(Db, UpdateSeq),
+ throw({finished, Remaining + PurgesToPush + Changes})
+ end
+
+ Acc0#acc{purgeid = LocalPurgeId}
+ end).
+
+
+push_purges(#acc{} = Acc0) ->
+ #acc{
+ batch_size = BatchSize,
+ purgeid = LocalPurgeId,
+ seq = UpdateSeq,
+ target = Target
+ } = Acc0,
+ #shard{
+ node = TgtNode,
+ name = TgtDbName
+ } = Target,
+
+ with_src_db(Acc0, fun(Db) ->
+ FoldFun = fun({PSeq, UUID, Id, Revs}, {Count, Infos, _}) ->
+ NewCount = Count + length(Revs),
+ NewInfos = [{UUID, Id, Revs} | Infos],
+ Status = if NewCount < BatchSize -> ok; true -> stop end,
+ {Status, {NewCount, NewInfos, PSeq}}
+ end,
+ InitAcc = {0, [], StartSeq}
+ {ok, {_, Infos, ThroughSeq}} =
+ couch_db:fold_purge_infos(Db, StartSeq, FoldFun, InitAcc),
+
+ if Infos == [] -> ok; true ->
+ ok = purge_on_target(TgtNode, TgtDbName, Infos),
+ Doc = #doc{
+ id = LocalPurgeId,
+ body = purge_cp_body(Acc0, ThroughSeq)
+ },
+ {ok, _} = couch_db:update_doc(Db, Doc, [])
+ end,
+
+ {ok, PurgeSeq} = couch_db:get_purge_seq(Db),
+ if ThroughSeq >= PurgeSeq -> ok; true ->
+ Remaining = PurgeSeq - ThroughSeq,
+ Changes = couch_db:count_changes_since(Db, UpdateSeq),
+ throw({finished, Remaining + Changes})
+ end,
+
+ Acc0
+ end).
+
+
+push_changes(#acc{} = Acc0) ->
+ #acc{
+ db = Db0,
+ seq = Seq
+ } = Acc0,
+
+ % Avoid needless rewriting the internal replication
+ % checkpoint document if nothing is replicated.
+ if Seq < couch_db:get_update_seq(Db0) -> ok; true ->
+ throw({finished, 0})
+ end
+
+ with_src_db(Acc0, fun(Db) ->
+ Acc1 = Acc#acc{db = Db},
+ Fun = fun ?MODULE:changes_enumerator/2,
+ {ok, Acc2} = couch_db:fold_changes(Db, Seq, Fun, Acc1),
+ {ok, #acc{seq = LastSeq}} = replicate_batch(Acc2),
+ {ok, couch_db:count_changes_since(Db, LastSeq)}
+ end).
+
+
calculate_start_seq(Acc) ->
#acc{
db = Db,
@@ -323,6 +456,15 @@ save_on_target(Node, Name, Docs) ->
ok.
+purge_on_target(Node, Name, PurgeInfos) ->
+ {ok, _} = mem3_rpc:purge_docs(Node, Name, PurgeInfos, [
+ replicated_changes,
+ full_commit,
+ ?ADMIN_CTX,
+ {io_priority, {internal_repl, Name}}
+ ]),
+ ok.
+
update_locals(Acc) ->
#acc{seq=Seq, db=Db, target=Target, localid=Id, history=History} = Acc,
#shard{name=Name, node=Node} = Target,
@@ -336,6 +478,26 @@ update_locals(Acc) ->
{ok, _} = couch_db:update_doc(Db, #doc{id = Id, body = NewBody}, []).
+purge_cp_body(#acc{} = Acc, PurgeSeq) ->
+ #acc{
+ source = Source,
+ target = Target
+ } = Acc,
+ {Mega, Secs, _} = os:timestamp(),
+ NowSecs = Mega * 1000000 + Secs,
+ Body = {[
+ {<<"type">>, <<"internal_replication">>},
+ {<<"updated_on">>, NowSecs},
+ {<<"purge_seq">>, PurgeSeq},
+ {<<"verify_module">>, <<"mem3_rep">>},
+ {<<"verify_function">>, <<"verify_purge_checkpoint">>},
+ {<<"dbname">>, Source#shard.dbname},
+ {<<"source">>, atom_to_binary(Source#shard.node)},
+ {<<"target">>, atom_to_binary(Target#shard.node)},
+ {<<"range">>, Source#shard.range}
+ ]}.
+
+
find_repl_doc(SrcDb, TgtUUIDPrefix) ->
SrcUUID = couch_db:get_uuid(SrcDb),
S = couch_util:encodeBase64Url(crypto:hash(md5, term_to_binary(SrcUUID))),
@@ -368,6 +530,15 @@ find_repl_doc(SrcDb, TgtUUIDPrefix) ->
end.
+with_src_db(#acc{source = Source}, Fun) ->
+ {ok, Db} = couch_db:open(Source#shard.name, [?ADMIN_CTX]),
+ try
+ Fun(Db)
+ after
+ couch_db:close(Db)
+ end.
+
+
is_prefix(Prefix, Subject) ->
binary:longest_common_prefix([Prefix, Subject]) == size(Prefix).
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index c2bd58f..ab293d7 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -20,14 +20,21 @@
get_missing_revs/4,
update_docs/4,
load_checkpoint/4,
- save_checkpoint/6
+ save_checkpoint/6,
+
+ load_purge_infos/4,
+ save_purge_checkpoint/4,
+ purge_docs/4
]).
% Private RPC callbacks
-export([
find_common_seq_rpc/3,
load_checkpoint_rpc/3,
- save_checkpoint_rpc/5
+ save_checkpoint_rpc/5,
+
+ load_purge_infos_rpc/3,
+ save_purge_checkpoint_rpc/3
]).
@@ -58,6 +65,20 @@ find_common_seq(Node, DbName, SourceUUID, SourceEpochs) ->
rexi_call(Node, {mem3_rpc, find_common_seq_rpc, Args}).
+load_purge_infos(Node, DbName, SourceUUID, Count) ->
+ Args = [DbName, SourceUUID, Count],
+ rexi_call(Node, {mem3_rpc, load_purge_infos_rpc, Args}).
+
+
+save_purge_checkpoint(Node, DbName, PurgeDocId, Body) ->
+ Args = [DbName, PurgeDocId, Body],
+ rexi_call(Node, {mem3_rpc, save_purge_checkpoint, Args}).
+
+
+purge_docs(Node, DbName, PurgeInfos, Options) ->
+ rexi_call(Node, {fabric_rpc, purge_docs, [DbName, PurgeInfos, Options]}).
+
+
load_checkpoint_rpc(DbName, SourceNode, SourceUUID) ->
erlang:put(io_priority, {internal_repl, DbName}),
case get_or_create_db(DbName, [?ADMIN_CTX]) of
@@ -128,6 +149,53 @@ find_common_seq_rpc(DbName, SourceUUID, SourceEpochs) ->
end.
+load_purge_infos_rpc(DbName, SrcUUID, BatchSize) ->
+ erlang:put(io_priority, {internal_repl, DbName}),
+ case get_or_create_db(DbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ TgtUUID = couch_db:get_uuid(Db),
+ PurgeDocId = mem3_rep:make_purge_id(SrcUUID, TgtUUID),
+ StartSeq = case couch_db:open_doc(Db, PurgeDocId, []) of
+ {ok, #doc{props = {Props}}} ->
+ couch_util:get_value(<<"purge_seq">>, Props);
+ {not_found, _} ->
+ {ok, Oldest} = couch_db:get_oldest_purge_seq(Db),
+ Oldest
+ end,
+ FoldFun = fun({PSeq, UUID, Id, Revs}, {Count, Infos, _}) ->
+ NewCount = Count + length(Revs),
+ NewInfos = [{UUID, Id, Revs} | Infos],
+ Status = if NewCount < BatchSize -> ok; true -> stop end,
+ {Status, {NewCount, NewInfos, PSeq}}
+ end,
+ InitAcc = {0, [], StartSeq},
+ {ok, {_, PurgeInfos, ThroughSeq}} =
+ couch_db:fold_purge_infos(Db, StartSeq, FoldFun, InitAcc),
+ {ok, PurgeSeq} = couch_db:get_purge_seq(Db),
+ Remaining = PurgeSeq - ThroughSeq,
+ rexi:reply({ok, PurgeDocId, PurgeInfos, ThroughSeq, Remainin})
+ Else ->
+ rexi:reply(Else)
+ end.
+
+
+save_purge_checkpoint_rpc(DbName, PurgeDocId, Body) ->
+ erlang:put(io_priority, {internal_repl, DbName}),
+ case get_or_create_db(DbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ Doc = #doc{id = PurgeDocId, body = Body},
+ Resp = try couch_db:update_doc(Db, Doc, []) of
+ {ok, _} -> ok;
+ Else -> {error, Else}
+ catch T:R ->
+ {T, R}
+ end,
+ rexi:reply(Resp);
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
%% @doc Return the sequence where two files with the same UUID diverged.
compare_epochs(SourceEpochs, TargetEpochs) ->
compare_rev_epochs(
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 05/07: Add metrics for clustered purge
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 633d83883c67a2f999a1b31b61d32b2fad57d649
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Tue Jun 27 14:59:03 2017 -0400
Add metrics for clustered purge
COUCHDB-3326
---
src/chttpd/src/chttpd_db.erl | 1 +
src/couch/priv/stats_descriptions.cfg | 12 ++++++++++++
src/couch/src/couch_db.erl | 1 +
src/couch/src/couch_httpd_db.erl | 1 +
src/couch_index/src/couch_index_updater.erl | 17 ++++++++++++++---
src/couch_mrview/src/couch_mrview_index.erl | 11 ++++++++---
src/couch_mrview/src/couch_mrview_updater.erl | 14 +++++++++-----
7 files changed, 46 insertions(+), 11 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index f469b98..fbfae66 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -496,6 +496,7 @@ db_req(#httpd{path_parts=[_, <<"_bulk_get">>]}=Req, _Db) ->
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
+ couch_stats:increment_counter([couchdb, httpd, purge_requests]),
chttpd:validate_ctype(Req, "application/json"),
W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
Options = [{user_ctx, Req#httpd.user_ctx}, {w, W}],
diff --git a/src/couch/priv/stats_descriptions.cfg b/src/couch/priv/stats_descriptions.cfg
index f091978..bceb0ce 100644
--- a/src/couch/priv/stats_descriptions.cfg
+++ b/src/couch/priv/stats_descriptions.cfg
@@ -34,6 +34,10 @@
{type, counter},
{desc, <<"number of times a document was read from a database">>}
]}.
+{[couchdb, database_purges], [
+ {type, counter},
+ {desc, <<"number of times a database was purged">>}
+]}.
{[couchdb, db_open_time], [
{type, histogram},
{desc, <<"milliseconds required to open a database">>}
@@ -46,6 +50,10 @@
{type, counter},
{desc, <<"number of document write operations">>}
]}.
+{[couchdb, document_purges], [
+ {type, counter},
+ {desc, <<"number of document purge operations">>}
+]}.
{[couchdb, local_document_writes], [
{type, counter},
{desc, <<"number of _local document write operations">>}
@@ -74,6 +82,10 @@
{type, counter},
{desc, <<"number of clients for continuous _changes">>}
]}.
+{[couchdb, httpd, purge_requests], [
+ {type, counter},
+ {desc, <<"number of purge requests">>}
+]}.
{[couchdb, httpd_request_methods, 'COPY'], [
{type, counter},
{desc, <<"number of HTTP COPY requests">>}
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 3ef6ab0..2a2095c 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -382,6 +382,7 @@ get_full_doc_infos(Db, Ids) ->
Rev :: {non_neg_integer(), binary()},
Reply :: {ok, []} | {ok, [Rev]}.
purge_docs(#db{main_pid = Pid}, UUIdsIdsRevs) ->
+ increment_stat(Db, [couchdb, database_purges]),
gen_server:call(Pid, {purge_docs, UUIdsIdsRevs});
-spec get_purge_infos(#db{}, [UUId]) -> [PurgeInfo] when
diff --git a/src/couch/src/couch_httpd_db.erl b/src/couch/src/couch_httpd_db.erl
index ed8a47b..b407c02 100644
--- a/src/couch/src/couch_httpd_db.erl
+++ b/src/couch/src/couch_httpd_db.erl
@@ -376,6 +376,7 @@ db_req(#httpd{path_parts=[_,<<"_bulk_docs">>]}=Req, _Db) ->
send_method_not_allowed(Req, "POST");
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
+ couch_stats:increment_counter([couchdb, httpd, purge_requests]),
couch_httpd:validate_ctype(Req, "application/json"),
{IdRevs} = couch_httpd:json_body_obj(Req),
PurgeReqs = lists:map(fun({Id, JsonRevs} ->
diff --git a/src/couch_index/src/couch_index_updater.erl b/src/couch_index/src/couch_index_updater.erl
index b416d17..6dc587d 100644
--- a/src/couch_index/src/couch_index_updater.erl
+++ b/src/couch_index/src/couch_index_updater.erl
@@ -141,10 +141,11 @@ update(Idx, Mod, IdxState) ->
DbUpdateSeq = couch_db:get_update_seq(Db),
DbCommittedSeq = couch_db:get_committed_update_seq(Db),
+ NumUpdateChanges = couch_db:count_changes_since(Db, CurrSeq),
+ NumPurgeChanges = count_pending_purged_docs_since(Db, Mod, IdxState),
+ TotalChanges = NumUpdateChanges + NumPurgeChanges,
{ok, PurgedIdxState} = purge_index(Db, Mod, IdxState),
- NumChanges = couch_db:count_changes_since(Db, CurrSeq),
-
GetSeq = fun
(#full_doc_info{update_seq=Seq}) -> Seq;
(#doc_info{high_seq=Seq}) -> Seq
@@ -182,8 +183,13 @@ update(Idx, Mod, IdxState) ->
{ok, {NewSt, true}}
end
end,
+ {ok, InitIdxState} = Mod:start_update(
+ Idx,
+ PurgedIdxState,
+ TotalChanges,
+ NumPurgeChanges
+ ),
- {ok, InitIdxState} = Mod:start_update(Idx, PurgedIdxState, NumChanges),
Acc0 = {InitIdxState, true},
{ok, Acc} = couch_db:fold_changes(Db, CurrSeq, Proc, Acc0, []),
{ProcIdxSt, SendLast} = Acc,
@@ -223,3 +229,8 @@ purge_index(Db, Mod, IdxState) ->
Mod:update_local_purge_doc(Db, NewStateAcc),
{ok, NewStateAcc}
end.
+
+count_pending_purged_docs_since(Db, Mod, IdxState) ->
+ {ok, DbPurgeSeq} = couch_db:get_purge_seq(Db),
+ IdxPurgeSeq = Mod:get(purge_seq, IdxState),
+ DbPurgeSeq - IdxPurgeSeq.
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 6c63eaf..3f04f8f 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -15,7 +15,7 @@
-export([get/2]).
-export([init/2, open/2, close/1, reset/1, delete/1, shutdown/1]).
--export([start_update/3, purge/4, process_doc/3, finish_update/1, commit/1]).
+-export([start_update/4, purge/4, process_doc/3, finish_update/1, commit/1]).
-export([compact/3, swap_compacted/2, remove_compacted/1]).
-export([index_file_exists/1]).
-export([update_local_purge_doc/2, verify_index_exists/1]).
@@ -172,8 +172,13 @@ reset(State) ->
end).
-start_update(PartialDest, State, NumChanges) ->
- couch_mrview_updater:start_update(PartialDest, State, NumChanges).
+start_update(PartialDest, State, NumChanges, NumChangesDone) ->
+ couch_mrview_updater:start_update(
+ PartialDest,
+ State,
+ NumChanges,
+ NumChangesDone
+ ).
purge(Db, PurgeSeq, PurgedIdRevs, State) ->
diff --git a/src/couch_mrview/src/couch_mrview_updater.erl b/src/couch_mrview/src/couch_mrview_updater.erl
index 214f487..3383b49 100644
--- a/src/couch_mrview/src/couch_mrview_updater.erl
+++ b/src/couch_mrview/src/couch_mrview_updater.erl
@@ -12,15 +12,14 @@
-module(couch_mrview_updater).
--export([start_update/3, purge/4, process_doc/3, finish_update/1]).
+-export([start_update/4, purge/4, process_doc/3, finish_update/1]).
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
-define(REM_VAL, removed).
-
-start_update(Partial, State, NumChanges) ->
+start_update(Partial, State, NumChanges, NumChangesDone) ->
MaxSize = config:get_integer("view_updater", "queue_memory_cap", 100000),
MaxItems = config:get_integer("view_updater", "queue_item_cap", 500),
QueueOpts = [{max_size, MaxSize}, {max_items, MaxItems}],
@@ -36,14 +35,19 @@ start_update(Partial, State, NumChanges) ->
},
Self = self(),
+
MapFun = fun() ->
+ Progress = case NumChanges of
+ 0 -> 0;
+ _ -> (NumChangesDone * 100) div NumChanges
+ end,
couch_task_status:add_task([
{indexer_pid, ?l2b(pid_to_list(Partial))},
{type, indexer},
{database, State#mrst.db_name},
{design_document, State#mrst.idx_name},
- {progress, 0},
- {changes_done, 0},
+ {progress, Progress},
+ {changes_done, NumChangesDone},
{total_changes, NumChanges}
]),
couch_task_status:set_update_frequency(500),
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 02/07: Implement clustered purge API
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 509548f0c6f3525a560d356af40d03057318c2c6
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon May 1 15:43:43 2017 -0400
Implement clustered purge API
* Implement clustered endpoint to purge docs on all nodes
- implement fabric:purge_docs(DbName, IdsRevs, Options)
- generate a unique ID: "UUID" for every purge request before
sending them to workers on specific node.
- fabric_rpc:purge_docs has an option of "replicated_changes" or
or "interactive edit" that are passed to couch_db:purge_docs/3.
This done so that "replicated_changes" updates will not reapply
purges with UUIDs that already exist on a node.
* Fix read-repair
- fix read-repair so not to recreate docs that have been purged before
on a certain node from nodes that are out of sync.
- In the open docs calls we track which nodes sent which revisions. If we
detect the need for read repair, we send this list of (node, rev) pairs
as an option to the fabric:update_docs call. When fabric:update_docs
receives a list of (node, rev) pairs it will use this information to know
whether it should apply the update or ignore it. It checks the
_local/purge-mem3.. docs to see if the purge_seq is up to date.
If not it should ignore the update request. As an optimization,
if the purge_seq is less than a configurable limit
out of sync, the updater sequentially scans the purge_seq tree
looking for purge requests for the given revision and if not found can
continue with the write.
* Implement clustered endpoint to set purged_docs_limit of Db on all nodes
- implement fabric:set_purged_docs_limit(DbName, Limit, Options)
* Implement clustered endpoint to get purged_docs_limit
- implement fabric:get_purged_docs_limit(DbName)
COUCHDB-3326
---
src/fabric/src/fabric.erl | 35 ++-
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 | 414 ++++++++++++++++++++++++++++++++++++
src/fabric/src/fabric_rpc.erl | 102 ++++++++-
6 files changed, 612 insertions(+), 36 deletions(-)
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 4a07271..7221654 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_purged_docs_limit/1, set_purged_docs_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_purged_docs_limit(dbname(), pos_integer(), [option()]) -> ok.
+set_purged_docs_limit(DbName, Limit, Options)
+ when is_integer(Limit), Limit > 0 ->
+ fabric_db_meta:set_purged_docs_limit(dbname(DbName), Limit, opts(Options)).
+
+%% @doc retrieves the upper bound for the number of stored purge requests
+-spec get_purged_docs_limit(dbname()) -> pos_integer() | no_return().
+get_purged_docs_limit(DbName) ->
+ {ok, Db} = fabric_util:get_db(dbname(DbName), [?ADMIN_CTX]),
+ try couch_db:get_purged_docs_limit(Db) after catch couch_db:close(Db) end.
+
get_security(DbName) ->
get_security(DbName, [?ADMIN_CTX]).
@@ -267,8 +280,24 @@ 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()]) ->
+ {Health, {PurgeSeq, [{Health, [revision()]}] }} when
+ Health :: ok | accepted,
+ PurgeSeq :: any().
+purge_docs(DbName, IdsRevs, Options) when is_list(IdsRevs) ->
+ IdsRevs2 = [idrevs(IdRs) || IdRs <- IdsRevs],
+ case fabric_doc_purge:go(dbname(DbName), IdsRevs2, opts(Options)) of
+ {ok, Results} ->
+ {ok, Results};
+ {accepted, Results} ->
+ {accepted, Results};
+ Error ->
+ throw(Error)
+ end.
+
%% @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..4fd9365 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_purged_docs_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_purged_docs_limit(DbName, Limit, Options) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, set_purged_docs_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..24e8c66
--- /dev/null
+++ b/src/fabric/src/fabric_doc_purge.erl
@@ -0,0 +1,414 @@
+% 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").
+-include_lib("couch/include/couch_db.hrl").
+
+go(_, [], _) ->
+ {ok, []};
+go(DbName, AllIdsRevs, Opts) ->
+ % tag each purge request with UUId
+ {AllUUIDs, AllUUIDsIdsRevs, DocCount} = tag_docs(AllIdsRevs),
+
+ Options = lists:delete(all_or_nothing, Opts),
+ % Counters -> [{Worker, UUIDs}]
+ {Counters, Workers} = dict:fold(fun(Shard, UUIDsIdsRevs, {Cs,Ws}) ->
+ UUIDs = [UUID || {UUID, _Id, _Revs} <-UUIDsIdsRevs],
+ #shard{name=Name, node=Node} = Shard,
+ Ref = rexi:cast(Node,
+ {fabric_rpc, purge_docs, [Name, UUIDsIdsRevs, Options]}),
+ Worker = Shard#shard{ref=Ref},
+ {[{Worker, UUIDs}|Cs], [Worker|Ws]}
+ end, {[], []}, group_idrevs_by_shard(DbName, AllUUIDsIdsRevs)),
+
+ RexiMon = fabric_util:create_monitors(Workers),
+ W = couch_util:get_value(w, Options, integer_to_list(mem3:quorum(DbName))),
+ Acc = {length(Workers), DocCount, list_to_integer(W), Counters, dict:new()},
+ Timeout = fabric_util:request_timeout(),
+ try rexi_utils:recv(Workers, #shard.ref,
+ fun handle_message/3, Acc, infinity, Timeout) of
+ {ok, {Health, Results}} when Health =:= ok; Health =:= accepted ->
+ % Results-> [{UUID, {ok, Revs}}]
+ {Health, [R || R <-
+ couch_util:reorder_results(AllUUIDs, Results)]};
+ {timeout, Acc1} ->
+ {_, _, W1, Counters1, DocReplDict0} = Acc1,
+ {DefunctWorkers, _} = lists:unzip(Counters1),
+ fabric_util:log_timeout(DefunctWorkers, "purge_docs"),
+ DocReplDict = lists:foldl(fun({_W, Docs}, Dict) ->
+ Replies = [{error, timeout} || _D <- Docs],
+ append_purge_replies(Docs, Replies, Dict)
+ end, DocReplDict0, Counters1),
+ {Health, _, Resp} = dict:fold(
+ fun force_reply/3, {ok, W1, []}, DocReplDict),
+ case Health of
+ error -> timeout;
+ _ -> {Health, [R || R <-
+ couch_util:reorder_results(AllUUIDs, Resp)]}
+
+ end;
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, Acc0) ->
+ {_, DocCount, W, Counters, DocsDict0} = Acc0,
+ {FailCounters, NewCounters} = lists:partition(fun({#shard{node=N}, _}) ->
+ N == NodeRef
+ end, Counters),
+ % fill DocsDict with error messages for relevant Docs
+ DocsDict = lists:foldl(fun({_W, Docs}, CDocsDict) ->
+ Replies = [{error, internal_server_error} || _D <- Docs],
+ append_purge_replies(Docs, Replies, CDocsDict)
+ end, DocsDict0, FailCounters),
+ skip_message({length(NewCounters), DocCount, W, NewCounters, DocsDict});
+handle_message({rexi_EXIT, _}, Worker, Acc0) ->
+ {WC, DocCount , W, Counters, DocsDict0} = Acc0,
+ % fill DocsDict with error messages for relevant Docs
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ Replies = [{error, internal_server_error} || _D <- Docs],
+ DocsDict = append_purge_replies(Docs, Replies, DocsDict0),
+ skip_message({WC-1, DocCount, W, NewCounters, DocsDict});
+handle_message({ok, Replies0}, Worker, Acc0) ->
+ {WCount, DocCount, W, Counters, DocsDict0} = Acc0,
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ DocsDict = append_purge_replies(Docs, Replies0, DocsDict0),
+ case {WCount, dict:size(DocsDict)} of
+ {1, _} ->
+ % last message has arrived, we need to conclude things
+ {Health, W, Replies} = dict:fold(fun force_reply/3, {ok, W, []},
+ DocsDict),
+ {stop, {Health, Replies}};
+ {_, DocCount} ->
+ % we've got at least one reply for each document, let's take a look
+ case dict:fold(fun maybe_reply/3, {stop,W,[]}, DocsDict) of
+ continue ->
+ {ok, {WCount - 1, DocCount, W, NewCounters, DocsDict}};
+ {stop, W, Replies} ->
+ {stop, {ok, Replies}}
+ end;
+ _ ->
+ {ok, {WCount - 1, DocCount, W, NewCounters, DocsDict}}
+ end;
+handle_message({error, purged_docs_limit_exceeded}=Error, Worker, Acc0) ->
+ {WC, DocCount , W, Counters, DocsDict0} = Acc0,
+ % fill DocsDict with error messages for relevant Docs
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ Replies = [Error || _D <- Docs],
+ DocsDict = append_purge_replies(Docs, Replies, DocsDict0),
+ skip_message({WC-1, DocCount, W, NewCounters, DocsDict});
+handle_message({bad_request, Msg}, _, _) ->
+ throw({bad_request, Msg}).
+
+
+tag_docs(AllIdsRevs) ->
+ {UUIDs, UUIDsIdsRevs, DocCount} = lists:foldl(fun(
+ {Id, Revs}, {UAcc, UIRAcc, C}) ->
+ UUID = couch_uuids:new(),
+ {[UUID|UAcc], [{UUID, Id, Revs}|UIRAcc], C+1}
+ end, {[], [], 0}, AllIdsRevs),
+ {lists:reverse(UUIDs), lists:reverse(UUIDsIdsRevs), DocCount}.
+
+
+force_reply(Doc, Replies, {Health, W, Acc}) ->
+ case update_quorum_met(W, Replies) of
+ {true, FinalReply} ->
+ {Health, W, [{Doc, FinalReply} | Acc]};
+ false ->
+ case [Reply || {ok, Reply} <- Replies] of
+ [] ->
+ UReplies = lists:usort(Replies),
+ case UReplies of
+ [{error, internal_server_error}] ->
+ {error, W, [{Doc, {error, internal_server_error}} | Acc]};
+ [{error, timeout}] ->
+ {error, W, [{Doc, {error, timeout}} | Acc]};
+ [FirstReply|[]] ->
+ % check if all errors are identical, if so inherit health
+ {Health, W, [{Doc, FirstReply} | Acc]};
+ _ ->
+ {error, W, [{Doc, UReplies} | Acc]}
+ end;
+ AcceptedReplies0 ->
+ NewHealth = case Health of ok -> accepted; _ -> Health end,
+ AcceptedReplies = lists:usort(lists:flatten(AcceptedReplies0)),
+ {NewHealth, W, [{Doc, {accepted, AcceptedReplies}} | Acc]}
+ end
+ end.
+
+
+maybe_reply(_, _, continue) ->
+ % we didn't meet quorum for all docs, so we're fast-forwarding the fold
+ continue;
+maybe_reply(Doc, Replies, {stop, W, Acc}) ->
+ case update_quorum_met(W, Replies) of
+ {true, Reply} ->
+ {stop, W, [{Doc, Reply} | Acc]};
+ false ->
+ continue
+ end.
+
+update_quorum_met(W, Replies) ->
+ OkReplies = lists:foldl(fun(Reply, PrevsAcc) ->
+ case Reply of
+ {ok, PurgedRevs} -> [PurgedRevs | PrevsAcc];
+ _ -> PrevsAcc
+ end
+ end, [], Replies),
+ if length(OkReplies) < W -> false; true ->
+ % make a union of PurgedRevs
+ FinalReply = {ok, lists:usort(lists:flatten(OkReplies))},
+ {true, FinalReply}
+ end.
+
+
+group_idrevs_by_shard(DbName, UUIDsIdsRevs) ->
+ lists:foldl(fun({_UUID, Id, _Revs} = UUIDIdRevs, D0) ->
+ lists:foldl(fun(Shard, D1) ->
+ dict:append(Shard, UUIDIdRevs, D1)
+ end, D0, mem3:shards(DbName, Id))
+ end, dict:new(), UUIDsIdsRevs).
+
+
+append_purge_replies([], [], DocReplyDict) ->
+ DocReplyDict;
+append_purge_replies([Doc|Rest1], [Reply|Rest2], Dict0) ->
+ append_purge_replies(Rest1, Rest2, dict:append(Doc, Reply, Dict0)).
+
+
+skip_message({0, _, W, _, DocsDict}) ->
+ {Health, W, Reply} = dict:fold(fun force_reply/3, {ok, W, []}, DocsDict),
+ {stop, {Health, Reply}};
+skip_message(Acc0) ->
+ {ok, Acc0}.
+
+
+% 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..6e2c05f 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_purged_docs_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_purged_docs_limit(DbName, Limit, Options) ->
+ with_db(DbName, Options, {couch_db, set_purged_docs_limit, [Limit]}).
+
open_doc(DbName, DocId, Options) ->
with_db(DbName, Options, {couch_db, open_doc, [DocId, Options]}).
@@ -236,14 +240,31 @@ get_missing_revs(DbName, IdRevsList, Options) ->
end).
update_docs(DbName, Docs0, Options) ->
+ X = case proplists:get_value(replicated_changes, Options) of
+ true -> replicated_changes;
+ _ -> interactive_edit
+ end,
+ 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) ->
case proplists:get_value(replicated_changes, Options) of
- true ->
- X = replicated_changes;
- _ ->
- X = interactive_edit
+ true ->
+ X = replicated_changes;
+ _ ->
+ X = interactive_edit
end,
- Docs = make_att_readers(Docs0),
- with_db(DbName, Options, {couch_db, update_docs, [Docs, Options, X]}).
+ with_db(DbName, Options, {couch_db, purge_docs, [UUIdsIdsRevs, X]}).
%% @equiv group_info(DbName, DDocId, [])
group_info(DbName, DDocId) ->
@@ -298,6 +319,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.
[couchdb] 04/07: Implement clustered purge HTTP endpoint
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 90ac409e2f2c42b58f0a1f4708fe3e1252371bc9
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon May 1 16:44:25 2017 -0400
Implement clustered purge HTTP endpoint
* Implement clustered purge endpoint
* Add endpoint for setting purged_docs_limit
* Add endpoint for getting purged_docs_limit
COUCHDB-3326
---
src/chttpd/src/chttpd_db.erl | 42 ++++++++---
src/chttpd/test/chttpd_purge_tests.erl | 130 +++++++++++++++++++++++++++++++++
2 files changed, 161 insertions(+), 11 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 2c3ec63..f469b98 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -497,23 +497,22 @@ db_req(#httpd{path_parts=[_, <<"_bulk_get">>]}=Req, _Db) ->
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
chttpd:validate_ctype(Req, "application/json"),
+ W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
+ Options = [{user_ctx, Req#httpd.user_ctx}, {w, W}],
{IdsRevs} = chttpd:json_body_obj(Req),
IdsRevs2 = [{Id, couch_doc:parse_revs(Revs)} || {Id, Revs} <- IdsRevs],
- case fabric:purge_docs(Db, IdsRevs2) of
- {ok, PurgeSeq, PurgedIdsRevs} ->
- PurgedIdsRevs2 = [{Id, couch_doc:revs_to_strs(Revs)} || {Id, Revs}
- <- PurgedIdsRevs],
- send_json(Req, 200, {[
- {<<"purge_seq">>, PurgeSeq},
- {<<"purged">>, {PurgedIdsRevs2}}
- ]});
- Error ->
- throw(Error)
- end;
+ {Status, Results} = fabric:purge_docs(Db, IdsRevs2, Options),
+ Code = case Status of
+ ok -> 201;
+ accepted -> 202
+ end,
+ Purged = lists:zipwith(fun purge_result_to_json/2, IdsRevs2, Results),
+ send_json(Req, Code, {[{<<"purged">>, {Purged}}]});
db_req(#httpd{path_parts=[_,<<"_purge">>]}=Req, _Db) ->
send_method_not_allowed(Req, "POST");
+
db_req(#httpd{method='GET',path_parts=[_,OP]}=Req, Db) when ?IS_ALL_DOCS(OP) ->
case chttpd:qs_json_value(Req, "keys", nil) of
Keys when is_list(Keys) ->
@@ -608,6 +607,20 @@ db_req(#httpd{method='GET',path_parts=[_,<<"_revs_limit">>]}=Req, Db) ->
db_req(#httpd{path_parts=[_,<<"_revs_limit">>]}=Req, _Db) ->
send_method_not_allowed(Req, "PUT,GET");
+db_req(#httpd{method='PUT',path_parts=[_,<<"_purged_docs_limit">>]}=Req, Db) ->
+ Limit = chttpd:json_body(Req),
+ Options = [{user_ctx, Req#httpd.user_ctx}],
+ case chttpd:json_body(Req) of
+ Limit when is_integer(Limit), Limit > 0 ->
+ ok = fabric:set_purged_docs_limit(Db, Limit, Options),
+ send_json(Req, {[{<<"ok">>, true}]});
+ _->
+ throw({bad_request, "`purged_docs_limit` must be positive integer"})
+ end;
+
+db_req(#httpd{method='GET',path_parts=[_,<<"_purged_docs_limit">>]}=Req, Db) ->
+ send_json(Req, fabric:get_purged_docs_limit(Db));
+
% Special case to enable using an unencoded slash in the URL of design docs,
% as slashes in document IDs must otherwise be URL encoded.
db_req(#httpd{method='GET', mochi_req=MochiReq, path_parts=[_DbName, <<"_design/", _/binary>> | _]}=Req, _Db) ->
@@ -955,6 +968,13 @@ update_doc_result_to_json(DocId, Error) ->
{_Code, ErrorStr, Reason} = chttpd:error_info(Error),
{[{id, DocId}, {error, ErrorStr}, {reason, Reason}]}.
+purge_result_to_json({DocId, _Revs}, {ok, PRevs}) ->
+ {DocId, {[{purged, couch_doc:revs_to_strs(PRevs)}, {ok, true}]}};
+purge_result_to_json({DocId, _Revs}, {accepted, PRevs}) ->
+ {DocId, {[{purged, couch_doc:revs_to_strs(PRevs)}, {accepted, true}]}};
+purge_result_to_json({DocId, _Revs}, Error) ->
+ {_Code, ErrorStr, Reason} = chttpd:error_info(Error),
+ {DocId, {[{error, ErrorStr}, {reason, Reason}]}}.
send_updated_doc(Req, Db, DocId, Json) ->
send_updated_doc(Req, Db, DocId, Json, []).
diff --git a/src/chttpd/test/chttpd_purge_tests.erl b/src/chttpd/test/chttpd_purge_tests.erl
new file mode 100644
index 0000000..7900090
--- /dev/null
+++ b/src/chttpd/test/chttpd_purge_tests.erl
@@ -0,0 +1,130 @@
+% 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(chttpd_purge_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+-define(USER, "chttpd_db_test_admin").
+-define(PASS, "pass").
+-define(AUTH, {basic_auth, {?USER, ?PASS}}).
+-define(CONTENT_JSON, {"Content-Type", "application/json"}).
+
+
+setup() ->
+ ok = config:set("admins", ?USER, ?PASS, _Persist=false),
+ TmpDb = ?tempdb(),
+ Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Port = mochiweb_socket_server:get(chttpd, port),
+ Url = lists:concat(["http://", Addr, ":", Port, "/", ?b2l(TmpDb)]),
+ create_db(Url),
+ Url.
+
+
+teardown(Url) ->
+ delete_db(Url),
+ ok = config:delete("admins", ?USER, _Persist=false).
+
+
+create_db(Url) ->
+ {ok, Status, _, _} = test_request:put(Url, [?CONTENT_JSON, ?AUTH], "{}"),
+ ?assert(Status =:= 201 orelse Status =:= 202).
+
+
+create_doc(Url, Id) ->
+ test_request:put(Url ++ "/" ++ Id,
+ [?CONTENT_JSON, ?AUTH], "{\"mr\": \"rockoartischocko\"}").
+
+
+delete_db(Url) ->
+ {ok, 200, _, _} = test_request:delete(Url, [?AUTH]).
+
+
+purge_test_() ->
+ {
+ "chttpd db tests",
+ {
+ setup,
+ fun chttpd_test_util:start_couch/0,
+ fun chttpd_test_util:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ fun test_empty_purge_request/1,
+ fun test_ok_purge_request/1,
+ fun should_error_set_purged_docs_limit_to0/1
+ ]
+ }
+ }
+ }.
+
+
+test_empty_purge_request(Url) ->
+ ?_test(begin
+ IdsRevs = "{}",
+ {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+ [?CONTENT_JSON, ?AUTH], IdsRevs),
+ ResultJson = ?JSON_DECODE(ResultBody),
+ ?assert(Status =:= 201 orelse Status =:= 202),
+ ?assertEqual({[{<<"purged">>,{[]}}]}, ResultJson)
+ end).
+
+
+test_ok_purge_request(Url) ->
+ ?_test(begin
+ {ok, _, _, Body} = create_doc(Url, "doc1"),
+ {Json} = ?JSON_DECODE(Body),
+ Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+ {ok, _, _, Body2} = create_doc(Url, "doc2"),
+ {Json2} = ?JSON_DECODE(Body2),
+ Rev2 = couch_util:get_value(<<"rev">>, Json2, undefined),
+ {ok, _, _, Body3} = create_doc(Url, "doc3"),
+ {Json3} = ?JSON_DECODE(Body3),
+ Rev3 = couch_util:get_value(<<"rev">>, Json3, undefined),
+ IdsRevs = "{\"doc1\": [\"" ++ ?b2l(Rev1) ++ "\"], \"doc2\": [\"" ++
+ ?b2l(Rev2) ++ "\"], \"doc3\": [\"" ++ ?b2l(Rev3) ++ "\"] }",
+
+ {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+ [?CONTENT_JSON, ?AUTH], IdsRevs),
+ ResultJson = ?JSON_DECODE(ResultBody),
+ ?assert(Status =:= 201 orelse Status =:= 202),
+ ?assertEqual(
+ {[{<<"purged">>, {[
+ {<<"doc1">>, {[
+ {<<"purged">>,[Rev1]},
+ {<<"ok">>,true}
+ ]}},
+ {<<"doc2">>, {[
+ {<<"purged">>,[Rev2]},
+ {<<"ok">>,true}
+ ]}},
+ {<<"doc3">>, {[
+ {<<"purged">>,[Rev3]},
+ {<<"ok">>,true}
+ ]}}
+ ]}}]},
+ ResultJson
+ )
+ end).
+
+
+should_error_set_purged_docs_limit_to0(Url) ->
+ ?_test(begin
+ {ok, Status, _, _} = test_request:put(Url ++ "/_purged_docs_limit/",
+ [?CONTENT_JSON, ?AUTH], "0"),
+ ?assert(Status =:= 400)
+ end).
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 07/07: Temporarily disable
should_compare_compression_methods/1
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 79e3d2293fba35e38d32694e3188671018e8f67a
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Tue May 16 09:50:19 2017 +0800
Temporarily disable should_compare_compression_methods/1
COUCHDB-3326
---
src/couch/test/couchdb_file_compression_tests.erl | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/src/couch/test/couchdb_file_compression_tests.erl b/src/couch/test/couchdb_file_compression_tests.erl
index 8f0fe5b..e5ef74c 100644
--- a/src/couch/test/couchdb_file_compression_tests.erl
+++ b/src/couch/test/couchdb_file_compression_tests.erl
@@ -57,8 +57,8 @@ couch_file_compression_test_() ->
fun should_use_none/1,
fun should_use_deflate_1/1,
fun should_use_deflate_9/1,
- fun should_use_snappy/1,
- fun should_compare_compression_methods/1
+ fun should_use_snappy/1
+ %fun should_compare_compression_methods/1
]
}
}
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 06/07: Fix tests to work with clustered purge
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d3a2b599e9d7c0a86c32ab8c315e60f305a734d4
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Thu Mar 1 17:46:17 2018 +0800
Fix tests to work with clustered purge
COUCHDB-3326
---
test/javascript/tests/erlang_views.js | 5 +++--
test/javascript/tests/purge.js | 27 ++++++++-------------------
2 files changed, 11 insertions(+), 21 deletions(-)
diff --git a/test/javascript/tests/erlang_views.js b/test/javascript/tests/erlang_views.js
index ec78e65..9b15e10 100644
--- a/test/javascript/tests/erlang_views.js
+++ b/test/javascript/tests/erlang_views.js
@@ -56,7 +56,7 @@ couchTests.erlang_views = function(debug) {
' {Info} = couch_util:get_value(<<"info">>, Req, {[]}), ' +
' Purged = couch_util:get_value(<<"purge_seq">>, Info, -1), ' +
' Verb = couch_util:get_value(<<"method">>, Req, <<"not_get">>), ' +
- ' R = list_to_binary(io_lib:format("~b - ~s", [Purged, Verb])), ' +
+ ' R = list_to_binary(io_lib:format("~s - ~s", [Purged, Verb])), ' +
' {[{<<"code">>, 200}, {<<"headers">>, {[]}}, {<<"body">>, R}]} ' +
'end.'
},
@@ -85,7 +85,8 @@ couchTests.erlang_views = function(debug) {
var url = "/" + db_name + "/_design/erlview/_show/simple/1";
var xhr = CouchDB.request("GET", url);
T(xhr.status == 200, "standard get should be 200");
- T(xhr.responseText == "0 - GET");
+ T(/0-/.test(xhr.responseText));
+ T(/- GET/.test(xhr.responseText));
var url = "/" + db_name + "/_design/erlview/_list/simple_list/simple_view";
var xhr = CouchDB.request("GET", url);
diff --git a/test/javascript/tests/purge.js b/test/javascript/tests/purge.js
index 38eca8d..6bfba02 100644
--- a/test/javascript/tests/purge.js
+++ b/test/javascript/tests/purge.js
@@ -11,7 +11,6 @@
// the License.
couchTests.purge = function(debug) {
- return console.log('TODO: this feature is not yet implemented');
var db_name = get_random_db_name();
var db = new CouchDB(db_name, {"X-Couch-Full-Commit":"false"});
db.createDb();
@@ -53,21 +52,13 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
body: JSON.stringify({"1":[doc1._rev], "2":[doc2._rev]})
});
- console.log(xhr.status);
- console.log(xhr.responseText);
- T(xhr.status == 200);
+ T(xhr.status == 201);
var result = JSON.parse(xhr.responseText);
var newInfo = db.info();
-
- // purging increments the update sequence
- T(info.update_seq+1 == newInfo.update_seq);
- // and it increments the purge_seq
- T(info.purge_seq+1 == newInfo.purge_seq);
- T(result.purge_seq == newInfo.purge_seq);
- T(result.purged["1"][0] == doc1._rev);
- T(result.purged["2"][0] == doc2._rev);
+ T(result.purged["1"].purged[0] == doc1._rev);
+ T(result.purged["2"].purged[0] == doc2._rev);
T(db.open("1") == null);
T(db.open("2") == null);
@@ -85,7 +76,6 @@ couchTests.purge = function(debug) {
// compaction isn't instantaneous, loop until done
while (db.info().compact_running) {};
var compactInfo = db.info();
- T(compactInfo.purge_seq == newInfo.purge_seq);
// purge documents twice in a row without loading views
// (causes full view rebuilds)
@@ -97,15 +87,14 @@ couchTests.purge = function(debug) {
body: JSON.stringify({"3":[doc3._rev]})
});
- T(xhr.status == 200);
+ T(xhr.status == 201);
xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
body: JSON.stringify({"4":[doc4._rev]})
});
- T(xhr.status == 200);
+ T(xhr.status == 201);
result = JSON.parse(xhr.responseText);
- T(result.purge_seq == db.info().purge_seq);
var rows = db.view("test/all_docs_twice").rows;
for (var i = 4; i < numDocs; i++) {
@@ -129,7 +118,7 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docA._rev]})
});
- TEquals(200, xhr.status, "single rev purge after replication succeeds");
+ TEquals(201, xhr.status, "single rev purge after replication succeeds");
var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docA._rev);
TEquals(404, xhr.status, "single rev purge removes revision");
@@ -137,14 +126,14 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docB._rev]})
});
- TEquals(200, xhr.status, "single rev purge after replication succeeds");
+ TEquals(201, xhr.status, "single rev purge after replication succeeds");
var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docB._rev);
TEquals(404, xhr.status, "single rev purge removes revision");
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docA._rev, docB._rev]})
});
- TEquals(200, xhr.status, "all rev purge after replication succeeds");
+ TEquals(201, xhr.status, "all rev purge after replication succeeds");
// cleanup
db.deleteDb();
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 03/07: Update fabric_doc_open_revs to handle purges
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 57719ecc3635230977db3779f7cd44eefdd1d8d6
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Tue Oct 3 15:56:55 2017 -0400
Update fabric_doc_open_revs to handle purges
We need to account for the possibility that a document is opened while a
purge is still propogating between shards. This means we need to inform
the read-repair algorithms that a difference in revisions may be due to
a purge request in progress. If we don't do this its possible that a
read-repair may race the purge request and effectively undo the purge.
---
src/fabric/src/fabric_doc_open_revs.erl | 148 ++++++++++++++++++++++----------
src/fabric/src/fabric_rpc.erl | 67 ++++++++-------
2 files changed, 137 insertions(+), 78 deletions(-)
diff --git a/src/fabric/src/fabric_doc_open_revs.erl b/src/fabric/src/fabric_doc_open_revs.erl
index 096722f..dbe02bf 100644
--- a/src/fabric/src/fabric_doc_open_revs.erl
+++ b/src/fabric/src/fabric_doc_open_revs.erl
@@ -29,7 +29,8 @@
revs,
latest,
replies = [],
- repair = false
+ repair = false,
+ replies_by_node=[] %[{Node, Reply}] used for read_repair
}).
go(DbName, Id, Revs, Options) ->
@@ -82,6 +83,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
worker_count = WorkerCount,
workers = Workers,
replies = PrevReplies,
+ replies_by_node = PrevNReplies,
r = R,
revs = Revs,
latest = Latest,
@@ -92,13 +94,14 @@ handle_message({ok, RawReplies}, Worker, State) ->
IsTree = Revs == all orelse Latest,
% Do not count error replies when checking quorum
-
RealReplyCount = ReplyCount + 1 - ReplyErrorCount,
QuorumReplies = RealReplyCount >= R,
{NewReplies, QuorumMet, Repair} = case IsTree of
true ->
- {NewReplies0, AllInternal, Repair0} =
+ {NewReplies0, AllInternal, Repair00} =
tree_replies(PrevReplies, tree_sort(RawReplies)),
+ % don't set Repair=true on the first reply
+ Repair0 = (ReplyCount > 0) and Repair00,
NumLeafs = couch_key_tree:count_leafs(PrevReplies),
SameNumRevs = length(RawReplies) == NumLeafs,
QMet = AllInternal andalso SameNumRevs andalso QuorumReplies,
@@ -107,6 +110,10 @@ handle_message({ok, RawReplies}, Worker, State) ->
{NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
{NewReplies0, MinCount >= R, false}
end,
+ NewNReplies = case Worker of
+ nil -> PrevNReplies;
+ _ -> [{Worker#shard.node, RawReplies}|PrevNReplies]
+ end,
Complete = (ReplyCount =:= (WorkerCount - 1)),
@@ -117,6 +124,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
DbName,
IsTree,
NewReplies,
+ NewNReplies,
ReplyCount + 1,
InRepair orelse Repair
),
@@ -124,6 +132,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
false ->
{ok, State#state{
replies = NewReplies,
+ replies_by_node = NewNReplies,
reply_count = ReplyCount + 1,
workers = lists:delete(Worker, Workers),
repair = InRepair orelse Repair
@@ -180,7 +189,7 @@ dict_replies(Dict, [Reply | Rest]) ->
dict_replies(NewDict, Rest).
-maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+maybe_read_repair(Db, IsTree, Replies, NodeReplies0, ReplyCount, DoRepair) ->
Docs = case IsTree of
true -> tree_repair_docs(Replies, DoRepair);
false -> dict_repair_docs(Replies, ReplyCount)
@@ -189,7 +198,11 @@ maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
[] ->
ok;
_ ->
- erlang:spawn(fun() -> read_repair(Db, Docs) end)
+ NodeReplies = lists:foldl(fun({Node, NReplies}, Acc) ->
+ NewAcc = [{Node, Doc} || {ok, Doc} <- NReplies],
+ NewAcc ++ Acc
+ end, [], NodeReplies0),
+ erlang:spawn(fun() -> read_repair(Db, Docs, NodeReplies) end)
end.
@@ -208,8 +221,9 @@ dict_repair_docs(Replies, ReplyCount) ->
end.
-read_repair(Db, Docs) ->
- Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
+read_repair(Db, Docs, NodeReplies) ->
+ Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NodeReplies}],
+ Res = fabric:update_docs(Db, Docs, Opts),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
@@ -268,20 +282,24 @@ filter_reply(Replies) ->
setup() ->
config:start_link([]),
meck:new([fabric, couch_stats, couch_log]),
+ meck:new(fabric_util, [passthrough]),
meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:expect(couch_log, notice, fun(_, _) -> ok end).
+ meck:expect(couch_log, notice, fun(_, _) -> ok end),
+ meck:expect(fabric_util, cleanup, fun(_) -> ok end).
+
teardown(_) ->
- (catch meck:unload([fabric, couch_stats, couch_log])),
+ (catch meck:unload([fabric, couch_stats, couch_log, fabric_util])),
config:stop().
state0(Revs, Latest) ->
#state{
worker_count = 3,
- workers = [w1, w2, w3],
+ workers =
+ [#shard{node='node1'}, #shard{node='node2'}, #shard{node='node3'}],
r = 2,
revs = Revs,
latest = Latest
@@ -334,27 +352,35 @@ open_doc_revs_test_() ->
check_empty_response_not_quorum() ->
% Simple smoke test that we don't think we're
% done with a first empty response
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
?_assertMatch(
- {ok, #state{workers = [w2, w3]}},
- handle_message({ok, []}, w1, state0(all, false))
+ {ok, #state{workers = [W2, W3]}},
+ handle_message({ok, []}, W1, state0(all, false))
).
check_basic_response() ->
% Check that we've handle a response
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
?_assertMatch(
- {ok, #state{reply_count = 1, workers = [w2, w3]}},
- handle_message({ok, [foo1(), bar1()]}, w1, state0(all, false))
+ {ok, #state{reply_count = 1, workers = [W2, W3]}},
+ handle_message({ok, [foo1(), bar1()]}, W1, state0(all, false))
).
check_finish_quorum() ->
% Two messages with the same revisions means we're done
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
Expect = {stop, [bar1(), foo1()]},
- ?assertEqual(Expect, handle_message({ok, [foo1(), bar1()]}, w2, S1))
+ ?assertEqual(Expect, handle_message({ok, [foo1(), bar1()]}, W2, S1))
end).
@@ -363,11 +389,13 @@ check_finish_quorum_newer() ->
% foo1 should count for foo2 which means we're finished.
% We also validate that read_repair was triggered.
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
Expect = {stop, [bar1(), foo2()]},
ok = meck:reset(fabric),
- ?assertEqual(Expect, handle_message({ok, [foo2(), bar1()]}, w2, S1)),
+ ?assertEqual(Expect, handle_message({ok, [foo2(), bar1()]}, W2, S1)),
ok = meck:wait(fabric, update_docs, '_', 5000),
?assertMatch(
[{_, {fabric, update_docs, [_, _, _]}, _}],
@@ -380,11 +408,14 @@ check_no_quorum_on_second() ->
% Quorum not yet met for the foo revision so we
% would wait for w3
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
?assertMatch(
- {ok, #state{workers = [w3]}},
- handle_message({ok, [bar1()]}, w2, S1)
+ {ok, #state{workers = [W3]}},
+ handle_message({ok, [bar1()]}, W2, S1)
)
end).
@@ -394,11 +425,14 @@ check_done_on_third() ->
% what. Every revision seen in this pattern should be
% included.
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
- {ok, S2} = handle_message({ok, [bar1()]}, w2, S1),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
+ {ok, S2} = handle_message({ok, [bar1()]}, W2, S1),
Expect = {stop, [bar1(), foo1()]},
- ?assertEqual(Expect, handle_message({ok, [bar1()]}, w3, S2))
+ ?assertEqual(Expect, handle_message({ok, [bar1()]}, W3, S2))
end).
@@ -407,108 +441,128 @@ check_done_on_third() ->
check_specific_revs_first_msg() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), false),
?assertMatch(
- {ok, #state{reply_count = 1, workers = [w2, w3]}},
- handle_message({ok, [foo1(), bar1(), bazNF()]}, w1, S0)
+ {ok, #state{reply_count = 1, workers = [W2, W3]}},
+ handle_message({ok, [foo1(), bar1(), bazNF()]}, W1, S0)
)
end).
check_revs_done_on_agreement() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), false),
Msg = {ok, [foo1(), bar1(), bazNF()]},
- {ok, S1} = handle_message(Msg, w1, S0),
+ {ok, S1} = handle_message(Msg, W1, S0),
Expect = {stop, [bar1(), foo1(), bazNF()]},
- ?assertEqual(Expect, handle_message(Msg, w2, S1))
+ ?assertEqual(Expect, handle_message(Msg, W2, S1))
end).
check_latest_true() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo2(), bar1(), bazNF()]},
Msg2 = {ok, [foo2(), bar1(), bazNF()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
+ {ok, S1} = handle_message(Msg1, W1, S0),
Expect = {stop, [bar1(), foo2(), bazNF()]},
- ?assertEqual(Expect, handle_message(Msg2, w2, S1))
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1))
end).
check_ancestor_counted_in_quorum() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), bazNF()]},
Msg2 = {ok, [foo2(), bar1(), bazNF()]},
Expect = {stop, [bar1(), foo2(), bazNF()]},
% Older first
- {ok, S1} = handle_message(Msg1, w1, S0),
- ?assertEqual(Expect, handle_message(Msg2, w2, S1)),
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1)),
% Newer first
- {ok, S2} = handle_message(Msg2, w2, S0),
- ?assertEqual(Expect, handle_message(Msg1, w1, S2))
+ {ok, S2} = handle_message(Msg2, W2, S0),
+ ?assertEqual(Expect, handle_message(Msg1, W1, S2))
end).
check_not_found_counts_for_descendant() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), bazNF()]},
Msg2 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
% not_found first
- {ok, S1} = handle_message(Msg1, w1, S0),
- ?assertEqual(Expect, handle_message(Msg2, w2, S1)),
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1)),
% not_found second
- {ok, S2} = handle_message(Msg2, w2, S0),
- ?assertEqual(Expect, handle_message(Msg1, w1, S2))
+ {ok, S2} = handle_message(Msg2, W2, S0),
+ ?assertEqual(Expect, handle_message(Msg1, W1, S2))
end).
check_worker_error_skipped() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), baz1()]},
Msg2 = {rexi_EXIT, reason},
Msg3 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
check_quorum_only_counts_valid_responses() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {rexi_EXIT, reason},
Msg2 = {rexi_EXIT, reason},
Msg3 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
check_empty_list_when_no_workers_reply() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {rexi_EXIT, reason},
Msg2 = {rexi_EXIT, reason},
Msg3 = {rexi_DOWN, nodedown, {nil, node()}, nil},
Expect = {stop, all_workers_died},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 6e2c05f..2c4d5f4 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -344,48 +344,53 @@ update_docs_read_repair(DbName, DocsByNode, Options) ->
% 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,
+% This is done for read-repair from fabric_doc_open or fabric_doc_open_revs,
% 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,
+ % go through _local/purge-mem3-.. docs
+ % and assemble NodePSeqs = [{Node1, NodePSeq1}, ...]
+ % NodePSeq1 - purge_seq of this node known to Node1
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}
+ NodePSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ {ok, [{Node, NodePSeq} | Acc]}
end,
- {ok, Docs} = couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
- Docs.
+ {ok, NodePSeqs} =
+ couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
+
+ % go through all doc_updates and
+ % filter out updates from nodes that are behind in purges synchronization
+ 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,
+ lists:foldl(fun({Node, Doc}, Docs) ->
+ NodePSeq = case lists:keyfind(Node, 1, NodePSeqs) of
+ {Node, NodePSeq0} -> NodePSeq0;
+ false -> 0
+ end,
+ if NodePSeq == DbPSeq ->
+ [Doc|Docs];
+ (NodePSeq+AllowedPSeqLag) < DbPSeq ->
+ % Node is very out of sync, ignore updates from it
+ Docs;
+ true -> %(NodePSeq+ClientAllowedPSeqLag) >= DbPSeq
+ % if Doc has been purged recently -> 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 -> Docs;
+ false -> [Doc|Docs]
+ end
+ end
+ end, [], DocsByNode).
get_or_create_db(DbName, Options) ->
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.