fix simple repair on devel (#38)

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

Author: ttt161

elvis.config

@@ -27,6 +27,7 @@
                     {elvis_style, no_spec_with_records},
                     {elvis_style, no_debug_call, #{}},
                     {elvis_style, export_used_types, disable},
+                    {elvis_style, god_modules, disable},
                     %% FIXME Maybe refactor code blocks
                     {elvis_style, dont_repeat_yourself, #{
                         ignore => [

include/progressor.hrl

@@ -130,7 +130,7 @@
 
 -type history_range() :: #{
     offset => non_neg_integer(),
-    limit => pos_integer(),
+    limit => non_neg_integer(),
     direction => forward | backward
 }.
 

src/prg_storage.erl

@@ -12,6 +12,7 @@
 -export([put_process_data/4]).
 -export([process_trace/3]).
 -export([get_process_with_initialization/4]).
+-export([repair_process/3]).
 
 %% scan functions
 -export([search_timers/4]).
@@ -90,6 +91,10 @@ prepare_call(#{client := Handler, options := HandlerOpts}, NsId, ProcessId, Task
 prepare_repair(#{client := Handler, options := HandlerOpts}, NsId, ProcessId, RepairTask) ->
     Handler:prepare_repair(HandlerOpts, NsId, ProcessId, RepairTask).
 
+-spec repair_process(storage_opts(), namespace_id(), id()) -> ok | no_return().
+repair_process(#{client := Handler, options := HandlerOpts}, NsId, ProcessId) ->
+    Handler:repair_process(HandlerOpts, NsId, ProcessId).
+
 %%%%%%%%%%%%%%%%%
 %% Scan functions
 %%%%%%%%%%%%%%%%%

src/prg_worker.erl

@@ -171,14 +171,15 @@ handle_result(
     #prg_worker_state{
         ns_id = NsId,
         ns_opts = #{storage := StorageOpts} = NsOpts,
-        process = #{process_id := ProcessId} = Process,
+        process = #{process_id := ProcessId, status := OldStatus} = Process,
         sidecar_pid = Pid
     } = State
 ) ->
     Now = erlang:system_time(second),
-    ProcessUpdated = update_process(
-        maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
-    ),
+    #{status := NewStatus} =
+        ProcessUpdated = update_process(
+            maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
+        ),
     Response = response(maps:get(response, Result, undefined)),
     TaskResult = #{
         task_id => TaskId,
@@ -196,7 +197,7 @@ handle_result(
         attempts_count => 0
     },
     ok = prg_worker_sidecar:lifecycle_sink(
-        Pid, Deadline, NsOpts, extract_task_type(TaskHeader), ProcessId
+        Pid, Deadline, NsOpts, lifecycle_event(TaskHeader, OldStatus, NewStatus), ProcessId
     ),
     ok = prg_worker_sidecar:event_sink(Pid, Deadline, NsOpts, ProcessId, Events),
     SaveResult = prg_worker_sidecar:complete_and_continue(
@@ -249,17 +250,18 @@ handle_result(
     #prg_worker_state{
         ns_id = NsId,
         ns_opts = #{storage := StorageOpts} = NsOpts,
-        process = #{process_id := ProcessId} = Process,
+        process = #{process_id := ProcessId, status := OldStatus} = Process,
         sidecar_pid = Pid
     } = State
 ) ->
+    #{status := NewStatus} =
+        ProcessUpdated = update_process(
+            maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
+        ),
     ok = prg_worker_sidecar:lifecycle_sink(
-        Pid, Deadline, NsOpts, extract_task_type(TaskHeader), ProcessId
+        Pid, Deadline, NsOpts, lifecycle_event(TaskHeader, OldStatus, NewStatus), ProcessId
     ),
     ok = prg_worker_sidecar:event_sink(Pid, Deadline, NsOpts, ProcessId, Events),
-    ProcessUpdated = update_process(
-        maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
-    ),
     Response = response(maps:get(response, Result, undefined)),
     TaskResult = #{
         task_id => TaskId,
@@ -303,7 +305,7 @@ handle_result(
 ) ->
     Now = erlang:system_time(second),
     ok = prg_worker_sidecar:lifecycle_sink(
-        Pid, Deadline, NsOpts, extract_task_type(TaskHeader), ProcessId
+        Pid, Deadline, NsOpts, repair, ProcessId
     ),
     ok = prg_worker_sidecar:event_sink(Pid, Deadline, NsOpts, ProcessId, Events),
     ProcessUpdated = update_process(
@@ -368,17 +370,18 @@ handle_result(
     #prg_worker_state{
         ns_id = NsId,
         ns_opts = #{storage := StorageOpts} = NsOpts,
-        process = #{process_id := ProcessId} = Process,
+        process = #{process_id := ProcessId, status := OldStatus} = Process,
         sidecar_pid = Pid
     } = State
 ) ->
+    #{status := NewStatus} =
+        ProcessUpdated = update_process(
+            maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
+        ),
     ok = prg_worker_sidecar:lifecycle_sink(
-        Pid, Deadline, NsOpts, extract_task_type(TaskHeader), ProcessId
+        Pid, Deadline, NsOpts, lifecycle_event(TaskHeader, OldStatus, NewStatus), ProcessId
     ),
     ok = prg_worker_sidecar:event_sink(Pid, Deadline, NsOpts, ProcessId, Events),
-    ProcessUpdated = update_process(
-        maps:without([detail, corrupted_by], Process#{status => <<"running">>}), Result
-    ),
     Response = response(maps:get(response, Result, undefined)),
     TaskResult = #{
         task_id => TaskId,
@@ -601,3 +604,10 @@ last_event_id([]) ->
 last_event_id(History) ->
     [#{event_id := Id} | _] = lists:reverse(History),
     Id.
+
+lifecycle_event({timeout, _}, <<"error">>, <<"running">>) ->
+    repair;
+lifecycle_event({timeout, _}, _, _) ->
+    timeout;
+lifecycle_event({TaskType, _}, _, _) ->
+    TaskType.

src/progressor.erl

@@ -88,12 +88,11 @@ simple_repair(Req) ->
         [
             fun add_ns_opts/1,
             fun check_idempotency/1,
-            fun(Data) -> check_process_status(Data, <<"error">>) end,
-            fun add_task/1,
+            fun check_process_continuation/1,
             fun(Data) -> prepare_postponed(fun prg_storage:prepare_call/4, Data) end,
-            fun(_Data) -> {ok, ok} end
+            fun do_simple_repair/1
         ],
-        Req#{type => timeout}
+        Req
     ).
 
 -spec get(request()) -> {ok, _Result} | {error, _Reason}.
@@ -205,6 +204,33 @@ check_process_status(
         {error, _} = Error -> Error
     end.
 
+check_process_continuation(
+    #{ns_opts := #{storage := StorageOpts}, id := Id, ns := NsId} = Opts
+) ->
+    case prg_storage:get_process(external, StorageOpts, NsId, Id, #{limit => 0}) of
+        {ok, #{status := <<"running">>}} ->
+            {error, <<"process is running">>};
+        {ok, #{status := <<"error">>, corrupted_by := TaskId}} when is_integer(TaskId) ->
+            case prg_storage:get_task(external, StorageOpts, NsId, TaskId) of
+                {ok, #{task_type := Type}} when Type =:= <<"timeout">>; Type =:= <<"remove">> ->
+                    add_task(Opts#{type => Type});
+                {ok, _} ->
+                    Opts
+            end;
+        {ok, #{status := <<"error">>}} ->
+            Opts;
+        {error, _} = Error ->
+            Error
+    end.
+
+do_simple_repair(#{task := _T}) ->
+    %% process will repaired via timeout task
+    {ok, ok};
+do_simple_repair(#{ns_opts := #{storage := StorageOpts} = NsOpts, id := Id, ns := NsId}) ->
+    ok = prg_storage:repair_process(StorageOpts, NsId, Id),
+    ok = prg_notifier:lifecycle_sink(NsOpts, repair, Id),
+    {ok, ok}.
+
 prepare(
     Fun,
     #{ns_opts := #{storage := StorageOpts} = NsOpts, ns := NsId, id := ProcessId, task := Task} =
@@ -251,7 +277,10 @@ prepare_postponed(
             Req#{task => Task#{task_id => TaskId}};
         {error, _} = Error ->
             Error
-    end.
+    end;
+prepare_postponed(_Fun, Req) ->
+    %% Req without task, skip this step
+    Req.
 
 get_task_result(#{
     ns_opts := #{storage := StorageOpts} = NsOpts, ns := NsId, idempotency_key := IdempotencyKey
@@ -377,7 +406,9 @@ convert_task_type(notify) ->
 convert_task_type(timeout) ->
     <<"timeout">>;
 convert_task_type(repair) ->
-    <<"repair">>.
+    <<"repair">>;
+convert_task_type(Type) when is_binary(Type) ->
+    Type.
 
 maybe_add_idempotency(Task, undefined) ->
     Task;

src/storage/postgres/prg_pg_backend.erl

@@ -15,6 +15,7 @@
 -export([put_process_data/4]).
 -export([process_trace/3]).
 -export([get_process_with_initialization/4]).
+-export([repair_process/3]).
 
 %% scan functions
 -export([collect_zombies/3]).
@@ -163,6 +164,19 @@ get_process_with_initialization(PgOpts, NsId, ProcessId, HistoryRange) ->
     ),
     parse_process_info(RawResult, HistoryRange).
 
+-spec repair_process(pg_opts(), namespace_id(), id()) -> ok | no_return().
+repair_process(PgOpts, NsId, ProcessId) ->
+    Pool = get_pool(external, PgOpts),
+    #{processes := ProcessesTable} = prg_pg_utils:tables(NsId),
+    {ok, 1} = epg_pool:query(
+        Pool,
+        "UPDATE " ++ ProcessesTable ++
+            " SET status = 'running', corrupted_by = null, detail = null "
+            " WHERE process_id = $1",
+        [ProcessId]
+    ),
+    ok.
+
 -spec put_process_data(
     pg_opts(),
     namespace_id(),

test/prg_base_SUITE.erl

@@ -22,6 +22,7 @@
 -export([postponed_call_to_suspended_process_test/1]).
 -export([multiple_calls_test/1]).
 -export([simple_repair_after_non_retriable_error_test/1]).
+-export([simple_repair_after_call_error_test/1]).
 -export([repair_after_non_retriable_error_test/1]).
 -export([error_after_max_retries_test/1]).
 -export([repair_after_call_error_test/1]).
@@ -69,6 +70,7 @@ groups() ->
             postponed_call_to_suspended_process_test,
             multiple_calls_test,
             simple_repair_after_non_retriable_error_test,
+            simple_repair_after_call_error_test,
             repair_after_non_retriable_error_test,
             error_after_max_retries_test,
             repair_after_call_error_test,
@@ -487,6 +489,36 @@ simple_repair_after_non_retriable_error_test(C) ->
     unmock_processor(),
     ok.
 
+-spec simple_repair_after_call_error_test(_) -> _.
+simple_repair_after_call_error_test(C) ->
+    %% steps:
+    %% 1. init ->    [],       undefined
+    %% 2. call ->    {error, retry_this}
+    %% 3. simple_repair
+    _ = mock_processor(simple_repair_after_call_error_test),
+    Id = gen_id(),
+    {ok, ok} = progressor:init(#{ns => ?NS(C), id => Id, args => <<"init_args">>}),
+    {error, retry_this} = progressor:call(#{ns => ?NS(C), id => Id, args => <<"call_args">>}),
+    2 = expect_steps_counter(2),
+    timer:sleep(?AWAIT_TIMEOUT(C)),
+    {ok, #{
+        detail := <<"retry_this">>,
+        metadata := #{<<"k">> := <<"v">>},
+        history := [],
+        process_id := Id,
+        status := <<"error">>
+    }} = progressor:get(#{ns => ?NS(C), id => Id}),
+    {ok, ok} = progressor:simple_repair(#{ns => ?NS(C), id => Id}),
+    timer:sleep(?AWAIT_TIMEOUT(C)),
+    {ok, #{
+        metadata := #{<<"k">> := <<"v">>},
+        history := [],
+        process_id := Id,
+        status := <<"running">>
+    }} = progressor:get(#{ns => ?NS(C), id => Id}),
+    unmock_processor(),
+    ok.
+
 -spec repair_after_non_retriable_error_test(_) -> _.
 repair_after_non_retriable_error_test(C) ->
     %% steps:
@@ -1144,6 +1176,23 @@ mock_processor(simple_repair_after_non_retriable_error_test = TestCase) ->
     end,
     mock_processor(TestCase, MockProcessor);
 %%
+mock_processor(simple_repair_after_call_error_test = TestCase) ->
+    Self = self(),
+    MockProcessor = fun
+        ({init, <<"init_args">>, _Process}, _Opts, _Ctx) ->
+            Result = #{
+                metadata => #{<<"k">> => <<"v">>},
+                events => []
+            },
+            Self ! 1,
+            {ok, Result};
+        ({call, <<"call_args">>, #{history := []} = _Process}, _Opts, _Ctx) ->
+            Self ! 2,
+            %% retriable error for call must be ignore and process set error status
+            {error, retry_this}
+    end,
+    mock_processor(TestCase, MockProcessor);
+%%
 mock_processor(repair_after_non_retriable_error_test = TestCase) ->
     Self = self(),
     MockProcessor = fun