diff --git a/deploy/self-hosting/.env.example b/deploy/self-hosting/.env.example index 3944df993..f2b8ceed5 100644 --- a/deploy/self-hosting/.env.example +++ b/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 diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index 6facc4069..a5ba68379 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/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 diff --git a/fluxer_api/pkgs/postgres/src/Client.ts b/fluxer_api/pkgs/postgres/src/Client.ts index d488da930..fabdbd511 100644 --- a/fluxer_api/pkgs/postgres/src/Client.ts +++ b/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 = [], name?: string, ): Promise> { - return this.getPool().query({text, values, name}); + return this.getPool().query({text, values, name: this.statementName(name)}); + } + + private statementName(name: string | undefined): string | undefined { + return this.config.preparedStatements === false ? undefined : name; } async transaction(fn: (client: PostgresQueryable) => Promise): Promise { 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: (text: string, values: Array = [], name?: string) => - client.query({text, values, name}), + client.query({text, values, name: preparedStatements ? name : undefined}), }; } diff --git a/fluxer_api/src/api/Config.ts b/fluxer_api/src/api/Config.ts index 272c26b32..4fe027d29 100644 --- a/fluxer_api/src/api/Config.ts +++ b/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, diff --git a/fluxer_api/src/api/config/APIConfig.ts b/fluxer_api/src/api/config/APIConfig.ts index 8d498b992..f3bce6abc 100644 --- a/fluxer_api/src/api/config/APIConfig.ts +++ b/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'; diff --git a/fluxer_api/src/api/database/PostgresKvStatementNames.test.ts b/fluxer_api/src/api/database/PostgresKvStatementNames.test.ts index 76f8527ff..ab233836a 100644 --- a/fluxer_api/src/api/database/PostgresKvStatementNames.test.ts +++ b/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,12 +172,103 @@ describe('PostgresKvQueryExecutor statement names', () => { }); }); +async function exerciseKvShapes(executor: PostgresKvQueryExecutor): Promise { + for (let index = 0; index < 4; index += 1) { + await executor.executeQuery({ + cql: '__stmt_seed', + params: {owner_id: `o${index % 2}`, item_id: `i${index}`, payload: `p${index}`} as CassandraParams, + kvMeta: meta(Composite, 'upsert', []) as KvQueryMeta, + }); + } + for (let index = 0; index < 12; index += 1) { + const point = await executor.executeQuery({ + cql: '__stmt_point', + params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, + kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, + }); + expect(point.map((row) => row.payload)).toEqual(['p0']); + const range = await executor.executeQuery({ + cql: '__stmt_range', + params: {owner_id: 'o0'} as CassandraParams, + kvMeta: meta(Composite, 'select', [eq('owner_id')]) as KvQueryMeta, + }); + expect(range).toHaveLength(2); + } + await executor.executeQuery({ + cql: '__stmt_patch', + params: {owner_id: 'o0', item_id: 'i0', payload: 'patched'} as CassandraParams, + kvMeta: meta(Composite, 'patch', [eq('owner_id'), eq('item_id')], {patchKeys: ['payload']}) as KvQueryMeta, + }); + const reapplied = await executor.executeQuery({ + cql: '__stmt_lwt', + params: {owner_id: 'o0', item_id: 'i0', payload: 'ignored'} as CassandraParams, + kvMeta: meta(Composite, 'upsert', [], {ifNotExists: true}) as KvQueryMeta, + }); + expect(reapplied).toEqual([{'[applied]': false}]); + const claimed = await executor.executeQuery({ + cql: '__stmt_lwt', + params: {owner_id: 'o9', item_id: 'i9', payload: 'claimed'} as CassandraParams, + kvMeta: meta(Composite, 'upsert', [], {ifNotExists: true}) as KvQueryMeta, + }); + expect(claimed).toEqual([{'[applied]': true}]); + await executor.executeQuery({ + cql: '__stmt_patch_ttl', + params: {owner_id: 'o9', item_id: 'i9', payload: 'expiring', ttl_: 600} as CassandraParams, + kvMeta: meta(Composite, 'patch', [eq('owner_id'), eq('item_id')], { + patchKeys: ['payload'], + ttlParamName: 'ttl_', + }) as KvQueryMeta, + }); + const expiring = await executor.executeQuery({ + cql: '__stmt_point', + params: {owner_id: 'o9', item_id: 'i9'} as CassandraParams, + kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, + }); + expect(expiring.map((row) => row.payload)).toEqual(['expiring']); + const patched = await executor.executeQuery({ + cql: '__stmt_point', + params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, + kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, + }); + expect(patched.map((row) => row.payload)).toEqual(['patched']); + await executor.executeQuery({ + cql: '__stmt_delete', + params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, + kvMeta: meta(Composite, 'delete', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, + }); + const remaining = await executor.executeQuery({ + cql: '__stmt_range', + params: {owner_id: 'o0'} as CassandraParams, + 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({ + 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 () => { - const port = await freePort(); + port = await freePort(); startDockerContainer([ 'run', '-d', @@ -225,75 +317,7 @@ describe.skipIf(!dockerAvailable)('PostgresKvQueryExecutor statement names again }); 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', - params: {owner_id: `o${index % 2}`, item_id: `i${index}`, payload: `p${index}`} as CassandraParams, - kvMeta: meta(Composite, 'upsert', []) as KvQueryMeta, - }); - } - for (let index = 0; index < 12; index += 1) { - const point = await executor.executeQuery({ - cql: '__stmt_point', - params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, - kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, - }); - expect(point.map((row) => row.payload)).toEqual(['p0']); - const range = await executor.executeQuery({ - cql: '__stmt_range', - params: {owner_id: 'o0'} as CassandraParams, - kvMeta: meta(Composite, 'select', [eq('owner_id')]) as KvQueryMeta, - }); - expect(range).toHaveLength(2); - } - await executor.executeQuery({ - cql: '__stmt_patch', - params: {owner_id: 'o0', item_id: 'i0', payload: 'patched'} as CassandraParams, - kvMeta: meta(Composite, 'patch', [eq('owner_id'), eq('item_id')], {patchKeys: ['payload']}) as KvQueryMeta, - }); - const reapplied = await executor.executeQuery({ - cql: '__stmt_lwt', - params: {owner_id: 'o0', item_id: 'i0', payload: 'ignored'} as CassandraParams, - kvMeta: meta(Composite, 'upsert', [], {ifNotExists: true}) as KvQueryMeta, - }); - expect(reapplied).toEqual([{'[applied]': false}]); - const claimed = await executor.executeQuery({ - cql: '__stmt_lwt', - params: {owner_id: 'o9', item_id: 'i9', payload: 'claimed'} as CassandraParams, - kvMeta: meta(Composite, 'upsert', [], {ifNotExists: true}) as KvQueryMeta, - }); - expect(claimed).toEqual([{'[applied]': true}]); - await executor.executeQuery({ - cql: '__stmt_patch_ttl', - params: {owner_id: 'o9', item_id: 'i9', payload: 'expiring', ttl_: 600} as CassandraParams, - kvMeta: meta(Composite, 'patch', [eq('owner_id'), eq('item_id')], { - patchKeys: ['payload'], - ttlParamName: 'ttl_', - }) as KvQueryMeta, - }); - const expiring = await executor.executeQuery({ - cql: '__stmt_point', - params: {owner_id: 'o9', item_id: 'i9'} as CassandraParams, - kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, - }); - expect(expiring.map((row) => row.payload)).toEqual(['expiring']); - const patched = await executor.executeQuery({ - cql: '__stmt_point', - params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, - kvMeta: meta(Composite, 'select', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, - }); - expect(patched.map((row) => row.payload)).toEqual(['patched']); - await executor.executeQuery({ - cql: '__stmt_delete', - params: {owner_id: 'o0', item_id: 'i0'} as CassandraParams, - kvMeta: meta(Composite, 'delete', [eq('owner_id'), eq('item_id')]) as KvQueryMeta, - }); - const remaining = await executor.executeQuery({ - cql: '__stmt_range', - params: {owner_id: 'o0'} as CassandraParams, - kvMeta: meta(Composite, 'select', [eq('owner_id')]) as KvQueryMeta, - }); - expect(remaining).toHaveLength(1); + 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:fluxer@127.0.0.1:${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([]); + }); }); diff --git a/packages/config/src/ConfigLoader.ts b/packages/config/src/ConfigLoader.ts index 6fe38c39b..99e7377f5 100644 --- a/packages/config/src/ConfigLoader.ts +++ b/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; } diff --git a/packages/config/src/MasterConfig.ts b/packages/config/src/MasterConfig.ts index c29cfbd66..a2c268d1f 100644 --- a/packages/config/src/MasterConfig.ts +++ b/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?: { diff --git a/packages/config/src/__tests__/ConfigLoader.test.ts b/packages/config/src/__tests__/ConfigLoader.test.ts index da9dc9c8c..87bdba516 100644 --- a/packages/config/src/__tests__/ConfigLoader.test.ts +++ b/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'); diff --git a/packages/config/src/config_loader/EnvironmentOverrides.ts b/packages/config/src/config_loader/EnvironmentOverrides.ts index 764b96c0e..a3a7ac9f8 100644 --- a/packages/config/src/config_loader/EnvironmentOverrides.ts +++ b/packages/config/src/config_loader/EnvironmentOverrides.ts @@ -47,6 +47,7 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record = { 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']},