Allow specifying FDB transaction options

With the latest erlfdb release v1.1.0 we have the ability to set default
transaction options on the database handle. Once set, those are inherited by
every transaction started from that handle.

Use this feature to give advanced users a way to experiment with various
transaction options. Descriptions of those options in the default.ini file have
been mostly a copy and paste from the fdb_c_option.g.h file from the client
library.

In addition, specify some safer default values for transaction timeouts (1min)
and retry limit (100). These quite conservative and are basically something
less that "infinity". In the future these may be adjusted lower.

4 files changed, 245 insertions, 12 deletions

diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index dfc67f7fb..a1e3c5851 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -709,4 +709,58 @@ compaction = false
 
 [couch_rate.views]
 limiter = couch_rate_limiter
-opts = #{budget => 100, target => 2500, window => 60000, sensitivity => 1000}
\ No newline at end of file
+opts = #{budget => 100, target => 2500, window => 60000, sensitivity => 1000}
+
+
+; Some low-level FDB transaction options. These options will be applied to the
+; database handle and inherited by each transaction started with that handle.
+; The description of these can be found in fdb_c_option.g.h include file from
+; the client libraries. The default values which were not specified the
+; fdb_c_option.g.h file were not included here either.
+
+[fdb_tx_options]
+; Specify the machine ID that was passed to fdbserver processes running on the
+; same machine as this client, for better location-aware load balancing.
+; Type is a hexadecimal string, less than 16 bytes in size.
+;machine_id =
+
+; Specify the datacenter ID that was passed to fdbserver processes running in
+; the same datacenter as this client, for better location-aware load balancing.
+; Type is hexadecimal string, less than 16 bytes in size.
+;datacenter_id =
+
+; Sets the maximum escaped length of key and value fields to be logged to the
+; trace file via the LOG_TRANSACTION option, after which the field will be
+; truncated. A negative value disables truncation.
+;transaction_logging_max_field_length =
+
+; Set a timeout in milliseconds which, when elapsed, will cause the transaction
+; automatically to be cancelled. Valid parameter values are [0, INT_MAX].
+; If set to 0, will disable all timeouts. All pending and any future uses of
+; the transaction will throw an exception. The transaction can be used again
+; after it is reset.
+;timeout = 60000
+
+; Set a maximum number of retries after which additional calls to 'on_error`
+; will throw the most recently seen error code. Valid parameter values are
+; [-1, INT_MAX]. If set to -1, will disable the retry limit.
+;retry_limit = 100
+
+; Set the maximum amount of backoff delay incurred in the call to 'on_error'
+; if the error is retryable. Defaults to 1000 ms. Valid parameter values are
+; [0, INT_MAX]. If the maximum retry delay is less than the current retry
+; delay of the transaction, then the current retry delay will be clamped to the
+; maximum retry delay. The retry limit is not reset after an
+; 'on_erro' call.
+;max_retry_delay = 1000
+
+; Set the transaction size limit in bytes. The size is calculated by combining
+; the sizes of all keys and values written or mutated, all key ranges cleared,
+; and all read and write conflict ranges. (In other words, it includes the
+; total size of all data included in the request to the cluster to commit the
+; transaction.) Large transactions can cause performance problems on
+; FoundationDB clusters, so setting this limit to a smaller value than the
+; default can help prevent the client from accidentally degrading the cluster's
+; performance. This value must be at least 10000 and cannot be set to higher than
+; 10000000, the default transaction size limit.
+;size_limit = 10000000
diff --git a/src/fabric/include/fabric2.hrl b/src/fabric/include/fabric2.hrl
index 234c5291e..27f3d61c2 100644
--- a/src/fabric/include/fabric2.hrl
+++ b/src/fabric/include/fabric2.hrl
@@ -77,6 +77,7 @@
 -define(FUTURE_VERSION, 1009).
 -define(COMMIT_UNKNOWN_RESULT, 1021).
 -define(TRANSACTION_CANCELLED, 1025).
+-define(TRANSACTION_TOO_LARGE, 2101).
 
 
 -define(DEFAULT_BINARY_CHUNK_SIZE, 100000).
diff --git a/src/fabric/src/fabric2_server.erl b/src/fabric/src/fabric2_server.erl
index 1de60f798..204246ae2 100644
--- a/src/fabric/src/fabric2_server.erl
+++ b/src/fabric/src/fabric2_server.erl
@@ -42,6 +42,21 @@
 -define(FDB_DIRECTORY, fdb_directory).
 -define(FDB_CLUSTER, fdb_cluster).
 -define(DEFAULT_FDB_DIRECTORY, <<"couchdb">>).
+-define(TX_OPTIONS_SECTION, "fdb_tx_options").
+-define(RELISTEN_DELAY, 1000).
+
+-define(DEFAULT_TIMEOUT_MSEC, "60000").
+-define(DEFAULT_RETRY_LIMIT, "100").
+
+-define(TX_OPTIONS, #{
+    machine_id                           => {binary,  undefined},
+    datacenter_id                        => {binary,  undefined},
+    transaction_logging_max_field_length => {integer, undefined},
+    timeout                              => {integer, ?DEFAULT_TIMEOUT_MSEC},
+    retry_limit                          => {integer, ?DEFAULT_RETRY_LIMIT},
+    max_retry_delay                      => {integer, undefined},
+    size_limit                           => {integer, undefined}
+}).
 
 
 start_link() ->
@@ -79,16 +94,7 @@ init(_) ->
             {read_concurrency, true},
             {write_concurrency, true}
         ]),
-
-    {Cluster, Db} = case application:get_env(fabric, eunit_run) of
-        {ok, true} ->
-            {<<"eunit_test">>, erlfdb_util:get_test_db([empty])};
-        undefined ->
-            ClusterFileStr = config:get("erlfdb", "cluster_file", ?CLUSTER_FILE),
-            {ok, ConnectionStr} = file:read_file(ClusterFileStr),
-            DbHandle = erlfdb:open(iolist_to_binary(ClusterFileStr)),
-            {string:trim(ConnectionStr), DbHandle}
-    end,
+    {Cluster, Db} = get_db_and_cluster([empty]),
     application:set_env(fabric, ?FDB_CLUSTER, Cluster),
     application:set_env(fabric, db, Db),
 
@@ -99,7 +105,7 @@ init(_) ->
             [?DEFAULT_FDB_DIRECTORY]
     end,
     application:set_env(fabric, ?FDB_DIRECTORY, Dir),
-
+    config:subscribe_for_changes([?TX_OPTIONS_SECTION]),
     {ok, nil}.
 
 
@@ -115,6 +121,27 @@ handle_cast(Msg, St) ->
     {stop, {bad_cast, Msg}, St}.
 
 
+handle_info({config_change, ?TX_OPTIONS_SECTION, _K, deleted, _}, St) ->
+    % Since we don't know the exact default values to reset the options
+    % to we recreate the db handle instead which will start with a default
+    % handle and re-apply all the options
+    {_Cluster, NewDb} = get_db_and_cluster([]),
+    application:set_env(fabric, db, NewDb),
+    {noreply, St};
+
+handle_info({config_change, ?TX_OPTIONS_SECTION, K, V, _}, St) ->
+    {ok, Db} = application:get_env(fabric, db),
+    apply_tx_options(Db, [{K, V}]),
+    {noreply, St};
+
+handle_info({gen_event_EXIT, _Handler, _Reason}, St) ->
+    erlang:send_after(?RELISTEN_DELAY, self(), restart_config_listener),
+    {noreply, St};
+
+handle_info(restart_config_listener, St) ->
+    config:subscribe_for_changes([?TX_OPTIONS_SECTION]),
+    {noreply, St};
+
 handle_info(Msg, St) ->
     {stop, {bad_info, Msg}, St}.
 
@@ -142,3 +169,51 @@ get_env(Key) ->
         Value ->
             Value
     end.
+
+
+get_db_and_cluster(EunitDbOpts) ->
+     {Cluster, Db} = case application:get_env(fabric, eunit_run) of
+        {ok, true} ->
+            {<<"eunit_test">>, erlfdb_util:get_test_db(EunitDbOpts)};
+        undefined ->
+            ClusterFileStr = config:get("erlfdb", "cluster_file", ?CLUSTER_FILE),
+            {ok, ConnectionStr} = file:read_file(ClusterFileStr),
+            DbHandle = erlfdb:open(iolist_to_binary(ClusterFileStr)),
+            {string:trim(ConnectionStr), DbHandle}
+    end,
+    apply_tx_options(Db, config:get(?TX_OPTIONS_SECTION)),
+    {Cluster, Db}.
+
+
+apply_tx_options(Db, Cfg) ->
+    maps:map(fun(Option, {Type, Default}) ->
+        case lists:keyfind(atom_to_list(Option), 1, Cfg) of
+            false ->
+                case Default of
+                    undefined -> ok;
+                    _Defined -> apply_tx_option(Db, Option, Default, Type)
+                end;
+            {_K, Val} ->
+                apply_tx_option(Db, Option, Val, Type)
+        end
+    end, ?TX_OPTIONS).
+
+
+apply_tx_option(Db, Option, Val, integer) ->
+    try
+        erlfdb:set_option(Db, Option, list_to_integer(Val))
+    catch
+        error:badarg ->
+            Msg = "~p : Invalid integer tx option ~p = ~p",
+            couch_log:error(Msg, [?MODULE, Option, Val])
+    end;
+
+apply_tx_option(Db, Option, Val, binary) ->
+    BinVal = list_to_binary(Val),
+    case size(BinVal) < 16 of
+        true ->
+            erlfdb:set_option(Db, Option, BinVal);
+        false ->
+            Msg = "~p : String tx option ~p is larger than 16 bytes",
+            couch_log:error(Msg, [?MODULE, Option])
+    end.
diff --git a/src/fabric/test/fabric2_tx_options_tests.erl b/src/fabric/test/fabric2_tx_options_tests.erl
new file mode 100644
index 000000000..34cb6e180
--- /dev/null
+++ b/src/fabric/test/fabric2_tx_options_tests.erl
@@ -0,0 +1,103 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+%   http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(fabric2_tx_options_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include("fabric2_test.hrl").
+-include("fabric2.hrl").
+
+
+fdb_tx_options_test_() ->
+    {
+        "Test setting default transaction options",
+        setup,
+        fun() ->
+            meck:new(erlfdb, [passthrough]),
+            % erlfdb, rexi and mem3 are all dependent apps for fabric. We make
+            % sure to start them so when fabric is started during the test it
+            % already has its dependencies
+            test_util:start_couch([erlfdb, rexi, mem3, ctrace, fabric])
+        end,
+        fun(Ctx) ->
+            meck:unload(),
+
+            config:delete("fdb_tx_options", "size_limit", false),
+            config:delete("fdb_tx_options", "max_retry_delay", false),
+            config:delete("fdb_tx_options", "machine_id", false),
+            config:delete("fdb_tx_options", "datacenter_id", false),
+
+            test_util:stop_couch(Ctx)
+        end,
+        with([
+            ?TDEF(options_take_effect),
+            ?TDEF(can_configure_options_at_runtime)
+        ])
+    }.
+
+
+options_take_effect(_) ->
+    ok = application:stop(fabric),
+
+    % Try one of each type including some invalid values
+    config:set("fdb_tx_options", "size_limit", "150000", false),
+    config:set("fdb_tx_options", "max_retry_delay", "badness", false),
+    config:set("fdb_tx_options", "machine_id", "123abc", false),
+    TooLong = ["x" || _ <- lists:seq(1, 1000)],
+    config:set("fdb_tx_options", "datacenter_id", TooLong, false),
+    ok = application:start(fabric),
+
+    DbName = ?tempdb(),
+    {ok, Db} = fabric2_db:create(DbName, [?ADMIN_CTX]),
+    ?assertError({erlfdb_error, ?TRANSACTION_TOO_LARGE},
+        add_large_doc(Db, 200000)),
+    ok = fabric2_db:delete(DbName, [?ADMIN_CTX]).
+
+
+can_configure_options_at_runtime(_) ->
+    meck:expect(erlfdb, set_option, fun(Fdb, Option, Val) ->
+        meck:passthrough([Fdb, Option, Val])
+    end),
+
+    meck:reset(erlfdb),
+
+    config:set("fdb_tx_options", "size_limit", "150000", false),
+    meck:wait(erlfdb, set_option, ['_', size_limit, 150000], 4000),
+
+    DbName = ?tempdb(),
+
+    {ok, Db} = fabric2_db:create(DbName, [?ADMIN_CTX]),
+    ?assertError({erlfdb_error, ?TRANSACTION_TOO_LARGE},
+        add_large_doc(Db, 200000)),
+
+    meck:reset(erlfdb),
+
+    config:delete("fdb_tx_options", "size_limit", false),
+    % Assert that we get a new handle and are setting our default values
+    meck:wait(erlfdb, set_option, ['_', timeout, '_'], 4000),
+    erase(?PDICT_DB_KEY),
+
+    {ok, Db1} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+    ?assertMatch({ok, _}, add_large_doc(Db1, 200000)),
+
+    ok = fabric2_db:delete(DbName, [?ADMIN_CTX]).
+
+
+add_large_doc(Db, Size) ->
+    Doc = #doc{
+        id = fabric2_util:uuid(),
+        body = {[{<<"x">>, crypto:strong_rand_bytes(Size)}]}
+    },
+    fabric2_db:update_doc(Db, Doc).