anchor.diff:              ?warning("Unknown stream crash ~p: ~p", [Stream, Reason])
       end;
   handle({async_init_stream, Since, Until, Queue}) ->
  -    state_update(changes_stream, fun(_) ->
  +    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]),
  
positive.code:     lists:foreach(
          fun(SpaceId) ->
              CurrentSeq = get_current_seq(SpaceId),
  %%            ?info("DBSync broadcast for space ~p: ~p", [SpaceId, CurrentSeq]),
              {ok, Providers} = oz_spaces:get_providers(provider, SpaceId),
              dbsync_proto:status_report(SpaceId, Providers -- [oneprovider:get_provider_id()], CurrentSeq)
          end, SpaceIds).
  
negative.code: %% 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(
  