Develop 3.0 -> 3.1 merge

.gitignore

@@ -25,3 +25,4 @@ src/riak_core_pb.erl
 *@*
 undefined
 log/crash.log
+.eqc/*counterexample.eqc

.travis.yml

@@ -0,0 +1,11 @@
+language: erlang
+otp_release:
+  - 20.3.8
+  - 21.3
+  - 22.3
+script:
+  - chmod u+x rebar3
+  - ./rebar3 do upgrade, compile, dialyzer, xref, eunit
+notifications:
+  slack:
+    secure: SE0EMU9HenZlLBuNg7l6WLMxJPkfyAEGgodvAqMEuQmICtrh1nV019D/A8ykejYYPPsJafWVOfypOSDrSHCndzXvEZvU8l45nJ6XLdUdrDYEmvcSqN3EqmVSsuf9H3g99bvKygXaY27MkTS5ixLil7PzybG+YpwMnQGcQxYo5Eg=

docs/ReapErase.md

@@ -0,0 +1,75 @@
+# Reap and Erase
+
+## Background
+
+The Reaper and Eraser are now processes supported in Riak as at release 2.9.1.  They are intended to be a starting point of improving a number of situations:
+
+* Riak supports [multiple delete modes](https://docs.riak.com/riak/kv/2.2.3/using/reference/object-deletion/index.html#configuring-object-deletion), and the recommended mode for safety is `keep`.  However, a delete_mode of `keep` creates an unresolved problem of permanently uncollected garbage - the tombstones are never erased.  Note, even with an `interval` delete mode, tombstones may fail to be cleared and then continue to exist forever.
+
+* Riak supports time-to-live for objects configured within the backend, also known as [Global Object Expiration](https://riak.com/products/riak-kv/global-object-expiration/index.html?p=12768.html).  However, there are two flaws with this form of automated expiration:
+
+ * It considers only when the object is added to the backend, not when the object is added to the database (i.e. the object's last modified time).  An object may extend and survive expiry through handoffs.
+
+ * Object expiry is not coordinated with anti-entropy changes, and so as objects expire they may be resurrected by anti-entropy, which in turn will reset their expiry data to further in the future.  It can also lead to huge cascades in false repair action when AAE trees are rebuilt - as one vnode's AAE trees suddenly rebuild without all the expired objects.  This has led to some customers attempting to coordinate AAE tree rebuilds to make the occur concurrently, which can have a significant performance impact during the rebuild period.
+
+To begin to address these problems, the `riak_kv_reaper` and `riak_kv_eraser` have been introduced in Riak KV 2.9.1.
+
+## Riak KV Reaper
+
+The `riak_kv_reaper` is a process that receives requests to reap tombstones, queues those requests, and continuously reaps tombstones from that queue when processing time is available.  
+
+The reaper queue can be fed via a tictac aae fold from the riak client - `aae_reap_tombs`:
+
+```
+-spec aae_reap_tombs(pid(),
+                    riakc_obj:bucket(), key_range(),
+                    segment_filter(),
+                    modified_range(),
+                    change_method()) ->
+                        {ok, non_neg_integer()} | {error, any()}.
+```
+
+Reaping tombstones can be done only against a single specific bucket at a time, and can be further restricted to a key range within the bucket.  A segment filter can be added to only reap tombstones within a given part of the AAE tree; the segment filter may be useful when trying to break up reaping activity to do only a proportion of required reaps at a time.  A modified range should be passed so that the reap can be restricted only to tombstones which have existed beyond a certain point - for example to only reap tombstones more than one month old.
+
+The reap fold will then discover keys to be reaped, and queue them for reaping (or count them if `change_method` is set to count).  To run reaps in parallel across nodes in the cluster use `local` as the `change_method`.  To have a single queue of reaps for a single process dedicated to this fold then `{job, JobID}` can be passed as the `change_method`.
+
+The actual reap will remove both the tombstone from the backend as well as removing the reference from the Active Anti-Entropy system.  Before attempting a reap a check is made to ensure all primary vnodes in the preflist are online - and if not the reap will be deferred by switching it to the back of the queue.  If a reap were to proceed without a primary being available, then it is likely to be eventually resurrected through anti-entropy.
+
+The reaping itself will only act if:
+
+* the object to be reaped is confirmed as a tombstone, and;
+
+* the object to be reaped has the same vector clock as when the reap requirement was discovered (the comparison is based on a hash of the sorted vector clock).
+
+Note that when using `riak_kv_ttaaefs_manager` for full-sync, or any riak_repl full-sync mechanism, that is reap jobs are not co-ordinated between clusters tombstones will be resurrected by full-sync jobs.
+
+## Riak KV Eraser
+
+The `riak_kv_eraser` is a process that receives requests to delete objects, queues those requests, and continuously delete objects from that queue when processing time is available.  The eraser is simply an unscheduled garbage collection process at the moment, but is planned to be extended in 2.9.2 to be part of a more complete TTL management solution.
+
+The eraser queue can be fed via a tictac aae fold from the riak client - `aae_erase_keys`:
+
+```
+-spec aae_erase_keys(pid(),
+                    riakc_obj:bucket(), key_range(),
+                    segment_filter(),
+                    modified_range(),
+                    change_method()) ->
+                        {ok, non_neg_integer()} | {error, any()}.
+```
+
+The function inputs are the same as with `aae_reap_tombs`.  For this fold, the results will be queued for the `riak_kv_eraser`.  If all primary vnodes are not up, then as with the reap the delete will not be attempted, but will be re-queued.
+
+The delete will only complete if the object to be deleted has a vector clock equal to that discovered at the time the delete was queued.
+
+
+## Outstanding TODO
+
+* Allow for scheduled reap and erase jobs to generate reap and erase activity.
+
+ * This makes more sense for erase jobs to have them auto-scheduled, as reap jobs won't naturally co-ordinate between clusters - so tombstones may resurrect through full-sync.
+
+
+* Change replication so that it will filter the sending of tombstones beyond a certain modified time, so as not to resurrect old tombstones via full-sync.
+
+* Have a TTL bucket property so that GETs can be filtered beyond that modified time.  Need to consider local GETs (e.g. at a vnode before a PUT).

eqc/backend_eqc.erl

@@ -62,6 +62,8 @@
          async_put/5,
          init_backend/3]).
 
+-export([fold_keys/5, check/1, check2/1]).
+
 -define(TEST_SECONDS, 120).
 
 -record(qcst, {backend, % Backend module under test

eqc/ec_eqc.erl

@@ -670,11 +670,15 @@ get_fsm_proc(ReqId, #params{n = N, r = R}) ->
     NotFoundOk = true,
     AllowMult = true,
     DeletedVclock = true,
+    ExpectedVclock = false,
+    NodeConfirms = 0,
     GetCore = riak_kv_get_core:init(N, R,
                                     0, %% SLF hack
                                     FailThreshold,
                                     NotFoundOk, AllowMult, DeletedVclock,
-                                    [{Idx, primary} || Idx <- lists:seq(1, N)] %% SLF hack
+                                    [{Idx, primary} || Idx <- lists:seq(1, N)], %% SLF hack
+                                    ExpectedVclock,
+                                    NodeConfirms
                                    ),
     #proc{name = {get_fsm, ReqId}, handler = get_fsm,
           procst = #getfsmst{getcore = GetCore}}.
@@ -689,7 +693,7 @@ get_fsm(#msg{from = {kv_vnode, Idx, _}, c = {r, Result, Idx, _ReqId}},
                                               reply_to = ReplyTo,
                                               responded = Responded,
                                               getcore = GetCore} = ProcSt} = P) ->
-    UpdGetCore1 = riak_kv_get_core:add_result(Idx, Result, GetCore),
+    UpdGetCore1 = riak_kv_get_core:add_result(Idx, Result, node(), GetCore),
     {ReplyMsgs, UpdGetCore3, UpdResponded} =
         case riak_kv_get_core:enough(UpdGetCore1) of
             true when Responded == false ->

eqc/eraser_eqc.erl

@@ -0,0 +1,90 @@
+-module(eraser_eqc).
+
+-include_lib("eqc/include/eqc.hrl").
+-include_lib("eqc/include/eqc_component.hrl").
+
+-compile([export_all, nowarn_export_all]).
+
+%% -- State ------------------------------------------------------------------
+initial_state() ->
+  #{ deletes => [] }.
+
+%% -- Operations -------------------------------------------------------------
+
+%% --- Operation: start ---
+start_pre(S) -> not maps:is_key(pid, S).
+
+start_args(_S) ->
+  [gen_delete_mode()].
+
+start(DelMode) ->
+  {ok, Pid} = gen_server:start_link(riak_kv_eraser, [eqc_job, fun erase/2, DelMode], []),
+  Pid.
+
+start_next(S, Pid, [DelMode]) ->
+  S#{ delete_mode => DelMode, pid => Pid }.
+
+%% --- Operation: delete ---
+delete_pre(S) -> maps:is_key(pid, S).
+
+delete_args(#{ pid := Pid }) ->
+  [Pid, gen_delete_ref()].
+
+delete(Pid, {Ref, Retries}) ->
+  ets:insert(?MODULE, {Ref, Retries}),
+  riak_kv_eraser:request_delete(Pid, Ref).
+
+delete_next(S = #{ deletes := Ds }, _Value, [_Pid, {Ref, Retries}]) ->
+  S#{ deletes := [{Ref, Retries} | Ds] }.
+
+delete_features(_, [_, {_, N}], _) ->
+  [{retries, N}].
+
+%% -- Generators -------------------------------------------------------------
+gen_delete_mode() ->
+  elements([keep, immediate]).
+
+gen_delete_ref() ->
+  ?LET(Ref, gen_reference(),
+    weighted_default({4, {Ref, 1}}, {1, {Ref, choose(2, 5)}})).
+
+gen_reference() ->
+  os:timestamp().
+
+%% -- Properties -------------------------------------------------------------
+
+prop_eraser() ->
+  application:set_env(riak_kv, eraser_redo_timeout, 20),
+  ?FORALL(Cmds, commands(?MODULE),
+  begin
+    ets:new(?MODULE, [named_table, public]),
+    {H, S, Res} = run_commands(Cmds),
+    StopRes = stop_job(maps:get(pid, S, undefined)),
+    ETSRes = ets:tab2list(?MODULE),
+    ets:delete(?MODULE),
+    aggregate(call_features(H),
+      pretty_commands(?MODULE, Cmds, {H, S, Res},
+        conjunction([
+          {result, equals(Res, ok)},
+          {stop,   equals(StopRes, ok)},
+          {ets,    equals(ETSRes, [])}])))
+  end).
+
+stop_job(undefined) -> ok;
+stop_job(Pid) ->
+  MonRef = monitor(process, Pid),
+  riak_kv_eraser:stop_job(Pid),
+  receive
+    {'DOWN', MonRef, _, _, _} -> ok
+  after 1000 ->
+    timeout_stop_job
+  end.
+
+
+erase(Ref, _) ->
+  case ets:lookup(?MODULE, Ref) of
+    [{_, 1}] ->
+      ets:delete(?MODULE, Ref), true;
+    [{_, N}] ->
+      ets:insert(?MODULE, {Ref, N - 1}), false
+  end.

eqc/fsm_eqc_vnode.erl

@@ -27,9 +27,15 @@
 %% -------------------------------------------------------------------
 
 -module(fsm_eqc_vnode).
--behaviour(gen_fsm_compat).
+-behaviour(gen_fsm).
 -include("include/riak_kv_vnode.hrl").
 
+-compile({nowarn_deprecated_function, 
+            [{gen_fsm, start_link, 3},
+                {gen_fsm, start_link, 4},
+                {gen_fsm, sync_send_all_state_event, 2}]}).
+
+
 -export([start_link/0, start_link/1, set_data/2, set_vput_replies/1, 
          get_history/0, get_put_history/0,
          get_reply_history/0, log_postcommit/1, get_postcommits/0]).
@@ -66,30 +72,30 @@ start_link() ->
     start_link(?MODULE).
 
 start_link(undefined) ->
-    gen_fsm_compat:start_link(?MODULE, [], []);
+    gen_fsm:start_link(?MODULE, [], []);
 start_link(RegName) ->
-    gen_fsm_compat:start_link({local, RegName}, ?MODULE, [], []).
+    gen_fsm:start_link({local, RegName}, ?MODULE, [], []).
 
 set_data(Objs, PartVals) ->
-    ok = gen_fsm_compat:sync_send_all_state_event(?MODULE, {set_data, Objs, PartVals}).
+    ok = gen_fsm:sync_send_all_state_event(?MODULE, {set_data, Objs, PartVals}).
 
 set_vput_replies(VPutReplies) ->
-    ok = gen_fsm_compat:sync_send_all_state_event(?MODULE, {set_vput_replies, VPutReplies}).
+    ok = gen_fsm:sync_send_all_state_event(?MODULE, {set_vput_replies, VPutReplies}).
 
 get_history() ->
-    gen_fsm_compat:sync_send_all_state_event(?MODULE, get_history).
+    gen_fsm:sync_send_all_state_event(?MODULE, get_history).
 
 get_put_history() ->
-    gen_fsm_compat:sync_send_all_state_event(?MODULE, get_put_history).
+    gen_fsm:sync_send_all_state_event(?MODULE, get_put_history).
 
 get_reply_history() ->
-    gen_fsm_compat:sync_send_all_state_event(?MODULE, get_reply_history).
+    gen_fsm:sync_send_all_state_event(?MODULE, get_reply_history).
 
 log_postcommit(Obj) ->
-    gen_fsm_compat:sync_send_all_state_event(?MODULE, {log_postcommit, Obj}).
+    gen_fsm:sync_send_all_state_event(?MODULE, {log_postcommit, Obj}).
 
 get_postcommits() ->
-    gen_fsm_compat:sync_send_all_state_event(?MODULE, get_postcommits).
+    gen_fsm:sync_send_all_state_event(?MODULE, get_postcommits).
 
 %% ====================================================================
 %% gen_fsm callbacks

eqc/reaper_eqc.erl

@@ -0,0 +1,87 @@
+-module(reaper_eqc).
+
+-include_lib("eqc/include/eqc.hrl").
+-include_lib("eqc/include/eqc_component.hrl").
+
+-compile([export_all, nowarn_export_all]).
+
+%% -- State ------------------------------------------------------------------
+initial_state() ->
+  #{ reaps => [] }.
+
+%% -- Operations -------------------------------------------------------------
+
+%% --- Operation: start ---
+start_pre(S) -> not maps:is_key(pid, S).
+
+start_args(_S) ->
+  [].
+
+start() ->
+  {ok, Pid} = gen_server:start_link(riak_kv_reaper, [eqc_job, fun reaper/1], []),
+  Pid.
+
+start_next(S, Pid, []) ->
+  S#{ pid => Pid }.
+
+%% --- Operation: reap ---
+reap_pre(S) -> maps:is_key(pid, S).
+
+reap_args(#{ pid := Pid }) ->
+  [Pid, gen_reap_ref()].
+
+reap(Pid, {Ref, Retries}) ->
+  ets:insert(?MODULE, {Ref, Retries}),
+  riak_kv_reaper:request_reap(Pid, Ref).
+
+reap_next(S = #{ reaps := Ds }, _Value, [_Pid, {Ref, Retries}]) ->
+  S#{ reaps := [{Ref, Retries} | Ds] }.
+
+reap_features(_, [_, {_, N}], _) ->
+  [{retries, N}].
+
+%% -- Generators -------------------------------------------------------------
+gen_reap_ref() ->
+  ?LET(Ref, gen_reference(),
+    weighted_default({4, {Ref, 1}}, {1, {Ref, choose(2, 5)}})).
+
+gen_reference() ->
+  os:timestamp().
+
+%% -- Properties -------------------------------------------------------------
+
+prop_reaper() ->
+  application:set_env(riak_kv, reaper_redo_timeout, 20),
+  ?FORALL(Cmds, commands(?MODULE),
+  begin
+    ets:new(?MODULE, [named_table, public]),
+    {H, S, Res} = run_commands(Cmds),
+    StopRes = stop_job(maps:get(pid, S, undefined)),
+    ETSRes = ets:tab2list(?MODULE),
+    ets:delete(?MODULE),
+    aggregate(call_features(H),
+      pretty_commands(?MODULE, Cmds, {H, S, Res},
+        conjunction([
+          {result, equals(Res, ok)},
+          {stop,   equals(StopRes, ok)},
+          {ets,    equals(ETSRes, [])}])))
+  end).
+
+stop_job(undefined) -> ok;
+stop_job(Pid) ->
+  MonRef = monitor(process, Pid),
+  riak_kv_reaper:stop_job(Pid),
+  receive
+    {'DOWN', MonRef, _, _, _} -> ok
+  after 1000 ->
+    timeout_stop_job
+  end.
+
+
+reaper(Ref) ->
+  case ets:lookup(?MODULE, Ref) of
+    [{_, 1}] ->
+      ets:delete(?MODULE, Ref), true;
+    [{_, N}] ->
+      ets:insert(?MODULE, {Ref, N - 1}), false
+  end.