Merge pull request #10453 from rabbitmq/rabbitmq-server-10416

Handle old child id format in federation_exchange_link_sup_sup (take 3)

Author: michaelklishin

deps/rabbitmq_federation/src/rabbit_federation_exchange_link_sup_sup.erl

@@ -66,42 +66,87 @@ start_link() ->
 %% Note that the next supervisor down, rabbit_federation_link_sup, is common
 %% between exchanges and queues.
 start_child(X) ->
-    case mirrored_supervisor:start_child(
-           ?SUPERVISOR,
-           {id(X), {rabbit_federation_link_sup, start_link, [X]},
-            transient, ?SUPERVISOR_WAIT, supervisor,
-            [rabbit_federation_link_sup]}) of
-        {ok, _Pid}               -> ok;
-        {error, {already_started, _Pid}} ->
-          #exchange{name = ExchangeName} = X,
-          rabbit_log_federation:debug("Federation link for exchange ~tp was already started",
-                                      [rabbit_misc:rs(ExchangeName)]),
-          ok;
-        %% A link returned {stop, gone}, the link_sup shut down, that's OK.
-        {error, {shutdown, _}} -> ok
+    Result =
+        case child_exists(X) orelse
+            mirrored_supervisor:start_child(
+              ?SUPERVISOR,
+              {id(X), {rabbit_federation_link_sup, start_link, [X]},
+               transient, ?SUPERVISOR_WAIT, supervisor,
+               [rabbit_federation_link_sup]}) of
+            true ->
+                already_started;
+            {ok, _Pid} ->
+                ok;
+            {error, {already_started, _Pid}} ->
+                already_started;
+            %% A link returned {stop, gone}, the link_sup shut down, that's OK.
+            {error, {shutdown, _}} ->
+                ok
+        end,
+    case Result of
+        ok ->
+            ok;
+        already_started ->
+            #exchange{name = ExchangeName} = X,
+            rabbit_log_federation:debug("Federation link for exchange ~tp was already started",
+                                        [rabbit_misc:rs(ExchangeName)]),
+            ok
     end.
 
+
+child_exists(Name) ->
+    Id = id(Name),
+    %% older format, pre-3.13.0
+    OldId = old_id(Name),
+    lists:any(fun ({ChildId, _, _, _}) ->
+                      ChildId =:= Id orelse ChildId =:= OldId
+              end,
+              mirrored_supervisor:which_children(?SUPERVISOR)).
+
 adjust({clear_upstream, VHost, UpstreamName}) ->
-    _ = [rabbit_federation_link_sup:adjust(Pid, X, {clear_upstream, UpstreamName}) ||
-            {{_, #exchange{name = Name} = X}, Pid, _, _} <- mirrored_supervisor:which_children(?SUPERVISOR),
-            Name#resource.virtual_host == VHost],
+    _ = [rabbit_federation_link_sup:adjust(Pid, exchange_record_from_child_id(Id), {clear_upstream, UpstreamName}) ||
+            {Id, Pid, _, _} <- mirrored_supervisor:which_children(?SUPERVISOR),
+            virtual_host_name_from_child_id(Id) =:= VHost
+        ],
     ok;
 adjust(Reason) ->
-    _ = [rabbit_federation_link_sup:adjust(Pid, X, Reason) ||
-            {{_, X}, Pid, _, _} <- mirrored_supervisor:which_children(?SUPERVISOR)],
+    _ = [rabbit_federation_link_sup:adjust(Pid, exchange_record_from_child_id(Id), Reason) ||
+            {Id, Pid, _, _} <- mirrored_supervisor:which_children(?SUPERVISOR)
+        ],
     ok.
 
 stop_child(X) ->
-    case mirrored_supervisor:terminate_child(?SUPERVISOR, id(X)) of
-      ok -> ok;
-      {error, Err} ->
-        #exchange{name = ExchangeName} = X,
-        rabbit_log_federation:warning(
-          "Attempt to stop a federation link for exchange ~tp failed: ~tp",
-          [rabbit_misc:rs(ExchangeName), Err]),
-        ok
-    end,
-    ok = mirrored_supervisor:delete_child(?SUPERVISOR, id(X)).
+    Result =
+        case stop_and_delete_child(id(X)) of
+            ok -> ok;
+            {error, not_found} = Error ->
+                case rabbit_khepri:is_enabled() of
+                    true ->
+                        %% Old id format is not supported by Khepri and cannot exist there
+                        Error;
+                    false ->
+                        %% try old format, pre-3.13.0
+                        stop_and_delete_child(old_id(X))
+                end
+        end,
+    case Result of
+        ok ->
+            ok;
+        {error, Err} ->
+            #exchange{name = ExchangeName} = X,
+            rabbit_log_federation:warning(
+              "Attempt to stop a federation link for exchange ~tp failed: ~tp",
+              [rabbit_misc:rs(ExchangeName), Err]),
+            ok
+    end.
+
+stop_and_delete_child(Id) ->
+    case mirrored_supervisor:terminate_child(?SUPERVISOR, Id) of
+        ok ->
+            ok = mirrored_supervisor:delete_child(?SUPERVISOR, Id);
+        {error, not_found} = Error ->
+            Error
+    end.
 
 %%----------------------------------------------------------------------------
 
@@ -115,3 +160,20 @@ id(X = #exchange{policy = Policy}) ->
 
 simple_id(#exchange{name = #resource{virtual_host = VHost, name = Name}}) ->
     [exchange, VHost, Name].
+
+%% Old child id format, pre 3.13.0
+old_id(X = #exchange{policy = Policy}) ->
+    X1 = rabbit_exchange:immutable(X),
+    X1#exchange{policy = Policy}.
+
+%% New child id format, introduced in 3.13.0 for Khepri
+exchange_record_from_child_id({_, #exchange{} = XR}) ->
+    XR;
+%% Old child id format, pre-3.13.0
+exchange_record_from_child_id(#exchange{} = XR) ->
+    XR.
+
+virtual_host_name_from_child_id({_, #exchange{name = Res}}) ->
+    Res#resource.virtual_host;
+virtual_host_name_from_child_id(#exchange{name = Res}) ->
+    Res#resource.virtual_host.