Allow applying per-transaction options with fabric2_fdb:transactional/3

1) First, as a cleanup, remove DB `Options` from the `init_db/3 call. We always
follow `init_db/3` (sometimes called through the
`fabric2_fdb:transactional(DbName, ...)` with a `create(TxDb, Options)` or
`open(TxDb, Options)` call, where we overrode `Options` anyway. The only time
we didn't follow it up with a `create/2` or `open/2` is when dbs are deleted
where `Options` wouldn't matter.

2) Add a new `fabric2_fdb:transactional(DbName|Db, TxOptions, Fun)` call which
allows specifying per-transaction TX options in the `TxOptions` arg. The format
of `TxOptions` is `#{option_name_as_atom => integer | binary}`

3 files changed, 71 insertions, 24 deletions

diff --git a/src/fabric/src/fabric2_db.erl b/src/fabric/src/fabric2_db.erl
index a310470ea..6d8bb72e9 100644
--- a/src/fabric/src/fabric2_db.erl
+++ b/src/fabric/src/fabric2_db.erl
@@ -173,7 +173,7 @@
 create(DbName, Options) ->
     case validate_dbname(DbName) of
         ok ->
-            Result = fabric2_fdb:transactional(DbName, Options, fun(TxDb) ->
+            Result = fabric2_fdb:transactional(DbName, fun(TxDb) ->
                 case fabric2_fdb:exists(TxDb) of
                     true ->
                         {error, file_exists};
@@ -205,7 +205,7 @@ open(DbName, Options) ->
             Db2 = maybe_set_interactive(Db1, Options),
             {ok, require_member_check(Db2)};
         undefined ->
-            Result = fabric2_fdb:transactional(DbName, Options, fun(TxDb) ->
+            Result = fabric2_fdb:transactional(DbName, fun(TxDb) ->
                 fabric2_fdb:open(TxDb, Options)
             end),
             % Cache outside the transaction retry loop
@@ -227,7 +227,7 @@ delete(DbName, Options) ->
     Options1 = lists:keystore(user_ctx, 1, Options, ?ADMIN_CTX),
     case lists:keyfind(deleted_at, 1, Options1) of
         {deleted_at, TimeStamp} ->
-            fabric2_fdb:transactional(DbName, Options1, fun(TxDb) ->
+            fabric2_fdb:transactional(DbName, fun(TxDb) ->
                 fabric2_fdb:remove_deleted_db(TxDb, TimeStamp)
             end);
         false ->
@@ -245,7 +245,7 @@ delete(DbName, Options) ->
 undelete(DbName, TgtDbName, TimeStamp, Options) ->
     case validate_dbname(TgtDbName) of
         ok ->
-            Resp = fabric2_fdb:transactional(DbName, Options,
+            Resp = fabric2_fdb:transactional(DbName,
                 fun(TxDb) ->
                     fabric2_fdb:undelete(TxDb, TgtDbName, TimeStamp)
                 end
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index 1690f2f2c..8c60d960c 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -15,8 +15,8 @@
 
 -export([
     transactional/1,
-    transactional/3,
     transactional/2,
+    transactional/3,
 
     with_snapshot/2,
 
@@ -118,18 +118,25 @@
 
 
 transactional(Fun) ->
-    do_transaction(Fun, undefined).
+    do_transaction(Fun, undefined, #{}).
 
 
-transactional(DbName, Options, Fun) when is_binary(DbName) ->
-    with_span(Fun, #{'db.name' => DbName}, fun() ->
-        transactional(fun(Tx) ->
-            Fun(init_db(Tx, DbName, Options))
-        end)
-    end).
+transactional(DbName, Fun) when is_binary(DbName), is_function(Fun) ->
+    transactional(DbName, #{}, Fun);
+
+transactional(#{} = Db, Fun) when is_function(Fun) ->
+    transactional(Db, #{}, Fun).
 
 
-transactional(#{tx := undefined} = Db, Fun) ->
+
+transactional(DbName, #{} = TxOptions, Fun) when is_binary(DbName) ->
+    with_span(Fun, #{'db.name' => DbName}, fun() ->
+        do_transaction(fun(Tx) ->
+            Fun(init_db(Tx, DbName))
+        end, undefined, TxOptions)
+    end);
+
+transactional(#{tx := undefined} = Db, #{} = TxOptions, Fun) ->
     DbName = maps:get(name, Db, undefined),
     try
         Db1 = refresh(Db),
@@ -145,30 +152,31 @@ transactional(#{tx := undefined} = Db, Fun) ->
                     true -> Fun(reopen(Db2#{tx => Tx}));
                     false -> Fun(Db2#{tx => Tx})
                 end
-            end, LayerPrefix)
+            end, LayerPrefix, TxOptions)
         end)
     catch throw:{?MODULE, reopen} ->
         with_span('db.reopen', #{'db.name' => DbName}, fun() ->
             transactional(Db#{reopen => true}, Fun)
         end)
     end;
-transactional(#{tx := {erlfdb_snapshot, _}} = Db, Fun) ->
+transactional(#{tx := {erlfdb_snapshot, _}} = Db, #{} = _TxOptions, Fun) ->
     DbName = maps:get(name, Db, undefined),
     with_span(Fun, #{'db.name' => DbName}, fun() ->
         Fun(Db)
     end);
 
-transactional(#{tx := {erlfdb_transaction, _}} = Db, Fun) ->
+transactional(#{tx := {erlfdb_transaction, _}} = Db, #{} = _TxOptions, Fun) ->
     DbName = maps:get(name, Db, undefined),
     with_span(Fun, #{'db.name' => DbName}, fun() ->
         Fun(Db)
     end).
 
 
-do_transaction(Fun, LayerPrefix) when is_function(Fun, 1) ->
+do_transaction(Fun, LayerPrefix, #{} = TxOptions) when is_function(Fun, 1) ->
     Db = get_db_handle(),
     try
         erlfdb:transactional(Db, fun(Tx) ->
+            apply_tx_options(Tx, TxOptions),
             case get(erlfdb_trace) of
                 Name when is_binary(Name) ->
                     UId = erlang:unique_integer([positive]),
@@ -190,6 +198,12 @@ do_transaction(Fun, LayerPrefix) when is_function(Fun, 1) ->
     end.
 
 
+apply_tx_options(Tx, #{} = TxOptions) ->
+    maps:map(fun(K, V) ->
+        erlfdb:set_option(Tx, K, V)
+    end, TxOptions).
+
+
 with_snapshot(#{tx := {erlfdb_transaction, _} = Tx} = TxDb, Fun) ->
     SSDb = TxDb#{tx := erlfdb:snapshot(Tx)},
     Fun(SSDb);
@@ -365,7 +379,7 @@ reopen(#{} = OldDb) ->
         interactive := Interactive
     } = OldDb,
     Options1 = lists:keystore(user_ctx, 1, Options, {user_ctx, UserCtx}),
-    NewDb = open(init_db(Tx, DbName, Options1), Options1),
+    NewDb = open(init_db(Tx, DbName), Options1),
 
     % Check if database was re-created
     case {Interactive, maps:get(uuid, NewDb)} of
@@ -1226,7 +1240,7 @@ debug_cluster(Start, End) ->
     end).
 
 
-init_db(Tx, DbName, Options) ->
+init_db(Tx, DbName) ->
     Prefix = get_dir(Tx),
     Version = erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)),
     #{
@@ -1236,7 +1250,7 @@ init_db(Tx, DbName, Options) ->
         md_version => Version,
 
         security_fun => undefined,
-        db_options => Options
+        db_options => []
     }.
 
 
diff --git a/src/fabric/test/fabric2_tx_options_tests.erl b/src/fabric/test/fabric2_tx_options_tests.erl
index 323292287..cff383e57 100644
--- a/src/fabric/test/fabric2_tx_options_tests.erl
+++ b/src/fabric/test/fabric2_tx_options_tests.erl
@@ -43,7 +43,9 @@ fdb_tx_options_test_() ->
         end,
         with([
             ?TDEF(options_take_effect, 15),
-            ?TDEF(can_configure_options_at_runtime, 15)
+            ?TDEF(can_configure_options_at_runtime, 15),
+            ?TDEF(can_apply_options_to_db_name_transactions),
+            ?TDEF(can_apply_options_to_db_handle_transactions)
         ])
     }.
 
@@ -101,9 +103,40 @@ can_configure_options_at_runtime(_) ->
     ok = fabric2_db:delete(DbName, [?ADMIN_CTX]).
 
 
+can_apply_options_to_db_name_transactions(_) ->
+    DbName = ?tempdb(),
+
+    TxFun = fun(TxDb) ->
+        #{tx := Tx} = TxDb,
+        fabric2_fdb:create(TxDb, [?ADMIN_CTX]),
+        erlfdb:wait(erlfdb:get(Tx, <<16#FF, "/primaryDatacenter">>))
+    end,
+    TxOpts = #{read_system_keys => <<>>},
+    ?assertEqual(<<>>, fabric2_fdb:transactional(DbName, TxOpts, TxFun)),
+
+    ok = fabric2_db:delete(DbName, [?ADMIN_CTX]).
+
+
+can_apply_options_to_db_handle_transactions(_) ->
+    DbName = ?tempdb(),
+    {ok, Db} = fabric2_db:create(DbName, [?ADMIN_CTX]),
+
+    TxFun = fun(TxDb) ->
+        fabric2_db:update_doc(TxDb, large_doc(200000))
+    end,
+    TxOpts = #{size_limit => 150000},
+    ?assertError({erlfdb_error, ?TRANSACTION_TOO_LARGE},
+        fabric2_fdb:transactional(Db, TxOpts, TxFun)),
+
+    ok = fabric2_db:delete(DbName, [?ADMIN_CTX]).
+
+
 add_large_doc(Db, Size) ->
-    Doc = #doc{
+    fabric2_db:update_doc(Db, large_doc(Size)).
+
+
+large_doc(Size) ->
+    #doc{
         id = fabric2_util:uuid(),
         body = {[{<<"x">>, crypto:strong_rand_bytes(Size)}]}
-    },
-    fabric2_db:update_doc(Db, Doc).
+    }.