anchor.diff:      {noreply, NewState :: state()} |
       {noreply, NewState :: state(), timeout() | hibernate} |
       {stop, Reason :: term(), NewState :: state()}.
  -handle_cast(_Request, #state{mode = Mode} = State)
  -    when Mode =:= retry orelse Mode =:= resuming ->
  -    % ignore because stream is in retry mode
  -    {noreply, 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}};
  
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:     space_id = SpaceId,
      stream = Stream
  }, Docs = [Doc | 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]),
              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]).
  