From 61c75a5e6ee563dfe2b8663e4c7bccdc2660b18d Mon Sep 17 00:00:00 2001 From: Agent Date: Mon, 2 Nov 2026 09:42:07 +0100 Subject: [PATCH] fix(worker): coordinate pending starts with invocation handshakes Debounce measured HTTP dispatch time rather than readiness, so slow initialization let cron enqueue duplicate live workers. Reserve one database-clock invocation per function, then atomically accept its current unexpired token with ready-worker registration instead of extending debounce. Use a 60-second pending deadline, idempotent acknowledgement, safe dismissal, and bounded runtime/SQL startup recovery. Authenticated manual and local requests share reservations; process workers stay outside HTTP scheduling. Include the generated migration, types, regression coverage, and SQL-first mixed-version rollout guidance. Lifetime ownership and task fencing remain in #697. Review feedback: reserve_worker_invocation is now a declarative language sql function. Its eligibility predicates are NULL-safe because a fresh registration has no invocation state; the previous `not (status = 'pending' and ...)` form filtered fresh rows out under WHERE semantics while the plpgsql IF treated the NULL as false. reconcile_worker_invocation and accept_worker_invocation stay plpgsql: they are multi-exit state machines whose early returns read clearer imperatively. invocation_worker_id stays a soft pointer without a foreign key: pgflow never deletes worker rows, so the constraint would add migration churn without behavior. A startup troubleshooting section joins the worker management docs. The four baseline array-spacing lint failures were a local sqruff 0.39.0 defect in function signatures; CI installs a current sqruff that passes the original compact form, so no reformat ships and the earlier style commit was dropped. Validation: full pgTAP (1702 assertions), the real-SQL worker-invocation integration file (11 tests), Edge Function E2E (13 tests), portable-runtime E2E (14 tests), migration verification, generated-type checks, and the website build all pass, with the full 81-test integration suite from the pre-review delivery. Fixes #696 --- .changeset/worker-startup-handshake.md | 6 + .../schemas/0056_table_worker_functions.sql | 22 +- ...058_function_reserve_worker_invocation.sql | 52 ++ ...8_function_worker_invocation_handshake.sql | 143 ++++++ .../schemas/0059_function_ensure_workers.sql | 147 ++---- pkgs/core/src/database-types.ts | 56 ++ ...104928_pgflow_worker_startup_handshake.sql | 233 +++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- .../credentials_from_vault.test.sql | 84 +-- .../credentials_local_fallback.test.sql | 71 +-- .../credentials_pgflow_auth_secret.test.sql | 6 + .../tests/ensure_workers/debounce.test.sql | 8 +- .../disabled_functions.test.sql | 39 +- .../http_request_queued.test.sql | 104 +--- .../invocation_handshake.test.sql | 54 ++ .../invocation_recovery.test.sql | 47 ++ .../local_mode_always_pings.test.sql | 52 -- .../local_mode_handshake.test.sql | 23 + .../ensure_workers/pending_start.test.sql | 28 + .../request_id_returned.test.sql | 49 +- .../returns_functions_to_invoke.test.sql | 9 + .../skips_process_start_mode.test.sql | 3 + .../worker_death_detection.test.sql | 9 + pkgs/edge-worker/src/core/Queries.ts | 82 ++- pkgs/edge-worker/src/core/WorkerLifecycle.ts | 9 +- pkgs/edge-worker/src/core/types.ts | 2 + .../src/core/withStartupTimeout.ts | 17 + .../src/flow/FlowWorkerLifecycle.ts | 14 +- .../src/platform/SupabasePlatformAdapter.ts | 126 ++++- pkgs/edge-worker/src/queue/Queue.ts | 7 +- .../integration/workerInvocation.test.ts | 484 ++++++++++++++++++ .../platform/SupabasePlatformAdapter.test.ts | 54 +- .../content/docs/build/local-development.mdx | 8 +- .../docs/concepts/startup-compilation.mdx | 22 +- .../docs/concepts/worker-lifecycle.mdx | 26 +- .../deploy/supabase/update-deployed-flows.mdx | 30 +- .../src/content/docs/deploy/update-pgflow.mdx | 4 + .../content/docs/deploy/worker-management.mdx | 91 +++- 38 files changed, 1667 insertions(+), 557 deletions(-) create mode 100644 .changeset/worker-startup-handshake.md create mode 100644 pkgs/core/schemas/0058_function_reserve_worker_invocation.sql create mode 100644 pkgs/core/schemas/0058_function_worker_invocation_handshake.sql create mode 100644 pkgs/core/supabase/migrations/20261102104928_pgflow_worker_startup_handshake.sql create mode 100644 pkgs/core/supabase/tests/ensure_workers/invocation_handshake.test.sql create mode 100644 pkgs/core/supabase/tests/ensure_workers/invocation_recovery.test.sql delete mode 100644 pkgs/core/supabase/tests/ensure_workers/local_mode_always_pings.test.sql create mode 100644 pkgs/core/supabase/tests/ensure_workers/local_mode_handshake.test.sql create mode 100644 pkgs/core/supabase/tests/ensure_workers/pending_start.test.sql create mode 100644 pkgs/edge-worker/src/core/withStartupTimeout.ts create mode 100644 pkgs/edge-worker/tests/integration/workerInvocation.test.ts diff --git a/.changeset/worker-startup-handshake.md b/.changeset/worker-startup-handshake.md new file mode 100644 index 000000000..ca843a4d0 --- /dev/null +++ b/.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. diff --git a/pkgs/core/schemas/0056_table_worker_functions.sql b/pkgs/core/schemas/0056_table_worker_functions.sql index 92377b48e..5a6fb2c01 100644 --- a/pkgs/core/schemas/0056_table_worker_functions.sql +++ b/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'; diff --git a/pkgs/core/schemas/0058_function_reserve_worker_invocation.sql b/pkgs/core/schemas/0058_function_reserve_worker_invocation.sql new file mode 100644 index 000000000..001d05534 --- /dev/null +++ b/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.'; diff --git a/pkgs/core/schemas/0058_function_worker_invocation_handshake.sql b/pkgs/core/schemas/0058_function_worker_invocation_handshake.sql new file mode 100644 index 000000000..03130e7ed --- /dev/null +++ b/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.'; diff --git a/pkgs/core/schemas/0059_function_ensure_workers.sql b/pkgs/core/schemas/0059_function_ensure_workers.sql index ab5774d88..63e23be6c 100644 --- a/pkgs/core/schemas/0059_function_ensure_workers.sql +++ b/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.'; diff --git a/pkgs/core/src/database-types.ts b/pkgs/core/src/database-types.ts index 737d0c72b..89008428a 100644 --- a/pkgs/core/src/database-types.ts +++ b/pkgs/core/src/database-types.ts @@ -359,6 +359,11 @@ export type Database = { debounce: unknown enabled: boolean function_name: string + invocation_error: string | null + invocation_expires_at: string | null + invocation_id: string | null + invocation_status: string | null + invocation_worker_id: string | null last_invoked_at: string | null start_mode: string updated_at: string @@ -368,6 +373,11 @@ export type Database = { debounce?: unknown enabled?: boolean function_name: string + invocation_error?: string | null + invocation_expires_at?: string | null + invocation_id?: string | null + invocation_status?: string | null + invocation_worker_id?: string | null last_invoked_at?: string | null start_mode?: string updated_at?: string @@ -377,6 +387,11 @@ export type Database = { debounce?: unknown enabled?: boolean function_name?: string + invocation_error?: string | null + invocation_expires_at?: string | null + invocation_id?: string | null + invocation_status?: string | null + invocation_worker_id?: string | null last_invoked_at?: string | null start_mode?: string updated_at?: string @@ -451,6 +466,31 @@ export type Database = { Args: { p_flow_slug: string; p_step_index: number; p_step_slug: string } Returns: string } + accept_worker_invocation: { + Args: { + function_name: string + invocation_id: string + pgflow_version: string + queue_name: string + worker_id: string + } + Returns: { + deprecated_at: string | null + function_name: string + last_heartbeat_at: string + pgflow_version: string | null + queue_name: string + started_at: string + stopped_at: string | null + worker_id: string + }[] + SetofOptions: { + from: "*" + to: "workers" + isOneToOne: false + isSetofReturn: true + } + } add_step: { Args: { base_delay?: number @@ -565,6 +605,10 @@ export type Database = { Args: { p_flow_slug: string } Returns: undefined } + dismiss_worker_invocation: { + Args: { function_name: string; invocation_id: string; reason: string } + Returns: boolean + } ensure_flow_compiled: { Args: { flow_slug: string @@ -638,7 +682,19 @@ export type Database = { isSetofReturn: true } } + reconcile_worker_invocation: { + Args: { + function_name: string + invocation_id: string + worker_id?: string + } + Returns: string + } requeue_stalled_tasks: { Args: never; Returns: number } + reserve_worker_invocation: { + Args: { function_name: string } + Returns: string + } set_vt_batch: { Args: { msg_ids: number[]; queue_name: string; vt_offsets: number[] } Returns: { diff --git a/pkgs/core/supabase/migrations/20261102104928_pgflow_worker_startup_handshake.sql b/pkgs/core/supabase/migrations/20261102104928_pgflow_worker_startup_handshake.sql new file mode 100644 index 000000000..4e343ab50 --- /dev/null +++ b/pkgs/core/supabase/migrations/20261102104928_pgflow_worker_startup_handshake.sql @@ -0,0 +1,233 @@ +-- Modify "worker_functions" table +ALTER TABLE "pgflow"."worker_functions" ADD CONSTRAINT "worker_functions_invocation_status_check" CHECK (invocation_status = ANY (ARRAY['pending'::text, 'ready'::text, 'failed'::text])), ADD COLUMN "invocation_id" uuid NULL, ADD COLUMN "invocation_expires_at" timestamptz NULL, ADD COLUMN "invocation_status" text NULL, ADD COLUMN "invocation_worker_id" uuid NULL, ADD COLUMN "invocation_error" text NULL; +-- Set comment to column: "last_invoked_at" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."last_invoked_at" IS 'When startup was last reserved (retry throttle, not a readiness signal)'; +-- Set comment to column: "invocation_id" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."invocation_id" IS 'Current startup handshake ID; retained after readiness or failure for idempotent reconciliation'; +-- Set comment to column: "invocation_expires_at" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."invocation_expires_at" IS 'Database-clock deadline for pending startup. An expired pending invocation is retryable'; +-- Set comment to column: "invocation_status" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."invocation_status" IS 'pending: awaiting readiness; ready: handshake completed, not worker lifetime; failed: retryable after debounce'; +-- Set comment to column: "invocation_worker_id" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."invocation_worker_id" IS 'Worker that completed the current startup handshake'; +-- Set comment to column: "invocation_error" on table: "worker_functions" +COMMENT ON COLUMN "pgflow"."worker_functions"."invocation_error" IS 'Reason the current startup handshake failed, if explicitly dismissed'; +-- Create "reserve_worker_invocation" function +CREATE 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; +$$; +-- Set comment to function: "reserve_worker_invocation" +COMMENT ON FUNCTION "pgflow"."reserve_worker_invocation" 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.'; +-- Modify "ensure_workers" function +CREATE OR REPLACE FUNCTION "pgflow"."ensure_workers" () RETURNS TABLE ("function_name" text, "invoked" boolean, "request_id" bigint) LANGUAGE sql AS $$ +with + -- 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 + ), + credentials as materialized ( + select + 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 + ), + -- 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 + cross join credentials as c + 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 + ) + -- 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; +$$; +-- Set comment to function: "ensure_workers" +COMMENT ON FUNCTION "pgflow"."ensure_workers" IS '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.'; +-- Create "accept_worker_invocation" function +CREATE 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; +$$; +-- Set comment to function: "accept_worker_invocation" +COMMENT ON FUNCTION "pgflow"."accept_worker_invocation" IS 'Atomically consumes a current pending invocation and registers its ready worker. +Idempotent for the accepted healthy worker; no lifetime ownership or task fencing.'; +-- Create "dismiss_worker_invocation" function +CREATE 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); +$$; +-- Set comment to function: "dismiss_worker_invocation" +COMMENT ON FUNCTION "pgflow"."dismiss_worker_invocation" IS 'Makes only the matching pending invocation retryable after debounce; +cannot undo readiness or dismiss a newer invocation.'; +-- Create "reconcile_worker_invocation" function +CREATE FUNCTION "pgflow"."reconcile_worker_invocation" ("function_name" text, "invocation_id" uuid, "worker_id" uuid DEFAULT NULL::uuid) 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; +$$; +-- Set comment to function: "reconcile_worker_invocation" +COMMENT ON FUNCTION "pgflow"."reconcile_worker_invocation" 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.'; diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index 771681f6d..6215f95e3 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:sOlj3XB6+Tss5pInYvuwe/yxDhipAVGK/ZF49e+gLAo= +h1:mY/dlzJLq3/OBDZj5QD3exFFZ1ba5tO9ScipRaJcr5E= 20250429164909_pgflow_initial.sql h1:I3n/tQIg5Q5nLg7RDoU3BzqHvFVjmumQxVNbXTPG15s= 20250517072017_pgflow_fix_poll_for_tasks_to_use_separate_statement_for_polling.sql h1:wTuXuwMxVniCr3ONCpodpVWJcHktoQZIbqMZ3sUHKMY= 20250609105135_pgflow_add_start_tasks_and_started_status.sql h1:ggGanW4Wyt8Kv6TWjnZ00/qVb3sm+/eFVDjGfT8qyPg= @@ -26,3 +26,4 @@ h1:sOlj3XB6+Tss5pInYvuwe/yxDhipAVGK/ZF49e+gLAo= 20260919152659_pgflow_fix_force_skip_multi_queue.sql h1:DC7uQRwjpOm3A4NtDYb3SjbSSqQix4MlQ4+BnhGQ/Q8= 20260920093533_pgflow_telemetry.sql h1:dp3gNRSUpTNoPH9w4n9P7CBhXFoJspQ+hP9MfQ1u3Fg= 20261001194538_pgflow_telemetry_endpoint_fix.sql h1:/ROJiUnHzOt6xsw9f877wY6KlrVOFxpjGBuFuweX6Is= +20261102104928_pgflow_worker_startup_handshake.sql h1:TnfsZkNyx4CItISg+jQWsIbteVeHBfLIyvaOWlDWEE0= diff --git a/pkgs/core/supabase/tests/ensure_workers/credentials_from_vault.test.sql b/pkgs/core/supabase/tests/ensure_workers/credentials_from_vault.test.sql index b48f4e8cb..f2a043e2f 100644 --- a/pkgs/core/supabase/tests/ensure_workers/credentials_from_vault.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/credentials_from_vault.test.sql @@ -1,79 +1,17 @@ --- Test: ensure_workers() retrieves credentials from Vault +-- Production credentials remain independent of the startup handshake. begin; select plan(4); select pgflow_tests.reset_db(); - --- Setup: Create Vault secrets -select vault.create_secret( - 'test-service-role-key-from-vault', - 'supabase_service_role_key' -); -select vault.create_secret( - 'vaultproject123', - 'supabase_project_id' -); - --- Setup: Register a worker function +select vault.create_secret('test-service-role-key-from-vault', 'supabase_service_role_key'); +select vault.create_secret('vaultproject123', 'supabase_project_id'); select pgflow.track_worker_function('my-function'); -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Simulate production mode (non-local jwt_secret) -set local app.settings.jwt_secret = 'production-secret-different-from-local'; - --- TEST: In production mode WITH Vault secrets, function IS invoked -with result as ( - select * from pgflow.ensure_workers() -) -select is( - (select count(*) from result), - 1::bigint, - 'Production mode with Vault secrets invokes functions' -); - --- TEST: request_id is returned (proves Vault credentials were used for HTTP call) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -) -select ok( - (select request_id is not null from result limit 1), - 'Vault credentials allow HTTP invocation in production mode' -); - --- TEST: HTTP request URL uses base URL from Vault -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Store result in temp table to ensure ensure_workers() executes before we query the queue -select * into temporary test3_result from pgflow.ensure_workers(); - -select ok( - (select url = 'https://vaultproject123.supabase.co/functions/v1/my-function' - from net.http_request_queue - where id = (select request_id from test3_result limit 1)), - 'HTTP request URL is constructed from Vault supabase_project_id' -); - -drop table test3_result; - --- TEST: HTTP request Authorization header uses service role key from Vault -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Store result in temp table to ensure ensure_workers() executes before we query the queue -select * into temporary test4_result from pgflow.ensure_workers(); - -select ok( - (select headers->>'Authorization' = 'Bearer test-service-role-key-from-vault' - from net.http_request_queue - where id = (select request_id from test4_result limit 1)), - 'HTTP request Authorization header contains Vault service role key' -); - -drop table test4_result; - +set local app.settings.jwt_secret = 'production-secret'; +create temporary table invocation as select * from pgflow.ensure_workers(); +select is((select count(*) from invocation), 1::bigint, 'production Vault credentials enqueue startup'); +select ok((select request_id is not null from invocation), 'returns the queued request ID'); +select is((select url from net.http_request_queue where id = (select request_id from invocation)), + 'https://vaultproject123.supabase.co/functions/v1/my-function', 'uses the project ID from Vault'); +select is((select headers->>'Authorization' from net.http_request_queue where id = (select request_id from invocation)), + 'Bearer test-service-role-key-from-vault', 'preserves service-role authentication'); select finish(); rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/credentials_local_fallback.test.sql b/pkgs/core/supabase/tests/ensure_workers/credentials_local_fallback.test.sql index 5c3445f05..55b031ccf 100644 --- a/pkgs/core/supabase/tests/ensure_workers/credentials_local_fallback.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/credentials_local_fallback.test.sql @@ -1,72 +1,15 @@ --- Test: ensure_workers() uses local fallback credentials when Vault is empty begin; select plan(4); select pgflow_tests.reset_db(); - --- Ensure no Vault secrets exist delete from vault.secrets where name in ('supabase_service_role_key', 'supabase_project_id'); - --- Setup: Register a worker function select pgflow.track_worker_function('my-function'); -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Simulate local mode set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; - --- TEST: In local mode without Vault secrets, function IS invoked (uses fallback) -with result as ( - select * from pgflow.ensure_workers() -) -select is( - (select count(*) from result), - 1::bigint, - 'Local mode uses fallback credentials when Vault is empty' -); - --- TEST: request_id is returned (proves HTTP call was made) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -) -select ok( - (select request_id is not null from result limit 1), - 'Local fallback credentials allow HTTP invocation' -); - --- TEST: HTTP request URL uses local fallback base URL -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Store result in temp table to ensure ensure_workers() executes before we query the queue -select * into temporary test3_result from pgflow.ensure_workers(); - -select ok( - (select url = 'http://kong:8000/functions/v1/my-function' - from net.http_request_queue - where id = (select request_id from test3_result limit 1)), - 'HTTP request URL uses local fallback (http://kong:8000/functions/v1)' -); - -drop table test3_result; - --- TEST: HTTP request has no Authorization header in local mode -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Store result in temp table to ensure ensure_workers() executes before we query the queue -select * into temporary test4_result from pgflow.ensure_workers(); - -select ok( - (select headers->>'Authorization' is null - from net.http_request_queue - where id = (select request_id from test4_result limit 1)), - 'HTTP request has no Authorization header in local mode' -); - -drop table test4_result; - +create temporary table invocation as select * from pgflow.ensure_workers(); +select is((select count(*) from invocation), 1::bigint, 'local startup works without Vault credentials'); +select ok((select request_id is not null from invocation), 'local startup returns the request ID'); +select is((select url from net.http_request_queue where id = (select request_id from invocation)), + 'http://kong:8000/functions/v1/my-function', 'uses the local URL'); +select ok((select headers->>'Authorization' is null from net.http_request_queue where id = (select request_id from invocation)), + 'local startup preserves the local authentication fallback'); select finish(); rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/credentials_pgflow_auth_secret.test.sql b/pkgs/core/supabase/tests/ensure_workers/credentials_pgflow_auth_secret.test.sql index 596bf5e5a..45e1a8612 100644 --- a/pkgs/core/supabase/tests/ensure_workers/credentials_pgflow_auth_secret.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/credentials_pgflow_auth_secret.test.sql @@ -32,6 +32,9 @@ select ok( drop table test1_result; +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next credentials') +from pgflow.worker_functions; + -- Cleanup secrets for next test delete from vault.secrets where name in ('pgflow_auth_secret', 'supabase_service_role_key'); @@ -54,6 +57,9 @@ select ok( drop table test2_result; +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next credentials') +from pgflow.worker_functions; + -- Cleanup secrets for next test delete from vault.secrets where name = 'supabase_service_role_key'; diff --git a/pkgs/core/supabase/tests/ensure_workers/debounce.test.sql b/pkgs/core/supabase/tests/ensure_workers/debounce.test.sql index 413093452..6fb181e71 100644 --- a/pkgs/core/supabase/tests/ensure_workers/debounce.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/debounce.test.sql @@ -1,5 +1,4 @@ -- Test: ensure_workers() respects debounce window in production mode --- Note: Debounce only applies in production mode; local mode bypasses debounce begin; select plan(4); select pgflow_tests.reset_db(); @@ -42,6 +41,10 @@ select is( 'Function with last_invoked_at beyond timeout IS returned' ); +-- End the previous attempt before checking another retry-throttle scenario. +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test retry') +from pgflow.worker_functions; + -- TEST: Function with NULL last_invoked_at IS returned update pgflow.worker_functions set last_invoked_at = null @@ -53,6 +56,9 @@ select is( 'Function with NULL last_invoked_at IS returned' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test retry') +from pgflow.worker_functions; + -- TEST: Debounce uses debounce interval from function config update pgflow.worker_functions set debounce = '3 seconds'::interval, diff --git a/pkgs/core/supabase/tests/ensure_workers/disabled_functions.test.sql b/pkgs/core/supabase/tests/ensure_workers/disabled_functions.test.sql index a71bb0c74..07aa7123a 100644 --- a/pkgs/core/supabase/tests/ensure_workers/disabled_functions.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/disabled_functions.test.sql @@ -1,43 +1,12 @@ --- Test: ensure_workers() skips disabled functions begin; select plan(2); select pgflow_tests.reset_db(); - --- Setup: Register worker functions select pgflow.track_worker_function('enabled-function'); select pgflow.track_worker_function('disabled-function'); - --- Disable one function -update pgflow.worker_functions -set enabled = false -where function_name = 'disabled-function'; - --- Set last_invoked_at to past (beyond debounce window) for both -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- TEST: Only enabled functions are returned +update pgflow.worker_functions set enabled = false where function_name = 'disabled-function'; set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; - --- Use a CTE to capture results in one call (avoid debounce between assertions) -with results as ( - select * from pgflow.ensure_workers() -) -select is( - (select count(*) from results), - 1::bigint, - 'Only enabled functions are returned' -); - --- Reset debounce for second test -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -select is( - (select count(*) from pgflow.ensure_workers() where ensure_workers.function_name = 'enabled-function'), - 1::bigint, - 'Enabled function is returned' -); - +create temporary table invocation as select * from pgflow.ensure_workers(); +select is((select count(*) from invocation), 1::bigint, 'only enabled functions are returned'); +select is((select function_name from invocation), 'enabled-function', 'returns the enabled function'); select finish(); rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/http_request_queued.test.sql b/pkgs/core/supabase/tests/ensure_workers/http_request_queued.test.sql index 0faffd1b2..0c19478d5 100644 --- a/pkgs/core/supabase/tests/ensure_workers/http_request_queued.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/http_request_queued.test.sql @@ -1,93 +1,25 @@ --- Test: ensure_workers() queues HTTP request via pg_net +-- Inspect the same enqueued batch; later cron calls must not replace pending requests. begin; -select plan(5); +select plan(6); select pgflow_tests.reset_db(); - --- Clear any existing HTTP requests -delete from net._http_response; - --- Setup: Register a worker function select pgflow.track_worker_function('my-function'); - --- Set last_invoked_at to past (beyond debounce window) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Simulate local mode (will use local fallback credentials) +select pgflow.track_worker_function('my-function-b'); set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; - --- Execute ensure_workers() and capture request_id -with result as ( - select * from pgflow.ensure_workers() -) -select is( - (select count(*) from result), - 1::bigint, - 'One function was invoked' +create temporary table invocation as select * from pgflow.ensure_workers(); +select is((select count(*) from invocation), 2::bigint, 'queues distinct function replicas'); +select ok((select bool_and(request_id is not null) from invocation), 'returns queued request IDs'); +select is((select count(distinct request_id) from invocation), 2::bigint, 'each replica has a unique request ID'); +select results_eq( + $$select url from net.http_request_queue where id in (select request_id from invocation) order by url$$, + $$values ('http://kong:8000/functions/v1/my-function'::text), ('http://kong:8000/functions/v1/my-function-b'::text)$$, + 'addresses each distinct function name' ); - --- Check that an HTTP request was queued --- Note: pg_net queues requests but doesn't execute until transaction commits --- We check the internal request table to verify the request was created - --- TEST: Request was queued in net._http_response --- Note: The request may be in a pending state in net's internal tables --- For now, we verify the request_id returned is valid -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -) -select ok( - (select request_id is not null from result limit 1), - 'HTTP request was queued (request_id returned)' -); - --- TEST: HTTP request URL format is correct (base_url/function_name) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Store result in temp table to ensure ensure_workers() executes before we query the queue -select * into temporary test_url_result from pgflow.ensure_workers(); - -select ok( - (select url LIKE '%/functions/v1/my-function' - from net.http_request_queue - where id = (select request_id from test_url_result limit 1)), - 'HTTP request URL ends with /functions/v1/{function_name}' -); - -drop table test_url_result; - --- TEST: Multiple functions each get their own request -select pgflow.track_worker_function('function-two'); -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -) -select is( - (select count(distinct request_id) from result), - 2::bigint, - 'Each function gets its own request_id' -); - --- TEST: request_ids are unique -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -), -ids as ( - select request_id from result -) -select ok( - (select count(*) = count(distinct request_id) from ids), - 'All request_ids are unique' -); - +select ok(( + select bool_and(convert_from(q.body, 'UTF8')::jsonb->>'invocation_id' = wf.invocation_id::text) + from invocation as i + join net.http_request_queue as q on q.id = i.request_id + join pgflow.worker_functions as wf on wf.function_name = i.function_name +), 'HTTP body carries the matching reserved invocation ID'); +select is((select count(*) from pgflow.ensure_workers()), 0::bigint, 'another cron call cannot replace pending requests'); select finish(); rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/invocation_handshake.test.sql b/pkgs/core/supabase/tests/ensure_workers/invocation_handshake.test.sql new file mode 100644 index 000000000..25c2e32de --- /dev/null +++ b/pkgs/core/supabase/tests/ensure_workers/invocation_handshake.test.sql @@ -0,0 +1,54 @@ +begin; +select plan(10); +select pgflow_tests.reset_db(); +select pgflow.track_worker_function('handshake'); +create temporary table invocation as +select pgflow.reserve_worker_invocation('handshake') as id; + +select is( + pgflow.reconcile_worker_invocation('handshake', (select id from invocation)), + 'pending', + 'receipt validates without completing the handshake' +); +select is( + (select invocation_status from pgflow.worker_functions where function_name = 'handshake'), + 'pending', 'receipt keeps startup pending' +); +select is( + (select count(*) from pgflow.accept_worker_invocation('handshake', (select id from invocation), + '69600000-0000-0000-0000-000000000001', 'test_queue', 'test')), + 1::bigint, 'readiness atomically accepts and registers a worker' +); +select ok( + (select last_heartbeat_at > transaction_timestamp() from pgflow.workers + where worker_id = '69600000-0000-0000-0000-000000000001'), + 'readiness uses the database wall clock, not the older transaction timestamp' +); +select is( + (select invocation_worker_id::text from pgflow.worker_functions where function_name = 'handshake'), + '69600000-0000-0000-0000-000000000001', 'ready state identifies the accepted worker' +); +select is( + (select count(*) from pgflow.accept_worker_invocation('handshake', (select id from invocation), + '69600000-0000-0000-0000-000000000001', 'test_queue', 'test')), + 1::bigint, 'lost acknowledgement can recover the same registration' +); +select is( + (select count(*) from pgflow.accept_worker_invocation('handshake', (select id from invocation), + '69600000-0000-0000-0000-000000000002', 'test_queue', 'test')), + 0::bigint, 'another runtime cannot accept the same invocation' +); +select is( + pgflow.dismiss_worker_invocation('handshake', (select id from invocation), 'late failure'), + false, 'late dismissal cannot undo readiness' +); +select is( + pgflow.reconcile_worker_invocation('other-function', (select id from invocation)), + 'rejected', 'IDs cannot address another function' +); +select is( + (select count(*) from pgflow.ensure_workers() where function_name = 'handshake'), + 0::bigint, 'completed handshake is not a reason to invoke a healthy worker' +); +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/invocation_recovery.test.sql b/pkgs/core/supabase/tests/ensure_workers/invocation_recovery.test.sql new file mode 100644 index 000000000..078bf9b22 --- /dev/null +++ b/pkgs/core/supabase/tests/ensure_workers/invocation_recovery.test.sql @@ -0,0 +1,47 @@ +begin; +select plan(14); +select pgflow_tests.reset_db(); +select pgflow.track_worker_function('recovery'); +create temporary table tokens as +select pgflow.reserve_worker_invocation('recovery') as old_id, null::uuid as new_id; + +select ok(pgflow.dismiss_worker_invocation('recovery', (select old_id from tokens), 'initialization failed'), + 'startup failure dismisses the current pending invocation'); +select is((select invocation_error from pgflow.worker_functions where function_name = 'recovery'), + 'initialization failed', 'failure reason is observable'); +select is(pgflow.reconcile_worker_invocation('recovery', (select old_id from tokens)), + 'rejected', 'failed invocation cannot authorize startup'); +select is(pgflow.reserve_worker_invocation('recovery'), null::uuid, + 'dismissal preserves debounce as a retry throttle'); +update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds'; +update tokens set new_id = pgflow.reserve_worker_invocation('recovery'); +select isnt((select new_id from tokens), (select old_id from tokens), 'retry has a new invocation ID'); +select is(pgflow.dismiss_worker_invocation('recovery', (select old_id from tokens), 'late failure'), + false, 'old dismissal cannot clear a newer pending invocation'); +select is(pgflow.reconcile_worker_invocation('recovery', (select old_id from tokens)), + 'rejected', 'superseded delivery is rejected'); + +-- Simulate delivery loss or initialization that outlives the bounded timeout. +update pgflow.worker_functions +set invocation_expires_at = clock_timestamp() - interval '1 second', + last_invoked_at = clock_timestamp() - interval '10 seconds'; +select is(pgflow.reconcile_worker_invocation('recovery', (select new_id from tokens)), + 'rejected', 'late delivery after expiry is rejected'); +select is((select count(*) from pgflow.accept_worker_invocation('recovery', (select new_id from tokens), + '69600000-0000-0000-0000-000000000003', 'queue', 'test')), + 0::bigint, 'expiry during initialization rejects final readiness'); +select ok(pgflow.reserve_worker_invocation('recovery') is not null, + 'lost delivery recovers after pending timeout without acknowledgement'); +select is((select count(*) from pgflow.accept_worker_invocation('recovery', null, + '69600000-0000-0000-0000-000000000003', 'queue', 'test')), + 0::bigint, 'NULL tokens cannot authorize new processing'); + +update pgflow.worker_functions set enabled = false; +select is(pgflow.reconcile_worker_invocation('recovery', + (select invocation_id from pgflow.worker_functions where function_name = 'recovery')), + 'rejected', 'disabling a registration rejects pending deliveries'); +select is(pgflow.reserve_worker_invocation('recovery'), null::uuid, 'disabled functions cannot reserve'); +select pgflow.track_worker_function('process-only', 'process'); +select is(pgflow.reserve_worker_invocation('process-only'), null::uuid, 'process workers stay outside HTTP scheduling'); +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/local_mode_always_pings.test.sql b/pkgs/core/supabase/tests/ensure_workers/local_mode_always_pings.test.sql deleted file mode 100644 index 95761d657..000000000 --- a/pkgs/core/supabase/tests/ensure_workers/local_mode_always_pings.test.sql +++ /dev/null @@ -1,52 +0,0 @@ --- Test: ensure_workers() in local mode always returns functions (ignores worker state) -begin; -select plan(3); -select pgflow_tests.reset_db(); - --- Setup: Register worker function -select pgflow.track_worker_function('my-function'); -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Setup: Create an alive worker -insert into pgflow.workers (worker_id, queue_name, function_name, started_at, last_heartbeat_at) -values ('11111111-1111-1111-1111-111111111111', 'test_queue', 'my-function', now(), now()); - --- Simulate local mode -set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; - --- TEST: Local mode returns function even with alive worker -select is( - (select count(*) from pgflow.ensure_workers()), - 1::bigint, - 'Local mode returns function even with alive worker' -); - --- TEST: Local mode returns function even when worker has recent heartbeat -update pgflow.workers -set last_heartbeat_at = now() -where worker_id = '11111111-1111-1111-1111-111111111111'; - --- Reset last_invoked_at to past again (it was updated by previous call) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -select is( - (select count(*) from pgflow.ensure_workers()), - 1::bigint, - 'Local mode returns function even when worker has fresh heartbeat' -); - --- TEST: Debounce is bypassed in local mode --- Reset last_invoked_at to now (will be within debounce window) -update pgflow.worker_functions -set last_invoked_at = now(); - -select is( - (select count(*) from pgflow.ensure_workers()), - 1::bigint, - 'Debounce is bypassed in local mode' -); - -select finish(); -rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/local_mode_handshake.test.sql b/pkgs/core/supabase/tests/ensure_workers/local_mode_handshake.test.sql new file mode 100644 index 000000000..734e42ec5 --- /dev/null +++ b/pkgs/core/supabase/tests/ensure_workers/local_mode_handshake.test.sql @@ -0,0 +1,23 @@ +-- Local development uses the same pending/readiness protocol as production. +begin; +select plan(4); +select pgflow_tests.reset_db(); +set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; +select pgflow.track_worker_function('my-function'); +insert into pgflow.workers (worker_id, queue_name, function_name) +values ('11111111-1111-1111-1111-111111111111', 'test_queue', 'my-function'); +select is((select count(*) from pgflow.ensure_workers()), 0::bigint, + 'local cron does not invoke healthy functions'); +update pgflow.workers set stopped_at = clock_timestamp() +where worker_id = '11111111-1111-1111-1111-111111111111'; +update pgflow.worker_functions set last_invoked_at = clock_timestamp(); +select is((select count(*) from pgflow.ensure_workers()), 0::bigint, + 'local cron preserves the retry throttle'); +update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds'; +select is((select count(*) from pgflow.ensure_workers()), 1::bigint, + 'local cron reserves startup for an unhealthy function'); +update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds'; +select is((select count(*) from pgflow.ensure_workers()), 0::bigint, + 'local cron does not replace pending startup after debounce'); +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/pending_start.test.sql b/pkgs/core/supabase/tests/ensure_workers/pending_start.test.sql new file mode 100644 index 000000000..57009999d --- /dev/null +++ b/pkgs/core/supabase/tests/ensure_workers/pending_start.test.sql @@ -0,0 +1,28 @@ +-- A request remains pending after the retry throttle passes, before readiness. +begin; +select plan(2); +select pgflow_tests.reset_db(); +set local app.settings.jwt_secret = 'production-secret'; +select vault.create_secret('test-key', 'supabase_service_role_key'); +select vault.create_secret('testproject', 'supabase_project_id'); +select pgflow.track_worker_function('slow-start'); + +select is( + (select count(*) from pgflow.ensure_workers()), + 1::bigint, + 'queues the first startup request' +); + +-- Initialization has not registered a worker. Advance past debounce without +-- waiting on the wall clock; the pending timeout must still prevent a retry. +update pgflow.worker_functions +set last_invoked_at = clock_timestamp() - interval '10 seconds' +where function_name = 'slow-start'; + +select is( + (select count(*) from pgflow.ensure_workers()), + 0::bigint, + 'does not queue another request while initialization is pending beyond debounce' +); +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/request_id_returned.test.sql b/pkgs/core/supabase/tests/ensure_workers/request_id_returned.test.sql index bdc6eec96..a878cecd0 100644 --- a/pkgs/core/supabase/tests/ensure_workers/request_id_returned.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/request_id_returned.test.sql @@ -1,52 +1,11 @@ --- Test: ensure_workers() returns request_id in result begin; select plan(3); select pgflow_tests.reset_db(); - --- Setup: Register a worker function select pgflow.track_worker_function('my-function'); - --- Set last_invoked_at to past (beyond debounce window) -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- Simulate local mode (will use local fallback credentials) set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'; - --- TEST: Result includes request_id column (verify by selecting it) -with result as ( - select function_name, invoked, request_id from pgflow.ensure_workers() -) -select ok( - (select count(*) = 1 from result where request_id is not null), - 'ensure_workers() returns request_id column' -); - --- Reset debounce for next test -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - --- TEST: request_id is NOT NULL when function is invoked -with result as ( - select * from pgflow.ensure_workers() -) -select ok( - (select request_id is not null from result where function_name = 'my-function'), - 'request_id is NOT NULL when function is invoked' -); - --- TEST: request_id is a valid bigint (positive number from pg_net) --- Reset debounce for next call -update pgflow.worker_functions -set last_invoked_at = now() - interval '10 seconds'; - -with result as ( - select * from pgflow.ensure_workers() -) -select ok( - (select request_id > 0 from result where function_name = 'my-function'), - 'request_id is a positive bigint from pg_net' -); - +create temporary table invocation as select function_name, invoked, request_id from pgflow.ensure_workers(); +select is((select count(*) from invocation where request_id is not null), 1::bigint, 'returns the request_id column'); +select ok((select request_id is not null from invocation where function_name = 'my-function'), 'request_id is not NULL'); +select ok((select request_id > 0 from invocation), 'request_id is a positive bigint from pg_net'); select finish(); rollback; diff --git a/pkgs/core/supabase/tests/ensure_workers/returns_functions_to_invoke.test.sql b/pkgs/core/supabase/tests/ensure_workers/returns_functions_to_invoke.test.sql index ef5d60bf9..550c15d7a 100644 --- a/pkgs/core/supabase/tests/ensure_workers/returns_functions_to_invoke.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/returns_functions_to_invoke.test.sql @@ -25,6 +25,9 @@ select is( 'In local mode, returns all enabled worker functions' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next environment') +from pgflow.worker_functions; + -- Reset last_invoked_at (was updated by previous ensure_workers call) update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds'; @@ -39,6 +42,9 @@ select is( 'In production mode with no workers, returns all functions' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next worker state') +from pgflow.worker_functions; + -- Reset last_invoked_at again update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds'; @@ -54,6 +60,9 @@ select is( 'In production mode, function with alive worker is NOT returned' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next worker state') +from pgflow.worker_functions; + -- Reset last_invoked_at for function-b only (function-a shouldn't be returned anyway) update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds' diff --git a/pkgs/core/supabase/tests/ensure_workers/skips_process_start_mode.test.sql b/pkgs/core/supabase/tests/ensure_workers/skips_process_start_mode.test.sql index 1447e7dcb..2b27dcd79 100644 --- a/pkgs/core/supabase/tests/ensure_workers/skips_process_start_mode.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/skips_process_start_mode.test.sql @@ -47,6 +47,9 @@ select results_eq( -- Clear the request queue from the local test delete from net._http_response; +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next environment') +from pgflow.worker_functions; + -- Reset debounce timestamps update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds'; diff --git a/pkgs/core/supabase/tests/ensure_workers/worker_death_detection.test.sql b/pkgs/core/supabase/tests/ensure_workers/worker_death_detection.test.sql index 1e0e15e79..6c57383ce 100644 --- a/pkgs/core/supabase/tests/ensure_workers/worker_death_detection.test.sql +++ b/pkgs/core/supabase/tests/ensure_workers/worker_death_detection.test.sql @@ -41,6 +41,9 @@ select is( 'Function with stopped worker IS returned' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next worker state') +from pgflow.worker_functions; + -- Reset stopped_at, set deprecated_at instead update pgflow.workers set stopped_at = null, deprecated_at = now() @@ -57,6 +60,9 @@ select is( 'Function with deprecated worker IS returned' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next worker state') +from pgflow.worker_functions; + -- Reset deprecated_at, make heartbeat stale update pgflow.workers set deprecated_at = null, @@ -74,6 +80,9 @@ select is( 'Function with stale heartbeat worker IS returned' ); +select pgflow.dismiss_worker_invocation(function_name, invocation_id, 'test next worker state') +from pgflow.worker_functions; + -- TEST: Mix of alive and dead workers - function with at least one alive is NOT returned -- Reset first worker to dead update pgflow.workers diff --git a/pkgs/edge-worker/src/core/Queries.ts b/pkgs/edge-worker/src/core/Queries.ts index 44220dc6a..f787c0fa0 100644 --- a/pkgs/edge-worker/src/core/Queries.ts +++ b/pkgs/edge-worker/src/core/Queries.ts @@ -1,4 +1,5 @@ import type postgres from 'postgres'; +import { withStartupTimeout } from './withStartupTimeout.js'; import type { WorkerRow, WorkerStartMode } from './types.js'; import type { FlowShape, @@ -19,6 +20,13 @@ export interface EnsureFlowCompiledResult { mismatchKind?: EnsureFlowMismatchKind; } +export class WorkerInvocationRejectedError extends Error { + constructor() { + super('Worker invocation is expired, superseded, dismissed, or already accepted'); + this.name = 'WorkerInvocationRejectedError'; + } +} + export class Queries { constructor(private readonly sql: postgres.Sql) {} @@ -27,12 +35,28 @@ export class Queries { workerId, edgeFunctionName, pgflowVersion, + invocationId, }: { queueName: string; workerId: string; edgeFunctionName: string; pgflowVersion: string; + invocationId?: string; }): Promise { + if (invocationId) { + const accept = () => withStartupTimeout(this.sql, (sql) => sql` + SELECT * FROM pgflow.accept_worker_invocation( + ${edgeFunctionName}, ${invocationId}::uuid, ${workerId}::uuid, + ${queueName}, ${pgflowVersion} + ) + `, invocationId); + // A connection can fail after COMMIT. Repeat the idempotent operation + // once to recover the same worker, never to authorize another runtime. + const [worker] = await accept().catch(() => accept()); + if (!worker) throw new WorkerInvocationRejectedError(); + return worker; + } + const [worker] = await this.sql` INSERT INTO pgflow.workers (queue_name, worker_id, function_name, pgflow_version) VALUES (${queueName}, ${workerId}, ${edgeFunctionName}, ${pgflowVersion}) @@ -68,7 +92,8 @@ export class Queries { flowSlug: string, shape: FlowShape, queueMode: QueueMode = 'flow', - routes: readonly StepRoute[] | null = null + routes: readonly StepRoute[] | null = null, + invocationId?: string ): Promise { // SAFETY: FlowShape is JSON-compatible by construction (only strings, numbers, // arrays, and plain objects), but TypeScript can't prove this because FlowShape @@ -84,21 +109,21 @@ export class Queries { : this.sql.json( routes.map(({ stepSlug, queueName }) => ({ stepSlug, queueName })) as unknown as Json ); - const [result] = await this.sql<{ result: EnsureFlowCompiledResult }[]>` + const [result] = await withStartupTimeout(this.sql, (sql) => sql<{ result: EnsureFlowCompiledResult }[]>` SELECT pgflow.ensure_flow_compiled( ${flowSlug}, ${shapeJson}::jsonb, ${queueMode}::text, ${routesJson}::jsonb ) as result - `; + `, invocationId); return result.result; } /** * Registers an edge function for monitoring by ensure_workers() cron. - * Called by workers on startup. Sets last_invoked_at to prevent cron from - * pinging during startup (debounce). + * Called by process workers and tokenless low-level bootstrap. HTTP handlers + * reserve a startup invocation before initialization instead. */ async trackWorkerFunction(functionName: string, startMode: WorkerStartMode = 'http'): Promise { if (startMode === 'http') { @@ -113,6 +138,53 @@ export class Queries { `; } + async registerHttpWorkerFunction(functionName: string): Promise { + // Manual startup may discover an unregistered function, but cannot enable + // a disabled registration or convert a process worker into an HTTP worker. + await this.sql` + INSERT INTO pgflow.worker_functions (function_name) + VALUES (${functionName}) ON CONFLICT (function_name) DO NOTHING + `; + } + + async isWorkerHealthy(functionName: string, workerId: string): Promise { + const [result] = await this.sql<{ healthy: boolean }[]>` + SELECT EXISTS ( + SELECT 1 FROM pgflow.workers as w + JOIN pgflow.worker_functions as wf ON wf.function_name = w.function_name + WHERE wf.function_name = ${functionName} AND w.worker_id = ${workerId}::uuid + AND wf.enabled AND wf.start_mode = 'http' + AND w.stopped_at IS NULL AND w.deprecated_at IS NULL + AND w.last_heartbeat_at > clock_timestamp() - wf.debounce + ) as healthy + `; + return result.healthy; + } + + async reserveWorkerInvocation(functionName: string): Promise { + const [result] = await this.sql<{ id: string | null }[]>` + SELECT pgflow.reserve_worker_invocation(${functionName}) as id + `; + return result.id; + } + + async reconcileWorkerInvocation( + functionName: string, invocationId: string, workerId: string | null + ): Promise<'pending' | 'ready' | 'rejected'> { + const [result] = await this.sql<{ status: 'pending' | 'ready' | 'rejected' }[]>` + SELECT pgflow.reconcile_worker_invocation( + ${functionName}, ${invocationId}::uuid, ${workerId}::uuid + ) as status + `; + return result.status; + } + + async dismissWorkerInvocation(functionName: string, invocationId: string, reason: string): Promise { + await this.sql` + SELECT pgflow.dismiss_worker_invocation(${functionName}, ${invocationId}::uuid, ${reason}) + `; + } + /** * Marks a worker as stopped for graceful shutdown signaling. * Called by workers on beforeunload to allow ensure_workers() to detect death immediately. diff --git a/pkgs/edge-worker/src/core/WorkerLifecycle.ts b/pkgs/edge-worker/src/core/WorkerLifecycle.ts index 03e667782..9d1b0400c 100644 --- a/pkgs/edge-worker/src/core/WorkerLifecycle.ts +++ b/pkgs/edge-worker/src/core/WorkerLifecycle.ts @@ -32,10 +32,15 @@ export class WorkerLifecycle implements InternalLifecycle // Register this edge function for monitoring by ensure_workers() cron. const startMode = workerBootstrap.startMode ?? 'http'; - await this.queries.trackWorkerFunction(workerBootstrap.edgeFunctionName, startMode); + if (!workerBootstrap.invocationId) { + if (startMode === 'http') { + this.logger.warn('Tokenless low-level bootstrap bypasses the HTTP startup handshake; use EdgeWorker.start() on Supabase'); + } + await this.queries.trackWorkerFunction(workerBootstrap.edgeFunctionName, startMode); + } this.logger.debug(`Ensuring queue '${this.queue.queueName}' exists...`); - await this.queue.safeCreate(); + await this.queue.safeCreate(workerBootstrap.invocationId); this.workerRow = await this.queries.onWorkerStarted({ queueName: this.queueName, diff --git a/pkgs/edge-worker/src/core/types.ts b/pkgs/edge-worker/src/core/types.ts index 6ac3d63a9..df71b3af7 100644 --- a/pkgs/edge-worker/src/core/types.ts +++ b/pkgs/edge-worker/src/core/types.ts @@ -63,4 +63,6 @@ export interface WorkerBootstrap { edgeFunctionName: string; workerId: string; startMode?: WorkerStartMode; + /** Current SQL startup reservation, required by the Supabase HTTP handler. */ + invocationId?: string; } diff --git a/pkgs/edge-worker/src/core/withStartupTimeout.ts b/pkgs/edge-worker/src/core/withStartupTimeout.ts new file mode 100644 index 000000000..b0143231d --- /dev/null +++ b/pkgs/edge-worker/src/core/withStartupTimeout.ts @@ -0,0 +1,17 @@ +import type postgres from 'postgres'; + +/** Bound initialization SQL without changing timeouts for task handlers. */ +export async function withStartupTimeout( + sql: postgres.Sql, + query: (sql: postgres.Sql | postgres.TransactionSql) => PromiseLike, + invocationId?: string +): Promise { + if (!invocationId) return await query(sql); + return await sql.begin(async (transaction) => { + await transaction`SELECT + set_config('statement_timeout', '60s', true), + set_config('idle_in_transaction_session_timeout', '60s', true) + `; + return await query(transaction); + }) as T; +} diff --git a/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts b/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts index 93cd3cb8d..56b4b9b28 100644 --- a/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts +++ b/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts @@ -56,11 +56,16 @@ export class FlowWorkerLifecycle implements InternalLifec this._edgeFunctionName = workerBootstrap.edgeFunctionName; // Compile/verify the flow before any registration write - const compilationStatus = await this.ensureFlowCompiled(); + const compilationStatus = await this.ensureFlowCompiled(workerBootstrap.invocationId); // Register this edge function for monitoring by ensure_workers() cron. const startMode = workerBootstrap.startMode ?? 'http'; - await this.queries.trackWorkerFunction(workerBootstrap.edgeFunctionName, startMode); + if (!workerBootstrap.invocationId) { + if (startMode === 'http') { + this.logger.warn('Tokenless low-level bootstrap bypasses the HTTP startup handshake; use EdgeWorker.start() on Supabase'); + } + await this.queries.trackWorkerFunction(workerBootstrap.edgeFunctionName, startMode); + } // Log startup banner with compilation status this.logStartupBanner(compilationStatus); @@ -74,7 +79,7 @@ export class FlowWorkerLifecycle implements InternalLifec this.workerState.transitionTo(States.Running); } - private async ensureFlowCompiled(): Promise { + private async ensureFlowCompiled(invocationId?: string): Promise { const shape = extractFlowShape(this.flow); // Queue mode and the complete ordered route map travel with the shape as @@ -84,7 +89,8 @@ export class FlowWorkerLifecycle implements InternalLifec this.flow.slug, shape, this.routing.queueMode, - this.routing.routes + this.routing.routes, + invocationId ); if (result.status === 'mismatch') { diff --git a/pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts b/pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts index bb790252e..486045beb 100644 --- a/pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts +++ b/pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts @@ -15,7 +15,7 @@ import { resolveConnectionString, resolveSqlConnection, } from './resolveConnection.js'; -import { Queries } from '../core/Queries.js'; +import { Queries, WorkerInvocationRejectedError } from '../core/Queries.js'; import { getPlatformDeps, type SupabasePlatformDeps } from './deps.js'; /** @@ -37,6 +37,10 @@ interface SupabaseEnv extends Record { */ const SUPABASE_SHUTDOWN_DEADLINE_MS = 5_000; +// A local watchdog releases a hung runtime's startup guard. SQL's 60-second +// database-clock expiry still authorizes readiness; this timer never does. +const SUPABASE_STARTUP_TIMEOUT_MS = 60_000; + /** * Supabase platform adapter for Deno runtime environment. * IMPORTANT: This class assumes it is running within a Deno environment @@ -334,36 +338,70 @@ export class SupabasePlatformAdapter implements PlatformAdapter this.logger.error('Could not dismiss failed startup', dismissError)); + } + if (error instanceof WorkerInvocationRejectedError) { + return Response.json({ status: 'invocation_rejected', invocationId }, { status: 409 }); + } this.logger.error('Worker startup failed', error); return createServerErrorResponse(); } }); } - private needsWorkerReplacement(): boolean { - return !this.worker || this.worker.isDeprecated || this.worker.isStopped; - } - /** * Serializes worker startup across concurrent HTTP requests: every request * waits for an in-flight replacement, exactly one request owns each * replacement, and once `stopPromise` exists no request can admit a new * replacement or report readiness after shutdown began. */ - private async ensureWorkerStarted(req: Request, createWorkerFn: CreateWorkerFn): Promise { + private async ensureWorkerStarted( + functionName: string, invocationId: string, createWorkerFn: CreateWorkerFn + ): Promise { for (;;) { if (this.stopPromise) { throw new Error('Worker startup rejected: shutdown in progress'); @@ -377,13 +415,26 @@ export class SupabasePlatformAdapter implements PlatformAdapter { + private async replaceWorker( + functionName: string, invocationId: string, createWorkerFn: CreateWorkerFn + ): Promise { const previousWorkerId = this.workerId; if (this.worker) { @@ -411,14 +464,9 @@ export class SupabasePlatformAdapter implements PlatformAdapter | undefined; + let timedOut = false; try { - await worker.startOnlyOnce({ - edgeFunctionName: this.edgeFunctionName, - workerId, - startMode: 'http', - }); + await Promise.race([ + worker.startOnlyOnce({ + edgeFunctionName: this.edgeFunctionName, + workerId, + startMode: 'http', + invocationId, + }), + new Promise((_resolve, reject) => { + timer = setTimeout(() => { + timedOut = true; + reject(new Error(`Worker startup timed out after ${SUPABASE_STARTUP_TIMEOUT_MS}ms`)); + }, SUPABASE_STARTUP_TIMEOUT_MS); + }), + ]); } catch (error) { + if (timedOut) { + // stop() aborts before it awaits initialization. Do not await a hung + // startup here: if it resumes, the aborted worker cannot process. + void worker.stop().catch((stopError) => this.logger.error('Timed-out startup cleanup failed', stopError)); + } if (this.worker === worker) { this.worker = null; } throw error; + } finally { + clearTimeout(timer); } } diff --git a/pkgs/edge-worker/src/queue/Queue.ts b/pkgs/edge-worker/src/queue/Queue.ts index 11afb2e73..13fc1fcca 100644 --- a/pkgs/edge-worker/src/queue/Queue.ts +++ b/pkgs/edge-worker/src/queue/Queue.ts @@ -1,4 +1,5 @@ import type postgres from 'postgres'; +import { withStartupTimeout } from '../core/withStartupTimeout.js'; import type { PgmqMessageRecord } from './types.js'; import type { Json } from '../core/types.js'; import type { Logger } from '../platform/types.js'; @@ -18,14 +19,14 @@ export class Queue { * Creates a queue if it doesn't exist. * If the queue already exists, this method does nothing. */ - async safeCreate() { + async safeCreate(invocationId?: string) { this.logger.debug(`Creating queue '${this.queueName}' if it doesn't exist`); - return await this.sql` + return await withStartupTimeout(this.sql, (sql) => sql` select * from pgmq.create(${this.queueName}) where not exists ( select 1 from pgmq.list_queues() where queue_name = ${this.queueName} ); - `; + `, invocationId); } /** diff --git a/pkgs/edge-worker/tests/integration/workerInvocation.test.ts b/pkgs/edge-worker/tests/integration/workerInvocation.test.ts new file mode 100644 index 000000000..4700b5123 --- /dev/null +++ b/pkgs/edge-worker/tests/integration/workerInvocation.test.ts @@ -0,0 +1,484 @@ +import { assertEquals, assert } from '@std/assert'; +import { delay } from '@std/async'; +import { FakeTime } from '@std/testing/time'; +import postgres from 'postgres'; +import { Flow } from '@pgflow/dsl'; +import { SupabasePlatformAdapter } from '../../src/platform/SupabasePlatformAdapter.ts'; +import { createFlowWorker } from '../../src/flow/createFlowWorker.ts'; +import { createQueueWorker } from '../../src/queue/createQueueWorker.ts'; +import { withPgNoTransaction } from '../db.ts'; +import { integrationConfig } from '../config.ts'; +import { waitForRunCompletion } from './flow/_testHelpers.ts'; + +const functionName = 'invocation-test'; +const flow = new Flow({ slug: 'invocation_test' }).step( + { slug: 'one' }, () => 'done' +); + +async function runtime(options: { local?: boolean; queue?: boolean; loseAcknowledgement?: boolean; hangCompilation?: boolean; holdReconciliation?: boolean; maxConnections?: number } = {}) { + let handle!: (req: Request) => Response | Promise; + let loseAcknowledgement = options.loseAcknowledgement ?? false; + let hangCompilation = options.hangCompilation ?? false; + let holdReconciliation = options.holdReconciliation ?? false; + let releaseReconciliation = () => {}; + let reconciliationHeld = () => {}; + const reconciled = new Promise((resolve) => { reconciliationHeld = resolve; }); + let releaseCompilation = () => {}; + let compilationHeld = () => {}; + const held = new Promise((resolve) => { compilationHeld = resolve; }); + const connection = postgres(integrationConfig.dbUrl, { + prepare: false, + max: options.maxConnections, + connection: { + 'app.settings.jwt_secret': options.local + ? 'super-secret-jwt-token-with-at-least-32-characters-long' : 'production-secret', + }, + }); + // Fault injection at the database transport boundary: commit the real + // acceptance, then lose its response before the caller observes it. + let loseOnCommit = false; + let hangOnCommit = false; + function instrument(client: T): T { + return new Proxy(client, { + get(target, key, receiver) { + if (key === 'begin') return async (callback: (tx: postgres.TransactionSql) => Promise) => { + const rows = await (target as postgres.Sql).begin((tx) => callback(instrument(tx))); + if (loseOnCommit) { + loseOnCommit = false; + throw new Error('acknowledgement response lost'); + } + if (hangOnCommit) { + hangOnCommit = false; + return new Promise((resolve) => { + releaseCompilation = () => resolve(rows); + compilationHeld(); + }); + } + return rows; + }; + return Reflect.get(target, key, receiver); + }, + apply(target, thisArg, args) { + const query = args[0].join(''); + if (hangCompilation && query.includes('ensure_flow_compiled')) { + hangCompilation = false; + hangOnCommit = true; + } + if (loseAcknowledgement && query.includes('accept_worker_invocation')) { + loseAcknowledgement = false; + loseOnCommit = true; + } + const result = Reflect.apply(target, thisArg, args); + if (holdReconciliation && query.includes('reconcile_worker_invocation')) { + holdReconciliation = false; + return result.then((rows: unknown) => new Promise((resolve) => { + releaseReconciliation = () => resolve(rows); + reconciliationHeld(); + })); + } + return result; + }, + }); + } + const sql = instrument(connection); + const adapter = new SupabasePlatformAdapter({ sql }, { + getEnv: () => ({ + SUPABASE_URL: options.local ? 'http://kong:8000' : 'https://test.supabase.co', + SUPABASE_ANON_KEY: 'test-anon', + SUPABASE_SERVICE_ROLE_KEY: 'test-service-key', + SB_EXECUTION_ID: crypto.randomUUID(), + EDGE_WORKER_LOG_LEVEL: 'error', + }), + serve: (handler) => { handle = handler; }, + onShutdown: () => {}, + extendLifetime: () => {}, + }); + if (options.hangCompilation || options.maxConnections === 1) await connection`select 1`; + await adapter.startWorker((logger) => options.queue + ? createQueueWorker(() => {}, { sql, queueName: flow.slug, maxPollSeconds: 1 }, logger, adapter) + : createFlowWorker(flow, { sql, maxPollSeconds: 1, pollIntervalMs: 10 }, logger, adapter) + ); + return { + adapter, + held, + reconciled, + releaseReconciliation: () => releaseReconciliation(), + releaseCompilation: () => releaseCompilation(), + request: (invocationId?: string, auth = true, name = functionName) => handle(new Request(`https://test/${name}`, { + method: 'POST', + headers: { ...(auth ? { Authorization: 'Bearer test-service-key' } : {}), 'Content-Type': 'application/json' }, + body: JSON.stringify(invocationId === undefined ? {} : { invocation_id: invocationId }), + })), + }; +} + +Deno.test('independent runtimes cannot start two workers for duplicate delivery', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.create_flow(${flow.slug})`; + await sql`select pgflow.add_step(${flow.slug}, 'one')`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const runtimes = await Promise.all([runtime(), runtime({ queue: true })]); + try { + const responses = await Promise.all(runtimes.map((r) => r.request(id))); + assertEquals(responses.map((r) => r.status).sort(), [200, 409]); + const [{ count }] = await sql` + select count(*)::int as count from pgflow.workers where function_name = ${functionName} + `; + assertEquals(count, 1); + } finally { + await Promise.all(runtimes.map((r) => r.adapter.stopWorker())); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('a delayed reconciliation cannot replace the worker started by a concurrent request', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const r = await runtime({ holdReconciliation: true }); + const delayed = Promise.resolve(r.request(id)); + try { + await r.reconciled; + const first = await r.request(id); + assertEquals(first.status, 200); + const started = await first.json(); + r.releaseReconciliation(); + const duplicate = await delayed; + assertEquals(duplicate.status, 200); + assertEquals(await duplicate.json(), { ...started, status: 'running' }); + const [{ run_id }] = await sql`select run_id from pgflow.start_flow(${flow.slug}, '{}'::jsonb)`; + const completed = await waitForRunCompletion(sql, run_id, { timeoutMs: 5_000, pollIntervalMs: 20 }); + assertEquals(completed.status, 'completed'); + assertEquals(completed.output, { one: 'done' }); + } finally { + r.releaseReconciliation(); + await delayed; + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +async function ensure(sql: postgres.Sql) { + return await sql.begin(async (tx) => { + await tx`set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long'`; + return await tx`select * from pgflow.ensure_workers()`; + }); +} + +async function waitForCompilation(sql: postgres.Sql) { + for (let attempt = 0; attempt < 100; attempt++) { + const [{ waiting }] = await sql` + select exists ( + select 1 from pg_stat_activity + where wait_event = 'advisory' and query like '%ensure_flow_compiled%' + ) as waiting + `; + if (waiting) return; + await delay(10); + } + throw new Error('Startup did not reach the compilation barrier'); +} + +Deno.test('initialization beyond debounce keeps one pending enqueue across concurrent cron calls', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.track_worker_function(${functionName})`; + await sql`update pgflow.worker_functions set debounce = interval '1 second' where function_name = ${functionName}`; + const queued = await ensure(sql); + assertEquals(queued.length, 1); + const [{ invocation_id: id }] = await sql`select invocation_id from pgflow.worker_functions where function_name = ${functionName}`; + const lock = await sql.reserve(); + await lock`select pg_advisory_lock(1, hashtext(${flow.slug}))`; + const r = await runtime(); + const response = r.request(id); + try { + await waitForCompilation(sql); + await sql`select pg_sleep(1.05)`; + const [{ pending }] = await sql` + select invocation_status = 'pending' and invocation_expires_at > clock_timestamp() + and last_invoked_at < clock_timestamp() - debounce as pending + from pgflow.worker_functions where function_name = ${functionName} + `; + assert(pending, 'receipt must remain pending beyond debounce until readiness'); + const retries = await Promise.all(Array.from({ length: 8 }, () => ensure(sql))); + assertEquals(retries.flat().length, 0, 'concurrent cron calls cannot enqueue while initialization is held'); + const [{ count }] = await sql`select count(*)::int as count from pgflow.workers where function_name = ${functionName}`; + assertEquals(count, 0, 'receipt alone must not register an unready worker'); + } finally { + await lock`select pg_advisory_unlock(1, hashtext(${flow.slug}))`; + lock.release(); + assertEquals((await response).status, 200); + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('concurrent cron transactions reserve one request per distinct replica', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + for (const name of [functionName, `${functionName}-b`]) { + await sql`select pgflow.track_worker_function(${name})`; + } + try { + const batches = await Promise.all(Array.from({ length: 8 }, () => ensure(sql))); + assertEquals(batches.flat().map((row) => row.function_name).sort(), [functionName, `${functionName}-b`]); + const [{ count }] = await sql`select count(distinct invocation_id)::int as count from pgflow.worker_functions`; + assertEquals(count, 2); + } finally { + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('lost readiness response recovers the committed worker and duplicate delivery does not start another loop', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const r = await runtime({ loseAcknowledgement: true }); + try { + const first = await r.request(id); + assertEquals(first.status, 200); + const started = await first.json(); + const again = await r.request(id); + assertEquals(again.status, 200); + assertEquals(await again.json(), { ...started, status: 'running' }); + const [{ count }] = await sql`select count(*)::int as count from pgflow.workers where function_name = ${functionName}`; + assertEquals(count, 1); + } finally { + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('expiry or supersession during initialization rejects readiness and the same runtime handles the retry', + withPgNoTransaction(async (sql) => { + for (const supersede of [false, true]) { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const r = await runtime(); + const lock = await sql.reserve(); + await lock`select pg_advisory_lock(1, hashtext(${flow.slug}))`; + const response = r.request(id); + let newerId: string | undefined; + try { + await waitForCompilation(sql); + await sql`update pgflow.worker_functions set invocation_expires_at = clock_timestamp() - interval '1 second', + last_invoked_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + if (supersede) { + const [newer] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + newerId = newer.id; + } + } finally { + await lock`select pg_advisory_unlock(1, hashtext(${flow.slug}))`; + lock.release(); + } + try { + assertEquals((await response).status, 409, 'final readiness must revalidate the token'); + assertEquals((await r.request(id)).status, 409, 'late delivery cannot revive the old token'); + const [{ count }] = await sql`select count(*)::int as count from pgflow.workers where function_name = ${functionName}`; + assertEquals(count, 0); + if (supersede) { + const [pending] = await sql`select invocation_id, invocation_status from pgflow.worker_functions where function_name = ${functionName}`; + assertEquals(pending.invocation_id, newerId); + assertEquals(pending.invocation_status, 'pending', 'old failure must not dismiss the newer invocation'); + } else { + const [newer] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + newerId = newer.id; + } + assertEquals((await r.request(newerId)).status, 200, 'retry needs neither runtime eviction nor redeployment'); + } finally { + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + } + }) +); + +Deno.test('startup SQL timeout frees a single-connection runtime without redeployment', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const lock = await sql.reserve(); + await lock`select pg_advisory_lock(1, hashtext(${flow.slug}))`; + const r = await runtime({ maxConnections: 1 }); + const response = Promise.resolve(r.request(id)); + const deadline = new AbortController(); + try { + try { + await waitForCompilation(sql); + const settled = await Promise.race([ + response.then(() => true), + delay(65_000, { signal: deadline.signal }).then(() => false).catch(() => false), + ]); + assert(settled, 'SQL startup must release its connection without waiting for the compilation lock'); + assertEquals((await response).status, 500); + } finally { + deadline.abort(); + await lock`select pg_advisory_unlock(1, hashtext(${flow.slug}))`; + lock.release(); + await response; + } + await sql`update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + const [{ id: next }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + assertEquals((await r.request(next)).status, 200); + } finally { + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('hung initialization releases the runtime for a later invocation without redeployment', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + const r = await runtime({ hangCompilation: true }); + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id: firstId }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const realSetTimeout = globalThis.setTimeout; + const time = new FakeTime(); + let restored = false; + const first = Promise.resolve(r.request(firstId)); + let settled = false; + let held = false; + first.then(() => { settled = true; }); + r.held.then(() => { held = true; }); + try { + // Pump mocked timers while real SQL I/O reaches the transport barrier. + for (let i = 0; i < 100 && !held; i++) { + await time.tickAsync(1); + await new Promise((resolve) => realSetTimeout(resolve, 10)); + } + assert(held, 'initialization must reach the transport barrier'); + await time.tickAsync(60_000); + time.restore(); + restored = true; + // Dismissal uses real SQL I/O after the local watchdog fires. + for (let i = 0; i < 100 && !settled; i++) await delay(10); + assert(settled, 'a hung startup must not own the runtime startup guard forever'); + assertEquals((await first).status, 500); + // Database clock, not the local watchdog, authorizes the new reservation. + await sql`update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + assert(id); + const next = await r.request(id); + assertEquals(next.status, 200, 'the surviving runtime accepts a later invocation before the old initialization settles'); + r.releaseCompilation(); + const [{ run_id }] = await sql`select run_id from pgflow.start_flow(${flow.slug}, '{}'::jsonb)`; + const completed = await waitForRunCompletion(sql, run_id, { timeoutMs: 5_000, pollIntervalMs: 20 }); + assertEquals(completed.status, 'completed'); + assertEquals(completed.output, { one: 'done' }); + const [{ count }] = await sql`select count(*)::int as count from pgflow.workers where function_name = ${functionName}`; + assertEquals(count, 1); + } finally { + if (!restored) time.restore(); + r.releaseCompilation(); + await first; + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('startup failure dismisses the token and a later request recovers in the same runtime', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + await sql`select pgflow.create_flow(${flow.slug})`; + await sql`select pgflow.add_step(${flow.slug}, 'different')`; + await sql`select pgflow.track_worker_function(${functionName})`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + const r = await runtime(); + try { + assertEquals((await r.request(id)).status, 500); + const [failed] = await sql`select invocation_status, invocation_error from pgflow.worker_functions where function_name = ${functionName}`; + assertEquals(failed.invocation_status, 'failed'); + assert(failed.invocation_error.includes('mismatch')); + assertEquals((await r.request(id)).status, 409, 'dismissed tokens cannot authorize processing'); + await sql`select pgflow.delete_flow_and_data(${flow.slug})`; + await sql`update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + const [{ id: retry }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + assertEquals((await r.request(retry)).status, 200); + } finally { + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('authentication, manual startup and local development preserve the handshake and registration settings', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + const r = await runtime(); + const local = await runtime({ local: true }); + try { + assertEquals((await r.request(undefined, false)).status, 401); + assertEquals((await r.request('not-a-uuid')).status, 400); + const [{ count }] = await sql`select count(*)::int as count from pgflow.worker_functions`; + assertEquals(count, 0, 'unauthenticated and malformed requests cannot reserve'); + await sql`select pgflow.track_worker_function(${functionName})`; + await sql`update pgflow.worker_functions set enabled = false where function_name = ${functionName}`; + assertEquals((await r.request()).status, 409); + await sql`update pgflow.worker_functions set enabled = true, start_mode = 'process' where function_name = ${functionName}`; + assertEquals((await r.request()).status, 409); + const [process] = await sql`select start_mode, invocation_id from pgflow.worker_functions where function_name = ${functionName}`; + assertEquals(process.start_mode, 'process'); + assertEquals(process.invocation_id, null, 'manual requests cannot turn process workers into HTTP workers'); + await sql`delete from pgflow.worker_functions where function_name = ${functionName}`; + const replies = await Promise.all([r.request(), local.request(undefined, false)]); + assertEquals(replies.map((reply) => reply.status).sort(), [200, 409], 'manual/local starts share SQL reservations'); + const [{ invocation_status }] = await sql`select invocation_status from pgflow.worker_functions where function_name = ${functionName}`; + assertEquals(invocation_status, 'ready'); + } finally { + await Promise.all([r.adapter.stopWorker(), local.adapter.stopWorker()]); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); + +Deno.test('a healthy surviving runtime acknowledges a later valid invocation without another worker', + withPgNoTransaction(async (sql) => { + await sql`select pgflow_tests.reset_db()`; + const r = await runtime(); + try { + const first = await r.request(); + assertEquals(first.status, 200); + const started = await first.json(); + // Heartbeat loss can schedule a replacement (#697 owns lifetime fencing). + await sql`update pgflow.workers set last_heartbeat_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + await sql`update pgflow.worker_functions set last_invoked_at = clock_timestamp() - interval '10 seconds' where function_name = ${functionName}`; + const [{ id }] = await sql`select pgflow.reserve_worker_invocation(${functionName}) as id`; + assert(id); + await sql`update pgflow.workers set last_heartbeat_at = clock_timestamp() where function_name = ${functionName}`; + const ack = await r.request(id); + assertEquals(ack.status, 200); + assertEquals((await ack.json()).workerId, started.workerId); + assertEquals((await r.request(started.invocationId)).status, 409, 'superseded requests fail even in a ready runtime'); + assertEquals((await r.request(id)).status, 200, 'duplicate acknowledgement is idempotent'); + assertEquals((await ensure(sql)).length, 0, 'ready is not a reason to invoke again'); + const [{ count }] = await sql`select count(*)::int as count from pgflow.workers where function_name = ${functionName}`; + assertEquals(count, 1); + } finally { + await r.adapter.stopWorker(); + await sql`delete from pgflow.workers where function_name = ${functionName}`; + await sql`select pgflow_tests.reset_db()`; + } + }) +); diff --git a/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts b/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts index afe0cc349..48d2fc344 100644 --- a/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts +++ b/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts @@ -34,6 +34,19 @@ function createMockDeps(overrides?: Partial): SupabasePlat }; } +const invocationId = '69600000-0000-0000-0000-000000000001'; + +// SQL is a transport boundary here. Protocol arbitration uses real SQL in +// integration/workerInvocation.test.ts; these tests keep adapter lifecycle seams. +function startupResult(strings: TemplateStringsArray, values: unknown[]) { + const query = strings.join(''); + return [{ + id: invocationId, + healthy: true, + status: query.includes('reconcile_worker_invocation') && values[2] ? 'ready' : 'pending', + }]; +} + // ============================================================ // Environment Validation Tests // ============================================================ @@ -360,7 +373,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -382,6 +395,7 @@ Deno.test({ edgeFunctionName: 'functions/v1/my-worker', workerId: 'test-exec-id', startMode: 'http', + invocationId, }); }, }); @@ -399,7 +413,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -414,6 +428,7 @@ Deno.test({ }) : Promise.resolve(), stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -433,7 +448,7 @@ Deno.test({ responsePromise.then(() => { responseSettled = true; }); - await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 0)); assertEquals(responseSettled, false, 'Response must wait for worker readiness'); @@ -476,6 +491,7 @@ Deno.test({ requestShutdown?.(); return Promise.resolve(); }, + get isRunning() { return false; }, get isDeprecated() { return true; }, @@ -488,6 +504,7 @@ Deno.test({ workerIds.push(workerBootstrap.workerId); }, stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -502,7 +519,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -554,6 +571,7 @@ Deno.test({ releaseStop = release; }); }, + get isRunning() { return false; }, get isDeprecated() { return true; }, @@ -567,6 +585,7 @@ Deno.test({ workerIds.push(workerBootstrap.workerId); }, stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -580,7 +599,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -712,7 +731,7 @@ Deno.test({ }); const events: string[] = []; - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => { @@ -739,6 +758,7 @@ Deno.test({ events.push('worker.stop'); return Promise.resolve(); }, + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -795,7 +815,7 @@ Deno.test({ }); const events: string[] = []; - const sql = (() => Promise.resolve([])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => { @@ -821,6 +841,7 @@ Deno.test({ workerStopCalls++; return Promise.resolve(); }, + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -985,6 +1006,7 @@ Deno.test({ releaseOldStop = release; }); }, + get isRunning() { return false; }, get isDeprecated() { return true; }, @@ -1009,6 +1031,7 @@ Deno.test({ events.push('replacement.stop'); return Promise.resolve(); }, + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -1022,7 +1045,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => { @@ -1196,6 +1219,7 @@ Deno.test({ events.push('old.stop'); return Promise.resolve(); }, + get isRunning() { return false; }, get isDeprecated() { return true; }, @@ -1209,6 +1233,7 @@ Deno.test({ events.push('new.startOnlyOnce'); }, stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -1229,7 +1254,7 @@ Deno.test({ events.push('markWorkerStopped'); workerIds.push(values[0] as string); } - return Promise.resolve([{ has_active: true }]); + return Promise.resolve(startupResult(strings, values)); }) as unknown as { end: () => Promise }; sql.end = () => Promise.resolve(); @@ -1285,7 +1310,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -1301,6 +1326,7 @@ Deno.test({ resolveStartup = resolve; }), stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -1350,7 +1376,7 @@ Deno.test({ serveHandler = h; }, }); - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => Promise.resolve(); @@ -1366,6 +1392,7 @@ Deno.test({ rejectStartup = reject; }), stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, @@ -1415,7 +1442,7 @@ Deno.test({ let releaseSqlEnd = () => {}; let sqlEndCalls = 0; - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: () => Promise; }; sql.end = () => { @@ -1569,7 +1596,7 @@ Deno.test({ }); const endCalls: unknown[] = []; - const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + const sql = ((strings: TemplateStringsArray, ...values: unknown[]) => Promise.resolve(startupResult(strings, values))) as unknown as { end: (options?: unknown) => Promise; }; sql.end = (options) => { @@ -1589,6 +1616,7 @@ Deno.test({ const worker = { startOnlyOnce: () => new Promise(() => undefined), stop: () => Promise.resolve(), + get isRunning() { return true; }, get isDeprecated() { return false; }, diff --git a/pkgs/website/src/content/docs/build/local-development.mdx b/pkgs/website/src/content/docs/build/local-development.mdx index fa1355904..867995ba8 100644 --- a/pkgs/website/src/content/docs/build/local-development.mdx +++ b/pkgs/website/src/content/docs/build/local-development.mdx @@ -9,7 +9,7 @@ import { Aside, CardGrid, LinkCard } from "@astrojs/starlight/components"; ## Filtering Console Noise -The Supabase Edge Runtime can produce verbose output during local development. This is especially noticeable because pgflow sends HTTP requests every 1 second to respawn workers quickly. To get cleaner logs that focus on your worker output, pipe through grep: +The Supabase Edge Runtime can produce verbose output during local development. The cron checks worker health every second, but sends HTTP only when no healthy worker or current pending startup exists. To get cleaner logs that focus on your worker output, pipe through grep: ```bash frame="none" npx supabase functions serve --no-verify-jwt 2>&1 | \ @@ -33,9 +33,9 @@ Then just run: `supabase-serve` ## Auto-Restart Behavior During local development, pgflow automatically: -- Restarts workers when you change flow code +- Starts workers after a code reload once the previous worker stops or its heartbeat becomes stale - Recompiles flows when the shape changes -- Keeps workers running via cron (every 1 second) +- Checks worker health via cron every second, with one pending startup per function See [Worker Lifecycle](/concepts/worker-lifecycle/) for details on how this works. @@ -47,7 +47,7 @@ The first time you start a worker, you need to trigger it once: curl http://localhost:54321/functions/v1/my-worker ``` -After this initial registration, pgflow's cron keeps the worker running automatically. +After this initial registration, pgflow's cron keeps the worker running automatically. Local startup uses the same [pending invocation handshake](/deploy/worker-management/#startup-handshake) as production. A lost request can delay retry for up to 60 seconds, plus the cron interval and any remaining debounce. An HTTP 409 while another startup is pending does not authorize another worker. ## Related diff --git a/pkgs/website/src/content/docs/concepts/startup-compilation.mdx b/pkgs/website/src/content/docs/concepts/startup-compilation.mdx index a0ecdf30d..abb19ad07 100644 --- a/pkgs/website/src/content/docs/concepts/startup-compilation.mdx +++ b/pkgs/website/src/content/docs/concepts/startup-compilation.mdx @@ -7,21 +7,22 @@ sidebar: import { Aside, CardGrid, LinkCard } from "@astrojs/starlight/components"; -When a flow worker starts, the worker compiles or verifies its imported flow definition before any registration or polling. Worker startup is the only supported flow deployment path. +When a flow worker starts, the worker compiles or verifies its imported flow definition before ready-worker registration or polling. Worker startup is the only supported flow deployment path. ## How It Works -Every startup follows this ordered contract: +HTTP startup reserves and validates an invocation before initialization. Process workers skip that HTTP reservation. Both paths compile or verify before registering a ready worker: ```text -extract complete shape +HTTP only: reserve and validate invocation ID + -> extract complete shape -> compile or verify under the flow lock - -> track the worker function - -> insert the worker row + -> HTTP: atomically accept current ID and insert the ready worker row + process: track the function and insert the worker row -> start polling ``` -Compilation happens under the flow's advisory lock, so concurrent workers for the same flow serialize safely. If compilation fails, the worker performs no registration write: the `worker_functions` row and the `pgflow.workers` row stay unchanged, and startup fails. +Compilation happens under the flow's advisory lock, so concurrent workers for the same flow serialize safely. If compilation fails, no ready worker row appears and no polling starts. HTTP startup retains its function registration and records the pending invocation as failed with a reason. Process startup still leaves registration unchanged on compilation failure. See [Startup handshake](/deploy/worker-management/#startup-handshake) for reservation, expiry, and retry behavior. ```d2 ...@../../../assets/pgflow-theme.d2 @@ -55,7 +56,7 @@ recompile.class: step_completed fail: "Fail\nto start" fail.class: step_failed -register: "Register\nworker" +register: "Accept readiness\n& register worker" register.class: step_started polling: "Start\npolling" @@ -71,7 +72,8 @@ local -> fail: "production" compile -> register verified -> register recompile -> register -register -> polling +register -> polling: "accepted" +register -> fail: "HTTP ID expired\nor superseded" ``` When a worker starts, it: @@ -79,7 +81,7 @@ When a worker starts, it: 1. **Extracts the complete flow shape** from the imported TypeScript definition 2. **Compiles the flow if missing** - PostgreSQL creates the flow, steps, and queue from the shape 3. **Verifies the shape if the flow exists** - PostgreSQL compares the worker's shape with the database definition -4. **Registers the worker** - tracks the edge function, inserts the worker row, then starts polling +4. **Registers readiness** - HTTP startup atomically accepts its still-current, unexpired invocation and inserts the worker row; process startup tracks the function and inserts the row. Only then does polling begin. ## Local Development @@ -105,7 +107,7 @@ The workflow is fully automatic: In production, shape mismatches leave flow data unchanged and cause the worker to **refuse to start**: - No flow definition, run, or task data is deleted -- No worker registration is created or changed +- No ready worker row is created; HTTP invocation state records the failure - Worker fails with `FlowShapeMismatchError` - Resolution: create a new concrete flow version (see [Version Flows](/build/version-flows/)) diff --git a/pkgs/website/src/content/docs/concepts/worker-lifecycle.mdx b/pkgs/website/src/content/docs/concepts/worker-lifecycle.mdx index 7300b3a1e..effb8f2a7 100644 --- a/pkgs/website/src/content/docs/concepts/worker-lifecycle.mdx +++ b/pkgs/website/src/content/docs/concepts/worker-lifecycle.mdx @@ -22,7 +22,7 @@ startup.class: step_started polling: "Worker polls\nfor tasks" polling.class: step_completed -heartbeat: "Sends heartbeat\nto DB (every 6s)" +heartbeat: "Sends heartbeat\nto DB (every 5s)" heartbeat.class: step_completed timeout: "Edge Function\ntime limit hit" @@ -34,23 +34,23 @@ shutdown.class: warning cron_check: "Cron detects\nmissing heartbeat" cron_check.class: info -restart: "Cron sends\nHTTP request" +restart: "Cron reserves ID\nand sends HTTP" restart.class: info startup -> polling: "begins\npolling" -polling -> heartbeat: "pings DB\nevery 6s" +polling -> heartbeat: "pings DB\nevery 5s" heartbeat -> polling: "continues" {style.stroke-dash: 3} polling -> timeout: "after\n150-400s" timeout -> shutdown: "graceful\nshutdown" shutdown -> cron_check: "worker\ngone" cron_check -> restart: "needs\nrestart" -restart -> startup: "starts new\nworker" +restart -> startup: "validates ID\nat readiness" ``` -1. **Startup** - Worker begins polling for tasks -2. **Heartbeats** - Every 6 seconds, worker pings the database +1. **Startup** - The handler validates its invocation ID, finishes initialization, and atomically accepts readiness before polling +2. **Heartbeats** - Every 5 seconds, worker pings the database 3. **Graceful shutdown** - `stopped_at` timestamp recorded when worker stops -4. **Auto-restart** - Cron detects missing heartbeat, starts new instance +4. **Auto-restart** - Cron reserves one pending invocation when no healthy worker or unexpired pending startup exists ## Why Workers Stop @@ -81,7 +81,7 @@ auto_compile: "Worker compiles\nnew flow & polls" auto_compile.class: step_completed code_change -> fn_restart: "file watch\ndetects change" -fn_restart -> cron_trigger: "cron runs\nevery 1s" +fn_restart -> cron_trigger: "old worker stops\nor heartbeat expires" cron_trigger -> auto_compile: "starts\nworker" auto_compile -> code_change: "ready for\nnext change" {style.stroke-dash: 3} ``` @@ -94,18 +94,20 @@ When you modify your flow code locally, everything restarts automatically (see [ 4. **Worker starts** - Worker compares flow shape with database 5. **Auto-recompile if needed** - If shape changed, deletes old flow + data, compiles new one -You don't need to manually curl or restart anything after the initial setup. +After initial registration, cron starts a replacement once the previous worker stops or its heartbeat becomes stale. A pending request suppresses additional startup for up to 60 seconds. See [Startup handshake](/deploy/worker-management/#startup-handshake) for retries, manual requests, and rollout limits. ## Local vs Production Behavior | Aspect | Local | Production | | ------------------- | ---------------------------- | ------------------------------------- | | Cron interval | 1 second | 1 second | -| HTTP request | Always for enabled functions | Only if not enough active workers | -| Debounce | Bypassed | Applied (prevents too frequent pings) | -| Active worker check | Bypassed | Required (enabled, non-deprecated) | +| HTTP request | No healthy worker or unexpired pending invocation | Same | +| Debounce | Retry throttle | Retry throttle | +| Active worker check | Fresh, non-stopped, non-deprecated worker | Same | | Detection | Automatic (is_local) | Automatic | +The pending-start handshake does not prevent every overlap after heartbeat loss. A previously ready worker can resume after replacement; strict ownership and fencing are separate from startup acknowledgement. External side effects still need retry-safe idempotency. + ## Related diff --git a/pkgs/website/src/content/docs/deploy/supabase/update-deployed-flows.mdx b/pkgs/website/src/content/docs/deploy/supabase/update-deployed-flows.mdx index 1aacd4908..eeec4ee19 100644 --- a/pkgs/website/src/content/docs/deploy/supabase/update-deployed-flows.mdx +++ b/pkgs/website/src/content/docs/deploy/supabase/update-deployed-flows.mdx @@ -10,10 +10,14 @@ banner: import { Aside, Steps } from "@astrojs/starlight/components"; -When updating flows that are already deployed to production, follow this sequence to prevent old and new worker versions from processing tasks simultaneously. The order matters: the `ensure_workers()` cron restarts any function whose `worker_functions` row is `enabled`, so you must disable the row before you deprecate workers. +When updating flows that are already deployed to production, disable restarts and drain old workers before deploying the new version. The order matters: the `ensure_workers()` cron restarts any function whose `worker_functions` row is `enabled`, so you must disable the row before you deprecate workers. See [Worker Management](/deploy/worker-management/) for how worker registration, deprecation, and heartbeats work. + + @@ -42,7 +46,7 @@ Throughout, replace `your-worker-name` with your Edge Function name. WHERE function_name = 'your-worker-name'; ``` - While `enabled` is `false`, the cron stops pinging this function. Disable the row before deprecating workers, otherwise the cron restarts the old worker within a second of the deprecation. + While `enabled` is `false`, the cron stops pinging this function. Disable the row before deprecating workers, otherwise the cron can request another worker with the old code once no healthy worker or unexpired pending invocation remains. 3. ### Deprecate live workers @@ -96,7 +100,7 @@ Throughout, replace `your-worker-name` with your Edge Function name. npx supabase functions deploy your-worker-name ``` - The function row stays disabled during deployment, so nothing restarts the worker while the new code deploys. + The function row stays disabled during deployment, so automatic restarts remain paused while the new code deploys. 6. ### Restore the recorded enabled state @@ -118,7 +122,7 @@ Throughout, replace `your-worker-name` with your Edge Function name. AND last_heartbeat_at > now() - interval '6 seconds'; ``` - Expect a row with a current heartbeat and a `started_at` after your deploy. The cron pings the re-enabled function within seconds and the new worker starts. + Expect a row with a current heartbeat and a `started_at` after your deploy. The cron can reserve startup when no healthy worker or unexpired pending invocation remains. A lost pending request can delay retry for up to 60 seconds, plus the cron interval and any remaining debounce. @@ -171,7 +175,7 @@ Deploy the per-step-queue definition as a new flow version: keep the old slug ru ### Replace the worker set without changing routing -Use an in-place update when the function names stay the same. Fence the complete function set before the first deploy so one flow cannot run a mixture of handler versions. +Use an in-place update when the function names stay the same. Disable the complete function set before the first deploy and drain the old workers before replacing code. @@ -196,7 +200,7 @@ Use an in-place update when the function names stay the same. Fence the complete WHERE function_name IN ('flow-step-a', 'flow-step-b'); ``` - This stops `ensure_workers()` from restarting the old code. Pause every system that invokes these functions directly before continuing; disabling the rows does not block direct HTTP requests. + This stops `ensure_workers()` from restarting the old code. Pause every system that invokes these functions directly before continuing. New HTTP workers reject startup while disabled; older workers can ignore the setting on direct requests. 3. ### Deprecate and drain every old worker @@ -248,9 +252,9 @@ Use new function names only when the old and new handlers are safe to run side b Two mechanisms overlap during an update: - **Worker lifetime.** An Edge Function keeps running for up to 150s (free tier) or 400s (paid) after deployment. Without deprecation, the old worker keeps processing tasks while the new version starts, so both versions run simultaneously. -- **Automatic restart.** The `ensure_workers()` cron runs every second and pings every function whose `worker_functions` row is `enabled` and has no alive worker. Deprecated workers do not count as alive. If you deprecate workers while the row stays enabled, the cron pings the function within a second and starts another worker with the old, still-deployed code. The deprecation undoes itself. +- **Automatic restart.** The `ensure_workers()` cron checks every second. An enabled HTTP function can receive a new startup request when no healthy worker or unexpired pending invocation exists and debounce permits it. Deprecated workers do not count as healthy. Deprecation alone therefore allows another startup with the old, still-deployed code. -This is why the sequence disables the function row first: deprecation then drains the workers, deployment swaps the code, and re-enabling starts exactly one worker with the new code. +This is why the sequence disables the function row first: deprecation then drains the workers, deployment swaps the code, and re-enabling permits startup under the pending-invocation protocol. It does not create a lifetime ownership lease. ## When the Drain or Restart Does Not Complete @@ -265,7 +269,7 @@ WHERE function_name = 'your-worker-name' ORDER BY started_at DESC; ``` -A row with `deprecated_at IS NULL` and a `started_at` after your deprecation means something started the function again. Two sources do this: the cron (the function row was re-enabled) and any direct HTTP request to the function (this starts a worker even while the row is disabled, because the function is still deployed). Confirm `enabled = false`, find the source of the restart, then repeat step 3 to deprecate the new worker. +A row with `deprecated_at IS NULL` and a `started_at` after your deprecation means something started the function again. Two sources do this: the cron (the function row was re-enabled) and direct HTTP requests handled by older worker versions, which can start while the row is disabled. New HTTP workers respect the disabled registration, but disabling does not stop an already-ready worker. Confirm `enabled = false`, find the source of the restart, then repeat step 3 to deprecate the new worker. ### Tasks stay in started @@ -290,13 +294,15 @@ The stalled-task cron requeues such tasks after the step timeout plus 30 seconds Check whether the cron is pinging the function: ```sql -SELECT function_name, enabled, last_invoked_at +SELECT function_name, enabled, last_invoked_at, invocation_id, + invocation_status, invocation_expires_at, invocation_worker_id, invocation_error FROM pgflow.worker_functions WHERE function_name = 'your-worker-name'; ``` - `enabled` is `false`: step 6 did not restore the row. Restore it. -- `last_invoked_at` does not advance: the cron is not invoking the function. Check the cron job exists and the vault secrets are set, because the cron builds the function URL and authorizes the ping from them: +- `invocation_status` is `pending` with a future `invocation_expires_at`: wait for readiness or expiry; cron deliberately suppresses another request. +- `last_invoked_at` does not advance and no healthy worker or unexpired pending invocation exists: check the retry throttle, then check the cron job exists and the vault secrets are set, because the cron builds the function URL and authorizes the ping from them: ```sql SELECT jobname, schedule, command @@ -306,7 +312,7 @@ WHERE function_name = 'your-worker-name'; See [Configure Secrets](/deploy/supabase/configure-secrets/). -- `last_invoked_at` advances but no worker row appears: the ping reaches Supabase but the function fails to start. Check the Edge Function logs in the Supabase dashboard. See [Monitor workers health](/deploy/monitor-workers-health/). +- `last_invoked_at` advances but no worker row appears: SQL reserved and enqueued a request, not proof of delivery or readiness. Check `invocation_status`, `invocation_error`, expiry, pg_net responses, and the Edge Function logs. See [Monitor workers health](/deploy/monitor-workers-health/). ## Monitoring Workers diff --git a/pkgs/website/src/content/docs/deploy/update-pgflow.mdx b/pkgs/website/src/content/docs/deploy/update-pgflow.mdx index 687601cc3..9ef6946cf 100644 --- a/pkgs/website/src/content/docs/deploy/update-pgflow.mdx +++ b/pkgs/website/src/content/docs/deploy/update-pgflow.mdx @@ -16,6 +16,10 @@ This guide explains how to update pgflow to the latest version, including packag All pgflow packages (npm and jsr) are versioned together. When you update one pgflow package, all others should be updated to the same version to maintain compatibility. + + ## What needs to be updated? When updating pgflow, you need to update: diff --git a/pkgs/website/src/content/docs/deploy/worker-management.mdx b/pkgs/website/src/content/docs/deploy/worker-management.mdx index 3320882c7..3ce66207d 100644 --- a/pkgs/website/src/content/docs/deploy/worker-management.mdx +++ b/pkgs/website/src/content/docs/deploy/worker-management.mdx @@ -27,7 +27,49 @@ curl http://localhost:54321/functions/v1/my-worker SELECT pgflow.track_worker_function('my-worker'); ``` -Once registered, pgflow tracks the worker in the `worker_functions` table and keeps it alive automatically. +Once registered, pgflow tracks the function in `worker_functions` and starts it when no healthy worker or current pending invocation exists. + +### Startup handshake + +Before sending an HTTP request, `pgflow.ensure_workers()` reserves a unique `invocation_id` on the function's registration. Reservation and HTTP enqueue share one transaction. Concurrent cron calls cannot reserve another current pending invocation for that function. + +The HTTP handler authenticates the request before checking the ID. The ID correlates a request; it is not a credential. Receipt alone leaves the invocation pending. After initialization, SQL atomically accepts the still-current, unexpired ID and registers the ready worker before processing starts. Duplicate deliveries to independent runtimes can accept only one new worker. A healthy existing runtime acknowledges the specific request without another processing loop. + +Pending startup expires after **60 seconds**, measured by the database clock. This allows slow initialization beyond the default six-second debounce without another request. Lost delivery or hung startup therefore delays retry for up to the pending timeout, plus the cron interval and any remaining debounce. Initialization that exceeds the deadline cannot start processing; it needs a new invocation. An explicit initialization failure records a reason and permits retry after debounce. + +Old IDs cannot dismiss a newer pending invocation. A lost readiness response can recover the same committed worker registration. + +A separate 60-second runtime watchdog aborts a hung start and releases the local startup guard, so a later valid request can use the surviving runtime. SQL still decides whether a token can authorize readiness; the watchdog cannot extend or replace its database-clock expiry. Initialization SQL uses transaction-local 60-second statement and idle-transaction timeouts, so a blocked start cannot occupy the only database connection forever. These timeouts do not change task-handler SQL or process-worker startup. + +`debounce` remains a retry throttle and the heartbeat freshness threshold; it is not the pending-start correctness mechanism. Function names remain independent: `my-worker` and `my-worker-b` can intentionally serve the same queue. Node/Bun registrations with `start_mode = 'process'` never enter HTTP scheduling. + + + +### Direct requests and local development + +Authenticated production requests may omit `invocation_id`, including an empty-body `curl` request. The handler registers an unknown function without overwriting an existing registration, then reserves through the same SQL protocol. It cannot enable a disabled function, convert a process registration, or bypass a pending startup. A healthy worker in the same runtime returns `running` without another loop. Otherwise, unavailable startup returns HTTP **409** (`startup_unavailable`); retry only after the current attempt or retry throttle permits it. + +Requests with an ID must supply a JSON object containing a UUID `invocation_id`. Expired, dismissed, superseded, or differently accepted IDs return HTTP **409** (`invocation_rejected`), even when a runtime survives. Malformed bodies or IDs return **400**. Production authentication remains `Authorization: Bearer ` with the existing service-role-key fallback. Missing or invalid credentials return **401**; an invocation ID never bypasses this check. + +Local requests keep the existing local authentication exemption, but use the same reservation, expiry, and readiness checks. Local cron no longer pings every enabled function unconditionally. After a code reload, it waits for the old worker to stop or its heartbeat to become stale. A lost request can add up to 60 seconds before retry. The initial local `curl` above still works; a 409 during startup does not mean another worker should start. + +Internal low-level worker/lifecycle bootstrap without an invocation ID remains a legacy, unprotected path. Use `EdgeWorker.start()` on Supabase for the HTTP handshake; process workers self-start separately. + +### Upgrade existing SQL and workers + +Apply the SQL migration **before deploying the new worker package**. New HTTP workers fail closed against old SQL; there is no tokenless fallback. Old workers can run against the added SQL fields, but ignore invocation IDs and never acknowledge them. Mixed old/new deployments do not provide the pending-start guarantee for old runtimes. + +For a controlled rollout: + +1. Record each HTTP function's current `enabled` value. Preserve intentionally disabled functions. +2. Disable restarts and drain existing workers using [Update Deployed Flows](/deploy/supabase/update-deployed-flows/). Pause external/manual pings too: older workers can ignore `enabled` on direct requests. Disabling does not evict a hosted runtime. +3. Install and apply the engine migration through your normal database migration process. Keep restarts disabled. +4. Deploy matching new worker packages to every HTTP function. Do not rely on handshake protection while old runtimes still serve requests. +5. Restore each function's exact recorded `enabled` value. Check pending state, the accepted worker ID, package version, and fresh heartbeats before treating the rollout as complete. + +The handshake adds no stronger drain or eviction guarantee. A completed handshake does not mean the worker stopped. ## Per-step queue workers @@ -106,7 +148,7 @@ trigger: "1. Trigger" { deprecate: "You deprecate\nworkers by function" deprecate.class: neutral - heartbeat: "Worker sends\nheartbeat (every 6s)" + heartbeat: "Worker sends\nheartbeat (every 5s)" heartbeat.class: info response: "DB returns\nis_deprecated: true" @@ -128,7 +170,7 @@ shutdown: "2. Graceful Shutdown" { exit: "Worker\nexits" exit.class: step_completed - skip: "Cron skips\n(not restarted)" + skip: "Cron restarts\nif enabled" skip.class: step_failed stop -> finish @@ -144,7 +186,7 @@ trigger.response -> shutdown.stop: "isRunning = false" 3. **Worker stops polling** - Worker stops accepting new tasks 4. **Finish in-flight work** - Worker completes any currently running task 5. **Graceful shutdown** - Worker exits cleanly -6. **No restart** - Cron skips deprecated workers when checking for restarts +6. **Restart if enabled** - Cron can invoke the function again unless its registration is disabled ## Tracking Workers @@ -154,7 +196,36 @@ Query the `worker_functions` table to see registered workers: SELECT * FROM pgflow.worker_functions; ``` -Query the `workers` table to see active worker instances: +Check startup state separately from worker lifetime: + +```sql +SELECT + wf.function_name, + wf.invocation_id, + wf.invocation_status, + wf.invocation_expires_at, + wf.invocation_worker_id, + wf.invocation_error, + CASE + WHEN NOT wf.enabled THEN 'disabled' + WHEN wf.start_mode = 'process' THEN 'process' + WHEN 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 + ) THEN 'ready' + WHEN wf.invocation_status = 'pending' + AND wf.invocation_expires_at > clock_timestamp() THEN 'pending' + ELSE 'retryable' + END AS startup_state +FROM pgflow.worker_functions AS wf +ORDER BY wf.function_name; +``` + +`invocation_status = 'ready'` records a completed handshake, not current liveness. An expired `pending` invocation is retryable. Retry still obeys debounce. The query reports readiness only when a healthy worker exists. + +Query the `workers` table to see non-stopped worker instances; also check heartbeat freshness and deprecation before treating one as healthy: ```sql SELECT @@ -169,6 +240,16 @@ WHERE stopped_at IS NULL ORDER BY started_at DESC; ``` +## Troubleshooting worker startup + +Run the `worker_functions` query above first. `startup_state` and `invocation_error` explain most startup situations. + +- `startup_state = 'pending'` with a future `invocation_expires_at` means a start is in flight. Do not send another request; retry happens automatically after the 60-second expiry plus the cron interval and any remaining debounce. +- `startup_state = 'retryable'` with `invocation_error` set means initialization failed and pgflow recorded the reason. Fix the underlying cause in the function logs; the invocation clears on the next successful reservation and needs no manual cleanup. +- HTTP **409** `invocation_rejected` on a manual request means its `invocation_id` expired, was superseded, or was already accepted. Repeat the request without the ID; the handler reserves through the same protocol. +- HTTP **409** `startup_unavailable` means the runtime could not start a worker. Retry after the current attempt finishes or debounce passes. +- Two live workers for one function: the handshake prevents duplicate pending starts, not runtime overlap. A worker accepted earlier can stay live while a replacement starts. Deprecate the stale worker instead of disabling the function. + ## Prerequisites For automatic worker management to work in production, you need vault secrets configured.