Merge pull request #220 from martinsumner/mas-i157-indexttl

Add test of index expiry

test/end_to_end/iterator_SUITE.erl

@@ -6,7 +6,8 @@
 -define(KEY_ONLY, {false, undefined}).
 
 -export([all/0]).
--export([breaking_folds/1,
+-export([expiring_indexes/1,
+            breaking_folds/1,
             single_object_with2i/1,
             small_load_with2i/1,
             query_count/1,
@@ -15,6 +16,7 @@
             rotating_objects/1]).
 
 all() -> [
+            expiring_indexes,
             breaking_folds,
             single_object_with2i,
             small_load_with2i,
@@ -25,6 +27,113 @@ all() -> [
             ].
 
 
+expiring_indexes(_Config) ->
+    % Add objects to ths tore with index entries, where the objects (and hence
+    % the indexes have an expiry time.  Confirm that the indexes and the
+    % objects are no longer present after the expiry time (and are present
+    % before).  Confirm that replacing an object has the expected outcome, if
+    % the IndexSpecs are updated as part of the request.
+    KeyCount = 50000,
+    Future = 120,
+        % 2 minutes - if running tests on a slow machine, may need to increase
+        % this value
+    RootPath = testutil:reset_filestructure(),
+    StartOpts1 = [{root_path, RootPath},
+                    {max_journalsize, 100000000},
+                    {sync_strategy, testutil:sync_strategy()}],
+    {ok, Bookie1} = leveled_bookie:book_start(StartOpts1),
+    
+    SW1 = os:timestamp(),
+    IBKL1 = testutil:stdload_expiring(Bookie1, KeyCount, Future),
+    LoadTime = timer:now_diff(os:timestamp(), SW1)/1000000,
+    io:format("Load of ~w std objects in ~w seconds~n",  [KeyCount, LoadTime]),
+
+
+    FilterFun = fun({I, _B, _K}) -> lists:member(I, [5, 6, 7, 8]) end,
+    LoadedEntriesInRange = lists:sort(lists:filter(FilterFun, IBKL1)),
+
+    true = LoadTime < (Future - 30), 
+        % need 30 seconds spare to run query
+        % and add replacements
+
+    {I0, B0, K0} = hd(IBKL1),
+    false = FilterFun(hd(IBKL1)), 
+        % The head entry should not have index between 5 and 8
+
+    CountI0Fold =
+        fun() ->
+            leveled_bookie:book_indexfold(Bookie1,
+                                            B0,
+                                            {fun(_BF, _KT, Acc) -> Acc + 1 end,
+                                                0},
+                                            {<<"temp_int">>, I0, I0},
+                                            {true, undefined})
+        end,
+    {async, I0Counter1} = CountI0Fold(),
+    I0Count1 = I0Counter1(),
+
+    FoldFun = fun(BF, {IdxV, KeyF}, Acc) -> [{IdxV, BF, KeyF}|Acc] end,
+    InitAcc = [],
+    IndexFold = 
+        fun() ->
+            leveled_bookie:book_indexfold(Bookie1,
+                                            B0,
+                                            {FoldFun, InitAcc},
+                                            {<<"temp_int">>, 5, 8},
+                                            {true, undefined})
+        end,
+
+    {async, Folder1} = IndexFold(),
+    QR1 = Folder1(),
+    true = lists:sort(QR1) == LoadedEntriesInRange,
+    % Replace object with one with an index value of 6
+    testutil:stdload_object(Bookie1, B0, K0, 6, <<"value">>,
+                            leveled_util:integer_now() + 600),
+    % Confirm that this has reduced the index entries in I0 by 1
+    {async, I0Counter2} = CountI0Fold(),
+    I0Count2 = I0Counter2(),
+    io:format("Count with index value ~w changed from ~w to ~w~n",
+                [I0, I0Count1, I0Count2]),
+    true = I0Count2 == (I0Count1 - 1),
+    % Now replace again, shortening the timeout to 15s,
+    % this time index value of 6
+    testutil:stdload_object(Bookie1, B0, K0, 5, <<"value">>,
+                            leveled_util:integer_now() + 15),
+    {async, Folder2} = IndexFold(),
+        leveled_bookie:book_indexfold(Bookie1,
+                                        B0,
+                                        {FoldFun, InitAcc},
+                                        {<<"temp_int">>, 5, 8},
+                                        {true, undefined}),
+    QR2 = Folder2(),
+    io:format("Query with additional entry length ~w~n", [length(QR2)]),
+    true = lists:sort(QR2) == lists:sort([{5, B0, K0}|LoadedEntriesInRange]),
+    % Wait for a 15s timeout + lus a second to be sure
+    timer:sleep(15000 + 1000),
+    {async, Folder3} = IndexFold(),
+    QR3 = Folder3(),
+    % Now the entry should be missing (and the 600s TTL entry should not have
+    % resurfaced)
+    io:format("Query results length ~w following sleep~n", [length(QR3)]),
+    true = lists:sort(QR3) == LoadedEntriesInRange,
+
+    FoldTime = timer:now_diff(os:timestamp(), SW1)/1000000 - LoadTime,
+    io:format("Query returned ~w entries in ~w seconds - 3 queries + 15s wait~n",
+                [length(QR1), FoldTime]),
+    true = (LoadTime + FoldTime) < Future,
+    SleepTime = round((Future - (LoadTime + FoldTime)) * 1000 + 1000), % add a second
+    io:format("Sleeping ~w s for all to expire~n", [SleepTime/1000]),
+    timer:sleep(SleepTime),
+
+    % Index entries should now have expired
+    {async, Folder4} = IndexFold(),
+    QR4 = Folder4(),
+    true = QR4 == [],
+
+    ok = leveled_bookie:book_close(Bookie1),
+    testutil:reset_filestructure().
+
+
 breaking_folds(_Config) ->
     % Run various iterators and show that they can be broken by throwing an
     % exception from within the fold

test/end_to_end/testutil.erl

@@ -8,6 +8,8 @@
             book_riakhead/3,
             riakload/2,
             stdload/2,
+            stdload_expiring/3,
+            stdload_object/6,
             reset_filestructure/0,
             reset_filestructure/1,
             check_bucket_stats/2,
@@ -209,6 +211,43 @@ stdload(Bookie, Count, Acc) ->
     end,
     stdload(Bookie, Count - 1, [{B, K, erlang:phash2(V)}|Acc]).
 
+stdload_expiring(Book, KeyCount, When) ->
+    % Adds KeyCount object that will expire When seconds in the future.
+    % Each object will have a single entry on the <<"temp_int">> index.
+    ExpiryTime = leveled_util:integer_now() + When,
+    V = get_compressiblevalue(),
+    stdload_expiring(Book, KeyCount, ExpiryTime, V, []).
+
+stdload_expiring(_Book, 0, _TLL, _V, Acc) ->
+    lists:sort(Acc);
+stdload_expiring(Book, KeyCount, TTL, V, Acc) ->
+    B = <<"Bucket">>,
+    K = list_to_binary(leveled_util:generate_uuid()),
+    I = KeyCount rem 1000,
+    stdload_object(Book, B, K, I, V, TTL),
+    stdload_expiring(Book, KeyCount - 1, TTL, V, [{I, B, K}|Acc]).
+
+stdload_object(Book, B, K, I, V, TTL) ->
+    Obj = [{index, I}, {value, V}],
+    IdxSpecs = 
+        case leveled_bookie:book_get(Book, B, K) of
+            {ok, PrevObj} ->
+                {index, OldI} = lists:keyfind(index, 1, PrevObj),
+                io:format("Remove index ~w for ~w~n", [OldI, I]),
+                [{remove, <<"temp_int">>, OldI}, {add, <<"temp_int">>, I}];
+            not_found ->
+                [{add, <<"temp_int">>, I}]
+        end,
+    R = leveled_bookie:book_tempput(Book, B, K, Obj, IdxSpecs, ?STD_TAG, TTL),
+    case R of
+        ok -> 
+            ok;
+        pause -> 
+            io:format("Slow offer needed~n"),
+            timer:sleep(?SLOWOFFER_DELAY)
+    end.
+
+
 
 reset_filestructure() ->
     reset_filestructure(0, ?ROOT_PATH).