Merge pull request #4402 from esl/fix-cockroachdb-consistency

Fix cockroachdb consistency

Author: JanuszJakubiec

big_tests/test.config

@@ -286,7 +286,7 @@
   workers = 10
 [outgoing_pools.rdbms.default]
   scope = \"global\"
-  workers = 1
+  workers = 5
   connection.driver = \"cockroachdb\"
   connection.host = \"localhost\"
   connection.port = 26257

big_tests/tests/pubsub_SUITE.erl

@@ -22,6 +22,7 @@
                              require_rpc_nodes/1,
                              subhost_pattern/1,
                              rpc/4]).
+-import(domain_helper, [host_type/0]).
 
 %%--------------------------------------------------------------------
 %% Suite configuration
@@ -49,17 +50,26 @@ group_is_compatible(hometree_specific, OnlyNodetreeTree) -> OnlyNodetreeTree =:=
 group_is_compatible(_, _) -> true.
 
 base_groups() ->
-    [{basic, [parallel], basic_tests()},
-     {service_config, [parallel], service_config_tests()},
-     {node_config, [parallel], node_config_tests()},
-     {node_affiliations, [parallel], node_affiliations_tests()},
-     {manage_subscriptions, [parallel], manage_subscriptions_tests()},
+    [{basic, parallel_props(), basic_tests()},
+     {service_config, parallel_props(), service_config_tests()},
+     {node_config, parallel_props(), node_config_tests()},
+     {node_affiliations, parallel_props(), node_affiliations_tests()},
+     {manage_subscriptions, parallel_props(), manage_subscriptions_tests()},
      {collection, [sequence], collection_tests()},
-     {collection_config, [parallel], collection_config_tests()},
-     {debug_calls, [parallel], debug_calls_tests()},
-     {pubsub_item_publisher_option, [parallel], pubsub_item_publisher_option_tests()},
+     {collection_config, parallel_props(), collection_config_tests()},
+     {debug_calls, parallel_props(), debug_calls_tests()},
+     {pubsub_item_publisher_option, parallel_props(), pubsub_item_publisher_option_tests()},
      {hometree_specific, [sequence], hometree_specific_tests()},
-     {last_item_cache, [parallel], last_item_cache_tests()}].
+     {last_item_cache, parallel_props(), last_item_cache_tests()}].
+
+parallel_props() ->
+    case rpc(mim(), mongoose_rdbms, db_engine, [host_type()]) of
+        cockroachdb ->
+            %% Parallel pubsub tests are flaky on CockroachDB
+            [parallel, {repeat_until_all_ok, 5}];
+        _ ->
+            [parallel]
+    end.
 
 basic_tests() ->
     [

big_tests/tests/rdbms_SUITE.erl

@@ -82,6 +82,7 @@ rdbms_queries_cases() ->
      test_request_transaction,
      test_restart_transaction_with_execute,
      test_restart_transaction_with_execute_eventually_passes,
+     prepare_without_execute_does_not_cause_inconsistency,
      test_failed_transaction_with_execute_wrapped,
      test_failed_wrapper_transaction,
      test_incremental_upsert,
@@ -104,8 +105,7 @@ init_per_suite(Config) ->
          orelse mongoose_helper:is_rdbms_enabled(host_type()) of
         false -> {skip, rdbms_or_ct_not_running};
         true ->
-            %% Warning: inject_module does not really work well with --rerun-big-tests flag
-            mongoose_helper:inject_module(?MODULE),
+            mongoose_helper:inject_module(?MODULE, reload),
             Config1 = mongoose_helper:backup_and_set_config_option(Config, [instrumentation, probe_interval], 1),
             escalus:init_per_suite(Config1)
     end.
@@ -115,15 +115,15 @@ end_per_suite(Config) ->
     escalus:end_per_suite(Config).
 
 init_per_group(tagged_rdbms_queries, Config) ->
-    ExtraConfig = stop_global_default_pool(),
-    instrument_helper:start(declared_events(tagged_rdbms_queries)),
+    ExtraConfig = stop_global_default_pool() ++ [{scope, host_type()}, {tag, tag()}],
+    instrument_helper:start(declared_events(ExtraConfig)),
     start_local_host_type_pool(ExtraConfig),
     ExtraConfig ++ Config;
 init_per_group(global_rdbms_queries, Config) ->
-    ExtraConfig = stop_global_default_pool(),
-    instrument_helper:start(declared_events(global_rdbms_queries)),
+    ExtraConfig = stop_global_default_pool() ++ [{scope, global}, {tag, default}],
+    instrument_helper:start(declared_events(ExtraConfig)),
     restart_global_default_pool(ExtraConfig),
-    [{tag, global} | Config].
+    ExtraConfig ++ Config.
 
 end_per_group(tagged_rdbms_queries, Config) ->
     stop_local_host_type_pool(),
@@ -136,6 +136,14 @@ end_per_group(global_rdbms_queries, Config) ->
 init_per_testcase(test_incremental_upsert, Config) ->
     erase_inbox(Config),
     escalus:init_per_testcase(test_incremental_upsert, Config);
+init_per_testcase(Case = prepare_without_execute_does_not_cause_inconsistency, Config) ->
+    case is_pgsql() orelse is_cockroachdb() of
+        true ->
+            erase_inbox(Config),
+            escalus:init_per_testcase(Case, Config);
+        false ->
+            {skip, "Test for a Postgres-specific issue"}
+    end;
 init_per_testcase(CaseName, Config) ->
     escalus:init_per_testcase(CaseName, Config).
 
@@ -147,18 +155,25 @@ end_per_testcase(CaseName, Config)
          CaseName =:= test_failed_wrapper_transaction ->
     rpc(mim(), meck, unload, []),
     escalus:end_per_testcase(CaseName, Config);
+end_per_testcase(Case = prepare_without_execute_does_not_cause_inconsistency, Config) ->
+    erase_inbox(Config),
+    rpc(mim(), meck, unload, []),
+    escalus:end_per_testcase(Case, Config);
 end_per_testcase(test_incremental_upsert, Config) ->
     erase_inbox(Config),
     escalus:end_per_testcase(test_incremental_upsert, Config);
+end_per_testcase(Case = test_wrapped_request, Config) ->
+    Tag = ?config(tag, Config),
+    ok = rpc(mim(), mongoose_instrument, tear_down, [wrapper_event_spec(Tag)]),
+    escalus:end_per_testcase(Case, Config);
 end_per_testcase(CaseName, Config) ->
     escalus:end_per_testcase(CaseName, Config).
 
-declared_events(tagged_rdbms_queries) ->
-    instrument_helper:declared_events(mongoose_wpool_rdbms, [host_type(), tag()]) ++
-    [{test_wrapped_request, #{pool_tag => tag()}}];
-declared_events(global_rdbms_queries) ->
-    instrument_helper:declared_events(mongoose_wpool_rdbms, [global, default]) ++
-    [{test_wrapped_request, #{pool_tag => global}}].
+declared_events(Config) ->
+    Scope = ?config(scope, Config),
+    Tag = ?config(tag, Config),
+    instrument_helper:declared_events(mongoose_wpool_rdbms, [Scope, Tag]) ++
+    [{test_wrapped_request, #{pool_tag => Tag}}].
 
 %%--------------------------------------------------------------------
 %% Data for cases
@@ -369,11 +384,11 @@ read_prep_boolean_case(Config) ->
 
 select_current_timestamp_case(Config) ->
     ok = rpc(mim(), mongoose_rdbms_timestamp, prepare, []),
-    Res = case ?config(tag, Config) of
-              global ->
+    Res = case {?config(scope, Config), ?config(tag, Config)} of
+              {global, default} ->
                   rpc(mim(), mongoose_rdbms_timestamp, select, []);
-              Tag ->
-                  rpc(mim(), mongoose_rdbms_timestamp, select, [host_type(), Tag])
+              {Scope, Tag} ->
+                  rpc(mim(), mongoose_rdbms_timestamp, select, [Scope, Tag])
           end,
     assert_is_integer(Res).
 
@@ -540,6 +555,31 @@ test_restart_transaction_with_execute_eventually_passes(Config) ->
     called_times(3),
     ok.
 
+prepare_without_execute_does_not_cause_inconsistency(Config) ->
+    prepare_insert_int8(Config),
+    Args = [rdbms, ?config(scope, Config), ?config(tag, Config)],
+    Pool = rpc(mim(), mongoose_wpool, make_pool_name, Args),
+    [Key1, Key2] = make_hash_keys(2, Pool),
+
+    %% Simulate a call to 'prepare' with missing subsequent 'execute'
+    ok = rpc(mim(), meck, new, [mongoose_rdbms_backend, [passthrough, no_link]]),
+    ok = rpc(mim(), meck, expect, [mongoose_rdbms_backend, execute, 4, ok]),
+    ok = call_worker(Args, Key1, {sql_execute, insert_int8, [1]}),
+
+    %% Insert the data using a different worker
+    Insert = <<"INSERT INTO inbox VALUES ('alice', 'localhost', 'bob@localhost', "
+               "'msg-id', 'inbox', 'content', 14, 0, 0)">>,
+    {updated, 1} = call_worker(Args, Key2, {sql_query, Insert}),
+
+    %% Make sure that worker 1 is not stuck in the previous transaction after 'prepare'
+    {selected, [{<<"14">>}]} = call_worker(Args, Key1, {sql_query, <<"SELECT timestamp FROM inbox">>}).
+
+%% Send the SQL command to a particular DB worker
+call_worker(Args, HashKey, Operation) ->
+    TS = rpc(mim(), erlang, monotonic_time, [millisecond]),
+    Command = {sql_cmd, Operation, TS},
+    rpc(mim(), mongoose_wpool, call, Args ++ [HashKey, Command]).
+
 test_failed_transaction_with_execute_wrapped(Config) ->
     % given
     HostType = host_type(),
@@ -687,11 +727,11 @@ test_upsert_many2_replaces_existing(Config) ->
 
 pool_probe_metrics_are_updated(Config) ->
     Tag = ?config(tag, Config),
-    {Event, Labels} = case Tag of
+    {Event, Labels} = case ?config(scope, Config) of
                           global ->
-                              {wpool_global_rdbms_stats, #{pool_tag => default}};
-                          Tag ->
-                              {wpool_rdbms_stats, #{host_type => host_type(), pool_tag => Tag}}
+                              {wpool_global_rdbms_stats, #{pool_tag => Tag}};
+                          Scope ->
+                              {wpool_rdbms_stats, #{host_type => Scope, pool_tag => Tag}}
                       end,
     #{recv_oct := Recv, send_oct := Send} = rpc(mim(), mongoose_wpool_rdbms, probe, [Event, Labels]),
 
@@ -718,10 +758,12 @@ tag() ->
     extra_tag.
 
 scope_and_tag(Config) ->
-    case ?config(tag, Config) of
-        global -> [host_type()];
-        Tag -> [host_type(), Tag]
-    end.
+    skip_default_tag([?config(scope, Config), ?config(tag, Config)]).
+
+skip_default_tag([Scope, default]) ->
+    [Scope];
+skip_default_tag(ScopeAndTag) ->
+    ScopeAndTag.
 
 sql_query(Config, Query) ->
     ScopeAndTag = scope_and_tag(Config),
@@ -1285,7 +1327,7 @@ stop_global_default_pool() ->
     [GlobalRdbmsPool] = [Pool || Pool = #{type := rdbms, scope := global, tag := default} <- Pools],
     ok = rpc(mim(), mongoose_wpool, stop, [rdbms, global, default]),
     Extra = maybe_stop_service_domain_db(),
-    [{tag, tag()}, {global_default_rdbms_pool, GlobalRdbmsPool} | Extra].
+    [{global_default_rdbms_pool, GlobalRdbmsPool} | Extra].
 
 restart_global_default_pool(Config) ->
     GlobalRdbmsPool = ?config(global_default_rdbms_pool, Config),
@@ -1390,3 +1432,24 @@ check_like_not_matching_prep(SelName, Config, _TextValue, NotMatching, Info) ->
                         SelectResult,
                         Info#{pattern => NotMatching,
                               select_result => SelectResult}).
+
+%% Generate a list of Num numerical keys resolving to different Pool workers using hash_worker
+make_hash_keys(Num, Pool) ->
+    Workers = rpc(mim(), wpool_pool, get_workers, [Pool]),
+    if Num =< length(Workers) ->
+            lists:reverse(make_hash_keys(Num, Pool, Workers, 1, []));
+       true ->
+            ct:fail("Not enough workers in ~p (needed: ~p, actual: ~p)",
+                    [Pool, Num, length(Workers)])
+    end.
+
+make_hash_keys(RemainingNum, Pool, Workers, Key, Acc) when RemainingNum > 0 ->
+    Worker = rpc(mim(), wpool_pool, hash_worker, [Pool, Key]),
+    case lists:member(Worker, Workers) of
+        true ->
+            make_hash_keys(RemainingNum - 1, Pool, Workers -- [Worker], Key + 1, [Key | Acc]);
+        false ->
+            make_hash_keys(RemainingNum, Pool, Workers, Key + 1, Acc)
+    end;
+make_hash_keys(0, _WorkerNum, _Workers, _Key, Acc) ->
+    Acc.

src/rdbms/mongoose_rdbms_pgsql.erl

@@ -70,8 +70,15 @@ prepare(Connection, Name, _Table, _Fields, Statement) ->
     BinName = [atom_to_binary(Name, latin1)],
     ReplacedStatement = replace_question_marks(Statement),
     case epgsql:parse(Connection, BinName, ReplacedStatement, []) of
-        {ok, _} -> epgsql:describe(Connection, statement, BinName);
-        Error   -> Error
+        {ok, PreparedStatement} ->
+            %% If the statement is not executed for some reason, the transaction would remain open.
+            %% The safest way to fix this is to issue the Sync message.
+            %% Performance impact is minimal, because 'prepare' is called only once.
+            %% See https://www.postgresql.org/docs/current/protocol-flow.html for details
+            ok = epgsql:sync(Connection),
+            {ok, PreparedStatement};
+        {error, Reason} ->
+            {error, Reason}
     end.
 
 -spec execute(Connection :: term(), StatementRef :: term(), Params :: [term()],