fix zombie collection

.env

@@ -2,4 +2,4 @@ SERVICE_NAME=progressor
 OTP_VERSION=27.1.2
 REBAR_VERSION=3.24
 THRIFT_VERSION=0.14.2.3
-CONFLUENT_PLATFORM_VERSION=5.1.2
+CONFLUENT_PLATFORM_VERSION=7.2.15

docker-compose.yml

@@ -30,9 +30,6 @@ services:
       POSTGRES_DB: "progressor_db"
       POSTGRES_USER: "progressor"
       POSTGRES_PASSWORD: "progressor"
-      PGDATA: "/tmp/postgresql/data/pgdata"
-    volumes:
-      - progressor-data:/tmp/postgresql/data
     ports:
       - "5432:5432"
     healthcheck:
@@ -64,7 +61,7 @@ services:
     depends_on:
       - zookeeper
     healthcheck:
-      test: ["CMD", "kafka-topics", "--list", "--zookeeper", "zookeeper:2181"]
+      test: ["CMD", "kafka-topics", "--list", "--bootstrap-server", "localhost:9092"]
       interval: 5s
       timeout: 10s
       retries: 5
@@ -86,6 +83,3 @@ services:
       KAFKA_BROKER_ID: 3
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka3:9092
-
-volumes:
-  progressor-data:

src/prg_utils.erl

@@ -7,6 +7,7 @@
 -export([pipe/2]).
 -export([format/1]).
 -export([make_ns_opts/2]).
+-export([unixtime_to_datetime/1]).
 -export([with_observe/3]).
 -export([with_observe/4]).
 
@@ -40,6 +41,10 @@ make_ns_opts(NsId, NsOpts) ->
     Defaults = maps:merge(PresetDefaults, ConfigDefaults),
     maps:merge(Defaults, NsOpts).
 
+-spec unixtime_to_datetime(timestamp_sec()) -> calendar:datetime().
+unixtime_to_datetime(TimestampSec) ->
+    calendar:gregorian_seconds_to_datetime(TimestampSec + ?EPOCH_DIFF).
+
 -spec with_observe(_Fun, atom(), [list() | binary()]) -> any().
 with_observe(Fun, MetricKey, Labels) ->
     with_observe(Fun, histogram, MetricKey, Labels).

src/storage/postgres/prg_pg_backend.erl

@@ -218,25 +218,29 @@ collect_zombies(PgOpts, NsId, Timeout) ->
     #{
         processes := ProcessesTable,
         tasks := TaskTable,
+        schedule := ScheduleTable,
         running := RunningTable
     } = prg_pg_utils:tables(NsId),
     NowSec = erlang:system_time(second),
     Now = unixtime_to_datetime(NowSec),
     TsBackward = unixtime_to_datetime(NowSec - (Timeout + ?PROTECT_TIMEOUT)),
-    epg_pool:transaction(
+    {ok, _, _} = epg_pool:transaction(
         Pool,
         fun(Connection) ->
-            {ok, _, _} = epg_pool:query(
+            epg_pool:query(
                 Connection,
                 "WITH zombie_tasks as ("
                 "  DELETE FROM " ++ RunningTable ++
                     " WHERE running_time < $1 "
                     "    RETURNING process_id, task_id"
                     "  ), "
                     "  t1 AS (UPDATE " ++ TaskTable ++
-                    " SET status = 'cancelled' WHERE process_id IN (SELECT process_id FROM zombie_tasks))"
+                    " SET status = 'cancelled' WHERE status IN ('waiting', 'blocked') "
+                    "    AND process_id IN (SELECT process_id FROM zombie_tasks)),"
                     "  t2 AS (UPDATE " ++ TaskTable ++
-                    " SET status = 'error', finished_time = $2 WHERE task_id IN (SELECT task_id FROM zombie_tasks))"
+                    " SET status = 'error', finished_time = $2 WHERE task_id IN (SELECT task_id FROM zombie_tasks)),"
+                    "  t3 AS (DELETE FROM " ++ ScheduleTable ++
+                    " WHERE process_id IN (SELECT process_id FROM zombie_tasks))"
                     "MERGE INTO " ++ ProcessesTable ++
                     " AS pt USING zombie_tasks AS zt ON pt.process_id = zt.process_id "
                     "  WHEN MATCHED THEN UPDATE SET"

test/prg_base_SUITE.erl

@@ -28,6 +28,7 @@
 -export([remove_by_timer_test/1]).
 -export([remove_without_timer_test/1]).
 -export([put_process_test/1]).
+-export([put_process_zombie_test/1]).
 -export([put_process_with_timeout_test/1]).
 -export([put_process_with_remove_test/1]).
 
@@ -72,6 +73,7 @@ groups() ->
             remove_by_timer_test,
             remove_without_timer_test,
             put_process_test,
+            put_process_zombie_test,
             put_process_with_timeout_test,
             put_process_with_remove_test
         ]},
@@ -682,6 +684,73 @@ put_process_with_timeout_test(C) ->
     unmock_processor(),
     ok.
 %%
+-spec put_process_zombie_test(_) -> _.
+put_process_zombie_test(C) ->
+    %% steps:
+    %% 1. put     -> [event1], timer 1s
+    %% 2. insert running task from past
+    %% 3. zombie collecttion
+    Id = gen_id(),
+    Args = #{
+        process => #{
+            process_id => Id,
+            status => <<"running">>,
+            history => [event(1)]
+        }
+    },
+    {ok, ok} = progressor:put(#{ns => ?NS(C), id => Id, args => Args}),
+    Now = erlang:system_time(second),
+    ZombieTs = prg_utils:unixtime_to_datetime(Now - 30),
+    NS = erlang:atom_to_list(?NS(C)),
+    %% TODO: rework it via storage backend
+    %% START SQL INJECTION
+    {ok, _, _, [{TaskId}]} = epg_pool:query(
+        default_pool,
+        "INSERT INTO \"" ++ NS ++
+            "_tasks\" "
+            "  (process_id, task_type, status, scheduled_time, running_time, args, last_retry_interval, attempts_count)"
+            "  VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING task_id",
+        [
+            Id,
+            <<"timeout">>,
+            <<"running">>,
+            ZombieTs,
+            ZombieTs,
+            <<>>,
+            0,
+            0
+        ]
+    ),
+    {ok, 1} = epg_pool:query(
+        default_pool,
+        "INSERT INTO \"" ++ NS ++
+            "_running\" "
+            "  (task_id, process_id, task_type, status, scheduled_time, running_time, args, "
+            "   last_retry_interval, attempts_count)"
+            "  VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
+        [
+            TaskId,
+            Id,
+            <<"timeout">>,
+            <<"running">>,
+            ZombieTs,
+            ZombieTs,
+            <<>>,
+            0,
+            0
+        ]
+    ),
+    %% END SQL INJECTION
+
+    %% await zombie collection (process step timeout (10s) + random part (2s))
+    timer:sleep(12010),
+    {ok, #{
+        process_id := Id,
+        status := <<"error">>,
+        detail := <<"zombie detected">>
+    }} = progressor:get(#{ns => ?NS(C), id => Id}),
+    ok.
+%%
 -spec put_process_with_remove_test(_) -> _.
 put_process_with_remove_test(C) ->
     %% steps: