anchor.diff:          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).
   
   receive_results(Workers, State, Timeout, Callback) ->
       case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3, State,
  
positive.code:             end, find_replacement_shards(Shard, AllLiveShards))
          end
      end, unpack_seqs(PackedSeqs, DbName)),
      {Workers, _} = lists:unzip(Seqs),
      RexiMon = fabric_util:create_monitors(Workers),
      State = #collector{
          query_args = ChangesArgs,
          callback = Callback,
          counters = orddict:from_list(Seqs),
  
negative.code:         limit = Limit,
          user_acc = AccIn
      } = St,
      true = fabric_dict:is_key(Worker, S0),
      S1 = fabric_dict:store(Worker, Key, S0),
      % Temporary hack for FB 23637
      Interval = erlang:get(changes_seq_interval),
      if (Interval == undefined) orelse (Limit rem Interval == 0) ->
          Row = Row0#change{key = pack_seqs(S1)};
      true ->
          Row = Row0#change{key = null}
      end,
      {Go, Acc} = Callback(changes_row(Row, IncludeDocs), AccIn),
      rexi:stream_ack(From),
      {Go, St#collector{counters=S1, limit=Limit-1, user_acc=Acc}};
  
  handle_message({no_pass, Seq}, {Worker, From}, St) ->
      #collector{counters = S0} = St,
      true = fabric_dict:is_key(Worker, S0),
      S1 = fabric_dict:store(Worker, Seq, S0),
      rexi:stream_ack(From),
      {ok, St#collector{counters=S1}};
  
  handle_message({complete, Key}, Worker, State) ->
      #collector{
          counters = S0,
          total_rows = Completed % override
      } = State,
      true = fabric_dict:is_key(Worker, S0),
      S1 = fabric_dict:store(Worker, Key, S0),
      NewState = State#collector{counters=S1, total_rows=Completed+1},
      % We're relying on S1 having exactly the numnber of workers that
      % are participtaing in this response. With the new stream_start
      % that's a bit more obvious but historically it wasn't quite
      % so clear. The Completed variable is just a hacky override
      % of the total_rows field in the #collector{} record.
      NumWorkers = fabric_dict:size(S1),
      Go = case NumWorkers =:= (Completed+1) of
          true -> stop;
          false -> ok
      end,
      {Go, NewState}.
  
  make_changes_args(#changes_args{style=Style, filter=undefined}=Args) ->
      Args#changes_args{filter = Style};
  