Ensure backlog is tackled (#394)

Even when level 0 work is continuous.

src/leveled_log.erl

@@ -109,7 +109,7 @@
         p0019 =>
             {info, <<"Rolling level zero to filename ~s at ledger sqn ~w">>},
         p0024 =>
-            {info, <<"Outstanding compaction work items of ~w with backlog status of ~w">>},
+            {info, <<"Outstanding compaction work items of ~w with backlog status of ~w L0 full ~w">>},
         p0029 =>
             {info, <<"L0 completion confirmed and will transition to not pending">>},
         p0030 =>

src/leveled_penciller.erl

@@ -1070,11 +1070,20 @@ handle_cast({levelzero_complete, FN, StartKey, EndKey, Bloom}, State) ->
                             manifest=UpdMan,
                             persisted_sqn=State#state.ledger_sqn}};
 handle_cast(work_for_clerk, State) ->
-    case {State#state.levelzero_pending,
+    case {(State#state.levelzero_pending or State#state.work_ongoing),
-            State#state.work_ongoing,
             leveled_pmanifest:levelzero_present(State#state.manifest)} of
-        {false, false, false} ->
+        {true, _L0Present} ->
-            % If the penciller memory needs rolling, prompt this now
+            % Work is blocked by ongoing activity
+            {noreply, State};
+        {false, true} ->
+            % If L0 present, and no work ongoing - dropping L0 to L1 is the
+            % priority
+            ok = leveled_pclerk:clerk_push(
+                State#state.clerk, {0, State#state.manifest}),
+            {noreply, State#state{work_ongoing=true}};
+        {false, false} ->
+            % No impediment to work - see what other work may be required
+            % See if the in-memory cache requires rolling now
             CacheOverSize =
                 maybe_cache_too_big(
                     State#state.levelzero_size,
@@ -1082,9 +1091,16 @@ handle_cast(work_for_clerk, State) ->
                     State#state.levelzero_cointoss),
             CacheAlreadyFull =
                 leveled_pmem:cache_full(State#state.levelzero_cache),
-            case (CacheAlreadyFull or CacheOverSize) of
+            % Check for a backlog of work
-                true ->
+            {WL, WC} = leveled_pmanifest:check_for_work(State#state.manifest),
+            case {WC, (CacheAlreadyFull or CacheOverSize)} of
+                {0, false} ->
+                    % No work required
+                    {noreply, State#state{work_backlog = false}};
+                {WC, true} when WC < ?WORKQUEUE_BACKLOG_TOLERANCE ->
                     % Rolling the memory to create a new Level Zero file
+                    % Must not do this if there is a work backlog beyond the 
+                    % tolerance, as then the backlog may never be addressed.
                     NextSQN =
                         leveled_pmanifest:get_manifest_sqn(
                             State#state.manifest) + 1,
@@ -1100,34 +1116,23 @@ handle_cast(work_for_clerk, State) ->
                     {noreply,
                         State#state{
                             levelzero_pending = true,
-                            levelzero_constructor=Constructor}};
+                            levelzero_constructor = Constructor,
-                false ->
+                            work_backlog = false}}; 
-                    {WL, WC} =
+                {WC, L0Full} ->
-                        leveled_pmanifest:check_for_work(State#state.manifest),
+                    % Address the backlog of work, either because there is no
-                    case WC of
+                    % L0 work to do, or because the backlog has grown beyond
-                        0 ->
+                    % tolerance
-                            % Should do some tidy-up work here?
+                    Backlog = WC >= ?WORKQUEUE_BACKLOG_TOLERANCE,
-                            {noreply, State#state{work_backlog=false}};
+                    leveled_log:log(p0024, [WC, Backlog, L0Full]),
-                        N ->
-                            Backlog = N > ?WORKQUEUE_BACKLOG_TOLERANCE,
-                            leveled_log:log(p0024, [N, Backlog]),
                     [TL|_Tail] = WL,
                     ok =
                         leveled_pclerk:clerk_push(
-                                    State#state.clerk,
+                            State#state.clerk, {TL, State#state.manifest}),
-                                    {TL, State#state.manifest}),
                     {noreply,
                         State#state{
                             work_backlog = Backlog, work_ongoing = true}}
             end
     end;
-        {false, false, true} ->
-            ok = leveled_pclerk:clerk_push(
-                State#state.clerk, {0, State#state.manifest}),
-            {noreply, State#state{work_ongoing=true}};
-        _ ->
-            {noreply, State}
-    end;
 handle_cast({fetch_levelzero, Slot, ReturnFun}, State) ->
     ReturnFun(lists:nth(Slot, State#state.levelzero_cache)),
     {noreply, State};