Handle Postgres network scan failures

packages/connector-postgres/src/connector.test.ts

@@ -339,4 +339,38 @@ describe('KtxPostgresScanConnector', () => {
     expect(snapshot.tables.length).toBeGreaterThan(0);
     expect(endCalled).toBe(true);
   });
+
+  it('attaches an error listener to the pg pool', async () => {
+    const on = vi.fn();
+    const poolFactory: KtxPostgresPoolFactory = {
+      createPool() {
+        return {
+          on,
+          async connect() {
+            return {
+              query: vi.fn(async () => ({ rows: [{ '?column?': 1 }], fields: [{ name: '?column?', dataTypeID: 23 }] })),
+              release: vi.fn(),
+            };
+          },
+          end: vi.fn(async () => undefined),
+        };
+      },
+    };
+    const connector = new KtxPostgresScanConnector({
+      connectionId: 'warehouse',
+      connection: {
+        driver: 'postgres',
+        host: 'db.example.test',
+        database: 'analytics',
+        username: 'reader',
+        password: 'test-password', // pragma: allowlist secret
+        readonly: true,
+      },
+      poolFactory,
+    });
+
+    await expect(connector.testConnection()).resolves.toEqual({ success: true });
+
+    expect(on).toHaveBeenCalledWith('error', expect.any(Function));
+  });
 });

packages/connector-postgres/src/connector.ts

@@ -89,6 +89,7 @@ interface KtxPostgresClient {
 interface KtxPostgresPool {
   connect(): Promise<KtxPostgresClient>;
   end(): Promise<void>;
+  on?(event: 'error', listener: (error: Error) => void): void;
 }
 
 export interface KtxPostgresPoolFactory {
@@ -349,6 +350,7 @@ export class KtxPostgresScanConnector implements KtxScanConnector {
   private readonly now: () => Date;
   private readonly dialect = new KtxPostgresDialect();
   private pool: KtxPostgresPool | null = null;
+  private lastIdlePoolError: Error | null = null;
   private resolvedEndpoint: KtxPostgresResolvedEndpoint | null = null;
 
   constructor(options: KtxPostgresScanConnectorOptions) {
@@ -667,11 +669,15 @@ export class KtxPostgresScanConnector implements KtxScanConnector {
         config = { ...config, host: endpoint.host, port: endpoint.port };
       }
       this.pool = this.poolFactory.createPool(config);
+      this.pool.on?.('error', (error) => {
+        this.lastIdlePoolError = error;
+      });
     }
     return this.pool;
   }
 
   private async queryRaw<T>(sql: string, params?: unknown[]): Promise<T[]> {
+    this.throwIdlePoolErrorIfPresent();
     const pool = await this.getPool();
     const client = await pool.connect();
     try {
@@ -683,6 +689,7 @@ export class KtxPostgresScanConnector implements KtxScanConnector {
   }
 
   private async query(sql: string, params?: Record<string, unknown> | unknown[]): Promise<KtxQueryResult> {
+    this.throwIdlePoolErrorIfPresent();
     const pool = await this.getPool();
     const client = await pool.connect();
     try {
@@ -704,4 +711,13 @@ export class KtxPostgresScanConnector implements KtxScanConnector {
       throw new Error(`PostgreSQL connector ${this.connectionId} cannot run scan for ${connectionId}`);
     }
   }
+
+  private throwIdlePoolErrorIfPresent(): void {
+    if (!this.lastIdlePoolError) {
+      return;
+    }
+    const error = this.lastIdlePoolError;
+    this.lastIdlePoolError = null;
+    throw error;
+  }
 }