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 2025/03/13 18:25:27 UTC
(couchdb) branch auto-delete-3 updated (f1d1e8baf -> c5e6a5235)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch auto-delete-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
discard f1d1e8baf tidier?
discard f32920b36 don't go lower than latest sync
discard 92da6e598 wip
discard 59b838eeb wip
discard 7d75543b9 wip
discard 8c0cf6e53 wip
discard 19630d0cd wip
discard 02e5dcb24 wip
discard d4ebb1c06 wip
discard 67d9b0f5b remove uuid_prefix_len for drop_seq safety
discard e8ca91233 wip
discard 9ab85a15b wip
discard 2124fe35a wip
discard b64a645a3 Introduce drop_seq
new c5e6a5235 Introduce drop_seq
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 (f1d1e8baf)
\
N -- N -- N refs/heads/auto-delete-3 (c5e6a5235)
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:
rel/overlay/etc/default.ini | 1 +
src/couch/src/couch_bt_engine.erl | 30 +++++++++----
src/couch/src/couch_bt_engine_header.erl | 3 +-
src/couch/src/couch_db.erl | 9 ++++
src/couch/src/couch_db_engine.erl | 15 +++++++
src/couch/src/couch_db_updater.erl | 9 ++++
src/fabric/src/fabric_drop_seq.erl | 72 ++++++++++++++++++++++++++++----
src/fabric/src/fabric_rpc.erl | 6 ++-
src/fabric/src/fabric_util.erl | 3 +-
9 files changed, 127 insertions(+), 21 deletions(-)
(couchdb) 01/01: Introduce drop_seq
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-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c5e6a52351d02be5eefa4de3768797a22a4dff1f
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Nov 5 14:14:15 2024 +0000
Introduce drop_seq
Any deleted documents with an update seq equal or lower than
drop seq is completely removed during compaction.
---
src/couch/src/couch_bt_engine.erl | 27 +++
src/couch/src/couch_bt_engine_compactor.erl | 10 +
src/couch/src/couch_bt_engine_header.erl | 12 +-
src/couch/src/couch_db.erl | 9 +
src/couch/src/couch_db_engine.erl | 22 ++
src/couch/src/couch_db_updater.erl | 9 +
.../src/couch_replicator_scheduler_job.erl | 10 +
src/fabric/src/fabric.erl | 8 +-
src/fabric/src/fabric_drop_seq.erl | 224 +++++++++++++++++++++
src/fabric/src/fabric_rpc.erl | 6 +-
src/mem3/src/mem3_rpc.erl | 1 +
11 files changed, 333 insertions(+), 5 deletions(-)
diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index 1cd3f6d00..13efbba61 100644
--- a/src/couch/src/couch_bt_engine.erl
+++ b/src/couch/src/couch_bt_engine.erl
@@ -44,6 +44,7 @@
get_size_info/1,
get_partition_info/2,
get_update_seq/1,
+ get_drop_seq/1,
get_uuid/1,
set_revs_limit/2,
@@ -52,6 +53,7 @@
set_props/2,
set_update_seq/2,
+ set_drop_seq/3,
open_docs/2,
open_local_docs/2,
@@ -326,6 +328,9 @@ get_props(#st{header = Header} = St) ->
get_update_seq(#st{header = Header}) ->
couch_bt_engine_header:get(Header, update_seq).
+get_drop_seq(#st{header = Header}) ->
+ couch_bt_engine_header:get(Header, drop_seq).
+
get_uuid(#st{header = Header}) ->
couch_bt_engine_header:get(Header, uuid).
@@ -802,6 +807,28 @@ set_update_seq(#st{header = Header} = St, UpdateSeq) ->
needs_commit = true
}}.
+set_drop_seq(#st{header = Header} = St, ExpectedUuidPrefix, NewDropSeq) when
+ is_binary(ExpectedUuidPrefix), is_integer(NewDropSeq), NewDropSeq > 0
+->
+ CurrentDropSeq = get_drop_seq(St),
+ Uuid = get_uuid(St),
+ ActualUuidPrefix = binary:part(Uuid, 0, byte_size(ExpectedUuidPrefix)),
+
+ if
+ NewDropSeq < CurrentDropSeq ->
+ {error, drop_seq_cant_decrease};
+ ExpectedUuidPrefix /= ActualUuidPrefix ->
+ {error, uuid_mismatch};
+ true ->
+ NewSt = St#st{
+ header = couch_bt_engine_header:set(Header, [
+ {drop_seq, NewDropSeq}
+ ]),
+ needs_commit = true
+ },
+ {ok, increment_update_seq(NewSt)}
+ end.
+
copy_security(#st{header = Header} = St, SecProps) ->
Options = [{compression, St#st.compression}],
{ok, Ptr, _} = couch_file:append_term(St#st.fd, SecProps, Options),
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 8ed55b5c3..619200bce 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -295,6 +295,7 @@ copy_compact(#comp_st{} = CompSt) ->
Compression = couch_compress:get_compression_method(),
NewSt = NewSt0#st{compression = Compression},
NewUpdateSeq = couch_bt_engine:get_update_seq(NewSt0),
+ DropSeq = couch_bt_engine:get_drop_seq(NewSt0),
TotalChanges = couch_bt_engine:count_changes_since(St, NewUpdateSeq),
BufferSize = list_to_integer(
config:get("database_compaction", "doc_buffer_size", "524288")
@@ -318,9 +319,18 @@ copy_compact(#comp_st{} = CompSt) ->
#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
+ Deleted andalso Seq =< DropSeq ->
+ %% drop this document completely
+ {ok, {AccNewSt, AccUncopied, AccUncopiedSize, AccCopiedSize}};
AccUncopiedSize2 >= BufferSize ->
NewSt2 = copy_docs(
St, AccNewSt, lists:reverse([DocInfo | AccUncopied]), Retry
diff --git a/src/couch/src/couch_bt_engine_header.erl b/src/couch/src/couch_bt_engine_header.erl
index 3581b1e39..9f5285d49 100644
--- a/src/couch/src/couch_bt_engine_header.erl
+++ b/src/couch/src/couch_bt_engine_header.erl
@@ -38,7 +38,8 @@
revs_limit/1,
uuid/1,
epochs/1,
- compacted_seq/1
+ compacted_seq/1,
+ drop_seq/1
]).
-include_lib("stdlib/include/assert.hrl").
@@ -71,7 +72,8 @@
epochs,
compacted_seq,
purge_infos_limit = 1000,
- props_ptr
+ props_ptr,
+ drop_seq = 0
}).
-define(PARTITION_DISK_VERSION, 8).
@@ -87,7 +89,8 @@ from(Header0) ->
#db_header{
uuid = Header#db_header.uuid,
epochs = Header#db_header.epochs,
- compacted_seq = Header#db_header.compacted_seq
+ compacted_seq = Header#db_header.compacted_seq,
+ drop_seq = Header#db_header.drop_seq
}.
is_header(Header) ->
@@ -176,6 +179,9 @@ epochs(Header) ->
compacted_seq(Header) ->
get_field(Header, compacted_seq).
+drop_seq(Header) ->
+ get_field(Header, drop_seq).
+
purge_infos_limit(Header) ->
get_field(Header, purge_infos_limit).
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 2ef89ced3..662792e08 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -50,6 +50,7 @@
get_revs_limit/1,
get_security/1,
get_update_seq/1,
+ get_drop_seq/1,
get_user_ctx/1,
get_uuid/1,
get_purge_seq/1,
@@ -66,6 +67,7 @@
set_purge_infos_limit/2,
set_security/2,
set_user_ctx/2,
+ set_drop_seq/3,
load_validation_funs/1,
reload_validation_funs/1,
@@ -553,6 +555,9 @@ get_committed_update_seq(#db{committed_update_seq = Seq}) ->
get_update_seq(#db{} = Db) ->
couch_db_engine:get_update_seq(Db).
+get_drop_seq(#db{} = Db) ->
+ couch_db_engine:get_drop_seq(Db).
+
get_user_ctx(#db{user_ctx = UserCtx}) ->
UserCtx;
get_user_ctx(?OLD_DB_REC = Db) ->
@@ -795,6 +800,10 @@ set_security(_, _) ->
set_user_ctx(#db{} = Db, UserCtx) ->
{ok, Db#db{user_ctx = UserCtx}}.
+set_drop_seq(#db{main_pid = Pid} = Db, UuidPrefix, DropSeq) ->
+ check_is_admin(Db),
+ gen_server:call(Pid, {set_drop_seq, UuidPrefix, DropSeq}).
+
validate_security_object(SecProps) ->
Admins = couch_util:get_value(<<"admins">>, SecProps, {[]}),
% we fallback to readers here for backwards compatibility
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 54f2c1482..1d89066e8 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -263,6 +263,9 @@
% the database.
-callback get_update_seq(DbHandle :: db_handle()) -> UpdateSeq :: non_neg_integer().
+% The current drop sequence of the database.
+-callback get_drop_seq(DbHandle :: db_handle()) -> DropSeq :: non_neg_integer().
+
% Whenever a database is created it should generate a
% persistent UUID for identification in case the shard should
% ever need to be moved between nodes in a cluster.
@@ -299,6 +302,14 @@
) ->
{ok, NewDbHandle :: db_handle()}.
+% Set the drop sequence of the database.
+-callback set_drop_seq(
+ DbHandle :: db_handle(),
+ UuidPrefix :: binary,
+ DropSeq :: non_neg_integer()
+) ->
+ {ok, NewDbHandle :: db_handle()}.
+
% This function will be called by many processes concurrently.
% It should return a #full_doc_info{} record or not_found for
% every provided DocId in the order those DocId's appear in
@@ -680,6 +691,7 @@
get_size_info/1,
get_partition_info/2,
get_update_seq/1,
+ get_drop_seq/1,
get_uuid/1,
set_revs_limit/2,
@@ -688,6 +700,7 @@
set_props/2,
set_update_seq/2,
+ set_drop_seq/3,
open_docs/2,
open_local_docs/2,
@@ -841,6 +854,10 @@ get_update_seq(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:get_update_seq(EngineState).
+get_drop_seq(#db{} = Db) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:get_drop_seq(EngineState).
+
get_uuid(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:get_uuid(EngineState).
@@ -870,6 +887,11 @@ set_update_seq(#db{} = Db, UpdateSeq) ->
{ok, NewSt} = Engine:set_update_seq(EngineState, UpdateSeq),
{ok, Db#db{engine = {Engine, NewSt}}}.
+set_drop_seq(#db{} = Db, UuidPrefix, UpdateSeq) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ {ok, NewSt} = Engine:set_drop_seq(EngineState, UuidPrefix, UpdateSeq),
+ {ok, Db#db{engine = {Engine, NewSt}}}.
+
open_docs(#db{} = Db, DocIds) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:open_docs(EngineState, DocIds).
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index 3f6c8886d..cdf020f36 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -92,6 +92,15 @@ handle_call({set_purge_infos_limit, Limit}, _From, Db) ->
{ok, Db2} = couch_db_engine:set_purge_infos_limit(Db, Limit),
ok = couch_server:db_updated(Db2),
{reply, ok, Db2};
+handle_call({set_drop_seq, UuidPrefix, DropSeq}, _From, Db) ->
+ case couch_db_engine:set_drop_seq(Db, UuidPrefix, DropSeq) of
+ {ok, Db2} ->
+ Db3 = commit_data(Db2),
+ ok = couch_server:db_updated(Db3),
+ {reply, ok, Db3};
+ {error, Reason} ->
+ {reply, {error, Reason}, Db}
+ end;
handle_call({purge_docs, [], _}, _From, Db) ->
{reply, {ok, []}, Db};
handle_call({purge_docs, PurgeReqs0, Options}, _From, Db) ->
diff --git a/src/couch_replicator/src/couch_replicator_scheduler_job.erl b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
index 544c5602a..62e604b5e 100644
--- a/src/couch_replicator/src/couch_replicator_scheduler_job.erl
+++ b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
@@ -886,6 +886,7 @@ do_checkpoint(State) ->
{TgtRevPos, TgtRevId} = update_checkpoint(
Target, TargetLog#doc{body = NewRepHistory}, target
),
+ %% TODO update_checkpoint(Source, peer_checkpoint_doc(State), source),
NewState = State#rep_state{
checkpoint_history = NewRepHistory,
committed_seq = NewTsSeq,
@@ -912,6 +913,15 @@ do_checkpoint(State) ->
>>}
end.
+peer_checkpoint_doc(#rep_state{} = State) ->
+ #rep_state{
+ session_id = SessionId
+ } = State,
+ #doc{
+ id = <<"peer-checkpoint-", SessionId/binary>>,
+ body = {[{<<"update_seq">>, State#rep_state.committed_seq}]}
+ }.
+
update_checkpoint(Db, Doc, DbType) ->
try
update_checkpoint(Db, Doc)
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 93a71b035..9f3fbb767 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -33,7 +33,8 @@
set_purge_infos_limit/3,
get_purged_infos/1,
compact/1, compact/2,
- get_partition_info/2
+ get_partition_info/2,
+ calculate_drop_seq/1
]).
% Documents
@@ -119,6 +120,11 @@ get_db_info(DbName) ->
get_partition_info(DbName, Partition) ->
fabric_db_partition_info:go(dbname(DbName), Partition).
+-spec calculate_drop_seq(dbname()) ->
+ {ok, [{node(), binary(), non_neg_integer()}]}.
+calculate_drop_seq(DbName) ->
+ fabric_drop_seq:go(dbname(DbName)).
+
%% @doc the number of docs in a database
%% @equiv get_doc_count(DbName, <<"_all_docs">>)
get_doc_count(DbName) ->
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
new file mode 100644
index 000000000..c00ea24ea
--- /dev/null
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -0,0 +1,224 @@
+-module(fabric_drop_seq).
+
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-export([go/1]).
+
+-type range() :: [non_neg_integer()].
+
+-type uuid() :: binary().
+
+-type seq() :: non_neg_integer().
+
+-type peer_checkpoints() :: #{{range(), Node :: node()} => {Uuid :: uuid(), Seq :: seq()}}.
+
+-type history_item() :: {
+ SourceUuid :: uuid(), SourceSeq :: seq(), TargetUuid :: uuid(), TargetSeq :: seq()
+}.
+
+-type shard_sync_history() :: #{
+ {Range :: range(), SourceNode :: node(), TargetNode :: node()} => [history_item()]
+}.
+
+go(DbName) ->
+ {PeerCheckpoints0, ShardSyncHistory} = parse_local_docs(DbName),
+ ShardSyncCheckpoints = latest_shard_sync_checkpoints(ShardSyncHistory),
+ PeerCheckpoints1 = maps:merge_with(fun merge_peers/3, PeerCheckpoints0, ShardSyncCheckpoints),
+ PeerCheckpoints2 = crossref(PeerCheckpoints1, ShardSyncHistory),
+ Shards = mem3:live_shards(DbName, [node() | nodes()]),
+ RexiMon = fabric_util:create_monitors(Shards),
+ Workers = lists:filtermap(
+ fun(Shard) ->
+ #shard{range = Range, node = Node, name = ShardName} = Shard,
+ case maps:find({Range, Node}, PeerCheckpoints2) of
+ {ok, {UuidPrefix, DropSeq}} ->
+ Ref = rexi:cast(
+ Node,
+ {fabric_rpc, set_drop_seq, [ShardName, UuidPrefix, DropSeq, [?ADMIN_CTX]]}
+ ),
+ {true, Shard#shard{ref = Ref}};
+ error ->
+ false
+ end
+ end,
+ Shards
+ ),
+ Acc0 = {Workers, length(Workers) - 1},
+ try
+ case fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Acc0) of
+ {ok, ok} ->
+ ok;
+ {timeout, {WorkersDict, _}} ->
+ DefunctWorkers = fabric_util:remove_done_workers(
+ WorkersDict,
+ nil
+ ),
+ fabric_util:log_timeout(
+ DefunctWorkers,
+ "set_drop_seq"
+ ),
+ {error, timeout};
+ {error, Reason} ->
+ {error, Reason}
+ end
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+handle_message(ok, _Worker, {_Workers, 0}) ->
+ {stop, ok};
+handle_message(ok, Worker, {Workers, Waiting}) ->
+ {ok, {lists:delete(Worker, Workers), Waiting - 1}};
+handle_message(Error, _, _Acc) ->
+ {error, Error}.
+
+crossref(PeerCheckpoints0, ShardSyncHistory) ->
+ PeerCheckpoints1 = maps:fold(
+ fun({Range, Node}, {Uuid, Seq}, Acc1) ->
+ Others = maps:filter(
+ fun({R, _S, T}, _V) -> R == Range andalso T /= Node end, ShardSyncHistory
+ ),
+ maps:fold(
+ fun({R, _S, T}, H, Acc2) ->
+ case
+ lists:search(
+ fun({SU, SS, _TU, _TS}) -> Uuid == SU andalso SS =< Seq end,
+ H
+ )
+ of
+ {value, {_SU, _SS, TU, TS}} ->
+ maps:merge_with(fun merge_peers/3, #{{R, T} => {TU, TS}}, Acc2);
+ false ->
+ Acc2
+ end
+ end,
+ Acc1,
+ Others
+ )
+ end,
+ PeerCheckpoints0,
+ PeerCheckpoints0
+ ),
+
+ %% mem3 sync is not hub-spoke, each iteration of crossref will add
+ %% some crossreferences. we call it again if the map changes as new
+ %% crossreferences may be possible.
+ if
+ PeerCheckpoints0 == PeerCheckpoints1 ->
+ PeerCheckpoints1;
+ true ->
+ crossref(PeerCheckpoints1, ShardSyncHistory)
+ end.
+
+-spec parse_local_docs(DbName :: binary()) -> {peer_checkpoints(), shard_sync_history()}.
+parse_local_docs(DbName) ->
+ {ok, Result} = fabric:all_docs(
+ DbName, fun parse_local_docs_cb/2, {#{}, #{}}, all_docs_mrargs()
+ ),
+ Result.
+
+parse_local_docs_cb({row, Row}, Acc) ->
+ case lists:keyfind(doc, 1, Row) of
+ false ->
+ {ok, Acc};
+ {doc, Doc} ->
+ parse_local_doc(couch_doc:from_json_obj(Doc), Acc)
+ end;
+parse_local_docs_cb(_Else, Acc) ->
+ {ok, Acc}.
+
+parse_local_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
+ {Props} = Doc#doc.body,
+ case couch_util:get_value(<<"dbname">>, Props) of
+ undefined ->
+ %% not yet upgraded with new property
+ {ok, Acc};
+ DbName ->
+ Range = mem3:range(DbName),
+ {[{_SrcNode, History}]} = couch_util:get_value(<<"history">>, Props),
+ {PeerCheckpoints, ShardSyncHistory0} = Acc,
+ KeyFun = fun({Item}) ->
+ {Range, binary_to_existing_atom(couch_util:get_value(<<"source_node">>, Item)),
+ binary_to_existing_atom(couch_util:get_value(<<"target_node">>, Item))}
+ end,
+ ValueFun = fun({Item}) ->
+ {
+ couch_util:get_value(<<"source_uuid">>, Item),
+ couch_util:get_value(<<"source_seq">>, Item),
+ couch_util:get_value(<<"target_uuid">>, Item),
+ couch_util:get_value(<<"target_seq">>, Item)
+ }
+ end,
+
+ ShardSyncHistory1 = maps:merge(
+ maps:groups_from_list(KeyFun, ValueFun, History), ShardSyncHistory0
+ ),
+ {ok, {PeerCheckpoints, ShardSyncHistory1}}
+ end;
+parse_local_doc(#doc{id = <<"_local/peer-checkpoint-", _/binary>>} = Doc, Acc) ->
+ {Props} = Doc#doc.body,
+ case couch_util:get_value(<<"update_seq">>, Props) of
+ undefined ->
+ {ok, Acc};
+ UpdateSeq ->
+ {PeerCheckpoints0, ShardSyncHistory} = Acc,
+ PeerCheckpoints1 = maps:merge_with(
+ fun merge_peers/3, decode_seq(UpdateSeq), PeerCheckpoints0
+ ),
+ {ok, {PeerCheckpoints1, ShardSyncHistory}}
+ end;
+parse_local_doc(_Doc, Acc) ->
+ {ok, Acc}.
+
+merge_peers(_Key, {Uuid1, Val1}, {Uuid2, Val2}) when is_integer(Val1), is_integer(Val2) ->
+ PrefixLen = min(byte_size(Uuid1), byte_size(Uuid2)),
+ case binary:longest_common_prefix([Uuid1, Uuid2]) == PrefixLen of
+ true ->
+ {Uuid1, min(Val1, Val2)};
+ false ->
+ {Uuid2, Val2}
+ end.
+
+decode_seq(OpaqueSeq) ->
+ Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
+ lists:foldl(
+ fun
+ ({_Node, [S, E], {Seq, Uuid, Node}}, Acc) when
+ is_integer(S),
+ is_integer(E),
+ is_integer(Seq),
+ S >= 0,
+ E > S,
+ Seq > 0,
+ is_binary(Uuid),
+ is_atom(Node)
+ ->
+ Acc#{{[S, E], Node} => {Uuid, Seq}};
+ (_Else, Acc) ->
+ Acc
+ end,
+ #{},
+ Decoded
+ ).
+
+all_docs_mrargs() ->
+ #mrargs{
+ view_type = map,
+ include_docs = true,
+ extra = [
+ {include_system, true},
+ {namespace, <<"_local">>}
+ ]
+ }.
+
+latest_shard_sync_checkpoints(ShardSyncHistory) ->
+ maps:fold(
+ fun({R, SN, _TN}, History, Acc) ->
+ {SU, SS, _TU, _TS} = hd(History),
+ maps:merge_with(fun merge_peers/3, #{{R, SN} => {SU, SS}}, Acc)
+ end,
+ #{},
+ ShardSyncHistory
+ ).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 9b00d9501..1128ab757 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,
+ set_drop_seq/4
]).
-include_lib("fabric/include/fabric.hrl").
@@ -347,6 +348,9 @@ compact(ShardName, DesignName) ->
get_uuid(DbName) ->
with_db(DbName, [], {couch_db, get_uuid, []}).
+set_drop_seq(DbName, UuidPrefix, DropSeq, Options) ->
+ with_db(DbName, Options, {couch_db, set_drop_seq, [UuidPrefix, DropSeq]}).
+
%%
%% internal
%%
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index 70fc797da..b00eb5637 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -141,6 +141,7 @@ save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
},
Body =
{[
+ {<<"dbname">>, DbName},
{<<"seq">>, SourceSeq},
{<<"target_uuid">>, couch_db:get_uuid(Db)},
{<<"history">>, add_checkpoint(NewEntry, History0)}