anchor.diff:                         after
                              riak_sock_pool:checkin(?CONN_POOL, Conn)
                          end end),
  -    receive_fold_results(Acc, 0),
  -    {reply, Acc, State}.
  +    FinalAcc = receive_fold_results(Folder, Acc),
  +    {reply, FinalAcc, State}.
   
   is_empty(State) ->
       Partition = to_binary(State#state.partition),
  
positive.code:             CatalogProcessPid ! {fold_stream, done, StreamRef};
          {stream, StreamRef, Value, Props}=_Msg2 ->
              case Props of
                  <<"">> ->
                      Props2 = [];
                  _ ->
                      Props2 = binary_to_term(Props)
              end,
              IndexBin = riak_search_utils:to_binary(Index),
              FieldTermBin = riak_search_utils:to_binary([Field, ".", Term]),
              BObj = term_to_binary({Field, Term, Value, Props2}),
              FoldResultPid ! {fold_result, Fun0({IndexBin, FieldTermBin}, BObj, Acc)},
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term);
          _Msg ->
              %% io:format("fold_stream_process: unknown message: ~p~n", [_Msg]),
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term)
          after ?FOLD_TIMEOUT ->
              CatalogProcessPid ! {fold_stream, done, StreamRef},
              error_logger:warning_msg("fold_stream_process: table timed out: ~p.~p.~p~n",
                                        [Index, Field, Term])
      end.
  
  %% receive the Fun0(processed) objects from all the "buckets" on this partition, accumulate them
  %%   and return them
  receive_fold_results(Acc, Count) ->
      receive
          {fold_result, done} ->
              %% io:format("receive_fold_results: fold complete [~p objects].~n",
              %%           [Count]),
              Acc;
          {fold_result, _Obj} ->
              receive_fold_results(Acc, Count+1)
      end.
  
  %%%
  
negative.code: %%   off a stream process for each one in parallel
  fold_catalog_process(CatRef,
                       FoldResultPid,
                       CatalogDone,
                       StreamProcessCount,
                       FinishedStreamProcessCount,
  