Handle transaction timeouts in list_dbs and list_dbs_infohandle-timouts-all-dbs-and-docs

Previously those endpoints would break when transactions time-out and are
retried. To fix it we re-use the mechanism from changes feeds.

There is a longer discussion about this on the mailing list:

https://lists.apache.org/thread.html/r02cee7045cac4722e1682bb69ba0ec791f5cce025597d0099fb34033%40%3Cdev.couchdb.apache.org%3E

2 files changed, 186 insertions, 14 deletions

diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index b4a4fd6a2..8bc87926d 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -340,7 +340,11 @@ get_dir(Tx) ->
     erlfdb_directory:get_name(CouchDB).
 
 
-list_dbs(Tx, Callback, AccIn, Options) ->
+list_dbs(Tx, Callback, AccIn, Options0) ->
+    Options = case fabric2_util:get_value(restart_tx, Options0) of
+        undefined -> [{restart_tx, true} | Options0];
+        _AlreadySet -> Options0
+    end,
     LayerPrefix = get_dir(Tx),
     Prefix = erlfdb_tuple:pack({?ALL_DBS}, LayerPrefix),
     fold_range({tx, Tx}, Prefix, fun({K, _V}, Acc) ->
@@ -349,7 +353,11 @@ list_dbs(Tx, Callback, AccIn, Options) ->
     end, AccIn, Options).
 
 
-list_dbs_info(Tx, Callback, AccIn, Options) ->
+list_dbs_info(Tx, Callback, AccIn, Options0) ->
+    Options = case fabric2_util:get_value(restart_tx, Options0) of
+        undefined -> [{restart_tx, true} | Options0];
+        _AlreadySet -> Options0
+    end,
     LayerPrefix = get_dir(Tx),
     Prefix = erlfdb_tuple:pack({?ALL_DBS}, LayerPrefix),
     fold_range({tx, Tx}, Prefix, fun({DbNameKey, DbPrefix}, Acc) ->
diff --git a/src/fabric/test/fabric2_db_crud_tests.erl b/src/fabric/test/fabric2_db_crud_tests.erl
index 943b55f3f..25dbfd639 100644
--- a/src/fabric/test/fabric2_db_crud_tests.erl
+++ b/src/fabric/test/fabric2_db_crud_tests.erl
@@ -18,27 +18,63 @@
 -include("fabric2_test.hrl").
 
 
+-define(PDICT_ERROR_IN_FOLD_RANGE, '$fabric2_error_in_fold_range').
+-define(PDICT_ERROR_IN_USER_FUN, '$fabric2_error_throw_in_user_fun').
+
+
 crud_test_() ->
     {
         "Test database CRUD operations",
         {
             setup,
-            fun() -> test_util:start_couch([fabric]) end,
-            fun test_util:stop_couch/1,
-            with([
-                ?TDEF(create_db),
-                ?TDEF(open_db),
-                ?TDEF(delete_db),
-                ?TDEF(list_dbs),
-                ?TDEF(list_dbs_user_fun),
-                ?TDEF(list_dbs_user_fun_partial),
-                ?TDEF(list_dbs_info),
-                ?TDEF(list_dbs_info_partial)
-            ])
+            fun setup_all/0,
+            fun teardown_all/1,
+            {
+                foreach,
+                fun setup/0,
+                fun cleanup/1,
+                [
+                    ?TDEF_FE(create_db),
+                    ?TDEF_FE(open_db),
+                    ?TDEF_FE(delete_db),
+                    ?TDEF_FE(list_dbs),
+                    ?TDEF_FE(list_dbs_user_fun),
+                    ?TDEF_FE(list_dbs_user_fun_partial),
+                    ?TDEF_FE(list_dbs_info),
+                    ?TDEF_FE(list_dbs_info_partial),
+                    ?TDEF_FE(list_dbs_tx_too_long),
+                    ?TDEF_FE(list_dbs_tx_too_long_with_user_fun),
+                    ?TDEF_FE(list_dbs_info_tx_too_long),
+                    ?TDEF_FE(list_dbs_info_tx_too_long_with_user_fun)
+                ]
+            }
         }
     }.
 
 
+setup_all() ->
+    Ctx = test_util:start_couch([fabric]),
+    meck:new(erlfdb, [passthrough]),
+    Ctx.
+
+
+teardown_all(Ctx) ->
+    meck:unload(),
+    test_util:stop_couch(Ctx).
+
+
+setup() ->
+    meck:expect(erlfdb, fold_range, fun(Tx, Start, End, Callback, Acc, Opts) ->
+        maybe_tx_too_long(?PDICT_ERROR_IN_FOLD_RANGE),
+        meck:passthrough([Tx, Start, End, Callback, Acc, Opts])
+    end),
+    ok.
+
+
+cleanup(_) ->
+    reset_error_counts().
+
+
 create_db(_) ->
     DbName = ?tempdb(),
     ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
@@ -132,6 +168,103 @@ list_dbs_info_partial(_) ->
     ?assertEqual([{meta, []}], UserAcc).
 
 
+list_dbs_tx_too_long(_) ->
+    DbName1 = ?tempdb(),
+    DbName2 = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName1, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName2, [])),
+    Dbs = fabric2_db:list_dbs(),
+
+    % Blow up in fold range
+    tx_too_long_errors(0, 1),
+    ?assertEqual(Dbs, fabric2_db:list_dbs()),
+
+    % Blow up in fold_range after emitting one row
+    tx_too_long_errors(0, {1, 1}),
+    ?assertEqual(Dbs, fabric2_db:list_dbs()),
+
+    ok = fabric2_db:delete(DbName1, []),
+    ok = fabric2_db:delete(DbName2, []).
+
+
+list_dbs_tx_too_long_with_user_fun(_) ->
+    DbName1 = ?tempdb(),
+    DbName2 = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName1, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName2, [])),
+
+    UserFun = fun(Row, Acc) ->
+        maybe_tx_too_long(?PDICT_ERROR_IN_USER_FUN),
+        {ok, [Row, Acc]}
+    end,
+
+    % Get get expected output without any transactions timing out
+    Dbs = fabric2_db:list_dbs(UserFun, [], []),
+
+    % Blow up in user fun
+    tx_too_long_errors(1, 0),
+    ?assertEqual(Dbs, fabric2_db:list_dbs(UserFun, [], [])),
+
+    % Blow up in user fun after emitting one row
+    tx_too_long_errors({1, 1}, 0),
+    ?assertEqual(Dbs, fabric2_db:list_dbs(UserFun, [], [])),
+
+    % Blow up in in user fun and fold range
+    tx_too_long_errors(1, {1, 1}),
+    ?assertEqual(Dbs, fabric2_db:list_dbs(UserFun, [], [])),
+
+    ok = fabric2_db:delete(DbName1, []),
+    ok = fabric2_db:delete(DbName2, []).
+
+
+list_dbs_info_tx_too_long(_) ->
+    DbName1 = ?tempdb(),
+    DbName2 = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName1, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName2, [])),
+    {ok, DbInfos} = fabric2_db:list_dbs_info(),
+
+    % Blow up in fold range
+    tx_too_long_errors(0, 1),
+    ?assertEqual({ok, DbInfos}, fabric2_db:list_dbs_info()),
+
+    % Blow up in fold_range after emitting one row
+    tx_too_long_errors(0, {1, 1}),
+    ?assertEqual({ok, DbInfos}, fabric2_db:list_dbs_info()),
+
+    ok = fabric2_db:delete(DbName1, []),
+    ok = fabric2_db:delete(DbName2, []).
+
+
+list_dbs_info_tx_too_long_with_user_fun(_) ->
+    DbName1 = ?tempdb(),
+    DbName2 = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName1, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName2, [])),
+
+    UserFun = fun(Row, Acc) ->
+        maybe_tx_too_long(?PDICT_ERROR_IN_USER_FUN),
+        {ok, [Row, Acc]}
+    end,
+
+    {ok, DbInfos} = fabric2_db:list_dbs_info(UserFun, [], []),
+
+    % Blow up in user fun
+    tx_too_long_errors(1, 0),
+    ?assertEqual({ok, DbInfos}, fabric2_db:list_dbs_info(UserFun, [], [])),
+
+    % Blow up in user fun after emitting one row
+    tx_too_long_errors({1, 1}, 0),
+    ?assertEqual({ok, DbInfos}, fabric2_db:list_dbs_info(UserFun, [], [])),
+
+    % Blow up in in user fun and fold range
+    tx_too_long_errors(1, {1, 1}),
+    ?assertEqual({ok, DbInfos}, fabric2_db:list_dbs_info(UserFun, [], [])),
+
+    ok = fabric2_db:delete(DbName1, []),
+    ok = fabric2_db:delete(DbName2, []).
+
+
 is_db_info_member(_, []) ->
     false;
 
@@ -142,3 +275,34 @@ is_db_info_member(DbName, [DbInfo | RestInfos]) ->
         _E ->
             is_db_info_member(DbName, RestInfos)
     end.
+
+
+tx_too_long_errors(UserFunCount, FoldErrors) when is_integer(UserFunCount) ->
+    tx_too_long_errors({0, UserFunCount}, FoldErrors);
+
+tx_too_long_errors(UserFunErrors, FoldCount) when is_integer(FoldCount) ->
+    tx_too_long_errors(UserFunErrors, {0, FoldCount});
+
+tx_too_long_errors({UserFunSkip, UserFunCount}, {FoldSkip, FoldCount}) ->
+    reset_error_counts(),
+    put(?PDICT_ERROR_IN_USER_FUN, {UserFunSkip, UserFunCount}),
+    put(?PDICT_ERROR_IN_FOLD_RANGE, {FoldSkip, FoldCount}).
+
+
+reset_error_counts() ->
+    erase(?PDICT_ERROR_IN_FOLD_RANGE),
+    erase(?PDICT_ERROR_IN_USER_FUN).
+
+
+maybe_tx_too_long(Key) ->
+    case get(Key) of
+        {Skip, Count} when is_integer(Skip), Skip > 0 ->
+            put(Key, {Skip - 1, Count});
+        {0, Count} when is_integer(Count), Count > 0 ->
+            put(Key, {0, Count - 1}),
+            error({erlfdb_error, 1007});
+        {0, 0} ->
+            ok;
+        undefined ->
+            ok
+    end.