Add support for AAE hashing version 0 (#1148)

* Add support for AAE hashing version 0

Additionally, change verify_aae to use 3 nodes for more complete
testing.

* Update repl_aae_fullsync.erl

* Update verify_aae.erl

* Do hashtree upgrade verification first

* Fix intercepts in aae_fullsync test

* Increase node count for more accurate test

intercepts/riak_kv_index_hashtree_intercepts.erl

@@ -32,11 +32,17 @@ delayed_compare(_IndexN, _Remote, _AccFun, _TreePid) ->
 
 %% @doc When attempting to get the lock on a hashtree, return the
 %%      not_built atom which means the tree has not been computed yet.
-not_built(_TreePid, _Type) ->
+not_built(_TreePid, _Type, _Version, _Pid) ->
     not_built.
 
 %% @doc When attempting to get the lock on a hashtree, return the
 %%      already_locked atom which means the tree is locked by another
 %%      process.
-already_locked(_TreePid, _Type) ->
+already_locked(_TreePid, _Type, _Version, _Pid) ->
     already_locked.
+
+%% @doc When attempting to get the lock on a hashtree, return the
+%%      bad_version atom which means the local tree does not match
+%%      the requested version
+bad_version(_TreePid, _Type, _Version, _Pid) ->
+    bad_version.

tests/repl_aae_fullsync.erl

@@ -542,7 +542,7 @@ validate_intercepted_fullsync(InterceptTarget,
     %% not_built to defer fullsync process.
     validate_intercepted_fullsync(InterceptTarget,
                                   {riak_kv_index_hashtree,
-                                   [{{get_lock, 2}, not_built}]},
+                                   [{{get_lock, 4}, not_built}]},
                                   ReplicationLeader,
                                   ReplicationCluster,
                                   NumIndicies),
@@ -551,7 +551,16 @@ validate_intercepted_fullsync(InterceptTarget,
     %% already_locked to defer fullsync process.
     validate_intercepted_fullsync(InterceptTarget,
                                   {riak_kv_index_hashtree,
-                                   [{{get_lock, 2}, already_locked}]},
+                                   [{{get_lock, 4}, already_locked}]},
+                                  ReplicationLeader,
+                                  ReplicationCluster,
+                                  NumIndicies),
+
+    %% Before enabling fullsync, ensure trees on one source node return
+    %% bad_version to defer fullsync process.
+    validate_intercepted_fullsync(InterceptTarget,
+                                  {riak_kv_index_hashtree,
+                                   [{{get_lock, 4}, bad_version}]},
                                   ReplicationLeader,
                                   ReplicationCluster,
                                   NumIndicies),

tests/verify_aae.erl

@@ -60,7 +60,7 @@
            {ring_creation_size, ?DEFAULT_RING_SIZE}
           ]}]
        ).
--define(NUM_NODES, 1).
+-define(NUM_NODES, 3).
 -define(NUM_KEYS, 1000).
 -define(BUCKET, <<"test_bucket">>).
 -define(N_VAL, 3).
@@ -88,7 +88,11 @@ verify_throttle_config(Nodes) ->
 
 verify_aae(Nodes) ->
     Node1 = hd(Nodes),
-    % First, recovery without tree rebuilds
+
+    % Verify that AAE eventually upgrades to version 0(or already has)
+    wait_until_hashtree_upgrade(Nodes), 
+
+    % Recovery without tree rebuilds
 
     % Test recovery from to few replicas written
     KV1 = test_data(1, 1000),
@@ -101,7 +105,7 @@ verify_aae(Nodes) ->
     lager:info("Run similar tests now with tree rebuilds enabled"),
     start_tree_rebuilds(Nodes),
 
-    % Test recovery from to few replicas written
+    % Test recovery from too few replicas written
     KV3 = test_data(1001, 2000),
     test_less_than_n_writes(Node1, KV3),
 
@@ -320,3 +324,26 @@ max_aae_repairs(Node) when is_atom(Node) ->
     LastCounts = [Last || {_, _, _, {Last, _, _, _}} <- Info],
     MaxCount = lists:max(LastCounts),
     MaxCount.
+
+wait_until_hashtree_upgrade(Nodes) ->
+    lager:info("Verifying AAE hashtrees eventually all upgrade to version 0"),
+    rt:wait_until(fun() -> all_hashtrees_upgraded(Nodes) end).
+
+all_hashtrees_upgraded(Nodes) when is_list(Nodes) ->
+    [Check|_] = lists:usort([all_hashtrees_upgraded(Node) || Node <- Nodes]),
+    Check;
+
+all_hashtrees_upgraded(Node) when is_atom(Node) ->
+    case rpc:call(Node, riak_kv_entropy_manager, get_version, []) of
+        0 ->
+            Trees = rpc:call(Node, riak_kv_entropy_manager, get_trees_version, []),
+            case [Idx || {Idx, undefined} <- Trees] of
+                [] ->
+                    true;
+                _ ->
+                    false
+            end;
+        _ ->
+            false
+     end.
+