fix ssl 心跳机制

docs/efka_iot_protocol.md

@@ -219,6 +219,22 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
 
 `iot` 会为每个 `{UUID, TaskId}` 维护一个独立的任务进程，用于缓存最近的部署日志、支持多个 SSE listener，并在收到 close 事件后结束事件流。
 
+### efka -> iot: ping
+
+```erlang
+{<<"message">>, <<"ping">>}
+```
+
+用于 TLS 长连接的应用层保活。`efka` 在鉴权成功后周期发送，当前发送间隔为 30 秒。
+
+`iot` 收到后回复：
+
+```erlang
+{<<"message">>, <<"pong">>}
+```
+
+`ping/pong` 只表示 TLS 连接仍可读写，不参与 host online/offline 判定。host 上下线仍由 UDP 心跳和 `iot_host` 本地连接状态共同维护。
+
 ### iot -> efka: pub
 
 ```erlang
@@ -235,6 +251,7 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
 
 - `efka` 鉴权超时时间：5 秒。
 - `iot` command inflight 超时时间：60 秒。
+- `iot` SSL channel 空闲超时时间：120 秒。120 秒内没有收到任何 TLS 包，包括 `ping`、业务 `message`、`request`、`command_response`，`iot` 会主动关闭该连接。
 - `iot` 管理多个 `efka` 时，每个连接有独立 `ssl_channel` 和独立 inflight 表。
 - command 超时后，`iot` 删除 inflight 记录；之后如果迟到的 `command_response` 到达，会被视为未预期响应。
 

src/transport/efka_client.erl

@@ -28,11 +28,13 @@
 -define(STATE_AUTH, auth).
 %% 激活状态下
 -define(STATE_ACTIVATED, activated).
+-define(SSL_PING_INTERVAL, 30000).
 
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
     auth_ref = undefined :: undefined | binary(),
+    ping_timer_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
@@ -143,6 +145,20 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
     schedule_reconnect(),
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 
+handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
+    Packet = term_to_binary({<<"message">>, <<"ping">>}),
+    case ssl:send(Socket, Packet) of
+        ok ->
+            {keep_state, schedule_ssl_ping(State)};
+        {error, Reason} ->
+            logger:warning("[efka_client] send ssl ping failed, reason: ~p", [Reason]),
+            disconnect(Socket),
+            schedule_reconnect(),
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+    end;
+handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
+    {keep_state, State};
+
 %% 将缓存中的数据推送到服务器端
 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     case efka_client_cache:fetch_next() of
@@ -165,19 +181,22 @@ handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket})
         error:Error ->
             logger:warning("[efka_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
             disconnect(Socket),
+            cancel_ssl_ping(State),
             schedule_reconnect(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     logger:debug("[efka_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
+    cancel_ssl_ping(State),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
     logger:debug("[efka_client] ssl closed"),
     disconnect(Socket),
+    cancel_ssl_ping(State),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 
 %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
@@ -193,7 +212,8 @@ handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateN
 %% 处理response
 handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_client] auth success"),
-    {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
+    State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
+    {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
 handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
     logger:debug("[efka_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
@@ -207,6 +227,8 @@ handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State)
     {keep_state, State};
 
 %% 处理Pub/Sub机制
+handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
+    {keep_state, State};
 handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     efka_subscription:publish(Topic, Qos, Content),
@@ -254,7 +276,8 @@ handle_container_command(Ref, Request, Socket) ->
     ok.
 
 -spec terminate(term(), atom(), #state{}) -> ok.
-terminate(Reason, _StateName, _State = #state{socket = Socket}) ->
+terminate(Reason, _StateName, State = #state{socket = Socket}) ->
+    cancel_ssl_ping(State),
     disconnect(Socket),
     efka_client_cache:close(),
     logger:notice("[efka_client] terminate with reason: ~p", [Reason]),
@@ -305,6 +328,22 @@ disconnect(Socket) ->
 schedule_reconnect() ->
     erlang:start_timer(5000, self(), create_transport).
 
+-spec schedule_ssl_ping(#state{}) -> #state{}.
+schedule_ssl_ping(State = #state{ping_timer_ref = TimerRef}) ->
+    cancel_timer(TimerRef),
+    State#state{ping_timer_ref = erlang:start_timer(?SSL_PING_INTERVAL, self(), ssl_ping)}.
+
+-spec cancel_ssl_ping(#state{}) -> ok.
+cancel_ssl_ping(#state{ping_timer_ref = TimerRef}) ->
+    cancel_timer(TimerRef).
+
+-spec cancel_timer(undefined | reference()) -> ok.
+cancel_timer(undefined) ->
+    ok;
+cancel_timer(TimerRef) ->
+    _ = erlang:cancel_timer(TimerRef),
+    ok.
+
 -spec request_ref() -> binary().
 request_ref() ->
     crypto:strong_rand_bytes(16).