Khepri: handle pairs Table-Key received during metadata store migration

dcorbacho · dcorbacho · commit 8c35c8efd88d · 2023-10-27T07:42:37.000+02:00
These are generated by Mnesia when there are writes in-flight,
but do not contain enough info to copy the record.
An event with Table-Record is generated at the same time, which is
properly processed after the initial copy of data.
diff --git a/deps/rabbit/app.bzl b/deps/rabbit/app.bzl
@@ -539,6 +539,7 @@ def all_srcs(name = "all_srcs"):
             "include/amqqueue.hrl",
             "include/amqqueue_v2.hrl",
             "include/gm_specs.hrl",
+            "include/internal_user.hrl",
             "include/mc.hrl",
             "include/rabbit_global_counters.hrl",
             "include/vhost.hrl",
diff --git a/deps/rabbit/include/internal_user.hrl b/deps/rabbit/include/internal_user.hrl
@@ -0,0 +1,4 @@
+-define(is_internal_user(U),
+        (?is_internal_user_v2(U))).
+
+-define(is_internal_user_v2(U), is_record(U, internal_user, 6)).
diff --git a/deps/rabbit/src/rabbit_db_binding_m2k_converter.erl b/deps/rabbit/src/rabbit_db_binding_m2k_converter.erl
@@ -87,7 +87,7 @@ copy_to_khepri(rabbit_route = Table,
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl b/deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl
@@ -45,7 +45,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(
-  rabbit_exchange = Table, Record,
+  rabbit_exchange = Table, #exchange{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     Name = Record#exchange.name,
     ?LOG_DEBUG(
@@ -82,7 +82,7 @@ copy_to_khepri(rabbit_exchange_serial = Table,
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl b/deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl
@@ -45,7 +45,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(
-  rabbit_node_maintenance_states = Table, Record,
+  rabbit_node_maintenance_states = Table, #node_maintenance_state{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     Name = Record#node_maintenance_state.node,
     ?LOG_DEBUG(
@@ -64,7 +64,7 @@ copy_to_khepri(
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_msup_m2k_converter.erl b/deps/rabbit/src/rabbit_db_msup_m2k_converter.erl
@@ -64,7 +64,7 @@ copy_to_khepri(mirrored_sup_childspec = Table,
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_queue_m2k_converter.erl b/deps/rabbit/src/rabbit_db_queue_m2k_converter.erl
@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("amqqueue.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -45,7 +46,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(rabbit_queue = Table, Record,
-               #?MODULE{store_id = StoreId} = State) ->
+               #?MODULE{store_id = StoreId} = State) when ?is_amqqueue(Record) ->
     Name = amqqueue:get_name(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",
@@ -63,7 +64,7 @@ copy_to_khepri(rabbit_queue = Table, Record,
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_rtparams_m2k_converter.erl b/deps/rabbit/src/rabbit_db_rtparams_m2k_converter.erl
@@ -63,7 +63,7 @@ copy_to_khepri(
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_user_m2k_converter.erl b/deps/rabbit/src/rabbit_db_user_m2k_converter.erl
@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("internal_user.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -46,7 +47,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 
 copy_to_khepri(
   rabbit_user = Table, Record,
-  #?MODULE{store_id = StoreId} = State) ->
+  #?MODULE{store_id = StoreId} = State) when ?is_internal_user(Record) ->
     Username = internal_user:get_username(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",
@@ -62,7 +63,7 @@ copy_to_khepri(
         Error -> Error
     end;
 copy_to_khepri(
-  rabbit_user_permission = Table, Record,
+  rabbit_user_permission = Table, #user_permission{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     #user_permission{
        user_vhost = #user_vhost{
@@ -89,7 +90,7 @@ copy_to_khepri(
         Error -> Error
     end;
 copy_to_khepri(
-  rabbit_topic_permission = Table, Record,
+  rabbit_topic_permission = Table, #topic_permission{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     #topic_permission{
        topic_permission_key =
@@ -122,7 +123,7 @@ copy_to_khepri(
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
diff --git a/deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl b/deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl
@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("vhost.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -46,7 +47,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 
 copy_to_khepri(
   rabbit_vhost = Table, Record,
-  #?MODULE{store_id = StoreId} = State) ->
+  #?MODULE{store_id = StoreId} = State) when ?is_vhost(Record) ->
     Name = vhost:get_name(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",
@@ -64,7 +65,7 @@ copy_to_khepri(
 copy_to_khepri(Table, Record, State) ->
     ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
                [Table, Record, State]),
-    {error, unexpected_record}.
+    {ok, State}.
 
 -spec delete_from_khepri(Table, Key, Priv) -> Ret when
       Table :: mnesia_to_khepri:mnesia_table(),
