diff --git a/apps/emqx/src/emqx_persistent_session_ds.erl b/apps/emqx/src/emqx_persistent_session_ds.erl index 514cc5bcf..76b54e34a 100644 --- a/apps/emqx/src/emqx_persistent_session_ds.erl +++ b/apps/emqx/src/emqx_persistent_session_ds.erl @@ -70,6 +70,8 @@ do_ensure_all_iterators_closed/1 ]). +-export([print_session/1]). + -ifdef(TEST). -export([ session_open/1, @@ -226,6 +228,25 @@ info(await_rel_timeout, #{props := Conf}) -> stats(Session) -> info(?STATS_KEYS, Session). +%% Debug/troubleshooting +-spec print_session(emqx_types:client_id()) -> map() | undefined. +print_session(ClientId) -> + catch ro_transaction( + fun() -> + case mnesia:read(?SESSION_TAB, ClientId) of + [Session] -> + #{ + session => Session, + streams => mnesia:read(?SESSION_STREAM_TAB, ClientId), + pubranges => session_read_pubranges(ClientId), + subscriptions => session_read_subscriptions(ClientId) + }; + [] -> + undefined + end + end + ). + %%-------------------------------------------------------------------- %% Client -> Broker: SUBSCRIBE / UNSUBSCRIBE %%-------------------------------------------------------------------- @@ -563,7 +584,7 @@ session_drop(DSSessionId) -> -spec session_drop_subscriptions(id()) -> ok. session_drop_subscriptions(DSSessionId) -> - Subscriptions = session_read_subscriptions(DSSessionId), + Subscriptions = session_read_subscriptions(DSSessionId, write), lists:foreach( fun(#ds_sub{id = DSSubId} = DSSub) -> TopicFilter = subscription_id_to_topic_filter(DSSubId), @@ -626,13 +647,27 @@ session_del_subscription(DSSessionId, TopicFilter) -> session_del_subscription(#ds_sub{id = DSSubId}) -> mnesia:delete(?SESSION_SUBSCRIPTIONS_TAB, DSSubId, write). -session_read_subscriptions(DSSessionId) -> +session_read_subscriptions(DSSessionID) -> + session_read_subscriptions(DSSessionID, read). + +session_read_subscriptions(DSSessionId, LockKind) -> MS = ets:fun2ms( fun(Sub = #ds_sub{id = {Sess, _}}) when Sess =:= DSSessionId -> Sub end ), - mnesia:select(?SESSION_SUBSCRIPTIONS_TAB, MS, read). + mnesia:select(?SESSION_SUBSCRIPTIONS_TAB, MS, LockKind). + +session_read_pubranges(DSSessionID) -> + session_read_pubranges(DSSessionID, read). + +session_read_pubranges(DSSessionId, LockKind) -> + MS = ets:fun2ms( + fun(#ds_pubrange{id = {Sess, First}}) when Sess =:= DSSessionId -> + {DSSessionId, First} + end + ), + mnesia:select(?SESSION_PUBRANGE_TAB, MS, LockKind). -spec new_subscription_id(id(), topic_filter()) -> {subscription_id(), integer()}. new_subscription_id(DSSessionId, TopicFilter) -> @@ -735,12 +770,7 @@ session_drop_streams(DSSessionId) -> %% must be called inside a transaction -spec session_drop_pubranges(id()) -> ok. session_drop_pubranges(DSSessionId) -> - MS = ets:fun2ms( - fun(#ds_pubrange{id = {DSSessionId0, First}}) when DSSessionId0 =:= DSSessionId -> - {DSSessionId, First} - end - ), - RangeIds = mnesia:select(?SESSION_PUBRANGE_TAB, MS, write), + RangeIds = session_read_pubranges(DSSessionId, write), lists:foreach( fun(RangeId) -> mnesia:delete(?SESSION_PUBRANGE_TAB, RangeId, write)