Take better advantage of metadata version key featurecheck-metadata-first

FDB's metadata version key allows more efficient metadata invalidation
(see https://github.com/apple/foundationdb/pull/1213). To take advantage of
that feature update the caching logic to check the metadata version first, and
if it is current, skip checking the db version altogether.

When db version is bumped we now update the metadata version as well.

There is a bit of a subtlety when the metadata version is stale. In that case
we check the db version, and if that is current, we still don't reopen the
database, instead we continue with the transaction. Then, after the transaction
succeeds, we update the cached metadata version for that db handle. Next client
would get the updated db metadata, it will be current, and they won't need to
check the db version. If the db version is stale as well, then we throw a
`reopen` exception and the handle gets removed from the cache and reopened.

Note: this commit doesn't actually use the new metadata version key, it still
uses the old plain key. That update will be a separate commit where we also start
setting a new API version (610) and will only work on FDB version 6.1.x

3 files changed, 102 insertions, 30 deletions

diff --git a/src/fabric/include/fabric2.hrl b/src/fabric/include/fabric2.hrl
index 3e224987d..fe11e6b8d 100644
--- a/src/fabric/include/fabric2.hrl
+++ b/src/fabric/include/fabric2.hrl
@@ -59,8 +59,10 @@
 -define(PDICT_DB_KEY, '$fabric_db_handle').
 -define(PDICT_LAYER_CACHE, '$fabric_layer_id').
 -define(PDICT_CHECKED_DB_IS_CURRENT, '$fabric_checked_db_is_current').
+-define(PDICT_CHECKED_MD_IS_CURRENT, '$fabric_checked_md_is_current').
 -define(PDICT_TX_ID_KEY, '$fabric_tx_id').
 -define(PDICT_TX_RES_KEY, '$fabric_tx_result').
+-define(PDICT_ON_COMMIT_FUN, '$fabric_on_commit_fun').
 -define(COMMIT_UNKNOWN_RESULT, 1021).
 
 
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index 0f55d9175..1ee4dffff 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -864,6 +864,31 @@ bump_metadata_version(Tx) ->
     erlfdb:set_versionstamped_value(Tx, ?METADATA_VERSION_KEY, <<0:112>>).
 
 
+check_metadata_version(#{} = Db) ->
+    #{
+        tx := Tx,
+        layer_prefix := LayerPrefix,
+        name := DbName,
+        md_version := Version
+    } = Db,
+
+    AlreadyChecked = get(?PDICT_CHECKED_MD_IS_CURRENT),
+    if AlreadyChecked == true -> {current, Db}; true ->
+        case erlfdb:wait(erlfdb:get_ss(Tx, ?METADATA_VERSION_KEY)) of
+            Version ->
+                put(?PDICT_CHECKED_MD_IS_CURRENT, true),
+                % We want to set a read conflict on the db version as we'd want
+                % to to conflict with any writes to this particular db
+                DbPrefix = erlfdb_tuple:pack({?DBS, DbName}, LayerPrefix),
+                DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
+                erlfdb:add_read_conflict_key(Tx, DbVersionKey),
+                {current, Db};
+            NewVersion ->
+                {stale, Db#{md_version := NewVersion}}
+        end
+    end.
+
+
 bump_db_version(#{} = Db) ->
     #{
         tx := Tx,
@@ -872,7 +897,30 @@ bump_db_version(#{} = Db) ->
 
     DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
     DbVersion = fabric2_util:uuid(),
-    ok = erlfdb:set(Tx, DbVersionKey, DbVersion).
+    ok = erlfdb:set(Tx, DbVersionKey, DbVersion),
+    ok = bump_metadata_version(Tx).
+
+
+check_db_version(#{} = Db, CheckDbVersion) ->
+    #{
+        tx := Tx,
+        db_prefix := DbPrefix,
+        db_version := DbVersion
+    } = Db,
+
+    AlreadyChecked = get(?PDICT_CHECKED_DB_IS_CURRENT),
+    if not CheckDbVersion orelse AlreadyChecked == true -> Db; true ->
+        DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
+        case erlfdb:wait(erlfdb:get(Tx, DbVersionKey)) of
+            DbVersion ->
+                put(?PDICT_CHECKED_DB_IS_CURRENT, true),
+                on_commit(Tx, fun() -> fabric2_server:store(Db) end),
+                Db;
+            _NewDBVersion ->
+                fabric2_server:remove(maps:get(name, Db)),
+                throw({?MODULE, reopen})
+        end
+    end.
 
 
 write_doc_body(#{} = Db0, #doc{} = Doc) ->
@@ -1171,34 +1219,9 @@ ensure_current(Db) ->
 
 ensure_current(#{} = Db, CheckDbVersion) ->
     require_transaction(Db),
-
-    #{
-        tx := Tx,
-        md_version := MetaDataVersion
-    } = Db,
-
-    case erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)) of
-        MetaDataVersion -> Db;
-        _NewVersion -> throw({?MODULE, reopen})
-    end,
-
-    AlreadyChecked = get(?PDICT_CHECKED_DB_IS_CURRENT),
-    if not CheckDbVersion orelse AlreadyChecked == true -> Db; true ->
-        #{
-            db_prefix := DbPrefix,
-            db_version := DbVersion
-        } = Db,
-
-        DbVersionKey = erlfdb_tuple:pack({?DB_VERSION}, DbPrefix),
-
-        case erlfdb:wait(erlfdb:get(Tx, DbVersionKey)) of
-            DbVersion ->
-                put(?PDICT_CHECKED_DB_IS_CURRENT, true),
-                Db;
-            _NewDBVersion ->
-                fabric2_server:remove(maps:get(name, Db)),
-                throw({?MODULE, reopen})
-        end
+    case check_metadata_version(Db) of
+        {current, Db1} -> Db1;
+        {stale, Db1} -> check_db_version(Db1, CheckDbVersion)
     end.
 
 
@@ -1222,12 +1245,14 @@ execute_transaction(Tx, Fun, LayerPrefix) ->
             erlfdb:set(Tx, get_transaction_id(Tx, LayerPrefix), <<>>),
             put(?PDICT_TX_RES_KEY, Result)
     end,
+    ok = run_on_commit_fun(Tx),
     Result.
 
 
 clear_transaction() ->
     fabric2_txids:remove(get(?PDICT_TX_ID_KEY)),
     erase(?PDICT_CHECKED_DB_IS_CURRENT),
+    erase(?PDICT_CHECKED_MD_IS_CURRENT),
     erase(?PDICT_TX_ID_KEY),
     erase(?PDICT_TX_RES_KEY).
 
@@ -1259,3 +1284,25 @@ new_versionstamp(Tx) ->
     TxId = erlfdb:get_next_tx_id(Tx),
     {versionstamp, 16#FFFFFFFFFFFFFFFF, 16#FFFF, TxId}.
 
+
+on_commit(Tx, Fun) when is_function(Fun, 0) ->
+    % Here we rely on Tx objects matching. However they contain a nif resource
+    % object. Before Erlang 20.0 those would have been represented as empty
+    % binaries and would have compared equal to each other. See
+    % http://erlang.org/doc/man/erl_nif.html for more info. We assume we run on
+    % Erlang 20+ here and don't worry about that anymore.
+    case get({?PDICT_ON_COMMIT_FUN, Tx}) of
+        undefined -> put({?PDICT_ON_COMMIT_FUN, Tx}, Fun);
+        _ -> error({?MODULE, on_commit_function_already_set})
+    end.
+
+
+run_on_commit_fun(Tx) ->
+    case get({?PDICT_ON_COMMIT_FUN, Tx}) of
+        undefined ->
+            ok;
+        Fun when is_function(Fun, 0) ->
+            Fun(),
+            erase({?PDICT_ON_COMMIT_FUN, Tx}),
+            ok
+    end.
diff --git a/src/fabric/test/fabric2_db_misc_tests.erl b/src/fabric/test/fabric2_db_misc_tests.erl
index 8e6405632..dc28f9269 100644
--- a/src/fabric/test/fabric2_db_misc_tests.erl
+++ b/src/fabric/test/fabric2_db_misc_tests.erl
@@ -16,6 +16,7 @@
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch/include/couch_eunit.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include("fabric2.hrl").
 
 
 -define(TDEF(A), {atom_to_list(A), fun A/1}).
@@ -34,7 +35,8 @@ misc_test_() ->
                 fun set_revs_limit/1,
                 fun set_security/1,
                 fun is_system_db/1,
-                fun ensure_full_commit/1
+                fun ensure_full_commit/1,
+                fun metadata_bump/1
             ]}
         }
     }.
@@ -111,3 +113,24 @@ is_system_db({DbName, Db, _}) ->
 ensure_full_commit({_, Db, _}) ->
     ?assertEqual({ok, 0}, fabric2_db:ensure_full_commit(Db)),
     ?assertEqual({ok, 0}, fabric2_db:ensure_full_commit(Db, 5)).
+
+
+metadata_bump({DbName, _, _}) ->
+    % Call open again here to make sure we have a version in the cache
+    % as we'll be checking if that version gets its metadata bumped
+    {ok, Db} = fabric2_db:open(DbName, [{user_ctx, ?ADMIN_USER}]),
+
+    % Emulate a remote client bumping the metadataversion
+    {ok, Fdb} = application:get_env(fabric, db),
+    erlfdb:transactional(Fdb, fun(Tx) ->
+        erlfdb:set_versionstamped_value(Tx, ?METADATA_VERSION_KEY, <<0:112>>)
+    end),
+    NewMDVersion =  erlfdb:transactional(Fdb, fun(Tx) ->
+        erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY))
+    end),
+
+    % Perform a random operation which calls ensure_current
+    {ok, _} = fabric2_db:get_db_info(Db),
+
+    % Check that db handle in the cache got the new metadata version
+    ?assertMatch(#{md_version := NewMDVersion}, fabric2_server:fetch(DbName)).