fix: properly pass maxPgConnections config through connection chain (#589)

# Fix maxPgConnections Configuration

This PR fixes an issue where the `maxPgConnections` configuration parameter was being ignored in the edge worker. The parameter is now properly passed through the connection chain with a default value of 4 (previously hardcoded to 10).

## Changes:
- Added `maxPgConnections` parameter to all relevant interfaces and function calls
- Updated the connection resolution logic to use the configured value or default to 4
- Added test functions to verify both default and custom connection limits
- Added unit tests to validate the connection configuration behavior
- Added E2E tests to ensure the configuration is properly applied

This change ensures that users can now properly control the maximum number of PostgreSQL connections used by the edge worker, which is important for resource management and performance tuning.

Author: jumski

.changeset/fix-max-pg-connections-config.md

@@ -0,0 +1,5 @@
+---
+'@pgflow/edge-worker': patch
+---
+
+Fix maxPgConnections config being ignored - now properly passed through the connection chain with default of 4

pkgs/edge-worker/deno.lock

@@ -143,6 +143,7 @@ export class EdgeWorker {
     const platform = await createAdapter({
       sql: config.sql,
       connectionString: config.connectionString,
+      maxPgConnections: config.maxPgConnections,
     });
     this.platform = platform;
 
@@ -197,6 +198,7 @@ export class EdgeWorker {
     const platform = await createAdapter({
       sql: config.sql,
       connectionString: config.connectionString,
+      maxPgConnections: config.maxPgConnections,
     });
     this.platform = platform;
 

pkgs/edge-worker/src/EdgeWorker.ts

@@ -54,7 +54,7 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
   private loggingFactory!: ReturnType<typeof createLoggingFactory>;
 
   constructor(
-    options?: { sql?: postgres.Sql; connectionString?: string },
+    options?: { sql?: postgres.Sql; connectionString?: string; maxPgConnections?: number },
     deps: SupabasePlatformDeps = getPlatformDeps()
   ) {
     this.deps = deps;

pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts

@@ -7,6 +7,7 @@ import { getPlatformDeps } from './deps.js';
 interface AdapterOptions {
   sql?: postgres.Sql;
   connectionString?: string;
+  maxPgConnections?: number;
 }
 
 /**

pkgs/edge-worker/src/platform/createAdapter.ts

@@ -24,6 +24,7 @@ export interface ConnectionOptions {
 export interface SqlConnectionOptions {
   sql?: postgres.Sql;
   connectionString?: string;
+  maxPgConnections?: number;
 }
 
 /**
@@ -82,19 +83,21 @@ export function resolveSqlConnection(
     return options.sql;
   }
 
+  const max = options?.maxPgConnections ?? 4;
+
   // 2. config.connectionString
   if (options?.connectionString) {
-    return postgres(options.connectionString, { prepare: false, max: 10 });
+    return postgres(options.connectionString, { prepare: false, max });
   }
 
   // 3. EDGE_WORKER_DB_URL
   if (env.EDGE_WORKER_DB_URL) {
-    return postgres(env.EDGE_WORKER_DB_URL, { prepare: false, max: 10 });
+    return postgres(env.EDGE_WORKER_DB_URL, { prepare: false, max });
   }
 
   // 4. Local Supabase detection + docker URL
   if (isLocalSupabaseEnv(env)) {
-    return postgres(DOCKER_TRANSACTION_POOLER_URL, { prepare: false, max: 10 });
+    return postgres(DOCKER_TRANSACTION_POOLER_URL, { prepare: false, max });
   }
 
   throw new Error(

pkgs/edge-worker/src/platform/resolveConnection.ts

@@ -0,0 +1,20 @@
+import { EdgeWorker } from '@pgflow/edge-worker';
+
+// Tests that default maxPgConnections is 4 (not the old hardcoded 10)
+// NO maxPgConnections specified - should default to 4
+EdgeWorker.start(
+  async (_payload, { sql }) => {
+    const queueName = 'conn_max_pg_default';
+    const actualMax = sql.options.max;
+    const status = actualMax === 4 ? 'success' : 'error';
+    const errorMessage = actualMax === 4 ? null : `Expected max=4, got ${actualMax}`;
+
+    await sql`
+      INSERT INTO e2e_test_results (queue_name, status, actual, error_message)
+      VALUES (${queueName}, ${status}, ${sql.json({ max: actualMax })}, ${errorMessage})
+    `;
+
+    return { max: actualMax };
+  },
+  { queueName: 'conn_max_pg_default' }
+);

pkgs/edge-worker/supabase/functions/conn_max_pg_default/index.ts

@@ -0,0 +1,22 @@
+import { EdgeWorker } from '@pgflow/edge-worker';
+
+// Tests that explicit maxPgConnections: 7 is respected
+EdgeWorker.start(
+  async (_payload, { sql }) => {
+    const queueName = 'conn_max_pg_override';
+    const actualMax = sql.options.max;
+    const status = actualMax === 7 ? 'success' : 'error';
+    const errorMessage = actualMax === 7 ? null : `Expected max=7, got ${actualMax}`;
+
+    await sql`
+      INSERT INTO e2e_test_results (queue_name, status, actual, error_message)
+      VALUES (${queueName}, ${status}, ${sql.json({ max: actualMax })}, ${errorMessage})
+    `;
+
+    return { max: actualMax };
+  },
+  {
+    queueName: 'conn_max_pg_override',
+    maxPgConnections: 7,
+  }
+);

pkgs/edge-worker/supabase/functions/conn_max_pg_override/index.ts

@@ -1,13 +1,21 @@
 import { withSql } from '../sql.ts';
-import { assertGreaterOrEqual } from 'jsr:@std/assert';
+import { assertEquals, assertGreaterOrEqual } from 'jsr:@std/assert';
 import {
   sendBatch,
   seqLastValue,
   startWorker,
+  waitFor,
   waitForSeqToIncrementBy,
 } from './_helpers.ts';
 import type postgres from 'postgres';
 
+interface ConnTestResult {
+  queue_name: string;
+  status: string;
+  actual: Record<string, unknown> | null;
+  error_message: string | null;
+}
+
 async function setupTest(sql: postgres.Sql, queueName: string) {
   await sql`CREATE SEQUENCE IF NOT EXISTS conn_test_seq`;
   await sql`ALTER SEQUENCE conn_test_seq RESTART WITH 1`;
@@ -24,6 +32,53 @@ async function setupTest(sql: postgres.Sql, queueName: string) {
   `;
 }
 
+async function setupMaxPgConnectionsTest(sql: postgres.Sql, queueName: string) {
+  // Create results table if not exists (generic, reusable for other tests)
+  await sql`
+    CREATE TABLE IF NOT EXISTS e2e_test_results (
+      id SERIAL PRIMARY KEY,
+      queue_name TEXT NOT NULL,
+      status TEXT NOT NULL,
+      actual JSONB,
+      error_message TEXT,
+      created_at TIMESTAMPTZ DEFAULT NOW()
+    )
+  `;
+  // Clear previous results for this queue
+  await sql`DELETE FROM e2e_test_results WHERE queue_name = ${queueName}`;
+  // Standard queue setup
+  await sql`
+    SELECT * FROM pgmq.drop_queue(${queueName})
+    WHERE EXISTS (
+      SELECT 1 FROM pgmq.list_queues() WHERE queue_name = ${queueName}
+    )
+  `;
+  await sql`SELECT pgmq.create(${queueName})`;
+  await sql`
+    DELETE FROM pgflow.workers
+    WHERE last_heartbeat_at < NOW() - INTERVAL '6 seconds'
+  `;
+}
+
+async function waitForTestResult(
+  sql: postgres.Sql,
+  queueName: string,
+  timeoutMs = 10000
+): Promise<ConnTestResult> {
+  return await waitFor(
+    async () => {
+      const rows = await sql<ConnTestResult[]>`
+        SELECT queue_name, status, actual, error_message
+        FROM e2e_test_results
+        WHERE queue_name = ${queueName}
+        LIMIT 1
+      `;
+      return rows.length > 0 ? rows[0] : false;
+    },
+    { timeoutMs, description: `test result for ${queueName}` }
+  );
+}
+
 Deno.test(
   {
     name: 'connection config - zero config uses Docker pooler',
@@ -112,3 +167,51 @@ Deno.test(
     });
   }
 );
+
+Deno.test(
+  {
+    name: 'connection config - default maxPgConnections is 4',
+    sanitizeOps: false,
+    sanitizeResources: false,
+  },
+  async () => {
+    await withSql(async (sql) => {
+      const queueName = 'conn_max_pg_default';
+      await setupMaxPgConnectionsTest(sql, queueName);
+      await startWorker(queueName);
+      await sendBatch(1, queueName);
+
+      const result = await waitForTestResult(sql, queueName);
+      assertEquals(
+        result.status,
+        'success',
+        `Expected success but got error: ${result.error_message} (actual=${JSON.stringify(result.actual)})`
+      );
+      assertEquals(result.actual?.max, 4);
+    });
+  }
+);
+
+Deno.test(
+  {
+    name: 'connection config - maxPgConnections override works',
+    sanitizeOps: false,
+    sanitizeResources: false,
+  },
+  async () => {
+    await withSql(async (sql) => {
+      const queueName = 'conn_max_pg_override';
+      await setupMaxPgConnectionsTest(sql, queueName);
+      await startWorker(queueName);
+      await sendBatch(1, queueName);
+
+      const result = await waitForTestResult(sql, queueName);
+      assertEquals(
+        result.status,
+        'success',
+        `Expected success but got error: ${result.error_message} (actual=${JSON.stringify(result.actual)})`
+      );
+      assertEquals(result.actual?.max, 7);
+    });
+  }
+);

pkgs/edge-worker/tests/e2e/connection-config.test.ts

@@ -219,3 +219,21 @@ Deno.test('resolveSqlConnection - sql takes priority over everything', () => {
   assertEquals(result, mockSql);
   mockSql.end();
 });
+
+// ============================================================
+// maxPgConnections tests
+// ============================================================
+
+Deno.test('resolveSqlConnection - uses default maxPgConnections of 4', () => {
+  const env = { SUPABASE_URL: LOCAL_SUPABASE_URL };
+  const result = resolveSqlConnection(env);
+  assertEquals(result.options.max, 4);
+  result.end();
+});
+
+Deno.test('resolveSqlConnection - respects maxPgConnections override', () => {
+  const env = { SUPABASE_URL: LOCAL_SUPABASE_URL };
+  const result = resolveSqlConnection(env, { maxPgConnections: 7 });
+  assertEquals(result.options.max, 7);
+  result.end();
+});