Fix: bound the per-table catalog fan-out in connection-test

`connectionTestListTables` fetched `/connection_catalog/.../tables/{t}/columns`
for every table of the schema with an unbounded `Promise.all`. A schema with
~100 tables became ~100 concurrent connections-api Lambda invocations, each
cold container opening its own SQLAlchemy connection on the shared
postgres-microservices-prd RDS (max_connections ~85). This saturated the
database three times on 03-04/set/2026 (65, 85 and 83 connections).

- Only fetch per-table primary keys when the plugin is CDC (the only consumer);
  batch plugins get `table_list` and an empty `tables`, which the frontend
  already handles.
- Run the remaining fan-outs (CDC list-tables and getTableMetadata) through
  `mapWithConcurrency` with CONNECTION_CATALOG_CONCURRENCY (default 4, env
  override).
- Unit tests for the bound, the non-CDC short-circuit and the helper.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

src/modules/connection-test/connection-test.service.spec.ts

@@ -1,4 +1,7 @@
-import { ConnectionTestService } from './connection-test.service';
+import {
+  CONNECTION_CATALOG_CONCURRENCY,
+  ConnectionTestService,
+} from './connection-test.service';
 import { RequestUser } from 'src/decorators/user.decorator';
 
 describe('ConnectionTestService catalog cache', () => {
@@ -45,7 +48,29 @@ describe('ConnectionTestService catalog cache', () => {
     });
   });
 
-  it('lists tables and enriches each with its cached primary keys', async () => {
+  it('lists tables for a batch plugin with a single catalog call (no per-table fan-out)', async () => {
+    connectionsApiService.proxy.mockResolvedValueOnce({
+      tables: [{ table_name: 'customers' }, { table_name: 'orders' }],
+    });
+
+    await expect(
+      service.connectionTestListTables(
+        {
+          connection_id: 'config-id',
+          plugin: 'postgresql',
+          schema: 'public',
+        },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      table_list: ['customers', 'orders'],
+      tables: [],
+    });
+    expect(connectionsApiService.proxy).toHaveBeenCalledTimes(1);
+  });
+
+  it('lists tables for a CDC plugin and enriches each with its cached primary keys', async () => {
     connectionsApiService.proxy
       // list-tables call (names only from the catalog cache)
       .mockResolvedValueOnce({
@@ -68,7 +93,7 @@ describe('ConnectionTestService catalog cache', () => {
       service.connectionTestListTables(
         {
           connection_id: 'config-id',
-          plugin: 'postgresql',
+          plugin: 'postgresql_cdc',
           schema: 'public',
         },
         user,
@@ -83,6 +108,39 @@ describe('ConnectionTestService catalog cache', () => {
     });
   });
 
+  it('bounds the per-table columns fan-out to CONNECTION_CATALOG_CONCURRENCY', async () => {
+    const tableCount = 25;
+    let inFlight = 0;
+    let maxInFlight = 0;
+    connectionsApiService.proxy.mockImplementation(async (_method, path) => {
+      if (path.endsWith('/tables')) {
+        return {
+          tables: Array.from({ length: tableCount }, (_, i) => ({
+            table_name: `t${i}`,
+          })),
+        };
+      }
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 2));
+      inFlight--;
+      return {
+        columns: [{ column_name: 'id', data_type: 'bigint', is_primary_key: true }],
+      };
+    });
+
+    const result = await service.connectionTestListTables(
+      { connection_id: 'config-id', plugin: 'mysql_cdc', schema: 'public' },
+      user,
+    );
+
+    expect(result.tables).toHaveLength(tableCount);
+    expect(result.tables[7]).toEqual({ table_name: 't7', primary_keys: ['id'] });
+    expect(connectionsApiService.proxy).toHaveBeenCalledTimes(tableCount + 1);
+    expect(maxInFlight).toBeLessThanOrEqual(CONNECTION_CATALOG_CONCURRENCY);
+    expect(maxInFlight).toBeGreaterThan(1);
+  });
+
   it('maps cached columns and derives references from the engine allowlist', async () => {
     connectionsApiService.proxy.mockResolvedValue({
       columns: [

src/modules/connection-test/connection-test.service.ts

@@ -28,6 +28,19 @@ import { RequestUser } from 'src/decorators/user.decorator';
 import { PackTheMetadata } from 'src/utils/PackTheMetadata';
 import { ConnectionsApiService } from '../connections-api/connections-api.service';
 import { PlatformApiService } from '../platform-api/platform-api.service';
+import { isCdcPlugin } from 'src/utils/cdc';
+import { mapWithConcurrency } from 'src/utils/concurrency';
+
+/**
+ * Max parallel `/connection_catalog/.../columns` calls per request. Each
+ * in-flight call is one connections-api Lambda container holding one
+ * connection on the shared microservices RDS (max_connections ~85), so this
+ * must stay small. Override with CONNECTION_CATALOG_CONCURRENCY.
+ */
+export const CONNECTION_CATALOG_CONCURRENCY = Math.max(
+  1,
+  Number(process.env.CONNECTION_CATALOG_CONCURRENCY) || 4,
+);
 
 @Injectable()
 export class ConnectionTestService {
@@ -180,13 +193,23 @@ export class ConnectionTestService {
       user,
     );
     const table_names: string[] = result.tables.map((table) => table.table_name);
-    // CDC create needs the primary keys per table (used to build the deduped
-    // Iceberg identifier-fields). The catalog-cache list-tables endpoint returns
-    // only names, so fetch each table's columns from the cache and keep the ones
-    // flagged is_primary_key. Reads hit the stored catalog snapshot (populated by
-    // refresh-catalog), never the live connection.
-    const tables = await Promise.all(
-      table_names.map(async (table_name) => {
+
+    // Only CDC create needs the primary keys per table (used to build the
+    // deduped Iceberg identifier-fields). Batch pipelines get the names only:
+    // fetching every table's columns costs one connections-api call per table
+    // in the schema, which is what saturated the microservices RDS.
+    if (!isCdcPlugin(body.plugin)) {
+      return { operation_result: true, table_list: table_names, tables: [] };
+    }
+
+    // The catalog-cache list-tables endpoint returns only names, so fetch each
+    // table's columns from the cache and keep the ones flagged is_primary_key.
+    // Reads hit the stored catalog snapshot (populated by refresh-catalog),
+    // never the live connection. Bounded fan-out: see CONNECTION_CATALOG_CONCURRENCY.
+    const tables = await mapWithConcurrency(
+      table_names,
+      CONNECTION_CATALOG_CONCURRENCY,
+      async (table_name) => {
         const columns = await this.connectionsApiService.proxy(
           'GET',
           `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
@@ -200,7 +223,7 @@ export class ConnectionTestService {
             .filter((column) => column.is_primary_key)
             .map((column) => column.column_name),
         };
-      }),
+      },
     );
     return {
       operation_result: true,
@@ -227,8 +250,12 @@ export class ConnectionTestService {
       allowedDataTypes.map((type) => type.toLowerCase()),
     );
 
-    const tables_metadata = await Promise.all(
-      body.table_list.map(async (table_name) => {
+    // One catalog call per *selected* table, bounded so a user picking dozens
+    // of tables cannot spin up dozens of Lambda containers at once.
+    const tables_metadata = await mapWithConcurrency(
+      body.table_list,
+      CONNECTION_CATALOG_CONCURRENCY,
+      async (table_name) => {
         const result = await this.connectionsApiService.proxy(
           'GET',
           `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
@@ -249,7 +276,7 @@ export class ConnectionTestService {
           columns,
           references,
         };
-      }),
+      },
     );
     return { operation_result: true, tables_metadata };
   }

src/utils/concurrency.spec.ts

@@ -0,0 +1,39 @@
+import { mapWithConcurrency } from './concurrency';
+
+describe('mapWithConcurrency', () => {
+  it('preserves input order and never exceeds the limit', async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const items = Array.from({ length: 12 }, (_, i) => i);
+
+    const result = await mapWithConcurrency(items, 3, async (item) => {
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      // Slower items first so ordering would break if we appended results.
+      await new Promise((resolve) => setTimeout(resolve, 12 - item));
+      inFlight--;
+      return item * 2;
+    });
+
+    expect(result).toEqual(items.map((i) => i * 2));
+    expect(maxInFlight).toBe(3);
+  });
+
+  it('handles empty input and a limit below one', async () => {
+    await expect(mapWithConcurrency([], 4, async (x) => x)).resolves.toEqual(
+      [],
+    );
+    await expect(
+      mapWithConcurrency([1, 2], 0, async (x) => x + 1),
+    ).resolves.toEqual([2, 3]);
+  });
+
+  it('rejects with the first worker error', async () => {
+    await expect(
+      mapWithConcurrency([1, 2, 3], 2, async (x) => {
+        if (x === 2) throw new Error('boom');
+        return x;
+      }),
+    ).rejects.toThrow('boom');
+  });
+});

src/utils/concurrency.ts

@@ -0,0 +1,30 @@
+/**
+ * Runs `worker` over `items` with at most `limit` promises in flight at a
+ * time, preserving the input order in the result (like `Promise.all(map)`,
+ * but bounded).
+ *
+ * Why: every fan-out to the connections-api / platform-api Lambdas turns each
+ * in-flight call into its own Lambda container, and each container opens its
+ * own connection on the shared microservices RDS. An unbounded `Promise.all`
+ * over ~100 tables saturated that database (04/set/2026).
+ */
+export async function mapWithConcurrency<T, R>(
+  items: readonly T[],
+  limit: number,
+  worker: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  const size = Math.max(1, Math.floor(limit) || 1);
+  const results: R[] = new Array(items.length);
+  let next = 0;
+
+  const run = async (): Promise<void> => {
+    while (next < items.length) {
+      const index = next++;
+      results[index] = await worker(items[index], index);
+    }
+  };
+
+  const runners = Array.from({ length: Math.min(size, items.length) }, run);
+  await Promise.all(runners);
+  return results;
+}