Skip to content

Commit 9e4a27e

Browse files
committed
fix(db): give billing and auth their own connection pools
postgres.js pipelines new queries onto busy connections when none is idle, so slow reads saturating the process pool made short usage-ledger transactions wait seconds for BEGIN until Postgres killed them. - Add `billing` (4) and `auth` (3) dbFor roles, with DATABASE_URL_BILLING and DATABASE_URL_AUTH overrides falling back to the process URL. - Billing pool: recordCumulativeUsage, recordUsage without a caller tx, the execution logger's usage reconciliation, both threshold settlement transactions, and recordJobRunSpend. Each moved transaction only queries through its tx handle; readers stay on the process pool. - Auth pool: getAuthPool() backs the Better Auth adapter, getAuthDatabase() outside a transaction, the OAuth adapter guard defaults, and the API-key hash lookup. - isPrimaryPoolClient() in @sim/db replaces the `executor === db` checks in the security, session, and SSO policy caches, so reads through the auth pool keep hitting them instead of costing two queries per cached session. - Integration test saturates the process pool and requires the billing write, an adapter session read, and an API-key lookup to finish anyway.
1 parent 2fc2008 commit 9e4a27e

23 files changed

Lines changed: 287 additions & 37 deletions

‎apps/sim/app/api/billing/update-cost/replay.live.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ vi.mock('@sim/db', async () => {
4747
})
4848
state.client = client
4949
const db = drizzle(client)
50-
return { db, dbReplica: db }
50+
return { db, dbReplica: db, dbFor: () => db }
5151
})
5252

5353
/** Subscription and payment-provider fixtures; callback, ledger and replay code are real. */

‎apps/sim/app/api/billing/update-cost/worker.live.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ vi.mock('@sim/db', async () => {
4747
})
4848
state.client = client
4949
const db = drizzle(client)
50-
return { db, dbReplica: db }
50+
return { db, dbReplica: db, dbFor: () => db }
5151
})
5252

5353
/** Subscription and payment-provider fixtures; callback, ledger and replay code are real. */

‎apps/sim/lib/api-key/service.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { apiKey as apiKeyTable, user as userTable } from '@sim/db/schema'
33
import { createLogger, setRequestAuth } from '@sim/logger'
44
import { and, eq, isNull, lt, or } from 'drizzle-orm'
55
import { hashApiKey } from '@/lib/api-key/crypto'
6+
import { getAuthPool } from '@/lib/auth/database-context'
67
import { createDetachedTouch } from '@/lib/core/utils/background'
78
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
89
import { getWorkspaceBillingSettings, type WorkspaceBillingSettings } from '@/lib/workspaces/utils'
@@ -62,7 +63,7 @@ export async function authenticateApiKeyFromHeader(
6263
}
6364

6465
const keyHash = hashApiKey(apiKeyHeader)
65-
const rows: HashCandidate[] = await db
66+
const rows: HashCandidate[] = await getAuthPool()
6667
.select({
6768
id: apiKeyTable.id,
6869
userId: apiKeyTable.userId,

‎apps/sim/lib/auth/database-context.ts‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,21 @@
11
import { AsyncLocalStorage } from 'node:async_hooks'
2-
import { db } from '@sim/db'
2+
import { type db, dbFor } from '@sim/db'
33
import type { DbOrTx } from '@/lib/db/types'
44

55
const authDatabase = new AsyncLocalStorage<DbOrTx>()
66

7-
/** The Drizzle executor backing the current Better Auth transaction, or the root database. */
7+
/**
8+
* The auth pool's root client. Session, adapter, and API-key lookups gate every
9+
* authenticated request, so they keep their own connections when slow work
10+
* saturates the process pool.
11+
*/
12+
export function getAuthPool(): typeof db {
13+
return dbFor('auth')
14+
}
15+
16+
/** The Drizzle executor backing the current Better Auth transaction, or the auth pool. */
817
export function getAuthDatabase(): DbOrTx {
9-
return authDatabase.getStore() ?? db
18+
return authDatabase.getStore() ?? getAuthPool()
1019
}
1120

1221
/** Keeps database hooks on the adapter's connection so they can see uncommitted auth records. */

‎apps/sim/lib/auth/oauth-provider-adapter-guard.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
import { AsyncLocalStorage } from 'node:async_hooks'
2-
import { db } from '@sim/db'
2+
import type { db } from '@sim/db'
33
import { oauthConsent, oauthTokenFamily } from '@sim/db/schema'
44
import { generateId } from '@sim/utils/id'
55
import type { drizzleAdapter } from 'better-auth/adapters/drizzle'
66
import { APIError } from 'better-auth/api'
77
import { inArray } from 'drizzle-orm'
8+
import { getAuthPool } from '@/lib/auth/database-context'
89
import { getOAuthIssuedResource } from '@/lib/auth/oauth-resource'
910
import { capabilityRefusal } from '@/lib/permission-groups/capabilities'
1011
import { isCapabilityWithheldForUser } from '@/lib/permission-groups/user-scope.server'
@@ -37,7 +38,7 @@ async function deleteTrackedFamilies(
3738
*/
3839
export async function withOAuthProviderIssuanceCompensation<T extends Response>(
3940
work: () => Promise<T>,
40-
database: AuthDatabase = db
41+
database: AuthDatabase = getAuthPool()
4142
): Promise<T> {
4243
const familyIds = new Set<string>()
4344
try {
@@ -87,7 +88,7 @@ function requireConsentInsert(data: Record<string, unknown>): OAuthConsentInsert
8788
*/
8889
export function guardOAuthProviderWrites(
8990
adapter: BetterAuthAdapter,
90-
database: AuthDatabase = db
91+
database: AuthDatabase = getAuthPool()
9192
): BetterAuthAdapter {
9293
return {
9394
...adapter,

‎apps/sim/lib/auth/security-policy.test.ts‎

Lines changed: 36 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { db } from '@sim/db'
2-
import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
3-
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
2+
import { databaseMockFns, dbChainMockFns, resetDbChainMock } from '@sim/testing'
3+
import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
44
import {
55
getMemberOrganizationId,
66
getSecurityPolicyVersion,
@@ -146,3 +146,37 @@ describe('getSessionCookieCacheVersion', () => {
146146
expect(limit).toHaveBeenCalledTimes(2)
147147
})
148148
})
149+
150+
describe('shared caches behind a sub-pool client', () => {
151+
beforeEach(() => {
152+
resetDbChainMock()
153+
invalidateMembershipCache('pool-user')
154+
invalidateSecurityPolicyVersionCache('pool-org')
155+
})
156+
157+
afterEach(() => {
158+
databaseMockFns.mockIsPrimaryPoolClient.mockReset()
159+
})
160+
161+
/**
162+
* Better Auth reads the cookie-cache version on every cached-session request through the auth
163+
* pool, not `db`; caching only reads made through `db` would cost two queries per request.
164+
*/
165+
it('serves a repeat cookie-cache version from cache when read through the auth pool', async () => {
166+
const authPool: typeof db = Object.create(db)
167+
databaseMockFns.mockIsPrimaryPoolClient.mockImplementation(
168+
(executor) => executor === db || executor === authPool
169+
)
170+
dbChainMockFns.limit
171+
.mockResolvedValueOnce([{ organizationId: 'pool-org' }])
172+
.mockResolvedValueOnce([{ version: 3 }])
173+
174+
await expect(getSessionCookieCacheVersion({ userId: 'pool-user' }, authPool)).resolves.toBe(
175+
'pool-org:3'
176+
)
177+
await expect(getSessionCookieCacheVersion({ userId: 'pool-user' }, authPool)).resolves.toBe(
178+
'pool-org:3'
179+
)
180+
expect(dbChainMockFns.limit).toHaveBeenCalledTimes(2)
181+
})
182+
})

‎apps/sim/lib/auth/security-policy.ts‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { db } from '@sim/db'
1+
import { db, isPrimaryPoolClient } from '@sim/db'
22
import { member, organization } from '@sim/db/schema'
33
import { createLogger } from '@sim/logger'
44
import { eq } from 'drizzle-orm'
@@ -50,7 +50,7 @@ export async function getSecurityPolicyVersion(
5050
): Promise<number> {
5151
if (!organizationId) return DEFAULT_VERSION
5252

53-
const cached = executor === db ? versionCache.get(organizationId) : undefined
53+
const cached = isPrimaryPoolClient(executor) ? versionCache.get(organizationId) : undefined
5454
if (cached !== undefined) return cached
5555

5656
try {
@@ -61,7 +61,7 @@ export async function getSecurityPolicyVersion(
6161
.limit(1)
6262

6363
const version = row?.version ?? DEFAULT_VERSION
64-
if (executor === db) versionCache.set(organizationId, version)
64+
if (isPrimaryPoolClient(executor)) versionCache.set(organizationId, version)
6565
return version
6666
} catch (error) {
6767
logger.error('Failed to resolve security policy version; using default', {
@@ -133,7 +133,7 @@ export async function getMemberOrganizationId(
133133
if (!userId) return null
134134

135135
/** A transaction may have just created or moved the membership; shared cache entries cannot see it. */
136-
const cached = executor === db ? membershipCache.get(userId) : undefined
136+
const cached = isPrimaryPoolClient(executor) ? membershipCache.get(userId) : undefined
137137
if (cached) return cached.organizationId
138138

139139
try {
@@ -144,7 +144,7 @@ export async function getMemberOrganizationId(
144144
.limit(1)
145145

146146
const organizationId = row?.organizationId ?? null
147-
if (executor === db) {
147+
if (isPrimaryPoolClient(executor)) {
148148
membershipCache.set(userId, { organizationId }, { ttl: membershipCacheTtlMs(organizationId) })
149149
}
150150
return organizationId

‎apps/sim/lib/auth/session-policy.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { db } from '@sim/db'
1+
import { db, isPrimaryPoolClient } from '@sim/db'
22
import type { SessionPolicySettings } from '@sim/db/schema'
33
import { organization } from '@sim/db/schema'
44
import { createLogger } from '@sim/logger'
@@ -57,7 +57,7 @@ export async function getSessionPolicy(
5757
if (!organizationId) return NO_POLICY
5858

5959
/** Uncommitted policy reads must neither consume nor populate the shared cache. */
60-
const cached = executor === db ? policyCache.get(organizationId) : undefined
60+
const cached = isPrimaryPoolClient(executor) ? policyCache.get(organizationId) : undefined
6161
if (cached) return cached
6262

6363
try {
@@ -78,7 +78,7 @@ export async function getSessionPolicy(
7878
idleTimeoutHours: settings.idleTimeoutHours ?? null,
7979
}
8080
: NO_POLICY
81-
if (executor === db) policyCache.set(organizationId, policy)
81+
if (isPrimaryPoolClient(executor)) policyCache.set(organizationId, policy)
8282
return policy
8383
} catch (error) {
8484
logger.error('Failed to resolve session policy; applying no policy', {

‎apps/sim/lib/auth/sim-auth-adapter.ts‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,9 @@
1-
import { db } from '@sim/db'
21
import * as schema from '@sim/db/schema'
32
import type { BetterAuthOptions } from 'better-auth'
43
import { drizzleAdapter } from 'better-auth/adapters/drizzle'
54
import { APIError } from 'better-auth/api'
65
import { and, eq } from 'drizzle-orm'
7-
import { runWithAuthDatabase } from '@/lib/auth/database-context'
6+
import { getAuthPool, runWithAuthDatabase } from '@/lib/auth/database-context'
87
import {
98
type AuthDatabase,
109
guardOAuthProviderWrites,
@@ -66,7 +65,7 @@ function createTransactionAdapter(
6665
*/
6766
export function createSimAuthAdapter(
6867
options: BetterAuthOptions,
69-
database: AuthDatabase = db
68+
database: AuthDatabase = getAuthPool()
7069
): BetterAuthAdapter {
7170
const guarded = createGuardedAdapter(options, database)
7271
const transaction: BetterAuthAdapter['transaction'] = (callback) =>

‎apps/sim/lib/auth/sso-policy.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { db } from '@sim/db'
1+
import { db, isPrimaryPoolClient } from '@sim/db'
22
import { organization } from '@sim/db/schema'
33
import { createLogger } from '@sim/logger'
44
import { APIError } from 'better-auth/api'
@@ -43,7 +43,7 @@ export async function isSsoRequiredForOrganization(
4343
if (!organizationId) return false
4444

4545
/** Uncommitted reads must neither consume nor populate the shared cache. */
46-
const cached = executor === db ? requirementCache.get(organizationId) : undefined
46+
const cached = isPrimaryPoolClient(executor) ? requirementCache.get(organizationId) : undefined
4747
if (cached !== undefined) return cached
4848

4949
const [row] = await executor
@@ -64,7 +64,7 @@ export async function isSsoRequiredForOrganization(
6464
onError: 'throw',
6565
})) &&
6666
(await hasSignInCapableSsoProvider(organizationId, executor))
67-
if (executor === db) requirementCache.set(organizationId, required)
67+
if (isPrimaryPoolClient(executor)) requirementCache.set(organizationId, required)
6868
return required
6969
}
7070

0 commit comments

Comments
 (0)