Merge pull request #11987 from zmstone/1120-do-not-crash-on-einval-after-check-cache

fix(emqx_connection): handle socket activation error return

apps/emqx/src/emqx_connection.erl

@@ -552,13 +552,13 @@ handle_msg({quic, Data, _Stream, #{len := Len}}, State) when is_binary(Data) ->
     inc_counter(incoming_bytes, Len),
     ok = emqx_metrics:inc('bytes.received', Len),
     when_bytes_in(Len, Data, State);
-handle_msg(check_cache, #state{limiter_buffer = Cache} = State) ->
-    case queue:peek(Cache) of
+handle_msg(check_limiter_buffer, #state{limiter_buffer = Buffer} = State) ->
+    case queue:peek(Buffer) of
         empty ->
-            activate_socket(State);
+            handle_info(activate_socket, State);
         {value, #pending_req{need = Needs, data = Data, next = Next}} ->
-            State2 = State#state{limiter_buffer = queue:drop(Cache)},
-            check_limiter(Needs, Data, Next, [check_cache], State2)
+            State2 = State#state{limiter_buffer = queue:drop(Buffer)},
+            check_limiter(Needs, Data, Next, [check_limiter_buffer], State2)
     end;
 handle_msg(
     {incoming, Packet = ?CONNECT_PACKET(ConnPkt)},
@@ -1036,13 +1036,13 @@ check_limiter(
     Data,
     WhenOk,
     _Msgs,
-    #state{limiter_buffer = Cache} = State
+    #state{limiter_buffer = Buffer} = State
 ) ->
     %% if there has a retry timer,
-    %% cache the operation and execute it after the retry is over
-    %% the maximum length of the cache queue is equal to the active_n
+    %% Buffer the operation and execute it after the retry is over
+    %% the maximum length of the buffer queue is equal to the active_n
     New = #pending_req{need = Needs, data = Data, next = WhenOk},
-    {ok, State#state{limiter_buffer = queue:in(New, Cache)}}.
+    {ok, State#state{limiter_buffer = queue:in(New, Buffer)}}.
 
 %% try to perform a retry
 -spec retry_limiter(state()) -> _.
@@ -1053,7 +1053,7 @@ retry_limiter(#state{limiter = Limiter} = State) ->
         {ok, Limiter2} ->
             Next(
                 Data,
-                [check_cache],
+                [check_limiter_buffer],
                 State#state{
                     limiter = Limiter2,
                     limiter_timer = undefined

apps/emqx/src/emqx_ws_connection.erl

@@ -94,7 +94,7 @@
     limiter :: container(),
 
     %% cache operation when overload
-    limiter_cache :: queue:queue(cache()),
+    limiter_buffer :: queue:queue(cache()),
 
     %% limiter timers
     limiter_timer :: undefined | reference()
@@ -326,7 +326,7 @@ websocket_init([Req, Opts]) ->
                     zone = Zone,
                     listener = {Type, Listener},
                     limiter_timer = undefined,
-                    limiter_cache = queue:new()
+                    limiter_buffer = queue:new()
                 },
                 hibernate};
         {denny, Reason} ->
@@ -462,13 +462,13 @@ websocket_info(
     State
 ) ->
     return(retry_limiter(State));
-websocket_info(check_cache, #state{limiter_cache = Cache} = State) ->
-    case queue:peek(Cache) of
+websocket_info(check_limiter_buffer, #state{limiter_buffer = Buffer} = State) ->
+    case queue:peek(Buffer) of
         empty ->
             return(enqueue({active, true}, State#state{sockstate = running}));
         {value, #cache{need = Needs, data = Data, next = Next}} ->
-            State2 = State#state{limiter_cache = queue:drop(Cache)},
-            return(check_limiter(Needs, Data, Next, [check_cache], State2))
+            State2 = State#state{limiter_buffer = queue:drop(Buffer)},
+            return(check_limiter(Needs, Data, Next, [check_limiter_buffer], State2))
     end;
 websocket_info({timeout, TRef, Msg}, State) when is_reference(TRef) ->
     handle_timeout(TRef, Msg, State);
@@ -630,10 +630,10 @@ check_limiter(
     Data,
     WhenOk,
     _Msgs,
-    #state{limiter_cache = Cache} = State
+    #state{limiter_buffer = Buffer} = State
 ) ->
     New = #cache{need = Needs, data = Data, next = WhenOk},
-    State#state{limiter_cache = queue:in(New, Cache)}.
+    State#state{limiter_buffer = queue:in(New, Buffer)}.
 
 -spec retry_limiter(state()) -> state().
 retry_limiter(#state{limiter = Limiter} = State) ->
@@ -644,7 +644,7 @@ retry_limiter(#state{limiter = Limiter} = State) ->
         {ok, Limiter2} ->
             Next(
                 Data,
-                [check_cache],
+                [check_limiter_buffer],
                 State#state{
                     limiter = Limiter2,
                     limiter_timer = undefined

changes/ce/fix-11987.en.md

@@ -0,0 +1,3 @@
+Fix connection crash when trying to set TCP/SSL socket `active_n` option.
+
+Prior to this fix, if a socket is already closed when connection process tries to set `active_n` option, it causes a `case_clause` crash.