fix(sessds): Stricter checks for PacketIds
This commit is contained in:
parent
eec56b0d6b
commit
2c6a776318
|
@ -909,30 +909,28 @@ update_seqno(Track, PacketId, Session = #{id := SessionId, s := S}) ->
|
||||||
SeqNo = packet_id_to_seqno(PacketId, S),
|
SeqNo = packet_id_to_seqno(PacketId, S),
|
||||||
case Track of
|
case Track of
|
||||||
puback ->
|
puback ->
|
||||||
MinTrack = ?committed(?QOS_1),
|
QoS = ?QOS_1,
|
||||||
MaxTrack = ?next(?QOS_1);
|
SeqNoKey = ?committed(?QOS_1);
|
||||||
pubrec ->
|
pubrec ->
|
||||||
MinTrack = ?rec,
|
QoS = ?QOS_2,
|
||||||
MaxTrack = ?next(?QOS_2);
|
SeqNoKey = ?rec;
|
||||||
pubcomp ->
|
pubcomp ->
|
||||||
MinTrack = ?committed(?QOS_2),
|
QoS = ?QOS_2,
|
||||||
MaxTrack = ?next(?QOS_2)
|
SeqNoKey = ?committed(?QOS_2)
|
||||||
end,
|
end,
|
||||||
Min = emqx_persistent_session_ds_state:get_seqno(MinTrack, S),
|
Current = emqx_persistent_session_ds_state:get_seqno(SeqNoKey, S),
|
||||||
Max = emqx_persistent_session_ds_state:get_seqno(MaxTrack, S),
|
case inc_seqno(QoS, Current) of
|
||||||
case Min =< SeqNo andalso SeqNo =< Max of
|
SeqNo ->
|
||||||
true ->
|
|
||||||
%% TODO: we pass a bogus message into the hook:
|
%% TODO: we pass a bogus message into the hook:
|
||||||
Msg = emqx_message:make(SessionId, <<>>, <<>>),
|
Msg = emqx_message:make(SessionId, <<>>, <<>>),
|
||||||
{ok, Msg, Session#{s => emqx_persistent_session_ds_state:put_seqno(MinTrack, SeqNo, S)}};
|
{ok, Msg, Session#{s => emqx_persistent_session_ds_state:put_seqno(SeqNoKey, SeqNo, S)}};
|
||||||
false ->
|
Expected ->
|
||||||
?SLOG(warning, #{
|
?SLOG(warning, #{
|
||||||
msg => "out-of-order_commit",
|
msg => "out-of-order_commit",
|
||||||
track => Track,
|
track => Track,
|
||||||
packet_id => PacketId,
|
packet_id => PacketId,
|
||||||
seqno => SeqNo,
|
seqno => SeqNo,
|
||||||
min => Min,
|
expected => Expected
|
||||||
max => Max
|
|
||||||
}),
|
}),
|
||||||
{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}
|
{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}
|
||||||
end.
|
end.
|
||||||
|
|
|
@ -495,7 +495,7 @@ consume(It) ->
|
||||||
end.
|
end.
|
||||||
|
|
||||||
receive_messages(Count) ->
|
receive_messages(Count) ->
|
||||||
receive_messages(Count, 5_000).
|
receive_messages(Count, 10_000).
|
||||||
|
|
||||||
receive_messages(Count, Timeout) ->
|
receive_messages(Count, Timeout) ->
|
||||||
lists:reverse(receive_messages(Count, [], Timeout)).
|
lists:reverse(receive_messages(Count, [], Timeout)).
|
||||||
|
|
Loading…
Reference in New Issue