diff --git a/fluxer_gateway/src/gateway/fluxer_gateway_config.erl b/fluxer_gateway/src/gateway/fluxer_gateway_config.erl index 05095de5a..f33bfe2f1 100644 --- a/fluxer_gateway/src/gateway/fluxer_gateway_config.erl +++ b/fluxer_gateway/src/gateway/fluxer_gateway_config.erl @@ -81,6 +81,9 @@ env_gateway_base_config() -> <<"push_clear_notifications_enabled">> => env_bool( "FLUXER_GATEWAY_PUSH_CLEAR_NOTIFICATIONS_ENABLED", true ), + <<"push_endpoint_guard_enabled">> => env_bool( + "FLUXER_GATEWAY_PUSH_ENDPOINT_GUARD_ENABLED", true + ), <<"push_outbox_request_timeout_ms">> => env_int( "FLUXER_GATEWAY_PUSH_OUTBOX_REQUEST_TIMEOUT_MS", 100000 ), @@ -256,6 +259,9 @@ build_push_config(Service, Public) -> push_clear_notifications_enabled => get_bool( Service, <<"push_clear_notifications_enabled">>, true ), + push_endpoint_guard_enabled => get_bool( + Service, <<"push_endpoint_guard_enabled">>, true + ), push_outbox_max_queue => get_int(Service, <<"push_outbox_max_queue">>, 10000), push_outbox_max_inflight => get_int(Service, <<"push_outbox_max_inflight">>, 64), push_outbox_request_timeout_ms => get_int( diff --git a/fluxer_gateway/src/push/push_endpoint_guard.erl b/fluxer_gateway/src/push/push_endpoint_guard.erl index 2fdf55837..6eaec2b5e 100644 --- a/fluxer_gateway/src/push/push_endpoint_guard.erl +++ b/fluxer_gateway/src/push/push_endpoint_guard.erl @@ -3,24 +3,46 @@ -module(push_endpoint_guard). -typing([eqwalizer]). --export([check/1, check/2]). +-export([check/1, check/2, check/3]). --export_type([resolver/0]). +-export_type([resolver/0, cache_mode/0, verdict/0]). -define(RESOLVE_TIMEOUT_MS, 3000). -define(MAX_HOST_LENGTH, 253). -define(MAX_LABEL_LENGTH, 63). +-define(ALLOWED_VERDICT_TTL_SECONDS, 300). +-define(REFUSED_VERDICT_TTL_SECONDS, 30). -type resolver() :: fun((string()) -> {ok, [inet:ip_address()]} | {error, term()}). +-type cache_mode() :: cached | uncached. +-type verdict() :: ok | {error, term()}. --spec check(binary()) -> ok | {error, term()}. +-spec check(binary()) -> verdict(). check(Endpoint) -> - check(Endpoint, fun resolve/1). + check(Endpoint, fun resolve/1, cached). --spec check(binary(), resolver()) -> ok | {error, term()}. +-spec check(binary(), resolver()) -> verdict(). check(Endpoint, Resolver) -> + check(Endpoint, Resolver, uncached). + +-spec check(binary(), resolver(), cache_mode()) -> verdict(). +check(Endpoint, Resolver, CacheMode) -> + case enabled() of + true -> check_endpoint(Endpoint, Resolver, CacheMode); + false -> ok + end. + +-spec enabled() -> boolean(). +enabled() -> + case fluxer_gateway_env:get(push_endpoint_guard_enabled) of + Enabled when is_boolean(Enabled) -> Enabled; + _ -> true + end. + +-spec check_endpoint(binary(), resolver(), cache_mode()) -> verdict(). +check_endpoint(Endpoint, Resolver, CacheMode) -> case parse_endpoint(Endpoint) of - {ok, Host} -> check_host(Host, Resolver); + {ok, Host} -> check_host(Host, Resolver, CacheMode); {error, Reason} -> {error, Reason} end. @@ -64,21 +86,41 @@ allowed_port(80) -> true; allowed_port(443) -> true; allowed_port(_Port) -> false. --spec check_host(string(), resolver()) -> ok | {error, term()}. -check_host(Host, Resolver) -> +-spec check_host(string(), resolver(), cache_mode()) -> verdict(). +check_host(Host, Resolver, CacheMode) -> case inet:parse_address(Host) of {ok, Address} -> check_addresses([Address]); - {error, _Reason} -> check_hostname(Host, Resolver) + {error, _Reason} -> check_hostname(Host, Resolver, CacheMode) end. --spec check_hostname(string(), resolver()) -> ok | {error, term()}. -check_hostname(Host, Resolver) -> +-spec check_hostname(string(), resolver(), cache_mode()) -> verdict(). +check_hostname(Host, Resolver, CacheMode) -> case is_fqdn(Host) of - true -> resolve_and_check(Host, Resolver); + true -> resolve_and_check(Host, Resolver, CacheMode); false -> {error, endpoint_rejected} end. --spec resolve_and_check(string(), resolver()) -> ok | {error, term()}. +-spec resolve_and_check(string(), resolver(), cache_mode()) -> verdict(). +resolve_and_check(Host, Resolver, uncached) -> + resolve_and_check(Host, Resolver); +resolve_and_check(Host, Resolver, cached) -> + CacheKey = list_to_binary(Host), + case push_ets_cache:get_endpoint_verdict(CacheKey) of + {ok, Verdict} -> Verdict; + undefined -> store_verdict(CacheKey, resolve_and_check(Host, Resolver)) + end. + +-spec store_verdict(binary(), verdict()) -> verdict(). +store_verdict(CacheKey, Verdict) -> + Ttl = verdict_ttl_seconds(Verdict), + ok = push_ets_cache:put_endpoint_verdict(CacheKey, Verdict, Ttl), + Verdict. + +-spec verdict_ttl_seconds(verdict()) -> pos_integer(). +verdict_ttl_seconds(ok) -> ?ALLOWED_VERDICT_TTL_SECONDS; +verdict_ttl_seconds({error, _Reason}) -> ?REFUSED_VERDICT_TTL_SECONDS. + +-spec resolve_and_check(string(), resolver()) -> verdict(). resolve_and_check(Host, Resolver) -> case Resolver(Host) of {ok, Addresses} -> check_addresses(Addresses); diff --git a/fluxer_gateway/src/push/push_ets_cache.erl b/fluxer_gateway/src/push/push_ets_cache.erl index e018644e1..5b5fff9aa 100644 --- a/fluxer_gateway/src/push/push_ets_cache.erl +++ b/fluxer_gateway/src/push/push_ets_cache.erl @@ -25,6 +25,8 @@ reserve_badge_counts/1, get_bearer_token/1, put_bearer_token/3, + get_endpoint_verdict/1, + put_endpoint_verdict/3, release/1, rebalance/0, rebalance_async/0, @@ -33,16 +35,20 @@ table_size/1 ]). --export_type([fill/0]). +-export_type([fill/0, endpoint_verdict/0]). -define(USER_GUILD_SETTINGS, push_user_guild_settings). -define(SUBSCRIPTIONS, push_subscriptions). -define(BLOCKED_IDS, push_blocked_ids). -define(BADGE_COUNTS, push_badge_counts). -define(BEARER_TOKENS, push_bearer_tokens). +-define(ENDPOINT_VERDICTS, push_endpoint_verdicts). -define(MAX_TABLE_ENTRIES, 500000). -define(MAX_BEARER_TOKENS, 10000). +-define(MAX_ENDPOINT_VERDICTS, 2048). +-define(MAX_ENDPOINT_HOST_BYTES, 253). +-define(ENDPOINT_VERDICT_EVICT_BATCH, 512). -define(EVICT_BATCH, 4096). -define(MAX_EVICT_RESEEKS, 8). -define(RESERVATION_TTL_MS, 120000). @@ -55,6 +61,7 @@ ]). -type fill() :: {atom(), pos_integer(), [term()]}. +-type endpoint_verdict() :: ok | {error, term()}. -spec init() -> ok. init() -> @@ -63,6 +70,7 @@ init() -> ensure_table(?BLOCKED_IDS), ensure_table(?BADGE_COUNTS), ensure_table(?BEARER_TOKENS), + ensure_table(?ENDPOINT_VERDICTS), ok. -spec get_user_guild_settings(integer(), integer()) -> map() | undefined. @@ -219,6 +227,48 @@ put_bearer_token(Key, Token, ExpiresAt) when is_binary(Token), is_integer(Expire error:badarg -> ok end. +-spec get_endpoint_verdict(binary()) -> {ok, endpoint_verdict()} | undefined. +get_endpoint_verdict(Host) when is_binary(Host) -> + try ets:lookup(?ENDPOINT_VERDICTS, Host) of + [{_, ok, ExpiresAt}] when is_integer(ExpiresAt) -> + live_endpoint_verdict(ok, ExpiresAt); + [{_, {error, Reason}, ExpiresAt}] when is_integer(ExpiresAt) -> + live_endpoint_verdict({error, Reason}, ExpiresAt); + _ -> + undefined + catch + error:badarg -> undefined + end. + +-spec live_endpoint_verdict(endpoint_verdict(), integer()) -> + {ok, endpoint_verdict()} | undefined. +live_endpoint_verdict(Verdict, ExpiresAt) -> + case erlang:system_time(second) < ExpiresAt of + true -> {ok, Verdict}; + false -> undefined + end. + +-spec put_endpoint_verdict(binary(), endpoint_verdict(), pos_integer()) -> ok. +put_endpoint_verdict(Host, Verdict, TtlSeconds) when + is_binary(Host), is_integer(TtlSeconds), TtlSeconds > 0 +-> + case byte_size(Host) =< ?MAX_ENDPOINT_HOST_BYTES of + true -> insert_endpoint_verdict(Host, Verdict, TtlSeconds); + false -> ok + end. + +-spec insert_endpoint_verdict(binary(), endpoint_verdict(), pos_integer()) -> ok. +insert_endpoint_verdict(Host, Verdict, TtlSeconds) -> + guard_table_size( + ?ENDPOINT_VERDICTS, ?MAX_ENDPOINT_VERDICTS, ?ENDPOINT_VERDICT_EVICT_BATCH + ), + ExpiresAt = erlang:system_time(second) + TtlSeconds, + try ets:insert(?ENDPOINT_VERDICTS, {Host, Verdict, ExpiresAt}) of + _ -> ok + catch + error:badarg -> ok + end. + -spec write(atom(), tuple()) -> ok. write(Table, Row) -> guard_table_size(Table, ?MAX_TABLE_ENTRIES), @@ -292,7 +342,8 @@ cache_stats() -> push_subscriptions_size => table_size(?SUBSCRIPTIONS), blocked_ids_size => table_size(?BLOCKED_IDS), badge_counts_size => table_size(?BADGE_COUNTS), - bearer_tokens_size => table_size(?BEARER_TOKENS) + bearer_tokens_size => table_size(?BEARER_TOKENS), + endpoint_verdicts_size => table_size(?ENDPOINT_VERDICTS) }. -spec evict_tables(map()) -> ok. @@ -300,6 +351,7 @@ evict_tables(MaxEntries) -> Now = erlang:system_time(second), select_delete(?BLOCKED_IDS, expired_rows(Now)), select_delete(?BEARER_TOKENS, expired_rows(Now)), + select_delete(?ENDPOINT_VERDICTS, expired_rows(Now)), lists:foreach( fun expire_reservations/1, [?USER_GUILD_SETTINGS, ?SUBSCRIPTIONS, ?BLOCKED_IDS, ?BADGE_COUNTS] @@ -309,6 +361,7 @@ evict_tables(MaxEntries) -> evict_table(?BLOCKED_IDS, maps:get(blocked_ids, MaxEntries, undefined)), evict_table(?BADGE_COUNTS, maps:get(badge_counts, MaxEntries, undefined)), evict_table(?BEARER_TOKENS, ?MAX_BEARER_TOKENS), + evict_table(?ENDPOINT_VERDICTS, ?MAX_ENDPOINT_VERDICTS), ok. -spec expired_rows(integer()) -> ets:match_spec(). @@ -330,8 +383,12 @@ select_delete(Table, MatchSpec) -> -spec guard_table_size(atom(), non_neg_integer()) -> ok. guard_table_size(Table, MaxEntries) -> + guard_table_size(Table, MaxEntries, ?EVICT_BATCH). + +-spec guard_table_size(atom(), non_neg_integer(), pos_integer()) -> ok. +guard_table_size(Table, MaxEntries, EvictBatch) -> case table_size(Table) >= MaxEntries of - true -> evict_table(Table, max(0, MaxEntries - ?EVICT_BATCH)); + true -> evict_table(Table, max(0, MaxEntries - EvictBatch)); false -> ok end. diff --git a/fluxer_gateway/test/fluxer_gateway_config_tests.erl b/fluxer_gateway/test/fluxer_gateway_config_tests.erl index 5e2777116..0f7793ac8 100644 --- a/fluxer_gateway/test/fluxer_gateway_config_tests.erl +++ b/fluxer_gateway/test/fluxer_gateway_config_tests.erl @@ -80,6 +80,16 @@ cluster_static_peers_accepts_valid_node_names_test() -> maps:get(cluster_static_peers, Config) ). +push_endpoint_guard_enabled_defaults_on_test() -> + Config = fluxer_gateway_config:load(), + ?assertEqual(true, maps:get(push_endpoint_guard_enabled, Config)). + +push_endpoint_guard_enabled_can_be_turned_off_test() -> + with_env("FLUXER_GATEWAY_PUSH_ENDPOINT_GUARD_ENABLED", "false", fun() -> + Config = fluxer_gateway_config:load(), + ?assertEqual(false, maps:get(push_endpoint_guard_enabled, Config)) + end). + presence_push_buffer_env_defaults_test() -> with_env("FLUXER_GATEWAY_PRESENCE_PUSH_BUFFER_MAX_ENTRIES", "7", fun() -> with_env("FLUXER_GATEWAY_PRESENCE_PUSH_BUFFER_MAX_BYTES", "4096", fun() -> diff --git a/fluxer_gateway/test/push_endpoint_guard_tests.erl b/fluxer_gateway/test/push_endpoint_guard_tests.erl index 94fcc70f9..7eb6e7aa3 100644 --- a/fluxer_gateway/test/push_endpoint_guard_tests.erl +++ b/fluxer_gateway/test/push_endpoint_guard_tests.erl @@ -6,6 +6,8 @@ -include_lib("eunit/include/eunit.hrl"). -define(ENDPOINT, <<"https://push.example.com/wpush/v2/abc">>). +-define(VERDICT_TABLE, push_endpoint_verdicts). +-define(MAX_VERDICTS, 2048). resolves_to(Addresses) -> fun(_Host) -> {ok, Addresses} end. @@ -13,6 +15,9 @@ resolves_to(Addresses) -> fails_with(Reason) -> fun(_Host) -> {error, Reason} end. +never_resolves() -> + fun(_Host) -> erlang:error(resolver_called) end. + link_local_metadata_address_is_refused_test() -> ?assertEqual( {error, endpoint_blocked}, @@ -185,8 +190,162 @@ userinfo_is_refused_test() -> ) ). +a_repeat_lookup_for_the_same_host_does_not_resolve_again_test() -> + with_verdict_cache(fun() -> + Counter = counters:new(1, []), + Endpoint = endpoint("cache-repeat.example.com"), + Resolver = counting_resolver(Counter, [{93, 184, 216, 34}]), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(1, counters:get(Counter, 1)) + end). + +a_cache_hit_returns_the_verdict_the_uncached_path_returns_test() -> + with_verdict_cache(fun() -> + lists:foreach( + fun assert_cached_matches_uncached/1, + [ + {"cache-allow.example.com", resolves_to([{93, 184, 216, 34}]), ok}, + {"cache-block.example.com", resolves_to([{10, 0, 0, 1}]), + {error, endpoint_blocked}}, + {"cache-empty.example.com", resolves_to([]), {error, nxdomain}}, + {"cache-timeout.example.com", fails_with(timeout), {error, timeout}} + ] + ) + end). + +assert_cached_matches_uncached({Host, Resolver, Expected}) -> + Endpoint = endpoint(Host), + ?assertEqual(Expected, push_endpoint_guard:check(Endpoint, Resolver)), + ?assertEqual(Expected, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(Expected, push_endpoint_guard:check(Endpoint, never_resolves(), cached)). + +an_uncached_check_never_writes_the_cache_test() -> + with_verdict_cache(fun() -> + Endpoint = endpoint("cache-bypass.example.com"), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, resolves_to([{1, 1, 1, 1}]))), + ?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE)) + end). + +an_ip_literal_is_never_cached_test() -> + with_verdict_cache(fun() -> + ?assertEqual( + ok, push_endpoint_guard:check(<<"https://93.184.216.34/sub">>, never_resolves()) + ), + ?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE)) + end). + +a_cached_verdict_expires_test() -> + with_verdict_cache(fun() -> + Counter = counters:new(1, []), + Endpoint = endpoint("cache-expiry.example.com"), + Resolver = counting_resolver(Counter, [{93, 184, 216, 34}]), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(1, counters:get(Counter, 1)), + expire_verdict(<<"cache-expiry.example.com">>), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assertEqual(2, counters:get(Counter, 1)) + end). + +a_refused_verdict_expires_sooner_than_an_allowed_one_test() -> + with_verdict_cache(fun() -> + Allowed = endpoint("cache-ttl-allowed.example.com"), + Refused = endpoint("cache-ttl-refused.example.com"), + ?assertEqual( + ok, push_endpoint_guard:check(Allowed, resolves_to([{93, 184, 216, 34}]), cached) + ), + ?assertEqual( + {error, timeout}, push_endpoint_guard:check(Refused, fails_with(timeout), cached) + ), + AllowedExpiry = expires_at(<<"cache-ttl-allowed.example.com">>), + RefusedExpiry = expires_at(<<"cache-ttl-refused.example.com">>), + ?assert(RefusedExpiry < AllowedExpiry) + end). + +many_distinct_hosts_cannot_grow_the_cache_without_bound_test() -> + with_verdict_cache(fun() -> + Resolver = resolves_to([{93, 184, 216, 34}]), + lists:foreach( + fun(N) -> flood_one_host(N, Resolver) end, + lists:seq(1, 20000) + ), + Size = push_ets_cache:table_size(?VERDICT_TABLE), + ?assert(Size >= 1500), + ?assert(Size =< ?MAX_VERDICTS), + ?assert(verdict_table_bytes() =< 4 * 1024 * 1024) + end). + +flood_one_host(N, Resolver) -> + Endpoint = endpoint("flood-" ++ integer_to_list(N) ++ ".example.com"), + ?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)), + ?assert(push_ets_cache:table_size(?VERDICT_TABLE) =< ?MAX_VERDICTS). + +the_guard_is_enabled_by_default_test() -> + ?assertEqual( + {error, endpoint_blocked}, + push_endpoint_guard:check(?ENDPOINT, resolves_to([{10, 0, 0, 1}])) + ). + +a_disabled_guard_passes_everything_through_without_resolving_test() -> + with_verdict_cache(fun() -> + with_guard_disabled(fun() -> + Never = never_resolves(), + ?assertEqual(ok, push_endpoint_guard:check(?ENDPOINT, Never)), + ?assertEqual(ok, push_endpoint_guard:check(?ENDPOINT, Never, cached)), + ?assertEqual( + ok, push_endpoint_guard:check(<<"https://169.254.169.254/latest">>, Never) + ), + ?assertEqual( + ok, push_endpoint_guard:check(<<"http://push.example.com/sub">>, Never) + ), + ?assertEqual(ok, push_endpoint_guard:check(<<"not-a-url">>, Never)), + ?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE)) + end) + end). + +counting_resolver(Counter, Addresses) -> + fun(_Host) -> + counters:add(Counter, 1, 1), + {ok, Addresses} + end. + +endpoint(Host) -> + list_to_binary("https://" ++ Host ++ "/sub"). + +expires_at(Host) -> + [{Host, _Verdict, ExpiresAt}] = ets:lookup(?VERDICT_TABLE, Host), + ExpiresAt. + +expire_verdict(Host) -> + [{Host, Verdict, _ExpiresAt}] = ets:lookup(?VERDICT_TABLE, Host), + true = ets:insert(?VERDICT_TABLE, {Host, Verdict, erlang:system_time(second) - 1}), + ok. + +verdict_table_bytes() -> + ets:info(?VERDICT_TABLE, memory) * erlang:system_info(wordsize). + +with_verdict_cache(Fun) -> + ok = push_ets_cache:init(), + true = ets:delete_all_objects(?VERDICT_TABLE), + try + Fun() + after + ets:delete_all_objects(?VERDICT_TABLE) + end. + +with_guard_disabled(Fun) -> + Original = fluxer_gateway_env:get(push_endpoint_guard_enabled), + _ = fluxer_gateway_env:patch(#{push_endpoint_guard_enabled => false}), + try + Fun() + after + _ = fluxer_gateway_env:patch(#{push_endpoint_guard_enabled => Original}) + end. + ip_literals_skip_dns_and_are_screened_directly_test() -> - Never = fun(_Host) -> erlang:error(resolver_called) end, + Never = never_resolves(), ?assertEqual( {error, endpoint_blocked}, push_endpoint_guard:check(<<"https://169.254.169.254/latest">>, Never) @@ -204,7 +363,7 @@ ip_literals_skip_dns_and_are_screened_directly_test() -> ?assertEqual(ok, push_endpoint_guard:check(<<"https://93.184.216.34/sub">>, Never)). malformed_and_non_fqdn_hosts_are_refused_test() -> - Never = fun(_Host) -> erlang:error(resolver_called) end, + Never = never_resolves(), lists:foreach( fun(Endpoint) -> ?assertEqual( diff --git a/fluxer_gateway/test/push_ets_cache_tests.erl b/fluxer_gateway/test/push_ets_cache_tests.erl index 02590fc36..53bd702ce 100644 --- a/fluxer_gateway/test/push_ets_cache_tests.erl +++ b/fluxer_gateway/test/push_ets_cache_tests.erl @@ -11,6 +11,7 @@ init_creates_tables_test() -> ?assertNotEqual(undefined, ets:whereis(push_subscriptions)), ?assertNotEqual(undefined, ets:whereis(push_blocked_ids)), ?assertNotEqual(undefined, ets:whereis(push_badge_counts)), + ?assertNotEqual(undefined, ets:whereis(push_endpoint_verdicts)), cleanup_tables(). init_idempotent_test() -> @@ -112,6 +113,69 @@ rebalance_evicts_remote_owned_entries_test() -> persistent_term:erase({gateway_cluster_membership, members_by_role}), cleanup_tables(). +endpoint_verdict_round_trip_test() -> + cleanup_tables(), + ok = push_ets_cache:init(), + ?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(<<"push.example.com">>)), + ok = push_ets_cache:put_endpoint_verdict(<<"push.example.com">>, ok, 300), + ?assertEqual({ok, ok}, push_ets_cache:get_endpoint_verdict(<<"push.example.com">>)), + ok = push_ets_cache:put_endpoint_verdict( + <<"bad.example.com">>, {error, endpoint_blocked}, 30 + ), + ?assertEqual( + {ok, {error, endpoint_blocked}}, + push_ets_cache:get_endpoint_verdict(<<"bad.example.com">>) + ), + cleanup_tables(). + +endpoint_verdicts_expire_and_are_reclaimed_test() -> + cleanup_tables(), + ok = push_ets_cache:init(), + ok = push_ets_cache:put_endpoint_verdict(<<"stale.example.com">>, ok, 300), + Stale = erlang:system_time(second) - 1, + true = ets:insert(push_endpoint_verdicts, {<<"stale.example.com">>, ok, Stale}), + ?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(<<"stale.example.com">>)), + ok = push_ets_cache:evict_tables(#{}), + ?assertEqual([], ets:lookup(push_endpoint_verdicts, <<"stale.example.com">>)), + cleanup_tables(). + +an_oversized_host_is_never_cached_test() -> + cleanup_tables(), + ok = push_ets_cache:init(), + Oversized = binary:copy(<<"a">>, 254), + ok = push_ets_cache:put_endpoint_verdict(Oversized, ok, 300), + ?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(Oversized)), + ?assertEqual(0, push_ets_cache:table_size(push_endpoint_verdicts)), + AtLimit = binary:copy(<<"a">>, 253), + ok = push_ets_cache:put_endpoint_verdict(AtLimit, ok, 300), + ?assertEqual({ok, ok}, push_ets_cache:get_endpoint_verdict(AtLimit)), + cleanup_tables(). + +endpoint_verdicts_stay_bounded_under_max_length_hosts_test() -> + cleanup_tables(), + ok = push_ets_cache:init(), + lists:foreach(fun seed_max_length_verdict/1, lists:seq(1, 20000)), + Size = push_ets_cache:table_size(push_endpoint_verdicts), + ?assert(Size >= 1500), + ?assert(Size =< 2048), + Bytes = ets:info(push_endpoint_verdicts, memory) * erlang:system_info(wordsize), + ?assert(Bytes =< 4 * 1024 * 1024), + cleanup_tables(). + +seed_max_length_verdict(N) -> + Suffix = integer_to_binary(N), + Host = <<(binary:copy(<<"a">>, 253 - byte_size(Suffix)))/binary, Suffix/binary>>, + ok = push_ets_cache:put_endpoint_verdict(Host, ok, 300), + ?assert(push_ets_cache:table_size(push_endpoint_verdicts) =< 2048). + +endpoint_verdicts_are_reported_in_cache_stats_test() -> + cleanup_tables(), + ok = push_ets_cache:init(), + ok = push_ets_cache:put_endpoint_verdict(<<"push.example.com">>, ok, 300), + Stats = push_ets_cache:cache_stats(), + ?assertEqual(1, maps:get(endpoint_verdicts_size, Stats)), + cleanup_tables(). + seed_subscriptions(UserId, Subscriptions) -> push_ets_cache:put_subscriptions( UserId, Subscriptions, push_ets_cache:reserve_subscriptions([UserId]) @@ -143,6 +207,7 @@ cleanup_tables() -> delete_table(push_blocked_ids), delete_table(push_badge_counts), delete_table(push_bearer_tokens), + delete_table(push_endpoint_verdicts), ok. delete_table(Table) ->