QQ: emit release cursors after consumer cancel

If this is not done apps that consume/cancel from empty queues in a loop
will grow the raft log in an unbounded manner. This could also be the
case for the garbage_collect command.

(cherry picked from commit eaa216d)
(cherry picked from commit a787a92)

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -340,10 +340,13 @@ apply(#{index := Index,
                     {State, Reply, Effects}
             end
     end;
-apply(Meta, #checkout{spec = cancel, consumer_id = ConsumerId}, State0) ->
-    {State, Effects} = cancel_consumer(Meta, ConsumerId, State0, [],
-                                       consumer_cancel),
-    checkout(Meta, State0, State, Effects);
+apply(#{index := Idx} = Meta,
+      #checkout{spec = cancel,
+                consumer_id = ConsumerId}, State0) ->
+    {State1, Effects1} = cancel_consumer(Meta, ConsumerId, State0, [],
+                                         consumer_cancel),
+    {State, Reply, Effects} = checkout(Meta, State0, State1, Effects1),
+    update_smallest_raft_index(Idx, Reply, State, Effects);
 apply(Meta, #checkout{spec = Spec, meta = ConsumerMeta,
                       consumer_id = {_, Pid} = ConsumerId},
       State0) ->
@@ -372,8 +375,8 @@ apply(#{index := Index}, #purge{},
     {State, _, Effects} = evaluate_limit(Index, false, State0,
                                          State1, Effects0),
     update_smallest_raft_index(Index, Reply, State, Effects);
-apply(_Meta, #garbage_collection{}, State) ->
-    {State, ok, [{aux, garbage_collection}]};
+apply(#{index := Idx}, #garbage_collection{}, State) ->
+    update_smallest_raft_index(Idx, ok, State, [{aux, garbage_collection}]);
 apply(#{system_time := Ts} = Meta, {down, Pid, noconnection},
       #?MODULE{consumers = Cons0,
                cfg = #cfg{consumer_strategy = single_active},

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -397,9 +397,11 @@ cancelled_checkout_empty_queue_test(_) ->
     {State1, _} = check_auto(Cid, 2, test_init(test)),
     % cancelled checkout should clear out service_queue also, else we'd get a
     % build up of these
-    {State2, _, _} = apply(meta(3), rabbit_fifo:make_checkout(Cid, cancel, #{}), State1),
+    {State2, _, Effects} = apply(meta(3), rabbit_fifo:make_checkout(Cid, cancel, #{}), State1),
     ?assertEqual(0, map_size(State2#rabbit_fifo.consumers)),
     ?assertEqual(0, priority_queue:len(State2#rabbit_fifo.service_queue)),
+    ct:pal("Effs: ~p", [Effects]),
+    ?ASSERT_EFF({release_cursor, _, _}, Effects),
     ok.
 
 cancelled_checkout_out_test(_) ->