anchor.diff:      ActiveSenders=proplists:get_value(active,Senders),
       get_concurrency_limit() =< (ActiveReceivers + ActiveSenders).
   
  -%% checks to see if the handoff is already off to the races, or changed
  -enqueue_handoff(Mod,Idx,Node,VnodePid,Handoffs) ->
  -    lists:foldl(
  -      fun (#handoff_status{direction=inbound},Flag) ->
  -              Flag;
  -          (H=#handoff_status{handoff={M,I,N},vnode_pid=Pid},Flag) ->
  -              %% a different module or index, add it
  -              case (M==Mod) and (I==Idx) of
  -                  false ->
  -                      Flag;
  -                  true ->
  -                      %% or if the same target and vnode, keep it
  -                      case (N==Node) and (Pid==VnodePid) of
  -                          true ->
  -                              false;
  -                          false ->
  -                              %% kill the old one
  -                              erlang:exit(H#handoff_status.transport_pid,
  -                                          resubmit_handoff_change),
  -                              true
  -                      end
  -              end
  -      end,
  -      true,
  -      Handoffs).
  -
   %% spawn a sender process
  -send_handoff (Module,Index,TargetNode,VnodePid) ->
  +send_handoff (Mod,Idx,Node,Vnode,HS) ->
       case handoff_concurrency_limit_reached() of
           true ->
               {error, max_concurrency};
           false ->
  -            {ok,Pid}=riak_core_handoff_sender_sup:start_sender(TargetNode,
  -                                                               Module,
  -                                                               Index,
  -                                                               VnodePid),
  -            erlang:monitor(process,Pid),
  -
  -            %% successfully started up a new sender handoff
  -            {ok, #handoff_status{ transport_pid=Pid,
  -                                  direction=outbound,
  -                                  timestamp=now(),
  -                                  handoff={Module,Index,TargetNode},
  -                                  vnode_pid=VnodePid
  -                                }
  -            }
  +            ShouldHandoff=
  +                case lists:keyfind({Mod,Idx},#handoff_status.modindex,HS) of
  +                    false ->
  +                        true;
  +                    Handoff=#handoff_status{node=Node,vnode_pid=Vnode} ->
  +                        {false,Handoff};
  +                    #handoff_status{transport_pid=Sender} ->
  +                        %% found a running handoff with a different vnode
  +                        %% source or a different arget ndoe, kill the current
  +                        %% one and the new one will start up
  +                        erlang:exit(Sender,resubmit_handoff_change),
  +                        true
  +                end,
  +
  +            case ShouldHandoff of
  +                true ->
  +                    %% start the sender process
  +                    {ok,Pid}=riak_core_handoff_sender_sup:start_sender(Node,
  +                                                                       Mod,
  +                                                                       Idx,
  +                                                                       Vnode),
  +                    erlang:monitor(process,Pid),
  +
  +                    %% successfully started up a new sender handoff
  +                    {ok, #handoff_status{ transport_pid=Pid,
  +                                          direction=outbound,
  +                                          timestamp=now(),
  +                                          node=Node,
  +                                          modindex={Mod,Idx},
  +                                          vnode_pid=Vnode
  +                                        }
  +                    };
  +                {false,CurrentHandoff} ->
  +                    %% handoff already going, just return it
  +                    {ok, CurrentHandoff}
  +            end
       end.
   
   %% spawn a receiver process
  
positive.code:                     normal ->
                          false;
                      max_concurrency ->
                          lager:info("An ~w handoff of partition ~w ~w was terminated for reason: ~w~n", [Dir,Mod,Index,Reason]),
                          true;
                      _ ->
                          lager:error("An ~w handoff of partition ~w ~w was terminated for reason: ~w~n", [Dir,Mod,Index,Reason]),
                          true
                  end,
  
              %% if we have the vnode process pid, tell the vnode why the
              %% handoff stopped so it can clean up its state
              case WarnVnode andalso H#handoff_status.vnode_pid of
                  Vnode when is_pid(Vnode) ->
                      riak_core_vnode:handoff_error(Vnode,'DOWN',Reason);
                  _ ->
                      ok
  
negative.code: 
  -type mod()   :: atom().
  -type index() :: integer().
  
  -record(handoff_status,
          { modindex      :: {mod(),index()},
            node          :: atom(),
            direction     :: inbound | outbound,
            transport_pid :: pid(),
            timestamp     :: tuple(),
  