mnkv

%%%=============================================================================
%%%
%%%               |  o __   _|  _  __  |_   _       _ _   (TM)
%%%               |_ | | | (_| (/_ | | |_) (_| |_| | | |
%%%
%%% @copyright (C) 2014, Lindenbaum GmbH
%%%
%%% Permission to use, copy, modify, and/or distribute this software for any
%%% purpose with or without fee is hereby granted, provided that the above
%%% copyright notice and this permission notice appear in all copies.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
%%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
%%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
%%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
%%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
%%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
%%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
%%%
%%% @doc
%%% This module implements the `mnkv' table merge strategy. Currently this
%%% strategy is based on vector clocks provided in {@link mnkv_vclock}. If
%%% the algorithm encounters diverged entries for a specific key, it tries to
%%% call a user defined callback for the respective table. As last resort one
%%% of the nodes with conflicting tables will be restarted.
%%%
%%% For more information about user defined callbacks, refer to the {@mnkv}
%%% behaviour description.
%%%
%%% This code is inspired by the work put in the `unsplit' project by Ulf Wiger,
%%% the man deserves some credit!
%%%
%%% @see https://github.com/uwiger/unsplit
%%% @end
%%%=============================================================================

-module(mnkv_merge).

%% Internal API
-export([tables/2]).

%% Remoting API
-export([handle_actions/1]).

-include("mnkv.hrl").

%%%=============================================================================
%%% Internal API
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @private
%% This function runs inside the {@link mnesia_schema:merge_schema/1}
%% transaction locking all tables to merge. However, since the merged schema
%% must first be committed to be able to make ACID compliant writes, all table
%% merge actions must be dirty opertations.
%%
%% It is sufficient to merge from an arbitrary node from the passed island. The
%% other island should already be consistent. Although dirty, merge actions will
%% be replicated to the other nodes of the island.
%%------------------------------------------------------------------------------
-spec tables([mnkv:table()], [node()]) -> ok | {error, term()}.
tables(_Tables, []) ->
    ok;
tables(Tables, [Node | _]) ->
    ?MNKV_DBG("Merging with ~s:~n", [Node]),
    tables(Tables, Node, ok).
tables([Table | Tables], Node, ok) ->
    ?MNKV_DBG(" * ~w~n", [Table]),
    tables(Tables, Node, merge_table(Node, Table));
tables(_, _, Result) ->
    Result.

%%------------------------------------------------------------------------------
%% @private
%% This is an internal remoting API function that handles remote merge actions.
%%------------------------------------------------------------------------------
-spec handle_actions([{atom(), [term()]}]) -> ok.
handle_actions(Actions) -> lists:foreach(fun handle_action/1, Actions).

%%%=============================================================================
%%% Internal functions
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @private
%% Merges the values found in `Table' from the local and `Remote' node.
%%------------------------------------------------------------------------------
merge_table(Remote, Table) ->
    Keys = get_all_keys([node(), Remote], Table),
    case merge_entries(Keys, node(), Remote, Table, {[], []}) of
        {ok, {LocalActions, RemoteActions}} ->
            case rpc_merge(Remote, RemoteActions) of
                ok    -> handle_actions(LocalActions);
                Error -> Error
            end;
        Error ->
            Error
    end.

%%------------------------------------------------------------------------------
%% @private
%% Returns the local and remote merge actions for a table.
%%------------------------------------------------------------------------------
merge_entries([], _, _, _, Acc) ->
    {ok, Acc};
merge_entries([Key | Keys], Local, Remote, Table, Acc = {LAcc, RAcc}) ->
    case merge_entry(Local, Remote, Table, Key) of
        {all, Action} ->
            ?MNKV_DBG("   - ~w => {all,~w}~n", [Key, Action]),
            NewAcc = {[Action | LAcc], [Action | RAcc]},
            merge_entries(Keys, Local, Remote, Table, NewAcc);
        {local, Action} ->
            ?MNKV_DBG("   - ~w => {local,~w}~n", [Key, Action]),
            NewAcc = {[Action | LAcc], RAcc},
            merge_entries(Keys, Local, Remote, Table, NewAcc);
        {remote, Action} ->
            ?MNKV_DBG("   - ~w => {remote,~w}~n", [Key, Action]),
            NewAcc = {LAcc, [Action | RAcc]},
            merge_entries(Keys, Local, Remote, Table, NewAcc);
        noop ->
            ?MNKV_DBG("   - ~w => noop~n", [Key]),
            merge_entries(Keys, Local, Remote, Table, Acc);
        Error = {error, _} ->
            Error
    end.

%%------------------------------------------------------------------------------
%% @private
%% Return the merge action for `Key' in `Table'. All dirty mnesia operations
%% are allowed as merge actions. The returned action must be of the form
%% `noop' or `{all | local | remote, {DirtyMnesiaFunction, FunctionArgs}}'.
%%------------------------------------------------------------------------------
merge_entry(Local, Remote, Table, Key) ->
    case {get_records(Local, Table, Key), get_records(Remote, Table, Key)} of
        {Records, Records} ->
            noop;
        {[Record], []} ->
            {remote, {dirty_write, [Table, Record]}};
        {[], [Record]} ->
            {local, {dirty_write, [Table, Record]}};
        {[#mnkv{val = V}], [Record = #mnkv{val = V}]} ->
            {local, {dirty_write, [Table, Record]}};
        {[L = #mnkv{ver = LVer}], [R = #mnkv{ver = RVer}]} ->
            case mnkv_vclock:descends(LVer, RVer) of
                true ->
                    {remote, {dirty_write, [Table, L]}};
                false ->
                    case mnkv_vclock:descends(RVer, LVer) of
                        true  -> {local, {dirty_write, [Table, R]}};
                        false -> user_callback(Table, Key, L, R)
                    end
            end;
        {[LRecord], [RRecord]} -> %% merging non-mnkv table
            user_callback(Table, Key, LRecord, RRecord);
        {{error, Reason}, _} ->
            {error, {Local, Reason}};
        {_, {error, Reason}} ->
            {error, {Remote, Reason}}
    end.

%%------------------------------------------------------------------------------
%% @private
%% Call a user provided function to handle a conflicting entry. This can happen
%% on an arbitrary node (the one that connects the nodes and merges the
%% schemas).
%%
%% For more information refer to the {@mnkv} behaviour description.
%%
%% Why is this function written as it is (no pattern matching on #mnkv{})?
%% This hidden feature could (in the future) be used to call the user-provided
%% callback to merge non-mnkv tables ;)
%%------------------------------------------------------------------------------
user_callback(Table, Key, LRecord, RRecord) when is_atom(Table) ->
    case code:ensure_loaded(Table) of
        {module, Table} ->
            case erlang:function_exported(Table, handle_conflict, 3) of
                true ->
                    LVal = get_value(LRecord),
                    RVal = get_value(RRecord),
                    try {Table:handle_conflict(Key, LVal, RVal), LRecord} of
                        {{value, LVal}, _} ->
                            {remote, {dirty_write, [Table, LRecord]}};
                        {{value, RVal}, _} ->
                            {local, {dirty_write, [Table, RRecord]}};
                        {{value, Val}, #mnkv{ver = OldVer}} ->
                            Ver = mnkv_vclock:increment(node(), OldVer),
                            Record = #mnkv{key = Key, val = Val, ver = Ver},
                            {all, {dirty_write, [Table, Record]}};
                        {{value, Record}, _} ->
                            {all, {dirty_write, [Table, Record]}};
                        {delete, _} ->
                            {all, {dirty_delete, [Table, Key]}};
                        _ ->
                            noop
                    catch Class:Exception ->
                            error_logger:error_msg(
                              "~w:handle_conflict/3 raised ~w on key ~w: ~w",
                              [Table, Class, Key, Exception]),
                            {error, {diverged, Table, Key}}
                    end;
                false ->
                    {error, {diverged, Table, Key}}
            end;
        _ ->
            {error, {diverged, Table, Key}}
    end;
user_callback(Table, Key, _, _) ->
    {error, {diverged, Table, Key}}.

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
get_value(#mnkv{val = Val}) -> Val;
get_value(Val)                -> Val.

%%------------------------------------------------------------------------------
%% @private
%% Returns the record for `Key' on `Node'.
%%------------------------------------------------------------------------------
get_records(Node, Table, Key) -> rpc_mnesia(Node, dirty_read, [Table, Key]).

%%------------------------------------------------------------------------------
%% @private
%% Return the list of keys of `Table' on `Nodes'.
%%------------------------------------------------------------------------------
get_all_keys(Nodes, Table) ->
    lists:usort([K || N <- Nodes, K <- rpc_mnesia(N, dirty_all_keys, [Table])]).

%%------------------------------------------------------------------------------
%% @private
%% Make an RPC call to the mnesia module on node `Node'. The `rpc' module knows
%% when a call is local and optimizes that.
%%------------------------------------------------------------------------------
rpc_mnesia(Node, Function, Args) ->
    Timeout = application:get_env(mnkv, rpc_timeout, ?MNKV_RPC_TIMEOUT),
    check_rpc(rpc:call(Node, mnesia, Function, Args, Timeout)).

%%------------------------------------------------------------------------------
%% @private
%% Make subsequent RPC calls to this module on `Node' handing over merge
%% actions in batches of a configurable size. This is done to limit the size
%% of terms sent over the distributed erlang connection. Not doing so might
%% result in connection loss due to inter-node heartbeats timing out.
%%------------------------------------------------------------------------------
rpc_merge(Node, Actions) ->
    BatchSize = application:get_env(mnkv, batch_size, 10),
    Timeout = application:get_env(mnkv, rpc_timeout, ?MNKV_RPC_TIMEOUT),
    rpc_merge(Node, Actions, BatchSize, Timeout + BatchSize * 200).
rpc_merge(_Node, [], _BatchSize, _Timeout) ->
    ok;
rpc_merge(Node, Actions, BatchSize, Timeout) ->
    {Current, Remaining} = split(BatchSize, Actions),
    case rpc:call(Node, ?MODULE, handle_actions, [Current], Timeout) of
        ok     -> rpc_merge(Node, Remaining, BatchSize, Timeout);
        Result -> check_rpc(Result)
    end.

%%------------------------------------------------------------------------------
%% @private
%% Similar to {@link lists:split/2}. However, it is not an error when `N'
%% exceeds the length of the list.
%%------------------------------------------------------------------------------
split(N, List) ->
    try
        lists:split(N, List)
    catch
        error:badarg -> {List, []}
    end.

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
handle_action({Function, Args}) -> erlang:apply(mnesia, Function, Args).

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
check_rpc({badrpc, Reason}) -> {error, Reason};
check_rpc(Result)           -> Result.