anchor.diff:  %% :: list()'.
   -spec collect_sink(#mrc_ctx{}) ->
            {ok, grouped_results()}
  -       | {error, {Reason :: term(), Outputs :: ungrouped_results()}}.
  +       | {error, receive_sink_error()}.
   collect_sink(#mrc_ctx{keeps=NumKeeps}=Ctx) ->
       case collect_sink_loop(Ctx, []) of
           {ok, Outputs} ->
  
positive.code: %% your code should not also call it.
  -spec receive_sink(#mrc_ctx{}) ->
            {ok, Done::boolean(), Results::grouped_results()}
          | {error, Reason::term(), 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
  
negative.code: -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) ->
  