diff --git a/pkgs/edge-worker/deno.lock b/pkgs/edge-worker/deno.lock index 1e9a70266..9cac9b9ba 100644 --- a/pkgs/edge-worker/deno.lock +++ b/pkgs/edge-worker/deno.lock @@ -8,6 +8,7 @@ "jsr:@std/assert@0.224": "0.224.0", "jsr:@std/async@0.224": "0.224.2", "jsr:@std/crypto@0.224": "0.224.0", + "jsr:@std/data-structures@0.224": "0.224.1", "jsr:@std/fmt@0.224": "0.224.0", "jsr:@std/fmt@1.0.3": "1.0.3", "jsr:@std/internal@0.224": "0.224.0", @@ -46,6 +47,9 @@ "jsr:@std/assert@0.224" ] }, + "@std/data-structures@0.224.1": { + "integrity": "266365f90014e5c52bccf1eadd65f7782a2af2c84e4052ef8316698a3572cac3" + }, "@std/fmt@0.224.0": { "integrity": "e20e9a2312a8b5393272c26191c0a68eda8d2c4b08b046bad1673148f1d69851" }, @@ -64,7 +68,9 @@ "@std/testing@0.224.0": { "integrity": "371b8a929aa7132240d5dd766a439be8f780ef5c176ab194e0bcab72370c761e", "dependencies": [ - "jsr:@std/assert@0.224" + "jsr:@std/assert@0.224", + "jsr:@std/async", + "jsr:@std/data-structures" ] } }, diff --git a/pkgs/edge-worker/deno.test.json b/pkgs/edge-worker/deno.test.json index dd777ec6a..e587fb08e 100644 --- a/pkgs/edge-worker/deno.test.json +++ b/pkgs/edge-worker/deno.test.json @@ -11,6 +11,7 @@ "@std/crypto/timing-safe-equal": "jsr:@std/crypto@^0.224.0/timing-safe-equal", "@std/log": "jsr:@std/log@^0.224.13", "@std/testing/mock": "jsr:@std/testing@^0.224.0/mock", + "@std/testing/time": "jsr:@std/testing@^0.224.0/time", "postgres": "jsr:@oscar6echo/postgres@3.4.5-d", "@pgflow/core": "../core/src/index.ts", "@pgflow/dsl": "../dsl/src/index.ts", diff --git a/pkgs/edge-worker/src/core/Worker.ts b/pkgs/edge-worker/src/core/Worker.ts index da3ea960d..7ff49b3e1 100644 --- a/pkgs/edge-worker/src/core/Worker.ts +++ b/pkgs/edge-worker/src/core/Worker.ts @@ -1,6 +1,11 @@ import type { IBatchProcessor, ILifecycle, WorkerBootstrap } from './types.js'; import type { Logger } from '../platform/types.js'; +/** Initial delay before retrying a failed main-loop iteration. */ +const RETRY_DELAY_MS = 100; +/** Maximum delay for consecutive failed main-loop iterations. */ +const MAX_RETRY_DELAY_MS = 5_000; + export interface WorkerOptions { requestShutdown?: () => void; cleanup?: () => Promise; @@ -57,12 +62,17 @@ export class Worker { } private async runMainLoop() { + let consecutiveFailures = 0; + try { while (this.isMainLoopActive) { + let iterationFailed = false; + try { await this.lifecycle.sendHeartbeat(); } catch (error: unknown) { this.logger.error(`Error sending heartbeat: ${error}`); + iterationFailed = true; } if (!this.isMainLoopActive) { @@ -77,6 +87,19 @@ export class Worker { await this.batchProcessor.processBatch(); } catch (error: unknown) { this.logger.error(`Error processing batch: ${error}`); + iterationFailed = true; + } + + if (iterationFailed) { + consecutiveFailures++; + if (this.isMainLoopActive) { + await this.waitForRetry( + Math.min(RETRY_DELAY_MS * 2 ** (consecutiveFailures - 1), MAX_RETRY_DELAY_MS) + ); + } + } else { + // Only a fully successful iteration resets the backoff. + consecutiveFailures = 0; } } } catch (error) { @@ -85,6 +108,29 @@ export class Worker { } } + /** + * Abort-aware backoff wait: resolves after `ms`, or immediately when the + * worker's abort signal fires so stop() never waits out a retry delay. + */ + private waitForRetry(ms: number): Promise { + return new Promise((resolve) => { + if (this.isAborted) { + resolve(); + return; + } + + const onAbort = () => { + clearTimeout(timer); + resolve(); + }; + const timer = setTimeout(() => { + this.abortController.signal.removeEventListener('abort', onAbort); + resolve(); + }, ms); + this.abortController.signal.addEventListener('abort', onAbort, { once: true }); + }); + } + onDeprecated(handler: () => void): void { this.deprecationHandler = handler; } @@ -99,15 +145,22 @@ export class Worker { return; } - this.lifecycle.transitionToStopping(); - try { this.logDeprecation(); this.requestShutdown?.(); this.abortController.abort(); + // Wait for startup to settle before transitioning: a stop during + // Starting must not attempt an invalid transition, and the abort + // above already keeps the main loop from processing any batch. + if (this.startupPromise) { + await this.startupPromise; + } + + this.lifecycle.transitionToStopping(); + + this.logger.debug('-> Waiting for main loop to complete'); try { - this.logger.debug('-> Waiting for main loop to complete'); await this.mainLoopPromise; } catch (error) { this.logger.error( diff --git a/pkgs/edge-worker/src/core/WorkerLifecycle.ts b/pkgs/edge-worker/src/core/WorkerLifecycle.ts index 37b08aa0f..051e3d06f 100644 --- a/pkgs/edge-worker/src/core/WorkerLifecycle.ts +++ b/pkgs/edge-worker/src/core/WorkerLifecycle.ts @@ -47,10 +47,6 @@ export class WorkerLifecycle implements InternalLifecycle acknowledgeStop() { this.workerState.transitionTo(States.Stopping); - if (!this.workerRow) { - throw new Error('Cannot stop worker: workerRow not set'); - } - try { this.logger.debug('Acknowledging worker stop...'); diff --git a/pkgs/edge-worker/src/core/WorkerState.ts b/pkgs/edge-worker/src/core/WorkerState.ts index be7b7cac2..828617967 100644 --- a/pkgs/edge-worker/src/core/WorkerState.ts +++ b/pkgs/edge-worker/src/core/WorkerState.ts @@ -22,7 +22,7 @@ export enum States { } export const Transitions: Record = { - [States.Created]: [States.Starting], + [States.Created]: [States.Starting, States.Stopping], [States.Starting]: [States.Running], [States.Running]: [States.Deprecated, States.Stopping], [States.Deprecated]: [States.Stopping], diff --git a/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts b/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts index a6608fd8e..cec9562ae 100644 --- a/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts +++ b/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts @@ -114,10 +114,6 @@ export class FlowWorkerLifecycle implements InternalLifec acknowledgeStop() { this.workerState.transitionTo(States.Stopping); - if (!this.workerRow) { - throw new Error('Cannot stop worker: workerRow not set'); - } - try { this.logger.debug('Acknowledging worker stop...'); this.workerState.transitionTo(States.Stopped); diff --git a/pkgs/edge-worker/src/flow/StepTaskPoller.ts b/pkgs/edge-worker/src/flow/StepTaskPoller.ts index f61d9e869..e7de964d5 100644 --- a/pkgs/edge-worker/src/flow/StepTaskPoller.ts +++ b/pkgs/edge-worker/src/flow/StepTaskPoller.ts @@ -115,7 +115,9 @@ export class StepTaskPoller return taskWithMessages; } catch (err: unknown) { this.logger.error(`Error in two-phase polling for flow tasks: ${err}`); - return []; + // Rethrow so Worker can distinguish a failed poll (which drives its + // retry backoff) from an empty successful poll. + throw err; } } diff --git a/pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts b/pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts index c9cece8a4..fddfab8cf 100644 --- a/pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts +++ b/pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts @@ -43,6 +43,7 @@ export class ProcessPlatformAdapter implements PlatformAdapter | null = null; private signalHandlersRegistered = false; private signalCount = 0; + private startupCompleted = false; private readonly signalHandler = () => this.handleSignal(); constructor( @@ -75,7 +76,12 @@ export class ProcessPlatformAdapter implements PlatformAdapter { - await this.cleanup(); + try { + await this.cleanup(); + } catch (cleanupError) { + // A cleanup failure must not replace the startup failure. + this.logger.error('Cleanup after startup failure failed', cleanupError); + } throw error; } ); @@ -141,6 +147,7 @@ export class ProcessPlatformAdapter implements PlatformAdapter { this.requestShutdown(); + let operationError: { error: unknown } | null = null; try { await this.startupPromise; if (this.worker) { @@ -172,8 +180,22 @@ export class ProcessPlatformAdapter implements PlatformAdapter { } +/** + * Hard ceiling for the whole Supabase shutdown operation: replacement + * settlement, worker drain, marking, and SQL close must all fit inside it. + */ +const SUPABASE_SHUTDOWN_DEADLINE_MS = 5_000; + /** * Supabase platform adapter for Deno runtime environment. * IMPORTANT: This class assumes it is running within a Deno environment @@ -44,6 +50,7 @@ export class SupabasePlatformAdapter implements PlatformAdapter | null = null; + private stopPromise: Promise | null = null; private logger: Logger; private abortController: AbortController; private _platformResources: SupabaseResources; @@ -105,18 +112,134 @@ export class SupabasePlatformAdapter implements PlatformAdapter { + stopWorker(): Promise { + this.stopPromise ??= this.performStopWorker(); + return this.stopPromise; + } + + /** + * Single shared stop operation: concurrent callers share it and SQL closes + * exactly once. Awaits any in-flight worker replacement so the worker row + * is guaranteed to exist before it is marked stopped. + * + * The whole operation (replacement settlement, worker drain, marking) runs + * under one deadline; when the deadline expires SQL is force-closed and the + * stop rejects with a timeout error. + */ + private async performStopWorker(): Promise { this.requestShutdown(); + const startedAt = Date.now(); + let deadlineTimer: ReturnType | undefined; + const deadlinePromise = new Promise((resolve) => { + deadlineTimer = setTimeout(() => { + resolve(); + }, SUPABASE_SHUTDOWN_DEADLINE_MS); + }); + + // Boxed so a rejection value of undefined cannot be mistaken for success, + // and so a synchronously throwing peer can never reject this promise. + const work = this.drainAndMark().then( + (failure) => ({ failure }), + (error) => ({ failure: { error } }) + ); + + const settled = await Promise.race([ + work.then(() => 'work' as const), + deadlinePromise.then(() => 'deadline' as const), + ]); + clearTimeout(deadlineTimer); + + if (settled === 'deadline') { + this.logger.warn( + `Supabase shutdown exceeded the ${SUPABASE_SHUTDOWN_DEADLINE_MS}ms deadline; forcing sql close` + ); + try { + await this._platformResources.sql.end({ timeout: 0 }); + } catch (closeError) { + this.logger.error('Failed to force close sql connection', closeError); + } + // The detached operation may still settle later; observe it so it + // cannot surface as an unhandled rejection. + work.then(({ failure }) => { + if (failure) { + this.logger.error('Supabase shutdown operation failed after the deadline', failure.error); + } + }); + throw new Error(`Supabase shutdown timed out after ${SUPABASE_SHUTDOWN_DEADLINE_MS}ms`); + } + + const { failure } = await work; + + const remainingMs = Math.max(SUPABASE_SHUTDOWN_DEADLINE_MS - (Date.now() - startedAt), 0); try { - if (this.worker) { - await this.worker.stop(); + await this._platformResources.sql.end({ timeout: remainingMs / 1000 }); + } catch (closeError) { + if (!failure) { + throw closeError; } - } finally { - await this._platformResources.sql.end(); + // A failing close must not replace the earlier operation error. + this.logger.error('Failed to close sql connection', closeError); + } + + if (failure) { + throw failure.error; } } + /** + * Waits for any in-flight replacement, then drains the current worker and + * marks it stopped. The drain starts before marking so database bookkeeping + * can never delay it, and both peers are awaited so one failure cannot skip + * the other. Resolves with the first operational failure (if any), never + * rejects. + */ + private async drainAndMark(): Promise<{ error: unknown } | null> { + // Capture the promise before awaiting: its owner clears the field in a + // finally block, but shutdown must await the replacement it observed. + const replacement = this.workerReplacementPromise; + if (replacement) { + // Startup/replacement rejection is already handled by the HTTP + // startup path; shutdown continues with cleanup. + await replacement.catch(() => undefined); + } + + // Snapshot after replacement settlement: replaceWorker may still be + // swapping worker and worker id. + const worker = this.worker; + const workerId = this.workerId; + if (!worker) { + return null; + } + + // Signal death to ensure_workers() cron by setting stopped_at. + // This allows the cron to immediately ping for a replacement worker. + const [drainResult, markResult] = await Promise.allSettled([ + worker.stop(), + workerId ? this.queries.markWorkerStopped(workerId) : Promise.resolve(), + ]); + + const failures: unknown[] = []; + if (drainResult.status === 'rejected') { + this.logger.error('Failed to drain worker', drainResult.reason); + failures.push(drainResult.reason); + } + if (markResult.status === 'rejected') { + this.logger.error('Failed to mark worker stopped', markResult.reason); + failures.push(markResult.reason); + } + + if (failures.length === 0) { + return null; + } + if (failures.length === 1) { + return { error: failures[0] }; + } + return { + error: new AggregateError(failures, 'Worker drain and worker marking both failed'), + }; + } + requestShutdown(): void { this.abortController.abort(); } @@ -181,15 +304,6 @@ export class SupabasePlatformAdapter implements PlatformAdapter { this.logger.debug('Shutting down...'); - - if (this.worker) { - // Signal death to ensure_workers() cron by setting stopped_at. - // This allows the cron to immediately ping for a replacement worker. - if (this.workerId) { - await this.queries.markWorkerStopped(this.workerId); - } - } - await this.stopWorker(); }); } @@ -243,23 +357,50 @@ export class SupabasePlatformAdapter implements PlatformAdapter { - while (this.needsWorkerReplacement()) { - if (this.workerReplacementPromise) { - await this.workerReplacementPromise; + for (;;) { + if (this.stopPromise) { + throw new Error('Worker startup rejected: shutdown in progress'); + } + + // Wait for an in-flight replacement before deciding anything: the + // worker reference exists while its startup is still pending. + const inFlight = this.workerReplacementPromise; + if (inFlight) { + await inFlight; continue; } - this.workerReplacementPromise = this.replaceWorker(req, createWorkerFn); + if (!this.needsWorkerReplacement()) { + return false; + } + + // Assign and own the replacement in one synchronous section so + // shutdown always observes an admitted replacement. + const owned = this.replaceWorker(req, createWorkerFn); + this.workerReplacementPromise = owned; try { - await this.workerReplacementPromise; - return true; + await owned; } finally { - this.workerReplacementPromise = null; + if (this.workerReplacementPromise === owned) { + this.workerReplacementPromise = null; + } } - } - return false; + // Re-check shutdown after the awaited replacement instead of reporting + // readiness directly: once stopPromise exists, no request may report + // a started worker. + if (this.stopPromise) { + throw new Error('Worker startup rejected: shutdown in progress'); + } + return true; + } } private async replaceWorker(req: Request, createWorkerFn: CreateWorkerFn): Promise { @@ -270,7 +411,11 @@ export class SupabasePlatformAdapter implements PlatformAdapter processLike.on?.(signal, handler), - offSignal: (signal, handler) => processLike.off?.(signal, handler), + offSignal: processLike.off + ? (signal, handler) => processLike.off?.(signal, handler) + : undefined, exit: (code) => processLike.exit!(code), setExitCode: (code) => { processLike.exitCode = code; diff --git a/pkgs/edge-worker/tests/unit/FlowWorkerLifecycle.deprecation.test.ts b/pkgs/edge-worker/tests/unit/FlowWorkerLifecycle.deprecation.test.ts index 50ab16e46..c1cb89da0 100644 --- a/pkgs/edge-worker/tests/unit/FlowWorkerLifecycle.deprecation.test.ts +++ b/pkgs/edge-worker/tests/unit/FlowWorkerLifecycle.deprecation.test.ts @@ -167,6 +167,18 @@ Deno.test('FlowWorkerLifecycle - deprecated state transitions', async () => { assertEquals(lifecycle.isStopped, true); }); +Deno.test('FlowWorkerLifecycle - stopping a never-started lifecycle reaches Stopped without a worker row', () => { + const mockQueries = new MockQueries(); + const mockFlow = createMockFlow(); + const lifecycle = new FlowWorkerLifecycle(mockQueries, mockFlow, logger); + + lifecycle.transitionToStopping(); + assertEquals(lifecycle.isStopping, true); + + lifecycle.acknowledgeStop(); + assertEquals(lifecycle.isStopped, true); +}); + Deno.test('FlowWorkerLifecycle - cannot transition to deprecated from non-running states', () => { const mockQueries = new MockQueries(); const mockFlow = createMockFlow(); diff --git a/pkgs/edge-worker/tests/unit/Poller.batchSize.test.ts b/pkgs/edge-worker/tests/unit/Poller.batchSize.test.ts index c1aeec6ef..44cd94f91 100644 --- a/pkgs/edge-worker/tests/unit/Poller.batchSize.test.ts +++ b/pkgs/edge-worker/tests/unit/Poller.batchSize.test.ts @@ -1,4 +1,4 @@ -import { assertEquals } from '@std/assert'; +import { assertEquals, assertRejects } from '@std/assert'; import { ReadWithPollPoller } from '../../src/queue/ReadWithPollPoller.ts'; import { StepTaskPoller } from '../../src/flow/StepTaskPoller.ts'; import { fakeLogger } from '../fakes.ts'; @@ -182,3 +182,26 @@ Deno.test('StepTaskPoller uses configured batchSize without limit', async () => assertEquals(readQty, 5); }); + +Deno.test('StepTaskPoller rethrows readMessages failures instead of returning an empty batch', async () => { + const adapter = { + readMessages: () => Promise.reject(new Error('connection refused')), + startTasks: () => Promise.resolve([]), + }; + + const poller = new StepTaskPoller( + adapter as never, + new AbortController().signal, + { + batchSize: 5, + queueName: 'test_flow', + visibilityTimeout: 10, + maxPollSeconds: 1, + pollIntervalMs: 100, + }, + () => 'worker-id', + fakeLogger + ); + + await assertRejects(() => poller.poll(), Error, 'connection refused'); +}); diff --git a/pkgs/edge-worker/tests/unit/Worker.mainLoop.test.ts b/pkgs/edge-worker/tests/unit/Worker.mainLoop.test.ts new file mode 100644 index 000000000..7438922c1 --- /dev/null +++ b/pkgs/edge-worker/tests/unit/Worker.mainLoop.test.ts @@ -0,0 +1,196 @@ +import { assertEquals } from '@std/assert'; +import { FakeTime } from '@std/testing/time'; +import { Worker } from '../../src/core/Worker.ts'; +import type { IBatchProcessor, ILifecycle, WorkerBootstrap } from '../../src/core/types.ts'; +import { States, WorkerState } from '../../src/core/WorkerState.ts'; +import { fakeLogger } from '../fakes.ts'; + +/** + * Lifecycle backed by a real WorkerState: acknowledgeStart transitions + * Created -> Starting -> Running, so the main loop runs until stop(). + */ +function createRunningLifecycle(): ILifecycle { + const workerState = new WorkerState(fakeLogger); + + return { + acknowledgeStart: () => { + workerState.transitionTo(States.Starting); + workerState.transitionTo(States.Running); + return Promise.resolve(); + }, + acknowledgeStop: () => { + workerState.transitionTo(States.Stopped); + }, + sendHeartbeat: () => Promise.resolve(), + get edgeFunctionName() { + return 'test-function'; + }, + get queueName() { + return 'test-queue'; + }, + get isCreated() { + return workerState.isCreated; + }, + get isStarting() { + return workerState.isStarting; + }, + get isRunning() { + return workerState.isRunning; + }, + get isDeprecated() { + return workerState.isDeprecated; + }, + get isStopping() { + return workerState.isStopping; + }, + get isStopped() { + return workerState.isStopped; + }, + transitionToStopping: () => { + workerState.transitionTo(States.Stopping); + }, + }; +} + +function createBatchProcessor( + outcomeFor: (attempt: number) => Promise, + batchTimes: number[] +): IBatchProcessor { + return { + processBatch: () => { + batchTimes.push(Date.now()); + return outcomeFor(batchTimes.length); + }, + awaitCompletion: () => Promise.resolve(), + }; +} + +/** + * Advances the fake clock by `ms` and settles the loop continuation that the + * fired timer unblocks. Sync tick + microtask drain keeps the clock at each + * timer's due time, so Date.now() stamps land exactly on schedule. + */ +async function advance(time: FakeTime, ms: number) { + time.tick(ms); + await time.runMicrotasks(); + await time.runMicrotasks(); +} + +const workerBootstrap: WorkerBootstrap = { + edgeFunctionName: 'test-function', + workerId: 'test-worker-id', +}; + +const fail = () => Promise.reject(new Error('db down')); +const succeed = () => Promise.resolve(); + +Deno.test('Worker main loop delays the next iteration after a failed iteration', async () => { + const time = new FakeTime(); + try { + const batchTimes: number[] = []; + const worker = new Worker( + createBatchProcessor(fail, batchTimes), + createRunningLifecycle(), + fakeLogger + ); + + await worker.startOnlyOnce(workerBootstrap); + await time.runMicrotasks(); + assertEquals(batchTimes.length, 1); + + await advance(time, 50); + assertEquals(batchTimes.length, 1, 'second iteration must not run immediately after a failure'); + + await advance(time, 50); + assertEquals(batchTimes.length, 2, 'second iteration runs after the initial 100ms backoff'); + assertEquals(batchTimes[1]! - batchTimes[0]!, 100); + + await worker.stop(); + } finally { + time.restore(); + } +}); + +Deno.test('Worker main loop backoff grows exponentially and caps at five seconds', async () => { + const time = new FakeTime(); + try { + const batchTimes: number[] = []; + const worker = new Worker( + createBatchProcessor(fail, batchTimes), + createRunningLifecycle(), + fakeLogger + ); + + await worker.startOnlyOnce(workerBootstrap); + await time.runMicrotasks(); + assertEquals(batchTimes.length, 1); + + for (const delay of [100, 200, 400, 800, 1600, 3200, 5000, 5000]) { + await advance(time, delay); + } + assertEquals(batchTimes.length, 9); + + await worker.stop(); + + const deltas = batchTimes.slice(1).map((t, i) => t - batchTimes[i]!); + assertEquals(deltas, [100, 200, 400, 800, 1600, 3200, 5000, 5000]); + } finally { + time.restore(); + } +}); + +Deno.test('Worker main loop resets the backoff after a fully successful iteration', async () => { + const time = new FakeTime(); + try { + const batchTimes: number[] = []; + // fail, fail, succeed, fail, fail - the fifth call only exists to time + // the delay after the fourth (post-reset) failure. + const script = [fail, fail, succeed, fail, fail]; + const worker = new Worker( + createBatchProcessor((attempt) => script[Math.min(attempt - 1, script.length - 1)](), batchTimes), + createRunningLifecycle(), + fakeLogger + ); + + await worker.startOnlyOnce(workerBootstrap); + await time.runMicrotasks(); + assertEquals(batchTimes.length, 1); + + await advance(time, 100); // second failure at +100, backoff grows to 200 + await advance(time, 200); // success at +300, loop continues without delay + await advance(time, 100); // fourth failure at +300, backoff reset to 100 + assertEquals(batchTimes.length, 5); + + await worker.stop(); + + const deltas = batchTimes.slice(1).map((t, i) => t - batchTimes[i]!); + // Failures grow 100 -> 200, the successful iteration proceeds without a + // delay (0), and the next failure starts over at the initial 100ms. + assertEquals(deltas, [100, 200, 0, 100]); + } finally { + time.restore(); + } +}); + +Deno.test('Worker stop completes immediately while a retry delay is pending', async () => { + const time = new FakeTime(); + try { + const batchTimes: number[] = []; + const worker = new Worker( + createBatchProcessor(fail, batchTimes), + createRunningLifecycle(), + fakeLogger + ); + + await worker.startOnlyOnce(workerBootstrap); + await time.runMicrotasks(); + assertEquals(batchTimes.length, 1); + + // No clock advance happens here: stop must not wait out the 100ms delay. + await worker.stop(); + + assertEquals(batchTimes.length, 1, 'no further iteration after stop'); + } finally { + time.restore(); + } +}); diff --git a/pkgs/edge-worker/tests/unit/Worker.stop.test.ts b/pkgs/edge-worker/tests/unit/Worker.stop.test.ts index 7a851d85a..5f6e7785d 100644 --- a/pkgs/edge-worker/tests/unit/Worker.stop.test.ts +++ b/pkgs/edge-worker/tests/unit/Worker.stop.test.ts @@ -1,6 +1,7 @@ -import { assertEquals } from '@std/assert'; +import { assertEquals, assertRejects } from '@std/assert'; import { Worker } from '../../src/core/Worker.ts'; -import type { IBatchProcessor, ILifecycle } from '../../src/core/types.ts'; +import type { IBatchProcessor, ILifecycle, WorkerBootstrap } from '../../src/core/types.ts'; +import { States, WorkerState } from '../../src/core/WorkerState.ts'; import { fakeLogger } from '../fakes.ts'; function createRunningLifecycle(): ILifecycle { @@ -50,6 +51,69 @@ function createBatchProcessor(): IBatchProcessor { }; } +/** + * Lifecycle backed by a real WorkerState so stop tests exercise real + * transition validation, with controllable startup readiness. + */ +function createStatefulLifecycle() { + const workerState = new WorkerState(fakeLogger); + let resolveStart: (() => void) | null = null; + + const lifecycle: ILifecycle = { + acknowledgeStart: () => { + workerState.transitionTo(States.Starting); + return new Promise((resolve) => { + resolveStart = () => { + workerState.transitionTo(States.Running); + resolve(); + }; + }); + }, + acknowledgeStop: () => { + workerState.transitionTo(States.Stopped); + }, + sendHeartbeat: () => Promise.resolve(), + get edgeFunctionName() { + return 'test-function'; + }, + get queueName() { + return 'test-queue'; + }, + get isCreated() { + return workerState.isCreated; + }, + get isStarting() { + return workerState.isStarting; + }, + get isRunning() { + return workerState.isRunning; + }, + get isDeprecated() { + return false; + }, + get isStopping() { + return workerState.isStopping; + }, + get isStopped() { + return workerState.isStopped; + }, + transitionToStopping: () => { + workerState.transitionTo(States.Stopping); + }, + }; + + return { + lifecycle, + workerState, + resolveStart: () => resolveStart?.(), + }; +} + +const workerBootstrap: WorkerBootstrap = { + edgeFunctionName: 'test-function', + workerId: 'test-worker-id', +}; + Deno.test('Worker.stop calls provided cleanup callback', async () => { let cleanupCalled = false; const worker = new Worker( @@ -105,3 +169,114 @@ Deno.test('Worker.stop side effects run once', async () => { assertEquals(transitionCount, 1, 'transitionToStopping should be called once'); }); + +Deno.test('Worker.stop during startup aborts immediately and transitions only after readiness', async () => { + const { lifecycle, workerState, resolveStart } = createStatefulLifecycle(); + let shutdownRequests = 0; + let transitionsToStopping = 0; + let stopAcknowledgements = 0; + let processBatchCalls = 0; + let cleanupCalls = 0; + + lifecycle.transitionToStopping = () => { + transitionsToStopping++; + workerState.transitionTo(States.Stopping); + }; + lifecycle.acknowledgeStop = () => { + stopAcknowledgements++; + workerState.transitionTo(States.Stopped); + }; + + const batchProcessor: IBatchProcessor = { + processBatch: () => { + processBatchCalls++; + return Promise.resolve(); + }, + awaitCompletion: () => Promise.resolve(), + }; + + const worker = new Worker(batchProcessor, lifecycle, fakeLogger, { + requestShutdown: () => { + shutdownRequests++; + }, + cleanup: () => { + cleanupCalls++; + return Promise.resolve(); + }, + }); + + const startup = worker.startOnlyOnce(workerBootstrap); + assertEquals(workerState.current, States.Starting, 'lifecycle must be Starting while startup is pending'); + + const stopPromise = worker.stop(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + assertEquals(shutdownRequests, 1, 'shutdown must be requested while startup is pending'); + assertEquals(transitionsToStopping, 0, 'no transition while startup is pending'); + assertEquals(workerState.current, States.Starting, 'state must stay Starting while startup is pending'); + + resolveStart(); + await startup; + await stopPromise; + + assertEquals(transitionsToStopping, 1, 'exactly one transition to Stopping'); + assertEquals(stopAcknowledgements, 1, 'exactly one stop acknowledgement'); + assertEquals(workerState.current, States.Stopped); + assertEquals(processBatchCalls, 0, 'no batch may start after shutdown was requested during startup'); + assertEquals(cleanupCalls, 1, 'cleanup must run once'); +}); + +Deno.test('Worker.stop during startup propagates the startup error', async () => { + const { lifecycle } = createStatefulLifecycle(); + lifecycle.acknowledgeStart = () => { + return Promise.reject(new Error('bootstrap failed')); + }; + + const worker = new Worker(createBatchProcessor(), lifecycle, fakeLogger); + + const startup = worker.startOnlyOnce(workerBootstrap); + const stopPromise = worker.stop(); + + await assertRejects(() => startup, Error, 'bootstrap failed'); + await assertRejects(() => stopPromise, Error, 'bootstrap failed'); +}); + +Deno.test('Worker.stop on a never-started worker reaches Stopped and cleans up once', async () => { + const { lifecycle, workerState } = createStatefulLifecycle(); + let cleanupCalls = 0; + let stopAcknowledgements = 0; + lifecycle.acknowledgeStop = () => { + stopAcknowledgements++; + workerState.transitionTo(States.Stopped); + }; + + const worker = new Worker(createBatchProcessor(), lifecycle, fakeLogger, { + cleanup: () => { + cleanupCalls++; + return Promise.resolve(); + }, + }); + + await worker.stop(); + + assertEquals(workerState.current, States.Stopped, 'Created -> Stopping -> Stopped must be allowed'); + assertEquals(stopAcknowledgements, 1); + assertEquals(cleanupCalls, 1, 'cleanup must run once'); +}); + +Deno.test('Worker.startOnlyOnce is ignored after a Created-state stop', async () => { + const { lifecycle } = createStatefulLifecycle(); + let acknowledgeStartCalls = 0; + lifecycle.acknowledgeStart = () => { + acknowledgeStartCalls++; + return Promise.resolve(); + }; + + const worker = new Worker(createBatchProcessor(), lifecycle, fakeLogger); + await worker.stop(); + + await worker.startOnlyOnce(workerBootstrap); + + assertEquals(acknowledgeStartCalls, 0, 'start must be ignored after the worker already stopped'); + assertEquals(lifecycle.isStopped, true); +}); diff --git a/pkgs/edge-worker/tests/unit/WorkerLifecycle.deprecation.test.ts b/pkgs/edge-worker/tests/unit/WorkerLifecycle.deprecation.test.ts index f428b3724..22be02bf1 100644 --- a/pkgs/edge-worker/tests/unit/WorkerLifecycle.deprecation.test.ts +++ b/pkgs/edge-worker/tests/unit/WorkerLifecycle.deprecation.test.ts @@ -237,6 +237,18 @@ Deno.test('WorkerLifecycle - deprecated state transitions', async () => { assertEquals(lifecycle.isStopped, true); }); +Deno.test('WorkerLifecycle - stopping a never-started lifecycle reaches Stopped without a worker row', () => { + const mockQueries = new MockQueries(); + const mockQueue = new MockQueue('test-queue'); + const lifecycle = new WorkerLifecycle(mockQueries, mockQueue, logger); + + lifecycle.transitionToStopping(); + assertEquals(lifecycle.isStopping, true); + + lifecycle.acknowledgeStop(); + assertEquals(lifecycle.isStopped, true); +}); + Deno.test( 'WorkerLifecycle - cannot transition to deprecated from non-running states', () => { diff --git a/pkgs/edge-worker/tests/unit/WorkerState.test.ts b/pkgs/edge-worker/tests/unit/WorkerState.test.ts index 0cbceb847..42932b151 100644 --- a/pkgs/edge-worker/tests/unit/WorkerState.test.ts +++ b/pkgs/edge-worker/tests/unit/WorkerState.test.ts @@ -61,6 +61,20 @@ Deno.test('WorkerState - valid deprecation transitions', () => { assertEquals(state.current, States.Stopped); }); +Deno.test('WorkerState - valid cancellation from Created before startup', () => { + const state = new WorkerState(logger); + + // Created -> Stopping (explicit cancellation) + state.transitionTo(States.Stopping); + assertEquals(state.current, States.Stopping); + assertEquals(state.isStopping, true); + + // Stopping -> Stopped + state.transitionTo(States.Stopped); + assertEquals(state.current, States.Stopped); + assertEquals(state.isStopped, true); +}); + Deno.test('WorkerState - invalid state transitions should throw', () => { const state = new WorkerState(logger); @@ -81,6 +95,17 @@ Deno.test('WorkerState - invalid state transitions should throw', () => { TransitionError, 'Cannot transition from created to stopped' ); + + // Cannot transition from Starting to Stopping + const startingState = new WorkerState(logger); + startingState.transitionTo(States.Starting); + assertThrows( + () => { + startingState.transitionTo(States.Stopping); + }, + TransitionError, + 'Cannot transition from starting to stopping' + ); }); Deno.test('WorkerState - transitioning to same state should be no-op', () => { diff --git a/pkgs/edge-worker/tests/unit/platform/ProcessPlatformAdapter.test.ts b/pkgs/edge-worker/tests/unit/platform/ProcessPlatformAdapter.test.ts index 7fdb295b8..0788fb8b9 100644 --- a/pkgs/edge-worker/tests/unit/platform/ProcessPlatformAdapter.test.ts +++ b/pkgs/edge-worker/tests/unit/platform/ProcessPlatformAdapter.test.ts @@ -271,9 +271,35 @@ Deno.test('ProcessPlatformAdapter deprecation drains and exits zero', async () = assertEquals(exit.calls, [[0]]); }); -Deno.test('ProcessPlatformAdapter handles a first signal during startup after readiness', async () => { +Deno.test('ProcessPlatformAdapter first signal during startup exits immediately with zero', async () => { + const { deps, handlers, exit } = createDeps(validEnv()); + const sql = createSqlStub(); + const worker = createWorkerStub(); + worker.startOnlyOnce.implementation = () => new Promise(() => undefined); + const adapter = new ProcessPlatformAdapter({ sql }, deps); + const end = createSpy<[], Promise>(() => Promise.resolve()); + (adapter.sql as unknown as { end: typeof end }).end = end; + + const startupPromise = adapter.startWorker(() => worker as never); + + assertEquals(handlers.size, 3, 'Signal handlers must exist during startup'); + assertEquals(adapter.shutdownSignal.aborted, false); + + await assertRejects(async () => await handlers.get('SIGTERM')?.(), Error, 'exit:0'); + + assertEquals(adapter.shutdownSignal.aborted, true, 'first signal must abort the shutdown signal'); + assertEquals((deps.setExitCode as SpyFn<[number], void>).calls, [[0]]); + assertEquals(exit.calls, [[0]], 'exit must not wait for the hung startup'); + assertEquals(worker.stop.calls.length, 0, 'no worker drain before the hard exit'); + assertEquals(sql.calls.length, 0, 'no row marking before the hard exit'); + assertEquals(end.calls.length, 0, 'no owned-SQL cleanup before the hard exit'); + + void startupPromise; +}); + +Deno.test('ProcessPlatformAdapter manual stop during startup waits for readiness without exiting', async () => { let resolveStartup = () => {}; - const { deps, handlers } = createDeps(validEnv()); + const { deps, exit } = createDeps(validEnv()); const sql = createSqlStub(); const worker = createWorkerStub(); worker.startOnlyOnce.implementation = () => new Promise((resolve) => { @@ -282,19 +308,19 @@ Deno.test('ProcessPlatformAdapter handles a first signal during startup after re const adapter = new ProcessPlatformAdapter({ sql }, deps); const startupPromise = adapter.startWorker(() => worker as never); + const stopPromise = adapter.stopWorker(); + await new Promise((resolve) => setTimeout(resolve, 0)); - assertEquals(handlers.size, 3, 'Signal handlers must exist during startup'); - - const shutdownPromise = handlers.get('SIGTERM')?.(); - await Promise.resolve(); - assertEquals(worker.stop.calls.length, 0, 'Worker must not stop while startup is pending'); + assertEquals(worker.stop.calls.length, 0, 'manual stop must wait for startup'); + assertEquals(exit.calls.length, 0, 'manual stop must not exit the process'); resolveStartup(); await startupPromise; - await assertRejects(async () => await shutdownPromise, Error, 'exit:0'); + await stopPromise; assertEquals(worker.stop.calls.length, 1); - assertEquals(handlers.size, 0); + assertStringIncludes(sql.calls.at(-1) ?? '', 'pgflow.mark_worker_stopped'); + assertEquals(exit.calls.length, 0); }); Deno.test('ProcessPlatformAdapter startup failure cleans owned resources once', async () => { @@ -341,3 +367,84 @@ Deno.test('ProcessPlatformAdapter shutdown cleans owned resources once', async ( assertEquals(offSignal.calls.length, 3); assertEquals(handlers.size, 0); }); + +// ============================================================ +// Dual-Failure Error Precedence Tests (review follow-up) +// ============================================================ + +/** + * Replaces the adapter's logger.error with a recorder so tests can assert + * that cleanup failures are logged without being thrown. + */ +function captureErrorLogs(adapter: ProcessPlatformAdapter): string[] { + const messages: string[] = []; + const logger = (adapter as unknown as { + logger: { error: (message: string) => void }; + }).logger; + logger.error = (message: string) => { + messages.push(message); + }; + return messages; +} + +function rejectEnd(adapter: ProcessPlatformAdapter, endCalls: unknown[][]): void { + (adapter.sql as unknown as { + end: (options?: unknown) => Promise; + }).end = (options) => { + endCalls.push([options]); + return Promise.reject(new Error('sql close failed')); + }; +} + +Deno.test('ProcessPlatformAdapter startup failure keeps the startup error when owned sql close fails', async () => { + const { deps } = createDeps(validEnv()); + const worker = createWorkerStub(); + worker.startOnlyOnce.implementation = () => Promise.reject(new Error('startup failed')); + const adapter = new ProcessPlatformAdapter(undefined, deps); + const endCalls: unknown[][] = []; + rejectEnd(adapter, endCalls); + const loggedErrors = captureErrorLogs(adapter); + + await assertRejects(() => adapter.startWorker(() => worker as never), Error, 'startup failed'); + + assertEquals(endCalls.length, 1, 'owned sql.end must run once'); + assertStringIncludes(loggedErrors.join('\n'), 'Cleanup after startup failure failed'); +}); + +Deno.test('ProcessPlatformAdapter stop failure keeps the drain error when owned sql close fails', async () => { + const { deps } = createDeps(validEnv()); + const worker = createWorkerStub(); + worker.stop.implementation = () => Promise.reject(new Error('drain failed')); + const adapter = new ProcessPlatformAdapter(undefined, deps); + const endCalls: unknown[][] = []; + rejectEnd(adapter, endCalls); + const loggedErrors = captureErrorLogs(adapter); + + await adapter.startWorker(() => worker as never); + await assertRejects(() => adapter.stopWorker(), Error, 'drain failed'); + + assertEquals(endCalls.length, 1, 'owned sql.end must run once'); + assertStringIncludes(loggedErrors.join('\n'), 'Cleanup during shutdown failed'); +}); + +Deno.test('ProcessPlatformAdapter keeps the marking error when drain succeeds but owned sql close fails', async () => { + const { deps } = createDeps(validEnv()); + const worker = createWorkerStub(); + const adapter = new ProcessPlatformAdapter(undefined, deps); + const markWorkerStopped = createSpy<[string], Promise>(() => + Promise.reject(new Error('mark failed')) + ); + (adapter as unknown as { + queries: { markWorkerStopped: typeof markWorkerStopped }; + }).queries.markWorkerStopped = markWorkerStopped; + const endCalls: unknown[][] = []; + rejectEnd(adapter, endCalls); + const loggedErrors = captureErrorLogs(adapter); + + await adapter.startWorker(() => worker as never); + await assertRejects(() => adapter.stopWorker(), Error, 'mark failed'); + + assertEquals(worker.stop.calls.length, 1); + assertEquals(endCalls.length, 1, 'owned sql.end must run once'); + assertStringIncludes(loggedErrors.join('\n'), 'Cleanup during shutdown failed'); +}); diff --git a/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts b/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts index 24d03366d..afe0cc349 100644 --- a/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts +++ b/pkgs/edge-worker/tests/unit/platform/SupabasePlatformAdapter.test.ts @@ -1,4 +1,5 @@ -import { assertEquals, assertThrows } from '@std/assert'; +import { assertEquals, assertInstanceOf, assertRejects, assertStringIncludes, assertThrows } from '@std/assert'; +import { FakeTime } from '@std/testing/time'; import { SupabasePlatformAdapter } from '../../../src/platform/SupabasePlatformAdapter.ts'; import type { SupabasePlatformDeps } from '../../../src/platform/deps.ts'; import type { Worker } from '../../../src/core/Worker.ts'; @@ -695,6 +696,492 @@ Deno.test({ }, }); +Deno.test({ + name: 'shutdown during pending startup aborts immediately and marks only after startup settles', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + let shutdownHandler: (() => void | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + onShutdown: (h) => { + shutdownHandler = h; + }, + }); + + const events: string[] = []; + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: () => Promise; + }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.resolve(); + }; + + let resolveStartup = () => {}; + let workerStopCalls = 0; + const worker = { + startOnlyOnce: () => new Promise((resolve) => { + resolveStartup = resolve; + }), + stop: () => { + workerStopCalls++; + events.push('worker.stop'); + return Promise.resolve(); + }, + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + + await adapter.startWorker(() => worker); + + const request = () => new Request('http://localhost/functions/v1/my-worker', { + headers: { authorization: 'Bearer test-service-key' }, + }); + const responsePromise = serveHandler!(request()); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const shutdownPromise = shutdownHandler!(); + + assertEquals( + adapter.shutdownSignal.aborted, + true, + 'shutdown signal must abort while startup is still pending' + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + assertEquals(events, [], 'no mark, worker stop, or sql close before startup settles'); + assertEquals(workerStopCalls, 0); + + resolveStartup(); + await responsePromise; + await shutdownPromise; + + assertEquals( + events, + ['worker.stop', 'markWorkerStopped', 'sql.end'], + 'after startup settles the order must be drain, mark, close' + ); + assertEquals(workerStopCalls, 1); + }, +}); + +Deno.test({ + name: 'shutdown during failing startup closes sql without stopping the cleared worker', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + let shutdownHandler: (() => void | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + onShutdown: (h) => { + shutdownHandler = h; + }, + }); + + const events: string[] = []; + const sql = (() => Promise.resolve([])) as unknown as { + end: () => Promise; + }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.resolve(); + }; + + let rejectStartup = (_error: Error) => {}; + let workerStopCalls = 0; + const worker = { + startOnlyOnce: () => new Promise((_resolve, reject) => { + rejectStartup = reject; + }), + stop: () => { + workerStopCalls++; + return Promise.resolve(); + }, + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + + await adapter.startWorker(() => worker); + + const request = () => new Request('http://localhost/functions/v1/my-worker', { + headers: { authorization: 'Bearer test-service-key' }, + }); + const responsePromise = serveHandler!(request()); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const shutdownPromise = shutdownHandler!(); + + rejectStartup(new Error('startup failed')); + const response = await responsePromise; + assertEquals(response.status, 500); + + await shutdownPromise; + + assertEquals(events, ['sql.end'], 'sql must close and nothing must be marked or stopped'); + assertEquals(workerStopCalls, 0, 'cleared worker must not be stopped'); + }, +}); + +Deno.test({ + name: 'stopWorker still stops worker and closes sql when marking rejects', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return Promise.resolve(); + }, + } as unknown as Worker; + (adapter as unknown as { workerId: string | null }).workerId = 'worker-1'; + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.reject(new Error('mark failed')); + }; + + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + await assertRejects(() => adapter.stopWorker(), Error, 'mark failed'); + + assertEquals( + events, + ['worker.stop', 'markWorkerStopped', 'sql.end'], + 'worker must stop and sql must close even when marking rejects' + ); + }, +}); + +Deno.test({ + name: 'concurrent stopWorker calls share one stop operation', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return Promise.resolve(); + }, + } as unknown as Worker; + (adapter as unknown as { workerId: string | null }).workerId = 'worker-1'; + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.resolve(); + }; + + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const p1 = adapter.stopWorker(); + const p2 = adapter.stopWorker(); + + assertEquals(p1 === p2, true, 'Concurrent stopWorker calls should return the same promise'); + + await p1; + + assertEquals(events, ['worker.stop', 'markWorkerStopped', 'sql.end']); + }, +}); + +Deno.test({ + name: 'stopWorker before the first HTTP request closes sql without creating or marking a worker', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + let createCount = 0; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.resolve(); + }; + + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + await adapter.startWorker(() => { + createCount++; + return createMockWorker(); + }); + await adapter.stopWorker(); + + assertEquals(createCount, 0, 'no worker may be created before the first HTTP request'); + assertEquals(events, ['sql.end'], 'nothing to mark, sql closes once'); + }, +}); + +Deno.test({ + name: 'shutdown during deprecated worker replacement keeps the signal aborted', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + const events: string[] = []; + const markedIds: string[] = []; + + let releaseOldStop = () => {}; + let resolveOldStopStarted: (() => void) | null = null; + const oldStopStarted = new Promise((resolve) => { + resolveOldStopStarted = resolve; + }); + + const deprecatedWorker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('old.stop'); + resolveOldStopStarted?.(); + return new Promise((release) => { + releaseOldStop = release; + }); + }, + get isDeprecated() { + return true; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + + let resolveReplacementStartup = () => {}; + let resolveFactoryStarted: (() => void) | null = null; + const factoryStarted = new Promise((resolve) => { + resolveFactoryStarted = resolve; + }); + + const replacementWorker = { + startOnlyOnce: () => + new Promise((resolve) => { + events.push('replacement.startOnlyOnce'); + resolveReplacementStartup = resolve; + }), + stop: () => { + events.push('replacement.stop'); + return Promise.resolve(); + }, + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + }); + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: () => Promise; + }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (workerId) => { + markedIds.push(workerId); + return Promise.resolve(); + }; + + let createCount = 0; + await adapter.startWorker(() => { + createCount++; + if (createCount === 1) { + return deprecatedWorker; + } + resolveFactoryStarted?.(); + return replacementWorker; + }); + + const handler = serveHandler as unknown as (req: Request) => Response | Promise; + const request = () => + new Request('http://localhost/functions/v1/my-worker', { + headers: { authorization: 'Bearer test-service-key' }, + }); + await handler(request()); + + const replacementResponse = handler(request()); + await oldStopStarted; + + const stopPromise = adapter.stopWorker(); + assertEquals( + adapter.shutdownSignal.aborted, + true, + 'stopWorker must abort the signal synchronously' + ); + + releaseOldStop(); + await factoryStarted; + await new Promise((resolve) => setTimeout(resolve, 0)); + + assertEquals( + adapter.shutdownSignal.aborted, + true, + 'replacement must not make the adapter signal live again during stop' + ); + + resolveReplacementStartup(); + const response = await replacementResponse; + // Startup admission is permanently closed by stopPromise, so the + // in-flight request must not report the replacement as running. + assertEquals(response.status, 500); + await stopPromise; + + assertEquals(markedIds.length, 2); + assertEquals(markedIds[0], 'test-exec-id'); + assertEquals(markedIds[1] !== markedIds[0], true); + assertEquals(events, [ + 'old.stop', + 'replacement.startOnlyOnce', + 'replacement.stop', + 'sql.end', + ]); + }, +}); + +Deno.test({ + name: 'stopWorker reports worker stop error when sql close also rejects', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return Promise.reject(new Error('worker stop failed')); + }, + } as unknown as Worker; + + let sqlEndCalls = 0; + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + sqlEndCalls++; + events.push('sql.end'); + return Promise.reject(new Error('sql close failed')); + }; + + await assertRejects(() => adapter.stopWorker(), Error, 'worker stop failed'); + + assertEquals(events, ['worker.stop', 'sql.end']); + assertEquals(sqlEndCalls, 1); + }, +}); + +Deno.test({ + name: 'stopWorker reports marking error when sql close also rejects', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return Promise.resolve(); + }, + } as unknown as Worker; + (adapter as unknown as { workerId: string | null }).workerId = 'worker-1'; + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.reject(new Error('mark failed')); + }; + + let sqlEndCalls = 0; + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + sqlEndCalls++; + events.push('sql.end'); + return Promise.reject(new Error('sql close failed')); + }; + + await assertRejects(() => adapter.stopWorker(), Error, 'mark failed'); + + assertEquals(events, ['worker.stop', 'markWorkerStopped', 'sql.end']); + assertEquals(sqlEndCalls, 1); + }, +}); + +Deno.test({ + name: 'stopWorker rejects with the sql close error when no earlier failure', + sanitizeResources: false, + fn: async () => { + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + let sqlEndCalls = 0; + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + sqlEndCalls++; + return Promise.reject(new Error('sql close failed')); + }; + + await assertRejects(() => adapter.stopWorker(), Error, 'sql close failed'); + assertEquals(sqlEndCalls, 1); + }, +}); + Deno.test({ name: 'replacement marks old worker stopped before starting replacement', sanitizeResources: false, @@ -777,3 +1264,378 @@ Deno.test({ ); }, }); + +// ============================================================ +// Startup Admission Serialization Tests (review follow-up) +// ============================================================ + +function authedRequest(): Request { + return new Request('http://localhost/functions/v1/my-worker', { + headers: { authorization: 'Bearer test-service-key' }, + }); +} + +Deno.test({ + name: 'second HTTP request waits for an in-flight startup instead of reporting running', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + }); + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: () => Promise; + }; + sql.end = () => Promise.resolve(); + + let resolveStartup = () => {}; + let createCount = 0; + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + await adapter.startWorker(() => { + createCount++; + return { + startOnlyOnce: () => + new Promise((resolve) => { + resolveStartup = resolve; + }), + stop: () => Promise.resolve(), + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + }); + + const handler = serveHandler as unknown as (req: Request) => Response | Promise; + + const firstResponse = handler(authedRequest()); + await new Promise((resolve) => setTimeout(resolve, 0)); + + // replaceWorker() has published the worker, but startup is still pending. + let secondSettled = false; + const secondResponse = Promise.resolve(handler(authedRequest())); + secondResponse.then( + () => { + secondSettled = true; + }, + () => { + secondSettled = true; + } + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + assertEquals(secondSettled, false, 'second request must wait for the in-flight startup'); + + resolveStartup(); + const firstBody = await (await firstResponse).json(); + const secondBody = await (await secondResponse).json(); + + assertEquals(firstBody.status, 'started'); + assertEquals(secondBody.status, 'running'); + assertEquals(createCount, 1, 'both requests must share one worker'); + }, +}); + +Deno.test({ + name: 'shared startup rejection fails every waiting request', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + }); + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: () => Promise; + }; + sql.end = () => Promise.resolve(); + + let rejectStartup = (_error: Error) => {}; + let createCount = 0; + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + await adapter.startWorker(() => { + createCount++; + return { + startOnlyOnce: () => + new Promise((_resolve, reject) => { + rejectStartup = reject; + }), + stop: () => Promise.resolve(), + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + }); + + const handler = serveHandler as unknown as (req: Request) => Response | Promise; + + const firstResponse = handler(authedRequest()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const secondResponse = handler(authedRequest()); + await new Promise((resolve) => setTimeout(resolve, 0)); + + rejectStartup(new Error('sensitive database details')); + const first = await firstResponse; + const second = await secondResponse; + const firstBody = await first.text(); + const secondBody = await second.text(); + + assertEquals(first.status, 500); + assertEquals(second.status, 500); + assertEquals(firstBody.includes('sensitive database details'), false); + assertEquals(secondBody.includes('sensitive database details'), false); + assertEquals(firstBody.includes('running'), false); + assertEquals(secondBody.includes('running'), false); + assertEquals(createCount, 1); + }, +}); + +Deno.test({ + name: 'HTTP request after shutdown begins fails without creating a worker', + sanitizeResources: false, + fn: async () => { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + let shutdownHandler: (() => void | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + onShutdown: (h) => { + shutdownHandler = h; + }, + }); + + let releaseSqlEnd = () => {}; + let sqlEndCalls = 0; + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: () => Promise; + }; + sql.end = () => { + sqlEndCalls++; + return new Promise((resolve) => { + releaseSqlEnd = resolve; + }); + }; + + let createCount = 0; + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + await adapter.startWorker(() => { + createCount++; + return createMockWorker(); + }); + + const handler = serveHandler as unknown as (req: Request) => Response | Promise; + + const shutdownPromise = shutdownHandler!(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const response = await handler(authedRequest()); + + assertEquals(response.status, 500); + assertEquals(createCount, 0, 'no worker may be created after shutdown begins'); + + releaseSqlEnd(); + await shutdownPromise; + assertEquals(sqlEndCalls, 1); + }, +}); + +// ============================================================ +// Bounded Shutdown Tests (review follow-up) +// ============================================================ + +Deno.test({ + name: 'stopWorker starts the drain before marking and closes sql only after both settle', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + let releaseStop = () => {}; + let releaseMark = () => {}; + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return new Promise((resolve) => { + releaseStop = resolve; + }); + }, + } as unknown as Worker; + (adapter as unknown as { workerId: string | null }).workerId = 'worker-1'; + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return new Promise((resolve) => { + releaseMark = resolve; + }); + }; + + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const stopPromise = adapter.stopWorker(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + assertEquals( + events, + ['worker.stop', 'markWorkerStopped'], + 'drain must start first and sql must stay open while both are pending' + ); + + releaseStop(); + await new Promise((resolve) => setTimeout(resolve, 0)); + assertEquals( + events, + ['worker.stop', 'markWorkerStopped'], + 'sql must stay open until marking settles too' + ); + + releaseMark(); + await stopPromise; + assertEquals(events, ['worker.stop', 'markWorkerStopped', 'sql.end']); + }, +}); + +Deno.test({ + name: 'stopWorker rejects with an AggregateError when drain and marking both fail', + sanitizeResources: false, + fn: async () => { + const events: string[] = []; + const deps = createMockDeps(); + const adapter = new SupabasePlatformAdapter(undefined, deps); + + (adapter as unknown as { worker: Worker | null }).worker = { + startOnlyOnce: () => {}, + stop: () => { + events.push('worker.stop'); + return Promise.reject(new Error('drain failed')); + }, + } as unknown as Worker; + (adapter as unknown as { workerId: string | null }).workerId = 'worker-1'; + (adapter as unknown as { + queries: { markWorkerStopped: (workerId: string) => Promise }; + }).queries.markWorkerStopped = (_workerId) => { + events.push('markWorkerStopped'); + return Promise.reject(new Error('mark failed')); + }; + + const sql = adapter.sql as unknown as { end: () => Promise }; + sql.end = () => { + events.push('sql.end'); + return Promise.resolve(); + }; + + const error = await assertRejects(() => adapter.stopWorker()); + + assertInstanceOf(error, AggregateError); + assertEquals( + error.errors.map((e) => (e as Error).message), + ['drain failed', 'mark failed'], + 'AggregateError must preserve both errors in drain-then-mark order' + ); + assertEquals(events, ['worker.stop', 'markWorkerStopped', 'sql.end']); + }, +}); + +Deno.test({ + name: 'shutdown deadline force-closes sql when startup never settles', + sanitizeResources: false, + fn: async () => { + const time = new FakeTime(); + try { + let serveHandler: ((req: Request) => Response | Promise) | null = null; + let shutdownHandler: (() => void | Promise) | null = null; + const deps = createMockDeps({ + serve: (h) => { + serveHandler = h; + }, + onShutdown: (h) => { + shutdownHandler = h; + }, + }); + + const endCalls: unknown[] = []; + const sql = (() => Promise.resolve([{ has_active: true }])) as unknown as { + end: (options?: unknown) => Promise; + }; + sql.end = (options) => { + endCalls.push(options); + return Promise.resolve(); + }; + + const adapter = new SupabasePlatformAdapter({ sql: sql as never }, deps); + const warns: string[] = []; + (adapter as unknown as { + logger: { warn: (message: string) => void }; + }).logger.warn = (message) => { + warns.push(message); + }; + + // Startup never settles, so shutdown can only finish via the deadline. + const worker = { + startOnlyOnce: () => new Promise(() => undefined), + stop: () => Promise.resolve(), + get isDeprecated() { + return false; + }, + get isStopped() { + return false; + }, + } as unknown as Worker; + await adapter.startWorker(() => worker); + + const handler = serveHandler as unknown as (req: Request) => Response | Promise; + const responsePromise = Promise.resolve(handler(authedRequest())); + await time.tickAsync(0); + + const shutdownPromise = Promise.resolve(shutdownHandler!()); + // Capture the rejection immediately: a transiently unhandled rejection + // would fail the test runner before the assertions below. + const shutdownError = shutdownPromise.then( + () => new Error('expected shutdown to reject'), + (error: unknown) => error + ); + assertEquals(adapter.shutdownSignal.aborted, true); + + await time.tickAsync(5_000); + await time.tickAsync(0); + + const error = (await shutdownError) as Error; + assertStringIncludes(error.message, 'timed out after 5000ms'); + assertEquals(error instanceof Error && /5000ms/.test(error.message), true); + assertEquals(endCalls, [{ timeout: 0 }], 'sql must be force-closed once at the deadline'); + assertEquals(warns.length, 1, 'the deadline expiry must be logged once'); + assertStringIncludes(warns[0], '5000'); + assertEquals(adapter.shutdownSignal.aborted, true, 'signal must stay aborted'); + + let responseSettled = false; + responsePromise.then( + () => { + responseSettled = true; + }, + () => { + responseSettled = true; + } + ); + await time.tickAsync(0); + assertEquals(responseSettled, false, 'startup stays unresolved after the deadline'); + void shutdownPromise; + } finally { + time.restore(); + } + }, +}); diff --git a/pkgs/edge-worker/tests/unit/platform/processDeps.test.ts b/pkgs/edge-worker/tests/unit/platform/processDeps.test.ts new file mode 100644 index 000000000..a72b41291 --- /dev/null +++ b/pkgs/edge-worker/tests/unit/platform/processDeps.test.ts @@ -0,0 +1,59 @@ +import { assertEquals, assertThrows } from '@std/assert'; +import { getProcessDeps } from '../../../src/platform/processDeps.ts'; + +/** + * Replaces globalThis.process with a fake for the duration of one check and + * restores the original property descriptor afterwards. + */ +async function withFakeProcess( + processLike: Record | undefined, + check: () => Promise | void +): Promise { + const original = Object.getOwnPropertyDescriptor(globalThis, 'process'); + try { + Object.defineProperty(globalThis, 'process', { + value: processLike, + configurable: true, + writable: true, + enumerable: true, + }); + await check(); + } finally { + if (original) { + Object.defineProperty(globalThis, 'process', original); + } else { + delete (globalThis as { process?: unknown }).process; + } + } +} + +Deno.test('getProcessDeps works without process.off and omits offSignal', async () => { + await withFakeProcess( + { env: { SUPABASE_URL: 'https://example.supabase.co' }, on: () => {}, exit: () => {} }, + () => { + const deps = getProcessDeps(); + + assertEquals(typeof deps.onSignal, 'function'); + assertEquals(typeof deps.exit, 'function'); + assertEquals(typeof deps.randomUUID, 'function'); + assertEquals(deps.offSignal, undefined, 'offSignal must be optional when process.off is absent'); + } + ); +}); + +Deno.test('getProcessDeps exposes offSignal when process.off exists', async () => { + await withFakeProcess( + { env: {}, on: () => {}, off: () => {}, exit: () => {} }, + () => { + const deps = getProcessDeps(); + + assertEquals(typeof deps.offSignal, 'function'); + } + ); +}); + +Deno.test('getProcessDeps still requires env, on, exit, and randomUUID', async () => { + await withFakeProcess({ off: () => {} }, () => { + assertThrows(() => getProcessDeps(), Error, 'Process runtime is not available'); + }); +}); diff --git a/pkgs/website/src/content/docs/deploy/node-bun-process-workers.mdx b/pkgs/website/src/content/docs/deploy/node-bun-process-workers.mdx index 2906dfb7e..27437231a 100644 --- a/pkgs/website/src/content/docs/deploy/node-bun-process-workers.mdx +++ b/pkgs/website/src/content/docs/deploy/node-bun-process-workers.mdx @@ -104,7 +104,11 @@ Your process manager should keep this command running. On Railway, Fly, Docker, Process workers drain on `SIGTERM`, `SIGINT`, and `SIGQUIT`. -On the first signal, the worker stops accepting new work, lets in-flight tasks finish, marks the worker stopped in pgflow, and exits with code `0`. A second signal exits immediately with code `1`. +After the worker is ready, the first signal stops accepting new work, lets in-flight tasks finish, marks the worker stopped in pgflow, and exits with code `0`. + +During startup, before the worker is ready, the first signal exits immediately with code `0`. No tasks have been accepted yet, so there is nothing to drain, and waiting on startup could block termination. + +A second signal exits immediately with code `1`. ## Health diff --git a/pkgs/website/src/content/docs/news/pgflow-0-15-0-portable-workers-node-bun.mdx b/pkgs/website/src/content/docs/news/pgflow-0-15-0-portable-workers-node-bun.mdx new file mode 100644 index 000000000..6b63193f0 --- /dev/null +++ b/pkgs/website/src/content/docs/news/pgflow-0-15-0-portable-workers-node-bun.mdx @@ -0,0 +1,43 @@ +--- +draft: true +title: 'pgflow 0.15.0: Portable Workers with Node and Bun Support' +description: 'Run pgflow workers anywhere with the new @pgflow/edge-worker npm package, including long-running Node and Bun processes' +date: 2026-08-18 +authors: + - jumski +tags: + - feature + - minor +featured: false +--- + +import { Aside } from '@astrojs/starlight/components'; + +pgflow workers are no longer tied to Supabase Edge Functions. This release publishes `@pgflow/edge-worker` to npm and adds a process runtime that runs workers as long-lived Node or Bun processes — on Railway, Fly.io, Docker, or any VM. + +## Run a worker with Node or Bun + +Install the package, set your connection variables, and start a worker script: + +```ts title="worker.ts" +import { EdgeWorker } from '@pgflow/edge-worker'; +import { ProcessOrders } from './flows/process-orders.js'; + +await EdgeWorker.start(ProcessOrders); +``` + +Process workers handle `SIGTERM`, `SIGINT`, and `SIGQUIT` with graceful drain: tasks finish, the worker row is marked stopped, and connections close before exit. Deprecation via the control plane stops the process cleanly, and `ensure_workers()` can replace it. + +See the [Node/Bun Process Workers guide](/deploy/node-bun-process-workers/) for deployment recipes and configuration. + +## Supabase Edge Function improvements + +Edge Function workers also got lifecycle hardening in this release: + +- **Bounded shutdown.** Shutdown now finishes within a five-second deadline. Worker drain starts before database bookkeeping, and PostgreSQL connections are force-closed if the deadline expires. +- **Serialized startup.** Concurrent HTTP requests share one worker startup; none can observe a half-started worker, and shutdown permanently closes startup admission. +- **Retry backoff.** Persistent database failures now back off exponentially (100 ms to 5 s) instead of hot-looping, and the backoff resets after a healthy iteration. + +