test(quic): improve coverage

apps/emqx/include/emqx_quic.hrl

@@ -0,0 +1,24 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-ifndef(EMQX_QUIC_HRL).
+-define(EMQX_QUIC_HRL, true).
+
+%% MQTT Over QUIC Shutdown Error code.
+-define(MQTT_QUIC_CONN_NOERROR, 0).
+-define(MQTT_QUIC_CONN_ERROR_OVERLOADED, 2).
+
+-endif.

apps/emqx/src/emqx_quic_connection.erl

@@ -20,6 +20,7 @@
 -include("logger.hrl").
 -ifndef(BUILD_WITHOUT_QUIC).
 -include_lib("quicer/include/quicer.hrl").
+-include_lib("emqx/include/emqx_quic.hrl").
 -else.
 -define(QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0).
 -endif.
@@ -36,9 +37,9 @@
     local_address_changed/3,
     peer_address_changed/3,
     streams_available/3,
-    peer_needs_streams/3,
+    % @TODO wait for newer quicer
+    %peer_needs_streams/3,
     resumed/3,
-    nst_received/3,
     new_stream/3
 ]).
 
@@ -120,11 +121,16 @@ new_conn(
                     ok = quicer:async_handshake(Conn),
                     {ok, S#{conn := Conn, ctrl_pid := CtrlPid}};
                 {'EXIT', _Pid, _Reason} ->
-                    {error, stream_accept_error}
+                    {stop, stream_accept_error, S}
             end;
         true ->
             emqx_metrics:inc('olp.new_conn'),
-            {error, overloaded}
+            quicer:async_shutdown_connection(
+                Conn,
+                ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE,
+                ?MQTT_QUIC_CONN_ERROR_OVERLOADED
+            ),
+            {stop, normal, S}
     end.
 
 %% @doc callback when connection is connected.
@@ -132,8 +138,8 @@ new_conn(
     {ok, cb_state()} | {error, any()}.
 connected(Conn, Props, #{slow_start := false} = S) ->
     ?SLOG(debug, Props),
-    {ok, _Pid} = emqx_connection:start_link(emqx_quic_stream, Conn, S),
-    {ok, S};
+    {ok, Pid} = emqx_connection:start_link(emqx_quic_stream, Conn, S),
+    {ok, S#{ctrl_pid => Pid}};
 connected(_Conn, Props, S) ->
     ?SLOG(debug, Props),
     {ok, S}.
@@ -147,12 +153,6 @@ resumed(Conn, Data, #{resumed_callback := ResumeFun} = S) when
 resumed(_Conn, _Data, S) ->
     {ok, S#{is_resumed := true}}.
 
-%% @doc callback for receiving nst, should never happen on server.
--spec nst_received(quicer:connection_handle(), TicketBin :: binary(), cb_state()) -> cb_ret().
-nst_received(_Conn, _Data, S) ->
-    %% As server we should not recv NST!
-    {stop, no_nst_for_server, S}.
-
 %% @doc callback for handling orphan data streams
 %%      depends on the connecion state and control stream state.
 -spec new_stream(quicer:stream_handle(), quicer:new_stream_props(), cb_state()) -> cb_ret().
@@ -233,9 +233,9 @@ streams_available(_C, {BidirCnt, UnidirCnt}, S) ->
 %%      should cope with rate limiting
 %% @TODO this is not going to get triggered in current version
 %% for https://github.com/microsoft/msquic/issues/3120
--spec peer_needs_streams(quicer:connection_handle(), undefined, cb_state()) -> cb_ret().
-peer_needs_streams(_C, undefined, S) ->
-    {ok, S}.
+%% -spec peer_needs_streams(quicer:connection_handle(), undefined, cb_state()) -> cb_ret().
+%% peer_needs_streams(_C, undefined, S) ->
+%%     {ok, S}.
 
 %% @doc handle API calls
 handle_call(
@@ -296,5 +296,6 @@ init_cb_state(#{zone := _Zone} = Map) ->
         streams => [],
         parse_state => undefined,
         channel => undefined,
-        serialize => undefined
+        serialize => undefined,
+        is_resumed => false
     }.

apps/emqx/src/emqx_quic_data_stream.erl

@@ -240,6 +240,10 @@ do_handle_appl_msg({incoming, #mqtt_packet{} = Packet}, #{channel := Channel} =
     Channel =/= undefined
 ->
     with_channel(handle_in, [Packet], S);
+do_handle_appl_msg({incoming, {frame_error, _} = FE}, #{channel := Channel} = S) when
+    Channel =/= undefined
+->
+    with_channel(handle_in, [FE], S);
 do_handle_appl_msg({close, Reason}, S) ->
     %% @TODO shall we abort shutdown or graceful shutdown?
     with_channel(handle_info, [{sock_closed, Reason}], S);

apps/emqx/src/emqx_quic_stream.erl

@@ -108,15 +108,15 @@ wait({ConnOwner, Conn, ConnInfo}) ->
         %% Connection owner process down
         {'EXIT', ConnOwner, _Reason} ->
             {error, enotconn}
-    end;
-%% For ownership handover
-wait({PrevOwner, Conn, Stream, SocketInfo}) ->
-    case quicer:wait_for_handoff(PrevOwner, Stream) of
-        ok ->
-            {ok, socket(Conn, Stream, SocketInfo)};
-        owner_down ->
-            {error, owner_down}
     end.
+%% UNUSED, for ownership handover,
+%% wait({PrevOwner, Conn, Stream, SocketInfo}) ->
+%%     case quicer:wait_for_handoff(PrevOwner, Stream) of
+%%         ok ->
+%%             {ok, socket(Conn, Stream, SocketInfo)};
+%%         owner_down ->
+%%             {error, owner_down}
+%%     end.
 
 type(_) ->
     quic.