From 2f13bfd4527074d6a5868c406a531edda371079b Mon Sep 17 00:00:00 2001
From: Thales Macedo Garitezi <thalesmg@gmail.com>
Date: Wed, 4 Jan 2023 09:47:14 -0300
Subject: [PATCH 1/2] fix(retainer): change mnesia table storage types during
 update

https://emqx.atlassian.net/browse/EMQX-8650
---
 apps/emqx_retainer/src/emqx_retainer.app.src  |  2 +-
 apps/emqx_retainer/src/emqx_retainer.erl      | 23 ++++++---
 .../test/emqx_retainer_api_SUITE.erl          | 51 +++++++++++++++++++
 changes/v5.0.14-en.md                         |  2 +
 changes/v5.0.14-zh.md                         |  2 +
 5 files changed, 71 insertions(+), 9 deletions(-)

diff --git a/apps/emqx_retainer/src/emqx_retainer.app.src b/apps/emqx_retainer/src/emqx_retainer.app.src
index f61468d9b..d151ad4e7 100644
--- a/apps/emqx_retainer/src/emqx_retainer.app.src
+++ b/apps/emqx_retainer/src/emqx_retainer.app.src
@@ -2,7 +2,7 @@
 {application, emqx_retainer, [
     {description, "EMQX Retainer"},
     % strict semver, bump manually!
-    {vsn, "5.0.8"},
+    {vsn, "5.0.9"},
     {modules, []},
     {registered, [emqx_retainer_sup]},
     {applications, [kernel, stdlib, emqx]},
diff --git a/apps/emqx_retainer/src/emqx_retainer.erl b/apps/emqx_retainer/src/emqx_retainer.erl
index aa1260033..b81ea2446 100644
--- a/apps/emqx_retainer/src/emqx_retainer.erl
+++ b/apps/emqx_retainer/src/emqx_retainer.erl
@@ -321,16 +321,23 @@ update_config(
     OldConf
 ) ->
     #{
-        backend := BackendCfg,
+        backend := #{
+            type := BackendType,
+            storage_type := StorageType
+        },
         msg_clear_interval := ClearInterval
     } = NewConf,
 
-    #{backend := OldBackendCfg} = OldConf,
-
-    StorageType = maps:get(type, BackendCfg),
-    OldStrorageType = maps:get(type, OldBackendCfg),
-    case OldStrorageType of
-        StorageType ->
+    #{
+        backend := #{
+            type := OldBackendType,
+            storage_type := OldStorageType
+        }
+    } = OldConf,
+    SameBackendType = BackendType =:= OldBackendType,
+    SameStorageType = StorageType =:= OldStorageType,
+    case SameBackendType andalso SameStorageType of
+        true ->
             State#{
                 clear_timer := check_timer(
                     ClearTimer,
@@ -338,7 +345,7 @@ update_config(
                     clear_expired
                 )
             };
-        _ ->
+        false ->
             State2 = disable_retainer(State),
             enable_retainer(State2, NewConf)
     end.
diff --git a/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl b/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
index aee6aa4e4..a64e9a7df 100644
--- a/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
+++ b/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
@@ -31,6 +31,7 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
+    emqx_common_test_helpers:clear_screen(),
     application:load(emqx_conf),
     ok = ekka:start(),
     ok = mria_rlog:wait_for_shards([?CLUSTER_RPC_SHARD], infinity),
@@ -219,6 +220,56 @@ t_lookup_and_delete(_) ->
 
     ok = emqtt:disconnect(C1).
 
+t_change_storage_type(_Config) ->
+    Path = api_path(["mqtt", "retainer"]),
+    {ok, ConfJson} = request_api(get, Path),
+    RawConf = emqx_json:decode(ConfJson, [return_maps]),
+    %% pre-conditions
+    ?assertMatch(
+        #{
+            <<"backend">> := #{
+                <<"type">> := <<"built_in_database">>,
+                <<"storage_type">> := <<"ram">>
+            },
+            <<"enable">> := true
+        },
+        RawConf
+    ),
+    ?assertEqual(ram_copies, mnesia:table_info(?TAB_INDEX_META, storage_type)),
+    ?assertEqual(ram_copies, mnesia:table_info(?TAB_MESSAGE, storage_type)),
+    ?assertEqual(ram_copies, mnesia:table_info(?TAB_INDEX, storage_type)),
+
+    ChangedConf = emqx_map_lib:deep_merge(
+        RawConf,
+        #{
+            <<"backend">> =>
+                #{<<"storage_type">> => <<"disc">>}
+        }
+    ),
+    {ok, UpdateResJson} = request_api(
+        put,
+        Path,
+        [],
+        auth_header_(),
+        ChangedConf
+    ),
+    UpdatedRawConf = emqx_json:decode(UpdateResJson, [return_maps]),
+    ?assertMatch(
+        #{
+            <<"backend">> := #{
+                <<"type">> := <<"built_in_database">>,
+                <<"storage_type">> := <<"disc">>
+            },
+            <<"enable">> := true
+        },
+        UpdatedRawConf
+    ),
+    ?assertEqual(disc_copies, mnesia:table_info(?TAB_INDEX_META, storage_type)),
+    ?assertEqual(disc_copies, mnesia:table_info(?TAB_MESSAGE, storage_type)),
+    ?assertEqual(disc_copies, mnesia:table_info(?TAB_INDEX, storage_type)),
+
+    ok.
+
 %%--------------------------------------------------------------------
 %% HTTP Request
 %%--------------------------------------------------------------------
diff --git a/changes/v5.0.14-en.md b/changes/v5.0.14-en.md
index f61211bb8..35dd836bf 100644
--- a/changes/v5.0.14-en.md
+++ b/changes/v5.0.14-en.md
@@ -14,3 +14,5 @@
 - Fix an issue where testing the GCP PubSub could leak memory, and an issue where its JWT token would fail to refresh a second time. [#9641](https://github.com/emqx/emqx/pull/9641)
 
 - Fix the problem of data loss and bad match when the MySQL driver is disconnected [#9638](https://github.com/emqx/emqx/pull/9638).
+
+- Fixed an issue where changing the storage type of the built-in database retainer would not take effect without restarting the node [#9676](https://github.com/emqx/emqx/pull/9676).
diff --git a/changes/v5.0.14-zh.md b/changes/v5.0.14-zh.md
index c96bef305..f097b8559 100644
--- a/changes/v5.0.14-zh.md
+++ b/changes/v5.0.14-zh.md
@@ -14,3 +14,5 @@
 - 修复了测试GCP PubSub可能泄露内存的问题，以及其JWT令牌第二次刷新失败的问题。 [#9640](https://github.com/emqx/emqx/pull/9640)
 
 - 修复 MySQL 驱动断开连接时出现的数据丢失和匹配错误的问题 [#9638](https://github.com/emqx/emqx/pull/9638)。
+
+- 修复了如果不重新启动节点，改变保留消息的存储类型将不会生效的问题 [#9676](https://github.com/emqx/emqx/pull/9676)。

From 51ad27cb4bb9fe3539dc39a0f160582ee43a9fcb Mon Sep 17 00:00:00 2001
From: Thales Macedo Garitezi <thalesmg@gmail.com>
Date: Wed, 4 Jan 2023 12:11:35 -0300
Subject: [PATCH 2/2] test(retainer): assert that retained messages are not
 lost when changing storage type

---
 .../src/emqx_retainer_mnesia.erl              |  4 +-
 .../test/emqx_retainer_api_SUITE.erl          | 59 +++++++++++++++----
 2 files changed, 51 insertions(+), 12 deletions(-)

diff --git a/apps/emqx_retainer/src/emqx_retainer_mnesia.erl b/apps/emqx_retainer/src/emqx_retainer_mnesia.erl
index 69a6a877a..cadb9110f 100644
--- a/apps/emqx_retainer/src/emqx_retainer_mnesia.erl
+++ b/apps/emqx_retainer/src/emqx_retainer_mnesia.erl
@@ -146,7 +146,9 @@ store_retained(_, Msg = #message{topic = Topic}) ->
                 reason => table_is_full
             });
         false ->
-            do_store_retained(Msg, Tokens, ExpiryTime)
+            do_store_retained(Msg, Tokens, ExpiryTime),
+            ?tp(message_retained, #{topic => Topic}),
+            ok
     end.
 
 clear_expired(_) ->
diff --git a/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl b/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
index a64e9a7df..ba96887a2 100644
--- a/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
+++ b/apps/emqx_retainer/test/emqx_retainer_api_SUITE.erl
@@ -31,7 +31,6 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
-    emqx_common_test_helpers:clear_screen(),
     application:load(emqx_conf),
     ok = ekka:start(),
     ok = mria_rlog:wait_for_shards([?CLUSTER_RPC_SHARD], infinity),
@@ -104,11 +103,12 @@ t_messages(_) ->
     end,
 
     ?check_trace(
-        ?wait_async_action(
-            lists:foreach(Each, lists:seq(1, 5)),
-            #{?snk_kind := message_retained, topic := <<"retained/A">>},
-            500
-        ),
+        {ok, {ok, _}} =
+            ?wait_async_action(
+                lists:foreach(Each, lists:seq(1, 5)),
+                #{?snk_kind := message_retained, topic := <<"retained/A">>},
+                500
+            ),
         []
     ),
 
@@ -150,11 +150,12 @@ t_messages_page(_) ->
     end,
 
     ?check_trace(
-        ?wait_async_action(
-            lists:foreach(Each, lists:seq(1, 5)),
-            #{?snk_kind := message_retained, topic := <<"retained/A">>},
-            500
-        ),
+        {ok, {ok, _}} =
+            ?wait_async_action(
+                lists:foreach(Each, lists:seq(1, 5)),
+                #{?snk_kind := message_retained, topic := <<"retained/A">>},
+                500
+            ),
         []
     ),
     Page = 4,
@@ -238,6 +239,23 @@ t_change_storage_type(_Config) ->
     ?assertEqual(ram_copies, mnesia:table_info(?TAB_INDEX_META, storage_type)),
     ?assertEqual(ram_copies, mnesia:table_info(?TAB_MESSAGE, storage_type)),
     ?assertEqual(ram_copies, mnesia:table_info(?TAB_INDEX, storage_type)),
+    %% insert some retained messages
+    {ok, C0} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
+    {ok, _} = emqtt:connect(C0),
+    ok = snabbkaffe:start_trace(),
+    Topic = <<"retained">>,
+    Payload = <<"retained">>,
+    {ok, {ok, _}} =
+        ?wait_async_action(
+            emqtt:publish(C0, Topic, Payload, [{qos, 0}, {retain, true}]),
+            #{?snk_kind := message_retained, topic := Topic},
+            500
+        ),
+    emqtt:stop(C0),
+    ok = snabbkaffe:stop(),
+    {ok, MsgsJson0} = request_api(get, api_path(["mqtt", "retainer", "messages"])),
+    #{data := Msgs0, meta := _} = decode_json(MsgsJson0),
+    ?assertEqual(1, length(Msgs0)),
 
     ChangedConf = emqx_map_lib:deep_merge(
         RawConf,
@@ -267,6 +285,25 @@ t_change_storage_type(_Config) ->
     ?assertEqual(disc_copies, mnesia:table_info(?TAB_INDEX_META, storage_type)),
     ?assertEqual(disc_copies, mnesia:table_info(?TAB_MESSAGE, storage_type)),
     ?assertEqual(disc_copies, mnesia:table_info(?TAB_INDEX, storage_type)),
+    %% keep retained messages
+    {ok, MsgsJson1} = request_api(get, api_path(["mqtt", "retainer", "messages"])),
+    #{data := Msgs1, meta := _} = decode_json(MsgsJson1),
+    ?assertEqual(1, length(Msgs1)),
+    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
+    {ok, _} = emqtt:connect(C1),
+    {ok, _, _} = emqtt:subscribe(C1, Topic),
+
+    receive
+        {publish, #{topic := T, payload := P, retain := R}} ->
+            ?assertEqual(Payload, P),
+            ?assertEqual(Topic, T),
+            ?assert(R),
+            ok
+    after 500 ->
+        emqtt:stop(C1),
+        ct:fail("should have preserved retained messages")
+    end,
+    emqtt:stop(C1),
 
     ok.