add 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/ssl_channel.erl

@@ -11,6 +11,7 @@
 -behaviour(ranch_protocol).
 
 -define(INFLIGHT_TIMEOUT, 60000).
+-define(SSL_IDLE_TIMEOUT, 120000).
 
 %% API
 -export([pub/4, container_call/4, cancel_command_call/2]).
@@ -29,7 +30,9 @@
     host_pid = undefined,
 
     %% iot 发起的 command 与 command_response 的对应关系
-    inflight = #{}
+    inflight = #{},
+
+    idle_timer_ref :: undefined | reference()
 }).
 
 -record(inflight_command, {
@@ -72,7 +75,8 @@ init(Ref, Transport, _Opts = []) ->
     {ok, Socket} = ranch:handshake(Ref),
     logger:debug("[ssl_channel] get a new connection: ~p", [Socket]),
     Transport:setopts(Socket, [binary, {active, true}, {packet, 4}]),
-    gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
+    IdleTimerRef = start_idle_timer(),
+    gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket, idle_timer_ref = IdleTimerRef}).
 
 handle_call({cancel_command_call, Ref}, _From, State = #state{inflight = Inflight}) ->
     case maps:take(Ref, Inflight) of
@@ -120,6 +124,12 @@ handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight
             {noreply, State}
     end;
 
+handle_info({timeout, TimerRef, ssl_idle_timeout}, State = #state{idle_timer_ref = TimerRef}) ->
+    logger:notice("[ssl_channel] ssl channel idle timeout"),
+    {stop, ssl_idle_timeout, State#state{idle_timer_ref = undefined}};
+handle_info({timeout, _TimerRef, ssl_idle_timeout}, State) ->
+    {noreply, State};
+
 %% 关闭当前通道
 handle_info({stop, Reason}, State) ->
     {stop, Reason, State};
@@ -130,19 +140,20 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h
     {stop, Reason, State};
 
 handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket}) when is_binary(PacketBin) ->
+    State1 = reset_idle_timer(State),
     try binary_to_term(PacketBin, [safe]) of
         {<<"request">>, Ref, Body} ->
-            handle_request_frame(Ref, Body, State);
+            handle_request_frame(Ref, Body, State1);
         {<<"message">>, Body} ->
-            handle_message_frame(Body, State);
+            handle_message_frame(Body, State1);
         {<<"command_response">>, Ref, Response} ->
-            handle_command_response_frame(Ref, Response, State);
+            handle_command_response_frame(Ref, Response, State1);
         Other ->
             logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
-            {stop, bad_packet, State}
+            {stop, bad_packet, State1}
     catch error:Error ->
         logger:warning("[ssl_channel] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
-        {stop, bad_packet, State}
+        {stop, bad_packet, State1}
     end;
 
 handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
@@ -161,7 +172,8 @@ handle_info(Info, State) ->
     logger:warning("[ssl_channel] get a unknown message: ~p, state: ~p", [Info, State]),
     {noreply, State}.
 
-terminate(Reason, #state{inflight = Inflight, transport = Transport, socket = Socket}) ->
+terminate(Reason, #state{inflight = Inflight, transport = Transport, socket = Socket, idle_timer_ref = IdleTimerRef}) ->
+    cancel_timer(IdleTimerRef),
     maps:foreach(fun(Ref, CommandInfo) -> reply_command_closed(Ref, CommandInfo, Reason) end, Inflight),
     Transport:close(Socket),
     logger:warning("[ssl_channel] stop with reason: ~p", [Reason]),
@@ -205,8 +217,17 @@ handle_request_frame(Ref, Body, State) ->
     logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, Body]),
     {stop, normal, State}.
 
--spec handle_message_frame(tuple(), #state{}) ->
+-spec handle_message_frame(term(), #state{}) ->
-    {noreply, #state{}}.
+    {noreply, #state{}} | {stop, term(), #state{}}.
+handle_message_frame(<<"ping">>, State = #state{transport = Transport, socket = Socket}) ->
+    Packet = term_to_binary({<<"message">>, <<"pong">>}),
+    case Transport:send(Socket, Packet) of
+        ok ->
+            {noreply, State};
+        {error, Reason} ->
+            logger:warning("[ssl_channel] send pong failed, reason: ~p", [Reason]),
+            {stop, {send_failed, Reason}, State}
+    end;
 handle_message_frame({<<"data">>, #{<<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
     iot_host:handle(HostPid, {data, RouteKey, Metric}),
     {noreply, State};
@@ -329,3 +350,19 @@ safe_term(Value) when is_tuple(Value) ->
     list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
 safe_term(Value) ->
     Value.
+
+-spec start_idle_timer() -> reference().
+start_idle_timer() ->
+    erlang:start_timer(?SSL_IDLE_TIMEOUT, self(), ssl_idle_timeout).
+
+-spec reset_idle_timer(#state{}) -> #state{}.
+reset_idle_timer(State = #state{idle_timer_ref = TimerRef}) ->
+    cancel_timer(TimerRef),
+    State#state{idle_timer_ref = start_idle_timer()}.
+
+-spec cancel_timer(undefined | reference()) -> ok.
+cancel_timer(undefined) ->
+    ok;
+cancel_timer(TimerRef) ->
+    _ = erlang:cancel_timer(TimerRef),
+    ok.