Add ready to start backend callback

This will be used, for example, to indicate that Riak KV ensemble peers
should wait for the riak_kv service to be up before starting to avoid
some nasty races.

src/riak_ensemble_backend.erl

@@ -116,6 +116,8 @@
 -callback handle_down(reference(), pid(), term(), state()) -> false |
                                                               {ok, state()} |
                                                               {reset, state()}.
+%% Callback used to determine if peers using this backend can be started.
+-callback ready_to_start() -> boolean().
 
 %%===================================================================
 

src/riak_ensemble_basic_backend.erl

@@ -33,7 +33,7 @@
 -export([init/3, new_obj/4]).
 -export([obj_epoch/1, obj_seq/1, obj_key/1, obj_value/1]).
 -export([set_obj_epoch/2, set_obj_seq/2, set_obj_value/2]).
--export([get/3, put/4, tick/5, ping/2]).
+-export([get/3, put/4, tick/5, ping/2, ready_to_start/0]).
 -export([trusted/1, sync_request/2, sync/2]).
 -export([handle_down/4]).
 
@@ -157,6 +157,9 @@ ping(_From, State) ->
 trusted(_State) ->
     false.
 
+ready_to_start() ->
+    true.
+
 %%===================================================================
 
 -spec handle_down(reference(), pid(), term(), state()) -> false.

src/riak_ensemble_manager.erl

@@ -615,8 +615,14 @@ state_changed(State=#state{ensemble_data=EnsData, cluster_state=CS}) ->
     _ = [case Change of
              {add, {Ensemble, Id}, Info} ->
                  #ensemble_info{mod=Mod, args=Args} = Info,
-                 %% Start new peer
-                 _ = riak_ensemble_peer_sup:start_peer(Mod, Ensemble, Id, Args);
+                 %% Maybe start new peer
+                 case Mod:ready_to_start() of
+                     true ->
+                         riak_ensemble_peer_sup:start_peer(Mod, Ensemble,
+                                                           Id, Args);
+                     _ ->
+                         ok
+                 end;
              {del, {Ensemble, Id}} ->
                  %% Stop running peer
                  %% io:format("Should stop: ~p~n", [{Ensemble, Id}]),