anchor.diff:  -spec state_update(Key :: term(), UpdateFun :: fun((OldValue :: term()) -> NewValue :: term())) ->
       ok | no_return().
   state_update(Key, UpdateFun) when is_function(UpdateFun) ->
  -    DoUpdate = fun() ->
  +    DoUpdate = fun(_) ->
           OldValue = state_get(Key),
           NewValue = UpdateFun(OldValue),
           case OldValue of
  
positive.code: %% Append change to given queue
  handle({QueueKey, #change{seq = Seq, doc = #document{key = Key, rev = Rev} = Doc} = Change}) ->
      ?debug("[ DBSync ] Received change on queue ~p with seq ~p: ~p", [QueueKey, Seq, Doc]),
      dbsync_utils:temp_put(last_change, erlang:monotonic_time(milli_seconds), 0),
      Rereplication = QueueKey =:= global andalso dbsync_utils:temp_get({replicated, Key, Rev}) =:= true,
      case {has_sync_context(Doc), Rereplication} of
          {true, false} ->
  
negative.code: 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 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
  