Make channel_operation_timeout_test_queue more resilient

Now the CQ state can be changed without breaking this test
all the time. Only the #msg_status{} record remains a potential
problem but thankfully it does not change often.

Author: lhoguin

deps/rabbit/src/rabbit_variable_queue.erl

@@ -22,7 +22,7 @@
 -export([start/2, stop/1]).
 
 %% exported for testing only
--export([start_msg_store/3, stop_msg_store/1]).
+-export([start_msg_store/3, stop_msg_store/1, ram_pending_acks/1]).
 
 -include("mc.hrl").
 -include_lib("stdlib/include/qlc.hrl").
@@ -1584,6 +1584,9 @@ prepare_to_store(Msg) ->
 %% Internal gubbins for acks
 %%----------------------------------------------------------------------------
 
+ram_pending_acks(#vqstate{ ram_pending_ack = RPA }) ->
+    RPA.
+
 record_pending_ack(#msg_status { seq_id = SeqId } = MsgStatus,
                    State = #vqstate { ram_pending_ack  = RPA,
                                       disk_pending_ack = DPA,

deps/rabbit/test/channel_operation_timeout_test_queue.erl

@@ -5,7 +5,6 @@
 %% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
-%% @todo This module also needs to be updated when variable queue changes.
 -module(channel_operation_timeout_test_queue).
 
 -export([init/3, terminate/2, delete_and_terminate/2, delete_crashed/1,
@@ -28,69 +27,11 @@
 %% the test message has been published, and is awaiting acknowledgement in the
 %% queue index. Test message is "timeout_test_msg!".
 %%
+%% Only access the #msg_status{} record needs to be updated if it changes.
 %%----------------------------------------------------------------------------
 
 -behaviour(rabbit_backing_queue).
 
--record(vqstate,
-        { q_head,
-          q_tail,
-          next_seq_id,
-          %% seq_id() of first undelivered message
-          %% everything before this seq_id() was delivered at least once
-          next_deliver_seq_id,
-          ram_pending_ack,    %% msgs still in RAM
-          disk_pending_ack,   %% msgs in store, paged out
-          index_state,
-          store_state,
-          msg_store_clients,
-          durable,
-          transient_threshold,
-          qi_embed_msgs_below,
-
-          bytes,              %% w/o unacked
-          unacked_bytes,
-          persistent_count,   %% w   unacked
-          persistent_bytes,   %% w   unacked
-
-          ram_msg_count,      %% w/o unacked
-          ram_msg_count_prev,
-          ram_ack_count_prev,
-          ram_bytes,          %% w   unacked
-          out_counter,
-          in_counter,
-          rates,
-          %% There are two confirms paths: either store/index produce confirms
-          %% separately (v2 with per-vhost message store) or the confirms
-          %% are produced all at once while syncing/flushing (v2 with per-queue
-          %% message store). The latter is more efficient as it avoids many
-          %% sets operations.
-          msgs_on_disk,
-          msg_indices_on_disk,
-          unconfirmed,
-          confirmed,
-          ack_out_counter,
-          ack_in_counter,
-          %% Unlike the other counters these two do not feed into
-          %% #rates{} and get reset
-          disk_read_count,
-          disk_write_count,
-
-          %% Fast path for confirms handling. Instead of having
-          %% index/store keep track of confirms separately and
-          %% doing intersect/subtract/union we just put the messages
-          %% here and on sync move them to 'confirmed'.
-          %%
-          %% Note: This field used to be 'memory_reduction_run_count'.
-          unconfirmed_simple,
-          %% Queue data is grouped by VHost. We need to store it
-          %% to work with queue index.
-          virtual_host,
-          waiting_bump = false
-        }).
-
--record(rates, { in, out, ack_in, ack_out, timestamp }).
-
 -record(msg_status,
         { seq_id,
           msg_id,
@@ -103,79 +44,9 @@
           msg_props
         }).
 
--record(q_tail,
-        { start_seq_id, %% start_seq_id is inclusive
-          count,
-          end_seq_id    %% end_seq_id is exclusive
-        }).
-
-
 -include_lib("rabbit_common/include/rabbit.hrl").
--define(QUEUE, lqueue).
 -define(TIMEOUT_TEST_MSG, <<"timeout_test_msg!">>).
 
-%%----------------------------------------------------------------------------
-
--type seq_id()  :: non_neg_integer().
-
--type rates() :: #rates { in        :: float(),
-                          out       :: float(),
-                          ack_in    :: float(),
-                          ack_out   :: float(),
-                          timestamp :: rabbit_types:timestamp()}.
-
--type q_tail() :: #q_tail { start_seq_id :: non_neg_integer(),
-                            count        :: non_neg_integer(),
-                            end_seq_id   :: non_neg_integer() }.
-
-%% The compiler (rightfully) complains that ack() and state() are
-%% unused. For this reason we duplicate a -spec from
-%% rabbit_backing_queue with the only intent being to remove
-%% warnings. The problem here is that we can't parameterise the BQ
-%% behaviour by these two types as we would like to. We still leave
-%% these here for documentation purposes.
--type ack() :: seq_id().
--type state() :: #vqstate {
-             q_head                :: ?QUEUE:?QUEUE(),
-             q_tail                :: q_tail(),
-             next_seq_id           :: seq_id(),
-             next_deliver_seq_id   :: seq_id(),
-             ram_pending_ack       :: map(),
-             disk_pending_ack      :: map(),
-             index_state           :: any(),
-             store_state           :: any(),
-             msg_store_clients     :: 'undefined' | {{any(), binary()},
-                                                    {any(), binary()}},
-             durable               :: boolean(),
-             transient_threshold   :: non_neg_integer(),
-             qi_embed_msgs_below   :: non_neg_integer(),
-
-             bytes                 :: non_neg_integer(),
-             unacked_bytes         :: non_neg_integer(),
-             persistent_count      :: non_neg_integer(),
-             persistent_bytes      :: non_neg_integer(),
-
-             ram_msg_count         :: non_neg_integer(),
-             ram_msg_count_prev    :: non_neg_integer(),
-             ram_ack_count_prev    :: non_neg_integer(),
-             ram_bytes             :: non_neg_integer(),
-             out_counter           :: non_neg_integer(),
-             in_counter            :: non_neg_integer(),
-             rates                 :: rates(),
-             msgs_on_disk          :: gb_sets:set(),
-             msg_indices_on_disk   :: gb_sets:set(),
-             unconfirmed           :: gb_sets:set(),
-             confirmed             :: gb_sets:set(),
-             ack_out_counter       :: non_neg_integer(),
-             ack_in_counter        :: non_neg_integer(),
-             disk_read_count       :: non_neg_integer(),
-             disk_write_count      :: non_neg_integer(),
-
-             unconfirmed_simple    :: sets:set()}.
-
-%% Duplicated from rabbit_backing_queue
--spec ack([ack()], state()) -> {[rabbit_guid:guid()], state()}.
-
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
@@ -198,12 +69,8 @@ delete_and_terminate(Reason, State) ->
 delete_crashed(Q) ->
     rabbit_variable_queue:delete_crashed(Q).
 
-purge(State = #vqstate { ram_pending_ack= QPA }) ->
-    maybe_delay(QPA),
-    rabbit_variable_queue:purge(State);
-%% For v4.2.x because the state has changed.
 purge(State) ->
-    QPA = element(9, State),
+    QPA = ram_pending_acks(State),
     maybe_delay(QPA),
     rabbit_variable_queue:purge(State).
 
@@ -236,24 +103,16 @@ drop(AckRequired, State) ->
 ack(List, State) ->
     rabbit_variable_queue:ack(List, State).
 
-requeue(AckTags, #vqstate { ram_pending_ack = QPA } = State) ->
-    maybe_delay(QPA),
-    rabbit_variable_queue:requeue(AckTags, State);
-%% For v4.2.x because the state has changed.
 requeue(AckTags, State) ->
-    QPA = element(9, State),
+    QPA = ram_pending_acks(State),
     maybe_delay(QPA),
     rabbit_variable_queue:requeue(AckTags, State).
 
 ackfold(MsgFun, Acc, State, AckTags) ->
     rabbit_variable_queue:ackfold(MsgFun, Acc, State, AckTags).
 
-len(#vqstate { ram_pending_ack = QPA } = State) ->
-    maybe_delay(QPA),
-    rabbit_variable_queue:len(State);
-%% For v4.2.x because the state has changed.
 len(State) ->
-    QPA = element(9, State),
+    QPA = ram_pending_acks(State),
     maybe_delay(QPA),
     rabbit_variable_queue:len(State).
 
@@ -298,6 +157,13 @@ set_queue_mode(_, State) ->
 set_queue_version(_, State) ->
     State.
 
+ram_pending_acks(State) ->
+    case erlang:function_exported(rabbit_variable_queue, ram_pending_acks, 1) of
+        true -> rabbit_variable_queue:ram_pending_acks(State);
+        %% For v4.2.x because the state has changed.
+        false -> element(9, State)
+    end.
+
 %% Delay
 maybe_delay(QPA) ->
   %% The structure for ram_pending_acks has changed to maps in 3.12.