Search the loader's mock cache .. (#354)

... in the correct direction - otherwise frequently updated objects may not be indexed correctly on reload.

src/leveled_bookie.erl

@@ -2281,7 +2281,7 @@ recalcfor_ledgercache(_InkTag,
         case check_in_ledgercache(LK, KeyH, LedgerCache, loader) of
             false ->
                 leveled_penciller:pcl_fetch(Penciller, LK, KeyH, true);
-            {value, KV} ->
+            KV ->
                 KV
         end,
     OldMetadata =
@@ -2341,7 +2341,7 @@ addto_ledgercache({H, SQN, KeyChanges}, Cache, loader) ->
                             leveled_codec:segment_hash(),
                             ledger_cache(),
                             loader) ->
-                                false | {value, leveled_codec:ledger_kv()}.
+                                false | leveled_codec:ledger_kv().
 %% @doc
 %% Check the ledger cache for a Key, when the ledger cache is in loader mode
 %% and so is populating a queue not an ETS table
@@ -2350,18 +2350,9 @@ check_in_ledgercache(PK, Hash, Cache, loader) ->
         [] ->
             false;
         _ ->
-            search(fun({K,_V}) -> K == PK end,
-                            lists:reverse(Cache#ledger_cache.load_queue))
+            lists:keyfind(PK, 1, Cache#ledger_cache.load_queue)
     end.
 
--spec search(fun((any()) -> boolean()), list()) -> {value, any()}|false.
-search(Pred, [Hd|Tail]) ->
-    case Pred(Hd) of
-        true -> {value, Hd};
-        false -> search(Pred, Tail)
-    end;
-search(Pred, []) when is_function(Pred, 1) ->
-    false.
 
 -spec maybepush_ledgercache(integer(), ledger_cache(), pid()) 
                                             -> {ok|returned, ledger_cache()}.
@@ -3302,10 +3293,6 @@ sqnorder_mutatefold_test() ->
 
     ok = book_destroy(Bookie1).
 
-search_test() ->
-    ?assertMatch({value, 5}, search(fun(X) -> X == 5 end, lists:seq(1, 10))),
-    ?assertMatch(false, search(fun(X) -> X == 55 end, lists:seq(1, 10))).
-
 check_notfound_test() ->
     ProbablyFun = fun() -> probably end,
     MissingFun = fun() -> missing end,

test/end_to_end/recovery_SUITE.erl

@@ -4,6 +4,7 @@
 -export([all/0]).
 -export([
             recovery_with_samekeyupdates/1,
+            same_key_rotation_withindexes/1,
             hot_backup_simple/1,
             hot_backup_changes/1,
             retain_strategy/1,
@@ -22,6 +23,7 @@
 
 all() -> [
             recovery_with_samekeyupdates,
+            same_key_rotation_withindexes,
             hot_backup_simple,
             hot_backup_changes,
             retain_strategy,
@@ -152,6 +154,80 @@ recovery_with_samekeyupdates(_Config) ->
     testutil:reset_filestructure(BackupPath),
     testutil:reset_filestructure().
 
+same_key_rotation_withindexes(_Config) ->
+    % If we have the same key - but the indexes change.  Do we consistently
+    % recalc the indexes correctly, even when the key exists multiple times
+    % in the loader's mock ledger cache
+    RootPath = testutil:reset_filestructure(),
+    BookOpts = [{root_path, RootPath},
+                    {cache_size, 2000},
+                    {max_journalsize, 20000000},
+                    {reload_strategy, [{?RIAK_TAG, recalc}]},
+                    {sync_strategy, testutil:sync_strategy()}],
+    {ok, Book1} = leveled_bookie:book_start(BookOpts),
+    IndexGenFun =
+        fun(ID) ->
+            fun() ->
+                [{add, list_to_binary("binary_bin"), <<ID:32/integer>>}]
+            end
+        end,
+
+    Bucket = <<"TestBucket">>,
+
+    ObjectGenFun =
+        fun(KeyID, IndexID) ->
+            Key = list_to_binary("Key" ++ integer_to_list(KeyID)),
+            Value = <<IndexID:32/integer>>,
+            GenRemoveFun = IndexGenFun(IndexID - 1),
+            testutil:set_object(Bucket,
+                                Key,
+                                Value,
+                                IndexGenFun(IndexID),
+                                GenRemoveFun())
+        end,
+
+    IdxCnt = 8,
+    KeyCnt = 50,
+
+    Sequence =
+        lists:map(fun(K) -> lists:map(fun(I) -> {K, I} end, lists:seq(1, IdxCnt)) end,
+                    lists:seq(1, KeyCnt)),
+    ObjList =
+        lists:map(fun({K, I}) -> ObjectGenFun(K, I) end, lists:flatten(Sequence)),
+
+    lists:foreach(
+            fun({Obj, SpcL}) -> testutil:book_riakput(Book1, Obj, SpcL) end,
+            ObjList),
+
+    FoldKeysFun = fun(_B, K, Acc) -> [K|Acc] end,
+    CheckFun =
+        fun(Bookie) ->
+            {async, R} =
+                leveled_bookie:book_indexfold(Bookie,
+                                                {Bucket, <<>>},
+                                                {FoldKeysFun, []},
+                                                {list_to_binary("binary_bin"),
+                                                    <<0:32/integer>>, 
+                                                    <<255:32/integer>>},
+                                                {true, undefined}),
+            QR = R(),
+            BadAnswers =
+                lists:filter(fun({I, _K}) -> I =/= <<IdxCnt:32/integer>> end, QR),
+            io:format("Results ~w BadAnswers ~w~n",
+                        [length(QR), length(BadAnswers)]),
+            true = length(QR) == KeyCnt,
+            true = [] == BadAnswers
+        end,
+
+    CheckFun(Book1),
+    ok = leveled_bookie:book_close(Book1),
+
+    {ok, Book2} = leveled_bookie:book_start(BookOpts),
+    CheckFun(Book2),
+    ok = leveled_bookie:book_close(Book2),
+
+    testutil:reset_filestructure().
+
 
 hot_backup_simple(_Config) ->
     % The journal may have a hot backup.  This allows for an online Bookie
@@ -235,6 +311,8 @@ hot_backup_changes(_Config) ->
 
     ok = testutil:check_indexed_objects(BookBackup, B, KSpcL2, V2),
 
+    ok = leveled_bookie:book_close(BookBackup),
+
     testutil:reset_filestructure("backup0"),
     testutil:reset_filestructure().