fix(gateway): stop presence evictions suppressing repair (#2305)

fluxer_gateway/src/presence/presence_cache.erl

@@ -126,14 +126,13 @@ handoff_to_target(TargetNode) ->
 -spec put_local(integer(), map(), state()) -> {ok, state()}.
 put_local(UserId, Presence, State) ->
     {_Reply, NewState} = presence_cache_shards:forward_put(UserId, Presence, State),
-    Tracked = presence_cache_rebalance:record_put_tombstone(UserId, Presence, NewState),
+    Tracked = presence_cache_anti_entropy:record_put(UserId, Presence, NewState),
     {ok, presence_cache_rebalance:increment_generation(Tracked)}.
 
 -spec delete_local(integer(), state()) -> {ok, state()}.
 delete_local(UserId, State) ->
-    {_Reply, NewState} = presence_cache_shards:forward_delete(UserId, State),
-    Tracked = presence_cache_rebalance:record_delete_tombstone(UserId, NewState),
-    {ok, presence_cache_rebalance:increment_generation(Tracked)}.
+    NewState = presence_cache_rebalance:evict_local(UserId, State),
+    {ok, presence_cache_anti_entropy:record_delete(UserId, NewState)}.
 
 -spec local_snapshot(state()) -> #{integer() => map()}.
 local_snapshot(State) ->

fluxer_gateway/src/presence/presence_cache_rebalance.erl

@@ -6,12 +6,11 @@
 -export([
     rebalance_ownership/1,
     handoff_all_to_target/2,
+    evict_local/2,
     perform_anti_entropy/1,
     handle_anti_entropy_request/3,
     handle_anti_entropy_digest_request/3,
     merge_anti_entropy_entries/2,
-    record_delete_tombstone/2,
-    record_put_tombstone/3,
     schedule_anti_entropy/0,
     cancel_anti_entropy_timer/1,
     start_nodedown_grace/2,
@@ -82,14 +81,6 @@ handle_anti_entropy_digest_request(FromNode, RemoteDigest, State) ->
 merge_anti_entropy_entries(Entries, State) ->
     presence_cache_anti_entropy:merge_anti_entropy_entries(Entries, State).
 
--spec record_delete_tombstone(integer(), state()) -> state().
-record_delete_tombstone(UserId, State) ->
-    presence_cache_anti_entropy:record_delete(UserId, State).
-
--spec record_put_tombstone(integer(), map(), state()) -> state().
-record_put_tombstone(UserId, Presence, State) ->
-    presence_cache_anti_entropy:record_put(UserId, Presence, State).
-
 -spec start_nodedown_grace(node(), state()) -> state().
 start_nodedown_grace(Node, State) ->
     PendingCleanups = maps:get(pending_nodedown_cleanups, State, #{}),
@@ -155,6 +146,11 @@ ensure_pending_retry_timer(State) ->
 cancel_pending_retry_timer(State) ->
     presence_cache_pending:cancel_pending_retry_timer(State).
 
+-spec evict_local(integer(), state()) -> state().
+evict_local(UserId, State) ->
+    {_Reply, NewState} = presence_cache_shards:forward_delete(UserId, State),
+    increment_generation(NewState).
+
 -spec increment_generation(state()) -> state().
 increment_generation(State) ->
     Gen = maps:get(generation, State, 0),
@@ -251,8 +247,7 @@ handoff_entry_to_target(TargetNode, _UserId, _Presence, State) when TargetNode =
 handoff_entry_to_target(TargetNode, UserId, Presence, State) ->
     case remote_apply_op(TargetNode, {put_local, UserId, Presence}, State) of
         {ok, State1} ->
-            {_Reply, State2} = presence_cache:delete_local(UserId, State1),
-            State2;
+            evict_local(UserId, State1);
         {error, State1} ->
             State1
     end.
@@ -276,8 +271,7 @@ finalize_rebalance(UserId, {put, _Presence}, OwnerNodes, true, State) ->
         true ->
             {ok, State};
         false ->
-            {_Reply, State1} = presence_cache:delete_local(UserId, State),
-            {ok, State1}
+            {ok, evict_local(UserId, State)}
     end;
 finalize_rebalance(_UserId, {put, _Presence}, _OwnerNodes, false, State) ->
     {error, State}.

fluxer_gateway/test/presence_cache_tests.erl

@@ -285,6 +285,93 @@ assert_tombstone_cleared_on_put(Pid) ->
     {_SecondPut, State3} = presence_cache:put_local(UserId, Presence, State2),
     ?assertNot(maps:is_key(UserId, maps:get(delete_tombstones, State3))).
 
+rebalance_drop_does_not_suppress_repair_test() ->
+    RemoteNode = '[email protected]',
+    with_presence_cache(fun(Pid) -> assert_rebalance_drop_repairs(RemoteNode, Pid) end).
+
+handoff_drop_does_not_suppress_repair_test() ->
+    with_local_presence_cache(fun(Pid) -> assert_handoff_drop_repairs(Pid) end).
+
+rebalance_delete_still_tombstones_test() ->
+    RemoteNode = '[email protected]',
+    with_presence_cache(fun(Pid) -> assert_rebalance_delete_tombstones(RemoteNode, Pid) end).
+
+handle_delete_still_tombstones_test() ->
+    with_local_presence_cache(fun(Pid) -> assert_handle_delete_tombstones(Pid) end).
+
+assert_rebalance_drop_repairs(RemoteNode, Pid) ->
+    State0 = cache_state(sys:get_state(Pid)),
+    {UserId, Presence, State2} = with_presence_member_nodes([node(), RemoteNode], fun() ->
+        MovedUserId = remote_owned_user_id(RemoteNode),
+        MovedPresence = anti_entropy_presence(MovedUserId),
+        {_PutReply, State1} = presence_cache:put_local(MovedUserId, MovedPresence, State0),
+        Dropped = with_reachable_remote(fun() ->
+            presence_cache_rebalance:rebalance_ownership(State1)
+        end),
+        {MovedUserId, MovedPresence, Dropped}
+    end),
+    ?assertMatch({not_found, _}, presence_cache_ops:get_local(UserId, State2)),
+    with_presence_member_nodes([node()], fun() ->
+        State3 = presence_cache_rebalance:merge_anti_entropy_entries(
+            #{UserId => Presence}, State2
+        ),
+        ?assertMatch({{ok, Presence}, _}, presence_cache_ops:get_local(UserId, State3))
+    end),
+    ?assertNot(maps:is_key(UserId, maps:get(delete_tombstones, State2))).
+
+assert_handoff_drop_repairs(Pid) ->
+    UserId = 77005,
+    Presence = anti_entropy_presence(UserId),
+    State0 = cache_state(sys:get_state(Pid)),
+    {_PutReply, State1} = presence_cache:put_local(UserId, Presence, State0),
+    State2 = with_reachable_remote(fun() ->
+        presence_cache_rebalance:handoff_all_to_target('[email protected]', State1)
+    end),
+    ?assertMatch({not_found, _}, presence_cache_ops:get_local(UserId, State2)),
+    State3 = presence_cache_rebalance:merge_anti_entropy_entries(#{UserId => Presence}, State2),
+    ?assertMatch({{ok, Presence}, _}, presence_cache_ops:get_local(UserId, State3)),
+    ?assertNot(maps:is_key(UserId, maps:get(delete_tombstones, State2))).
+
+assert_rebalance_delete_tombstones(RemoteNode, Pid) ->
+    State0 = cache_state(sys:get_state(Pid)),
+    {UserId, Presence, State3} = with_presence_member_nodes([node(), RemoteNode], fun() ->
+        GoneUserId = remote_owned_user_id(RemoteNode),
+        GonePresence = anti_entropy_presence(GoneUserId),
+        {_PutReply, State1} = presence_cache:put_local(GoneUserId, GonePresence, State0),
+        State2 = presence_cache_rebalance:queue_pending_operation(GoneUserId, delete, State1),
+        Deleted = with_reachable_remote(fun() ->
+            presence_cache_rebalance:rebalance_ownership(State2)
+        end),
+        {GoneUserId, GonePresence, Deleted}
+    end),
+    presence_cache_rebalance:cancel_pending_retry_timer(State3),
+    ?assert(maps:is_key(UserId, maps:get(delete_tombstones, State3))),
+    with_presence_member_nodes([node()], fun() ->
+        State4 = presence_cache_rebalance:merge_anti_entropy_entries(
+            #{UserId => Presence}, State3
+        ),
+        ?assertMatch({not_found, _}, presence_cache_ops:get_local(UserId, State4))
+    end).
+
+assert_handle_delete_tombstones(Pid) ->
+    UserId = 77006,
+    Presence = anti_entropy_presence(UserId),
+    State0 = cache_state(sys:get_state(Pid)),
+    {_PutReply, State1} = presence_cache:put_local(UserId, Presence, State0),
+    State2 = presence_cache_ops:handle_delete(UserId, State1),
+    ?assert(maps:is_key(UserId, maps:get(delete_tombstones, State2))),
+    State3 = presence_cache_rebalance:merge_anti_entropy_entries(#{UserId => Presence}, State2),
+    ?assertMatch({not_found, _}, presence_cache_ops:get_local(UserId, State3)).
+
+with_reachable_remote(Fun) ->
+    meck:new(presence_cache_bulk, [passthrough, no_link]),
+    meck:expect(presence_cache_bulk, safe_remote_call, fun(_Node, _Request, _Fallback) -> ok end),
+    try
+        Fun()
+    after
+        meck:unload(presence_cache_bulk)
+    end.
+
 anti_entropy_presence(UserId) ->
     #{
         <<"status">> => <<"online">>,