<commit_before>% 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
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(fabric_doc_open).

-export([go/3]).

-include("fabric.hrl").
-include_lib("mem3/include/mem3.hrl").
-include_lib("couch/include/couch_db.hrl").

go(DbName, Id, Options) ->
    Workers = fabric_util:submit_jobs(mem3:shards(DbName,Id), open_doc,
        [Id, [deleted|Options]]),
    SuppressDeletedDoc = not lists:member(deleted, 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
    {ok, {ok, #doc{deleted=true}}} when SuppressDeletedDoc ->
        {not_found, deleted};
    {ok, Else} ->
        Else;
    Error ->
        Error
    end.

handle_message({ok, Doc}, _Worker0, {_WaitingCount, 1, _GroupedReplies0}) ->
    {stop, {ok, Doc}};
handle_message(NewReply, Worker0, {WaitingCount, R, GroupedReplies0}) ->
    GroupedReplies =
    case NewReply of
    {rexi_DOWN, _RexiMonPid, _ServerPid, _Reason} ->
        GroupedReplies0;
    {rexi_EXIT, _Reason} ->
        GroupedReplies0;
    _ ->
        [{Worker0, NewReply} | GroupedReplies0]
    end,
    N = list_to_integer(couch_config:get("cluster","n","3")),
    NumReplies = length(GroupedReplies),
    Quorum = find_r_equal_versions(N div 2 + 1, GroupedReplies),
    case Quorum of
    {_Worker, Reply} ->
        case agree(GroupedReplies) of
        false when Reply =/= {not_found, missing} ->
            read_quorum_repairer(GroupedReplies, Reply);
        true ->
            ok
        end,
        {stop, Reply};
    false when NumReplies =:= N ->
        {_Worker, Reply} = get_highest_version(GroupedReplies),
        case Reply of
        {not_found, missing} ->
            ok;
        _ ->
            spawn(fun() ->
                      read_quorum_repairer(GroupedReplies, Reply)
                  end)
        end,
        {stop, Reply};
    false when WaitingCount =:= 1 ->
        {_Worker, Reply} = get_highest_version(GroupedReplies),
        {stop, Reply};
    false when NumReplies >= R ->
        case agree(GroupedReplies) of
        false ->
            {ok, {WaitingCount-1, R, GroupedReplies}};
        true ->
            {_Worker, Reply} = get_highest_version(GroupedReplies),
            {stop, Reply}
        end;
    false ->
        {ok, {WaitingCount-1, R, GroupedReplies}}
    end.

-spec agree({#shard{}, {ok, #doc{}}|{not_found, missing}}) -> boolean().
agree([H|T]) -><commit_msg>The previous version allowed for a response to be returned to the client, in case that R > 1 but only one worker has successfully returned a reply.<commit_after>% 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
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(fabric_doc_open).

-export([go/3]).

-include("fabric.hrl").
-include_lib("mem3/include/mem3.hrl").
-include_lib("couch/include/couch_db.hrl").

go(DbName, Id, Options) ->
    Workers = fabric_util:submit_jobs(mem3:shards(DbName,Id), open_doc,
        [Id, [deleted|Options]]),
    SuppressDeletedDoc = not lists:member(deleted, 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
    {ok, {ok, #doc{deleted=true}}} when SuppressDeletedDoc ->
        {not_found, deleted};
    {ok, Else} ->
        Else;
    Error ->
        Error
    end.

handle_message({ok, Doc}, _Worker0, {_WaitingCount, 1, _GroupedReplies0}) ->
    {stop, {ok, Doc}};
handle_message(NewReply, Worker0, {WaitingCount, R, GroupedReplies0}) ->
    GroupedReplies =
    case NewReply of
    {rexi_DOWN, _RexiMonPid, _ServerPid, _Reason} ->
        GroupedReplies0;
    {rexi_EXIT, _Reason} ->
        GroupedReplies0;
    _ ->
        [{Worker0, NewReply} | GroupedReplies0]
    end,
    N = list_to_integer(couch_config:get("cluster","n","3")),
    NumReplies = length(GroupedReplies),
    Quorum = find_r_equal_versions(N div 2 + 1, GroupedReplies),
    case Quorum of
    {_Worker, Reply} ->
        case agree(GroupedReplies) of
        false when Reply =/= {not_found, missing} ->
            read_quorum_repairer(GroupedReplies, Reply);
        true ->
            ok
        end,
        {stop, Reply};
    false when NumReplies =:= N ->
        {_Worker, Reply} = get_highest_version(GroupedReplies),
        case Reply of
        {not_found, missing} ->
            ok;
        _ ->
            spawn(fun() ->
                      read_quorum_repairer(GroupedReplies, Reply)
                  end)
        end,
        {stop, Reply};
    false when NumReplies >= R ->
        case agree(GroupedReplies) of
        false ->
            {ok, {WaitingCount-1, R, GroupedReplies}};
        true ->
            {_Worker, Reply} = get_highest_version(GroupedReplies),
            {stop, Reply}
        end;
    false when WaitingCount =:= 1 ->
        rexi;
    false ->
        {ok, {WaitingCount-1, R, GroupedReplies}}
    end.

-spec agree({#shard{}, {ok, #doc{}}|{not_found, missing}}) -> boolean().
agree([H|T]) ->
{'fraction_non_alphanumeric': 0.1277461350691619, 'fraction_numerical': 0.010252237591537835, 'mean_word_length': 2.8032178217821784, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2629695', 'n_tokens_mistral': 2038, 'n_tokens_neox': 2019, 'n_words': 516}