fix(worker): coordinate pending starts with invocation handshakes

.changeset/worker-startup-handshake.md

@@ -0,0 +1,6 @@
+---
+"@pgflow/core": patch
+"@pgflow/edge-worker": patch
+---
+
+Coordinate pending worker starts with invocation handshakes. `ensure_workers()` now reserves a unique `invocation_id` per function under a row lock and enqueues it in one transaction; the HTTP worker start atomically consumes the still-current, unexpired invocation when it registers readiness. Duplicate cron starts, duplicate delivery to independent runtimes, and manual/local requests now share one reservation protocol with a 60-second pending deadline, idempotent acknowledgement, explicit dismissal on failure, and bounded watchdog/SQL-timeout recovery. `debounce` becomes a retry throttle instead of the correctness mechanism. Includes the SQL migration and generated types; requires applying the migration before deploying new workers.

pkgs/core/schemas/0056_table_worker_functions.sql

@@ -9,6 +9,11 @@ create table if not exists pgflow.worker_functions (
   debounce interval not null default '6 seconds'
   check (debounce >= '1 second'),
   last_invoked_at timestamptz,
+  invocation_id uuid,
+  invocation_expires_at timestamptz,
+  invocation_status text check (invocation_status in ('pending', 'ready', 'failed')),
+  invocation_worker_id uuid,
+  invocation_error text,
   created_at timestamptz not null default now(),
   updated_at timestamptz not null default now()
 );
@@ -29,4 +34,19 @@ comment on column pgflow.worker_functions.debounce is
 'Minimum interval between invocation attempts for this function';
 
 comment on column pgflow.worker_functions.last_invoked_at is
-'When ensure_workers() last pinged this function (used for debouncing)';
+'When startup was last reserved (retry throttle, not a readiness signal)';
+
+comment on column pgflow.worker_functions.invocation_id is
+'Current startup handshake ID; retained after readiness or failure for idempotent reconciliation';
+
+comment on column pgflow.worker_functions.invocation_expires_at is
+'Database-clock deadline for pending startup. An expired pending invocation is retryable';
+
+comment on column pgflow.worker_functions.invocation_status is
+'pending: awaiting readiness; ready: handshake completed, not worker lifetime; failed: retryable after debounce';
+
+comment on column pgflow.worker_functions.invocation_worker_id is
+'Worker that completed the current startup handshake';
+
+comment on column pgflow.worker_functions.invocation_error is
+'Reason the current startup handshake failed, if explicitly dismissed';

pkgs/core/schemas/0058_function_reserve_worker_invocation.sql

@@ -0,0 +1,52 @@
+-- Shared reservation for cron and authenticated direct HTTP startup.
+create or replace function pgflow.reserve_worker_invocation(function_name text)
+returns uuid
+language sql
+as $$
+  with
+    -- Lock the registration row only when it is eligible: enabled, HTTP-started,
+    -- past debounce, with no unexpired pending invocation and no healthy worker.
+    -- Concurrent reservations re-check eligibility under the row lock, so at most
+    -- one current pending invocation can exist per function.
+    eligible as (
+      select wf.function_name
+      from pgflow.worker_functions as wf
+      where wf.function_name = reserve_worker_invocation.function_name
+        and wf.enabled
+        and wf.start_mode = 'http'
+        and (
+          wf.invocation_status is distinct from 'pending'
+          or wf.invocation_expires_at is null
+          or wf.invocation_expires_at <= clock_timestamp()
+        )
+        and (
+          wf.last_invoked_at is null
+          or wf.last_invoked_at <= clock_timestamp() - wf.debounce
+        )
+        and not exists (
+          select 1 from pgflow.workers as w
+          where w.function_name = wf.function_name
+            and w.stopped_at is null
+            and w.deprecated_at is null
+            and w.last_heartbeat_at > clock_timestamp() - wf.debounce
+        )
+      for update
+    ),
+    reserved as (
+      update pgflow.worker_functions as wf
+      set invocation_id = gen_random_uuid(),
+          invocation_expires_at = clock_timestamp() + interval '60 seconds',
+          invocation_status = 'pending',
+          invocation_worker_id = null,
+          invocation_error = null,
+          last_invoked_at = clock_timestamp()
+      from eligible as e
+      where wf.function_name = e.function_name
+      returning wf.invocation_id
+    )
+  select invocation_id from reserved;
+$$;
+
+comment on function pgflow.reserve_worker_invocation(text) is
+'Reserves one pending HTTP startup for 60 seconds under a per-function row lock.
+Shared by cron and manual startup; returns NULL while healthy, pending, disabled, process-started or debounced.';

pkgs/core/schemas/0058_function_worker_invocation_handshake.sql

@@ -0,0 +1,143 @@
+-- Validate every HTTP delivery, including deliveries to surviving runtimes.
+-- A receipt never clears pending state. An existing healthy worker may finish
+-- the handshake without registering a new worker or starting another loop.
+create or replace function pgflow.reconcile_worker_invocation(
+  function_name text,
+  invocation_id uuid,
+  worker_id uuid default null
+) returns text
+language plpgsql
+as $$
+declare
+  registration pgflow.worker_functions%rowtype;
+begin
+  select wf.* into registration from pgflow.worker_functions as wf
+  where wf.function_name = reconcile_worker_invocation.function_name
+  for update;
+
+  if not found or not registration.enabled or registration.start_mode <> 'http'
+    or reconcile_worker_invocation.invocation_id is null
+    or registration.invocation_id is distinct from reconcile_worker_invocation.invocation_id then
+    return 'rejected';
+  end if;
+
+  if registration.invocation_status = 'pending'
+    and registration.invocation_expires_at > clock_timestamp() then
+    if reconcile_worker_invocation.worker_id is null then
+      return 'pending';
+    end if;
+    if exists (
+      select 1 from pgflow.workers as w
+      where w.worker_id = reconcile_worker_invocation.worker_id
+        and w.function_name = registration.function_name
+        and w.stopped_at is null and w.deprecated_at is null
+        and w.last_heartbeat_at > clock_timestamp() - registration.debounce
+    ) then
+      update pgflow.worker_functions as wf
+      set invocation_status = 'ready', invocation_worker_id = reconcile_worker_invocation.worker_id
+      where wf.function_name = registration.function_name;
+      return 'ready';
+    end if;
+    return 'pending';
+  elsif registration.invocation_status = 'ready'
+    and registration.invocation_worker_id = reconcile_worker_invocation.worker_id
+    and exists (
+      select 1 from pgflow.workers as w
+      where w.worker_id = reconcile_worker_invocation.worker_id
+        and w.function_name = registration.function_name
+        and w.stopped_at is null and w.deprecated_at is null
+        and w.last_heartbeat_at > clock_timestamp() - registration.debounce
+    ) then
+    return 'ready';
+  end if;
+
+  return 'rejected';
+end;
+$$;
+
+-- Called only after initialization completes, immediately before processing.
+create or replace function pgflow.accept_worker_invocation(
+  function_name text,
+  invocation_id uuid,
+  worker_id uuid,
+  queue_name text,
+  pgflow_version text
+) returns setof pgflow.workers
+language plpgsql
+as $$
+declare
+  registration pgflow.worker_functions%rowtype;
+begin
+  select wf.* into registration from pgflow.worker_functions as wf
+  where wf.function_name = accept_worker_invocation.function_name
+  for update;
+
+  if not found or not registration.enabled or registration.start_mode <> 'http'
+    or accept_worker_invocation.invocation_id is null
+    or registration.invocation_id is distinct from accept_worker_invocation.invocation_id then
+    return;
+  end if;
+
+  -- A committed acceptance whose response was lost is safe to repeat only
+  -- for that same healthy worker, never for a different runtime.
+  if registration.invocation_status = 'ready'
+    and registration.invocation_worker_id = accept_worker_invocation.worker_id then
+    return query select w.* from pgflow.workers as w
+    where w.worker_id = accept_worker_invocation.worker_id
+      and w.function_name = registration.function_name
+      and w.queue_name = accept_worker_invocation.queue_name
+      and w.stopped_at is null and w.deprecated_at is null
+      and w.last_heartbeat_at > clock_timestamp() - registration.debounce;
+    return;
+  end if;
+
+  if registration.invocation_status is distinct from 'pending'
+    or registration.invocation_expires_at is null
+    or registration.invocation_expires_at <= clock_timestamp() then
+    return;
+  end if;
+
+  return query
+  insert into pgflow.workers (
+    worker_id, function_name, queue_name, pgflow_version, started_at, last_heartbeat_at
+  )
+  values (accept_worker_invocation.worker_id, registration.function_name,
+          accept_worker_invocation.queue_name, accept_worker_invocation.pgflow_version,
+          clock_timestamp(), clock_timestamp())
+  returning *;
+
+  update pgflow.worker_functions as wf
+  set invocation_status = 'ready', invocation_worker_id = accept_worker_invocation.worker_id
+  where wf.function_name = registration.function_name;
+end;
+$$;
+
+create or replace function pgflow.dismiss_worker_invocation(
+  function_name text,
+  invocation_id uuid,
+  reason text
+) returns boolean
+language sql
+as $$
+  with dismissed as (
+    update pgflow.worker_functions as wf
+    set invocation_status = 'failed', invocation_error = left(dismiss_worker_invocation.reason, 2000)
+    where wf.function_name = dismiss_worker_invocation.function_name
+      and wf.invocation_id = dismiss_worker_invocation.invocation_id
+      and wf.invocation_status = 'pending'
+    returning 1
+  )
+  select exists (select 1 from dismissed);
+$$;
+
+comment on function pgflow.reconcile_worker_invocation(text, uuid, uuid) is
+'Validates a startup delivery. Receipt returns pending without consuming it; a healthy existing worker
+can acknowledge readiness. Old, expired, failed or superseded IDs are rejected.';
+
+comment on function pgflow.accept_worker_invocation(text, uuid, uuid, text, text) is
+'Atomically consumes a current pending invocation and registers its ready worker.
+Idempotent for the accepted healthy worker; no lifetime ownership or task fencing.';
+
+comment on function pgflow.dismiss_worker_invocation(text, uuid, text) is
+'Makes only the matching pending invocation retryable after debounce;
+cannot undo readiness or dismiss a newer invocation.';

pkgs/core/schemas/0059_function_ensure_workers.sql

@@ -1,116 +1,61 @@
--- Ensure Workers
--- Returns which worker functions should be invoked, makes HTTP requests, and updates last_invoked_at
--- Called by cron job to keep workers running
-
+-- Reserve and enqueue in the same transaction: delivery cannot precede the
+-- committed pending state, and enqueue failure rolls the reservation back.
+-- Startup eligibility (debounce, pending state, healthy workers) lives in
+-- pgflow.reserve_worker_invocation, shared with manual startup.
 create or replace function pgflow.ensure_workers()
 returns table (function_name text, invoked boolean, request_id bigint)
 language sql
 as $$
   with
-    -- Detect environment
-    env as (
+    -- Local development targets Kong directly without auth; production
+    -- resolves the project URL and service credentials from Vault.
+    env as materialized (
       select pgflow.is_local() as is_local
     ),
-
-    -- Get credentials: Local mode uses hardcoded URL, production uses vault secrets
-    -- Empty strings are treated as NULL using nullif()
-    -- pgflow_auth_secret takes priority over supabase_service_role_key for production auth
-    credentials as (
+    credentials as materialized (
       select
-        case
-          when (select is_local from env) then null
-          else coalesce(
-            nullif((select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_auth_secret'), ''),
-            nullif((select decrypted_secret from vault.decrypted_secrets where name = 'supabase_service_role_key'), '')
-          )
-        end as service_role_key,
-        case
-          when (select is_local from env) then 'http://kong:8000/functions/v1'
-          else (select 'https://' || nullif(decrypted_secret, '') || '.supabase.co/functions/v1' from vault.decrypted_secrets where name = 'supabase_project_id')
-        end as base_url
+        case when e.is_local then null else coalesce(
+          nullif((select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_auth_secret'), ''),
+          nullif((select decrypted_secret from vault.decrypted_secrets where name = 'supabase_service_role_key'), '')
+        ) end as service_role_key,
+        case when e.is_local then 'http://kong:8000/functions/v1' else (
+          select 'https://' || nullif(decrypted_secret, '') || '.supabase.co/functions/v1'
+          from vault.decrypted_secrets where name = 'supabase_project_id'
+        ) end as base_url,
+        e.is_local
+      from env as e
     ),
-
-    -- Only HTTP-started worker functions are supervised via HTTP pings.
-    http_worker_functions as (
-      select wf.function_name, wf.debounce, wf.last_invoked_at
+    -- One reservation per eligible function; unreserved functions send nothing.
+    reservations as materialized (
+      select wf.function_name, c.base_url, c.service_role_key, c.is_local,
+             pgflow.reserve_worker_invocation(wf.function_name) as invocation_id
       from pgflow.worker_functions as wf
-      where wf.enabled = true
-        and wf.start_mode = 'http'
-    ),
-
-    -- Find functions that pass the debounce check
-    debounce_passed as (
-      select wf.function_name, wf.debounce
-      from http_worker_functions as wf
-      where wf.last_invoked_at is null
-        or wf.last_invoked_at < now() - wf.debounce
-    ),
-
-    -- Find functions that have at least one alive worker
-    functions_with_alive_workers as (
-      select distinct w.function_name
-      from pgflow.workers as w
-      inner join debounce_passed as dp on w.function_name = dp.function_name
-      where w.stopped_at is null
-        and w.deprecated_at is null
-        and w.last_heartbeat_at > now() - dp.debounce
-    ),
-
-    -- Determine which functions should be invoked
-    -- Local mode: all enabled functions (bypass debounce AND alive workers check)
-    -- Production mode: only functions that pass debounce AND have no alive workers
-    functions_to_invoke as (
-      select wf.function_name
-      from http_worker_functions as wf
-      where pgflow.is_local() = true  -- Local: all enabled HTTP functions
-        or (
-          -- Production: debounce + no alive workers
-          wf.function_name in (select dp.function_name from debounce_passed as dp)
-          and wf.function_name not in (select faw.function_name from functions_with_alive_workers as faw)
-        )
-    ),
-
-    -- Make HTTP requests and capture request_ids
-    http_requests as (
-      select
-        fti.function_name,
-        net.http_post(
-          url => c.base_url || '/' || fti.function_name,
-          headers => case
-            when e.is_local then '{}'::jsonb
-            else jsonb_build_object(
-              'Content-Type', 'application/json',
-              'Authorization', 'Bearer ' || c.service_role_key
-            )
-          end,
-          body => '{}'::jsonb
-        ) as request_id
-      from functions_to_invoke as fti
       cross join credentials as c
-      cross join env as e
-      where c.base_url is not null
-        and (e.is_local or c.service_role_key is not null)
-    ),
-
-    -- Update last_invoked_at for invoked functions
-    updated as (
-      update pgflow.worker_functions as wf
-      set last_invoked_at = clock_timestamp()
-      from http_requests as hr
-      where wf.function_name = hr.function_name
-      returning wf.function_name
+      where wf.enabled and wf.start_mode = 'http'
+        and c.base_url is not null
+        and (c.is_local or c.service_role_key is not null)
+      -- Concurrent cron transactions acquire registration locks in one order.
+      order by wf.function_name
     )
-
-  select u.function_name, true as invoked, hr.request_id
-  from updated as u
-  inner join http_requests as hr on u.function_name = hr.function_name
+  -- pg_net delivers asynchronously after this transaction commits.
+  select r.function_name, true,
+    net.http_post(
+      url => r.base_url || '/' || r.function_name,
+      headers => case when r.is_local then '{"Content-Type":"application/json"}'::jsonb
+        else jsonb_build_object(
+          'Content-Type', 'application/json',
+          'Authorization', 'Bearer ' || r.service_role_key
+        ) end,
+      body => jsonb_build_object('invocation_id', r.invocation_id),
+      timeout_milliseconds => 60000
+    )
+  from reservations as r
+  where r.invocation_id is not null;
 $$;
 
 comment on function pgflow.ensure_workers() is
-'Ensures worker functions are running by pinging them via HTTP when needed.
-In local mode: pings ALL enabled functions (ignores debounce AND alive workers check).
-In production mode: only pings functions that pass debounce AND have no alive workers.
-Debounce: skips functions pinged within their debounce interval (production only).
-Credentials: Uses Vault secrets (pgflow_auth_secret with fallback to supabase_service_role_key, supabase_project_id) or local fallbacks.
-URL is built from project_id: https://{project_id}.supabase.co/functions/v1
-Returns request_id from pg_net for each HTTP request made.';
+'Enqueues HTTP startup only when no healthy worker or unexpired pending invocation exists.
+Reservations last 60 seconds; debounce throttles retries. Local development uses the same handshake
+with local URL/auth fallbacks. Credentials: Vault pgflow_auth_secret, falling back to
+supabase_service_role_key, and supabase_project_id. Returns the pg_net request_id;
+handshake state is in worker_functions.';