Roll in fix for intermittently failing test

As descibed in https://github.com/martinsumner/leveled/issues/92

Only the first fix was made.

Just to eb safe - archiving means renaming to another file with a different extension.  Assumption is that renamed files cna be manually reaped if necessary.

src/leveled_codec.erl

@@ -620,7 +620,7 @@ build_metadata_object(PrimaryKey, MD) ->
 %% The metadata object should be returned with the full list of last 
 %% modified dates (which will be used for recent anti-entropy index creation)
 riak_extract_metadata(delete, Size) ->
-    {{delete, null, null, null, Size}, []};
+    {{delete, null, null, Size}, []};
 riak_extract_metadata(ObjBin, Size) ->
     {VclockBin, SibBin, LastMods} = riak_metadata_from_binary(ObjBin),
     {{SibBin, 

src/leveled_log.erl

@@ -120,7 +120,8 @@
     {"P0030",
         {warn, "We're doomed - intention recorded to destroy all files"}},
     {"P0031",
-        {info, "Completion of update to levelzero"}},
+        {info, "Completion of update to levelzero"
+                    ++ " with cache size status ~w ~w"}},
     {"P0032",
         {info, "Head timing for result ~w is sample ~w total ~w and max ~w"}},
     {"P0033",
@@ -141,6 +142,8 @@
     {"P0039",
         {info, "Failed to release pid=~w "
                     ++ "leaving SnapshotCount=~w and MinSQN=~w"}},
+    {"P0040",
+        {info, "Archiving filename ~s as unused at startup"}},
         
     {"PC001",
         {info, "Penciller's clerk ~w started with owner ~w"}},

src/leveled_penciller.erl

@@ -206,6 +206,8 @@
 -define(FILES_FP, "ledger_files").
 -define(CURRENT_FILEX, "crr").
 -define(PENDING_FILEX, "pnd").
+-define(SST_FILEX, ".sst").
+-define(ARCHIVE_FILEX, ".bak").
 -define(MEMTABLE, mem).
 -define(MAX_TABLESIZE, 28000). % This is less than max - but COIN_SIDECOUNT
 -define(SUPER_MAX_TABLE_SIZE, 40000).
@@ -819,7 +821,8 @@ sst_rootpath(RootPath) ->
     FP.
 
 sst_filename(ManSQN, Level, Count) ->
-    lists:flatten(io_lib:format("./~w_~w_~w.sst", [ManSQN, Level, Count])).
+    lists:flatten(io_lib:format("./~w_~w_~w" ++ ?SST_FILEX, 
+                                    [ManSQN, Level, Count])).
     
 
 %%%============================================================================
@@ -859,41 +862,73 @@ start_from_file(PCLopts) ->
             Pid
         end,
     SQNFun = fun leveled_sst:sst_getmaxsequencenumber/1,
-    {MaxSQN, Manifest1} = leveled_pmanifest:load_manifest(Manifest0,
+    {MaxSQN, Manifest1, FileList} = 
-                                                            OpenFun,
+        leveled_pmanifest:load_manifest(Manifest0, OpenFun, SQNFun),
-                                                            SQNFun),
     leveled_log:log("P0014", [MaxSQN]),
     ManSQN = leveled_pmanifest:get_manifest_sqn(Manifest1),
     leveled_log:log("P0035", [ManSQN]),
     %% Find any L0 files
     L0FN = sst_filename(ManSQN + 1, 0, 0),
-    case filelib:is_file(filename:join(sst_rootpath(RootPath), L0FN)) of
+    {State0, FileList0} = 
-        true ->
+        case filelib:is_file(filename:join(sst_rootpath(RootPath), L0FN)) of
-            leveled_log:log("P0015", [L0FN]),
+            true ->
-            L0Open = leveled_sst:sst_open(sst_rootpath(RootPath), L0FN),
+                leveled_log:log("P0015", [L0FN]),
-            {ok, L0Pid, {L0StartKey, L0EndKey}} = L0Open,
+                L0Open = leveled_sst:sst_open(sst_rootpath(RootPath), L0FN),
-            L0SQN = leveled_sst:sst_getmaxsequencenumber(L0Pid),
+                {ok, L0Pid, {L0StartKey, L0EndKey}} = L0Open,
-            L0Entry = #manifest_entry{start_key = L0StartKey,
+                L0SQN = leveled_sst:sst_getmaxsequencenumber(L0Pid),
-                                        end_key = L0EndKey,
+                L0Entry = #manifest_entry{start_key = L0StartKey,
-                                        filename = L0FN,
+                                            end_key = L0EndKey,
-                                        owner = L0Pid},
+                                            filename = L0FN,
-            Manifest2 = leveled_pmanifest:insert_manifest_entry(Manifest1,
+                                            owner = L0Pid},
-                                                                ManSQN + 1,
+                Manifest2 = leveled_pmanifest:insert_manifest_entry(Manifest1,
-                                                                0,
+                                                                    ManSQN + 1,
-                                                                L0Entry),
+                                                                    0,
-            leveled_log:log("P0016", [L0SQN]),
+                                                                    L0Entry),
-            LedgerSQN = max(MaxSQN, L0SQN),
+                leveled_log:log("P0016", [L0SQN]),
-            {ok,
+                LedgerSQN = max(MaxSQN, L0SQN),
-                InitState#state{manifest = Manifest2,
+                {InitState#state{manifest = Manifest2,
                                     ledger_sqn = LedgerSQN,
-                                    persisted_sqn = LedgerSQN}};
+                                    persisted_sqn = LedgerSQN},
-        false ->
+                    [L0FN|FileList]};
-            leveled_log:log("P0017", []),
+            false ->
-            {ok,
+                leveled_log:log("P0017", []),
-                InitState#state{manifest = Manifest1,
+                {InitState#state{manifest = Manifest1,
                                     ledger_sqn = MaxSQN,
-                                    persisted_sqn = MaxSQN}}
+                                    persisted_sqn = MaxSQN},
-    end.
+                    FileList}
+        end,
+    ok = archive_files(RootPath, FileList0),
+    {ok, State0}.
+
+archive_files(RootPath, FileList) ->
+    {ok, AllFiles} = file:list_dir(sst_rootpath(RootPath)),
+    FileCheckFun =
+        fun(FN, UnusedFiles) ->
+            FN0 = "./" ++ FN,
+            case filename:extension(FN0) of 
+                ?SST_FILEX ->
+                    case lists:member(FN0, FileList) of 
+                        true ->
+                            UnusedFiles;
+                        false ->
+                            leveled_log:log("P0040", [FN0]),
+                            [FN0|UnusedFiles]
+                    end;
+                _ ->
+                    UnusedFiles
+            end
+        end,
+    RenameFun =
+        fun(FN) ->
+            AltName = filename:join(sst_rootpath(RootPath), 
+                                        filename:basename(FN, ?SST_FILEX))
+                        ++ ?ARCHIVE_FILEX,
+            file:rename(filename:join(sst_rootpath(RootPath), FN), 
+                            AltName)
+        end,
+    FilesToArchive = lists:foldl(FileCheckFun, [], AllFiles),
+    lists:foreach(RenameFun, FilesToArchive),
+    ok.
 
 
 update_levelzero(L0Size, {PushedTree, PushedIdx, MinSQN, MaxSQN},
@@ -934,11 +969,13 @@ update_levelzero(L0Size, {PushedTree, PushedIdx, MinSQN, MaxSQN},
             case {CacheTooBig, L0Free, JitterCheck, NoPendingManifestChange} of
                 {true, true, true, true}  ->
                     L0Constructor = roll_memory(UpdState, false),
-                    leveled_log:log_timer("P0031", [], SW),
+                    leveled_log:log_timer("P0031", [true, true], SW),
                     UpdState#state{levelzero_pending=true,
                                     levelzero_constructor=L0Constructor};
                 _ ->
-                    leveled_log:log_timer("P0031", [], SW),
+                    leveled_log:log_timer("P0031", 
+                                            [CacheTooBig, JitterCheck], 
+                                            SW),
                     UpdState
             end
     end.

src/leveled_pmanifest.erl

@@ -136,7 +136,8 @@ copy_manifest(Manifest) ->
     % about is switched to undefined
     Manifest#manifest{snapshots = undefined, pending_deletes = undefined}.
 
--spec load_manifest(manifest(), fun(), fun()) -> {integer(), manifest()}.
+-spec load_manifest(manifest(), fun(), fun()) -> 
+                                        {integer(), manifest(), list()}.
 %% @doc
 %% Roll over the manifest starting a process to manage each file in the
 %% manifest.  The PidFun should be able to return the Pid of a file process
@@ -144,13 +145,15 @@ copy_manifest(Manifest) ->
 %% of that file, if passed the Pid that owns it.
 load_manifest(Manifest, PidFun, SQNFun) ->
     UpdateLevelFun =
-        fun(LevelIdx, {AccMaxSQN, AccMan}) ->
+        fun(LevelIdx, {AccMaxSQN, AccMan, AccFL}) ->
             L0 = array:get(LevelIdx, AccMan#manifest.levels),
-            {L1, SQN1} = load_level(LevelIdx, L0, PidFun, SQNFun),
+            {L1, SQN1, FileList} = load_level(LevelIdx, L0, PidFun, SQNFun),
             UpdLevels = array:set(LevelIdx, L1, AccMan#manifest.levels),
-            {max(AccMaxSQN, SQN1), AccMan#manifest{levels = UpdLevels}}
+            {max(AccMaxSQN, SQN1), 
+                AccMan#manifest{levels = UpdLevels},
+                AccFL ++ FileList}
         end,
-    lists:foldl(UpdateLevelFun, {0, Manifest},
+    lists:foldl(UpdateLevelFun, {0, Manifest, []},
                     lists:seq(0, Manifest#manifest.basement)).
 
 -spec close_manifest(manifest(), fun()) -> ok.
@@ -488,27 +491,33 @@ levelzero_present(Manifest) ->
 
 load_level(LevelIdx, Level, PidFun, SQNFun) ->
     HigherLevelLoadFun =
-        fun(ME, {L_Out, L_MaxSQN}) ->
+        fun(ME, {L_Out, L_MaxSQN, FileList}) ->
             FN = ME#manifest_entry.filename,
             P = PidFun(FN),
             SQN = SQNFun(P),
-            {[ME#manifest_entry{owner=P}|L_Out], max(SQN, L_MaxSQN)}
+            {[ME#manifest_entry{owner=P}|L_Out], 
+                max(SQN, L_MaxSQN),
+                [FN|FileList]}
         end,
     LowerLevelLoadFun =
-        fun({EK, ME}, {L_Out, L_MaxSQN}) ->
+        fun({EK, ME}, {L_Out, L_MaxSQN, FileList}) ->
             FN = ME#manifest_entry.filename,
             P = PidFun(FN),
             SQN = SQNFun(P),
-            {[{EK, ME#manifest_entry{owner=P}}|L_Out], max(SQN, L_MaxSQN)}
+            {[{EK, ME#manifest_entry{owner=P}}|L_Out], 
+                max(SQN, L_MaxSQN),
+                [FN|FileList]}
         end,
     case LevelIdx =< 1 of
         true ->
-            lists:foldr(HigherLevelLoadFun, {[], 0}, Level);
+            lists:foldr(HigherLevelLoadFun, {[], 0, []}, Level);
         false ->
-            {L0, MaxSQN} = lists:foldr(LowerLevelLoadFun,
+            {L0, MaxSQN, Flist} = lists:foldr(LowerLevelLoadFun,
-                                        {[], 0},
+                                        {[], 0, []},
                                         leveled_tree:to_list(Level)),
-            {leveled_tree:from_orderedlist(L0, ?TREE_TYPE, ?TREE_WIDTH), MaxSQN}
+            {leveled_tree:from_orderedlist(L0, ?TREE_TYPE, ?TREE_WIDTH), 
+                MaxSQN, 
+                Flist}
     end.
 
 close_level(LevelIdx, Level, CloseEntryFun) when LevelIdx =< 1 ->

test/end_to_end/basic_SUITE.erl

@@ -580,22 +580,59 @@ space_clear_ondelete(_Config) ->
     io:format("This should cause a final ledger merge event~n"),
     io:format("Will require the penciller to resolve the issue of creating" ++
                 " an empty file as all keys compact on merge~n"),
-    timer:sleep(12000),
+    
+    CheckFun = 
+        fun(X, FileCount) ->
+            case FileCount of
+                0 ->
+                    0;
+                _ ->
+                    timer:sleep(X),
+                    {ok, NewFC} = 
+                        file:list_dir(RootPath ++ "/ledger/ledger_files"),
+                    io:format("Looping with ledger file count ~w~n", 
+                                [length(NewFC)]),
+                    length(strip_nonsst(NewFC))
+            end
+        end,
+    
+    FC = lists:foldl(CheckFun, infinity, [2000, 3000, 5000, 8000]),
     ok = leveled_bookie:book_close(Book3),
+    case FC of 
+        0 ->
+            ok;
+        _ ->
+            {ok, Book4} = leveled_bookie:book_start(StartOpts1),
+            lists:foldl(CheckFun, infinity, [2000, 3000, 5000, 8000]),
+            leveled_bookie:book_close(Book4)
+    end,
+
     {ok, FNsD_L} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
     io:format("FNsD - Bookie has ~w ledger files " ++
-                    "after second close~n", [length(FNsD_L)]),
+                    "after second close~n", [length(strip_nonsst(FNsD_L))]),
     lists:foreach(fun(FN) -> 
                         io:format("FNsD - Ledger file is ~s~n", [FN]) 
                     end, 
                     FNsD_L),
     true = PointB_Journals < length(FNsA_J),
-    true = length(FNsD_L) < length(FNsA_L),
+    true = length(strip_nonsst(FNsD_L)) < length(strip_nonsst(FNsA_L)),
-    true = length(FNsD_L) < length(FNsB_L),
+    true = length(strip_nonsst(FNsD_L)) < length(strip_nonsst(FNsB_L)),
-    true = length(FNsD_L) < length(FNsC_L),
+    true = length(strip_nonsst(FNsD_L)) < length(strip_nonsst(FNsC_L)),
-    true = length(FNsD_L) == 0.
+    true = length(strip_nonsst(FNsD_L)) == 0.
 
 
+strip_nonsst(FileList) ->
+    SSTOnlyFun = 
+        fun(FN, Acc) ->
+            case filename:extension(FN) of 
+                ".sst" -> 
+                    [FN|Acc];
+                _ ->
+                    Acc 
+            end
+        end,
+    lists:foldl(SSTOnlyFun, [], FileList).
+
 
 is_empty_test(_Config) ->
     RootPath = testutil:reset_filestructure(),