QQ: annotate messages with the raft index they were written at

kjnilsson · kjnilsson · commit 810e6db1f37c · 2026-07-06T14:27:33.000+01:00
Add an "x-opt-index" message annotation to every message read from a
quorum queue - via basic.get, normal delivery, or dead-lettering -
containing the raft log index the message currently occupies. This
lets a client identify exactly which point in the log a message came
from, e.g. to correlate it with the queue's own operator-facing raft
index.
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
@@ -1430,7 +1430,8 @@ handle_aux(_, _, {get_checked_out, ConsumerKey, MsgIds}, Aux0, RaAux0) ->
                           %% crashed and the message got removed
                           case ra_aux:log_fetch(Idx, S0) of
                               {{_Term, _Meta, Cmd}, S} ->
-                                  {S, [{MsgId, {Header, get_msg_from_cmd(Cmd)}} | Acc]};
+                                  RawMsg = annotate_index(Idx, get_msg_from_cmd(Cmd)),
+                                  {S, [{MsgId, {Header, RawMsg}} | Acc]};
                               {undefined, S} ->
                                   {S, Acc}
                           end
@@ -2331,25 +2332,36 @@ get_header(Key, Header)
 annotate_msg(Header, Msg0) ->
     case mc:is(Msg0) of
         true when is_map(Header) ->
-            Msg1 = maps:fold(fun (K, V, Acc) ->
-                                     mc:set_annotation(K, V, Acc)
-                             end, Msg0, maps:get(anns, Header, #{})),
-            Msg = case Header of
+            Msg1 = maps:fold(fun mc:set_annotation/3, Msg0,
+                             maps:get(anns, Header, #{})),
+            Msg2 = case Header of
                       #{acquired_count := AcqCount} ->
                           mc:set_annotation(acquired_count, AcqCount, Msg1);
                       _ ->
                           Msg1
                   end,
             case Header of
                 #{delivery_count := DelCount} ->
-                    mc:set_annotation(delivery_count, DelCount, Msg);
+                    mc:set_annotation(delivery_count, DelCount, Msg2);
                 _ ->
-                    Msg
+                    Msg2
             end;
         _ ->
             Msg0
     end.
 
+%% Stamps the raft index a message currently occupies onto the message
+%% itself. Applied server-side (never over the wire) so that clients -
+%% including ones running an older release that predates this annotation -
+%% don't need to understand any new wire format to benefit from it.
+annotate_index(Idx, Msg0) ->
+    case mc:is(Msg0) of
+        true ->
+            mc:set_annotation(<<"x-opt-index">>, Idx, Msg0);
+        false ->
+            Msg0
+    end.
+
 return_one(#{system_time := Ts} = Meta, MsgId,
            ?C_MSG(Msg0), DeliveryFailed, Anns,
            #?STATE{returns = Returns,
@@ -2798,19 +2810,18 @@ peek_next_msg(#?STATE{returns = Returns0,
 delivery_effect(ConsumerKey, [{MsgId, ?MSG(Idx,  Header)}],
                 #?STATE{msg_cache = {Idx, RawMsg}} = State) ->
     {CTag, CPid} = consumer_id(ConsumerKey, State),
-    {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, RawMsg}}]},
+    {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, annotate_index(Idx, RawMsg)}}]},
      ?DELIVERY_SEND_MSG_OPTS};
 delivery_effect(ConsumerKey, [{MsgId, Msg}],
                 #?STATE{msg_cache = {Idx, RawMsg}} = State)
   when is_integer(Msg) andalso ?PACKED_IDX(Msg) == Idx ->
     Header = get_msg_header(Msg),
     {CTag, CPid} = consumer_id(ConsumerKey, State),
-    {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, RawMsg}}]},
+    {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, annotate_index(Idx, RawMsg)}}]},
      ?DELIVERY_SEND_MSG_OPTS};
 delivery_effect(ConsumerKey, Msgs, #?STATE{} = State) ->
     {CTag, CPid} = consumer_id(ConsumerKey, State),
     {RaftIdxs, _Num} = lists:foldr(fun ({_, Msg}, {Acc, N}) ->
-
                                            {[get_msg_idx(Msg) | Acc], N+1}
                                    end, {[], 0}, Msgs),
     {log_ext, RaftIdxs,
@@ -2837,9 +2848,10 @@ reply_log_effect(RaftIdx, MsgId, Header, Ready, From) ->
      fun ([]) ->
              [];
          ([Cmd]) ->
+             RawMsg = annotate_index(RaftIdx, get_msg_from_cmd(Cmd)),
              [{reply, From,
                {wrap_reply,
-                {dequeue, {MsgId, {Header, get_msg_from_cmd(Cmd)}}, Ready}}}]
+                {dequeue, {MsgId, {Header, RawMsg}}, Ready}}}]
      end}.
 
 checkout_one(#{system_time := Ts} = Meta, ExpiredMsg0, InitState0, Effects0) ->
@@ -3931,7 +3943,8 @@ exec_read(Flru0, ReadPlan, Msgs) ->
                                Idx = get_msg_idx(Msg),
                                Header = get_msg_header(Msg),
                                Cmd = maps:get(Idx, Entries),
-                               {MsgId, {Header, get_msg_from_cmd(Cmd)}}
+                               RawMsg = annotate_index(Idx, get_msg_from_cmd(Cmd)),
+                               {MsgId, {Header, RawMsg}}
                        end, Msgs), Flru}
     catch exit:{missing_key, _}
             when Flru0 =/= undefined ->
@@ -4209,7 +4222,8 @@ discard_or_dead_letter(Msgs0, Reason, {at_most_once, {Mod, Fun, Args}}, State) -
                                   Cmd = maps:get(Idx, Lookup),
                                   %% ensure header delivery count
                                   %% is copied to the message container
-                                  annotate_msg(Hdr, rabbit_fifo:get_msg_from_cmd(Cmd))
+                                  annotate_index(Idx,
+                                                 annotate_msg(Hdr, rabbit_fifo:get_msg_from_cmd(Cmd)))
                               end || Msg <- Msgs0],
                       [{mod_call, Mod, Fun, Args ++ [Reason, Msgs]}]
               end},
@@ -4274,14 +4288,15 @@ dlx_delivery_effects(_CPid, []) ->
     [];
 dlx_delivery_effects(CPid, Msgs0) ->
     Msgs1 = lists:reverse(Msgs0),
-    {RaftIdxs, RsnIds} = lists:unzip(Msgs1),
+    {RaftIdxs, _RsnIds} = lists:unzip(Msgs1),
     [{log, RaftIdxs,
       fun(Log) ->
               Msgs = lists:zipwith(
-                       fun (Cmd, {Reason, H, MsgId}) ->
+                       fun (Cmd, {Idx, {Reason, H, MsgId}}) ->
                                {MsgId, {Reason,
-                                        annotate_msg(H, rabbit_fifo:get_msg_from_cmd(Cmd))}}
-                       end, Log, RsnIds),
+                                        annotate_index(Idx,
+                                                       annotate_msg(H, rabbit_fifo:get_msg_from_cmd(Cmd)))}}
+                       end, Log, Msgs1),
               [{send_msg, CPid, {dlx_event, self(), {dlx_delivery, Msgs}}, [cast]}]
       end}].
 
diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl
@@ -94,6 +94,7 @@ groups() ->
        sync_get_settled_classic_queue,
        sync_get_settled_quorum_queue,
        sync_get_settled_stream,
+       x_opt_index_quorum_queue,
        timed_get_classic_queue,
        timed_get_quorum_queue,
        timed_get_stream,
@@ -2775,6 +2776,31 @@ sync_get_settled(QType, Config) ->
 timed_get_classic_queue(Config) ->
     timed_get(<<"classic">>, Config).
 
+%% Every message carries the "x-opt-index" message annotation, the raft
+%% index the message was written at.
+x_opt_index_quorum_queue(Config) ->
+    QName = atom_to_binary(?FUNCTION_NAME),
+    {_Conn0, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config),
+    #'queue.declare_ok'{} = amqp_channel:call(
+                              Ch, #'queue.declare'{
+                                     queue = QName,
+                                     durable = true,
+                                     arguments = [{<<"x-queue-type">>, longstr, <<"quorum">>}]}),
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+    Address = rabbitmq_amqp_address:queue(QName),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"test-sender">>, Address),
+    ok = wait_for_credit(Sender),
+    ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag1">>, <<"m1">>, true)),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(
+                       Session, <<"test-receiver">>, Address, unsettled),
+    {ok, Msg} = amqp10_client:get_msg(Receiver),
+    ?assertMatch(#{<<"x-opt-index">> := V} when is_integer(V),
+                 amqp10_msg:message_annotations(Msg)),
+    ok = amqp10_client:settle_msg(Receiver, Msg, accepted),
+    ok = amqp10_client:close_connection(Connection).
+
 timed_get_quorum_queue(Config) ->
     timed_get(<<"quorum">>, Config).
 
diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl
@@ -91,6 +91,7 @@ groups() ->
                                             consume_in_minority,
                                             get_in_minority,
                                             reject_after_leader_transfer,
+                                            x_opt_index_annotation,
                                             delete_members,
                                             rebalance,
                                             node_removal_is_not_quorum_critical,
@@ -2494,6 +2495,74 @@ reject_after_leader_transfer(Config) ->
                                                 requeue = true}),
     ok.
 
+%% Every message carries the "x-opt-index" annotation, the raft index the
+%% message was written at. Check that it is present for basic.get, for
+%% deliveries to a consumer on a node with a member, and for deliveries to a
+%% consumer on a node without a member (i.e. read remotely via log_ext).
+x_opt_index_annotation(Config) ->
+    check_quorum_queues_v9_compat(Config),
+
+    [Server0, Server1, Server2] =
+        rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    QQ = ?config(queue_name, Config),
+    RaName = ra_name(QQ),
+    IndexHeader = <<"x-opt-index">>,
+
+    Ch0 = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch0, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    ?awaitMatch(3, count_online_nodes(Server0, <<"/">>, QQ), ?DEFAULT_AWAIT),
+
+    %% basic.get
+    publish(Ch0, QQ),
+    wait_for_messages_ready([Server0, Server1, Server2], RaName, 1),
+    {#'basic.get_ok'{delivery_tag = GetTag},
+     #amqp_msg{props = #'P_basic'{headers = GetHeaders}}} =
+        basic_get(Ch0, QQ, false, 10),
+    ?assertMatch({IndexHeader, long, _},
+                 rabbit_basic:header(IndexHeader, GetHeaders)),
+    ok = amqp_channel:call(Ch0, #'basic.ack'{delivery_tag = GetTag}),
+
+    %% delivery to a consumer on a node that has a member
+    publish(Ch0, QQ),
+    wait_for_messages_ready([Server0, Server1, Server2], RaName, 1),
+    Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    subscribe(Ch1, QQ, false),
+    receive
+        {#'basic.deliver'{delivery_tag = LocalTag},
+         #amqp_msg{props = #'P_basic'{headers = LocalHeaders}}} ->
+            ?assertMatch({IndexHeader, long, _},
+                         rabbit_basic:header(IndexHeader, LocalHeaders)),
+            amqp_channel:cast(Ch1, #'basic.ack'{delivery_tag = LocalTag})
+    after ?TIMEOUT ->
+              flush(10),
+              exit(basic_deliver_timeout)
+    end,
+    amqp_channel:close(Ch1),
+
+    %% remove Server2's member so it no longer holds a copy of the raft log
+    ?assertEqual(ok,
+                 rpc:call(Server0, rabbit_queue_type_ra, delete_member,
+                          [<<"/">>, QQ, Server2])),
+    ?awaitMatch(2, count_online_nodes(Server0, <<"/">>, QQ), ?DEFAULT_AWAIT),
+
+    %% delivery to a consumer on a node that does not have a member
+    publish(Ch0, QQ),
+    wait_for_messages_ready([Server0, Server1], RaName, 1),
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server2),
+    subscribe(Ch2, QQ, false),
+    receive
+        {#'basic.deliver'{delivery_tag = RemoteTag},
+         #amqp_msg{props = #'P_basic'{headers = RemoteHeaders}}} ->
+            ?assertMatch({IndexHeader, long, _},
+                         rabbit_basic:header(IndexHeader, RemoteHeaders)),
+            amqp_channel:cast(Ch2, #'basic.ack'{delivery_tag = RemoteTag})
+    after ?TIMEOUT ->
+              flush(10),
+              exit(basic_deliver_timeout)
+    end,
+    ok.
+
 delete_members(Config) ->
     [Server0, Server1, Server2] =
         rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -5436,8 +5505,10 @@ amqpl_headers(Config) ->
      #amqp_msg{props = #'P_basic'{headers = Headers2Received}}
     } = amqp_channel:call(Ch, #'basic.get'{queue = QQ}),
 
-    ?assertEqual(Headers1Sent, Headers1Received),
-    ?assertEqual(Headers2Sent, Headers2Received),
+    %% every message carries the raft index it was written at as the
+    %% "x-opt-index" header, even when the client didn't set any headers
+    ?assertMatch([{<<"x-opt-index">>, long, _}], Headers1Received),
+    ?assertMatch([{<<"x-opt-index">>, long, _}], Headers2Received),
 
     ok = amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag,
                                             multiple = true}).
@@ -6631,9 +6702,15 @@ basic_get(Ch, Q, NoAck, Attempt) ->
     end.
 
 check_quorum_queues_v8_compat(Config) ->
+    check_quorum_queues_vn_compat(8, Config).
+
+check_quorum_queues_v9_compat(Config) ->
+    check_quorum_queues_vn_compat(9, Config).
+
+check_quorum_queues_vn_compat(Version, Config) ->
     Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     MacVer = lists:min([V || {ok, V} <- erpc:multicall(Nodes, rabbit_fifo, version, [])]),
-    case MacVer >= 8 of
+    case MacVer >= Version of
         true ->
             ok;
         false ->
