Merge branch 'develop-2.2' into develop

Author: Ian Milligan

rebar.config

@@ -9,11 +9,11 @@
 {xref_queries, [{"(XC - UC) || (XU - X - B - cluster_info : Mod)", []}]}.
 
 {deps, [
-        {lager, "2.0.3", {git, "git://github.com/basho/lager.git", {tag, "2.0.3"}}},
+        {lager, ".*", {git, "git://github.com/basho/lager.git", {tag, "3.2.2"}}},
         {ranch, "0.4.0-p1", {git, "git://github.com/basho/ranch.git", {tag, "0.4.0-p1"}}},
         {ebloom, ".*", {git, "git://github.com/basho/ebloom.git", {tag, "2.0.0"}}},
-        {riak_kv, ".*", {git, "git://github.com/basho/riak_kv.git", {branch, "develop"}}},
-        {riak_repl_pb_api, ".*", {git, "[email protected]:basho/riak_repl_pb_api.git", {branch, "develop"}}}
+        {riak_kv, ".*", {git, "git://github.com/basho/riak_kv.git", {tag, "2.1.3"}}},
+        {riak_repl_pb_api, ".*", {git, "[email protected]:basho/riak_repl_pb_api.git", {tag, "2.4.0"}}}
        ]}.
 
 {edoc_opts, [{preprocess, true}]}.

src/riak_core_connection_mgr.erl

@@ -584,14 +584,17 @@ locate_endpoints({Type, Name}, Strategy, Locators) ->
 %% our book keeping for that endpoint. Black-list it, and
 %% adjust a backoff timer so that we wait a while before
 %% trying this endpoint again.
+
+-spec fail_endpoint(ip_addr(), term(), proto_id(), #state{}) -> #state{}.
 fail_endpoint(Addr, Reason, ProtocolId, State) ->
     %% update the stats module
-    Stat = {conn_error, Reason},
+    Err = reason_to_atom(Reason),
+    Stat = {conn_error, Err},
     riak_core_connection_mgr_stats:update(Stat, Addr, ProtocolId),
     %% update the endpoint
     Fun = fun(EP=#ep{backoff_delay = Backoff, failures = Failures}) ->
                   erlang:send_after(Backoff, self(), {backoff_timer, Addr}),
-                  EP#ep{failures = orddict:update_counter(Reason, 1, Failures),
+                  EP#ep{failures = orddict:update_counter(Err, 1, Failures),
                         nb_failures = EP#ep.nb_failures + 1,
                         backoff_delay = increase_backoff(Backoff),
                         last_fail_time = os:timestamp(),
@@ -600,6 +603,18 @@ fail_endpoint(Addr, Reason, ProtocolId, State) ->
           end,
     update_endpoint(Addr, Fun, State).
 
+%% Attempt to extract atom from an error reason.
+
+-spec reason_to_atom(term()) -> atom().
+reason_to_atom({{Err, _Val}, _Stack}) when is_atom(Err) ->
+    Err;
+reason_to_atom({Err, _Stack}) when is_atom(Err) ->
+    Err;
+reason_to_atom(Reason) when is_atom(Reason) ->
+    Reason;
+reason_to_atom(_Reason) ->
+    unknown_reason.
+
 connect_endpoint(Addr, State) ->
     update_endpoint(Addr, fun(EP) ->
                                   EP#ep{is_black_listed = false,

src/riak_repl2_fscoordinator.erl

@@ -56,7 +56,6 @@
     socket,
     transport,
     largest_n,
-    owners = [],
     connection_ref,
     partition_queue = queue:new(),
     retries = dict:new(),
@@ -361,7 +360,6 @@ handle_cast(start_fullsync,  State) ->
             Partitions = sort_partitions(Ring),
             State2 = State#state{
                 largest_n = N,
-                owners = riak_core_ring:all_owners(Ring),
                 partition_queue = queue:from_list(Partitions),
                 retries = dict:new(),
                 reserve_retries = dict:new(),
@@ -389,7 +387,6 @@ handle_cast(stop_fullsync, State) ->
                 <- State#state.running_sources],
     State2 = State#state{
         largest_n = undefined,
-        owners = [],
         partition_queue = queue:new(),
         retries = dict:new(),
         reserve_retries = dict:new(),
@@ -770,26 +767,26 @@ send_next_whereis_req(State) ->
 % two specs:  is the local node available, and does our cache of remote nodes
 % say the remote node is available.
 determine_best_partition(State) ->
-    #state{partition_queue = Queue, busy_nodes = Busies, owners = Owners, whereis_waiting = Waiting} = State,
+    #state{partition_queue = Queue, busy_nodes = Busies, whereis_waiting = Waiting} = State,
     SeedPart = queue:out(Queue),
     lager:debug("Starting partition search"),
-    determine_best_partition(SeedPart, Busies, Owners, Waiting, queue:new()).
+    determine_best_partition(SeedPart, Busies, Waiting, queue:new()).
 
-determine_best_partition({empty, _Q}, _Business, _Owners, _Waiting, AccQ) ->
+determine_best_partition({empty, _Q}, _Business, _Waiting, AccQ) ->
     lager:debug("No partition in the queue that will not exceed a limit; will try again later."),
     % there is no best partition, try again later
     {undefined, AccQ};
 
-determine_best_partition({{value, Part}, Queue}, Busies, Owners, Waiting, AccQ) ->
-    case node_available(Part, Owners, Waiting) of
+determine_best_partition({{value, Part}, Queue}, Busies, Waiting, AccQ) ->
+    case node_available(Part, Waiting) of
         false ->
-            determine_best_partition(queue:out(Queue), Busies, Owners, Waiting, queue:in(Part, AccQ));
+            determine_best_partition(queue:out(Queue), Busies, Waiting, queue:in(Part, AccQ));
         skip ->
-            determine_best_partition(queue:out(Queue), Busies, Owners, Waiting, AccQ);
+            determine_best_partition(queue:out(Queue), Busies, Waiting, AccQ);
         true ->
             case remote_node_available(Part, Busies) of
                 false ->
-                    determine_best_partition(queue:out(Queue), Busies, Owners, Waiting, queue:in(Part, AccQ));
+                    determine_best_partition(queue:out(Queue), Busies, Waiting, queue:in(Part, AccQ));
                 true ->
                     {Part, queue:join(Queue, AccQ)}
             end
@@ -801,8 +798,10 @@ below_max_sources(State) ->
     Max = app_helper:get_env(riak_repl, max_fssource_cluster, ?DEFAULT_SOURCE_PER_CLUSTER),
     ( length(State#state.running_sources) + length(State#state.whereis_waiting) ) < Max.
 
-node_available(PartitionInfo, Owners, Waiting) ->
+node_available(PartitionInfo, Waiting) ->
     Partition = PartitionInfo#partition_info.index,
+    {ok, Ring} = riak_core_ring_manager:get_my_ring(),
+    Owners = riak_core_ring:all_owners(Ring),
     LocalNode = proplists:get_value(Partition, Owners),
     Max = app_helper:get_env(riak_repl, max_fssource_node, ?DEFAULT_SOURCE_PER_NODE),
     try riak_repl2_fssource_sup:enabled(LocalNode) of
@@ -834,8 +833,8 @@ remote_node_available(Partition, Busies) ->
 
 start_fssource(PartitionVal, Ip, Port, State) ->
     Partition = PartitionVal#partition_info.index,
-    #state{owners = Owners} = State,
-    LocalNode = proplists:get_value(Partition, Owners),
+    {ok, Ring} = riak_core_ring_manager:get_my_ring(),
+    LocalNode = riak_core_ring:index_owner(Ring, Partition),
     case riak_repl2_fssource_sup:enable(LocalNode, Partition, {Ip, Port}) of
         {ok, Pid} ->
             link(Pid),

src/riak_repl2_fssink.erl

@@ -148,8 +148,9 @@ handle_info({Proto, Socket, Data},
         State=#state{socket=Socket,transport=Transport}) when Proto==tcp; Proto==ssl ->
     %% aae strategy will not receive messages here
     Transport:setopts(Socket, [{active, once}]),
-    case decode_obj_msg(Data) of
-        {fs_diff_obj, RObj} ->
+    case binary_to_term(Data) of
+        {fs_diff_obj, BinObj} ->
+            RObj = riak_repl_util:decode_bin_obj(BinObj),
             riak_repl_util:do_repl_put(RObj);
         Other ->
             gen_fsm:send_event(State#state.fullsync_worker, Other)
@@ -185,18 +186,6 @@ handle_info(init_ack, State=#state{socket=Socket,
 handle_info(_Msg, State) ->
     {noreply, State}.
 
-decode_obj_msg(Data) ->
-    Msg = binary_to_term(Data),
-    case Msg of
-        {fs_diff_obj, BObj} when is_binary(BObj) ->
-            RObj = riak_repl_util:from_wire(BObj),
-            {fs_diff_obj, RObj};
-        {fs_diff_obj, _RObj} ->
-            Msg;
-        Other ->
-            Other
-    end.
-
 terminate(_Reason, #state{fullsync_worker=FSW, work_dir=WorkDir, strategy=Strategy}) ->
     %% TODO: define a fullsync worker behavior and call it's stop function
     case is_pid(FSW) andalso is_process_alive(FSW) of

src/riak_repl2_pg_block_provider.erl

@@ -16,7 +16,7 @@
     terminate/2, code_change/3, status/1, status/2]).
 
 %% send a message every KEEPALIVE milliseconds to make sure the service is running on the sink
--define(KEEPALIVE, 1000).
+-define(KEEPALIVE, 60000).
 
 -record(state,
         {
@@ -27,7 +27,6 @@
           connection_ref,
           worker,
           client,
-          keepalive_timer,
           proxy_gets_provided = 0
         }).
 
@@ -75,16 +74,16 @@ handle_call({connected, Socket, Transport, _Endpoint, _Proto, Props}, _From,
     lager:debug("Keeping stats for " ++ SocketTag),
     riak_core_tcp_mon:monitor(Socket, {?TCP_MON_PROXYGET_APP, source,
                                        SocketTag}, Transport),
-    TRef = keepalive_timer(),
     Transport:setopts(Socket, [{active, once}]),
     {ok, Client} = riak:local_client(),
-    {reply, ok, State#state{
+    State2 = State#state{
                   transport=Transport,
                   socket=Socket,
                   other_cluster=Cluster,
-                  client=Client,
-                  keepalive_timer=TRef
-                 }};
+                  client=Client
+    },
+    _ = keepalive_timer(),
+    {reply, ok, State2};
 handle_call(status, _From, State=#state{socket=Socket,
                                         proxy_gets_provided=PGCount}) ->
     SocketStats = riak_core_tcp_mon:socket_status(Socket),
@@ -109,6 +108,7 @@ handle_cast(_Msg, State) ->
 handle_info(keepalive, State=#state{socket=Socket, transport=Transport}) ->
     Data = term_to_binary(stay_awake),
     Transport:send(Socket, Data),
+    _ = keepalive_timer(),
     {noreply, State};
 handle_info({tcp_closed, Socket}, State=#state{socket=Socket}) ->
     lager:info("Connection for proxy_get ~p closed", [State#state.other_cluster]),
@@ -125,14 +125,11 @@ handle_info({ssl_error, _Socket, Reason}, State) ->
         [State#state.other_cluster, Reason]),
     {stop, socket_closed, State};
 handle_info({Proto, Socket, Data},
-            State0=#state{socket=Socket,transport=Transport, keepalive_timer=TRef})
+            State0=#state{socket=Socket,transport=Transport})
         when Proto==tcp; Proto==ssl ->
     Transport:setopts(Socket, [{active, once}]),
-    _ = timer:cancel(TRef),
     Msg = binary_to_term(Data),
-    %% restart the timer after each message has been processed
-    State = State0#state{keepalive_timer=keepalive_timer()},
-    handle_msg(Msg, State);
+    handle_msg(Msg, State0);
 handle_info(_Msg, State) ->
     {noreply, State}.
 
@@ -161,5 +158,4 @@ code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
 keepalive_timer() ->
-    timer:send_interval(?KEEPALIVE, keepalive).
-
+    erlang:send_after(?KEEPALIVE, self(), keepalive).

src/riak_repl2_rtq.erl

@@ -453,12 +453,8 @@ terminate(Reason, #state{cs = Cs}) ->
     %% when started from tests, we may not be registered
     catch(erlang:unregister(?SERVER)),
     flush_pending_pushes(),
-    _ = [case DeliverFun of
-         undefined ->
-             ok;
-         _ ->
-            catch(DeliverFun({error, {terminate, Reason}}))
-     end || #c{deliver = DeliverFun} <- Cs],
+    _ = [deliver_error(DeliverFun, {terminate, Reason}) ||
+	    #c{deliver = DeliverFun} <- Cs],
     ok.
 
 %% @private
@@ -502,8 +498,8 @@ unregister_q(Name, State = #state{qtab = QTab, cs = Cs}) ->
             case C#c.deliver of
                 undefined ->
                     ok;
-                Deliver ->
-                    Deliver({error, unregistered})
+                DeliverFun ->
+                    deliver_error(DeliverFun, {error, unregistered})
             end,
             MinSeq = case Cs2 of
                          [] ->
@@ -551,8 +547,8 @@ pull(Name, DeliverFun, State = #state{qtab = QTab, qseq = QSeq, cs = Cs}) ->
                 {value, C, Cs2} ->
                     [maybe_pull(QTab, QSeq, C, CsNames, DeliverFun) | Cs2];
                 false ->
-                    lager:info("not_registered"),
-                    DeliverFun({error, not_registered})
+                    lager:error("Consumer ~p pulled from RTQ, but was not registered", [Name]),
+                    deliver_error(DeliverFun, not_registered)
             end,
     State#state{cs = UpdCs}.
 
@@ -625,6 +621,13 @@ deliver_item(C, DeliverFun, {Seq,_NumItem, _Bin, _Meta} = QEntry) ->
             C#c{errs = C#c.errs + 1, deliver = undefined}
     end.
 
+%% Deliver an error if a delivery function is registered.
+deliver_error(DeliverFun, Reason) when is_function(DeliverFun)->
+    catch DeliverFun({error, Reason}),
+    ok;
+deliver_error(_NotAFun, _Reason) ->
+    ok.
+
 % if nothing has been delivered, the sink assumes nothing was skipped
 % fulfill that expectation.
 set_skip_meta(QEntry, _Seq, _C = #c{delivered = false}) ->

src/riak_repl2_rtsink_conn.erl

@@ -135,6 +135,7 @@ handle_call(status, _From, State = #state{remote = Remote,
               {hb_last, HBLast},
               %%{peer, peername(State)},
               {helper, riak_repl_util:safe_pid_to_list(Helper)},
+              {helper_msgq_len, riak_repl_util:safe_get_msg_q_len(Helper)},
               {active, Active},
               {deactivated, Deactivated},
               {source_drops, SourceDrops},
@@ -242,8 +243,9 @@ handle_info(report_bt_drops, State=#state{bt_drops = DropDict}) ->
     Report = app_helper:get_env(riak_repl, bucket_type_drop_report_interval, ?DEFAULT_INTERVAL_MILLIS),
     {noreply, State#state{bt_drops = dict:new(), bt_timer = undefined, bt_interval = Report}}.
 
-terminate(_Reason, _State) ->
+terminate(_Reason, State) ->
     %% TODO: Consider trying to do something graceful with poolboy?
+    catch riak_repl2_rtsink_helper:stop(State#state.helper),
     ok.
 
 code_change(_OldVsn, State, _Extra) ->

src/riak_repl2_rtsource_conn.erl

@@ -326,8 +326,9 @@ handle_info(Msg, State) ->
     lager:warning("Unhandled info:  ~p", [Msg]),
     {noreply, State}.
 
-terminate(_Reason, #state{helper_pid=_HelperPid, remote=Remote}) ->
+terminate(_Reason, #state{helper_pid=HelperPid, remote=Remote}) ->
     riak_core_connection_mgr:disconnect({rt_repl, Remote}),
+    catch riak_repl2_rtsource_helper:stop(HelperPid),
     ok.
 
 code_change(_OldVsn, State, _Extra) ->

src/riak_repl_aae_sink.erl

@@ -152,13 +152,19 @@ process_msg(?MSG_PUT_OBJ, {fs_diff_obj, BObj}, State) ->
 %% replies: ok | not_responsible
 process_msg(?MSG_UPDATE_TREE, IndexN, State=#state{tree_pid=TreePid}) ->
     ResponseMsg = riak_kv_index_hashtree:update(IndexN, TreePid),
+    send_reply(ResponseMsg, State);
+
+%% replies: ok | not_built | already_locked | bad_version
+process_msg(?MSG_LOCK_TREE, Version, State=#state{tree_pid=TreePid}) ->
+    ResponseMsg = riak_kv_index_hashtree:get_lock(TreePid, fullsync_sink, Version),
     send_reply(ResponseMsg, State).
 
-%% replies: ok | not_built | already_locked
-process_msg(?MSG_LOCK_TREE, State=#state{tree_pid=TreePid}) ->
+%% replies: ok | not_built | already_locked | bad_version
+process_msg(?MSG_LOCK_TREE, State) ->
     %% NOTE: be sure to die if tcp connection dies, to give back lock
-    ResponseMsg = riak_kv_index_hashtree:get_lock(TreePid, fullsync_sink),
-    send_reply(ResponseMsg, State);
+    %% Message coming from an old aae source. Only allow lock if old undefined version
+    %% is in use locally.
+    process_msg(?MSG_LOCK_TREE, legacy, State);
 
 %% no reply
 process_msg(?MSG_COMPLETE, State=#state{owner=Owner}) ->