Packages
macula
14.10.0
15.0.0-rc.2
15.0.0-rc.1
14.10.0
14.9.0
14.8.2
14.8.1
14.8.0
14.7.0
14.6.0
14.5.0
14.4.0
14.3.0
14.2.1
14.2.0
14.1.0
14.0.0
13.6.0
13.5.0
13.4.0
13.3.0
13.2.2
13.2.1
13.2.0
13.1.0
13.0.1
13.0.0
12.12.0
12.11.1
12.11.0
12.10.0
12.9.1
12.9.0
12.8.0
12.7.0
12.6.0
12.5.1
12.5.0
12.4.0
12.3.0
12.2.1
12.2.0
12.1.0
12.0.0
11.5.0
11.4.0
11.3.1
11.3.0
11.2.0
11.1.0
11.0.0
10.25.0
10.24.0
10.23.0
10.22.0
10.21.0
10.20.3
10.20.2
10.20.0
10.19.2
10.19.1
10.19.0
10.18.0
10.17.0
10.16.0
10.15.0
10.14.5
10.14.4
10.14.2
10.14.1
10.14.0
10.13.2
10.13.1
10.11.0
10.10.2
10.10.1
10.10.0
10.9.1
10.9.0
10.8.0
10.7.0
10.5.8
10.5.7
10.5.6
10.5.5
10.5.4
10.5.3
10.5.2
10.5.1
10.5.0
10.4.0
10.2.0
10.1.1
10.1.0
10.0.2
10.0.1
10.0.0
9.13.8
9.8.2
9.8.1
9.8.0
9.5.0
9.4.0
9.3.1
9.3.0
9.2.0
9.1.1
9.1.0
9.0.0
8.7.0
8.6.0
8.5.0
8.4.1
8.4.0
8.3.0
8.2.0
8.1.0
8.0.2
8.0.1
8.0.0
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_stream_sessions.erl
%%%-------------------------------------------------------------------
%%% @doc The node's count of served stream sessions, and of the bytes their
%%% streams keep unread.
%%%
%%% A station link asks `admit/2' before it serves a session it has
%%% started, naming the verified caller whose STREAM_OPEN the session
%%% serves. The session is admitted while that caller holds fewer than
%%% `max_served_sessions_per_caller' sessions on the node and the node
%%% fewer than `max_served_sessions', both read from the macula application
%%% env when it asks. An admitted session holds its place until its stream
%%% process ends, however it ends: this process monitors it.
%%%
%%% The caps are 16 per caller and 1000 on the node by default. A cap is
%%% kept per caller and not per link, because a link to a station carries
%%% every caller that station sends here: a cap per link would let one busy
%%% caller take the places of all the others. 16 leaves a caller room for
%%% many sessions at once, and it takes at least 63 callers to fill the
%%% node. The node cap bounds the processes all served sessions hold
%%% together, two for each session: its handler and its stream. A node that
%%% serves more sets its own value.
%%%
%%% A served stream charges the bytes of each chunk it queues unread with
%%% `charge/2', and gives them back with `release/2' when a reader takes the
%%% chunk. Both run in the stream's own process against the table, so no
%%% chunk waits on another process. A charge is refused when it would take
%%% the bytes of the stream's caller past `max_served_inbox_bytes_per_caller'
%%% (16 MiB by default) or the node's past `max_served_inbox_bytes' (256 MiB
%%% by default). The caller budget is one stream's own bound, so a caller
%%% with many sessions keeps no more unread than one session may; the node
%%% budget is sixteen callers' worth. What a stream still has charged when
%%% its process ends comes back then, through this process's monitor.
%%%
%%% A charge adds to the caller's count, then the node's, then the stream's
%%% own, and a release takes off in the opposite order; a refused add is
%%% taken back at once. So a stream that ends in the middle of one leaves at
%%% most that chunk counted too much, never too little, and this process
%%% clears what is left when the caller holds no session any more, and on
%%% the node when it holds none. Those clears run where a session ends, and
%%% a new session is admitted by this same process before its stream can
%%% charge anything.
%%%
%%% The sessions and bytes are kept in a table that
%%% `macula_stream_sessions_keeper' holds, so they outlive this process.
%%% When it starts again it counts the kept sessions anew and monitors their
%%% streams, and a stream that ended in between frees its place and its
%%% bytes at once.
%%%
%%% A refusal is counted by its reason: `caller_limit' or `node_limit' for a
%%% session, `caller_budget' or `node_budget' for a charge. A refusal is
%%% logged at once when no refusal was logged within the last
%%% `served_session_refusal_log_interval_ms'; the ones after it are logged
%%% together when that interval has passed.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_stream_sessions).
-behaviour(gen_server).
-export([start_link/0, new_table/0, admit/2, sessions/0, sessions/1,
charge/2, release/2, inbox_bytes/0, refusals/0, session_share/0, has_room/1]).
%% Reads of the node's counts that nothing inside macula calls yet; the tests
%% read them.
-ignore_xref([{macula_stream_sessions, sessions, 0}]).
-ignore_xref([{macula_stream_sessions, inbox_bytes, 0}]).
-ignore_xref([{macula_stream_sessions, refusals, 0}]).
%% Started from macula_root's child spec, which xref does not count as a call.
-ignore_xref([{macula_stream_sessions, start_link, 0}]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-export_type([refusal/0]).
-define(SERVER, ?MODULE).
%% The table's rows: `{{stream, Pid}, Caller}' for each admitted session,
%% `{{caller, Caller}, Count}' for each caller that holds a session,
%% `{sessions, Count}', `{{stream_bytes, Pid}, Bytes}',
%% `{{caller_bytes, Caller}, Bytes}', `{inbox_bytes, Bytes}', and
%% `{{refused, Reason}, Count}'.
-define(TABLE, macula_stream_sessions).
-define(MAX_SESSIONS_PER_CALLER, 16).
-define(MAX_SESSIONS, 1000).
-define(MAX_INBOX_BYTES_PER_CALLER, 16#1000000).
-define(MAX_INBOX_BYTES, 16#10000000).
-define(REFUSAL_LOG_INTERVAL_MS, 60000).
%% The counter only does table operations, so a link waits at most this long
%% for an admission before refusing it as `unavailable'.
-define(ADMIT_TIMEOUT_MS, 1000).
-type refusal() :: caller_limit | node_limit | caller_budget | node_budget.
-record(state, {
%% Refusals not logged yet, when refusals were last logged, and the
%% timer that logs the ones held back.
unlogged = #{} :: #{refusal() => pos_integer()},
logged_at = undefined :: undefined | integer(),
flush = undefined :: undefined | reference()
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
%% @doc Create the table the sessions are kept in, owned by the calling
%% process. `macula_stream_sessions_keeper' creates it, so it outlives this
%% process.
-spec new_table() -> ets:table().
new_table() ->
ets:new(?TABLE, [set, public, named_table]).
%% @doc The unread bytes one served session may keep and still leave every
%% other session its caller may hold the same: the caller budget over the
%% caller's session cap (1 MiB by default). A provider that credits its
%% caller no more than this never has a healthy session refused for
%% another one's stall.
-spec session_share() -> pos_integer().
session_share() ->
limit(max_served_inbox_bytes_per_caller, ?MAX_INBOX_BYTES_PER_CALLER)
div limit(max_served_sessions_per_caller, ?MAX_SESSIONS_PER_CALLER).
%% @doc Admit the session that `Stream' serves for the verified `Caller', or
%% refuse it when that caller or the node already holds as many sessions as
%% its cap allows. A stream already admitted keeps its one place.
%%
%% Admission fails closed and never holds up the link that asks: when this
%% process does not answer within a second, because it is restarting, gone
%% or held up, the session is refused as `unavailable'.
-spec admit(binary(), pid()) -> ok | {error, caller_limit | node_limit | unavailable}.
admit(Caller, Stream) when is_binary(Caller), is_pid(Stream) ->
try gen_server:call(?SERVER, {admit, Caller, Stream}, ?ADMIT_TIMEOUT_MS)
catch exit:{_Why, {gen_server, call, _Args}} -> {error, unavailable}
end.
%% @doc Whether `Caller' and the node have room for one more served session
%% now, read from the counts without asking this process: what a link checks
%% before it pays for opening a sealed STREAM_OPEN, so a caller at its cap
%% costs no decapsulation. `admit/2' still decides; a count that cannot be
%% read leaves the decision to it.
-spec has_room(binary()) -> ok | {error, caller_limit | node_limit}.
has_room(Caller) when is_binary(Caller) ->
try admission(false, Caller)
catch error:badarg -> ok
end.
%% @doc The number of served sessions the node holds now.
-spec sessions() -> non_neg_integer().
sessions() ->
gen_server:call(?SERVER, sessions).
%% @doc The number of served sessions `Caller' holds now: the count
%% `max_served_sessions_per_caller' is decided on, for one caller rather
%% than the node. A caller that holds none has no row and counts 0.
-spec sessions(binary()) -> non_neg_integer().
sessions(Caller) when is_binary(Caller) ->
gen_server:call(?SERVER, {sessions, Caller}).
%% @doc Charge `Bytes' that `Stream' keeps unread to its caller and the node,
%% or refuse when that would take either past its budget, or when `Stream' is
%% not an admitted session. Called in the stream's own process.
-spec charge(pid(), non_neg_integer()) -> ok | {error, caller_budget | node_budget | not_admitted}.
charge(Stream, Bytes) when is_pid(Stream), is_integer(Bytes), Bytes >= 0 ->
charge_caller(ets:lookup(?TABLE, {stream, Stream}), Stream, Bytes).
%% @doc Give back `Bytes' that `Stream' charged and no longer keeps. Called in
%% the stream's own process.
-spec release(pid(), non_neg_integer()) -> ok.
release(Stream, Bytes) when is_pid(Stream), is_integer(Bytes), Bytes >= 0 ->
release_caller(ets:lookup(?TABLE, {stream, Stream}), Stream, Bytes).
%% @doc The bytes the node's served streams keep unread now.
-spec inbox_bytes() -> integer().
inbox_bytes() ->
ets:lookup_element(?TABLE, inbox_bytes, 2, 0).
%% @doc The refusals since the macula application started, by reason.
-spec refusals() -> #{refusal() => pos_integer()}.
refusals() ->
gen_server:call(?SERVER, refusals).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init([]) ->
ok = count_kept_sessions(ets:match_object(?TABLE, {{stream, '_'}, '_'})),
{ok, #state{}}.
handle_call({admit, Caller, Stream}, _From, S) ->
on_admission(admission(ets:member(?TABLE, {stream, Stream}), Caller), Caller, Stream, S);
handle_call(sessions, _From, S) ->
{reply, count(sessions), S};
handle_call({sessions, Caller}, _From, S) ->
{reply, count({caller, Caller}), S};
handle_call(refusals, _From, S) ->
{reply, maps:from_list(refusal_counts(ets:match_object(?TABLE, {{refused, '_'}, '_'}))), S}.
handle_cast({charge_refused, Reason}, S) ->
{noreply, count_refusal(Reason, S)};
handle_cast(_Msg, S) ->
{noreply, S}.
handle_info({'DOWN', _Mon, process, Stream, _Reason}, S) ->
ok = release_session(ets:lookup(?TABLE, {stream, Stream})),
{noreply, S};
handle_info(log_refusals, #state{unlogged = Unlogged} = S) when map_size(Unlogged) =:= 0 ->
{noreply, S#state{flush = undefined}};
handle_info(log_refusals, S) ->
{noreply, log_refusals(S#state{flush = undefined}, erlang:monotonic_time(millisecond))};
handle_info(_Info, S) ->
{noreply, S}.
terminate(_Reason, _S) ->
ok.
%%%===================================================================
%%% Internal: sessions
%%%===================================================================
%% Counts the kept sessions anew from their stream rows and monitors each
%% stream again. A stream that ended while no counter ran frees its place
%% when its monitor fires, at once.
count_kept_sessions(Streams) ->
true = ets:match_delete(?TABLE, {{caller, '_'}, '_'}),
true = ets:insert(?TABLE, [{sessions, length(Streams)} | caller_rows(Streams)]),
_ = [erlang:monitor(process, Stream) || {{stream, Stream}, _Caller} <- Streams],
ok.
caller_rows(Streams) ->
Counts = lists:foldl(fun({_Key, Caller}, Acc) -> bump(Caller, Acc) end, #{}, Streams),
[{{caller, Caller}, N} || {Caller, N} <- maps:to_list(Counts)].
refusal_counts(Rows) ->
[{Reason, N} || {{refused, Reason}, N} <- Rows].
admission(true, _Caller) ->
admitted;
admission(false, Caller) ->
admission_for(count(sessions) < limit(max_served_sessions, ?MAX_SESSIONS),
count({caller, Caller}) < limit(max_served_sessions_per_caller,
?MAX_SESSIONS_PER_CALLER)).
admission_for(false, _CallerHasRoom) -> {error, node_limit};
admission_for(true, false) -> {error, caller_limit};
admission_for(true, true) -> ok.
limit(Key, Default) ->
application:get_env(macula, Key, Default).
count(Key) ->
ets:lookup_element(?TABLE, Key, 2, 0).
on_admission(ok, Caller, Stream, S) ->
_ = erlang:monitor(process, Stream),
true = ets:insert(?TABLE, [{{stream, Stream}, Caller},
{{caller, Caller}, count({caller, Caller}) + 1},
{sessions, count(sessions) + 1}]),
{reply, ok, S};
on_admission(admitted, _Caller, _Stream, S) ->
{reply, ok, S};
on_admission({error, Reason} = Refused, _Caller, _Stream, S) ->
{reply, Refused, count_refusal(Reason, S)}.
%% A session's bytes and place free when its stream process ends. The
%% stream's own count is taken first, so a restart in between never gives
%% the same bytes back twice.
release_session([{{stream, Stream} = Key, Caller}]) ->
ok = give_back(Caller, taken_bytes(ets:take(?TABLE, {stream_bytes, Stream}))),
true = ets:delete(?TABLE, Key),
ok = one_less_session(count(sessions)),
one_less(count({caller, Caller}), Caller);
release_session([]) ->
ok.
taken_bytes([{_Key, Bytes}]) -> Bytes;
taken_bytes([]) -> 0.
give_back(Caller, Bytes) ->
ok = add(inbox_bytes, -Bytes),
add({caller_bytes, Caller}, -Bytes).
%% A node that holds no session keeps no unread bytes: what a stream that
%% ended in the middle of an update left counted is cleared with the last one.
one_less_session(1) ->
true = ets:insert(?TABLE, {sessions, 0}),
true = ets:delete(?TABLE, inbox_bytes),
ok;
one_less_session(N) ->
true = ets:insert(?TABLE, {sessions, N - 1}),
ok.
%% A caller that holds no session keeps no unread bytes either.
one_less(1, Caller) ->
true = ets:delete(?TABLE, {caller, Caller}),
true = ets:delete(?TABLE, {caller_bytes, Caller}),
ok;
one_less(N, Caller) ->
true = ets:insert(?TABLE, {{caller, Caller}, N - 1}),
ok.
bump(Key, Counts) ->
maps:update_with(Key, fun(N) -> N + 1 end, 1, Counts).
%%%===================================================================
%%% Internal: the budget on unread bytes
%%%===================================================================
charge_caller([], _Stream, _Bytes) ->
{error, not_admitted};
charge_caller([{_Key, Caller}], Stream, Bytes) ->
charged_caller(add_within({caller_bytes, Caller}, Bytes,
limit(max_served_inbox_bytes_per_caller, ?MAX_INBOX_BYTES_PER_CALLER)),
Caller, Stream, Bytes).
charged_caller(refused, _Caller, _Stream, _Bytes) ->
refuse_charge(caller_budget);
charged_caller(added, Caller, Stream, Bytes) ->
charged_node(add_within(inbox_bytes, Bytes, limit(max_served_inbox_bytes, ?MAX_INBOX_BYTES)),
Caller, Stream, Bytes).
charged_node(refused, Caller, _Stream, Bytes) ->
ok = add({caller_bytes, Caller}, -Bytes),
refuse_charge(node_budget);
charged_node(added, _Caller, Stream, Bytes) ->
add({stream_bytes, Stream}, Bytes).
%% Adds Bytes to the count at Key, and takes them back at once when that took
%% it past Max: for that moment the count is over by only the refused bytes.
add_within(Key, Bytes, Max) ->
within(ets:update_counter(?TABLE, Key, {2, Bytes}, {Key, 0}) =< Max, Key, Bytes).
within(true, _Key, _Bytes) ->
added;
within(false, Key, Bytes) ->
ok = add(Key, -Bytes),
refused.
add(Key, Bytes) ->
_ = ets:update_counter(?TABLE, Key, {2, Bytes}, {Key, 0}),
ok.
%% This process counts a refused charge and logs it with the other refusals.
refuse_charge(Reason) ->
ok = gen_server:cast(?SERVER, {charge_refused, Reason}),
{error, Reason}.
release_caller([], _Stream, _Bytes) ->
ok;
release_caller([{_Key, Caller}], Stream, Bytes) ->
ok = add({stream_bytes, Stream}, -Bytes),
ok = add(inbox_bytes, -Bytes),
add({caller_bytes, Caller}, -Bytes).
%%%===================================================================
%%% Internal: refusals
%%%===================================================================
count_refusal(Reason, #state{unlogged = Unlogged} = S) ->
_ = ets:update_counter(?TABLE, {refused, Reason}, 1, {{refused, Reason}, 0}),
report(S#state{unlogged = bump(Reason, Unlogged)}, erlang:monotonic_time(millisecond)).
%% Logs now when nothing was logged within the interval; otherwise holds the
%% refusals back for one timer that logs them when the interval has passed.
report(#state{logged_at = undefined} = S, Now) ->
log_refusals(S, Now);
report(#state{logged_at = At} = S, Now) ->
report_after(Now - At >= log_interval(), S, Now).
report_after(true, S, Now) ->
log_refusals(S, Now);
report_after(false, #state{flush = undefined, logged_at = At} = S, Now) ->
S#state{flush = erlang:send_after(max(0, At + log_interval() - Now), self(), log_refusals)};
report_after(false, S, _Now) ->
S.
log_refusals(#state{unlogged = Unlogged, flush = Flush} = S, Now) ->
ok = cancel(Flush),
logger:warning("[macula_stream_sessions] served stream sessions refused since the last report: ~p",
[Unlogged]),
S#state{unlogged = #{}, logged_at = Now, flush = undefined}.
cancel(undefined) -> ok;
cancel(Timer) -> _ = erlang:cancel_timer(Timer), ok.
log_interval() ->
limit(served_session_refusal_log_interval_ms, ?REFUSAL_LOG_INTERVAL_MS).