feat: add ProcessPlatformAdapter for Node.js process environments#627
feat: add ProcessPlatformAdapter for Node.js process environments#627jumski wants to merge 1 commit into
ProcessPlatformAdapter for Node.js process environments#627Conversation
|
|
Warning This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
This stack of pull requests is managed by Graphite. Learn more about stacking. |
|
View your CI Pipeline Execution ↗ for commit 5e9c0a9
💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗ ☁️ Nx Cloud last updated this comment at |
![graphite-app[bot]](https://avatars.githubusercontent.com/in/158384?s=60&v=4){kind=small}
| async stopWorker(): Promise<void> { | ||
| this.requestShutdown(); | ||
|
|
||
| try { | ||
| if (this.worker) { | ||
| await this.worker.stop(); | ||
| } | ||
| if (this.workerId) { | ||
| await this.queries.markWorkerStopped(this.workerId); | ||
| } | ||
| } finally { | ||
| if (this.ownsSql) { | ||
| await this._platformResources.sql.end(); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Race condition: stopWorker() can be called concurrently
If stopWorker() is called manually and then a signal arrives before it completes, handleSignal() will call stopWorker() again since shutdownStarted is only set in handleSignal(), not in stopWorker(). This causes:
worker.stop()called twice (may not be idempotent)markWorkerStopped()called twice (could fail or create duplicate entries)sql.end()potentially called twice (will error on second call)
Fix: Add a guard in stopWorker() or set the shutdown flag there as well:
async stopWorker(): Promise<void> {
if (this.shutdownStarted) {
return;
}
this.shutdownStarted = true;
this.requestShutdown();
// ... rest of implementation
}| async stopWorker(): Promise<void> { | |
| this.requestShutdown(); | |
| try { | |
| if (this.worker) { | |
| await this.worker.stop(); | |
| } | |
| if (this.workerId) { | |
| await this.queries.markWorkerStopped(this.workerId); | |
| } | |
| } finally { | |
| if (this.ownsSql) { | |
| await this._platformResources.sql.end(); | |
| } | |
| } | |
| } | |
| async stopWorker(): Promise<void> { | |
| if (this.shutdownStarted) { | |
| return; | |
| } | |
| this.shutdownStarted = true; | |
| this.requestShutdown(); | |
| try { | |
| if (this.worker) { | |
| await this.worker.stop(); | |
| } | |
| if (this.workerId) { | |
| await this.queries.markWorkerStopped(this.workerId); | |
| } | |
| } finally { | |
| if (this.ownsSql) { | |
| await this._platformResources.sql.end(); | |
| } | |
| } | |
| } | |
Spotted by Graphite
Is this helpful? React 👍 or 👎 to let us know.
jumski
force-pushed
the
portable-worker-process-adapter
branch
from
3cb8c81 to
efd6d99
Compare
jumski
force-pushed
the
portable-worker-start
branch
from
ca1dd9b to
8f4bb3e
Compare
jumski
force-pushed
the
portable-worker-process-adapter
branch
from
efd6d99 to
6b47ef1
Compare
jumski
force-pushed
the
portable-worker-process-adapter
branch
from
6b47ef1 to
5e9c0a9
Compare
jumski
force-pushed
the
portable-worker-start
branch
from
8f4bb3e to
30f8253
Compare
1 participant
Adds
ProcessPlatformAdapterto support running pgflow workers in Node.js (or anyprocess-based runtime), complementing the existingSupabasePlatformAdapterfor Deno/Edge environments.The adapter handles OS signal registration (
SIGTERM,SIGINT,SIGQUIT) for graceful shutdown, including double-signal force-exit behavior. It manages worker lifecycle by callingstartOnlyOncewithstartMode: 'process', marking the worker stopped in the database on shutdown, and conditionally closing the SQL connection only when the adapter owns it.A
processDepsabstraction is introduced to make the adapter fully testable without a real Node.js process, allowing signal handlers,process.exit, exit codes, andcrypto.randomUUIDto be injected in tests.DATABASE_URLis added as a connection resolution option, slotting in betweenconnectionStringandEDGE_WORKER_DB_URLin priority order.createAdapteris updated to detect a process environment and return aProcessPlatformAdapterautomatically.CurrentPlatformEnvis broadened fromSupabaseEnvtoRecord<string, string | undefined>to accommodate non-Supabase environments.