fix(push): run the stale DM read check on presence nodes (#3042)

fluxer_gateway/src/presence/presence_update.erl

@@ -166,11 +166,17 @@ flush_push_buffer(#{push_buffer := []} = State) ->
     State;
 flush_push_buffer(#{push_buffer := Buffer} = State) ->
     ok = push:handle_buffered_message_creates([
-        (maps:get(params, Entry))#{buffered_at => maps:get(buffered_at, Entry, undefined)}
+        (maps:get(params, Entry))#{buffered_at => entry_buffered_at(Entry)}
      || Entry <- lists:reverse(Buffer)
     ]),
     State#{push_buffer := []}.
 
+-spec entry_buffered_at(push_buffer_entry()) -> integer() | undefined.
+entry_buffered_at(#{buffered_at := BufferedAt}) ->
+    BufferedAt;
+entry_buffered_at(#{message_id := MessageId}) ->
+    snowflake_util:extract_timestamp(MessageId).
+
 -spec maybe_update_push_eligibility(state()) -> state().
 maybe_update_push_eligibility(State) ->
     Eligible = push_eligible(State),

fluxer_gateway/src/push/push.erl

@@ -36,7 +36,7 @@
 -define(CNT_READ_STATE_SKIPPED, push_read_state_fetch_skipped).
 -define(MAX_FETCH_RPCS, 8).
 -define(READ_STATE_FETCH_TIMEOUT_MS, 2000).
--define(READ_STATE_FETCH_MAX_POOL_PCT, 50).
+-define(READ_STATE_TABLE, push_read_state_counters).
 -define(READ_STATE_FETCH_SLOTS, read_state_fetches_in_flight).
 -define(MAX_READ_STATE_FETCHES, 64).
 -define(DEFAULT_FETCH_USERS, 2000).
@@ -334,28 +334,26 @@ publish_unread(Channels, Stale) ->
 -spec publish_unless_read(read_marks(), {read_key(), map()}) -> ok.
 publish_unless_read(ReadMarks, {{UserId, ChannelId, MessageId}, Params}) ->
     case MessageId =< maps:get({UserId, ChannelId}, ReadMarks, 0) of
-        true -> bump_counter(?CNT_READ_STATE_SUPPRESSED);
+        true -> bump_read_state_counter(?CNT_READ_STATE_SUPPRESSED, 1);
         false -> handle_message_create(Params)
     end.
 
 -spec read_state_fetch_allowed(non_neg_integer()) -> boolean().
 read_state_fetch_allowed(Fetches) ->
-    case
+    case acquire_read_state_fetch_slot() of
-        push_worker_pool:utilization_pct() < ?READ_STATE_FETCH_MAX_POOL_PCT andalso
-            acquire_read_state_fetch_slot()
-    of
         true ->
             true;
         false ->
-            bump_counter(?CNT_READ_STATE_SKIPPED, Fetches),
+            bump_read_state_counter(?CNT_READ_STATE_SKIPPED, Fetches),
             false
     end.
 
 -spec acquire_read_state_fetch_slot() -> boolean().
 acquire_read_state_fetch_slot() ->
+    ok = ensure_read_state_table(),
     try
         ets:update_counter(
-            ?PUSH_COUNTER_TABLE, ?READ_STATE_FETCH_SLOTS, {2, 1}, {?READ_STATE_FETCH_SLOTS, 0}
+            ?READ_STATE_TABLE, ?READ_STATE_FETCH_SLOTS, {2, 1}, {?READ_STATE_FETCH_SLOTS, 0}
         )
     of
         InFlight when InFlight > ?MAX_READ_STATE_FETCHES ->
@@ -369,12 +367,41 @@ acquire_read_state_fetch_slot() ->
 
 -spec release_read_state_fetch_slot() -> ok.
 release_read_state_fetch_slot() ->
-    try ets:update_counter(?PUSH_COUNTER_TABLE, ?READ_STATE_FETCH_SLOTS, {2, -1, 0, 0}) of
+    try ets:update_counter(?READ_STATE_TABLE, ?READ_STATE_FETCH_SLOTS, {2, -1, 0, 0}) of
         _InFlight -> ok
     catch
         error:badarg -> ok
     end.
 
+-spec bump_read_state_counter(atom(), non_neg_integer()) -> ok.
+bump_read_state_counter(Key, Increment) ->
+    ok = ensure_read_state_table(),
+    try ets:update_counter(?READ_STATE_TABLE, Key, {2, Increment}, {Key, 0}) of
+        _Value -> ok
+    catch
+        error:badarg -> ok
+    end.
+
+-spec ensure_read_state_table() -> ok.
+ensure_read_state_table() ->
+    case ets:whereis(?READ_STATE_TABLE) of
+        undefined ->
+            guild_ets_utils:ensure_table(?READ_STATE_TABLE, [
+                named_table, public, set, {write_concurrency, true}
+            ]);
+        _Table ->
+            ok
+    end.
+
+-spec read_state_counter(atom()) -> non_neg_integer() | unavailable.
+read_state_counter(Key) ->
+    try ets:lookup(?READ_STATE_TABLE, Key) of
+        [{Key, Value}] when is_integer(Value), Value >= 0 -> Value;
+        _ -> unavailable
+    catch
+        error:badarg -> unavailable
+    end.
+
 -spec fetch_each_read_mark([{integer(), integer()}], read_marks()) -> read_marks().
 fetch_each_read_mark([], ReadMarks) ->
     ReadMarks;
@@ -383,7 +410,7 @@ fetch_each_read_mark([{UserId, ChannelId} = Key | Rest], ReadMarks) ->
         {ok, ReadMessageId} ->
             fetch_each_read_mark(Rest, ReadMarks#{Key => ReadMessageId});
         error ->
-            bump_counter(?CNT_READ_STATE_SKIPPED, length(Rest)),
+            bump_read_state_counter(?CNT_READ_STATE_SKIPPED, length(Rest)),
             ReadMarks
     end.
 
@@ -400,7 +427,7 @@ fetch_read_message_id(UserId, ChannelId) ->
         {ok, _Data} ->
             {ok, 0};
         {error, Reason} ->
-            bump_counter(?CNT_READ_STATE_FAILURES),
+            bump_read_state_counter(?CNT_READ_STATE_FAILURES, 1),
             logger:debug("Push: read state fetch failed", #{
                 reason => Reason, user_id => UserId, channel_id => ChannelId
             }),
@@ -791,10 +818,10 @@ push_loss_counters() ->
     #{
         counters => counter_table_status(),
         worker_pool_dropped => read_counter(?CNT_WORKER_POOL),
-        read_state_suppressed => read_counter(?CNT_READ_STATE_SUPPRESSED),
+        read_state_suppressed => read_state_counter(?CNT_READ_STATE_SUPPRESSED),
-        read_state_fetch_failures => read_counter(?CNT_READ_STATE_FAILURES),
+        read_state_fetch_failures => read_state_counter(?CNT_READ_STATE_FAILURES),
-        read_state_fetch_skipped => read_counter(?CNT_READ_STATE_SKIPPED),
+        read_state_fetch_skipped => read_state_counter(?CNT_READ_STATE_SKIPPED),
-        read_state_fetches_in_flight => read_counter(?READ_STATE_FETCH_SLOTS)
+        read_state_fetches_in_flight => read_state_counter(?READ_STATE_FETCH_SLOTS)
     }.
 
 -spec counter_table_status() -> live | unavailable.
@@ -1037,19 +1064,21 @@ a_buffered_dm_without_a_buffer_time_is_published_without_a_read_state_fetch_test
     ),
     ?assertEqual({1, 0}, Published).
 
-a_dm_buffered_past_the_outbox_window_is_published_without_a_fetch_when_the_pool_is_busy_test() ->
+a_dm_buffered_past_the_outbox_window_is_published_without_a_fetch_when_all_slots_are_taken_test() ->
     Skipped = read_counter_value(?CNT_READ_STATE_SKIPPED),
-    ok = meck:new(push_worker_pool, [passthrough, no_link]),
+    ok = ensure_read_state_table(),
+    true = ets:insert(?READ_STATE_TABLE, {?READ_STATE_FETCH_SLOTS, ?MAX_READ_STATE_FETCHES}),
     try
-        ok = meck:expect(push_worker_pool, utilization_pct, fun() ->
+        Published = with_buffered_dm_push(read_through(#{123 => <<"456">>}), fun() ->
-            ?READ_STATE_FETCH_MAX_POOL_PCT
+            ok = handle_buffered_message_creates([
+                buffered_dm_params(stale_buffered_at(), 123, 456)
+            ]),
+            length(published_message_ids(0))
         end),
-        Published = with_buffered_dms(
+        ?assertEqual({1, 0}, Published),
-            [{stale_buffered_at(), 123, 456}], read_through(#{123 => <<"456">>})
+        ?assertEqual(?MAX_READ_STATE_FETCHES, read_state_counter(?READ_STATE_FETCH_SLOTS))
-        ),
-        ?assertEqual({1, 0}, Published)
     after
-        meck:unload(push_worker_pool)
+        true = ets:insert(?READ_STATE_TABLE, {?READ_STATE_FETCH_SLOTS, 0})
     end,
     ?assertEqual(Skipped + 1, read_counter_value(?CNT_READ_STATE_SKIPPED)).
 
@@ -1105,8 +1134,7 @@ read_through(ReadMessageIds) ->
     fun(ChannelId) -> {ok, #{<<"last_message_id">> => maps:get(ChannelId, ReadMessageIds)}} end.
 
 read_counter_value(Key) ->
-    ok = push_worker_pool:init_counter(),
+    case read_state_counter(Key) of
-    case read_counter(Key) of
         Value when is_integer(Value) -> Value;
         unavailable -> 0
     end.
@@ -1126,7 +1154,7 @@ with_buffered_dms(Entries, ReadStateReply) ->
 wait_for_read_state_fetches(0) ->
     erlang:error(read_state_fetch_still_in_flight);
 wait_for_read_state_fetches(Attempts) ->
-    case read_counter(?READ_STATE_FETCH_SLOTS) of
+    case read_state_counter(?READ_STATE_FETCH_SLOTS) of
         InFlight when is_integer(InFlight), InFlight > 0 ->
             timer:sleep(20),
             wait_for_read_state_fetches(Attempts - 1);

fluxer_gateway/test/push_read_dm_tests.erl

@@ -85,15 +85,16 @@ a_dm_older_than_the_reply_arriving_after_it_is_not_pushed_test() ->
     end),
     ?assertEqual([?REPLY_MSG + 1], Pushed).
 
-a_buffered_dm_without_a_buffer_time_is_still_pushed_test() ->
+a_recent_buffered_dm_without_a_buffer_time_is_pushed_test() ->
+    Recent = recent_message_id(),
     Pushed = with_desktop_presence(fun(PresencePid, _SessionPid) ->
         sys:replace_state(PresencePid, fun(State) ->
             State#{
                 push_buffer := [
                     #{
                         channel_id => ?DM,
-                        message_id => ?PARTNER_MSG,
+                        message_id => Recent,
-                        params => #{message_data => dm_message(?PARTNER_MSG, ?PARTNER)}
+                        params => #{message_data => dm_message(Recent, ?PARTNER)}
                     }
                 ]
             }
@@ -102,7 +103,7 @@ a_buffered_dm_without_a_buffer_time_is_still_pushed_test() ->
         sync(PresencePid),
         pushed_message_ids([])
     end),
-    ?assertEqual([?PARTNER_MSG], Pushed).
+    ?assertEqual([Recent], Pushed).
 
 a_message_ack_drops_the_buffered_dm_push_test() ->
     Pushed = with_desktop_presence(fun(PresencePid, _SessionPid) ->
@@ -198,12 +199,113 @@ stale_read_state_fetches_are_capped_per_node_and_fail_open_past_the_cap_test() -
     ?assertEqual(64, InFlight),
     ?assertEqual(Flushes - 64, Published),
     ?assertEqual(0, Late),
-    ?assertEqual(0, read_state_fetches_in_flight()).
+    ?assertEqual(0, read_state_fetches_in_flight()),
+    ?assertEqual(undefined, ets:whereis(push_worker_counter)).
+
+a_read_dm_buffered_past_the_outbox_window_is_not_pushed_by_a_presence_node_test() ->
+    Self = self(),
+    {Fetches, Published, Suppressed} = with_presence_node(fun() ->
+        with_desktop_presence(fun(PresencePid, _SessionPid) ->
+            ok = meck:expect(push, handle_buffered_message_creates, fun(ParamsList) ->
+                meck:passthrough([ParamsList])
+            end),
+            ok = meck:expect(rpc_client, call, fun(Request, _Timeout) ->
+                Self ! {fetching, maps:get(<<"channel_id">>, Request)},
+                {ok, #{<<"last_message_id">> => integer_to_binary(?PARTNER_MSG)}}
+            end),
+            dispatch(PresencePid, message_create, dm_message(?PARTNER_MSG, ?PARTNER)),
+            sys:replace_state(PresencePid, fun(#{push_buffer := Buffer} = State) ->
+                State#{
+                    push_buffer := [
+                        Entry#{buffered_at => erlang:system_time(millisecond) - 600000}
+                     || Entry <- Buffer
+                    ]
+                }
+            end),
+            gen_server:cast(PresencePid, {presence_update, desktop_request(true)}),
+            sync(PresencePid),
+            {Blocked, Immediate} = collect_read_state_traffic(1, [], 0),
+            Late = collect_published(0),
+            {Blocked, Immediate + Late, read_state_counters(push_read_state_suppressed)}
+        end)
+    end),
+    ?assertEqual([integer_to_binary(?DM)], Fetches),
+    ?assertEqual(0, Published),
+    ?assertEqual([{push_read_state_suppressed, 1}], Suppressed),
+    ?assertEqual(undefined, ets:whereis(push_worker_counter)).
+
+a_read_dm_buffered_before_buffer_times_existed_is_aged_from_its_id_test() ->
+    Self = self(),
+    {Fetches, Published, Suppressed} = with_presence_node(fun() ->
+        with_desktop_presence(fun(PresencePid, _SessionPid) ->
+            ok = meck:expect(push, handle_buffered_message_creates, fun(ParamsList) ->
+                meck:passthrough([ParamsList])
+            end),
+            ok = meck:expect(rpc_client, call, fun(Request, _Timeout) ->
+                Self ! {fetching, maps:get(<<"channel_id">>, Request)},
+                {ok, #{<<"last_message_id">> => integer_to_binary(?PARTNER_MSG)}}
+            end),
+            dispatch(PresencePid, message_create, dm_message(?PARTNER_MSG, ?PARTNER)),
+            sys:replace_state(PresencePid, fun(#{push_buffer := Buffer} = State) ->
+                State#{push_buffer := [maps:remove(buffered_at, Entry) || Entry <- Buffer]}
+            end),
+            gen_server:cast(PresencePid, {presence_update, desktop_request(true)}),
+            sync(PresencePid),
+            {Blocked, Immediate} = collect_read_state_traffic(1, [], 0),
+            Late = collect_published(0),
+            {Blocked, Immediate + Late, read_state_counters(push_read_state_suppressed)}
+        end)
+    end),
+    ?assertEqual([integer_to_binary(?DM)], Fetches),
+    ?assertEqual(0, Published),
+    ?assertEqual([{push_read_state_suppressed, 1}], Suppressed).
+
+recent_message_id() ->
+    (erlang:system_time(millisecond) - 1420070400000) bsl 22.
+
+with_presence_node(Fun) ->
+    Self = self(),
+    ok = delete_table(push_worker_counter),
+    ok = delete_table(push_read_state_counters),
+    ok = meck:new(fluxer_gateway_env, [passthrough, no_link]),
+    ok = meck:new(gateway_node_router, [passthrough, no_link]),
+    ok = meck:new(rpc_client, [passthrough, no_link]),
+    try
+        ok = meck:expect(fluxer_gateway_env, get, fun
+            (push_enabled) -> true;
+            (Key) -> meck:passthrough([Key])
+        end),
+        ok = meck:expect(gateway_node_router, owner_node_result, fun(_Key, push) ->
+            Self ! published,
+            {error, test}
+        end),
+        with_ets_owner(Fun)
+    after
+        meck:unload(rpc_client),
+        meck:unload(gateway_node_router),
+        meck:unload(fluxer_gateway_env)
+    end.
+
+with_ets_owner(Fun) ->
+    {ok, Owner} = guild_ets_owner:start_link(),
+    unlink(Owner),
+    try
+        Fun()
+    after
+        stop_quietly(Owner)
+    end.
+
+delete_table(Table) ->
+    try ets:delete(Table) of
+        true -> ok
+    catch
+        error:badarg -> ok
+    end.
 
 with_stale_read_state_mocks(Fun) ->
     Self = self(),
-    ok = push_worker_pool:init_counter(),
+    ok = delete_table(push_worker_counter),
-    true = ets:delete(push_worker_counter, read_state_fetches_in_flight),
+    ok = delete_table(push_read_state_counters),
     ok = meck:new(fluxer_gateway_env, [passthrough, no_link]),
     ok = meck:new(gateway_node_router, [passthrough, no_link]),
     ok = meck:new(rpc_client, [passthrough, no_link]),
@@ -224,7 +326,7 @@ with_stale_read_state_mocks(Fun) ->
                 {error, timeout}
             end
         end),
-        Fun()
+        with_ets_owner(Fun)
     after
         meck:unload(rpc_client),
         meck:unload(gateway_node_router),
@@ -271,11 +373,18 @@ wait_for_released_fetch_slots(Attempts) ->
     end.
 
 read_state_fetches_in_flight() ->
-    case ets:lookup(push_worker_counter, read_state_fetches_in_flight) of
+    case read_state_counters(read_state_fetches_in_flight) of
         [{_, InFlight}] -> InFlight;
         [] -> 0
     end.
 
+read_state_counters(Key) ->
+    try
+        ets:lookup(push_read_state_counters, Key)
+    catch
+        error:badarg -> []
+    end.
+
 an_outbox_read_watermark_filters_a_later_job_for_the_read_message_test() ->
     {Sent, Stats} = with_outbox(60000, fun() ->
         ok = push_outbox:truncate_read(?USER, ?DM, ?PARTNER_MSG),