anchor.diff:                      ?info("[ DBSync ] Queue ~p removed!", [QueueKey]),
                       undefined;
                   false ->
  -                    timer:send_after(?FLUSH_QUEUE_INTERVAL, whereis(dbsync_worker), {timer, {flush_queue, QueueKey}}),
  +                    timer:send_after(FlushAgainAfter, whereis(dbsync_worker), {timer, {flush_queue, QueueKey}}),
                       Queue#queue{batch_map = NewBatchMap}
               end;
               (undefined) when QueueKey =:= global ->
  -                timer:send_after(?FLUSH_QUEUE_INTERVAL, whereis(dbsync_worker), {timer, {flush_queue, QueueKey}}),
  +                timer:send_after(FlushAgainAfter, whereis(dbsync_worker), {timer, {flush_queue, QueueKey}}),
                   undefined;
               (undefined) ->
                   ?warning("Unknown operation on empty queue ~p", [QueueKey]),
  
positive.code: %%--------------------------------------------------------------------
  -spec ensure_global_stream_active() -> ok.
  ensure_global_stream_active() ->
      case is_valid_stream(state_get(changes_stream)) of
          true ->
              CTime = erlang:monotonic_time(milli_seconds),
              MaxIdleTime = timer:seconds(10),
              case dbsync_utils:temp_get(last_change) of
                  undefined ->
  
negative.code: handle({flush_queue, QueueKey}) ->
      ?debug("[ DBSync ] Flush queue ~p", [QueueKey]),
      ensure_global_stream_active(),
  
      FlushInterval = ?FLUSH_QUEUE_INTERVAL,
      FlushAgainAfter = FlushInterval + crypto:rand_uniform(0, round(?FLUSH_QUEUE_INTERVAL / 2)),
  
      case QueueKey of
          global ->
              CTime = erlang:monotonic_time(milli_seconds),
              case dbsync_utils:temp_get(last_global_flush) of
                  FTime when  is_integer(FTime),
                              FTime + FlushInterval / 2 > CTime ->
                      ?info("[ DBSync ] Flush loop is too fast, breaking this one."),
                      throw({too_many_flush_loops, {flush_queue, QueueKey}});
                  _ ->
                      dbsync_utils:temp_put(last_global_flush, CTime, 0)
              end;
          _ -> ok
      end,
  
      state_update({queue, QueueKey},
          fun(#queue{batch_map = BatchMap, removed = IsRemoved} = Queue) ->
              NewBatchMap = maps:map(
  