0217 improve frame too large logging context

apps/emqx/src/emqx_channel.erl

@@ -540,13 +540,17 @@ handle_in(?AUTH_PACKET(), Channel) ->
     handle_out(disconnect, ?RC_IMPLEMENTATION_SPECIFIC_ERROR, Channel);
 handle_in({frame_error, Reason}, Channel = #channel{conn_state = idle}) ->
     shutdown(shutdown_count(frame_error, Reason), Channel);
-handle_in({frame_error, frame_too_large}, Channel = #channel{conn_state = connecting}) ->
+handle_in(
+    {frame_error, #{cause := frame_too_large} = R}, Channel = #channel{conn_state = connecting}
+) ->
     shutdown(
-        shutdown_count(frame_error, frame_too_large), ?CONNACK_PACKET(?RC_PACKET_TOO_LARGE), Channel
+        shutdown_count(frame_error, R), ?CONNACK_PACKET(?RC_PACKET_TOO_LARGE), Channel
     );
 handle_in({frame_error, Reason}, Channel = #channel{conn_state = connecting}) ->
     shutdown(shutdown_count(frame_error, Reason), ?CONNACK_PACKET(?RC_MALFORMED_PACKET), Channel);
-handle_in({frame_error, frame_too_large}, Channel = #channel{conn_state = ConnState}) when
+handle_in(
+    {frame_error, #{cause := frame_too_large}}, Channel = #channel{conn_state = ConnState}
+) when
     ConnState =:= connected orelse ConnState =:= reauthenticating
 ->
     handle_out(disconnect, {?RC_PACKET_TOO_LARGE, frame_too_large}, Channel);
@@ -2327,6 +2331,8 @@ shutdown(Reason, Reply, Packet, Channel) ->
 
 %% process exits with {shutdown, #{shutdown_count := Kind}} will trigger
 %% the connection supervisor (esockd) to keep a shutdown-counter grouped by Kind
+shutdown_count(_Kind, #{cause := Cause} = Reason) when is_atom(Cause) ->
+    Reason#{shutdown_count => Cause};
 shutdown_count(Kind, Reason) when is_map(Reason) ->
     Reason#{shutdown_count => Kind};
 shutdown_count(Kind, Reason) ->

apps/emqx/src/emqx_frame.erl

@@ -168,7 +168,7 @@ parse_remaining_len(Rest, Header, Options) ->
 parse_remaining_len(_Bin, _Header, _Multiplier, Length, #{max_size := MaxSize}) when
     Length > MaxSize
 ->
-    ?PARSE_ERR(frame_too_large);
+    ?PARSE_ERR(#{cause => frame_too_large, limit => MaxSize, received => Length});
 parse_remaining_len(<<>>, Header, Multiplier, Length, Options) ->
     {more, {{len, #{hdr => Header, len => {Multiplier, Length}}}, Options}};
 %% Match DISCONNECT without payload
@@ -189,12 +189,12 @@ parse_remaining_len(
 parse_remaining_len(
     <<0:8, _Rest/binary>>, _Header = #mqtt_packet_header{type = ?PINGRESP}, 1, 0, _Options
 ) ->
-    ?PARSE_ERR(#{hint => unexpected_packet, header_type => 'PINGRESP'});
+    ?PARSE_ERR(#{cause => unexpected_packet, header_type => 'PINGRESP'});
 %% All other types of messages should not have a zero remaining length.
 parse_remaining_len(
     <<0:8, _Rest/binary>>, Header, 1, 0, _Options
 ) ->
-    ?PARSE_ERR(#{hint => zero_remaining_len, header_type => Header#mqtt_packet_header.type});
+    ?PARSE_ERR(#{cause => zero_remaining_len, header_type => Header#mqtt_packet_header.type});
 %% Match PUBACK, PUBREC, PUBREL, PUBCOMP, UNSUBACK...
 parse_remaining_len(<<0:1, 2:7, Rest/binary>>, Header, 1, 0, Options) ->
     parse_frame(Rest, Header, 2, Options);
@@ -213,7 +213,7 @@ parse_remaining_len(
 ) ->
     FrameLen = Value + Len * Multiplier,
     case FrameLen > MaxSize of
-        true -> ?PARSE_ERR(frame_too_large);
+        true -> ?PARSE_ERR(#{cause => frame_too_large, limit => MaxSize, received => FrameLen});
         false -> parse_frame(Rest, Header, FrameLen, Options)
     end.
 
@@ -267,7 +267,7 @@ packet(Header, Variable, Payload) ->
     #mqtt_packet{header = Header, variable = Variable, payload = Payload}.
 
 parse_connect(FrameBin, StrictMode) ->
-    {ProtoName, Rest} = parse_utf8_string_with_hint(FrameBin, StrictMode, invalid_proto_name),
+    {ProtoName, Rest} = parse_utf8_string_with_cause(FrameBin, StrictMode, invalid_proto_name),
     case ProtoName of
         <<"MQTT">> ->
             ok;
@@ -277,7 +277,7 @@ parse_connect(FrameBin, StrictMode) ->
             %% from spec: the server MAY send disconnect with reason code 0x84
             %% we chose to close socket because the client is likely not talking MQTT anyway
             ?PARSE_ERR(#{
-                hint => invalid_proto_name,
+                cause => invalid_proto_name,
                 expected => <<"'MQTT' or 'MQIsdp'">>,
                 received => ProtoName
             })
@@ -296,7 +296,7 @@ parse_connect2(
         1 -> ?PARSE_ERR(reserved_connect_flag)
     end,
     {Properties, Rest3} = parse_properties(Rest2, ProtoVer, StrictMode),
-    {ClientId, Rest4} = parse_utf8_string_with_hint(Rest3, StrictMode, invalid_clientid),
+    {ClientId, Rest4} = parse_utf8_string_with_cause(Rest3, StrictMode, invalid_clientid),
     ConnPacket = #mqtt_packet_connect{
         proto_name = ProtoName,
         proto_ver = ProtoVer,
@@ -315,25 +315,29 @@ parse_connect2(
     {Username, Rest6} = parse_optional(
         Rest5,
         fun(Bin) ->
-            parse_utf8_string_with_hint(Bin, StrictMode, invalid_username)
+            parse_utf8_string_with_cause(Bin, StrictMode, invalid_username)
         end,
         bool(UsernameFlag)
     ),
     {Password, Rest7} = parse_optional(
         Rest6,
         fun(Bin) ->
-            parse_utf8_string_with_hint(Bin, StrictMode, invalid_password)
+            parse_utf8_string_with_cause(Bin, StrictMode, invalid_password)
         end,
         bool(PasswordFlag)
     ),
     case Rest7 of
         <<>> ->
             ConnPacket1#mqtt_packet_connect{username = Username, password = Password};
         _ ->
-            ?PARSE_ERR(malformed_connect_data)
+            ?PARSE_ERR(#{
+                cause => malformed_connect,
+                unexpected_trailing_bytes => size(Rest7)
+            })
     end;
-parse_connect2(_ProtoName, _, _) ->
-    ?PARSE_ERR(malformed_connect_header).
+parse_connect2(_ProtoName, Bin, _StrictMode) ->
+    %% sent less than 32 bytes
+    ?PARSE_ERR(#{cause => malformed_connect, header_bytes => Bin}).
 
 parse_packet(
     #mqtt_packet_header{type = ?CONNECT},
@@ -362,7 +366,7 @@ parse_packet(
     Bin,
     #{strict_mode := StrictMode, version := Ver}
 ) ->
-    {TopicName, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_topic),
+    {TopicName, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_topic),
     {PacketId, Rest1} =
         case QoS of
             ?QOS_0 -> {undefined, Rest};
@@ -474,7 +478,7 @@ parse_packet(
     {Properties, <<>>} = parse_properties(Rest, ?MQTT_PROTO_V5, StrictMode),
     #mqtt_packet_auth{reason_code = ReasonCode, properties = Properties};
 parse_packet(Header, _FrameBin, _Options) ->
-    ?PARSE_ERR(#{hint => malformed_packet, header_type => Header#mqtt_packet_header.type}).
+    ?PARSE_ERR(#{cause => malformed_packet, header_type => Header#mqtt_packet_header.type}).
 
 parse_will_message(
     Packet = #mqtt_packet_connect{
@@ -485,8 +489,8 @@ parse_will_message(
     StrictMode
 ) ->
     {Props, Rest} = parse_properties(Bin, Ver, StrictMode),
-    {Topic, Rest1} = parse_utf8_string_with_hint(Rest, StrictMode, invalid_topic),
-    {Payload, Rest2} = parse_binary_data(Rest1),
+    {Topic, Rest1} = parse_utf8_string_with_cause(Rest, StrictMode, invalid_topic),
+    {Payload, Rest2} = parse_will_payload(Rest1),
     {
         Packet#mqtt_packet_connect{
             will_props = Props,
@@ -518,7 +522,7 @@ parse_properties(Bin, ?MQTT_PROTO_V5, StrictMode) ->
             {parse_property(PropsBin, #{}, StrictMode), Rest1};
         _ ->
             ?PARSE_ERR(#{
-                hint => user_property_not_enough_bytes,
+                cause => user_property_not_enough_bytes,
                 parsed_key_length => Len,
                 remaining_bytes_length => byte_size(Rest)
             })
@@ -531,10 +535,10 @@ parse_property(<<16#01, Val, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#02, Val:32/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Message-Expiry-Interval' => Val}, StrictMode);
 parse_property(<<16#03, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_content_type),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_content_type),
     parse_property(Rest, Props#{'Content-Type' => Val}, StrictMode);
 parse_property(<<16#08, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_response_topic),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_response_topic),
     parse_property(Rest, Props#{'Response-Topic' => Val}, StrictMode);
 parse_property(<<16#09, Len:16/big, Val:Len/binary, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Correlation-Data' => Val}, StrictMode);
@@ -544,12 +548,12 @@ parse_property(<<16#0B, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#11, Val:32/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Session-Expiry-Interval' => Val}, StrictMode);
 parse_property(<<16#12, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_assigned_client_id),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_assigned_client_id),
     parse_property(Rest, Props#{'Assigned-Client-Identifier' => Val}, StrictMode);
 parse_property(<<16#13, Val:16, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Server-Keep-Alive' => Val}, StrictMode);
 parse_property(<<16#15, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_authn_method),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_authn_method),
     parse_property(Rest, Props#{'Authentication-Method' => Val}, StrictMode);
 parse_property(<<16#16, Len:16/big, Val:Len/binary, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Authentication-Data' => Val}, StrictMode);
@@ -560,13 +564,13 @@ parse_property(<<16#18, Val:32, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#19, Val, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Request-Response-Information' => Val}, StrictMode);
 parse_property(<<16#1A, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_response_info),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_response_info),
     parse_property(Rest, Props#{'Response-Information' => Val}, StrictMode);
 parse_property(<<16#1C, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_server_reference),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_server_reference),
     parse_property(Rest, Props#{'Server-Reference' => Val}, StrictMode);
 parse_property(<<16#1F, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_reason_string),
+    {Val, Rest} = parse_utf8_string_with_cause(Bin, StrictMode, invalid_reason_string),
     parse_property(Rest, Props#{'Reason-String' => Val}, StrictMode);
 parse_property(<<16#21, Val:16/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Receive-Maximum' => Val}, StrictMode);
@@ -635,7 +639,7 @@ parse_utf8_pair(<<LenK:16/big, Rest/binary>>, _StrictMode) when
     LenK > byte_size(Rest)
 ->
     ?PARSE_ERR(#{
-        hint => user_property_not_enough_bytes,
+        cause => user_property_not_enough_bytes,
         parsed_key_length => LenK,
         remaining_bytes_length => byte_size(Rest)
     });
@@ -644,7 +648,7 @@ parse_utf8_pair(<<LenK:16/big, _Key:LenK/binary, LenV:16/big, Rest/binary>>, _St
     LenV > byte_size(Rest)
 ->
     ?PARSE_ERR(#{
-        hint => malformed_user_property_value,
+        cause => malformed_user_property_value,
         parsed_key_length => LenK,
         parsed_value_length => LenV,
         remaining_bytes_length => byte_size(Rest)
@@ -653,16 +657,16 @@ parse_utf8_pair(Bin, _StrictMode) when
     4 > byte_size(Bin)
 ->
     ?PARSE_ERR(#{
-        hint => user_property_not_enough_bytes,
+        cause => user_property_not_enough_bytes,
         total_bytes => byte_size(Bin)
     }).
 
-parse_utf8_string_with_hint(Bin, StrictMode, Hint) ->
+parse_utf8_string_with_cause(Bin, StrictMode, Cause) ->
     try
         parse_utf8_string(Bin, StrictMode)
     catch
         throw:{?FRAME_PARSE_ERROR, Reason} when is_map(Reason) ->
-            ?PARSE_ERR(Reason#{hint => Hint})
+            ?PARSE_ERR(Reason#{cause => Cause})
     end.
 
 parse_optional(Bin, F, true) ->
@@ -678,7 +682,7 @@ parse_utf8_string(<<Len:16/big, Rest/binary>>, _) when
     Len > byte_size(Rest)
 ->
     ?PARSE_ERR(#{
-        hint => malformed_utf8_string,
+        cause => malformed_utf8_string,
         parsed_length => Len,
         remaining_bytes_length => byte_size(Rest)
     });
@@ -687,20 +691,24 @@ parse_utf8_string(Bin, _) when
 ->
     ?PARSE_ERR(#{reason => malformed_utf8_string_length}).
 
-parse_binary_data(<<Len:16/big, Data:Len/binary, Rest/binary>>) ->
+parse_will_payload(<<Len:16/big, Data:Len/binary, Rest/binary>>) ->
     {Data, Rest};
-parse_binary_data(<<Len:16/big, Rest/binary>>) when
+parse_will_payload(<<Len:16/big, Rest/binary>>) when
     Len > byte_size(Rest)
 ->
     ?PARSE_ERR(#{
-        hint => malformed_binary_data,
+        cause => malformed_will_payload,
         parsed_length => Len,
-        remaining_bytes_length => byte_size(Rest)
+        remaining_bytes => byte_size(Rest)
     });
-parse_binary_data(Bin) when
+parse_will_payload(Bin) when
     2 > byte_size(Bin)
 ->
-    ?PARSE_ERR(malformed_binary_data_length).
+    ?PARSE_ERR(#{
+        cause => malformed_will_payload,
+        length_bytes => size(Bin),
+        expected_bytes => 2
+    }).
 
 %%--------------------------------------------------------------------
 %% Serialize MQTT Packet

apps/emqx/src/emqx_packet.erl

@@ -493,8 +493,8 @@ format(#mqtt_packet{header = Header, variable = Variable, payload = Payload}, Pa
         "" -> [HeaderIO, ")"];
         VarIO -> [HeaderIO, ", ", VarIO, ")"]
     end;
-%% receive a frame error packet, such as {frame_error,frame_too_large} or
-%% {frame_error,#{expected => <<"'MQTT' or 'MQIsdp'">>,hint => invalid_proto_name,received => <<"bad_name">>}}
+%% receive a frame error packet, such as {frame_error,#{cause := frame_too_large}} or
+%% {frame_error,#{expected => <<"'MQTT' or 'MQIsdp'">>,cause => invalid_proto_name,received => <<"bad_name">>}}
 format(FrameError, _PayloadEncode) ->
     lists:flatten(io_lib:format("~tp", [FrameError])).
 

apps/emqx/test/emqx_channel_SUITE.erl

@@ -427,19 +427,32 @@ t_handle_in_auth(_) ->
 
 t_handle_in_frame_error(_) ->
     IdleChannel = channel(#{conn_state => idle}),
-    {shutdown, #{shutdown_count := frame_error, reason := frame_too_large}, _Chan} =
-        emqx_channel:handle_in({frame_error, frame_too_large}, IdleChannel),
+    {shutdown, #{shutdown_count := frame_too_large, cause := frame_too_large}, _Chan} =
+        emqx_channel:handle_in({frame_error, #{cause => frame_too_large}}, IdleChannel),
     ConnectingChan = channel(#{conn_state => connecting}),
     ConnackPacket = ?CONNACK_PACKET(?RC_PACKET_TOO_LARGE),
-    {shutdown, #{shutdown_count := frame_error, reason := frame_too_large}, ConnackPacket, _} =
-        emqx_channel:handle_in({frame_error, frame_too_large}, ConnectingChan),
+    {shutdown,
+        #{
+            shutdown_count := frame_too_large,
+            cause := frame_too_large,
+            limit := 100,
+            received := 101
+        },
+        ConnackPacket,
+        _} =
+        emqx_channel:handle_in(
+            {frame_error, #{cause => frame_too_large, received => 101, limit => 100}},
+            ConnectingChan
+        ),
     DisconnectPacket = ?DISCONNECT_PACKET(?RC_PACKET_TOO_LARGE),
     ConnectedChan = channel(#{conn_state => connected}),
-    {ok, [{outgoing, DisconnectPacket}, {close, frame_too_large}], _} =
-        emqx_channel:handle_in({frame_error, frame_too_large}, ConnectedChan),
+    ?assertMatch(
+        {ok, [{outgoing, DisconnectPacket}, {close, frame_too_large}], _},
+        emqx_channel:handle_in({frame_error, #{cause => frame_too_large}}, ConnectedChan)
+    ),
     DisconnectedChan = channel(#{conn_state => disconnected}),
     {ok, DisconnectedChan} =
-        emqx_channel:handle_in({frame_error, frame_too_large}, DisconnectedChan).
+        emqx_channel:handle_in({frame_error, #{cause => frame_too_large}}, DisconnectedChan).
 
 t_handle_in_expected_packet(_) ->
     Packet = ?DISCONNECT_PACKET(?RC_PROTOCOL_ERROR),