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: %%--------------------------------------------------------------------
  -spec init_stream(Since :: non_neg_integer(), Until :: non_neg_integer() | infinity, Queue :: queue()) ->
      {ok, pid()} | {error, Reason :: term()}.
  init_stream(undefined, Until, Queue) ->
      init_stream(0, Until, Queue);
  init_stream(Since, Until, Queue) ->
      ?info("[ DBSync ]: Starting stream ~p ~p ~p", [Since, Until, Queue]),
      case Queue of
  