Merge partial work of slf/chain-manager/cp-mode4 into tmp-mergeit

.gitignore

@@ -10,3 +10,9 @@ edoc
 
 # PB artifacts for Erlang
 include/machi_pb.hrl
+
+# Misc Scott cruft
+*.patch
+current_counterexample.eqc
+foo*
+typescript*

src/machi_flu1.erl

@@ -837,7 +837,7 @@ make_listener_regname(BaseName) ->
 %% registers.
 
 make_projection_server_regname(BaseName) ->
-    list_to_atom(atom_to_list(BaseName) ++ "_pstore2").
+    list_to_atom(atom_to_list(BaseName) ++ "_pstore").
 
 %% @doc Encode `Offset + Size + TaggedCSum' into an `iolist()' type for
 %% internal storage by the FLU.

src/machi_projection.erl

@@ -31,7 +31,8 @@
          compare/2,
          get_epoch_id/1,
          make_summary/1,
-         make_members_dict/1
+         make_members_dict/1,
+         make_epoch_id/1
         ]).
 
 %% @doc Create a new projection record.
@@ -110,8 +111,26 @@ new(EpochNum, MyName, [] = _MembersDict0, _Down_list, _UPI_list,_Repairing_list,
 %% @doc Update the checksum element of a projection record.
 
 update_checksum(P) ->
+    %% Fields that we ignore when calculating checksum:
+    %% * epoch_csum
+    %% * dbg2: humming consensus participants may modify this at will without
+    %%         voiding the identity of the projection as a whole.
+    %% * flap: In some cases in CP mode, coode upstream of C120 may have
+    %%         updated the flapping information.  That's OK enough: we aren't
+    %%         going to violate chain replication safety rules (or
+    %%         accidentally encourage someone else sometime later) by
+    %%         replacing flapping information with our own local view at
+    %%         this instant in time.
+    %% * creation_time: With CP mode & inner projections, it's damn annoying
+    %%                  to have to copy this around 100% correctly.  {sigh}
+    %%                  That's a negative state of the code.  However, there
+    %%                  isn't a safety violation if the creation_time is
+    %%                  altered for any reason: it's there only for human
+    %%                  benefit for debugging.
     CSum = crypto:hash(sha,
                        term_to_binary(P#projection_v1{epoch_csum= <<>>,
+                                                      creation_time=undefined,
+                                                      flap=undefined,
                                                       dbg2=[]})),
     P#projection_v1{epoch_csum=CSum}.
 
@@ -146,6 +165,7 @@ get_epoch_id(#projection_v1{epoch_number=Epoch, epoch_csum=CSum}) ->
 %% @doc Create a proplist-style summary of a projection record.
 
 make_summary(#projection_v1{epoch_number=EpochNum,
+                            epoch_csum= <<_CSum4:4/binary, _/binary>>,
                             all_members=_All_list,
                             mode=CMode,
                             witnesses=Witness_list,
@@ -161,8 +181,8 @@ make_summary(#projection_v1{epoch_number=EpochNum,
                    true ->
                         []
                 end,
-    [{epoch,EpochNum},{author,Author},
+    [{epoch,EpochNum}, {csum,_CSum4},
-     {mode,CMode},{witnesses, Witness_list},
+     {author,Author}, {mode,CMode},{witnesses, Witness_list},
      {upi,UPI_list},{repair,Repairing_list},{down,Down_list}] ++
         InnerInfo ++
         [{flap, Flap}] ++
@@ -201,3 +221,6 @@ make_members_dict(Ps) ->
                     exit({badarg, {make_members_dict, lists:filter(F_neither, Ps)}})
             end
     end.
+
+make_epoch_id(#projection_v1{epoch_number=Epoch, epoch_csum=CSum}) ->
+    {Epoch, CSum}.

src/machi_projection_store.erl

@@ -59,7 +59,8 @@
          get_all_projections/2, get_all_projections/3,
          list_all_projections/2, list_all_projections/3
         ]).
--export([set_wedge_notify_pid/2]).
+-export([set_wedge_notify_pid/2, get_wedge_notify_pid/1,
+         set_consistency_mode/2]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -72,7 +73,8 @@
           private_dir = ""       :: string(),
           wedge_notify_pid       :: pid() | atom(),
           max_public_epochid =  ?NO_EPOCH :: {-1 | non_neg_integer(), binary()},
-          max_private_epochid = ?NO_EPOCH :: {-1 | non_neg_integer(), binary()}
+          max_private_epochid = ?NO_EPOCH :: {-1 | non_neg_integer(), binary()},
+          consistency_mode=ap_mode :: 'ap_mode' | 'cp_mode'
          }).
 
 %% @doc Start a new projection store server.
@@ -159,7 +161,16 @@ list_all_projections(PidSpec, ProjType, Timeout)
     g_call(PidSpec, {list_all_projections, ProjType}, Timeout).
 
 set_wedge_notify_pid(PidSpec, NotifyWedgeStateChanges) ->
-    gen_server:call(PidSpec, {set_wedge_notify_pid, NotifyWedgeStateChanges}).
+    gen_server:call(PidSpec, {set_wedge_notify_pid, NotifyWedgeStateChanges},
+                    infinity).
+
+get_wedge_notify_pid(PidSpec) ->
+    gen_server:call(PidSpec, {get_wedge_notify_pid},
+                    infinity).
+
+set_consistency_mode(PidSpec, CMode)
+  when CMode == ap_mode; CMode == cp_mode ->
+    gen_server:call(PidSpec, {set_consistency_mode, CMode}, infinity).
 
 %%%%%%%%%%%%%%%%%%%%%%%%%%%
 
@@ -224,6 +235,10 @@ handle_call({{list_all_projections, ProjType}, LC1}, _From, S) ->
     {reply, {{ok, find_all(Dir)}, LC2}, S};
 handle_call({set_wedge_notify_pid, NotifyWedgeStateChanges}, _From, S) ->
     {reply, ok, S#state{wedge_notify_pid=NotifyWedgeStateChanges}};
+handle_call({get_wedge_notify_pid}, _From, S) ->
+    {reply, {ok, S#state.wedge_notify_pid}, S};
+handle_call({set_consistency_mode, CMode}, _From, S) ->
+    {reply, ok, S#state{consistency_mode=CMode}};
 handle_call(_Request, _From, S) ->
     Reply = {whaaaaaaaaaaaaazz, _Request},
     {reply, Reply, S}.
@@ -261,62 +276,90 @@ do_proj_read(ProjType, Epoch, S_or_Dir) ->
             {{error, Else}, S_or_Dir}
     end.
 
-do_proj_write(public=ProjType, Proj, S) ->
+do_proj_write(ProjType, Proj, S) ->
-    do_proj_write2(ProjType, Proj, S);
+    do_proj_write2(ProjType, Proj, S).
-do_proj_write(private=ProjType, #projection_v1{epoch_number=Epoch}=Proj, S) ->
+
-    case S#state.max_public_epochid of
+do_proj_write2(ProjType, #projection_v1{epoch_csum=CSum}=Proj, S) ->
-        {PublicEpoch, _} when PublicEpoch =< Epoch ->
+    case (machi_projection:update_checksum(Proj))#projection_v1.epoch_csum of
-            do_proj_write2(ProjType, Proj, S);
+        CSum2 when CSum2 == CSum ->
-        {PublicEpoch, _} ->
+            do_proj_write3(ProjType, Proj, S);
+        _Else ->
             {{error, bad_arg}, S}
     end.
 
-do_proj_write2(ProjType, #projection_v1{epoch_number=Epoch}=Proj, S) ->
+do_proj_write3(ProjType, #projection_v1{epoch_number=Epoch,
+                                        epoch_csum=CSum}=Proj, S) ->
     %% TODO: We probably ought to check the projection checksum for sanity, eh?
     Dir = pick_path(ProjType, S),
     Path = filename:join(Dir, epoch2name(Epoch)),
-    case file:read_file_info(Path) of
+    case file:read_file(Path) of
-        {ok, _FI} ->
+        {ok, _Bin} when ProjType == public ->
             {{error, written}, S};
+        {ok, Bin} when ProjType == private ->
+            #projection_v1{epoch_number=CurEpoch,
+                           epoch_csum=CurCSum} = _CurProj = binary_to_term(Bin),
+            %% We've already checked that CSum is correct matches the
+            %% contents of this new projection version.  If the epoch_csum
+            %% values match, and if we trust the value on disk (TODO paranoid
+            %% check that, also), then the only difference must be the dbg2
+            %% list, which is ok.
+            if CurEpoch == Epoch, CurCSum == CSum ->
+                    do_proj_write4(ProjType, Proj, Path, Epoch, S);
+               true ->
+                    %% io:format(user, "OUCH: on disk: ~w\n", [machi_projection:make_summary(binary_to_term(Bin))]),
+                    %% io:format(user, "OUCH: clobber: ~w\n", [machi_projection:make_summary(Proj)]),
+                    %% io:format(user, "OUCH: clobber: ~p\n", [Proj#projection_v1.dbg2]),
+                    %% {{error, written, CurEpoch, Epoch, CurCSum, CSum}, S}
+                    {{error, written}, S}
+            end;
         {error, enoent} ->
-            {ok, FH} = file:open(Path, [write, raw, binary]),
+            do_proj_write4(ProjType, Proj, Path, Epoch, S);
-            ok = file:write(FH, term_to_binary(Proj)),
-            ok = file:sync(FH),
-            ok = file:close(FH),
-            EffectiveProj = machi_chain_manager1:inner_projection_or_self(Proj),
-            EffectiveEpoch = EffectiveProj#projection_v1.epoch_number,
-            EpochId = {Epoch, Proj#projection_v1.epoch_csum},
-            EffectiveEpochId = {EffectiveEpoch, EffectiveProj#projection_v1.epoch_csum},
-            %% 
-            NewS = if ProjType == public,
-                      Epoch > element(1, S#state.max_public_epochid) ->
-                           if Epoch == EffectiveEpoch ->
-                                   %% This is a regular projection, i.e.,
-                                   %% does not have an inner proj.
-                                   update_wedge_state(
-                                     S#state.wedge_notify_pid, true,
-                                     EffectiveEpochId);
-                              Epoch /= EffectiveEpoch ->
-                                   %% This projection has an inner proj.
-                                   %% The outer proj is flapping, so we do
-                                   %% not bother wedging.
-                                   ok
-                           end,
-                           S#state{max_public_epochid=EpochId};
-                      ProjType == private,
-                      Epoch > element(1, S#state.max_private_epochid) ->
-                           update_wedge_state(
-                                 S#state.wedge_notify_pid, false,
-                             EffectiveEpochId),
-                           S#state{max_private_epochid=EpochId};
-                      true ->
-                           S
-                   end,
-            {ok, NewS};
         {error, Else} ->
             {{error, Else}, S}
     end.
 
+do_proj_write4(ProjType, Proj, Path, Epoch, #state{consistency_mode=CMode}=S) ->
+    {ok, FH} = file:open(Path, [write, raw, binary]),
+    ok = file:write(FH, term_to_binary(Proj)),
+    ok = file:sync(FH),
+    ok = file:close(FH),
+    EffectiveProj = machi_chain_manager1:inner_projection_or_self(Proj),
+    EffectiveEpoch = EffectiveProj#projection_v1.epoch_number,
+    EpochId = {Epoch, Proj#projection_v1.epoch_csum},
+    EffectiveEpochId = {EffectiveEpoch, EffectiveProj#projection_v1.epoch_csum},
+    %%
+    NewS = if ProjType == public,
+              Epoch > element(1, S#state.max_public_epochid) ->
+                   if Epoch == EffectiveEpoch ->
+                           %% This is a regular projection, i.e.,
+                           %% does not have an inner proj.
+                           update_wedge_state(
+                             S#state.wedge_notify_pid, true,
+                             EffectiveEpochId);
+                      Epoch /= EffectiveEpoch ->
+                           %% This projection has an inner proj.
+                           %% The outer proj is flapping, so we do
+                           %% not bother wedging.
+                           ok
+                   end,
+                   S#state{max_public_epochid=EpochId};
+              ProjType == private,
+              Epoch > element(1, S#state.max_private_epochid) ->
+                   if CMode == ap_mode ->
+                           update_wedge_state(
+                             S#state.wedge_notify_pid, false,
+                             EffectiveEpochId);
+                      true ->
+                           %% If ProjType == private and CMode == cp_mode, then
+                           %% the unwedge action is not performed here!
+                           ok
+                   end,
+                   S#state{max_private_epochid=EpochId};
+              true ->
+                   S
+           end,
+    {ok, NewS}.
+
 update_wedge_state(PidSpec, Boolean, {0,_}=EpochId) ->
     %% Epoch #0 is a special case: no projection has been written yet.
     %% However, given the way that machi_flu_psup starts the

test/machi_chain_manager1_converge_demo.erl

@@ -239,6 +239,14 @@ convergence_demo_testfun(NumFLUs, MgrOpts0) ->
 
       %% machi_partition_simulator:reset_thresholds(10, 50),
       %% io:format(user, "\nLet loose the dogs of war!\n", []),
+      %% [DoIt(20, 0, 0) || _ <- lists:seq(1,9)],
+      %% %% io:format(user, "\nVariations of puppies and dogs of war!\n", []),
+      %% %% [begin
+      %% %%      machi_partition_simulator:reset_thresholds(90, 90),
+      %% %%      DoIt(7, 0, 0),
+      %% %%      machi_partition_simulator:always_these_partitions([]),
+      %% %%      DoIt(7, 0, 0)
+      %% %%  end || _ <- lists:seq(1, 3)],
       machi_partition_simulator:always_these_partitions([]),
       io:format(user, "\nPuppies for everyone!\n", []),
       [DoIt(20, 0, 0) || _ <- lists:seq(1,9)],
@@ -372,12 +380,14 @@ convergence_demo_testfun(NumFLUs, MgrOpts0) ->
 make_partition_list(All_list) ->
     [
      [{b,c}],
-     [],
+     [{a,c},{b,c}]
-     [{c,d}],
+     %% [{b,c}],
-     [],
+     %% [],
-     [{d,e}],
+     %% [{c,d}],
-     [],
+     %% [],
-     [{c,e}]
+     %% [{d,e}],
+     %% [],
+     %% [{c,e}]
     ].
 
     %% _X_Ys1 = [[{X,Y}] || X <- All_list, Y <- All_list, X /= Y],
@@ -542,9 +552,14 @@ todo_why_does_this_crash_sometimes(FLUName, FLU, PPPepoch) ->
     end.
 
 private_projections_are_stable(Namez, PollFunc) ->
-    Private1 = [{Name, get_latest_inner_proj_summ(FLU)} || {Name,FLU} <- Namez],
+    FilterNoneProj = fun({_EpochID,[],[],_Dn,_W,InnerP}) -> false;
+                        (_)                              -> true
+                     end,
+    Private1x = [{Name, get_latest_inner_proj_summ(FLU)} || {Name,FLU} <- Namez],
+    Private1 = [X || X={_,Proj} <- Private1x, FilterNoneProj(Proj)],
     [PollFunc(15, 1, 10) || _ <- lists:seq(1,6)],
-    Private2 = [{Name, get_latest_inner_proj_summ(FLU)} || {Name,FLU} <- Namez],
+    Private2x = [{Name, get_latest_inner_proj_summ(FLU)} || {Name,FLU} <- Namez],
+    Private2 = [X || X={_,Proj} <- Private2x, FilterNoneProj(Proj)],
     %% Is = [Inner_p || {_,_,_,_,Inner_p} <- Private1],
     put(stable, lists:sort(Private1)),
     %% We want either all true or all false (inner or not) ... except
@@ -623,15 +638,22 @@ private_projections_are_stable(Namez, PollFunc) ->
                         io:format(user, "Priv2: EID ~W e ~w u ~w\n", [EpochID, 7, ExpectedFLUs, UsingFLUs]),
                         ordsets:is_subset(ordsets:from_list(ExpectedFLUs),
                                           ordsets:from_list(UsingFLUs));
-                    _Else ->
+                    [{1=_Count,_EpochID}|_] ->
-                        io:format(user, "Priv2: Else ~p\n", [_Else]),
+                        %% Our list is sorted & reversed, so 1=_Count
+                        %% is biggest.  If a majority is using the none proj,
+                        %% then we're OK.
+                        Private2None = [X || {_,{_,[],[],_,_,_}}=X <- Private2],
+                        length(Private2None) >= FullMajority;
+                    Else ->
+                        %% This is bad: we have a count that's less than
+                        %% FullMajority but greater than 1.
                         false
                 end;
            CMode == ap_mode ->
                 true
         end,
 
-    io:format(user, "\nPriv1 ~P\n1==2 ~w ap_disjoint ~w u_all_peers ~w cp_mode_agree ~w\n", [lists:sort(Private1), 20, Private1 == Private2, AP_mode_disjoint_test_p, Unanimous_with_all_peers_p, CP_mode_agree_test_p]),
+    io:format(user, "\nPriv1 ~p\nPriv2 ~p\n1==2 ~w ap_disjoint ~w u_all_peers ~w cp_mode_agree ~w\n", [lists:sort(Private1), lists:sort(Private2), Private1 == Private2, AP_mode_disjoint_test_p, Unanimous_with_all_peers_p, CP_mode_agree_test_p]),
     Private1 == Private2 andalso
         AP_mode_disjoint_test_p andalso
         (
@@ -651,12 +673,12 @@ private_projections_are_stable(Namez, PollFunc) ->
 
 get_latest_inner_proj_summ(FLU) ->
     {ok, Proj} = ?FLU_PC:read_latest_projection(FLU, private),
-    #projection_v1{epoch_number=E, epoch_csum=CSum,
+    #projection_v1{epoch_number=E, epoch_csum= <<CSum4:4/binary, _/binary>>,
                    upi=UPI, repairing=Repairing,
                    witnesses=Witnesses, down=Down} =
         machi_chain_manager1:inner_projection_or_self(Proj),
     Inner_p = machi_chain_manager1:inner_projection_exists(Proj),
-    EpochID = {E, CSum},
+    EpochID = {E, CSum4},
     {EpochID, UPI, Repairing, Down, Witnesses, Inner_p}.
 
 random_sort(L) ->

test/machi_cr_client_test.erl

@@ -205,7 +205,7 @@ witness_smoke_test2() ->
 
         %% Whew ... ok, now start some damn tests.
         {ok, C1} = machi_cr_client:start_link([P || {_,P}<-orddict:to_list(D)]),
-        machi_cr_client:append_chunk(C1, Prefix, Chunk1),
+        {ok, _} = machi_cr_client:append_chunk(C1, Prefix, Chunk1),
         {ok, {Off1,Size1,File1}} =
             machi_cr_client:append_chunk(C1, Prefix, Chunk1),
         Chunk1_badcs = {<<?CSUM_TAG_CLIENT_SHA:8, 0:(8*20)>>, Chunk1},
@@ -215,7 +215,8 @@ witness_smoke_test2() ->
 
         %% Stop 'b' and let the chain reset.
         ok = machi_flu_psup:stop_flu_package(b),
-        run_ticks([a_chmgr,c_chmgr]),
+        %% Run ticks enough times to force auto-unwedge of both a & c.
+        [run_ticks([a_chmgr,c_chmgr]) || _ <- [1,2,3,4] ],
 
         %% The chain should now be [a,c].
         %% Let's wedge OurWitness and see what happens: timeout/partition.

test/machi_flu1_test.erl

@@ -186,7 +186,10 @@ flu_projection_common(Host, TcpPort, T) ->
     P_a = #p_srvr{name=a, address="localhost", port=4321},
     P1 = machi_projection:new(1, a, [P_a], [], [a], [], []),
     ok = ?FLU_C:write_projection(Host, TcpPort, T, P1),
-    {error, written} = ?FLU_C:write_projection(Host, TcpPort, T, P1),
+    case ?FLU_C:write_projection(Host, TcpPort, T, P1) of
+        {error, written} when T == public  -> ok;
+        ok               when T == private -> ok
+    end,
     {ok, P1} = ?FLU_C:read_projection(Host, TcpPort, T, 1),
     {ok, {1,_}} = ?FLU_C:get_latest_epochid(Host, TcpPort, T),
     {ok, P1} = ?FLU_C:read_latest_projection(Host, TcpPort, T),

test/machi_projection_store_test.erl

@@ -0,0 +1,68 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2007-2015 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_projection_store_test).
+
+-ifdef(TEST).
+-ifndef(PULSE).
+
+-compile(export_all).
+-define(PS, machi_projection_store).
+
+-include("machi_projection.hrl").
+
+smoke_test() ->
+    {ok, SupPid} = machi_flu_sup:start_link(),
+    PortBase = 64820,
+    Dir = "./data.a",
+    Os = [{ignore_stability_time, true}, {active_mode, false}],
+    os:cmd("rm -rf " ++ Dir),
+    {ok,Yo}=machi_flu_psup:start_flu_package(a, PortBase, "./data.a", Os),
+
+    try
+        P1 = machi_projection:new(1, a, [], [], [], [], []),
+        ok = ?PS:write(a_pstore, public, P1),
+        {error, written} = ?PS:write(a_pstore, public, P1),
+
+        Pbad = P1#projection_v1{epoch_number=99238}, % break checksum
+        {error, bad_arg} = ?PS:write(a_pstore, public, Pbad),
+
+        ok = ?PS:write(a_pstore, private, P1),
+        P1a = machi_projection:update_checksum(P1#projection_v1{dbg=[diff_yo]}),
+        {error, written} = ?PS:write(a_pstore, private, P1a),
+
+        P1b = P1#projection_v1{dbg2=[version_b]},
+        ok = ?PS:write(a_pstore, private, P1b),
+        P1c = P1#projection_v1{dbg2=[version_c]},
+        ok = ?PS:write(a_pstore, private, P1c),
+        {error, written} = ?PS:write(a_pstore, private, P1a),
+
+        ok = ?PS:set_consistency_mode(a_pstore, ap_mode),
+        ok = ?PS:set_consistency_mode(a_pstore, cp_mode),
+
+        ok
+    after
+        machi_flu_psup:stop_flu_package(a),
+        exit(SupPid, normal),
+        timer:sleep(10)
+    end.
+
+-endif. % !PULSE
+-endif. % TEST

test/machi_proxy_flu1_client_test.erl

@@ -128,7 +128,8 @@ flu_restart_test() ->
                                           infinity),
             P_a = #p_srvr{name=a, address="localhost", port=6622},
             P1   = machi_projection:new(1, a, [P_a], [], [a], [], []),
-            P1xx = P1#projection_v1{dbg2=["not exactly the same as P1!!!"]},
+            P1xx = P1#projection_v1{dbg2=["dbg2 changes are ok"]},
+            P1yy = P1#projection_v1{dbg=["not exactly the same as P1!!!"]},
             EpochID = {P1#projection_v1.epoch_number,
                        P1#projection_v1.epoch_csum},
             ok = ?MUT:write_projection(Prox1, public, P1),
@@ -202,12 +203,19 @@ flu_restart_test() ->
                     (line) -> io:format("line ~p, ", [?LINE]);
                     (stop) -> ?MUT:write_projection(Prox1, public, P1xx)
                  end,
-                 fun(run) -> {error, written} =
+
+                 fun(run) -> ok = %% P1xx is difference only in dbg2
                                  ?MUT:write_projection(Prox1, private, P1xx),
                              ok;
                     (line) -> io:format("line ~p, ", [?LINE]);
                     (stop) -> ?MUT:write_projection(Prox1, private, P1xx)
                  end,
+                 fun(run) -> {error, bad_arg} = % P1yy has got bad checksum
+                                 ?MUT:write_projection(Prox1, private, P1yy),
+                             ok;
+                    (line) -> io:format("line ~p, ", [?LINE]);
+                    (stop) -> ?MUT:write_projection(Prox1, private, P1yy)
+                 end,
 
                  fun(run) -> {ok, [_]} =
                                  ?MUT:get_all_projections(Prox1, public),