anchor.diff:      space_id = SpaceId,
       stream = Stream
   }, Docs = [Doc | RestDocs]) ->
  -    try
  -        State2 = consume_change(State, Doc),
  -        consume_change_and_enter_retry_mode_on_error(State2, RestDocs)
  +    State2 = try
  +        consume_change(State, Doc)
       catch
           Error:Reason ->
               ?error_stacktrace("Unexpected error ~p:~p in harvest_stream ~p "
               "handling changes for harvester ~p in space ~p",
                   [Error, Reason, Id, HarvesterId, SpaceId]),
  -            catch gen_server:stop(Stream),
  +            stop_changes_stream(Stream),
               enter_retry_mode(Docs, State)
  +    end,
  +    case State2#state.mode of
  +        retry ->
  +            State2;
  +        _ ->
  +            consume_change_and_enter_retry_mode_on_error(State2, RestDocs)
       end;
   consume_change_and_enter_retry_mode_on_error(State, Doc = #document{}) ->
       consume_change_and_enter_retry_mode_on_error(State, [Doc]).
  
positive.code:     {noreply, consume_change_and_enter_retry_mode_on_error(State, DocOrDocs)};
  handle_cast({change, {error, _Seq, Reason}}, State = #state{mode = stream}) ->
      {stop, Reason, State};
  handle_cast(Request, #state{} = State) ->
      ?log_bad_request(Request),
      {noreply, State}.
  
negative.code:     {noreply, NewState :: state()} |
      {noreply, NewState :: state(), timeout() | hibernate} |
      {stop, Reason :: term(), NewState :: state()}.
  handle_cast({change, {ok, end_of_stream}}, State = #state{mode = retry}) ->
      % all changes sent by changes_stream before stopping it have already arrived
      {noreply, State#state{changes_flushed = true}};
  