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

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.

Author: dcorbacho

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(),

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(),

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(),

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(),

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(),

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(),

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(),

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(),