You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2023/06/30 19:53:48 UTC
[couchdb] 01/01: Fix undefined range in mem3_rep purge replication logic
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch undefined-in-mem3-rep
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit ad4a7dd024d3e30cdd1b3ef3709c7598cfc1bcd9
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Fri Jun 30 15:35:00 2023 -0400
Fix undefined range in mem3_rep purge replication logic
When replication jobs are started with just shard names (paths), mem3_rep
creates `#shard{}` records from those and fills in some fields like `name` and
`node`. However we had left the `range` field as `undefined`. Until a recent
fix to avoid duplicating document purges [1], we didn't care about ranges, so
it was just untidy but not a bit problem. Since the fix, we rely on having a
proper ranges and would get function or case clause errors during purge infos
pulling and pushing operations.
The fix is to use the `mem3:range/1` function to obtain the range based on the
shard path if we can. In some cases we may replicate node local databases, so
we still have to account for the cases that ranges may be `undefined`.
In addition to the fix, improve the tests coverage by testing replication of
purges and local endpoints. Also, take the opportunity to clean up the
mem3_rep_tests suite to use the newer ?TDEF_FE macro.
[1] https://github.com/apache/couchdb/pull/4626
---
src/mem3/src/mem3_rep.erl | 50 +++++++-
src/mem3/test/eunit/mem3_rep_test.erl | 230 ++++++++++++++++++++++------------
2 files changed, 195 insertions(+), 85 deletions(-)
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
index 3cc381494..b452fd2fa 100644
--- a/src/mem3/src/mem3_rep.erl
+++ b/src/mem3/src/mem3_rep.erl
@@ -59,7 +59,9 @@ go(Source, Target) ->
go(DbName, Node, Opts) when is_binary(DbName), is_atom(Node) ->
go(#shard{name = DbName, node = node()}, #shard{name = DbName, node = Node}, Opts);
-go(#shard{} = Source, #shard{} = Target, Opts) ->
+go(#shard{} = Source0, #shard{} = Target0, Opts) ->
+ Source = add_range(Source0),
+ Target = add_range(Target0),
case mem3:db_is_current(Source) of
true ->
go(Source, targets_map(Source, Target), Opts);
@@ -366,7 +368,15 @@ pull_purges(Db, Count, #shard{} = SrcShard, #tgt{} = Tgt0, HashFun) ->
% out using the same pickfun which we use when picking documents
#shard{range = SrcRange} = SrcShard,
BelongsFun = fun({_UUID, Id, _Revs}) when is_binary(Id) ->
- mem3_reshard_job:pickfun(Id, [SrcRange], HashFun) =:= SrcRange
+ case SrcRange of
+ [B, E] when is_integer(B), is_integer(E) ->
+ mem3_reshard_job:pickfun(Id, [SrcRange], HashFun) =:= SrcRange;
+ undefined ->
+ % We may replicate node-local databases
+ % which are not associated with a shard range. In that case
+ % range will be undefined.
+ true
+ end
end,
Infos1 = lists:filter(BelongsFun, Infos),
{ok, _} = couch_db:purge_docs(Db, Infos1, [?REPLICATED_CHANGES]),
@@ -418,7 +428,14 @@ push_purges(Db, BatchSize, SrcShard, Tgt, HashFun) ->
erlang:max(0, Oldest - 1)
end,
BelongsFun = fun(Id) when is_binary(Id) ->
- mem3_reshard_job:pickfun(Id, [TgtRange], HashFun) =:= TgtRange
+ case TgtRange of
+ [B, E] when is_integer(B), is_integer(E) ->
+ mem3_reshard_job:pickfun(Id, [TgtRange], HashFun) =:= TgtRange;
+ undefined ->
+ % We may replicate node-local databases which are not associated
+ % with a shard range. In that case range will be undefined.
+ true
+ end
end,
FoldFun = fun({PSeq, UUID, Id, Revs}, {Count, Infos, _}) ->
case BelongsFun(Id) of
@@ -875,10 +892,37 @@ reset_remaining(#{} = Targets) ->
Targets
).
+add_range(#shard{name = DbName} = Shard) when is_binary(DbName) ->
+ case DbName of
+ <<"shards/", _Start:8/binary, "-", _End:8/binary, "/", _/binary>> ->
+ Shard#shard{range = mem3:range(DbName)};
+ <<_/binary>> ->
+ % We may replicate local dbs which do not have a shard range.
+ Shard
+ end.
+
-ifdef(TEST).
-include_lib("couch/include/couch_eunit.hrl").
+name_node_to_shard_local_db_test() ->
+ DbName = <<"foo">>,
+ Node = 'n1@bar.net',
+ Shard = add_range(#shard{name = DbName, node = Node}),
+ ?assertMatch(#shard{}, Shard),
+ ?assertEqual(DbName, Shard#shard.name),
+ ?assertEqual(Node, Shard#shard.node),
+ ?assertEqual(undefined, Shard#shard.range).
+
+name_node_to_shard_local_shard_test() ->
+ DbName = <<"shards/00000000-7fffffff/db.1687450595">>,
+ Node = 'n2@baz.org',
+ Shard = add_range(#shard{name = DbName, node = Node}),
+ ?assertMatch(#shard{}, Shard),
+ ?assertEqual(DbName, Shard#shard.name),
+ ?assertEqual(Node, Shard#shard.node),
+ ?assertEqual([0, 2147483647], Shard#shard.range).
+
find_source_seq_int_test_() ->
{
setup,
diff --git a/src/mem3/test/eunit/mem3_rep_test.erl b/src/mem3/test/eunit/mem3_rep_test.erl
index 31a6d9b77..6da1621d2 100644
--- a/src/mem3/test/eunit/mem3_rep_test.erl
+++ b/src/mem3/test/eunit/mem3_rep_test.erl
@@ -24,15 +24,29 @@
setup() ->
{AllSrc, AllTgt} = {?tempdb(), ?tempdb()},
{PartSrc, PartTgt} = {?tempdb(), ?tempdb()},
+ Localdb = ?tempdb(),
create_db(AllSrc, [{q, 1}, {n, 1}]),
create_db(AllTgt, [{q, 2}, {n, 1}]),
PartProps = [{partitioned, true}, {hash, [couch_partition, hash, []]}],
create_db(PartSrc, [{q, 1}, {n, 1}, {props, PartProps}]),
create_db(PartTgt, [{q, 2}, {n, 1}, {props, PartProps}]),
- #{allsrc => AllSrc, alltgt => AllTgt, partsrc => PartSrc, parttgt => PartTgt}.
+ create_local_db(Localdb),
+ #{
+ allsrc => AllSrc,
+ alltgt => AllTgt,
+ partsrc => PartSrc,
+ parttgt => PartTgt,
+ localdb => Localdb
+ }.
teardown(#{} = Dbs) ->
- maps:map(fun(_, Db) -> delete_db(Db) end, Dbs).
+ maps:map(
+ fun
+ (localdb, Db) -> delete_local_db(Db);
+ (_, Db) -> delete_db(Db)
+ end,
+ Dbs
+ ).
start_couch() ->
test_util:start_couch([mem3, fabric]).
@@ -52,102 +66,134 @@ mem3_reshard_db_test_() ->
fun setup/0,
fun teardown/1,
[
- fun replicate_basics/1,
- fun replicate_small_batches/1,
- fun replicate_low_batch_count/1,
- fun replicate_with_partitions/1
+ ?TDEF_FE(replicate_basics, ?TIMEOUT),
+ ?TDEF_FE(replicate_small_batches, ?TIMEOUT),
+ ?TDEF_FE(replicate_low_batch_count, ?TIMEOUT),
+ ?TDEF_FE(replicate_with_partitions, ?TIMEOUT),
+ ?TDEF_FE(replicate_to_and_from_local, ?TIMEOUT),
+ ?TDEF_FE(replicate_with_purges, ?TIMEOUT)
]
}
}
}.
replicate_basics(#{allsrc := AllSrc, alltgt := AllTgt}) ->
- {timeout, ?TIMEOUT,
- ?_test(begin
- DocSpec = #{docs => 10, delete => [5, 9]},
- add_test_docs(AllSrc, DocSpec),
- SDocs = get_all_docs(AllSrc),
-
- [Src] = lists:sort(mem3:local_shards(AllSrc)),
- [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
- #shard{range = R1} = Tgt1,
- #shard{range = R2} = Tgt2,
- TMap = #{R1 => Tgt1, R2 => Tgt2},
- Opts = [{batch_size, 1000}, {batch_count, all}],
- ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
-
- ?assertEqual(SDocs, get_all_docs(AllTgt))
- end)}.
+ DocSpec = #{docs => 10, delete => [5, 9]},
+ add_test_docs(AllSrc, DocSpec),
+ SDocs = get_all_docs(AllSrc),
+
+ [Src] = lists:sort(mem3:local_shards(AllSrc)),
+ [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
+ #shard{range = R1} = Tgt1,
+ #shard{range = R2} = Tgt2,
+ TMap = #{R1 => Tgt1, R2 => Tgt2},
+ Opts = [{batch_size, 1000}, {batch_count, all}],
+ ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
+
+ ?assertEqual(SDocs, get_all_docs(AllTgt)).
replicate_small_batches(#{allsrc := AllSrc, alltgt := AllTgt}) ->
- {timeout, ?TIMEOUT,
- ?_test(begin
- DocSpec = #{docs => 10, delete => [5, 9]},
- add_test_docs(AllSrc, DocSpec),
- SDocs = get_all_docs(AllSrc),
-
- [Src] = lists:sort(mem3:local_shards(AllSrc)),
- [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
- #shard{range = R1} = Tgt1,
- #shard{range = R2} = Tgt2,
- TMap = #{R1 => Tgt1, R2 => Tgt2},
- Opts = [{batch_size, 2}, {batch_count, all}],
- ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
-
- ?assertEqual(SDocs, get_all_docs(AllTgt))
- end)}.
+ DocSpec = #{docs => 10, delete => [5, 9]},
+ add_test_docs(AllSrc, DocSpec),
+ SDocs = get_all_docs(AllSrc),
+
+ [Src] = lists:sort(mem3:local_shards(AllSrc)),
+ [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
+ #shard{range = R1} = Tgt1,
+ #shard{range = R2} = Tgt2,
+ TMap = #{R1 => Tgt1, R2 => Tgt2},
+ Opts = [{batch_size, 2}, {batch_count, all}],
+ ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
+
+ ?assertEqual(SDocs, get_all_docs(AllTgt)).
replicate_low_batch_count(#{allsrc := AllSrc, alltgt := AllTgt}) ->
- {timeout, ?TIMEOUT,
- ?_test(begin
- DocSpec = #{docs => 10, delete => [5, 9]},
- add_test_docs(AllSrc, DocSpec),
- SDocs = get_all_docs(AllSrc),
+ DocSpec = #{docs => 10, delete => [5, 9]},
+ add_test_docs(AllSrc, DocSpec),
+ SDocs = get_all_docs(AllSrc),
- [Src] = lists:sort(mem3:local_shards(AllSrc)),
- [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
- #shard{range = R1} = Tgt1,
- #shard{range = R2} = Tgt2,
- TMap = #{R1 => Tgt1, R2 => Tgt2},
+ [Src] = lists:sort(mem3:local_shards(AllSrc)),
+ [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
+ #shard{range = R1} = Tgt1,
+ #shard{range = R2} = Tgt2,
+ TMap = #{R1 => Tgt1, R2 => Tgt2},
- Opts1 = [{batch_size, 2}, {batch_count, 1}],
- ?assertMatch({ok, 8}, mem3_rep:go(Src, TMap, Opts1)),
+ Opts1 = [{batch_size, 2}, {batch_count, 1}],
+ ?assertMatch({ok, 8}, mem3_rep:go(Src, TMap, Opts1)),
- Opts2 = [{batch_size, 1}, {batch_count, 2}],
- ?assertMatch({ok, 6}, mem3_rep:go(Src, TMap, Opts2)),
+ Opts2 = [{batch_size, 1}, {batch_count, 2}],
+ ?assertMatch({ok, 6}, mem3_rep:go(Src, TMap, Opts2)),
- Opts3 = [{batch_size, 1000}, {batch_count, all}],
- ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts3)),
+ Opts3 = [{batch_size, 1000}, {batch_count, all}],
+ ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts3)),
- ?assertEqual(SDocs, get_all_docs(AllTgt))
- end)}.
+ ?assertEqual(SDocs, get_all_docs(AllTgt)).
replicate_with_partitions(#{partsrc := PartSrc, parttgt := PartTgt}) ->
- {timeout, ?TIMEOUT,
- ?_test(begin
- DocSpec = #{
- pdocs => #{
- <<"PX">> => 15,
- <<"PY">> => 19
- }
- },
- add_test_docs(PartSrc, DocSpec),
- SDocs = get_all_docs(PartSrc),
- PXSrc = get_partition_info(PartSrc, <<"PX">>),
- PYSrc = get_partition_info(PartSrc, <<"PY">>),
-
- [Src] = lists:sort(mem3:local_shards(PartSrc)),
- [Tgt1, Tgt2] = lists:sort(mem3:local_shards(PartTgt)),
- #shard{range = R1} = Tgt1,
- #shard{range = R2} = Tgt2,
- TMap = #{R1 => Tgt1, R2 => Tgt2},
- Opts = [{batch_size, 1000}, {batch_count, all}],
- ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
-
- ?assertEqual(PXSrc, get_partition_info(PartTgt, <<"PX">>)),
- ?assertEqual(PYSrc, get_partition_info(PartTgt, <<"PY">>)),
- ?assertEqual(SDocs, get_all_docs(PartTgt))
- end)}.
+ DocSpec = #{
+ pdocs => #{
+ <<"PX">> => 15,
+ <<"PY">> => 19
+ }
+ },
+ add_test_docs(PartSrc, DocSpec),
+ SDocs = get_all_docs(PartSrc),
+ PXSrc = get_partition_info(PartSrc, <<"PX">>),
+ PYSrc = get_partition_info(PartSrc, <<"PY">>),
+
+ [Src] = lists:sort(mem3:local_shards(PartSrc)),
+ [Tgt1, Tgt2] = lists:sort(mem3:local_shards(PartTgt)),
+ #shard{range = R1} = Tgt1,
+ #shard{range = R2} = Tgt2,
+ TMap = #{R1 => Tgt1, R2 => Tgt2},
+ Opts = [{batch_size, 1000}, {batch_count, all}],
+ ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
+
+ ?assertEqual(PXSrc, get_partition_info(PartTgt, <<"PX">>)),
+ ?assertEqual(PYSrc, get_partition_info(PartTgt, <<"PY">>)),
+ ?assertEqual(SDocs, get_all_docs(PartTgt)).
+
+replicate_with_purges(#{allsrc := AllSrc, alltgt := AllTgt}) ->
+ DocSpec = #{docs => 10, delete => [5, 9], purge => [2, 4]},
+ add_test_docs(AllSrc, DocSpec),
+ % Add and purge some docs on target to excercise the pull_purges
+ % code path
+ add_test_docs(AllTgt, #{docs => 3, purge => [0, 2]}),
+
+ [Src] = lists:sort(mem3:local_shards(AllSrc)),
+ [Tgt1, Tgt2] = lists:sort(mem3:local_shards(AllTgt)),
+ #shard{range = R1} = Tgt1,
+ #shard{range = R2} = Tgt2,
+ TMap = #{R1 => Tgt1, R2 => Tgt2},
+ Opts = [{batch_size, 1000}, {batch_count, all}],
+ ?assertMatch({ok, 0}, mem3_rep:go(Src, TMap, Opts)),
+
+ SDocs = get_all_docs(AllSrc),
+ % Purges from the target should have been pulled and removed docs 0,1,2
+ % Source should have no live docs
+ ?assertEqual(#{}, SDocs),
+ ?assertEqual(#{}, get_all_docs(AllTgt)).
+
+replicate_to_and_from_local(#{localdb := LocalDb, allsrc := ClusteredDb}) ->
+ % We'll just tests that we can pull purges from the target
+ add_test_docs(ClusteredDb, #{docs => 6, purge => [0, 2]}),
+
+ [#shard{name = TgtDbName}] = mem3:local_shards(ClusteredDb),
+ Opts = [{batch_size, 1000}, {batch_count, all}],
+ Src1 = #shard{name = LocalDb, node = node()},
+ Tgt1 = #shard{name = TgtDbName, node = node()},
+ ?assertMatch({ok, 0}, mem3_rep:go(Src1, Tgt1, Opts)),
+
+ % Purge a few more docs ClusteredDb
+ add_test_docs(ClusteredDb, #{purge => [3, 4]}),
+
+ % Replicate the other way from clustered to source
+ Src2 = #shard{name = TgtDbName, node = node()},
+ Tgt2 = #shard{name = LocalDb, node = node()},
+ ?assertMatch({ok, 0}, mem3_rep:go(Src2, Tgt2, Opts)),
+ SDocs = get_all_docs(ClusteredDb),
+ ?assertEqual(1, map_size(SDocs)),
+ ?assertMatch(#{<<"00005">> := #{}}, SDocs).
get_partition_info(DbName, Partition) ->
with_proc(fun() ->
@@ -197,6 +243,13 @@ delete_db(DbName) ->
GL = erlang:group_leader(),
with_proc(fun() -> fabric:delete_db(DbName, [?ADMIN_CTX]) end, GL).
+create_local_db(DbName) ->
+ {ok, _} = couch_server:create(DbName, []),
+ ok.
+
+delete_local_db(DbName) ->
+ couch_server:delete(DbName, []).
+
with_proc(Fun) ->
with_proc(Fun, undefined, 30000).
@@ -237,7 +290,7 @@ add_test_docs(DbName, #{} = DocSpec) ->
[] -> ok;
[_ | _] = Deleted -> update_docs(DbName, Deleted)
end,
- ok.
+ purge_docs(DbName, maps:get(purge, DocSpec, [])).
update_docs(DbName, Docs) ->
with_proc(fun() ->
@@ -261,6 +314,19 @@ delete_docs([S, E], Docs) when E >= S ->
delete_docs(_, _) ->
[].
+purge_docs(DbName, [S, E]) when E >= S ->
+ Ids = [doc_id(<<"">>, I) || I <- lists:seq(S, E)],
+ IdRevs = [{Id, get_revs(DbName, Id)} || Id <- Ids],
+ {ok, _} = fabric:purge_docs(DbName, IdRevs, []),
+ ok;
+purge_docs(_DbName, []) ->
+ ok.
+
+get_revs(DbName, DocId) ->
+ FDI = fabric:get_full_doc_info(DbName, DocId, []),
+ #doc_info{revs = Revs} = couch_doc:to_doc_info(FDI),
+ [Rev#rev_info.rev || Rev <- Revs].
+
pdocs(#{} = PMap) ->
maps:fold(
fun(Part, DocSpec, DocsAcc) ->