Test rdbms_SUITE entirely on host_type pool

big_tests/tests/rdbms_SUITE.erl

@@ -17,6 +17,7 @@
 -module(rdbms_SUITE).
 -compile([export_all, nowarn_export_all]).
 
+-include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
 %% We need assert from it
@@ -91,13 +92,18 @@ init_per_suite(Config) ->
          orelse mongoose_helper:is_rdbms_enabled(host_type()) of
         false -> {skip, rdbms_or_ct_not_running};
         true ->
+            Pools = rpc(mim(), mongoose_config, get_opt, [outgoing_pools]),
+            GlobalRdbmsPool = stop_global_default_pool(Pools),
+            start_local_host_type_pool(Pools),
             %% Warning: inject_module does not really work well with --rerun-big-tests flag
             mongoose_helper:inject_module(?MODULE),
-            escalus:init_per_suite(Config)
+            escalus:init_per_suite([{global_default_rdbms_pool, GlobalRdbmsPool} | Config])
     end.
 
 end_per_suite(Config) ->
-    escalus:end_per_suite(Config).
+    escalus:end_per_suite(Config),
+    GlobalRdbmsPool = ?config(global_default_rdbms_pool, Config),
+    rpc(mim(), mongoose_wpool, start_configured_pools, [[GlobalRdbmsPool]]).
 
 init_per_testcase(test_incremental_upsert, Config) ->
     erase_inbox(Config),
@@ -325,7 +331,7 @@ read_prep_boolean_case(Config) ->
 
 select_current_timestamp_case(Config) ->
     ok = rpc(mim(), mongoose_rdbms_timestamp, prepare, []),
-    assert_is_integer(rpc(mim(), mongoose_rdbms_timestamp, select, [])).
+    assert_is_integer(rpc(mim(), mongoose_rdbms_timestamp, select, [host_type(), tag()])).
 
 assert_is_integer(X) when is_integer(X) ->
     X.
@@ -590,9 +596,11 @@ select_like_prep_case(Config) ->
 %%--------------------------------------------------------------------
 %% Helpers
 %%--------------------------------------------------------------------
+tag() ->
+    extra_tag.
 
 sql_query(_Config, Query) ->
-    slow_rpc(mongoose_rdbms, sql_query, [host_type(), Query]).
+    slow_rpc(mongoose_rdbms, sql_query, [host_type(), tag(), Query]).
 
 sql_prepare(_Config, Name, Table, Fields, Query) ->
     escalus_ejabberd:rpc(mongoose_rdbms, prepare, [Name, Table, Fields, Query]).
@@ -601,38 +609,38 @@ sql_prepare_upsert(_Config, Name, Table, Insert, Update, Unique, Incr) ->
     escalus_ejabberd:rpc(rdbms_queries, prepare_upsert, [host_type(), Name, Table, Insert, Update, Unique, Incr]).
 
 sql_execute(_Config, Name, Parameters) ->
-    slow_rpc(mongoose_rdbms, execute, [host_type(), Name, Parameters]).
+    slow_rpc(mongoose_rdbms, execute, [host_type(), tag(), Name, Parameters]).
 
 sql_execute_cast(_Config, Name, Parameters) ->
-    slow_rpc(mongoose_rdbms, execute_cast, [host_type(), Name, Parameters]).
+    slow_rpc(mongoose_rdbms, execute_cast, [host_type(), tag(), Name, Parameters]).
 
 sql_query_cast(_Config, Query) ->
-    slow_rpc(mongoose_rdbms, sql_query_cast, [host_type(), Query]).
+    slow_rpc(mongoose_rdbms, sql_query_cast, [host_type(), tag(), Query]).
 
 sql_execute_request(_Config, Name, Parameters) ->
-    slow_rpc(mongoose_rdbms, execute_request, [host_type(), Name, Parameters]).
+    slow_rpc(mongoose_rdbms, execute_request, [host_type(), tag(), Name, Parameters]).
 
 sql_execute_wrapped_request(_Config, Name, Parameters, WrapperFun) ->
-    slow_rpc(mongoose_rdbms, execute_wrapped_request, [host_type(), Name, Parameters, WrapperFun]).
+    slow_rpc(mongoose_rdbms, execute_wrapped_request, [host_type(), tag(), Name, Parameters, WrapperFun]).
 
 sql_execute_wrapped_request_and_wait_response(_Config, Name, Parameters, WrapperFun) ->
-    slow_rpc(?MODULE, execute_wrapped_request_and_wait_response, [host_type(), Name, Parameters, WrapperFun]).
+    slow_rpc(?MODULE, execute_wrapped_request_and_wait_response, [host_type(), tag(), Name, Parameters, WrapperFun]).
 
-execute_wrapped_request_and_wait_response(HostType, Name, Parameters, WrapperFun) ->
-    RequestId = mongoose_rdbms:execute_wrapped_request(HostType, Name, Parameters, WrapperFun),
+execute_wrapped_request_and_wait_response(HostType, Tag, Name, Parameters, WrapperFun) ->
+    RequestId = mongoose_rdbms:execute_wrapped_request(HostType, Tag, Name, Parameters, WrapperFun),
     gen_server:wait_response(RequestId, 100).
 
 sql_execute_upsert(_Config, Name, Insert, Update, Unique) ->
-    slow_rpc(rdbms_queries, execute_upsert, [host_type(), Name, Insert, Update, Unique]).
+    slow_rpc(rdbms_queries, execute_upsert, [host_type(), tag(), Name, Insert, Update, Unique]).
 
 sql_query_request(_Config, Query) ->
-    slow_rpc(mongoose_rdbms, sql_query_request, [host_type(), Query]).
+    slow_rpc(mongoose_rdbms, sql_query_request, [host_type(), tag(), Query]).
 
 sql_transaction_request(_Config, Query) ->
-    slow_rpc(mongoose_rdbms, sql_transaction_request, [host_type(), Query]).
+    slow_rpc(mongoose_rdbms, sql_transaction_request, [host_type(), tag(), Query]).
 
 sql_transaction(_Config, F) ->
-    slow_rpc(mongoose_rdbms, sql_transaction, [host_type(), F]).
+    slow_rpc(mongoose_rdbms, sql_transaction, [host_type(), tag(), F]).
 
 escape_null(_Config) ->
     escalus_ejabberd:rpc(mongoose_rdbms, escape_null, []).
@@ -1118,6 +1126,16 @@ is_pgsql() ->
 is_mysql() ->
     db_engine() == mysql.
 
+stop_global_default_pool(Pools) ->
+    [GlobalRdbmsPool] = [Pool || Pool = #{type := rdbms, scope := global, tag := default} <- Pools],
+    ok = rpc(mim(), mongoose_wpool, stop, [rdbms, global, default]),
+    GlobalRdbmsPool.
+
+start_local_host_type_pool(Pools) ->
+    [GlobalRdbmsPool] = [Pool || Pool = #{type := rdbms, scope := global, tag := default} <- Pools],
+    LocalHostTypePool = GlobalRdbmsPool#{scope := host_type(), tag := tag()},
+    rpc(mim(), mongoose_wpool, start_configured_pools, [[LocalHostTypePool], [host_type()]]).
+
 escape_column(Name) ->
     case is_mysql() of
         true ->

src/mongoose_cets_discovery_rdbms.erl

@@ -156,7 +156,7 @@ delete_node_from_db(ClusterName, Node) ->
 timestamp() ->
     % We could use Erlang timestamp os:system_time(second).
     % But we use the database server time as a central source of truth.
-    mongoose_rdbms_timestamp:select().
+    mongoose_rdbms_timestamp:select(global, default).
 
 %% Returns a next free node id based on the currently registered ids
 first_free_num(Nums) ->

src/rdbms/mongoose_rdbms.erl

@@ -158,7 +158,7 @@
                }).
 -type state() :: #state{}.
 
--define(DEFAULT_POOL, default).
+-define(DEFAULT_POOL_NAME, default).
 -define(STATE_KEY, mongoose_rdbms_state).
 -define(MAX_TRANSACTION_RESTARTS, 10).
 -define(TRANSACTION_TIMEOUT, 60000). % milliseconds
@@ -240,7 +240,7 @@ prepared(Name) ->
 -spec execute(HostType :: server(), Name :: atom(), Parameters :: [term()]) ->
                      query_result().
 execute(HostType, Name, Parameters) ->
-    execute(HostType, ?DEFAULT_POOL, Name, Parameters).
+    execute(HostType, ?DEFAULT_POOL_NAME, Name, Parameters).
 
 -spec execute(HostType :: server(), PoolName :: atom(), Name :: atom(), Parameters :: [term()]) ->
     query_result().
@@ -250,7 +250,7 @@ execute(HostType, PoolName, Name, Parameters) when is_atom(PoolName), is_atom(Na
 -spec execute_cast(HostType :: server(), Name :: atom(), Parameters :: [term()]) ->
     query_result().
 execute_cast(HostType, Name, Parameters) ->
-    execute_cast(HostType, ?DEFAULT_POOL, Name, Parameters).
+    execute_cast(HostType, ?DEFAULT_POOL_NAME, Name, Parameters).
 
 -spec execute_cast(HostType :: server(), PoolName :: atom(), Name :: atom(), Parameters :: [term()]) ->
     query_result().
@@ -260,7 +260,7 @@ execute_cast(HostType, PoolName, Name, Parameters) when is_atom(PoolName), is_at
 -spec execute_request(HostType :: server(), Name :: atom(), Parameters :: [term()]) ->
     request_id().
 execute_request(HostType, Name, Parameters) when is_atom(Name), is_list(Parameters) ->
-    execute_request(HostType, ?DEFAULT_POOL, Name, Parameters).
+    execute_request(HostType, ?DEFAULT_POOL_NAME, Name, Parameters).
 
 -spec execute_request(HostType :: server(), PoolName :: atom(), Name :: atom(), Parameters :: [term()]) ->
     request_id().
@@ -273,7 +273,7 @@ execute_request(HostType, PoolName, Name, Parameters) when is_atom(PoolName), is
     Parameters :: [term()],
     Wrapper :: request_wrapper()) -> request_id().
 execute_wrapped_request(HostType, Name, Parameters, Wrapper) ->
-    execute_wrapped_request(HostType, ?DEFAULT_POOL, Name, Parameters, Wrapper).
+    execute_wrapped_request(HostType, ?DEFAULT_POOL_NAME, Name, Parameters, Wrapper).
 
 -spec execute_wrapped_request(
     HostType :: server(),
@@ -289,7 +289,7 @@ execute_wrapped_request(HostType, PoolName, Name, Parameters, Wrapper)
 -spec execute_successfully(HostType :: server(), Name :: atom(), Parameters :: [term()]) ->
     query_result().
 execute_successfully(HostType, Name, Parameters) ->
-    execute_successfully(HostType, ?DEFAULT_POOL, Name, Parameters).
+    execute_successfully(HostType, ?DEFAULT_POOL_NAME, Name, Parameters).
 
 -spec execute_successfully(HostType :: server(), PoolName :: atom(), Name :: atom(), Parameters :: [term()]) ->
     query_result().
@@ -324,23 +324,23 @@ query_name_to_string(Name) ->
 
 -spec sql_query(HostType :: server(), Query :: any()) -> query_result().
 sql_query(HostType, Query) ->
-    sql_query(HostType, ?DEFAULT_POOL, Query).
+    sql_query(HostType, ?DEFAULT_POOL_NAME, Query).
 
 -spec sql_query(HostType :: server(), PoolName :: atom(), Query :: any()) -> query_result().
 sql_query(HostType, PoolName, Query) ->
     sql_call(HostType, PoolName, {sql_query, Query}).
 
 -spec sql_query_request(HostType :: server(), Query :: any()) -> request_id().
 sql_query_request(HostType, Query) ->
-    sql_query_request(HostType, ?DEFAULT_POOL, Query).
+    sql_query_request(HostType, ?DEFAULT_POOL_NAME, Query).
 
 -spec sql_query_request(HostType :: server(), PoolName :: atom(), Query :: any()) -> request_id().
 sql_query_request(HostType, PoolName, Query) ->
     sql_request(HostType, PoolName, {sql_query, Query}).
 
 -spec sql_query_cast(HostType :: server(), Query :: any()) -> query_result().
 sql_query_cast(HostType, Query) ->
-    sql_query_cast(HostType, ?DEFAULT_POOL, Query).
+    sql_query_cast(HostType, ?DEFAULT_POOL_NAME, Query).
 
 -spec sql_query_cast(HostType :: server(), PoolName :: atom(), Query :: any()) -> query_result().
 sql_query_cast(HostType, PoolName, Query) ->
@@ -349,7 +349,7 @@ sql_query_cast(HostType, PoolName, Query) ->
 %% @doc SQL transaction based on a list of queries
 -spec sql_transaction(server(), fun() | maybe_improper_list()) -> transaction_result().
 sql_transaction(HostType, Msg) ->
-    sql_transaction(HostType, ?DEFAULT_POOL, Msg).
+    sql_transaction(HostType, ?DEFAULT_POOL_NAME, Msg).
 
 -spec sql_transaction(server(), atom(), fun() | maybe_improper_list()) -> transaction_result().
 sql_transaction(HostType, PoolName, Queries) when is_atom(PoolName), is_list(Queries) ->
@@ -362,7 +362,7 @@ sql_transaction(HostType, PoolName, F) when is_atom(PoolName), is_function(F) ->
 %% @doc SQL transaction based on a list of queries
 -spec sql_transaction_request(server(), fun() | maybe_improper_list()) -> request_id().
 sql_transaction_request(HostType, Queries) ->
-    sql_transaction_request(HostType, ?DEFAULT_POOL, Queries).
+    sql_transaction_request(HostType, ?DEFAULT_POOL_NAME, Queries).
 
 -spec sql_transaction_request(server(), atom(), fun() | maybe_improper_list()) -> request_id().
 sql_transaction_request(HostType, PoolName, Queries) when is_atom(PoolName), is_list(Queries) ->
@@ -400,7 +400,7 @@ do_transaction_with_delayed_retry(HostType, F, Retries, Delay, Info) ->
 
 -spec sql_dirty(server(), fun()) -> any() | no_return().
 sql_dirty(HostType, F) ->
-    sql_dirty(HostType, ?DEFAULT_POOL, F).
+    sql_dirty(HostType, ?DEFAULT_POOL_NAME, F).
 
 -spec sql_dirty(server(), atom(), fun()) -> any() | no_return().
 sql_dirty(HostType, PoolName, F) when is_function(F) ->

src/rdbms/mongoose_rdbms_timestamp.erl

@@ -1,6 +1,6 @@
 -module(mongoose_rdbms_timestamp).
 -export([prepare/0,
-         select/0]).
+         select/2]).
 
 -spec prepare() -> ok.
 prepare() ->
@@ -19,7 +19,7 @@ select_query() ->
            error({prepare_timestamp_query_failed, Other})
    end.
 
--spec select() -> integer().
-select() ->
-    Res = mongoose_rdbms:execute_successfully(global, mim_timestamp, []),
+-spec select(mongooseim:host_type_or_global(), atom()) -> integer().
+select(HostType, PoolName) ->
+    Res = mongoose_rdbms:execute_successfully(HostType, PoolName, mim_timestamp, []),
     mongoose_rdbms:selected_to_integer(Res). %% ensure it is an integer

src/rdbms/rdbms_queries.erl

@@ -43,11 +43,12 @@
 -export([join/2,
          prepare_upsert/6,
          prepare_upsert/7,
-         execute_upsert/5,
+         execute_upsert/5, execute_upsert/6,
          request_upsert/5]).
 
 -ignore_xref([
-    count_records_where/3, get_db_specific_limits/1, get_db_specific_offset/2, get_db_type/0
+    execute_upsert/6, count_records_where/3,
+    get_db_specific_limits/1, get_db_specific_offset/2, get_db_type/0
 ]).
 
 %% We have only two compile time options for db queries:
@@ -88,13 +89,22 @@ get_db_type() ->
                      UpdateParams :: [any()],
                      UniqueKeyValues :: [any()]) -> mongoose_rdbms:query_result().
 execute_upsert(Host, Name, InsertParams, UpdateParams, UniqueKeyValues) ->
+    execute_upsert(Host, default, Name, InsertParams, UpdateParams, UniqueKeyValues).
+
+-spec execute_upsert(Host :: mongoose_rdbms:server(),
+                     PoolName :: atom(),
+                     Name :: atom(),
+                     InsertParams :: [any()],
+                     UpdateParams :: [any()],
+                     UniqueKeyValues :: [any()]) -> mongoose_rdbms:query_result().
+execute_upsert(Host, PoolName, Name, InsertParams, UpdateParams, UniqueKeyValues) ->
     case {mongoose_rdbms:db_engine(Host), mongoose_rdbms:db_type()} of
         {mysql, _} ->
-            mongoose_rdbms:execute(Host, Name, InsertParams ++ UpdateParams);
+            mongoose_rdbms:execute(Host, PoolName, Name, InsertParams ++ UpdateParams);
         {pgsql, _} ->
-            mongoose_rdbms:execute(Host, Name, InsertParams ++ UpdateParams);
+            mongoose_rdbms:execute(Host, PoolName, Name, InsertParams ++ UpdateParams);
         {odbc, mssql} ->
-            mongoose_rdbms:execute(Host, Name, UniqueKeyValues ++ InsertParams ++ UpdateParams);
+            mongoose_rdbms:execute(Host, PoolName, Name, UniqueKeyValues ++ InsertParams ++ UpdateParams);
         NotSupported -> erlang:error({rdbms_not_supported, NotSupported})
     end.