diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl index 9b09c5df386..aaf41ee5186 100644 --- a/deps/rabbit/src/rabbit_core_ff.erl +++ b/deps/rabbit/src/rabbit_core_ff.erl @@ -232,3 +232,15 @@ stability => stable, depends_on => ['rabbitmq_4.3.0'] }}). + +-rabbit_feature_flag( + {tie_binding_to_dest_with_keep_while_cond, + #{desc => + "Use keep_while condition to tie a binding record lifetime to its " + "destination record in Khepri", + stability => stable, + depends_on => ['rabbitmq_4.3.0'], + callbacks => #{enable => + {rabbit_db_queue, + tie_binding_to_dest_with_keep_while_cond_enable}} + }}). diff --git a/deps/rabbit/src/rabbit_db_binding.erl b/deps/rabbit/src/rabbit_db_binding.erl index 60d52295bda..90202164573 100644 --- a/deps/rabbit/src/rabbit_db_binding.erl +++ b/deps/rabbit/src/rabbit_db_binding.erl @@ -31,7 +31,9 @@ delete_for_destination_in_khepri/2, delete_all_for_exchange_in_khepri/3, has_for_source_in_khepri/1, - match_source_and_destination_in_khepri_tx/2 + match_source_and_destination_in_khepri_tx/2, + khepri_ret_to_deletions/2, + put_options/1 ]). -export([ @@ -123,6 +125,12 @@ create(#binding{source = SrcName, case ChecksFun(Src, Dst) of ok -> RoutePath = khepri_route_path(Binding), + FeatureFlag = rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond), + PutOptions = case FeatureFlag of + true -> put_options(DstName); + false -> #{} + end, MaybeSerial = rabbit_exchange:serialise_events(Src), Serial = rabbit_khepri:transaction( fun() -> @@ -132,11 +140,17 @@ create(#binding{source = SrcName, true -> already_exists; false -> - ok = khepri_tx:put(RoutePath, sets:add_element(Binding, Set)), + ok = khepri_tx:put( + RoutePath, + sets:add_element(Binding, Set), + PutOptions), serial_in_khepri(MaybeSerial, Src) end; _ -> - ok = khepri_tx:put(RoutePath, sets:add_element(Binding, sets:new([{version, 2}]))), + ok = khepri_tx:put( + RoutePath, + sets:add_element(Binding, sets:new([{version, 2}])), + PutOptions), serial_in_khepri(MaybeSerial, Src) end end, rw), @@ -166,6 +180,30 @@ lookup_resource(#resource{kind = exchange} = Name) -> _ -> [] end. +put_options(#resource{virtual_host = VHost, name = DstName, kind = Kind}) -> + put_options(VHost, DstName, Kind); +put_options(BindingPath) when is_list(BindingPath) -> + { + VHost, + _SrcName, + Kind, + DstName, + _RoutingKey + } = rabbit_db_binding:khepri_route_path_to_args(BindingPath), + put_options(VHost, DstName, Kind). + +put_options(VHost, DstName, Kind) -> + DstPath = case Kind of + queue -> + rabbit_db_queue:khepri_queue_path( + VHost, DstName); + exchange -> + rabbit_db_exchange:khepri_exchange_path( + VHost, DstName) + end, + KeepWhile = #{DstPath => #if_node_exists{}}, + #{keep_while => KeepWhile}. + serial_in_khepri(false, _) -> none; serial_in_khepri(true, X) -> @@ -190,8 +228,18 @@ serial_in_khepri(true, X) -> %% %% @private -delete(#binding{source = SrcName, - destination = DstName} = Binding, ChecksFun) -> +delete(Binding, ChecksFun) -> + FeatureFlag = rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond), + case FeatureFlag of + true -> + delete_v2(Binding, ChecksFun); + false -> + delete_v1(Binding, ChecksFun) + end. + +delete_v1(#binding{source = SrcName, + destination = DstName} = Binding, ChecksFun) -> Path = khepri_route_path(Binding), case rabbit_khepri:transaction( fun () -> @@ -204,7 +252,7 @@ delete(#binding{source = SrcName, true -> case ChecksFun(Src, Dst) of ok -> - ok = delete_in_khepri(Binding), + ok = delete_in_khepri_tx_v1(Binding), maybe_auto_delete_exchange_in_khepri(Binding#binding.source, [Binding], rabbit_binding:new_deletions(), false); {error, _} = Err -> Err @@ -224,6 +272,40 @@ delete(#binding{source = SrcName, {ok, Deletions} end. +delete_v2(#binding{source = SrcName, + destination = DstName} = Binding, ChecksFun) -> + Path = khepri_route_path(Binding), + case rabbit_khepri:transaction( + fun () -> + case {lookup_resource_in_khepri_tx(SrcName), + lookup_resource_in_khepri_tx(DstName)} of + {[Src], [Dst]} -> + case exists_in_khepri(Path, Binding) of + false -> + ok; + true -> + case ChecksFun(Src, Dst) of + ok -> + delete_in_khepri_tx_v2(Binding); + {error, _} = Err -> + Err + end + end; + _Errs -> + %% No absent queues, always present on disk + ok + end + end) of + ok -> + ok; + {error, _} = Err -> + Err; + {ok, Deleted} -> + Deletions = khepri_ret_to_deletions(Deleted, false), + ok = rabbit_binding:process_deletions(Deletions), + {ok, Deletions} + end. + exists_in_khepri(Path, Binding) -> case khepri_tx:get(Path) of {ok, Set} -> @@ -232,7 +314,7 @@ exists_in_khepri(Path, Binding) -> false end. -delete_in_khepri(Binding) -> +delete_in_khepri_tx_v1(Binding) -> Path = khepri_route_path(Binding), case khepri_tx:get(Path) of {ok, Set0} -> @@ -247,6 +329,21 @@ delete_in_khepri(Binding) -> ok end. +delete_in_khepri_tx_v2(Binding) -> + Path = khepri_route_path(Binding), + case khepri_tx:get(Path) of + {ok, Set0} -> + Set = sets:del_element(Binding, Set0), + case sets:is_empty(Set) of + true -> + khepri_tx_adv:delete(Path); + false -> + ok = khepri_tx:put(Path, Set) + end; + _ -> + ok + end. + maybe_auto_delete_exchange_in_khepri(XName, Bindings, Deletions, OnlyDurable) -> case rabbit_db_exchange:maybe_auto_delete_in_khepri(XName, OnlyDurable) of {not_deleted, undefined} -> @@ -562,6 +659,38 @@ delete_for_destination_in_khepri(#resource{virtual_host = VHost, kind = Kind, na rabbit_binding:group_bindings_fold(fun maybe_auto_delete_exchange_in_khepri/4, lists:keysort(#binding.source, Bindings), OnlyDurable). +khepri_ret_to_deletions(Deleted, OnlyDurable) -> + Bindings0 = maps:fold( + fun(Path, Props, Acc) -> + case {Path, Props} of + {?RABBITMQ_KHEPRI_ROUTE_PATH( + _VHost, _SrcName, _Kind, _Name, _RoutingKey), + #{data := Set}} -> + sets:to_list(Set) ++ Acc; + {_, _} -> + Acc + end + end, [], Deleted), + Bindings1 = lists:keysort(#binding.source, Bindings0), + rabbit_binding:group_bindings_fold( + fun(XName, Bindings, Deletions, _OnlyDurable) -> + ExchangePath = rabbit_db_exchange:khepri_exchange_path(XName), + case Deleted of + #{ExchangePath := #{data := X}} -> + rabbit_binding:add_deletion( + XName, X, deleted, Bindings, Deletions); + _ -> + case rabbit_db_exchange:get(XName) of + {ok, X} -> + rabbit_binding:add_deletion( + XName, X, not_deleted, Bindings, Deletions); + _ -> + Deletions + end + end + end, + Bindings1, OnlyDurable). + %% ------------------------------------------------------------------- %% has_for_source_in_khepri(). %% ------------------------------------------------------------------- diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl index 97cea1ee7f3..4b128547a99 100644 --- a/deps/rabbit/src/rabbit_db_exchange.erl +++ b/deps/rabbit/src/rabbit_db_exchange.erl @@ -41,7 +41,8 @@ get_in_khepri_tx/1, update_in_khepri_tx/2, clear_exchanges_in_khepri/0, - clear_exchange_serials_in_khepri/0 + clear_exchange_serials_in_khepri/0, + put_options/1 ]). %% For testing @@ -284,7 +285,13 @@ create_or_get(#exchange{name = XName} = X) -> Path0, [#if_any{conditions = [#if_node_exists{exists = false}, #if_has_payload{has_payload = false}]}]), - case rabbit_khepri:put(Path1, X) of + FeatureFlag = rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond), + PutOptions = case FeatureFlag of + true -> put_options(X); + false -> #{} + end, + case rabbit_khepri:put(Path1, X, PutOptions) of ok -> {new, X}; {error, {khepri, mismatching_node, #{node_props := #{data := ExistingX}}}} -> @@ -293,6 +300,23 @@ create_or_get(#exchange{name = XName} = X) -> Err end. +put_options(Exchange) -> + case Exchange of + #exchange{name = #resource{virtual_host = VHost, + name = Name}, + auto_delete = true} -> + Path = rabbit_db_binding:khepri_route_path( + VHost, + Name, + _Kind = ?KHEPRI_WILDCARD_STAR, + _DstName = ?KHEPRI_WILDCARD_STAR, + _RoutingKey = ?KHEPRI_WILDCARD_STAR), + KeepWhile = #{Path => #if_has_data{}}, + #{keep_while => KeepWhile}; + _ -> + #{} + end. + %% ------------------------------------------------------------------- %% set(). %% ------------------------------------------------------------------- diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl index d4570b894e8..37e3b07b8d7 100644 --- a/deps/rabbit/src/rabbit_db_queue.erl +++ b/deps/rabbit/src/rabbit_db_queue.erl @@ -8,6 +8,7 @@ -module(rabbit_db_queue). -include_lib("khepri/include/khepri.hrl"). +-include_lib("rabbit_common/include/logging.hrl"). -include_lib("rabbit_common/include/rabbit.hrl"). -include("amqqueue.hrl"). @@ -63,6 +64,9 @@ get_in_khepri_tx/1 ]). +%% Used for the `tie_binding_to_dest_with_keep_while_cond' feature flag. +-export([tie_binding_to_dest_with_keep_while_cond_enable/1]). + %% For testing -export([clear/0]). @@ -283,33 +287,53 @@ delete(QueueName, OnlyDurable) -> Conditions :: [khepri_condition:condition()], OnlyDurable :: boolean(), Ret :: ok | - Deletions :: rabbit_binding:deletions() | - rabbit_khepri:timeout_error(). + Deletions :: rabbit_binding:deletions() | + rabbit_khepri:timeout_error(). delete_if(QueueName, Conditions, OnlyDurable) -> - rabbit_khepri:transaction( - fun () -> - Path0 = khepri_queue_path(QueueName), - Path = khepri_path:combine_with_conditions(Path0, Conditions), - UsesUniformWriteRet = try - khepri_tx:does_api_comply_with(uniform_write_ret) - catch - error:undef -> - false - end, - case khepri_tx_adv:delete(Path) of - {ok, #{Path0 := #{data := _}}} when UsesUniformWriteRet -> - %% we want to execute some things, as decided by rabbit_exchange, - %% after the transaction. - rabbit_db_binding:delete_for_destination_in_khepri(QueueName, OnlyDurable); - {ok, #{data := _}} when not UsesUniformWriteRet -> - %% we want to execute some things, as decided by rabbit_exchange, - %% after the transaction. - rabbit_db_binding:delete_for_destination_in_khepri(QueueName, OnlyDurable); - {ok, _} -> - ok - end - end, rw). + Path = khepri_queue_path(QueueName), + Pattern = khepri_path:combine_with_conditions(Path, Conditions), + FeatureFlag = rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond), + case FeatureFlag of + true -> + case rabbit_khepri:adv_delete(Pattern) of + {ok, #{Path := #{data := _}} = Deleted} -> + rabbit_db_binding:khepri_ret_to_deletions( + Deleted, OnlyDurable); + {ok, _} -> + ok; + {error, _} = Error -> + Error + end; + false -> + rabbit_khepri:transaction( + fun () -> + UsesUniformWriteRet = ( + try + khepri_tx:does_api_comply_with(uniform_write_ret) + catch + error:undef -> + false + end), + case khepri_tx_adv:delete(Pattern) of + {ok, #{Path := #{data := _}}} + when UsesUniformWriteRet -> + %% we want to execute some things, as decided by + %% rabbit_exchange, after the transaction. + rabbit_db_binding:delete_for_destination_in_khepri( + QueueName, OnlyDurable); + {ok, #{data := _}} + when not UsesUniformWriteRet -> + %% we want to execute some things, as decided by + %% rabbit_exchange, after the transaction. + rabbit_db_binding:delete_for_destination_in_khepri( + QueueName, OnlyDurable); + {ok, _} -> + ok + end + end, rw) + end. %% ------------------------------------------------------------------- %% get_targets(). @@ -978,6 +1002,72 @@ list_with_possible_retry(Fun) -> Ret end. +tie_binding_to_dest_with_keep_while_cond_enable( + #{feature_name := FeatureName}) -> + ?LOG_DEBUG( + "Feature flag `~s`: send transaction to add `keep_while` conditions", + [FeatureName], + #{domain => ?RMQLOG_DOMAIN_DB}), + Ret = rabbit_khepri:transaction( + fun() -> + %% Auto-delete exchange. + ?LOG_DEBUG( + "Feature flag `~s`: add `keep_while` condition to " + "auto-delete exchanges", + [FeatureName], + #{domain => ?RMQLOG_DOMAIN_DB}), + ExchangePattern = rabbit_db_exchange:khepri_exchange_path( + ?KHEPRI_WILDCARD_STAR, + ?KHEPRI_WILDCARD_STAR), + ok = khepri_tx:foreach( + ExchangePattern, + fun(ExchangePath, #{data := Exchange}) -> + PutOptions = rabbit_db_exchange:put_options( + Exchange), + case PutOptions =:= #{} of + true -> + ok; + false -> + ok = khepri_tx:put( + ExchangePath, Exchange, + PutOptions) + end + end), + + %% Bindings. + ?LOG_DEBUG( + "Feature flag `~s`: add `keep_while` condition to " + "bindings", + [FeatureName], + #{domain => ?RMQLOG_DOMAIN_DB}), + BindingPattern = rabbit_db_binding:khepri_route_path( + ?KHEPRI_WILDCARD_STAR, + ?KHEPRI_WILDCARD_STAR, + ?KHEPRI_WILDCARD_STAR, + ?KHEPRI_WILDCARD_STAR, + ?KHEPRI_WILDCARD_STAR), + ok = khepri_tx:foreach( + BindingPattern, + fun(BindingPath, #{data := Set}) -> + PutOptions = rabbit_db_binding:put_options( + BindingPath), + ok = khepri_tx:put( + BindingPath, Set, + PutOptions) + end), + ok + end, rw), + ?LOG_DEBUG( + "Feature flag `~s`: transaction return value: ~p", + [FeatureName, Ret], + #{domain => ?RMQLOG_DOMAIN_DB}), + case Ret of + {ok, ok} -> + ok; + Error -> + Error + end. + %% -------------------------------------------------------------- %% Khepri paths %% -------------------------------------------------------------- diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl index aa3e380ea25..7210d23c71a 100644 --- a/deps/rabbit/src/rabbit_khepri.erl +++ b/deps/rabbit/src/rabbit_khepri.erl @@ -366,7 +366,7 @@ await_replication() -> %% Then it resets the store. This includes removing it from its cluster if %% any, and deleting all tree nodes. %% -%% Finally, it stops the store and deteles files on disk. +%% Finally, it stops the store and deletes files on disk. %% %% The Khepri application is left running. %% diff --git a/deps/rabbit/test/rabbit_db_binding_SUITE.erl b/deps/rabbit/test/rabbit_db_binding_SUITE.erl index fd096a41188..629f28cdcc2 100644 --- a/deps/rabbit/test/rabbit_db_binding_SUITE.erl +++ b/deps/rabbit/test/rabbit_db_binding_SUITE.erl @@ -8,6 +8,7 @@ -module(rabbit_db_binding_SUITE). -include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("common_test/include/ct.hrl"). -include_lib("eunit/include/eunit.hrl"). -export([all/0, @@ -21,8 +22,8 @@ create/1, create1/1, exists/1, exists1/1, - delete/1, delete1/1, - auto_delete/1, auto_delete1/1, + delete_v1/1, delete_v2/1, delete1/2, + auto_delete_v1/1, auto_delete_v2/1, auto_delete1/2, get_all/1, get_all1/1, get_all_by_vhost/1, get_all_by_vhost1/1, get_all_for_source/1, get_all_for_source1/1, @@ -33,27 +34,34 @@ get_all_for_source_and_destination_reverse1/1, fold/1, fold1/1, match/1, match1/1, - match_routing_key/1, match_routing_key1/1 + match_routing_key/1, match_routing_key1/1, + tie_binding_to_dest_with_keep_while_cond/1, + tie_binding_to_dest_with_keep_while_cond1/1 ]). -define(VHOST, <<"/">>). all() -> [ - {group, all_tests} + {group, all_tests}, + {group, feature_flags} ]. groups() -> [ - {all_tests, [], all_tests()} + {all_tests, [], all_tests()}, + {feature_flags, [], + [delete_v1, + delete_v2, + auto_delete_v1, + auto_delete_v2, + tie_binding_to_dest_with_keep_while_cond]} ]. all_tests() -> [ create, exists, - delete, - auto_delete, get_all, get_all_by_vhost, get_all_for_source, @@ -81,7 +89,15 @@ init_per_group(Group, Config) -> {rmq_nodename_suffix, Group}, {rmq_nodes_count, 1} ]), - rabbit_ct_helpers:run_steps(Config1, + Config2 = case Group of + feature_flags -> + rabbit_ct_helpers:merge_app_env( + Config1, + {rabbit, [{forced_feature_flags_on_init, []}]}); + _ -> + Config1 + end, + rabbit_ct_helpers:run_steps(Config2, rabbit_ct_broker_helpers:setup_steps() ++ rabbit_ct_client_helpers:setup_steps()). @@ -91,7 +107,16 @@ end_per_group(_Group, Config) -> rabbit_ct_broker_helpers:teardown_steps()). init_per_testcase(Testcase, Config) -> - rabbit_ct_helpers:testcase_started(Config, Testcase). + rabbit_ct_helpers:testcase_started(Config, Testcase), + case lists:member({name, feature_flags}, ?config(tc_group_properties, Config)) of + true -> + ok = rabbit_ct_broker_helpers:stop_broker(Config, 0), + ok = rabbit_ct_broker_helpers:reset_node(Config, 0), + ok = rabbit_ct_broker_helpers:start_broker(Config, 0); + false -> + ok + end, + Config. end_per_testcase(Testcase, Config) -> rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_db_exchange, clear, []), @@ -140,10 +165,30 @@ exists1(_Config) -> ?assertEqual(true, rabbit_db_binding:exists(Binding)), passed. -delete(Config) -> - passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete1, [Config]). - -delete1(_Config) -> +delete_v1(Config) -> + passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete1, [Config, v1]). + +delete_v2(Config) -> + Ret = rabbit_ct_broker_helpers:enable_feature_flag( + Config, tie_binding_to_dest_with_keep_while_cond), + case Ret of + ok -> + passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete1, [Config, v2]); + {skip, _} = Skip -> + Skip + end. + +delete1(_Config, Version) -> + case Version of + v1 -> + ?assertNot( + rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond)); + v2 -> + ?assert( + rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond)) + end, XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>), XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>), Exchange1 = #exchange{name = XName1, durable = true, auto_delete = false, decorators = {[], []}}, @@ -161,10 +206,30 @@ delete1(_Config) -> ?assertEqual(false, rabbit_db_binding:exists(Binding)), passed. -auto_delete(Config) -> - passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, auto_delete1, [Config]). - -auto_delete1(_Config) -> +auto_delete_v1(Config) -> + passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, auto_delete1, [Config, v1]). + +auto_delete_v2(Config) -> + Ret = rabbit_ct_broker_helpers:enable_feature_flag( + Config, tie_binding_to_dest_with_keep_while_cond), + case Ret of + ok -> + passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, auto_delete1, [Config, v2]); + {skip, _} = Skip -> + Skip + end. + +auto_delete1(_Config, Version) -> + case Version of + v1 -> + ?assertNot( + rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond)); + v2 -> + ?assert( + rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond)) + end, XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>), XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>), Exchange1 = #exchange{name = XName1, durable = true, auto_delete = true, decorators = {[], []}}, @@ -356,3 +421,150 @@ match_routing_key1(_Config) -> ?assertEqual([], rabbit_db_binding:match_routing_key(XName1, [<<"a.b.c">>], false)), ?assertEqual([XName2], rabbit_db_binding:match_routing_key(XName1, [<<"a.b">>], false)), passed. + +tie_binding_to_dest_with_keep_while_cond(Config) -> + passed = 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) -> + %% The testcase uses `rabbit_db_*' modules directly to test specifically + %% the content of the metadata store and how it behaves. + %% + %% It creates two exchanges, one with auto_delete=false, the other one with + %% auto_delete=true. It then creates a queue and binds it to the two + %% exchanges. It finally deletes the queue and check what happens with the + %% bindings and the exchanges. + %% + %% This scenario is executed three times: + %% 1. with the `tie_binding_to_dest_with_keep_while_cond' feature flag + %% disabled + %% 2. with the feature flag enabled in the middle + %% 3. with the feature flag already eabled + %% The behaviour should be identical in all cases. + + %% Prepare resource records. + NamePrefix = atom_to_binary(?FUNCTION_NAME), + + XName1 = rabbit_misc:r(?VHOST, exchange, <>), + XName2 = rabbit_misc:r(?VHOST, exchange, <>), + XPath1 = rabbit_db_exchange:khepri_exchange_path(XName1), + XPath2 = rabbit_db_exchange:khepri_exchange_path(XName2), + Exchange1 = #exchange{ + name = XName1, + durable = true, auto_delete = true, decorators = {[], []}}, + Exchange2 = #exchange{ + name = XName2, + durable = true, auto_delete = false, decorators = {[], []}}, + QName = rabbit_misc:r(?VHOST, queue, <>), + QPath = rabbit_db_queue:khepri_queue_path(QName), + Queue = amqqueue:new(QName, none, true, false, none, [], ?VHOST, #{}), + Binding1 = #binding{ + source = XName1, key = <<"">>, + destination = QName, args = #{}}, + Binding2 = #binding{ + source = XName2, key = <<"">>, + destination = QName, args = #{}}, + BPath1 = rabbit_db_binding:khepri_route_path(Binding1), + BPath2 = rabbit_db_binding:khepri_route_path(Binding2), + + %% 1. Scenario with the feature flag disabled. + ?assertNot( + rabbit_feature_flags:is_enabled( + tie_binding_to_dest_with_keep_while_cond)), + + ?assertMatch( + {new, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange1)), + ?assertMatch( + {new, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange2)), + + ?assertMatch({created, Queue}, rabbit_db_queue:create_or_get(Queue)), + + ?assertMatch(ok, rabbit_db_binding:create(Binding1, fun(_, _) -> ok end)), + ?assertMatch(ok, rabbit_db_binding:create(Binding2, fun(_, _) -> ok end)), + + {ok, KWCS1} = khepri_machine:get_keep_while_conds_state( + rabbit_khepri:get_store_id(), #{}), + ?assertNotMatch(#{XPath1 := _}, KWCS1), + ?assertNotMatch(#{XPath2 := _}, KWCS1), + ?assertNotMatch(#{BPath1 := _}, KWCS1), + ?assertNotMatch(#{BPath2 := _}, KWCS1), + + Deletions1 = rabbit_db_queue:delete(QName, false), + ?assertMatch( + #{XName1 := _, + XName2 := _}, + Deletions1), + + ?assertNot(rabbit_khepri:exists(QPath)), + ?assertNot(rabbit_khepri:exists(XPath1)), + ?assert(rabbit_khepri:exists(XPath2)), + ?assertNot(rabbit_khepri:exists(BPath1)), + ?assertNot(rabbit_khepri:exists(BPath2)), + + %% 2. Scenario with the feature flag enabled in the process. + + ?assertMatch( + {new, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange1)), + ?assertMatch( + {existing, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange2)), + ?assertMatch({created, Queue}, rabbit_db_queue:create_or_get(Queue)), + ?assertMatch(ok, rabbit_db_binding:create(Binding1, fun(_, _) -> ok end)), + ?assertMatch(ok, rabbit_db_binding:create(Binding2, fun(_, _) -> ok end)), + + ?assertEqual( + ok, + rabbit_feature_flags:enable( + tie_binding_to_dest_with_keep_while_cond)), + + {ok, KWCS2} = khepri_machine:get_keep_while_conds_state( + rabbit_khepri:get_store_id(), #{}), + ?assertMatch(#{XPath1 := _}, KWCS2), + ?assertNotMatch(#{XPath2 := _}, KWCS2), + ?assertMatch(#{BPath1 := _}, KWCS2), + ?assertMatch(#{BPath2 := _}, KWCS2), + + Deletions2 = rabbit_db_queue:delete(QName, false), + ?assertEqual(Deletions1, Deletions2), + + ?assertNot(rabbit_khepri:exists(QPath)), + ?assertNot(rabbit_khepri:exists(XPath1)), + ?assert(rabbit_khepri:exists(XPath2)), + ?assertNot(rabbit_khepri:exists(BPath1)), + ?assertNot(rabbit_khepri:exists(BPath2)), + + %% 3. Scenario with the feature flag already enabled. + ?assert( + rabbit_feature_flags:is_enabled(tie_binding_to_dest_with_keep_while_cond)), + + ?assertMatch( + {new, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange1)), + ?assertMatch( + {existing, #exchange{}}, + rabbit_db_exchange:create_or_get(Exchange2)), + ?assertMatch({created, Queue}, rabbit_db_queue:create_or_get(Queue)), + ?assertMatch(ok, rabbit_db_binding:create(Binding1, fun(_, _) -> ok end)), + ?assertMatch(ok, rabbit_db_binding:create(Binding2, fun(_, _) -> ok end)), + + {ok, KWCS3} = khepri_machine:get_keep_while_conds_state( + rabbit_khepri:get_store_id(), #{}), + ?assertMatch(#{XPath1 := _}, KWCS3), + ?assertNotMatch(#{XPath2 := _}, KWCS3), + ?assertMatch(#{BPath1 := _}, KWCS3), + ?assertMatch(#{BPath2 := _}, KWCS3), + + Deletions3 = rabbit_db_queue:delete(QName, false), + ?assertEqual(Deletions1, Deletions3), + + ?assertNot(rabbit_khepri:exists(QPath)), + ?assertNot(rabbit_khepri:exists(XPath1)), + ?assert(rabbit_khepri:exists(XPath2)), + ?assertNot(rabbit_khepri:exists(BPath1)), + ?assertNot(rabbit_khepri:exists(BPath2)), + + passed.