chore(persistent_session): add description for ps db backend

apps/emqx/etc/emqx.conf

@@ -1646,6 +1646,12 @@ persistent_session_store {
     ## ValueType: Boolean
     ## Default: false
     enabled = false
+    ## Which database backend should be used
+    ##
+    ## @doc persistent_session_store.db_backend
+    ## ValueType: mnesia_ram | mnesia_disc
+    ## Default: mnesia_ram
+    db_backend = mnesia_ram
 
     ## How long are undelivered messages retained in the store
     ##

apps/emqx/etc/emqx_cloud/vm.args

@@ -116,3 +116,7 @@
 
 ## patches dir
 -pa {{ platform_data_dir }}/patches
+
+## Mnesia thresholds
+-mnesia dump_log_write_threshold 5000
+-mnesia dump_log_time_threshold 60000

apps/emqx/etc/emqx_edge/vm.args

@@ -114,3 +114,7 @@
 
 ## patches dir
 -pa {{ platform_data_dir }}/patches
+
+## Mnesia thresholds
+-mnesia dump_log_write_threshold 5000
+-mnesia dump_log_time_threshold 60000

apps/emqx/src/emqx_channel.erl

@@ -1179,20 +1179,27 @@ terminate(_, #channel{conn_state = idle}) -> ok;
 terminate(normal, Channel) ->
     run_terminate_hook(normal, Channel);
 terminate({shutdown, kicked}, Channel) ->
-    _ = emqx_persistent_session:persist(Channel#channel.clientinfo,
+    persist_if_session(Channel),
-                                        Channel#channel.conninfo,
-                                        Channel#channel.session),
     run_terminate_hook(kicked, Channel);
 terminate({shutdown, Reason}, Channel) when Reason =:= discarded;
                                             Reason =:= takeovered ->
     run_terminate_hook(Reason, Channel);
 terminate(Reason, Channel = #channel{will_msg = WillMsg}) ->
     (WillMsg =/= undefined) andalso publish_will_msg(WillMsg),
-    _ = emqx_persistent_session:persist(Channel#channel.clientinfo,
+    persist_if_session(Channel),
-                                        Channel#channel.conninfo,
-                                        Channel#channel.session),
     run_terminate_hook(Reason, Channel).
 
+persist_if_session(#channel{session = Session} = Channel) ->
+    case emqx_session:is_session(Session) of
+        true ->
+            _ = emqx_persistent_session:persist(Channel#channel.clientinfo,
+                                                Channel#channel.conninfo,
+                                                Channel#channel.session),
+            ok;
+        false ->
+            ok
+    end.
+
 run_terminate_hook(_Reason, #channel{session = undefined}) -> ok;
 run_terminate_hook(Reason, #channel{clientinfo = ClientInfo, session = Session}) ->
     emqx_session:terminate(ClientInfo, Reason, Session).

apps/emqx/src/emqx_cm.erl

@@ -58,7 +58,9 @@
         , lookup_channels/2
         ]).
 
--export([all_channels/0]).
+-export([ all_channels/0
+        , all_client_ids/0
+        ]).
 
 %% gen_server callbacks
 -export([ init/1
@@ -400,6 +402,11 @@ all_channels() ->
     Pat = [{{'_', '$1'}, [], ['$1']}],
     ets:select(?CHAN_TAB, Pat).
 
+all_client_ids() ->
+    Pat = [{{'$1', '_'}, [], ['$1']}],
+    ets:select(?CHAN_TAB, Pat).
+
+
 %% @doc Lookup channels.
 -spec(lookup_channels(emqx_types:clientid()) -> list(chan_pid())).
 lookup_channels(ClientId) ->

apps/emqx/src/emqx_persistent_session.erl

@@ -82,12 +82,17 @@
 init_db_backend() ->
     case is_store_enabled() of
         true  ->
+            Backend =
+                case emqx_config:get(?db_backend_key) of
+                    mnesia_ram  -> emqx_persistent_session_mnesia_ram_backend;
+                    mnesia_disc -> emqx_persistent_session_mnesia_disc_backend
+                end,
+            persistent_term:put(?db_backend_module, Backend),
+            Backend:create_tables(),
             ok = emqx_trie:create_session_trie(),
-            emqx_persistent_session_mnesia_backend:create_tables(),
-            persistent_term:put(?db_backend_key, emqx_persistent_session_mnesia_backend),
             ok;
         false ->
-            persistent_term:put(?db_backend_key, emqx_persistent_session_dummy_backend),
+            persistent_term:put(?db_backend_module, emqx_persistent_session_dummy_backend),
             ok
     end.
 

apps/emqx/src/emqx_persistent_session.hrl

@@ -14,10 +14,6 @@
 %% limitations under the License.
 %%--------------------------------------------------------------------
 
--define(SESSION_STORE, emqx_session_store).
--define(SESS_MSG_TAB, emqx_session_msg).
--define(MSG_TAB, emqx_persistent_msg).
-
 -record(session_store, { client_id        :: binary()
                        , expiry_interval  :: non_neg_integer()
                        , ts               :: non_neg_integer()
@@ -30,4 +26,5 @@
 -define(is_enabled_key, [persistent_session_store, enabled]).
 -define(msg_retain, [persistent_session_store, max_retain_undelivered]).
 
--define(db_backend, (persistent_term:get(?db_backend_key))).
+-define(db_backend_module, [persistent_session_store, db_backend_module]).
+-define(db_backend, (persistent_term:get(?db_backend_module))).

apps/emqx/src/emqx_persistent_session_mnesia_disc_backend.erl

@@ -14,11 +14,15 @@
 %% limitations under the License.
 %%--------------------------------------------------------------------
 
--module(emqx_persistent_session_mnesia_backend).
+-module(emqx_persistent_session_mnesia_disc_backend).
 
 -include("emqx.hrl").
 -include("emqx_persistent_session.hrl").
 
+-define(SESSION_STORE, emqx_session_store_disc).
+-define(SESS_MSG_TAB, emqx_session_msg_disc).
+-define(MSG_TAB, emqx_persistent_msg_disc).
+
 -export([ create_tables/0
         , first_message_id/0
         , next_message_id/1

apps/emqx/src/emqx_persistent_session_mnesia_ram_backend.erl

@@ -0,0 +1,114 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_persistent_session_mnesia_ram_backend).
+
+-include("emqx.hrl").
+-include("emqx_persistent_session.hrl").
+
+-define(SESSION_STORE, emqx_session_store_ram).
+-define(SESS_MSG_TAB, emqx_session_msg_ram).
+-define(MSG_TAB, emqx_persistent_msg_ram).
+
+-export([ create_tables/0
+        , first_message_id/0
+        , next_message_id/1
+        , delete_message/1
+        , first_session_message/0
+        , next_session_message/1
+        , delete_session_message/1
+        , put_session_store/1
+        , delete_session_store/1
+        , lookup_session_store/1
+        , put_session_message/1
+        , put_message/1
+        , get_message/1
+        , ro_transaction/1
+        ]).
+
+create_tables() ->
+    ok = mria:create_table(?SESSION_STORE, [
+                {type, set},
+                {rlog_shard, ?PERSISTENT_SESSION_SHARD},
+                {storage, ram_copies},
+                {record_name, session_store},
+                {attributes, record_info(fields, session_store)},
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]),
+
+    ok = mria:create_table(?SESS_MSG_TAB, [
+                {type, ordered_set},
+                {rlog_shard, ?PERSISTENT_SESSION_SHARD},
+                {storage, ram_copies},
+                {record_name, session_msg},
+                {attributes, record_info(fields, session_msg)},
+                {storage_properties, [{ets, [{read_concurrency, true},
+                                             {write_concurrency, true}]}]}]),
+
+    ok = mria:create_table(?MSG_TAB, [
+                {type, ordered_set},
+                {rlog_shard, ?PERSISTENT_SESSION_SHARD},
+                {storage, ram_copies},
+                {record_name, message},
+                {attributes, record_info(fields, message)},
+                {storage_properties, [{ets, [{read_concurrency, true},
+                                             {write_concurrency, true}]}]}]).
+
+first_session_message() ->
+    mnesia:dirty_first(?SESS_MSG_TAB).
+
+next_session_message(Key) ->
+    mnesia:dirty_next(?SESS_MSG_TAB, Key).
+
+first_message_id() ->
+    mnesia:dirty_first(?MSG_TAB).
+
+next_message_id(Key) ->
+    mnesia:dirty_next(?MSG_TAB, Key).
+
+delete_message(Key) ->
+    mria:dirty_delete(?MSG_TAB, Key).
+
+delete_session_message(Key) ->
+    mria:dirty_delete(?SESS_MSG_TAB, Key).
+
+put_session_store(SS) ->
+    mria:dirty_write(?SESSION_STORE, SS).
+
+delete_session_store(ClientID) ->
+    mria:dirty_delete(?SESSION_STORE, ClientID).
+
+lookup_session_store(ClientID) ->
+    case mnesia:dirty_read(?SESSION_STORE, ClientID) of
+        [] -> none;
+        [SS] -> {value, SS}
+    end.
+
+put_session_message(SessMsg) ->
+    mria:dirty_write(?SESS_MSG_TAB, SessMsg).
+
+put_message(Msg) ->
+    mria:dirty_write(?MSG_TAB, Msg).
+
+get_message(MsgId) ->
+    case mnesia:read(?MSG_TAB, MsgId) of
+        [] -> error({msg_not_found, MsgId});
+        [Msg] -> Msg
+    end.
+
+ro_transaction(Fun) ->
+    {atomic, Res} = mria:ro_transaction(?PERSISTENT_SESSION_SHARD, Fun),
+    Res.
+

apps/emqx/src/emqx_schema.erl

@@ -160,6 +160,18 @@ fields("persistent_session_store") ->
        sc(boolean(),
           #{ default => "false"
            })},
+      {"db_backend",
+       sc(hoconsc:union([mnesia_ram, mnesia_disc]),
+          #{ default => "mnesia_ram"
+           , desc => """
+Which database backend should be used for persistent session storage.<br>
+mnesia_ram: Mnesia table, RAM only.<br>
+mnesia_disc: Mnesia table, in RAM but also copy to disc.<br>
+When running in a single node, it's recommended to have the session states copied to disc.<br>
+When clustered, session states are replicated to all the nodes in the cluster,
+most of the time, RAM only is good enough.
+"""
+           })},
       {"max_retain_undelivered",
        sc(duration(),
           #{ default => "1h"

apps/emqx/src/emqx_session.erl

@@ -58,6 +58,7 @@
 
 -export([ info/1
         , info/2
+        , is_session/1
         , stats/1
         ]).
 
@@ -202,6 +203,9 @@ init(Opts) ->
 %% Info, Stats
 %%--------------------------------------------------------------------
 
+is_session(#session{}) -> true;
+is_session(_) -> false.
+
 %% @doc Get infos of the session.
 -spec(info(session()) -> emqx_types:infos()).
 info(Session) ->

apps/emqx/src/emqx_trie.erl

@@ -17,6 +17,7 @@
 -module(emqx_trie).
 
 -include("emqx.hrl").
+-include("emqx_persistent_session.hrl").
 
 %% Mnesia bootstrap
 -export([ mnesia/1
@@ -48,7 +49,8 @@
 -endif.
 
 -define(TRIE, emqx_trie).
--define(SESSION_TRIE, emqx_session_trie).
+-define(SESSION_TRIE_RAM, emqx_session_trie_ram).
+-define(SESSION_TRIE_DISC, emqx_session_trie_disc).
 -define(PREFIX(Prefix), {Prefix, 0}).
 -define(TOPIC(Topic), {Topic, 1}).
 
@@ -57,6 +59,12 @@
         , count = 0 :: non_neg_integer()
         }).
 
+-define(ACTIVE_SESSION_TRIE, case emqx_config:get(?db_backend_key) of
+                                 mnesia_ram  -> ?SESSION_TRIE_RAM;
+                                 mnesia_disc -> ?SESSION_TRIE_DISC
+                             end).
+
+
 %%--------------------------------------------------------------------
 %% Mnesia bootstrap
 %%--------------------------------------------------------------------
@@ -79,7 +87,14 @@ create_session_trie() ->
     StoreProps = [{ets, [{read_concurrency, true},
                          {write_concurrency, true}
                         ]}],
-    ok = mria:create_table(?SESSION_TRIE,
+    ok = mria:create_table(?SESSION_TRIE_RAM,
+                           [{rlog_shard, ?ROUTE_SHARD},
+                            {storage, ram_copies},
+                            {record_name, ?TRIE},
+                            {attributes, record_info(fields, ?TRIE)},
+                            {type, ordered_set},
+                            {storage_properties, StoreProps}]),
+    ok = mria:create_table(?SESSION_TRIE_DISC,
                            [{rlog_shard, ?ROUTE_SHARD},
                             {storage, disc_copies},
                             {record_name, ?TRIE},
@@ -87,6 +102,7 @@ create_session_trie() ->
                             {type, ordered_set},
                             {storage_properties, StoreProps}]).
 
+
 %%--------------------------------------------------------------------
 %% Topics APIs
 %%--------------------------------------------------------------------
@@ -98,7 +114,7 @@ insert(Topic) when is_binary(Topic) ->
 
 -spec(insert_session(emqx_topic:topic()) -> ok).
 insert_session(Topic) when is_binary(Topic) ->
-    insert(Topic, ?SESSION_TRIE).
+    insert(Topic, ?ACTIVE_SESSION_TRIE).
 
 insert(Topic, Trie) when is_binary(Topic) ->
     {TopicKey, PrefixKeys} = make_keys(Topic),
@@ -115,7 +131,7 @@ delete(Topic) when is_binary(Topic) ->
 %% @doc Delete a topic filter from the trie.
 -spec(delete_session(emqx_topic:topic()) -> ok).
 delete_session(Topic) when is_binary(Topic) ->
-    delete(Topic, ?SESSION_TRIE).
+    delete(Topic, ?ACTIVE_SESSION_TRIE).
 
 delete(Topic, Trie) when is_binary(Topic) ->
     {TopicKey, PrefixKeys} = make_keys(Topic),
@@ -131,7 +147,7 @@ match(Topic) when is_binary(Topic) ->
 
 -spec(match_session(emqx_topic:topic()) -> list(emqx_topic:topic())).
 match_session(Topic) when is_binary(Topic) ->
-    match(Topic, ?SESSION_TRIE).
+    match(Topic, ?ACTIVE_SESSION_TRIE).
 
 match(Topic, Trie) when is_binary(Topic) ->
     Words = emqx_topic:words(Topic),
@@ -154,7 +170,7 @@ match(Topic, Trie) when is_binary(Topic) ->
 empty() -> empty(?TRIE).
 
 empty_session() ->
-    empty(?SESSION_TRIE).
+    empty(?ACTIVE_SESSION_TRIE).
 
 empty(Trie) -> ets:first(Trie) =:= '$end_of_table'.
 
@@ -164,7 +180,7 @@ lock_tables() ->
 
 -spec lock_session_tables() -> ok.
 lock_session_tables() ->
-    mnesia:write_lock_table(?SESSION_TRIE).
+    mnesia:write_lock_table(?ACTIVE_SESSION_TRIE).
 
 %%--------------------------------------------------------------------
 %% Internal functions

apps/emqx/test/emqx_persistent_session_SUITE.erl

@@ -113,6 +113,8 @@ init_per_group(snabbkaffe, Config) ->
     [ {kill_connection_process, true} | Config];
 init_per_group(gc_tests, Config) ->
     %% We need to make sure the system does not interfere with this test group.
+    [maybe_kill_connection_process(ClientId, [{kill_connection_process, true}])
+     || ClientId <- emqx_cm:all_client_ids()],
     emqx_common_test_helpers:stop_apps([]),
     SessionMsgEts = gc_tests_session_store,
     MsgEts = gc_tests_msg_store,
@@ -122,15 +124,15 @@ init_per_group(gc_tests, Config) ->
                         receive stop -> ok end
                 end),
     meck:new(mnesia, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(mnesia, dirty_first, fun(?SESS_MSG_TAB) -> ets:first(SessionMsgEts);
+    meck:expect(mnesia, dirty_first, fun(emqx_session_msg_ram) -> ets:first(SessionMsgEts);
-                                        (?MSG_TAB) -> ets:first(MsgEts);
+                                        (emqx_persistent_msg_ram) -> ets:first(MsgEts);
                                         (X) -> meck:passthrough(X)
                                      end),
-    meck:expect(mnesia, dirty_next, fun(?SESS_MSG_TAB, X) -> ets:next(SessionMsgEts, X);
+    meck:expect(mnesia, dirty_next, fun(emqx_session_msg_ram, X) -> ets:next(SessionMsgEts, X);
-                                       (?MSG_TAB, X) -> ets:next(MsgEts, X);
+                                       (emqx_persistent_msg_ram, X) -> ets:next(MsgEts, X);
                                        (Tab, X) -> meck:passthrough([Tab, X])
                                     end),
-    meck:expect(mnesia, dirty_delete, fun(?MSG_TAB, X) -> ets:delete(MsgEts, X);
+    meck:expect(mnesia, dirty_delete, fun(emqx_persistent_msg_ram, X) -> ets:delete(MsgEts, X);
                                          (Tab, X) -> meck:passthrough([Tab, X])
                                       end),
     [{store_owner, Pid}, {session_msg_store, SessionMsgEts}, {msg_store, MsgEts} | Config].
@@ -230,50 +232,70 @@ receive_messages(Count, Msgs) ->
 maybe_kill_connection_process(ClientId, Config) ->
     case ?config(kill_connection_process, Config) of
         true ->
-            [ConnectionPid] = emqx_cm:lookup_channels(ClientId),
+            case emqx_cm:lookup_channels(ClientId) of
-            ?assert(is_pid(ConnectionPid)),
+                [] ->
-            Ref = monitor(process, ConnectionPid),
+                    ok;
-            ConnectionPid ! die_if_test,
+                [ConnectionPid] ->
-            receive {'DOWN', Ref, process, ConnectionPid, normal} -> ok
+                    ?assert(is_pid(ConnectionPid)),
-            after 3000 -> error(process_did_not_die)
+                    Ref = monitor(process, ConnectionPid),
+                    ConnectionPid ! die_if_test,
+                    receive {'DOWN', Ref, process, ConnectionPid, normal} -> ok
+                    after 3000 -> error(process_did_not_die)
+                    end,
+                    wait_for_cm_unregister(ClientId)
             end;
         false ->
             ok
     end.
 
-snabbkaffe_sync_publish(Topic, Payloads, Config) ->
+wait_for_cm_unregister(ClientId) ->
+    wait_for_cm_unregister(ClientId, 10).
+
+wait_for_cm_unregister(_ClientId, 0) ->
+    error(cm_did_not_unregister);
+wait_for_cm_unregister(ClientId, N) ->
+    case emqx_cm:lookup_channels(ClientId) of
+        [] -> ok;
+        [_] -> timer:sleep(100), wait_for_cm_unregister(ClientId, N - 1)
+    end.
+
+snabbkaffe_sync_publish(Topic, Payloads) ->
     Fun = fun(Client, Payload) ->
                   ?wait_async_action( {ok, _} = emqtt:publish(Client, Topic, Payload, 2)
                                     , #{?snk_kind := ps_persist_msg, payload := Payload}
                                     )
           end,
-    do_publish(Payloads, Fun, Config).
+    do_publish(Payloads, Fun, true).
 
-publish(Topic, Payloads, Config) ->
+publish(Topic, Payloads) ->
     Fun = fun(Client, Payload) ->
                   {ok, _} = emqtt:publish(Client, Topic, Payload, 2)
           end,
-    do_publish(Payloads, Fun, Config).
+    do_publish(Payloads, Fun, false).
 
-do_publish(Payloads = [_|_], PublishFun, Config) ->
+do_publish(Payloads = [_|_], PublishFun, WaitForUnregister) ->
     %% Publish from another process to avoid connection confusion.
     {Pid, Ref} =
         spawn_monitor(
           fun() ->
                   %% For convenience, always publish using tcp.
                   %% The publish path is not what we are testing.
+                  ClientID = <<"ps_SUITE_publisher">>,
                   {ok, Client} = emqtt:start_link([ {proto_ver, v5}
+                                                  , {clientid, ClientID}
                                                   , {port, 1883} ]),
                   {ok, _} = emqtt:connect(Client),
                   lists:foreach(fun(Payload) -> PublishFun(Client, Payload) end, Payloads),
-                  ok = emqtt:disconnect(Client)
+                  ok = emqtt:disconnect(Client),
+                  %% Snabbkaffe sometimes fails unless all processes are gone.
+                  [wait_for_cm_unregister(ClientID) || WaitForUnregister]
           end),
     receive
         {'DOWN', Ref, process, Pid, normal} -> ok;
         {'DOWN', Ref, process, Pid, What} -> error({failed_publish, What})
     end;
-do_publish(Payload, PublishFun, Config) ->
+do_publish(Payload, PublishFun, WaitForUnregister) ->
-    do_publish([Payload], PublishFun, Config).
+    do_publish([Payload], PublishFun, WaitForUnregister).
 
 %%--------------------------------------------------------------------
 %% Test Cases
@@ -297,7 +319,7 @@ t_connect_session_expiry_interval(Config) ->
 
     maybe_kill_connection_process(ClientId, Config),
 
-    publish(Topic, Payload, Config),
+    publish(Topic, Payload),
 
     {ok, Client2} = emqtt:start_link([ {clientid, ClientId},
                                        {proto_ver, v5},
@@ -424,7 +446,7 @@ t_process_dies_session_expires(Config) ->
 
     maybe_kill_connection_process(ClientId, Config),
 
-    ok = publish(Topic, [Payload], Config),
+    ok = publish(Topic, [Payload]),
 
     SessionId =
         case ?config(persistent_store_enabled, Config) of
@@ -498,7 +520,7 @@ t_publish_while_client_is_gone(Config) ->
     ok = emqtt:disconnect(Client1),
     maybe_kill_connection_process(ClientId, Config),
 
-    ok = publish(Topic, [Payload1, Payload2], Config),
+    ok = publish(Topic, [Payload1, Payload2]),
 
     {ok, Client2} = emqtt:start_link([ {proto_ver, v5},
                                        {clientid, ClientId},
@@ -544,7 +566,7 @@ t_clean_start_drops_subscriptions(Config) ->
     maybe_kill_connection_process(ClientId, Config),
 
     %% 2.
-    ok = publish(Topic, Payload1, Config),
+    ok = publish(Topic, Payload1),
 
     %% 3.
     {ok, Client2} = emqtt:start_link([ {proto_ver, v5},
@@ -556,7 +578,7 @@ t_clean_start_drops_subscriptions(Config) ->
     ?assertEqual(0, client_info(session_present, Client2)),
     {ok, _, [2]} = emqtt:subscribe(Client2, STopic, qos2),
 
-    ok = publish(Topic, Payload2, Config),
+    ok = publish(Topic, Payload2),
     [Msg1] = receive_messages(1),
     ?assertEqual({ok, iolist_to_binary(Payload2)}, maps:find(payload, Msg1)),
 
@@ -571,7 +593,7 @@ t_clean_start_drops_subscriptions(Config) ->
                                      | Config]),
     {ok, _} = emqtt:ConnFun(Client3),
 
-    ok = publish(Topic, Payload3, Config),
+    ok = publish(Topic, Payload3),
     [Msg2] = receive_messages(1),
     ?assertEqual({ok, iolist_to_binary(Payload3)}, maps:find(payload, Msg2)),
 
@@ -625,7 +647,7 @@ t_multiple_subscription_matches(Config) ->
 
     maybe_kill_connection_process(ClientId, Config),
 
-    publish(Topic, Payload, Config),
+    publish(Topic, Payload),
 
     {ok, Client2} = emqtt:start_link([ {clientid, ClientId},
                                        {proto_ver, v5},
@@ -675,9 +697,9 @@ t_lost_messages_because_of_gc(Config) ->
     {ok, _, [2]} = emqtt:subscribe(Client1, STopic, qos2),
     emqtt:disconnect(Client1),
     maybe_kill_connection_process(ClientId, Config),
-    publish(Topic, Payload1, Config),
+    publish(Topic, Payload1),
     timer:sleep(2 * Retain),
-    publish(Topic, Payload2, Config),
+    publish(Topic, Payload2),
     emqx_persistent_session_gc:message_gc_worker(),
     {ok, Client2} = emqtt:start_link([ {clientid, ClientId},
                                        {clean_start, false},
@@ -790,7 +812,7 @@ t_snabbkaffe_pending_messages(Config) ->
 
     ?check_trace(
        begin
-           snabbkaffe_sync_publish(Topic, Payloads, Config),
+           snabbkaffe_sync_publish(Topic, Payloads),
            {ok, Client2} = emqtt:start_link([{clean_start, false} | EmqttOpts]),
            {ok, _} = emqtt:ConnFun(Client2),
            Msgs = receive_messages(length(Payloads)),
@@ -829,7 +851,7 @@ t_snabbkaffe_buffered_messages(Config) ->
     ok = emqtt:disconnect(Client1),
     maybe_kill_connection_process(ClientId, Config),
 
-    publish(Topic, Payloads1, Config),
+    publish(Topic, Payloads1),
 
     ?check_trace(
        begin
@@ -838,7 +860,7 @@ t_snabbkaffe_buffered_messages(Config) ->
                             #{ ?snk_kind := ps_resume_end }),
            spawn_link(fun() ->
                               ?block_until(#{ ?snk_kind := ps_marker_pendings_msgs }, infinity, 5000),
-                              publish(Topic, Payloads2, Config)
+                              publish(Topic, Payloads2)
                       end),
            {ok, Client2} = emqtt:start_link([{clean_start, false} | EmqttOpts]),
            {ok, _} = emqtt:ConnFun(Client2),