anchor.diff:  handoff_finished(_TargetNode, State) ->
       {ok, State}.
   
  -handle_handoff_data(Data, #state{log=Log}=State) ->
  -    {Key, Operation} = binary_to_term(Data),
  -    Response = dets:insert_new(Log, {Key, Operation}),
  -    {reply, Response, State}.
  +handle_handoff_data(Data, #state{logs_map=Map}=State) ->
  +    {Key, #operation{op_number=OpId, payload=Payload}} = binary_to_term(Data),
  +	DocIdx = riak_core_util:chash_key({?BUCKET,
  +                                       term_to_binary(Key)}),
  +    Preflist = riak_core_apl:get_primary_apl(DocIdx, ?N, replication),
  +	case get_log_from_map(Map, Preflist) of
  +		{ok, Log} ->
  +			Response = insert_operation(Log, Key, OpId, Payload),
  +    		{reply, Response, State};
  +		{error, Reason} ->
  +    		{reply, {error, Reason}, State}
  +	end.
   
   encode_handoff_item(Key, Operation) ->
       term_to_binary({Key, Operation}).
  
positive.code: 			threshold_prune(Rest, From)
  	end.
  
  -ifdef(TEST).
  
  %% @doc Testing threshold_prune works as expected
  
negative.code: 
  -ignore_xref([start_vnode/1]).
  
  -record(state, {partition, logs_map}).
  
  %% API
  start_vnode(I) ->
  