chore: update ecpool usage

Author: savonarola

.ci/Makefile

@@ -1,5 +1,5 @@
 up:
-	docker compose --file docker-compose-file/docker-compose-hstreamdb.yaml up --detach --build
+	docker compose --file docker-compose-file/docker-compose-hstreamdb.yaml up --detach --build --force-recreate
 
 down:
 	docker compose --file docker-compose-file/docker-compose-hstreamdb.yaml down --remove-orphans

Makefile

@@ -46,6 +46,8 @@ coveralls: $(REBAR)
 clean:
 	@rm -rf _build
 	@rm -rf rebar3
+	@rm -rf rebar.lock
+	@rm -rf *.crashdump
 	@rm -rf *_crash.dump
 	@rm -rf hstreamdb_erl_*_plt
 

config/sys.config

@@ -1 +0,0 @@
-[{kernel, [{logger_level, debug}]}].

src/hstreamdb.erl

@@ -54,6 +54,7 @@
 -export([
     start_reader/2,
     stop_reader/1,
+    wait_for_reader/1,
     read_stream_key/3,
     read_stream_key/4
 ]).
@@ -201,6 +202,10 @@ start_reader(Name, ReaderOptions) ->
 stop_reader(Name) ->
     hstreamdb_readers_sup:stop(Name).
 
+-spec wait_for_reader(ecpool:pool_name()) -> ok | no_such_reader.
+wait_for_reader(Name) ->
+    hstreamdb_reader:wait_for_start(Name).
+
 %% @doc fetch only records that have the same key as the one provided, using
 %% server-side filtering.
 

src/hstreamdb_client.erl

@@ -370,6 +370,8 @@ stop_channel(SupRef, Channel) ->
     case stop_channel_pool(SupRef, Channel) of
         ok ->
             ?LOG_DEBUG("[hstreamdb] stop_channel ok: ~p", [Channel]);
+        {error, not_found} ->
+            ?LOG_DEBUG("[hstreamdb] stop_channel not_found: ~p", [Channel]);
         {error, _} = Error ->
             ?LOG_ERROR("[hstreamdb] stop_channel error: ~p, ~p", [Channel, Error])
     end.

src/reader/hstreamdb_reader.erl

@@ -23,7 +23,8 @@
 
 -export([
     read_key/3,
-    read_key/4
+    read_key/4,
+    wait_for_start/1
 ]).
 
 -export([
@@ -69,6 +70,17 @@ read_key(Reader, Key, Limits) ->
 read_key(Reader, Key, Limits, Fold) ->
     do_read_key(Reader, Key, Limits, Fold).
 
+-spec wait_for_start(ecpool:pool_name()) -> ok | no_such_reader.
+wait_for_start(Reader) ->
+    case ecpool:get_client(Reader) of
+        Pid when is_pid(Pid) -> ok;
+        false ->
+            timer:sleep(100),
+            wait_for_start(Reader);
+        no_such_pool ->
+            no_such_reader
+    end.
+
 %%-------------------------------------------------------------------------------------------------
 %% ecpool part
 
@@ -138,22 +150,26 @@ code_change(_OldVsn, State, _Extra) ->
 %%-------------------------------------------------------------------------------------------------
 
 do_read_key(Reader, Key, Limits, {FoldFun, InitAcc}) ->
-    LookupClient = ecpool_request(Reader, get_lookup_client),
-    case ?MEASURE({lookup_key, self(), Key}, hstreamdb_client:lookup_key(LookupClient, Key)) of
-        {ok, {_Host, _Port} = Addr} ->
-            case ecpool_request(Reader, {get_key_gstream, Key, Addr}) of
-                {ok, Stream, GStream} ->
-                    ?MEASURE(
-                        {fold_key_read_gstream, Stream, Key},
-                        hstreamdb_client:fold_key_read_gstream(
-                            GStream, Stream, Key, Limits, FoldFun, InitAcc
-                        )
-                    );
+    case ecpool_request(Reader, get_lookup_client) of
+        {error, _} = Error ->
+            Error;
+        LookupClient ->
+            case ?MEASURE({lookup_key, self(), Key}, hstreamdb_client:lookup_key(LookupClient, Key)) of
+                {ok, {_Host, _Port} = Addr} ->
+                    case ecpool_request(Reader, {get_key_gstream, Key, Addr}) of
+                        {ok, Stream, GStream} ->
+                            ?MEASURE(
+                                {fold_key_read_gstream, Stream, Key},
+                                hstreamdb_client:fold_key_read_gstream(
+                                    GStream, Stream, Key, Limits, FoldFun, InitAcc
+                                )
+                            );
+                        {error, _} = Error ->
+                            Error
+                    end;
                 {error, _} = Error ->
                     Error
-            end;
-        {error, _} = Error ->
-            Error
+            end
     end.
 
 do_get_key_gstream(
@@ -198,5 +214,7 @@ fold_stream_key_fun(Key) ->
 ecpool_request(Reader, Request) ->
     ecpool:with_client(
         Reader,
-        fun(Pid) -> gen_server:call(Pid, Request) end
+        fun(Pid) ->
+            gen_server:call(Pid, Request)
+        end
     ).

test/hstreamdb_producer_SUITE.erl

@@ -43,6 +43,8 @@ end).
 all() ->
     hstreamdb_test_helpers:test_cases(?MODULE).
 
+suite() -> [{timetrap, {minutes, 5}}].
+
 init_per_suite(Config) ->
     _ = application:ensure_all_started(hstreamdb_erl),
     Config.
@@ -799,6 +801,7 @@ t_append_sync_timeout_while_success(Config) ->
 
     Reader = "reader_" ++ atom_to_list(?FUNCTION_NAME),
     ok = hstreamdb:start_reader(Reader, ReaderOptions),
+    ok = hstreamdb:wait_for_reader(Reader),
 
     % Read all records
 

test/hstreamdb_read_SUITE.erl

@@ -27,6 +27,8 @@
 all() ->
     hstreamdb_test_helpers:test_cases(?MODULE).
 
+suite() -> [{timetrap, {minutes, 5}}].
+
 init_per_suite(Config) ->
     _ = application:ensure_all_started(hstreamdb_erl),
     Config.
@@ -64,6 +66,7 @@ t_read_stream_key(Config) ->
 
     Reader = "reader_" ++ atom_to_list(?FUNCTION_NAME),
     ok = hstreamdb:start_reader(Reader, ReaderOptions),
+    ok = hstreamdb:wait_for_reader(Reader),
 
     % Try to read with invalid limits
 
@@ -150,6 +153,7 @@ t_read_stream_recreated_key(Config) ->
 
     Reader = "reader_" ++ atom_to_list(?FUNCTION_NAME),
     ok = hstreamdb:start_reader(Reader, ReaderOptions),
+    ok = hstreamdb:wait_for_reader(Reader),
 
     Limits = #{
         from => #{offset => {specialOffset, 0}},
@@ -221,6 +225,7 @@ t_trim(Config) ->
 
     Reader = "reader_" ++ atom_to_list(?FUNCTION_NAME),
     ok = hstreamdb:start_reader(Reader, ReaderOptions),
+    ok = hstreamdb:wait_for_reader(Reader),
 
     Limits = #{
         from => #{offset => {specialOffset, 0}},