Merge pull request #16301 from rabbitmq/mergify/bp/v4.3.x/pr-16300

michaelklishin · web-flow · commit 09671cab92e1 · 2026-05-04T15:27:43.000-07:00
Apply the per-node `connection_max` limit to AMQP 1.0 connections (backport #16300)
diff --git a/deps/rabbit/Makefile b/deps/rabbit/Makefile
@@ -268,7 +268,7 @@ PARALLEL_CT_SET_4_B = per_user_connection_tracking per_vhost_connection_limit ra
 PARALLEL_CT_SET_4_C = msg_size_metrics unit_msg_size_metrics per_vhost_msg_store per_vhost_queue_limit priority_queue upgrade_preparation vhost
 PARALLEL_CT_SET_4_D = per_user_connection_channel_tracking product_info queue_type rabbitmq_queues_cli_integration rabbitmq_streams_cli_integration rabbitmqctl_integration rabbitmqctl_shutdown routing rabbit_amqqueue
 
-PARALLEL_CT_SET_5_A = consumer_recheck_prop pid_codec pid_codec_prop rabbit_direct_reply_to_prop rabbit_quorum_queue_prop prop_stream_arg_validation direct_reply_to_amqpl direct_reply_to_amqp classic_queue
+PARALLEL_CT_SET_5_A = amqp10_connection_max unit_amqp10_connection_max consumer_recheck_prop pid_codec pid_codec_prop rabbit_direct_reply_to_prop rabbit_quorum_queue_prop prop_stream_arg_validation direct_reply_to_amqpl direct_reply_to_amqp classic_queue
 PARALLEL_CT_SET_5_B = feature_flags_v2 backing_queue transactions
 PARALLEL_CT_SET_5_C = cluster_upgrade maintenance_mode
 PARALLEL_CT_SET_5_D = rabbit_fifo_dlx_integration publisher_confirms_parallel rabbit_exchange_type_modulus_hash
diff --git a/deps/rabbit/include/rabbit_amqp_reader.hrl b/deps/rabbit/include/rabbit_amqp_reader.hrl
@@ -43,6 +43,10 @@
 -record(v1,
         {parent :: pid(),
          helper_sup :: pid(),
+         %% A Ranch listener reference, used to enforce the per-node
+         %% `connection_max` limit.
+         %% Will be `undefined` for AMQP-over-WebSockets connections.
+         ranch_ref = undefined :: undefined | ranch:ref(),
          writer = none :: none | pid(),
          heartbeater = none :: none | rabbit_heartbeat:heartbeaters(),
          session_sup = none :: none | pid(),
diff --git a/deps/rabbit/src/rabbit_amqp_reader.erl b/deps/rabbit/src/rabbit_amqp_reader.erl
@@ -29,18 +29,24 @@
          handle_other/2,
          ensure_stats_timer/1]).
 
+-ifdef(TEST).
+-export([check_node_connection_limit/1]).
+-endif.
+
 -import(rabbit_amqp_util, [protocol_error/3]).
 
 -define(IS_RUNNING(State), State#v1.connection_state =:= running).
 
 unpack_from_0_9_1(
   {Sock, PendingRecv, SupPid, Buf, BufLen, ProxySocket,
-   ConnectionName, Host, PeerHost, Port, PeerPort, ConnectedAt, StatsTimer},
+   ConnectionName, Host, PeerHost, Port, PeerPort, ConnectedAt, StatsTimer,
+   RanchRef},
   Parent) ->
     logger:update_process_metadata(#{connection => ConnectionName}),
     #v1{parent           = Parent,
         websocket        = false,
         sock             = Sock,
+        ranch_ref        = RanchRef,
         callback         = {frame_header, sasl},
         pending_recv     = PendingRecv,
         helper_sup       = SupPid,
@@ -435,13 +441,16 @@ handle_connection_frame(
                                    user = User = #user{username = Username},
                                    auth_mechanism = {Mechanism, _Mod}
                                   },
+      ranch_ref = RanchRef,
       helper_sup = HelperSupPid,
       sock = Sock} = State0) ->
     Vhost = vhost(Hostname),
     logger:update_process_metadata(#{amqp_container => ContainerId,
                                      vhost => Vhost,
                                      user => Username}),
     ok = check_user_loopback(State0),
+    %% Enforced before per-vhost and per-user limits for efficiency.
+    ok = check_node_connection_limit(RanchRef),
     ok = check_vhost_exists(Vhost, State0),
     ok = check_vhost_alive(Vhost),
     ok = rabbit_access_control:check_vhost_access(User, Vhost, {socket, Sock}, #{}),
@@ -886,6 +895,27 @@ check_vhost_alive(Vhost) ->
                            [Vhost])
     end.
 
+-spec check_node_connection_limit(ranch:ref() | undefined) -> ok.
+check_node_connection_limit(undefined) ->
+    %% AMQP-over-WebSockets connections won't have an associated Ranch listener
+    ok;
+check_node_connection_limit(RanchRef) ->
+    case application:get_env(rabbit, connection_max, infinity) of
+        infinity ->
+            ok;
+        Limit when is_integer(Limit), Limit >= 0 ->
+            #{active_connections := ActiveConns} = ranch:info(RanchRef),
+            case ActiveConns > Limit of
+                false ->
+                    ok;
+                true ->
+                    protocol_error(
+                      ?V_1_0_AMQP_ERROR_RESOURCE_LIMIT_EXCEEDED,
+                      "connection refused: node connection limit (~p) is reached",
+                      [Limit])
+            end
+    end.
+
 check_vhost_connection_limit(Vhost, Username) ->
     case rabbit_vhost_limit:is_over_connection_limit(Vhost) of
         false ->
diff --git a/deps/rabbit/src/rabbit_reader.erl b/deps/rabbit/src/rabbit_reader.erl
@@ -1711,6 +1711,7 @@ become_10(State) ->
     State#v1{connection_state = {become, Fun}}.
 
 pack_for_1_0(Buf, BufLen, #v1{sock         = Sock,
+                              ranch_ref    = RanchRef,
                               pending_recv = PendingRecv,
                               helper_sup = {_HelperSup091, HelperSup10},
                               proxy_socket = ProxySocket,
@@ -1723,7 +1724,8 @@ pack_for_1_0(Buf, BufLen, #v1{sock         = Sock,
                                               peer_port = PeerPort,
                                               connected_at = ConnectedAt}}) ->
     {Sock, PendingRecv, HelperSup10, Buf, BufLen, ProxySocket,
-     Name, Host, PeerHost, Port, PeerPort, ConnectedAt, StatsTimer}.
+     Name, Host, PeerHost, Port, PeerPort, ConnectedAt, StatsTimer,
+     RanchRef}.
 
 respond_and_close(State, Channel, Reason, LogErr) ->
     log_hard_error(State, Channel, LogErr),
diff --git a/deps/rabbit/test/amqp10_connection_max_SUITE.erl b/deps/rabbit/test/amqp10_connection_max_SUITE.erl
@@ -0,0 +1,184 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(amqp10_connection_max_SUITE).
+
+-compile([export_all, nowarn_export_all]).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("amqp10_common/include/amqp10_framing.hrl").
+
+-define(TIMEOUT, 30_000).
+
+all() ->
+    [{group, single_node}].
+
+groups() ->
+    [{single_node, [],
+      [
+       node_connection_limit_amqp10,
+       node_limit_infinity_allows_amqp10,
+       node_limit_room_after_close,
+       node_connection_limit_shared_with_amqp091,
+       amqp10_connection_is_tracked
+      ]}].
+
+suite() ->
+    [{timetrap, {minutes, 3}}].
+
+init_per_suite(Config) ->
+    {ok, _} = application:ensure_all_started(amqp10_client),
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(Group, Config0) ->
+    Config1 = rabbit_ct_helpers:set_config(Config0,
+                                           [{rmq_nodename_suffix, Group},
+                                            {rmq_nodes_count, 1}]),
+    rabbit_ct_helpers:run_steps(Config1,
+                                rabbit_ct_broker_helpers:setup_steps() ++
+                                    rabbit_ct_client_helpers:setup_steps()).
+
+end_per_group(_Group, Config) ->
+    rabbit_ct_helpers:run_steps(Config,
+                                rabbit_ct_client_helpers:teardown_steps() ++
+                                    rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_testcase(Testcase, Config) ->
+    set_node_limit(Config, infinity),
+    rabbit_ct_helpers:testcase_started(Config, Testcase).
+
+end_per_testcase(Testcase, Config) ->
+    set_node_limit(Config, infinity),
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+%% -------------------------------------------------------------------
+%% Test cases
+%% -------------------------------------------------------------------
+
+node_connection_limit_amqp10(Config) ->
+    set_node_limit(Config, 0),
+    {refused, _} = open_amqp10(Config),
+
+    set_node_limit(Config, 5),
+    Conns = [open_amqp10_ok(Config) || _ <- lists:seq(1, 5)],
+    {refused, _} = open_amqp10(Config),
+    [amqp_utils:close_connection_sync(C) || C <- Conns],
+    ok.
+
+node_limit_infinity_allows_amqp10(Config) ->
+    set_node_limit(Config, infinity),
+    Conns = [open_amqp10_ok(Config) || _ <- lists:seq(1, 10)],
+    [amqp_utils:close_connection_sync(C) || C <- Conns],
+    ok.
+
+node_limit_room_after_close(Config) ->
+    set_node_limit(Config, 2),
+    C1 = open_amqp10_ok(Config),
+    C2 = open_amqp10_ok(Config),
+    {refused, _} = open_amqp10(Config),
+    amqp_utils:close_connection_sync(C1),
+    {ok, C3} = retry_open_amqp10(Config, 30),
+    [amqp_utils:close_connection_sync(C) || C <- [C2, C3]],
+    ok.
+
+%% Historically only AMQP 0-9-1 honored `connection_max`,
+%% so this example uses both protocols.
+node_connection_limit_shared_with_amqp091(Config) ->
+    set_node_limit(Config, 2),
+    C091 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0),
+    true = is_pid(C091),
+    C10 = open_amqp10_ok(Config),
+    {refused, _} = open_amqp10(Config),
+    {error, not_allowed} = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0),
+    amqp_utils:close_connection_sync(C10),
+    rabbit_ct_client_helpers:close_connection(C091),
+    ok.
+
+amqp10_connection_is_tracked(Config) ->
+    ok = event_recorder:start(Config),
+    try
+        C = open_amqp10_ok(Config),
+        ServerPid = wait_for_created_pid(Config),
+        Local = rabbit_ct_broker_helpers:rpc(
+                  Config, 0, rabbit_networking, local_connections, []),
+        ?assert(lists:member(ServerPid, Local)),
+        amqp_utils:close_connection_sync(C),
+        wait_for_event_types(Config, [connection_closed])
+    after
+        ok = event_recorder:stop(Config)
+    end,
+    ok.
+
+%% -------------------------------------------------------------------
+%% Helpers
+%% -------------------------------------------------------------------
+
+set_node_limit(Config, Limit) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+                                      application, set_env,
+                                      [rabbit, connection_max, Limit]).
+
+open_amqp10_ok(Config) ->
+    {ok, Conn} = open_amqp10(Config),
+    Conn.
+
+open_amqp10(Config) ->
+    {ok, Conn} = amqp10_client:open_connection(amqp_utils:connection_config(Config)),
+    receive
+        {amqp10_event, {connection, Conn, opened}} ->
+            {ok, Conn};
+        {amqp10_event, {connection, Conn, {closed, Reason}}} ->
+            {refused, Reason}
+    after ?TIMEOUT ->
+              ct:fail({open_amqp10_timeout, Conn})
+    end.
+
+%% Covers the brief window between client-side close and Ranch
+%% actually dropping the active TCP connection.
+retry_open_amqp10(_Config, 0) ->
+    ct:fail(retry_open_amqp10_exhausted);
+retry_open_amqp10(Config, Attempts) ->
+    case open_amqp10(Config) of
+        {ok, _} = Result -> Result;
+        {refused, _} ->
+            timer:sleep(100),
+            retry_open_amqp10(Config, Attempts - 1)
+    end.
+
+wait_for_created_pid(Config) ->
+    wait_for_created_pid(Config, [], 30).
+
+wait_for_created_pid(_Config, _Acc, 0) ->
+    ct:fail(missing_connection_created_event);
+wait_for_created_pid(Config, Acc, N) ->
+    Acc1 = Acc ++ event_recorder:get_events(Config),
+    case [proplists:get_value(pid, Props)
+          || #event{type = connection_created, props = Props} <- Acc1] of
+        [Pid | _] when is_pid(Pid) -> Pid;
+        []                          -> wait_for_created_pid(Config, Acc1, N - 1)
+    end.
+
+wait_for_event_types(Config, Wanted) ->
+    wait_for_event_types(Config, Wanted, 30).
+
+wait_for_event_types(_Config, [], _N) ->
+    ok;
+wait_for_event_types(_Config, Wanted, 0) ->
+    ct:fail({missing_event_types, Wanted});
+wait_for_event_types(Config, Wanted, N) ->
+    Seen = [T || #event{type = T} <- event_recorder:get_events(Config)],
+    Remaining = [T || T <- Wanted, not lists:member(T, Seen)],
+    case Remaining of
+        [] -> ok;
+        _  -> wait_for_event_types(Config, Remaining, N - 1)
+    end.
diff --git a/deps/rabbit/test/unit_amqp10_connection_max_SUITE.erl b/deps/rabbit/test/unit_amqp10_connection_max_SUITE.erl
