anchor.diff:                          do_apply_batch_changes(ProviderId, SpaceId, Batch)
                        end, SortedKeys);
                   {_, To} when To < Stashed ->
  +                    % other proc is working - wait
                       timer:sleep(timer:seconds(5)),
                       NewCurrentUntil = get_current_seq(ProviderId, SpaceId),
                       consume_batches(ProviderId, SpaceId, NewCurrentUntil, NewBranchSince, NewBranchUntil);
  
positive.code: %%--------------------------------------------------------------------
  -spec state_put(Key :: term(), Value :: term()) -> ok.
  state_put(Key, Value) ->
      {ok, _} = dbsync_state:save(#document{key = {dbsync_state, Key}, value = #dbsync_state{entry = Value}}),
      ok.
  
  %%--------------------------------------------------------------------
  
negative.code:                     do_request_changes(ProviderId, Stashed, NewBranchSince)
              end;
          _ ->
              % set this pid as batches consumer (only one proc can consume batches)
              MyPid = self(),
              state_update({current_consumer, ProviderId, SpaceId},
                  fun
  