Merge pull request #10059 from rabbitmq/cache-oldest-timestamp

QQ: Cache oldest_entry_timestamp

Author: michaelklishin

deps/rabbit/src/rabbit_fifo.erl

@@ -983,7 +983,7 @@ which_module(3) -> ?MODULE.
                capacity :: term(),
                gc = #aux_gc{} :: #aux_gc{},
                tick_pid,
-               unused2}).
+               cache = #{} :: map()}).
 
 init_aux(Name) when is_atom(Name) ->
     %% TODO: catch specific exception throw if table already exists
@@ -1102,21 +1102,31 @@ handle_aux(_RaState, cast, tick, #?AUX{name = Name,
 handle_aux(_RaState, cast, eol, #?AUX{name = Name} = Aux, Log, _) ->
     ets:delete(rabbit_fifo_usage, Name),
     {no_reply, Aux, Log};
-handle_aux(_RaState, {call, _From}, oldest_entry_timestamp, Aux,
+handle_aux(_RaState, {call, _From}, oldest_entry_timestamp,
+           #?AUX{cache = Cache} = Aux0,
            Log0, #?MODULE{} = State) ->
-    {Ts, Log} = case smallest_raft_index(State) of
-                    %% if there are no entries, we return current timestamp
-                    %% so that any previously obtained entries are considered
-                    %% older than this
-                    undefined ->
-                        {erlang:system_time(millisecond), Log0};
-                    Idx when is_integer(Idx) ->
-                        %% TODO: make more defensive to avoid potential crash
-                        {{_, _, {_, Meta, _, _}}, Log1} = ra_log:fetch(Idx, Log0),
-                        #{ts := Timestamp} = Meta,
-                        {Timestamp, Log1}
-                end,
-    {reply, {ok, Ts}, Aux, Log};
+    {CachedIdx, CachedTs} = maps:get(oldest_entry, Cache, {undefined, undefined}),
+    case smallest_raft_index(State) of
+        %% if there are no entries, we return current timestamp
+        %% so that any previously obtained entries are considered
+        %% older than this
+        undefined ->
+            Aux1 = Aux0#?AUX{cache = maps:remove(oldest_entry, Cache)},
+            {reply, {ok, erlang:system_time(millisecond)}, Aux1, Log0};
+        CachedIdx ->
+            %% cache hit
+            {reply, {ok, CachedTs}, Aux0, Log0};
+        Idx when is_integer(Idx) ->
+            case ra_log:fetch(Idx, Log0) of
+                {{_, _, {_, #{ts := Timestamp}, _, _}}, Log1} ->
+                    Aux1 = Aux0#?AUX{cache = Cache#{oldest_entry =>
+                                                    {Idx, Timestamp}}},
+                    {reply, {ok, Timestamp}, Aux1, Log1};
+                {undefined, Log1} ->
+                    %% fetch failed
+                    {reply, {error, failed_to_get_timestamp}, Aux0, Log1}
+            end
+    end;
 handle_aux(_RaState, {call, _From}, {peek, Pos}, Aux0,
            Log0, MacState) ->
     case rabbit_fifo:query_peek(Pos, MacState) of

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -156,6 +156,7 @@ all_tests() ->
      delete_if_unused,
      queue_ttl,
      peek,
+     oldest_entry_timestamp,
      peek_with_wrong_queue_type,
      message_ttl,
      message_ttl_policy,
@@ -2729,6 +2730,45 @@ peek(Config) ->
     wait_for_messages(Config, [[QQ, <<"2">>, <<"2">>, <<"0">>]]),
     ok.
 
+oldest_entry_timestamp(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-max-in-memory-length">>, long, 2}])),
+
+    Msg1 = <<"msg1">>,
+    VHost = <<"%2F">>,
+    ServerId = binary_to_atom(<<VHost/binary, "_", QQ/binary>>, utf8),
+
+    ?assertMatch({ok, Ts} when is_integer(Ts),
+                 rabbit_ct_broker_helpers:rpc(Config, 0, ra,
+                                              aux_command,
+                                              [ServerId, oldest_entry_timestamp])),
+    publish(Ch, QQ, Msg1),
+    wait_for_messages(Config, [[QQ, <<"1">>, <<"1">>, <<"0">>]]),
+
+    ?assertMatch({ok, Ts} when is_integer(Ts),
+                 rabbit_ct_broker_helpers:rpc(Config, 0, ra,
+                                              aux_command,
+                                              [ServerId, oldest_entry_timestamp])),
+    ?assertMatch({ok, Ts} when is_integer(Ts),
+                 rabbit_ct_broker_helpers:rpc(Config, 0, ra,
+                                              aux_command,
+                                              [ServerId, oldest_entry_timestamp])),
+
+    {'queue.purge_ok', 1} = amqp_channel:call(Ch, #'queue.purge'{queue = QQ}),
+    Now = erlang:system_time(millisecond),
+    timer:sleep(100),
+    ?assertMatch({ok, Ts2} when Ts2 > Now,
+                 rabbit_ct_broker_helpers:rpc(Config, 0, ra,
+                                              aux_command,
+                                              [ServerId, oldest_entry_timestamp])),
+
+    ok.
+
 -define(STATUS_MATCH(N, T),
         [{<<"Node Name">>, N},
          {<<"Raft State">>, _},