add internal timers for schedule (#40)

* add internal timers for schedule

* cleanup

---------

Co-authored-by: ttt161 <losto@nix>

Author: ttt161

include/progressor.hrl

@@ -88,6 +88,7 @@
     worker_pool_size => pos_integer(),
     process_step_timeout => timeout_sec(),
     task_scan_timeout => timeout_sec(),
+    call_scan_timeout => timeout_sec(),
     last_timer_repair => boolean()
 }.
 
@@ -196,12 +197,15 @@
 -type timestamp_ms() :: non_neg_integer().
 -type timestamp_sec() :: non_neg_integer().
 -type timeout_sec() :: non_neg_integer().
+-type timeout_ms() :: non_neg_integer().
 
 %%%
 %%% Constants
 %%%
 -define(DEFAULT_STEP_TIMEOUT_SEC, 60).
 
+-define(DEFAULT_CALL_SCAN_TIMEOUT_SEC, 3).
+
 -define(DEFAULT_RETRY_POLICY, #{
     %% second
     initial_timeout => 5,

src/prg_scanner.erl

@@ -30,9 +30,15 @@ start_link({NsId, _NsOpts} = NS) ->
     gen_server:start_link({local, RegName}, ?MODULE, NS, []).
 
 init(
-    {NsId, #{task_scan_timeout := RescanTimeoutSec, process_step_timeout := StepTimeoutSec} = Opts}
+    {NsId,
+        #{
+            task_scan_timeout := RescanTimeoutSec,
+            call_scan_timeout := CallRescanTimeoutSec,
+            process_step_timeout := StepTimeoutSec
+        } = Opts}
 ) ->
     RescanTimeoutMs = RescanTimeoutSec * 1000,
+    CallRescanTimeoutMs = CallRescanTimeoutSec * 1000,
     StepTimeoutMs = StepTimeoutSec * 1000,
     State = #prg_scanner_state{
         ns_id = NsId,
@@ -44,7 +50,7 @@ init(
         case maps:get(worker_pool_size, Opts) > 0 of
             true ->
                 _ = start_rescan_timers(RescanTimeoutMs),
-                _ = start_rescan_calls((RescanTimeoutMs div 3) + 100),
+                _ = start_rescan_calls(CallRescanTimeoutMs),
                 _ = start_zombie_collector(StepTimeoutMs);
             false ->
                 skip_scanning

src/prg_scheduler.erl

@@ -21,6 +21,9 @@
 -export([capture_worker/2]).
 -export([return_worker/3]).
 -export([release_worker/3]).
+-export([schedule_task/4]).
+
+%% Deprecated
 -export([continuation_task/3]).
 
 -record(prg_scheduler_state, {ns_id, ns_opts, ready, free_workers, owners, wrk_monitors}).
@@ -39,6 +42,7 @@ pop_task(NsId, Worker) ->
     RegName = prg_utils:registered_name(NsId, "_scheduler"),
     gen_server:call(RegName, {pop_task, Worker}, infinity).
 
+%% Deprecated
 -spec continuation_task(namespace_id(), pid(), task()) -> {task_header(), task()} | ok.
 continuation_task(NsId, Worker, Task) ->
     RegName = prg_utils:registered_name(NsId, "_scheduler"),
@@ -54,16 +58,23 @@ capture_worker(NsId, Owner) ->
     RegName = prg_utils:registered_name(NsId, "_scheduler"),
     gen_server:call(RegName, {capture_worker, Owner}, infinity).
 
+%% worker is alive and free
 -spec return_worker(namespace_id(), pid(), pid()) -> ok.
 return_worker(NsId, Owner, Worker) ->
     RegName = prg_utils:registered_name(NsId, "_scheduler"),
     gen_server:cast(RegName, {return_worker, Owner, Worker}).
 
+%% worker is alive and busy (processes task)
 -spec release_worker(namespace_id(), pid(), pid()) -> ok.
 release_worker(NsId, Owner, Pid) ->
     RegName = prg_utils:registered_name(NsId, "_scheduler"),
     gen_server:cast(RegName, {release_worker, Owner, Pid}).
 
+-spec schedule_task(namespace_id(), id(), task_id(), timeout_ms()) -> ok.
+schedule_task(NsId, ProcessId, TaskId, Timeout) ->
+    RegName = prg_utils:registered_name(NsId, "_scheduler"),
+    gen_server:cast(RegName, {schedule_task, ProcessId, TaskId, Timeout}).
+
 %%%===================================================================
 %%% Spawning and gen_server implementation
 %%%===================================================================
@@ -156,6 +167,12 @@ handle_cast(
                 State#prg_scheduler_state{owners = maps:without([Owner], Owners)}
         end,
     {noreply, NewState};
+handle_cast(
+    {schedule_task, ProcessId, TaskId, Timeout},
+    #prg_scheduler_state{} = State
+) ->
+    _TRef = erlang:start_timer(Timeout, self(), {process_scheduled_task, ProcessId, TaskId}),
+    {noreply, State};
 handle_cast(_Request, #prg_scheduler_state{} = State) ->
     {noreply, State}.
 
@@ -173,6 +190,19 @@ handle_info(
     MRef = erlang:monitor(process, NewWrk),
     NewWrkMonitors = maps:put(NewWrk, MRef, maps:without([Pid], WrkMonitors)),
     {noreply, State#prg_scheduler_state{wrk_monitors = NewWrkMonitors}};
+handle_info(
+    {timeout, _TRef, {process_scheduled_task, ProcessId, TaskId}},
+    #prg_scheduler_state{free_workers = FreeWorkers} = State
+) ->
+    NewState =
+        case queue:out(FreeWorkers) of
+            {{value, Worker}, NewWorkers} ->
+                ok = prg_worker:process_scheduled_task(Worker, ProcessId, TaskId),
+                State#prg_scheduler_state{free_workers = NewWorkers};
+            {empty, _} ->
+                State
+        end,
+    {noreply, NewState};
 handle_info(_Info, #prg_scheduler_state{} = State) ->
     {noreply, State}.
 

src/prg_storage.erl

@@ -25,6 +25,7 @@
 -export([complete_and_unlock/5]).
 -export([complete_and_error/4]).
 -export([remove_process/3]).
+-export([capture_task/3]).
 
 %% shared functions
 -export([get_task/3]).
@@ -147,6 +148,10 @@ complete_and_unlock(
 remove_process(#{client := Handler, options := HandlerOpts}, NsId, ProcessId) ->
     Handler:remove_process(HandlerOpts, NsId, ProcessId).
 
+-spec capture_task(storage_opts(), namespace_id(), task_id()) -> [task()].
+capture_task(#{client := Handler, options := HandlerOpts}, NsId, TaskId) ->
+    Handler:capture_task(HandlerOpts, NsId, TaskId).
+
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% Shared functions (recipient required)
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

src/prg_utils.erl

@@ -35,6 +35,7 @@ make_ns_opts(NsId, NsOpts) ->
         worker_pool_size => ?DEFAULT_WORKER_POOL_SIZE,
         process_step_timeout => ?DEFAULT_STEP_TIMEOUT_SEC,
         task_scan_timeout => (?DEFAULT_STEP_TIMEOUT_SEC div 2) + 1,
+        call_scan_timeout => ?DEFAULT_CALL_SCAN_TIMEOUT_SEC,
         last_timer_repair => false
     },
     ConfigDefaults = application:get_env(progressor, defaults, #{}),

src/prg_worker.erl

@@ -18,10 +18,12 @@
 -export([process_task/3]).
 -export([continuation_task/3]).
 -export([next_task/1]).
+-export([process_scheduled_task/3]).
 
 -record(prg_worker_state, {ns_id, ns_opts, process, sidecar_pid}).
 
 -define(DEFAULT_RANGE, #{direction => forward}).
+-define(CAPTURE_DEFENSE_INTERVAL_MS, 100).
 
 %%%
 %%% API
@@ -39,6 +41,10 @@ continuation_task(Worker, TaskHeader, Task) ->
 next_task(Worker) ->
     gen_server:cast(Worker, next_task).
 
+-spec process_scheduled_task(pid(), id(), task_id()) -> ok.
+process_scheduled_task(Worker, ProcessId, TaskId) ->
+    gen_server:cast(Worker, {process_scheduled_task, ProcessId, TaskId}).
+
 %%%===================================================================
 %%% Spawning and gen_server implementation
 %%%===================================================================
@@ -90,6 +96,32 @@ handle_cast(
     Deadline = erlang:system_time(millisecond) + TimeoutSec * 1000,
     NewState = do_process_task(TaskHeader, Task, Deadline, State),
     {noreply, NewState};
+handle_cast(
+    {process_scheduled_task, ProcessId, TaskId},
+    #prg_worker_state{
+        ns_id = NsId,
+        ns_opts = #{storage := StorageOpts, process_step_timeout := TimeoutSec} = _NsOpts,
+        sidecar_pid = Pid
+    } = State
+) ->
+    try prg_storage:capture_task(StorageOpts, NsId, TaskId) of
+        [] ->
+            %% task cancelled, blocked, already running or finished
+            ok = next_task(self()),
+            {noreply, State};
+        [#{status := <<"running">>} = Task] ->
+            Deadline = erlang:system_time(millisecond) + TimeoutSec * 1000,
+            HistoryRange = maps:get(range, maps:get(metadata, Task, #{}), #{}),
+            {ok, Process} = prg_worker_sidecar:get_process(Pid, Deadline, StorageOpts, NsId, ProcessId, HistoryRange),
+            TaskHeader = create_header(Task),
+            NewState = do_process_task(TaskHeader, Task, Deadline, State#prg_worker_state{process = Process}),
+            {noreply, NewState}
+    catch
+        Class:Term:Stacktrace ->
+            logger:error("process ~p. task capturing exception: ~p", [ProcessId, [Class, Term, Stacktrace]]),
+            ok = next_task(self()),
+            {noreply, State}
+    end;
 handle_cast(next_task, #prg_worker_state{sidecar_pid = CurrentPid}) ->
     %% kill sidecar and restart to clear memory
     true = erlang:unlink(CurrentPid),
@@ -235,7 +267,9 @@ success_and_continue(Intent, TaskHeader, Task, Deadline, State) ->
     ),
     _ = maybe_reply(TaskHeader, Response),
     case SaveResult of
-        {ok, []} ->
+        {ok, [#{status := <<"waiting">>, task_id := NextTaskId, scheduled_time := Ts} | _]} ->
+            RunAfterMs = (Ts - Now) * 1000 - ?CAPTURE_DEFENSE_INTERVAL_MS,
+            ok = prg_scheduler:schedule_task(NsId, ProcessId, NextTaskId, RunAfterMs),
             ok = next_task(self()),
             State#prg_worker_state{process = undefined};
         {ok, [ContinuationTask | _]} ->
@@ -383,6 +417,7 @@ success_and_unlock(Intent, TaskHeader, Task, Deadline, State) ->
         process = #{process_id := ProcessId, status := OldStatus} = Process,
         sidecar_pid = Pid
     } = State,
+    Now = erlang:system_time(second),
     {#{status := NewStatus} = ProcessUpdated, Updates} = update_process(Process, Intent),
     ok = prg_worker_sidecar:lifecycle_sink(
         Pid, Deadline, NsOpts, lifecycle_event(TaskHeader, OldStatus, NewStatus), ProcessId
@@ -409,7 +444,12 @@ success_and_unlock(Intent, TaskHeader, Task, Deadline, State) ->
         {ok, []} ->
             ok = next_task(self()),
             State#prg_worker_state{process = undefined};
-        {ok, [ContinuationTask | _]} ->
+        {ok, [#{status := <<"waiting">>, task_id := NextTaskId, scheduled_time := Ts} | _]} ->
+            RunAfterMs = (Ts - Now) * 1000 - ?CAPTURE_DEFENSE_INTERVAL_MS,
+            ok = prg_scheduler:schedule_task(NsId, ProcessId, NextTaskId, RunAfterMs),
+            ok = next_task(self()),
+            State#prg_worker_state{process = undefined};
+        {ok, [#{status := <<"running">>} = ContinuationTask | _]} ->
             NewHistory = maps:get(history, Process) ++ Events,
             ok = continuation_task(self(), create_header(ContinuationTask), ContinuationTask),
             State#prg_worker_state{
@@ -475,7 +515,17 @@ error_and_retry({error, Reason} = Response, TaskHeader, Task, Deadline, State) -
                 );
             NewTask ->
                 Updates = #{process_id => ProcessId},
-                {ok, _} = prg_worker_sidecar:complete_and_continue(
+                %% prg_storage guarantees that when saving a task with the error status,
+                %% all deferred tasks of all types will be completed with the canceled status,
+                %% so calling complete_and_continue is guaranteed to return the retrieval task,
+                %% and not any other deferred task
+                {ok, [
+                    #{
+                        status := <<"waiting">>,
+                        task_id := NextTaskId,
+                        scheduled_time := Ts
+                    }
+                ]} = prg_worker_sidecar:complete_and_continue(
                     Pid,
                     Deadline,
                     StorageOpts,
@@ -484,7 +534,10 @@ error_and_retry({error, Reason} = Response, TaskHeader, Task, Deadline, State) -
                     Updates,
                     [],
                     NewTask
-                )
+                ),
+                Now = erlang:system_time(second),
+                RunAfterMs = (Ts - Now) * 1000 - ?CAPTURE_DEFENSE_INTERVAL_MS,
+                ok = prg_scheduler:schedule_task(NsId, ProcessId, NextTaskId, RunAfterMs)
         end,
     ok = next_task(self()),
     State#prg_worker_state{process = undefined}.

src/progressor_app.erl

@@ -51,56 +51,56 @@ start_namespace(NsID, NsOpts) ->
 %% internal functions
 
 create_metrics() ->
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_calls_scanning_duration_ms},
         {help, "Calls (call, repair) scanning durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_timers_scanning_duration_ms},
         {help, "Timers (timeout, remove) scanning durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_zombie_collection_duration_ms},
         {help, "Zombie tasks collecting durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_request_preparing_duration_ms},
         {help, "Preparing request (init, call, repair) durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace, task_type]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_task_processing_duration_ms},
         {help, "Task processing durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace, task_type]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_task_completion_duration_ms},
         {help, "Task completion durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace, completion_type]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_process_removing_duration_ms},
         {help, "Task completion durations in millisecond"},
         {buckets, [50, 150, 300, 500, 750, 1000]},
         {labels, [prg_namespace]}
     ]),
 
-    _ = prometheus_histogram:new([
+    _ = prometheus_histogram:declare([
         {name, progressor_notification_duration_ms},
         {help, "Notification durations in millisecond"},
         {buckets, [10, 50, 150, 300, 500, 1000]},