From 07e39efbd90e7916be866f84984d116911366d49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Fri, 12 Jun 2026 00:32:37 +0300 Subject: [PATCH 1/6] Refactor action types in progressor.hrl to improve clarity and structure. Introduced distinct types for set_timer, remove, and scheduled_remove actions, ensuring mutually exclusive processor step actions. Updated processor_intent to allow action to be undefined. --- include/progressor.hrl | 12 +++++- src/progressor_action.erl | 90 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+), 2 deletions(-) create mode 100644 src/progressor_action.erl diff --git a/include/progressor.hrl b/include/progressor.hrl index 0a58de1..62b9ce7 100644 --- a/include/progressor.hrl +++ b/include/progressor.hrl @@ -157,7 +157,7 @@ -type processor_intent() :: #{ events := [event()], - action => action(), + action => action() | undefined, response => term(), aux_state => binary(), metadata => map() @@ -186,7 +186,15 @@ %% (i.e., attempts are not exhausted, and the error is not marked as %% non-retryable). --type action() :: #{set_timer := timestamp_sec(), remove => true} | unset_timer. +-type set_timer_action() :: #{set_timer := timestamp_sec()}. +-type remove_action() :: #{remove := true}. +-type scheduled_remove_action() :: #{set_timer := timestamp_sec(), remove := true}. +%% Mutually exclusive processor step actions (at most one per intent). +-type action() :: + set_timer_action() + | scheduled_remove_action() + | remove_action() + | unset_timer. -type task_result() :: #{ task_id := task_id(), diff --git a/src/progressor_action.erl b/src/progressor_action.erl new file mode 100644 index 0000000..c8d3daa --- /dev/null +++ b/src/progressor_action.erl @@ -0,0 +1,90 @@ +-module(progressor_action). + +-include("progressor.hrl"). + +-export([new/0]). +-export([instant/0]). +-export([set_timeout/1]). +-export([set_timeout/2]). +-export([set_deadline/1]). +-export([set_deadline/2]). +-export([set_timer/1]). +-export([set_timer/2]). +-export([unset_timer/0]). +-export([unset_timer/1]). +-export([remove/0]). +-export([remove/1]). +-export([mark_removal/0]). +-export([mark_removal/1]). +-export([marshal_timer/1]). + +-type seconds() :: timeout_sec(). +-type datetime() :: calendar:datetime() | binary(). +-type timer() :: {timeout, seconds()} | {deadline, datetime()}. +-type t() :: undefined | action(). + +-export_type([t/0, action/0, timer/0, seconds/0]). + +-spec new() -> t(). +new() -> + undefined. + +-spec instant() -> t(). +instant() -> + set_timeout(0). + +-spec set_timeout(seconds()) -> t(). +set_timeout(Seconds) -> + set_timeout(Seconds, new()). + +-spec set_timeout(seconds(), t()) -> set_timer_action(). +set_timeout(Seconds, _Action) when is_integer(Seconds), Seconds >= 0 -> + #{set_timer => marshal_timer({timeout, Seconds})}. + +-spec set_deadline(datetime()) -> t(). +set_deadline(Deadline) -> + set_deadline(Deadline, new()). + +-spec set_deadline(datetime(), t()) -> set_timer_action(). +set_deadline(Deadline, _Action) -> + #{set_timer => marshal_timer({deadline, Deadline})}. + +-spec set_timer(timer()) -> t(). +set_timer(Timer) -> + set_timer(Timer, new()). + +-spec set_timer(timer(), t()) -> set_timer_action(). +set_timer(Timer, _Action) -> + #{set_timer => marshal_timer(Timer)}. + +-spec unset_timer() -> unset_timer. +unset_timer() -> + 'unset_timer'. + +-spec unset_timer(t()) -> unset_timer. +unset_timer(_Action) -> + 'unset_timer'. + +-spec remove() -> remove_action(). +remove() -> + #{remove => true}. + +-spec remove(t()) -> remove_action(). +remove(_Action) -> + #{remove => true}. + +-spec mark_removal() -> remove_action(). +mark_removal() -> + remove(). + +-spec mark_removal(t()) -> remove_action(). +mark_removal(Action) -> + remove(Action). + +-spec marshal_timer(timer()) -> timestamp_sec(). +marshal_timer({timeout, Seconds}) when is_integer(Seconds), Seconds >= 0 -> + erlang:system_time(second) + Seconds; +marshal_timer({deadline, {_, _} = Dt}) -> + calendar:datetime_to_gregorian_seconds(Dt) - ?EPOCH_DIFF; +marshal_timer({deadline, Bin}) when is_binary(Bin) -> + calendar:rfc3339_to_system_time(unicode:characters_to_list(Bin), [{unit, second}]). From 4f6d78a01854b8e03612719be4418dd8402a090e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Fri, 12 Jun 2026 12:17:30 +0300 Subject: [PATCH 2/6] Refactor timer actions in progressor and related modules to use a unified scheduling approach. Updated action types to replace set_timer with timeout and introduced a schedule structure for better clarity. Adjusted processor logic and tests to accommodate these changes, ensuring consistent handling of scheduled actions. --- README.md | 4 +- .../base_bench/src/base_bench_processor.erl | 16 +- docs/step-effect-migration.md | 215 ++++++++++++++++++ include/progressor.hrl | 21 +- src/prg_echo_processor.erl | 2 +- src/prg_worker.erl | 40 ++-- src/progressor.erl | 30 ++- src/progressor_action.erl | 90 -------- test/prg_base_SUITE.erl | 44 ++-- 9 files changed, 301 insertions(+), 161 deletions(-) create mode 100644 docs/step-effect-migration.md delete mode 100644 src/progressor_action.erl diff --git a/README.md b/README.md index 78f511a..641d833 100644 --- a/README.md +++ b/README.md @@ -147,7 +147,7 @@ progressor:put(#{ status => <<"running">>, aux_state => <<"state_data">> }, - action => #{set_timer => 1640995200} + action => {schedule, #{at => 1640995200, action => timeout}} } }). ``` @@ -278,7 +278,7 @@ process({TaskType, Args, Process}, Options, Context) -> case should_set_timer(Process, TaskType) of true -> TimerTime = erlang:system_time(second) + 60, - {ok, Result#{action => #{set_timer => TimerTime}}}; + {ok, Result#{action => {schedule, #{at => TimerTime, action => timeout}}}}; false -> {ok, Result} end. diff --git a/benchmark/base_bench/src/base_bench_processor.erl b/benchmark/base_bench/src/base_bench_processor.erl index 71e3898..45353b1 100644 --- a/benchmark/base_bench/src/base_bench_processor.erl +++ b/benchmark/base_bench/src/base_bench_processor.erl @@ -7,19 +7,17 @@ process({init, Args, _Process}, _Opts, _Ctx) -> Result = #{ metadata => #{finish => Fin}, events => [event(1)], - action => #{set_timer => erlang:system_time(second)} + action => timeout }, {ok, Result}; %% process({timeout, _Args, #{history := History, metadata := Meta} = _Process}, _Opts, _Ctx) -> - %Random = rand:uniform(40), - %timer:sleep(60 + Random), #{finish := FinishTime} = Meta, - Action = case FinishTime > erlang:system_time(second) of - true -> #{set_timer => erlang:system_time(second)}; - false -> unset_timer - end, - %Action = #{set_timer => erlang:system_time(second)}, + Action = + case FinishTime > erlang:system_time(second) of + true -> timeout; + false -> suspend + end, NextId = erlang:length(History) + 1, Result = #{ events => [event(NextId)], @@ -31,7 +29,7 @@ process({call, _Args, #{history := History} = _Process}, _Opts, _Ctx) -> Result = #{ response => erlang:length(History), events => [], - action => unset_timer + action => suspend }, {ok, Result}. %% diff --git a/docs/step-effect-migration.md b/docs/step-effect-migration.md new file mode 100644 index 0000000..42c18cc --- /dev/null +++ b/docs/step-effect-migration.md @@ -0,0 +1,215 @@ +# Миграция action (progressor) + +План доработки progressor: убрать размытую legacy-семантику (`#{set_timer => ...}`, +`unset_timer`, пересечение ключей map) **изнутри runtime** и оставить только явную +алгебру `action()` из `progressor.hrl`. + +Связанный план для hellgate: `hellgate/docs/step-effect-hg-migration.md` (wire `action()`, адаптер в HG). + +--- + +## Принцип + +**Progressor — runtime, не слой совместимости.** + +- Внутри репозитория progressor после миграции **нет** map/atom legacy-action, **нет** + `normalize/1`, **нет** dual-field (`action` + `effect`), **нет** pattern match по + `#{set_timer := _}`. +- Процессор отдаёт в intent только `action()` из `progressor.hrl`. +- Отсутствие поля `action` в intent = `idle` (единственное допустимое «пустое» значение). +- Конвертация MG/map legacy — **на границе у потребителей** (hellgate / `prg_machine`), + не в `prg_worker`. + +Старый план с choke point `normalize(maps:get(action, ...))` **отменён**: он оставлял +старую семантику внутри progressor навсегда. + +--- + +## Проблема (сейчас) + +| Исход | Legacy в intent | Проблема | +|-------|-----------------|----------| +| unlock | `undefined` | неявно | +| suspend | `unset_timer` | atom, не в типе | +| continue сейчас | `#{set_timer => now}` | map + timestamp | +| continue позже | `#{set_timer => Ts}` | тот же map, другой смысл | +| remove сейчас | `#{remove => true}` | пересечение ключей | +| remove позже | `#{set_timer => Ts, remove => true}` | порядок клауз в `prg_worker` | + +`progressor_action` выдаёт те же map/atom — не решает проблему внутри runtime. + +--- + +## Целевое состояние + +### Wire-тип `action()` (`progressor.hrl`) + +```erlang +-type scheduled_action() :: timeout | remove. + +-type schedule() :: #{ + at := timestamp_sec(), %% абсолютный unix sec + action := scheduled_action() %% вид отложенной задачи +}. + +-type action() :: + idle + | suspend + | scheduled_action() %% timeout | remove на top-level + | {schedule, schedule()}. +``` + +Top-level `timeout` = «продолжить по timeout-задаче сразу» (legacy instant / timer 0). + +Wire-значения пишутся в intent как есть, без helper-модуля: + +```erlang +idle | suspend | timeout | remove +{schedule, #{at := UnixSec, action := timeout | remove}} +``` + +`at` — всегда абсолютный unix sec; относительное время — `erlang:system_time(second) + N` +на стороне автора. + +### Таблица dispatch (единственный источник правды в runtime) + +| `action()` | Worker path | `task_type` | +|------------|-------------|-------------| +| `idle` (или поле отсутствует) | `success_and_unlock` | — | +| `suspend` | `success_and_suspend` | — | +| `remove` | `success_and_remove` | — | +| `timeout` | `success_and_continue` | `<<"timeout">>` (scheduled_time = now) | +| `{schedule, #{action := timeout, at := Ts}}` | `success_and_continue` | `<<"timeout">>` | +| `{schedule, #{action := remove, at := Ts}}` | `success_and_continue` | `<<"remove">>` | + +Один `dispatch_action/5` — без чувствительного порядка клауз. + +### `processor_intent()` + +```erlang +-type processor_intent() :: #{ + events := [event()], + action => action(), %% отсутствие = idle + response => term(), + aux_state => binary(), + metadata => map() +}. +``` + +Старые map/atom на поле `action` после релиза невалидны. + +--- + +## Граница с потребителями + +``` +┌──────────────────────────────────────┐ +│ hellgate / ff / кастомный процессор │ доменная логика +│ hg_machine_action (опционально) │ legacy MG → action() ТОЛЬКО здесь +└──────────────┬───────────────────────┘ + │ processor_intent.action :: action() +┌──────────────▼───────────────────────┐ +│ progressor (runtime) │ dispatch_action, action_to_task +│ никаких map/atom legacy │ +└──────────────────────────────────────┘ +``` + +До обновления hellgate: `prg_machine:marshal_intent` конвертирует старые map в `action()` +**в репозитории hellgate**, не в progressor. + +--- + +## Фаза 0. Контракт + +Зафиксировать: + +- таблицу dispatch выше; +- top-level `timeout` = instant continue (не путать с `task_type`); +- `at` в schedule — только абсолютный unix sec; +- **breaking change**: tag `vX.Y.0`, старые map/atom в intent не поддерживаются; +- список файлов progressor с legacy (grep: `set_timer`, `unset_timer`, `#{remove`). + +**Критерий:** ревью контракта + согласование с hellgate по порядку релизов. + +--- + +## Фаза 1. Типы в `progressor.hrl` + +1. `scheduled_action/0`, `schedule/0`, `action/0` — wire-алгебра. +2. `processor_intent()` — `action => action()`. + +**Критерий:** компилируется, dialyzer зелёный. + +--- + +## Фаза 2. Runtime — чистый cut + +Одним проходом, без transitional choke point: + +1. **`prg_worker`** + - `handle_result_success/5` → `dispatch_action(action(), ...)`. + - Удалить case по `#{set_timer}`, `unset_timer`, `#{remove := true}`. + - `action_to_task_type/1` — по `scheduled_action()` (`timeout | remove`). + +2. **`progressor.erl`** + - `action_to_task/3` принимает только `action()`; absent → `idle` через `maps:get/3`. + +3. **Удалить** `src/progressor_action.erl`. + +**Критерий:** в `src/` нет `set_timer`, `unset_timer`, `#{remove =>` (кроме комментариев/миграций БД). + +--- + +## Фаза 3. Dogfooding и CT + +В том же PR / сразу после фазы 2 — **не откладывать**: + +1. `prg_echo_processor`, `benchmark/base_bench_processor` → wire `action()`. +2. Все моки в `prg_base_SUITE` → wire `action()` (не raw maps). +3. README / примеры процессора. + +**Критерий:** `rebar3 ct` зелёный; grep по `test/` и `src/` не находит legacy action maps. + +--- + +## Фаза 4. Релиз и потребители + +1. CHANGELOG: breaking — формат `processor_intent.action`. +2. Migration guide **для внешних авторов**: таблица legacy → `action()` (в доке hellgate). +3. Tag `vX.Y.0`. +4. Hellgate: bump tag, адаптер в `prg_machine` / `hg_machine_action`, затем домены. + +**Критерий:** progressor tag опубликован; hellgate компилируется со своим адаптером. + +--- + +## Порядок + +``` +Фаза 0 → 1 → 2 + 3 (один PR) → 4 +``` + +--- + +## Не делать + +- `normalize/1` / dual `action`+`effect` **внутри progressor**. +- Отдельный модуль-обёртка только ради типов — типы в `progressor.hrl`. +- «Зелёный CT без смены моков» — откладывает legacy внутри репозитория. +- Deprecated map-типы в `progressor.hrl` «на несколько фаз». +- `{set_timer, #{...}}` как wire-формат — переносит путаницу в кортежи. +- Authoring-типы (`{timeout, N}`) в `processor_intent` — только `at` / `timeout` / `{schedule, ...}`. +- Доменный аккумулятор hellgate (`set_timeout(0, Action)` по шагам) в progressor. + +--- + +## Чеклист «миграция завершена» + +- [x] типы `action/0`, `schedule/0` в `progressor.hrl` +- [x] `prg_worker` — только `dispatch_action/5` по `action()` +- [x] `progressor.erl` — только `action()` в `action_to_task` +- [x] `progressor_action.erl` удалён +- [x] grep `set_timer|unset_timer` в `src/` и `test/` — пусто +- [x] CT зелёный (`make wdeps-test`) +- [ ] CHANGELOG + tag +- [ ] hellgate на отдельном треке с адаптером на своей границе diff --git a/include/progressor.hrl b/include/progressor.hrl index 62b9ce7..ca535df 100644 --- a/include/progressor.hrl +++ b/include/progressor.hrl @@ -157,7 +157,7 @@ -type processor_intent() :: #{ events := [event()], - action => action() | undefined, + action => action(), %% отсутствие ключа = idle response => term(), aux_state => binary(), metadata => map() @@ -186,15 +186,18 @@ %% (i.e., attempts are not exhausted, and the error is not marked as %% non-retryable). --type set_timer_action() :: #{set_timer := timestamp_sec()}. --type remove_action() :: #{remove := true}. --type scheduled_remove_action() :: #{set_timer := timestamp_sec(), remove := true}. -%% Mutually exclusive processor step actions (at most one per intent). +-type scheduled_action() :: timeout | remove. + +-type schedule() :: #{ + at := timestamp_sec(), + action := scheduled_action() +}. + -type action() :: - set_timer_action() - | scheduled_remove_action() - | remove_action() - | unset_timer. + idle + | suspend + | scheduled_action() + | {schedule, schedule()}. -type task_result() :: #{ task_id := task_id(), diff --git a/src/prg_echo_processor.erl b/src/prg_echo_processor.erl index d96f9af..c8429fd 100644 --- a/src/prg_echo_processor.erl +++ b/src/prg_echo_processor.erl @@ -13,7 +13,7 @@ process({_, _, #{history := History} = _Process}, _Opts, _Ctx) -> Count -> Result = #{ events => [event(Count + 1)], - action => #{set_timer => erlang:system_time(second)} + action => timeout }, {ok, Result} end. diff --git a/src/prg_worker.erl b/src/prg_worker.erl index 8939522..9c2cb99 100644 --- a/src/prg_worker.erl +++ b/src/prg_worker.erl @@ -229,17 +229,21 @@ maybe_restore_history(_, State) -> State. handle_result_success(Intent, TaskHeader, Task, Deadline, State) -> - Action = maps:get(action, Intent, undefined), - case Action of - #{set_timer := _Timestamp} -> - success_and_continue(Intent, TaskHeader, Task, Deadline, State); - #{remove := true} -> - success_and_remove(Intent, TaskHeader, Task, Deadline, State); - unset_timer -> - success_and_suspend(Intent, TaskHeader, Task, Deadline, State); - undefined -> - success_and_unlock(Intent, TaskHeader, Task, Deadline, State) - end. + Action = maps:get(action, Intent, idle), + dispatch_action(Action, Intent, TaskHeader, Task, Deadline, State). + +dispatch_action(idle, Intent, TaskHeader, Task, Deadline, State) -> + success_and_unlock(Intent, TaskHeader, Task, Deadline, State); +dispatch_action(suspend, Intent, TaskHeader, Task, Deadline, State) -> + success_and_suspend(Intent, TaskHeader, Task, Deadline, State); +dispatch_action(remove, Intent, TaskHeader, Task, Deadline, State) -> + success_and_remove(Intent, TaskHeader, Task, Deadline, State); +dispatch_action(timeout, Intent, TaskHeader, Task, Deadline, State) -> + success_and_continue( + Intent, TaskHeader, Task, Deadline, State, timeout, erlang:system_time(second) + ); +dispatch_action({schedule, #{at := Timestamp0, action := Action}}, Intent, TaskHeader, Task, Deadline, State) -> + success_and_continue(Intent, TaskHeader, Task, Deadline, State, Action, Timestamp0). handle_result_error(Result, {TaskType, _} = TaskHeader, Task, Deadline, State) when TaskType =:= timeout; @@ -253,8 +257,8 @@ handle_result_error(Result, {TaskType, _} = TaskHeader, Task, Deadline, State) w -> error_and_stop(Result, TaskHeader, Task, Deadline, State). -success_and_continue(Intent, TaskHeader, Task, Deadline, State) -> - #{action := #{set_timer := Timestamp0} = Action, events := Events} = Intent, +success_and_continue(Intent, TaskHeader, Task, Deadline, State, Action, Timestamp0) -> + #{events := Events} = Intent, #{context := Context} = Task, #prg_worker_state{ ns_id = NsId, @@ -323,7 +327,7 @@ success_and_remove(Intent, TaskHeader, _Task, Deadline, State) -> State#prg_worker_state{process = undefined}. success_and_suspend(Intent, TaskHeader, Task, Deadline, State) -> - #{events := Events, action := unset_timer} = Intent, + #{events := Events} = Intent, #prg_worker_state{ ns_id = NsId, ns_opts = #{storage := StorageOpts} = NsOpts, @@ -747,10 +751,10 @@ create_header(#{task_type := <<"repair">>}) -> create_header(#{task_type := <<"notify">>}) -> {notify, undefined}. %% -action_to_task_type(#{remove := true}) -> - <<"remove">>; -action_to_task_type(#{set_timer := _}) -> - <<"timeout">>. +action_to_task_type(timeout) -> + <<"timeout">>; +action_to_task_type(remove) -> + <<"remove">>. last_event_id([]) -> 0; diff --git a/src/progressor.erl b/src/progressor.erl index 0351b10..79a6197 100644 --- a/src/progressor.erl +++ b/src/progressor.erl @@ -373,7 +373,7 @@ do_put( #{ process_id := ProcessId } = Process, - Action = maps:get(action, Args, undefined), + Action = maps:get(action, Args, idle), Context = maps:get(context, Opts, <<>>), Now = erlang:system_time(microsecond), InitTask = #{ @@ -513,27 +513,35 @@ check_for_run(undefined) -> check_for_run(Pid) when is_pid(Pid) -> <<"running">>. -action_to_task(undefined, _ProcessId, _Ctx) -> +action_to_task(idle, _ProcessId, _Ctx) -> undefined; -action_to_task(unset_timer, _ProcessId, _Ctx) -> +action_to_task(suspend, _ProcessId, _Ctx) -> undefined; -action_to_task(#{set_timer := Timestamp} = Action, ProcessId, Context) -> - TaskType = - case maps:get(remove, Action, false) of - true -> <<"remove">>; - false -> <<"timeout">> - end, +action_to_task(remove, _ProcessId, _Ctx) -> + undefined; +action_to_task(timeout, ProcessId, Context) -> + action_to_task( + {schedule, #{at => erlang:system_time(second), action => timeout}}, + ProcessId, + Context + ); +action_to_task({schedule, #{at := Timestamp0, action := Action}}, ProcessId, Context) -> #{ process_id => ProcessId, - task_type => TaskType, + task_type => action_to_task_type(Action), status => <<"waiting">>, args => <<>>, context => Context, - scheduled_time => prg_utils:to_microseconds(Timestamp), + scheduled_time => prg_utils:to_microseconds(Timestamp0), last_retry_interval => 0, attempts_count => 0 }. +action_to_task_type(timeout) -> + <<"timeout">>; +action_to_task_type(remove) -> + <<"remove">>. + maybe_add_key(undefined, _Key, Map) -> Map; maybe_add_key(Value, Key, Map) -> diff --git a/src/progressor_action.erl b/src/progressor_action.erl deleted file mode 100644 index c8d3daa..0000000 --- a/src/progressor_action.erl +++ /dev/null @@ -1,90 +0,0 @@ --module(progressor_action). - --include("progressor.hrl"). - --export([new/0]). --export([instant/0]). --export([set_timeout/1]). --export([set_timeout/2]). --export([set_deadline/1]). --export([set_deadline/2]). --export([set_timer/1]). --export([set_timer/2]). --export([unset_timer/0]). --export([unset_timer/1]). --export([remove/0]). --export([remove/1]). --export([mark_removal/0]). --export([mark_removal/1]). --export([marshal_timer/1]). - --type seconds() :: timeout_sec(). --type datetime() :: calendar:datetime() | binary(). --type timer() :: {timeout, seconds()} | {deadline, datetime()}. --type t() :: undefined | action(). - --export_type([t/0, action/0, timer/0, seconds/0]). - --spec new() -> t(). -new() -> - undefined. - --spec instant() -> t(). -instant() -> - set_timeout(0). - --spec set_timeout(seconds()) -> t(). -set_timeout(Seconds) -> - set_timeout(Seconds, new()). - --spec set_timeout(seconds(), t()) -> set_timer_action(). -set_timeout(Seconds, _Action) when is_integer(Seconds), Seconds >= 0 -> - #{set_timer => marshal_timer({timeout, Seconds})}. - --spec set_deadline(datetime()) -> t(). -set_deadline(Deadline) -> - set_deadline(Deadline, new()). - --spec set_deadline(datetime(), t()) -> set_timer_action(). -set_deadline(Deadline, _Action) -> - #{set_timer => marshal_timer({deadline, Deadline})}. - --spec set_timer(timer()) -> t(). -set_timer(Timer) -> - set_timer(Timer, new()). - --spec set_timer(timer(), t()) -> set_timer_action(). -set_timer(Timer, _Action) -> - #{set_timer => marshal_timer(Timer)}. - --spec unset_timer() -> unset_timer. -unset_timer() -> - 'unset_timer'. - --spec unset_timer(t()) -> unset_timer. -unset_timer(_Action) -> - 'unset_timer'. - --spec remove() -> remove_action(). -remove() -> - #{remove => true}. - --spec remove(t()) -> remove_action(). -remove(_Action) -> - #{remove => true}. - --spec mark_removal() -> remove_action(). -mark_removal() -> - remove(). - --spec mark_removal(t()) -> remove_action(). -mark_removal(Action) -> - remove(Action). - --spec marshal_timer(timer()) -> timestamp_sec(). -marshal_timer({timeout, Seconds}) when is_integer(Seconds), Seconds >= 0 -> - erlang:system_time(second) + Seconds; -marshal_timer({deadline, {_, _} = Dt}) -> - calendar:datetime_to_gregorian_seconds(Dt) - ?EPOCH_DIFF; -marshal_timer({deadline, Bin}) when is_binary(Bin) -> - calendar:rfc3339_to_system_time(unicode:characters_to_list(Bin), [{unit, second}]). diff --git a/test/prg_base_SUITE.erl b/test/prg_base_SUITE.erl index 8d30359..b423495 100644 --- a/test/prg_base_SUITE.erl +++ b/test/prg_base_SUITE.erl @@ -39,6 +39,8 @@ -define(NS(C), proplists:get_value(ns_id, C, 'default/default')). -define(AWAIT_TIMEOUT(C), proplists:get_value(repl_timeout, C, 5)). +-define(TIMEOUT_IN(Sec), {schedule, #{at => erlang:system_time(second) + (Sec), action => timeout}}). +-define(REMOVE_IN(Sec), {schedule, #{at => erlang:system_time(second) + (Sec), action => remove}}). init_per_suite(Config) -> Config. @@ -844,7 +846,7 @@ put_process_with_timeout_test(C) -> status => <<"running">>, history => [event(1)] }, - action => #{set_timer => erlang:system_time(microsecond) + 1000000} + action => ?TIMEOUT_IN(1) }, {ok, ok} = progressor:put(#{ns => ?NS(C), id => Id, args => Args}), timer:sleep(?AWAIT_TIMEOUT(C)), @@ -942,7 +944,7 @@ put_process_with_remove_test(C) -> status => <<"running">>, history => [event(1)] }, - action => #{set_timer => erlang:system_time(microsecond) + 1000000, remove => true} + action => ?REMOVE_IN(1) }, {ok, ok} = progressor:put(#{ns => ?NS(C), id => Id, args => Args}), timer:sleep(?AWAIT_TIMEOUT(C)), @@ -987,7 +989,7 @@ mock_processor(simple_timers_test = TestCase) -> events => [event(1)], metadata => #{<<"k">> => <<"v">>}, %% postponed timer - action => #{set_timer => erlang:system_time(microsecond) + 2000000}, + action => ?TIMEOUT_IN(2), aux_state => erlang:term_to_binary(<<"aux_state1">>) }, Self ! 1, @@ -996,7 +998,7 @@ mock_processor(simple_timers_test = TestCase) -> Result = #{ events => [event(2)], %% continuation timer - action => #{set_timer => erlang:system_time(microsecond)}, + action => timeout, aux_state => erlang:term_to_binary(<<"aux_state2">>) }, Self ! 2, @@ -1017,7 +1019,7 @@ mock_processor(simple_call_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000} + action => ?TIMEOUT_IN(2) }, Self ! 1, {ok, Result}; @@ -1047,7 +1049,7 @@ mock_processor(reschedule_after_call_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000} + action => ?TIMEOUT_IN(2) }, Self ! 1, {ok, Result}; @@ -1106,7 +1108,7 @@ mock_processor(simple_call_with_range_test = TestCase) -> Result = #{ response => <<"response">>, events => [event(6)], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 3, {ok, Result}; @@ -1127,7 +1129,7 @@ mock_processor(call_replace_timer_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000, remove => true} + action => ?REMOVE_IN(2) }, Self ! 1, {ok, Result}; @@ -1136,7 +1138,7 @@ mock_processor(call_replace_timer_test = TestCase) -> Result = #{ response => <<"response">>, events => [event(2), event(3)], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 2, {ok, Result}; @@ -1157,7 +1159,7 @@ mock_processor(call_unset_timer_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000} + action => ?TIMEOUT_IN(2) }, Self ! 1, {ok, Result}; @@ -1166,7 +1168,7 @@ mock_processor(call_unset_timer_test = TestCase) -> Result = #{ response => <<"response">>, events => [], - action => unset_timer + action => suspend }, Self ! 2, {ok, Result}; @@ -1187,7 +1189,7 @@ mock_processor(postponed_call_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 1, {ok, Result}; @@ -1195,7 +1197,7 @@ mock_processor(postponed_call_test = TestCase) -> timer:sleep(3000), Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 2, {ok, Result}; @@ -1223,7 +1225,7 @@ mock_processor(postponed_call_to_suspended_process_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 1, {ok, Result}; @@ -1271,7 +1273,7 @@ mock_processor(simple_repair_after_non_retriable_error_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 1, {ok, Result}; @@ -1282,7 +1284,7 @@ mock_processor(simple_repair_after_non_retriable_error_test = TestCase) -> %% timeout via simple repair Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 3, {ok, Result}; @@ -1302,7 +1304,7 @@ mock_processor(repair_after_non_retriable_error_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 1, {ok, Result}; @@ -1331,7 +1333,7 @@ mock_processor(error_after_max_retries_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{set_timer => erlang:system_time(microsecond)} + action => timeout }, Self ! 1, {ok, Result}; @@ -1391,7 +1393,7 @@ mock_processor(remove_by_timer_test = TestCase) -> MockProcessor = fun({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1), event(2)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000, remove => true} + action => ?REMOVE_IN(2) }, {ok, Result} end, @@ -1403,14 +1405,14 @@ mock_processor(remove_without_timer_test = TestCase) -> ({init, <<"init_args">>, _Process}, _Opts, _Ctx) -> Result = #{ events => [event(1)], - action => #{set_timer => erlang:system_time(microsecond) + 2000000} + action => ?TIMEOUT_IN(2) }, Self ! 1, {ok, Result}; ({timeout, <<>>, _Process}, _Opts, _Ctx) -> Result = #{ events => [], - action => #{remove => true} + action => remove }, Self ! 2, {ok, Result} From f54a33e871dc53679833ceab688bd559c9b2460e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Fri, 12 Jun 2026 15:00:06 +0300 Subject: [PATCH 3/6] Update migration documentation for hellgate and refine processor_intent structure. Changed references to the migration guide and clarified action types in progressor.hrl, ensuring consistency in the documentation and code structure. --- docs/step-effect-migration.md | 8 ++++---- include/progressor.hrl | 3 ++- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/docs/step-effect-migration.md b/docs/step-effect-migration.md index 42c18cc..eb78f78 100644 --- a/docs/step-effect-migration.md +++ b/docs/step-effect-migration.md @@ -4,7 +4,7 @@ `unset_timer`, пересечение ключей map) **изнутри runtime** и оставить только явную алгебру `action()` из `progressor.hrl`. -Связанный план для hellgate: `hellgate/docs/step-effect-hg-migration.md` (wire `action()`, адаптер в HG). +Связанный документ для hellgate: `hellgate/docs/prg-machine.md`. --- @@ -105,7 +105,7 @@ idle | suspend | timeout | remove ``` ┌──────────────────────────────────────┐ │ hellgate / ff / кастомный процессор │ доменная логика -│ hg_machine_action (опционально) │ legacy MG → action() ТОЛЬКО здесь +│ prg_action (hellgate, опционально) │ timer tuple → action(); MG/repair — граница └──────────────┬───────────────────────┘ │ processor_intent.action :: action() ┌──────────────▼───────────────────────┐ @@ -177,7 +177,7 @@ idle | suspend | timeout | remove 1. CHANGELOG: breaking — формат `processor_intent.action`. 2. Migration guide **для внешних авторов**: таблица legacy → `action()` (в доке hellgate). 3. Tag `vX.Y.0`. -4. Hellgate: bump tag, адаптер в `prg_machine` / `hg_machine_action`, затем домены. +4. Hellgate: bump tag, `prg_action` + wire в доменах (миграция завершена, см. hellgate `docs/prg-machine.md`). **Критерий:** progressor tag опубликован; hellgate компилируется со своим адаптером. @@ -212,4 +212,4 @@ idle | suspend | timeout | remove - [x] grep `set_timer|unset_timer` в `src/` и `test/` — пусто - [x] CT зелёный (`make wdeps-test`) - [ ] CHANGELOG + tag -- [ ] hellgate на отдельном треке с адаптером на своей границе +- [x] hellgate: wire `action()`, `prg_action`, CI green (до tag bump) diff --git a/include/progressor.hrl b/include/progressor.hrl index ca535df..31e4e0b 100644 --- a/include/progressor.hrl +++ b/include/progressor.hrl @@ -157,7 +157,8 @@ -type processor_intent() :: #{ events := [event()], - action => action(), %% отсутствие ключа = idle + %% отсутствие ключа = idle + action => action(), response => term(), aux_state => binary(), metadata => map() From af3c8c8c9b4939d36062a46467d59773ad578478 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Fri, 12 Jun 2026 21:18:03 +0300 Subject: [PATCH 4/6] Update timestamp handling in progressor.hrl to use microsecond precision. The timestamp is now stored as timestamptz, allowing both seconds and microseconds for input, enhancing flexibility in time representation. --- include/progressor.hrl | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/include/progressor.hrl b/include/progressor.hrl index 31e4e0b..733bdf1 100644 --- a/include/progressor.hrl +++ b/include/progressor.hrl @@ -50,7 +50,10 @@ process_id := id(), task_id := task_id(), event_id := event_id(), - timestamp := timestamp_sec(), + %% Stored as timestamptz with microsecond precision; the unit is auto-detected + %% (prg_utils:split_timestamp/to_microseconds), so both seconds and microseconds + %% are accepted on write. + timestamp := timestamp_us(), metadata => #{format => pos_integer()}, payload := binary() }. From 8f18b309279f0401283e8a18a0166825a8717980 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Sat, 13 Jun 2026 02:12:53 +0300 Subject: [PATCH 5/6] Update timestamp handling across modules to use microsecond precision. Adjusted scheduling logic in progressor and related files to ensure consistent handling of timestamps, enhancing the accuracy of scheduled actions. Updated documentation to reflect changes in timestamp representation from seconds to microseconds. --- README.md | 4 ++-- docs/step-effect-migration.md | 10 +++++----- include/progressor.hrl | 3 ++- src/prg_worker.erl | 11 ++++------- src/progressor.erl | 2 +- test/prg_base_SUITE.erl | 4 ++-- 6 files changed, 16 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 641d833..047671d 100644 --- a/README.md +++ b/README.md @@ -147,7 +147,7 @@ progressor:put(#{ status => <<"running">>, aux_state => <<"state_data">> }, - action => {schedule, #{at => 1640995200, action => timeout}} + action => {schedule, #{at => 1640995200000000, action => timeout}} } }). ``` @@ -277,7 +277,7 @@ process({TaskType, Args, Process}, Options, Context) -> % Опционально установить таймер case should_set_timer(Process, TaskType) of true -> - TimerTime = erlang:system_time(second) + 60, + TimerTime = erlang:system_time(microsecond) + 60 * 1000000, {ok, Result#{action => {schedule, #{at => TimerTime, action => timeout}}}}; false -> {ok, Result} diff --git a/docs/step-effect-migration.md b/docs/step-effect-migration.md index eb78f78..33b809a 100644 --- a/docs/step-effect-migration.md +++ b/docs/step-effect-migration.md @@ -48,7 +48,7 @@ -type scheduled_action() :: timeout | remove. -type schedule() :: #{ - at := timestamp_sec(), %% абсолютный unix sec + at := timestamp_us(), %% абсолютный unix us action := scheduled_action() %% вид отложенной задачи }. @@ -65,11 +65,11 @@ Wire-значения пишутся в intent как есть, без helper-м ```erlang idle | suspend | timeout | remove -{schedule, #{at := UnixSec, action := timeout | remove}} +{schedule, #{at := UnixUs, action := timeout | remove}} ``` -`at` — всегда абсолютный unix sec; относительное время — `erlang:system_time(second) + N` -на стороне автора. +`at` — абсолютный unix us; относительное время — `erlang:system_time(microsecond) + N * 1000000` +на стороне автора. `prg_utils:to_microseconds/1` на входе runtime по-прежнему принимает sec/ms/us. ### Таблица dispatch (единственный источник правды в runtime) @@ -125,7 +125,7 @@ idle | suspend | timeout | remove - таблицу dispatch выше; - top-level `timeout` = instant continue (не путать с `task_type`); -- `at` в schedule — только абсолютный unix sec; +- `at` в schedule — абсолютный unix us; - **breaking change**: tag `vX.Y.0`, старые map/atom в intent не поддерживаются; - список файлов progressor с legacy (grep: `set_timer`, `unset_timer`, `#{remove`). diff --git a/include/progressor.hrl b/include/progressor.hrl index 733bdf1..7b9c3d7 100644 --- a/include/progressor.hrl +++ b/include/progressor.hrl @@ -193,7 +193,8 @@ -type scheduled_action() :: timeout | remove. -type schedule() :: #{ - at := timestamp_sec(), + %% Absolute unix time; unit is auto-detected on write (prg_utils:to_microseconds). + at := timestamp_us(), action := scheduled_action() }. diff --git a/src/prg_worker.erl b/src/prg_worker.erl index 9c2cb99..ff3b2d2 100644 --- a/src/prg_worker.erl +++ b/src/prg_worker.erl @@ -240,7 +240,7 @@ dispatch_action(remove, Intent, TaskHeader, Task, Deadline, State) -> success_and_remove(Intent, TaskHeader, Task, Deadline, State); dispatch_action(timeout, Intent, TaskHeader, Task, Deadline, State) -> success_and_continue( - Intent, TaskHeader, Task, Deadline, State, timeout, erlang:system_time(second) + Intent, TaskHeader, Task, Deadline, State, timeout, erlang:system_time(microsecond) ); dispatch_action({schedule, #{at := Timestamp0, action := Action}}, Intent, TaskHeader, Task, Deadline, State) -> success_and_continue(Intent, TaskHeader, Task, Deadline, State, Action, Timestamp0). @@ -722,12 +722,9 @@ is_retryable(Error, {timeout, undefined}, RetryPolicy, Timeout, Attempts) -> is_retryable(_Error, _TaskHeader, _RetryPolicy, _Timeout, _Attempts) -> false. -%% Due to the difference in the time scales used for storage (microseconds) -%% and the schedule time (seconds), the following logic is required: -%% - If the difference between the schedule and the current time is less than a ~1 second -%% the task is assigned the status "running" and is processed immediately -%% - If the difference between the schedule and the current time exceeds ~1 second -%% the task is assigned the status "waiting" and is saved to the schedule +%% Sub-second schedules are coerced to immediate execution: if the gap to +%% `scheduled_time` is below ~1s (scheduler overhead), the task stays `running` +%% instead of `waiting`. create_status(Timestamp, Now) when Timestamp =< Now -> <<"running">>; create_status(Timestamp, Now) -> diff --git a/src/progressor.erl b/src/progressor.erl index 79a6197..ece81a6 100644 --- a/src/progressor.erl +++ b/src/progressor.erl @@ -521,7 +521,7 @@ action_to_task(remove, _ProcessId, _Ctx) -> undefined; action_to_task(timeout, ProcessId, Context) -> action_to_task( - {schedule, #{at => erlang:system_time(second), action => timeout}}, + {schedule, #{at => erlang:system_time(microsecond), action => timeout}}, ProcessId, Context ); diff --git a/test/prg_base_SUITE.erl b/test/prg_base_SUITE.erl index b423495..42c5cb2 100644 --- a/test/prg_base_SUITE.erl +++ b/test/prg_base_SUITE.erl @@ -39,8 +39,8 @@ -define(NS(C), proplists:get_value(ns_id, C, 'default/default')). -define(AWAIT_TIMEOUT(C), proplists:get_value(repl_timeout, C, 5)). --define(TIMEOUT_IN(Sec), {schedule, #{at => erlang:system_time(second) + (Sec), action => timeout}}). --define(REMOVE_IN(Sec), {schedule, #{at => erlang:system_time(second) + (Sec), action => remove}}). +-define(TIMEOUT_IN(Sec), {schedule, #{at => erlang:system_time(microsecond) + (Sec) * 1000000, action => timeout}}). +-define(REMOVE_IN(Sec), {schedule, #{at => erlang:system_time(microsecond) + (Sec) * 1000000, action => remove}}). init_per_suite(Config) -> Config. From c42706e40eb0ef94d97f062206c0aa43f948c56f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D1=80=D1=82=D0=B5=D0=BC?= Date: Tue, 23 Jun 2026 12:12:28 +0300 Subject: [PATCH 6/6] Refactor action type handling in progressor and related modules. Moved action_to_task_type function to prg_utils for better modularity and consistency. Updated references in prg_worker and progressor to utilize the new utility function, ensuring a unified approach to task type determination. Removed legacy action type definitions from progressor. This change supports the ongoing migration to a cleaner action handling structure. --- docs/step-effect-migration.md | 215 ---------------------------------- src/prg_utils.erl | 7 ++ src/prg_worker.erl | 7 +- src/progressor.erl | 7 +- 4 files changed, 9 insertions(+), 227 deletions(-) delete mode 100644 docs/step-effect-migration.md diff --git a/docs/step-effect-migration.md b/docs/step-effect-migration.md deleted file mode 100644 index 33b809a..0000000 --- a/docs/step-effect-migration.md +++ /dev/null @@ -1,215 +0,0 @@ -# Миграция action (progressor) - -План доработки progressor: убрать размытую legacy-семантику (`#{set_timer => ...}`, -`unset_timer`, пересечение ключей map) **изнутри runtime** и оставить только явную -алгебру `action()` из `progressor.hrl`. - -Связанный документ для hellgate: `hellgate/docs/prg-machine.md`. - ---- - -## Принцип - -**Progressor — runtime, не слой совместимости.** - -- Внутри репозитория progressor после миграции **нет** map/atom legacy-action, **нет** - `normalize/1`, **нет** dual-field (`action` + `effect`), **нет** pattern match по - `#{set_timer := _}`. -- Процессор отдаёт в intent только `action()` из `progressor.hrl`. -- Отсутствие поля `action` в intent = `idle` (единственное допустимое «пустое» значение). -- Конвертация MG/map legacy — **на границе у потребителей** (hellgate / `prg_machine`), - не в `prg_worker`. - -Старый план с choke point `normalize(maps:get(action, ...))` **отменён**: он оставлял -старую семантику внутри progressor навсегда. - ---- - -## Проблема (сейчас) - -| Исход | Legacy в intent | Проблема | -|-------|-----------------|----------| -| unlock | `undefined` | неявно | -| suspend | `unset_timer` | atom, не в типе | -| continue сейчас | `#{set_timer => now}` | map + timestamp | -| continue позже | `#{set_timer => Ts}` | тот же map, другой смысл | -| remove сейчас | `#{remove => true}` | пересечение ключей | -| remove позже | `#{set_timer => Ts, remove => true}` | порядок клауз в `prg_worker` | - -`progressor_action` выдаёт те же map/atom — не решает проблему внутри runtime. - ---- - -## Целевое состояние - -### Wire-тип `action()` (`progressor.hrl`) - -```erlang --type scheduled_action() :: timeout | remove. - --type schedule() :: #{ - at := timestamp_us(), %% абсолютный unix us - action := scheduled_action() %% вид отложенной задачи -}. - --type action() :: - idle - | suspend - | scheduled_action() %% timeout | remove на top-level - | {schedule, schedule()}. -``` - -Top-level `timeout` = «продолжить по timeout-задаче сразу» (legacy instant / timer 0). - -Wire-значения пишутся в intent как есть, без helper-модуля: - -```erlang -idle | suspend | timeout | remove -{schedule, #{at := UnixUs, action := timeout | remove}} -``` - -`at` — абсолютный unix us; относительное время — `erlang:system_time(microsecond) + N * 1000000` -на стороне автора. `prg_utils:to_microseconds/1` на входе runtime по-прежнему принимает sec/ms/us. - -### Таблица dispatch (единственный источник правды в runtime) - -| `action()` | Worker path | `task_type` | -|------------|-------------|-------------| -| `idle` (или поле отсутствует) | `success_and_unlock` | — | -| `suspend` | `success_and_suspend` | — | -| `remove` | `success_and_remove` | — | -| `timeout` | `success_and_continue` | `<<"timeout">>` (scheduled_time = now) | -| `{schedule, #{action := timeout, at := Ts}}` | `success_and_continue` | `<<"timeout">>` | -| `{schedule, #{action := remove, at := Ts}}` | `success_and_continue` | `<<"remove">>` | - -Один `dispatch_action/5` — без чувствительного порядка клауз. - -### `processor_intent()` - -```erlang --type processor_intent() :: #{ - events := [event()], - action => action(), %% отсутствие = idle - response => term(), - aux_state => binary(), - metadata => map() -}. -``` - -Старые map/atom на поле `action` после релиза невалидны. - ---- - -## Граница с потребителями - -``` -┌──────────────────────────────────────┐ -│ hellgate / ff / кастомный процессор │ доменная логика -│ prg_action (hellgate, опционально) │ timer tuple → action(); MG/repair — граница -└──────────────┬───────────────────────┘ - │ processor_intent.action :: action() -┌──────────────▼───────────────────────┐ -│ progressor (runtime) │ dispatch_action, action_to_task -│ никаких map/atom legacy │ -└──────────────────────────────────────┘ -``` - -До обновления hellgate: `prg_machine:marshal_intent` конвертирует старые map в `action()` -**в репозитории hellgate**, не в progressor. - ---- - -## Фаза 0. Контракт - -Зафиксировать: - -- таблицу dispatch выше; -- top-level `timeout` = instant continue (не путать с `task_type`); -- `at` в schedule — абсолютный unix us; -- **breaking change**: tag `vX.Y.0`, старые map/atom в intent не поддерживаются; -- список файлов progressor с legacy (grep: `set_timer`, `unset_timer`, `#{remove`). - -**Критерий:** ревью контракта + согласование с hellgate по порядку релизов. - ---- - -## Фаза 1. Типы в `progressor.hrl` - -1. `scheduled_action/0`, `schedule/0`, `action/0` — wire-алгебра. -2. `processor_intent()` — `action => action()`. - -**Критерий:** компилируется, dialyzer зелёный. - ---- - -## Фаза 2. Runtime — чистый cut - -Одним проходом, без transitional choke point: - -1. **`prg_worker`** - - `handle_result_success/5` → `dispatch_action(action(), ...)`. - - Удалить case по `#{set_timer}`, `unset_timer`, `#{remove := true}`. - - `action_to_task_type/1` — по `scheduled_action()` (`timeout | remove`). - -2. **`progressor.erl`** - - `action_to_task/3` принимает только `action()`; absent → `idle` через `maps:get/3`. - -3. **Удалить** `src/progressor_action.erl`. - -**Критерий:** в `src/` нет `set_timer`, `unset_timer`, `#{remove =>` (кроме комментариев/миграций БД). - ---- - -## Фаза 3. Dogfooding и CT - -В том же PR / сразу после фазы 2 — **не откладывать**: - -1. `prg_echo_processor`, `benchmark/base_bench_processor` → wire `action()`. -2. Все моки в `prg_base_SUITE` → wire `action()` (не raw maps). -3. README / примеры процессора. - -**Критерий:** `rebar3 ct` зелёный; grep по `test/` и `src/` не находит legacy action maps. - ---- - -## Фаза 4. Релиз и потребители - -1. CHANGELOG: breaking — формат `processor_intent.action`. -2. Migration guide **для внешних авторов**: таблица legacy → `action()` (в доке hellgate). -3. Tag `vX.Y.0`. -4. Hellgate: bump tag, `prg_action` + wire в доменах (миграция завершена, см. hellgate `docs/prg-machine.md`). - -**Критерий:** progressor tag опубликован; hellgate компилируется со своим адаптером. - ---- - -## Порядок - -``` -Фаза 0 → 1 → 2 + 3 (один PR) → 4 -``` - ---- - -## Не делать - -- `normalize/1` / dual `action`+`effect` **внутри progressor**. -- Отдельный модуль-обёртка только ради типов — типы в `progressor.hrl`. -- «Зелёный CT без смены моков» — откладывает legacy внутри репозитория. -- Deprecated map-типы в `progressor.hrl` «на несколько фаз». -- `{set_timer, #{...}}` как wire-формат — переносит путаницу в кортежи. -- Authoring-типы (`{timeout, N}`) в `processor_intent` — только `at` / `timeout` / `{schedule, ...}`. -- Доменный аккумулятор hellgate (`set_timeout(0, Action)` по шагам) в progressor. - ---- - -## Чеклист «миграция завершена» - -- [x] типы `action/0`, `schedule/0` в `progressor.hrl` -- [x] `prg_worker` — только `dispatch_action/5` по `action()` -- [x] `progressor.erl` — только `action()` в `action_to_task` -- [x] `progressor_action.erl` удалён -- [x] grep `set_timer|unset_timer` в `src/` и `test/` — пусто -- [x] CT зелёный (`make wdeps-test`) -- [ ] CHANGELOG + tag -- [x] hellgate: wire `action()`, `prg_action`, CI green (до tag bump) diff --git a/src/prg_utils.erl b/src/prg_utils.erl index f8d9ea0..782d169 100644 --- a/src/prg_utils.erl +++ b/src/prg_utils.erl @@ -18,6 +18,7 @@ -export([to_seconds/1]). -export([split_timestamp/1]). -export([format_microseconds/1]). +-export([action_to_task_type/1]). -type time_unit() :: second | millisecond | microsecond. @@ -141,3 +142,9 @@ format_microseconds(Val) -> Bin = integer_to_binary(Val), Pad = 6 - byte_size(Bin), <<(binary:copy(<<"0">>, Pad))/binary, Bin/binary>>. + +-spec action_to_task_type(scheduled_action()) -> task_type(). +action_to_task_type(timeout) -> + <<"timeout">>; +action_to_task_type(remove) -> + <<"remove">>. diff --git a/src/prg_worker.erl b/src/prg_worker.erl index ff3b2d2..7a665d6 100644 --- a/src/prg_worker.erl +++ b/src/prg_worker.erl @@ -273,7 +273,7 @@ success_and_continue(Intent, TaskHeader, Task, Deadline, State, Action, Timestam TaskResult = task_result(Task, <<"finished">>, Response), NewTask = #{ process_id => ProcessId, - task_type => action_to_task_type(Action), + task_type => prg_utils:action_to_task_type(Action), status => create_status(Timestamp, Now), scheduled_time => Timestamp, context => Context, @@ -747,11 +747,6 @@ create_header(#{task_type := <<"repair">>}) -> {repair, undefined}; create_header(#{task_type := <<"notify">>}) -> {notify, undefined}. -%% -action_to_task_type(timeout) -> - <<"timeout">>; -action_to_task_type(remove) -> - <<"remove">>. last_event_id([]) -> 0; diff --git a/src/progressor.erl b/src/progressor.erl index ece81a6..723317e 100644 --- a/src/progressor.erl +++ b/src/progressor.erl @@ -528,7 +528,7 @@ action_to_task(timeout, ProcessId, Context) -> action_to_task({schedule, #{at := Timestamp0, action := Action}}, ProcessId, Context) -> #{ process_id => ProcessId, - task_type => action_to_task_type(Action), + task_type => prg_utils:action_to_task_type(Action), status => <<"waiting">>, args => <<>>, context => Context, @@ -537,11 +537,6 @@ action_to_task({schedule, #{at := Timestamp0, action := Action}}, ProcessId, Con attempts_count => 0 }. -action_to_task_type(timeout) -> - <<"timeout">>; -action_to_task_type(remove) -> - <<"remove">>. - maybe_add_key(undefined, _Key, Map) -> Map; maybe_add_key(Value, Key, Map) ->