fix(knex): handle errors and stalls of the LISTEN probe client

The probe client had no error listener: a connection failure while it
waited for its notification emitted an unhandled error event and
killed the process, instead of rejecting watch(). Its errors now reject
the probe.

The probe client also ended gracefully without a bound, so a connection
stalled in its handshake kept its socket open and could hold the
process at shutdown. It now ends like the LISTEN client: its socket is
destroyed when end() does not settle in time. Both keep an error
listener while they end.

Author: RomainLanz

src/drivers/knex_adapter.ts

@@ -96,6 +96,21 @@ function wakeChannelName(table: string): string {
   return `queue_jobs_${createHash('sha256').update(table).digest('hex').slice(0, 40)}`
 }
 
+/**
+ * End a LISTEN or probe client. A graceful end waits for the server, which a
+ * stalled connection never answers: its socket is destroyed after a while.
+ */
+async function endListenClient(client: PgListenClient): Promise<void> {
+  client.removeAllListeners()
+  client.on('error', () => {})
+  const timer = setTimeout(
+    () => client.connection?.stream?.destroy(),
+    subscriptionHeartbeat.timeout
+  )
+  await client.end().catch(() => {})
+  clearTimeout(timer)
+}
+
 /** The message of a job of `queue` ready now, or due at `at`. */
 function jobMessage(queue: string, at?: number): WakeMessage {
   return at === undefined ? { ready: [queue] } : { due: { [queue]: at } }
@@ -358,14 +373,7 @@ export class KnexAdapter implements Adapter {
       const current = client
       client = undefined
       if (!current) return
-      current.removeAllListeners()
-      // A graceful end waits for the server, which a stalled connection never answers.
-      const timer = setTimeout(
-        () => current.connection?.stream?.destroy(),
-        subscriptionHeartbeat.timeout
-      )
-      await current.end().catch(() => {})
-      clearTimeout(timer)
+      await endListenClient(current)
     }
 
     // A connection config provider is called again once its settings expire,
@@ -437,6 +445,10 @@ export class KnexAdapter implements Adapter {
       }
       const { received, payload } = probe
       const sender = await createClient()
+      // An error event without a listener would crash the process.
+      const failed = Promise.withResolvers<never>()
+      failed.promise.catch(() => {})
+      sender.on('error', (error) => failed.reject(error))
       let timer: NodeJS.Timeout | undefined
       const timeout = new Promise<never>((_resolve, reject) => {
         timer = setTimeout(
@@ -458,11 +470,11 @@ export class KnexAdapter implements Adapter {
       })()
       exchange.catch(() => {})
       try {
-        await Promise.race([exchange, timeout, closing.promise])
+        await Promise.race([exchange, failed.promise, timeout, closing.promise])
       } finally {
         clearTimeout(timer)
         probe = undefined
-        void sender.end().catch(() => {})
+        void endListenClient(sender)
       }
     }
 

tests/adapter.spec.ts

@@ -2611,11 +2611,15 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => {
    * client gets the behavior `behave` returns for its 1-based creation
    * number: 'real' for a working connection, or a fake one.
    */
+  type FakeBehavior = {
+    connect?: () => Promise<void>
+    query?: () => Promise<void>
+    end?: () => Promise<void>
+  }
+
   function fakeListenClients(
     cleanup: (callback: () => void) => void,
-    behave: (
-      index: number
-    ) => 'real' | { connect?: () => Promise<void>; query?: () => Promise<void> }
+    behave: (index: number) => 'real' | FakeBehavior
   ) {
     type PgClient = new (settings: unknown) => {
       end(): Promise<void>
@@ -2630,16 +2634,24 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => {
       client.driver = driver
     })
 
-    const created: Array<{ ended: boolean; emit: (event: string, error?: Error) => void }> = []
+    const created: Array<{
+      ended: boolean
+      destroyed?: boolean
+      emit: (event: string, error?: Error) => void
+    }> = []
     const FakeClient = class {
       #listeners = new Map<string, (error?: Error) => void>()
-      #behavior: { connect?: () => Promise<void>; query?: () => Promise<void> }
+      #behavior: FakeBehavior
       state = {
         ended: false,
+        destroyed: false,
         emit: (event: string, error?: Error) => this.#listeners.get(event)?.(error),
       }
 
-      constructor(behavior: { connect?: () => Promise<void>; query?: () => Promise<void> }) {
+      // As node-postgres: the socket, destroyed when end() does not settle.
+      connection = { stream: { destroy: () => (this.state.destroyed = true) } }
+
+      constructor(behavior: FakeBehavior) {
         this.#behavior = behavior
         created.push(this.state)
       }
@@ -2658,6 +2670,7 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => {
       }
       async end() {
         this.state.ended = true
+        await this.#behavior.end?.()
       }
     }
     client.driver.Client = function (settings: unknown) {
@@ -2693,6 +2706,45 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => {
     await assert.rejects(() => watching.watch!(['default']), /do not enable notifications/)
   }).timeout(5000)
 
+  test('rejects a LISTEN probe whose connection fails', async ({ assert, cleanup }) => {
+    // The probe sends its notification, then its connection fails.
+    const created = fakeListenClients(cleanup, (index) =>
+      index === 1
+        ? 'real'
+        : {
+            async query() {
+              setTimeout(() => created[1].emit('error', new Error('Connection terminated')), 10)
+            },
+          }
+    )
+    const watching = new KnexAdapter({
+      connection,
+      tableName,
+      schedulesTableName,
+      notifications: true,
+    })
+
+    await assert.rejects(() => watching.watch!(['default']), 'Connection terminated')
+  })
+
+  test('destroys a LISTEN probe connection that does not end', async ({ assert, cleanup }) => {
+    shortenHeartbeat(cleanup)
+    const never = () => new Promise<void>(() => {})
+    const created = fakeListenClients(cleanup, (index) =>
+      index === 1 ? 'real' : { connect: never, end: never }
+    )
+    const watching = new KnexAdapter({
+      connection,
+      tableName,
+      schedulesTableName,
+      notifications: true,
+    })
+
+    await assert.rejects(() => watching.watch!(['default']), /do not enable notifications/)
+    await new Promise((resolve) => setTimeout(resolve, 400))
+    assert.isTrue(created[1].destroyed)
+  }).timeout(5000)
+
   test('aborts a first LISTEN connection still in its handshake', async ({ assert, cleanup }) => {
     const created = fakeListenClients(cleanup, () => ({ connect: () => new Promise(() => {}) }))
     const watching = new KnexAdapter({