Merge pull request 'Cache connection info for faster and more robust reconnect' (#17) from uw-cache-conn into master

Reviewed-on: #17
Reviewed-by: Craig Everett <craigeverett@qpq.swiss>

.gitignore

@@ -19,7 +19,7 @@ VERSION
 *.aes~
 *.config~
 *.args~
-data/mnesia
+data/*
 _checkouts*/
 tmp/
 rebar3.crashdump

Makefile

@@ -0,0 +1,10 @@
+
+BUILD=_build/default/lib/gmhive_client
+SCHEMA_OUT=$(BUILD)/priv/gmhc_schema.json
+
+schema:
+	ERL_LIBS=_build/default/lib \
+	erl -run gmhc_config_schema export $(SCHEMA_OUT) -run init stop
+	scripts/update_readme_json.sh "README.md" "$(SCHEMA_OUT)"
+
+

README.md

@@ -82,6 +82,20 @@ only does this inform the client of the network for which the client should
 mine, but it also changes what URL the client will use to find the hive
 server. (For `"testnet"` this will be `"test.gajumining.com"`.)
 
+### Caching of connection data
+
+The client retrieves the Hive Server connection info from gajumining.com.
+This information is account-specific, and may change over time.
+
+To speed up the reconnecting process, should the Hive Server restart,
+this connection information is cached locally for up to 24 hours.
+The normal place for the cached data would be
+`$ZOMP_DIR//var/uwiger/gmhive_client/Vsn/setup.data/mainnet/gmhc_eureka.XXXXXXXX.cache`,
+where `Vsn` is the current version of `gmhive_client`, and `XXXXXXXX` is a fragment
+of the current account ID. The data location can be changed with the command-line
+option `-setup data_dir Dir`.
+
+
 ## JSON Schema
 
 ```json
@@ -89,6 +103,10 @@ server. (For `"testnet"` this will be `"test.gajumining.com"`.)
     "$schema": "http://json-schema.org/draft-04/schema#",
     "additionalProperties": false,
     "properties": {
+        "cache_dir": {
+            "description": "Location of cache, default is 'setup:data_dir()'",
+            "type": "string"
+        },
         "extra_pubkeys": {
             "default": [],
             "description": "Additional worker pubkeys, sharing rewards",
@@ -155,6 +173,16 @@ server. (For `"testnet"` this will be `"test.gajumining.com"`.)
             "pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
             "type": "string"
         },
+        "report": {
+            "default": "silent",
+            "description": "Progress reporting",
+            "enum": [
+                "debug",
+                "progress",
+                "silent"
+            ],
+            "type": "string"
+        },
         "type": {
             "default": "worker",
             "description": "monitor mode can be used to see if a pool is alive",
@@ -174,7 +202,7 @@ server. (For `"testnet"` this will be `"test.gajumining.com"`.)
                 "properties": {
                     "executable": {
                         "default": "mean29-generic",
-                        "description": "Executable binary of the worker. Can be a fully qualified path, but the software may apply default logic to locate a plain basename.",
+                        "description": "Executable binary of the worker. Can be a fully qualified path, \nbut the software may apply default logic to locate a plain basename.",
                         "type": "string"
                     },
                     "extra_args": {

ebin/gmhive_client.app

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

gmhc_config-testnet-local.eterm

@@ -1,4 +1,5 @@
 #{ pubkey => <<"ak_zjyvDmMbXafMADYrxDs4uMiYM5zEVJBqWgv619NUQ37Gkwt7z">>
+ , report => <<"progress">>
  , pool_admin => #{url => <<"local">>}
  , pool => #{id => <<"ct_26xqeE3YKmZV8jrks57JSgZRCHSuG4RGzpnvdz6AAiSSTVbJRM">>,
              host => <<"127.0.0.1">>}

gmhc_config.eterm

@@ -1,5 +1,5 @@
 #{ pubkey => <<"ak_2KAcA2Pp1nrR8Wkt3FtCkReGzAi8vJ9Snxa4PcmrthVx8AhPe8">>
  , pool => #{id => <<"ct_LRbi65kmLtE7YMkG6mvG5TxAXTsPJDZjAtsPuaXtRyPA7gnfJ">>}
- , mining =>
+ , workers =>
        [#{executable => <<"mean15-generic">>}]
  }.

priv/gmhc_schema.json

@@ -0,0 +1,139 @@
+{
+    "$schema": "http://json-schema.org/draft-04/schema#",
+    "additionalProperties": false,
+    "properties": {
+        "cache_dir": {
+            "description": "Location of cache, default is 'setup:data_dir()'",
+            "type": "string"
+        },
+        "extra_pubkeys": {
+            "default": [],
+            "description": "Additional worker pubkeys, sharing rewards",
+            "items": {
+                "pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
+                "type": "string"
+            },
+            "type": "array"
+        },
+        "network": {
+            "default": "mainnet",
+            "type": "string"
+        },
+        "pool": {
+            "additionalProperties": false,
+            "properties": {
+                "host": {
+                    "default": "127.0.0.1",
+                    "description": "Hostname of hive server",
+                    "example": "0.0.0.0",
+                    "type": "string"
+                },
+                "id": {
+                    "description": "Pool contract id",
+                    "pattern": "^ct_[1-9A-HJ-NP-Za-km-z]*$",
+                    "type": "string"
+                },
+                "port": {
+                    "default": 17888,
+                    "description": "Hive server listen port",
+                    "minimum": 1,
+                    "type": "integer"
+                }
+            },
+            "type": "object"
+        },
+        "pool_admin": {
+            "additionalProperties": false,
+            "properties": {
+                "default_per_network": {
+                    "additionalProperties": false,
+                    "properties": {
+                        "mainnet": {
+                            "default": "https://gajumining.com/api/workers/{CLIENT_ID}",
+                            "type": "string"
+                        },
+                        "testnet": {
+                            "default": "https://test.gajumining.com/api/workers/{CLIENT_ID}",
+                            "type": "string"
+                        }
+                    },
+                    "type": "object"
+                },
+                "url": {
+                    "default": "https://test.gajumining.com/api/workers/{CLIENT_ID}",
+                    "description": "URL of Eureka worker api",
+                    "type": "string"
+                }
+            },
+            "type": "object"
+        },
+        "pubkey": {
+            "description": "Primary client pubkey",
+            "pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
+            "type": "string"
+        },
+        "report": {
+            "default": "silent",
+            "description": "Progress reporting",
+            "enum": [
+                "debug",
+                "progress",
+                "silent"
+            ],
+            "type": "string"
+        },
+        "type": {
+            "default": "worker",
+            "description": "monitor mode can be used to see if a pool is alive",
+            "enum": [
+                "worker",
+                "monitor"
+            ],
+            "type": "string"
+        },
+        "workers": {
+            "default": [
+                { "executable": "mean29-generic" }
+            ],
+            "description": "Definitions of workers' configurations. If no worker are configured one worker is used as default, i.e. 'mean29-generic' executable without any extra args.",
+            "items": {
+                "additionalProperties": false,
+                "properties": {
+                    "executable": {
+                        "default": "mean29-generic",
+                        "description": "Executable binary of the worker. Can be a fully qualified path, \nbut the software may apply default logic to locate a plain basename.",
+                        "type": "string"
+                    },
+                    "extra_args": {
+                        "default": "",
+                        "description": "Extra arguments to pass to the worker executable binary. The safest choice is specifying no arguments i.e. empty string.",
+                        "type": "string"
+                    },
+                    "hex_encoded_header": {
+                        "default": false,
+                        "description": "Hexadecimal encode the header argument that is send to the worker executable. CUDA executables expect hex encoded header.",
+                        "type": "boolean"
+                    },
+                    "instances": {
+                        "description": "Instances used by the worker in case of Multi-GPU mining. Numbers on the configuration list represent GPU devices that are to be addressed by the worker.",
+                        "example": [0,1,2,3],
+                        "items": { "type": "integer" },
+                        "minItems": 1,
+                        "type": "array"
+                    },
+                    "repeats": {
+                        "default": 1,
+                        "description": "Number of tries to do in each worker context - WARNING: it should be set so the worker process runs for 3-5s or else the node risk missing out on new micro blocks.",
+                        "type": "integer"
+                    }
+                },
+                "required": [
+                    "executable"
+                ],
+                "type": "object"
+            },
+            "type": "array"
+        }
+    },
+    "type": "object"
+}

rebar.config

@@ -5,11 +5,13 @@
 {erl_opts, [debug_info]}.
 {plugins, [rebar3_hex]}.
 
+{post_hooks, [{compile, "make schema"}]}.
+
 {deps, [
         {enoise, {git, "https://git.qpq.swiss/QPQ-AG/enoise.git", {ref, "029292817e"}}},
         {gmhive_protocol,
          {git, "https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
-          {ref, "818ce33"}}},
+          {ref, "8d4652a"}}},
         {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"}}},

rebar.lock

@@ -25,7 +25,7 @@
   1},
  {<<"gmhive_protocol">>,
   {git,"https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
-       {ref,"818ce33cc1dec74c020515be48fb548a5207befd"}},
+       {ref,"8d4652a79a2ad8f51e1fe560c15c698d4c92485c"}},
   0},
  {<<"gmhive_worker">>,
   {git,"https://git.qpq.swiss/QPQ-AG/gmhive_worker",

scripts/update_readme_json.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+README="$1"
+SCHEMA="$2"
+
+if [[ ! -f "$README" || ! -f "$SCHEMA" ]]; then
+  echo "Usage: $0 README.md schema.json"
+  exit 1
+fi
+
+tmpfile=$(mktemp)
+
+awk -v schema_file="$SCHEMA" '
+  BEGIN {
+    in_json_block = 0
+    in_schema_section = 0
+  }
+
+  /^##[[:space:]]+JSON[[:space:]]+Schema/ {
+    in_schema_section = 1
+    print
+    next
+  }
+
+  /^##[[:space:]]+/ && in_schema_section {
+    # Another section starts, end the schema section
+    in_schema_section = 0
+  }
+
+  in_schema_section && /^```json/ {
+    print "```json"
+    while ((getline line < schema_file) > 0) print line
+    close(schema_file)
+    in_json_block = 1
+    next
+  }
+
+  in_json_block && /^```$/ {
+    print "```"
+    in_json_block = 0
+    next
+  }
+
+  !(in_schema_section && in_json_block) {
+    print
+  }
+' "$README" > "$tmpfile"
+
+mv "$tmpfile" "$README"
+echo "✅ Updated JSON schema block in $README"

src/gmhc_app.erl

@@ -1,6 +1,6 @@
 %% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
 -module(gmhc_app).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -behaviour(application).
 

src/gmhc_config.erl

@@ -1,5 +1,5 @@
 -module(gmhc_config).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([ load_config/0
         , get_config/1

src/gmhc_config_schema.erl

@@ -1,8 +1,9 @@
 -module(gmhc_config_schema).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([ schema/0
-        , to_json/0 ]).
+        , to_json/0
+        , export/1 ]).
 
 -import(gmconfig_schema_helpers,
         [ str/1
@@ -22,6 +23,17 @@
 to_json() ->
     json:encode(schema()).
 
+export(ToFile) ->
+    case file:open(ToFile, [write]) of
+        {ok, Fd} ->
+            try ok = io:put_chars(Fd, json:format(schema(), #{indent => 4}))
+            after
+                file:close(Fd)
+            end;
+        {error, _} = Error ->
+            Error
+    end.
+
 schema() ->
     obj(schema_init(),
         #{
@@ -39,6 +51,7 @@ schema() ->
          , pool => pool()
          , pool_admin => pool_admin()
          , workers => workers()
+         , cache_dir => str(#{description => <<"Location of cache, default is 'setup:data_dir()'">>})
          , report => str(#{ enum  => [<<"debug">>, <<"progress">>, <<"silent">>]
                           , default => <<"silent">>
                           , description => <<"Progress reporting">> })

src/gmhc_connector.erl

@@ -1,5 +1,5 @@
 -module(gmhc_connector).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -behaviour(gen_server).
 
@@ -56,7 +56,10 @@
               id :: non_neg_integer()
             , auto_connect = true :: boolean()
             , econn
+            , connected = false :: boolean()
+            , reconnect = true :: boolean()
             , reconnect_timer :: timer_ref() | 'undefined'
+            , recache_timer   :: reference() | 'undefined'
             , awaiting_connect = [] :: list()
             , protocol   :: binary() | 'undefined'
             , version    :: binary() | 'undefined'
@@ -141,6 +144,9 @@ init(#{id := Id} = Opts) when is_map(Opts) ->
                     {ok, S} ->
                         ?LOG_DEBUG("Initial connect succeeded", []),
                         S;
+                    {error, rejected} ->
+                        ?LOG_WARNING("Connection rejected; will not retry", []),
+                        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})
@@ -179,6 +185,7 @@ handle_call({connect, Opts}, From, #st{awaiting_connect = Waiters} = S) ->
                                end,
                     S1 = start_reconnect_timer(
                            S#st{ auto_connect = true
+                               , reconnect = true
                                , awaiting_connect = Waiters1 }, Opts),
                     {noreply, S1};
                 false ->
@@ -239,12 +246,22 @@ handle_info({timeout, TRef, {reconnect, Opts}}, #st{ reconnect_timer = {TRef, _,
     end;
 handle_info({tcp_closed, _Port}, #st{} = S) ->
     ?LOG_DEBUG("got tcp_closed", []),
-    disconnected(S#st.id),
-    S1 = case S#st.auto_connect of
-             true  -> start_reconnect_timer(S#st{econn = undefined});
-             false -> S#st{econn = undefined}
+    S1 = disconnected(S#st.id, S),
+    S2 = case S1#st.auto_connect of
+             true  -> start_reconnect_timer(S1#st{econn = undefined});
+             false -> S1#st{econn = undefined}
          end,
-    {noreply, S1};
+    {noreply, S2};
+handle_info({timeout, _, recache}, #st{opts = Opts, econn = EConn} = S0) ->
+    S = S0#st{recache_timer = undefined},
+    case (EConn =/= undefined) andalso S#st.connected of
+        false ->
+            {noreply, S};
+        true ->
+            ?LOG_NOTICE("Recaching eureka info", []),
+            cache_eureka_info(Opts),
+            {noreply, ensure_recache_timer(S)}
+    end;
 handle_info(Msg, S) ->
     ?LOG_DEBUG("Discarding msg (auto_connect=~p): ~p", [S#st.auto_connect, Msg]),
     {noreply, S}.
@@ -264,6 +281,8 @@ code_change(_FromVsn, S, _Extra) ->
 try_connect(Opts, S) ->
     try try_connect_(Opts, S)
     catch
+        error:rejected:_ ->
+            {error, rejected};
         error:E:T ->
             ?LOG_ERROR("Unexpected error connecting: ~p / ~p", [E, T]),
             {error, E}
@@ -286,12 +305,12 @@ try_connect_(Opts0, S) ->
 
 eureka_get_host_port() ->
     case gmhc_eureka:get_pool_address() of
-        #{<<"address">> := Host,
-          <<"port">>    := Port,
-          <<"pool_id">> := PoolId} ->
-            #{host => binary_to_list(Host),
+        {ok, #{host    := Host,
+               port    := Port,
+               pool_id := PoolId}} ->
+            #{host => unicode:characters_to_list(Host),
               port => Port,
-              pool_id => binary_to_list(PoolId)};
+              pool_id => unicode:characters_to_list(PoolId)};
         {error, _} = Error ->
             Error
     end.
@@ -333,6 +352,8 @@ default_tcp_opts() ->
 enoise_opts() ->
     [{noise, <<"Noise_NN_25519_ChaChaPoly_BLAKE2b">>}].
 
+start_reconnect_timer(#st{reconnect = false} = S) ->
+    S;
 start_reconnect_timer(#st{} = S) ->
     start_reconnect_timer(S, #{}).
 
@@ -344,11 +365,18 @@ start_reconnect_timer(#st{} = S, Opts) ->
         false ->
             ?LOG_DEBUG("starting reconnect timer ...", []),
             TRef = start_timer(1000, Opts),
-            S#st{reconnect_timer = {TRef, 10, 1000, Opts}}
+            S#st{reconnect_timer = {TRef, 10, 8000 + gmhc_lib:rand(3000), Opts}}
     end.
 
+restart_reconnect_timer(#st{reconnect = false} = S) ->
+    cancel_reconnect_timer(S);
 restart_reconnect_timer(#st{reconnect_timer = {_, 0, T, Opts}} = S) ->
-    NewT = max(T * 2, ?MAX_RETRY_INTERVAL),
+    NewT = max(T + 5000 + gmhc_lib:rand(1000), ?MAX_RETRY_INTERVAL),
+    if (NewT > T andalso NewT =:= ?MAX_RETRY_INTERVAL) ->
+            gmhc_eureka:invalidate_cache();
+       true ->
+            ok
+    end,
     TRef = start_timer(NewT, Opts),
     S#st{reconnect_timer = {TRef, 10, NewT, Opts}};
 restart_reconnect_timer(#st{reconnect_timer = {_, N, T, Opts}} = S) ->
@@ -373,11 +401,24 @@ notify_deadline(D, #st{awaiting_connect = Waiters} = S) ->
                     end, [], Waiters),
     S#st{awaiting_connect = Waiters1}.
 
-notify_connected(#st{id = Id, awaiting_connect = Waiters} = S) ->
+cache_eureka_info(Opts) ->
+    gmhc_eureka:cache_good_address(maps:with([host, port, pool_id], Opts)).
+
+notify_connected(#st{id = Id, awaiting_connect = Waiters, opts = Opts} = S) ->
     gmhc_events:publish(connected, #{id => Id}),
     [gen_server:reply(From, ok) || {From, _} <- Waiters],
     gmhc_handler:pool_connected(S#st.id, S#st.opts),
-    S#st{awaiting_connect = []}.
+    cache_eureka_info(Opts),
+    ensure_recache_timer(S#st{awaiting_connect = []}).
+
+ensure_recache_timer(#st{recache_timer = T} = S) ->
+    case T of
+        undefined ->
+            TRef = erlang:start_timer(1*60*1000, self(), recache),
+            S#st{recache_timer = TRef};
+        _ when is_reference(T) ->
+            S
+    end.
 
 cancel_reconnect_timer(#st{reconnect_timer = T} = S) ->
     case T of
@@ -411,7 +452,20 @@ protocol_connect(Opts, #st{econn = EConn} = S) ->
                   , nonces => gmhc_server:total_nonces()
                   , signature => ""},
     ?LOG_DEBUG("ConnectReq = ~p", [ConnectReq]),
-    Msg = gmhp_msgs:encode_connect(ConnectReq, RId),
+    try gmhp_msgs:encode_connect(ConnectReq, RId) of
+        Msg ->
+            send_connect(EConn, RId, Msg, ConnectReq, Opts, S)
+    catch error:Error ->
+            ErrMsg = unicode:characters_to_binary(io_lib:fwrite("~p", [Error])),
+            disconnected(S#st.id, #{error =>
+                                        #{code    => gmhp_msgs:error_code(invalid_input),
+                                          message => ErrMsg}}, S)
+    end.
+
+send_connect(EConn, RId, Msg, #{pubkey        := Pubkey,
+                                extra_pubkeys := Extra,
+                                pool_id       := PoolId,
+                                type          := Type}, Opts, S) ->
     enoise:send(EConn, Msg),
     receive
         {noise, EConn, Data} ->
@@ -420,17 +474,20 @@ protocol_connect(Opts, #st{econn = EConn} = S) ->
                             , result := #{connect_ack := #{ protocol := P
                                                           , version  := V }}
                             }} ->
-                    connected(S#st.id, Type),
+                    S1 = connected(S#st.id, Type, S),
                     Opts1 = Opts#{ pubkey  => Pubkey
                                  , extra   => Extra
                                  , pool_id => PoolId
                                  , type    => Type },
-                    notify_connected(S#st{protocol = P, version = V, opts = Opts1});
-                #{error := #{message := Msg}} ->
-                    ?LOG_ERROR("Connect error: ~s", [Msg]),
-                    error(protocol_connect)
+                    notify_connected(S1#st{protocol = P, version = V, opts = Opts1});
+                #{error := #{code := _, message := ErrMsg}} = ErrReply ->
+                    ?LOG_ERROR("Connect error: ~s", [ErrMsg]),
+                    disconnected(S#st.id, ErrReply, S),
+                    gmhc_eureka:invalidate_cache(),
+                    error(rejected)
             end
     after 10000 ->
+            gmhc_eureka:invalidate_cache(),
             error(protocol_connect_timeout)
     end.
 
@@ -443,12 +500,17 @@ to_atom(A) when is_atom(A) -> A;
 to_atom(S) ->
     binary_to_existing_atom(iolist_to_binary(S), utf8).
 
-connected(Id, Type) when Type==worker; Type==monitor ->
-    gmhc_server:connected(Id, Type).
+connected(Id, Type, S) when Type==worker; Type==monitor ->
+    gmhc_server:connected(Id, Type),
+    S#st{connected = true}.
 
-disconnected(Id) ->
-    gmhc_events:publish(disconnected, #{id => Id}),
-    gmhc_server:disconnected(Id).
+disconnected(Id, S) ->
+    disconnected(Id, #{}, S).
+
+disconnected(Id, Msg, S) ->
+    gmhc_events:publish(disconnected, Msg#{id => Id}),
+    gmhc_server:disconnected(Id),
+    S#st{connected = false}.
 
 opt_autoconnect(#{auto_connect := Bool}) when is_boolean(Bool) ->
     Bool;

src/gmhc_connectors_sup.erl

@@ -1,5 +1,5 @@
 -module(gmhc_connectors_sup).
--vsn("0.6.1").
+-vsn("0.8.3").
 -behavior(supervisor).
 
 -export([ start_link/0

src/gmhc_counters.erl

@@ -1,5 +1,5 @@
 -module(gmhc_counters).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([ initialize/0 ]).
 

src/gmhc_eureka.erl

@@ -1,20 +1,137 @@
 -module(gmhc_eureka).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([get_pool_address/0]).
+-export([cache_good_address/1,
+         invalidate_cache/0,
+         cached_address/0,
+         cache_filename/0,
+         cache_dir/0]).
 
 -include_lib("kernel/include/logger.hrl").
 -include("gmhc_events.hrl").
 
 get_pool_address() ->
+    case cached_address() of
+        {ok, _} = Ok -> Ok;
+        {error, _} ->
+            get_pool_address_()
+    end.
+
+cached_address() ->
+    I0 = cache_info(),
+    CacheF = cache_filename(I0),
+    ?LOG_DEBUG("Eureka cache filename: ~p", [CacheF]),
+    case file:read_file(CacheF) of
+        {ok, Bin} ->
+            NowTS = erlang:system_time(seconds),
+            OldestTS = NowTS - 24*60*60,
+            try binary_to_term(Bin) of
+                #{ ts      := TS
+                 , network := N
+                 , pubkey  := PK
+                 , host    := _
+                 , port    := _
+                 , pool_id := _} = Map when N  == map_get(network, I0),
+                                            PK == map_get(pubkey, I0) ->
+                    if TS >= OldestTS ->
+                            Result = maps:remove(ts, Map),
+                            ?LOG_DEBUG("Cached eureka info: ~p", [Result]),
+                            {ok, Result};
+                       true ->
+                            {error, outdated}
+                    end;
+                Other ->
+                    {error, {invalid_cache_term, Other}}
+            catch
+                error:E ->
+                    {error, {invalid_cache_data, E}}
+            end;
+        {error, _} = Err ->
+            Err
+    end.
+
+cache_good_address(#{host := _,
+                     port := _,
+                     pool_id := _} = I0) ->
+    CacheInfo = cache_info(I0),
+    CacheF = cache_filename(CacheInfo),
+    
+    ToCache = CacheInfo#{ts => erlang:system_time(seconds)},
+    case filelib:ensure_dir(CacheF) of
+        ok ->
+            case file:write_file(CacheF, term_to_binary(ToCache)) of
+                ok ->
+                    ?LOG_DEBUG("Cached eureka info in: ~p", [CacheF]),
+                    {ok, ToCache};
+                {error, _} = Err ->
+                    ?LOG_DEBUG("Couldn't cache eureka in ~p: ~p", [CacheF, Err]),
+                    Err
+            end;
+        {error, _} = Err ->
+            ?LOG_ERROR("Cannot save cached info to ~s", [CacheF]),
+            Err
+    end.
+
+invalidate_cache() ->
+    CacheF = cache_filename(),
+    case file:delete(CacheF) of
+        ok ->
+            ?LOG_DEBUG("Eureka cache file removed (~p)", [CacheF]),
+            ok;
+        {error, _} = Err ->
+            ?LOG_DEBUG("Couldn't remove Eureka cache (~p): ~p", [CacheF, Err]),
+            Err
+    end.
+
+cache_info(#{ host := Addr
+            , port := Port
+            , pool_id := PoolId }) ->
+    I0 = cache_info(),
+    I0#{ host => unicode:characters_to_binary(Addr)
+       , port => Port
+       , pool_id => unicode:characters_to_binary(PoolId)}.
+
+cache_info() ->
+    Pubkey = gmhc_config:get_config([<<"pubkey">>]),
+    Network = gmhc_config:get_config([<<"network">>]),
+    #{ pubkey => Pubkey
+     , network => Network }.
+
+cache_filename() ->
+    cache_filename(cache_info()).
+
+cache_filename(#{network := Network, pubkey := Pubkey}) ->
+    Path = filename:join(cache_dir(), Network),
+    <<"ak_", PKShort:8/binary, _/binary>> = Pubkey,
+    filename:join(Path, "gmhc_eureka." ++ binary_to_list(PKShort) ++ ".cache").
+
+cache_dir() ->
+    case gmconfig:find_config([<<"cache_dir">>]) of
+        {ok, D} ->
+            D;
+        undefined ->
+            case setup_zomp:is_zomp_context() of
+                true ->
+                    cache_dir_zomp();
+                false ->
+                    filename:join(setup:data_dir(), "gmhive.cache")
+            end
+    end.
+
+cache_dir_zomp() ->
+    #{package_id := {Realm, App, _}} = zx_daemon:meta(),
+    filename:join(zx_lib:ppath(var, {Realm, App}), "gmhive.cache").
+
+get_pool_address_() ->
     case gmconfig:find_config([<<"pool_admin">>, <<"url">>], [user_config]) of
         {ok, URL0} ->
             case expand_url(URL0) of
                 <<"local">> ->
-                    #{<<"address">> => <<"127.0.0.1">>,
-                      <<"port">>    => gmconfig:get_config(
-                                         [<<"pool">>, <<"port">>], [schema_default]),
-                      <<"pool_id">> => gmhc_config:get_config([<<"pool">>, <<"id">>]) };
+                    {ok, #{host    => <<"127.0.0.1">>,
+                           port    => gmconfig:get_config(
+                                        [<<"pool">>, <<"port">>], [schema_default]),
+                           pool_id => gmhc_config:get_config([<<"pool">>, <<"id">>]) }};
                 URL ->
                     ?LOG_INFO("Trying to connect to ~p", [URL]),
                     connect1(URL)
@@ -49,9 +166,13 @@ connect1(URL0) ->
             Error
     end.
 
-get_host_port(Data) ->
+get_host_port(#{ <<"address">> := Addr
+               , <<"port">>    := Port
+               , <<"pool_id">> := PoolId } = Data) ->
     ?LOG_DEBUG("Data = ~p", [Data]),
-    maps:with([<<"address">>, <<"port">>, <<"pool_id">>], Data).
+    {ok, #{ host => Addr
+          , port => Port
+          , pool_id => PoolId }}.
 
 request(get, URL) ->
     case request(get, URL, []) of

src/gmhc_events.erl

@@ -1,5 +1,5 @@
 -module(gmhc_events).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([subscribe/1,
          ensure_subscribed/1,

src/gmhc_handler.erl

@@ -1,5 +1,5 @@
 -module(gmhc_handler).
--vsn("0.6.1").
+-vsn("0.8.3").
 -behavior(gen_server).
 
 -export([ start_link/0

src/gmhc_lib.erl

@@ -0,0 +1,9 @@
+-module(gmhc_lib).
+-vsn("0.8.3").
+
+-export([ rand/1 ]).
+
+
+rand(Range) ->
+    <<Rx:32>> = crypto:strong_rand_bytes(4),
+    Range - (Rx rem Range).

src/gmhc_server.erl

@@ -1,5 +1,5 @@
 -module(gmhc_server).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -behaviour(gen_server).
 
@@ -162,16 +162,11 @@ handle_info({'EXIT', Pid, Reason}, #st{ workers = Workers
             {noreply, S}
     end;
 handle_info({timeout, _, check_workers}, #st{workers = Workers} = S) ->
-    case [W || #worker{cand = undefined} = W <- Workers] of
-        [] ->
-            {noreply, S};
-        Idle ->
-            S1 = maybe_request_nonces(S),
-            S2 = lists:foldl(fun(W, Sx) ->
-                                 maybe_restart_worker(W, Sx)
-                             end, S1, Idle),
-            {noreply, S2}
-    end;
+    S1 = maybe_request_nonces(S),
+    S2 = lists:foldl(fun(W, Sx) ->
+                             maybe_restart_worker(W, Sx)
+                     end, S1, Workers),
+    {noreply, S2};
 handle_info(Msg, St) ->
     ?LOG_DEBUG("Unknown msg: ~p", [Msg]),
     {noreply, St}.
@@ -216,7 +211,6 @@ maybe_request_nonces(#st{ candidate = #{via := Via, seq := Seq, nonces := Nonces
                         , nonces = N} = S) when ?CONNECTED(S) ->
     case Nonces == [] of
         true ->
-            %% ?LOG_DEBUG("Request more nonces, Seq = ~p, N = ~p", [Seq, N]),
             Res = gmhc_handler:call(#{via => Via,
                                        get_nonces => #{ seq => Seq
                                                       , n   => N }}),
@@ -230,7 +224,6 @@ maybe_request_nonces(S) ->
 nonces_result(#{nonces := #{seq := Seq, nonces := Nonces}}, Seq0, S) ->
     case Seq == Seq0 of
         true ->
-            %% ?LOG_DEBUG("Got nonces = ~p", [Nonces]),
             #st{candidate = Cand} = S,
             S#st{candidate = Cand#{nonces => Nonces}};
         false ->
@@ -240,8 +233,18 @@ nonces_result(#{nonces := #{seq := Seq, nonces := Nonces}}, Seq0, S) ->
 nonces_result({error, Reason}, Seq0, S) ->
     ?LOG_DEBUG("Got error on nonce request: ~p", [Reason]),
     Workers = stop_workers_for_seq(Seq0, S#st.workers),
+    case Reason of
+        {timeout, _} ->
+            Timeout = retry_timeout(1000, 3000),
+            erlang:start_timer(Timeout, self(), check_workers);
+        _ ->
+            ok
+    end,
     S#st{workers = Workers}.
 
+retry_timeout(Floor, Range) ->
+    Floor + gmhc_lib:rand(Range).
+
 handle_worker_result({worker_result, Result}, W, S) ->
     %% ?LOG_DEBUG("worker result: ~p", [Result]),
     case Result of

src/gmhc_sup.erl

@@ -1,6 +1,6 @@
 %% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
 -module(gmhc_sup).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -behaviour(supervisor).
 

src/gmhc_workers.erl

@@ -8,7 +8,7 @@
 %%%-------------------------------------------------------------------
 
 -module(gmhc_workers).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([
           get_worker_configs/0

src/gmhive_client.erl

@@ -1,5 +1,5 @@
 -module(gmhive_client).
--vsn("0.6.1").
+-vsn("0.8.3").
 
 -export([ connect/1
         , disconnect/1

zomp.meta

@@ -1,16 +1,16 @@
 {name,"gmhive_client"}.
 {type,app}.
 {modules,[]}.
-{author,"Ulf Wiger, QPQ AG"}.
 {prefix,"gmhc"}.
 {desc,"Gajumaru Hive Client"}.
-{package_id,{"uwiger","gmhive_client",{0,6,2}}}.
-{deps,[{"uwiger","gmhive_worker",{0,5,1}},
+{author,"Ulf Wiger, QPQ AG"}.
+{package_id,{"uwiger","gmhive_client",{0,8,3}}}.
+{deps,[{"uwiger","gmhive_protocol",{0,2,0}},
+       {"uwiger","gmhive_worker",{0,5,1}},
        {"uwiger","gmcuckoo",{1,2,4}},
        {"otpr","eblake2",{1,0,1}},
        {"otpr","base58",{0,1,1}},
        {"otpr","gmserialization",{0,1,3}},
-       {"uwiger","gmhive_protocol",{0,1,1}},
        {"uwiger","setup",{2,2,4}},
        {"uwiger","gproc",{1,0,1}},
        {"uwiger","gmconfig",{0,1,2}},

zompify.sh

@@ -38,5 +38,12 @@ rm "$IGNORE_TEMP"
 cp "$PWD/zomp.meta" "$DST/"
 cp "$PWD/Emakefile" "$DST/"
 
+# copy generated schema
+SCHEMA="$SRC/priv/gmhc_schema.json"
+if [ -e "$SCHEMA" ]; then
+    mkdir -p "$DST/priv"
+    cp -a "$SCHEMA" "$DST/priv/$(basename "$SCHEMA")"
+fi
+
 # Clean up beam files just in case
 [ -d "$DST/ebin" ] && find "$DST/ebin" -name '*.beam' -exec rm -f {} + || true