Skip to content

Commit e06cd6b

Browse files
jumskijumski
authored
fix: preserve portable worker compatibility (#633)
1 parent 64a2321 commit e06cd6b

16 files changed

Lines changed: 399 additions & 100 deletions

‎pkgs/edge-worker/src/core/Queries.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,13 @@ export class Queries {
8282
* pinging during startup (debounce).
8383
*/
8484
async trackWorkerFunction(functionName: string, startMode: WorkerStartMode = 'http'): Promise<void> {
85+
if (startMode === 'http') {
86+
await this.sql`
87+
SELECT pgflow.track_worker_function(${functionName})
88+
`;
89+
return;
90+
}
91+
8592
await this.sql`
8693
SELECT pgflow.track_worker_function(${functionName}, ${startMode})
8794
`;

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ export class Worker {
6767

6868
if (!this.isMainLoopActive) {
6969
this.logDeprecation();
70-
if (this.lifecycle.isDeprecated) {
70+
if (this.isDeprecated) {
7171
this.deprecationHandler?.();
7272
}
7373
break;
@@ -142,15 +142,15 @@ export class Worker {
142142
}
143143

144144
get isStarting() {
145-
return this.lifecycle.isStarting;
145+
return this.lifecycle.isStarting ?? false;
146146
}
147147

148148
get isRunning() {
149149
return this.lifecycle.isRunning;
150150
}
151151

152152
get isDeprecated() {
153-
return this.lifecycle.isDeprecated;
153+
return this.lifecycle.isDeprecated ?? false;
154154
}
155155

156156
get isStopped() {

‎pkgs/edge-worker/src/core/WorkerLifecycle.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import type { Queries } from './Queries.js';
22
import type { Queue } from '../queue/Queue.js';
3-
import type { ILifecycle, Json, WorkerBootstrap, WorkerRow } from './types.js';
3+
import type { InternalLifecycle, Json, WorkerBootstrap, WorkerRow } from './types.js';
44
import { States, WorkerState } from './WorkerState.js';
55
import type { Logger } from '../platform/types.js';
66

@@ -9,7 +9,7 @@ export interface LifecycleConfig {
99
heartbeatInterval?: number;
1010
}
1111

12-
export class WorkerLifecycle<IMessage extends Json> implements ILifecycle {
12+
export class WorkerLifecycle<IMessage extends Json> implements InternalLifecycle {
1313
private workerState: WorkerState;
1414
private logger: Logger;
1515
private queries: Queries;

‎pkgs/edge-worker/src/core/types.ts‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,15 +27,20 @@ export interface ILifecycle {
2727
get edgeFunctionName(): string | undefined;
2828
get queueName(): string;
2929
get isCreated(): boolean;
30-
get isStarting(): boolean;
30+
readonly isStarting?: boolean;
3131
get isRunning(): boolean;
32-
get isDeprecated(): boolean;
32+
readonly isDeprecated?: boolean;
3333
get isStopping(): boolean;
3434
get isStopped(): boolean;
3535

3636
transitionToStopping(): void;
3737
}
3838

39+
export interface InternalLifecycle extends ILifecycle {
40+
readonly isStarting: boolean;
41+
readonly isDeprecated: boolean;
42+
}
43+
3944
export interface IBatchProcessor {
4045
processBatch(): Promise<void>;
4146
awaitCompletion(): Promise<void>;

‎pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import type { Queries } from '../core/Queries.js';
2-
import type { ILifecycle, WorkerBootstrap, WorkerRow } from '../core/types.js';
2+
import type { InternalLifecycle, WorkerBootstrap, WorkerRow } from '../core/types.js';
33
import type { Logger, StartupContext } from '../platform/types.js';
44
import { States, WorkerState } from '../core/WorkerState.js';
55
import type { AnyFlow } from '@pgflow/dsl';
@@ -21,7 +21,7 @@ type CompilationStatus = 'compiled' | 'verified' | 'recompiled' | 'mismatch';
2121
/**
2222
* A specialized WorkerLifecycle for Flow-based workers that is aware of the Flow's step types
2323
*/
24-
export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
24+
export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements InternalLifecycle {
2525
private workerState: WorkerState;
2626
private logger: Logger;
2727
private queries: Queries;

‎pkgs/edge-worker/src/platform/ProcessPlatformAdapter.ts‎

Lines changed: 64 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -37,9 +37,13 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
3737
private readonly queries: Queries;
3838
private worker: Worker | null = null;
3939
private workerId: string | null = null;
40+
private startupPromise: Promise<void> | null = null;
4041
private stopPromise: Promise<void> | null = null;
4142
private gracefulExitPromise: Promise<void> | null = null;
43+
private cleanupPromise: Promise<void> | null = null;
44+
private signalHandlersRegistered = false;
4245
private signalCount = 0;
46+
private readonly signalHandler = () => this.handleSignal();
4347

4448
constructor(
4549
options?: ProcessAdapterOptions,
@@ -48,41 +52,34 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
4852
this.deps = deps;
4953
this.assertProcessEnv(deps.env);
5054
this.validatedEnv = deps.env;
51-
this._connectionString = resolveConnectionString(this.validatedEnv, {
55+
const connectionOptions = {
56+
...options,
5257
hasSql: !!options?.sql,
53-
connectionString: options?.connectionString,
54-
});
58+
allowDatabaseUrl: true,
59+
};
60+
this._connectionString = resolveConnectionString(
61+
this.validatedEnv,
62+
connectionOptions
63+
);
5564
this.ownsSql = !options?.sql;
5665
this.loggingFactory = createLoggingFactory(this.validatedEnv);
5766
this.logger = this.loggingFactory.createLogger('ProcessPlatformAdapter');
5867
this._platformResources = {
59-
sql: resolveSqlConnection(this.validatedEnv, options),
68+
sql: resolveSqlConnection(this.validatedEnv, connectionOptions),
6069
supabase: createServiceSupabaseClient(this.validatedEnv),
6170
};
6271
this.queries = new Queries(this._platformResources.sql);
6372
}
6473

65-
async startWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
66-
const workerName = this.validatedEnv.WORKER_NAME || 'pgflow-worker';
67-
const workerId = this.deps.randomUUID();
68-
69-
this.workerId = workerId;
70-
this.loggingFactory.setWorkerId(workerId);
71-
this.loggingFactory.setWorkerName(workerName);
72-
73-
this.worker = createWorkerFn(this.loggingFactory.createLogger);
74-
75-
this.worker.onDeprecated(() => {
76-
this.handleDeprecation().catch(() => undefined);
77-
});
78-
79-
await this.worker.startOnlyOnce({
80-
edgeFunctionName: workerName,
81-
workerId,
82-
startMode: 'process',
83-
});
84-
74+
startWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
8575
this.registerSignalHandlers();
76+
this.startupPromise ??= this.performStartWorker(createWorkerFn).catch(
77+
async (error) => {
78+
await this.cleanup();
79+
throw error;
80+
}
81+
);
82+
return this.startupPromise;
8683
}
8784

8885
stopWorker(): Promise<void> {
@@ -126,27 +123,68 @@ export class ProcessPlatformAdapter implements PlatformAdapter<SupabaseResources
126123
return this._platformResources.supabase;
127124
}
128125

126+
private async performStartWorker(createWorkerFn: CreateWorkerFn): Promise<void> {
127+
const workerName = this.validatedEnv.WORKER_NAME || 'pgflow-worker';
128+
const workerId = this.deps.randomUUID();
129+
130+
this.workerId = workerId;
131+
this.loggingFactory.setWorkerId(workerId);
132+
this.loggingFactory.setWorkerName(workerName);
133+
134+
this.worker = createWorkerFn(this.loggingFactory.createLogger);
135+
this.worker.onDeprecated(() => {
136+
this.handleDeprecation().catch(() => undefined);
137+
});
138+
139+
await this.worker.startOnlyOnce({
140+
edgeFunctionName: workerName,
141+
workerId,
142+
startMode: 'process',
143+
});
144+
}
145+
129146
private registerSignalHandlers(): void {
147+
if (this.signalHandlersRegistered) return;
148+
149+
this.signalHandlersRegistered = true;
130150
for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
131-
this.deps.onSignal(signal, () => this.handleSignal());
151+
this.deps.onSignal(signal, this.signalHandler);
152+
}
153+
}
154+
155+
private removeSignalHandlers(): void {
156+
if (!this.signalHandlersRegistered) return;
157+
158+
this.signalHandlersRegistered = false;
159+
for (const signal of ['SIGTERM', 'SIGINT', 'SIGQUIT'] satisfies ProcessSignal[]) {
160+
this.deps.offSignal?.(signal, this.signalHandler);
132161
}
133162
}
134163

135164
private async performStopWorker(): Promise<void> {
136165
this.requestShutdown();
137166

138167
try {
168+
await this.startupPromise;
139169
if (this.worker) {
140170
await this.worker.stop();
141171
}
142172
if (this.workerId) {
143173
await this.queries.markWorkerStopped(this.workerId);
144174
}
145175
} finally {
176+
await this.cleanup();
177+
}
178+
}
179+
180+
private cleanup(): Promise<void> {
181+
this.cleanupPromise ??= (async () => {
182+
this.removeSignalHandlers();
146183
if (this.ownsSql) {
147184
await this._platformResources.sql.end();
148185
}
149-
}
186+
})();
187+
return this.cleanupPromise;
150188
}
151189

152190
private gracefulExit(): Promise<void> {

‎pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts‎

Lines changed: 29 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -222,15 +222,20 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
222222

223223
this.logger.debug(`HTTP Request: ${this.edgeFunctionName}`);
224224

225-
const wasStarted = await this.ensureWorkerStarted(req, createWorkerFn);
226-
227-
return new Response(JSON.stringify({
228-
status: wasStarted ? 'started' : 'running',
229-
workerId: this.workerId,
230-
functionName: this.edgeFunctionName,
231-
}), {
232-
headers: { 'Content-Type': 'application/json' },
233-
});
225+
try {
226+
const wasStarted = await this.ensureWorkerStarted(req, createWorkerFn);
227+
228+
return new Response(JSON.stringify({
229+
status: wasStarted ? 'started' : 'running',
230+
workerId: this.workerId,
231+
functionName: this.edgeFunctionName,
232+
}), {
233+
headers: { 'Content-Type': 'application/json' },
234+
});
235+
} catch (error) {
236+
this.logger.error('Worker startup failed', error);
237+
return createServerErrorResponse();
238+
}
234239
});
235240
}
236241

@@ -279,12 +284,21 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
279284
this.loggingFactory.setWorkerName(this.edgeFunctionName);
280285

281286
// Create the worker using the factory function and the logger
282-
this.worker = createWorkerFn(this.loggingFactory.createLogger);
283-
void this.worker.startOnlyOnce({
284-
edgeFunctionName: this.edgeFunctionName,
285-
workerId,
286-
startMode: 'http',
287-
});
287+
const worker = createWorkerFn(this.loggingFactory.createLogger);
288+
this.worker = worker;
289+
290+
try {
291+
await worker.startOnlyOnce({
292+
edgeFunctionName: this.edgeFunctionName,
293+
workerId,
294+
startMode: 'http',
295+
});
296+
} catch (error) {
297+
if (this.worker === worker) {
298+
this.worker = null;
299+
}
300+
throw error;
301+
}
288302
}
289303

290304
/**

‎pkgs/edge-worker/src/platform/processDeps.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ export type ProcessSignal = 'SIGTERM' | 'SIGINT' | 'SIGQUIT';
33
export type ProcessDeps = {
44
env: Record<string, string | undefined>;
55
onSignal: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
6+
offSignal?: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
67
exit: (code: number) => never;
78
setExitCode: (code: number) => void;
89
randomUUID: () => string;
@@ -11,6 +12,7 @@ export type ProcessDeps = {
1112
type ProcessLike = {
1213
env?: Record<string, string | undefined>;
1314
on?: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
15+
off?: (signal: ProcessSignal, handler: () => void | Promise<void>) => void;
1416
exit?: (code: number) => never;
1517
exitCode?: number;
1618
};
@@ -23,13 +25,20 @@ export function getProcessDeps(): ProcessDeps {
2325
const processLike = (globalThis as { process?: ProcessLike }).process;
2426
const cryptoLike = (globalThis as { crypto?: CryptoLike }).crypto;
2527

26-
if (!processLike?.env || !processLike.on || !processLike.exit || !cryptoLike?.randomUUID) {
28+
if (
29+
!processLike?.env ||
30+
!processLike.on ||
31+
!processLike.off ||
32+
!processLike.exit ||
33+
!cryptoLike?.randomUUID
34+
) {
2735
throw new Error('Process runtime is not available');
2836
}
2937

3038
return {
3139
env: processLike.env,
3240
onSignal: (signal, handler) => processLike.on?.(signal, handler),
41+
offSignal: (signal, handler) => processLike.off?.(signal, handler),
3342
exit: (code) => processLike.exit!(code),
3443
setExitCode: (code) => {
3544
processLike.exitCode = code;

0 commit comments

Comments
 (0)