feat(ds): Add a flag that forces all sessions to become durable

apps/emqx/src/emqx_persistent_message.erl

@@ -19,7 +19,7 @@
 -include("emqx.hrl").
 
 -export([init/0]).
--export([is_persistence_enabled/0]).
+-export([is_persistence_enabled/0, force_ds/0]).
 
 %% Message persistence
 -export([
@@ -54,6 +54,12 @@ is_persistence_enabled() ->
 storage_backend() ->
     storage_backend(emqx_config:get([session_persistence, storage])).
 
+%% Dev-only option: force all messages to go through
+%% `emqx_persistent_session_ds':
+-spec force_ds() -> boolean().
+force_ds() ->
+    emqx_config:get([session_persistence, force_ds]).
+
 storage_backend(#{
     builtin := #{enable := true, n_shards := NShards, replication_factor := ReplicationFactor}
 }) ->

apps/emqx/src/emqx_session.erl

@@ -626,12 +626,18 @@ choose_impl_candidates(#{expiry_interval := EI}) ->
 choose_impl_candidates(_, _IsPSStoreEnabled = false) ->
     [emqx_session_mem];
 choose_impl_candidates(0, _IsPSStoreEnabled = true) ->
-    %% NOTE
-    %% If ExpiryInterval is 0, the natural choice is `emqx_session_mem`. Yet we still
-    %% need to look the existing session up in the `emqx_persistent_session_ds` store
-    %% first, because previous connection may have set ExpiryInterval to a non-zero
-    %% value.
-    [emqx_session_mem, emqx_persistent_session_ds];
+    case emqx_persistent_message:force_ds() of
+        false ->
+            %% NOTE
+            %% If ExpiryInterval is 0, the natural choice is
+            %% `emqx_session_mem'. Yet we still need to look the
+            %% existing session up in the `emqx_persistent_session_ds'
+            %% store first, because previous connection may have set
+            %% ExpiryInterval to a non-zero value.
+            [emqx_session_mem, emqx_persistent_session_ds];
+        true ->
+            [emqx_persistent_session_ds]
+    end;
 choose_impl_candidates(EI, _IsPSStoreEnabled = true) when EI > 0 ->
     [emqx_persistent_session_ds].