anchor.diff:  %% Push changes from queue to providers
   handle({flush_queue, QueueKey}) ->
       ?debug("[ DBSync ] Flush queue ~p", [QueueKey]),
  +    ensure_global_stream_active(),
       state_update({queue, QueueKey},
           fun(#queue{batch_map = BatchMap, removed = IsRemoved} = Queue) ->
               NewBatchMap = maps:map(
  
positive.code:     pong;
  
  handle(healthcheck) ->
      case state_get(changes_stream) of
          Stream when is_pid(Stream) ->
              ok;
  
negative.code:             ?warning("Unknown stream crash ~p: ~p", [Stream, Reason])
      end;
  handle({async_init_stream, Since, Until, Queue}) ->
      state_update(changes_stream, fun(OldStream) ->
          case catch init_stream(Since, infinity, Queue) of
              {ok, Pid} ->
                  catch exit(OldStream, shutdown),
                  Pid;
              Reason ->
                  ?warning("Unable to start stream ~p (since ~p until ~p) due to: ~p", [Queue, Since, Until, Reason]),
  