fix: pool settings improvements (#347)

Author: fenos

package-lock.json

@@ -25,10 +25,10 @@
     "node": ">= 14.0.0"
   },
   "dependencies": {
-    "@aws-sdk/client-s3": "^3.350.0",
-    "@aws-sdk/lib-storage": "^3.182.0",
-    "@aws-sdk/node-http-handler": "^3.178.0",
-    "@aws-sdk/s3-request-presigner": "^3.182.0",
+    "@aws-sdk/client-s3": "3.350.0",
+    "@aws-sdk/lib-storage": "3.350.0",
+    "@aws-sdk/node-http-handler": "3.350.0",
+    "@aws-sdk/s3-request-presigner": "3.350.0",
     "@fastify/multipart": "^7.6.0",
     "@fastify/rate-limit": "^7.6.0",
     "@fastify/swagger": "^8.3.1",

package.json

@@ -119,7 +119,7 @@ export function getConfig(): StorageConfigType {
       10
     ),
     databaseConnectionTimeout: parseInt(
-      getOptionalConfigFromEnv('DATABASE_CONNECTION_TIMEOUT') || '30000',
+      getOptionalConfigFromEnv('DATABASE_CONNECTION_TIMEOUT') || '3000',
       10
     ),
     region: getConfigFromEnv('REGION'),

src/config.ts

@@ -2,7 +2,6 @@ import pg from 'pg'
 import { Knex, knex } from 'knex'
 import { JwtPayload } from 'jsonwebtoken'
 import { getConfig } from '../config'
-import { logger } from '../monitoring'
 import { DbActiveConnection, DbActivePool } from '../monitoring/metrics'
 
 // https://github.com/knex/knex/issues/387#issuecomment-51554522
@@ -30,12 +29,13 @@ export interface User {
 }
 
 export const connections = new Map<string, Knex>()
+const searchPath = ['storage', 'public', 'extensions']
 
 export class TenantConnection {
   public readonly role: string
 
   protected constructor(
-    public readonly pool: Knex,
+    protected readonly pool: Knex,
     protected readonly options: TenantConnectionOptions
   ) {
     this.role = options.user.payload.role || 'anon'
@@ -56,20 +56,19 @@ export class TenantConnection {
     let knexPool = connections.get(connectionString)
 
     if (knexPool) {
-      return new this(await knexPool, options)
+      return new this(knexPool, options)
     }
 
     const isExternalPool = Boolean(options.isExternalPool)
 
     knexPool = knex({
       client: 'pg',
-      searchPath: ['public', 'storage', 'extensions'],
+      searchPath: isExternalPool ? undefined : searchPath,
       pool: {
         min: 0,
         max: isExternalPool ? 1 : options.maxConnections || databaseMaxConnections,
-        propagateCreateError: false,
         acquireTimeoutMillis: databaseConnectionTimeout,
-        idleTimeoutMillis: isExternalPool ? 100 : undefined,
+        idleTimeoutMillis: isExternalPool ? 100 : databaseFreePoolAfterInactivity,
         reapIntervalMillis: isExternalPool ? 110 : undefined,
       },
       connection: connectionString,
@@ -97,38 +96,12 @@ export class TenantConnection {
     })
 
     if (!isExternalPool) {
-      let freePoolIntervalFn: NodeJS.Timeout | undefined
-
       knexPool.client.pool.on('poolDestroySuccess', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-        }
-
         if (connections.get(connectionString) === knexPool) {
           connections.delete(connectionString)
         }
       })
 
-      knexPool.client.pool.on('stopReaping', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-        }
-
-        freePoolIntervalFn = setTimeout(async () => {
-          connections.delete(connectionString)
-          knexPool?.destroy().catch((e) => {
-            logger.error(e, 'Error destroying pool')
-          })
-          clearTimeout(freePoolIntervalFn)
-        }, databaseFreePoolAfterInactivity)
-      })
-
-      knexPool.client.pool.on('startReaping', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-          freePoolIntervalFn = undefined
-        }
-      })
       connections.set(connectionString, knexPool)
     }
 
@@ -141,10 +114,20 @@ export class TenantConnection {
     }
   }
 
-  transaction(isolation?: Knex.IsolationLevels, instance?: Knex) {
-    return (instance || this.pool).transactionProvider({
-      isolationLevel: isolation,
-    })
+  async transaction(instance?: Knex) {
+    const pool = instance || this.pool
+    const tnx = await pool.transaction()
+
+    if (!instance && this.options.isExternalPool) {
+      await tnx.raw(`SELECT set_config('search_path', ?, true)`, [searchPath.join(', ')])
+    }
+    return tnx
+  }
+
+  transactionProvider(instance?: Knex): Knex.TransactionProvider {
+    return async () => {
+      return this.transaction(instance)
+    }
   }
 
   asSuperUser() {

src/database/connection.ts

@@ -42,10 +42,7 @@ export class StorageKnexDB implements Database {
 
     while (retryLeft > 0) {
       try {
-        const tnx = await this.connection.transaction(
-          transactionOptions?.isolation as Knex.IsolationLevels,
-          this.options.tnx
-        )()
+        const tnx = await this.connection.transactionProvider(this.options.tnx)()
 
         try {
           await this.connection.setScope(tnx)
@@ -475,7 +472,7 @@ export class StorageKnexDB implements Database {
     const needsNewTransaction = !tnx || differentScopes
 
     if (!tnx || needsNewTransaction) {
-      tnx = await this.connection.transaction(isolation, this.options.tnx)()
+      tnx = await this.connection.transactionProvider(this.options.tnx)()
       tnx.on('query-error', (error: DatabaseError) => {
         throw DBError.fromDBError(error)
       })

src/storage/database/knex.ts

@@ -12,11 +12,13 @@ import { StorageBackendError } from '../storage'
 import { useMockObject, useMockQueue } from './common'
 import { getPostgresConnection } from '../database'
 import { getServiceKeyUser } from '../database/tenant'
+import { Knex } from 'knex'
 
 dotenv.config({ path: '.env.test' })
 
 const { anonKey, jwtSecret, serviceKey, tenantId } = getConfig()
 
+let tnx: Knex.Transaction | undefined
 async function getSuperuserPostgrestClient() {
   const superUser = await getServiceKeyUser(tenantId)
 
@@ -26,14 +28,20 @@ async function getSuperuserPostgrestClient() {
     tenantId,
     host: 'localhost',
   })
-  const tnx = await conn.pool
+  tnx = await conn.transaction()
 
   return tnx
 }
 
 useMockObject()
 useMockQueue()
 
+afterEach(async () => {
+  if (tnx) {
+    await tnx.commit()
+  }
+})
+
 /*
  * GET /object/:id
  */

src/test/object.test.ts

@@ -132,7 +132,7 @@ describe('RLS policies', () => {
 
   afterAll(async () => {
     await db.destroy()
-    await (storage.db as StorageKnexDB).connection.pool.destroy()
+    await (storage.db as StorageKnexDB).connection.dispose()
   })
 
   testSpec.tests.forEach((_test, index) => {

src/test/rls.test.ts

@@ -294,7 +294,7 @@ describe('Webhooks', () => {
 
 async function createObject(pg: TenantConnection, bucketId: string) {
   const objectName = Date.now()
-  const tnx = pg.pool
+  const tnx = await pg.transaction()
 
   const [data] = await tnx
     .from<Obj>('objects')
@@ -316,5 +316,7 @@ async function createObject(pg: TenantConnection, bucketId: string) {
     ])
     .returning('*')
 
+  await tnx.commit()
+
   return data as Obj
 }

src/test/webhooks.test.ts

@@ -12,7 +12,8 @@
     "target": "ES2020",
     "outDir": "dist",
     "sourceMap": true,
-    "strict": true
+    "strict": true,
+    "skipLibCheck": true
   },
   "include": ["./src/**/*"],
   "exclude": ["node_modules", "dist"],