anchor.diff:  %%--------------------------------------------------------------------
   -spec state_put(Key :: term(), Value :: term()) -> ok.
   state_put(Key, Value) ->
  -    {ok, _} = dbsync_state:save(#document{key = Key, value = #dbsync_state{entry = Value}}),
  +    {ok, _} = dbsync_state:save(#document{key = {dbsync_state, Key}, value = #dbsync_state{entry = Value}}),
       ok.
   
   %%--------------------------------------------------------------------
  
positive.code:         Stream ->
              Since = state_get(global_resume_seq),
              state_update(changes_stream,
                  fun(_) ->
                      {ok, NewStream} = init_stream(Since, infinity, global),
                      NewStream
                  end);
          _ ->
              ?warning("Unknown stream crash ~p: ~p", [Stream, Reason])
  
negative.code: %% @end
  %%--------------------------------------------------------------------
  -spec get_space_id(datastore:document()) ->
      {ok, SpaceId :: binary() | {space_doc, SpaceId :: binary()}} | {error, Reason :: term()}.
  get_space_id(#document{key = Id, value = #monitoring_state{}}) ->
      case monitoring_state:decode_id(Id) of
          {space, SpaceId, _, _} ->
  