1 files changed, 277 insertions, 41 deletions

diff --git a/apps/fabric/src/fabric_doc_open_revs.erl b/apps/fabric/src/fabric_doc_open_revs.erl
index 8dbd1aba..722d23a5 100644
--- a/apps/fabric/src/fabric_doc_open_revs.erl
+++ b/apps/fabric/src/fabric_doc_open_revs.erl
@@ -1,5 +1,5 @@
 % Copyright 2010 Cloudant
-% 
+%
 % 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
@@ -19,65 +19,301 @@
 -include("fabric.hrl").
 -include_lib("mem3/include/mem3.hrl").
 -include_lib("couch/include/couch_db.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-record(state, {
+    dbname,
+    worker_count,
+    reply_count = 0,
+    r,
+    revs,
+    latest,
+    replies = []
+}).
 
 go(DbName, Id, Revs, Options) ->
     Workers = fabric_util:submit_jobs(mem3:shards(DbName,Id), open_revs,
         [Id, Revs, Options]),
     R = couch_util:get_value(r, Options, couch_config:get("cluster","r","2")),
-    Acc0 = {length(Workers), list_to_integer(R), []},
-    case fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Acc0) of
+    State = #state{
+        dbname = DbName,
+        worker_count = length(Workers),
+        r = list_to_integer(R),
+        revs = Revs,
+        latest = lists:member(latest, Options),
+        replies = case Revs of all -> []; Revs -> [{Rev,[]} || Rev <- Revs] end
+    },
+    case fabric_util:recv(Workers, #shard.ref, fun handle_message/3, State) of
     {ok, {ok, Reply}} ->
         {ok, Reply};
     Else ->
         Else
     end.
 
-handle_message({rexi_DOWN, _, _, _}, _Worker, Acc0) ->
-    skip_message(Acc0);
-handle_message({rexi_EXIT, _}, _Worker, Acc0) ->
-    skip_message(Acc0);
-handle_message(Reply, _Worker, {WaitingCount, R, Replies}) ->
-    case merge_read_reply(make_key(Reply), Reply, Replies) of
-    {_, KeyCount} when KeyCount =:= R ->
-        {stop, Reply};
-    {NewReplies, KeyCount} when KeyCount < R ->
-        if WaitingCount =:= 1 ->
-            % last message arrived, but still no quorum
-            repair_read_quorum_failure(NewReplies);
-        true ->
-            {ok, {WaitingCount-1, R, NewReplies}}
+handle_message({rexi_DOWN, _, _, _}, _Worker, State) ->
+    skip(State);
+handle_message({rexi_EXIT, _}, _Worker, State) ->
+    skip(State);
+handle_message({ok, RawReplies}, _Worker, #state{revs = all} = State) ->
+    #state{
+        dbname = DbName,
+        reply_count = ReplyCount,
+        worker_count = WorkerCount,
+        replies = All0,
+        r = R
+    } = State,
+    All = lists:foldl(fun(Reply,D) -> orddict:update_counter(Reply,1,D) end,
+        All0, RawReplies),
+    Reduced = remove_ancestors(All, []),
+    Complete = (ReplyCount =:= (WorkerCount - 1)),
+    Repair = case Reduced of All -> false; _ ->
+        [D || {{ok,D}, _} <- Reduced]
+    end,
+    case maybe_reply(DbName, Reduced, Complete, Repair, R) of
+    noreply ->
+        {ok, State#state{replies = All, reply_count = ReplyCount+1}};
+    {reply, FinalReply} ->
+        {stop, FinalReply}
+    end;
+handle_message({ok, RawReplies0}, _Worker, State) ->
+    % we've got an explicit revision list, but if latest=true the workers may
+    % return a descendant of the requested revision.  Take advantage of the
+    % fact that revisions are returned in order to keep track.
+    RawReplies = strip_not_found_missing(RawReplies0),
+    #state{
+        dbname = DbName,
+        reply_count = ReplyCount,
+        worker_count = WorkerCount,
+        replies = All0,
+        r = R
+    } = State,
+    All = lists:zipwith(fun({Rev, D}, Reply) ->
+        if Reply =:= error -> {Rev, D}; true ->
+            {Rev, orddict:update_counter(Reply, 1, D)}
         end
+    end, All0, RawReplies),
+    Reduced = [remove_ancestors(X, []) || {_, X} <- All],
+    FinalReplies = [choose_winner(X, R) || X <- Reduced],
+    Complete = (ReplyCount =:= (WorkerCount - 1)),
+    case is_repair_needed(All, FinalReplies) of
+    true ->
+        Repair = [D || {{ok,D}, _} <- lists:flatten(Reduced)];
+    false ->
+        Repair = false
+    end,
+    case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
+    noreply ->
+        {ok, State#state{replies = All, reply_count = ReplyCount+1}};
+    {reply, FinalReply} ->
+        {stop, FinalReply}
     end.
 
-skip_message({1, _R, Replies}) ->
-    repair_read_quorum_failure(Replies);
-skip_message({WaitingCount, R, Replies}) ->
-    {ok, {WaitingCount-1, R, Replies}}.
+skip(#state{revs=all} = State) ->
+    handle_message({ok, []}, nil, State);
+skip(#state{revs=Revs} = State) ->
+    handle_message({ok, [error || _Rev <- Revs]}, nil, State).
 
-merge_read_reply(Key, Reply, Replies) ->
-    case lists:keyfind(Key, 1, Replies) of
+maybe_reply(DbName, ReplyDict, IsComplete, RepairDocs, R) ->
+    case lists:all(fun({_, C}) -> C >= R end, ReplyDict) of
+    true ->
+        maybe_execute_read_repair(DbName, RepairDocs),
+        {reply, unstrip_not_found_missing(orddict:fetch_keys(ReplyDict))};
     false ->
-        {[{Key, Reply, 1} | Replies], 1};
-    {Key, _, N} ->
-        {lists:keyreplace(Key, 1, Replies, {Key, Reply, N+1}), N+1}
+        case IsComplete of false -> noreply; true ->
+            maybe_execute_read_repair(DbName, RepairDocs),
+            {reply, unstrip_not_found_missing(orddict:fetch_keys(ReplyDict))}
+        end
     end.
 
-make_key({ok, L}) when is_list(L) ->
-    make_key(L);
-make_key([]) ->
-    [];
-make_key([{ok, #doc{revs= {Pos,[RevId | _]}}} | Rest]) ->
-    [{ok, {Pos, RevId}} | make_key(Rest)];
-make_key([{{not_found, missing}, Rev} | Rest]) ->
-    [{not_found, Rev} | make_key(Rest)].
+choose_winner(Options, R) ->
+    case lists:dropwhile(fun({_Reply, C}) -> C < R end, Options) of
+    [] ->
+        case [Elem || {{ok, #doc{}}, _} = Elem <- Options] of
+        [] ->
+            hd(Options);
+        Docs ->
+            lists:last(lists:sort(Docs))
+        end;
+    [QuorumMet | _] ->
+        QuorumMet
+    end.
 
-repair_read_quorum_failure(Replies) ->
-    case [Doc || {_Key, {ok, Doc}, _Count} <- Replies] of
+% repair needed if any reply other than the winner has been received for a rev
+is_repair_needed([], []) ->
+    false;
+is_repair_needed([{_Rev, [Reply]} | Tail1], [Reply | Tail2]) ->
+    is_repair_needed(Tail1, Tail2);
+is_repair_needed([H1|_], [H2|_]) ->
+    true.
+
+% this presumes the incoming list is sorted, i.e. shorter revlists come first
+remove_ancestors([], Acc) ->
+    lists:reverse(Acc);
+remove_ancestors([{{not_found, _}, Count} = Head | Tail], Acc) ->
+    % any document is a descendant
+    case lists:filter(fun({{ok, #doc{}}, _}) -> true; (_) -> false end, Tail) of
+    [{{ok, #doc{}} = Descendant, _} | _] ->
+        remove_ancestors(orddict:update_counter(Descendant, Count, Tail), Acc);
     [] ->
-        {stop, {not_found, missing}};
-    [Doc|_] ->
-        % TODO merge docs to find the winner as determined by replication
-        {stop, {ok, Doc}}
+        remove_ancestors(Tail, [Head | Acc])
+    end;
+remove_ancestors([{{ok, #doc{revs = {Pos, Revs}}}, Count} = Head | Tail], Acc) ->
+    Descendants = lists:dropwhile(fun
+    ({{ok, #doc{revs = {Pos2, Revs2}}}, _}) ->
+        case lists:nthtail(Pos2 - Pos, Revs2) of
+        [] ->
+            % impossible to tell if Revs2 is a descendant - assume no
+            true;
+        History ->
+            % if Revs2 is a descendant, History is a prefix of Revs
+            not lists:prefix(History, Revs)
+        end
+    end, Tail),
+    case Descendants of [] ->
+        remove_ancestors(Tail, [Head | Acc]);
+    [{Descendant, _} | _] ->
+        remove_ancestors(orddict:update_counter(Descendant, Count, Tail), Acc)
     end.
 
-    
\ No newline at end of file
+maybe_execute_read_repair(_Db, false) ->
+    ok;
+maybe_execute_read_repair(Db, Docs) ->
+    spawn(fun() ->
+        [#doc{id=Id} | _] = Docs,
+        Ctx = #user_ctx{roles=[<<"_admin">>]},
+        Res = fabric:update_docs(Db, Docs, [replicated_changes, {user_ctx,Ctx}]),
+        ?LOG_INFO("read_repair ~s ~s ~p", [Db, Id, Res])
+    end).
+
+% hackery required so that not_found sorts first
+strip_not_found_missing([]) ->
+    [];
+strip_not_found_missing([{{not_found, missing}, Rev} | Rest]) ->
+    [{not_found, Rev} | strip_not_found_missing(Rest)];
+strip_not_found_missing([Else | Rest]) ->
+    [Else | strip_not_found_missing(Rest)].
+
+unstrip_not_found_missing([]) ->
+    [];
+unstrip_not_found_missing([{not_found, Rev} | Rest]) ->
+    [{{not_found, missing}, Rev} | unstrip_not_found_missing(Rest)];
+unstrip_not_found_missing([Else | Rest]) ->
+    [Else | unstrip_not_found_missing(Rest)].
+
+remove_ancestors_test() ->
+    Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
+    Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
+    Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
+    Bar2 = {not_found, {1,<<"bar">>}},
+    ?assertEqual(
+        [{Bar1,1}, {Foo1,1}],
+        remove_ancestors([{Bar1,1}, {Foo1,1}], [])
+    ),
+    ?assertEqual(
+        [{Bar1,1}, {Foo2,2}],
+        remove_ancestors([{Bar1,1}, {Foo1,1}, {Foo2,1}], [])
+    ),
+    ?assertEqual(
+        [{Bar1,2}],
+        remove_ancestors([{Bar2,1}, {Bar1,1}], [])
+    ).
+
+all_revs_test() ->
+    State0 = #state{worker_count = 3, r = 2, revs = all},
+    Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
+    Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
+    Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
+    ?assertMatch(
+        {ok, #state{}},
+        handle_message({ok, [Foo1, Bar1]}, nil, State0)
+    ),
+    {ok, State1} = handle_message({ok, [Foo1, Bar1]}, nil, State0),
+
+    % the normal case - workers agree
+    ?assertEqual(
+        {stop, [Bar1, Foo1]},
+        handle_message({ok, [Foo1, Bar1]}, nil, State1)
+    ),
+
+    % a case where the 2nd worker has a newer Foo - currently we're considering
+    % Foo to have reached quorum and execute_read_repair()
+    ?assertEqual(
+        {stop, [Bar1, Foo2]},
+        handle_message({ok, [Foo2, Bar1]}, nil, State1)
+    ),
+
+    % a case where quorum has not yet been reached for Foo
+    ?assertMatch(
+        {ok, #state{}},
+        handle_message({ok, [Bar1]}, nil, State1)
+    ),
+    {ok, State2} = handle_message({ok, [Bar1]}, nil, State1),
+
+    % still no quorum, but all workers have responded.  We include Foo1 in the
+    % response and execute_read_repair()
+    ?assertEqual(
+        {stop, [Bar1, Foo1]},
+        handle_message({ok, [Bar1]}, nil, State2)
+    ).
+
+specific_revs_test() ->
+    Revs = [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}],
+    State0 = #state{
+        worker_count = 3,
+        r = 2,
+        revs = Revs,
+        latest = false,
+        replies = [{Rev,[]} || Rev <- Revs]
+    },
+    Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
+    Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
+    Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
+    Baz1 = {{not_found, missing}, {1,<<"baz">>}},
+    Baz2 = {ok, #doc{revs = {1, [<<"baz">>]}}},
+
+    ?assertMatch(
+        {ok, #state{}},
+        handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0)
+    ),
+    {ok, State1} = handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0),
+
+    % the normal case - workers agree
+    ?assertEqual(
+        {stop, [Foo1, Bar1, Baz1]},
+        handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1)
+    ),
+
+    % latest=true, worker responds with Foo2 and we return it
+    State0L = State0#state{latest = true},
+    ?assertMatch(
+        {ok, #state{}},
+        handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L)
+    ),
+    {ok, State1L} = handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L),
+    ?assertEqual(
+        {stop, [Foo2, Bar1, Baz1]},
+        handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State1L)
+    ),
+
+    % Foo1 is included in the read quorum for Foo2
+    ?assertEqual(
+        {stop, [Foo2, Bar1, Baz1]},
+        handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1L)
+    ),
+
+    % {not_found, missing} is included in the quorum for any found revision
+    ?assertEqual(
+        {stop, [Foo2, Bar1, Baz2]},
+        handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State1L)
+    ),
+
+    % a worker failure is skipped
+    ?assertMatch(
+        {ok, #state{}},
+        handle_message({rexi_EXIT, foo}, nil, State1L)
+    ),
+    {ok, State2L} = handle_message({rexi_EXIT, foo}, nil, State1L),
+    ?assertEqual(
+        {stop, [Foo2, Bar1, Baz2]},
+        handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State2L)
+    ).