anchor.diff:          {ok, NewState}
       end.
   
  -handle_message({rexi_DOWN, _, {_, NodeRef}, _}, nil, State) ->
  -    fabric_view:remove_down_shards(State, NodeRef);
  +handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
  +    fabric_view:check_down_shards(State, NodeRef);
   
   handle_message({rexi_EXIT, Reason}, Worker, State) ->
  -    #collector{
  -        callback=Callback,
  -        counters=Counters0,
  -        rows = Seqs0,
  -        user_acc=Acc
  -    } = State,
  -    Counters = fabric_dict:erase(Worker, Counters0),
  -    Seqs = fabric_dict:erase(Worker, Seqs0),
  -    case fabric_view:is_progress_possible(Counters) of
  -    true ->
  -        {ok, State#collector{counters = Counters, rows=Seqs}};
  -    false ->
  -        {ok, Resp} = Callback({error, fabric_util:error_info(Reason)}, Acc),
  -        {error, Resp}
  -    end;
  +    fabric_view:handle_worker_exit(State, Worker, Reason);
   
   handle_message(_, _, #collector{limit=0} = State) ->
       {stop, State};
  
positive.code:         rows = Seqs % store sequence positions instead
      },
      %% TODO: errors need to be handled here
      try
          receive_results(Workers, State, Timeout, Callback)
      after
          rexi_monitor:stop(RexiMon),
          fabric_util:cleanup(Workers)
      end.
  
  receive_results(Workers, State, Timeout, Callback) ->
      case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3, State,
  
negative.code:             end, find_replacement_shards(Shard, AllLiveShards))
          end
      end, unpack_seqs(PackedSeqs, DbName)),
      {Workers0, _} = lists:unzip(Seqs),
      RexiMon = fabric_util:create_monitors(Workers0),
      try
          case fabric_util:stream_start(Workers0, #shard.ref) of
              {ok, Workers} ->
                  try
                      LiveSeqs = lists:filter(fun({W, _S}) ->
                          lists:member(W, Workers)
                      end, Seqs),
                      send_changes(DbName, Workers, LiveSeqs, ChangesArgs,
                              Callback, AccIn, Timeout)
                  after
                      fabric_util:cleanup(Workers)
                  end;
              Else ->
                  Callback(Else, AccIn)
          end
      after
          rexi_monitor:stop(RexiMon)
      end.
  
  send_changes(DbName, Workers, Seqs, ChangesArgs, Callback, AccIn, Timeout) ->
      State = #collector{
          db_name = DbName,
          query_args = ChangesArgs,
          callback = Callback,
          counters = orddict:from_list(Seqs),
  