fix(api): allow disabling named Postgres prepared statements (#2251)

deploy/self-hosting/.env.example

@@ -113,3 +113,10 @@ FLUXER_DISCOVERY_ENABLED=true
 # from starting far more schedulers than the container can actually use.
 #FLUXER_ERLANG_SCHEDULERS_MIN=2
 #FLUXER_ERLANG_SCHEDULERS_MAX=16
+
+# The api names its fixed Postgres statement shapes so the server can reuse their
+# plans. Named prepared statements require a session that outlives the
+# transaction, so set this to false if you put a transaction-pooling connection
+# pooler such as PgBouncer in front of Postgres. The bundled compose talks to
+# Postgres directly, where naming is a win and the default is correct.
+#FLUXER_POSTGRES_PREPARED_STATEMENTS=true

deploy/self-hosting/docker-compose.yml

@@ -17,6 +17,7 @@ x-fluxer-env: &fluxer-env
   FLUXER_POSTGRES_USERNAME: fluxer
   FLUXER_POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}
   FLUXER_POSTGRES_SSL: "false"
+  FLUXER_POSTGRES_PREPARED_STATEMENTS: ${FLUXER_POSTGRES_PREPARED_STATEMENTS:-true}
 
   FLUXER_KV_URL: redis://valkey:6379/0
   FLUXER_NATS_URL: nats://nats:4222

fluxer_api/pkgs/postgres/src/Client.ts

@@ -14,6 +14,7 @@ interface PostgresConfig {
 	sslCa?: string;
 	maxConnections?: number;
 	kvTable?: string;
+	preparedStatements?: boolean;
 }
 
 export interface PostgresQueryable {
@@ -98,14 +99,18 @@ class PostgresClient implements IPostgresClient {
 		values: Array<unknown> = [],
 		name?: string,
 	): Promise<QueryResult<T>> {
-		return this.getPool().query<T>({text, values, name});
+		return this.getPool().query<T>({text, values, name: this.statementName(name)});
+	}
+
+	private statementName(name: string | undefined): string | undefined {
+		return this.config.preparedStatements === false ? undefined : name;
 	}
 
 	async transaction<T>(fn: (client: PostgresQueryable) => Promise<T>): Promise<T> {
 		const client = await this.getPool().connect();
 		try {
 			await client.query('BEGIN');
-			const result = await fn(poolClientQueryable(client));
+			const result = await fn(poolClientQueryable(client, this.config.preparedStatements !== false));
 			await client.query('COMMIT');
 			return result;
 		} catch (error) {
@@ -128,10 +133,10 @@ class PostgresClient implements IPostgresClient {
 	}
 }
 
-function poolClientQueryable(client: PoolClient): PostgresQueryable {
+function poolClientQueryable(client: PoolClient, preparedStatements: boolean): PostgresQueryable {
 	return {
 		query: <T extends QueryResultRow = QueryResultRow>(text: string, values: Array<unknown> = [], name?: string) =>
-			client.query<T>({text, values, name}),
+			client.query<T>({text, values, name: preparedStatements ? name : undefined}),
 	};
 }
 

fluxer_api/src/api/Config.ts

@@ -185,6 +185,7 @@ export function buildAPIConfigFromMaster(master: MasterConfig): APIConfig {
 			sslCa: postgresSource?.ssl_ca ?? '',
 			maxConnections: postgresSource?.max_connections ?? 20,
 			kvTable: postgresSource?.kv_table ?? 'fluxer_kv',
+			preparedStatements: postgresSource?.prepared_statements ?? true,
 		},
 		database: {
 			backend: master.database.backend,

fluxer_api/src/api/config/APIConfig.ts

@@ -56,6 +56,7 @@ export interface APIConfig {
 		sslCa: string;
 		maxConnections: number;
 		kvTable: string;
+		preparedStatements: boolean;
 	};
 	database: {
 		backend: 'cassandra' | 'postgres';

fluxer_api/src/api/database/PostgresKvStatementNames.test.ts

@@ -21,6 +21,7 @@ interface Statement {
 }
 
 const KV_TABLE = 'kv_stmt_names';
+const KV_TABLE_POOLED = 'kv_stmt_names_pooled';
 const CONTAINER = `fluxer-kvstmt-${process.pid.toString(36)}-${Date.now().toString(36)}`;
 const dockerAvailable = spawnSync('docker', ['version'], {stdio: 'ignore'}).status === 0;
 
@@ -171,60 +172,7 @@ describe('PostgresKvQueryExecutor statement names', () => {
 	});
 });
 
-describe.skipIf(!dockerAvailable)('PostgresKvQueryExecutor statement names against postgres', () => {
+async function exerciseKvShapes(executor: PostgresKvQueryExecutor): Promise<void> {
-	let raw: IPostgresClient;
-	let executor: PostgresKvQueryExecutor;
-
-	beforeAll(async () => {
-		const port = await freePort();
-		startDockerContainer([
-			'run',
-			'-d',
-			'--name',
-			CONTAINER,
-			'-e',
-			'POSTGRES_USER=fluxer',
-			'-e',
-			'POSTGRES_PASSWORD=fluxer',
-			'-e',
-			'POSTGRES_DB=fluxer',
-			'-p',
-			`127.0.0.1:${port}:5432`,
-			'postgres:16-alpine',
-			'-c',
-			'fsync=off',
-		]);
-		let ready = false;
-		for (let attempt = 0; attempt < 180 && !ready; attempt += 1) {
-			await sleep(500);
-			const probe = spawnSync('docker', ['exec', CONTAINER, 'pg_isready', '-U', 'fluxer', '-d', 'fluxer'], {
-				stdio: 'ignore',
-			});
-			if (probe.status !== 0) continue;
-			try {
-				await initPostgres({
-					url: `postgres://fluxer:[email protected]:${port}/fluxer`,
-					maxConnections: 1,
-					kvTable: KV_TABLE,
-				});
-				await getDefaultPostgresClient().query('SELECT 1');
-				ready = true;
-			} catch {
-				await shutdownPostgres().catch(() => {});
-			}
-		}
-		if (!ready) throw new Error('postgres never came up');
-		raw = getDefaultPostgresClient();
-		await ensurePostgresKvSchema(raw);
-		executor = new PostgresKvQueryExecutor(raw);
-	}, 900_000);
-
-	afterAll(async () => {
-		await shutdownPostgres().catch(() => {});
-		spawnSync('docker', ['rm', '-f', CONTAINER], {stdio: 'ignore'});
-	});
-
-	it('prepares each named shape server side and keeps reading the right rows', async () => {
 	for (let index = 0; index < 4; index += 1) {
 		await executor.executeQuery({
 			cql: '__stmt_seed',
@@ -294,6 +242,82 @@ describe.skipIf(!dockerAvailable)('PostgresKvQueryExecutor statement names again
 		kvMeta: meta(Composite, 'select', [eq('owner_id')]) as KvQueryMeta,
 	});
 	expect(remaining).toHaveLength(1);
+	await executor.executeBatch([
+		{
+			query: '__stmt_batch',
+			params: {owner_id: 'o7', item_id: 'i7', payload: 'batched'},
+			meta: meta(Composite, 'upsert', []) as KvQueryMeta,
+		},
+		{
+			query: '__stmt_batch',
+			params: {owner_id: 'o7', item_id: 'i8', payload: 'batched'},
+			meta: meta(Composite, 'upsert', []) as KvQueryMeta,
+		},
+	]);
+	const batched = await executor.executeQuery<Row>({
+		cql: '__stmt_range',
+		params: {owner_id: 'o7'} as CassandraParams,
+		kvMeta: meta(Composite, 'select', [eq('owner_id')]) as KvQueryMeta,
+	});
+	expect(batched.map((row) => row.payload)).toEqual(['batched', 'batched']);
+}
+
+describe.skipIf(!dockerAvailable)('PostgresKvQueryExecutor statement names against postgres', () => {
+	let raw: IPostgresClient;
+	let executor: PostgresKvQueryExecutor;
+	let port: number;
+
+	beforeAll(async () => {
+		port = await freePort();
+		startDockerContainer([
+			'run',
+			'-d',
+			'--name',
+			CONTAINER,
+			'-e',
+			'POSTGRES_USER=fluxer',
+			'-e',
+			'POSTGRES_PASSWORD=fluxer',
+			'-e',
+			'POSTGRES_DB=fluxer',
+			'-p',
+			`127.0.0.1:${port}:5432`,
+			'postgres:16-alpine',
+			'-c',
+			'fsync=off',
+		]);
+		let ready = false;
+		for (let attempt = 0; attempt < 180 && !ready; attempt += 1) {
+			await sleep(500);
+			const probe = spawnSync('docker', ['exec', CONTAINER, 'pg_isready', '-U', 'fluxer', '-d', 'fluxer'], {
+				stdio: 'ignore',
+			});
+			if (probe.status !== 0) continue;
+			try {
+				await initPostgres({
+					url: `postgres://fluxer:[email protected]:${port}/fluxer`,
+					maxConnections: 1,
+					kvTable: KV_TABLE,
+				});
+				await getDefaultPostgresClient().query('SELECT 1');
+				ready = true;
+			} catch {
+				await shutdownPostgres().catch(() => {});
+			}
+		}
+		if (!ready) throw new Error('postgres never came up');
+		raw = getDefaultPostgresClient();
+		await ensurePostgresKvSchema(raw);
+		executor = new PostgresKvQueryExecutor(raw);
+	}, 900_000);
+
+	afterAll(async () => {
+		await shutdownPostgres().catch(() => {});
+		spawnSync('docker', ['rm', '-f', CONTAINER], {stdio: 'ignore'});
+	});
+
+	it('prepares each named shape server side and keeps reading the right rows', async () => {
+		await exerciseKvShapes(executor);
 		const prepared = await raw.query<{name: string}>('SELECT name FROM pg_prepared_statements ORDER BY name');
 		expect(prepared.rows.map((row) => row.name)).toEqual([
 			'kv_del_expired',
@@ -306,4 +330,18 @@ describe.skipIf(!dockerAvailable)('PostgresKvQueryExecutor statement names again
 			'kv_upsert',
 		]);
 	});
+
+	it('prepares nothing server side when prepared statements are disabled', async () => {
+		await initPostgres({
+			url: `postgres://fluxer:[email protected]:${port}/fluxer`,
+			maxConnections: 1,
+			kvTable: KV_TABLE_POOLED,
+			preparedStatements: false,
+		});
+		const pooled = getDefaultPostgresClient();
+		await ensurePostgresKvSchema(pooled);
+		await exerciseKvShapes(new PostgresKvQueryExecutor(pooled));
+		const prepared = await pooled.query<{name: string}>('SELECT name FROM pg_prepared_statements ORDER BY name');
+		expect(prepared.rows).toEqual([]);
+	});
 });

packages/config/src/ConfigLoader.ts

@@ -71,6 +71,7 @@ function defaultConfig(): MasterConfig {
 				ssl_ca: '',
 				max_connections: 20,
 				kv_table: 'fluxer_kv',
+				prepared_statements: true,
 			},
 		},
 		s3: {
@@ -356,6 +357,7 @@ function validatePostgresConfig(config: MasterConfig): void {
 	assertIntegerInRange(postgres.max_connections, 'FLUXER_POSTGRES_MAX_CONNECTIONS', 1, 1000);
 	assertBoolean(postgres.ssl, 'FLUXER_POSTGRES_SSL');
 	assertIdentifier(postgres.kv_table, 'FLUXER_POSTGRES_KV_TABLE');
+	assertBoolean(postgres.prepared_statements, 'FLUXER_POSTGRES_PREPARED_STATEMENTS');
 	if (config.env !== 'production' || config.database.backend !== 'postgres') {
 		return;
 	}

packages/config/src/MasterConfig.ts

@@ -61,6 +61,7 @@ export interface MasterConfig {
 			ssl_ca: string;
 			max_connections: number;
 			kv_table: string;
+			prepared_statements: boolean;
 		};
 	};
 	s3?: {

packages/config/src/__tests__/ConfigLoader.test.ts

@@ -110,6 +110,7 @@ describe('ConfigLoader', () => {
 			FLUXER_POSTGRES_PORT: '5544',
 			FLUXER_POSTGRES_MAX_CONNECTIONS: '7',
 			FLUXER_POSTGRES_SSL_CA: '-----BEGIN CERTIFICATE-----\\n-----END CERTIFICATE-----',
+			FLUXER_POSTGRES_PREPARED_STATEMENTS: 'false',
 			FLUXER_API_WORKER_MODE: 'single_task',
 			FLUXER_API_WORKER_TASK: 'processStripeWebhook',
 			FLUXER_GATEWAY_PUSH_ENABLED: 'false',
@@ -127,6 +128,7 @@ describe('ConfigLoader', () => {
 		expect(config.database.postgres.port).toBe(5544);
 		expect(config.database.postgres.max_connections).toBe(7);
 		expect(config.database.postgres.ssl_ca).toBe('-----BEGIN CERTIFICATE-----\\n-----END CERTIFICATE-----');
+		expect(config.database.postgres.prepared_statements).toBe(false);
 		expect(config.services.api.worker?.mode).toBe('single_task');
 		expect(config.services.api.worker?.task).toBe('processStripeWebhook');
 		expect(config.services.gateway.push_enabled).toBe(false);
@@ -148,6 +150,16 @@ describe('ConfigLoader', () => {
 		await expect(loadConfig()).rejects.toThrow('FLUXER_POSTGRES_PORT');
 	});
 
+	test('keeps Postgres prepared statements on by default', async () => {
+		stubMinimalEnv();
+		expect((await loadConfig()).database.postgres.prepared_statements).toBe(true);
+	});
+
+	test('rejects a non-boolean Postgres prepared statements value', async () => {
+		stubMinimalEnv({FLUXER_POSTGRES_PREPARED_STATEMENTS: 'maybe'});
+		await expect(loadConfig()).rejects.toThrow('FLUXER_POSTGRES_PREPARED_STATEMENTS');
+	});
+
 	test('rejects unsafe production Postgres defaults', async () => {
 		stubMinimalEnv({FLUXER_ENV: 'production'});
 		await expect(loadConfig()).rejects.toThrow('FLUXER_POSTGRES_HOST');

packages/config/src/config_loader/EnvironmentOverrides.ts

@@ -47,6 +47,7 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record<string, NamedEnvOverride> = {
 	FLUXER_POSTGRES_SSL_CA: {path: ['database', 'postgres', 'ssl_ca']},
 	FLUXER_POSTGRES_MAX_CONNECTIONS: {path: ['database', 'postgres', 'max_connections'], parse: parseEnvValue},
 	FLUXER_POSTGRES_KV_TABLE: {path: ['database', 'postgres', 'kv_table']},
+	FLUXER_POSTGRES_PREPARED_STATEMENTS: {path: ['database', 'postgres', 'prepared_statements'], parse: parseEnvValue},
 	FLUXER_DATABASE_BACKEND: {path: ['database', 'backend']},
 	FLUXER_KV_URL: {path: ['internal', 'kv']},
 	FLUXER_INTERNAL_API_ENDPOINT: {path: ['internal', 'api']},