diff --git a/fluxer_gateway/src/presence/presence_update.erl b/fluxer_gateway/src/presence/presence_update.erl index b3cff273d..2791558fc 100644 --- a/fluxer_gateway/src/presence/presence_update.erl +++ b/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), diff --git a/fluxer_gateway/src/push/push.erl b/fluxer_gateway/src/push/push.erl index 393559305..8119dca8d 100644 --- a/fluxer_gateway/src/push/push.erl +++ b/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 - push_worker_pool:utilization_pct() < ?READ_STATE_FETCH_MAX_POOL_PCT andalso - acquire_read_state_fetch_slot() - of + case 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_fetch_failures => read_counter(?CNT_READ_STATE_FAILURES), - read_state_fetch_skipped => read_counter(?CNT_READ_STATE_SKIPPED), - read_state_fetches_in_flight => read_counter(?READ_STATE_FETCH_SLOTS) + read_state_suppressed => read_state_counter(?CNT_READ_STATE_SUPPRESSED), + read_state_fetch_failures => read_state_counter(?CNT_READ_STATE_FAILURES), + read_state_fetch_skipped => read_state_counter(?CNT_READ_STATE_SKIPPED), + 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() -> - ?READ_STATE_FETCH_MAX_POOL_PCT + Published = with_buffered_dm_push(read_through(#{123 => <<"456">>}), fun() -> + ok = handle_buffered_message_creates([ + buffered_dm_params(stale_buffered_at(), 123, 456) + ]), + length(published_message_ids(0)) end), - Published = with_buffered_dms( - [{stale_buffered_at(), 123, 456}], read_through(#{123 => <<"456">>}) - ), - ?assertEqual({1, 0}, Published) + ?assertEqual({1, 0}, Published), + ?assertEqual(?MAX_READ_STATE_FETCHES, read_state_counter(?READ_STATE_FETCH_SLOTS)) 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_counter(Key) of + case read_state_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); diff --git a/fluxer_gateway/test/push_read_dm_tests.erl b/fluxer_gateway/test/push_read_dm_tests.erl index 77bdbcc38..b2beb9728 100644 --- a/fluxer_gateway/test/push_read_dm_tests.erl +++ b/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, - params => #{message_data => dm_message(?PARTNER_MSG, ?PARTNER)} + message_id => Recent, + 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(), - true = ets:delete(push_worker_counter, read_state_fetches_in_flight), + 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]), @@ -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),