diff --git a/deps/rabbit/Makefile b/deps/rabbit/Makefile index a0ba08999d86..58775fee7e69 100644 --- a/deps/rabbit/Makefile +++ b/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 diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl index e7e8a7a0991f..5e291255e7b1 100644 --- a/deps/rabbit/src/rabbit_db_queue.erl +++ b/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), diff --git a/deps/rabbit/test/rabbit_db_binding_SUITE.erl b/deps/rabbit/test/rabbit_db_binding_SUITE.erl index 0af2aba2366a..ebf3f651e0ca 100644 --- a/deps/rabbit/test/rabbit_db_binding_SUITE.erl +++ b/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, diff --git a/deps/rabbit/test/rabbit_db_binding_prop_SUITE.erl b/deps/rabbit/test/rabbit_db_binding_prop_SUITE.erl new file mode 100644 index 000000000000..2fbd29a23f47 --- /dev/null +++ b/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).