Update pipe_verify_handoff_blocking to provoke vnode_down errors.

Use intercepts to provoke an otherwise-occasional error where a vnode 
has been transferred after the vnode proxy has gotten its pid
but before the work can be sent. Will fail until fix to
`riak_pipe_vnode:queue_work_wait` is merged. See basho/riak_pipe#106
for the fix.

Author: Doug Rohrer

intercepts/riak_core_vnode_master_intercepts.erl

@@ -8,6 +8,19 @@
     forwardable :: boolean(),
     opts = [] :: list()}).
 
+-record(fitting,
+{
+    pid :: pid(),
+    ref :: reference(),
+    chashfun,
+    nval
+}).
+
+-record(cmd_enqueue, {fitting :: #fitting{},
+    input :: term(),
+    timeout,
+    usedpreflist}).
+
 -define(M, riak_core_vnode_master_orig).
 
 
@@ -32,4 +45,33 @@ stop_vnode_after_bloom_fold_request_succeeds(IndexNode, Req, Sender, VMaster) ->
                     ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
             end;
         false -> ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
+    end.
+
+stop_pipe_vnode_after_request_sent(IndexNode, Req, Sender, VMaster) ->
+    case Req of
+        #cmd_enqueue{} = _Req ->
+            %% ?I_INFO("Intercepting riak_core_vnode_master:command_returning_vnode"),
+            random:seed(os:timestamp()),
+            case random:uniform(20) of
+                5 ->
+                    %% Simulate what happens when a VNode completes handoff between command_returning_vnode
+                    %% and the fold attempting to start - other attempts to intercept and slow
+                    %% certain parts of Riak to invoke the particular race condition were unsuccessful
+                    ?I_INFO("Replaced VNode with spawned function in command_returning_vnode"),
+                    Runner = self(),
+                    VNodePid = spawn(fun() ->
+                                        Runner ! go,
+                                        exit(normal)
+                                     end),
+                    receive
+                        go -> ok
+                    end,
+                    %% Still need to send the work
+                    ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster),
+                    {ok, VNodePid};
+                _ ->
+                    ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
+            end;
+        _ ->
+            ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
     end.

tests/pipe_verify_handoff_blocking.erl

@@ -43,17 +43,6 @@
 %% archive), but we don't want them to process so many inputs that
 %% they consume their blocking queues before handing off.
 
-%% Please Note: Under rare circumstances, this test may fail with a 
-%% "{badmatch,{error,[{vnode_down,noproc}]}}' error. This is not a 
-%% failure of this test but rather a side effect of a race condition 
-%% in riak_core_vnode_proxy. It manifests due to the fact that the
-%% test is attempting to send a command to a vnode that is in fact 
-%% down, however monitor only works by issuing a command and getting
-%% a PID. In some instances, get_vnode_pid fails because vnode shutdown
-%% is queued up in the mailbox before monitor node. Unfortunately, the
-%% fix would require a fundamental shift in the architecture of
-%% riak_core, which at the time of this writing is not feasible for
-%% this rare failure case.
 -module(pipe_verify_handoff_blocking).
 
 -export([
@@ -80,9 +69,12 @@ confirm() ->
     Inputs = lists:seq(1, 20),
 
     lager:info("Start ~b nodes", [?NODE_COUNT]),
-    NodeDefs = lists:duplicate(?NODE_COUNT, {current, default}),
+    Config = [{riak_core, [{ring_creation_size, 8},
+        {vnode_management_timer, 1000},
+        {handoff_concurrency, 100},
+        {vnode_inactivity_timeout, 1000}]}],
     Services = [riak_pipe],
-    [Primary,Secondary] = Nodes = rt:deploy_nodes(NodeDefs, Services),
+    [Primary,Secondary] = Nodes = rt:deploy_nodes(?NODE_COUNT, Config, Services),
     %% Ensure each node owns 100% of it's own ring
     [?assertEqual([Node], rt:owners_according_to(Node)) || Node <- Nodes],
 
@@ -114,14 +106,18 @@ confirm() ->
 
     lager:info("Unpause workers"),
     Runner ! go,
+    set_up_vnode_crashing_intercept(Primary),
 
     ok = rt:wait_until_transfers_complete(Nodes),
 
     FillerInputCount = stop_fillers(Fillers),
 
     %% if we make it this far, then no filler ever saw the vnode_down
     %% error message; otherwise badmatches in queue_filler/4 will have
-    %% halted the test
+    %% halted the test. Note that we should not see this test fail
+    %% with `{error,[{vnode_down,noproc}]}}` errors, as the `noproc` case
+    %% should be handled similarly to the `normal` exit case in
+    %% `riak_pipe_vnode:queue_work_wait`
 
     _Status2 = pipe_status(Primary, Pipe),
 
@@ -139,7 +135,13 @@ confirm() ->
     lager:info("~s: PASS", [atom_to_list(?MODULE)]),
     pass.
 
-%%% queue filling
+set_up_vnode_crashing_intercept(Primary) ->
+    lager:info("Add intercept to kill vnode before calling the wait function"),
+    rt_intercept:add(Primary, {riak_core_vnode_master,
+        [{{command_return_vnode, 4},
+            stop_pipe_vnode_after_request_sent}]}).
+
+%% queue filling
 
 %% @doc fill pipe vnode queues by repeatedly sending each input in the
 %% input list until the queue reports timeout.