refactor: improve worker function invocation logic in local and production modes (#527)

# Improve worker function management in local and production modes

This PR refines the `ensure_workers()` function to better handle different behaviors between local and production environments:

- **Local mode**: Now pings ALL enabled functions, completely bypassing both the debounce check and alive workers check
- **Production mode**: Only pings functions that pass debounce AND have no alive workers

The key changes include:

1. Restructured the `functions_to_invoke` CTE to more clearly separate local vs production logic
2. Updated function comments to better document the behavior differences
3. Fixed tests to align with the new behavior, particularly that debounce is now bypassed in local mode
4. Renamed migration file with updated timestamp

This change improves developer experience in local environments by ensuring all enabled functions are always pinged, while maintaining efficient worker management in production.

Author: jumski

pkgs/core/schemas/0059_function_ensure_workers.sql

@@ -48,12 +48,20 @@ as $$
     ),
 
     -- 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 dp.function_name
-      from debounce_passed as dp
-      where
-        pgflow.is_local() = true
-        or dp.function_name not in (select faw.function_name from functions_with_alive_workers as faw)
+      select wf.function_name
+      from pgflow.worker_functions as wf
+      where wf.enabled = true
+        and (
+          pgflow.is_local() = true  -- Local: all enabled 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
@@ -94,9 +102,9 @@ $$;
 
 comment on function pgflow.ensure_workers() is
 'Ensures worker functions are running by pinging them via HTTP when needed.
-In local mode: always pings all enabled functions (for fast restart after code changes).
-In production mode: only pings functions that have no alive workers.
-Respects debounce: skips functions pinged within their heartbeat_timeout_seconds window.
+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 heartbeat_timeout_seconds window (production only).
 Credentials: Uses Vault secrets (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.';

…51208093149_pgflow_worker_management.sql …51209074533_pgflow_worker_management.sqlpkgs/core/supabase/migrations/20251208093149_pgflow_worker_management.sql renamed to pkgs/core/supabase/migrations/20251209074533_pgflow_worker_management.sql pkgs/core/supabase/migrations/20251208093149_pgflow_worker_management.sql renamed to pkgs/core/supabase/migrations/20251209074533_pgflow_worker_management.sql

@@ -1,6 +1,6 @@
--- Create extension "pg_net" if not exists (already present in Supabase)
+-- Create extension "pg_net" if not exists
 CREATE EXTENSION IF NOT EXISTS "pg_net" WITH SCHEMA "public";
--- Create extension "pg_cron" if not exists (already present in Supabase)
+-- Create extension "pg_cron" if not exists
 CREATE EXTENSION IF NOT EXISTS "pg_cron";
 -- Modify "workers" table
 ALTER TABLE "pgflow"."workers" ADD COLUMN "stopped_at" timestamptz NULL;
@@ -142,12 +142,20 @@ with
     ),
 
     -- 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 dp.function_name
-      from debounce_passed as dp
-      where
-        pgflow.is_local() = true
-        or dp.function_name not in (select faw.function_name from functions_with_alive_workers as faw)
+      select wf.function_name
+      from pgflow.worker_functions as wf
+      where wf.enabled = true
+        and (
+          pgflow.is_local() = true  -- Local: all enabled 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
@@ -187,9 +195,9 @@ with
 $$;
 -- Set comment to function: "ensure_workers"
 COMMENT ON FUNCTION "pgflow"."ensure_workers" IS 'Ensures worker functions are running by pinging them via HTTP when needed.
-In local mode: always pings all enabled functions (for fast restart after code changes).
-In production mode: only pings functions that have no alive workers.
-Respects debounce: skips functions pinged within their heartbeat_timeout_seconds window.
+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 heartbeat_timeout_seconds window (production only).
 Credentials: Uses Vault secrets (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.';

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:rp08kLQLd1h/gfTMMGK0y+skRiqSfksIcbQYTe/7gXI=
+h1:+LFk0wyD7sBjvwZW93t7THECms67qWV2Vm0E5GHTCxQ=
 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=
@@ -12,4 +12,4 @@ h1:rp08kLQLd1h/gfTMMGK0y+skRiqSfksIcbQYTe/7gXI=
 20251103222045_pgflow_fix_broadcast_order_and_timestamp_handling.sql h1:K/XnZpOmxfelsaNoJbR5HxhBrs/oW4aYja222h5cps4=
 20251104080523_pgflow_upgrade_pgmq_1_5_1.sql h1:Fw7zpMWnjhAHQ0qBJAprAvGl7dJMd8ExNHg8aKvkzTg=
 20251130000000_pgflow_auto_compilation.sql h1:qs+3qq1Vsyo0ETzbxDnmkVtSUa6XHkd/K9wF/3W46jM=
-20251208093149_pgflow_worker_management.sql h1:/YCw+E63GSUiTkCx4jk4XgRApqKt+9by/NYoPsdJoIQ=
+20251209074533_pgflow_worker_management.sql h1:Yi+WZEgL+AcK0eAfKJD8v/uPBVBM8WcL+ZOL4ilkYz8=

pkgs/core/supabase/tests/ensure_workers/debounce.test.sql

@@ -1,8 +1,19 @@
--- Test: ensure_workers() respects debounce window
+-- 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();
 
+-- Setup: Create Vault secrets for production mode
+select vault.create_secret(
+  'test-service-role-key',
+  'supabase_service_role_key'
+);
+select vault.create_secret(
+  'testproject123',
+  'supabase_project_id'
+);
+
 -- Setup: Register a worker function with 6 second heartbeat timeout
 select pgflow.track_worker_function('my-function');
 
@@ -12,7 +23,8 @@ update pgflow.worker_functions
 set last_invoked_at = now()
 where function_name = 'my-function';
 
-set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long';
+-- Simulate production mode (non-local jwt_secret)
+set local app.settings.jwt_secret = 'production-secret-different-from-local';
 select is(
   (select count(*) from pgflow.ensure_workers()),
   0::bigint,

pkgs/core/supabase/tests/ensure_workers/local_mode_always_pings.test.sql

@@ -37,15 +37,15 @@ select is(
   'Local mode returns function even when worker has fresh heartbeat'
 );
 
--- TEST: Debounce still applies in local mode
+-- 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()),
-  0::bigint,
-  'Debounce still applies in local mode'
+  1::bigint,
+  'Debounce is bypassed in local mode'
 );
 
 select finish();