Fix a tie_binding_to_dest_with_keep_while_cond exception on certain auto-delete exchanges

deps/rabbit/Makefile

@@ -253,7 +253,7 @@ PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql am
 PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange cluster_limit cluster_minority default_queue_type_prop passive_declare_permission term_to_binary_compat_prop topic_permission unicode unit_access_control user_tags_count_limit
 PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel channel_interceptor channel_operation_timeout classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit server_named_queue_prefix
 
-PARALLEL_CT_SET_2_A = cluster confirms_rejects consumer_timeout rabbit_access_control rabbit_confirms rabbit_core_metrics_gc rabbit_cuttlefish rabbit_db_binding rabbit_db_exchange
+PARALLEL_CT_SET_2_A = cluster confirms_rejects consumer_timeout rabbit_access_control rabbit_confirms rabbit_core_metrics_gc rabbit_cuttlefish rabbit_db_binding rabbit_db_binding_prop rabbit_db_exchange
 PARALLEL_CT_SET_2_B = crashing_queues deprecated_features direct_exchange_routing_v2 disconnect_detected_during_alarm exchanges unit_gen_server2
 PARALLEL_CT_SET_2_C = disk_monitor dynamic_qq unit_disk_monitor unit_log_management unit_operator_policy prevent_startup_if_node_was_reset
 PARALLEL_CT_SET_2_D = queue_length_limits queue_parallel quorum_queue_member_reconciliation rabbit_fifo rabbit_fifo_dlx rabbit_stream_coordinator

deps/rabbit/src/rabbit_db_queue.erl

@@ -1052,9 +1052,22 @@ tie_binding_to_dest_with_keep_while_cond_enable(
                                        true ->
                                            ok;
                                        false ->
-                                           ok = khepri_tx:put(
-                                                  ExchangePath, Exchange,
-                                                  PutOptions)
+                                           %% An auto-delete exchange's
+                                           %% `keep_while' condition requires at
+                                           %% least one binding where it is the
+                                           %% source. Adding it to an exchange
+                                           %% without such bindings is rejected
+                                           %% by Khepri with
+                                           %% `keep_while_conditions_not_met'.
+                                           case rabbit_db_binding:has_for_source_in_khepri(
+                                                  Exchange#exchange.name) of
+                                               true ->
+                                                   ok = khepri_tx:put(
+                                                          ExchangePath, Exchange,
+                                                          PutOptions);
+                                               false ->
+                                                   ok
+                                           end
                                    end
                            end),
 

deps/rabbit/test/rabbit_db_binding_SUITE.erl

@@ -40,7 +40,13 @@
          tie_binding_to_dest_with_keep_while_cond_orphan_nodes/1,
          tie_binding_to_dest_with_keep_while_cond_orphan_nodes1/1,
          tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission/1,
-         tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission1/1
+         tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission1/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings1/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange1/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings/1,
+         tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings1/1
         ]).
 
 -define(VHOST, <<"/">>).
@@ -61,7 +67,10 @@ groups() ->
        auto_delete_v2,
        tie_binding_to_dest_with_keep_while_cond,
        tie_binding_to_dest_with_keep_while_cond_orphan_nodes,
-       tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission]}
+       tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission,
+       tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings,
+       tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange,
+       tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings]}
     ].
 
 all_tests() ->
@@ -623,6 +632,133 @@ tie_binding_to_dest_with_keep_while_cond_orphan_nodes1(_Config) ->
 
     passed.
 
+tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, 0, ?MODULE,
+               tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings1,
+               [Config]).
+
+tie_binding_to_dest_with_keep_while_cond_auto_delete_without_bindings1(_Config) ->
+    %% Regression test for #16823: enabling the feature flag crashed with
+    %% `keep_while_conditions_not_met' on an auto-delete exchange that has no
+    %% binding where it is the source.
+    ?assertNot(
+       rabbit_feature_flags:is_enabled(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    XName = rabbit_misc:r(?VHOST, exchange, <<"auto-delete-x">>),
+    Exchange = #exchange{name = XName,
+                         type = headers,
+                         durable = true, auto_delete = true,
+                         decorators = {[], []}},
+    ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange)),
+
+    ?assertEqual(
+       ok,
+       rabbit_feature_flags:enable(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    ?assert(rabbit_khepri:exists(
+              rabbit_db_exchange:khepri_exchange_path(?VHOST, XName#resource.name))),
+    passed.
+
+tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, 0, ?MODULE,
+               tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange1,
+               [Config]).
+
+tie_binding_to_dest_with_keep_while_cond_auto_delete_alternate_exchange1(_Config) ->
+    %% Reproduces the topology from rabbitmq/rabbitmq-server#16823: `DeadEnd' is
+    %% an auto-delete exchange that is only an exchange-to-exchange destination
+    %% and an alternate-exchange target, so it has no binding where it is the
+    %% source.
+    ?assertNot(
+       rabbit_feature_flags:is_enabled(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    DestinationName = rabbit_misc:r(?VHOST, exchange, <<"x.destination">>),
+    AltName = rabbit_misc:r(?VHOST, exchange, <<"alt-target">>),
+    DeadEndName = rabbit_misc:r(?VHOST, exchange, <<"dead-end">>),
+    SourceName = rabbit_misc:r(?VHOST, exchange, <<"source">>),
+    Destination = #exchange{name = DestinationName, type = fanout,
+                            durable = true, auto_delete = false,
+                            decorators = {[], []}},
+    Alt = #exchange{name = AltName, type = fanout,
+                    durable = true, auto_delete = true,
+                    decorators = {[], []}},
+    DeadEnd = #exchange{name = DeadEndName, type = headers,
+                        durable = true, auto_delete = true,
+                        arguments = [{<<"alternate-exchange">>, longstr, <<"alt-target">>}],
+                        decorators = {[], []}},
+    Source = #exchange{name = SourceName, type = topic,
+                       durable = true, auto_delete = false,
+                       decorators = {[], []}},
+    [?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(X))
+     || X <- [Destination, Alt, DeadEnd, Source]],
+    AltToDestination = #binding{source = AltName, key = <<>>,
+                                destination = DestinationName, args = #{}},
+    SourceToDeadEnd = #binding{source = SourceName, key = <<>>,
+                              destination = DeadEndName, args = #{}},
+    [?assertEqual(ok, rabbit_db_binding:create(B, fun(_, _) -> ok end))
+     || B <- [AltToDestination, SourceToDeadEnd]],
+
+    ?assertEqual(
+       ok,
+       rabbit_feature_flags:enable(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    ?assert(rabbit_khepri:exists(
+              rabbit_db_exchange:khepri_exchange_path(
+                ?VHOST, DeadEndName#resource.name))),
+    passed.
+
+tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, 0, ?MODULE,
+               tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings1,
+               [Config]).
+
+tie_binding_to_dest_with_keep_while_cond_auto_delete_with_bindings1(_Config) ->
+    %% The migration attaches a `keep_while' condition to an auto-delete
+    %% exchange that is the source of a binding, but not to one that is not.
+    ?assertNot(
+       rabbit_feature_flags:is_enabled(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    DestinationName = rabbit_misc:r(?VHOST, exchange, <<"x.destination">>),
+    WithBindingName = rabbit_misc:r(?VHOST, exchange, <<"with-binding">>),
+    WithoutBindingName = rabbit_misc:r(?VHOST, exchange, <<"without-binding">>),
+    Destination = #exchange{name = DestinationName, type = fanout,
+                            durable = true, auto_delete = false,
+                            decorators = {[], []}},
+    WithBinding = #exchange{name = WithBindingName, type = headers,
+                            durable = true, auto_delete = true,
+                            decorators = {[], []}},
+    WithoutBinding = #exchange{name = WithoutBindingName, type = headers,
+                               durable = true, auto_delete = true,
+                               decorators = {[], []}},
+    [?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(X))
+     || X <- [Destination, WithBinding, WithoutBinding]],
+    Binding = #binding{source = WithBindingName, key = <<>>,
+                       destination = DestinationName, args = #{}},
+    ?assertEqual(ok, rabbit_db_binding:create(Binding, fun(_, _) -> ok end)),
+
+    ?assertEqual(
+       ok,
+       rabbit_feature_flags:enable(
+         tie_binding_to_dest_with_keep_while_cond)),
+
+    {ok, KWCS} = khepri_machine:get_keep_while_conds_state(
+                   rabbit_khepri:get_store_id(), #{}),
+    WithBindingPath = rabbit_db_exchange:khepri_exchange_path(
+                        ?VHOST, WithBindingName#resource.name),
+    WithoutBindingPath = rabbit_db_exchange:khepri_exchange_path(
+                           ?VHOST, WithoutBindingName#resource.name),
+    ?assertMatch(#{WithBindingPath := _}, KWCS),
+    ?assertNotMatch(#{WithoutBindingPath := _}, KWCS),
+    passed.
+
 tie_binding_to_dest_with_keep_while_cond_orphan_topic_permission(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(
                Config, 0,

deps/rabbit/test/rabbit_db_binding_prop_SUITE.erl

@@ -0,0 +1,149 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(rabbit_db_binding_prop_SUITE).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("proper/include/proper.hrl").
+
+-export([all/0,
+         groups/0,
+         init_per_suite/1,
+         end_per_suite/1,
+         init_per_group/2,
+         end_per_group/2,
+         init_per_testcase/2,
+         end_per_testcase/2,
+
+         tie_binding_to_dest_with_keep_while_cond/1,
+         tie_binding_to_dest_with_keep_while_cond1/1]).
+
+-define(VHOST, <<"/">>).
+-define(FEATURE_FLAG, tie_binding_to_dest_with_keep_while_cond).
+-define(NUM_TESTS, 100).
+
+all() ->
+    [{group, property_based}].
+
+groups() ->
+    [{property_based, [], [tie_binding_to_dest_with_keep_while_cond]}].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(Group, Config) ->
+    Config1 = rabbit_ct_helpers:set_config(
+                Config, [{rmq_nodename_suffix, Group},
+                         {rmq_nodes_count, 1}]),
+    %% The feature flag must start disabled so the property can run the
+    %% migration itself.
+    Config2 = rabbit_ct_helpers:merge_app_env(
+                Config1, {rabbit, [{forced_feature_flags_on_init, []}]}),
+    rabbit_ct_helpers:run_steps(
+      Config2, rabbit_ct_broker_helpers:setup_steps()).
+
+end_per_group(_Group, Config) ->
+    rabbit_ct_helpers:run_steps(
+      Config, rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    Config.
+
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+tie_binding_to_dest_with_keep_while_cond(Config) ->
+    true = rabbit_ct_broker_helpers:rpc(
+             Config, 0, ?MODULE,
+             tie_binding_to_dest_with_keep_while_cond1, [Config]).
+
+tie_binding_to_dest_with_keep_while_cond1(_Config) ->
+    proper:quickcheck(
+      prop_tie_binding_to_dest_with_keep_while_cond(),
+      [{numtests, ?NUM_TESTS}]).
+
+%% Each exchange is described by whether it is auto-delete and whether it has a
+%% binding where it is the source.
+exchange_spec() ->
+    {boolean(), boolean()}.
+
+topology() ->
+    ?LET(N, integer(0, 12), vector(N, exchange_spec())).
+
+%% For any topology, the migration must succeed, keep every exchange, and add a
+%% `keep_while' condition to exactly those auto-delete exchanges that have a
+%% source binding.
+prop_tie_binding_to_dest_with_keep_while_cond() ->
+    ?FORALL(
+       Specs, topology(),
+       begin
+           ok = clear(),
+           Destination = declare_exchange(<<"x.destination">>, fanout, false),
+           Names = declare_topology(Specs, Destination),
+           Ret = run_migration(),
+           AllExist = lists:all(
+                        fun(N) -> rabbit_khepri:exists(xpath(N)) end,
+                        [Destination | Names]),
+           CondsOk = lists:all(
+                       fun({{AutoDelete, HasBinding}, N}) ->
+                               has_keep_while_cond(N) =:=
+                                   (AutoDelete andalso HasBinding)
+                       end,
+                       lists:zip(Specs, Names)),
+           ok = clear(),
+           Ret =:= ok andalso AllExist andalso CondsOk
+       end).
+
+declare_topology(Specs, Destination) ->
+    [declare_spec(I, Spec, Destination) || {I, Spec} <- lists:enumerate(Specs)].
+
+declare_spec(I, {AutoDelete, HasBinding}, Destination) ->
+    XName = declare_exchange(
+              <<"x-", (integer_to_binary(I))/binary>>, headers, AutoDelete),
+    case HasBinding of
+        true  -> ok = bind(XName, Destination);
+        false -> ok
+    end,
+    XName.
+
+declare_exchange(Name, Type, AutoDelete) ->
+    XName = rabbit_misc:r(?VHOST, exchange, Name),
+    X = #exchange{name = XName, type = Type,
+                  durable = true, auto_delete = AutoDelete,
+                  decorators = {[], []}},
+    {new, #exchange{}} = rabbit_db_exchange:create_or_get(X),
+    XName.
+
+bind(SrcName, DstName) ->
+    Binding = #binding{source = SrcName, key = <<>>,
+                       destination = DstName, args = #{}},
+    rabbit_db_binding:create(Binding, fun(_, _) -> ok end).
+
+%% Invokes the migration callback directly. Unlike enabling the feature flag,
+%% this can be run repeatedly on a single node, which the property relies on.
+run_migration() ->
+    rabbit_db_queue:tie_binding_to_dest_with_keep_while_cond_enable(
+      #{feature_name => ?FEATURE_FLAG}).
+
+clear() ->
+    ok = rabbit_db_exchange:clear(),
+    ok = rabbit_db_binding:clear(),
+    ok.
+
+xpath(#resource{name = Name}) ->
+    rabbit_db_exchange:khepri_exchange_path(?VHOST, Name).
+
+has_keep_while_cond(XName) ->
+    {ok, Conds} = khepri_machine:get_keep_while_conds_state(
+                    rabbit_khepri:get_store_id(), #{}),
+    maps:is_key(xpath(XName), Conds).