fix: make startOnlyOnce awaitable, deduplicate stop calls, and add packed-tarball smoke test

pkgs/core/schemas/0050_tables_definitions.sql

@@ -15,7 +15,7 @@ create table pgflow.flows (
 
 -- Steps table - stores individual steps within flows
 create table pgflow.steps (
-  flow_slug text not null references pgflow.flows (flow_slug),
+  flow_slug text not null references pgflow.flows(flow_slug),
   step_slug text not null,
   step_type text not null default 'single',
   step_index int not null default 0,
@@ -49,15 +49,15 @@ create table pgflow.steps (
 
 -- Dependencies table - stores relationships between steps
 create table pgflow.deps (
-  flow_slug text not null references pgflow.flows (flow_slug),
+  flow_slug text not null references pgflow.flows(flow_slug),
   dep_slug text not null, -- slug of the dependency
   step_slug text not null, -- slug of the dependent
   created_at timestamptz not null default now(),
   primary key (flow_slug, dep_slug, step_slug),
   foreign key (flow_slug, dep_slug)
-  references pgflow.steps (flow_slug, step_slug),
+  references pgflow.steps(flow_slug, step_slug),
   foreign key (flow_slug, step_slug)
-  references pgflow.steps (flow_slug, step_slug),
+  references pgflow.steps(flow_slug, step_slug),
   check (dep_slug != step_slug)  -- Prevent self-dependencies
 );
 

pkgs/core/schemas/0060_tables_runtime.sql

@@ -3,7 +3,7 @@
 -- Runs table - tracks flow execution instances
 create table pgflow.runs (
   run_id uuid primary key not null default gen_random_uuid(),
-  flow_slug text not null references pgflow.flows (flow_slug), -- denormalized
+  flow_slug text not null references pgflow.flows(flow_slug), -- denormalized
   status text not null default 'started',
   input jsonb not null,
   output jsonb,
@@ -22,8 +22,8 @@ create index if not exists idx_runs_status on pgflow.runs (status);
 
 -- Step states table - tracks the state of individual steps within a run
 create table pgflow.step_states (
-  flow_slug text not null references pgflow.flows (flow_slug),
-  run_id uuid not null references pgflow.runs (run_id),
+  flow_slug text not null references pgflow.flows(flow_slug),
+  run_id uuid not null references pgflow.runs(run_id),
   step_slug text not null,
   status text not null default 'created',
   remaining_tasks int null,  -- NULL = not started, >0 = active countdown
@@ -39,7 +39,7 @@ create table pgflow.step_states (
   skipped_at timestamptz,
   primary key (run_id, step_slug),
   foreign key (flow_slug, step_slug)
-  references pgflow.steps (flow_slug, step_slug),
+  references pgflow.steps(flow_slug, step_slug),
   constraint status_is_valid check (status in ('created', 'started', 'completed', 'failed', 'skipped')),
   constraint status_and_remaining_tasks_match check (status != 'completed' or remaining_tasks = 0),
   -- Add constraint to ensure remaining_tasks is only set when step has started
@@ -82,8 +82,8 @@ create index if not exists idx_step_states_run_id on pgflow.step_states (run_id)
 
 -- Step tasks table - tracks units of work for step
 create table pgflow.step_tasks (
-  flow_slug text not null references pgflow.flows (flow_slug),
-  run_id uuid not null references pgflow.runs (run_id),
+  flow_slug text not null references pgflow.flows(flow_slug),
+  run_id uuid not null references pgflow.runs(run_id),
   step_slug text not null,
   message_id bigint,
   task_index int not null default 0,
@@ -95,14 +95,14 @@ create table pgflow.step_tasks (
   started_at timestamptz,
   completed_at timestamptz,
   failed_at timestamptz,
-  last_worker_id uuid references pgflow.workers (worker_id) on delete set null,
+  last_worker_id uuid references pgflow.workers(worker_id) on delete set null,
   -- Requeue tracking columns
   requeued_count int not null default 0,
   last_requeued_at timestamptz,
   permanently_stalled_at timestamptz,
   constraint step_tasks_pkey primary key (run_id, step_slug, task_index),
   foreign key (run_id, step_slug)
-  references pgflow.step_states (run_id, step_slug),
+  references pgflow.step_states(run_id, step_slug),
   constraint valid_status check (
     status in ('queued', 'started', 'completed', 'failed')
   ),

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

@@ -1,29 +1,73 @@
+-- Test: ensure_workers() never pings process-started workers
 begin;
-select plan(3);
-
+select plan(2);
 select pgflow_tests.reset_db();
 
+-- Clear any existing HTTP requests
+delete from net._http_response;
+
+-- 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 two worker functions — one HTTP-started, one process-started
 select pgflow.track_worker_function('http-worker', 'http');
 select pgflow.track_worker_function('process-worker', 'process');
 
-select is(
-  (select count(*)::int from pgflow.ensure_workers() where function_name = 'process-worker'),
-  0,
-  'ensure_workers skips process workers in production mode'
-);
+-- ============================================================
+-- TEST 1: Local mode pings HTTP worker, skips process worker
+-- ============================================================
+
+-- Activate local mode via known local JWT secret
+set local app.settings.jwt_secret = 'super-secret-jwt-token-with-at-least-32-characters-long';
+
+-- Reset debounce timestamps so functions are eligible for invocation
+update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds';
 
-select ok(
-  (select count(*)::int from pgflow.ensure_workers() where function_name = 'http-worker') >= 0,
-  'ensure_workers still evaluates http workers'
+-- Force evaluation into a temp table so pg_net requests are queued before we inspect them
+select * into temporary local_result from pgflow.ensure_workers();
+
+-- Assert: only the HTTP worker URL was queued (process worker must NOT appear)
+select results_eq(
+  $$
+    select url from net.http_request_queue
+    where url like 'http://kong:8000/%'
+    order by url
+  $$,
+  $$
+    select 'http://kong:8000/functions/v1/http-worker'::text as url
+  $$,
+  'Local mode: only HTTP worker is pinged, process worker is skipped'
 );
 
-select set_config('app.settings.is_local', 'true', true);
+-- ============================================================
+-- TEST 2: Production mode pings HTTP worker, skips process worker
+-- ============================================================
+
+-- Clear the request queue from the local test
+delete from net._http_response;
+
+-- Reset debounce timestamps
+update pgflow.worker_functions set last_invoked_at = now() - interval '10 seconds';
+
+-- Switch to production mode (jwt_secret does NOT match the local value)
+set local app.settings.jwt_secret = 'production-secret-different-from-local';
+
+-- Force evaluation into a temp table so pg_net requests are queued before we inspect them
+select * into temporary prod_result from pgflow.ensure_workers();
 
-select is(
-  (select count(*)::int from pgflow.ensure_workers() where function_name = 'process-worker'),
-  0,
-  'ensure_workers skips process workers in local mode'
+-- Assert: only the HTTP worker URL was queued (process worker must NOT appear)
+select results_eq(
+  $$
+    select url from net.http_request_queue
+    where url like 'https://%'
+    order by url
+  $$,
+  $$
+    select 'https://testproject123.supabase.co/functions/v1/http-worker'::text as url
+  $$,
+  'Production mode: only HTTP worker is pinged, process worker is skipped'
 );
 
-select * from finish();
+select finish();
 rollback;

pkgs/edge-worker/README.md

@@ -25,7 +25,7 @@ import { EdgeWorker } from 'jsr:@pgflow/edge-worker';
 
 `@pgflow/edge-worker` is published to both registries:
 
-- Use JSR for Supabase Edge Functions and Deno deployments.
+- Use JSR for Supabase Edge Functions.
 - Use npm for Node and Bun long-running process deployments.
 
 Supabase Edge Functions remain the primary deployment path. Node and Bun process hosting is intended for Railway, Docker, and similar hosts that run workers as normal long-running processes.

pkgs/edge-worker/project.json

@@ -147,11 +147,26 @@
     },
     "test:node": {
       "executor": "nx:run-commands",
+      "dependsOn": ["build"],
       "options": {
-        "command": "pnpm vitest run tests/unit/platform/ProcessPlatformAdapter.test.ts",
+        "command": "node ../../scripts/smoke-edge-worker-dist.mjs",
         "cwd": "pkgs/edge-worker"
       }
     },
+    "verify-exports": {
+      "executor": "nx:run-commands",
+      "cache": true,
+      "dependsOn": ["build"],
+      "inputs": [
+        "production",
+        "^production",
+        "{workspaceRoot}/scripts/smoke-edge-worker-dist.mjs"
+      ],
+      "options": {
+        "command": "node ./scripts/smoke-edge-worker-dist.mjs",
+        "cwd": "{workspaceRoot}"
+      }
+    },
     "smoke:bun": {
       "executor": "nx:run-commands",
       "dependsOn": ["build"],

pkgs/edge-worker/src/_internal/flow.ts

@@ -2,4 +2,4 @@ export { createFlowWorker } from '../flow/createFlowWorker.js';
 export { FlowWorkerLifecycle } from '../flow/FlowWorkerLifecycle.js';
 export { StepTaskExecutor } from '../flow/StepTaskExecutor.js';
 export { StepTaskPoller } from '../flow/StepTaskPoller.js';
-export type * from '../flow/types.js';
+export type * from '@pgflow/core';

pkgs/edge-worker/src/core/Worker.ts

@@ -15,7 +15,10 @@ export class Worker {
 
   private batchProcessor: IBatchProcessor;
   private mainLoopPromise: Promise<void> | undefined;
+  private startupPromise: Promise<void> | undefined;
+  private stopPromise: Promise<void> | undefined;
   private deprecationLogged = false;
+  private deprecationHandler?: () => void;
 
   constructor(
     batchProcessor: IBatchProcessor,
@@ -30,37 +33,50 @@ export class Worker {
     this.cleanup = options.cleanup;
   }
 
-  startOnlyOnce(workerBootstrap: WorkerBootstrap) {
+  startOnlyOnce(workerBootstrap: WorkerBootstrap): Promise<void> {
+    if (this.startupPromise) {
+      return this.startupPromise;
+    }
+
     if (!this.lifecycle.isCreated) {
       this.logger.debug('Worker not in Created state, ignoring start request');
-      return;
+      return Promise.resolve();
     }
-    this.mainLoopPromise = this.start(workerBootstrap);
+
+    this.startupPromise = this.lifecycle
+      .acknowledgeStart(workerBootstrap)
+      .then(() => {
+        this.mainLoopPromise = this.runMainLoop();
+      })
+      .catch((error) => {
+        this.logger.error(`Error in worker startup: ${error}`);
+        throw error;
+      });
+
+    return this.startupPromise;
   }
 
-  private async start(workerBootstrap: WorkerBootstrap) {
+  private async runMainLoop() {
     try {
-      await this.lifecycle.acknowledgeStart(workerBootstrap);
-
       while (this.isMainLoopActive) {
         try {
           await this.lifecycle.sendHeartbeat();
         } catch (error: unknown) {
           this.logger.error(`Error sending heartbeat: ${error}`);
-          // Continue execution - a failed heartbeat shouldn't stop processing
         }
 
-        // Check if deprecated after heartbeat
         if (!this.isMainLoopActive) {
           this.logDeprecation();
+          if (this.lifecycle.isDeprecated) {
+            this.deprecationHandler?.();
+          }
           break;
         }
 
         try {
           await this.batchProcessor.processBatch();
         } catch (error: unknown) {
           this.logger.error(`Error processing batch: ${error}`);
-          // Continue to next iteration - failed batch shouldn't stop the worker
         }
       }
     } catch (error) {
@@ -69,16 +85,23 @@ export class Worker {
     }
   }
 
-  async stop() {
-    // If the worker is already stopping or stopped, do nothing
+  onDeprecated(handler: () => void): void {
+    this.deprecationHandler = handler;
+  }
+
+  stop(): Promise<void> {
+    this.stopPromise ??= Promise.resolve().then(() => this.performStop());
+    return this.stopPromise;
+  }
+
+  private async performStop() {
     if (this.lifecycle.isStopping || this.lifecycle.isStopped) {
       return;
     }
 
     this.lifecycle.transitionToStopping();
 
     try {
-      // Signal deprecation (which includes "Stopped accepting new messages")
       this.logDeprecation();
       this.requestShutdown?.();
       this.abortController.abort();
@@ -93,7 +116,6 @@ export class Worker {
         throw error;
       }
 
-      // Signal waiting for pending tasks
       this.logger.shutdown('waiting');
       await this.batchProcessor.awaitCompletion();
 
@@ -104,7 +126,6 @@ export class Worker {
         await this.cleanup();
       }
 
-      // Signal graceful stop complete
       this.logger.shutdown('stopped');
     } catch (error) {
       this.logger.debug(`Error during worker stop: ${error}`);

pkgs/edge-worker/src/core/context.ts

@@ -2,7 +2,7 @@
 import type { BaseContext, AnyFlow, AllStepInputs, ExtractFlowInput } from '@pgflow/dsl';
 import type { Json } from './types.js';
 import type { PgmqMessageRecord } from '../queue/types.js';
-import type { StepTaskRecord } from '../flow/types.js';
+import type { StepTaskRecord } from '@pgflow/core';
 import type { QueueWorkerConfig, FlowWorkerConfig } from './workerConfigTypes.js';
 
 /* ──────────────────────────────────────────────────────────────────────

pkgs/edge-worker/src/core/test-context-utils.ts

@@ -5,7 +5,7 @@ import type {
 import type {
   MessageContext, StepTaskContext
 } from './context.js';
-import type { StepTaskRecord } from '../flow/types.js';
+import type { StepTaskRecord } from '@pgflow/core';
 import type { PgmqMessageRecord } from '../queue/types.js';
 
 export function createMessageTestContext<

pkgs/edge-worker/src/flow/StepTaskExecutor.ts

@@ -1,5 +1,6 @@
 import type { AnyFlow } from '@pgflow/dsl';
-import type { IPgflowClient } from './types.js';
+import type { StepTaskRecord } from '@pgflow/core';
+import type { IPgflowClient } from '@pgflow/core';
 import type { IExecutor } from '../core/types.js';
 import type { Logger, TaskLogContext } from '../platform/types.js';
 import type { StepTaskHandlerContext } from '../core/context.js';
@@ -39,7 +40,7 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
   }
 
   // Convenience getters to avoid drilling into context
-  get stepTask() {
+  get stepTask(): StepTaskRecord<TFlow> {
     return this.context.stepTask;
   }
 

pkgs/edge-worker/src/flow/StepTaskPoller.ts

@@ -1,4 +1,4 @@
-import type { IPgflowClient } from './types.js';
+import type { IPgflowClient } from '@pgflow/core';
 import type { IPoller, Supplier } from '../core/types.js';
 import type { Logger } from '../platform/types.js';
 import type { AnyFlow, AllStepInputs } from '@pgflow/dsl';

pkgs/edge-worker/src/index.ts

@@ -13,7 +13,7 @@ export { ControlPlane } from './control-plane/index.js';
 export * from './platform/index.js';
 
 // Export types
-export type { StepTaskRecord } from './flow/types.js';
+export type { StepTaskRecord } from '@pgflow/core';
 export type { FlowWorkerConfig } from './flow/createFlowWorker.js';
 export type { StepTaskPollerConfig } from './flow/StepTaskPoller.js';