machi/test/machi_chain_manager1_test.erl

%% -------------------------------------------------------------------
%%
%% Machi: a small village of replicated files
%%
%% Copyright (c) 2014 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you 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(machi_chain_manager1_test).

-include("machi.hrl").
-include("machi_projection.hrl").

-define(MGR, machi_chain_manager1).

-define(D(X), io:format(user, "~s ~p\n", [??X, X])).
-define(Dw(X), io:format(user, "~s ~w\n", [??X, X])).
-define(FLU_C,  machi_flu1_client).
-define(FLU_PC, machi_proxy_flu1_client).

-export([]).

-ifdef(TEST).

-ifdef(EQC).
-include_lib("eqc/include/eqc.hrl").
%% -include_lib("eqc/include/eqc_statem.hrl").
-define(QC_OUT(P),
        eqc:on_output(fun(Str, Args) -> io:format(user, Str, Args) end, P)).
-endif.

-include_lib("eunit/include/eunit.hrl").
-compile(export_all).

%% @doc Create a summary report of all of the *private* projections of
%%      each of the FLUs in the chain, and create a summary for each
%%      epoch number.
%%
%% Report format: list({EpochNumber::non_neg_integer(), Report::rpt()})
%%        rpt(): {ok_disjoint | bummer_NOT_DISJOINT, agree_list()}
%% agree_list(): list(agree())
%%      agree(): {agreed_membership, UPIandRepairing()} |
%%               {not_agreed, ......TODO LEFT OFF HERE
%%
%% Example:
%% TODO LEFT OFF HERE

unanimous_report(Namez) ->
    UniquePrivateEs =
        lists:usort(lists:flatten(
                      [element(2, ?FLU_PC:list_all_projections(FLU, private)) ||
                          {_FLUName, FLU} <- Namez])),
    [{Epoch, unanimous_report(Epoch, Namez)} || Epoch <- UniquePrivateEs,
                                                Epoch /= 0].

unanimous_report(Epoch, Namez) ->
    FLU_Projs = [{FLUName,
                  case ?FLU_PC:read_projection(FLU, private, Epoch) of
                      {ok, T} ->
                          machi_chain_manager1:inner_projection_or_self(T);
                      _Else ->
                          not_in_this_epoch
                  end} || {FLUName, FLU} <- Namez],
    unanimous_report2(Epoch, FLU_Projs).

unanimous_report2(Epoch, FLU_Projs) ->
    ProjsSumms = [{FLU, if is_tuple(P) ->
                                Summ = machi_projection:make_summary(P),
                                lists:flatten(io_lib:format("~w", [Summ]));
                           is_atom(P) ->
                                P
                        end} || {FLU, P} <- FLU_Projs],
    UPI_R_Sums = [{Proj#projection_v1.upi, Proj#projection_v1.repairing,
                   Proj#projection_v1.epoch_csum} ||
                     {_FLUname, Proj} <- FLU_Projs,
                     is_record(Proj, projection_v1)],
    UniqueUPIs = lists:usort([UPI || {UPI, _Repairing, _CSum} <- UPI_R_Sums]),
    Res =
        [begin
             case lists:usort([CSum || {U, _Repairing, CSum} <- UPI_R_Sums,
                                       U == UPI]) of
                 [_1CSum] ->
                     %% Yay, there's only 1 checksum.  Let's check
                     %% that all FLUs are in agreement.
                     {UPI, Repairing, _CSum} =
                         lists:keyfind(UPI, 1, UPI_R_Sums),
                     Tmp = [{FLU, case proplists:get_value(FLU, FLU_Projs) of
                                      P when is_record(P, projection_v1) ->
                                          P#projection_v1.epoch_csum;
                                      Else ->
                                          Else
                                  end} || FLU <- UPI ++ Repairing],
                     case lists:usort([CSum || {_FLU, CSum} <- Tmp]) of
                         [_] ->
                             {agreed_membership, {UPI, Repairing}};
                         _Else2 ->
                             {not_agreed, ProjsSumms}
                     end;
                 _Else ->
                     {not_agreed, [checksum_disagreement|ProjsSumms]}
             end
         end || UPI <- UniqueUPIs],
    AgreedResUPI_Rs = [UPI++Repairing ||
                          {agreed_membership, {UPI, Repairing}} <- Res],
    Tag = case lists:usort(lists:flatten(AgreedResUPI_Rs)) ==
               lists:sort(lists:flatten(AgreedResUPI_Rs)) of
              true ->
                  ok_disjoint;
              false ->
                  bummer_NOT_DISJOINT
          end,
    {Tag, Res}.

all_reports_are_disjoint(Report) ->
    [] == [X || {_Epoch, Tuple}=X <- Report,
                element(1, Tuple) /= ok_disjoint].

extract_chains_relative_to_flu(FLU, Report) ->
    {FLU, [{Epoch, UPI, Repairing} ||
              {Epoch, {ok_disjoint, Es}} <- Report,
              {agreed_membership, {UPI, Repairing}} <- Es,
              lists:member(FLU, UPI) orelse lists:member(FLU, Repairing)]}.

chain_to_projection(MyName, Epoch, UPI_list, Repairing_list, All_list) ->
    MemberDict = orddict:from_list([{FLU, #p_srvr{name=FLU}} ||
                                       FLU <- All_list]),
    machi_projection:new(Epoch, MyName, MemberDict,
                         All_list -- (UPI_list ++ Repairing_list),
                         UPI_list, Repairing_list, [{artificial_by, ?MODULE}]).

simple_chain_state_transition_is_sane(UPI1, Repair1, UPI2) ->
    {KeepsDels, Orders} = mk(UPI1, Repair1, UPI2),
    NumKeeps = length([x || keep <- KeepsDels]),
    NumOrders = length(Orders),
    false == lists:member(error, Orders)
        andalso Orders == lists:sort(Orders)
        andalso length(UPI2) == NumKeeps + NumOrders.

mk(UPI1, Repair1, UPI2) ->
    mk(UPI1, Repair1, UPI2, []).

mk([X|UPI1], Repair1, [X|UPI2], Acc) ->
    mk(UPI1, Repair1, UPI2, [keep|Acc]);
mk([X|UPI1], Repair1, UPI2, Acc) ->
    mk(UPI1, Repair1, UPI2 -- [X], [del|Acc]);
mk([], [], [], Acc) ->
    {lists:reverse(Acc), []};
mk([], Repair1, UPI2, Acc) ->
    {lists:reverse(Acc), mk_order(UPI2, Repair1)}.

mk_order(UPI2, Repair1) ->
    R1 = length(Repair1),
    Repair1_order_d = orddict:from_list(lists:zip(Repair1, lists:seq(1, R1))),
    UPI2_order = [case orddict:find(X, Repair1_order_d) of
                      {ok, Idx} -> Idx;
                      error     -> error
                  end || X <- UPI2],
    UPI2_order.

perms([]) -> [[]];
perms(L)  -> [[H|T] || H <- L, T <- perms(L--[H])].

combinations(L) ->
    lists:usort(perms(L) ++ lists:append([ combinations(L -- [X]) || X <- L])).

ordered_combinations(Master) ->
    [L || L <- combinations(Master), is_ordered(L, Master)].

is_ordered(L, Reference) ->
    L_order = mk_order(L, Reference),
    lists:all(fun(X) -> is_integer(X) end, L_order) andalso
        L_order == lists:sort(L_order).

mk_order_is_sane_test_() ->
    {timeout, 300, fun() -> mk_order_is_sane_test2() end}.

mk_order_is_sane_test2() ->
    %% All: A list of all FLUS for a particular test
    %% UPI1: some combination of All that represents UPI1
    %% Repair1: Some combination of (All -- UP1) that represents Repairing1
    %% ... then we test check_sane_func_with_upi_and_repair() with all
    %% possible UPI1 and Repair1.
    [true = check_sane_func_with_upi_and_repair(UPI1, Repair1) ||
        %% The five elements below runs on my MacBook Pro in about 4.8 seconds
        %% All <- [ [a], [a,b], [a,b,c], [a,b,c,d], [a,b,c,d,e] ],
        All <- [ [a], [a,b], [a,b,c], [a,b,c,d] ],
        UPI1 <- combinations(All),
        Repair1 <- combinations(All -- UPI1)].

%% Given a UPI1 and Repair1 list, we calculate all possible good UPI2
%% lists.  For all good {UPI1, Repair1} -> UPI2 transitions, then the
%% simple_chain_state_transition_is_sane() function must be true.  For
%% all other UPI2 transitions, simple_chain_state_transition_is_sane()
%% must be false.
%%
%% We include adding an extra possible participant, 'bogus', to the
%% list of all possible UPI2 transitions, just to demonstrate that
%% adding an extra element/participant/thingie is never sane.

check_sane_func_with_upi_and_repair([], []) ->
    true;
check_sane_func_with_upi_and_repair(UPI1, Repair1) ->
    Good_UPI2s = [ X ++ Y || X <- ordered_combinations(UPI1),
                             Y <- ordered_combinations(Repair1)],
    All_UPI2s = combinations(lists:usort(UPI1 ++ Repair1) ++ [bogus]),

    [true = simple_chain_state_transition_is_sane(UPI1, Repair1, UPI2) ||
        UPI2 <- Good_UPI2s],
    [false = simple_chain_state_transition_is_sane(UPI1, Repair1, UPI2) ||
        UPI2 <- (All_UPI2s -- Good_UPI2s)],

    All = UPI1 ++ Repair1,
    Me = hd(All),
    MembersDict = orddict:from_list([{X, #p_srvr{name=X}} || X <- All]),
    [true = begin
         Down1 = All -- (UPI1 ++ Repair1),
         Repair2 = [],
         Down2 = All -- (UPI2 ++ Repair2),
         P1 = machi_projection:new(1, Me, MembersDict,
                                   Down1, UPI1, Repair1, []),
         P2 = machi_projection:new(2, Me, MembersDict,
                                   Down2, UPI2, Repair2, []),
         QQ = machi_chain_manager1:projection_transition_is_sane(P1, P2, Me,
                                                                 false),
         if QQ /= true ->
                 io:format(user, "QQ ~p\n", [QQ]);
            true ->
                 ok
         end,
         QQ
     end || UPI2 <- Good_UPI2s],

    true.

-ifndef(PULSE).

smoke0_test() ->
    {ok, _} = machi_partition_simulator:start_link({1,2,3}, 50, 50),
    Host = "localhost",
    TcpPort = 6623,
    {ok, FLUa} = machi_flu1:start_link([{a,TcpPort,"./data.a"}]),
    Pa = #p_srvr{name=a, address=Host, port=TcpPort},
    Members_Dict = machi_projection:make_members_dict([Pa]),
    %% Egadz, more racing on startup, yay.  TODO fix.
    timer:sleep(1),
    {ok, FLUaP} = ?FLU_PC:start_link(Pa),
    {ok, M0} = ?MGR:start_link(a, Members_Dict, [{active_mode, false}]),
    _SockA = machi_util:connect(Host, TcpPort),
    try
        pong = ?MGR:ping(M0)
    after
        ok = ?MGR:stop(M0),
        ok = machi_flu1:stop(FLUa),
        ok = ?FLU_PC:quit(FLUaP),
        ok = machi_partition_simulator:stop()
    end.

smoke1_test() ->
    machi_partition_simulator:start_link({1,2,3}, 100, 0),
    TcpPort = 62777,
    FluInfo = [{a,TcpPort+0,"./data.a"}, {b,TcpPort+1,"./data.b"}, {c,TcpPort+2,"./data.c"}],
    P_s = [#p_srvr{name=Name, address="localhost", port=Port} ||
              {Name,Port,_Dir} <- FluInfo],

    [machi_flu1_test:clean_up_data_dir(Dir) || {_,_,Dir} <- FluInfo],
    FLUs = [element(2, machi_flu1:start_link([{Name,Port,Dir}])) ||
               {Name,Port,Dir} <- FluInfo],
    MembersDict = machi_projection:make_members_dict(P_s),
    {ok, M0} = ?MGR:start_link(a, MembersDict, [{active_mode,false}]),
    try
        {ok, P1} = ?MGR:test_calc_projection(M0, false),
        % DERP! Check for race with manager's proxy vs. proj listener
        case ?MGR:test_read_latest_public_projection(M0, false) of
            {error, partition} -> timer:sleep(500);
            _                  -> ok
        end,
        {local_write_result, ok,
         {remote_write_results, [{b,ok},{c,ok}]}} =
            ?MGR:test_write_public_projection(M0, P1),
        {unanimous, P1, Extra1} = ?MGR:test_read_latest_public_projection(M0, false),

        ok
    after
        ok = ?MGR:stop(M0),
        [ok = machi_flu1:stop(X) || X <- FLUs],
        ok = machi_partition_simulator:stop()
    end.

nonunanimous_setup_and_fix_test() ->
    machi_partition_simulator:start_link({1,2,3}, 100, 0),
    TcpPort = 62877,
    FluInfo = [{a,TcpPort+0,"./data.a"}, {b,TcpPort+1,"./data.b"}],
    P_s = [#p_srvr{name=Name, address="localhost", port=Port} ||
              {Name,Port,_Dir} <- FluInfo],
    
    [machi_flu1_test:clean_up_data_dir(Dir) || {_,_,Dir} <- FluInfo],
    FLUs = [element(2, machi_flu1:start_link([{Name,Port,Dir}])) ||
               {Name,Port,Dir} <- FluInfo],
    [Proxy_a, Proxy_b] = Proxies =
        [element(2,?FLU_PC:start_link(P)) || P <- P_s],
    MembersDict = machi_projection:make_members_dict(P_s),
    XX = [],
    %% XX = [{private_write_verbose,true}],
    {ok, Ma} = ?MGR:start_link(a, MembersDict, [{active_mode, false}]++XX),
    {ok, Mb} = ?MGR:start_link(b, MembersDict, [{active_mode, false}]++XX),
    try
        {ok, P1} = ?MGR:test_calc_projection(Ma, false),

        P1a = machi_projection:update_checksum(
                 P1#projection_v1{down=[b], upi=[a], dbg=[{hackhack, ?LINE}]}),
        P1b = machi_projection:update_checksum(
                 P1#projection_v1{author_server=b, creation_time=now(),
                                  down=[a], upi=[b], dbg=[{hackhack, ?LINE}]}),
        %% Scribble different projections
        ok = ?FLU_PC:write_projection(Proxy_a, public, P1a),
        ok = ?FLU_PC:write_projection(Proxy_b, public, P1b),

        %% ?D(x),
        {not_unanimous,_,_}=_XX = ?MGR:test_read_latest_public_projection(
                                     Ma, false),
        %% ?Dw(_XX),
        {not_unanimous,_,_}=_YY = ?MGR:test_read_latest_public_projection(
                                     Ma, true),
        %% The read repair here doesn't automatically trigger the creation of
        %% a new projection (to try to create a unanimous projection).  So
        %% we expect nothing to change when called again.
        {not_unanimous,_,_}=_YY = ?MGR:test_read_latest_public_projection(
                                     Ma, true),

        {now_using, _, EpochNum_a} = ?MGR:test_react_to_env(Ma),
        {no_change, _, EpochNum_a} = ?MGR:test_react_to_env(Ma),
        {unanimous,P2,_E2} = ?MGR:test_read_latest_public_projection(Ma, false),
        {ok, P2pa} = ?FLU_PC:read_latest_projection(Proxy_a, private),
        P2 = P2pa#projection_v1{dbg2=[]},

        %% FLUb should have nothing written to private because it hasn't
        %% reacted yet.
        {error, not_written} = ?FLU_PC:read_latest_projection(Proxy_b, private),

        %% Poke FLUb to react ... should be using the same private proj
        %% as FLUa.
        {now_using, _, EpochNum_a} = ?MGR:test_react_to_env(Mb),
        {ok, P2pb} = ?FLU_PC:read_latest_projection(Proxy_b, private),
        P2 = P2pb#projection_v1{dbg2=[]},

        ok
    after
        ok = ?MGR:stop(Ma),
        ok = ?MGR:stop(Mb),
        [ok = ?FLU_PC:quit(X) || X <- Proxies],
        [ok = machi_flu1:stop(X) || X <- FLUs],
        ok = machi_partition_simulator:stop()
    end.

foo_TODO_UNFINISHED_KEEP_PERHAPS_unanimous_report_test() ->
    TcpPort = 63877,
    FluInfo = [{a,TcpPort+0,"./data.a"}, {b,TcpPort+1,"./data.b"}],
    P_s = [#p_srvr{name=Name, address="localhost", port=Port} ||
              {Name,Port,_Dir} <- FluInfo],
    MembersDict = machi_projection:make_members_dict(P_s),

    E5 = 5,
    UPI5 = [a,b],
    Rep5 = [],
    P5 = machi_projection:new(E5, a, MembersDict, [], UPI5, Rep5, []),
    {ok_disjoint, [{agreed_membership, {UPI5, Rep5}}]} =
        unanimous_report2(E5, [{a, P5}, {b, P5}]),
    {ok_disjoint, [{not_agreed, _}]} =
        unanimous_report2(E5, [{a, not_in_this_epoch}, {b, P5}]),
    {ok_disjoint, [{not_agreed, _}]} =
        unanimous_report2(E5, [{a, P5}, {b, not_in_this_epoch}]),

    P5_other_csum = machi_projection:update_checksum(
                    P5#projection_v1{author_server=b}),
    {ok_disjoint, [{not_agreed, _}]} =
        unanimous_report2(E5, [{a, P5}, {b, P5_other_csum}]),

    UPI5_b = [a],
    Rep5_b = [b],
    P5_b = machi_projection:new(E5, b, MembersDict, [], UPI5_b, Rep5_b, []),
    XX =
        unanimous_report2(E5, [{a, P5}, {b, P5_b}]),

    io:format(user, "\nXX ~p\n", [XX]).

-endif. % !PULSE
-endif. % TEST