anchor.diff:      ok.
   
   %% don't let timer messages leak
  -cleanup_timer(#mrc_ctx{timer=Timer}) ->
  -    cleanup_timer(Timer);
   cleanup_timer({Tref, PipeRef}) when is_reference(Tref),
                                       is_reference(PipeRef) ->
       case erlang:cancel_timer(Tref) of
  
positive.code: %% @doc Tear down the async sender, sink, and timer pieces setup by
  %% {@link mapred_stream_sink/3}, and collect any messages they might
  %% have been delivering.
  -spec cleanup_sink(#mrc_ctx{}) -> ok.
  cleanup_sink(#mrc_ctx{sender=Sender, sink=Sink, timer=Timer}) ->
      cleanup_sender(Sender),
      cleanup_sink(Sink),
  
negative.code: %% eventually receive `worker_startup_error's from vnodes that can no
  %% longer find the fittings, but to help the process along, we kill
  %% them immediately here.
  cleanup_sender({SenderPid, SenderMon}) when is_pid(SenderPid),
                                              is_reference(SenderMon) ->
      erlang:demonitor(SenderMon, [flush]),
  