src/mem3/src/mem3_sync_security.erl

% 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(mem3_sync_security).

-export([maybe_sync/2, maybe_sync_int/2]).
-export([go/0, go/1]).

-include_lib("mem3/include/mem3.hrl").


maybe_sync(#shard{}=Src, #shard{}=Dst) ->
    case is_local(Src#shard.name) of
        false ->
            erlang:spawn(?MODULE, maybe_sync_int, [Src, Dst]);
        true ->
            ok
    end.

maybe_sync_int(#shard{name=Name}=Src, Dst) ->
    DbName = mem3:dbname(Name),
    case fabric:get_all_security(DbName, [{shards, [Src, Dst]}]) of
        {ok, WorkerObjs} ->
            Objs = [Obj || {_Worker, Obj} <- WorkerObjs],
            case length(lists:usort(Objs)) of
                1 -> ok;
                2 -> go(DbName)
            end;
        {error, no_majority} ->
            go(DbName);
        Else ->
            Args = [DbName, Else],
            couch_log:error("Error checking security objects for ~s :: ~p", Args)
    end.

go() ->
    {ok, Dbs} = fabric:all_dbs(),
    lists:foreach(fun handle_existing_db/1, Dbs).

go(DbName) when is_binary(DbName) ->
    handle_existing_db(DbName).

handle_existing_db(DbName) ->
    try handle_db(DbName) of
        _ -> ok
    catch
        error:database_does_not_exist->
            couch_log:error("Db was deleted while getting security"
                " object. DbName: ~p", [DbName]),
            ok
    end.

handle_db(DbName) ->
    ShardCount = length(mem3:shards(DbName)),
    case get_all_security(DbName) of
    {ok, SecObjs} ->
        case is_ok(SecObjs, ShardCount) of
        ok ->
            ok;
        {fixable, SecObj} ->
            couch_log:info("Sync security object for ~p: ~p", [DbName, SecObj]),
            case fabric:set_security(DbName, SecObj) of
                ok -> ok;
                Error ->
                    couch_log:error("Error setting security object in ~p: ~p",
                        [DbName, Error])
            end;
        broken ->
            couch_log:error("Bad security object in ~p: ~p", [DbName, SecObjs])
        end;
    Error ->
        couch_log:error("Error getting security objects for ~p: ~p", [
                DbName, Error])
    end.

get_all_security(DbName) ->
    case fabric:get_all_security(DbName) of
    {ok, SecObjs} ->
        SecObjsDict = lists:foldl(fun({_, SO}, Acc) ->
            dict:update_counter(SO, 1, Acc)
        end, dict:new(), SecObjs),
        {ok, dict:to_list(SecObjsDict)};
    Error ->
        Error
    end.

is_ok([_], _) ->
    % One security object is the happy case
    ok;
is_ok([_, _] = SecObjs0, ShardCount) ->
    % Figure out if we have a simple majority of security objects
    % and if so, use that as the correct value. Otherwise we abort
    % and rely on human intervention.
    {Count, SecObj} =  lists:max([{C, O} || {O, C} <- SecObjs0]),
    case Count >= ((ShardCount div 2) + 1) of
        true -> {fixable, SecObj};
        false -> broken
    end;
is_ok(_, _) ->
    % Anything else requires human intervention
    broken.


is_local(<<"shards/", _/binary>>) ->
    false;
is_local(_) ->
    true.