feat(ds): move session data from mnesia to emqx_ds

apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl

@@ -14,6 +14,8 @@
 
 -import(emqx_common_test_helpers, [on_exit/1]).
 
+-define(SESSION_DB, emqx_persistent_session).
+
 %%------------------------------------------------------------------------------
 %% CT boilerplate
 %%------------------------------------------------------------------------------
@@ -25,6 +27,7 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
+    emqx_common_test_helpers:clear_screen(),
     TCApps = emqx_cth_suite:start(
         app_specs(),
         #{work_dir => emqx_cth_suite:work_dir(Config)}
@@ -649,19 +652,26 @@ t_session_replay_retry(_Config) ->
         ClientsPub
     ),
 
-    Pubs0 = emqx_common_test_helpers:wait_publishes(NClients, 5_000),
+    Pubs0 = emqx_common_test_helpers:wait_publishes(NClients, 6_000),
     NPubs = length(Pubs0),
     ?assertEqual(NClients, NPubs, ?drainMailbox(1_500)),
 
     ok = emqtt:stop(ClientSub),
 
     %% Make `emqx_ds` believe that roughly half of the shards are unavailable.
     ok = emqx_ds_test_helpers:mock_rpc_result(
-        fun(_Node, emqx_ds_replication_layer, _Function, [_DB, Shard | _]) ->
-            case erlang:phash2(Shard) rem 2 of
-                0 -> unavailable;
-                1 -> passthrough
-            end
+        fun
+            (_Node, emqx_ds_replication_layer, _Function, [?SESSION_DB | _]) ->
+                %% When storing session data in DS, failing here could lead to a session
+                %% being considered new and not loaded, since we read non-atomically from
+                %% the DB.  If getting the streams fail due to rpc problems, it leniently
+                %% returns an empty list of streams, and that results in a "new" session.
+                passthrough;
+            (_Node, emqx_ds_replication_layer, _Function, [_DB, Shard | _]) ->
+                case erlang:phash2(Shard) rem 2 of
+                    0 -> unavailable;
+                    1 -> passthrough
+                end
         end
     ),
 

apps/emqx/src/emqx_ds_schema.erl

@@ -80,13 +80,22 @@ namespace() ->
 schema() ->
     [
         {messages,
-            ds_schema(#{
+            ds_schema(ref(builtin), 'durable_storage.messages', #{
                 default =>
                     #{
                         <<"backend">> => builtin
                     },
                 importance => ?IMPORTANCE_MEDIUM,
                 desc => ?DESC(messages)
+            })},
+        {sessions,
+            ds_schema(ref(builtin_session), 'durable_storage.sessions', #{
+                default =>
+                    #{
+                        <<"backend">> => builtin
+                    },
+                importance => ?IMPORTANCE_MEDIUM,
+                desc => ?DESC(sessions)
             })}
     ].
 
@@ -179,6 +188,19 @@ fields(builtin) ->
                 }
             )}
     ];
+fields(builtin_session) ->
+    lists:map(
+        fun
+            ({layout = Key, Sc0}) ->
+                Override = #{type => hoconsc:union(builtin_layouts_session())},
+                {Key, hocon_schema:override(Sc0, Override)};
+            (Field) ->
+                Field
+        end,
+        fields(builtin)
+    );
+fields(layout_builtin_wildcard_optimized_session) ->
+    fields(layout_builtin_wildcard_optimized);
 fields(builtin_local_write_buffer) ->
     [
         {max_items,
@@ -253,36 +275,50 @@ fields(layout_builtin_reference) ->
 
 desc(builtin) ->
     ?DESC(builtin);
+desc(builtin_session) ->
+    ?DESC(builtin);
 desc(builtin_local_write_buffer) ->
     ?DESC(builtin_local_write_buffer);
 desc(layout_builtin_wildcard_optimized) ->
     ?DESC(layout_builtin_wildcard_optimized);
+desc(layout_builtin_wildcard_optimized_session) ->
+    ?DESC(layout_builtin_wildcard_optimized);
 desc(_) ->
     undefined.
 
 %%================================================================================
 %% Internal functions
 %%================================================================================
 
-ds_schema(Options) ->
+ds_schema(MainRef, InjectionPoint, Options) ->
     sc(
         hoconsc:union([
-            ref(builtin)
-            | emqx_schema_hooks:injection_point('durable_storage.backends', [])
+            MainRef
+            | emqx_schema_hooks:union_option_injections(InjectionPoint, [])
         ]),
         Options
     ).
 
 -ifndef(TEST).
 builtin_layouts() ->
     [ref(layout_builtin_wildcard_optimized)].
+
+builtin_layouts_session() ->
+    [ref(layout_builtin_wildcard_optimized_session)].
 -else.
 builtin_layouts() ->
     %% Reference layout stores everything in one stream, so it's not
     %% suitable for production use. However, it's very simple and
     %% produces a very predictabale replay order, which can be useful
     %% for testing and debugging:
     [ref(layout_builtin_wildcard_optimized), ref(layout_builtin_reference)].
+
+builtin_layouts_session() ->
+    %% Reference layout stores everything in one stream, so it's not
+    %% suitable for production use. However, it's very simple and
+    %% produces a very predictabale replay order, which can be useful
+    %% for testing and debugging:
+    [ref(layout_builtin_wildcard_optimized_session), ref(layout_builtin_reference)].
 -endif.
 
 sc(Type, Meta) -> hoconsc:mk(Type, Meta).

apps/emqx/src/emqx_persistent_message.erl

@@ -56,7 +56,7 @@ init() ->
         Backend = storage_backend(),
         ok = emqx_ds:open_db(?PERSISTENT_MESSAGE_DB, Backend),
         ok = emqx_persistent_session_ds_router:init_tables(),
-        ok = emqx_persistent_session_ds:create_tables(),
+        ok = initialize_session_ds_state(),
         ok
     end).
 
@@ -72,6 +72,15 @@ is_persistence_enabled(Zone) ->
 storage_backend() ->
     storage_backend([durable_storage, messages]).
 
+-ifdef(STORE_STATE_IN_DS).
+initialize_session_ds_state() ->
+    ok = emqx_persistent_session_ds_state:open_db(storage_backend([durable_storage, sessions])).
+-else.
+initialize_session_ds_state() ->
+    ok = emqx_persistent_session_ds_state:create_tables().
+%% -ifdef(STORE_STATE_IN_DS).
+-endif.
+
 %% Dev-only option: force all messages to go through
 %% `emqx_persistent_session_ds':
 -spec force_ds(emqx_types:zone()) -> boolean().

apps/emqx/src/emqx_persistent_session_ds.erl

@@ -79,7 +79,7 @@
 ]).
 
 %% session table operations
--export([create_tables/0, sync/1]).
+-export([sync/1]).
 
 %% internal export used by session GC process
 -export([destroy_session/1]).
@@ -674,9 +674,6 @@ list_client_subscriptions(ClientId) ->
 %% Session tables operations
 %%--------------------------------------------------------------------
 
-create_tables() ->
-    emqx_persistent_session_ds_state:create_tables().
-
 %% @doc Force syncing of the transient state to persistent storage
 sync(ClientId) ->
     case emqx_cm:lookup_channels(ClientId) of

apps/emqx/src/emqx_persistent_session_ds.hrl

@@ -18,12 +18,18 @@
 
 -define(PERSISTENT_MESSAGE_DB, emqx_persistent_message).
 
+-ifdef(STORE_STATE_IN_DS).
+-define(PERSISTENT_SESSION_DB, emqx_persistent_session).
+%% ELSE ifdef(STORE_STATE_IN_DS).
+-else.
 -define(SESSION_TAB, emqx_ds_session).
 -define(SESSION_SUBSCRIPTIONS_TAB, emqx_ds_session_subscriptions).
 -define(SESSION_STREAM_TAB, emqx_ds_stream_tab).
 -define(SESSION_PUBRANGE_TAB, emqx_ds_pubrange_tab).
 -define(SESSION_COMMITTED_OFFSET_TAB, emqx_ds_committed_offset_tab).
 -define(DS_MRIA_SHARD, emqx_ds_session_shard).
+%% END ifdef(STORE_STATE_IN_DS).
+-endif.
 
 %%%%% Session sequence numbers: