Expanded AAE tests to include busted hashtable

Busted the hashtable in a Journal file, and demonstrated it can be fixed
by changing the extension name (no need to recover from backup if only
the hashtable is bust)

src/leveled_inker.erl

@@ -546,8 +546,9 @@ open_all_manifest(Man0, RootPath, CDBOpts) ->
                                     Pid} = leveled_cdb:cdb_open_reader(CFN),
                                 {LowSQN, FN, Pid};
                             false ->
-                                {ok,
+                                W = leveled_cdb:cdb_open_writer(PFN, CDBOpts),
-                                    Pid} = leveled_cdb:cdb_open_reader(PFN),
+                                {ok, Pid} = W,
+                                ok = leveled_cdb:cdb_roll(Pid),
                                 {LowSQN, FN, Pid}
                         end;
                     _ ->

test/end_to_end/recovery_SUITE.erl

@@ -8,9 +8,9 @@
             ]).
 
 all() -> [
-            retain_strategy,
+            % retain_strategy,
-            aae_bustedjournal,
+            aae_bustedjournal %,
-            journal_compaction_bustedjournal
+            % journal_compaction_bustedjournal
             ].
 
 retain_strategy(_Config) ->
@@ -56,7 +56,7 @@ aae_bustedjournal(_Config) ->
     CDBFiles = testutil:find_journals(RootPath),
     [HeadF|_Rest] = CDBFiles,
     io:format("Selected Journal for corruption of ~s~n", [HeadF]),
-    testutil:corrupt_journal(RootPath, HeadF, 1000),
+    testutil:corrupt_journal(RootPath, HeadF, 1000, 2048, 1000),
     {ok, Bookie2} = leveled_bookie:book_start(StartOpts),
     
     {async, KeyF} = leveled_bookie:book_returnfolder(Bookie2,
@@ -122,8 +122,74 @@ aae_bustedjournal(_Config) ->
                     length(KeyHashList3)]),
     
     ok = leveled_bookie:book_close(Bookie2),
+    {ok, BytesCopied} = testutil:restore_file(RootPath, HeadF),
+    io:format("File restored is of size ~w~n", [BytesCopied]),
+    {ok, Bookie3} = leveled_bookie:book_start(StartOpts),
+    
+    SW4 = os:timestamp(),
+    {async, HashTreeF4} = leveled_bookie:book_returnfolder(Bookie3,
+                                                            {foldobjects_allkeys,
+                                                                ?RIAK_TAG,
+                                                                FoldObjectsFun}),
+    KeyHashList4 = HashTreeF4(),
+    
+    true = length(KeyHashList4) == 20001,
+    io:format("Fetch of hashtree using fold objects took ~w microseconds" ++
+                " and found an object count of ~w~n",
+                [timer:now_diff(os:timestamp(), SW4), length(KeyHashList4)]),
+    
+    ok = leveled_bookie:book_close(Bookie3),
+    testutil:corrupt_journal(RootPath, HeadF, 500, BytesCopied - 8000, 14),
+    
+    {ok, Bookie4} = leveled_bookie:book_start(StartOpts),
+    
+    SW5 = os:timestamp(),
+    {async, HashTreeF5} = leveled_bookie:book_returnfolder(Bookie4,
+                                                            {foldobjects_allkeys,
+                                                                ?RIAK_TAG,
+                                                                FoldObjectsFun}),
+    KeyHashList5 = HashTreeF5(),
+    
+    true = length(KeyHashList5) > 19000,
+    true = length(KeyHashList5) < HeadCount,
+    Delta5 = length(lists:subtract(KeyHashList1, KeyHashList5)),
+    true = Delta5 < 1001,
+    io:format("Fetch of hashtree using fold objects took ~w microseconds" ++
+                " and found a Delta of ~w and an objects count of ~w~n",
+                [timer:now_diff(os:timestamp(), SW5),
+                    Delta5,
+                    length(KeyHashList5)]),
+    
+    {async, HashTreeF6} = leveled_bookie:book_returnfolder(Bookie4,
+                                                            {hashtree_query,
+                                                                ?RIAK_TAG,
+                                                                check_presence}),
+    KeyHashList6 = HashTreeF6(),
+    true = length(KeyHashList6) > 19000,
+    true = length(KeyHashList6) < HeadCount,
+    
+    ok = leveled_bookie:book_close(Bookie4),
+    
+    testutil:restore_topending(RootPath, HeadF),
+    
+    {ok, Bookie5} = leveled_bookie:book_start(StartOpts),
+    
+    SW6 = os:timestamp(),
+    {async, HashTreeF7} = leveled_bookie:book_returnfolder(Bookie5,
+                                                            {foldobjects_allkeys,
+                                                                ?RIAK_TAG,
+                                                                FoldObjectsFun}),
+    KeyHashList7 = HashTreeF7(),
+    
+    true = length(KeyHashList7) == 20001,
+    io:format("Fetch of hashtree using fold objects took ~w microseconds" ++
+                " and found an object count of ~w~n",
+                [timer:now_diff(os:timestamp(), SW6), length(KeyHashList7)]),
+    
+    ok = leveled_bookie:book_close(Bookie5),
     testutil:reset_filestructure().
 
+
 riak_hash(Obj=#r_object{}) ->
     Vclock = vclock(Obj),
     UpdObj = set_vclock(Obj, lists:sort(Vclock)),

test/end_to_end/testutil.erl

@@ -29,7 +29,9 @@
             put_altered_indexed_objects/4,
             check_indexed_objects/4,
             rotating_object_check/3,
-            corrupt_journal/3,
+            corrupt_journal/5,
+            restore_file/2,
+            restore_topending/2,
             find_journals/1]).
 
 -define(RETURN_TERMS, {true, undefined}).
@@ -382,16 +384,33 @@ rotating_object_check(RootPath, B, NumberOfObjects) ->
     ok = leveled_bookie:book_close(Book2),
     ok.
     
-corrupt_journal(RootPath, FileName, Corruptions) ->
+corrupt_journal(RootPath, FileName, Corruptions, BasePosition, GapSize) ->
-    {ok, Handle} = file:open(RootPath ++ "/journal/journal_files/" ++ FileName,
+    OriginalPath = RootPath ++ "/journal/journal_files/" ++ FileName,
-                                [binary, raw, read, write]),
+    BackupPath = RootPath ++ "/journal/journal_files/" ++
+                    filename:basename(FileName, ".cdb") ++ ".bak",
+    {ok, _BytesCopied} = file:copy(OriginalPath, BackupPath),
+    {ok, Handle} = file:open(OriginalPath, [binary, raw, read, write]),
     lists:foreach(fun(X) ->
-                        Position = X * 1000 + 2048,
+                        Position = X * GapSize + BasePosition,
                         ok = file:pwrite(Handle, Position, <<0:8/integer>>)
                         end,
                     lists:seq(1, Corruptions)),
     ok = file:close(Handle).
 
+
+restore_file(RootPath, FileName) ->
+    OriginalPath = RootPath ++ "/journal/journal_files/" ++ FileName,
+    BackupPath = RootPath ++ "/journal/journal_files/" ++
+                    filename:basename(FileName, ".cdb") ++ ".bak",
+    file:copy(BackupPath, OriginalPath).
+
+restore_topending(RootPath, FileName) ->
+    OriginalPath = RootPath ++ "/journal/journal_files/" ++ FileName,
+    PndPath = RootPath ++ "/journal/journal_files/" ++
+                    filename:basename(FileName, ".cdb") ++ ".pnd",
+    ok = file:rename(OriginalPath, PndPath),
+    false = filelib:is_file(OriginalPath).
+
 find_journals(RootPath) ->
     {ok, FNsA_J} = file:list_dir(RootPath ++ "/journal/journal_files"),
     {ok, Regex} = re:compile(".*\.cdb"),