Handle database failures when adding/removing bindings

This ensures that the call graph of `rabbit_db_binding:create/2` and
`rabbit_db_binding:delete/2` handle the `{error, timeout}` error
possible when Khepri is in a minority.

(cherry picked from commit f1be7ba)
(cherry picked from commit f4d2452)

Author: the-mikedavis

deps/rabbit/src/rabbit_binding.erl

@@ -41,7 +41,8 @@
 -type bind_ok_or_error() :: 'ok' | bind_errors() |
                             rabbit_types:error({'binding_invalid', string(), [any()]}) |
                             %% inner_fun() result
-                            rabbit_types:error(rabbit_types:amqp_error()).
+                            rabbit_types:error(rabbit_types:amqp_error()) |
+                            rabbit_khepri:timeout_error().
 -type bind_res() :: bind_ok_or_error() | rabbit_misc:thunk(bind_ok_or_error()).
 -type inner_fun() ::
         fun((rabbit_types:exchange(),

deps/rabbit/src/rabbit_channel.erl

@@ -1912,6 +1912,9 @@ binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
             rabbit_misc:protocol_error(precondition_failed, Fmt, Args);
         {error, #amqp_error{} = Error} ->
             rabbit_misc:protocol_error(Error);
+        {error, timeout} ->
+            rabbit_misc:protocol_error(
+              internal_error, "Could not ~s binding due to timeout", [Action]);
         ok ->
             ok
     end.

deps/rabbit/src/rabbit_db_binding.erl

@@ -150,8 +150,9 @@ not_found({[], []}, SrcName, DstName) ->
       Binding :: rabbit_types:binding(),
       Src :: rabbit_types:binding_source(),
       Dst :: rabbit_types:binding_destination(),
-      ChecksFun :: fun((Src, Dst) -> ok | {error, Reason :: any()}),
-      Ret :: ok | {error, Reason :: any()}.
+      ChecksFun :: fun((Src, Dst) -> ok | {error, ChecksErrReason}),
+      ChecksErrReason :: any(),
+      Ret :: ok | {error, ChecksErrReason} | rabbit_khepri:timeout_error().
 %% @doc Writes a binding if it doesn't exist already and passes the validation in
 %% `ChecksFun' i.e. exclusive access
 %%
@@ -255,8 +256,12 @@ serial_in_khepri(true, X) ->
       Binding :: rabbit_types:binding(),
       Src :: rabbit_types:binding_source(),
       Dst :: rabbit_types:binding_destination(),
-      ChecksFun :: fun((Src, Dst) -> ok | {error, Reason :: any()}),
-      Ret :: ok | {ok, rabbit_binding:deletions()} | {error, Reason :: any()}.
+      ChecksFun :: fun((Src, Dst) -> ok | {error, ChecksErrReason}),
+      ChecksErrReason :: any(),
+      Ret :: ok |
+             {ok, rabbit_binding:deletions()} |
+             {error, ChecksErrReason} |
+             rabbit_khepri:timeout_error().
 %% @doc Deletes a binding record from the database if it passes the validation in
 %% `ChecksFun'. It also triggers the deletion of auto-delete exchanges if needed.
 %%

deps/rabbit/src/rabbit_definitions.erl

@@ -886,12 +886,17 @@ add_binding(VHost, Binding, ActingUser) ->
                     rv(VHost, DestType, destination, Binding), ActingUser).
 
 add_binding_int(Binding, Source, Destination, ActingUser) ->
-    rabbit_binding:add(
-      #binding{source       = Source,
-               destination  = Destination,
-               key          = maps:get(routing_key, Binding, undefined),
-               args         = args(maps:get(arguments, Binding, undefined))},
-      ActingUser).
+    case rabbit_binding:add(
+           #binding{source       = Source,
+                    destination  = Destination,
+                    key          = maps:get(routing_key, Binding, undefined),
+                    args         = args(maps:get(arguments, Binding, undefined))},
+           ActingUser) of
+        ok ->
+            ok;
+        {error, _} = Err ->
+            throw(Err)
+    end.
 
 dest_type(Binding) ->
     rabbit_data_coercion:to_atom(maps:get(destination_type, Binding, undefined)).

deps/rabbit/test/cluster_minority_SUITE.erl

@@ -26,6 +26,7 @@ groups() ->
                               open_channel,
                               declare_exchange,
                               declare_binding,
+                              delete_binding,
                               declare_queue,
                               publish_to_exchange,
                               publish_and_consume_to_local_classic_queue,
@@ -86,7 +87,7 @@ init_per_group(Group, Config0) when Group == client_operations;
         {skip, _} ->
             Config1;
         _ ->
-            %% Before partitioning the cluster, create a policy and queue that can be used in
+            %% Before partitioning the cluster, create resources that can be used in
             %% the test cases. They're needed for delete and consume operations, which can list
             %% them but fail to operate anything else.
             %%
@@ -96,6 +97,10 @@ init_per_group(Group, Config0) when Group == client_operations;
             %% To be used in consume_from_queue
             #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = <<"test-queue">>,
                                                                            arguments = [{<<"x-queue-type">>, longstr, <<"classic">>}]}),
+            %% To be used in delete_binding
+            #'exchange.bind_ok'{} = amqp_channel:call(Ch, #'exchange.bind'{destination = <<"amq.fanout">>,
+                                                                           source = <<"amq.direct">>,
+                                                                           routing_key = <<"binding-to-be-deleted">>}),
 
             %% Lower the default Khepri command timeout. By default this is set
             %% to 30s in `rabbit_khepri:setup/1' which makes the cases in this
@@ -161,6 +166,14 @@ declare_binding(Config) ->
                                                        source = <<"amq.direct">>,
                                                        routing_key = <<"key">>})).
 
+delete_binding(Config) ->
+    [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
+    ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 541, _}}}, _},
+                amqp_channel:call(Ch, #'exchange.unbind'{destination = <<"amq.fanout">>,
+                                                         source = <<"amq.direct">>,
+                                                         routing_key = <<"binding-to-be-deleted">>})).
+
 declare_queue(Config) ->
     [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),

deps/rabbitmq_stream/src/rabbit_stream_manager.erl

@@ -879,6 +879,8 @@ add_super_stream_binding(VirtualHost,
             {error, {binding_invalid, rabbit_misc:format(Fmt, Args)}};
         {error, #amqp_error{} = Error} ->
             {error, {internal_error, rabbit_misc:format("~tp", [Error])}};
+        {error, timeout} ->
+            {error, {internal_error, "failed to add binding due to a timeout"}};
         ok ->
             ok
     end.