Prevent eviction of newer handles from fabric_server cache

Check metadata versions to ensure newer handles are not clobbered. The same
thing is done for removal, `maybe_remove/1` removes handle only if there isn't
a newer handle already there.

2 files changed, 70 insertions, 9 deletions

diff --git a/src/fabric/src/fabric2_server.erl b/src/fabric/src/fabric2_server.erl
index 957efff3b..b557da8c7 100644
--- a/src/fabric/src/fabric2_server.erl
+++ b/src/fabric/src/fabric2_server.erl
@@ -17,9 +17,15 @@
 
 -export([
     start_link/0,
+
     fetch/2,
+
     store/1,
+    maybe_update/1,
+
     remove/1,
+    maybe_remove/1,
+
     fdb_directory/0,
     fdb_cluster/0
 ]).
@@ -66,27 +72,57 @@ start_link() ->
 fetch(DbName, UUID) when is_binary(DbName) ->
     case {UUID, ets:lookup(?MODULE, DbName)} of
         {_, []} -> undefined;
-        {undefined, [{DbName, #{} = Db}]} -> Db;
-        {<<_/binary>>, [{DbName, #{uuid := UUID} = Db}]} -> Db;
-        {<<_/binary>>, [{DbName, #{} = _Db}]} -> undefined
+        {undefined, [{DbName, _UUID, _, #{} = Db}]} -> Db;
+        {<<_/binary>>, [{DbName, UUID, _, #{} = Db}]} -> Db;
+        {<<_/binary>>, [{DbName, _UUID, _, #{} = _Db}]} -> undefined
     end.
 
 
 store(#{name := DbName} = Db0) when is_binary(DbName) ->
-    Db1 = Db0#{
-        tx := undefined,
-        user_ctx := #user_ctx{},
-        security_fun := undefined
-    },
-    true = ets:insert(?MODULE, {DbName, Db1}),
+    #{
+        uuid := UUID,
+        md_version := MDVer
+    } = Db0,
+    Db1 = sanitize(Db0),
+    case ets:insert_new(?MODULE, {DbName, UUID, MDVer, Db1}) of
+        true -> ok;
+        false -> maybe_update(Db1)
+    end,
     ok.
 
 
+maybe_update(#{name := DbName} = Db0) when is_binary(DbName) ->
+    #{
+        uuid := UUID,
+        md_version := MDVer
+    } = Db0,
+    Db1 = sanitize(Db0),
+    Head = {DbName, UUID,  '$1', '_'},
+    Guard = {'=<', '$1', MDVer},
+    Body = {DbName, UUID, MDVer, {const, Db1}},
+    try
+        1 =:= ets:select_replace(?MODULE, [{Head, [Guard], [{Body}]}])
+    catch
+        error:badarg ->
+            false
+    end.
+
+
 remove(DbName) when is_binary(DbName) ->
     true = ets:delete(?MODULE, DbName),
     ok.
 
 
+maybe_remove(#{name := DbName} = Db) when is_binary(DbName) ->
+    #{
+        uuid := UUID,
+        md_version := MDVer
+    } = Db,
+    Head = {DbName, UUID, '$1', '_'},
+    Guard = {'=<', '$1', MDVer},
+    1 =:= ets:select_delete(?MODULE, [{Head, [Guard], [true]}]).
+
+
 init(_) ->
     ets:new(?MODULE, [
             public,
@@ -229,3 +265,11 @@ set_option(Db, Option, Val) ->
             Msg = "~p : Could not set fdb tx option ~p = ~p",
             couch_log:error(Msg, [?MODULE, Option, Val])
     end.
+
+
+sanitize(#{} = Db) ->
+    Db#{
+        tx := undefined,
+        user_ctx := #user_ctx{},
+        security_fun := undefined
+    }.
diff --git a/src/fabric/test/fabric2_db_misc_tests.erl b/src/fabric/test/fabric2_db_misc_tests.erl
index 9c95ca565..2fad73f1f 100644
--- a/src/fabric/test/fabric2_db_misc_tests.erl
+++ b/src/fabric/test/fabric2_db_misc_tests.erl
@@ -50,6 +50,7 @@ misc_test_() ->
                 ?TDEF(ensure_full_commit),
                 ?TDEF(metadata_bump),
                 ?TDEF(db_version_bump),
+                ?TDEF(db_cache_doesnt_evict_newer_handles),
                 ?TDEF(events_listener)
             ])
         }
@@ -375,6 +376,22 @@ db_version_bump({DbName, _, _}) ->
     ?assertMatch(#{db_version := NewDbVersion}, Db2).
 
 
+db_cache_doesnt_evict_newer_handles({DbName, _, _}) ->
+    {ok, Db} = fabric2_db:open(DbName, [{user_ctx, ?ADMIN_USER}]),
+    CachedDb = fabric2_server:fetch(DbName, undefined),
+
+    StaleDb = Db#{md_version := <<0>>},
+
+    ok = fabric2_server:store(StaleDb),
+    ?assertEqual(CachedDb, fabric2_server:fetch(DbName, undefined)),
+
+    ?assert(not fabric2_server:maybe_update(StaleDb)),
+    ?assertEqual(CachedDb, fabric2_server:fetch(DbName, undefined)),
+
+    ?assert(not fabric2_server:maybe_remove(StaleDb)),
+    ?assertEqual(CachedDb, fabric2_server:fetch(DbName, undefined)).
+
+
 events_listener({DbName, Db, _}) ->
     Opts = [
         {dbname, DbName},