Correct async fold

fold objects which snaps in the fold was implemented incorrectly - it
took information from the LedgeCache at the point of the request, not
at the point of the fold.  So the LedgerCache SQN may have been
surpassed in the Penciller by the time the fold was called.

src/leveled_bookie.erl

@@ -484,6 +484,9 @@ handle_call({head, Bucket, Key, Tag}, _From, State) ->
             end
     end;
 handle_call({snapshot, _Requestor, SnapType, _Timeout}, _From, State) ->
+    % TODO: clean-up passing of Requestor (which was previously just used in 
+    % logs) and so can now be ignored, and timeout which is ignored - but 
+    % probably shouldn't be.
     Reply = snapshot_store(State, SnapType),
     {reply, Reply, State};
 handle_call({return_folder, FolderType}, _From, State) ->
@@ -860,7 +863,7 @@ foldobjects_byindex(State, Tag, Bucket,
     foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, false).
 
 
-foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, DeferredFetch) ->
+foldobjects(_State, Tag, StartKey, EndKey, FoldObjectsFun, DeferredFetch) ->
     {FoldFun, InitAcc} = case is_tuple(FoldObjectsFun) of
                                 true ->
                                     FoldObjectsFun;
@@ -874,11 +877,17 @@ foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, DeferredFetch) ->
     % and the sweeper prompts the fold before checking to see if the fold is
     % ready to be run.  This may lead to the fold being called on an old
     % snapshot.
+    Self = self(),
     Folder =
         fun() ->
             {ok,
                 LedgerSnapshot,
-                JournalSnapshot} = snapshot_store(State, store),
+                JournalSnapshot} = book_snapshotstore(Self, Self, 5400),
+                % Timeout will be ignored, as will Requestor
+                %
+                % This uses the external snapshot - as the snpshot will need 
+                % to have consistent state between Bookie and Penciller when
+                % it is made.
             AccFun = accumulate_objects(FoldFun,
                                         JournalSnapshot,
                                         Tag,

src/leveled_penciller.erl

@@ -214,7 +214,7 @@
 -define(COIN_SIDECOUNT, 5).
 -define(SLOW_FETCH, 20000).
 -define(ITERATOR_SCANWIDTH, 4).
--define(SNAPSHOT_TIMEOUT_LONG, 5400).
+-define(SNAPSHOT_TIMEOUT_LONG, 3600).
 -define(SNAPSHOT_TIMEOUT_SHORT, 600).
 
 -record(state, {manifest, % a manifest record from the leveled_manifest module

test/end_to_end/basic_SUITE.erl

@@ -489,7 +489,7 @@ space_clear_ondelete(_Config) ->
     timer:sleep(10000), % Allow for any L0 file to be rolled
     {ok, FNsA_L} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
     {ok, FNsA_J} = file:list_dir(RootPath ++ "/journal/journal_files"),
-    io:format("Bookie created ~w journal files and ~w ledger files~n",
+    io:format("FNsA - Bookie created ~w journal files and ~w ledger files~n",
                     [length(FNsA_J), length(FNsA_L)]),
     
     % Get an iterator to lock the inker during compaction
@@ -499,6 +499,8 @@ space_clear_ondelete(_Config) ->
                                                         {foldobjects_allkeys,
                                                             ?RIAK_TAG,
                                                             FoldObjectsFun}),
+    {async, KF1} = leveled_bookie:book_returnfolder(Book1, AllKeyQuery),
+
     % Delete the keys
     SW2 = os:timestamp(),
     lists:foreach(fun({Bucket, Key}) ->
@@ -529,9 +531,12 @@ space_clear_ondelete(_Config) ->
                     lists:seq(1, 15)),
     io:format("Waiting for journal deletes - blocked~n"),
     timer:sleep(20000),
-    KeyHashList1 = HTreeF1(),
-    io:format("Key Hash List returned of length ~w~n", [length(KeyHashList1)]),
-    true = length(KeyHashList1) == 80000,
+    
+    % for this query snapshot is made at fold time
+    true = length(HTreeF1()) == 0,
+    % This query uses a genuine async fold on a snasphot made at request time
+    true = length(KF1()) == 80000, 
+    
     io:format("Waiting for journal deletes - unblocked~n"),
     timer:sleep(20000),
     {ok, FNsB_L} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
@@ -539,7 +544,7 @@ space_clear_ondelete(_Config) ->
     {ok, FNsB_PC} = file:list_dir(RootPath
                                     ++ "/journal/journal_files/post_compact"),
     PointB_Journals = length(FNsB_J) + length(FNsB_PC),
-    io:format("Bookie has ~w journal files and ~w ledger files " ++
+    io:format("FNsB - Bookie has ~w journal files and ~w ledger files " ++
                     "after deletes~n",
                 [PointB_Journals, length(FNsB_L)]),
     
@@ -566,7 +571,7 @@ space_clear_ondelete(_Config) ->
             end,
     ok = leveled_bookie:book_close(Book2),
     {ok, FNsC_L} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
-    io:format("Bookie has ~w ledger files " ++
+    io:format("FNsC - Bookie has ~w ledger files " ++
                     "after close~n", [length(FNsC_L)]),
     
     {ok, Book3} = leveled_bookie:book_start(StartOpts1),
@@ -576,8 +581,12 @@ space_clear_ondelete(_Config) ->
     timer:sleep(12000),
     ok = leveled_bookie:book_close(Book3),
     {ok, FNsD_L} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
-    io:format("Bookie has ~w ledger files " ++
+    io:format("FNsD - Bookie has ~w ledger files " ++
                     "after second close~n", [length(FNsD_L)]),
+    lists:foreach(fun(FN) -> 
+                        io:format("FNsD - Ledger file if ~s~n", [FN]) 
+                    end, 
+                    FNsD_L),
     true = PointB_Journals < length(FNsA_J),
     true = length(FNsD_L) < length(FNsA_L),
     true = length(FNsD_L) < length(FNsB_L),