Clean up old expiry key on update insert

When an existing key is inserted with different timestamps, the primary
key is the same but the primary value is different from the existing one.
Currently, this results in a new expiry key being inserted, but the old
one is not deleted and lingers until it is removed by the inexorable
advance of time via the `remove_expired` server messages.

This checks whether there's already primary key for the inserted key,
and if so, cleans up the existing expiry key before proceeding with the
insert.

2 files changed, 44 insertions, 2 deletions

diff --git a/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl b/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl
index dc337665c..7c4ad8f6f 100644
--- a/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl
+++ b/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl
@@ -13,6 +13,7 @@
 -module(couch_expiring_cache_fdb).
 
 -export([
+    get_range_to/3,
     insert/6,
     lookup/3,
     clear_all/1,
@@ -40,6 +41,16 @@
 insert(#{jtx := true} = JTx, Name, Key, Val, StaleTS, ExpiresTS) ->
     #{tx := Tx, layer_prefix := LayerPrefix} = couch_jobs_fdb:get_jtx(JTx),
     PK = primary_key(Name, Key, LayerPrefix),
+    case get_val(Tx, PK) of
+        not_found ->
+            ok;
+        {_OldVal, _OldStaleTS, OldExpiresTS} ->
+            % Clean up current expiry key for this primary key. No
+            % need to clean up the existing primary key since it will
+            % be overwritten below.
+            OldXK = expiry_key(OldExpiresTS, Name, Key, LayerPrefix),
+            ok = erlfdb:clear(Tx, OldXK)
+    end,
     PV = erlfdb_tuple:pack({Val, StaleTS, ExpiresTS}),
     ok = erlfdb:set(Tx, PK, PV),
     XK = expiry_key(ExpiresTS, Name, Key, LayerPrefix),
@@ -87,6 +98,22 @@ clear_range_to(Name, EndTS, Limit) when Limit > 0 ->
         end, 0).
 
 
+-spec get_range_to(Name :: binary(), EndTS :: ?TIME_UNIT,
+    Limit :: non_neg_integer()) ->
+        [{Key :: binary(), Val :: binary()}].
+get_range_to(Name, EndTS, Limit) when Limit > 0 ->
+    fold_range(Name, EndTS, Limit,
+        fun(Tx, PK, _XK, Key, _ExpiresTS, Acc) ->
+            case get_val(Tx, PK) of
+                not_found ->
+                    couch_log:error("~p:entry missing Key: ~p", [?MODULE, Key]),
+                    Acc;
+                Val ->
+                    [{Key, Val} | Acc]
+            end
+        end, []).
+
+
 %% Private
 
 
diff --git a/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl b/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl
index aeb1df6f0..2e06fcc5a 100644
--- a/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl
+++ b/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl
@@ -66,7 +66,7 @@ teardown(#{pid := Pid}) ->
 
 
 simple_lifecycle(_) ->
-    ?_test(begin
+    {timeout, 10, ?_test(begin
         Now = erlang:system_time(?TIME_UNIT),
         StaleTS = Now + 100,
         ExpiresTS = Now + 200,
@@ -76,14 +76,29 @@ simple_lifecycle(_) ->
 
         ?assertEqual(ok, couch_expiring_cache_fdb:clear_all(Name)),
         ?assertEqual(not_found, couch_expiring_cache:lookup(Name, Key)),
+        ?assertEqual([], entries(Name)),
         ?assertEqual(ok,
             couch_expiring_cache:insert(Name, Key, Val, StaleTS, ExpiresTS)),
         ?assertEqual({fresh, Val}, couch_expiring_cache:lookup(Name, Key)),
         ok = wait_lookup(Name, Key, {stale, Val}),
+
+        % Refresh the existing key with updated timestamps
+        ?assertEqual(ok,
+            couch_expiring_cache:insert(Name, Key, Val,
+                StaleTS + 100, ExpiresTS + 100)),
+        ?assertEqual({fresh, Val}, couch_expiring_cache:lookup(Name, Key)),
+        ?assertEqual(1, length(entries(Name))),
+        ok = wait_lookup(Name, Key, {stale, Val}),
         ok = wait_lookup(Name, Key, expired),
         ok = wait_lookup(Name, Key, not_found),
+        ?assertEqual([], entries(Name)),
         ?assertEqual(not_found, couch_expiring_cache:lookup(Name, Key))
-    end).
+    end)}.
+
+
+entries(Name) ->
+    FarFuture = erlang:system_time(?TIME_UNIT) * 2,
+    couch_expiring_cache_fdb:get_range_to(Name, FarFuture, _Limit=100).
 
 
 wait_lookup(Name, Key, Expect) ->