What happens when worker gets terminated mid job processing?

[mikz]

I'm facing issues where a bunch of jobs are getting stuck in started state and never recover.
Supabase can kill workers after two seconds of wall time. What happens if worker gets killed when the job is already in started state?

Codex went through the codebase and it claims this is a blind spot - introspecting the state of the db, logs and the source.
Is there a test to validate this? For anything serious this is quite important use case.

[jumski]

Hey @mikz, first off - thank you for reporting this! This is a significant issue and I appreciate you taking the time to report it.

I apologize for the comment spam earlier - I was thinking out loud as I investigated. I've consolidated everything into this single response.

What's Happening

This is a confirmed bug in pgflow's message handling. When a worker gets killed mid-processing (common in serverless environments), tasks get stuck in 'started' status while messages keep cycling in the queue indefinitely - which explains the 300+ attempts you're seeing.

I've reproduced this on our demo environment and are actively working on a fix. This is a top priority for me.

Diagnostic Queries

Could you run these and paste me the results to help me understand your specific case and confirm my understanding?
I already confirmed it on https://demo.pgflow.dev and if your case is similar I am 100% confident in this bug and the fix I plan to implement.

Important: Replace your_flow_slug with your actual flow name in queries 2, 4, and 5.

-- 1. Stuck 'started' tasks
SELECT flow_slug, step_slug, status, attempts_count,
       started_at, now() - started_at as stuck_for
FROM pgflow.step_tasks
WHERE status = 'started'
  AND started_at < now() - interval '5 minutes'
LIMIT 20;

-- 2. Messages cycling without tasks being started
-- (messages read multiple times but task still 'queued' or 'started')
SELECT
  m.msg_id,
  m.read_ct,
  m.vt,
  m.message->>'run_id' as run_id,
  m.message->>'step_slug' as step_slug,
  st.status as task_status,
  st.started_at,
  st.attempts_count,
  r.status as run_status
FROM pgmq.q_your_flow_slug m
LEFT JOIN pgflow.step_tasks st ON st.message_id = m.msg_id
LEFT JOIN pgflow.runs r ON r.run_id = st.run_id
WHERE m.read_ct > 1
ORDER BY m.read_ct DESC
LIMIT 20;

-- 3. Messages for failed runs still in queue (orphan type 1)
SELECT
  st.flow_slug,
  st.run_id,
  st.step_slug,
  st.message_id,
  st.status as task_status,
  r.status as run_status,
  r.failed_at
FROM pgflow.step_tasks st
JOIN pgflow.runs r ON r.run_id = st.run_id
WHERE r.status = 'failed'
  AND st.status IN ('queued', 'started')
  AND st.message_id IS NOT NULL;

-- 4. Check archived vs active messages ratio
SELECT 'active' as type, count(*) FROM pgmq.q_your_flow_slug
UNION ALL
SELECT 'archived' as type, count(*) FROM pgmq.a_your_flow_slug;

-- 5. High read_ct messages (sign of cycling orphans)
SELECT
  msg_id,
  read_ct,
  vt,
  enqueued_at,
  message
FROM pgmq.q_your_flow_slug
WHERE read_ct > 3
ORDER BY read_ct DESC
LIMIT 20;

Join Discord?

Would you be up for joining our Discord for some sync communication about this? Would love to chat directly and get this resolved quickly for you: https://pgflow.dev/discord/

Technical Details (click to expand)

Root Cause

pgflow uses two-phase polling to prevent duplicate task processing:

1. Phase 1: Read messages from queue (marks them invisible via visibility timeout)
2. Phase 2: start_tasks atomically claims tasks by setting status to 'started'

The bug: when a worker dies between starting a task and completing it, the task stays 'started' forever. When the message's visibility timeout expires, it returns to the queue, but start_tasks filters out non-'queued' tasks - so the message cycles indefinitely without being processed or archived.

Planned Fix

I'm implementing "orphan reclamation" - when start_tasks sees a message whose task has been 'started' longer than the step timeout allows, it will reset the task back to 'queued' so it can be processed again.

Tracking internally in bead pgf-nid.

[mikz]

@jumski no worries. I'll try. I already have a cron mitigation to reap stuck tasks, so I disabled that and reproduced the issue.

PGFlow run status:

select run_id, flow_slug, status, remaining_steps, started_at
from pgflow.runs
where run_id='ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24';

[
  {
    "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
    "flow_slug": "specification_run_v2",
    "status": "started",
    "remaining_steps": 3,
    "started_at": "2026-01-12 08:51:29.592685+00"
  }
]

The critical part: run_test_runs task is stuck in started with attempts_count=1:

select flow_slug, run_id, step_slug, task_index, status, attempts_count, message_id,
       started_at, now() - started_at as stuck_for, last_worker_id
from pgflow.step_tasks
where run_id='ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24'
  and step_slug='run_test_runs'
  and task_index=0;

[
  {
    "flow_slug": "specification_run_v2",
    "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
    "step_slug": "run_test_runs",
    "task_index": 0,
    "status": "started",
    "attempts_count": 1,
    "message_id": 377,
    "started_at": "2026-01-12 08:51:31.547455+00",
    "stuck_for": "00:39:01.871997",
    "last_worker_id": "42d7ce4d-ed2c-42d3-8f87-6ceeb9c76e0a"
  }
]

Meanwhile the corresponding PGMQ message keeps getting re-delivered (read count climbs rapidly; vt stays near “now”):

select msg_id, read_ct, vt, enqueued_at, message
from pgmq.q_specification_run_v2
where msg_id=377;

[
  {
    "msg_id": 377,
    "read_ct": 1181,
    "vt": "2026-01-12 09:31:35.344665+00",
    "enqueued_at": "2026-01-12 08:51:31.467492+00",
    "message": {
      "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
      "flow_slug": "specification_run_v2",
      "step_slug": "run_test_runs",
      "task_index": 0
    }
  }
]

We also see the worker that last held the task is marked stopped, but the task remains started:

select worker_id, queue_name, function_name, last_heartbeat_at, stopped_at
from pgflow.workers
where worker_id='42d7ce4d-ed2c-42d3-8f87-6ceeb9c76e0a';

[
  {
    "worker_id": "42d7ce4d-ed2c-42d3-8f87-6ceeb9c76e0a",
    "queue_name": "specification_run_v2",
    "function_name": "specification_run_worker",
    "last_heartbeat_at": "2026-01-12 08:51:30.392975+00",
    "stopped_at": "2026-01-12 08:56:14.099034+00"
  }
]

Diagnostic queries (requested in issue comment)

1) Stuck 'started' tasks

SELECT flow_slug, step_slug, status, attempts_count,
       started_at, now() - started_at as stuck_for
FROM pgflow.step_tasks
WHERE status = 'started'
  AND started_at < now() - interval '5 minutes'
LIMIT 20;

[
  {
    "flow_slug": "specification_run_v2",
    "step_slug": "run_test_runs",
    "status": "started",
    "attempts_count": 1,
    "started_at": "2026-01-12 08:51:31.547455+00",
    "stuck_for": "00:39:15.466033"
  },
  {
    "flow_slug": "test_run",
    "step_slug": "execute_entries",
    "status": "started",
    "attempts_count": 1,
    "started_at": "2026-01-12 08:51:59.647537+00",
    "stuck_for": "00:38:47.365951"
  }
]

2) Messages cycling without tasks being started

SELECT
  m.msg_id,
  m.read_ct,
  m.vt,
  m.message->>'run_id' as run_id,
  m.message->>'step_slug' as step_slug,
  st.status as task_status,
  st.started_at,
  st.attempts_count,
  r.status as run_status
FROM pgmq.q_specification_run_v2 m
LEFT JOIN pgflow.step_tasks st ON st.message_id = m.msg_id
LEFT JOIN pgflow.runs r ON r.run_id = st.run_id
WHERE m.read_ct > 1
ORDER BY m.read_ct DESC
LIMIT 20;

[
  {
    "msg_id": 377,
    "read_ct": 1193,
    "vt": "2026-01-12 09:31:59.728141+00",
    "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
    "step_slug": "run_test_runs",
    "task_status": "completed",
    "started_at": "2025-12-02 13:46:42.338874+00",
    "attempts_count": 1,
    "run_status": "completed"
  },
  {
    "msg_id": 377,
    "read_ct": 1193,
    "vt": "2026-01-12 09:31:59.728141+00",
    "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
    "step_slug": "run_test_runs",
    "task_status": "started",
    "started_at": "2026-01-12 08:51:31.547455+00",
    "attempts_count": 1,
    "run_status": "started"
  }
]

Note: the join on st.message_id = m.msg_id can return multiple rows in our DB because pgflow.step_tasks.message_id is not globally unique over time. Joining additionally on (run_id, flow_slug, step_slug, task_index) from the message payload gives a 1:1 mapping:

select
  m.msg_id,
  m.read_ct,
  m.vt,
  m.enqueued_at,
  m.message->>'run_id' as run_id,
  m.message->>'flow_slug' as flow_slug,
  m.message->>'step_slug' as step_slug,
  (m.message->>'task_index')::int as task_index,
  st.status as task_status,
  st.started_at,
  st.attempts_count,
  r.status as run_status
from pgmq.q_specification_run_v2 m
left join pgflow.step_tasks st
  on st.message_id = m.msg_id
 and st.run_id::text = m.message->>'run_id'
 and st.flow_slug = m.message->>'flow_slug'
 and st.step_slug = m.message->>'step_slug'
 and st.task_index = (m.message->>'task_index')::int
left join pgflow.runs r on r.run_id = st.run_id
where m.read_ct > 1
order by m.read_ct desc
limit 20;

[
  {
    "msg_id": 377,
    "read_ct": 1196,
    "vt": "2026-01-12 09:32:05.895671+00",
    "enqueued_at": "2026-01-12 08:51:31.467492+00",
    "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
    "flow_slug": "specification_run_v2",
    "step_slug": "run_test_runs",
    "task_index": 0,
    "task_status": "started",
    "started_at": "2026-01-12 08:51:31.547455+00",
    "attempts_count": 1,
    "run_status": "started"
  }
]

3) Messages for failed runs still in queue (orphan type 1)

SELECT
  st.flow_slug,
  st.run_id,
  st.step_slug,
  st.message_id,
  st.status as task_status,
  r.status as run_status,
  r.failed_at
FROM pgflow.step_tasks st
JOIN pgflow.runs r ON r.run_id = st.run_id
WHERE r.status = 'failed'
  AND st.status IN ('queued', 'started')
  AND st.message_id IS NOT NULL;

[]

4) Check archived vs active messages ratio

SELECT 'active' as type, count(*) FROM pgmq.q_specification_run_v2
UNION ALL
SELECT 'archived' as type, count(*) FROM pgmq.a_specification_run_v2;

[{ "type": "active", "count": 1 }, { "type": "archived", "count": 376 }]

5) High read_ct messages

SELECT
  msg_id,
  read_ct,
  vt,
  enqueued_at,
  message
FROM pgmq.q_specification_run_v2
WHERE read_ct > 3
ORDER BY read_ct DESC
LIMIT 20;

[
  {
    "msg_id": 377,
    "read_ct": 1190,
    "vt": "2026-01-12 09:31:53.646797+00",
    "enqueued_at": "2026-01-12 08:51:31.467492+00",
    "message": {
      "run_id": "ebcf5a37-cfaa-40c3-8cfd-e8e91ed45d24",
      "flow_slug": "specification_run_v2",
      "step_slug": "run_test_runs",
      "task_index": 0
    }
  }
]

FYI) Same symptom on test_run

The same pattern shows up for the test_run flow as well (from query #1 above, test_run.execute_entries is also stuck started).

SELECT
  m.msg_id,
  m.read_ct,
  m.vt,
  m.message->>'run_id' as run_id,
  m.message->>'step_slug' as step_slug,
  st.status as task_status,
  st.started_at,
  st.attempts_count,
  r.status as run_status
FROM pgmq.q_test_run m
LEFT JOIN pgflow.step_tasks st ON st.message_id = m.msg_id
LEFT JOIN pgflow.runs r ON r.run_id = st.run_id
WHERE m.read_ct > 1
ORDER BY m.read_ct DESC
LIMIT 20;

[
  {
    "msg_id": 32753,
    "read_ct": 1262,
    "vt": "2026-01-12 09:34:46.424693+00",
    "run_id": "b49ff59f-9adb-4691-913b-fa2374b17944",
    "step_slug": "execute_entries",
    "task_status": "started",
    "started_at": "2026-01-12 08:51:59.647537+00",
    "attempts_count": 1,
    "run_status": "started"
  }
]

SELECT 'active' as type, count(*) FROM pgmq.q_test_run
UNION ALL
SELECT 'archived' as type, count(*) FROM pgmq.a_test_run;

[{ "type": "active", "count": 1 }, { "type": "archived", "count": 32756 }]

SELECT
  msg_id,
  read_ct,
  vt,
  enqueued_at,
  message
FROM pgmq.q_test_run
WHERE read_ct > 3
ORDER BY read_ct DESC
LIMIT 20;

[
  {
    "msg_id": 32753,
    "read_ct": 1271,
    "vt": "2026-01-12 09:35:04.74796+00",
    "enqueued_at": "2026-01-12 08:51:52.381172+00",
    "message": {
      "run_id": "b49ff59f-9adb-4691-913b-fa2374b17944",
      "flow_slug": "test_run",
      "step_slug": "execute_entries",
      "task_index": 55
    }
  }
]

This matches the described root cause: the message becomes visible again, but the task stays started, so it cycles indefinitely until something transitions the task away from started (e.g. reclaim/reset to queued).

[jumski]

Thanks for the details @mikz, appreciated!
I'm gonna process it and confirm the root cause, will get back here with any findings