QQ: implement snapshot_installed/4 callback

kjnilsson · kjnilsson · commit 16772f38d119 · 2025-09-16T11:05:52.000+01:00
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
@@ -43,6 +43,7 @@
          init/1,
          apply/3,
          live_indexes/1,
+         snapshot_installed/4,
          state_enter/2,
          tick/2,
          overview/1,
@@ -487,7 +488,8 @@ apply(#{index := Idx} = Meta,
 apply(#{index := Index}, #purge{},
       #?STATE{messages_total = Total,
               returns = Returns,
-              ra_indexes = Indexes0
+              ra_indexes = Indexes0,
+              msg_bytes_enqueue = MsgBytesEnqueue
              } = State0) ->
     NumReady = messages_ready(State0),
     Indexes = case Total of
@@ -514,7 +516,9 @@ apply(#{index := Index}, #purge{},
                            returns = lqueue:new(),
                            msg_bytes_enqueue = 0
                           },
-    Effects0 = [{aux, force_checkpoint}, garbage_collection],
+    Effects0 = [{aux, {bytes_out, MsgBytesEnqueue}},
+                {aux, force_checkpoint},
+                garbage_collection],
     Reply = {purge, NumReady},
     {State, _, Effects} = evaluate_limit(Index, false, State0,
                                          State1, Effects0),
@@ -681,14 +685,56 @@ apply(_Meta, Cmd, State) ->
     ?LOG_DEBUG("rabbit_fifo: unhandled command ~W", [Cmd, 10]),
     {State, ok, []}.
 
--spec live_indexes(state()) ->
-    [ra:index()].
-live_indexes(#?STATE{returns = Returns,
+-spec live_indexes(state()) -> [ra:index()].
+live_indexes(#?STATE{cfg = #cfg{},
+                     returns = Returns,
                      messages = Messages,
+                     ra_indexes = Indexes,
                      dlx = Dlx}) ->
     DlxIndexes = rabbit_fifo_dlx:live_indexes(Dlx),
     RtnIndexes = [I || ?MSG(I, _) <- lqueue:to_list(Returns)],
-    DlxIndexes ++ RtnIndexes ++ rabbit_fifo_q:indexes(Messages).
+    CheckedIdxs = rabbit_fifo_index:indexes(Indexes),
+    CheckedIdxs ++
+    DlxIndexes ++
+    RtnIndexes ++
+    rabbit_fifo_q:indexes(Messages).
+
+
+-spec snapshot_installed(Meta, State, OldMeta, OldState) ->
+    ra_machine:effects() when
+      Meta :: ra_snapshot:meta(),
+      State :: state(),
+      OldMeta :: ra_snapshot:meta(),
+      OldState :: state().
+snapshot_installed(_Meta, #?MODULE{cfg = #cfg{resource = QR},
+                                   consumers = Consumers} = State,
+                   _OldMeta, _OldState) ->
+    %% here we need to redliver all pending consumer messages
+    %% to local consumers
+    %% TODO: with some additional state (raft indexes assigned to consumer)
+    %% we could reduce the number of resends but it is questionable if this
+    %% complexity is worth the effort. rabbit_fifo_index will de-duplicate
+    %% deliveries anyway
+    SendAcc = maps:fold(
+                fun (_ConsumerKey, #consumer{cfg = #consumer_cfg{tag = Tag,
+                                                                 pid = Pid},
+                                             checked_out = Checked},
+                     Acc) ->
+                        case node(Pid) == node() of
+                            true ->
+                                Acc#{{Tag, Pid} => maps:to_list(Checked)};
+                            false ->
+                                Acc
+                        end
+                end, #{}, Consumers),
+    ?LOG_DEBUG("~ts: rabbit_fifo: install snapshot sending ~p",
+                     [rabbit_misc:rs(QR), SendAcc]),
+    Effs = add_delivery_effects([], SendAcc, State),
+    ?LOG_DEBUG("~ts: rabbit_fifo: effs ~p",
+                     [rabbit_misc:rs(QR), Effs]),
+    Effs.
+
+
 
 convert_v3_to_v4(#{} = _Meta, StateV3) ->
     %% TODO: consider emitting release cursors as checkpoints
@@ -3026,28 +3072,29 @@ priority_tag(Msg) ->
     end.
 
 
-do_checkpoints(MacVer, Ts, #checkpoint{index = _ChIdx,
-                                    timestamp = _SnapTime},
+do_checkpoints(MacVer, Ts, #checkpoint{},
                RaAux, BytesIn, BytesOut, Force) when MacVer >= 8 ->
     do_checkpoints(MacVer, Ts, #snapshot{}, RaAux, BytesIn, BytesOut, Force);
 do_checkpoints(MacVer, Ts, #snapshot{index = _ChIdx,
-                                    timestamp = SnapTime,
-                                    bytes_out = LastBytesOut} = Snap0,
-               RaAux, _BytesIn, BytesOut, _Force) when MacVer >= 8 ->
+                                     timestamp = SnapTime,
+                                     bytes_out = LastBytesOut} = Snap0,
+               RaAux, _BytesIn, BytesOut, Force) when MacVer >= 8 ->
     LastAppliedIdx = ra_aux:last_applied(RaAux),
     #?STATE{} = MacState = ra_aux:machine_state(RaAux),
     TimeSince = Ts - SnapTime,
     MsgsTot = messages_total(MacState),
-    ra_aux:overview(RaAux),
+    % ra_aux:overview(RaAux),
     % MaxBytesFactor = max(1, MsgsTot / CheckMaxIndexes),
+    % TODO: snapshots also need to be triggered by non settled commands
+    % that aren't enqueues
     EnoughDataRemoved = BytesOut - LastBytesOut > ?SNAP_OUT_BYTES,
     {CheckMinInterval, _CheckMinIndexes, _CheckMaxIndexes} =
         persistent_term:get(quorum_queue_checkpoint_config,
                             {?CHECK_MIN_INTERVAL_MS, ?CHECK_MIN_INDEXES,
                              ?CHECK_MAX_INDEXES}),
     EnoughTimeHasPassed = TimeSince > CheckMinInterval,
-    case (EnoughTimeHasPassed andalso
-          EnoughDataRemoved) of
+    case (EnoughTimeHasPassed andalso EnoughDataRemoved) orelse
+         Force of
         true ->
             {#snapshot{index = LastAppliedIdx,
                        timestamp = Ts,
diff --git a/deps/rabbit/src/rabbit_fifo_client.erl b/deps/rabbit/src/rabbit_fifo_client.erl
@@ -687,6 +687,8 @@ handle_ra_event(QName, Leader, {applied, Seqs},
     end;
 handle_ra_event(QName, From, {machine, Del}, State0)
       when element(1, Del) == delivery ->
+    ?LOG_DEBUG("~ts: handle delivery , ",
+               [rabbit_misc:rs(QName)]),
     handle_delivery(QName, From, Del, State0);
 handle_ra_event(_QName, _From, {machine, Action}, State)
   when element(1, Action) =:= credit_reply orelse
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -1998,7 +1998,7 @@ make_ra_conf(Q, ServerId, Membership, MacVersion)
                  Membership, MacVersion).
 
 make_ra_conf(Q, ServerId, TickTimeout,
-             SnapshotInterval, CheckpointInterval,
+             _SnapshotInterval, CheckpointInterval,
              Membership, MacVersion) ->
     QName = amqqueue:get_name(Q),
     #resource{name = QNameBin} = QName,
@@ -2009,7 +2009,6 @@ make_ra_conf(Q, ServerId, TickTimeout,
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
                min_snapshot_interval => 0,
-               snapshot_interval => SnapshotInterval,
                min_checkpoint_interval => CheckpointInterval,
                max_checkpoints => 3},
     rabbit_misc:maps_put_truthy(membership, Membership,
diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl
@@ -105,7 +105,8 @@ groups() ->
                                             force_checkpoint,
                                             policy_repair,
                                             gh_12635,
-                                            replica_states
+                                            replica_states,
+                                            consumer_message_is_delevered_after_snapshot
                                            ]
                        ++ all_tests()},
                       {cluster_size_5, [], [start_queue,
@@ -1461,19 +1462,19 @@ force_checkpoint_on_queue(Config) ->
     rabbit_ct_helpers:await_condition(
       fun() ->
           {ok, State, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           LCI =:= undefined
       end),
     rabbit_ct_helpers:await_condition(
       fun() ->
           {ok, State, _} = rpc:call(Server1, ra, member_overview, [{RaName, Server1}]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           LCI =:= undefined
       end),
     rabbit_ct_helpers:await_condition(
       fun() ->
           {ok, State, _} = rpc:call(Server2, ra, member_overview, [{RaName, Server2}]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           LCI =:= undefined
       end),
 
@@ -1490,21 +1491,21 @@ force_checkpoint_on_queue(Config) ->
       fun() ->
           {ok, State, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
           ct:pal("Ra server state post forced checkpoint: ~tp~n", [State]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           (LCI =/= undefined) andalso (LCI >= N)
       end),
     rabbit_ct_helpers:await_condition(
       fun() ->
           {ok, State, _} = rpc:call(Server1, ra, member_overview, [{RaName, Server1}]),
           ct:pal("Ra server state post forced checkpoint: ~tp~n", [State]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           (LCI =/= undefined) andalso (LCI >= N)
       end),
     rabbit_ct_helpers:await_condition(
       fun() ->
           {ok, State, _} = rpc:call(Server2, ra, member_overview, [{RaName, Server2}]),
           ct:pal("Ra server state post forced checkpoint: ~tp~n", [State]),
-          #{log := #{latest_checkpoint_index := LCI}} = State,
+          #{log := #{snapshot_index := LCI}} = State,
           (LCI =/= undefined) andalso (LCI >= N)
       end).
 
@@ -1780,6 +1781,60 @@ dont_leak_file_handles(Config) ->
     rabbit_ct_client_helpers:close_channel(C),
     ok.
 
+consumer_message_is_delevered_after_snapshot(Config) ->
+    %% a consumer on a node that received a snapshot should have it's messages
+    %% delivered
+    [Server0, _Server1, Server2] =
+        rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, set_env,
+                                      [rabbit, quorum_min_checkpoint_interval, 1]),
+
+    Ch0 = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    #'confirm.select_ok'{} = amqp_channel:call(Ch0, #'confirm.select'{}),
+    QQ = ?config(queue_name, Config),
+    RaName = ra_name(QQ),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch0, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    %% stop server on a follower node
+    ok = rpc:call(Server2, ra, stop_server, [quorum_queues, {RaName, Server2}]),
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server2),
+    %% create a consumer
+    qos(Ch2, 2, false),
+    subscribe(Ch2, QQ, false),
+
+    %% publish some messages and make sure a snapshot has been taken
+    Msg = crypto:strong_rand_bytes(13_000),
+
+    [publish(Ch0, QQ, Msg) || _ <- lists:seq(1, 5000)],
+    amqp_channel:wait_for_confirms(Ch0, 5),
+    %% need to sleep here a bit as QQs wont take
+    %% snapshots more often than once every second
+    timer:sleep(1100),
+
+    %% then purge
+    #'queue.purge_ok'{} = amqp_channel:call(Ch0, #'queue.purge'{queue = QQ}),
+
+    rabbit_ct_helpers:await_condition_ignoring_exceptions(
+      fun () ->
+              {ok, #{log := Log}, _} = rpc:call(Server0, ra, member_overview,
+                                                [{RaName, Server0}]),
+              undefined =/= maps:get(snapshot_index, Log)
+      end),
+    %% restart stopped member
+    ok = rpc:call(Server2, ra, restart_server, [quorum_queues, {RaName, Server2}]),
+
+    %% messages should be delivered
+    receive
+        {#'basic.deliver'{delivery_tag = _DeliveryTag}, _} ->
+            ok
+    after 30000 ->
+              flush(1),
+              ct:fail("expected messages were not delivered")
+    end,
+    ok.
+
 gh_12635(Config) ->
     check_quorum_queues_v4_compat(Config),
 
@@ -1788,7 +1843,7 @@ gh_12635(Config) ->
         rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
     ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, set_env,
-                                      [rabbit, quorum_min_checkpoint_interval, 1]),
+                                      [rabbit, quorum_snapshot_interval, 1]),
 
     Ch0 = rabbit_ct_client_helpers:open_channel(Config, Server0),
     #'confirm.select_ok'{} = amqp_channel:call(Ch0, #'confirm.select'{}),
@@ -1810,7 +1865,7 @@ gh_12635(Config) ->
     rabbit_ct_helpers:await_condition(
       fun () ->
               {ok, #{log := Log}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
-              undefined =/= maps:get(latest_checkpoint_index, Log)
+              undefined =/= maps:get(snapshot_index, Log)
       end),
 
     %% publish 1 more message
diff --git a/deps/rabbit/test/rabbit_fifo_q_SUITE.erl b/deps/rabbit/test/rabbit_fifo_q_SUITE.erl
@@ -77,6 +77,8 @@ basics(_Config) ->
                      {no, ?MSG(4)},
                      {hi, ?MSG(5)}
                     ]),
+
+    ?assertEqual([1,2,3,4,5], lists:sort(rabbit_fifo_q:indexes(Q1))),
     {?MSG(1), Q2} = rabbit_fifo_q:out(Q1),
     {?MSG(3), Q3} = rabbit_fifo_q:out(Q2),
     {?MSG(2), Q4} = rabbit_fifo_q:out(Q3),
