Safely handle channel creation errors on initiating connections and top-level supervisors

deps/amqp_client/src/amqp_channel_sup.erl

@@ -39,6 +39,7 @@ start_link(Type, Connection, ConnName, InfraArgs, ChNumber,
             {ok, AState} = init_command_assembler(Type),
             {ok, Sup, {ChPid, AState}};
         {error, _}=Error ->
+            rabbit_misc:shutdown_supervisor(Sup),
             Error
     end.
 

deps/amqp_client/src/amqp_channels_manager.erl

@@ -105,12 +105,15 @@ handle_open_channel(ProposedNumber, Consumer, InfraArgs,
                     State = #state{channel_sup_sup = ChSupSup}) ->
     case new_number(ProposedNumber, State) of
         {ok, Number} ->
-            {ok, _ChSup, {Ch, AState}} =
-                amqp_channel_sup_sup:start_channel_sup(ChSupSup, InfraArgs,
-                                                       Number, Consumer),
-            NewState = internal_register(Number, Ch, AState, State),
-            erlang:monitor(process, Ch),
-            {reply, {ok, Ch}, NewState};
+            case amqp_channel_sup_sup:start_channel_sup(ChSupSup, InfraArgs,
+                                                        Number, Consumer) of
+                {ok, _ChSup, {Ch, AState}} ->
+                    NewState = internal_register(Number, Ch, AState, State),
+                    erlang:monitor(process, Ch),
+                    {reply, {ok, Ch}, NewState};
+                {error, _} = Error ->
+                    {reply, Error, State}
+            end;
         {error, _} = Error ->
             {reply, Error, State}
     end.

deps/rabbit/src/rabbit_channel_sup.erl

@@ -42,7 +42,8 @@
 
 %%----------------------------------------------------------------------------
 
--spec start_link(start_link_args()) -> {'ok', pid(), {pid(), any()}}.
+-spec start_link(start_link_args()) ->
+    {'ok', pid(), {pid(), any()}} | {'error', any()}.
 
 start_link({tcp, Sock, Channel, FrameMax, ReaderPid, ConnName, User,
             VHost, Capabilities, Collector}) ->
@@ -62,9 +63,14 @@ start_link({tcp, Sock, Channel, FrameMax, ReaderPid, ConnName, User,
                   shutdown => ?FAIR_WAIT,
                   type => worker,
                   modules => [rabbit_channel]},
-    {ok, ChannelPid} = supervisor:start_child(SupPid, ChildSpec),
-    {ok, AState} = rabbit_command_assembler:init(),
-    {ok, SupPid, {ChannelPid, AState}};
+    case supervisor:start_child(SupPid, ChildSpec) of
+        {ok, ChannelPid} ->
+            {ok, AState} = rabbit_command_assembler:init(),
+            {ok, SupPid, {ChannelPid, AState}};
+        {error, Reason} ->
+            rabbit_misc:shutdown_supervisor(SupPid),
+            {error, Reason}
+    end;
 start_link({direct, Channel, ClientChannelPid, ConnPid, ConnName,
             User, VHost, Capabilities, Collector, AmqpParams}) ->
     {ok, SupPid} = supervisor:start_link(
@@ -81,8 +87,13 @@ start_link({direct, Channel, ClientChannelPid, ConnPid, ConnName,
                   shutdown => ?FAIR_WAIT,
                   type => worker,
                   modules => [rabbit_channel]},
-    {ok, ChannelPid} = supervisor:start_child(SupPid, ChildSpec),
-    {ok, SupPid, {ChannelPid, none}}.
+    case supervisor:start_child(SupPid, ChildSpec) of
+        {ok, ChannelPid} ->
+            {ok, SupPid, {ChannelPid, none}};
+        {error, Reason} ->
+            rabbit_misc:shutdown_supervisor(SupPid),
+            {error, Reason}
+    end.
 
 %%----------------------------------------------------------------------------
 

deps/rabbit/src/rabbit_channel_sup_sup.erl

@@ -28,7 +28,7 @@ start_link() ->
     supervisor:start_link(?MODULE, []).
 
 -spec start_channel(pid(), rabbit_channel_sup:start_link_args()) ->
-          {'ok', pid(), {pid(), any()}}.
+          {'ok', pid(), {pid(), any()}} | {'error', any()}.
 
 start_channel(Pid, Args) ->
     supervisor:start_child(Pid, [Args]).

deps/rabbit/src/rabbit_direct.erl

@@ -209,7 +209,7 @@ connect1(User = #user{username = Username}, VHost, Pid, Infos) ->
         (rabbit_channel:channel_number(), pid(), pid(), string(),
          rabbit_types:protocol(), rabbit_types:user(), rabbit_types:vhost(),
          rabbit_framing:amqp_table(), pid(), any()) ->
-            {'ok', pid()}.
+            {'ok', pid()} | {'error', any()}.
 
 %% Kept for compatibility with older amqp_client versions (rpc:call).
 start_channel(Number, ClientChannelPid, ConnPid, ConnName, _Protocol,
@@ -221,19 +221,26 @@ start_channel(Number, ClientChannelPid, ConnPid, ConnName, _Protocol,
         (rabbit_channel:channel_number(), pid(), pid(), string(),
          rabbit_types:user(), rabbit_types:vhost(),
          rabbit_framing:amqp_table(), pid(), any()) ->
-            {'ok', pid()}.
+            {'ok', pid()} | {'error', any()}.
 
 start_channel(Number, ClientChannelPid, ConnPid, ConnName,
               User = #user{username = Username}, VHost, Capabilities,
               Collector, AmqpParams) ->
     case rabbit_auth_backend_internal:is_over_channel_limit(Username) of
         false ->
-            {ok, _, {ChannelPid, _}} =
-                supervisor:start_child(
-                  rabbit_direct_client_sup,
-                  [{direct, Number, ClientChannelPid, ConnPid, ConnName,
-                    User, VHost, Capabilities, Collector, AmqpParams}]),
-            {ok, ChannelPid};
+            case supervisor:start_child(
+                   rabbit_direct_client_sup,
+                   [{direct, Number, ClientChannelPid, ConnPid, ConnName,
+                     User, VHost, Capabilities, Collector, AmqpParams}]) of
+                {ok, _, {ChannelPid, _}} ->
+                    {ok, ChannelPid};
+                {error, Reason} ->
+                    ?LOG_ERROR(
+                        "Error on direct connection ~tp~n"
+                        "failed to open channel: ~tp",
+                        [ConnPid, Reason]),
+                    {error, channel_open_failed}
+            end;
         {true, Limit} ->
             ?LOG_ERROR(
                 "Error on direct connection ~tp~n"

deps/rabbit/src/rabbit_reader.erl

@@ -975,15 +975,24 @@ create_channel(Channel,
                    } = State) ->
     case is_over_limits(Username) of
         false ->
-            {ok, _ChSupPid, {ChPid, AState}} =
-                rabbit_channel_sup_sup:start_channel(
-                  ChanSupSup, {tcp, Sock, Channel, FrameMax, self(), Name,
-                               User, VHost, Capabilities,
-                               Collector}),
-            MRef = erlang:monitor(process, ChPid),
-            put({ch_pid, ChPid}, {Channel, MRef}),
-            put({channel, Channel}, {ChPid, AState}),
-            {ok, {ChPid, AState}, State#v1{channel_count = ChannelCount + 1}};
+            case rabbit_channel_sup_sup:start_channel(
+                   ChanSupSup, {tcp, Sock, Channel, FrameMax, self(), Name,
+                                User, VHost, Capabilities,
+                                Collector}) of
+                {ok, _ChSupPid, {ChPid, AState}} ->
+                    MRef = erlang:monitor(process, ChPid),
+                    put({ch_pid, ChPid}, {Channel, MRef}),
+                    put({channel, Channel}, {ChPid, AState}),
+                    {ok, {ChPid, AState}, State#v1{channel_count = ChannelCount + 1}};
+                {error, Reason} ->
+                    ?LOG_ERROR(
+                        "Connection ~ts failed to open channel ~tp: ~tp",
+                        [Name, Channel, Reason]),
+                    {error, rabbit_misc:amqp_error(
+                              internal_error,
+                              "failed to open channel ~tp",
+                              [Channel], none)}
+            end;
         {true, Limit, Fmt, FmtArg} ->
             {error, rabbit_misc:amqp_error(
                       not_allowed,

deps/rabbit/test/channel_interceptor_SUITE.erl

@@ -24,7 +24,9 @@ groups() ->
           register_interceptor,
           register_interceptor_failing_with_amqp_error,
           register_interceptor_crashing_with_amqp_error_exception,
-          register_failing_interceptors
+          register_failing_interceptors,
+          conflicting_interceptors_close_network_connections_gracefully,
+          conflicting_interceptors_close_direct_connections_gracefully
         ]}
     ].
 
@@ -213,6 +215,94 @@ register_failing_interceptors(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, register_interceptor1, [Config, failing_dummy_interceptor]).
 
+conflicting_interceptors_close_network_connections_gracefully(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, conflicting_interceptors_close_network_connections_gracefully1, [Config]).
+
+conflicting_interceptors_close_network_connections_gracefully1(Config) ->
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor">>,
+                                  dummy_interceptor),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"conflicting dummy interceptor">>,
+                                  dummy_interceptor_conflicting),
+    try
+        Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0),
+        ?assert(is_pid(Conn)),
+        ?assert(is_process_alive(Conn)),
+        ConnMRef = erlang:monitor(process, Conn),
+        Result = try amqp_connection:open_channel(Conn)
+                 catch _:_ -> {error, channel_open_failed}
+                 end,
+        ?assertMatch({error, _}, Result),
+        receive
+            {'DOWN', ConnMRef, process, Conn, ConnExitReason} ->
+                ?assertMatch(
+                    {shutdown, {server_initiated_close, _, _}},
+                    ConnExitReason)
+        after 10000 ->
+            ct:fail("Connection process did not terminate")
+        end
+    after
+        ok = rabbit_registry:unregister(channel_interceptor,
+                                        <<"dummy interceptor">>),
+        ok = rabbit_registry:unregister(channel_interceptor,
+                                        <<"conflicting dummy interceptor">>)
+    end,
+    %% Verify the server still accepts new connections and channels
+    Conn2 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0),
+    ?assert(is_pid(Conn2)),
+    {ok, Ch} = amqp_connection:open_channel(Conn2),
+    ?assert(is_pid(Ch)),
+    ok = amqp_connection:close(Conn2),
+    passed.
+
+conflicting_interceptors_close_direct_connections_gracefully(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, conflicting_interceptors_close_direct_connections_gracefully1, [Config]).
+
+conflicting_interceptors_close_direct_connections_gracefully1(Config) ->
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"dummy interceptor">>,
+                                  dummy_interceptor),
+    ok = rabbit_registry:register(channel_interceptor,
+                                  <<"conflicting dummy interceptor">>,
+                                  dummy_interceptor_conflicting),
+    try
+        Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+        Params = #amqp_params_direct{node = Node,
+                                     virtual_host = <<"/">>,
+                                     username = <<"guest">>,
+                                     password = <<"guest">>},
+        {ok, Conn} = amqp_connection:start(Params),
+        ?assert(is_pid(Conn)),
+        ?assert(is_process_alive(Conn)),
+        %% open_channel must return {error, _} gracefully, not crash.
+        %% If the connection process crashes (e.g. badmatch in
+        %% amqp_channels_manager), this call will throw and fail the test.
+        Result = amqp_connection:open_channel(Conn),
+        ?assertMatch({error, _}, Result),
+        ?assert(is_process_alive(Conn)),
+        ok = amqp_connection:close(Conn)
+    after
+        ok = rabbit_registry:unregister(channel_interceptor,
+                                        <<"dummy interceptor">>),
+        ok = rabbit_registry:unregister(channel_interceptor,
+                                        <<"conflicting dummy interceptor">>)
+    end,
+    %% Verify the server still accepts new direct connections and channels
+    Node2 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Params2 = #amqp_params_direct{node = Node2,
+                                  virtual_host = <<"/">>,
+                                  username = <<"guest">>,
+                                  password = <<"guest">>},
+    {ok, Conn2} = amqp_connection:start(Params2),
+    ?assert(is_pid(Conn2)),
+    {ok, Ch} = amqp_connection:open_channel(Conn2),
+    ?assert(is_pid(Ch)),
+    ok = amqp_connection:close(Conn2),
+    passed.
+
 check_send_receive(Ch1, QName, Send, Receive) ->
     amqp_channel:call(Ch1,
                         #'basic.publish'{routing_key = QName},

deps/rabbit/test/dummy_interceptor_conflicting.erl

@@ -0,0 +1,28 @@
+%% 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(dummy_interceptor_conflicting).
+
+-behaviour(rabbit_channel_interceptor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("rabbit_common/include/rabbit_framing.hrl").
+
+-compile(export_all).
+
+init(_Ch) ->
+    undefined.
+
+description() ->
+    [{description,
+      <<"Interceptor that conflicts with dummy_interceptor on queue.declare">>}].
+
+intercept(Method, Content, _IState) ->
+    {Method, Content}.
+
+applies_to() ->
+    ['queue.declare'].

deps/rabbit_common/include/rabbit_misc.hrl

@@ -5,5 +5,6 @@
 %% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
--define(RPC_TIMEOUT, 15000).
+-define(RPC_TIMEOUT, 15_000).
 -define(RPC_INFINITE_TIMEOUT, infinity).
+-define(DEFAULT_TIMEOUT, 5_000).

deps/rabbit_common/src/rabbit_misc.erl

@@ -79,6 +79,7 @@
 -export([raw_read_file/1]).
 -export([strip_bom/1]).
 -export([find_child/2]).
+-export([shutdown_supervisor/1]).
 -export([is_regular_file/1]).
 -export([safe_ets_update_counter/3, safe_ets_update_counter/4, safe_ets_update_counter/5,
          safe_ets_update_element/3, safe_ets_update_element/4, safe_ets_update_element/5]).
@@ -1435,6 +1436,16 @@ find_child(Supervisor, Name) ->
     [Pid || {Name1, Pid, _Type, _Modules} <- supervisor:which_children(Supervisor),
             Name1 =:= Name].
 
+shutdown_supervisor(SupPid) when is_pid(SupPid) ->
+    MRef = erlang:monitor(process, SupPid),
+    unlink(SupPid),
+    exit(SupPid, shutdown),
+    receive
+        {'DOWN', MRef, process, SupPid, _} -> ok
+    after ?DEFAULT_TIMEOUT ->
+        exit({shutdown_supervisor_timeout, SupPid})
+    end.
+
 %% -------------------------------------------------------------------------
 %% Begin copypasta from gen_server2.erl