Handle connection terminated errors in postgres eventstore adapter

packages/runtime/adapters/eventstore-adapters/eventstore-base/src/create-adapter.ts

@@ -111,10 +111,20 @@ const createAdapter = <
     disposed: false,
     validateEventFilter,
     isInitialized: false,
+    connectionErrors: [],
     maybeThrowResourceError,
     bucketSize,
     getNextCursor: getNextCursor.bind(null),
     counters: new Map(),
+    createGetConnectPromise: () => {
+      let p: Promise<void>
+      return () => {
+        if (p === undefined) {
+          p = connect.bind(null, adapterPool, connectionDependencies, config)()
+        }
+        return p
+      }
+    },
   }
 
   const emptyProps: Partial<ConnectedProps> = {}
@@ -123,9 +133,7 @@ const createAdapter = <
     ...emptyProps,
   }
 
-  adapterPool.connectPromise = new Promise((resolve) => {
-    adapterPool.connectPromiseResolve = resolve.bind(null, null)
-  }).then(connect.bind(null, adapterPool, connectionDependencies, config))
+  adapterPool.getConnectPromise = adapterPool.createGetConnectPromise()
 
   const connectedProps: Partial<ConnectedProps> = {
     injectEvent: wrapMethod(adapterPool, injectEvent),

packages/runtime/adapters/eventstore-adapters/eventstore-base/src/types.ts

@@ -254,8 +254,9 @@ export type AdapterPoolPrimalProps = {
   validateEventFilter: ValidateEventFilter
 
   isInitialized: boolean
-  connectPromiseResolve?: any
-  connectPromise?: Promise<any>
+  createGetConnectPromise: () => () => Promise<void>
+  getConnectPromise?: () => Promise<void>
+  connectionErrors: any[]
 
   maybeThrowResourceError: CheckForResourceError
   bucketSize: number

packages/runtime/adapters/eventstore-adapters/eventstore-base/src/wrap-dispose.ts

@@ -18,7 +18,7 @@ const wrapDispose = <ConnectedProps extends AdapterPoolConnectedProps>(
     return
   }
 
-  await pool.connectPromise
+  if (pool.getConnectPromise) await pool.getConnectPromise()
 
   await dispose(pool as AdapterPoolConnected<ConnectedProps>)
 }

packages/runtime/adapters/eventstore-adapters/eventstore-base/src/wrap-method.ts

@@ -18,8 +18,10 @@ const connectOnDemandAndCall = async <
   }
 
   pool.isInitialized = true
-  pool.connectPromiseResolve()
-  await pool.connectPromise
+  if (pool.getConnectPromise) {
+    const connectPromise = pool.getConnectPromise()
+    await connectPromise
+  }
 
   return await method(pool as AdapterPoolConnected<ConnectedProps>, ...args)
 }

packages/runtime/adapters/eventstore-adapters/eventstore-base/test/wrap-method.test.js

@@ -7,10 +7,8 @@ test('wrapMethod should wrap API methods to await connection', async () => {
 
   const pool = {
     config: {},
-    connectPromiseResolve: sinon.stub(),
-    connectPromise: Promise.resolve(),
-    initialPromiseResolve: sinon.stub(),
-    initialPromise: Promise.resolve(),
+    createGetConnectPromise: sinon.stub(),
+    getConnectPromise: sinon.stub(),
   }
 
   const wrappedApiMethod = wrapMethod(pool, apiMethod)
@@ -24,8 +22,7 @@ test('wrapMethod should wrap API methods to await connection', async () => {
 
   sinon.assert.calledWith(apiMethod, pool, methodArg)
 
-  sinon.assert.calledWith(pool.connectPromiseResolve)
-  sinon.assert.notCalled(pool.initialPromiseResolve)
+  sinon.assert.calledWith(pool.getConnectPromise)
 })
 
 test('wrapMethod handle dispose adapter state', async () => {

packages/runtime/adapters/eventstore-adapters/eventstore-postgresql/src/connect.ts

@@ -46,6 +46,9 @@ const connect = async (
     ...connectionOptions,
   })
 
+  connection.on('error', (error) => {
+    pool.connectionErrors.push(error)
+  })
   await connection.connect()
 
   Object.assign<

packages/runtime/adapters/eventstore-adapters/eventstore-postgresql/src/execute-statement.ts

@@ -13,15 +13,31 @@ const isRetryableError = (error: any): boolean =>
   ) ||
     checkFuzzyError(error, /Remaining connection slots are reserved/i) ||
     checkFuzzyError(error, /Too many clients already/i) ||
-    checkFuzzyError(error, /Connection terminated unexpectedly/i) ||
     checkFormalError(error, 'ECONNRESET') ||
     checkFormalError(error, '08000') ||
     checkFormalError(error, '08003') ||
-    checkFormalError(error, '08006'))
+    checkFormalError(error, '08006') ||
+    checkFuzzyError(error, /Connection terminated/i) ||
+    checkFuzzyError(error, /canceling statement due to statement timeout/i) ||
+    checkFuzzyError(error, /Query read timeout/i))
 
 const executeStatement = async (pool: AdapterPool, sql: any): Promise<any> => {
   while (true) {
     try {
+      if (pool.connectionErrors.length > 0) {
+        let summaryError = pool.connectionErrors[0]
+        if (pool.connectionErrors.length > 1) {
+          summaryError = new Error(
+            pool.connectionErrors.map(({ message }) => message).join('\n')
+          )
+          summaryError.stack = pool.connectionErrors
+            .map(({ stack }) => stack)
+            .join('\n')
+        }
+        pool.connectionErrors = []
+        throw summaryError
+      }
+
       const result = await pool.connection.query(sql)
 
       if (result != null && Array.isArray(result.rows)) {
@@ -30,7 +46,16 @@ const executeStatement = async (pool: AdapterPool, sql: any): Promise<any> => {
 
       return null
     } catch (error) {
-      if (!isRetryableError(error)) {
+      if (isRetryableError(error)) {
+        try {
+          await pool.connection.end()
+        } catch (error) {
+          // pass
+        }
+
+        pool.getConnectPromise = pool.createGetConnectPromise()
+        await pool.getConnectPromise()
+      } else {
         throw error
       }
     }

packages/runtime/adapters/eventstore-adapters/eventstore-postgresql/test/connect.unit.test.ts

@@ -11,13 +11,20 @@ import connect from '../src/connect'
 import executeStatement from '../src/execute-statement'
 
 const mPostgres = mocked(Postgres)
+Object.assign(mPostgres.prototype, {
+  on: (event: 'error', listener: (err: Error) => void) => {
+    return mPostgres
+  },
+})
 
 let pool: AdapterPool
 let connectionDependencies: ConnectionDependencies
 let config: PostgresqlAdapterConfig
 
 beforeEach(() => {
-  pool = {} as any
+  pool = {
+    connectionErrors: [],
+  } as any
   connectionDependencies = {
     Postgres,
     coercer: jest.fn(),
@@ -37,7 +44,6 @@ beforeEach(() => {
     snapshotsTableName: 'snapshots-table-name',
     secretsTableName: 'secrets-table-name',
   }
-  mPostgres.mockClear()
 })
 
 test('destination passed to postgres client', async () => {

packages/runtime/adapters/readmodel-adapters/readmodel-postgresql/src/passthrough-error.ts

@@ -57,6 +57,10 @@ const PassthroughError: PassthroughErrorFactory = Object.assign(
           checkFuzzyError(error, /Remaining connection slots are reserved/i) ||
           checkFuzzyError(error, /Too many clients already/i) ||
           checkFuzzyError(error, /Connection terminated unexpectedly/i) ||
+          checkFuzzyError(
+            error,
+            /canceling statement due to statement timeout/i
+          ) ||
           checkFormalError(error, 'ECONNRESET') ||
           checkFormalError(error, PostgresErrors.CONNECTION_EXCEPTION) ||
           checkFormalError(error, PostgresErrors.CONNECTION_DOES_NOT_EXIST) ||