anchor.diff:      {stop, Reason :: term(), NewState :: state()}.
   handle_call({change, {ok, DocOrDocs}}, From, State = #state{mode = streaming}) ->
       gen_server2:reply(From, ok),
  -    {noreply, consume_change_and_enter_retrying_mode_on_error(State, DocOrDocs)};
  +    {noreply, consume_change_and_retry_on_error(State, DocOrDocs)};
   handle_call({change, {ok, end_of_stream}}, _From, State = #state{mode = retrying}) ->
       % all changes sent by changes_stream before stopping it have already arrived
       % so we can start backoff algorithm
  
positive.code:             State
      end.
  
  -spec consume_change_and_enter_retrying_mode_on_error(state(),
      datastore:doc() | [datastore:doc()]) -> state().
  consume_change_and_enter_retrying_mode_on_error(State = #state{mode = streaming}, []) ->
      % all docs were successfully handled
      State;
  consume_change_and_enter_retrying_mode_on_error(State = #state{
      mode = retrying,
      stream_pid = undefined
  }, []) ->
      % all docs were successfully handled in retrying mode, get back to streaming mode
      enter_streaming_mode(State);
  consume_change_and_enter_retrying_mode_on_error(State = #state{
      id = Id,
      harvester_id = HarvesterId,
      space_id = SpaceId
  
negative.code:     stream_pid = undefined,
      docs_to_retry = Docs
  }) ->
      {noreply, consume_change_and_retry_on_error(State, Docs)};
  handle_info(Info, #state{} = State) ->
      ?log_bad_request(Info),
      {noreply, State}.
  