Support for read your writes consistency

include/fabric.hrl

@@ -31,7 +31,9 @@
     reducer,
     lang,
     sorted,
-    user_acc
+    user_acc,
+    seqs = [],
+    seqs_sent = false
 }).
 
 -record(view_row, {key, id, value, doc, worker}).

src/fabric.erl

@@ -32,6 +32,8 @@
 % Views
 -export([all_docs/4, changes/4, query_view/3, query_view/4, query_view/6,
     get_view_group_info/2]).
+-export([all_docs/5]).
+-export([query_view/7]).
 
 % miscellany
 -export([design_docs/1, reset_validation_funs/1, cleanup_index_files/0,
@@ -192,18 +194,18 @@ get_missing_revs(DbName, IdsRevs, Options) when is_list(IdsRevs) ->
     {ok, any()} | any().
 update_doc(DbName, Doc, Options) ->
     case update_docs(DbName, [Doc], opts(Options)) of
-    {ok, [{ok, NewRev}]} ->
-        {ok, NewRev};
-    {accepted, [{accepted, NewRev}]} ->
-        {accepted, NewRev};
-    {ok, [{{_Id, _Rev}, Error}]} ->
+    {ok, [{ok, NewRev}], EncShards} ->
+        {ok, NewRev, EncShards};
+    {accepted, [{accepted, NewRev}], EncShards} ->
+        {accepted, NewRev, EncShards};
+    {ok, [{{_Id, _Rev}, Error}], _} ->
         throw(Error);
-    {ok, [Error]} ->
+    {ok, [Error], _} ->
         throw(Error);
-    {ok, []} ->
+    {ok, [], EncShards} ->
         % replication success
         #doc{revs = {Pos, [RevId | _]}} = doc(Doc),
-        {ok, {Pos, RevId}}
+        {ok, {Pos, RevId}, EncShards}
     end.
 
 %% @doc update a list of docs
@@ -212,10 +214,10 @@ update_doc(DbName, Doc, Options) ->
 update_docs(DbName, Docs, Options) ->
     try
         fabric_doc_update:go(dbname(DbName), docs(Docs), opts(Options)) of
-        {ok, Results} ->
-            {ok, Results};
-        {accepted, Results} ->
-            {accepted, Results};
+        {ok, Results, EncShards} ->
+            {ok, Results, EncShards};
+        {accepted, Results, EncShards} ->
+            {accepted, Results, EncShards};
         Error ->
             throw(Error)
     catch {aborted, PreCommitFailures} ->
@@ -234,21 +236,24 @@ purge_docs(_DbName, _IdsRevs) ->
 att_receiver(Req, Length) ->
     fabric_doc_attachments:receiver(Req, Length).
 
+all_docs(DbName, Callback, Acc0, QueryArgs) ->
+    all_docs(DbName, Callback, Acc0, QueryArgs, []).
+
 %% @doc retrieves all docs. Additional query parameters, such as `limit',
 %%      `start_key' and `end_key', `descending', and `include_docs', can
 %%      also be passed to further constrain the query. See <a href=
 %%      "http://wiki.apache.org/couchdb/HTTP_Document_API#All_Documents">
 %%      all_docs</a> for details
--spec all_docs(dbname(), callback(), [] | tuple(), #view_query_args{}) ->
+-spec all_docs(dbname(), callback(), [] | tuple(), #view_query_args{}, []) ->
     {ok, [any()]}.
-all_docs(DbName, Callback, Acc0, #view_query_args{} = QueryArgs) when
+all_docs(DbName, Callback, Acc0, #view_query_args{} = QueryArgs, EncShards) when
         is_function(Callback, 2) ->
-    fabric_view_all_docs:go(dbname(DbName), QueryArgs, Callback, Acc0);
+    fabric_view_all_docs:go(dbname(DbName), QueryArgs, Callback, Acc0, EncShards);
 
 %% @doc convenience function that takes a keylist rather than a record
 %% @equiv all_docs(DbName, Callback, Acc0, kl_to_query_args(QueryArgs))
-all_docs(DbName, Callback, Acc0, QueryArgs) ->
-    all_docs(DbName, Callback, Acc0, kl_to_query_args(QueryArgs)).
+all_docs(DbName, Callback, Acc0, QueryArgs, EncShards) ->
+    all_docs(DbName, Callback, Acc0, kl_to_query_args(QueryArgs), EncShards).
 
 
 -spec changes(dbname(), callback(), any(), #changes_args{} | [{atom(),any()}]) ->
@@ -270,24 +275,27 @@ query_view(DbName, DesignName, ViewName) ->
 %%                     ViewName, fun default_callback/2, [], QueryArgs)
 query_view(DbName, DesignName, ViewName, QueryArgs) ->
     Callback = fun default_callback/2,
-    query_view(DbName, DesignName, ViewName, Callback, [], QueryArgs).
+    query_view(DbName, DesignName, ViewName, Callback, [], QueryArgs, "").
+
+query_view(DbName, DesignName, ViewName, Callback, Acc0, QueryArgs) ->
+    query_view(DbName, DesignName, ViewName, Callback, Acc0, QueryArgs, []).
 
 %% @doc execute a given view.
 %%      There are many additional query args that can be passed to a view,
 %%      see <a href="http://wiki.apache.org/couchdb/HTTP_view_API#Querying_Options">
 %%      query args</a> for details.
 -spec query_view(dbname(), #doc{} | binary(), iodata(), callback(), any(),
-        #view_query_args{}) ->
+        #view_query_args{}, any()) ->
     any().
-query_view(DbName, Design, ViewName, Callback, Acc0, QueryArgs) ->
+query_view(DbName, Design, ViewName, Callback, Acc0, QueryArgs, EncShards) ->
     Db = dbname(DbName), View = name(ViewName),
     case is_reduce_view(Db, Design, View, QueryArgs) of
     true ->
         Mod = fabric_view_reduce;
     false ->
         Mod = fabric_view_map
     end,
-    Mod:go(Db, Design, View, QueryArgs, Callback, Acc0).
+    Mod:go(Db, Design, View, QueryArgs, Callback, Acc0, EncShards).
 
 %% @doc retrieve info about a view group, disk size, language, whether compaction
 %%      is running and so forth
@@ -328,7 +336,7 @@ design_docs(DbName) ->
     ({error, Reason}, _Acc) ->
         {error, Reason}
     end,
-    fabric:all_docs(dbname(DbName), Callback, [], QueryArgs).
+    fabric:all_docs(dbname(DbName), Callback, [], QueryArgs, "").
 
 %% @doc forces a reload of validation functions, this is performed after
 %%      design docs are update

src/fabric_doc_update.erl

@@ -26,68 +26,73 @@ go(DbName, AllDocs, Opts) ->
     validate_atomic_update(DbName, AllDocs, lists:member(all_or_nothing, Opts)),
     Options = lists:delete(all_or_nothing, Opts),
     GroupedDocs = lists:map(fun({#shard{name=Name, node=Node} = Shard, Docs}) ->
-        Ref = rexi:cast(Node, {fabric_rpc, update_docs, [Name, Docs, Options]}),
+        Ref = rexi:cast(Node, {fabric_rpc, update_docs_seq, [Name, Docs, Options]}),
         {Shard#shard{ref=Ref}, Docs}
     end, group_docs_by_shard(DbName, AllDocs)),
     {Workers, _} = lists:unzip(GroupedDocs),
     RexiMon = fabric_util:create_monitors(Workers),
     W = couch_util:get_value(w, Options, integer_to_list(mem3:quorum(DbName))),
     Acc0 = {length(Workers), length(AllDocs), list_to_integer(W), GroupedDocs,
-        dict:from_list([{Doc,[]} || Doc <- AllDocs])},
+        dict:from_list([{Doc,[]} || Doc <- AllDocs]), []},
     Timeout = fabric_util:request_timeout(),
-    try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3, Acc0, infinity, Timeout) of
-    {ok, {Health, Results}} when Health =:= ok; Health =:= accepted ->
-        {Health, [R || R <- couch_util:reorder_results(AllDocs, Results), R =/= noreply]};
+    try fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Acc0, infinity, Timeout) of
+    {ok, {Health, Results, Responders}} when Health =:= ok; Health =:= accepted ->
+        {Health, [R || R <- couch_util:reorder_results(AllDocs, Results), R =/= noreply], Responders};
     {timeout, Acc} ->
-        {_, _, W1, _, DocReplDict} = Acc,
+        {_, _, W1, _, DocReplDict, Responders} = Acc,
         {Health, _, Resp} = dict:fold(fun force_reply/3, {ok, W1, []},
             DocReplDict),
-        {Health, [R || R <- couch_util:reorder_results(AllDocs, Resp), R =/= noreply]};
+        {Health, [R || R <- couch_util:reorder_results(AllDocs, Resp), R =/= noreply], Responders};
     Else ->
         Else
     after
         rexi_monitor:stop(RexiMon)
     end.
 
 handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, Acc0) ->
-    {_, LenDocs, W, GroupedDocs, DocReplyDict} = Acc0,
+    {_, LenDocs, W, GroupedDocs, DocReplyDict, Responders} = Acc0,
     NewGrpDocs = [X || {#shard{node=N}, _} = X <- GroupedDocs, N =/= NodeRef],
-    skip_message({length(NewGrpDocs), LenDocs, W, NewGrpDocs, DocReplyDict});
+    skip_message({length(NewGrpDocs), LenDocs, W, NewGrpDocs, DocReplyDict, Responders});
 
 handle_message({rexi_EXIT, _}, Worker, Acc0) ->
-    {WC,LenDocs,W,GrpDocs,DocReplyDict} = Acc0,
+    {WC,LenDocs,W,GrpDocs,DocReplyDict, Responders} = Acc0,
     NewGrpDocs = lists:keydelete(Worker,1,GrpDocs),
-    skip_message({WC-1,LenDocs,W,NewGrpDocs,DocReplyDict});
+    skip_message({WC-1,LenDocs,W,NewGrpDocs,DocReplyDict, Responders});
 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,
+    {WC,LenDocs,W,GrpDocs,DocReplyDict, Responders} = Acc0,
     NewGrpDocs = lists:keydelete(Worker,1,GrpDocs),
-    skip_message({WC-1,LenDocs,W,NewGrpDocs,DocReplyDict});
+    skip_message({WC-1,LenDocs,W,NewGrpDocs,DocReplyDict, Responders});
 handle_message(attachment_chunk_received, _Worker, Acc0) ->
     {ok, Acc0};
-handle_message({ok, Replies}, Worker, Acc0) ->
-    {WaitingCount, DocCount, W, GroupedDocs, DocReplyDict0} = Acc0,
+handle_message({{ok, Replies},DbSeq}, Worker, Acc0) ->
+    {WaitingCount, DocCount, W, GroupedDocs, DocReplyDict0, Responders} = Acc0,
+    NewResponders =
+        case lists:member({Worker,DbSeq}, Responders) of
+        true ->
+            Responders;
+        false ->
+            [{Worker,DbSeq} | Responders]
+        end,
     {value, {_, Docs}, NewGrpDocs} = lists:keytake(Worker, 1, GroupedDocs),
     DocReplyDict = append_update_replies(Docs, Replies, DocReplyDict0),
-    case {WaitingCount, dict:size(DocReplyDict)} of
-    {1, _} ->
+    if WaitingCount =:= 1 ->
         % last message has arrived, we need to conclude things
         {Health, W, Reply} = dict:fold(fun force_reply/3, {ok, W, []},
-           DocReplyDict),
-        {stop, {Health, Reply}};
-    {_, DocCount} ->
-        % we've got at least one reply for each document, let's take a look
+                                       DocReplyDict),
+        {stop, {Health, Reply, fabric_util:pack(NewResponders)}};
+       true ->
         case dict:fold(fun maybe_reply/3, {stop,W,[]}, DocReplyDict) of
         continue ->
-            {ok, {WaitingCount - 1, DocCount, W, NewGrpDocs, DocReplyDict}};
+            {ok, {WaitingCount - 1, DocCount, W, NewGrpDocs, DocReplyDict, NewResponders}};
         {stop, W, FinalReplies} ->
-            {stop, {ok, FinalReplies}}
+            {stop, {ok, FinalReplies, fabric_util:pack(NewResponders)}}
         end
     end;
 handle_message({missing_stub, Stub}, _, _) ->
     throw({missing_stub, Stub});
 handle_message({not_found, no_db_file} = X, Worker, Acc0) ->
-    {_, _, _, GroupedDocs, _} = Acc0,
+    {_, _, _, GroupedDocs, _, _} = Acc0,
     Docs = couch_util:get_value(Worker, GroupedDocs),
     handle_message({ok, [X || _D <- Docs]}, Worker, Acc0).
 
@@ -160,9 +165,9 @@ append_update_replies([Doc|Rest1], [Reply|Rest2], Dict0) ->
     % TODO what if the same document shows up twice in one update_docs call?
     append_update_replies(Rest1, Rest2, dict:append(Doc, Reply, Dict0)).
 
-skip_message({0, _, W, _, DocReplyDict}) ->
+skip_message({0, _, W, _, DocReplyDict, Responders}) ->
     {Health, W, Reply} = dict:fold(fun force_reply/3, {ok, W, []}, DocReplyDict),
-    {stop, {Health, Reply}};
+    {stop, {Health, Reply, Responders}};
 skip_message(Acc0) ->
     {ok, Acc0}.
 
@@ -195,40 +200,39 @@ doc_update1_test() ->
 
     % test for W = 2
     AccW2 = {length(Shards), length(Docs), list_to_integer("2"), GroupedDocs,
-        Dict},
+        Dict,[]},
 
-    {ok,{WaitingCountW2_1,_,_,_,_}=AccW2_1} =
-        handle_message({ok, [{ok, Doc1}]},hd(Shards),AccW2),
+    {ok,{WaitingCountW2_1,_,_,_,_,_}=AccW2_1} =
+        handle_message({{ok, [{ok, Doc1}]}, 0},hd(Shards),AccW2),
     ?assertEqual(WaitingCountW2_1,2),
-    {stop, FinalReplyW2 } =
-        handle_message({ok, [{ok, Doc1}]},lists:nth(2,Shards),AccW2_1),
-    ?assertEqual({ok, [{Doc1, {ok,Doc1}}]},FinalReplyW2),
+
+    {stop, {ok, [{Doc1, {ok,Doc1}}], _Responders}} =
+        handle_message({{ok, [{ok, Doc1}]}, 0},lists:nth(2,Shards),AccW2_1),
 
     % test for W = 3
     AccW3 = {length(Shards), length(Docs), list_to_integer("3"), GroupedDocs,
-        Dict},
+        Dict,[]},
 
-    {ok,{WaitingCountW3_1,_,_,_,_}=AccW3_1} =
-        handle_message({ok, [{ok, Doc1}]},hd(Shards),AccW3),
+    {ok,{WaitingCountW3_1,_,_,_,_,_}=AccW3_1} =
+        handle_message({{ok, [{ok, Doc1}]}, 0},hd(Shards),AccW3),
     ?assertEqual(WaitingCountW3_1,2),
 
-    {ok,{WaitingCountW3_2,_,_,_,_}=AccW3_2} =
-        handle_message({ok, [{ok, Doc1}]},lists:nth(2,Shards),AccW3_1),
+    {ok,{WaitingCountW3_2,_,_,_,_,_}=AccW3_2} =
+        handle_message({{ok, [{ok, Doc1}]}, 0},lists:nth(2,Shards),AccW3_1),
     ?assertEqual(WaitingCountW3_2,1),
 
-    {stop, FinalReplyW3 } =
-        handle_message({ok, [{ok, Doc1}]},lists:nth(3,Shards),AccW3_2),
-    ?assertEqual({ok, [{Doc1, {ok,Doc1}}]},FinalReplyW3),
+    {stop, {ok, [{Doc1, {ok,Doc1}}], _Responders1}} =
+        handle_message({{ok, [{ok, Doc1}]}, 0},lists:nth(3,Shards),AccW3_2),
 
     % test w quorum > # shards, which should fail immediately
 
     Shards2 = mem3_util:create_partition_map("foo",1,1,["node1"]),
     GroupedDocs2 = group_docs_by_shard_hack(<<"foo">>,Shards2,Docs),
 
     AccW4 =
-        {length(Shards2), length(Docs), list_to_integer("2"), GroupedDocs2, Dict},
+        {length(Shards2), length(Docs), list_to_integer("2"), GroupedDocs2, Dict, []},
     Bool =
-    case handle_message({ok, [{ok, Doc1}]},hd(Shards2),AccW4) of
+    case handle_message({{ok, [{ok, Doc1}]}, 0},hd(Shards2),AccW4) of
         {stop, _Reply} ->
             true;
         _ -> false
@@ -241,40 +245,33 @@ doc_update1_test() ->
     SA2 = #shard{node=a, range=2},
     SB2 = #shard{node=b, range=2},
     GroupedDocs3 = [{SA1,[Doc1]}, {SB1,[Doc1]}, {SA2,[Doc2]}, {SB2,[Doc2]}],
-    StW5_0 = {length(GroupedDocs3), length(Docs2), 2, GroupedDocs3, Dict2},
-    {ok, StW5_1} = handle_message({ok, [{ok, "A"}]}, SA1, StW5_0),
+    StW5_0 = {length(GroupedDocs3), length(Docs2), 2, GroupedDocs3, Dict2, []},
+    {ok, StW5_1} = handle_message({{ok, [{ok, "A"}]}, 0}, SA1, StW5_0),
     {ok, StW5_2} = handle_message({rexi_EXIT, nil}, SB1, StW5_1),
     {ok, StW5_3} = handle_message({rexi_EXIT, nil}, SA2, StW5_2),
-    {stop, ReplyW5} = handle_message({rexi_EXIT, nil}, SB2, StW5_3),
-    ?assertEqual(
-        {error, [{Doc1,{accepted,"A"}},{Doc2,{error,internal_server_error}}]},
-        ReplyW5
-    ).
-
+    {stop, {error, [{Doc1,{accepted,"A"}},{Doc2,{error,internal_server_error}}], _Responders2}} =
+        handle_message({rexi_EXIT, nil}, SB2, StW5_3).
 
 doc_update2_test() ->
     Doc1 = #doc{revs = {1,[<<"foo">>]}},
     Doc2 = #doc{revs = {1,[<<"bar">>]}},
     Docs = [Doc2, Doc1],
     Shards =
         mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
-    GroupedDocs = group_docs_by_shard_hack(<<"foo">>,Shards,Docs),
+    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])},
+        dict:from_list([{Doc,[]} || Doc <- Docs]), []},
 
-    {ok,{WaitingCount1,_,_,_,_}=Acc1} =
-        handle_message({ok, [{ok, Doc1},{ok, Doc2}]},hd(Shards),Acc0),
+    {ok,{WaitingCount1,_,_,_,_,_}=Acc1} =
+        handle_message({{ok, [{ok, Doc1},{ok, Doc2}]}, 0},hd(Shards),Acc0),
     ?assertEqual(WaitingCount1,2),
 
-    {ok,{WaitingCount2,_,_,_,_}=Acc2} =
+    {ok,{WaitingCount2,_,_,_,_,_}=Acc2} =
         handle_message({rexi_EXIT, 1},lists:nth(2,Shards),Acc1),
     ?assertEqual(WaitingCount2,1),
 
-    {stop, Reply} =
-        handle_message({rexi_EXIT, 1},lists:nth(3,Shards),Acc2),
-
-    ?assertEqual({accepted, [{Doc1,{accepted,Doc2}}, {Doc2,{accepted,Doc1}}]},
-        Reply).
+    {stop, {accepted, [{Doc1,{accepted,Doc2}}, {Doc2,{accepted,Doc1}}], _Responders}} =
+        handle_message({rexi_EXIT, 1},lists:nth(3,Shards),Acc2).
 
 doc_update3_test() ->
     Doc1 = #doc{revs = {1,[<<"foo">>]}},
@@ -284,20 +281,18 @@ doc_update3_test() ->
         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])},
+        dict:from_list([{Doc,[]} || Doc <- Docs]),[]},
 
-    {ok,{WaitingCount1,_,_,_,_}=Acc1} =
-        handle_message({ok, [{ok, Doc1},{ok, Doc2}]},hd(Shards),Acc0),
+    {ok,{WaitingCount1,_,_,_,_,_}=Acc1} =
+        handle_message({{ok, [{ok, Doc1},{ok, Doc2}]}, 0},hd(Shards),Acc0),
     ?assertEqual(WaitingCount1,2),
 
-    {ok,{WaitingCount2,_,_,_,_}=Acc2} =
+    {ok,{WaitingCount2,_,_,_,_,_}=Acc2} =
         handle_message({rexi_EXIT, 1},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, Doc2}},{Doc2, {ok,Doc1}}]},Reply).
+    {stop, {ok, [{Doc1, {ok, Doc2}},{Doc2, {ok,Doc1}}], _Responders}} =
+        handle_message({{ok, [{ok, Doc1},{ok, Doc2}]}, 0},lists:nth(3,Shards),Acc2).
 
 % needed for testing to avoid having to start the mem3 application
 group_docs_by_shard_hack(_DbName, Shards, Docs) ->