anchor.diff:              riak_pipe:destroy(Pipe),
               prevent_keepalive(),
               {{halt, 500}, send_error({error, timeout}, RD), State};
  -        {error, {Error, _Input}} ->
  -            %% TODO: jsonify Input for error message
  +        {error, Error} ->
               riak_pipe:destroy(Pipe),
               prevent_keepalive(),
               {{halt, 500}, send_error({error, Error}, RD), State}
  
positive.code:         {pipe_timeout, Ref} ->
              {error, timeout}
      end.
  
  pipe_mapred_chunked(RD, State, Pipe, Sender) ->
      Boundary = riak_core_util:unique_id_62(),
  
negative.code:             eoi;
          #pipe_result{ref=Ref, from=From, result=Result} ->
              {ok, {From, Result}};
          #pipe_log{ref=Ref, from=From, msg=Msg} ->
              case Msg of
                  {trace, [error], {error, {Error, Input}}} ->
                      %% map function returned error tuple
                      {error, [{phase, pipe_phase_index(From)},
                               {error, trunc_print(Error)},
                               {input, trunc_print(Input)}]};
                  {trace, [error], {error, Elist}} when is_list(Elist) ->
                      %% generic pipe fitting error
                      {error, [{phase, pipe_error_phase(Elist)},
                               {error, pipe_error_error(Elist)},
                               {input, pipe_error_input(Elist)},
                               {type, pipe_error_type(Elist)},
                               {stack, pipe_error_stack(Elist)}]};
                  _ ->
                      %% not a log message we're interested in
                      pipe_receive_output(Ref, {SenderPid, SenderRef})
  