anchor.diff:         true ->
               %% something went wrong sending inputs - tell the client
               %% about it, and shutdown the pipe
  +            erlang:cancel_timer(PipeCtx#pipe_ctx.timer),
               riak_pipe:destroy(PipeCtx#pipe_ctx.pipe),
               lager:error("Error sending inputs: ~p", [Reason]),
               NewState = send_error("~p", [bad_mapred_inputs], State),
               {noreply, NewState#state{req=undefined, req_ctx=undefined}}
       end;
  +handle_info({pipe_timeout, Ref},
  +            State=#state{req=#rpbmapredreq{},
  +                         req_ctx=#pipe_ctx{ref=Ref,
  +                                           sender=Sender,
  +                                           pipe=Pipe}}) ->
  +    NewState = send_error("timeout", [], State),
  +    riak_pipe:destroy(Pipe),
  +    case Sender of
  +        {SenderPid, _} ->
  +            erlang:exit(SenderPid, kill);
  +        undefined ->
  +            ok
  +    end,
  +    {noreply, NewState#state{req=undefined, req_ctx=undefined}};
   %% ignore #pipe_log for now, since riak_kv_mrc_pipe does not enable it
   
   %% LEGACY Handle response from mapred_stream/mapred_bucket_stream
  
positive.code: 
  -record(pipe_ctx, {pipe,     % pipe handling mapred request
                     ref,      % easier-access ref/reqid
                     sender}). % {pid(), monitor()} of process sending inputs
  
  -define(PROTO_MAJOR, 1).
  
negative.code:                          req_ctx=#pipe_ctx{ref=ReqId}=PipeCtx}) ->
      case encode_mapred_phase([Res], ContentType) of
          {error, Reason} ->
              erlang:cancel_timer(PipeCtx#pipe_ctx.timer),
              riak_pipe:destroy(PipeCtx#pipe_ctx.pipe),
              case PipeCtx#pipe_ctx.sender of
                  {SenderPid, _} ->
  