anchor.diff:  handle({async_init_stream, undefined, Until, Queue}) ->
       handle({async_init_stream, 0, Until, Queue});
   handle({async_init_stream, Since, Until, Queue}) ->
  +    CurrentStream = state_get(changes_stream),
       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]),
  -                timer:send_after(?GLOBAL_STREAM_RESTART_INTERVAL, whereis(dbsync_worker), {timer, {async_init_stream, Since, Until, Queue}}),
  -                undefined
  +        case CurrentStream of
  +            OldStream ->
  +                case catch init_stream(Since, infinity, Queue) of
  +                    {ok, Pid} ->
  +                            catch exit(OldStream, stream_replaced),
  +                        Pid;
  +                    Reason ->
  +                        ?warning("Unable to start stream ~p (since ~p until ~p) due to: ~p", [Queue, Since, Until, Reason]),
  +                        timer:send_after(?GLOBAL_STREAM_RESTART_INTERVAL, whereis(dbsync_worker), {timer, {async_init_stream, Since, Until, Queue}}),
  +                        undefined
  +                end;
  +            _ ->
  +                ?info("Ignoring stream ~p restart request: stream has been changed since
  +                        request was issued."),
  +                OldStream
           end
       end);
   %% Unknown request
  
positive.code: 
  %% Handle stream crashes
  %% todo: ensure VFS-1877 is resolved (otherwise it probably isn't working)
  handle({'EXIT', Stream, Reason}) ->
      case state_get(changes_stream) of
          Stream ->
  
negative.code: -spec state_update(Key :: term(), UpdateFun :: fun((OldValue :: term()) -> NewValue :: term())) ->
      ok | no_return().
  state_update(Key, UpdateFun) when is_function(UpdateFun) ->
      DoUpdate = fun(_) ->
          OldValue = state_get(Key),
          NewValue = UpdateFun(OldValue),
          case OldValue of
  