Handle database timeouts when declaring queues

This fixes a case-clause crash in the logs in `cluster_minority_SUITE`.
When the database is not available `rabbit_amqqueue:declare/6,7` should
return a `protocol_error` record with an error message rather than a
hard crash. Also included in this change is the necessary changes to
typespecs: `rabbit_db_queue:create_or_get/1` is the first function to
return a possible `{error,timeout}`. That bubbles up through
`rabbit_amqqueue:internal_declare/3` and must be handled in each
`rabbit_queue_type:declare/2` callback.

Author: the-mikedavis

deps/rabbit/src/rabbit_amqqueue.erl

@@ -251,8 +251,12 @@ get_queue_type(Args, DefaultQueueType) ->
             rabbit_queue_type:discover(V)
     end.
 
--spec internal_declare(amqqueue:amqqueue(), boolean()) ->
-    {created | existing, amqqueue:amqqueue()} | queue_absent().
+-spec internal_declare(Queue, Recover) -> Ret when
+      Queue :: amqqueue:amqqueue(),
+      Recover :: boolean(),
+      Ret :: {created | existing, amqqueue:amqqueue()} |
+             queue_absent() |
+             rabbit_khepri:timeout_error().
 
 internal_declare(Q, Recover) ->
     do_internal_declare(Q, Recover).

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -226,6 +226,12 @@ init_it2(Recover, From, State = #q{q                   = Q,
                 false ->
                     {stop, normal, {existing, Q1}, State}
             end;
+        {error, timeout} ->
+            Reason = {protocol_error, internal_error,
+                      "Could not declare ~ts on node '~ts' because the "
+                      "metadata store operation timed out",
+                      [rabbit_misc:rs(amqqueue:get_name(Q)), node()]},
+            {stop, normal, Reason, State};
         Err ->
             {stop, normal, Err, State}
     end.

deps/rabbit/src/rabbit_db_queue.erl

@@ -875,7 +875,10 @@ get_all_by_type_and_node_in_khepri(VHostName, Type, Node) ->
 
 -spec create_or_get(Queue) -> Ret when
       Queue :: amqqueue:amqqueue(),
-      Ret :: {created, Queue} | {existing, Queue} | {absent, Queue, nodedown}.
+      Ret :: {created, Queue} |
+             {existing, Queue} |
+             {absent, Queue, nodedown} |
+             rabbit_khepri:timeout_error().
 %% @doc Writes a queue record if it doesn't exist already or returns the existing one
 %%
 %% @returns the existing record if there is one in the database already, or the newly

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -289,7 +289,12 @@ start_cluster(Q) ->
                     declare_queue_error(Error, NewQ, LeaderNode, ActingUser)
             end;
         {existing, _} = Ex ->
-            Ex
+            Ex;
+        {error, timeout} ->
+            {protocol_error, internal_error,
+             "Could not declare quorum ~ts on node '~ts' because the metadata "
+             "store operation timed out",
+             [rabbit_misc:rs(QName), node()]}
     end.
 
 declare_queue_error(Error, Queue, Leader, ActingUser) ->

deps/rabbit/src/rabbit_stream_queue.erl

@@ -195,7 +195,12 @@ create_stream(Q0) ->
         {existing, Q} ->
             {existing, Q};
         {absent, Q, Reason} ->
-            {absent, Q, Reason}
+            {absent, Q, Reason};
+        {error, timeout} ->
+            {protocol_error, internal_error,
+             "Could not declare ~ts on node '~ts' because the metadata store "
+             "operation timed out",
+             [rabbit_misc:rs(QName), node()]}
     end.
 
 -spec delete(amqqueue:amqqueue(), boolean(),

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -70,8 +70,10 @@ is_stateful() ->
 
 -spec declare(amqqueue:amqqueue(), node()) ->
     {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()} |
-    {'absent', amqqueue:amqqueue(), rabbit_amqqueue:absent_reason()}.
+    {'absent', amqqueue:amqqueue(), rabbit_amqqueue:absent_reason()} |
+    {protocol_error, internal_error, string(), [string()]}.
 declare(Q0, _Node) ->
+    QName = amqqueue:get_name(Q0),
     Q1 = case amqqueue:get_pid(Q0) of
              none ->
                  %% declaring process becomes the queue
@@ -86,14 +88,19 @@ declare(Q0, _Node) ->
             Opts = amqqueue:get_options(Q),
             ActingUser = maps:get(user, Opts, ?UNKNOWN_USER),
             rabbit_event:notify(queue_created,
-                                [{name, amqqueue:get_name(Q)},
+                                [{name, QName},
                                  {durable, true},
                                  {auto_delete, false},
                                  {exclusive, true},
                                  {type, amqqueue:get_type(Q)},
                                  {arguments, amqqueue:get_arguments(Q)},
                                  {user_who_performed_action, ActingUser}]),
             {new, Q};
+        {error, timeout} ->
+            {protocol_error, internal_error,
+             "Could not declare ~ts because the metadata store operation "
+             "timed out",
+             [rabbit_misc:rs(QName)]};
         Other ->
             Other
     end.