QQ: use new log_ext effect

Author: kjnilsson

deps/rabbit/src/rabbit_channel.erl

@@ -564,8 +564,10 @@ prioritise_cast(Msg, _Len, _State) ->
     case Msg of
         {confirm,            _MsgSeqNos, _QPid} -> 5;
         {reject_publish,     _MsgSeqNos, _QPid} -> 5;
-        {queue_event, _, {confirm,            _MsgSeqNos, _QPid}} -> 5;
+        % {queue_event, _, {confirm,            _MsgSeqNos, _QPid}} -> 5;
         {queue_event, _, {reject_publish,     _MsgSeqNos, _QPid}} -> 5;
+        {method, #'basic.ack'{}, _Content, _Flow} -> 5;
+        % {queue_event, _, {delivery, _, _}} -> 0;
         _                                       -> 0
     end.
 

deps/rabbit/src/rabbit_fifo.erl

@@ -89,7 +89,10 @@
          make_purge/0,
          make_purge_nodes/1,
          make_update_config/1,
-         make_garbage_collection/0
+         make_garbage_collection/0,
+
+         exec_read/3
+
         ]).
 
 -ifdef(TEST).
@@ -2076,30 +2079,27 @@ delivery_effect(ConsumerKey, [{MsgId, ?MSG(Idx,  Header)}],
     {CTag, CPid} = consumer_id(ConsumerKey, State),
     {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, RawMsg}}]},
      ?DELIVERY_SEND_MSG_OPTS};
-delivery_effect(ConsumerKey, Msgs,
-                #?STATE{cfg = #cfg{resource = QR}} = State) ->
+delivery_effect(ConsumerKey, Msgs, #?STATE{} = State) ->
     {CTag, CPid} = consumer_id(ConsumerKey, State),
-    {RaftIdxs, Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
-                                          {[I | Acc], N+1}
-                                  end, {[], 0}, Msgs),
-    {log, RaftIdxs,
-     fun (Commands)
-           when length(Commands) < Num ->
-             %% the mandatory length/1 guard is a bit :(
-             rabbit_log:info("~ts: requested read consumer tag '~ts' of ~b "
-                             "indexes ~w but only ~b were returned. "
-                             "This is most likely a stale read request "
-                             "and can be ignored",
-                             [rabbit_misc:rs(QR), CTag, Num, RaftIdxs,
-                              length(Commands)]),
-             [];
-         (Commands) ->
-             DelMsgs = lists:zipwith(
-                         fun (Cmd, {MsgId, ?MSG(_Idx,  Header)}) ->
-                                 {MsgId, {Header, get_msg(Cmd)}}
-                         end, Commands, Msgs),
-             [{send_msg, CPid, {delivery, CTag, DelMsgs},
-               ?DELIVERY_SEND_MSG_OPTS}]
+    {RaftIdxs, _Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
+                                           {[I | Acc], N+1}
+                                   end, {[], 0}, Msgs),
+    {log_ext, RaftIdxs,
+     fun (ReadPlan) ->
+             case node(CPid) == node() of
+                 true ->
+                     [{send_msg, CPid, {delivery, CTag, ReadPlan, Msgs},
+                       ?DELIVERY_SEND_MSG_OPTS}];
+                 false ->
+                     %% if we got there we need to read the data on this node
+                     %% and send it to the consumer pid as it isn't availble
+                     %% locally
+                     {DelMsgs, Flru} = exec_read(undefined, ReadPlan, Msgs),
+                     %% we need to evict all cached items here
+                     _ = ra_flru:evict_all(Flru),
+                     [{send_msg, CPid, {delivery, CTag, DelMsgs},
+                       ?DELIVERY_SEND_MSG_OPTS}]
+             end
      end,
      {local, node(CPid)}}.
 
@@ -3014,3 +3014,11 @@ incr_msg(Msg0, DelFailed, Anns) ->
         false ->
             Msg2
     end.
+
+exec_read(Flru0, ReadPlan, Msgs) ->
+    {Entries, Flru} = ra_log_read_plan:execute(ReadPlan, Flru0),
+    %% return a list in original order
+    {lists:map(fun ({MsgId, ?MSG(Idx,  Header)}) ->
+                       Cmd = maps:get(Idx, Entries),
+                       {MsgId, {Header, get_msg(Cmd)}}
+               end, Msgs), Flru}.

deps/rabbit/src/rabbit_fifo_client.erl

@@ -69,7 +69,8 @@
                 pending = #{} :: #{seq() =>
                                    {term(), rabbit_fifo:command()}},
                 consumers = #{} :: #{rabbit_types:ctag() => #consumer{}},
-                timer_state :: term()
+                timer_state :: term(),
+                cached_segments :: undefined | ra_flru:state()
                }).
 
 -opaque state() :: #state{}.
@@ -167,7 +168,7 @@ enqueue(QName, Correlation, Msg,
 %% @param QueueName Name of the queue.
 %% @param Msg an arbitrary erlang term representing the message.
 %% @param State the current {@module} state.
-%% @returns
+%% @return's
 %% `{ok, State, Actions}' if the command was successfully sent.
 %% {@module} assigns a sequence number to every raft command it issues. The
 %% SequenceNumber can be correlated to the applied sequence numbers returned
@@ -633,7 +634,8 @@ handle_ra_event(QName, From, {applied, Seqs},
         _ ->
             {ok, State1, Actions}
     end;
-handle_ra_event(QName, From, {machine, {delivery, _ConsumerTag, _} = Del}, State0) ->
+handle_ra_event(QName, From, {machine, Del}, State0)
+      when element(1, Del) == delivery ->
     handle_delivery(QName, From, Del, State0);
 handle_ra_event(_QName, _From, {machine, Action}, State)
   when element(1, Action) =:= credit_reply orelse
@@ -835,7 +837,12 @@ handle_delivery(_QName, _Leader, {delivery, Tag, [_ | _] = IdMsgs},
     %% we should return all messages.
     MsgIntIds = [Id || {Id, _} <- IdMsgs],
     {State1, Deliveries} = return(Tag, MsgIntIds, State0),
-    {ok, State1, Deliveries}.
+    {ok, State1, Deliveries};
+handle_delivery(QName, Leader, {delivery, Tag, ReadState, Msgs},
+                #state{cached_segments = Cached0} = State) ->
+    {MsgIds, Cached} = rabbit_fifo:exec_read(Cached0, ReadState, Msgs),
+    handle_delivery(QName, Leader, {delivery, Tag, MsgIds},
+                    State#state{cached_segments = Cached}).
 
 transform_msgs(QName, QRef, Msgs) ->
     lists:map(

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -154,6 +154,8 @@ discard(Msgs, Reason, undefined, State) ->
               [Reason, rabbit_quorum_queue, disabled, length(Msgs)]}]};
 discard(Msgs0, Reason, {at_most_once, {Mod, Fun, Args}}, State) ->
     Idxs = [I || ?MSG(I, _) <- Msgs0],
+    %% TODO: this could be turned into a log_ext effect instead to avoid
+    %% reading from disk inside the qq process
     Effect = {log, Idxs,
               fun (Log) ->
                       Lookup = maps:from_list(lists:zip(Idxs, Log)),

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -107,7 +107,10 @@ groups() ->
                                             quorum_cluster_size_7,
                                             node_removal_is_not_quorum_critical,
                                             select_nodes_with_least_replicas,
-                                            select_nodes_with_least_replicas_node_down
+                                            select_nodes_with_least_replicas_node_down,
+                                            subscribe_from_each
+
+
                                            ]},
                       {clustered_with_partitions, [],
                        [
@@ -187,7 +190,8 @@ all_tests() ->
      priority_queue_fifo,
      priority_queue_2_1_ratio,
      requeue_multiple_true,
-     requeue_multiple_false
+     requeue_multiple_false,
+     subscribe_from_each
     ].
 
 memory_tests() ->
@@ -1462,6 +1466,43 @@ policy_repair(Config) ->
             consume_all(Ch, QQ)
     end.
 
+subscribe_from_each(Config) ->
+
+    [Server0 | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    #'confirm.select_ok'{} = amqp_channel:call(Ch, #'confirm.select'{}),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    [begin
+         publish_confirm(Ch, QQ)
+     end || _ <- Servers],
+    timer:sleep(100),
+    %% roll the wal to force consumer messages to be read from disk
+    [begin
+         ok = rpc:call(S, ra_log_wal, force_roll_over, [ra_log_wal])
+     end || S <- Servers],
+
+    [begin
+         ct:pal("NODE ~p", [S]),
+         C = rabbit_ct_client_helpers:open_channel(Config, S),
+         qos(C, 1, false),
+         subscribe(C, QQ, false),
+         receive
+             {#'basic.deliver'{delivery_tag = DeliveryTag}, _} ->
+                 amqp_channel:call(C, #'basic.ack'{delivery_tag = DeliveryTag})
+         after 5000 ->
+                   flush(1),
+                   ct:fail("basic.deliver timeout")
+         end,
+         timer:sleep(256),
+         rabbit_ct_client_helpers:close_channel(C),
+         flush(1)
+
+     end || S <- Servers],
+
+    ok.
 
 gh_12635(Config) ->
     % https://github.com/rabbitmq/rabbitmq-server/issues/12635
@@ -3633,6 +3674,7 @@ receive_and_ack(Ch) ->
             amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag,
                                                multiple = false})
     after 5000 ->
+              flush(1),
               ct:fail("receive_and_ack timed out", [])
     end.