Expiring cacheexpiring-cache

This is a library for creating an FDB backed key value cache, where
each entry has a `stale` and `expires` time associated with it. Once
the current time exceeds the `expires` time, the entry is automatically
removed. The `stale` time can be used to indicate that a refresh is
necessary, while still returning a non-expired value. It is potentially
useful for implementing e.g. caches to external systems of record, such
as OAuth 2.

9 files changed, 507 insertions, 0 deletions

diff --git a/rebar.config.script b/rebar.config.script
index 178cca8dd..29c3fde39 100644
--- a/rebar.config.script
+++ b/rebar.config.script
@@ -92,6 +92,7 @@ SubDirs = [
     "src/dreyfus",
     "src/fabric",
     "src/couch_jobs",
+    "src/couch_expiring_cache",
     "src/global_changes",
     "src/mango",
     "src/rexi",
diff --git a/src/couch_expiring_cache/README.md b/src/couch_expiring_cache/README.md
new file mode 100644
index 000000000..2ab1699db
--- /dev/null
+++ b/src/couch_expiring_cache/README.md
@@ -0,0 +1,71 @@
+# Couch Expiring Cache
+
+This is a library for creating an FDB backed key value cache, where
+each entry has a `stale` and `expires` time associated with it. Once
+the current time exceeds the `expires` time, the entry is
+automatically removed. The `stale` time can be used to indicate that a
+refresh is necessary, while still returning a non-expired value. It is
+potentially useful for implementing e.g. caches to external systems of
+record, such as OAuth 2.
+
+The data model is based on this [FDB forum discussion](
+https://forums.foundationdb.org/t/designing-key-value-expiration-in-fdb/156).
+
+```
+(?EXPIRING_CACHE, Name, ?PK, Key) := (Val, StaleTS, ExpireTS)
+(?EXPIRING_CACHE, Name, ?EXP, ExpireTS, Key) := ()
+```
+where `Name` is a unique namespace for a particular use case. N.B.
+that it's possible for cache data remain indefinitely in FDB when a
+`Name` is changed or retired with unexpired entries. For such cases,
+we provide `couch_expiring_cache_fdb:clear_all/1` to manually clean
+up those entries.
+
+## Example
+
+Typical usage for this library is to create a separate behaviour
+module for each `Name`, which internally starts a uniquely named
+`couch_expiring_cache_server` to handle expiration and removal of
+entries for that `Name`. For example, to cache authorization decisions
+from an external source, one could implement a module like the
+following:
+
+```erlang
+-module(auth_fdb_decision_cache).
+
+-behaviour(couch_expiring_cache_server).
+
+-export([
+    start_link/0
+]).
+
+
+-define(CACHE_NAME, <<"auth-decision">>).
+
+
+start_link() ->
+    Opts = #{
+        cache_name => ?CACHE_NAME,
+        period => 1000, % clear expired entries every second
+        batch_size => 500, % clear at most 500 entries each period
+        max_jitter => 10
+    },
+    couch_expiring_cache_server:start_link(?MODULE, Opts).
+```
+
+## Modules
+
+* `couch_expiring_cache`: The API module, it contains functions for
+  inserting and looking up cache entries, which are simply
+  pass-throughs to `couch_expiring_cache_fdb`.
+
+* `couch_expiring_cache_fdb`: The module which interacts with FDB, in
+  addition to insertion and lookup functions, it also contains a
+  function to clear an expired range, which is called periodically
+  from instances of `couch_expiring_cache_server`.
+
+* `couch_expiring_cache_server`: An "abstract" gen_server, a specific
+  behaviour of this module should be created for each `Name`, which
+  can override the default expiration parameters. It periodically
+  removes expired cache entries using configurable parameters for
+  period, jitter, and batch size.
diff --git a/src/couch_expiring_cache/include/couch_expiring_cache.hrl b/src/couch_expiring_cache/include/couch_expiring_cache.hrl
new file mode 100644
index 000000000..78e6a8552
--- /dev/null
+++ b/src/couch_expiring_cache/include/couch_expiring_cache.hrl
@@ -0,0 +1,17 @@
+% 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.
+
+-define(TIME_UNIT, millisecond).
+
+-type millisecond() :: non_neg_integer().
+
+-type jtx() :: map() | undefined | tuple(). % copied from couch_jobs.hrl
diff --git a/src/couch_expiring_cache/rebar.config b/src/couch_expiring_cache/rebar.config
new file mode 100644
index 000000000..362c8785e
--- /dev/null
+++ b/src/couch_expiring_cache/rebar.config
@@ -0,0 +1,14 @@
+% 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.
+
+{cover_enabled, true}.
+{cover_print_enabled, true}.
diff --git a/src/couch_expiring_cache/src/couch_expiring_cache.app.src b/src/couch_expiring_cache/src/couch_expiring_cache.app.src
new file mode 100644
index 000000000..27d58ee0e
--- /dev/null
+++ b/src/couch_expiring_cache/src/couch_expiring_cache.app.src
@@ -0,0 +1,27 @@
+% 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.
+
+{application, couch_expiring_cache, [
+    {description, "CouchDB Expiring Cache"},
+    {vsn, git},
+    {registered, []},
+    {applications, [
+        kernel,
+        stdlib,
+        erlfdb,
+        config,
+        couch_log,
+        couch_stats,
+        couch_jobs,
+        fabric
+    ]}
+]}.
diff --git a/src/couch_expiring_cache/src/couch_expiring_cache.erl b/src/couch_expiring_cache/src/couch_expiring_cache.erl
new file mode 100644
index 000000000..b26556e98
--- /dev/null
+++ b/src/couch_expiring_cache/src/couch_expiring_cache.erl
@@ -0,0 +1,56 @@
+% 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(couch_expiring_cache).
+
+-export([
+    insert/5,
+    insert/6,
+    lookup/2,
+    lookup/3
+]).
+
+
+-include_lib("couch_expiring_cache/include/couch_expiring_cache.hrl").
+
+
+-spec insert(Name :: binary(), Key :: binary(), Value :: binary(),
+    StaleTS :: ?TIME_UNIT(), ExpiresTS :: ?TIME_UNIT()) -> ok.
+insert(Name, Key, Value, StaleTS, ExpiresTS)
+        when is_binary(Name), is_binary(Key), is_binary(Value),
+        is_integer(StaleTS), is_integer(ExpiresTS) ->
+    insert(undefined, Name, Key, Value, StaleTS, ExpiresTS).
+
+
+-spec insert(Tx :: jtx(), Name :: binary(), Key :: binary(), Value :: binary(),
+    StaleTS :: ?TIME_UNIT(), ExpiresTS :: ?TIME_UNIT()) -> ok.
+insert(Tx, Name, Key, Value, StaleTS, ExpiresTS)
+        when is_binary(Name), is_binary(Key), is_binary(Value),
+        is_integer(StaleTS), is_integer(ExpiresTS) ->
+    couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+        couch_expiring_cache_fdb:insert(
+            JTx, Name, Key, Value, StaleTS, ExpiresTS)
+    end).
+
+
+-spec lookup(Name :: binary(), Key :: binary()) ->
+    not_found | {fresh, Val :: binary()} | {stale, Val :: binary()} | expired.
+lookup(Name, Key) when is_binary(Name), is_binary(Key) ->
+    lookup(undefined, Name, Key).
+
+
+-spec lookup(Tx :: jtx(), Name :: binary(), Key :: binary()) ->
+    not_found | {fresh, Val :: binary()} | {stale, Val :: binary()} | expired.
+lookup(Tx, Name, Key) when is_binary(Name), is_binary(Key) ->
+    couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+        couch_expiring_cache_fdb:lookup(JTx, Name, Key)
+    end).
diff --git a/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl b/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl
new file mode 100644
index 000000000..fa8508e14
--- /dev/null
+++ b/src/couch_expiring_cache/src/couch_expiring_cache_fdb.erl
@@ -0,0 +1,116 @@
+% 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(couch_expiring_cache_fdb).
+
+-export([
+    insert/6,
+    lookup/3,
+    clear_all/1,
+    clear_expired_range/3
+]).
+
+
+-define(EXPIRING_CACHE, 53). % coordinate with fabric2.hrl
+-define(PK, 1).
+-define(EXP, 2).
+
+
+-include_lib("couch_expiring_cache/include/couch_expiring_cache.hrl").
+
+
+% Data model
+% see: https://forums.foundationdb.org/t/designing-key-value-expiration-in-fdb/156
+%
+% (?EXPIRING_CACHE, Name, ?PK, Key) := (Val, StaleTS, ExpiresTS)
+% (?EXPIRING_CACHE, Name, ?EXP, ExpiresTS, Key) := ()
+
+
+-spec insert(JTx :: jtx(), Name :: binary(), Key :: binary(), Value :: binary(),
+    StaleTS :: ?TIME_UNIT, ExpiresTS :: ?TIME_UNIT) -> ok.
+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),
+    PV = erlfdb_tuple:pack({Val, StaleTS, ExpiresTS}),
+    XK = expiry_key(ExpiresTS, Name, Key, LayerPrefix),
+    XV = erlfdb_tuple:pack({}),
+    ok = erlfdb:set(Tx, PK, PV),
+    ok = erlfdb:set(Tx, XK, XV).
+
+
+-spec lookup(JTx :: jtx(), Name :: binary(), Key :: binary()) ->
+    not_found | {fresh, Val :: binary()} | {stale, Val :: binary()} | expired.
+lookup(#{jtx := true} = JTx, Name, Key) ->
+    #{tx := Tx, layer_prefix := LayerPrefix} = couch_jobs_fdb:get_jtx(JTx),
+    PK = primary_key(Name, Key, LayerPrefix),
+    case erlfdb:wait(erlfdb:get(Tx, PK)) of
+        not_found ->
+            not_found;
+        Bin when is_binary(Bin) ->
+            {Val, StaleTS, ExpiresTS} = erlfdb_tuple:unpack(Bin),
+            Now = erlang:system_time(?TIME_UNIT),
+            if
+                Now < StaleTS -> {fresh, Val};
+                Now < ExpiresTS -> {stale, Val};
+                true -> expired
+            end
+    end.
+
+
+-spec clear_all(Name :: binary()) ->
+    ok.
+clear_all(Name) ->
+    fabric2_fdb:transactional(fun(Tx) ->
+        LayerPrefix = fabric2_fdb:get_dir(Tx),
+        NamePrefix = erlfdb_tuple:pack({?EXPIRING_CACHE, Name}, LayerPrefix),
+        erlfdb:clear_range_startswith(Tx, NamePrefix)
+    end).
+
+
+-spec clear_expired_range(Name :: binary(), EndTS :: ?TIME_UNIT,
+    Limit :: non_neg_integer()) ->
+        OldestTS :: ?TIME_UNIT.
+clear_expired_range(Name, EndTS, Limit) when Limit > 0 ->
+    fabric2_fdb:transactional(fun(Tx) ->
+        LayerPrefix = fabric2_fdb:get_dir(Tx),
+        ExpiresPrefix = erlfdb_tuple:pack(
+            {?EXPIRING_CACHE, Name, ?EXP}, LayerPrefix),
+        fabric2_fdb:fold_range({tx, Tx}, ExpiresPrefix, fun({K, _V}, Acc) ->
+            Unpacked = erlfdb_tuple:unpack(K, ExpiresPrefix),
+            couch_log:debug("~p clearing ~p", [?MODULE, Unpacked]),
+            {ExpiresTS, Key} = Unpacked,
+            clear_expired(Tx, ExpiresTS, Name, Key, LayerPrefix),
+            oldest_ts(ExpiresTS, Acc)
+        end, 0, [{end_key, EndTS}, {limit, Limit}])
+    end).
+
+
+%% Private
+
+
+clear_expired(Tx, ExpiresTS, Name, Key, Prefix) ->
+    PK = primary_key(Name, Key, Prefix),
+    XK = expiry_key(ExpiresTS, Name, Key, Prefix),
+    ok = erlfdb:clear(Tx, PK),
+    ok = erlfdb:clear(Tx, XK).
+
+
+oldest_ts(TS, 0) -> TS; % handle initial Acc = 0 case
+oldest_ts(TS, OldestTS) -> min(TS, OldestTS).
+
+
+primary_key(Name, Key, Prefix) ->
+    erlfdb_tuple:pack({?EXPIRING_CACHE, Name, ?PK, Key}, Prefix).
+
+
+expiry_key(ExpiresTS, Name, Key, Prefix) ->
+    erlfdb_tuple:pack({?EXPIRING_CACHE, Name, ?EXP, ExpiresTS, Key}, Prefix).
diff --git a/src/couch_expiring_cache/src/couch_expiring_cache_server.erl b/src/couch_expiring_cache/src/couch_expiring_cache_server.erl
new file mode 100644
index 000000000..6f9dc1fd1
--- /dev/null
+++ b/src/couch_expiring_cache/src/couch_expiring_cache_server.erl
@@ -0,0 +1,110 @@
+% 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(couch_expiring_cache_server).
+
+-behaviour(gen_server).
+
+-callback start_link() -> {ok, pid()} | ignore | {error, term()}.
+
+-export([
+    start_link/2
+]).
+
+-export([
+    init/1,
+    terminate/2,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    code_change/3
+]).
+
+
+-define(DEFAULT_BATCH_SIZE, 1000).
+-define(DEFAULT_PERIOD_MSEC, 5000).
+-define(DEFAULT_MAX_JITTER_MSEC, 1000).
+
+
+-include_lib("couch_expiring_cache/include/couch_expiring_cache.hrl").
+
+
+start_link(Name, Opts) when is_atom(Name) ->
+    gen_server:start_link({local, Name}, ?MODULE, Opts#{name => Name}, []).
+
+
+init(Opts) ->
+    ?MODULE = ets:new(?MODULE, [named_table, public, {read_concurrency, true}]),
+    DefaultCacheName = atom_to_binary(maps:get(name, Opts), utf8),
+    Period = maps:get(period, Opts, ?DEFAULT_PERIOD_MSEC),
+    MaxJitter = maps:get(max_jitter, Opts, ?DEFAULT_MAX_JITTER_MSEC),
+    {ok, #{
+        cache_name => maps:get(cache_name, Opts, DefaultCacheName),
+        batch_size => maps:get(batch_size, Opts, ?DEFAULT_BATCH_SIZE),
+        period => Period,
+        max_jitter => MaxJitter,
+        timer_ref => schedule_remove_expired(Period, MaxJitter),
+        oldest_ts => 0,
+        elapsed => 0,
+        largest_elapsed => 0,
+        lag => 0}}.
+
+
+terminate(_, _) ->
+    ok.
+
+
+handle_call(Msg, _From, St) ->
+    {stop, {bad_call, Msg}, {bad_call, Msg}, St}.
+
+
+handle_cast(Msg, St) ->
+    {stop, {bad_cast, Msg}, St}.
+
+
+handle_info(remove_expired, St) ->
+    #{
+        cache_name := Name,
+        batch_size := BatchSize,
+        period := Period,
+        max_jitter := MaxJitter,
+        oldest_ts := OldestTS0,
+        largest_elapsed := LargestElapsed
+    } = St,
+
+    NowTS = erlang:system_time(?TIME_UNIT),
+    OldestTS = max(OldestTS0,
+        couch_expiring_cache_fdb:clear_expired_range(Name, NowTS, BatchSize)),
+    Elapsed = erlang:system_time(?TIME_UNIT) - NowTS,
+
+    {noreply, St#{
+        timer_ref := schedule_remove_expired(Period, MaxJitter),
+        oldest_ts := OldestTS,
+        elapsed := Elapsed,
+        largest_elapsed := max(Elapsed, LargestElapsed),
+        lag := NowTS - OldestTS}};
+
+handle_info(Msg, St) ->
+    {stop, {bad_info, Msg}, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+    {ok, St}.
+
+
+%% Private
+
+
+schedule_remove_expired(Timeout, MaxJitter) ->
+    Jitter = max(Timeout div 2, MaxJitter),
+    Wait = Timeout + rand:uniform(max(1, Jitter)),
+    erlang:send_after(Wait, self(), remove_expired).
diff --git a/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl b/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl
new file mode 100644
index 000000000..aeb1df6f0
--- /dev/null
+++ b/src/couch_expiring_cache/test/couch_expiring_cache_tests.erl
@@ -0,0 +1,95 @@
+% 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(couch_expiring_cache_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+-include_lib("couch_expiring_cache/include/couch_expiring_cache.hrl").
+
+
+-define(CACHE_NAME, <<"test">>).
+
+
+start_link() ->
+    Opts = #{
+        cache_name => ?CACHE_NAME,
+        period => 20,
+        max_jitter => 0
+    },
+    couch_expiring_cache_server:start_link(?MODULE, Opts).
+
+
+couch_expiring_cache_basic_test_() ->
+    {
+        "Test expiring cache basics",
+        {
+            setup,
+            fun setup_couch/0, fun teardown_couch/1,
+            {
+                foreach,
+                fun setup/0, fun teardown/1,
+                [
+                    fun simple_lifecycle/1
+                ]
+            }
+        }
+    }.
+
+
+setup_couch() ->
+    test_util:start_couch([fabric, couch_jobs]).
+
+
+teardown_couch(Ctx) ->
+    test_util:stop_couch(Ctx).
+
+
+setup() ->
+    {ok, Pid} = start_link(),
+    true = unlink(Pid),
+    #{pid => Pid}.
+
+
+teardown(#{pid := Pid}) ->
+    exit(Pid, kill).
+
+
+simple_lifecycle(_) ->
+    ?_test(begin
+        Now = erlang:system_time(?TIME_UNIT),
+        StaleTS = Now + 100,
+        ExpiresTS = Now + 200,
+        Name = ?CACHE_NAME,
+        Key = <<"key">>,
+        Val = <<"val">>,
+
+        ?assertEqual(ok, couch_expiring_cache_fdb:clear_all(Name)),
+        ?assertEqual(not_found, couch_expiring_cache:lookup(Name, Key)),
+        ?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}),
+        ok = wait_lookup(Name, Key, expired),
+        ok = wait_lookup(Name, Key, not_found),
+        ?assertEqual(not_found, couch_expiring_cache:lookup(Name, Key))
+    end).
+
+
+wait_lookup(Name, Key, Expect) ->
+    test_util:wait(fun() ->
+        case couch_expiring_cache:lookup(Name, Key) of
+            Expect -> ok;
+            _ -> wait
+        end
+    end, _Timeout = 1000, _PollingInterval = 10).