Merge pull request #16743 from rabbitmq/mergify/bp/v4.3.x/pr-16740

Handle unexpected failures during channel termination cleanup (backport #16740)

Author: michaelklishin

deps/rabbit/src/rabbit_channel.erl

@@ -713,26 +713,32 @@ terminate(_Reason,
           State = #ch{cfg = #conf{user = #user{username = Username}},
                       consumer_mapping = CM,
                       queue_states = QueueCtxs}) ->
-    rabbit_queue_type:close(QueueCtxs),
-    {_Res, _State1} = notify_queues(State),
-    pg_local:leave(rabbit_channels, self()),
-    rabbit_event:if_enabled(State, #ch.stats_timer,
-                            fun() -> emit_stats(State) end),
-    [delete_stats(Tag) || {Tag, _} <- get()],
-    maybe_decrease_global_publishers(State),
-    maps:foreach(
-      fun (_, _) ->
-              rabbit_global_counters:consumer_deleted(amqp091)
-      end, CM),
-    rabbit_core_metrics:channel_closed(self()),
-    rabbit_event:notify(channel_closed, [{pid, self()},
-                                         {user_who_performed_action, Username},
-                                         {consumer_count, maps:size(CM)}]),
-    case rabbit_confirms:size(State#ch.unconfirmed) of
-        0 -> ok;
-        NumConfirms ->
-            ?LOG_WARNING("Channel is stopping with ~b pending publisher confirms",
-                               [NumConfirms])
+    try
+        rabbit_queue_type:close(QueueCtxs),
+        {_Res, _State1} = notify_queues(State),
+        pg_local:leave(rabbit_channels, self()),
+        rabbit_event:if_enabled(State, #ch.stats_timer,
+                                fun() -> emit_stats(State) end),
+        [delete_stats(Tag) || {Tag, _} <- get()],
+        maybe_decrease_global_publishers(State),
+        maps:foreach(
+          fun (_, _) ->
+                  rabbit_global_counters:consumer_deleted(amqp091)
+          end, CM),
+        rabbit_core_metrics:channel_closed(self()),
+        rabbit_event:notify(channel_closed, [{pid, self()},
+                                             {user_who_performed_action, Username},
+                                             {consumer_count, maps:size(CM)}]),
+        case rabbit_confirms:size(State#ch.unconfirmed) of
+            0 -> ok;
+            NumConfirms ->
+                ?LOG_WARNING("Channel is stopping with ~b pending publisher confirms",
+                                   [NumConfirms])
+        end
+    catch
+        _Class:Reason ->
+            ?LOG_WARNING("Channel ~tp termination cleanup failed with reason: ~tp",
+                         [self(), Reason])
     end.
 
 code_change(_OldVsn, State, _Extra) ->

deps/rabbit/test/channel_SUITE.erl

@@ -22,7 +22,8 @@ all() ->
 groups() ->
     [
       {non_parallel_tests, [], [
-          ready_for_close_with_dead_writer
+          ready_for_close_with_dead_writer,
+          terminate_safely_handles_cleanup_failures
         ]}
     ].
 
@@ -94,6 +95,46 @@ ready_for_close_with_dead_writer1(_Config) ->
     end,
     passed.
 
+terminate_safely_handles_cleanup_failures(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, terminate_safely_handles_cleanup_failures1, [Config]).
+
+terminate_safely_handles_cleanup_failures1(_Config) ->
+    {Writer, Ch} = start_channel_and_writer(),
+    MRef = erlang:monitor(process, Ch),
+
+    %% Make rabbit_queue_type:close/1 crash to simulate an unexpected
+    %% failure during channel termination cleanup.
+    ok = meck:new(rabbit_queue_type, [passthrough]),
+    meck:expect(rabbit_queue_type, close,
+                fun(_) -> error(fake_cleanup_failure) end),
+
+    rabbit_channel_common:do(Ch, #'channel.close'{reply_code = 200,
+                                                  reply_text = <<"OK">>,
+                                                  class_id = 0,
+                                                  method_id = 0}),
+    receive
+        {channel_closing, Ch} -> ok
+    after ?TIMEOUT ->
+        meck:unload(rabbit_queue_type),
+        throw(failed_to_receive_channel_closing)
+    end,
+
+    exit(Writer, kill),
+
+    rabbit_channel_common:ready_for_close(Ch),
+    receive
+        {'DOWN', MRef, process, Ch, normal} ->
+            meck:unload(rabbit_queue_type),
+            passed;
+        {'DOWN', MRef, process, Ch, Reason} ->
+            meck:unload(rabbit_queue_type),
+            throw({channel_should_terminate_normally, Reason})
+    after ?TIMEOUT ->
+        meck:unload(rabbit_queue_type),
+        throw(channel_did_not_terminate)
+    end.
+
 start_channel_and_writer() ->
     {Writer, _Limiter, Ch} = rabbit_ct_broker_helpers:test_channel(),
     ok = rabbit_channel_common:do(Ch, #'channel.open'{}),