Make empty CQ init faster in case of clean shutdown

In case of a clean shutdown the value next_seq_id is stored in
recovery terms. This value can be utilized by the queue index to
provide better bounds in absence of segment files.

Before this patch starting an empty classic queue with next_seq_id =
100_000_000 used to take about 26 seconds. With this patch it takes
less than 1ms.

Author: gomoripeti

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -20,7 +20,8 @@
 %% queue implementation itself.
 -export([pre_publish/7, flush_pre_publish_cache/2,
          sync/1, needs_sync/1, flush/1,
-         bounds/1, next_segment_boundary/1]).
+         bounds/1, bounds/2,
+         next_segment_boundary/1]).
 
 %% Used to upgrade/downgrade from/to the v1 index.
 -export([init_for_conversion/3]).
@@ -1191,11 +1192,18 @@ flush_pre_publish_cache(TargetRamCount, State) ->
 -spec bounds(State) ->
                        {non_neg_integer(), non_neg_integer(), State}
                        when State::state().
+bounds(State) ->
+    bounds(State, undefined).
 
-bounds(State = #qi{ segments = Segments }) ->
+-spec bounds(State, non_neg_integer() | undefiend) ->
+                       {non_neg_integer(), non_neg_integer(), State}
+                       when State::state().
+bounds(State = #qi{ segments = Segments }, NextSeqIdHint) ->
     ?DEBUG("~0p", [State]),
     %% We must special case when we are empty to make tests happy.
     if
+        Segments =:= #{} andalso is_integer(NextSeqIdHint) ->
+            {NextSeqIdHint, NextSeqIdHint, State};
         Segments =:= #{} ->
             {0, 0, State};
         true ->

deps/rabbit/src/rabbit_variable_queue.erl

@@ -1172,7 +1172,13 @@ expand_delta(_SeqId, #delta { count       = Count,
 
 init(IsDurable, IndexState, StoreState, DeltaCount, DeltaBytes, Terms,
      PersistentClient, TransientClient, VHost) ->
-    {LowSeqId, HiSeqId, IndexState1} = rabbit_classic_queue_index_v2:bounds(IndexState),
+    NextSeqIdHint =
+        case Terms of
+            non_clean_shutdown -> undefined;
+            _ -> proplists:get_value(next_seq_id, Terms)
+        end,
+
+    {LowSeqId, HiSeqId, IndexState1} = rabbit_classic_queue_index_v2:bounds(IndexState, NextSeqIdHint),
 
     {NextSeqId, NextDeliverSeqId, DeltaCount1, DeltaBytes1} =
         case Terms of