anchor.diff:  -type ungrouped_results() :: [{From :: non_neg_integer(), Result :: term()}].
   -type grouped_results() :: [Results :: list()]
                            | list().
  +
  +%% The error reasons returned from collect_sink/1
  +-type receive_sink_error() :: {sender_died, Reason::term()}
  +                            | {sink_died, Reason::term()}
  +                            | timeout
  +                            | {From::non_neg_integer(), Info::term()}.
   
   %% @equiv mapred(Inputs, Query, 60000)
   mapred(Inputs, Query) ->
  
positive.code: %% :: list()'.
  -spec collect_sink(#mrc_ctx{}) ->
           {ok, grouped_results()}
         | {error, {Reason :: term(), Outputs :: ungrouped_results()}}.
  collect_sink(#mrc_ctx{keeps=NumKeeps}=Ctx) ->
      case collect_sink_loop(Ctx, []) of
          {ok, Outputs} ->
  
negative.code: %% your code should not also call it.
  -spec receive_sink(#mrc_ctx{}) ->
            {ok, Done::boolean(), Results::grouped_results()}
          | {error, receive_sink_error(), PartialResults::grouped_results()}.
  receive_sink(#mrc_ctx{sink={Sink,_}}=Ctx) ->
      %% the sender-DOWN-normal case loops to ignore that message, but
      %% we only want to send our next-request once
  