Rate limit restructure (server-side)

.github/workflows/on-demand.yml

@@ -46,6 +46,8 @@ on:
           - ar_intervals
           - ar_join
           - ar_kv
+          - arweave_limiter_group
+          - arweave_limiter_metrics_collector
           - ar_mempool_tests
           - ar_merkle
           - ar_mine_randomx_tests

.github/workflows/x-test-full.yml

@@ -61,6 +61,8 @@ jobs:
             ar_intervals,
             ar_join,
             ar_kv,
+            arweave_limiter_group,
+            arweave_limiter_metrics_collector,
             ar_merkle,
             ar_mining_cache,
             ar_mining_server,

.github/workflows/x-test-vdf.yml

@@ -59,6 +59,8 @@ jobs:
             ar_intervals,
             ar_join,
             ar_kv,
+            arweave_limiter_group,
+            arweave_limiter_metrics_collector,
             ar_merkle,
             ar_node,
             ar_node_utils,

.gitignore

@@ -35,4 +35,5 @@ release/output
 node_modules
 screenlog.0
 _*
+*~
 *.swp

apps/arweave/src/ar.erl

@@ -1192,6 +1192,7 @@ start(normal, _Args) ->
 	%% Start the Prometheus metrics subsystem.
 	prometheus_registry:register_collector(prometheus_process_collector),
 	prometheus_registry:register_collector(ar_metrics_collector),
+	prometheus_registry:register_collector(arweave_limiter_metrics_collector),
 
 	%% Register custom metrics.
 	ar_metrics:register(),
@@ -1323,9 +1324,15 @@ stop(_State) ->
 stop_dependencies() ->
 	?LOG_INFO("========== Stopping Arweave Node  =========="),
 	{ok, [_Kernel, _Stdlib, _SASL, _OSMon | Deps]} = application:get_key(arweave, applications),
+	application:stop(arweave_limiter),
 	lists:foreach(fun(Dep) -> application:stop(Dep) end, Deps).
 
 start_dependencies() ->
+	{ok, _Config} = arweave_config:get_env(),
+	%% We should start limiter automatically by having it as a dependency in the
+	%% .app.src, starting arweave_config should load the config, and arweave_limiter
+	%% should use arweave_config, rather than the legacy solution.
+	arweave_limiter:start(),
 	{ok, _} = application:ensure_all_started(arweave, permanent),
 	ok.
 

apps/arweave/src/ar_blacklist_middleware.erl

@@ -1,35 +1,17 @@
 -module(ar_blacklist_middleware).
 
--behaviour(cowboy_middleware).
-
--export([start/0, execute/2, reset/0, reset_rate_limit/3,
-		ban_peer/2, is_peer_banned/1, cleanup_ban/1, decrement_ip_addr/2]).
+-export([start/0, ban_peer/2, is_peer_banned/1, cleanup_ban/1]).
 -export([start_link/0]).
 
+-ifdef(AR_TEST).
+-export([reset/0]).
+-endif.
+
 -include_lib("arweave/include/ar.hrl").
 -include_lib("arweave_config/include/arweave_config.hrl").
 -include_lib("arweave/include/ar_blacklist_middleware.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
-execute(Req, Env) ->
-	IPAddr = requesting_ip_addr(Req),
-	{ok, Config} = arweave_config:get_env(),
-	case lists:member(blacklist, Config#config.disable) of
-		true ->
-			{ok, Req, Env};
-		_ ->
-			LocalIPs = [peer_to_ip_addr(Peer) || Peer <- Config#config.local_peers],
-			case lists:member(IPAddr, LocalIPs) of
-				true ->
-					{ok, Req, Env};
-				false ->
-					case increment_ip_addr(IPAddr, Req) of
-						{block, Limit} -> {stop, blacklisted(Limit, Req)};
-						pass -> {ok, Req, Env}
-					end
-			end
-	end.
-
 start_link() ->
 	{ok, spawn_link(fun() -> start() end)}.
 
@@ -43,6 +25,10 @@ start() ->
 		#{ skip_on_shutdown => false }
 	).
 
+reset() ->
+    true = ets:delete_all_objects(?MODULE),
+    ok.
+
 %% Ban a peer completely for TTLSeconds seoncds. Since we cannot trust the port,
 %% we ban the whole IP address.
 ban_peer(Peer, TTLSeconds) ->
@@ -83,71 +69,4 @@ cleanup_ban(TableID) ->
 	end.
 
 %private functions
-blacklisted(Limit, Req) ->
-	cowboy_req:reply(
-		429,
-		#{
-			<<"connection">> => <<"close">>,
-			<<"retry-after">> => integer_to_binary(?THROTTLE_PERIOD div 1000),
-			<<"x-rate-limit-limit">> => integer_to_binary(Limit)
-		},
-		<<"Too Many Requests">>,
-		Req
-	).
-
-reset() ->
-	true = ets:delete_all_objects(?MODULE),
-	ok.
-
-reset_rate_limit(TableID, IPAddr, Path) ->
-	case ets:whereis(?MODULE) of
-		TableID ->
-			ets:delete(?MODULE, {rate_limit, IPAddr, Path});
-		_ ->
-			table_owner_died
-	end.
-
-increment_ip_addr(IPAddr, Req) ->
-	case ets:whereis(?MODULE) of
-		undefined -> pass;
-		_ -> update_ip_addr(IPAddr, Req, 1)
-	end.
-
-decrement_ip_addr(IPAddr, Req) ->
-	case ets:whereis(?MODULE) of
-		undefined -> pass;
-		_ -> update_ip_addr(IPAddr, Req, -1)
-	end.
-
-update_ip_addr(IPAddr, Req, Delta) ->
-	{PathKey, Limit}  = get_key_limit(IPAddr, Req),
-	%% Divide by 2 as the throttle period is 30 seconds.
-	RequestLimit = Limit div 2,
-	Key = {rate_limit, IPAddr, PathKey},
-	case ets:update_counter(?MODULE, Key, {2, Delta}, {Key, 0}) of
-		1 ->
-			_ = ar_timer:apply_after(
-				?THROTTLE_PERIOD,
-				?MODULE,
-				reset_rate_limit,
-				[ets:whereis(?MODULE), IPAddr, PathKey],
-				#{ skip_on_shutdown => true }
-			),
-			pass;
-		Count when Count =< RequestLimit ->
-			pass;
-		_ ->
-			{block, Limit}
-	end.
-
-requesting_ip_addr(Req) ->
-	{IPAddr, _} = cowboy_req:peer(Req),
-	IPAddr.
-
 peer_to_ip_addr({A, B, C, D, _}) -> {A, B, C, D}.
-
-get_key_limit(IPAddr, Req) ->
-	Path = ar_http_iface_server:split_path(cowboy_req:path(Req)),
-	{ok, Config} = arweave_config:get_env(),
-	Map = maps:get(IPAddr, Config#config.requests_per_minute_limit_by_ip, #{}),
-	?RPM_BY_PATH(Path, Map)().