anchor.diff:  
   -record(pipe_ctx, {pipe,     % pipe handling mapred request
                      ref,      % easier-access ref/reqid
  +                   timer,    % ref() for timeout send_after
                      sender}). % {pid(), monitor()} of process sending inputs
   
   -define(PROTO_MAJOR, 1).
  
positive.code:             end
      end.
  
  pipe_mapreduce(Req, State, Inputs, Query, _Timeout) ->
      case valid_mapred_inputs(Inputs) of
          true ->
              {{ok, Pipe}, _NumKeeps} =
                  riak_kv_mrc_pipe:mapred_stream(Query),
              {InputSender, SenderMonitor} =
                  riak_kv_mrc_pipe:send_inputs_async(Pipe, Inputs),
              Ctx = #pipe_ctx{pipe=Pipe,
                              ref=(Pipe#pipe.sink)#fitting.ref,
                              sender={InputSender, SenderMonitor}},
              State#state{req=Req, req_ctx=Ctx};
          false ->
  
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, _} ->
  