This is an automated email from the ASF dual-hosted git repository. nickva pushed a commit to branch fix-groups-by-range-in-fabric-doc-update in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit bb78d62688a5300db5c03bfd978b259e5a62164c Author: Nick Vatamaniuc <[email protected]> AuthorDate: Wed Jun 10 17:29:07 2026 -0400 Improve document grouping in fabric_doc_update The idea of doc/shard grouping and rotation was that different databases would start their workers on different copies, spread those uniformly and consistently by `{Range, DbName}`. However, the order of entries were actually dependent on `dict:to_list/1` internals. So, if `dict` was changed to return sorted items, like for instance maps do, we'd end up suddently breaking the supposedly randomized spread (I found this out by actually trying to use a map there instead of a dict). To fix this, make the rotation explicit. Group the entries first, then sort each group, and then rotate by `{Range, DbName}`. While at it make it so we don't have to duplicate the main rotate function in tests and use the actual function we intend to test. Add some explicit tests for this behavior. One downside is those tests rely on the exact rotate_list/2 implementation (`crc32(?term_to_binary/1`) so if that changes we'll know and have to update the test. --- src/fabric/src/fabric_doc_update.erl | 119 +++++++++++++++++++++++++++-------- 1 file changed, 92 insertions(+), 27 deletions(-) diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl index 96f7c1b11..81abb0f48 100644 --- a/src/fabric/src/fabric_doc_update.erl +++ b/src/fabric/src/fabric_doc_update.erl @@ -339,29 +339,34 @@ good_reply(_, _) -> -spec group_docs_by_shard(binary(), [#doc{}]) -> [{#shard{}, [#doc{}]}]. group_docs_by_shard(DbName, Docs) -> + Grouped = group_docs([{Doc, mem3:shards(DbName, Id)} || #doc{id = Id} = Doc <- Docs]), + rotate_ranges(DbName, Grouped). + +group_docs(DocsShards) -> dict:to_list( lists:foldl( - fun(#doc{id = Id} = Doc, D0) -> + fun({Doc, Shards}, D0) -> lists:foldl( fun(Shard, D1) -> dict:append(Shard, Doc, D1) end, D0, - begin - case mem3:shards(DbName, Id) of - [] -> - []; - [#shard{range = Range} | _] = Shards -> - mem3_util:rotate_list({Range, DbName}, Shards) - end - end + Shards ) end, dict:new(), - Docs + DocsShards ) ). +% Deterministically rotate workers in each range. When serialized workers start +% the first entry of each range first. +rotate_ranges(DbName, Grouped) -> + % orddict makes it easy to append to a value + FoldF = fun({#shard{range = R}, _} = E, D) -> orddict:append(R, E, D) end, + ByRange = orddict:to_list(lists:foldl(FoldF, orddict:new(), Grouped)), + lists:append([mem3_util:rotate_list({R, DbName}, lists:sort(G)) || {R, G} <- ByRange]). + append_update_replies([], [], DocReplyDict) -> DocReplyDict; append_update_replies([Doc | Rest], [], Dict0) -> @@ -509,7 +514,11 @@ doc_update_test_() -> fun parallel_in_flight_after_conflict/0, fun serial_filters_conflicts_at_cast/0, fun sws_false_mode_conflict_not_final/0, - fun sws_false_mode_ok_can_outvote_conflict/0 + fun sws_false_mode_ok_can_outvote_conflict/0, + fun group_docs_content_and_order/0, + fun rotate_ranges_rotates_each_range/0, + fun rotate_ranges_varies_by_db/0, + fun rotate_ranges_preserves_entries/0 ] }. @@ -1144,22 +1153,78 @@ sws_false_mode_ok_can_outvote_conflict() -> lists:sort(Reply) ). -% needed for testing to avoid having to start the mem3 application -group_docs_by_shard_hack(_DbName, Shards, Docs) -> - dict:to_list( - lists:foldl( - fun(#doc{id = _Id} = Doc, D0) -> - lists:foldl( - fun(Shard, D1) -> - dict:append(Shard, Doc, D1) - end, - D0, - Shards - ) - end, - dict:new(), - Docs - ) +group_docs_content_and_order() -> + S1 = mk_shard(<<"r1">>, n1, [0, 10]), + S2 = mk_shard(<<"r1">>, n2, [0, 10]), + DocA = #doc{id = <<"a">>}, + DocB = #doc{id = <<"b">>}, + DocC = #doc{id = <<"c">>}, + Grouped = group_docs([ + {DocA, [S1, S2]}, + {DocB, [S1]}, + {DocC, [S2, S1]} + ]), + ?assertEqual(2, length(Grouped)), + % Every shard gets exactly its docs in original batch order + ?assertEqual([DocA, DocB, DocC], couch_util:get_value(S1, Grouped)), + ?assertEqual([DocA, DocC], couch_util:get_value(S2, Grouped)). + +rotate_ranges_rotates_each_range() -> + {[S11, S12, S13, S21, S22, S23], Entries} = rotate_fixture(), + % each range's group is sorted, rotated by {Range, DbName}, and its first + % entry is the copy start_workers/1 will start first + ?assertEqual( + [ + {S13, docs13}, + {S11, docs11}, + {S12, docs12}, + {S22, docs22}, + {S23, docs23}, + {S21, docs21} + ], + rotate_ranges(<<"dba">>, Entries) ). +rotate_ranges_varies_by_db() -> + {[_, _, S13, S21, S22, _], Entries} = rotate_fixture(), + HeadsFor = fun(DbName) -> + Rotated = rotate_ranges(DbName, Entries), + Ranges = lists:usort([R || {#shard{range = R}, _} <- Rotated]), + [hd([S || {#shard{range = R1} = S, _} <- Rotated, R1 =:= R]) || R <- Ranges] + end, + % different db names => different order + ?assertEqual([S13, S22], HeadsFor(<<"dba">>)), + ?assertEqual([S13, S21], HeadsFor(<<"dbb">>)), + ?assertNotEqual(HeadsFor(<<"dba">>), HeadsFor(<<"dbb">>)). + +rotate_ranges_preserves_entries() -> + {_, Entries} = rotate_fixture(), + ?assertEqual(lists:sort(Entries), lists:sort(rotate_ranges(<<"dba">>, Entries))), + ?assertEqual(lists:sort(Entries), lists:sort(rotate_ranges(<<"dbb">>, Entries))). + +rotate_fixture() -> + S11 = mk_shard(<<"r1">>, n1, [0, 10]), + S12 = mk_shard(<<"r1">>, n2, [0, 10]), + S13 = mk_shard(<<"r1">>, n3, [0, 10]), + S21 = mk_shard(<<"r2">>, n1, [11, 20]), + S22 = mk_shard(<<"r2">>, n2, [11, 20]), + S23 = mk_shard(<<"r2">>, n3, [11, 20]), + % In a jumbled order to see how rotate_ranges will sort it (or not) + Entries = [ + {S22, docs22}, + {S11, docs11}, + {S13, docs13}, + {S21, docs21}, + {S12, docs12}, + {S23, docs23} + ], + {[S11, S12, S13, S21, S22, S23], Entries}. + +mk_shard(Name, Node, Range) -> + #shard{name = Name, node = Node, range = Range}. + +% needed for testing to avoid having to start the mem3 application +group_docs_by_shard_hack(DbName, Shards, Docs) -> + rotate_ranges(DbName, group_docs([{Doc, Shards} || Doc <- Docs])). + -endif.
