Add double-backup test

Added a test with back-to-back backups.  This caused issues with the empty CDB file it created (on opening, couldn't cope with last key of empty).

So now backup won't roll the active journal if it is empty.

src/leveled_cdb.erl

@@ -100,7 +100,9 @@
             cdb_destroy/1,
             cdb_deletepending/1,
             cdb_deletepending/3,
-            cdb_isrolling/1,
+            cdb_isrolling/1]).
+
+-export([finished_rolling/1,
             hashtable_calc/2]).
 
 -include_lib("eunit/include/eunit.hrl").
@@ -780,6 +782,25 @@ terminate(_Reason, _StateName, _State) ->
 code_change(_OldVsn, StateName, State, _Extra) ->
     {ok, StateName, State}.
 
+
+%%%============================================================================
+%%% External functions
+%%%============================================================================
+
+
+finished_rolling(CDB) ->
+    RollerFun = 
+        fun(Sleep, FinishedRolling) ->
+            case FinishedRolling of 
+                true ->
+                    true;
+                false ->
+                    timer:sleep(Sleep),
+                    not leveled_cdb:cdb_isrolling(CDB)
+            end
+        end,
+    lists:foldl(RollerFun, false, [0, 1000, 10000, 100000]).
+
 %%%============================================================================
 %%% Internal functions
 %%%============================================================================
@@ -2127,6 +2148,19 @@ emptyvalue_fromdict_test() ->
     ?assertMatch(KVP, D_Result),
     ok = file:delete("../test/from_dict_test_ev.cdb").
 
+
+empty_roll_test() ->
+    file:delete("../test/empty_roll.cdb"),
+    file:delete("../test/empty_roll.pnd"),
+    {ok, P1} = cdb_open_writer("../test/empty_roll.pnd",
+                                #cdb_options{binary_mode=true}),
+    ok = cdb_roll(P1),
+    true = finished_rolling(P1),
+    {ok, P2} = cdb_open_reader("../test/empty_roll.cdb", 
+                                #cdb_options{binary_mode=true}),
+    ok = cdb_close(P2),
+    ok = file:delete("../test/empty_roll.cdb").
+
 find_lastkey_test() ->
     file:delete("../test/lastkey.pnd"),
     {ok, P1} = cdb_open_writer("../test/lastkey.pnd",

src/leveled_imanifest.erl

@@ -168,12 +168,9 @@ writer(Manifest, ManSQN, RootPath) ->
     TmpFN = filename:join(ManPath,
                             integer_to_list(ManSQN) ++ "." ++ ?PENDING_FILEX),
     MBin = term_to_binary(to_list(Manifest), [compressed]),
-    case filelib:is_file(NewFN) of
+    leveled_log:log("I0016", [ManSQN]),
-        false ->
+    ok = file:write_file(TmpFN, MBin),
-            leveled_log:log("I0016", [ManSQN]),
+    ok = file:rename(TmpFN, NewFN),
-            ok = file:write_file(TmpFN, MBin),
-            ok = file:rename(TmpFN, NewFN)
-    end,
     GC_SQN = ManSQN - ?MANIFESTS_TO_RETAIN,
     GC_Man = filename:join(ManPath,
                             integer_to_list(GC_SQN) ++ "." ++ ?MANIFEST_FILEX),

src/leveled_inker.erl

@@ -553,18 +553,23 @@ handle_call({trim, PersistedSQN}, _From, State) ->
     ok = leveled_iclerk:clerk_trim(State#state.clerk, self(), PersistedSQN),
     {reply, ok, State};
 handle_call(roll, _From, State) ->
-    NewSQN = State#state.journal_sqn + 1,
+    case leveled_cdb:cdb_lastkey(State#state.active_journaldb) of
-    {NewJournalP, Manifest1, NewManSQN} = 
+        empty ->
-        roll_active(State#state.active_journaldb, 
+            {reply, ok, State};
-                    State#state.manifest, 
+        _ ->
-                    NewSQN,
+            NewSQN = State#state.journal_sqn + 1,
-                    State#state.cdb_options,
+            {NewJournalP, Manifest1, NewManSQN} = 
-                    State#state.root_path,
+                roll_active(State#state.active_journaldb, 
-                    State#state.manifest_sqn),
+                            State#state.manifest, 
-    {reply, ok, State#state{journal_sqn = NewSQN,
+                            NewSQN,
-                                manifest = Manifest1,
+                            State#state.cdb_options,
-                                manifest_sqn = NewManSQN,
+                            State#state.root_path,
-                                active_journaldb = NewJournalP}};
+                            State#state.manifest_sqn),
+            {reply, ok, State#state{journal_sqn = NewSQN,
+                                        manifest = Manifest1,
+                                        manifest_sqn = NewManSQN,
+                                        active_journaldb = NewJournalP}}
+    end;
 handle_call({backup, BackupPath}, _from, State) 
                                         when State#state.is_snapshot == true ->
     SW = os:timestamp(),
@@ -576,7 +581,7 @@ handle_call({backup, BackupPath}, _from, State)
                 true ->
                     BaseFN = filename:basename(FN),
                     BackupName = filename:join(BackupJFP, BaseFN),
-                    false = when_not_rolling(PidR),
+                    true = leveled_cdb:finished_rolling(PidR),
                     case file:make_link(FN ++ "." ++ ?JOURNAL_FILEX, 
                                             BackupName ++ "." ++ ?JOURNAL_FILEX) of
                         ok ->
@@ -712,19 +717,6 @@ start_from_file(InkOpts) ->
                     clerk = Clerk}}.
 
 
-when_not_rolling(CDB) ->
-    RollerFun = 
-        fun(Sleep, WasRolling) ->
-            case WasRolling of 
-                false ->
-                    false;
-                true ->
-                    timer:sleep(Sleep),
-                    leveled_cdb:cdb_isrolling(CDB)
-            end
-        end,
-    lists:foldl(RollerFun, true, [0, 1000, 10000, 100000]).
-
 -spec shutdown_snapshots(list(tuple())) -> ok.
 %% @doc
 %% Shutdown any snapshots before closing the store
@@ -983,9 +975,7 @@ open_all_manifest(Man0, RootPath, CDBOpts) ->
             NewManEntry = start_new_activejournal(LastSQN + 1,
                                                         RootPath,
                                                         CDBOpts),
-            leveled_imanifest:add_entry(ManToHead,
+            leveled_imanifest:add_entry(ManToHead, NewManEntry, true);
-                                        NewManEntry,
-                                        true);
         false ->
             {ok, HeadW} = leveled_cdb:cdb_open_writer(PendingHeadFN,
                                                         CDBOpts),

src/leveled_log.erl

@@ -284,7 +284,7 @@
         {info, "Journal backup completed to path=~s with file_count=~w"}},
     {"I0021",
         {info, "Ingoring filename=~s with SQN=~w and JournalSQN=~w"}},
-    
+        
     {"IC001",
         {info, "Closed for reason ~w so maybe leaving garbage"}},
     {"IC002",

test/end_to_end/recovery_SUITE.erl

@@ -3,6 +3,7 @@
 -include("include/leveled.hrl").
 -export([all/0]).
 -export([hot_backup_simple/1,
+            hot_backup_double/1,
             retain_strategy/1,
             recovr_strategy/1,
             aae_missingjournal/1,
@@ -12,6 +13,7 @@
 
 all() -> [
             hot_backup_simple,
+            hot_backup_double,
             retain_strategy,
             recovr_strategy,
             aae_missingjournal,
@@ -50,7 +52,31 @@ hot_backup_simple(_Config) ->
     ok = leveled_bookie:book_close(BookBackup),
     BackupPath = testutil:reset_filestructure("backup0").
 
-
+hot_backup_double(_Config) ->
+    % As with simple test, but check that calling for backup twice doesn't have
+    % any side effects
+    RootPath = testutil:reset_filestructure(),
+    BookOpts = [{root_path, RootPath},
+                    {cache_size, 1000},
+                    {max_journalsize, 10000000},
+                    {sync_strategy, testutil:sync_strategy()}],
+    {ok, Spcl1, LastV1} = rotating_object_check(BookOpts, "Bucket1", 4000),
+    {ok, Book1} = leveled_bookie:book_start(BookOpts),
+    {async, BackupFun1} = leveled_bookie:book_hotbackup(Book1),
+    BackupPath = testutil:reset_filestructure("backup0"),
+    ok = BackupFun1(BackupPath),
+    {async, BackupFun2} = leveled_bookie:book_hotbackup(Book1),
+    ok = BackupFun2(BackupPath),
+    ok = leveled_bookie:book_close(Book1),
+    RootPath = testutil:reset_filestructure(),
+    BookOptsBackup = [{root_path, BackupPath},
+                        {cache_size, 2000},
+                        {max_journalsize, 20000000},
+                        {sync_strategy, testutil:sync_strategy()}],
+    {ok, BookBackup} = leveled_bookie:book_start(BookOptsBackup),
+    ok = testutil:check_indexed_objects(BookBackup, "Bucket1", Spcl1, LastV1),
+    ok = leveled_bookie:book_close(BookBackup),
+    BackupPath = testutil:reset_filestructure("backup0").
 
 retain_strategy(_Config) ->
     RootPath = testutil:reset_filestructure(),