anchor.diff:          sorted = Args#mrargs.sorted,
           user_acc = Acc0
       },
  -    RexiMon = fabric_util:create_monitors(Workers),
  -    try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
  +    case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
           State, infinity, 1000 * 60 * 60) of
       {ok, NewState} ->
           {ok, NewState#collector.user_acc};
  
positive.code:         Callback({error, timeout}, NewState#collector.user_acc);
      {error, Resp} ->
          {ok, Resp}
      after
          rexi_monitor:stop(RexiMon),
          fabric_util:cleanup(Workers)
      end.
  
  handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
      fabric_view:remove_down_shards(State, NodeRef);
  
  handle_message({rexi_EXIT, Reason}, Worker, State) ->
      #collector{callback=Callback, counters=Counters0, user_acc=Acc} = State,
      Counters = fabric_dict:erase(Worker, Counters0),
      case fabric_view:is_progress_possible(Counters) of
      true ->
          {ok, State#collector{counters = Counters}};
      false ->
          {ok, Resp} = Callback({error, fabric_util:error_info(Reason)}, Acc),
          {error, Resp}
      end;
  
  handle_message({meta, Meta0}, {Worker, From}, State) ->
      Tot = couch_util:get_value(total, Meta0, 0),
  
negative.code:     {ok, DDoc} = fabric:open_doc(DbName, <<"_design/", GroupId/binary>>, []),
      go(DbName, DDoc, View, Args, Callback, Acc0);
  
  go(DbName, DDoc, View, Args, Callback, Acc) ->
      Shards = fabric_view:get_shards(DbName, Args),
      Workers0 = fabric_util:submit_jobs(
              Shards, fabric_rpc, map_view, [DDoc, View, Args]),
      RexiMon = fabric_util:create_monitors(Workers0),
      try
          case fabric_util:stream_start(Workers0, #shard.ref) of
              {ok, Workers} ->
                  try
                      go(DbName, Workers, Args, Callback, Acc)
                  after
                      fabric_util:cleanup(Workers)
                  end;
              {timeout, _} ->
                  Callback({error, timeout}, Acc);
              {error, Error} ->
                  Callback({error, Error}, Acc)
          end
      after
          rexi_monitor:stop(RexiMon)
      end.
  
  go(DbName, Workers, Args, Callback, Acc0) ->
      #mrargs{limit = Limit, skip = Skip, keys = Keys} = Args,
      State = #collector{
          db_name=DbName,
  