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/03 21:00:37 UTC
(couchdb) branch auto-delete-3 created (now 59b838eeb)
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
at 59b838eeb wip
This branch includes the following new commits:
new b64a645a3 Introduce drop_seq
new 2124fe35a wip
new 9ab85a15b wip
new e8ca91233 wip
new 67d9b0f5b remove uuid_prefix_len for drop_seq safety
new d4ebb1c06 wip
new 02e5dcb24 wip
new 19630d0cd wip
new 8c0cf6e53 wip
new 7d75543b9 wip
new 59b838eeb wip
The 11 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.
(couchdb) 01/11: 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 b64a645a3c65f2c1bb63c70bbb0460860580da45
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 | 13 +++++++++++++
src/couch/src/couch_bt_engine_compactor.erl | 9 +++++++++
src/couch/src/couch_bt_engine_header.erl | 9 +++++++--
src/couch/src/couch_db_engine.erl | 7 +++++++
4 files changed, 36 insertions(+), 2 deletions(-)
diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index 1cd3f6d00..ade83075e 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/2,
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,14 @@ set_update_seq(#st{header = Header} = St, UpdateSeq) ->
needs_commit = true
}}.
+set_drop_seq(#st{header = Header} = St, DropSeq) when is_integer(DropSeq) ->
+ {ok, St#st{
+ header = couch_bt_engine_header:set(Header, [
+ {drop_seq, DropSeq}
+ ]),
+ needs_commit = true
+ }}.
+
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..5e9289d71 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,17 @@ 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;
+ #doc_info{} -> false % Older versions stored #doc_info records in the seq_tree.
+ 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..c661507e3 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).
@@ -176,6 +178,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_engine.erl b/src/couch/src/couch_db_engine.erl
index 54f2c1482..8fd91ddf7 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -299,6 +299,13 @@
) ->
{ok, NewDbHandle :: db_handle()}.
+% Set the drop sequence of the database.
+-callback set_drop_seq(
+ DbHandle :: db_handle(),
+ 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
(couchdb) 09/11: wip
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 8c0cf6e53022fda383cc2423ff4ffcebd19261cb
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Mar 3 20:50:56 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 52 +++++++++----
src/fabric/src/fabric_drop_seq2.erl | 148 ++++++++++++++++++++++++++++++++++++
2 files changed, 187 insertions(+), 13 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index d0f474bb7..501837ffe 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -37,16 +37,26 @@ go(DbName) ->
ShardSyncHistory = parse_shard_sync_docs(LocalDocs),
Shards = mem3:shards(DbName),
- ByRange = maps:groups_from_list(fun(Shard) ->Shard#shard.range end,
- fun(Shard) ->
- Shard#shard.node end, Shards),
+ ByRange = maps:groups_from_list(
+ fun(Shard) -> Shard#shard.range end,
+ fun(Shard) ->
+ Shard#shard.node
+ end,
+ Shards
+ ),
Fun1 = fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
OtherNodes = maps:get(Range, ByRange, []) -- [SrcNode],
Fun2 = fun(TgtNode, Acc2) ->
History = maps:get({Range, SrcNode, TgtNode}, ShardSyncHistory, []),
- case lists:search(fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
- SourceSeq =< Seq end, History) of
+ case
+ lists:search(
+ fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
+ Uuid == SourceUuid andalso SourceSeq =< Seq
+ end,
+ History
+ )
+ of
{value, {SourceUuid, SourceSeq, TargetUuid, TargetSeq}} ->
Acc2#{{Range, TgtNode} => {TargetUuid, TargetSeq}};
false ->
@@ -58,7 +68,14 @@ go(DbName) ->
Expanded = maps:fold(Fun1, PeerCheckpoints, PeerCheckpoints),
- {PeerCheckpoints, ShardSyncHistory, Expanded}.
+ Workers = lists:map(
+ fun(#shard{} = Shard) ->
+ {Shard, maps:get({Shard#shard.range, Shard#shard.node}, Expanded)}
+ end,
+ Shards
+ ),
+
+ {PeerCheckpoints, ShardSyncHistory, Expanded, Workers}.
all_docs_cb({row, Row}, Acc0) ->
case lists:keyfind(doc, 1, Row) of
@@ -105,10 +122,10 @@ parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) -
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)
+ 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,
maps:merge(maps:groups_from_list(KeyFun, ValueFun, History), Acc)
@@ -120,9 +137,18 @@ decode_seq(OpaqueSeq) ->
Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
lists:foldl(
fun
- ({_Node, Range, {Seq, Uuid, Node}}, Acc) ->
- Acc#{{Range, Node} => {Uuid, Seq}};
- ({_Node, _Range, _Seq}, Acc) ->
+ ({_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,
#{},
diff --git a/src/fabric/src/fabric_drop_seq2.erl b/src/fabric/src/fabric_drop_seq2.erl
new file mode 100644
index 000000000..13edfda37
--- /dev/null
+++ b/src/fabric/src/fabric_drop_seq2.erl
@@ -0,0 +1,148 @@
+-module(fabric_drop_seq2).
+
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-export([go/1]).
+
+go(DbName) ->
+ Shards = mem3:shards(DbName),
+ RangeToNodes = range_to_nodes(Shards),
+
+ {ok, {PeerCheckpoints, ShardSyncHistory}} = parse_local_docs(DbName),
+ F = lists:map(calculate_drop_seq(RangeToNodes, PeerCheckpoints, ShardSyncHistory), Shards),
+ {PeerCheckpoints, F}.
+
+calculate_drop_seq(RangeToNodes, PeerCheckpoints, ShardSyncHistory) ->
+ fun(#shard{node = Node, range = Range} = Shard) ->
+ {Shard, case maps:find({Range, Node}, PeerCheckpoints) of
+ error ->
+ {<<>>, 0};
+ {ok, {Uuid, Seq}} ->
+ PeerNodes = maps:get(Range, RangeToNodes, []) -- [Node],
+ lists:foldl(
+ fun(PeerNode, {U, S}) ->
+ case maps:find({Range, PeerNode, Node}, ShardSyncHistory) of
+ error ->
+ {U, S};
+ {ok, History} ->
+ case
+ lists:search(
+ fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
+ TargetUuid == U andalso TargetSeq =< S
+ end,
+ History
+ )
+ of
+ {value, {SU, SS, TU, TS}} ->
+ {SU, SS};
+ false ->
+ {U, S}
+ end
+ end
+ end,
+ {Uuid, Seq},
+ PeerNodes
+ )
+ end}
+ end.
+
+parse_local_docs(DbName) ->
+ fabric:all_docs(DbName, fun parse_local_docs_cb/2, {#{}, #{}}, all_docs_mrargs()).
+
+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),
+ 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,
+ {PeerCheckpoints, ShardSyncHistory0} = Acc,
+ 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}.
+
+range_to_nodes(Shards) ->
+ maps:groups_from_list(fun range/1, fun node/1, Shards).
+
+range(Shard) ->
+ Shard#shard.range.
+
+node(Shard) ->
+ Shard#shard.node.
+
+merge_peers(_Key, {Uuid, Val1}, {Uuid, Val2}) when is_integer(Val1), is_integer(Val2) ->
+ {Uuid, min(Val1, Val2)}.
+
+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">>}
+ ]
+ }.
(couchdb) 08/11: wip
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 19630d0cd63f2cc19a0f893787bbcf25f50e49b3
Author: Robert Newson <rn...@apache.org>
AuthorDate: Fri Feb 28 15:02:21 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 32 ++++++++++++++++++++------------
1 file changed, 20 insertions(+), 12 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index c3f260960..d0f474bb7 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -37,20 +37,28 @@ go(DbName) ->
ShardSyncHistory = parse_shard_sync_docs(LocalDocs),
Shards = mem3:shards(DbName),
- Foo = fun(#shard{} = Shard, Acc) ->
- case maps:find({Shard#shard.node, Shard#shard.range}, PeerCheckpoints) of
- {ok, Value} ->
- [{Shard, Value} | Acc];
- error ->
- %% cross ref with sync docs
- Pred = fun({_Node, Range}, _) -> Range == Shard#shard.range end,
- maps:filter(Pred, PeerCheckpoints),
- Acc
- end
+ ByRange = maps:groups_from_list(fun(Shard) ->Shard#shard.range end,
+ fun(Shard) ->
+ Shard#shard.node end, Shards),
+
+ Fun1 = fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
+ OtherNodes = maps:get(Range, ByRange, []) -- [SrcNode],
+ Fun2 = fun(TgtNode, Acc2) ->
+ History = maps:get({Range, SrcNode, TgtNode}, ShardSyncHistory, []),
+ case lists:search(fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
+ SourceSeq =< Seq end, History) of
+ {value, {SourceUuid, SourceSeq, TargetUuid, TargetSeq}} ->
+ Acc2#{{Range, TgtNode} => {TargetUuid, TargetSeq}};
+ false ->
+ Acc2
+ end
+ end,
+ lists:foldl(Fun2, Acc1, OtherNodes)
end,
- Workers = lists:foldl(Foo, [], Shards),
- {PeerCheckpoints, ShardSyncHistory, Workers}.
+ Expanded = maps:fold(Fun1, PeerCheckpoints, PeerCheckpoints),
+
+ {PeerCheckpoints, ShardSyncHistory, Expanded}.
all_docs_cb({row, Row}, Acc0) ->
case lists:keyfind(doc, 1, Row) of
(couchdb) 06/11: wip
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 d4ebb1c06e3681c3a368516e86c13b1d30b7b793
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Feb 24 15:27:52 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 35 ++++++++++++++++++++++-------------
1 file changed, 22 insertions(+), 13 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 8b2eacf4d..217fcaa3e 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -33,8 +33,8 @@ go(DbName) ->
% 2. use mem3_sync docs to find lowest update seq on any other {node,uuids}
% 3. issue rpc to matching shards and ask them to set drop seq if uuid matches
- PeerCheckpoints = lists:foldl(fun parse_peer_doc/2, #{}, LocalDocs),
- ShardSyncHistory = lists:foldl(fun parse_shard_sync_doc/2, [], LocalDocs),
+ PeerCheckpoints = parse_peer_docs(LocalDocs),
+ ShardSyncHistory = parse_shard_sync_docs(LocalDocs),
{PeerCheckpoints, ShardSyncHistory}.
all_docs_cb({row, Row}, Acc0) ->
@@ -49,6 +49,9 @@ all_docs_cb({row, Row}, Acc0) ->
all_docs_cb(_Else, Acc) ->
{ok, Acc}.
+parse_peer_docs(LocalDocs) ->
+ lists:foldl(fun parse_peer_doc/2, #{}, LocalDocs).
+
parse_peer_doc(#doc{} = Doc, Acc) ->
{Props} = Doc#doc.body,
Type = couch_util:get_value(<<"type">>, Props),
@@ -60,18 +63,24 @@ parse_peer_doc(#doc{} = Doc, Acc) ->
Acc
end.
+parse_shard_sync_docs(LocalDocs) ->
+ lists:foldl(fun parse_shard_sync_doc/2, #{}, LocalDocs).
+
parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
{Props} = Doc#doc.body,
{[{_Node, History}]} = couch_util:get_value(<<"history">>, Props),
- L = [
- {couch_util:get_value(<<"source_node">>, Item),
- couch_util:get_value(<<"source_uuid">>, Item),
- couch_util:get_value(<<"source_seq">>, Item),
- couch_util:get_value(<<"target_node">>, Item),
- couch_util:get_value(<<"target_uuid">>, Item),
- couch_util:get_value(<<"target_seq">>, Item)
- } || {Item} <- History],
- L ++ Acc;
+ KeyFun = fun({Item}) ->
+ {couch_util:get_value(<<"source_uuid">>, Item),
+ binary_to_existing_atom(couch_util:get_value(<<"target_node">>, Item))}
+ end,
+ ValueFun = fun({Item}) ->
+ {
+ couch_util:get_value(<<"target_uuid">>, Item),
+ couch_util:get_value(<<"source_seq">>, Item),
+ couch_util:get_value(<<"target_seq">>, Item)
+ }
+ end,
+ maps:merge(maps:groups_from_list(KeyFun, ValueFun, History), Acc);
parse_shard_sync_doc(_Doc, Acc) ->
Acc.
@@ -79,8 +88,8 @@ decode_seq(OpaqueSeq) ->
Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
lists:foldl(
fun
- ({_Node, _Range, {Seq, Uuid, Node}}, Acc) ->
- Acc#{{Node, Uuid} => Seq};
+ ({_Node, Range, {Seq, Uuid, Node}}, Acc) ->
+ Acc#{{Node, Range, Uuid} => Seq};
({_Node, _Range, _Seq}, Acc) ->
Acc
end,
(couchdb) 03/11: wip
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 9ab85a15bdaa10e468fa0b2df8906f223a7d72aa
Author: Robert Newson <rn...@apache.org>
AuthorDate: Fri Feb 21 16:48:57 2025 +0000
wip
---
.../src/couch_replicator_scheduler_job.erl | 12 +++++++--
src/fabric/src/fabric_drop_seq.erl | 31 +++++++++++++---------
2 files changed, 28 insertions(+), 15 deletions(-)
diff --git a/src/couch_replicator/src/couch_replicator_scheduler_job.erl b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
index 21f5360d9..62e604b5e 100644
--- a/src/couch_replicator/src/couch_replicator_scheduler_job.erl
+++ b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
@@ -854,8 +854,6 @@ do_checkpoint(State) ->
BaseHistory =
[
{<<"session_id">>, SessionId},
- {<<"source">>, ?l2b(SourceName)},
- {<<"target">>, ?l2b(TargetName)},
{<<"source_last_seq">>, NewSeq},
{<<"replication_id_version">>, ?REP_ID_VERSION}
] ++
@@ -888,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,
@@ -914,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_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 08562d815..4dd31339f 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -29,11 +29,13 @@ go(DbName) ->
]
},
{ok, LocalDocs} = fabric:all_docs(DbName, fun all_docs_cb/2, [], MrArgs),
-% 1. find lowest update seq per {node,uuid} pair in all checkpoints excluding mem3_sync
-% 2. use mem3_sync docs to find lowest update seq on any other {node,uuids}
-% 3. issue rpc to matching shards and ask them to set drop seq if uuid matches
+ % 1. find lowest update seq per {node,uuid} pair in all peer checkpoints excluding mem3_sync
+ % 2. use mem3_sync docs to find lowest update seq on any other {node,uuids}
+ % 3. issue rpc to matching shards and ask them to set drop seq if uuid matches
- lists:foldl(fun parse_replication_doc/2, [], LocalDocs).
+ PeerCheckpoints = lists:foldl(fun parse_peer_doc/2, #{}, LocalDocs),
+ ShardSyncHistory = lists:foldl(fun parse_shard_sync_doc/2, [], LocalDocs),
+ {PeerCheckpoints, ShardSyncHistory}.
all_docs_cb({row, Row}, Acc0) ->
Acc1 =
@@ -47,25 +49,26 @@ all_docs_cb({row, Row}, Acc0) ->
all_docs_cb(_Else, Acc) ->
{ok, Acc}.
-parse_replication_doc(#doc{} = Doc, Acc) ->
+parse_peer_doc(#doc{} = Doc, Acc) ->
{Props} = Doc#doc.body,
- ReplicationIdVersion = couch_util:get_value(<<"replication_id_version">>, Props),
- case ReplicationIdVersion of
- ?REP_ID_VERSION ->
- SourceLastSeq = couch_util:get_value(<<"source_last_seq">>, Props),
- [decode_seq(SourceLastSeq) | Acc];
+ Type = couch_util:get_value(<<"type">>, Props),
+ case Type of
+ <<"peer_checkpoint">> ->
+ UpdateSeq = couch_util:get_value(<<"update_seq">>, Props),
+ maps:merge_with(fun combine_peers/3, decode_seq(UpdateSeq), Acc);
_Else ->
Acc
end.
parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
{Props} = Doc#doc.body,
- History = couch_util:get_value(<<"history">>, Props),
- [History | Acc];
+ {History0} = couch_util:get_value(<<"history">>, Props),
+ {_, History1} = lists:unzip(History0),
+ History2 = [[maps:from_list(I) || {I} <- H] || H <- History1],
+ [History2 | Acc];
parse_shard_sync_doc(_Doc, Acc) ->
Acc.
-
decode_seq(OpaqueSeq) ->
Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
lists:foldl(
@@ -79,3 +82,5 @@ decode_seq(OpaqueSeq) ->
Decoded
).
+combine_peers(_Key, Val1, Val2) when is_integer(Val1), is_integer(Val2) ->
+ min(Val1, Val2).
(couchdb) 10/11: wip
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 7d75543b9a25fd0fee55a254989e4b313cf7a7a8
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Mar 3 20:56:15 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 139 +++++++++++++++------------------
src/fabric/src/fabric_drop_seq2.erl | 148 ------------------------------------
2 files changed, 62 insertions(+), 225 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 501837ffe..956386f40 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -1,72 +1,36 @@
-% 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_drop_seq).
+-module(fabric_drop_seq2).
-include_lib("mem3/include/mem3.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
--include_lib("couch_replicator/src/couch_replicator.hrl").
-export([go/1]).
go(DbName) ->
- MrArgs = #mrargs{
- view_type = map,
- include_docs = true,
- extra = [
- {include_system, true},
- {namespace, <<"_local">>}
- ]
- },
- {ok, LocalDocs} = fabric:all_docs(DbName, fun all_docs_cb/2, [], MrArgs),
- % 1. find lowest update seq per {node,uuid} pair in all peer checkpoints excluding mem3_sync
- % 2. use mem3_sync docs to find lowest update seq on any other {node,uuids}
- % 3. issue rpc to matching shards and ask them to set drop seq if uuid matches
-
- PeerCheckpoints = parse_peer_docs(LocalDocs),
- ShardSyncHistory = parse_shard_sync_docs(LocalDocs),
Shards = mem3:shards(DbName),
+ RangeToNodes = range_to_nodes(Shards),
- ByRange = maps:groups_from_list(
- fun(Shard) -> Shard#shard.range end,
- fun(Shard) ->
- Shard#shard.node
- end,
- Shards
- ),
+ {ok, {PeerCheckpoints, ShardSyncHistory}} = parse_local_docs(DbName),
- Fun1 = fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
- OtherNodes = maps:get(Range, ByRange, []) -- [SrcNode],
- Fun2 = fun(TgtNode, Acc2) ->
+ Expanded = maps:fold( fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
+ OtherNodes = maps:get(Range, RangeToNodes, []) -- [SrcNode],
+ lists:foldl(fun(TgtNode, Acc2) ->
History = maps:get({Range, SrcNode, TgtNode}, ShardSyncHistory, []),
case
lists:search(
- fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
+ fun({SourceUuid, SourceSeq, _TargetUuid, _TargetSeq}) ->
Uuid == SourceUuid andalso SourceSeq =< Seq
end,
History
)
of
- {value, {SourceUuid, SourceSeq, TargetUuid, TargetSeq}} ->
+ {value, {_SourceUuid, _SourceSeq, TargetUuid, TargetSeq}} ->
Acc2#{{Range, TgtNode} => {TargetUuid, TargetSeq}};
false ->
Acc2
end
- end,
- lists:foldl(Fun2, Acc1, OtherNodes)
- end,
-
- Expanded = maps:fold(Fun1, PeerCheckpoints, PeerCheckpoints),
+ end, Acc1, OtherNodes)
+ end, PeerCheckpoints, PeerCheckpoints),
Workers = lists:map(
fun(#shard{} = Shard) ->
@@ -75,44 +39,27 @@ go(DbName) ->
Shards
),
- {PeerCheckpoints, ShardSyncHistory, Expanded, Workers}.
+ Workers.
+
+parse_local_docs(DbName) ->
+ fabric:all_docs(DbName, fun parse_local_docs_cb/2, {#{}, #{}}, all_docs_mrargs()).
-all_docs_cb({row, Row}, Acc0) ->
+parse_local_docs_cb({row, Row}, Acc) ->
case lists:keyfind(doc, 1, Row) of
- {doc, Doc} ->
- {ok, [couch_doc:from_json_obj(Doc) | Acc0]};
false ->
- {ok, Acc0}
+ {ok, Acc};
+ {doc, Doc} ->
+ parse_local_doc(couch_doc:from_json_obj(Doc), Acc)
end;
-all_docs_cb(_Else, Acc) ->
+parse_local_docs_cb(_Else, Acc) ->
{ok, Acc}.
-parse_peer_docs(LocalDocs) ->
- lists:foldl(fun parse_peer_doc/2, #{}, LocalDocs).
-
-parse_peer_doc(#doc{} = Doc, Acc) ->
- {Props} = Doc#doc.body,
- Type = couch_util:get_value(<<"type">>, Props),
- case Type of
- <<"_peer_checkpoint">> ->
- UpdateSeq = couch_util:get_value(<<"update_seq">>, Props),
- maps:merge_with(fun combine_peers/3, decode_seq(UpdateSeq), Acc);
- _Else ->
- Acc
- end.
-
-combine_peers(_Key, {Uuid, Val1}, {Uuid, Val2}) when is_integer(Val1), is_integer(Val2) ->
- {Uuid, min(Val1, Val2)}.
-
-parse_shard_sync_docs(LocalDocs) ->
- lists:foldl(fun parse_shard_sync_doc/2, #{}, LocalDocs).
-
-parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, 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
- Acc;
+ {ok, Acc};
DbName ->
Range = mem3:range(DbName),
{[{_SrcNode, History}]} = couch_util:get_value(<<"history">>, Props),
@@ -128,10 +75,38 @@ parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) -
couch_util:get_value(<<"target_seq">>, Item)
}
end,
- maps:merge(maps:groups_from_list(KeyFun, ValueFun, History), Acc)
+ {PeerCheckpoints, ShardSyncHistory0} = Acc,
+ ShardSyncHistory1 = maps:merge(
+ maps:groups_from_list(KeyFun, ValueFun, History), ShardSyncHistory0
+ ),
+ {ok, {PeerCheckpoints, ShardSyncHistory1}}
end;
-parse_shard_sync_doc(_Doc, Acc) ->
- Acc.
+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}.
+
+range_to_nodes(Shards) ->
+ maps:groups_from_list(fun range/1, fun node/1, Shards).
+
+range(Shard) ->
+ Shard#shard.range.
+
+node(Shard) ->
+ Shard#shard.node.
+
+merge_peers(_Key, {Uuid, Val1}, {Uuid, Val2}) when is_integer(Val1), is_integer(Val2) ->
+ {Uuid, min(Val1, Val2)}.
decode_seq(OpaqueSeq) ->
Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
@@ -154,3 +129,13 @@ decode_seq(OpaqueSeq) ->
#{},
Decoded
).
+
+all_docs_mrargs() ->
+ #mrargs{
+ view_type = map,
+ include_docs = true,
+ extra = [
+ {include_system, true},
+ {namespace, <<"_local">>}
+ ]
+ }.
diff --git a/src/fabric/src/fabric_drop_seq2.erl b/src/fabric/src/fabric_drop_seq2.erl
deleted file mode 100644
index 13edfda37..000000000
--- a/src/fabric/src/fabric_drop_seq2.erl
+++ /dev/null
@@ -1,148 +0,0 @@
--module(fabric_drop_seq2).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
--include_lib("couch_mrview/include/couch_mrview.hrl").
-
--export([go/1]).
-
-go(DbName) ->
- Shards = mem3:shards(DbName),
- RangeToNodes = range_to_nodes(Shards),
-
- {ok, {PeerCheckpoints, ShardSyncHistory}} = parse_local_docs(DbName),
- F = lists:map(calculate_drop_seq(RangeToNodes, PeerCheckpoints, ShardSyncHistory), Shards),
- {PeerCheckpoints, F}.
-
-calculate_drop_seq(RangeToNodes, PeerCheckpoints, ShardSyncHistory) ->
- fun(#shard{node = Node, range = Range} = Shard) ->
- {Shard, case maps:find({Range, Node}, PeerCheckpoints) of
- error ->
- {<<>>, 0};
- {ok, {Uuid, Seq}} ->
- PeerNodes = maps:get(Range, RangeToNodes, []) -- [Node],
- lists:foldl(
- fun(PeerNode, {U, S}) ->
- case maps:find({Range, PeerNode, Node}, ShardSyncHistory) of
- error ->
- {U, S};
- {ok, History} ->
- case
- lists:search(
- fun({SourceUuid, SourceSeq, TargetUuid, TargetSeq}) ->
- TargetUuid == U andalso TargetSeq =< S
- end,
- History
- )
- of
- {value, {SU, SS, TU, TS}} ->
- {SU, SS};
- false ->
- {U, S}
- end
- end
- end,
- {Uuid, Seq},
- PeerNodes
- )
- end}
- end.
-
-parse_local_docs(DbName) ->
- fabric:all_docs(DbName, fun parse_local_docs_cb/2, {#{}, #{}}, all_docs_mrargs()).
-
-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),
- 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,
- {PeerCheckpoints, ShardSyncHistory0} = Acc,
- 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}.
-
-range_to_nodes(Shards) ->
- maps:groups_from_list(fun range/1, fun node/1, Shards).
-
-range(Shard) ->
- Shard#shard.range.
-
-node(Shard) ->
- Shard#shard.node.
-
-merge_peers(_Key, {Uuid, Val1}, {Uuid, Val2}) when is_integer(Val1), is_integer(Val2) ->
- {Uuid, min(Val1, Val2)}.
-
-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">>}
- ]
- }.
(couchdb) 05/11: remove uuid_prefix_len for drop_seq safety
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 67d9b0f5b0bf6c54c1586fe2bd56553891aba784
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Feb 24 13:58:42 2025 +0000
remove uuid_prefix_len for drop_seq safety
---
rel/overlay/etc/default.ini | 1 -
src/fabric/src/fabric_util.erl | 3 ++-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index de8664781..c4e22e34d 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -382,7 +382,6 @@ hash_algorithms = sha256, sha
;changes_duration =
;shard_timeout_factor = 2
;shard_timeout_min_msec = 100
-;uuid_prefix_len = 7
;request_timeout = 60000
;all_docs_timeout = 10000
;attachments_timeout = 60000
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index 4acb65c73..5d72cf9c7 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -393,7 +393,8 @@ worker_ranges(Workers) ->
lists:usort(Ranges).
get_uuid_prefix_len() ->
- config:get_integer("fabric", "uuid_prefix_len", 7).
+ %% no longer configurable for drop_seq safety.
+ 32.
% If we issue multiple fabric calls from the same process we have to isolate
% them so in case of error they don't pollute the processes dictionary or the
(couchdb) 07/11: wip
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 02e5dcb242014dfea93611c21535252a06399ad2
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Feb 26 15:17:55 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 73 ++++++++++++++++++++++++--------------
src/mem3/src/mem3_rpc.erl | 1 +
2 files changed, 48 insertions(+), 26 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 217fcaa3e..c3f260960 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -35,17 +35,30 @@ go(DbName) ->
PeerCheckpoints = parse_peer_docs(LocalDocs),
ShardSyncHistory = parse_shard_sync_docs(LocalDocs),
- {PeerCheckpoints, ShardSyncHistory}.
+ Shards = mem3:shards(DbName),
+
+ Foo = fun(#shard{} = Shard, Acc) ->
+ case maps:find({Shard#shard.node, Shard#shard.range}, PeerCheckpoints) of
+ {ok, Value} ->
+ [{Shard, Value} | Acc];
+ error ->
+ %% cross ref with sync docs
+ Pred = fun({_Node, Range}, _) -> Range == Shard#shard.range end,
+ maps:filter(Pred, PeerCheckpoints),
+ Acc
+ end
+ end,
+ Workers = lists:foldl(Foo, [], Shards),
+
+ {PeerCheckpoints, ShardSyncHistory, Workers}.
all_docs_cb({row, Row}, Acc0) ->
- Acc1 =
- case lists:keyfind(doc, 1, Row) of
- {doc, Doc} ->
- [couch_doc:from_json_obj(Doc) | Acc0];
- false ->
- Acc0
- end,
- {ok, Acc1};
+ case lists:keyfind(doc, 1, Row) of
+ {doc, Doc} ->
+ {ok, [couch_doc:from_json_obj(Doc) | Acc0]};
+ false ->
+ {ok, Acc0}
+ end;
all_docs_cb(_Else, Acc) ->
{ok, Acc}.
@@ -63,24 +76,35 @@ parse_peer_doc(#doc{} = Doc, Acc) ->
Acc
end.
+combine_peers(_Key, {Uuid, Val1}, {Uuid, Val2}) when is_integer(Val1), is_integer(Val2) ->
+ {Uuid, min(Val1, Val2)}.
+
parse_shard_sync_docs(LocalDocs) ->
lists:foldl(fun parse_shard_sync_doc/2, #{}, LocalDocs).
parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
{Props} = Doc#doc.body,
- {[{_Node, History}]} = couch_util:get_value(<<"history">>, Props),
- KeyFun = fun({Item}) ->
- {couch_util:get_value(<<"source_uuid">>, Item),
- binary_to_existing_atom(couch_util:get_value(<<"target_node">>, Item))}
- end,
- ValueFun = fun({Item}) ->
- {
- couch_util:get_value(<<"target_uuid">>, Item),
- couch_util:get_value(<<"source_seq">>, Item),
- couch_util:get_value(<<"target_seq">>, Item)
- }
- end,
- maps:merge(maps:groups_from_list(KeyFun, ValueFun, History), Acc);
+ case couch_util:get_value(<<"dbname">>, Props) of
+ undefined ->
+ %% not yet upgraded with new property
+ Acc;
+ DbName ->
+ Range = mem3:range(DbName),
+ {[{_SrcNode, History}]} = couch_util:get_value(<<"history">>, Props),
+ 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,
+ maps:merge(maps:groups_from_list(KeyFun, ValueFun, History), Acc)
+ end;
parse_shard_sync_doc(_Doc, Acc) ->
Acc.
@@ -89,13 +113,10 @@ decode_seq(OpaqueSeq) ->
lists:foldl(
fun
({_Node, Range, {Seq, Uuid, Node}}, Acc) ->
- Acc#{{Node, Range, Uuid} => Seq};
+ Acc#{{Range, Node} => {Uuid, Seq}};
({_Node, _Range, _Seq}, Acc) ->
Acc
end,
#{},
Decoded
).
-
-combine_peers(_Key, Val1, Val2) when is_integer(Val1), is_integer(Val2) ->
- min(Val1, Val2).
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)}
(couchdb) 04/11: wip
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 e8ca9123317ba5f5317d80230c96df300d886c39
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Feb 24 13:53:02 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 20 +++++++++++++-------
1 file changed, 13 insertions(+), 7 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 4dd31339f..8b2eacf4d 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -53,7 +53,7 @@ parse_peer_doc(#doc{} = Doc, Acc) ->
{Props} = Doc#doc.body,
Type = couch_util:get_value(<<"type">>, Props),
case Type of
- <<"peer_checkpoint">> ->
+ <<"_peer_checkpoint">> ->
UpdateSeq = couch_util:get_value(<<"update_seq">>, Props),
maps:merge_with(fun combine_peers/3, decode_seq(UpdateSeq), Acc);
_Else ->
@@ -62,10 +62,16 @@ parse_peer_doc(#doc{} = Doc, Acc) ->
parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
{Props} = Doc#doc.body,
- {History0} = couch_util:get_value(<<"history">>, Props),
- {_, History1} = lists:unzip(History0),
- History2 = [[maps:from_list(I) || {I} <- H] || H <- History1],
- [History2 | Acc];
+ {[{_Node, History}]} = couch_util:get_value(<<"history">>, Props),
+ L = [
+ {couch_util:get_value(<<"source_node">>, Item),
+ couch_util:get_value(<<"source_uuid">>, Item),
+ couch_util:get_value(<<"source_seq">>, Item),
+ couch_util:get_value(<<"target_node">>, Item),
+ couch_util:get_value(<<"target_uuid">>, Item),
+ couch_util:get_value(<<"target_seq">>, Item)
+ } || {Item} <- History],
+ L ++ Acc;
parse_shard_sync_doc(_Doc, Acc) ->
Acc.
@@ -73,8 +79,8 @@ decode_seq(OpaqueSeq) ->
Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
lists:foldl(
fun
- ({_Node, Range, {Seq, Uuid, Node}}, Acc) ->
- Acc#{{Node, Range, Uuid} => Seq};
+ ({_Node, _Range, {Seq, Uuid, Node}}, Acc) ->
+ Acc#{{Node, Uuid} => Seq};
({_Node, _Range, _Seq}, Acc) ->
Acc
end,
(couchdb) 02/11: wip
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 2124fe35ad6813ddac4bea3c5caeb72a57ea1abb
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Feb 4 13:14:08 2025 +0000
wip
---
src/couch/src/couch_bt_engine_compactor.erl | 5 +-
.../src/couch_replicator_scheduler_job.erl | 2 +
src/fabric/src/fabric.erl | 8 ++-
src/fabric/src/fabric_drop_seq.erl | 81 ++++++++++++++++++++++
4 files changed, 93 insertions(+), 3 deletions(-)
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 5e9289d71..619200bce 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -322,14 +322,15 @@ copy_compact(#comp_st{} = CompSt) ->
Deleted =
case DocInfo of
#full_doc_info{} -> DocInfo#full_doc_info.deleted;
- #doc_info{} -> false % Older versions stored #doc_info records in the seq_tree.
+ % 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}};
+ {ok, {AccNewSt, AccUncopied, AccUncopiedSize, AccCopiedSize}};
AccUncopiedSize2 >= BufferSize ->
NewSt2 = copy_docs(
St, AccNewSt, lists:reverse([DocInfo | AccUncopied]), Retry
diff --git a/src/couch_replicator/src/couch_replicator_scheduler_job.erl b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
index 544c5602a..21f5360d9 100644
--- a/src/couch_replicator/src/couch_replicator_scheduler_job.erl
+++ b/src/couch_replicator/src/couch_replicator_scheduler_job.erl
@@ -854,6 +854,8 @@ do_checkpoint(State) ->
BaseHistory =
[
{<<"session_id">>, SessionId},
+ {<<"source">>, ?l2b(SourceName)},
+ {<<"target">>, ?l2b(TargetName)},
{<<"source_last_seq">>, NewSeq},
{<<"replication_id_version">>, ?REP_ID_VERSION}
] ++
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..08562d815
--- /dev/null
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -0,0 +1,81 @@
+% 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_drop_seq).
+
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+-include_lib("couch_replicator/src/couch_replicator.hrl").
+
+-export([go/1]).
+
+go(DbName) ->
+ MrArgs = #mrargs{
+ view_type = map,
+ include_docs = true,
+ extra = [
+ {include_system, true},
+ {namespace, <<"_local">>}
+ ]
+ },
+ {ok, LocalDocs} = fabric:all_docs(DbName, fun all_docs_cb/2, [], MrArgs),
+% 1. find lowest update seq per {node,uuid} pair in all checkpoints excluding mem3_sync
+% 2. use mem3_sync docs to find lowest update seq on any other {node,uuids}
+% 3. issue rpc to matching shards and ask them to set drop seq if uuid matches
+
+ lists:foldl(fun parse_replication_doc/2, [], LocalDocs).
+
+all_docs_cb({row, Row}, Acc0) ->
+ Acc1 =
+ case lists:keyfind(doc, 1, Row) of
+ {doc, Doc} ->
+ [couch_doc:from_json_obj(Doc) | Acc0];
+ false ->
+ Acc0
+ end,
+ {ok, Acc1};
+all_docs_cb(_Else, Acc) ->
+ {ok, Acc}.
+
+parse_replication_doc(#doc{} = Doc, Acc) ->
+ {Props} = Doc#doc.body,
+ ReplicationIdVersion = couch_util:get_value(<<"replication_id_version">>, Props),
+ case ReplicationIdVersion of
+ ?REP_ID_VERSION ->
+ SourceLastSeq = couch_util:get_value(<<"source_last_seq">>, Props),
+ [decode_seq(SourceLastSeq) | Acc];
+ _Else ->
+ Acc
+ end.
+
+parse_shard_sync_doc(#doc{id = <<"_local/shard-sync-", _/binary>>} = Doc, Acc) ->
+ {Props} = Doc#doc.body,
+ History = couch_util:get_value(<<"history">>, Props),
+ [History | Acc];
+parse_shard_sync_doc(_Doc, Acc) ->
+ Acc.
+
+
+decode_seq(OpaqueSeq) ->
+ Decoded = fabric_view_changes:decode_seq(OpaqueSeq),
+ lists:foldl(
+ fun
+ ({_Node, Range, {Seq, Uuid, Node}}, Acc) ->
+ Acc#{{Node, Range, Uuid} => Seq};
+ ({_Node, _Range, _Seq}, Acc) ->
+ Acc
+ end,
+ #{},
+ Decoded
+ ).
+
(couchdb) 11/11: wip
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 59b838eebaf42288a3ffcb4f93ab721c60148ed3
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Mar 3 20:57:51 2025 +0000
wip
---
src/fabric/src/fabric_drop_seq.erl | 48 ++++++++++++++++++++++----------------
1 file changed, 28 insertions(+), 20 deletions(-)
diff --git a/src/fabric/src/fabric_drop_seq.erl b/src/fabric/src/fabric_drop_seq.erl
index 956386f40..98c8b4542 100644
--- a/src/fabric/src/fabric_drop_seq.erl
+++ b/src/fabric/src/fabric_drop_seq.erl
@@ -1,4 +1,4 @@
--module(fabric_drop_seq2).
+-module(fabric_drop_seq).
-include_lib("mem3/include/mem3.hrl").
-include_lib("couch/include/couch_db.hrl").
@@ -12,25 +12,33 @@ go(DbName) ->
{ok, {PeerCheckpoints, ShardSyncHistory}} = parse_local_docs(DbName),
- Expanded = maps:fold( fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
- OtherNodes = maps:get(Range, RangeToNodes, []) -- [SrcNode],
- lists:foldl(fun(TgtNode, Acc2) ->
- History = maps:get({Range, SrcNode, TgtNode}, ShardSyncHistory, []),
- case
- lists:search(
- fun({SourceUuid, SourceSeq, _TargetUuid, _TargetSeq}) ->
- Uuid == SourceUuid andalso SourceSeq =< Seq
- end,
- History
- )
- of
- {value, {_SourceUuid, _SourceSeq, TargetUuid, TargetSeq}} ->
- Acc2#{{Range, TgtNode} => {TargetUuid, TargetSeq}};
- false ->
- Acc2
- end
- end, Acc1, OtherNodes)
- end, PeerCheckpoints, PeerCheckpoints),
+ Expanded = maps:fold(
+ fun({Range, SrcNode}, {Uuid, Seq}, Acc1) ->
+ OtherNodes = maps:get(Range, RangeToNodes, []) -- [SrcNode],
+ lists:foldl(
+ fun(TgtNode, Acc2) ->
+ History = maps:get({Range, SrcNode, TgtNode}, ShardSyncHistory, []),
+ case
+ lists:search(
+ fun({SourceUuid, SourceSeq, _TargetUuid, _TargetSeq}) ->
+ Uuid == SourceUuid andalso SourceSeq =< Seq
+ end,
+ History
+ )
+ of
+ {value, {_SourceUuid, _SourceSeq, TargetUuid, TargetSeq}} ->
+ Acc2#{{Range, TgtNode} => {TargetUuid, TargetSeq}};
+ false ->
+ Acc2
+ end
+ end,
+ Acc1,
+ OtherNodes
+ )
+ end,
+ PeerCheckpoints,
+ PeerCheckpoints
+ ),
Workers = lists:map(
fun(#shard{} = Shard) ->