Bugfixes

7 files changed, 16 insertions, 77 deletions

diff --git a/deps/rabbit/src/rabbit_db_binding.erl b/deps/rabbit/src/rabbit_db_binding.erl
index b75fdf4f9f..b7d737c8b1 100644
--- a/deps/rabbit/src/rabbit_db_binding.erl
+++ b/deps/rabbit/src/rabbit_db_binding.erl
@@ -776,7 +776,7 @@ delete_all_for_exchange_in_khepri(X = #exchange{name = XName}, OnlyDurable, Remo
 
 delete_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
     Path = khepri_routes_path() ++ [VHost, Name],
-    {ok, Bindings} = khepri_tx:get_many(Path ++ [rabbit_db:if_has_data_wildcard()]),
+    {ok, Bindings} = khepri_tx:get_many(Path ++ [rabbit_khepri:if_has_data_wildcard()]),
     %% ok = khepri_tx:delete(Path),
     maps:fold(fun(P, Set, Acc) ->
                       %% TODO projections are not updated if we just use the previous
diff --git a/deps/rabbit/test/metadata_store_clustering_SUITE.erl b/deps/rabbit/test/metadata_store_clustering_SUITE.erl
index 506dc1f0e0..d60b39bce3 100644
--- a/deps/rabbit/test/metadata_store_clustering_SUITE.erl
+++ b/deps/rabbit/test/metadata_store_clustering_SUITE.erl
@@ -12,8 +12,6 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
--include_lib("khepri/include/khepri.hrl").
-
 -compile([nowarn_export_all, export_all]).
 
 suite() ->
diff --git a/deps/rabbit/test/metadata_store_phase1_SUITE.erl b/deps/rabbit/test/metadata_store_phase1_SUITE.erl
index 4cf5d75b20..bab69123af 100644
--- a/deps/rabbit/test/metadata_store_phase1_SUITE.erl
+++ b/deps/rabbit/test/metadata_store_phase1_SUITE.erl
@@ -199,7 +199,7 @@ setup_khepri(Config) ->
     Config.
 
 setup_code_mocking(Config) ->
-    %% Bypass rabbit_misc:execute_mnesia_transaction/1 (no worker_pool
+    %% Bypass rabbit_mnesia:execute_mnesia_transaction/1 (no worker_pool
     %% configured in particular) but keep the behavior of throwing the error.
     meck:new(rabbit_misc, [passthrough, no_link]),
     meck:expect(
@@ -2656,7 +2656,7 @@ list_vhost_records(khepri) ->
     lists:sort(rabbit_vhost:all_in_khepri()).
 
 update_vhost(mnesia, VHostName, Fun) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               rabbit_vhost:update_in_mnesia(VHostName, Fun)
       end);
@@ -2664,7 +2664,7 @@ update_vhost(khepri, VHostName, Fun) ->
     rabbit_vhost:update_in_khepri(VHostName, Fun).
 
 update_vhost(mnesia, VHostName, Description, Tags) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               rabbit_vhost:update_in_mnesia(VHostName, Description, Tags)
       end);
@@ -2677,7 +2677,7 @@ vhost_info(khepri, VHostName) ->
     rabbit_vhost:info_in_khepri(VHostName).
 
 delete_vhost(mnesia, VHostName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               Fun = rabbit_vhost:with_in_mnesia(
                       VHostName,
diff --git a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
index 855c2e51af..8bbe5b5f62 100644
--- a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
+++ b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
@@ -58,7 +58,7 @@ create(X) ->
        }).
 
 create_in_mnesia(X) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() -> create_in_mnesia_tx(X) end).
 
 create_in_mnesia_tx(X) ->
@@ -90,7 +90,7 @@ create_binding(Src, Dst, Weight, UpdateFun) ->
        }).
 
 create_binding_in_mnesia(Src, Dst, Weight, UpdateFun) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               create_binding_in_mnesia_tx(Src, Dst, Weight, UpdateFun)
       end).
@@ -171,7 +171,7 @@ delete(XName) ->
        }).
 
 delete_in_mnesia(XName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               mnesia:write_lock_table(?HASH_RING_STATE_TABLE),
               mnesia:delete({?HASH_RING_STATE_TABLE, XName})
@@ -187,7 +187,7 @@ delete_bindings(Bindings, DeleteFun) ->
        }).
 
 delete_bindings_in_mnesia(Bindings, DeleteFun) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               [delete_binding_in_mnesia(Binding, DeleteFun) || Binding <- Bindings]
       end).
diff --git a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_exchange_type_consistent_hash.erl b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_exchange_type_consistent_hash.erl
index 2bd4493e31..09a028e238 100644
--- a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_exchange_type_consistent_hash.erl
+++ b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_exchange_type_consistent_hash.erl
@@ -48,11 +48,7 @@
 
 init() ->
     rabbit_db_ch_exchange:setup_schema(),
-<<<<<<< HEAD
     _ = recover(),
-=======
-    recover(),
->>>>>>> 7b61c2e402 (Khepri: The One Commit)
     ok.
 
 info(_X) -> [].
@@ -173,7 +169,6 @@ add_binding(_Serial, _X, #binding{source = S, destination = D, key = K}) ->
             rabbit_log:debug("Consistent hashing exchange: adding binding from "
                              "exchange ~s to destination ~s with routing key '~s'",
                              [rabbit_misc:rs(S), rabbit_misc:rs(D), K])
-<<<<<<< HEAD
     end.
 
 chx_hash_ring_update_fun(#chx_hash_ring{bucket_map = BM0,
@@ -225,59 +220,6 @@ ch_hash_ring_delete_fun(#chx_hash_ring{bucket_map = BM0,
             NextN = NexN0 - N,
             Chx0#chx_hash_ring{bucket_map = BM1,
                                next_bucket_number = NextN}
-=======
->>>>>>> 7b61c2e402 (Khepri: The One Commit)
-    end.
-
-chx_hash_ring_update_fun(#chx_hash_ring{bucket_map = BM0,
-                                        next_bucket_number = NexN0} = Chx0,
-                         Dst, Weight) ->
-    case map_has_value(BM0, Dst) of
-        true ->
-            already_exists;
-        false ->
-            NextN   = NexN0 + Weight,
-            %% hi/lo bucket counters are 0-based but weight is 1-based
-            Range   = lists:seq(NexN0, (NextN - 1)),
-            BM      = lists:foldl(fun(Key, Acc) ->
-                                          maps:put(Key, Dst, Acc)
-                                  end, BM0, Range),
-            Chx0#chx_hash_ring{bucket_map = BM,
-                               next_bucket_number = NextN}
-    end.
-
-remove_bindings(_Serial, _X, Bindings) ->
-    Ret = rabbit_db_ch_exchange:delete_bindings(Bindings, fun ch_hash_ring_delete_fun/2),
-    [rabbit_log:warning("Can't remove binding: hash ring state for exchange ~s wasn't found",
-                        [rabbit_misc:rs(X)]) || {not_found, X} <- Ret],
-    ok.
-
-ch_hash_ring_delete_fun(#chx_hash_ring{bucket_map = BM0,
-                                       next_bucket_number = NexN0} = Chx0,
-                       Dst) ->
-    %% Buckets with lower numbers stay as is; buckets that
-    %% belong to this binding are removed; buckets with
-    %% greater numbers are updated (their numbers are adjusted downwards)
-    BucketsOfThisBinding = maps:filter(fun (_K, V) -> V =:= Dst end, BM0),
-    case maps:size(BucketsOfThisBinding) of
-        0             ->
-            not_found;
-        N when N >= 1 ->
-            KeysOfThisBinding  = lists:usort(maps:keys(BucketsOfThisBinding)),
-            LastBucket         = lists:last(KeysOfThisBinding),
-            FirstBucket        = hd(KeysOfThisBinding),
-            BucketsDownTheRing = maps:filter(fun (K, _) -> K > LastBucket end, BM0),
-            UnchangedBuckets   = maps:filter(fun (K, _) -> K < FirstBucket end, BM0),
-            
-            %% final state with "down the ring" buckets updated
-            NewBucketsDownTheRing = maps:fold(
-                                      fun(K0, V, Acc)  ->
-                                              maps:put(K0 - N, V, Acc)
-                                      end, #{}, BucketsDownTheRing),
-            BM1 = maps:merge(UnchangedBuckets, NewBucketsDownTheRing),
-            NextN = NexN0 - N,
-            Chx0#chx_hash_ring{bucket_map = BM1,
-                               next_bucket_number = NextN}
     end.
             
 -spec ring_state(vhost:name(), rabbit_misc:resource_name()) ->
diff --git a/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl
index 199a3e73d4..8a5086dc45 100644
--- a/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl
+++ b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl
@@ -69,7 +69,7 @@ create_or_update(XName, BindingKeyAndFun, ErrorFun) ->
        }).
 
 create_or_update_in_mnesia(XName, BindingKeyAndFun, ErrorFun) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               #?JMS_TOPIC_RECORD{x_selector_funs = BindingFuns} =
                   read_state_in_mnesia(XName, ErrorFun),
@@ -105,7 +105,7 @@ insert(XName, BFuns) ->
        }).
 
 insert_in_mnesia(XName, BFuns) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               write_state_fun_in_mnesia(XName, BFuns)
       end).
@@ -155,7 +155,7 @@ delete(XName) ->
        }).
 
 delete_in_mnesia(XName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() -> mnesia:delete(?JMS_TOPIC_TABLE, XName, write) end).
 
 delete_in_khepri(XName) ->
@@ -170,7 +170,7 @@ delete(XName, BindingKeys, ErrorFun) ->
        }).
 
 delete_in_mnesia(XName, BindingKeys, ErrorFun) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               #?JMS_TOPIC_RECORD{x_selector_funs = BindingFuns} =
                   read_state_in_mnesia(XName, ErrorFun),
diff --git a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
index 1e22e39aec..76f0b141ea 100644
--- a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
+++ b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
@@ -68,7 +68,7 @@ get(XName) ->
        }).
 
 get_in_mnesia(XName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() -> get_in_mnesia_tx(XName) end).
 
 get_in_khepri(XName) ->
@@ -99,7 +99,7 @@ insert(XName, Message, Length) ->
        }).
 
 insert_in_mnesia(XName, Message, Length) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun () ->
               Cached = get_in_mnesia_tx(XName),
               insert_in_mnesia(XName, Cached, Message, Length)
@@ -167,7 +167,7 @@ delete(XName) ->
        }).
 
 delete_in_mnesia(XName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_mnesia:execute_mnesia_transaction(
       fun() ->
               mnesia:delete(?RH_TABLE, XName, write)
       end).
@@ -225,4 +225,3 @@ khepri_recent_history_path() ->
 
 khepri_recent_history_path(#resource{virtual_host = VHost, name = Name}) ->
     [?MODULE, recent_history_exchange, VHost, Name].
->>>>>>> 7b61c2e402 (Khepri: The One Commit)