[couchdb] branch handle_all_dbs_active_in_fabric_doc_update created (now 6fac7cc63)

[va...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

vatamane pushed a change to branch handle_all_dbs_active_in_fabric_doc_update
in repository https://gitbox.apache.org/repos/asf/couchdb.git


      at 6fac7cc63 Handle all_dbs_active in fabric_doc_update

This branch includes the following new commits:

     new 6fac7cc63 Handle all_dbs_active in fabric_doc_update

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.

[couchdb] 01/01: Handle all_dbs_active in fabric_doc_update

[va...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

vatamane pushed a commit to branch handle_all_dbs_active_in_fabric_doc_update
in repository https://gitbox.apache.org/repos/asf/couchdb.git

commit 6fac7cc63291aeba9edac80c0c40162944d6b6e9
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Jun 6 15:22:14 2022 -0400

    Handle all_dbs_active in fabric_doc_update
    
    Previously fabric_doc_update message handler crashed with a function_clause
    error the first time it encountered an all_dbs_active error.
    
    Opt to handle it as a `rexi_EXIT` with a hope at that some workers would still
    return a valid result.
---
 src/fabric/src/fabric_doc_update.erl | 66 ++++++++++++++++++++++++++++++++++++
 1 file changed, 66 insertions(+)

diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl
index 69ad03da5..9de9c6580 100644
--- a/src/fabric/src/fabric_doc_update.erl
+++ b/src/fabric/src/fabric_doc_update.erl
@@ -66,6 +66,11 @@ handle_message({rexi_EXIT, _}, Worker, Acc0) ->
     {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0,
     NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs),
     skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict});
+handle_message({error, all_dbs_active}, Worker, Acc0) ->
+    % treat it like rexi_EXIT, the hope at least one copy will return successfully
+    {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0,
+    NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs),
+    skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict});
 handle_message(internal_server_error, Worker, Acc0) ->
     % happens when we fail to load validation functions in an RPC worker
     {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0,
@@ -329,6 +334,8 @@ doc_update_test_() ->
             fun doc_update1/0,
             fun doc_update2/0,
             fun doc_update3/0,
+            fun handle_all_dbs_active/0,
+            fun handle_two_all_dbs_actives/0,
             fun one_forbid/0,
             fun two_forbid/0,
             fun extend_tree_forbid/0,
@@ -467,6 +474,65 @@ doc_update3() ->
 
     ?assertEqual({ok, [{Doc1, {ok, Doc1}}, {Doc2, {ok, Doc2}}]}, Reply).
 
+handle_all_dbs_active() ->
+    Doc1 = #doc{revs = {1, [<<"foo">>]}},
+    Doc2 = #doc{revs = {1, [<<"bar">>]}},
+    Docs = [Doc1, Doc2],
+    Shards =
+        mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]),
+    GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs),
+    Acc0 = {
+        length(Shards),
+        length(Docs),
+        list_to_integer("2"),
+        GroupedDocs,
+        dict:from_list([{Doc, []} || Doc <- Docs])
+    },
+
+    {ok, {WaitingCount1, _, _, _, _} = Acc1} =
+        handle_message({ok, [{ok, Doc1}, {ok, Doc2}]}, hd(Shards), Acc0),
+    ?assertEqual(WaitingCount1, 2),
+
+    {ok, {WaitingCount2, _, _, _, _} = Acc2} =
+        handle_message({error, all_dbs_active}, lists:nth(2, Shards), Acc1),
+    ?assertEqual(WaitingCount2, 1),
+
+    {stop, Reply} =
+        handle_message({ok, [{ok, Doc1}, {ok, Doc2}]}, lists:nth(3, Shards), Acc2),
+
+    ?assertEqual({ok, [{Doc1, {ok, Doc1}}, {Doc2, {ok, Doc2}}]}, Reply).
+
+handle_two_all_dbs_actives() ->
+    Doc1 = #doc{revs = {1, [<<"foo">>]}},
+    Doc2 = #doc{revs = {1, [<<"bar">>]}},
+    Docs = [Doc1, Doc2],
+    Shards =
+        mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]),
+    GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs),
+    Acc0 = {
+        length(Shards),
+        length(Docs),
+        list_to_integer("2"),
+        GroupedDocs,
+        dict:from_list([{Doc, []} || Doc <- Docs])
+    },
+
+    {ok, {WaitingCount1, _, _, _, _} = Acc1} =
+        handle_message({ok, [{ok, Doc1}, {ok, Doc2}]}, hd(Shards), Acc0),
+    ?assertEqual(WaitingCount1, 2),
+
+    {ok, {WaitingCount2, _, _, _, _} = Acc2} =
+        handle_message({error, all_dbs_active}, lists:nth(2, Shards), Acc1),
+    ?assertEqual(WaitingCount2, 1),
+
+    {stop, Reply} =
+        handle_message({error, all_dbs_active}, lists:nth(3, Shards), Acc2),
+
+    ?assertEqual(
+        {accepted, [{Doc1, {accepted, Doc1}}, {Doc2, {accepted, Doc2}}]},
+        Reply
+    ).
+
 one_forbid() ->
     Doc1 = #doc{revs = {1, [<<"foo">>]}},
     Doc2 = #doc{revs = {1, [<<"bar">>]}},