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
This commit is contained in:
neon
2026-09-29 00:37:44 +00:00
parent 7e3bb1892e
commit acce1e15ac
2 changed files with 126 additions and 2 deletions
+84 -1
View File
@@ -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();
});
});
+42 -1
View File
@@ -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<PoolClient, TrackedClientState>();
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<PoolClient>;
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`,
);
}
}