anchor.diff:                          _C ->
                               try
                                   {_Partition, BadNode} = VNode,
  -                                NewKeys = prune_inputs(Keys, BadNode),
  +                                NewKeys = prune_input_nodes(Keys, BadNode),
                                   ClaimLists = riak_kv_mapred_planner:plan_map(NewKeys),
                                   {NewFSMs, _ClaimLists1, FsmKeys} = schedule_input(NewKeys, ClaimLists, QTerm, FSMs, State),
                                   MapperData1 = lists:keydelete(Id, 1, lists:keydelete(Pid, 1, MapperData ++ FsmKeys)),
  
positive.code:     erlang:monitor(process, MapperPid),
      {no_output, State#state{mapper_data=MapperData1}};
  
  handle_event({mapexec_reply, VNode, BKey, Reply, Executor}, #state{fsms=FSMs, mapper_data=MapperData,
                                                                     pending=Pending}=State) ->
      case dict:is_key(Executor, FSMs) of
          false ->
              % node retry case will produce dictionary miss
              maybe_done(State);
          true ->
              Pending1 = Pending ++ Reply,
              FSMs1 = update_counter(Executor, FSMs),
              MapperData1 = update_inputs(Executor, VNode, BKey, MapperData),
              maybe_done(State#state{fsms=FSMs1, mapper_data=MapperData1, pending=Pending1})
      end;
  
  handle_event({mapexec_error, _Executor, Reply}, State) ->
  
negative.code:         {FSMs1, ClaimLists, FsmKeys}
      catch
          exit:{{nodedown, Node}, _} ->
              Inputs2 = prune_input_nodes(Inputs1, Node),
              ClaimLists2 = riak_kv_mapred_planner:plan_map(Inputs2),
              schedule_input(Inputs2, ClaimLists2, QTerm, FSMs0, State);
          Error ->
              throw(Error)
      end.
  
  prune_input_nodes(Inputs, BadNode) ->
      prune_input_nodes(Inputs, BadNode, []).
  prune_input_nodes([], _BadNode, NewInputs) ->
      NewInputs;
  prune_input_nodes([Input|T], BadNode, NewInputs) ->
      #riak_kv_map_input{preflist=Targets} = Input,
      Targets2 = lists:keydelete(BadNode, 2, Targets),
      prune_input_nodes(T, BadNode, [Input#riak_kv_map_input{preflist=Targets2}|NewInputs]).
  
  prune_input_partitions(Inputs, BadNode) ->
      prune_input_partitions(Inputs, BadNode, []).
  prune_input_partitions([], _BadNode, NewInputs) ->
      NewInputs;
  prune_input_partitions([Input|T], BadPartition, NewInputs) ->
      #riak_kv_map_input{preflist=Targets} = Input,
      Targets2 = lists:keydelete(BadPartition, 1, Targets),
      prune_input_partitions(T, BadPartition, [Input#riak_kv_map_input{preflist=Targets2}|NewInputs]).
  
  
  build_input(I, Ring) ->
      {{Bucket, Key}, KD} = convert_input(I),
  