Fix dlx checkout command redirection (#16203)

This commit supersedes #15548.

 ## What?

Fix the following genuine CI flake:
```
make -C deps/rabbit ct-rabbit_fifo_dlx_integration t=cluster_size_3:single_dlx_worker
```

Sometimes this test case failed with the logs showing the following:
```text
2026-02-24 09:06:19.413770+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': vote granted for term 3 votes 2
2026-02-24 09:06:19.414048+00:00 [debug] <0.2377.0> started rabbit_fifo_dlx_worker <0.2600.0> for queue 'single_dlx_worker_source' in vhost '/'
2026-02-24 09:06:19.414096+00:00 [notice] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': candidate -> leader in term: 3 machine version: 7, last applied 5
2026-02-24 09:06:19.414388+00:00 [debug] <0.2602.0> queue 'single_dlx_worker_source' in vhost '/': updating leader record to current node rmq-ct-cluster_size_3-1-28000@localhost
2026-02-24 09:06:19.414279+00:00 [info] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': leader saw request_vote_rpc from {'%2F_single�[118;1:3u_dlx_worker_source','rmq-ct-cluster_size_3-3-28144@localhost'} for term 4 abdicates term: 3!
2026-02-24 09:06:19.417479+00:00 [notice] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': leader -> follower in term: 4 machine version: 7, last applied 5
2026-02-24 09:06:19.417533+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': is not new, setting election timeout.
2026-02-24 09:06:19.417740+00:00 [info] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': declining vote for {'%2F_single_dlx_worker_source','rmq-ct-cluster_size_3-3-28144@localhost'} for term 4, candidate last log {index, term} was: {5,2}  last log entry {index, term} is: {{6,3}}
2026-02-24 09:06:19.417824+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': leader call - leader not known. Command will be forwarded once leader is known.
2026-02-24 09:06:19.418190+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/' declining pre-vote to {'%2F_single_dlx_worker_source','rmq-ct-cluster_size_3-2-28072@localhost'} for term 3, current term 4
2026-02-24 09:06:19.428043+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': resetting last index to 5 from 6 in term 4
2026-02-24 09:06:19.428157+00:00 [info] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': detected a new leader {'%2F_single_dlx_worker_source','rmq-ct-cluster_size_3-3-28144@localhost'} in term 4
2026-02-24 09:06:19.428280+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': mem table overwriting detected whilst staging entries, opening new mem table
2026-02-24 09:06:19.436299+00:00 [debug] <0.2377.0> queue 'single_dlx_worker_source' in vhost '/': enabling ra cluster changes in 4, index 6
2026-02-24 09:06:19.436411+00:00 [debug] <0.2377.0> Terminating <31028.1894.0> since <31122.2516.0> becomes active rabbit_fifo_dlx_worker
2026-02-24 09:06:19.437003+00:00 [debug] <0.2377.0> Terminating <31122.2516.0> since <0.2600.0> becomes active rabbit_fifo_dlx_worker
2026-02-24 09:06:19.437107+00:00 [warning] <0.2600.0> Failed to process command {dlx,{checkout,<0.2600.0>,2}} on quorum queue leader {'%2F_single_dlx_worker_source',
2026-02-24 09:06:19.437107+00:00 [warning] <0.2600.0>                                                                                 'rmq-ct-cluster_size_3-1-28000@localhost'} because actual leader is {'%2F_single_dlx_worker_source',
2026-02-24 09:06:19.437107+00:00 [warning] <0.2600.0>                                                                                                                                                      'rmq-ct-cluster_size_3-3-28144@localhost'}.
```

 ## How?

This commit supersedes #15548.

In this commit, we use `ra:pipeline_command/4` with a selective receive instead of
`ra:process_command/3` for the dlx checkout command. This prevents Ra from
automatically redirecting the checkout command to a new leader if a failover
happens while the command is being processed.

Author: ansd

deps/rabbit/src/rabbit_fifo_dlx_client.erl

@@ -41,23 +41,38 @@ checkout(QResource, Leader, NumUnsettled) ->
     State = #state{queue_resource = QResource,
                    leader = Leader,
                    last_msg_id = -1},
-    process_command(Cmd, State, 5).
+    checkout0(Cmd, State, 5).
 
-process_command(_Cmd, _State, 0) ->
+checkout0(_Cmd, _State, 0) ->
     {error, ra_command_failed};
-process_command(Cmd, #state{leader = Leader} = State, Tries) ->
-    case ra:process_command(Leader, Cmd, 60_000) of
-        {ok, ok, Leader} ->
-            {ok, State#state{leader = Leader}};
-        {ok, ok, NonLocalLeader} ->
-            ?LOG_WARNING("Failed to process command ~tp on queue leader ~tp because actual leader is ~tp.",
-                               [Cmd, Leader, NonLocalLeader]),
-            {error, non_local_leader};
-        Err ->
-            ?LOG_WARNING("Failed to process command ~tp on queue leader ~tp: ~tp~n"
-                               "Trying ~b more time(s)...",
-                               [Cmd, Leader, Err, Tries]),
-            process_command(Cmd, State, Tries - 1)
+checkout0(Cmd, #state{leader = Leader} = State, Tries) ->
+    Correlation = make_ref(),
+    %% We use ra:pipeline_command/4 instead of ra:process_command/3 because the
+    %% latter internally redirects to the new leader which we don't want.
+    ra:pipeline_command(Leader, Cmd, Correlation, normal),
+    receive_applied(Cmd, Correlation, State, Tries).
+
+receive_applied(Cmd, Corr, #state{queue_resource = QName,
+                                  leader = Leader} = State, Tries) ->
+    receive
+        {'$gen_cast', {queue_event, QName, {Leader, {applied, Results}}}} ->
+            case lists:member({Corr, ok}, Results) of
+                true ->
+                    {ok, State};
+                false ->
+                    receive_applied(Cmd, Corr, State, Tries)
+            end;
+        {'$gen_cast', {queue_event, QName,
+                       {_From, {rejected, {not_leader, NonLocalLeader, Corr}}}}} ->
+            ?LOG_WARNING("failed to apply command ~tp on leader ~tp "
+                         "because actual leader is ~tp",
+                         [Cmd, Leader, NonLocalLeader]),
+            {error, non_local_leader}
+    after 60_000 ->
+              ?LOG_WARNING("timed out applying command ~tp on leader ~tp; "
+                           "trying ~b more time(s)...",
+                           [Cmd, Leader, Tries - 1]),
+              checkout0(Cmd, State, Tries - 1)
     end.
 
 -spec handle_ra_event(pid(), term(), state()) ->

deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl

@@ -38,35 +38,37 @@
 
 all() ->
     [
-     {group, single_node},
+     {group, cluster_size_1},
      {group, cluster_size_3}
     ].
 
 groups() ->
     [
-     {single_node, [shuffle], [
-                               expired,
-                               rejected,
-                               delivery_limit,
-                               target_queue_not_bound,
-                               target_queue_deleted,
-                               dlx_missing,
-                               cycle,
-                               stats,
-                               drop_head_falls_back_to_at_most_once,
-                               switch_strategy,
-                               reject_publish_source_queue_max_length,
-                               reject_publish_source_queue_max_length_bytes,
-                               reject_publish_target_classic_queue,
-                               reject_publish_max_length_target_quorum_queue,
-                               target_quorum_queue_delete_create
-                              ]},
-     {cluster_size_3, [], [
-                           reject_publish_max_length_target_quorum_queue,
-                           reject_publish_down_target_quorum_queue,
-                           many_target_queues,
-                           single_dlx_worker
-                          ]}
+     {cluster_size_1, [shuffle],
+      [
+       expired,
+       rejected,
+       delivery_limit,
+       target_queue_not_bound,
+       target_queue_deleted,
+       dlx_missing,
+       cycle,
+       stats,
+       drop_head_falls_back_to_at_most_once,
+       switch_strategy,
+       reject_publish_source_queue_max_length,
+       reject_publish_source_queue_max_length_bytes,
+       reject_publish_target_classic_queue,
+       reject_publish_max_length_target_quorum_queue,
+       target_quorum_queue_delete_create
+      ]},
+     {cluster_size_3, [],
+      [
+       reject_publish_max_length_target_quorum_queue,
+       reject_publish_down_target_quorum_queue,
+       many_target_queues,
+       single_dlx_worker
+      ]}
     ].
 
 init_per_suite(Config0) ->
@@ -86,7 +88,7 @@ init_per_suite(Config0) ->
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config).
 
-init_per_group(single_node = Group, Config) ->
+init_per_group(cluster_size_1 = Group, Config) ->
     init_per_group(Group, Config, 1);
 init_per_group(cluster_size_3 = Group, Config) ->
     init_per_group(Group, Config, 3).
@@ -954,15 +956,17 @@ single_dlx_worker(Config) ->
     ok = rabbit_ct_broker_helpers:kill_node(Config, Leader0),
     {ok, _, {_, Leader1}} = ?awaitMatch({ok, _, _},
                                         ra:members({RaName, Follower0}),
-                                        30000),
+                                        30_000),
     ?assertNotEqual(Leader0, Leader1),
     [Follower1] = [Server1, Follower0] -- [Leader1],
     assert_active_dlx_workers(0, Config, Follower1),
     assert_active_dlx_workers(1, Config, Leader1),
     ok = rabbit_ct_broker_helpers:start_node(Config, Leader0).
 
 assert_active_dlx_workers(N, Config, Server) ->
-    ?awaitMatch(N, length(rpc(Config, Server, supervisor, which_children, [rabbit_fifo_dlx_sup], 2000)), 60000).
+    ?awaitMatch(N,
+                length(rpc(Config, Server, supervisor, which_children, [rabbit_fifo_dlx_sup], 5000)),
+                60_000).
 
 declare_queue(Channel, Queue, Args) ->
     #'queue.declare_ok'{} = amqp_channel:call(Channel, #'queue.declare'{