diff --git a/fluxer_docs/src/content/docs/gateway/limits-and-rate-limits.md b/fluxer_docs/src/content/docs/gateway/limits-and-rate-limits.md index 338302485..15c650795 100644 --- a/fluxer_docs/src/content/docs/gateway/limits-and-rate-limits.md +++ b/fluxer_docs/src/content/docs/gateway/limits-and-rate-limits.md @@ -48,7 +48,7 @@ Presence Update accepts five commands per WebSocket in a rolling 20-second windo Voice State Update processes the first two commands per session in a rolling one-second window immediately. Later updates enter a per-session queue that holds at most 64 commands and drains one command every 500 ms. A newer update replaces an older queued update for the same `guild_id` and `connection_id` pair, and a full queue discards its oldest entry before accepting the new one. -Request Guild Members has one command-specific budget. A bot requesting a complete member list is limited to one accepted request per guild every 30 seconds, and a request inside that window produces [Rate Limited](/gateway/events/#rate-limited). The budget is keyed by the account and the guild together, so reconnecting does not reset it. +Request Guild Members has three command-specific budgets. Each account may send 12 requests in a rolling 10-second window, and each guild accepts 40 requests in a rolling 10-second window across all accounts. A request over either of these budgets is discarded without a result and without a close. A bot requesting a complete member list is limited to one accepted request per guild every 30 seconds, and a request inside that window produces [Rate Limited](/gateway/events/#rate-limited). The budget is keyed by the account and the guild together, so reconnecting does not reset it. ## Bounded commands diff --git a/fluxer_gateway/src/gateway/gateway_cluster_membership.erl b/fluxer_gateway/src/gateway/gateway_cluster_membership.erl index 1d28a5398..750a81361 100644 --- a/fluxer_gateway/src/gateway/gateway_cluster_membership.erl +++ b/fluxer_gateway/src/gateway/gateway_cluster_membership.erl @@ -140,8 +140,10 @@ handle_cast(_Msg, State) -> handle_info({cluster_peers_changed, Peers}, State) -> {noreply, apply_discovered(normalize_nodes(Peers), State)}; handle_info({nodeup, Node}, State) when is_atom(Node) -> + ok = presence_cache:handle_nodeup(Node), {noreply, maybe_add_member(Node, State)}; handle_info({nodedown, Node}, State) when is_atom(Node) -> + ok = maybe_start_presence_grace(Node, State), {noreply, maybe_remove_member(Node, State)}; handle_info(refresh_roles, State) -> State1 = reconcile_against_connected_nodes(State), @@ -211,6 +213,13 @@ force_discovery_refresh_safely() -> exit:_ -> ok end. +-spec maybe_start_presence_grace(node(), state()) -> ok. +maybe_start_presence_grace(Node, #{members := Members}) -> + case lists:member(Node, Members) andalso Node =/= node() of + true -> presence_cache:handle_nodedown(Node); + false -> ok + end. + -spec maybe_remove_member(node(), state()) -> state(). maybe_remove_member(Node, #{members := Members} = State) -> case lists:member(Node, Members) andalso Node =/= node() of diff --git a/fluxer_gateway/src/guild/guild.erl b/fluxer_gateway/src/guild/guild.erl index dedd2d892..e99e7e386 100644 --- a/fluxer_gateway/src/guild/guild.erl +++ b/fluxer_gateway/src/guild/guild.erl @@ -516,6 +516,12 @@ terminate(Reason, State) when is_map(State) -> end, "ets_cleanup" ), + safe_cleanup( + fun() -> + cleanup_passive_sync_registry(State) + end, + "passive_sync_cleanup" + ), safe_cleanup( fun() -> cleanup_voice_server(State) @@ -578,6 +584,15 @@ cleanup_per_guild_ets(State) -> safe_delete_ets(maps:get(viewable_channels_cache, State, undefined)), ok. +-spec cleanup_passive_sync_registry(guild_state()) -> ok. +cleanup_passive_sync_registry(#{id := GuildId} = State) when is_integer(GuildId) -> + maps:foreach( + fun(SessionId, _Session) -> passive_sync_registry:delete(SessionId, GuildId) end, + maps:get(sessions, State, #{}) + ); +cleanup_passive_sync_registry(_State) -> + ok. + -spec cleanup_voice_server(guild_state()) -> ok. cleanup_voice_server(State) -> case maps:get(voice_server_pid, State, undefined) of diff --git a/fluxer_gateway/src/guild/guild_request_members.erl b/fluxer_gateway/src/guild/guild_request_members.erl index 5f5c47ff3..737f6d44e 100644 --- a/fluxer_gateway/src/guild/guild_request_members.erl +++ b/fluxer_gateway/src/guild/guild_request_members.erl @@ -180,9 +180,16 @@ normalize_nonce(_) -> -spec process_request(map(), pid(), session_state()) -> ok | {error, atom()}. process_request(Request, SocketPid, SessionState) -> + UserId = snowflake_id:parse_optional(maps:get(user_id, SessionState)), + case guild_request_members_filter:check_request_rate_limit(UserId) of + ok -> process_admitted_request(Request, UserId, SocketPid, SessionState); + {error, Reason} -> {error, Reason} + end. + +-spec process_admitted_request(map(), integer() | undefined, pid(), session_state()) -> + ok | {error, atom()}. +process_admitted_request(Request, UserId, SocketPid, SessionState) -> #{guild_ids := ReqGuildIds, query := Query, limit := Limit, user_ids := UserIds} = Request, - UserIdBin = maps:get(user_id, SessionState), - UserId = snowflake_id:parse_optional(UserIdBin), IsBot = truthy(maps:get(bot, SessionState, false)), case guild_request_members_filter:enforce_single_guild_for_bots(IsBot, ReqGuildIds) of ok -> @@ -248,6 +255,22 @@ dispatch_guild_requests(GIds, UID, Bot, Q, L, UI, Req, SP, SS) -> session_state() ) -> ok | {error, atom()}. process_single_guild(GId, UID, Bot, Q, L, UI, Req, _SP, SS) -> + case guild_request_members_filter:check_guild_request_rate_limit(GId) of + ok -> process_admitted_guild(GId, UID, Bot, Q, L, UI, Req, SS); + {error, Reason} -> {error, Reason} + end. + +-spec process_admitted_guild( + integer(), + integer() | undefined, + boolean(), + binary(), + non_neg_integer(), + [integer()], + map(), + session_state() +) -> ok | {error, atom()}. +process_admitted_guild(GId, UID, Bot, Q, L, UI, Req, SS) -> IsFull = guild_request_members_filter:is_full_member_list(Q, L, UI), case guild_request_members_filter:check_full_list_bot_rate_limit(IsFull, Bot, UID, GId) of ok -> @@ -320,6 +343,22 @@ try_guild_or_keep_processed(GId, UID, Bot, Q, L, UI, Req, SS, Acc) -> session_state() ) -> boolean(). try_guild(GId, UID, Bot, Q, L, UI, Req, SS) -> + case guild_request_members_filter:check_guild_request_rate_limit(GId) of + ok -> try_admitted_guild(GId, UID, Bot, Q, L, UI, Req, SS); + {error, _Reason} -> false + end. + +-spec try_admitted_guild( + integer(), + integer() | undefined, + boolean(), + binary(), + non_neg_integer(), + [integer()], + map(), + session_state() +) -> boolean(). +try_admitted_guild(GId, UID, Bot, Q, L, UI, Req, SS) -> IsFull = guild_request_members_filter:is_full_member_list(Q, L, UI), case guild_request_members_filter:check_full_list_bot_rate_limit(IsFull, Bot, UID, GId) of ok -> diff --git a/fluxer_gateway/src/guild/guild_request_members_filter.erl b/fluxer_gateway/src/guild/guild_request_members_filter.erl index b06ccf1b8..25a1f1763 100644 --- a/fluxer_gateway/src/guild/guild_request_members_filter.erl +++ b/fluxer_gateway/src/guild/guild_request_members_filter.erl @@ -16,6 +16,10 @@ -export_type([session_state/0]). +-ifdef(TEST). +-export([prune_expired_entries/2]). +-endif. + -define(REQUEST_MEMBERS_RATE_LIMIT_TABLE, guild_request_members_rate_limit). -define(REQUEST_MEMBERS_RATE_LIMIT_WINDOW_MS, 10000). -define(REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS, 12). @@ -24,6 +28,7 @@ -define(REQUEST_MEMBERS_GUILD_RATE_LIMIT_MAX_EVENTS, 40). -define(FULL_LIST_BOT_RATE_LIMIT_TABLE, guild_request_members_bot_full_list_rate_limit). -define(FULL_LIST_BOT_RATE_LIMIT_WINDOW_MS, 30000). +-define(RATE_LIMIT_CLEANUP_INTERVAL_MS, 60000). -type session_state() :: map(). @@ -106,7 +111,7 @@ check_full_list_bot_rate_limit(true, true, _UserId, _GuildId) -> -spec check_bot_rate_limit_window(integer(), integer()) -> ok | {rate_limited, non_neg_integer()}. check_bot_rate_limit_window(UserId, GuildId) -> - ensure_ets_table(?FULL_LIST_BOT_RATE_LIMIT_TABLE), + ensure_ets_table(?FULL_LIST_BOT_RATE_LIMIT_TABLE, ?FULL_LIST_BOT_RATE_LIMIT_WINDOW_MS), Now = erlang:system_time(millisecond), Key = {UserId, GuildId}, Window = ?FULL_LIST_BOT_RATE_LIMIT_WINDOW_MS, @@ -120,25 +125,40 @@ check_bot_rate_limit_window(UserId, GuildId) -> -spec check_request_rate_limit(integer() | undefined) -> ok | {error, atom()}. check_request_rate_limit(UserId) when is_integer(UserId), UserId > 0 -> - ensure_ets_table(?REQUEST_MEMBERS_RATE_LIMIT_TABLE), - check_sliding_window( - ?REQUEST_MEMBERS_RATE_LIMIT_TABLE, - UserId, - ?REQUEST_MEMBERS_RATE_LIMIT_WINDOW_MS, - ?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS - ); + case gateway_handler_rate_limit:rate_limits_disabled() of + true -> + ok; + false -> + ensure_ets_table( + ?REQUEST_MEMBERS_RATE_LIMIT_TABLE, ?REQUEST_MEMBERS_RATE_LIMIT_WINDOW_MS + ), + check_sliding_window( + ?REQUEST_MEMBERS_RATE_LIMIT_TABLE, + UserId, + ?REQUEST_MEMBERS_RATE_LIMIT_WINDOW_MS, + ?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS + ) + end; check_request_rate_limit(_) -> {error, invalid_session}. -spec check_guild_request_rate_limit(integer()) -> ok | {error, atom()}. check_guild_request_rate_limit(GuildId) when is_integer(GuildId), GuildId > 0 -> - ensure_ets_table(?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE), - check_sliding_window( - ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE, - GuildId, - ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_WINDOW_MS, - ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_MAX_EVENTS - ); + case gateway_handler_rate_limit:rate_limits_disabled() of + true -> + ok; + false -> + ensure_ets_table( + ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE, + ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_WINDOW_MS + ), + check_sliding_window( + ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE, + GuildId, + ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_WINDOW_MS, + ?REQUEST_MEMBERS_GUILD_RATE_LIMIT_MAX_EVENTS + ) + end; check_guild_request_rate_limit(_) -> {error, invalid_guild_id}. @@ -166,30 +186,56 @@ check_window_count(Table, Key, Now, Recent, _MaxEvents) -> ets:insert(Table, {Key, [Now | Recent]}), ok. --spec ensure_ets_table(atom()) -> ok. -ensure_ets_table(Name) -> +-spec ensure_ets_table(atom(), pos_integer()) -> ok. +ensure_ets_table(Name, WindowMs) -> case ets:whereis(Name) of undefined -> - create_ets_table(Name); + create_ets_table(Name, WindowMs); _ -> ok end. --spec create_ets_table(atom()) -> ok. -create_ets_table(Name) -> +-spec create_ets_table(atom(), pos_integer()) -> ok. +create_ets_table(Name, WindowMs) -> try - _ = ets:new(Name, [ - named_table, - public, - set, - {read_concurrency, true}, - {write_concurrency, true} - ]), + Tid = ets:new( + Name, + [ + named_table, + public, + set, + {read_concurrency, true}, + {write_concurrency, true} + ] ++ guild_ets_utils:heir_options() + ), + _ = spawn(fun() -> cleanup_loop(Tid, WindowMs) end), ok catch error:badarg -> ok end. +-spec cleanup_loop(ets:table(), pos_integer()) -> ok. +cleanup_loop(Table, WindowMs) -> + ok = gateway_retry_timer:wait(?RATE_LIMIT_CLEANUP_INTERVAL_MS), + case prune_expired_entries(Table, WindowMs) of + ok -> cleanup_loop(Table, WindowMs); + gone -> ok + end. + +-spec prune_expired_entries(ets:table(), pos_integer()) -> ok | gone. +prune_expired_entries(Table, WindowMs) -> + Cutoff = erlang:system_time(millisecond) - WindowMs, + try + _ = ets:select_delete(Table, [ + {{'_', []}, [], [true]}, + {{'_', '$1'}, [{is_list, '$1'}, {'=<', {hd, '$1'}, Cutoff}], [true]}, + {{'_', '$1'}, [{is_integer, '$1'}, {'=<', '$1', Cutoff}], [true]} + ]), + ok + catch + error:badarg -> gone + end. + -spec dispatch_full_list_rate_limited( session_state(), integer(), map(), non_neg_integer() ) -> ok. diff --git a/fluxer_gateway/src/guild/guild_state.erl b/fluxer_gateway/src/guild/guild_state.erl index f6fef7fce..0b982dc9e 100644 --- a/fluxer_gateway/src/guild/guild_state.erl +++ b/fluxer_gateway/src/guild/guild_state.erl @@ -155,10 +155,7 @@ post_update_role(guild_role_delete, EventData, OldState, NewState) -> guild_state(). resync_roles_after_permission_change(RoleIds, OldState, NewState) -> Recomputed = guild_state_roles:recompute_visibility_for_roles(RoleIds, OldState, NewState), - lists:foreach( - fun(RoleId) -> guild_voice_permission_sync:sync_users_with_role(RoleId, Recomputed) end, - RoleIds - ), + ok = guild_voice_permission_sync:sync_roles_after_change(RoleIds, OldState, Recomputed), Recomputed. -spec post_update_channel(event(), event_data(), guild_state(), guild_state()) -> guild_state(). diff --git a/fluxer_gateway/src/guild/passive_sync_registry.erl b/fluxer_gateway/src/guild/passive_sync_registry.erl index 186bdaba2..0e53c0522 100644 --- a/fluxer_gateway/src/guild/passive_sync_registry.erl +++ b/fluxer_gateway/src/guild/passive_sync_registry.erl @@ -7,8 +7,7 @@ init/0, store/3, lookup/2, - delete/2, - delete_all_for_session/1 + delete/2 ]). -ifdef(TEST). @@ -72,12 +71,6 @@ delete(SessionId, GuildId) -> ets:delete(?TABLE, Key), ok. --spec delete_all_for_session(session_id()) -> ok. -delete_all_for_session(SessionId) -> - ensure_table(), - ets:match_delete(?TABLE, {{SessionId, '_'}, '_'}), - ok. - -spec ensure_table() -> ok. ensure_table() -> case ets:whereis(?TABLE) of @@ -152,38 +145,6 @@ delete_removes_entry_test() -> ), cleanup_table(). -delete_all_for_session_removes_all_guilds_test() -> - cleanup_table(), - ok = init(), - SessionId = <<"session_1">>, - OtherSessionId = <<"session_2">>, - State1 = make_passive_state(<<"ch1">>, <<"msg1">>), - State2 = make_passive_state(<<"ch2">>, <<"msg2">>), - OtherState = make_passive_state(<<"ch3">>, <<"msg3">>), - ok = store(SessionId, 100, State1), - ok = store(SessionId, 200, State2), - ok = store(OtherSessionId, 100, OtherState), - ok = delete_all_for_session(SessionId), - Empty = empty_passive_state(), - ?assertEqual(Empty, lookup(SessionId, 100)), - ?assertEqual(Empty, lookup(SessionId, 200)), - ?assertEqual(OtherState, lookup(OtherSessionId, 100)), - cleanup_table(). - -make_passive_state(Ch, Msg) -> - #{ - previous_passive_updates => #{Ch => Msg}, - previous_passive_channel_versions => #{}, - previous_passive_voice_states => #{} - }. - -empty_passive_state() -> - #{ - previous_passive_updates => #{}, - previous_passive_channel_versions => #{}, - previous_passive_voice_states => #{} - }. - store_overwrites_existing_test() -> cleanup_table(), ok = init(), diff --git a/fluxer_gateway/src/guild/voice/guild_voice_permission_sync.erl b/fluxer_gateway/src/guild/voice/guild_voice_permission_sync.erl index 80f5471e8..07e1cb2ac 100644 --- a/fluxer_gateway/src/guild/voice/guild_voice_permission_sync.erl +++ b/fluxer_gateway/src/guild/voice/guild_voice_permission_sync.erl @@ -6,8 +6,7 @@ -export([ sync_user_voice_permissions/2, sync_all_voice_permissions_for_channels/2, - sync_users_with_role/2, - maybe_sync_permissions_on_role_update/2, + sync_roles_after_change/3, maybe_sync_permissions_on_member_update/2 ]). @@ -96,14 +95,6 @@ sync_channel_voice_state_permissions(GuildId, VoiceState, State) -> UserId -> sync_voice_state_permissions(GuildId, UserId, VoiceState, State) end. --spec maybe_sync_permissions_on_role_update(map(), guild_state()) -> ok. -maybe_sync_permissions_on_role_update(RoleUpdate, State) -> - RoleId = snowflake_id:parse_optional(maps:get(<<"id">>, RoleUpdate, undefined)), - case RoleId of - undefined -> ok; - RoleIdInt -> sync_users_with_role(RoleIdInt, State) - end. - -spec maybe_sync_permissions_on_member_update(map(), guild_state()) -> ok. maybe_sync_permissions_on_member_update(MemberUpdate, State) -> UserId = get_member_user_id(MemberUpdate), @@ -241,54 +232,66 @@ enforce_voice_permissions_in_livekit( ok end. --spec sync_users_with_role(integer(), guild_state()) -> ok. -sync_users_with_role(RoleId, State) -> - VoiceStates = guild_voice_lifecycle:authoritative_voice_states(State), +-spec sync_roles_after_change([integer()], guild_state(), guild_state()) -> ok. +sync_roles_after_change([], _OldState, _State) -> + ok; +sync_roles_after_change(RoleIds, OldState, State) -> case state_guild_id(State) of undefined -> ok; GuildId -> - do_sync_users_with_role(GuildId, RoleId, VoiceStates, State) + Affected = affected_role_users(GuildId, RoleIds, OldState, State), + sync_affected_voice_states(GuildId, Affected, State) end. --spec do_sync_users_with_role(integer(), integer(), map(), guild_state()) -> ok. -do_sync_users_with_role(GuildId, RoleId, VoiceStates, State) -> - RoleUsers = user_ids_with_role(RoleId, State), +-spec sync_affected_voice_states(integer(), all | sets:set(user_id()), guild_state()) -> ok. +sync_affected_voice_states(GuildId, Affected, State) -> maps:foreach( fun(_ConnId, VoiceState) -> - maybe_sync_role_voice_state(GuildId, RoleUsers, VoiceState, State) + maybe_sync_affected_voice_state(GuildId, Affected, VoiceState, State) end, - VoiceStates - ), - ok. + guild_voice_lifecycle:authoritative_voice_states(State) + ). --spec maybe_sync_role_voice_state(integer(), sets:set(user_id()), voice_state(), guild_state()) -> - ok. -maybe_sync_role_voice_state(GuildId, RoleUsers, VoiceState, State) -> - UserId = voice_state_utils:voice_state_user_id(VoiceState), - case UserId of - undefined -> - ok; - _ -> - sync_voice_state_permissions_if_role_matches( - GuildId, RoleUsers, UserId, VoiceState, State +-spec affected_role_users(integer(), [integer()], guild_state(), guild_state()) -> + all | sets:set(user_id()). +affected_role_users(GuildId, RoleIds, OldState, State) -> + case lists:member(GuildId, RoleIds) of + true -> + all; + false -> + OldIndex = guild_data_index:member_role_index(maps:get(data, OldState, #{})), + NewIndex = guild_data_index:member_role_index(maps:get(data, State, #{})), + sets:from_list( + lists:append([ + maps:keys(maps:get(RoleId, OldIndex, #{})) ++ + maps:keys(maps:get(RoleId, NewIndex, #{})) + || RoleId <- RoleIds + ]) ) end. --spec sync_voice_state_permissions_if_role_matches( - integer(), sets:set(user_id()), user_id(), voice_state(), guild_state() +-spec maybe_sync_affected_voice_state( + integer(), all | sets:set(user_id()), voice_state(), guild_state() ) -> ok. -sync_voice_state_permissions_if_role_matches(GuildId, RoleUsers, UserId, VoiceState, State) -> - case sets:is_element(UserId, RoleUsers) of +maybe_sync_affected_voice_state(GuildId, Affected, VoiceState, State) -> + case voice_state_utils:voice_state_user_id(VoiceState) of + undefined -> ok; + UserId -> sync_if_affected(GuildId, UserId, Affected, VoiceState, State) + end. + +-spec sync_if_affected( + integer(), user_id(), all | sets:set(user_id()), voice_state(), guild_state() +) -> ok. +sync_if_affected(GuildId, UserId, Affected, VoiceState, State) -> + case is_affected(UserId, Affected) of true -> sync_voice_state_permissions(GuildId, UserId, VoiceState, State); false -> ok end. --spec user_ids_with_role(integer(), guild_state()) -> sets:set(user_id()). -user_ids_with_role(RoleId, State) -> - Data = maps:get(data, State, #{}), - RoleIndex = guild_data_index:member_role_index(Data), - sets:from_list(maps:keys(maps:get(RoleId, RoleIndex, #{}))). +-spec is_affected(user_id(), all | sets:set(user_id())) -> boolean(). +is_affected(_UserId, all) -> true; +is_affected(UserId, Affected) -> sets:is_element(UserId, Affected). -spec get_member_user_id(map()) -> user_id() | undefined. get_member_user_id(MemberUpdate) -> @@ -377,7 +380,7 @@ maybe_sync_permissions_on_member_update_no_role_change_test() -> }, ?assertEqual(ok, maybe_sync_permissions_on_member_update(MemberUpdate, State)). -maybe_sync_permissions_on_role_update_uses_role_index_test() -> +sync_roles_after_change_uses_role_index_test() -> TestFun = make_sync_test_fun(), GuildId = 42, RoleId = 999, @@ -408,9 +411,7 @@ maybe_sync_permissions_on_role_update_uses_role_index_test() -> GuildId, RoleId, OtherRoleId, UserId, OtherUserId, ChannelId, Permissions ) }, - ok = maybe_sync_permissions_on_role_update( - #{<<"id">> => integer_to_binary(RoleId)}, State - ), + ok = sync_roles_after_change([RoleId], State, State), receive {synced, GuildId, ChannelId, UserId, <<"conn1">>, _Perms} -> ok after 200 -> diff --git a/fluxer_gateway/test/gateway_cluster_membership_tests.erl b/fluxer_gateway/test/gateway_cluster_membership_tests.erl index 2e7082f36..a1770a1ee 100644 --- a/fluxer_gateway/test/gateway_cluster_membership_tests.erl +++ b/fluxer_gateway/test/gateway_cluster_membership_tests.erl @@ -291,6 +291,51 @@ update_members_role_lookups_run_concurrently_test() -> ?assertEqual(lists:usort(Nodes), maps:get(sessions, maps:get(members_by_role, State1))) end). +member_nodedown_starts_presence_grace_and_nodeup_cancels_it_test() -> + with_clean_cluster_terms(fun() -> + with_presence_cache(fun(CachePid) -> + Peer = 'peer@a', + State0 = (base_state(#{members => lists:usort([node(), Peer])}))#{ + discovered => [Peer] + }, + {noreply, State1} = gateway_cluster_membership:handle_info( + {nodedown, Peer}, State0 + ), + ?assertNot(lists:member(Peer, maps:get(members, State1))), + ?assert(maps:is_key(Peer, pending_nodedown_cleanups(CachePid))), + {noreply, State2} = gateway_cluster_membership:handle_info({nodeup, Peer}, State1), + ?assert(lists:member(Peer, maps:get(members, State2))), + ?assertEqual(#{}, pending_nodedown_cleanups(CachePid)) + end) + end). + +non_member_nodedown_does_not_start_presence_grace_test() -> + with_clean_cluster_terms(fun() -> + with_presence_cache(fun(CachePid) -> + State0 = base_state(#{}), + {noreply, _State1} = gateway_cluster_membership:handle_info( + {nodedown, 'remsh@host'}, State0 + ), + ?assertEqual(#{}, pending_nodedown_cleanups(CachePid)) + end) + end). + +with_presence_cache(Fun) -> + {ok, Pid} = + case whereis(presence_cache) of + undefined -> presence_cache:start_link(); + Existing -> {ok, Existing} + end, + try + Fun(Pid) + after + unlink(Pid), + gen_server:stop(Pid) + end. + +pending_nodedown_cleanups(CachePid) -> + maps:get(pending_nodedown_cleanups, sys:get_state(CachePid)). + base_state(Overrides) -> Defaults = #{ members => [node()], diff --git a/fluxer_gateway/test/guild_request_members_filter_tests.erl b/fluxer_gateway/test/guild_request_members_filter_tests.erl index 28aedb591..3c1ad58f8 100644 --- a/fluxer_gateway/test/guild_request_members_filter_tests.erl +++ b/fluxer_gateway/test/guild_request_members_filter_tests.erl @@ -210,6 +210,143 @@ check_guild_request_rate_limit_invalid_guild_test() -> guild_request_members_filter:check_guild_request_rate_limit(invalid_guild_id()) ). +handle_request_consumes_user_budget_test() -> + with_rate_limits_enabled(fun() -> + UserId = 555000001, + clear_request_rate_limit(UserId), + SessionState = #{user_id => integer_to_binary(UserId), guilds => #{}, bot => false}, + Data = #{<<"guild_id">> => <<"777000001">>, <<"query">> => <<"a">>, <<"limit">> => 10}, + Results = [ + guild_request_members:handle_request(Data, self(), SessionState) + || _ <- lists:seq(1, ?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS) + ], + ?assertEqual( + lists:duplicate(?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS, {error, guild_not_found}), + Results + ), + ?assertEqual( + {error, rate_limited}, + guild_request_members:handle_request(Data, self(), SessionState) + ), + clear_request_rate_limit(UserId) + end). + +handle_request_refuses_guild_over_budget_test() -> + with_rate_limits_enabled(fun() -> + UserId = 555000002, + GuildIds = [777000002, 777000003], + SessionState = spent_guild_session(UserId, GuildIds), + [GuildIdBin | _] = GuildIdBins = [integer_to_binary(Id) || Id <- GuildIds], + ?assertEqual( + {error, rate_limited}, + guild_request_members:handle_request( + #{<<"guild_id">> => GuildIdBin, <<"query">> => <<"a">>}, self(), SessionState + ) + ), + ?assertEqual( + {error, guild_not_found}, + guild_request_members:handle_request( + #{<<"guild_ids">> => GuildIdBins, <<"query">> => <<"a">>}, self(), SessionState + ) + ), + clear_request_rate_limit(UserId), + lists:foreach(fun clear_guild_request_rate_limit/1, GuildIds) + end). + +spent_guild_session(UserId, GuildIds) -> + clear_request_rate_limit(UserId), + ensure_ets_table(?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE), + Now = erlang:system_time(millisecond), + Spent = lists:duplicate(?REQUEST_MEMBERS_GUILD_RATE_LIMIT_MAX_EVENTS, Now - 1000), + ets:insert(?REQUEST_MEMBERS_GUILD_RATE_LIMIT_TABLE, [{Id, Spent} || Id <- GuildIds]), + #{ + user_id => integer_to_binary(UserId), + guilds => maps:from_list([{Id, {self(), make_ref()}} || Id <- GuildIds]), + bot => false + }. + +handle_request_budgets_disabled_by_env_test() -> + OldValue = os:getenv("FLUXER_DISABLE_RATE_LIMITS"), + os:putenv("FLUXER_DISABLE_RATE_LIMITS", "true"), + try + UserId = 555000004, + clear_request_rate_limit(UserId), + SessionState = #{user_id => integer_to_binary(UserId), guilds => #{}, bot => false}, + Data = #{<<"guild_id">> => <<"777000004">>, <<"query">> => <<"a">>}, + Results = [ + guild_request_members:handle_request(Data, self(), SessionState) + || _ <- lists:seq(1, ?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS + 1) + ], + ?assertEqual( + lists:duplicate( + ?REQUEST_MEMBERS_RATE_LIMIT_MAX_EVENTS + 1, {error, guild_not_found} + ), + Results + ) + after + restore_env("FLUXER_DISABLE_RATE_LIMITS", OldValue) + end. + +rate_limit_tables_outlive_the_request_worker_test() -> + with_rate_limits_enabled(fun() -> + UserId = 555000005, + drop_table(?REQUEST_MEMBERS_RATE_LIMIT_TABLE), + {ok, OwnerPid} = start_ets_owner(), + try + ok = check_rate_limit_in_worker(UserId), + ?assertMatch( + [{UserId, [_]}], ets:lookup(?REQUEST_MEMBERS_RATE_LIMIT_TABLE, UserId) + ), + ?assertEqual(OwnerPid, ets:info(?REQUEST_MEMBERS_RATE_LIMIT_TABLE, owner)) + after + gen_server:stop(OwnerPid), + drop_table(?REQUEST_MEMBERS_RATE_LIMIT_TABLE) + end + end). + +check_rate_limit_in_worker(UserId) -> + Self = self(), + {Worker, Ref} = spawn_monitor(fun() -> + Self ! {checked, guild_request_members_filter:check_request_rate_limit(UserId)} + end), + receive + {checked, ok} -> ok + after 1000 -> error(worker_timeout) + end, + receive + {'DOWN', Ref, process, Worker, _} -> ok + after 1000 -> error(worker_alive) + end. + +drop_table(Name) -> + case ets:whereis(Name) of + undefined -> ok; + _ -> ets:delete(Name) + end. + +prune_expired_entries_drops_idle_keys_test() -> + Table = ets:new(request_members_prune_test, [public, set]), + Now = erlang:system_time(millisecond), + ets:insert(Table, [ + {idle_window, [Now - 20000, Now - 25000]}, + {active_window, [Now - 100, Now - 20000]}, + {idle_bot, Now - 40000}, + {active_bot, Now - 100} + ]), + ?assertEqual(ok, guild_request_members_filter:prune_expired_entries(Table, 10000)), + ?assertEqual( + [active_bot, active_window], + lists:sort([Key || {Key, _} <- ets:tab2list(Table)]) + ), + ets:delete(Table), + ?assertEqual(gone, guild_request_members_filter:prune_expired_entries(Table, 10000)). + +start_ets_owner() -> + case whereis(guild_ets_owner) of + undefined -> guild_ets_owner:start_link(); + Existing -> {ok, Existing} + end. + restore_env(Key, false) -> os:unsetenv(Key); restore_env(Key, Value) -> diff --git a/fluxer_gateway/test/guild_sessions_connect_tests.erl b/fluxer_gateway/test/guild_sessions_connect_tests.erl index d4bd03cb6..44a96b6f8 100644 --- a/fluxer_gateway/test/guild_sessions_connect_tests.erl +++ b/fluxer_gateway/test/guild_sessions_connect_tests.erl @@ -141,6 +141,72 @@ normalize_connect_queue_test() -> ?assertEqual(Q, guild_sessions_connect_cleanup:normalize_connect_queue(Q)), ?assertEqual(undefined, guild_sessions_connect_cleanup:normalize_connect_queue(undefined)). +session_down_clears_passive_sync_entry_test() -> + ok = passive_sync_registry:init(), + SessionId = <<"passive-down">>, + ok = passive_sync_registry:store(SessionId, 42, stored_passive_state()), + {Ref, Session, State} = down_state(SessionId), + {noreply, _} = guild_sessions_connect:handle_session_down(Ref, {SessionId, Session}, State), + ?assertEqual([], ets:lookup(passive_sync_registry, {SessionId, 42})). + +remove_session_clears_passive_sync_entry_test() -> + ok = passive_sync_registry:init(), + SessionId = <<"passive-remove">>, + ok = passive_sync_registry:store(SessionId, 42, stored_passive_state()), + _ = guild_sessions_connect:remove_session(SessionId, remove_session_state(SessionId)), + ?assertEqual([], ets:lookup(passive_sync_registry, {SessionId, 42})). + +passive_sync_registry_stays_bounded_across_session_cycles_test() -> + ok = passive_sync_registry:init(), + GuildRows = fun() -> length(ets:match(passive_sync_registry, {{'_', 42}, '_'})) end, + Before = GuildRows(), + lists:foreach( + fun(N) -> + SessionId = <<"cycle-", (integer_to_binary(N))/binary>>, + ok = passive_sync_registry:store(SessionId, 42, stored_passive_state()), + {Ref, Session, State} = down_state(SessionId), + {noreply, _} = guild_sessions_connect:handle_session_down( + Ref, {SessionId, Session}, State + ) + end, + lists:seq(1, 500) + ), + ?assertEqual(Before, GuildRows()). + +guild_terminate_clears_passive_sync_entries_test() -> + ok = passive_sync_registry:init(), + GuildId = 4242, + OtherGuildId = 4343, + lists:foreach( + fun({SessionId, GId}) -> + ok = passive_sync_registry:store(SessionId, GId, stored_passive_state()) + end, + [{<<"t1">>, GuildId}, {<<"t2">>, GuildId}, {<<"t1">>, OtherGuildId}] + ), + ok = guild:terminate(shutdown, #{ + id => GuildId, + sessions => #{<<"t1">> => #{}, <<"t2">> => #{}} + }), + ?assertEqual([], ets:lookup(passive_sync_registry, {<<"t1">>, GuildId})), + ?assertEqual([], ets:lookup(passive_sync_registry, {<<"t2">>, GuildId})), + ?assertEqual( + stored_passive_state(), passive_sync_registry:lookup(<<"t1">>, OtherGuildId) + ), + ok = passive_sync_registry:delete(<<"t1">>, OtherGuildId). + +stored_passive_state() -> + #{ + previous_passive_updates => #{<<"1">> => <<"2">>}, + previous_passive_channel_versions => #{}, + previous_passive_voice_states => #{} + }. + +down_state(SessionId) -> + State0 = remove_session_state(SessionId), + Session = maps:get(SessionId, maps:get(sessions, State0)), + Ref = maps:get(mref, Session), + {Ref, Session, State0#{guild_session_refs => #{Ref => SessionId}}}. + remove_session_state(SessionId) -> SessionData = #{ session_id => SessionId, diff --git a/fluxer_gateway/test/presence_cache_tests.erl b/fluxer_gateway/test/presence_cache_tests.erl index 7afdeed15..cb8749e9e 100644 --- a/fluxer_gateway/test/presence_cache_tests.erl +++ b/fluxer_gateway/test/presence_cache_tests.erl @@ -90,6 +90,14 @@ nodeup_cancels_grace_period_test() -> timer:sleep(50), ?assertEqual(ok, gen_server:stop(Pid)). +nodedown_grace_expiry_clears_pending_cleanup_test() -> + {ok, Pid} = maybe_start_for_test(), + ok = presence_cache:handle_nodedown('lost@node'), + ?assertMatch(#{'lost@node' := _}, maps:get(pending_nodedown_cleanups, sys:get_state(Pid))), + Pid ! {nodedown_grace_expired, 'lost@node'}, + ?assertEqual(#{}, maps:get(pending_nodedown_cleanups, sys:get_state(Pid))), + ?assertEqual(ok, gen_server:stop(Pid)). + anti_entropy_no_op_when_in_sync_test() -> {ok, Pid} = maybe_start_for_test(), State = cache_state(sys:get_state(Pid)),