Add timeout for inker snapshots

So that they can also be released if they silently crash without closing neatly.

docs/STARTUP_OPTIONS.md

@@ -107,3 +107,15 @@ The `compaction_runs_perday` indicates for the leveled store how many times eahc
 The `compaction_low_hour` and `compaction_high_hour` are the hours of the day which support the compaction window - set to 0 and 23 respectively if compaction is required to be a continuous process.
 
 The `max_run_length` controls how many files can be compacted in a single compaction run.  The scoring of files and runs is controlled through `maxrunlength_compactionpercentage` and `singlefile_compactionpercentage`.
+
+
+## Snapshot Timeouts
+
+There are two snapshot timeouts that can be configured:
+
+- `snapshot_timeout_short`
+- `snapshot_timeout_long`
+
+These set the period in seconds before a snapshot which has not shutdown, is declared to have been released - so that any file deletions which are awaiting the snapshot's completion can go ahead.
+
+This covers only silently failing snapshots.  Snapshots that shutdown neatly will be released from locking deleted files when they shutdown.  The 'short' timeout is used for snapshots which support index queries and bucket listing.  The 'long' timeout is used for all other folds (e.g. key lists, head folds and object folds).

include/leveled.hrl

@@ -65,7 +65,8 @@
                         compress_on_receipt = false :: boolean(),
                         max_run_length,
                         singlefile_compactionperc :: float()|undefined,
-                        maxrunlength_compactionperc :: float()|undefined}).
+                        maxrunlength_compactionperc :: float()|undefined,
+                        snaptimeout_long :: pos_integer() | undefined}).
 
 -record(penciller_options,
                         {root_path :: string() | undefined,

src/leveled_bookie.erl

@@ -117,8 +117,8 @@
 -define(MAX_KEYCHECK_FREQUENCY, 100).
 -define(MIN_KEYCHECK_FREQUENCY, 1).
 -define(OPEN_LASTMOD_RANGE, {0, infinity}).
--define(PCL_SNAPTIMEOUT_SHORT, 900). % 15 minutes
+-define(SNAPTIMEOUT_SHORT, 900). % 15 minutes
--define(PCL_SNAPTIMEOUT_LONG, 43200). % 12 hours
+-define(SNAPTIMEOUT_LONG, 43200). % 12 hours
 -define(OPTION_DEFAULTS,
             [{root_path, undefined},
                 {snapshot_bookie, undefined},
@@ -137,8 +137,8 @@
                 {log_level, ?LOG_LEVEL},
                 {forced_logs, []},
                 {override_functions, []},
-                {pcl_snapshottimeout_short, ?PCL_SNAPTIMEOUT_SHORT},
+                {snapshot_timeout_short, ?SNAPTIMEOUT_SHORT},
-                {pcl_snapshottimeout_long, ?PCL_SNAPTIMEOUT_LONG}]).
+                {snapshot_timeout_long, ?SNAPTIMEOUT_LONG}]).
 
 -record(ledger_cache, {mem :: ets:tab(),
                         loader = leveled_tree:empty(?CACHE_TYPE)
@@ -334,12 +334,12 @@
         {override_functions, list(leveled_head:appdefinable_function_tuple())} |
             % Provide a list of override functions that will be used for
             % user-defined tags
-        {pcl_snapshottimeout_short, pos_integer()} |
+        {snapshot_timeout_short, pos_integer()} |
             % Time in seconds before a snapshot that has not been shutdown is
             % assumed to have failed, and so requires to be torndown.  The
             % short timeout is applied to queries where long_running is set to
             % false
-        {pcl_snapshottimeout_long, pos_integer()}
+        {snapshot_timeout_long, pos_integer()}
             % Time in seconds before a snapshot that has not been shutdown is
             % assumed to have failed, and so requires to be torndown.  The
             % short timeout is applied to queries where long_running is set to
@@ -1577,8 +1577,8 @@ set_options(Opts) ->
     SyncStrat = proplists:get_value(sync_strategy, Opts),
     WRP = proplists:get_value(waste_retention_period, Opts),
 
-    SnapTimeoutShort = proplists:get_value(pcl_snapshottimeout_short, Opts),
+    SnapTimeoutShort = proplists:get_value(snapshot_timeout_short, Opts),
-    SnapTimeoutLong = proplists:get_value(pcl_snapshottimeout_long, Opts),
+    SnapTimeoutLong = proplists:get_value(snapshot_timeout_long, Opts),
 
     AltStrategy = proplists:get_value(reload_strategy, Opts),
     ReloadStrategy = leveled_codec:inker_reload_strategy(AltStrategy),
@@ -1619,6 +1619,7 @@ set_options(Opts) ->
                         singlefile_compactionperc = SFL_CompPerc,
                         maxrunlength_compactionperc = MRL_CompPerc,
                         waste_retention_period = WRP,
+                        snaptimeout_long = SnapTimeoutLong,
                         compression_method = CompressionMethod,
                         compress_on_receipt = CompressOnReceipt,
                         cdb_options = 

src/leveled_inker.erl

@@ -151,6 +151,7 @@
                 is_snapshot = false :: boolean(),
                 compression_method = native :: lz4|native,
                 compress_on_receipt = false :: boolean(),
+                snap_timeout :: pos_integer() | undefined, % in seconds
                 source_inker :: pid() | undefined}).
 
 
@@ -541,6 +542,7 @@ handle_call({fold,
     end;
 handle_call({register_snapshot, Requestor}, _From , State) ->
     Rs = [{Requestor,
+            os:timestamp(),
             State#state.manifest_sqn}|State#state.registered_snapshots],
     leveled_log:log("I0002", [Requestor, State#state.manifest_sqn]),
     {reply, {State#state.manifest,
@@ -548,13 +550,28 @@ handle_call({register_snapshot, Requestor}, _From , State) ->
                 State#state.journal_sqn},
                 State#state{registered_snapshots=Rs}};
 handle_call({confirm_delete, ManSQN}, _From, State) ->
+    % Check there are no snapshots that may be aware of the file process that
+    % is waiting to delete itself.
     CheckSQNFun = 
-        fun({_R, SnapSQN}, Bool) ->
+        fun({_R, _TS, SnapSQN}, Bool) ->
+            % If the Snapshot SQN was at the same point the file was set to
+            % delete (or after), then the snapshot would not have been told
+            % of the file, and the snapshot should not hold up its deletion 
             (SnapSQN >= ManSQN) and Bool
         end,
+    CheckSnapshotExpiryFun =
+        fun({_R, TS, _SnapSQN}) ->
+            Expiry = leveled_util:integer_time(TS) + State#state.snap_timeout,
+                % If Expiry has passed this will be false, and the snapshot
+                % will be removed from the list of registered snapshots and
+                % so will not longer block deletes
+            leveled_util:integer_now() < Expiry
+        end,
+    RegisteredSnapshots0 =
+        lists:filter(CheckSnapshotExpiryFun, State#state.registered_snapshots),
     {reply, 
-        lists:foldl(CheckSQNFun, true, State#state.registered_snapshots), 
+        lists:foldl(CheckSQNFun, true, RegisteredSnapshots0), 
-        State};
+        State#state{registered_snapshots = RegisteredSnapshots0}};
 handle_call(get_manifest, _From, State) ->
     {reply, leveled_imanifest:to_list(State#state.manifest), State};
 handle_call({update_manifest,
@@ -791,6 +808,8 @@ start_from_file(InkOpts) ->
     MRL_CompactPerc = InkOpts#inker_options.maxrunlength_compactionperc,
     PressMethod = InkOpts#inker_options.compression_method,
     PressOnReceipt = InkOpts#inker_options.compress_on_receipt,
+    SnapTimeout = InkOpts#inker_options.snaptimeout_long,
+
     IClerkOpts = 
         #iclerk_options{inker = self(),
                             cdb_options=IClerkCDBOpts,
@@ -799,8 +818,7 @@ start_from_file(InkOpts) ->
                             compression_method = PressMethod,
                             max_run_length = MRL,
                             singlefile_compactionperc = SFL_CompactPerc,
-                            maxrunlength_compactionperc = MRL_CompactPerc
+                            maxrunlength_compactionperc = MRL_CompactPerc},
-                            },
     
     {ok, Clerk} = leveled_iclerk:clerk_new(IClerkOpts),
     
@@ -821,6 +839,7 @@ start_from_file(InkOpts) ->
                     cdb_options = CDBopts,
                     compression_method = PressMethod,
                     compress_on_receipt = PressOnReceipt,
+                    snap_timeout = SnapTimeout,
                     clerk = Clerk}}.