anchor.diff:          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
  +    % Assert that we don't have other messages from this
  +    % worker when the total_and_offset message arrives.
  +    0 = fabric_dict:lookup_element(Worker, Counters0),
  +    rexi:stream_ack(From),
  +    Counters1 = fabric_dict:update_counter(Worker, 1, Counters0),
  +    Total = Total0 + Tot,
  +    Offset = Offset0 + Off,
  +    case fabric_dict:any(0, Counters1) of
  +    true ->
  +        {ok, State#collector{
  +            counters = Counters1,
  +            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(Counters1),
  +            total_rows = Total,
  +            offset = FinalOffset,
  +            user_acc = Acc
  +        }}
       end;
   
   handle_message(#view_row{}, {_, _}, #collector{limit=0} = State) ->
  
positive.code:         sorted = Args#mrargs.sorted,
          user_acc = Acc0
      },
      RexiMon = fabric_util:create_monitors(Workers),
      try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
          State, infinity, 1000 * 60 * 60) of
      {ok, NewState} ->
          {ok, NewState#collector.user_acc};
  
negative.code:         Callback({error, timeout}, NewState#collector.user_acc);
      {error, Resp} ->
          {ok, Resp}
      end.
  
  handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
      fabric_view:check_down_shards(State, NodeRef);
  
  handle_message({rexi_EXIT, Reason}, Worker, State) ->
      fabric_view:handle_worker_exit(State, Worker, Reason);
  
  handle_message({rexi_EXIT, Reason}, _, State) ->
      #collector{callback=Callback, user_acc=Acc} = State,
      {ok, Resp} = Callback({error, fabric_util:error_info(Reason)}, Acc),
      {error, Resp};
  
  handle_message({meta, Meta0}, {Worker, From}, State) ->
      Tot = couch_util:get_value(total, Meta0, 0),
  