-
-
Notifications
You must be signed in to change notification settings - Fork 288
fix(database): isolate web and worker connection pools #764
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
GET / was throwing KnexTimeoutError ("Timeout acquiring a connection")
in production. Root cause was the connection pool, not the database:
the single process (SvelteKit + cron scheduler + BullMQ workers) shared
one pool capped at 10, while one GET / fans out ~6 queries. A couple of
concurrent page loads, or a per-minute monitor burst overlapping a load,
exceeded 10 and queued acquires blew past the 15s timeout. Postgres
itself had 97 free slots the whole time and no leak.
Split into two pools so background work can't starve page loads:
- web pool (DATABASE_POOL_MAX, default 10) serves HTTP requests
- worker pool (DATABASE_WORKER_POOL_MAX, default 5) serves background jobs
Routing is by execution context via AsyncLocalStorage: q.createWorker
(the single chokepoint all workers/schedulers flow through) runs each
processor inside a worker-pool context, and BaseRepository.knex resolves
the pool from that context, defaulting to the web pool. This keeps shared
controllers correct whether they run in a request or a job. SQLite has no
real pool and reuses a single connection, so the split is a no-op there.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>- Loading branch information
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| import DbImpl from "./dbimpl"; | ||
| import knexOb from "../../../../knexfile.js"; | ||
| import knexOb, { workerKnexOb } from "../../../../knexfile.js"; | ||
|
|
||
| const instance: DbImpl = new DbImpl(knexOb); | ||
| const instance: DbImpl = new DbImpl(knexOb, workerKnexOb); | ||
| export default instance; |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| import { AsyncLocalStorage } from "node:async_hooks"; | ||
| import type { Knex as KnexType } from "knex"; | ||
|
|
||
| // Per-execution-context selection of the database connection pool. | ||
| // | ||
| // Kener runs SvelteKit requests, the cron scheduler, and the BullMQ workers in | ||
| // a single process, all sharing one Knex instance. A burst of background jobs | ||
| // could therefore exhaust the connection pool and time out user-facing page | ||
|
Comment on lines
+6
to
+8
|
||
| // loads (KnexTimeoutError on acquire). To prevent that, background work runs | ||
| // against a dedicated worker pool: queues/q.ts wraps every job processor in | ||
| // runWithWorkerKnex(), and BaseRepository reads getWorkerKnex() so its queries | ||
| // route to that pool. Anything outside a job (requests, startup, migrations) | ||
| // has no store set and falls back to the web pool. | ||
| // | ||
| // See knexfile.ts for pool sizing and docs .../setup/database-setup.md. | ||
| const workerKnexStorage = new AsyncLocalStorage<KnexType>(); | ||
|
|
||
| /** Runs `fn` with all repository queries routed to the worker pool `knex`. */ | ||
| export function runWithWorkerKnex<T>(knex: KnexType, fn: () => Promise<T>): Promise<T> { | ||
| return workerKnexStorage.run(knex, fn); | ||
| } | ||
|
|
||
| /** The worker pool for the current context, or undefined when not in a job. */ | ||
| export function getWorkerKnex(): KnexType | undefined { | ||
| return workerKnexStorage.getStore(); | ||
| } | ||
| Original file line number | Diff line number | Diff line change | ||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -1,4 +1,5 @@ | ||||||||||||||
| import { redisIOConnection } from "../redisConnector.js"; | ||||||||||||||
| import db from "../db/db.js"; | ||||||||||||||
| import { | ||||||||||||||
| Queue, | ||||||||||||||
| Worker, | ||||||||||||||
|
|
@@ -40,7 +41,15 @@ export const createWorker = <T = unknown, R = unknown>( | |||||||||||||
| concurrency: 5, | ||||||||||||||
| ...options, | ||||||||||||||
| }; | ||||||||||||||
| return new Worker<T, R>(queue.name, processor, opts); | ||||||||||||||
| // Route every job's database access to the worker pool. This is the single | ||||||||||||||
| // chokepoint all BullMQ workers and schedulers flow through, so wrapping here | ||||||||||||||
| // isolates background work from the web request pool (see db/poolContext.ts). | ||||||||||||||
| // Sandboxed (string/URL) processors run out-of-process and pass through. | ||||||||||||||
| const wrapped: Processor<T, R> = | ||||||||||||||
| typeof processor === "function" | ||||||||||||||
| ? (job, token) => db.runInWorkerContext(() => Promise.resolve(processor(job, token))) | ||||||||||||||
| : processor; | ||||||||||||||
|
Comment on lines
+48
to
+51
Comment on lines
+49
to
+51
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time! |
||||||||||||||
| return new Worker<T, R>(queue.name, wrapped, opts); | ||||||||||||||
| }; | ||||||||||||||
|
|
||||||||||||||
| export default { | ||||||||||||||
|
|
||||||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ensure the worker pool is destroyed even if web pool cleanup fails.
Line 870 awaits the web pool destroy before starting worker cleanup, so a rejection skips Line 872 and can leave the new worker pool open during shutdown/test cleanup.
Proposed fix
async close(): Promise<void> { - await this.knex.destroy(); - if (this.workerKnex !== this.knex) { - await this.workerKnex.destroy(); - } + const workerKnex = this.workerKnex !== this.knex ? this.workerKnex : undefined; + try { + await this.knex.destroy(); + } finally { + if (workerKnex) { + await workerKnex.destroy(); + } + } }🤖 Prompt for AI Agents