Re-issue monitors to clean up stale listeners

References #4133

(cherry picked from commit 6ab1158)

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -529,20 +529,35 @@ apply(Meta, {machine_version, From = 1, To = 2}, State = #?MODULE{streams = Stre
                                                                   monitors = Monitors0}) ->
     rabbit_log:info("Stream coordinator machine version changes from ~p to ~p, updating state.",
                     [From, To]),
-    Streams1 = maps:fold(fun(S, #stream{listeners = L0} = S0, StreamAcc) ->
-                                 L1 = maps:fold(fun(ListPid, LeaderPid, LAcc) ->
-                                                        LAcc#{{ListPid, leader} => LeaderPid}
-                                                end, #{}, L0),
-                                 StreamAcc#{S => S0#stream{listeners = L1}}
-                         end, #{}, Streams0),
-    Monitors1 = maps:fold(fun(P, {StreamId, listener}, Acc) ->
+    %% conversion from old state to new state
+    %% additional operation: the stream listeners are never collected in the previous version
+    %% so we'll emit monitors for all listener PIDs
+    %% this way we'll get the DOWN event for dead listener PIDs and
+    %% we'll clean the stream listeners the in the DOWN event callback
+
+    %% transform the listeners of each stream and accumulate listener PIDs
+    {Streams1, Listeners} =
+    maps:fold(fun(S, #stream{listeners = L0} = S0, {StreamAcc, GlobalListAcc}) ->
+                      {L1, GlobalListAcc1} = maps:fold(fun(ListPid, LeaderPid, {LAcc, GLAcc}) ->
+                                             {LAcc#{{ListPid, leader} => LeaderPid},
+                                              GLAcc#{ListPid => S}}
+                                     end, {#{}, GlobalListAcc}, L0),
+                      {StreamAcc#{S => S0#stream{listeners = L1}}, GlobalListAcc1}
+              end, {#{}, #{}}, Streams0),
+    %% accumulate monitors for the map and create the effects to emit the monitors
+    {ExtraMonitors, Effects} = maps:fold(fun(P, StreamId, {MAcc, EAcc}) ->
+                                                 {MAcc#{P => {StreamId, listener}},
+                                                 [{monitor, process, P} | EAcc]}
+                          end, {#{}, []}, Listeners),
+    Monitors1 = maps:merge(Monitors0, ExtraMonitors),
+    Monitors2 = maps:fold(fun(P, {StreamId, listener}, Acc) ->
                                   Acc#{P => {#{StreamId => ok}, listener}};
                              (P, V, Acc) ->
                                   Acc#{P => V}
-                          end, #{}, Monitors0),
+                          end, #{}, Monitors1),
     return(Meta, State#?MODULE{streams = Streams1,
-                               monitors = Monitors1,
-                               listeners = undefined}, ok, []);
+                               monitors = Monitors2,
+                               listeners = undefined}, ok, Effects);
 apply(Meta, {machine_version, From, To}, State) ->
     rabbit_log:info("Stream coordinator machine version changes from ~p to ~p, no state changes required.",
                     [From, To]),

deps/rabbit/test/rabbit_stream_coordinator_SUITE.erl

@@ -200,22 +200,31 @@ listeners(_) ->
 machine_version_from_1_to_2(_) ->
     S = <<"stream">>,
     LeaderPid = spawn(fun() -> ok end),
-    ListPid = spawn(fun() -> ok end),
+    ListPid = spawn(fun() -> ok end), %% simulate a dead listener (not cleaned up)
+    DeadListPid = spawn(fun() -> ok end),
     State0 = #?STATE{streams = #{S =>
-                                 #stream{listeners = #{ListPid => LeaderPid}}},
+                                 #stream{listeners = #{ListPid => LeaderPid,
+                                                       DeadListPid => LeaderPid}}},
                      monitors = #{ListPid => {S, listener}}},
 
-    {State1, ok, []} = apply_cmd(#{index => 42}, {machine_version, 1, 2}, State0),
+    {State1, ok, Effects} = apply_cmd(#{index => 42}, {machine_version, 1, 2}, State0),
 
     Stream1 = maps:get(S, State1#?STATE.streams),
     ?assertEqual(
-       #{{ListPid, leader} => LeaderPid},
+       #{{ListPid, leader} => LeaderPid,
+         {DeadListPid, leader} => LeaderPid}, %% should be cleaned up on DOWN event
        Stream1#stream.listeners
       ),
     ?assertEqual(
-       #{ListPid => {#{S => ok}, listener}},
+       #{ListPid => {#{S => ok}, listener},
+         DeadListPid => {#{S => ok}, listener}},
        State1#?STATE.monitors
       ),
+    ?assertEqual(
+       [{monitor, process, DeadListPid}, %% will trigger an immediate DOWN event
+        {monitor, process, ListPid}],
+       Effects
+      ),
     ok.
 
 new_stream(_) ->