Handle database timeouts when declaring exchanges

the-mikedavis · the-mikedavis · commit 52a0d70e1588 · 2024-07-22T16:02:03.000-04:00
The spec of `rabbit_exchange:declare/7` needs to be updated to return
`{ok, Exchange} | {error, Reason}` instead of the old return value of
`rabbit_types:exchange()`. This is safe to do since `declare/7` is not
called by RPC - from the CLI or otherwise - outside of test suites, and
in test suites only through the CLI's `TestHelper.declare_exchange/7`.
Callers of this helper are updated in this commit.

Otherwise this commit updates callers to unwrap the `{ok, Exchange}`
and bubble up errors.
diff --git a/deps/rabbit/src/rabbit_amqp_management.erl b/deps/rabbit/src/rabbit_amqp_management.erl
@@ -210,9 +210,18 @@ handle_http_req(<<"PUT">>,
             {error, not_found} ->
                 ok = prohibit_cr_lf(XNameBin),
                 ok = prohibit_reserved_amq(XName),
-                rabbit_exchange:declare(
-                  XName, XTypeAtom, Durable, AutoDelete,
-                  Internal, XArgs, Username)
+                case rabbit_exchange:declare(
+                       XName, XTypeAtom, Durable, AutoDelete,
+                       Internal, XArgs, Username) of
+                    {ok, DeclaredX} ->
+                        DeclaredX;
+                    {error, timeout} ->
+                        throw(
+                          <<"500">>,
+                          "Could not create exchange '~ts' in vhost '~ts' "
+                          "because the operation timed out",
+                          [XName, Vhost])
+                end
         end,
     try rabbit_exchange:assert_equivalence(
           X, XTypeAtom, Durable, AutoDelete, Internal, XArgs) of
diff --git a/deps/rabbit/src/rabbit_channel.erl b/deps/rabbit/src/rabbit_channel.erl
@@ -2569,13 +2569,22 @@ handle_method(#'exchange.declare'{exchange    = XNameBin,
                                  check_write_permitted(AName, User, AuthzContext),
                                  ok
                 end,
-                rabbit_exchange:declare(ExchangeName,
-                                        CheckedType,
-                                        Durable,
-                                        AutoDelete,
-                                        Internal,
-                                        Args,
-                                        Username)
+                case rabbit_exchange:declare(ExchangeName,
+                                             CheckedType,
+                                             Durable,
+                                             AutoDelete,
+                                             Internal,
+                                             Args,
+                                             Username) of
+                    {ok, DeclaredX} ->
+                        DeclaredX;
+                    {error, timeout} ->
+                        rabbit_misc:protocol_error(
+                          internal_error,
+                          "failed to declare exchange '~ts' in vhost '~ts' "
+                          "because the operation timed out",
+                          [XNameBinStripped, VHostPath])
+                end
         end,
     ok = rabbit_exchange:assert_equivalence(X, CheckedType, Durable,
                                             AutoDelete, Internal, Args);
diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl
@@ -368,7 +368,9 @@ update_in_khepri_tx(Name, Fun) ->
 
 -spec create_or_get(Exchange) -> Ret when
       Exchange :: rabbit_types:exchange(),
-      Ret :: {new, Exchange} | {existing, Exchange}.
+      Ret :: {new, Exchange} |
+             {existing, Exchange} |
+             rabbit_khepri:timeout_error().
 %% @doc Writes an exchange record if it doesn't exist already or returns
 %% the existing one.
 %%
@@ -400,7 +402,9 @@ create_or_get_in_khepri(#exchange{name = XName} = X) ->
         ok ->
             {new, X};
         {error, {khepri, mismatching_node, #{node_props := #{data := ExistingX}}}} ->
-            {existing, ExistingX}
+            {existing, ExistingX};
+        {error, timeout} = Err ->
+            Err
     end.
 
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_definitions.erl b/deps/rabbit/src/rabbit_definitions.erl
@@ -863,13 +863,18 @@ add_exchange_int(Exchange, Name, ActingUser) ->
                            undefined -> false; %% =< 2.2.0
                            I         -> I
                        end,
-            rabbit_exchange:declare(Name,
-                                    rabbit_exchange:check_type(maps:get(type, Exchange, undefined)),
-                                    maps:get(durable,                         Exchange, undefined),
-                                    maps:get(auto_delete,                     Exchange, undefined),
-                                    Internal,
-                                    args(maps:get(arguments, Exchange, undefined)),
-                                    ActingUser)
+            case rabbit_exchange:declare(Name,
+                                         rabbit_exchange:check_type(maps:get(type, Exchange, undefined)),
+                                         maps:get(durable,                         Exchange, undefined),
+                                         maps:get(auto_delete,                     Exchange, undefined),
+                                         Internal,
+                                         args(maps:get(arguments, Exchange, undefined)),
+                                         ActingUser) of
+                {ok, _Exchange} ->
+                    ok;
+                {error, timeout} = Err ->
+                    throw(Err)
+            end
     end.
 
 add_binding(Binding, ActingUser) ->
diff --git a/deps/rabbit/src/rabbit_exchange.erl b/deps/rabbit/src/rabbit_exchange.erl
@@ -92,10 +92,16 @@ serial(X) ->
         true -> rabbit_db_exchange:next_serial(X#exchange.name)
     end.
 
--spec declare
-        (name(), type(), boolean(), boolean(), boolean(),
-         rabbit_framing:amqp_table(), rabbit_types:username())
-        -> rabbit_types:exchange().
+-spec declare(Name, Type, Durable, AutoDelete, Internal, Args, Username) ->
+    Ret when
+      Name :: name(),
+      Type :: type(),
+      Durable :: boolean(),
+      AutoDelete :: boolean(),
+      Internal :: boolean(),
+      Args :: rabbit_framing:amqp_table(),
+      Username :: rabbit_types:username(),
+      Ret :: {ok, rabbit_types:exchange()} | {error, timeout}.
 
 declare(XName, Type, Durable, AutoDelete, Internal, Args, Username) ->
     X = rabbit_exchange_decorator:set(
@@ -122,14 +128,16 @@ declare(XName, Type, Durable, AutoDelete, Internal, Args, Username) ->
                     Serial = serial(Exchange),
                     ok = callback(X, create, Serial, [Exchange]),
                     rabbit_event:notify(exchange_created, info(Exchange)),
-                    Exchange;
+                    {ok, Exchange};
                 {existing, Exchange} ->
-                    Exchange
+                    {ok, Exchange};
+                {error, timeout} = Err ->
+                    Err
             end;
         _ ->
             rabbit_log:warning("ignoring exchange.declare for exchange ~tp,
                                 exchange.delete in progress~n.", [XName]),
-            X
+            {ok, X}
     end.
 
 %% Used with binaries sent over the wire; the type may not exist.
diff --git a/deps/rabbit/src/rabbit_logger_exchange_h.erl b/deps/rabbit/src/rabbit_logger_exchange_h.erl
@@ -171,16 +171,21 @@ setup_proc(
 declare_exchange(
   #{config := #{exchange := #resource{name = Name,
                                       virtual_host = VHost} = Exchange}}) ->
-    try
-        %% Durable.
-        #exchange{} = rabbit_exchange:declare(
-                        Exchange, topic, true, false, true, [],
-                        ?INTERNAL_USER),
-        ?LOG_DEBUG(
-           "Declared exchange '~ts' in vhost '~ts'",
-           [Name, VHost],
-           #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-        ok
+    try rabbit_exchange:declare(
+          Exchange, topic, true, false, true, [], ?INTERNAL_USER) of
+        {ok, #exchange{}} ->
+            ?LOG_DEBUG(
+               "Declared exchange '~ts' in vhost '~ts'",
+               [Name, VHost],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+            ok;
+        {error, timeout} ->
+            ?LOG_DEBUG(
+               "Could not declare exchange '~ts' in vhost '~ts' because the "
+               "operation timed out",
+               [Name, VHost],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+            error
     catch
         Class:Reason ->
             ?LOG_DEBUG(
diff --git a/deps/rabbit/src/rabbit_vhost.erl b/deps/rabbit/src/rabbit_vhost.erl
@@ -201,33 +201,57 @@ do_add(Name, Metadata, ActingUser) ->
             ok
     end,
     rabbit_db_vhost_defaults:apply(Name, ActingUser),
-    _ = [begin
-         Resource = rabbit_misc:r(Name, exchange, ExchangeName),
-         rabbit_log:debug("Will declare an exchange ~tp", [Resource]),
-         _ = rabbit_exchange:declare(Resource, Type, true, false, Internal, [], ActingUser)
-     end || {ExchangeName, Type, Internal} <-
-            [{<<"">>,                   direct,  false},
-             {<<"amq.direct">>,         direct,  false},
-             {<<"amq.topic">>,          topic,   false},
-             %% per 0-9-1 pdf
-             {<<"amq.match">>,          headers, false},
-             %% per 0-9-1 xml
-             {<<"amq.headers">>,        headers, false},
-             {<<"amq.fanout">>,         fanout,  false},
-             {<<"amq.rabbitmq.trace">>, topic,   true}]],
-    case rabbit_vhost_sup_sup:start_on_all_nodes(Name) of
+    case declare_default_exchanges(Name, ActingUser) of
         ok ->
-            rabbit_event:notify(vhost_created, info(VHost)
-                                ++ [{user_who_performed_action, ActingUser},
-                                    {description, Description},
-                                    {tags, Tags}]),
-            ok;
-        {error, Reason} ->
-            Msg = rabbit_misc:format("failed to set up vhost '~ts': ~tp",
-                                     [Name, Reason]),
+            case rabbit_vhost_sup_sup:start_on_all_nodes(Name) of
+                ok ->
+                    rabbit_event:notify(vhost_created, info(VHost)
+                                        ++ [{user_who_performed_action, ActingUser},
+                                            {description, Description},
+                                            {tags, Tags}]),
+                    ok;
+                {error, Reason} ->
+                    Msg = rabbit_misc:format("failed to set up vhost '~ts': ~tp",
+                                             [Name, Reason]),
+                    {error, Msg}
+            end;
+        {error, timeout} ->
+            Msg = rabbit_misc:format(
+                    "failed to set up vhost '~ts' because a timeout occurred "
+                    "while adding default exchanges",
+                    [Name]),
             {error, Msg}
     end.
 
+-spec declare_default_exchanges(VHostName, ActingUser) -> Ret when
+      VHostName :: vhost:name(),
+      ActingUser :: rabbit_types:username(),
+      Ret :: ok | {error, timeout}.
+
+declare_default_exchanges(VHostName, ActingUser) ->
+    DefaultExchanges = [{<<"">>,                   direct,  false},
+                        {<<"amq.direct">>,         direct,  false},
+                        {<<"amq.topic">>,          topic,   false},
+                        %% per 0-9-1 pdf
+                        {<<"amq.match">>,          headers, false},
+                        %% per 0-9-1 xml
+                        {<<"amq.headers">>,        headers, false},
+                        {<<"amq.fanout">>,         fanout,  false},
+                        {<<"amq.rabbitmq.trace">>, topic,   true}],
+    rabbit_misc:for_each_while_ok(
+      fun({ExchangeName, Type, Internal}) ->
+              Resource = rabbit_misc:r(VHostName, exchange, ExchangeName),
+              rabbit_log:debug("Will declare an exchange ~tp", [Resource]),
+              case rabbit_exchange:declare(
+                     Resource, Type, true, false, Internal, [],
+                     ActingUser) of
+                   {ok, _} ->
+                       ok;
+                   {error, timeout} = Err ->
+                       Err
+              end
+      end, DefaultExchanges).
+
 -spec update_metadata(vhost:name(), vhost:metadata(), rabbit_types:username()) -> rabbit_types:ok_or_error(any()).
 update_metadata(Name, Metadata0, ActingUser) ->
     Metadata = maps:with([description, tags, default_queue_type], Metadata0),
diff --git a/deps/rabbit/test/routing_SUITE.erl b/deps/rabbit/test/routing_SUITE.erl
@@ -84,9 +84,9 @@ topic(Config) ->
 
 topic1(_Config) ->
     XName = rabbit_misc:r(?VHOST, exchange, <<"topic_matching-exchange">>),
-    X = rabbit_exchange:declare(
-          XName, topic, _Durable = true, _AutoDelete = false,
-          _Internal = false, _Args = [], ?USER),
+    {ok, X} = rabbit_exchange:declare(
+                XName, topic, _Durable = true, _AutoDelete = false,
+                _Internal = false, _Args = [], ?USER),
 
     %% add some bindings
     Bindings = [#binding{source = XName,
diff --git a/deps/rabbitmq_cli/test/ctl/list_exchanges_command_test.exs b/deps/rabbitmq_cli/test/ctl/list_exchanges_command_test.exs
@@ -96,7 +96,7 @@ defmodule ListExchangesCommandTest do
 
   test "run: default options test", context do
     exchange_name = "test_exchange"
-    declare_exchange(exchange_name, @vhost)
+    {:ok, _} = declare_exchange(exchange_name, @vhost)
 
     assert MapSet.new(run_command_to_list(@command, [["name", "type"], context[:opts]])) ==
              MapSet.new(
@@ -106,8 +106,8 @@ defmodule ListExchangesCommandTest do
   end
 
   test "run: list multiple exchanges", context do
-    declare_exchange("test_exchange_1", @vhost, :direct)
-    declare_exchange("test_exchange_2", @vhost, :fanout)
+    {:ok, _} = declare_exchange("test_exchange_1", @vhost, :direct)
+    {:ok, _} = declare_exchange("test_exchange_2", @vhost, :fanout)
 
     non_default_exchanges =
       run_command_to_list(@command, [["name", "type"], context[:opts]])
@@ -124,8 +124,8 @@ defmodule ListExchangesCommandTest do
   end
 
   test "run: info keys filter single key", context do
-    declare_exchange("test_exchange_1", @vhost)
-    declare_exchange("test_exchange_2", @vhost)
+    {:ok, _} = declare_exchange("test_exchange_1", @vhost)
+    {:ok, _} = declare_exchange("test_exchange_2", @vhost)
 
     non_default_exchanges =
       run_command_to_list(@command, [["name"], context[:opts]])
@@ -138,8 +138,8 @@ defmodule ListExchangesCommandTest do
   end
 
   test "run: info keys add additional keys", context do
-    declare_exchange("durable_exchange", @vhost, :direct, true)
-    declare_exchange("auto_delete_exchange", @vhost, :fanout, false, true)
+    {:ok, _} = declare_exchange("durable_exchange", @vhost, :direct, true)
+    {:ok, _} = declare_exchange("auto_delete_exchange", @vhost, :fanout, false, true)
 
     non_default_exchanges =
       run_command_to_list(@command, [["name", "type", "durable", "auto_delete"], context[:opts]])
@@ -162,8 +162,8 @@ defmodule ListExchangesCommandTest do
       delete_vhost(other_vhost)
     end)
 
-    declare_exchange("test_exchange_1", @vhost)
-    declare_exchange("test_exchange_2", other_vhost)
+    {:ok, _} = declare_exchange("test_exchange_1", @vhost)
+    {:ok, _} = declare_exchange("test_exchange_2", other_vhost)
 
     non_default_exchanges1 =
       run_command_to_list(@command, [["name"], context[:opts]])
diff --git a/deps/rabbitmq_event_exchange/src/rabbit_exchange_type_event.erl b/deps/rabbitmq_event_exchange/src/rabbit_exchange_type_event.erl
@@ -38,9 +38,13 @@ info(_X) -> [].
 info(_X, _) -> [].
 
 register() ->
-    _ = rabbit_exchange:declare(exchange(), topic, true, false, true, [],
-                                ?INTERNAL_USER),
-    gen_event:add_handler(rabbit_event, ?MODULE, []).
+    case rabbit_exchange:declare(exchange(), topic, true, false, true, [],
+                                 ?INTERNAL_USER) of
+        {ok, _Exchange} ->
+            gen_event:add_handler(rabbit_event, ?MODULE, []);
+        {error, timeout} = Err ->
+            Err
+    end.
 
 unregister() ->
     case rabbit_exchange:ensure_deleted(exchange(), false, ?INTERNAL_USER) of
diff --git a/deps/rabbitmq_federation/test/unit_inbroker_SUITE.erl b/deps/rabbitmq_federation/test/unit_inbroker_SUITE.erl
@@ -200,10 +200,12 @@ upstream_validation(_Config) ->
     ok.
 
 with_exchanges(Fun) ->
-    rabbit_exchange:declare(r(?US_NAME), fanout, false, false, false, [],
-                            <<"acting-user">>),
-    X = rabbit_exchange:declare(r(?DS_NAME), fanout, false, false, false, [],
-                                <<"acting-user">>),
+    {ok, _} = rabbit_exchange:declare(
+                r(?US_NAME), fanout, false, false, false, [],
+                <<"acting-user">>),
+    {ok, X} = rabbit_exchange:declare(
+                r(?DS_NAME), fanout, false, false, false, [],
+                <<"acting-user">>),
     Fun(X),
     %% Delete downstream first or it will recreate the upstream
     rabbit_exchange:delete(r(?DS_NAME), false, <<"acting-user">>),
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_manager.erl b/deps/rabbitmq_stream/src/rabbit_stream_manager.erl
