fix(gateway): split inbound and outbound rpc concurrency keys (#2186)

This commit is contained in:
Hampus authored and GitHub committed 2026-08-30 23:16:53 +02:00
1 parent 6d305bacdf
commit 299cc40ff5
5 files changed
+31 -4

No files matched your search

@@ -99,6 +99,9 @@ env_gateway_base_config() ->
<<"gateway_http_rpc_max_concurrency">> => env_int( <<"gateway_http_rpc_max_concurrency">> => env_int(
"FLUXER_GATEWAY_HTTP_RPC_MAX_CONCURRENCY", 512 "FLUXER_GATEWAY_HTTP_RPC_MAX_CONCURRENCY", 512
), ),
<<"gateway_nats_rpc_max_handlers">> => env_int(
"FLUXER_GATEWAY_NATS_RPC_MAX_HANDLERS", 512
),
<<"gateway_http_failure_threshold">> => env_int( <<"gateway_http_failure_threshold">> => env_int(
"FLUXER_GATEWAY_HTTP_FAILURE_THRESHOLD", 6 "FLUXER_GATEWAY_HTTP_FAILURE_THRESHOLD", 6
), ),
@@ -205,6 +208,8 @@ build_core_config(Service, Internal, Nats, Proxy) ->
nats_core_url => get_string(Nats, <<"core_url">>, "nats://127.0.0.1:4222"), nats_core_url => get_string(Nats, <<"core_url">>, "nats://127.0.0.1:4222"),
nats_auth_token => get_string(Nats, <<"auth_token">>, ""), nats_auth_token => get_string(Nats, <<"auth_token">>, ""),
rpc_auth_token => get_string(Service, <<"rpc_auth_token">>, ""), rpc_auth_token => get_string(Service, <<"rpc_auth_token">>, ""),
gateway_nats_rpc_max_handlers =>
get_int(Service, <<"gateway_nats_rpc_max_handlers">>, 512),
identify_rate_limit_enabled => get_bool( identify_rate_limit_enabled => get_bool(
Service, <<"identify_rate_limit_enabled">>, false Service, <<"identify_rate_limit_enabled">>, false
) )
@@ -22,7 +22,7 @@
]). ]).
-define(DEFAULT_POOL_SIZE, 16). -define(DEFAULT_POOL_SIZE, 16).
-define(DEFAULT_MAX_HANDLERS, 1024). -define(DEFAULT_MAX_HANDLERS, 512).
-define(RECONNECT_BASE_DELAY_MS, 2000). -define(RECONNECT_BASE_DELAY_MS, 2000).
-define(RECONNECT_MAX_DELAY_MS, 30000). -define(RECONNECT_MAX_DELAY_MS, 30000).
-define(CONNECT_TIMEOUT_MS, 10000). -define(CONNECT_TIMEOUT_MS, 10000).
@@ -244,7 +244,7 @@ pool_size() ->
-spec max_handlers() -> pos_integer(). -spec max_handlers() -> pos_integer().
max_handlers() -> max_handlers() ->
case fluxer_gateway_env:get(gateway_http_rpc_max_concurrency) of case fluxer_gateway_env:get(gateway_nats_rpc_max_handlers) of
Value when is_integer(Value), Value > 0 -> Value; Value when is_integer(Value), Value > 0 -> Value;
_ -> ?DEFAULT_MAX_HANDLERS _ -> ?DEFAULT_MAX_HANDLERS
end. end.
@@ -15,7 +15,7 @@
build_connect_opts/1 build_connect_opts/1
]). ]).
-define(DEFAULT_MAX_HANDLERS, 1024). -define(DEFAULT_MAX_HANDLERS, 512).
-define(RECONNECT_DELAY_MS, 2000). -define(RECONNECT_DELAY_MS, 2000).
-define(CONNECT_TIMEOUT_MS, 10000). -define(CONNECT_TIMEOUT_MS, 10000).
-define(RPC_SUBJECT_WILDCARD, <<"rpc.gateway.>">>). -define(RPC_SUBJECT_WILDCARD, <<"rpc.gateway.>">>).
@@ -218,7 +218,7 @@ schedule_reconnect(State) ->
-spec max_handlers() -> pos_integer(). -spec max_handlers() -> pos_integer().
max_handlers() -> max_handlers() ->
case fluxer_gateway_env:get(gateway_http_rpc_max_concurrency) of case fluxer_gateway_env:get(gateway_nats_rpc_max_handlers) of
Value when is_integer(Value), Value > 0 -> Value; Value when is_integer(Value), Value > 0 -> Value;
_ -> ?DEFAULT_MAX_HANDLERS _ -> ?DEFAULT_MAX_HANDLERS
end. end.
@@ -127,6 +127,24 @@ env_only_push_and_http_runtime_config_test() ->
end end
). ).
rpc_concurrency_keys_are_independent_test() ->
with_envs(
[
{"FLUXER_GATEWAY_HTTP_RPC_MAX_CONCURRENCY", "128"},
{"FLUXER_GATEWAY_NATS_RPC_MAX_HANDLERS", "2048"}
],
fun() ->
Config = fluxer_gateway_config:load(),
?assertEqual(128, maps:get(gateway_http_rpc_max_concurrency, Config)),
?assertEqual(2048, maps:get(gateway_nats_rpc_max_handlers, Config))
end
).
rpc_concurrency_key_defaults_test() ->
Config = fluxer_gateway_config:build_config(#{}),
?assertEqual(512, maps:get(gateway_nats_rpc_max_handlers, Config)),
?assertEqual(512, maps:get(gateway_http_rpc_max_concurrency, Config)).
optional_string_test() -> optional_string_test() ->
?assertEqual(undefined, fluxer_gateway_config:optional_string(undefined)), ?assertEqual(undefined, fluxer_gateway_config:optional_string(undefined)),
?assertEqual("hello", fluxer_gateway_config:optional_string(<<"hello">>)), ?assertEqual("hello", fluxer_gateway_config:optional_string(<<"hello">>)),
@@ -212,6 +212,10 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record<string, NamedEnvOverride> = {
path: ['services', 'gateway', 'gateway_http_rpc_max_concurrency'], path: ['services', 'gateway', 'gateway_http_rpc_max_concurrency'],
parse: parseEnvValue, parse: parseEnvValue,
}, },
FLUXER_GATEWAY_NATS_RPC_MAX_HANDLERS: {
path: ['services', 'gateway', 'gateway_nats_rpc_max_handlers'],
parse: parseEnvValue,
},
FLUXER_GATEWAY_SHUTDOWN_DRAIN_WAIT_MS: { FLUXER_GATEWAY_SHUTDOWN_DRAIN_WAIT_MS: {
path: ['services', 'gateway', 'shutdown_drain_wait_ms'], path: ['services', 'gateway', 'shutdown_drain_wait_ms'],
parse: parseEnvValue, parse: parseEnvValue,