fix(core): guarantee task visibility extension in start_tasks()

.changeset/start-tasks-visibility.md

@@ -0,0 +1,5 @@
+---
+"@pgflow/core": patch
+---
+
+Make `start_tasks()` apply the PGMQ visibility extension before it returns claimed tasks. The visibility update is now structurally required (referenced CTE instead of an unreferenced `SELECT` CTE PostgreSQL may skip), so a claimed task keeps the effective timeout (`coalesce(step timeout, flow timeout) + 2`) instead of only the initial read visibility, and a visibility-update failure rolls back the whole claim.

pkgs/core/project.json

@@ -16,6 +16,12 @@
     "pgtapTests": [
       "{projectRoot}/supabase/tests/**/*.sql",
       "{projectRoot}/scripts/run-test-with-colors"
+    ],
+    "upgradeFixture": [
+      "{projectRoot}/supabase/upgrade_fixture/**",
+      "{projectRoot}/scripts/run-upgrade-fixture",
+      "{projectRoot}/atlas/atlas.hcl",
+      "{projectRoot}/atlas/supabase-baseline-schema.sql"
     ]
   },
   "targets": {
@@ -115,7 +121,7 @@
       "options": {
         "cwd": "{projectRoot}",
         "commands": [
-          "sqruff --config=../../.sqruff fix --force --parsing-errors schemas/"
+          "sqruff --config=../../.sqruff fix --parsing-errors schemas/"
         ],
         "parallel": false
       }
@@ -198,13 +204,14 @@
       "executor": "nx:run-commands",
       "local": true,
       "dependsOn": ["verify-migrations"],
-      "inputs": ["schemas", "migrations", "pgtapTests"],
+      "inputs": ["schemas", "migrations", "pgtapTests", "upgradeFixture"],
       "cache": true,
       "options": {
         "cwd": "{projectRoot}",
         "commands": [
           "../../scripts/supabase-start-locked.sh .",
-          "scripts/run-test-with-colors"
+          "scripts/run-test-with-colors",
+          "scripts/run-upgrade-fixture"
         ],
         "parallel": false
       }

pkgs/core/schemas/0120_function_start_tasks.sql

@@ -90,14 +90,36 @@ as $$
     join pgflow.flows flow on flow.flow_slug = task.flow_slug
     join pgflow.steps step on step.flow_slug = task.flow_slug and step.step_slug = task.step_slug
   ),
-  -- Batch update visibility timeouts for all messages
-  set_vt_batch as (
+  -- Batch update visibility timeouts for all messages.
+  -- The final statement must force this CTE to run: an unreferenced SELECT
+  -- CTE is not guaranteed to execute, which would leave a claimed task with
+  -- only the shorter initial PGMQ read visibility (#656).
+  visibility_reset as (
     select pgflow.set_vt_batch(
       start_tasks.flow_slug,
       array_agg(t.message_id order by t.message_id),
       array_agg(t.vt_delay order by t.message_id)
     )
     from timeouts t
+  ),
+  -- Force execution of the visibility_reset CTE (same pattern as
+  -- requeue_stalled_tasks) and guard completeness: set_vt_batch updates
+  -- only queue rows it finds, so fewer returned rows than claimed tasks
+  -- means a visibility extension did not run (#656). SQL functions cannot
+  -- RAISE, so the mismatch branch casts a descriptive message to int4:
+  -- the cast error fails the whole statement, rolling back the task
+  -- transition and attempt increment, and returns nothing.
+  _vr as (
+    select case
+      when updated.updated_count = claimed.claimed_count then updated.updated_count
+      else format(
+          'start_tasks(): visibility updated %s of %s claimed messages',
+          updated.updated_count,
+          claimed.claimed_count
+        )::int4
+    end as visibility_updates
+    from (select count(*) as updated_count from visibility_reset) as updated
+    cross join (select count(*) as claimed_count from tasks) as claimed
   )
   select
     st.flow_slug,
@@ -190,4 +212,6 @@ as $$
   left join deps_outputs dep_out on
     dep_out.run_id = st.run_id and
     dep_out.step_slug = st.step_slug
+  cross join _vr
+  where _vr.visibility_updates >= 0
 $$;

pkgs/core/scripts/run-upgrade-fixture

@@ -0,0 +1,68 @@
+#!/bin/bash
+set -euo pipefail
+
+# 0.15.0 upgrade fixture for the consolidated task_lifecycle_hardening migration.
+#
+# Proves on a fresh database that:
+#   1. a database at 0.15.0 (migrations up to 20260607175525) upgrades cleanly,
+#   2. both migration-only data repairs run, in the required order,
+#   3. final runtime behavior works (start_tasks extends PGMQ visibility).
+#
+# Uses the same postgres image and baseline schema as the atlas dev database
+# (pg_cron / pg_net cannot be created in a non-postgres database of the
+# Supabase dev instance, so the fixture gets its own container).
+# Runs as part of `pnpm nx test:pgtap core`.
+
+cd "$(dirname "$0")/.."
+
+IMAGE="jumski/atlas-postgres-pgflow:17.6.1.054"
+CONTAINER=pgflow-upgrade-fixture
+# Last migration released in 0.15.0; everything after it is replaced by the
+# consolidated migration.
+BASELINE_MIGRATION="20260607175525_pgflow_worker_start_mode.sql"
+
+cleanup() { docker rm -f "$CONTAINER" >/dev/null 2>&1 || true; }
+trap cleanup EXIT
+cleanup
+
+echo "upgrade fixture: starting postgres container"
+docker run -d --name "$CONTAINER" "$IMAGE" >/dev/null
+
+for _ in $(seq 1 30); do
+  if docker exec "$CONTAINER" pg_isready -U postgres >/dev/null 2>&1; then
+    break
+  fi
+  sleep 1
+done
+
+psql_in() {
+  docker exec -i "$CONTAINER" psql -v ON_ERROR_STOP=1 -X -q -U postgres -d postgres
+}
+
+echo "upgrade fixture: applying supabase baseline schema"
+psql_in < atlas/supabase-baseline-schema.sql
+
+echo "upgrade fixture: applying pgflow migrations up to 0.15.0 ($BASELINE_MIGRATION)"
+reached_baseline=false
+for f in supabase/migrations/*.sql; do
+  echo "  $(basename "$f")"
+  psql_in < "$f"
+  if [[ "$(basename "$f")" == "$BASELINE_MIGRATION" ]]; then
+    reached_baseline=true
+    break
+  fi
+done
+if [[ "$reached_baseline" != true ]]; then
+  echo "upgrade fixture: baseline migration $BASELINE_MIGRATION not found" >&2
+  exit 1
+fi
+
+echo "upgrade fixture: seeding stale 0.15.0 data"
+psql_in < supabase/upgrade_fixture/seed.sql
+
+consolidated=$(ls supabase/migrations/*_pgflow_task_lifecycle_hardening.sql)
+echo "upgrade fixture: applying consolidated migration $(basename "$consolidated")"
+psql_in < "$consolidated"
+
+psql_in < supabase/upgrade_fixture/assertions.sql
+echo "upgrade fixture: PASS"