You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2024/12/11 16:35:08 UTC
(couchdb) branch auto-delete-2 updated (49f7066c6 -> b42197b7b)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch auto-delete-2
in repository https://gitbox.apache.org/repos/asf/couchdb.git
discard 49f7066c6 WIP CouchDB Auto-Tombstone Removal
new b42197b7b WIP CouchDB Auto-Tombstone Removal
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 (49f7066c6)
\
N -- N -- N refs/heads/auto-delete-2 (b42197b7b)
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 1 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:
src/couch/src/couch_bt_engine_compactor.erl | 30 +++++++++++++++++++++++++++++
src/couch_mrview/src/couch_mrview_index.erl | 1 -
src/couch_mrview/src/couch_mrview_util.erl | 7 +++----
3 files changed, 33 insertions(+), 5 deletions(-)
(couchdb) 01/01: WIP CouchDB Auto-Tombstone Removal
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch auto-delete-2
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit b42197b7b38f29d48b1e4a3f1cabc48281897569
Author: Robert Newson <rn...@apache.org>
AuthorDate: Thu Nov 28 10:08:57 2024 +0000
WIP CouchDB Auto-Tombstone Removal
CouchDB will completely remove deleted documents once all 'registered peers'
have checkpointed past the update seq of the delete.
In order to do this efficiently all peers must create a local document in the
database, so that the database can use map/reduce to calculate the lower bound.
All peers that wish to consume the changes feed must create a local document at
seq 0 before they start (to ensure deletes that occur while reading the changes
feed are preserved) and periodically update that local document as they
progress.
_view_cleanup has been enhanced to delete all checkpoint documents for
unreachable indexes (similar to how we clean up purge sync documents).
---
src/couch/src/couch_bt_engine.erl | 70 +++++++++++++++++++++-
src/couch/src/couch_bt_engine_compactor.erl | 30 ++++++++++
src/couch/src/couch_db.erl | 7 ++-
src/couch/src/couch_db_engine.erl | 14 ++++-
src/couch_mrview/src/couch_mrview.erl | 4 ++
src/couch_mrview/src/couch_mrview_cleanup.erl | 12 ++--
src/couch_mrview/src/couch_mrview_index.erl | 44 +++++++++++++-
src/couch_mrview/src/couch_mrview_util.erl | 20 +++++--
src/fabric/src/fabric.erl | 21 ++++---
.../src/fabric_registered_replication_peers.erl | 63 +++++++++++++++++++
src/fabric/src/fabric_rpc.erl | 6 +-
src/mem3/src/mem3_rpc.erl | 17 ++++--
12 files changed, 279 insertions(+), 29 deletions(-)
diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index 7bc02146e..16c0203da 100644
--- a/src/couch/src/couch_bt_engine.erl
+++ b/src/couch/src/couch_bt_engine.erl
@@ -79,7 +79,9 @@
count_changes_since/2,
start_compaction/4,
- finish_compaction/4
+ finish_compaction/4,
+
+ get_registered_replication_peers/1
]).
-export([
@@ -97,6 +99,7 @@
local_tree_split/1,
local_tree_join/2,
+ local_tree_reduce/2,
purge_tree_split/1,
purge_tree_join/2,
@@ -781,6 +784,67 @@ local_tree_join(Id, {Rev, BodyData}) when is_integer(Rev) ->
body = BodyData
}.
+local_tree_reduce(reduce, []) ->
+ #{};
+local_tree_reduce(reduce, Docs) ->
+ Vals = lists:map(fun extract_seqs/1, Docs),
+ local_tree_merge(Vals);
+local_tree_reduce(rereduce, Reds) ->
+ local_tree_merge(Reds).
+
+local_tree_merge(Vals) ->
+ lists:foldl(
+ fun(Map, Acc) ->
+ maps:merge_with(
+ fun(_, V1, V2) ->
+ erlang:min(V1, V2)
+ end,
+ Map,
+ Acc
+ )
+ end,
+ #{},
+ Vals
+ ).
+
+extract_seqs(#doc{} = Doc) ->
+ {Props} = Doc#doc.body,
+ case {Doc#doc.id, couch_util:get_value(<<"source_last_seq">>, Props)} of
+ {<<?LOCAL_DOC_PREFIX, "shard-sync-", _/binary>>, _} ->
+ Range = couch_util:get_value(<<"range">>, Props),
+ case Range of
+ undefined ->
+ #{};
+ [RS, RE] ->
+ {History} = couch_util:get_value(<<"history">>, Props, {[]}),
+ [{_SourceNode, [{Latest} | _]} | _] = History,
+ TargetNode = couch_util:get_value(<<"target_node">>, Latest),
+ TargetSeq = couch_util:get_value(<<"target_seq">>, Latest),
+ #{{binary_to_existing_atom(TargetNode), RS, RE} => TargetSeq}
+ end;
+ {<<?LOCAL_DOC_PREFIX, "mrview-", _/binary>>, _} ->
+ Node = couch_util:get_value(<<"node">>, Props),
+ [RS, RE] = couch_util:get_value(<<"range">>, Props),
+ Seq = couch_util:get_value(<<"seq">>, Props),
+ #{{Node, RS, RE} => Seq};
+ {<<?LOCAL_DOC_PREFIX, "purge-mrview-", _/binary>>, _} ->
+ #{};
+ {_, SourceLastSeq} ->
+ opaque_seq_to_seen_map(SourceLastSeq)
+ end.
+
+opaque_seq_to_seen_map(Opaque) ->
+ Seq = fabric_view_changes:decode_seq(Opaque),
+ maps:from_list(
+ lists:map(
+ fun
+ ({N, [RS, RE], S}) when is_integer(S) -> {{N, RS, RE}, S};
+ ({N, [RS, RE], {S, _, N}}) when is_integer(S) -> {{N, RS, RE}, S}
+ end,
+ Seq
+ )
+ ).
+
purge_tree_split({PurgeSeq, UUID, DocId, Revs}) ->
{UUID, {PurgeSeq, DocId, Revs}}.
@@ -877,6 +941,7 @@ init_state(FilePath, Fd, Header0, Options) ->
{ok, LocalTree} = couch_btree:open(LocalTreeState, Fd, [
{split, fun ?MODULE:local_tree_split/1},
{join, fun ?MODULE:local_tree_join/2},
+ {reduce, fun ?MODULE:local_tree_reduce/2},
{compression, Compression}
]),
@@ -1227,3 +1292,6 @@ is_file(Path) ->
{ok, #file_info{type = directory}} -> true;
_ -> false
end.
+
+get_registered_replication_peers(#st{} = St) ->
+ couch_btree:full_reduce(St#st.local_tree).
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 8ed55b5c3..0c339c95c 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -307,17 +307,44 @@ copy_compact(#comp_st{} = CompSt) ->
)
),
+ DropSeq =
+ case DbName of
+ <<"shards/", _/binary>> ->
+ [RS, RE] = mem3:range(DbName),
+ {ok, RegisteredPeers} = fabric:get_registered_replication_peers(
+ mem3:dbname(DbName)
+ ),
+ maps:get({node(), RS, RE}, RegisteredPeers, 0);
+ _ ->
+ 0
+ end,
+
+ couch_log:warning("Compaction process for db '~s' will drop deleted docs <= ~B", [
+ DbName, DropSeq
+ ]),
+
EnumBySeqFun =
fun(
DocInfo,
_Offset,
{AccNewSt, AccUncopied, AccUncopiedSize, AccCopiedSize}
) ->
+ Id =
+ case DocInfo of
+ #full_doc_info{} -> DocInfo#full_doc_info.id;
+ #doc_info{} -> DocInfo#doc_info.id
+ end,
Seq =
case DocInfo of
#full_doc_info{} -> DocInfo#full_doc_info.update_seq;
#doc_info{} -> DocInfo#doc_info.high_seq
end,
+ Deleted =
+ case DocInfo of
+ #full_doc_info{} -> DocInfo#full_doc_info.deleted;
+ % Older versions stored #doc_info records in the seq_tree.
+ #doc_info{} -> false
+ end,
AccUncopiedSize2 = AccUncopiedSize + ?term_size(DocInfo),
if
@@ -335,6 +362,9 @@ copy_compact(#comp_st{} = CompSt) ->
{ok, NewSt3} = couch_bt_engine:set_update_seq(NewSt2, Seq),
{ok, {NewSt3, [], 0, AccCopiedSize2}}
end;
+ Deleted andalso Seq =< DropSeq ->
+ couch_log:warning("dropped deleted doc ~s", [Id]),
+ {ok, {AccNewSt, AccUncopied, AccUncopiedSize, AccCopiedSize}};
true ->
{ok, {AccNewSt, [DocInfo | AccUncopied], AccUncopiedSize2, AccCopiedSize}}
end
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 2ef89ced3..bee20f6c5 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -129,7 +129,9 @@
validate_dbname/1,
make_doc/5,
- new_revid/1
+ new_revid/1,
+
+ get_registered_replication_peers/1
]).
-export([
@@ -2131,6 +2133,9 @@ purge_client_warn_lag(Updated, DbName, DocId) ->
Log = "~p : Purge checkpoint ~p : ~s has an invalid updated value: ~p",
couch_log:error(Log, [?MODULE, DbName, DocId, Updated]).
+get_registered_replication_peers(#db{} = Db) ->
+ couch_db_engine:get_registered_replication_peers(Db).
+
-ifdef(TEST).
-include_lib("couch/include/couch_eunit.hrl").
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 052a527e3..a5a412f86 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -658,6 +658,12 @@
) ->
{ok, CompactedDbHandle :: db_handle(), CompactorPid :: pid() | undefined}.
+-callback get_registered_replication_peers(DbHandle :: db_handle()) ->
+ {ok, #{
+ {node(), StartRange :: non_neg_integer(), EndRange :: non_neg_integer()} => Seq :: non_neg_integer()
+ }}
+ | {error, Reason :: term()}.
+
-export([
exists/2,
delete/4,
@@ -723,7 +729,9 @@
start_compaction/1,
finish_compaction/2,
- trigger_on_compact/1
+ trigger_on_compact/1,
+
+ get_registered_replication_peers/1
]).
exists(Engine, DbPath) ->
@@ -1032,3 +1040,7 @@ get_ddocs(DbName) ->
{ok, Docs} = couch_db:fold_design_docs(Db, FoldFun, [], []),
{ok, lists:reverse(Docs)}
end).
+
+get_registered_replication_peers(#db{} = Db) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:get_registered_replication_peers(EngineState).
diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl
index d126e30cc..b1e305331 100644
--- a/src/couch_mrview/src/couch_mrview.erl
+++ b/src/couch_mrview/src/couch_mrview.erl
@@ -578,6 +578,10 @@ should_skip_local_document(_IncludeSys = false, #doc{
id = <<"_local/shard-sync", _/binary>>, body = {Props}
}) ->
contains_fields([<<"target_uuid">>, <<"seq">>], Props);
+should_skip_local_document(_IncludeSys = false, #doc{
+ id = <<"_local/mrview", _/binary>>, body = {Props}
+}) ->
+ contains_fields([<<"source_last_seq">>], Props);
should_skip_local_document(_IncludeSys, _Doc) ->
false.
diff --git a/src/couch_mrview/src/couch_mrview_cleanup.erl b/src/couch_mrview/src/couch_mrview_cleanup.erl
index 5b5afbdce..f1b671dd5 100644
--- a/src/couch_mrview/src/couch_mrview_cleanup.erl
+++ b/src/couch_mrview/src/couch_mrview_cleanup.erl
@@ -14,7 +14,7 @@
-export([
run/1,
- cleanup_purges/3,
+ cleanup_checkpoints/3,
cleanup_indices/2
]).
@@ -25,14 +25,14 @@ run(Db) ->
Checkpoints = couch_mrview_util:get_purge_checkpoints(Db),
{ok, Db1} = couch_db:reopen(Db),
Sigs = couch_mrview_util:get_signatures(Db1),
- ok = cleanup_purges(Db1, Sigs, Checkpoints),
+ ok = cleanup_checkpoints(Db1, Sigs, Checkpoints),
ok = cleanup_indices(Sigs, Indices).
-cleanup_purges(DbName, Sigs, Checkpoints) when is_binary(DbName) ->
+cleanup_checkpoints(DbName, Sigs, Checkpoints) when is_binary(DbName) ->
couch_util:with_db(DbName, fun(Db) ->
- cleanup_purges(Db, Sigs, Checkpoints)
+ cleanup_checkpoints(Db, Sigs, Checkpoints)
end);
-cleanup_purges(Db, #{} = Sigs, #{} = CheckpointsMap) ->
+cleanup_checkpoints(Db, #{} = Sigs, #{} = CheckpointsMap) ->
InactiveMap = maps:without(maps:keys(Sigs), CheckpointsMap),
InactiveCheckpoints = maps:values(InactiveMap),
DeleteFun = fun(DocId) -> delete_checkpoint(Db, DocId) end,
@@ -57,7 +57,7 @@ delete_file(File) ->
delete_checkpoint(Db, DocId) ->
DbName = couch_db:name(Db),
- LogMsg = "~p : deleting inactive purge checkpoint ~s : ~s",
+ LogMsg = "~p : deleting inactive checkpoint ~s : ~s",
couch_log:debug(LogMsg, [?MODULE, DbName, DocId]),
try couch_db:open_doc(Db, DocId, []) of
{ok, Doc = #doc{}} ->
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 51777480c..eef6bca19 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -129,12 +129,14 @@ open(Db, State0) ->
NewSt = init_and_upgrade_state(Db, Fd, State, Header),
ok = commit(NewSt),
ensure_local_purge_doc(Db, NewSt),
+ write_peer_checkpoint(Db, NewSt),
{ok, NewSt};
% end of upgrade code for <= 2.x
{ok, {Sig, Header}} ->
% Matching view signatures.
NewSt = init_and_upgrade_state(Db, Fd, State, Header),
ensure_local_purge_doc(Db, NewSt),
+ write_peer_checkpoint(Db, NewSt),
check_collator_versions(DbName, NewSt),
{ok, NewSt};
{ok, {WrongSig, _}} ->
@@ -144,6 +146,7 @@ open(Db, State0) ->
),
NewSt = couch_mrview_util:reset_index(Db, Fd, State),
ensure_local_purge_doc(Db, NewSt),
+ write_peer_checkpoint(Db, NewSt),
{ok, NewSt};
{ok, Else} ->
couch_log:error(
@@ -152,10 +155,12 @@ open(Db, State0) ->
),
NewSt = couch_mrview_util:reset_index(Db, Fd, State),
ensure_local_purge_doc(Db, NewSt),
+ write_peer_checkpoint(Db, NewSt),
{ok, NewSt};
no_valid_header ->
NewSt = couch_mrview_util:reset_index(Db, Fd, State),
ensure_local_purge_doc(Db, NewSt),
+ write_peer_checkpoint(Db, NewSt),
{ok, NewSt}
end;
{error, Reason} = Error ->
@@ -210,7 +215,13 @@ finish_update(State) ->
commit(State) ->
Header = {State#mrst.sig, couch_mrview_util:make_header(State)},
- couch_file:write_header(State#mrst.fd, Header).
+ case couch_file:write_header(State#mrst.fd, Header) of
+ ok ->
+ write_peer_checkpoint(State),
+ ok;
+ Else ->
+ Else
+ end.
compact(Db, State, Opts) ->
couch_mrview_compactor:compact(Db, State, Opts).
@@ -361,3 +372,34 @@ check_collator_versions(DbName, #mrst{} = Mrst) ->
false ->
ok
end.
+
+write_peer_checkpoint(#mrst{} = State) ->
+ couch_util:with_db(State#mrst.db_name, fun(Db) ->
+ write_peer_checkpoint(Db, State)
+ end).
+
+write_peer_checkpoint(Db, #mrst{} = State) ->
+ Sig = ?l2b(couch_index_util:hexsig(State#mrst.sig)),
+ DocId = <<?LOCAL_DOC_PREFIX, "mrview-", Sig/binary>>,
+ BaseDoc = couch_doc:from_json_obj(
+ {[
+ {<<"_id">>, DocId},
+ {<<"type">>, <<"mrview-checkpoint">>},
+ {<<"node">>, node()},
+ {<<"range">>, range(State#mrst.db_name)},
+ {<<"seq">>, State#mrst.update_seq}
+ ]}
+ ),
+ Doc =
+ case couch_db:open_doc(Db, DocId, []) of
+ {ok, #doc{revs = Revs}} ->
+ BaseDoc#doc{revs = Revs};
+ {not_found, _} ->
+ BaseDoc
+ end,
+ {ok, _} = couch_db:update_doc(Db, Doc, []).
+
+range(<<"shards/", _/binary>> = DbName) ->
+ mem3:range(DbName);
+range(_) ->
+ [0, 1].
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index a478685da..33aaf8a9c 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -36,6 +36,7 @@
-export([get_collator_versions/1]).
-export([compact_on_collator_upgrade/0]).
-export([commit_on_header_upgrade/0]).
+-export([get_mrview_checkpoints/1]).
-define(MOD, couch_mrview_index).
-define(GET_VIEW_RETRY_COUNT, 1).
@@ -114,18 +115,27 @@ get_signatures(Db) ->
% Returns a map of `Sig => DocId` elements for all the purge view
% checkpoint docs. Sig is a hex-encoded binary.
%
-get_purge_checkpoints(DbName) when is_binary(DbName) ->
- couch_util:with_db(DbName, fun get_purge_checkpoints/1);
-get_purge_checkpoints(Db) ->
+get_purge_checkpoints(DbOrName) ->
+ get_registered_peers(DbOrName, <<?LOCAL_DOC_PREFIX, "purge-mrview-">>).
+
+% Returns a map of `Sig => DocId` elements for all the mrview
+% checkpoint docs. Sig is a hex-encoded binary.
+%
+get_mrview_checkpoints(DbOrName) ->
+ get_registered_peers(DbOrName, <<?LOCAL_DOC_PREFIX, "mrview-">>).
+
+get_registered_peers(DbName, Prefix) when is_binary(DbName), is_binary(Prefix) ->
+ couch_util:with_db(DbName, fun(Db) -> get_registered_peers(Db, Prefix) end);
+get_registered_peers(Db, Prefix) ->
FoldFun = fun(#doc{id = Id}, Acc) ->
case Id of
- <<?LOCAL_DOC_PREFIX, "purge-mrview-", Sig/binary>> ->
+ <<Prefix:(byte_size(Prefix))/binary, Sig/binary>> ->
{ok, Acc#{Sig => Id}};
_ ->
{stop, Acc}
end
end,
- Opts = [{start_key, <<?LOCAL_DOC_PREFIX, "purge-mrview-">>}],
+ Opts = [{start_key, Prefix}],
{ok, Signatures = #{}} = couch_db:fold_local_docs(Db, FoldFun, #{}, Opts),
Signatures.
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 93a71b035..c717e0040 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -23,6 +23,7 @@
delete_db/1,
delete_db/2,
get_db_info/1,
+ get_registered_replication_peers/1,
get_doc_count/1, get_doc_count/2,
set_revs_limit/3,
set_security/2, set_security/3,
@@ -107,6 +108,9 @@ all_dbs(Prefix) when is_list(Prefix) ->
get_db_info(DbName) ->
fabric_db_info:go(dbname(DbName)).
+get_registered_replication_peers(DbName) ->
+ fabric_registered_replication_peers:go(dbname(DbName)).
+
%% @doc returns the size of a given partition
-spec get_partition_info(dbname(), Partition :: binary()) ->
{ok, [
@@ -570,10 +574,10 @@ reset_validation_funs(DbName) ->
].
%% @doc clean up index files for all Dbs
--spec cleanup_index_files() -> [ok].
+-spec cleanup_index_files() -> ok.
cleanup_index_files() ->
{ok, Dbs} = fabric:all_dbs(),
- [cleanup_index_files(Db) || Db <- Dbs].
+ lists:foreach(fun cleanup_index_files/1, Dbs).
%% @doc clean up index files for a specific db
-spec cleanup_index_files(dbname()) -> ok.
@@ -591,15 +595,16 @@ cleanup_local_indices_and_purge_checkpoints([]) ->
cleanup_local_indices_and_purge_checkpoints([_ | _] = Dbs) ->
AllIndices = lists:map(fun couch_mrview_util:get_index_files/1, Dbs),
AllPurges = lists:map(fun couch_mrview_util:get_purge_checkpoints/1, Dbs),
+ AllCheckpoints = lists:map(fun couch_mrview_util:get_mrview_checkpoints/1, Dbs),
Sigs = couch_mrview_util:get_signatures(hd(Dbs)),
- ok = cleanup_purges(Sigs, AllPurges, Dbs),
+ ok = cleanup_checkpoints(Sigs, AllPurges ++ AllCheckpoints, Dbs),
ok = cleanup_indices(Sigs, AllIndices).
-cleanup_purges(Sigs, AllPurges, Dbs) ->
- Fun = fun(DbPurges, Db) ->
- couch_mrview_cleanup:cleanup_purges(Db, Sigs, DbPurges)
+cleanup_checkpoints(Sigs, AllCheckpoints, Dbs) ->
+ Fun = fun(Checkpoints, Db) ->
+ couch_mrview_cleanup:cleanup_checkpoints(Db, Sigs, Checkpoints)
end,
- lists:zipwith(Fun, AllPurges, Dbs),
+ lists:zipwith(Fun, AllCheckpoints, Dbs),
ok.
cleanup_indices(Sigs, AllIndices) ->
@@ -609,7 +614,7 @@ cleanup_indices(Sigs, AllIndices) ->
lists:foreach(Fun, AllIndices).
%% @doc clean up index files for a specific db on all nodes
--spec cleanup_index_files_all_nodes(dbname()) -> [reference()].
+-spec cleanup_index_files_all_nodes(dbname()) -> ok.
cleanup_index_files_all_nodes(DbName) ->
lists:foreach(
fun(Node) ->
diff --git a/src/fabric/src/fabric_registered_replication_peers.erl b/src/fabric/src/fabric_registered_replication_peers.erl
new file mode 100644
index 000000000..430edcca7
--- /dev/null
+++ b/src/fabric/src/fabric_registered_replication_peers.erl
@@ -0,0 +1,63 @@
+% 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_registered_replication_peers).
+
+-export([go/1]).
+
+-include_lib("fabric/include/fabric.hrl").
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+go(DbName) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, get_registered_replication_peers, []),
+ RexiMon = fabric_util:create_monitors(Shards),
+ Acc0 = {fabric_dict:init(Workers, nil), []},
+ try fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Acc0) of
+ {timeout, {WorkersDict, _}} ->
+ DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil),
+ fabric_util:log_timeout(DefunctWorkers, "get_registered_replication_peers"),
+ {error, timeout};
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _Shard, {Counters, Resps}) ->
+ case fabric_ring:node_down(NodeRef, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, {nodedown, <<"progress not possible">>}}
+ end;
+handle_message({rexi_EXIT, Reason}, Shard, {Counters, Resps}) ->
+ case fabric_ring:handle_error(Shard, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, Reason}
+ end;
+handle_message({ok, Reds}, Shard, {Counters, Resps}) ->
+ case fabric_ring:handle_response(Shard, Reds, Counters, Resps) of
+ {ok, {Counters1, Resps1}} ->
+ {ok, {Counters1, Resps1}};
+ {stop, Resps1} ->
+ FinalRed = fabric_dict:fold(
+ fun(_, Red, Acc) -> couch_bt_engine:local_tree_reduce(rereduce, [Red, Acc]) end,
+ #{},
+ Resps1
+ ),
+ {stop, FinalRed}
+ end;
+handle_message(Reason, Shard, {Counters, Resps}) ->
+ case fabric_ring:handle_error(Shard, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, Reason}
+ end.
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 9b00d9501..104cf6ae8 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -51,7 +51,8 @@
reduce_view/5,
group_info/3,
update_mrview/4,
- get_uuid/1
+ get_uuid/1,
+ get_registered_replication_peers/1
]).
-include_lib("fabric/include/fabric.hrl").
@@ -347,6 +348,9 @@ compact(ShardName, DesignName) ->
get_uuid(DbName) ->
with_db(DbName, [], {couch_db, get_uuid, []}).
+get_registered_replication_peers(DbName) ->
+ with_db(DbName, [], {couch_db, get_registered_replication_peers, []}).
+
%%
%% internal
%%
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index 70fc797da..0b7cffdc9 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -140,11 +140,13 @@ save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
] ++ NewEntry0
},
Body =
- {[
- {<<"seq">>, SourceSeq},
- {<<"target_uuid">>, couch_db:get_uuid(Db)},
- {<<"history">>, add_checkpoint(NewEntry, History0)}
- ]},
+ {
+ [
+ {<<"seq">>, SourceSeq},
+ {<<"target_uuid">>, couch_db:get_uuid(Db)},
+ {<<"history">>, add_checkpoint(NewEntry, History0)}
+ ] ++ range(DbName)
+ },
Doc = #doc{id = Id, body = Body},
rexi:reply(
try couch_db:update_doc(Db, Doc, []) of
@@ -163,6 +165,11 @@ save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
rexi:reply(Error)
end.
+range(<<"shards/", _/binary>> = DbName) ->
+ [{<<"range">>, mem3:range(DbName)}];
+range(_) ->
+ [].
+
find_common_seq_rpc(DbName, SourceUUID, SourceEpochs) ->
erlang:put(io_priority, {internal_repl, DbName}),
case get_or_create_db(DbName, [?ADMIN_CTX]) of