24400472712

Committed 14 Apr 2026 01:03PM UTC coverage: 82.476% (+0.06%) from 82.421%

Build # 24400472712

Build Type

Pull #1018

github

Committed by

web-flow

Commit Message

Merge 7bf796127 into 9fc93ca96

Pull Request Pull Request #1018: feat: add outcome logging for tenant pools

Coverage Stats

3678 of 7447 branches covered (49.39%)

Branch coverage included in aggregate %.

43 of 45 new or added lines in 2 files covered. (95.56%)

9 existing lines in 1 file now uncovered.

30462 of 33947 relevant lines covered (89.73%)

347.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.03

/src/internal/database/pool.ts

import { type CacheLookupOutcome, createTtlCache, TENANT_POOL_CACHE_NAME } from '@internal/cache'
import { wait } from '@internal/concurrency'
import { getSslSettings } from '@internal/database/ssl'
import { logger, logSchema } from '@internal/monitoring'
import {
  cacheEvictionsTotal,
  cacheRequestsTotal,
  dbActiveConnection,
  dbActivePool,
  dbInUseConnection,
  isMetricEnabled,
  meter,
} from '@internal/monitoring/metrics'
import { JWTPayload } from 'jose'
import { Knex, knex } from 'knex'
import { getConfig } from '../../config'

const {
  isMultitenant,
  databaseSSLRootCert,
  databaseMaxConnections,
  databaseFreePoolAfterInactivity,
  databaseConnectionTimeout,
  dbSearchPath,
  dbPostgresVersion,
  databaseApplicationName,
  tenantPoolCacheHitLogSampleRate,
  tenantPoolCacheMissLogSampleRate,
} = getConfig()

export const TENANT_POOL_CACHE_LOOKUP_LOG_TYPE = 'cache'
export const TENANT_POOL_CACHE_LOOKUP_LOG_MESSAGE = '[Cache] Tenant pool lookup'

export interface TenantConnectionOptions {
  user: User
  superUser: User

  tenantId: string
  dbUrl: string
  isExternalPool?: boolean
  isSingleUse?: boolean
  idleTimeoutMillis?: number
  reapIntervalMillis?: number
  maxConnections: number
  clusterSize?: number
  numWorkers?: number
  headers?: Record<string, string | undefined | string[]>
  method?: string
  path?: string
  operation?: () => string | undefined
}

export interface User {
  jwt: string
  payload: { role?: string } & JWTPayload
}

export interface PoolStats {
  used: number
  total: number
}

export interface PoolStrategy {
  acquire(): Knex
  rebalance(options: { clusterSize: number }): void
  destroy(): Promise<void>
  getPoolStats(): PoolStats | null
}

export const searchPath = ['storage', 'public', 'extensions', ...dbSearchPath.split(',')].filter(
  Boolean
)

const multiTenantTtlConfig = {
  ttl: 1000 * 10,
  updateAgeOnGet: true,
  checkAgeOnGet: true,
}

const manuallyDestroyedPools = new WeakSet<PoolStrategy>()

function logPoolDestroyError(error: unknown): void {
  logSchema.error(logger, 'pool was not able to be destroyed', {
    type: 'db',
    error,
  })
}

async function destroyPool(pool: PoolStrategy): Promise<void> {
  await pool.destroy()
}

async function destroyPoolSafely(pool: PoolStrategy): Promise<void> {
  try {
    await destroyPool(pool)
  } catch (e) {
    logPoolDestroyError(e)
  }
}

function recordTenantPoolCacheEviction(reason: string): void {
  // Explicit destroy paths are filtered before this helper is called.
  if (reason === 'stale' || reason === 'evict' || reason === 'delete') {
    cacheEvictionsTotal.add(1, {
      cache: TENANT_POOL_CACHE_NAME,
    })
  }
}

function recordTenantPoolCacheRequest(outcome: string): void {
  cacheRequestsTotal.add(1, {
    cache: TENANT_POOL_CACHE_NAME,
    outcome,
  })
}

function shouldLogTenantPoolCacheLookup(sampleRate: number): boolean {
  return sampleRate >= 1 || (sampleRate > 0 && Math.random() < sampleRate)
}

function logTenantPoolCacheLookup(
  settings: TenantConnectionOptions,
  isCacheable: boolean,
  outcome: CacheLookupOutcome
): void {
  const sampleRate =
    outcome === 'hit' ? tenantPoolCacheHitLogSampleRate : tenantPoolCacheMissLogSampleRate

  if (!shouldLogTenantPoolCacheLookup(sampleRate)) {
    return
  }

  logger.info(
    {
      type: TENANT_POOL_CACHE_LOOKUP_LOG_TYPE,
      cache: TENANT_POOL_CACHE_NAME,
      tenantId: settings.tenantId,
      project: settings.tenantId,
      outcome,
      sampleRate,
      sampleWeight: 1 / sampleRate,
      isCacheable,
      isExternalPool: Boolean(settings.isExternalPool),
      isSingleUse: Boolean(settings.isSingleUse),
    },
    TENANT_POOL_CACHE_LOOKUP_LOG_MESSAGE
  )
}

const tenantPools = createTtlCache<string, PoolStrategy>({
  ...(isMultitenant ? multiTenantTtlConfig : { max: 1, ttl: Infinity }),
  dispose: async (pool, _tenantId, reason) => {
    if (!pool || manuallyDestroyedPools.has(pool)) {
      return
    }

    recordTenantPoolCacheEviction(reason)

    await destroyPoolSafely(pool)
  },
})

// ============================================================================
// Pool stats collection — chunked to avoid blocking the event loop
// ============================================================================
interface PoolStatsSnapshot {
  poolCount: number
  totalConnections: number
  totalInUse: number
}

const STATS_CHUNK_SIZE = 100
const STATS_INTERVAL_MS = 5_000

let cachedPoolStats: PoolStatsSnapshot = {
  poolCount: 0,
  totalConnections: 0,
  totalInUse: 0,
}
let collectInProgress = false

async function collectPoolStats() {
  if (collectInProgress) return
  collectInProgress = true

  try {
    let poolCount = 0
    let totalConnections = 0
    let totalInUse = 0
    let chunkCount = 0

    for (const [, pool] of tenantPools.entries()) {
      poolCount++
      const stats = pool.getPoolStats()
      if (stats) {
        totalConnections += stats.total
        totalInUse += stats.used
      }
      // Yield to the event loop between chunks
      if (++chunkCount % STATS_CHUNK_SIZE === 0) {
        await new Promise<void>((resolve) => setImmediate(resolve))
      }
    }

    cachedPoolStats = {
      poolCount,
      totalConnections,
      totalInUse,
    }
  } finally {
    collectInProgress = false
  }
}

/**
 * PoolManager is a class that manages a pool of Knex connections.
 * It creates a new pool for each tenant and reuses existing pools.
 */
export class PoolManager {
  protected numWorkers: number = 1

  setNumWorkers(numWorkers: number) {
    this.numWorkers = Math.max(numWorkers ?? 1, 1)
  }

  monitor() {
    // Periodically collect stats in a non-blocking way
    const interval = setInterval(() => {
      void collectPoolStats()
    }, STATS_INTERVAL_MS)
    interval.unref()

    // Observable callback reads the cached snapshot — O(1)
    meter.addBatchObservableCallback(
      (observer) => {
        if (isMetricEnabled('db_active_local_pools')) {
          observer.observe(dbActivePool, cachedPoolStats.poolCount)
        }
        if (isMetricEnabled('db_connections')) {
          observer.observe(dbActiveConnection, cachedPoolStats.totalConnections)
        }
        if (isMetricEnabled('db_connections_in_use')) {
          observer.observe(dbInUseConnection, cachedPoolStats.totalInUse)
        }
      },
      [dbActivePool, dbActiveConnection, dbInUseConnection]
    )
  }

  rebalanceAll(data: { clusterSize: number }) {
    for (const pool of tenantPools.values()) {
      pool.rebalance({
        clusterSize: data.clusterSize,
      })
    }
  }

  rebalance(tenantId: string, data: { clusterSize: number }) {
    const pool = tenantPools.get(tenantId)
    if (pool) {
      pool.rebalance({
        clusterSize: data.clusterSize,
      })
    }
  }

  getPool(settings: TenantConnectionOptions) {
    const isCacheable = (settings.isSingleUse && !settings.isExternalPool) || !settings.isSingleUse
    const { value: existingPool, outcome } = tenantPools.getWithOutcome(settings.tenantId)

    if (existingPool) {
      recordTenantPoolCacheRequest(outcome)
      logTenantPoolCacheLookup(settings, isCacheable, outcome)

      return existingPool
    }

    if (!isCacheable) {
      return this.newPool({ ...settings, numWorkers: this.numWorkers })
    }

    recordTenantPoolCacheRequest(outcome)
    logTenantPoolCacheLookup(settings, isCacheable, outcome)

    const newPool = this.newPool({ ...settings, numWorkers: this.numWorkers })

    tenantPools.set(settings.tenantId, newPool)
    return newPool
  }

  destroy(tenantId: string) {
    const pool = tenantPools.get(tenantId)
    if (pool) {
      manuallyDestroyedPools.add(pool)
      tenantPools.delete(tenantId)
      return destroyPool(pool).finally(() => {
        manuallyDestroyedPools.delete(pool)
      })
    }
    return Promise.resolve()
  }

  destroyAll() {
    const promises: Promise<void>[] = []

    for (const [connectionString, pool] of tenantPools) {
      manuallyDestroyedPools.add(pool)
      tenantPools.delete(connectionString)
      promises.push(
        destroyPool(pool).finally(() => {
          manuallyDestroyedPools.delete(pool)
        })
      )
    }
    return Promise.allSettled(promises)
  }

  protected newPool(settings: TenantConnectionOptions): PoolStrategy {
    return new TenantPool(settings)
  }
}

/**
 * TenantPool create a new Knex pool for each tenant, with rebalance
 * functionality to adjust the number of connections based on the cluster size.
 */
class TenantPool implements PoolStrategy {
  protected pool?: Knex

  constructor(protected readonly options: TenantConnectionOptions) {}

  acquire() {
    if (this.pool) {
      return this.pool
    }

    this.pool = this.createKnexPool()
    return this.pool
  }

  destroy(): Promise<void> {
    const originalPool = this.pool

    if (!originalPool) {
      return Promise.resolve()
    }

    this.pool = undefined
    return this.drainPool(originalPool)
  }

  getPoolStats(): PoolStats | null {
    const tarnPool = this.pool?.client?.pool
    if (!tarnPool) return null
    return {
      used: tarnPool.numUsed(),
      total: tarnPool.numUsed() + tarnPool.numFree(),
    }
  }

  getSettings() {
    const isSingleUseExternalPool = this.options.isSingleUse && this.options.isExternalPool

    const numWorkers = Math.max(this.options.numWorkers ?? 1, 1)
    const clusterSize = this.options.clusterSize || 0
    let maxConnection = this.options.maxConnections || databaseMaxConnections

    const divisor = Math.max(clusterSize, 1) * numWorkers
    if (divisor > 1) {
      maxConnection = Math.ceil(maxConnection / divisor) || 1
    }

    if (isSingleUseExternalPool) {
      maxConnection = 1
    }

    return {
      ...this.options,
      searchPath: this.options.isExternalPool ? undefined : searchPath,
      idleTimeoutMillis: isSingleUseExternalPool ? 100 : databaseFreePoolAfterInactivity,
      reapIntervalMillis: isSingleUseExternalPool ? 50 : undefined,
      maxConnections: maxConnection,
    }
  }

  rebalance(options: { clusterSize: number }) {
    if (options.clusterSize === 0) {
      return
    }

    const originalPool = this.pool

    this.options.clusterSize = options.clusterSize
    this.pool = undefined

    if (originalPool) {
      this.drainPool(originalPool).catch((e) => {
        logger.error({ type: 'pool', error: e })
      })
    }
  }

  protected async drainPool(pool: Knex) {
    for (; pool?.client?.pool; ) {
      let waiting = 0
      waiting += pool.client.pool.numPendingAcquires()
      waiting += pool.client.pool.numPendingValidations()
      waiting += pool.client.pool.numPendingCreates()

      if (waiting === 0) {
        break
      }

      await wait(200)
    }

    return pool.destroy()
  }

  protected createKnexPool() {
    const settings = this.getSettings()
    const sslSettings = getSslSettings({
      connectionString: settings.dbUrl,
      databaseSSLRootCert,
    })

    const maxConnections = settings.maxConnections

    return knex({
      client: 'pg',
      version: dbPostgresVersion,
      searchPath: settings.searchPath,
      pool: {
        min: 0,
        max: maxConnections,
        acquireTimeoutMillis: databaseConnectionTimeout,
        idleTimeoutMillis: settings.idleTimeoutMillis,
        reapIntervalMillis: 1000,
      },
      connection: {
        connectionString: settings.dbUrl,
        connectionTimeoutMillis: databaseConnectionTimeout,
        ssl: sslSettings ? { ...sslSettings } : undefined,
        application_name: databaseApplicationName,
      },
      acquireConnectionTimeout: databaseConnectionTimeout,
    })
  }
}

1	import { type CacheLookupOutcome, createTtlCache, TENANT_POOL_CACHE_NAME } from '@internal/cache'	1✔
2	import { wait } from '@internal/concurrency'	1✔
3	import { getSslSettings } from '@internal/database/ssl'	1✔
4	import { logger, logSchema } from '@internal/monitoring'	1✔
5	import {	1✔
6	cacheEvictionsTotal,	1✔
7	cacheRequestsTotal,	1✔
8	dbActiveConnection,	1✔
9	dbActivePool,	1✔
10	dbInUseConnection,	1✔
11	isMetricEnabled,	1✔
12	meter,	1✔
13	} from '@internal/monitoring/metrics'	1✔
14	import { JWTPayload } from 'jose'	1✔
15	import { Knex, knex } from 'knex'	1✔
16	import { getConfig } from '../../config'	1✔
17		1✔
18	const {	1✔
19	isMultitenant,	1✔
20	databaseSSLRootCert,	1✔
21	databaseMaxConnections,	1✔
22	databaseFreePoolAfterInactivity,	1✔
23	databaseConnectionTimeout,	1✔
24	dbSearchPath,	1✔
25	dbPostgresVersion,	1✔
26	databaseApplicationName,	1✔
27	tenantPoolCacheHitLogSampleRate,	1✔
28	tenantPoolCacheMissLogSampleRate,	18✔
29	} = getConfig()	1✔
30		1✔
31	export const TENANT_POOL_CACHE_LOOKUP_LOG_TYPE = 'cache'	18✔
32	export const TENANT_POOL_CACHE_LOOKUP_LOG_MESSAGE = '[Cache] Tenant pool lookup'	18✔
33		1✔
34	export interface TenantConnectionOptions {	1✔
35	user: User	1✔
36	superUser: User	1✔
37		1✔
38	tenantId: string	1✔
39	dbUrl: string	1✔
40	isExternalPool?: boolean	1✔
41	isSingleUse?: boolean	1✔
42	idleTimeoutMillis?: number	1✔
43	reapIntervalMillis?: number	1✔
44	maxConnections: number	1✔
45	clusterSize?: number	1✔
46	numWorkers?: number	1✔
47	headers?: Record<string, string \| undefined \| string[]>	1✔
48	method?: string	1✔
49	path?: string	1✔
50	operation?: () => string \| undefined	1✔
51	}	1✔
52		1✔
53	export interface User {	1✔
54	jwt: string	1✔
55	payload: { role?: string } & JWTPayload	1✔
56	}	1✔
57		1✔
58	export interface PoolStats {	1✔
59	used: number	1✔
60	total: number	1✔
61	}	1✔
62		1✔
63	export interface PoolStrategy {	1✔
64	acquire(): Knex	1✔
65	rebalance(options: { clusterSize: number }): void	1✔
66	destroy(): Promise<void>	1✔
67	getPoolStats(): PoolStats \| null	1✔
68	}	1✔
69		1✔
70	export const searchPath = ['storage', 'public', 'extensions', ...dbSearchPath.split(',')].filter(	18✔
71	Boolean	1✔
72	)	1✔
73		1✔
74	const multiTenantTtlConfig = {	18✔
75	ttl: 1000 * 10,	1✔
76	updateAgeOnGet: true,	1✔
77	checkAgeOnGet: true,	1✔
78	}	1✔
79		1✔
80	const manuallyDestroyedPools = new WeakSet<PoolStrategy>()	18✔
81		1✔
82	function logPoolDestroyError(error: unknown): void {
83	logSchema.error(logger, 'pool was not able to be destroyed', {	×
84	type: 'db',
85	error,
86	})
87	}
88		1✔
89	async function destroyPool(pool: PoolStrategy): Promise<void> {
90	await pool.destroy()	21✔
91	}
92		1✔
93	async function destroyPoolSafely(pool: PoolStrategy): Promise<void> {
94	try {	6✔
95	await destroyPool(pool)	6✔
96	} catch (e) {
97	logPoolDestroyError(e)	×
98	}
99	}
100		1✔
101	function recordTenantPoolCacheEviction(reason: string): void {
102	// Explicit destroy paths are filtered before this helper is called.
103	if (reason === 'stale' \|\| reason === 'evict' \|\| reason === 'delete') {	6!
104	cacheEvictionsTotal.add(1, {	6✔
105	cache: TENANT_POOL_CACHE_NAME,
106	})
107	}
108	}
109		1✔
110	function recordTenantPoolCacheRequest(outcome: string): void {	4✔
111	cacheRequestsTotal.add(1, {	32✔
112	cache: TENANT_POOL_CACHE_NAME,	4✔
113	outcome,	4✔
114	})	4✔
115	}	4✔
116		1✔
117	function shouldLogTenantPoolCacheLookup(sampleRate: number): boolean {	4✔
118	return sampleRate >= 1 \|\| (sampleRate > 0 && Math.random() < sampleRate)	32!
119	}	4✔
120		1✔
121	function logTenantPoolCacheLookup(	4✔
122	settings: TenantConnectionOptions,	4✔
123	isCacheable: boolean,	4✔
124	outcome: CacheLookupOutcome	4✔
125	): void {	4✔
126	const sampleRate =	4✔
127	outcome === 'hit' ? tenantPoolCacheHitLogSampleRate : tenantPoolCacheMissLogSampleRate	32✔
128		4✔
129	if (!shouldLogTenantPoolCacheLookup(sampleRate)) {	32✔
130	return	30✔
131	}	4✔
132		×
133	logger.info(	2✔
134	{
135	type: TENANT_POOL_CACHE_LOOKUP_LOG_TYPE,
136	cache: TENANT_POOL_CACHE_NAME,
137	tenantId: settings.tenantId,
138	project: settings.tenantId,
139	outcome,
140	sampleRate,
141	sampleWeight: 1 / sampleRate,
142	isCacheable,
143	isExternalPool: Boolean(settings.isExternalPool),
144	isSingleUse: Boolean(settings.isSingleUse),
145	},
146	TENANT_POOL_CACHE_LOOKUP_LOG_MESSAGE
147	)
148	}
149		1✔
150	const tenantPools = createTtlCache<string, PoolStrategy>({	18✔
151	...(isMultitenant ? multiTenantTtlConfig : { max: 1, ttl: Infinity }),	1✔
152	dispose: async (pool, _tenantId, reason) => {	1✔
153	if (!pool \|\| manuallyDestroyedPools.has(pool)) {	21✔
154	return	15✔
155	}
156
157	recordTenantPoolCacheEviction(reason)	6✔
158
159	await destroyPoolSafely(pool)	6✔
160	},	1✔
161	})	1✔
162		1✔
163	// ============================================================================	1✔
164	// Pool stats collection — chunked to avoid blocking the event loop	1✔
165	// ============================================================================	1✔
166	interface PoolStatsSnapshot {	1✔
167	poolCount: number	1✔
168	totalConnections: number	1✔
169	totalInUse: number	1✔
170	}	1✔
171		1✔
172	const STATS_CHUNK_SIZE = 100	18✔
173	const STATS_INTERVAL_MS = 5_000	18✔
174		1✔
175	let cachedPoolStats: PoolStatsSnapshot = {	18✔
176	poolCount: 0,	1✔
177	totalConnections: 0,	1✔
178	totalInUse: 0,	1✔
179	}	1✔
180	let collectInProgress = false	18✔
181		1✔
182	async function collectPoolStats() {
183	if (collectInProgress) return	5!
184	collectInProgress = true	5✔
185
186	try {	5✔
187	let poolCount = 0	5✔
188	let totalConnections = 0	5✔
189	let totalInUse = 0	5✔
190	let chunkCount = 0	5✔
191
192	for (const [, pool] of tenantPools.entries()) {	5✔
193	poolCount++	2✔
194	const stats = pool.getPoolStats()	2✔
195	if (stats) {	2!
196	totalConnections += stats.total	2✔
197	totalInUse += stats.used	2✔
198	}
199	// Yield to the event loop between chunks
200	if (++chunkCount % STATS_CHUNK_SIZE === 0) {	2!
UNCOV 201	await new Promise<void>((resolve) => setImmediate(resolve))	×
202	}
203	}
204
205	cachedPoolStats = {	5✔
206	poolCount,
207	totalConnections,
208	totalInUse,
209	}
210	} finally {
211	collectInProgress = false	5✔
212	}
213	}
214		1✔
215	/**	1✔
216	* PoolManager is a class that manages a pool of Knex connections.	1✔
217	* It creates a new pool for each tenant and reuses existing pools.	1✔
218	*/	1✔
219	export class PoolManager {	1✔
220	protected numWorkers: number = 1	18✔
221		1✔
222	setNumWorkers(numWorkers: number) {	1✔
UNCOV 223	this.numWorkers = Math.max(numWorkers ?? 1, 1)	×
224	}
225		1✔
226	monitor() {	1✔
227	// Periodically collect stats in a non-blocking way
228	const interval = setInterval(() => {	1✔
229	void collectPoolStats()	5✔
230	}, STATS_INTERVAL_MS)
231	interval.unref()	1✔
232
233	// Observable callback reads the cached snapshot — O(1)
234	meter.addBatchObservableCallback(	1✔
235	(observer) => {
236	if (isMetricEnabled('db_active_local_pools')) {	1!
237	observer.observe(dbActivePool, cachedPoolStats.poolCount)	1✔
238	}
239	if (isMetricEnabled('db_connections')) {	1!
240	observer.observe(dbActiveConnection, cachedPoolStats.totalConnections)	1✔
241	}
242	if (isMetricEnabled('db_connections_in_use')) {	1!
243	observer.observe(dbInUseConnection, cachedPoolStats.totalInUse)	1✔
244	}
245	},
246	[dbActivePool, dbActiveConnection, dbInUseConnection]
247	)
248	}
249		1✔
250	rebalanceAll(data: { clusterSize: number }) {	1✔
251	for (const pool of tenantPools.values()) {	1✔
252	pool.rebalance({	2✔
253	clusterSize: data.clusterSize,
254	})
255	}
256	}
257		1✔
258	rebalance(tenantId: string, data: { clusterSize: number }) {	1✔
259	const pool = tenantPools.get(tenantId)	×
260	if (pool) {	×
UNCOV 261	pool.rebalance({	×
262	clusterSize: data.clusterSize,
263	})
264	}
265	}
266		1✔
267	getPool(settings: TenantConnectionOptions) {	1✔
268	const isCacheable = (settings.isSingleUse && !settings.isExternalPool) \|\| !settings.isSingleUse	1,209✔
269	const { value: existingPool, outcome } = tenantPools.getWithOutcome(settings.tenantId)	1,209✔
270		1,178✔
271	if (existingPool) {	1,209✔
272	recordTenantPoolCacheRequest(outcome)	7✔
273	logTenantPoolCacheLookup(settings, isCacheable, outcome)	7✔
274		1✔
275	return existingPool	7✔
276	}	1✔
277		1,177✔
278	if (!isCacheable) {	1,203✔
279	return this.newPool({ ...settings, numWorkers: this.numWorkers })	1,177✔
280	}	1,174✔
281		3✔
282	recordTenantPoolCacheRequest(outcome)	25✔
283	logTenantPoolCacheLookup(settings, isCacheable, outcome)	25✔
284		3✔
285	const newPool = this.newPool({ ...settings, numWorkers: this.numWorkers })	25✔
286		3✔
287	tenantPools.set(settings.tenantId, newPool)	25✔
288	return newPool	25✔
289	}	3✔
290		1✔
291	destroy(tenantId: string) {	1✔
292	const pool = tenantPools.get(tenantId)	18✔
293	if (pool) {	18!
294	manuallyDestroyedPools.add(pool)	2✔
295	tenantPools.delete(tenantId)	2✔
296	return destroyPool(pool).finally(() => {	2✔
297	manuallyDestroyedPools.delete(pool)	2✔
298	})
299	}
300	return Promise.resolve()	16✔
301	}	16✔
302		1✔
303	destroyAll() {	1✔
304	const promises: Promise<void>[] = []	15✔
305
306	for (const [connectionString, pool] of tenantPools) {	15✔
307	manuallyDestroyedPools.add(pool)	13✔
308	tenantPools.delete(connectionString)	13✔
309	promises.push(	13✔
310	destroyPool(pool).finally(() => {
311	manuallyDestroyedPools.delete(pool)	13✔
312	})
313	)
314	}
315	return Promise.allSettled(promises)	15✔
316	}
317		1✔
318	protected newPool(settings: TenantConnectionOptions): PoolStrategy {	1✔
319	return new TenantPool(settings)	1,177✔
320	}	1,177✔
321	}	1✔
322		1✔
323	/**	1✔
324	* TenantPool create a new Knex pool for each tenant, with rebalance	1✔
325	* functionality to adjust the number of connections based on the cluster size.	1✔
326	*/	1✔
327	class TenantPool implements PoolStrategy {	1✔
328	protected pool?: Knex	1✔
329		1✔
330	constructor(protected readonly options: TenantConnectionOptions) {}	1✔
331		1✔
332	acquire() {	1✔
333	if (this.pool) {	3,187!
334	return this.pool	2,062✔
335	}	2,062✔
336		1,125✔
337	this.pool = this.createKnexPool()	1,125✔
338	return this.pool	1,125✔
339	}	1,125✔
340		1✔
341	destroy(): Promise<void> {	1✔
342	const originalPool = this.pool	1,139✔
343		1,139✔
344	if (!originalPool) {	1,139!
345	return Promise.resolve()	58✔
346	}	58✔
347		1,081✔
348	this.pool = undefined	1,081✔
349	return this.drainPool(originalPool)	1,081✔
350	}	1,081✔
351		1✔
352	getPoolStats(): PoolStats \| null {	1✔
353	const tarnPool = this.pool?.client?.pool	×
354	if (!tarnPool) return null	×
UNCOV 355	return {	×
356	used: tarnPool.numUsed(),
357	total: tarnPool.numUsed() + tarnPool.numFree(),
358	}
359	}
360		1✔
361	getSettings() {	1✔
362	const isSingleUseExternalPool = this.options.isSingleUse && this.options.isExternalPool	1,125!
363		1,125✔
364	const numWorkers = Math.max(this.options.numWorkers ?? 1, 1)	1,125!
365	const clusterSize = this.options.clusterSize \|\| 0	1,125!
366	let maxConnection = this.options.maxConnections \|\| databaseMaxConnections	1,125!
367		1,125✔
368	const divisor = Math.max(clusterSize, 1) * numWorkers	1,125✔
369	if (divisor > 1) {	1,125!
UNCOV 370	maxConnection = Math.ceil(maxConnection / divisor) \|\| 1	×
371	}
372		1,125✔
373	if (isSingleUseExternalPool) {	1,125!
374	maxConnection = 1	1,122✔
375	}	1,122✔
376		1,125✔
377	return {	1,125✔
378	...this.options,	1,125✔
379	searchPath: this.options.isExternalPool ? undefined : searchPath,	1,125!
380	idleTimeoutMillis: isSingleUseExternalPool ? 100 : databaseFreePoolAfterInactivity,	1,125!
381	reapIntervalMillis: isSingleUseExternalPool ? 50 : undefined,	1,125!
382	maxConnections: maxConnection,	1,125✔
383	}	1,125✔
384	}	1,125✔
385		1✔
386	rebalance(options: { clusterSize: number }) {	1✔
387	if (options.clusterSize === 0) {	×
UNCOV 388	return	×
389	}
390
UNCOV 391	const originalPool = this.pool	×
392
393	this.options.clusterSize = options.clusterSize	×
UNCOV 394	this.pool = undefined	×
395
396	if (originalPool) {	×
NEW 397	this.drainPool(originalPool).catch((e) => {	×
NEW 398	logger.error({ type: 'pool', error: e })	×
399	})
400	}
401	}
402		1✔
403	protected async drainPool(pool: Knex) {	1✔
404	for (; pool?.client?.pool; ) {	1,081✔
405	let waiting = 0	1,081✔
406	waiting += pool.client.pool.numPendingAcquires()	1,081✔
407	waiting += pool.client.pool.numPendingValidations()	1,081✔
408	waiting += pool.client.pool.numPendingCreates()	1,081✔
409		1,081✔
410	if (waiting === 0) {	1,081!
411	break	1,081✔
412	}	1,081✔
413		×
UNCOV 414	await wait(200)	×
415	}
416		1,081✔
417	return pool.destroy()	1,081✔
418	}	1,081✔
419		1✔
420	protected createKnexPool() {	1✔
421	const settings = this.getSettings()	1,125✔
422	const sslSettings = getSslSettings({	1,125✔
423	connectionString: settings.dbUrl,	1,125✔
424	databaseSSLRootCert,	1,125✔
425	})	1,125✔
426		1,125✔
427	const maxConnections = settings.maxConnections	1,125✔
428		1,125✔
429	return knex({	1,125✔
430	client: 'pg',	1,125✔
431	version: dbPostgresVersion,	1,125✔
432	searchPath: settings.searchPath,	1,125✔
433	pool: {	1,125✔
434	min: 0,	1,125✔
435	max: maxConnections,	1,125✔
436	acquireTimeoutMillis: databaseConnectionTimeout,	1,125✔
437	idleTimeoutMillis: settings.idleTimeoutMillis,	1,125✔
438	reapIntervalMillis: 1000,	1,125✔
439	},	1,125✔
440	connection: {	1,125✔
441	connectionString: settings.dbUrl,	1,125✔
442	connectionTimeoutMillis: databaseConnectionTimeout,	1,125✔
443	ssl: sslSettings ? { ...sslSettings } : undefined,	1,125!
444	application_name: databaseApplicationName,	1,125✔
445	},	1,125✔
446	acquireConnectionTimeout: databaseConnectionTimeout,	1,125✔
447	})	1,125✔
448	}	1,125✔
449	}	1✔

supabase / storage / 24400472712

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous