Fix routing for empty binding key to topic exchange

Fixes #16221

 ## What?

Fix a bug in the Khepri v4 topic trie projection where
messages published with an empty routing key to a topic exchange
were incorrectly routed to queues bound to other topic exchanges
with an empty binding key.

 ## Why?

The MQTT 5.0 spec states:
> All Topic Names and Topic Filters MUST be at least one character long

In contrast, the AMQP 0.9.1 spec state:
> The routing key used for a topic exchange MUST consist of **zero** or more words
  delimited by dots. Each word may contain the letters A-Z and a-z and digits 0-9.
  The routing pattern follows the same rules as the routing key with the addition
  that * matches a single word, and # matches zero or more words

Hence zero words, i.e. empty routing keys and empty binding keys, are expliclity
allowed for the topic exchange type.

In the Khepri v4 projection, the topic trie used a global `root` atom
as the initial node ID for all topic exchanges. When a binding was
created with an empty binding key (`<<>>`), `split_topic_key_binary/1`
returned an empty list (`[]`). The trie traversal stopped immediately
at the root, meaning the binding was inserted into the
`rabbit_khepri_topic_binding_v4` ETS table with the global `root` atom
as the `LeafNodeId`.

Because the `LeafNodeId` lacked any exchange-specific context, all
empty bindings across all topic exchanges were
attached to the exact same global `root` node. Consequently, when a
message was published with an empty routing key to any topic exchange,
the routing logic (`trie_bindings/3`) would scan the ETS table starting
at the global `root` node and incorrectly match bindings belonging to
completely unrelated topic exchanges.

 ## How?

To fix this and ensure exchange isolation, the conceptual root of the
trie was changed from the global `root` atom to an exchange-specific
tuple: `{root, XSrc}` (where `XSrc` is `{VHost, ExchangeName}`).
- `rabbit_khepri:trie_follow_down_create/3` and `trie_follow_down_get_path/3`
  now initialize their trie traversal with `{root, XSrc}`.
- `rabbit_db_topic_exchange:trie_match/5` now initiates routing from
  `{root, {VHost, XName}}`.

This structural change guarantees that empty bindings are isolated per
exchange in the bindings ETS table, completely resolving the cross-exchange
leakage while fully supporting AMQP 0-9-1's allowance for empty binding
keys without breaking backward compatibility.

Author: ansd

deps/rabbit/src/rabbit_db_topic_exchange.erl

@@ -39,8 +39,9 @@ match(#resource{virtual_host = VHost, name = XName} = X, RoutingKey, Opts) ->
     Words = split_topic_key_binary(RoutingKey),
     case rabbit_khepri:get_effective_topic_binding_projection_version() of
         V when V >= 4 ->
+            XSrc = {VHost, XName},
             try
-                trie_match({VHost, XName}, root, Words, BKeys, [])
+                trie_match(XSrc, {root, XSrc}, Words, BKeys, [])
             catch
                 error:badarg ->
                     []

deps/rabbit/src/rabbit_khepri.erl

@@ -1733,7 +1733,7 @@ topic_binding_path_pattern() ->
 %% ChildCount tracks the number of outgoing edges from ChildNodeId.
 %% It is incremented when a new edge is created, decremented during GC.
 trie_follow_down_create(TrieTab, XSrc, Words) ->
-    trie_follow_down_create(TrieTab, XSrc, root, none, Words).
+    trie_follow_down_create(TrieTab, XSrc, {root, XSrc}, none, Words).
 
 trie_follow_down_create(_TrieTab, _XSrc, NodeId, _ParentKey, []) ->
     NodeId;
@@ -1757,7 +1757,7 @@ trie_follow_down_create(TrieTab, XSrc, ParentId, ParentKey, [Word | Rest]) ->
 %% Walk down the trie following the given words, collecting the path
 %% for later GC. Returns {ok, LeafNodeId, Path} or error.
 trie_follow_down_get_path(TrieTab, XSrc, Words) ->
-    trie_follow_down_get_path(TrieTab, XSrc, root, none, Words, []).
+    trie_follow_down_get_path(TrieTab, XSrc, {root, XSrc}, none, Words, []).
 
 trie_follow_down_get_path(_TrieTab, _XSrc, NodeId, _ParentKey, [], Path) ->
     {ok, NodeId, Path};

deps/rabbit/test/bindings_SUITE.erl

@@ -46,6 +46,7 @@ all_tests() ->
      unbind_from_volatile_queue,
      binding_args_direct_exchange,
      binding_args_fanout_exchange,
+     topic_exchange_zero_words,
 
      %% Exchange bindings
      bind_and_unbind_direct_exchange,
@@ -706,6 +707,71 @@ binding_args(Exchange, Config) ->
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"m2">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q, no_ack = true})).
 
+%% Test case for https://github.com/rabbitmq/rabbitmq-server/discussions/16221
+%%
+%% "The routing key used for a topic exchange MUST consist of zero or more words delimited by dots.
+%% Each word may contain the letters A-Z and a-z and digits 0-9.
+%% The routing pattern follows the same rules as the routing key with the addition
+%% that * matches a single word, and # matches zero or more words."
+%% [AMQP 0.9.1]
+%%
+%% Here, we test a zero words routing key and routing pattern.
+topic_exchange_zero_words(Config) ->
+    ZeroWords = <<>>,
+
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+
+    X1 = <<"x1">>,
+    X2 = <<"x2">>,
+    Q1 = <<"q1">>,
+    Q2 = <<"q2">>,
+
+    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = X1,
+                                                                         type = <<"topic">>}),
+    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = X2,
+                                                                         type = <<"topic">>}),
+    #'queue.declare_ok'{} = declare(Ch, Q1, []),
+    #'queue.declare_ok'{} = declare(Ch, Q2, []),
+
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = X1,
+                                                             queue = Q1,
+                                                             routing_key = ZeroWords}),
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = X2,
+                                                             queue = Q2,
+                                                             routing_key = ZeroWords}),
+
+    #'confirm.select_ok'{} = amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{exchange = X1,
+                                            routing_key = ZeroWords},
+                           #amqp_msg{payload = <<"m1">>}),
+    receive #'basic.ack'{} -> ok
+    after 9000 -> ct:fail(confirm_timeout)
+    end,
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"m1">>}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = Q1,
+                                                    no_ack = true})),
+
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{exchange = X2,
+                                            routing_key = ZeroWords},
+                           #amqp_msg{payload = <<"m2">>}),
+    receive #'basic.ack'{} -> ok
+    after 9000 -> ct:fail(confirm_timeout)
+    end,
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"m2">>}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = Q2, no_ack = true})),
+
+    ?assertMatch(#'queue.delete_ok'{message_count = 0},
+                 amqp_channel:call(Ch, #'queue.delete'{queue = Q1})),
+    ?assertMatch(#'queue.delete_ok'{message_count = 0},
+                 amqp_channel:call(Ch, #'queue.delete'{queue = Q2})),
+    #'exchange.delete_ok'{} = amqp_channel:call(Ch, #'exchange.delete'{exchange = X1}),
+    #'exchange.delete_ok'{} = amqp_channel:call(Ch, #'exchange.delete'{exchange = X2}),
+    rabbit_ct_client_helpers:close_channel(Ch).
+
 bind_and_unbind_direct_exchange(Config) ->
     bind_and_unbind_exchange(<<"direct">>, Config).