yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(sessds): set replay retry timer if initial `replay/3` fails

This commit is contained in:
Andrew Mayorov 2024-03-07 12:45:10 +01:00
parent 09905d78cd
commit e7e8771277
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
1 changed files with 5 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
apps/emqx/src/emqx_persistent_session_ds.erl
View File

@ -478,14 +478,7 @@ handle_timeout(ClientInfo, ?TIMER_PULL, Session0) ->
Session = emqx_session:ensure_timer(?TIMER_PULL, Timeout, Session1), Session = emqx_session:ensure_timer(?TIMER_PULL, Timeout, Session1),
{ok, Publishes, Session}; {ok, Publishes, Session};
handle_timeout(ClientInfo, ?TIMER_RETRY_REPLAY, Session0) -> handle_timeout(ClientInfo, ?TIMER_RETRY_REPLAY, Session0) ->
Session1 = replay_streams(Session0, ClientInfo), Session = replay_streams(Session0, ClientInfo),
Session =
case ?IS_REPLAY_ONGOING(Session1) of
true ->
emqx_session:ensure_timer(?TIMER_RETRY_REPLAY, ?TIMEOUT_RETRY_REPLAY, Session1);
false ->
Session1
end,
{ok, [], Session}; {ok, [], Session};
handle_timeout(_ClientInfo, ?TIMER_GET_STREAMS, Session0 = #{s := S0}) -> handle_timeout(_ClientInfo, ?TIMER_GET_STREAMS, Session0 = #{s := S0}) ->
S1 = emqx_persistent_session_ds_subs:gc(S0), S1 = emqx_persistent_session_ds_subs:gc(S0),
@ -533,13 +526,15 @@ replay_streams(Session0 = #{replay := [{_StreamKey, Srs0} | Rest]}, ClientInfo)
Session = #{} -> Session = #{} ->
replay_streams(Session#{replay := Rest}, ClientInfo); replay_streams(Session#{replay := Rest}, ClientInfo);
{error, recoverable, Reason} -> {error, recoverable, Reason} ->
RetryTimeout = ?TIMEOUT_RETRY_REPLAY,
?SLOG(warning, #{ ?SLOG(warning, #{
msg => "failed_to_fetch_replay_batch", msg => "failed_to_fetch_replay_batch",
stream => Srs0, stream => Srs0,
reason => Reason, reason => Reason,
class => recoverable class => recoverable,
retry_in_ms => RetryTimeout
}), }),
Session0 emqx_session:ensure_timer(?TIMER_RETRY_REPLAY, RetryTimeout, Session0)
%% TODO: Handle unrecoverable errors. %% TODO: Handle unrecoverable errors.
end; end;
replay_streams(Session0 = #{replay := []}, _ClientInfo) -> replay_streams(Session0 = #{replay := []}, _ClientInfo) ->