Add grisp-io protocol version handshake (#53)

 - Remove invalid configuration.
 - Fix dialyzer spec.
 - Add option to limit the number of connection retries, mostly for
   testing use-cases that will always fail to connect.
 - Implemente wait_connected in the client to support reaching maximum
   connection retries and get the last known error.

Author: sylane

config/dev.config

@@ -1,6 +1,5 @@
 [
     {grisp_cryptoauth, [
-        {tls_server_trusted_certs_cb, []},
         {tls_server_trusted_certs, {priv, grisp_connect, "server"}}
     ]},
 

config/local.config

@@ -4,7 +4,6 @@
     ]},
 
     {grisp_cryptoauth, [
-        {tls_server_trusted_certs_cb, []},
         {tls_server_trusted_certs, {priv, grisp_connect, "server"}},
         {tls_client_trusted_certs, {test, grisp_connect, "certs/CA.crt"}},
         {client_certs, {test, grisp_connect, "certs/client.crt"}},

config/test.config

@@ -4,7 +4,6 @@
     ]},
 
     {grisp_cryptoauth, [
-        {tls_server_trusted_certs_cb, []},
         {tls_client_trusted_certs, {test, grisp_connect, "certs/CA.crt"}},
         {client_certs, {test, grisp_connect, "certs/client.crt"}},
         {client_key, {test, grisp_connect, "certs/client.key"}},

rebar.lock

@@ -6,7 +6,7 @@
  {<<"gun">>,{pkg,<<"gun">>,<<"2.1.0">>},1},
  {<<"jarl">>,
   {git,"https://github.com/grisp/jarl.git",
-       {ref,"10085d38df19c67664d33ef61f515c92a8b0de56"}},
+       {ref,"24d53cc7b521b126588be1f36afecd1d4eb59db3"}},
   0},
  {<<"jsx">>,{pkg,<<"jsx">>,<<"3.1.0">>},0},
  {<<"mapz">>,{pkg,<<"mapz">>,<<"2.4.0">>},1}]}.

src/grisp_connect.app.src

@@ -25,6 +25,7 @@
         {ws_path, "/grisp-connect/ws"},
         {ws_request_timeout, 5_000},
         {ws_ping_timeout, 60_000},
+        {ws_max_retries, infinity},
         {logs_interval, 2_000},
         {logs_batch_size, 100},
         {logger, [

src/grisp_connect_api.erl

@@ -15,6 +15,7 @@
 % @doc Handles requests and notifications from grisp.io.
 -spec handle_msg(Msg) ->
     ok | {reply, Result :: term(), ReqRef :: binary() | integer()}
+    | {error, Code :: integer() | atom(), Message :: binary() | undefined, ErData :: term(), ReqRef :: binary() | integer()}
   when Msg :: {request, Method :: jarl:method(), Params :: map() | list(), ReqRef :: binary() | integer()}
             | {notification, jarl:method(), Params :: map() | list()}.
 handle_msg({notification, M, Params}) ->

src/grisp_connect_client.erl

@@ -13,6 +13,7 @@
 -export([start_link/0]).
 -export([connect/0]).
 -export([is_connected/0]).
+-export([wait_connected/1]).
 -export([request/3]).
 -export([notify/3]).
 
@@ -37,7 +38,10 @@
     ws_path :: binary(),
     ws_transport :: tcp | tls,
     conn :: undefined | pid(),
-    retry_count = 0 :: non_neg_integer()
+    retry_count = 0 :: non_neg_integer(),
+    last_error :: term(),
+    max_retries = infinity :: non_neg_integer() | infinity,
+    wait_calls = [] :: [gen_statem:from()]
 }).
 
 -type data() :: #data{}.
@@ -50,6 +54,7 @@
 
 %--- Macros --------------------------------------------------------------------
 
+-define(GRISP_IO_PROTOCOL, <<"grisp-io-v1">>).
 -define(FORMAT(FMT, ARGS), iolist_to_binary(io_lib:format(FMT, ARGS))).
 -define(STD_TIMEOUT, 1000).
 -define(CONNECT_TIMEOUT, 5000).
@@ -85,6 +90,11 @@ is_connected() ->
     catch exit:noproc -> false
     end.
 
+wait_connected(Timeout) ->
+    try gen_statem:call(?MODULE, ?FUNCTION_NAME, Timeout)
+    catch exit:noproc -> {error, noproc}
+    end.
+
 request(Method, Type, Params) ->
     gen_statem:call(?MODULE, {?FUNCTION_NAME, Method, Type, Params}).
 
@@ -107,11 +117,13 @@ init([]) ->
     Port = ?ENV(port, is_integer(V) andalso V >= 0 andalso V < 65536),
     WsTransport = ?ENV(ws_transport, V =:= tls orelse V =:= tcp),
     WsPath = ?ENV(ws_path, is_binary(V) orelse is_list(V), as_bin(V)),
+    MaxRetries = ?ENV(ws_max_retries, is_integer(V) orelse V =:= infinity),
     Data = #data{
         domain = Domain,
         port = Port,
         ws_transport = WsTransport,
-        ws_path = WsPath
+        ws_path = WsPath,
+        max_retries = MaxRetries
     },
     % The error list is put in a persistent term to not add noise to the state.
     persistent_term:put({?MODULE, self()}, generic_errors()),
@@ -133,8 +145,13 @@ callback_mode() -> [state_functions, state_enter].
 
 %--- Behaviour gen_statem State Callback Functions -----------------------------
 
-idle(enter, _OldState, _Data) ->
-    keep_state_and_data;
+idle(enter, _OldState,
+     Data = #data{wait_calls = WaitCalls, last_error = LastError}) ->
+    % When entering idle, we reply to all wait_connected calls with the last error
+    gen_statem:reply([{reply, F, {error, LastError}} || F <- WaitCalls]),
+    {keep_state, Data#data{wait_calls = [], last_error = undefined}};
+idle({call, From}, wait_connected, _) ->
+    {keep_state_and_data, [{reply, From, {error, not_connecting}}]};
 idle(cast, connect, Data) ->
     {next_state, waiting_ip, Data};
 ?HANDLE_COMMON.
@@ -152,14 +169,14 @@ waiting_ip(state_timeout, retry, Data = #data{retry_count = RetryCount}) ->
             {next_state, connecting, Data};
         invalid ->
             ?LOG_DEBUG(#{event => waiting_ip}),
-            {repeat_state, Data#data{retry_count = RetryCount + 1}}
+            {repeat_state, Data#data{retry_count = RetryCount + 1,
+                                     last_error = no_ip_available}}
     end;
 ?HANDLE_COMMON.
 
 connecting(enter, _OldState, Data) ->
     {keep_state, Data, [{state_timeout, 0, connect}]};
-connecting(state_timeout, connect,
-           Data = #data{conn = undefined, retry_count = RetryCount}) ->
+connecting(state_timeout, connect, Data = #data{conn = undefined}) ->
     ?LOG_INFO(#{description => <<"Connecting to grisp.io">>,
                 event => connecting}),
     case conn_start(Data) of
@@ -168,24 +185,25 @@ connecting(state_timeout, connect,
         {error, Reason} ->
             ?LOG_WARNING("Failed to connect to grisp.io: ~p", [Reason],
                          #{event => connection_failed, reason => Reason}),
-            {next_state, waiting_ip, Data#data{retry_count = RetryCount + 1}}
+            reconnect(Data, Reason)
     end;
-connecting(state_timeout, timeout, Data = #data{retry_count = RetryCount}) ->
+connecting(state_timeout, timeout, Data) ->
     Reason = connect_timeout,
     ?LOG_WARNING(#{description => <<"Timeout while connecting to grisp.io">>,
                    event => connection_failed, reason => Reason}),
-    Data2 = conn_close(Data, Reason),
-    {next_state, waiting_ip, Data2#data{retry_count = RetryCount + 1}};
-connecting(info, {jarl, Conn, connected}, Data = #data{conn = Conn}) ->
+    reconnect(conn_close(Data, Reason), Reason);
+connecting(info, {jarl, Conn, {connected, _}}, Data = #data{conn = Conn}) ->
     % Received from the connection process
     ?LOG_NOTICE(#{description => <<"Connected to grisp.io">>,
                   event => connected}),
     {next_state, connected, Data#data{retry_count = 0}};
 ?HANDLE_COMMON.
 
-connected(enter, _OldState, _Data) ->
+connected(enter, _OldState, Data = #data{wait_calls = WaitCalls}) ->
+    % When entering connected, we reply to all wait_connected calls with ok
+    gen_statem:reply([{reply, F, ok} || F <- WaitCalls]),
     grisp_connect_log_server:start(),
-    keep_state_and_data;
+    {keep_state, Data#data{wait_calls = [], last_error = undefined}};
 connected({call, From}, is_connected, _) ->
     {keep_state_and_data, [{reply, From, true}]};
 connected(info, {jarl, Conn, Msg}, Data = #data{conn = Conn}) ->
@@ -205,6 +223,9 @@ handle_common(cast, connect, State, _Data) when State =/= idle ->
     keep_state_and_data;
 handle_common({call, From}, is_connected, State, _) when State =/= connected ->
     {keep_state_and_data, [{reply, From, false}]};
+handle_common({call, From}, wait_connected, _State,
+              Data = #data{wait_calls = WaitCalls}) ->
+    {keep_state, Data#data{wait_calls = [From | WaitCalls]}};
 handle_common({call, From}, {request, _, _, _}, State, _Data)
 when State =/= connected ->
     {keep_state_and_data, [{reply, From, {error, disconnected}}]};
@@ -214,13 +235,12 @@ handle_common(cast, {notify, _Method, _Type, _Params}, _State, _Data) ->
 handle_common(info, reboot, _, _) ->
     init:stop(),
     keep_state_and_data;
-handle_common(info, {'EXIT', Conn, Reason}, _State,
-              Data = #data{conn = Conn, retry_count = RetryCount}) ->
+handle_common(info, {'EXIT', Conn, Reason}, _State, Data = #data{conn = Conn}) ->
     % The connection process died
     ?LOG_WARNING(#{description =>
                     ?FORMAT("The connection to grisp.io died: ~p", [Reason]),
                    event => connection_failed, reason => Reason}),
-    {next_state, waiting_ip, conn_died(Data#data{retry_count = RetryCount + 1})};
+    reconnect(conn_died(Data), Reason);
 handle_common(info, {'EXIT', _Conn, _Reason}, _State, _Data) ->
     % Ignore any EXIT from past jarl connections
     keep_state_and_data;
@@ -287,6 +307,20 @@ handle_connection_message(Data, Msg) ->
             keep_state_and_data
     end.
 
+reconnect(Data = #data{retry_count = RetryCount,
+                       max_retries = MaxRetries,
+                       last_error = LastError}, Reason)
+  when MaxRetries =/= infinity, RetryCount > MaxRetries ->
+    Error = case Reason of undefined -> LastError; E -> E end,
+    ?LOG_ERROR(#{description => <<"Max retries reached, giving up connecting to grisp.io">>,
+                 event => max_retries_reached, last_error => LastError}),
+    {next_state, idle, Data#data{retry_count = 0, last_error = Error}};
+reconnect(Data = #data{retry_count = RetryCount,
+                       last_error = LastError}, Reason) ->
+    Error = case Reason of undefined -> LastError; E -> E end,
+    {next_state, waiting_ip,
+     Data#data{retry_count = RetryCount + 1, last_error = Error}}.
+
 % Connection Functions
 
 conn_start(Data = #data{conn = undefined,
@@ -308,7 +342,8 @@ conn_start(Data = #data{conn = undefined,
         path => WsPath,
         errors => ErrorList,
         ping_timeout => WsPingTimeout,
-        request_timeout => WsReqTimeout
+        request_timeout => WsReqTimeout,
+        protocols => [?GRISP_IO_PROTOCOL]
     },
     case jarl:start_link(self(), ConnOpts) of
         {error, _Reason} = Error -> Error;

test/grisp_connect_api_SUITE.erl

@@ -32,15 +32,13 @@ all() ->
         lists:suffix("_test", atom_to_list(F))
     ].
 
-init_per_suite(Config) ->
-    CertDir = cert_dir(),
-    Apps = grisp_connect_test_server:start(CertDir),
-    [{apps, Apps} | Config].
-
-end_per_suite(Config) ->
-    grisp_connect_test_server:stop(?config(apps, Config)).
-
+init_per_testcase(TestCase, Config)
+  when TestCase =:= bad_client_version_test;
+       TestCase =:= bad_server_version_test  ->
+    Config;
 init_per_testcase(TestCase, Config) ->
+    CertDir = cert_dir(),
+    Apps = grisp_connect_test_server:start(#{cert_dir => CertDir}),
     {ok, _} = application:ensure_all_started(grisp_emulation),
     {ok, _} = application:ensure_all_started(grisp_connect),
     case TestCase of
@@ -49,16 +47,66 @@ init_per_testcase(TestCase, Config) ->
             ?assertEqual(ok, wait_connection()),
             grisp_connect_test_server:listen()
     end,
-    Config.
+    [{apps, Apps} | Config].
 
+end_per_testcase(TestCase, Config)
+    when TestCase =:= bad_client_version_test;
+         TestCase =:= bad_server_version_test  ->
+      Config;
 end_per_testcase(_, Config) ->
     ok = application:stop(grisp_connect),
     grisp_connect_test_server:wait_disconnection(),
     ?assertEqual([], flush()),
+    grisp_connect_test_server:stop(proplists:get_value(apps, Config)),
     Config.
 
 %--- Tests ---------------------------------------------------------------------
 
+bad_client_version_test(_) ->
+    CertDir = cert_dir(),
+    Apps = grisp_connect_test_server:start(#{
+        cert_dir => CertDir,
+        expected_protocol => <<"grisp-io-v42">>}),
+    try
+        {ok, _} = application:ensure_all_started(grisp_emulation),
+        application:load(grisp_connect),
+        application:set_env(grisp_connect, ws_max_retries, 2),
+        {ok, _} = application:ensure_all_started(grisp_connect),
+        try
+            ?assertMatch({error, ws_upgrade_failed}, wait_connection())
+        after
+            ok = application:stop(grisp_connect)
+        end
+    after
+        grisp_connect_test_server:wait_disconnection(),
+        ?assertEqual([], flush()),
+        grisp_connect_test_server:stop(Apps)
+    end,
+    ok.
+
+bad_server_version_test(_) ->
+    CertDir = cert_dir(),
+    Apps = grisp_connect_test_server:start(#{
+        cert_dir => CertDir,
+        selected_protocol => <<"grisp-io-v42">>}),
+    try
+        {ok, _} = application:ensure_all_started(grisp_emulation),
+        application:load(grisp_connect),
+        application:set_env(grisp_connect, ws_max_retries, 2),
+        {ok, _} = application:ensure_all_started(grisp_connect),
+        try
+            % There is no way to know the reason why gun closed the connection
+            ?assertMatch({error, {closed, _}}, wait_connection())
+        after
+            ok = application:stop(grisp_connect)
+        end
+    after
+        grisp_connect_test_server:wait_disconnection(),
+        ?assertEqual([], flush()),
+        grisp_connect_test_server:stop(Apps)
+    end,
+    ok.
+
 auto_connect_test(_) ->
     ?assertMatch(ok, wait_connection()).
 

test/grisp_connect_log_SUITE.erl

@@ -28,7 +28,7 @@ all() ->
 
 init_per_suite(Config) ->
     CertDir = cert_dir(),
-    Apps = grisp_connect_test_server:start(CertDir),
+    Apps = grisp_connect_test_server:start(#{cert_dir => CertDir}),
     [{apps, Apps} | Config].
 
 end_per_suite(Config) ->

test/grisp_connect_reconnect_SUITE.erl

@@ -36,7 +36,7 @@ end_per_suite(Config) ->
     [?assertEqual(ok, application:stop(App)) || App <- ?config(apps, Config)].
 
 init_per_testcase(_, Config) ->
-    start_cowboy(cert_dir()),
+    start_cowboy(#{cert_dir => cert_dir()}),
     {ok, _} = application:ensure_all_started(grisp_emulation),
     application:set_env(grisp_connect, ws_ping_timeout, 120_000),
     {ok, _} = application:ensure_all_started(grisp_connect),
@@ -62,7 +62,7 @@ reconnect_on_disconnection_test(Config) ->
     ?assertMatch(ok, wait_connection()),
     stop_cowboy(),
     ?assertMatch(ok, wait_disconnection()),
-    start_cowboy(cert_dir()),
+    start_cowboy(#{cert_dir => cert_dir()}),
     ?assertMatch(ok, wait_connection(1200)),
     Config.
 
@@ -88,7 +88,7 @@ reconnect_on_closed_frame_test(_) ->
 %--- Internal Functions --------------------------------------------------------
 
 connection_gun_pid() ->
-    {_, {data, _, _, _, _, ConnPid, _}} = sys:get_state(grisp_connect_client),
+    {_, {data, _, _, _, _, ConnPid, _, _, _, _}} = sys:get_state(grisp_connect_client),
     % Depends on the internal state of jarl_connection
-    {_, {data, _, _, _, _, _, _, _, _, _, _, _, GunPid, _, _, _}} = sys:get_state(ConnPid),
+    {_, {data, _, _, _, _, _, _, _, _, _, _, _, _, GunPid, _, _, _}} = sys:get_state(ConnPid),
     GunPid.