From 4b4d0f8f02e095e240bdc3e0045d3edbacfb9fa7 Mon Sep 17 00:00:00 2001 From: Ulf Wiger Date: Wed, 19 Aug 2026 08:38:13 +0200 Subject: [PATCH] Keep a second session on new_server and skip Eureka-retry apply_new_server opens a keep connector to the advertised host/port. A dead primary does not Eureka-retry while another keep session is up. Reconnect preserves host/port. Dialyzer: host type, watchdog tref, worker config, zx ppath. --- rebar.config | 4 +- rebar.lock | 15 ++++-- src/gmhc_connector.erl | 24 +++++++--- src/gmhc_connectors_sup.erl | 4 +- src/gmhc_eureka.erl | 2 +- src/gmhc_handler.erl | 94 ++++++++++++++++++++++++++++++------- src/gmhc_server.erl | 6 ++- src/gmhc_watchdog.erl | 2 +- 8 files changed, 119 insertions(+), 32 deletions(-) diff --git a/rebar.config b/rebar.config index 8ee56cc..9cb2525 100644 --- a/rebar.config +++ b/rebar.config @@ -12,9 +12,11 @@ {deps, [ {enoise, {git, "https://git.qpq.swiss/QPQ-AG/enoise.git", {ref, "029292817e"}}}, + {gm_enoise_session, {git, "https://git.qpq.swiss/QPQ-AG/gm_enoise_session.git", + {ref, "38b84133c7"}}}, {gmhive_protocol, {git, "https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git", - {ref, "9eec871"}}}, + {ref, "1852b20"}}}, {gmhive_worker, {git, "https://git.qpq.swiss/QPQ-AG/gmhive_worker", {ref, "cabd104114"}}}, {gmconfig, {git, "https://git.qpq.swiss/QPQ-AG/gmconfig.git", {ref, "38620ff9e2"}}}, diff --git a/rebar.lock b/rebar.lock index fa1d758..bff6806 100644 --- a/rebar.lock +++ b/rebar.lock @@ -15,6 +15,10 @@ {git,"https://git.qpq.swiss/QPQ-AG/enoise.git", {ref,"029292817ea1c685b6377bfae667976c0f4a36bc"}}, 0}, + {<<"gm_enoise_session">>, + {git,"https://git.qpq.swiss/QPQ-AG/gm_enoise_session.git", + {ref,"38b84133c7f09ee33daca3514d7770b9ac640445"}}, + 0}, {<<"gmconfig">>, {git,"https://git.qpq.swiss/QPQ-AG/gmconfig.git", {ref,"38620ff9e2289dff121377c382e4d5b08c719d87"}}, @@ -25,7 +29,7 @@ 1}, {<<"gmhive_protocol">>, {git,"https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git", - {ref,"9eec871e954b729cc5c9249da81abb91de03ad39"}}, + {ref,"1852b20fe6666cd34ac08ec2bdf62540d817e8d0"}}, 0}, {<<"gmhive_worker">>, {git,"https://git.qpq.swiss/QPQ-AG/gmhive_worker", @@ -33,9 +37,10 @@ 0}, {<<"gmserialization">>, {git,"https://git.qpq.swiss/QPQ-AG/gmserialization.git", - {ref,"0288719ae15814f3a53114c657502a24376bebfa"}}, + {ref,"dda5cac7a91ba0ad105815ec3cb1de551a04d719"}}, 1}, {<<"gproc">>,{pkg,<<"gproc">>,<<"1.0.0">>},0}, + {<<"ranch">>,{pkg,<<"ranch">>,<<"1.8.0">>},1}, {<<"setup">>, {git,"https://github.com/uwiger/setup", {ref,"0430661811269d007285f817467cbab7c4c31e96"}}, @@ -47,7 +52,9 @@ 0}]}. [ {pkg_hash,[ - {<<"gproc">>, <<"AA9EC57F6C9FF065B16D96924168D7C7157CD1FD457680EFE4B1274F456FA500">>}]}, + {<<"gproc">>, <<"AA9EC57F6C9FF065B16D96924168D7C7157CD1FD457680EFE4B1274F456FA500">>}, + {<<"ranch">>, <<"8C7A100A139FD57F17327B6413E4167AC559FBC04CA7448E9BE9057311597A1D">>}]}, {pkg_hash_ext,[ - {<<"gproc">>, <<"109F253C2787DE8A371A51179D4973230CBEC6239EE673FA12216A5CE7E4F902">>}]} + {<<"gproc">>, <<"109F253C2787DE8A371A51179D4973230CBEC6239EE673FA12216A5CE7E4F902">>}, + {<<"ranch">>, <<"49FBCFD3682FAB1F5D109351B61257676DA1A2FDBE295904176D5E521A2DDFE5">>}]} ]. diff --git a/src/gmhc_connector.erl b/src/gmhc_connector.erl index dfcf33a..fc9bb40 100644 --- a/src/gmhc_connector.erl +++ b/src/gmhc_connector.erl @@ -45,8 +45,13 @@ , nowait => boolean() %% default: false , tcp_opts => list() , enoise_opts => list() %% if present, MUST contain {'noise', _} - , tcp_opts => list() , connect_timeout => pos_integer() + , type => worker | monitor + , pubkey => binary() | string() + , extra => list() + , extra_pubkeys => list() + , pool_id => binary() | string() + , id => id() }. -export_type([ id/0 @@ -150,7 +155,7 @@ init(#{id := Id} = Opts) when is_map(Opts) -> S0#st{econn = undefined, reconnect = false}; {error, _} = Error -> ?LOG_WARNING("Could not connect to core server: ~p", [Error]), - start_reconnect_timer(S0#st{econn = undefined}) + start_reconnect_timer(S0#st{econn = undefined, opts = Opts}, Opts) end; false -> S0 @@ -289,21 +294,27 @@ try_connect(Opts, S) -> end. try_connect_(Opts0, S) -> - case eureka_get_host_port() of + case connect_pool_opts(Opts0) of {error, _} = Error -> Error; PoolOpts when is_map(PoolOpts) -> - case try_noise_connect(maps:merge(Opts0, PoolOpts)) of + %% Explicit host/port (new_server) win over Eureka / config. + case try_noise_connect(maps:merge(PoolOpts, Opts0)) of {ok, EConn, Opts1} -> S1 = protocol_connect(Opts1, S#st{ econn = EConn , status = connecting - , reconnect_timer = undefined }), + , reconnect_timer = undefined }), {ok, S1#st{status = connected}}; {error, _} = Error -> Error end end. +connect_pool_opts(#{host := _, port := _} = Opts) -> + maps:with([host, port, pool_id], Opts); +connect_pool_opts(_) -> + eureka_get_host_port(). + eureka_get_host_port() -> case gmhc_eureka:get_pool_address() of {ok, #{host := Host, @@ -437,7 +448,8 @@ start_timer(T, Opts0) -> erlang:start_timer(T, self(), {reconnect, Opts}). retry_opts(Opts) -> - maps:with([deadline], Opts). + maps:with([deadline, host, port, pool_id, type, pubkey, + extra, extra_pubkeys], Opts). protocol_connect(Opts, #st{econn = EConn} = S) -> Pubkey = to_bin(opt(pubkey, Opts, [<<"pubkey">>])), diff --git a/src/gmhc_connectors_sup.erl b/src/gmhc_connectors_sup.erl index e30f33b..5718eb6 100644 --- a/src/gmhc_connectors_sup.erl +++ b/src/gmhc_connectors_sup.erl @@ -17,11 +17,11 @@ start_link() -> case supervisor:start_link({local, ?MODULE}, ?MODULE, []) of {ok, _} = Ok -> RI0 = gmhc_watchdog:get_restart_info(), - ?LOG_ERROR("Restart info: ~p", [RI0]), + ?LOG_DEBUG("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]), + ?LOG_INFO("Sorted restart info: ~p", [RI]), maps:foreach(fun restart_connector/2, RI), Ok; Other -> diff --git a/src/gmhc_eureka.erl b/src/gmhc_eureka.erl index 95e3f3b..2f458d4 100644 --- a/src/gmhc_eureka.erl +++ b/src/gmhc_eureka.erl @@ -120,7 +120,7 @@ cache_dir() -> cache_dir_zomp() -> #{package_id := {Realm, App, _}} = zx_daemon:meta(), - filename:join(zx_lib:ppath(var, {Realm, App}), "gmhive.cache"). + filename:join(zx_lib:ppath(var, {Realm, App, {z, z, z}}), "gmhive.cache"). get_pool_address_() -> case gmconfig:find_config([<<"pool_admin">>, <<"url">>], [user_config]) of diff --git a/src/gmhc_handler.erl b/src/gmhc_handler.erl index 79ed001..6c1f4dc 100644 --- a/src/gmhc_handler.erl +++ b/src/gmhc_handler.erl @@ -19,10 +19,10 @@ -record(pool, { id :: gmhc_connector:id() , pid :: pid() | 'undefined' - , mref :: reference() + , mref :: reference() | 'undefined' , connected = false :: boolean() , keep = true :: boolean() - , host :: string() + , host :: binary() | string() , port :: pos_integer() , opts = #{} :: map() }). @@ -60,6 +60,7 @@ start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). init([]) -> + gmhc_events:ensure_subscribed(disconnected), {ok, #st{}}. handle_call({call, Req}, _From, #st{} = S) -> @@ -71,13 +72,14 @@ handle_call(_Req, _From, S) -> handle_cast({pool_connected, Id, Pid, Opts}, #st{pools = Pools} = S) -> MRef = erlang:monitor(process, Pid), + S1 = remember_connect_opts(Opts, S), case lists:keyfind(Id, #pool.id, Pools) of #pool{} = P -> P1 = P#pool{ connected = true , pid = Pid , mref = MRef }, Pools1 = lists:keyreplace(Id, #pool.id, Pools, P1), - {noreply, S#st{pools = Pools1}}; + {noreply, S1#st{pools = Pools1}}; false -> P = #pool{ id = Id , pid = Pid @@ -85,7 +87,7 @@ handle_cast({pool_connected, Id, Pid, Opts}, #st{pools = Pools} = S) -> , host = maps:get(host, Opts) , port = maps:get(port, Opts) , opts = Opts }, - {noreply, S#st{pools = [P | Pools]}} + {noreply, S1#st{pools = [P | Pools]}} end; handle_cast({notify, Msg}, #st{} = S) -> notify_connector(Msg), @@ -99,11 +101,20 @@ handle_info({from_pool, Msg}, S) -> #{notification := #{new_server := #{ host := _ , port := _ , keep := _ } = Server}} -> - {noreply, start_pool_connector(Server, S)}; + {noreply, apply_new_server(Server, S)}; _ -> gmhc_server:from_pool(Msg), {noreply, S} end; +handle_info({gproc_ps_event, disconnected, #{info := #{id := Id}}}, S) -> + {noreply, on_pool_down(Id, S)}; +handle_info({'DOWN', _MRef, process, Pid, _}, #st{pools = Pools} = S) -> + case lists:keyfind(Pid, #pool.pid, Pools) of + #pool{id = Id} -> + {noreply, on_pool_down(Id, S)}; + false -> + {noreply, S} + end; handle_info(Msg, S) -> ?LOG_DEBUG("Unknown msg: ~p", [Msg]), {noreply, S}. @@ -169,19 +180,35 @@ notify_connector(Msg0) -> {Via, Msg} = maps:take(via, Msg0), gmhc_connector:send(Via, #{notification => #{msg => Msg}}). -start_pool_connector(#{ host := Host - , port := Port - , keep := Keep }, #st{pools = Pools, opts = Opts} = S) -> +apply_new_server(#{host := Host0, port := Port, keep := false} = _Server, + #st{pools = Pools} = S) -> + Host = norm_host(Host0), case [P || #pool{host = H1, port = P1} = P <- Pools, - H1 == Host, - P1 == Port] of + norm_host(H1) == Host, P1 == Port] of + [#pool{id = Id} | _] -> + catch gmhive_client:disconnect(Id), + S#st{pools = [P || #pool{id = I} = P <- Pools, I =/= Id]}; [] -> - case gmhive_client:connect(Opts#{ host => Host - , port => Port - , nowait => true }) of + S + end; +apply_new_server(#{host := Host0, port := Port, keep := Keep}, + #st{pools = Pools, opts = Opts} = S) + when is_integer(Port), Port > 0 -> + Host = norm_host(Host0), + case [P || #pool{host = H1, port = P1} = P <- Pools, + norm_host(H1) == Host, P1 == Port] of + [] -> + ConnOpts = Opts#{ host => host_to_list(Host) + , port => Port + , type => worker + , nowait => true }, + case gmhive_client:connect(ConnOpts) of {ok, CId} -> Pid = gmhc_connector:whereis_id(CId), - MRef = erlang:monitor(process, Pid), + MRef = case is_pid(Pid) of + true -> erlang:monitor(process, Pid); + false -> undefined + end, P = #pool{ id = CId, host = Host, port = Port, keep = Keep, pid = Pid, mref = MRef }, S#st{pools = [P | Pools]}; @@ -190,7 +217,42 @@ start_pool_connector(#{ host := Host [Host, Port, Error]), S end; - [#pool{} = _P] -> + [#pool{} = P] -> + Pools1 = lists:keyreplace(P#pool.id, #pool.id, Pools, + P#pool{keep = Keep}), ?LOG_DEBUG("Already have a pool entry for ~p:~p", [Host, Port]), - S + S#st{pools = Pools1} end. + +remember_connect_opts(Opts, #st{opts = Old} = S) when is_map(Opts) -> + Keep = maps:with([pubkey, extra, extra_pubkeys, pool_id, type], Opts), + S#st{opts = maps:merge(Old, Keep)}; +remember_connect_opts(_, S) -> + S. + +on_pool_down(Id, #st{pools = Pools} = S) -> + Pools1 = case lists:keyfind(Id, #pool.id, Pools) of + #pool{} = P -> + lists:keyreplace(Id, #pool.id, Pools, + P#pool{connected = false, pid = undefined}); + false -> + Pools + end, + case has_kept_connection(Id, Pools1) of + true -> + %% Another standby/keep session is up — do not Eureka-retry this one. + catch gmhc_connector:disconnect(Id); + false -> + ok + end, + S#st{pools = Pools1}. + +has_kept_connection(ExceptId, Pools) -> + lists:any(fun(#pool{id = I, connected = C, keep = K}) -> + I =/= ExceptId andalso C andalso K + end, Pools). + +norm_host(H) when is_list(H) -> list_to_binary(H); +norm_host(H) when is_binary(H) -> H. + +host_to_list(H) when is_binary(H) -> binary_to_list(H). diff --git a/src/gmhc_server.erl b/src/gmhc_server.erl index 787cb48..6555b22 100644 --- a/src/gmhc_server.erl +++ b/src/gmhc_server.erl @@ -410,6 +410,9 @@ pick_nonce(N, [A, B]) when A < B -> %% Dialyzer doesn't like that the fun passed to spawn_link/1 %% doesn't have a local return (it communicates its result via the exit reason). -dialyzer({no_return, spawn_worker/2}). +%% Miner port success typing is only {error,{runtime,_}}; the ok / no_solution +%% clauses and worker_result/2 are the live path. +-dialyzer({nowarn_function, [init_worker/5, worker_result/2]}). spawn_worker(#worker{nonce = undefined} = W, _) -> W; @@ -430,7 +433,8 @@ spawn_worker(W, _) -> % Worker already has work. Don't disturb it. W. --spec init_worker(binary(), integer(), integer(), tuple(), pid()) -> no_return(). +-spec init_worker(binary(), integer(), integer(), + gmhw_pow_cuckoo:config(), pid()) -> no_return(). init_worker(Data, Nonce, Target, Config, Parent) -> gmhc_events:publish(puzzle, {Data, Target, Nonce, Config}), Res = gmhc_workers:generate_from_hash(Data, Target, Nonce, Config, undefined), diff --git a/src/gmhc_watchdog.erl b/src/gmhc_watchdog.erl index 36d8213..9d34dba 100644 --- a/src/gmhc_watchdog.erl +++ b/src/gmhc_watchdog.erl @@ -21,7 +21,7 @@ -record(svc, { n = 5 :: pos_integer() , n0 = 5 :: pos_integer() , interval = 5000 :: pos_integer() - , tref :: reference() + , tref :: reference() | undefined , mref :: reference() }).