Merge pull request #11837 from rabbitmq/mergify/bp/v3.13.x/pr-11833

rabbit_khepri: Retry register_projections during boot (backport #11741) (backport #11833)

Author: michaelklishin

deps/rabbit/src/rabbit_khepri.erl

@@ -263,7 +263,7 @@ setup(_) ->
     case khepri:start(?RA_SYSTEM, RaServerConfig) of
         {ok, ?STORE_ID} ->
             wait_for_leader(),
-            register_projections(),
+            wait_for_register_projections(),
             ?LOG_DEBUG(
                "Khepri-based " ?RA_FRIENDLY_NAME " ready",
                #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
@@ -306,6 +306,21 @@ wait_for_leader(Timeout, Retries) ->
             throw(Reason)
     end.
 
+wait_for_register_projections() ->
+    wait_for_register_projections(retry_timeout(), retry_limit()).
+
+wait_for_register_projections(_Timeout, 0) ->
+    exit(timeout_waiting_for_khepri_projections);
+wait_for_register_projections(Timeout, Retries) ->
+    rabbit_log:info("Waiting for Khepri projections for ~tp ms, ~tp retries left",
+                    [Timeout, Retries - 1]),
+    try
+        register_projections()
+    catch
+        throw : timeout ->
+            wait_for_register_projections(Timeout, Retries -1)
+    end.
+
 %% @private
 
 can_join_cluster(DiscoveryNode) when is_atom(DiscoveryNode) ->

deps/rabbitmq_cli/lib/rabbitmq/cli/default_output.ex

@@ -82,6 +82,10 @@ defmodule RabbitMQ.CLI.DefaultOutput do
     {:error, RabbitMQ.CLI.Core.ExitCodes.exit_tempfail(), khepri_timeout_error(node_name)}
   end
 
+  defp format_khepri_output({:error, :timeout_waiting_for_khepri_projections}, %{node: node_name}) do
+    {:error, RabbitMQ.CLI.Core.ExitCodes.exit_tempfail(), khepri_timeout_error(node_name)}
+  end
+
   defp format_khepri_output(result, _opts) do
     result
   end