fix task_event

src/host/iot_container_task.erl

@@ -0,0 +1,147 @@
+%%%-------------------------------------------------------------------
+%%% @doc Owns one container deployment task stream.
+%%% @end
+%%%-------------------------------------------------------------------
+-module(iot_container_task).
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/2]).
+-export([get_pid/2, subscribe/3, stream/3, close/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(MAX_EVENTS, 200).
+-define(STOP_AFTER_CLOSE, 300000).
+
+-record(state, {
+    uuid :: binary(),
+    task_id :: integer(),
+    status = pending :: pending | running | success | fail,
+    events = [] :: [{binary(), binary()}],
+    listeners = #{} :: #{pid() => reference()},
+    close_reason = undefined :: undefined | binary(),
+    stop_ref = undefined :: undefined | reference()
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec start_link(binary(), integer()) -> {ok, pid()} | ignore | {error, term()}.
+start_link(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) ->
+    gen_server:start_link({via, gproc, {n, l, name(UUID, TaskId)}}, ?MODULE, [UUID, TaskId], []).
+
+-spec get_pid(binary(), integer()) -> undefined | pid().
+get_pid(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) ->
+    gproc:whereis_name({n, l, name(UUID, TaskId)}).
+
+-spec subscribe(binary(), integer(), pid()) -> ok | {error, term()}.
+subscribe(UUID, TaskId, ListenerPid) when is_binary(UUID), is_integer(TaskId), is_pid(ListenerPid) ->
+    case iot_container_task_sup:ensure_started(UUID, TaskId) of
+        {ok, Pid} ->
+            gen_server:call(Pid, {subscribe, ListenerPid});
+        {error, Reason} ->
+            {error, Reason}
+    end.
+
+-spec stream(pid(), binary(), binary()) -> ok.
+stream(Pid, Type, Stream) when is_pid(Pid), is_binary(Type), is_binary(Stream) ->
+    gen_server:cast(Pid, {stream, Type, Stream}).
+
+-spec close(pid(), binary()) -> ok.
+close(Pid, Reason) when is_pid(Pid), is_binary(Reason) ->
+    gen_server:cast(Pid, {close, Reason}).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+-spec init([binary() | integer()]) -> {ok, #state{}}.
+init([UUID, TaskId]) ->
+    ok = iot_log:set_metadata(),
+    {ok, #state{uuid = UUID, task_id = TaskId}}.
+
+-spec handle_call(term(), {pid(), term()}, #state{}) ->
+    {reply, term(), #state{}}.
+handle_call({subscribe, ListenerPid}, _From, State0 = #state{task_id = TaskId, events = Events, listeners = Listeners, close_reason = CloseReason}) ->
+    MonRef = erlang:monitor(process, ListenerPid),
+    maps:get(ListenerPid, Listeners, undefined) =/= undefined andalso erlang:demonitor(maps:get(ListenerPid, Listeners), [flush]),
+    lists:foreach(fun({Type, Stream}) ->
+        ListenerPid ! {stream_data, TaskId, Type, Stream}
+    end, Events),
+    case CloseReason of
+        undefined ->
+            ok;
+        Reason ->
+            ListenerPid ! {stream_close, TaskId, Reason}
+    end,
+    {reply, ok, State0#state{listeners = maps:put(ListenerPid, MonRef, Listeners)}};
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+handle_cast({stream, Type, Stream}, State0 = #state{task_id = TaskId, events = Events0, listeners = Listeners}) ->
+    Event = {Type, Stream},
+    Events = trim_events(Events0 ++ [Event]),
+    maps:foreach(fun(ListenerPid, _MonRef) ->
+        ListenerPid ! {stream_data, TaskId, Type, Stream}
+    end, Listeners),
+    {noreply, State0#state{status = running, events = Events}};
+handle_cast({close, Reason}, State0 = #state{task_id = TaskId, listeners = Listeners}) ->
+    maps:foreach(fun(ListenerPid, _MonRef) ->
+        ListenerPid ! {stream_close, TaskId, Reason}
+    end, Listeners),
+    Status = close_status(Reason),
+    {noreply, ensure_stop_timer(State0#state{status = Status, close_reason = Reason})};
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(term(), #state{}) -> {noreply, #state{}} | {stop, normal, #state{}}.
+handle_info({'DOWN', _Ref, process, ListenerPid, _Reason}, State = #state{listeners = Listeners}) ->
+    {noreply, State#state{listeners = maps:remove(ListenerPid, Listeners)}};
+handle_info(stop_after_close, State) ->
+    {stop, normal, State};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+-spec terminate(term(), #state{}) -> ok.
+terminate(_Reason, _State) ->
+    ok.
+
+-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+-spec name(binary(), integer()) -> term().
+name(UUID, TaskId) ->
+    {iot_container_task, UUID, TaskId}.
+
+-spec trim_events([{binary(), binary()}]) -> [{binary(), binary()}].
+trim_events(Events) ->
+    Len = length(Events),
+    case Len > ?MAX_EVENTS of
+        true ->
+            lists:nthtail(Len - ?MAX_EVENTS, Events);
+        false ->
+            Events
+    end.
+
+-spec close_status(binary()) -> success | fail.
+close_status(<<"success">>) ->
+    success;
+close_status(_) ->
+    fail.
+
+-spec ensure_stop_timer(#state{}) -> #state{}.
+ensure_stop_timer(State = #state{stop_ref = undefined}) ->
+    Ref = erlang:send_after(?STOP_AFTER_CLOSE, self(), stop_after_close),
+    State#state{stop_ref = Ref};
+ensure_stop_timer(State) ->
+    State.

src/host/iot_container_task_sup.erl

@@ -0,0 +1,100 @@
+%%%-------------------------------------------------------------------
+%%% @doc Dynamic supervisor for container deployment tasks.
+%%% @end
+%%%-------------------------------------------------------------------
+-module(iot_container_task_sup).
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+-export([ensure_started/2, stream/4, close/3, fail/3]).
+
+%% supervisor callbacks
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+-spec ensure_started(binary(), integer()) -> {ok, pid()} | {error, term()}.
+ensure_started(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) ->
+    case iot_container_task:get_pid(UUID, TaskId) of
+        undefined ->
+            case supervisor:start_child(?SERVER, [UUID, TaskId]) of
+                {ok, Pid} ->
+                    {ok, Pid};
+                {ok, Pid, _Info} ->
+                    {ok, Pid};
+                {error, {already_started, Pid}} ->
+                    {ok, Pid};
+                {error, {shutdown, {failed_to_start_child, _Id, {already_started, Pid}}}} ->
+                    {ok, Pid};
+                {error, Reason} ->
+                    {error, Reason}
+            end;
+        Pid when is_pid(Pid) ->
+            {ok, Pid}
+    end.
+
+-spec stream(binary(), integer(), binary(), binary()) -> ok.
+stream(UUID, TaskId, Type, Stream)
+  when is_binary(UUID), is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+    case ensure_started(UUID, TaskId) of
+        {ok, Pid} ->
+            iot_container_task:stream(Pid, Type, Stream);
+        {error, Reason} ->
+            logger:warning("[iot_container_task_sup] start task failed, uuid: ~p, task_id: ~p, reason: ~p", [UUID, TaskId, Reason]),
+            ok
+    end.
+
+-spec close(binary(), integer(), binary()) -> ok.
+close(UUID, TaskId, Reason) when is_binary(UUID), is_integer(TaskId), is_binary(Reason) ->
+    case ensure_started(UUID, TaskId) of
+        {ok, Pid} ->
+            iot_container_task:close(Pid, Reason);
+        {error, StartReason} ->
+            logger:warning("[iot_container_task_sup] start task before close failed, uuid: ~p, task_id: ~p, reason: ~p", [UUID, TaskId, StartReason]),
+            ok
+    end.
+
+-spec fail(binary(), integer(), term()) -> ok.
+fail(UUID, TaskId, Reason) when is_binary(UUID), is_integer(TaskId) ->
+    ReasonBin = reason_to_binary(Reason),
+    ok = stream(UUID, TaskId, <<"error">>, ReasonBin),
+    close(UUID, TaskId, <<"fail">>).
+
+%%%===================================================================
+%%% supervisor callbacks
+%%%===================================================================
+
+-spec init([]) -> {ok, {{simple_one_for_one, non_neg_integer(), pos_integer()}, [supervisor:child_spec()]}}.
+init([]) ->
+    SupFlags = {simple_one_for_one, 10, 60},
+    ChildSpec = #{
+        id => iot_container_task,
+        start => {iot_container_task, start_link, []},
+        restart => temporary,
+        shutdown => 5000,
+        type => worker,
+        modules => [iot_container_task]
+    },
+    {ok, {SupFlags, [ChildSpec]}}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+-spec reason_to_binary(term()) -> binary().
+reason_to_binary(Reason) when is_binary(Reason) ->
+    Reason;
+reason_to_binary(timeout) ->
+    <<"timeout">>;
+reason_to_binary(Reason) ->
+    unicode:characters_to_binary(io_lib:format("~p", [Reason])).

src/host/iot_event_stream_observer.erl

@@ -1,129 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author anlicheng
-%%% @copyright (C) 2025, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 26. 9月 2025 12:19
-%%%-------------------------------------------------------------------
--module(iot_event_stream_observer).
--author("anlicheng").
-
--behaviour(gen_server).
-
-%% API
--export([start_link/0]).
--export([add_listener/2, stream_data/3, stream_close/2]).
-
-%% gen_server callbacks
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-
--define(SERVER, ?MODULE).
-
--record(state, {
-    listeners = #{}
-}).
-
-%%%===================================================================
-%%% API
-%%%===================================================================
-
--spec add_listener(ListenerPid :: pid(), TaskId :: integer()) -> ok.
-add_listener(ListenerPid, TaskId) when is_pid(ListenerPid), is_integer(TaskId) ->
-    gen_server:call(?SERVER, {add_listener, ListenerPid, TaskId}).
-
--spec stream_data(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> no_return().
-stream_data(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
-    gen_server:cast(?SERVER, {stream_data, TaskId, Type, Stream}).
-
--spec stream_close(TaskId :: integer(), Reason :: binary()) -> no_return().
-stream_close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
-    gen_server:cast(?SERVER, {stream_close, TaskId, Reason}).
-
-%% @doc Spawns the server and registers the local name (unique)
--spec(start_link() ->
-    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
-start_link() ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
-
-%%%===================================================================
-%%% gen_server callbacks
-%%%===================================================================
-
-%% @private
-%% @doc Initializes the server
--spec(init(Args :: term()) ->
-    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term()} | ignore).
-init([]) ->
-    ok = iot_log:set_metadata(),
-    {ok, #state{}}.
-
-%% @private
-%% @doc Handling call messages
--spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
-    State :: #state{}) ->
-    {reply, Reply :: term(), NewState :: #state{}} |
-    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_call({add_listener, ListenerPid, TaskId}, _From, State = #state{listeners = Listeners}) ->
-    erlang:monitor(process, ListenerPid),
-    {reply, ok, State#state{listeners = maps:put(TaskId, ListenerPid, Listeners)}}.
-
-%% @private
-%% @doc Handling cast messages
--spec(handle_cast(Request :: term(), State :: #state{}) ->
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast({stream_data, TaskId, Type, Stream}, State = #state{listeners = Listeners}) ->
-    case maps:find(TaskId, Listeners) of
-        error ->
-            ok;
-        {ok, ListenerPid} ->
-            is_process_alive(ListenerPid) andalso ListenerPid ! {stream_data, TaskId, Type, Stream}
-    end,
-    {noreply, State};
-handle_cast({stream_close, TaskId, Reason}, State = #state{listeners = Listeners}) ->
-    case maps:find(TaskId, Listeners) of
-        error ->
-            ok;
-        {ok, ListenerPid} ->
-            is_process_alive(ListenerPid) andalso ListenerPid ! {stream_close, TaskId, Reason}
-    end,
-    {noreply, State}.
-
-%% @private
-%% @doc Handling all non call/cast messages
--spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_info({'DOWN', _Ref, process, Pid, _Reason}, State = #state{listeners = Listeners}) ->
-    NListeners = maps:filter(fun(_, ListenerPid) -> ListenerPid /= Pid end, Listeners),
-    {noreply, State#state{listeners = NListeners}}.
-
-%% @private
-%% @doc This function is called by a gen_server when it is about to
-%% terminate. It should be the opposite of Module:init/1 and do any
-%% necessary cleaning up. When it returns, the gen_server terminates
-%% with Reason. The return value is ignored.
--spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
-    State :: #state{}) -> term()).
-terminate(_Reason, _State = #state{}) ->
-    ok.
-
-%% @private
-%% @doc Convert process state when code is changed
--spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
-    Extra :: term()) ->
-    {ok, NewState :: #state{}} | {error, Reason :: term()}).
-code_change(_OldVsn, State = #state{}, _Extra) ->
-    {ok, State}.
-
-%%%===================================================================
-%%% Internal functions
-%%%===================================================================

src/iot_sup.erl

@@ -29,12 +29,12 @@ init([]) ->
 
     Specs = [
         #{
-            id => 'iot_event_stream_observer',
+            id => 'iot_container_task_sup',
-            start => {'iot_event_stream_observer', start_link, []},
+            start => {'iot_container_task_sup', start_link, []},
             restart => permanent,
             shutdown => 2000,
-            type => worker,
+            type => supervisor,
-            modules => ['iot_event_stream_observer']
+            modules => ['iot_container_task_sup']
         },
 
         #{

src/transport/http/container_handler.erl

@@ -70,18 +70,11 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id"
         undefined ->
             {ok, 200, iot_util:json_error(404, <<"host not found">>)};
         Pid when is_pid(Pid) ->
-            case iot_host:deploy_container(Pid, TaskId, Config) of
+            case iot_container_task_sup:ensure_started(UUID, TaskId) of
-                {ok, Ref} ->
+                {ok, _TaskPid} ->
-                    case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
+                    handle_deploy_container(Pid, UUID, TaskId, Config);
-                        ok ->
-                            {ok, 200, request_success_response(<<"ok">>)};
-                        {ok, Result} ->
-                            {ok, 200, request_success_response(Result)};
                 {error, Reason} ->
-                            request_error_http_response(Reason)
+                    {ok, 500, iot_util:json_error(500, reason_to_binary(Reason))}
-                    end;
-                {error, Reason} when is_binary(Reason) ->
-                    {ok, 200, iot_util:json_error(400, Reason)}
             end
     end;
 
@@ -225,3 +218,22 @@ decode_json_bytes(Data) when is_binary(Data) ->
         Decoded ->
             {ok, Decoded}
     end.
+
+-spec handle_deploy_container(pid(), binary(), integer(), map()) ->
+    {ok, integer(), iolist()}.
+handle_deploy_container(Pid, UUID, TaskId, Config) ->
+    case iot_host:deploy_container(Pid, TaskId, Config) of
+        {ok, Ref} ->
+            case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
+                ok ->
+                    {ok, 200, request_success_response(<<"ok">>)};
+                {ok, Result} ->
+                    {ok, 200, request_success_response(Result)};
+                {error, Reason} ->
+                    ok = iot_container_task_sup:fail(UUID, TaskId, Reason),
+                    request_error_http_response(Reason)
+            end;
+        {error, Reason} when is_binary(Reason) ->
+            ok = iot_container_task_sup:fail(UUID, TaskId, Reason),
+            {ok, 200, iot_util:json_error(400, Reason)}
+    end.

src/transport/http/event_stream_handler.erl

@@ -19,20 +19,23 @@ init(Req0, Opts) ->
     GetParams0 = cowboy_req:parse_qs(Req0),
     GetParams = maps:from_list(GetParams0),
 
-    #{<<"task_id">> := TaskId0} = GetParams,
-    TaskId = binary_to_integer(TaskId0),
-
     logger:debug("method: ~p, path: ~p,  get: ~p", [Method, Path, GetParams]),
+    case parse_stream_params(GetParams) of
+        {ok, UUID, TaskId} ->
             Req1 = cowboy_req:stream_reply(200, #{
                 <<"Content-Type">> => <<"text/event-stream">>,
                 <<"Cache-Control">> => <<"no-cache">>,
                 <<"Connection">> => <<"keep-alive">>
             }, Req0),
 
-    ok = iot_event_stream_observer:add_listener(self(), TaskId),
+            ok = iot_container_task:subscribe(UUID, TaskId, self()),
             receiver_events(TaskId, Req1),
 
-    {ok, Req1, Opts}.
+            {ok, Req1, Opts};
+        {error, Reason} ->
+            Req1 = cowboy_req:reply(400, #{<<"Content-Type">> => <<"application/json">>}, iot_util:json_error(400, Reason), Req0),
+            {ok, Req1, Opts}
+    end.
 
 receiver_events(TaskId, Req) ->
     receive
@@ -45,4 +48,23 @@ receiver_events(TaskId, Req) ->
         {stream_close, TaskId, Reason} ->
             CloseFrame = iolist_to_binary([<<"event: close\n">>, <<"data: ", Reason/binary, "\n">>, <<"\n">>]),
             ok = cowboy_req:stream_body(CloseFrame, fin, Req)
+    after 30000 ->
+            ok = cowboy_req:stream_body(<<": heartbeat\n\n">>, nofin, Req),
+            receiver_events(TaskId, Req)
     end.
+
+-spec parse_stream_params(map()) -> {ok, binary(), integer()} | {error, binary()}.
+parse_stream_params(#{<<"uuid">> := UUID, <<"task_id">> := TaskId0}) when is_binary(UUID), is_binary(TaskId0) ->
+    try binary_to_integer(TaskId0) of
+        TaskId when TaskId >= 0 ->
+            {ok, UUID, TaskId};
+        _ ->
+            {error, <<"task_id must be non-negative">>}
+    catch
+        error:badarg ->
+            {error, <<"task_id must be integer">>}
+    end;
+parse_stream_params(#{<<"task_id">> := _TaskId0}) ->
+    {error, <<"uuid required">>};
+parse_stream_params(_) ->
+    {error, <<"uuid and task_id required">>}.

src/transport/tcp/ssl_channel.erl

@@ -207,19 +207,22 @@ handle_request_frame(Ref, Body, State) ->
 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};
-handle_message_frame({task_event, Event}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
+handle_message_frame({task_event, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) ->
-    handle_event_stream_frame(Event),
+    handle_event_stream_frame(UUID, Event),
     {noreply, State};
 handle_message_frame(Body, State) ->
     logger:warning("[ssl_channel] unsupported message body: ~p", [Body]),
     {noreply, State}.
 
--spec handle_event_stream_frame(map()) -> any().
+-spec handle_event_stream_frame(binary(), map()) -> ok.
-handle_event_stream_frame(#{task_id := TaskId, type := <<"close">>, stream := Reason}) ->
+handle_event_stream_frame(UUID, #{task_id := TaskId, type := <<"close">>, stream := Reason}) ->
-    iot_event_stream_observer:stream_close(TaskId, Reason);
+    iot_container_task_sup:close(UUID, TaskId, Reason);
-handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream}) ->
+handle_event_stream_frame(UUID, #{task_id := TaskId, type := Type, stream := Stream}) ->
-    logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
+    logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]),
-    iot_event_stream_observer:stream_data(TaskId, Type, Stream).
+    iot_container_task_sup:stream(UUID, TaskId, Type, Stream);
+handle_event_stream_frame(UUID, Event) ->
+    logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]),
+    ok.
 
 -spec handle_command_response_frame(reference(), tuple(), #state{}) ->
     {noreply, #state{}}.