Protect against accidental multiple concurrent connections

ebin/gmhive_client.app

@@ -1,6 +1,6 @@
 {application,gmhive_client,
              [{description,"Gajumaru Hive Client"},
-              {vsn,"0.9.1"},
+              {vsn,"0.9.2"},
               {registered,[]},
               {applications,[kernel,stdlib,sasl,gproc,inets,ssl,enoise,
                              gmconfig,gmhive_protocol,gmhive_worker]},

src/gmhc_connector.erl

@@ -442,12 +442,14 @@ protocol_connect(Opts, #st{econn = EConn} = S) ->
     Type   = to_atom(opt(type, Opts, [<<"type">>])),
     RId = erlang:unique_integer(),
     Vsns = gmhp_msgs:versions(),
+    %% Client = client_name(),
     Protocols = gmhp_msgs:protocols(hd(Vsns)),
     ConnectReq = #{ protocols => Protocols
                   , versions  => Vsns
                   , pool_id => PoolId
                   , pubkey => Pubkey
                   , extra_pubkeys => Extra
+                  %% , client => Client
                   , type => Type
                   , nonces => gmhc_server:total_nonces()
                   , signature => ""},
@@ -491,6 +493,38 @@ send_connect(EConn, RId, Msg, #{pubkey        := Pubkey,
             error(protocol_connect_timeout)
     end.
 
+%% client_name() ->
+%%     MyStr = app_string(gmhive_client),
+%%     case app_string(gajumine) of
+%%         <<>>  -> MyStr;
+%%         GMStr -> <<MyStr/binary,",",GMStr/binary>>
+%%     end.
+%%     %% {ok, Vsn} = application:get_key(gmhive_client, vsn),
+%%     %% maybe_add_gajumine(unicode:characters_to_binary(["gmhive_client-", Vsn])).
+
+%% maybe_add_gajumine(Str) ->
+%%     case setup_zomp:is_zomp_context() of
+%%         true ->
+%%             case zx_daemon:meta() of
+%%                 #{package_id := {"qpq","gajumine",_} = PId} ->
+%%                     {ok, PStr} = zx_lib:package_string(PId),
+%%                     GMStr = unicode:characters_to_binary([",", PStr]),
+%%                     <<Str/binary, GMStr/binary>>;
+%%                 _ ->
+%%                     Str
+%%             end;
+%%         false ->
+%%             Str
+%%     end.
+
+%% app_string(App) ->
+%%     case application:get_key(App, vsn) of
+%%         undefined ->
+%%             <<>>;
+%%         {ok, Vsn} ->
+%%             unicode:characters_to_binary([atom_to_binary(App),"-",Vsn])
+%%     end.
+
 to_bin(A) when is_atom(A) ->
     atom_to_binary(A, utf8);
 to_bin(S) ->

src/gmhc_connectors_sup.erl

@@ -9,10 +9,19 @@
         , start_connector/1
         , add_restart_info/2 ]).
 
+-export([ sort_restart_info/1 ]).
+
+-include_lib("kernel/include/logger.hrl").
+
 start_link() ->
     case supervisor:start_link({local, ?MODULE}, ?MODULE, []) of
         {ok, _} = Ok ->
-            RI = gmhc_watchdog:get_restart_info(),
+            RI0 = gmhc_watchdog:get_restart_info(),
+            ?LOG_ERROR("Restart info: ~p", [RI0]),
+            RI = sort_restart_info(RI0),
+            RemoveIds = maps:keys(maps:without(maps:keys(RI), RI0)),
+            gmhc_watchdog:remove_restart_info(RemoveIds),
+            ?LOG_ERROR("Sorted restart info: ~p", [RI]),
             maps:foreach(fun restart_connector/2, RI),
             Ok;
         Other ->
@@ -49,3 +58,13 @@ init([]) ->
                     , shutdown => 5000
                     , modules => [Mod] } ],
     {ok, {SupFlags, ChildSpecs}}.
+
+sort_restart_info(RI) ->
+    L = lists:sort(
+          [{Id, {T,H,P}, I}
+           || {{?MODULE,_} = Id, #{type := T, host := H, port := P} = I} <- maps:to_list(RI)]),
+    ConnTypes = [CT || {_, CT, _} <- lists:ukeysort(2, L)],
+    lists:foldl(fun(CT, Acc) ->
+                        {Id, _, I} = lists:keyfind(CT, 2, L),
+                        Acc#{Id => I}
+                end, #{}, ConnTypes).

src/gmhc_watchdog.erl

@@ -6,6 +6,7 @@
         , ping/0 ]).
 
 -export([ note_started/2
+        , remove_restart_info/1
         , get_restart_info/0 ]).
 
 -export([ start_link/0
@@ -41,6 +42,11 @@ note_started(Id, Info) ->
 get_restart_info() ->
     gen_server:call(?MODULE, get_restart_info).
 
+remove_restart_info([]) ->
+    ok;
+remove_restart_info(IDs) ->
+    gen_server:call(?MODULE, {remove_restart_info, IDs}).
+
 start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
@@ -50,6 +56,9 @@ init([]) ->
 handle_call({note_started, Id, Info}, _From, S) ->
     update_pt(Id, Info),
     {reply, ok, S};
+handle_call({remove_restart_info, IDs}, _From, S) ->
+    remove_restart_info_(IDs),
+    {reply, ok, S};
 handle_call(get_restart_info, _From, S) ->
     {reply, get_pt(), S};
 handle_call({watch, Pid, Interval, N}, _From, S) ->
@@ -65,9 +74,9 @@ handle_cast(Msg, S) ->
     ?LOG_DEBUG("Unknown cast: ~p", [Msg]),
     {noreply, S}.
 
-handle_info({timeout, _, Pid}, S) ->
+handle_info({timeout, TRef, Pid}, S) ->
     ?LOG_INFO("Timeout for pid ~p", [Pid]),
-    {noreply, ping_timeout(Pid, S)};
+    {noreply, ping_timeout(Pid, TRef, S)};
 handle_info({'DOWN', _, process, Pid, _}, S) ->
     {noreply, delete_watch(Pid, S)};
 handle_info(Msg, S) ->
@@ -108,9 +117,9 @@ delete_watch(Pid, #st{services = Svcs} = S) ->
             S
     end.
 
-ping_timeout(Pid, #st{services = Svcs} = S) ->
+ping_timeout(Pid, TRef, #st{services = Svcs} = S) ->
     case maps:find(Pid, Svcs) of
-        {ok, #svc{ n = N } = Svc} ->
+        {ok, #svc{ n = N, tref = TRef} = Svc} ->
             N1 = N-1,
             if N1 =< 0 ->
                     ?LOG_ERROR("Will exit Pid ~p", [Pid]),
@@ -120,7 +129,10 @@ ping_timeout(Pid, #st{services = Svcs} = S) ->
                     Svc1 = restart_timer(Pid, Svc#svc{n = N1}),
                     S#st{services = Svcs#{Pid := Svc1}}
             end;
-        error ->
+        {ok, _} ->
+            ?LOG_DEBUG("Timeout didn't match TRef - ignoring", []),
+            S;
+        _ ->
             S
     end.
 
@@ -136,6 +148,15 @@ update_pt(Id, Info) ->
     Pt = get_pt(),
     put_pt(Pt#{Id => Info}).
 
+remove_restart_info_(IDs) ->
+    RI = get_pt(),
+    case maps:without(IDs, RI) of
+        RI ->
+            ok;
+        NewRI ->
+            put_pt(NewRI)
+    end.
+
 get_pt() ->
     persistent_term:get(pt_key(), #{}).
 

zomp.meta

@@ -4,7 +4,7 @@
 {prefix,"gmhc"}.
 {desc,"Gajumaru Hive Client"}.
 {author,"Ulf Wiger, QPQ AG"}.
-{package_id,{"uwiger","gmhive_client",{0,9,1}}}.
+{package_id,{"uwiger","gmhive_client",{0,9,2}}}.
 {deps,[{"uwiger","gmhive_protocol",{0,2,0}},
        {"uwiger","gmhive_worker",{0,5,1}},
        {"uwiger","gmcuckoo",{1,2,4}},