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:47 UTC
(couchdb) 10/11: wip
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">>}
- ]
- }.