anchor.diff:          Callback(timeout, Acc0)
       end.
   
  +go(DbName, _Options, Workers, QueryArgs, Callback, Acc0) ->
  +    #mrargs{limit = Limit, skip = Skip} = QueryArgs,
  +    State = #collector{
  +        db_name = DbName,
  +        query_args = QueryArgs,
  +        callback = Callback,
  +        counters = fabric_dict:init(Workers, 0),
  +        skip = Skip,
  +        limit = Limit,
  +        user_acc = Acc0
  +    },
  +    case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
  +        State, infinity, 5000) of
  +    {ok, NewState} ->
  +        {ok, NewState#collector.user_acc};
  +    {timeout, NewState} ->
  +        Callback({error, timeout}, NewState#collector.user_acc);
  +    {error, Resp} ->
  +        {ok, Resp}
  +    end.
  +
   handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
  -    fabric_view:remove_down_shards(State, NodeRef);
  +    fabric_view:check_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;
  +    fabric_view:handle_worker_exit(State, Worker, Reason);
   
   handle_message({meta, Meta0}, {Worker, From}, State) ->
       Tot = couch_util:get_value(total, Meta0, 0),
  
positive.code:         offset = Offset0,
          user_acc = AccIn
      } = State,
      case fabric_dict:lookup_element(Worker, Counters0) of
      undefined ->
          % this worker lost the race with other partition copies, terminate
          gen_server:reply(From, stop),
          {ok, State};
      0 ->
          gen_server:reply(From, ok),
          Counters1 = fabric_dict:update_counter(Worker, 1, Counters0),
          Counters2 = fabric_view:remove_overlapping_shards(Worker, Counters1),
          Total = Total0 + Tot,
          Offset = Offset0 + Off,
          case fabric_dict:any(0, Counters2) of
          true ->
              {ok, State#collector{
                  counters = Counters2,
                  total_rows = Total,
                  offset = Offset
              }};
          false ->
              FinalOffset = erlang:min(Total, Offset+State#collector.skip),
              Meta = [{total, Total}, {offset, FinalOffset}],
              {Go, Acc} = Callback({meta, Meta}, AccIn),
              {Go, State#collector{
                  counters = fabric_dict:decrement_all(Counters2),
                  total_rows = Total,
                  offset = FinalOffset,
                  user_acc = Acc
              }}
          end
      end;
  
  handle_message(#view_row{} = Row, {Worker, From}, State) ->
  
negative.code: -include_lib("couch/include/couch_db.hrl").
  -include_lib("couch_mrview/include/couch_mrview.hrl").
  
  go(DbName, Options, #mrargs{keys=undefined} = QueryArgs, Callback, Acc) ->
      Shards = mem3:shards(DbName),
      Workers0 = fabric_util:submit_jobs(
              Shards, fabric_rpc, all_docs, [Options, QueryArgs]),
      RexiMon = fabric_util:create_monitors(Workers0),
      try
          case fabric_util:stream_start(Workers0, #shard.ref) of
              {ok, Workers} ->
                  try
                      go(DbName, Options, Workers, QueryArgs, 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;
  
  
  