From acce1e15ac3b47b57e55035c9f3a09df4740b054 Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Tue, 29 Sep 2026 00:37:44 +0000 Subject: [PATCH] fix(server): survive database connection loss Handle both idle pool errors and checked-out client errors so a database restart fails only the affected operation. Keep connection-loss logging sanitized and deduplicated while allowing the pool to reconnect. Fixes #1581 --- server/src/db/instrumented-pg-pool.test.ts | 85 +++++++++++++++++++++- server/src/db/instrumented-pg-pool.ts | 43 ++++++++++- 2 files changed, 126 insertions(+), 2 deletions(-) diff --git a/server/src/db/instrumented-pg-pool.test.ts b/server/src/db/instrumented-pg-pool.test.ts index 481ac7c31..9075b213b 100644 --- a/server/src/db/instrumented-pg-pool.test.ts +++ b/server/src/db/instrumented-pg-pool.test.ts @@ -1,8 +1,28 @@ +import { EventEmitter } from 'node:events'; + import { Logger } from '@nestjs/common'; -import { Pool, type PoolClient } from 'pg'; +import { Pool, type ClientBase, type PoolClient } from 'pg'; import { InstrumentedPgPool } from './instrumented-pg-pool'; +class FakePoolClient extends EventEmitter { + release!: (error?: Error) => void; + ended = false; + _ending = false; + _queryable = true; + + connect(callback: (error?: Error) => void): void { + callback(); + } + + end(callback?: () => void): void { + this.ended = true; + callback?.(); + } +} + +const fakeClientConstructor = FakePoolClient as unknown as new () => ClientBase; + describe('InstrumentedPgPool', () => { afterEach(() => { vi.restoreAllMocks(); @@ -58,4 +78,67 @@ describe('InstrumentedPgPool', () => { expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('[db.pool_acquire] [fail] acquisitionKind=idle')); }); + + it('handles an idle client disconnect without throwing and removes the failed client', async () => { + const warnSpy = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + const pool = new InstrumentedPgPool({ max: 1, Client: fakeClientConstructor }); + const client = await pool.connect(); + client.release(); + + expect(pool.idleCount).toBe(1); + expect(() => client.emit('error', new Error('idle connection lost'))).not.toThrow(); + + expect(warnSpy).toHaveBeenCalledTimes(1); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringMatching( + /^\[db\.connection\] \[fail\] clientState=idle durationMs=\d+ totalCount=0 idleCount=0 waitingCount=0 errorClass=Error error="idle connection lost" - database connection lost$/, + ), + ); + expect(pool.totalCount).toBe(0); + expect((client as unknown as FakePoolClient).ended).toBe(true); + + const replacement = await pool.connect(); + expect(replacement).not.toBe(client); + replacement.release(); + await pool.end(); + }); + + it('handles a checked-out client disconnect without throwing or double logging', async () => { + const warnSpy = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + const pool = new InstrumentedPgPool({ max: 1, Client: fakeClientConstructor }); + const client = await pool.connect(); + const error = new Error('checked-out connection lost'); + + expect(pool.idleCount).toBe(0); + expect(() => client.emit('error', error)).not.toThrow(); + expect(() => client.emit('error', new Error('follow-up socket error'))).not.toThrow(); + + expect(warnSpy).toHaveBeenCalledTimes(1); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringMatching( + /^\[db\.connection\] \[fail\] clientState=checked-out durationMs=\d+ totalCount=1 idleCount=0 waitingCount=0 errorClass=Error error="checked-out connection lost" - database connection lost$/, + ), + ); + + client.release(error); + expect(pool.totalCount).toBe(0); + + const replacement = await pool.connect(); + expect(replacement).not.toBe(client); + replacement.release(); + await pool.end(); + }); + + it('sanitizes connection errors before logging them', async () => { + const warnSpy = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + const pool = new InstrumentedPgPool({ max: 1, Client: fakeClientConstructor }); + const client = await pool.connect(); + + client.emit('error', new Error('socket closed\nfor "maintenance"')); + + expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('error="socket closed for \\"maintenance\\""')); + + client.release(new Error('connection failed')); + await pool.end(); + }); }); diff --git a/server/src/db/instrumented-pg-pool.ts b/server/src/db/instrumented-pg-pool.ts index c160cb435..43d7e714a 100644 --- a/server/src/db/instrumented-pg-pool.ts +++ b/server/src/db/instrumented-pg-pool.ts @@ -1,13 +1,37 @@ import { Logger } from '@nestjs/common'; -import { Pool, type PoolClient } from 'pg'; +import { Pool, type PoolClient, type PoolConfig } from 'pg'; import { sanitizeLogValue } from '../common/utils/log-sanitize.utils'; type ConnectCallback = (err: Error | undefined, client: PoolClient | undefined, done: (release?: unknown) => void) => void; type AcquisitionKind = 'idle' | 'new' | 'queued'; +type ClientState = 'checked-out' | 'idle'; + +interface TrackedClientState { + state: ClientState; + startedAt: number; + reportedError: boolean; +} export class InstrumentedPgPool extends Pool { private readonly logger = new Logger(InstrumentedPgPool.name); + private readonly clientStates = new WeakMap(); + + constructor(config?: PoolConfig) { + super(config); + + this.on('connect', (client) => { + this.clientStates.set(client, { state: 'checked-out', startedAt: Date.now(), reportedError: false }); + client.on('error', (error) => this.handleClientError(client, error)); + }); + this.on('acquire', (client) => { + this.clientStates.set(client, { state: 'checked-out', startedAt: Date.now(), reportedError: false }); + }); + this.on('release', (_error, client) => { + this.clientStates.set(client, { state: 'idle', startedAt: Date.now(), reportedError: false }); + }); + this.on('error', (error, client) => this.logConnectionError('idle', client, error)); + } override connect(): Promise; override connect(callback: ConnectCallback): void; @@ -57,4 +81,21 @@ export class InstrumentedPgPool extends Pool { `[db.pool_acquire] [end] acquisitionKind=${acquisitionKind} durationMs=${durationMs} totalCount=${this.totalCount} idleCount=${this.idleCount} waitingCount=${this.waitingCount} - database connection acquired`, ); } + + private handleClientError(client: PoolClient, error: Error): void { + const tracked = this.clientStates.get(client); + if (tracked?.state !== 'checked-out' || tracked.reportedError) return; + tracked.reportedError = true; + this.logConnectionError('checked-out', client, error); + } + + private logConnectionError(state: ClientState, client: PoolClient, error: unknown): void { + const tracked = this.clientStates.get(client); + const durationMs = tracked ? Date.now() - tracked.startedAt : 0; + const errorClass = error instanceof Error ? error.constructor.name : typeof error; + const message = sanitizeLogValue(error instanceof Error ? error.message : error); + this.logger.warn( + `[db.connection] [fail] clientState=${state} durationMs=${durationMs} totalCount=${this.totalCount} idleCount=${this.idleCount} waitingCount=${this.waitingCount} errorClass=${errorClass} error="${message}" - database connection lost`, + ); + } }