Support soft-deletion in fabric level

Instead of automatically and immediately removing data and index in
database after a delete operation, soft-deletion allows to restore
the deleted data back to original state due to a “fat finger”or
undesired delete operation, up to defined periods, such as 48 hours.

Co-Authored-By: Paul J. Davis <paul.joseph.davis@gmail.com>

5 files changed, 515 insertions, 14 deletions

diff --git a/src/fabric/include/fabric2.hrl b/src/fabric/include/fabric2.hrl
index 0c0757567..e12762260 100644
--- a/src/fabric/include/fabric2.hrl
+++ b/src/fabric/include/fabric2.hrl
@@ -22,6 +22,7 @@
 -define(CLUSTER_CONFIG, 0).
 -define(ALL_DBS, 1).
 -define(DB_HCA, 2).
+-define(DELETED_DBS, 3).
 -define(DBS, 15).
 -define(TX_IDS, 255).
 
diff --git a/src/fabric/src/fabric2_db.erl b/src/fabric/src/fabric2_db.erl
index fb6ae5176..3d6d9245e 100644
--- a/src/fabric/src/fabric2_db.erl
+++ b/src/fabric/src/fabric2_db.erl
@@ -17,6 +17,7 @@
     create/2,
     open/2,
     delete/2,
+    undelete/4,
 
     list_dbs/0,
     list_dbs/1,
@@ -26,6 +27,10 @@
     list_dbs_info/1,
     list_dbs_info/3,
 
+    list_deleted_dbs_info/0,
+    list_deleted_dbs_info/1,
+    list_deleted_dbs_info/3,
+
     check_is_admin/1,
     check_is_member/1,
 
@@ -202,12 +207,30 @@ delete(DbName, Options) ->
     % Delete doesn't check user_ctx, that's done at the HTTP API level
     % here we just care to get the `database_does_not_exist` error thrown
     Options1 = lists:keystore(user_ctx, 1, Options, ?ADMIN_CTX),
-    {ok, Db} = open(DbName, Options1),
-    Resp = fabric2_fdb:transactional(Db, fun(TxDb) ->
-        fabric2_fdb:delete(TxDb)
-    end),
-    if Resp /= ok -> Resp; true ->
-        fabric2_server:remove(DbName)
+    case lists:keyfind(deleted_at, 1, Options1) of
+        {deleted_at, TimeStamp} ->
+            fabric2_fdb:transactional(DbName, Options1, fun(TxDb) ->
+                fabric2_fdb:remove_deleted_db(TxDb, TimeStamp)
+            end);
+        false ->
+            {ok, Db} = open(DbName, Options1),
+            Resp = fabric2_fdb:transactional(Db, fun(TxDb) ->
+                fabric2_fdb:delete(TxDb)
+            end),
+            if Resp /= ok -> Resp; true ->
+                fabric2_server:remove(DbName)
+            end
+    end.
+
+
+undelete(DbName, TgtDbName, TimeStamp, Options) ->
+    case validate_dbname(TgtDbName) of
+        ok ->
+            fabric2_fdb:transactional(DbName, Options, fun(TxDb) ->
+                fabric2_fdb:undelete(TxDb, TgtDbName, TimeStamp)
+            end);
+        Error ->
+            Error
     end.
 
 
@@ -283,6 +306,87 @@ list_dbs_info(UserFun, UserAcc0, Options) ->
     end).
 
 
+list_deleted_dbs_info() ->
+    list_deleted_dbs_info([]).
+
+
+list_deleted_dbs_info(Options) ->
+    Callback = fun(Value, Acc) ->
+        NewAcc = case Value of
+            {meta, _} -> Acc;
+            {row, DbInfo} -> [DbInfo | Acc];
+            complete -> Acc
+        end,
+        {ok, NewAcc}
+    end,
+    {ok, DbInfos} = list_deleted_dbs_info(Callback, [], Options),
+    {ok, lists:reverse(DbInfos)}.
+
+
+list_deleted_dbs_info(UserFun, UserAcc0, Options0) ->
+    Dir = fabric2_util:get_value(dir, Options0, fwd),
+    StartKey0 = fabric2_util:get_value(start_key, Options0),
+    EndKey0 = fabric2_util:get_value(end_key, Options0),
+
+    {FirstBinary, LastBinary} = case Dir of
+        fwd -> {<<>>, <<255>>};
+        rev -> {<<255>>, <<>>}
+    end,
+
+    StartKey1 = case StartKey0 of
+        undefined ->
+            {FirstBinary};
+        DbName0 when is_binary(DbName0) ->
+            {DbName0, FirstBinary};
+        [DbName0, TimeStamp0] when is_binary(DbName0), is_binary(TimeStamp0) ->
+            {DbName0, TimeStamp0};
+        BadStartKey ->
+            erlang:error({invalid_start_key, BadStartKey})
+    end,
+    EndKey1 = case EndKey0 of
+        undefined ->
+            {LastBinary};
+        DbName1 when is_binary(DbName1) ->
+            {DbName1, LastBinary};
+        [DbName1, TimeStamp1] when is_binary(DbName1), is_binary(TimeStamp1) ->
+            {DbName1, TimeStamp1};
+        BadEndKey ->
+            erlang:error({invalid_end_key, BadEndKey})
+    end,
+
+    Options1 = Options0 -- [{start_key, StartKey0}, {end_key, EndKey0}],
+    Options2 = [
+        {start_key, StartKey1},
+        {end_key, EndKey1},
+        {wrap_keys, false}
+    ] ++ Options1,
+
+    FoldFun = fun(DbName, TimeStamp, InfoFuture, {FutureQ, Count, Acc}) ->
+        NewFutureQ = queue:in({DbName, TimeStamp, InfoFuture}, FutureQ),
+        drain_deleted_info_futures(NewFutureQ, Count + 1, UserFun, Acc)
+    end,
+    fabric2_fdb:transactional(fun(Tx) ->
+        try
+            UserAcc1 = maybe_stop(UserFun({meta, []}, UserAcc0)),
+            InitAcc = {queue:new(), 0, UserAcc1},
+            {FinalFutureQ, _, UserAcc2} = fabric2_fdb:list_deleted_dbs_info(
+                    Tx,
+                    FoldFun,
+                    InitAcc,
+                    Options2
+                ),
+            UserAcc3 = drain_all_deleted_info_futures(
+                    FinalFutureQ,
+                    UserFun,
+                    UserAcc2
+                ),
+            {ok, maybe_stop(UserFun(complete, UserAcc3))}
+        catch throw:{stop, FinalUserAcc} ->
+            {ok, FinalUserAcc}
+        end
+    end).
+
+
 is_admin(Db, {SecProps}) when is_list(SecProps) ->
     case fabric2_db_plugin:check_is_admin(Db) of
         true ->
@@ -1064,6 +1168,37 @@ drain_all_info_futures(FutureQ, UserFun, Acc) ->
     end.
 
 
+drain_deleted_info_futures(FutureQ, Count, _UserFun, Acc) when Count < 100 ->
+    {FutureQ, Count, Acc};
+
+drain_deleted_info_futures(FutureQ, Count, UserFun, Acc) when Count >= 100 ->
+    {{value, {DbName, TimeStamp, Future}}, RestQ} = queue:out(FutureQ),
+    BaseProps = fabric2_fdb:get_info_wait(Future),
+    DeletedProps = BaseProps ++ [
+        {deleted, true},
+        {timestamp, TimeStamp}
+    ],
+    DbInfo = make_db_info(DbName, DeletedProps),
+    NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
+    {RestQ, Count - 1, NewAcc}.
+
+
+drain_all_deleted_info_futures(FutureQ, UserFun, Acc) ->
+    case queue:out(FutureQ) of
+        {{value, {DbName, TimeStamp, Future}}, RestQ} ->
+            BaseProps = fabric2_fdb:get_info_wait(Future),
+            DeletedProps = BaseProps ++ [
+                {deleted, true},
+                {timestamp, TimeStamp}
+            ],
+            DbInfo = make_db_info(DbName, DeletedProps),
+            NewAcc = maybe_stop(UserFun({row, DbInfo}, Acc)),
+            drain_all_deleted_info_futures(RestQ, UserFun, NewAcc);
+        {empty, _} ->
+            Acc
+    end.
+
+
 new_revid(Db, Doc) ->
     #doc{
         id = DocId,
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index 2295a5648..430693329 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -22,12 +22,15 @@
     open/2,
     ensure_current/1,
     delete/1,
+    undelete/3,
+    remove_deleted_db/2,
     exists/1,
 
     get_dir/1,
 
     list_dbs/4,
     list_dbs_info/4,
+    list_deleted_dbs_info/4,
 
     get_info/1,
     get_info_future/2,
@@ -340,18 +343,70 @@ reopen(#{} = OldDb) ->
 
 
 delete(#{} = Db) ->
+    DoRecovery = fabric2_util:do_recovery(),
+    case DoRecovery of
+        true -> soft_delete_db(Db);
+        false -> hard_delete_db(Db)
+    end.
+
+
+undelete(#{} = Db0, TgtDbName, TimeStamp) ->
     #{
         name := DbName,
         tx := Tx,
-        layer_prefix := LayerPrefix,
-        db_prefix := DbPrefix
-    } = ensure_current(Db),
+        layer_prefix := LayerPrefix
+    } = ensure_current(Db0, false),
+    DbKey = erlfdb_tuple:pack({?ALL_DBS, TgtDbName}, LayerPrefix),
+    case erlfdb:wait(erlfdb:get(Tx, DbKey)) of
+        Bin when is_binary(Bin) ->
+            file_exists;
+        not_found ->
+            DeletedDbTupleKey = {
+                ?DELETED_DBS,
+                DbName,
+                TimeStamp
+            },
+            DeleteDbKey = erlfdb_tuple:pack(DeletedDbTupleKey, LayerPrefix),
+            case erlfdb:wait(erlfdb:get(Tx, DeleteDbKey)) of
+                not_found ->
+                    not_found;
+                DbPrefix ->
+                    erlfdb:set(Tx, DbKey, DbPrefix),
+                    erlfdb:clear(Tx, DeleteDbKey),
+                    bump_db_version(#{
+                        tx => Tx,
+                        db_prefix => DbPrefix
+                    }),
+                    ok
+            end
+    end.
 
-    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
-    erlfdb:clear(Tx, DbKey),
-    erlfdb:clear_range_startswith(Tx, DbPrefix),
-    bump_metadata_version(Tx),
-    ok.
+
+remove_deleted_db(#{} = Db0, TimeStamp) ->
+    #{
+        name := DbName,
+        tx := Tx,
+        layer_prefix := LayerPrefix
+    } = ensure_current(Db0, false),
+
+    DeletedDbTupleKey = {
+        ?DELETED_DBS,
+        DbName,
+        TimeStamp
+    },
+    DeletedDbKey = erlfdb_tuple:pack(DeletedDbTupleKey, LayerPrefix),
+    case erlfdb:wait(erlfdb:get(Tx, DeletedDbKey)) of
+        not_found ->
+            not_found;
+        DbPrefix ->
+            erlfdb:clear(Tx, DeletedDbKey),
+            erlfdb:clear_range_startswith(Tx, DbPrefix),
+            bump_db_version(#{
+                tx => Tx,
+                db_prefix => DbPrefix
+            }),
+            ok
+    end.
 
 
 exists(#{name := DbName} = Db) when is_binary(DbName) ->
@@ -401,6 +456,20 @@ list_dbs_info(Tx, Callback, AccIn, Options0) ->
     end, AccIn, Options).
 
 
+list_deleted_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({?DELETED_DBS}, LayerPrefix),
+    fold_range({tx, Tx}, Prefix, fun({DbKey, DbPrefix}, Acc) ->
+        {DbName, TimeStamp} = erlfdb_tuple:unpack(DbKey, Prefix),
+        InfoFuture = get_info_future(Tx, DbPrefix),
+        Callback(DbName, TimeStamp, InfoFuture, Acc)
+    end, AccIn, Options).
+
+
 get_info(#{} = Db) ->
     #{
         tx := Tx,
@@ -1186,6 +1255,45 @@ check_db_version(#{} = Db, CheckDbVersion) ->
     end.
 
 
+soft_delete_db(Db) ->
+    #{
+        name := DbName,
+        tx := Tx,
+        layer_prefix := LayerPrefix,
+        db_prefix := DbPrefix
+    } = ensure_current(Db),
+
+    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
+    Timestamp = list_to_binary(fabric2_util:iso8601_timestamp()),
+    DeletedDbKeyTuple = {?DELETED_DBS, DbName, Timestamp},
+    DeletedDbKey = erlfdb_tuple:pack(DeletedDbKeyTuple, LayerPrefix),
+    case erlfdb:wait(erlfdb:get(Tx, DeletedDbKey)) of
+        not_found ->
+            erlfdb:set(Tx, DeletedDbKey, DbPrefix),
+            erlfdb:clear(Tx, DbKey),
+            bump_db_version(Db),
+            ok;
+        _Val ->
+            {deletion_frequency_exceeded, DbName}
+    end.
+
+
+hard_delete_db(Db) ->
+    #{
+        name := DbName,
+        tx := Tx,
+        layer_prefix := LayerPrefix,
+        db_prefix := DbPrefix
+    } = ensure_current(Db),
+
+    DbKey = erlfdb_tuple:pack({?ALL_DBS, DbName}, LayerPrefix),
+
+    erlfdb:clear(Tx, DbKey),
+    erlfdb:clear_range_startswith(Tx, DbPrefix),
+    bump_metadata_version(Tx),
+    ok.
+
+
 write_doc_body(#{} = Db0, #doc{} = Doc) ->
     #{
         tx := Tx
@@ -1514,6 +1622,7 @@ get_fold_acc(Db, RangePrefix, UserCallback, UserAcc, Options)
     EndKeyGt = fabric2_util:get_value(end_key_gt, Options),
     EndKey0 = fabric2_util:get_value(end_key, Options, EndKeyGt),
     InclusiveEnd = EndKeyGt == undefined,
+    WrapKeys = fabric2_util:get_value(wrap_keys, Options) /= false,
 
     % CouchDB swaps the key meanings based on the direction
     % of the fold. FoundationDB does not so we have to
@@ -1527,6 +1636,8 @@ get_fold_acc(Db, RangePrefix, UserCallback, UserAcc, Options)
     StartKey2 = case StartKey1 of
         undefined ->
             <<RangePrefix/binary, 16#00>>;
+        SK2 when not WrapKeys ->
+            erlfdb_tuple:pack(SK2, RangePrefix);
         SK2 ->
             erlfdb_tuple:pack({SK2}, RangePrefix)
     end,
@@ -1534,9 +1645,14 @@ get_fold_acc(Db, RangePrefix, UserCallback, UserAcc, Options)
     EndKey2 = case EndKey1 of
         undefined ->
             <<RangePrefix/binary, 16#FF>>;
+        EK2 when Reverse andalso not WrapKeys ->
+            PackedEK = erlfdb_tuple:pack(EK2, RangePrefix),
+            <<PackedEK/binary, 16#FF>>;
         EK2 when Reverse ->
             PackedEK = erlfdb_tuple:pack({EK2}, RangePrefix),
             <<PackedEK/binary, 16#FF>>;
+        EK2 when not WrapKeys ->
+            erlfdb_tuple:pack(EK2, RangePrefix);
         EK2 ->
             erlfdb_tuple:pack({EK2}, RangePrefix)
     end,
diff --git a/src/fabric/src/fabric2_util.erl b/src/fabric/src/fabric2_util.erl
index 97bfedc2c..9b6d18c58 100644
--- a/src/fabric/src/fabric2_util.erl
+++ b/src/fabric/src/fabric2_util.erl
@@ -40,6 +40,9 @@
     encode_all_doc_key/1,
     all_docs_view_opts/1,
 
+    iso8601_timestamp/0,
+    do_recovery/0,
+
     pmap/2,
     pmap/3
 ]).
@@ -337,6 +340,19 @@ all_docs_view_opts(#mrargs{} = Args) ->
     ] ++ StartKeyOpts ++ EndKeyOpts.
 
 
+iso8601_timestamp() ->
+    Now = os:timestamp(),
+    {{Year, Month, Date}, {Hour, Minute, Second}} =
+        calendar:now_to_datetime(Now),
+    Format = "~4.10.0B-~2.10.0B-~2.10.0BT~2.10.0B:~2.10.0B:~2.10.0BZ",
+    io_lib:format(Format, [Year, Month, Date, Hour, Minute, Second]).
+
+
+do_recovery() ->
+    config:get_boolean("couchdb",
+        "enable_database_recovery", false).
+
+
 pmap(Fun, Args) ->
     pmap(Fun, Args, []).
 
diff --git a/src/fabric/test/fabric2_db_crud_tests.erl b/src/fabric/test/fabric2_db_crud_tests.erl
index f409389d6..d5025b987 100644
--- a/src/fabric/test/fabric2_db_crud_tests.erl
+++ b/src/fabric/test/fabric2_db_crud_tests.erl
@@ -37,6 +37,9 @@ crud_test_() ->
                     ?TDEF_FE(open_db),
                     ?TDEF_FE(delete_db),
                     ?TDEF_FE(recreate_db),
+                    ?TDEF_FE(undelete_db),
+                    ?TDEF_FE(remove_deleted_db),
+                    ?TDEF_FE(old_db_handle),
                     ?TDEF_FE(list_dbs),
                     ?TDEF_FE(list_dbs_user_fun),
                     ?TDEF_FE(list_dbs_user_fun_partial),
@@ -44,6 +47,10 @@ crud_test_() ->
                     ?TDEF_FE(list_dbs_info_partial),
                     ?TDEF_FE(list_dbs_tx_too_old),
                     ?TDEF_FE(list_dbs_info_tx_too_old),
+                    ?TDEF_FE(list_deleted_dbs_info),
+                    ?TDEF_FE(list_deleted_dbs_info_user_fun),
+                    ?TDEF_FE(list_deleted_dbs_info_user_fun_partial),
+                    ?TDEF_FE(list_deleted_dbs_info_with_timestamps),
                     ?TDEF_FE(get_info_wait_retry_on_tx_too_old),
                     ?TDEF_FE(get_info_wait_retry_on_tx_abort)
                 ]
@@ -68,6 +75,7 @@ setup() ->
 
 
 cleanup(_) ->
+    ok = config:set("couchdb", "enable_database_recovery", "false", false),
     fabric2_test_util:tx_too_old_reset_errors(),
     reset_fail_erfdb_wait(),
     meck:reset([erlfdb]).
@@ -139,6 +147,123 @@ recreate_db(_) ->
     ?assertError(database_does_not_exist, fabric2_db:open(DbName, BadOpts)).
 
 
+undelete_db(_) ->
+    DbName = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName, [])),
+
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(true, ets:member(fabric2_server, DbName)),
+
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+    ?assertEqual(false, ets:member(fabric2_server, DbName)),
+
+
+    {ok, Infos} = fabric2_db:list_deleted_dbs_info(),
+    [DeletedDbInfo] = [Info || Info <- Infos,
+        DbName == proplists:get_value(db_name, Info)
+    ],
+    Timestamp = proplists:get_value(timestamp, DeletedDbInfo),
+
+    OldTS = <<"2020-01-01T12:00:00Z">>,
+    ?assertEqual(not_found, fabric2_db:undelete(DbName, DbName, OldTS, [])),
+    BadDbName = <<"bad_dbname">>,
+    ?assertEqual(not_found,
+        fabric2_db:undelete(BadDbName, BadDbName, Timestamp, [])),
+
+    ok = fabric2_db:undelete(DbName, DbName, Timestamp, []),
+    {ok, AllDbInfos} = fabric2_db:list_dbs_info(),
+    ?assert(is_db_info_member(DbName, AllDbInfos)).
+
+
+remove_deleted_db(_) ->
+    DbName = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName, [])),
+
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(true, ets:member(fabric2_server, DbName)),
+
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+    ?assertEqual(false, ets:member(fabric2_server, DbName)),
+
+    {ok, Infos} = fabric2_db:list_deleted_dbs_info(),
+    [DeletedDbInfo] = [Info || Info <- Infos,
+        DbName == proplists:get_value(db_name, Info)
+    ],
+    Timestamp = proplists:get_value(timestamp, DeletedDbInfo),
+    OldTS = <<"2020-01-01T12:00:00Z">>,
+    ?assertEqual(not_found,
+        fabric2_db:delete(DbName, [{deleted_at, OldTS}])),
+    BadDbName = <<"bad_dbname">>,
+    ?assertEqual(not_found,
+        fabric2_db:delete(BadDbName, [{deleted_at, Timestamp}])),
+
+    ok = fabric2_db:delete(DbName, [{deleted_at, Timestamp}]),
+    {ok, Infos2} = fabric2_db:list_deleted_dbs_info(),
+    DeletedDbs = [proplists:get_value(db_name, Info) || Info <- Infos2],
+    ?assert(not lists:member(DbName, DeletedDbs)).
+
+
+old_db_handle(_) ->
+    % db hard deleted
+    DbName1 = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName1, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName1, [])),
+    {ok, Db1} = fabric2_db:open(DbName1, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db1)),
+    ?assertEqual(ok, fabric2_db:delete(DbName1, [])),
+    ?assertError(database_does_not_exist, fabric2_db:get_db_info(Db1)),
+
+    % db soft deleted
+    DbName2 = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName2, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName2, [])),
+    {ok, Db2} = fabric2_db:open(DbName2, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db2)),
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName2, [])),
+    ?assertError(database_does_not_exist, fabric2_db:get_db_info(Db2)),
+
+    % db soft deleted and re-created
+    DbName3 = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName3, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName3, [])),
+    {ok, Db3} = fabric2_db:open(DbName3, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db3)),
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName3, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName3, [])),
+    ?assertError(database_does_not_exist, fabric2_db:get_db_info(Db3)),
+
+    % db soft deleted and undeleted
+    DbName4 = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName4, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName4, [])),
+    {ok, Db4} = fabric2_db:open(DbName4, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db4)),
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName4, [])),
+    {ok, Infos} = fabric2_db:list_deleted_dbs_info(),
+    [DeletedDbInfo] = [Info || Info <- Infos,
+        DbName4 == proplists:get_value(db_name, Info)
+    ],
+    Timestamp = proplists:get_value(timestamp, DeletedDbInfo),
+    ok = fabric2_db:undelete(DbName4, DbName4, Timestamp, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db4)),
+
+    % db hard deleted and re-created
+    DbName5 = ?tempdb(),
+    ?assertError(database_does_not_exist, fabric2_db:delete(DbName5, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName5, [])),
+    {ok, Db5} = fabric2_db:open(DbName5, []),
+    ?assertMatch({ok, _}, fabric2_db:get_db_info(Db5)),
+    ok = config:set("couchdb", "enable_database_recovery", "false", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName5, [])),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName5, [])),
+    ?assertError(database_does_not_exist, fabric2_db:get_db_info(Db5)).
+
+
 list_dbs(_) ->
     DbName = ?tempdb(),
     AllDbs1 = fabric2_db:list_dbs(),
@@ -295,6 +420,108 @@ list_dbs_info_tx_too_old(_) ->
     end, DbNames).
 
 
+list_deleted_dbs_info(_) ->
+    DbName = ?tempdb(),
+    AllDbs1 = fabric2_db:list_dbs(),
+
+    ?assert(is_list(AllDbs1)),
+    ?assert(not lists:member(DbName, AllDbs1)),
+
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    AllDbs2 = fabric2_db:list_dbs(),
+    ?assert(lists:member(DbName, AllDbs2)),
+
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+
+    AllDbs3 = fabric2_db:list_dbs(),
+    ?assert(not lists:member(DbName, AllDbs3)),
+    {ok, DeletedDbsInfo} = fabric2_db:list_deleted_dbs_info(),
+    DeletedDbs4 = get_deleted_dbs(DeletedDbsInfo),
+    ?assert(lists:member(DbName, DeletedDbs4)).
+
+
+list_deleted_dbs_info_user_fun(_) ->
+    DbName = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+
+    UserFun = fun(Row, Acc) -> {ok, [Row | Acc]} end,
+    {ok, UserAcc} = fabric2_db:list_deleted_dbs_info(UserFun, [], []),
+    {ok, DeletedDbsInfo} = fabric2_db:list_deleted_dbs_info(),
+
+    Base = lists:foldl(fun(DbInfo, Acc) ->
+        [{row, DbInfo} | Acc]
+    end, [{meta, []}], DeletedDbsInfo),
+    Expect = lists:reverse(Base, [complete]),
+
+    ?assertEqual(Expect, lists:reverse(UserAcc)).
+
+
+list_deleted_dbs_info_user_fun_partial(_) ->
+    UserFun = fun(Row, Acc) -> {stop, [Row | Acc]} end,
+    {ok, UserAcc} = fabric2_db:list_deleted_dbs_info(UserFun, [], []),
+    ?assertEqual([{meta, []}], UserAcc).
+
+
+list_deleted_dbs_info_with_timestamps(_) ->
+    ok = config:set("couchdb", "enable_database_recovery", "true", false),
+
+    % Cycle our database three times to get multiple entries
+    DbName = ?tempdb(),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+    timer:sleep(1100),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+    timer:sleep(1100),
+    ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+    ?assertEqual(ok, fabric2_db:delete(DbName, [])),
+
+    UserFun = fun(Row, Acc) ->
+        case Row of
+            {row, Info} -> {ok, [Info | Acc]};
+            _ -> {ok, Acc}
+        end
+    end,
+
+    Options1 = [{start_key, DbName}, {end_key, <<DbName/binary, 255>>}],
+    {ok, Infos1} = fabric2_db:list_deleted_dbs_info(UserFun, [], Options1),
+    TimeStamps1 = [fabric2_util:get_value(timestamp, Info) || Info <- Infos1],
+    ?assertEqual(3, length(TimeStamps1)),
+
+    [FirstTS, MiddleTS, LastTS] = lists:sort(TimeStamps1),
+
+    % Check we can skip over the FirstTS
+    Options2 = [{start_key, [DbName, MiddleTS]}, {end_key, [DbName, LastTS]}],
+    {ok, Infos2} = fabric2_db:list_deleted_dbs_info(UserFun, [], Options2),
+    TimeStamps2 = [fabric2_util:get_value(timestamp, Info) || Info <- Infos2],
+    ?assertEqual(2, length(TimeStamps2)),
+    ?assertEqual([LastTS, MiddleTS], TimeStamps2), % because foldl reverses
+
+    % Check we an end before LastTS
+    Options3 = [{start_key, DbName}, {end_key, [DbName, MiddleTS]}],
+    {ok, Infos3} = fabric2_db:list_deleted_dbs_info(UserFun, [], Options3),
+    TimeStamps3 = [fabric2_util:get_value(timestamp, Info) || Info <- Infos3],
+    ?assertEqual([MiddleTS, FirstTS], TimeStamps3),
+
+    % Check that {dir, rev} works without timestamps
+    Options4 = [{start_key, DbName}, {end_key, DbName}, {dir, rev}],
+    {ok, Infos4} = fabric2_db:list_deleted_dbs_info(UserFun, [], Options4),
+    TimeStamps4 = [fabric2_util:get_value(timestamp, Info) || Info <- Infos4],
+    ?assertEqual([FirstTS, MiddleTS, LastTS], TimeStamps4),
+
+    % Check that reverse with keys returns correctly
+    Options5 = [
+        {start_key, [DbName, MiddleTS]},
+        {end_key, [DbName, FirstTS]},
+        {dir, rev}
+    ],
+    {ok, Infos5} = fabric2_db:list_deleted_dbs_info(UserFun, [], Options5),
+    TimeStamps5 = [fabric2_util:get_value(timestamp, Info) || Info <- Infos5],
+    ?assertEqual([FirstTS, MiddleTS], TimeStamps5).
+
+
 get_info_wait_retry_on_tx_too_old(_) ->
     DbName = ?tempdb(),
     ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
@@ -382,3 +609,9 @@ is_db_info_member(DbName, [DbInfo | RestInfos]) ->
         _E ->
             is_db_info_member(DbName, RestInfos)
     end.
+
+get_deleted_dbs(DeletedDbInfos)  ->
+    lists:foldl(fun(DbInfo, Acc) ->
+        DbName = fabric2_util:get_value(db_name, DbInfo),
+        [DbName | Acc]
+    end, [], DeletedDbInfos).