anchor.diff:  %% @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.
  +-spec cleanup_sink(#mrc_ctx{}|{pid(),reference()}|undefined) -> ok.
   cleanup_sink(#mrc_ctx{sender=Sender, sink=Sink, timer=Timer}) ->
       cleanup_sender(Sender),
       cleanup_sink(Sink),
  
positive.code:     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
  
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]),
  