diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 0000000..cee520b --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,33 @@ +# Copilot review instructions + +`macpaw/postgres-schema-bundle` is a public Symfony bundle. It switches the PostgreSQL `search_path` per request +through a Doctrine DBAL wrapper connection (`SchemaConnection`). Services configure it from `Kernel::boot()`. + +## Compatibility + +- PHP `>=8.3`, Doctrine DBAL `^3.4`, ORM `^2.17 || ^3.0`, Symfony `^6.4 || ^7.4 || ^8.0`. +- DBAL 3 driver interfaces declare return types in docblocks only. Native return types that narrow + them in subclasses are fine; their absence is not a fatal error. + +## Behaviour that must not change + +- `SearchPathMode::Session` is the default. With it, every statement must pass through exactly as before + the transaction mode existed. Flag any change that touches statements in Session mode. +- In `SearchPathMode::Transaction` the schema is applied only with `SET LOCAL`, never with a session `SET`. + A session `SET` leaks to other clients behind PgBouncer in transaction pooling. +- Transaction control SQL (`BEGIN`, `COMMIT`, `ROLLBACK`, savepoints) must run without a `SET LOCAL` before it. + +## Intentional decisions (do not flag) + +- `BEGIN; SET LOCAL search_path TO ...` is sent in one `exec()` to save a round trip. Only `pdo_pgsql` is + supported, and the native `pgsql` driver is rejected. +- Mode, resolver and logger are static setters, matching how the bundle has always been configured. +- Statements outside a transaction commit before the result is fetched; this relies on `pdo_pgsql` buffering. + +## Code and tests + +- `declare(strict_types=1)`, PSR-12, max line length 120, PHPStan level max. +- Unit tests in `tests/Doctrine` and `tests/Command` use mocks. Integration tests in `tests/Integration` need + PostgreSQL behind PgBouncer in transaction pooling (`docker compose`, or the CI services) and must fail, + not skip, in CI. +- Every change to `src/Doctrine/Middleware` needs an integration test that fails without it. diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 0564467..f182ded 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -8,6 +8,8 @@ on: jobs: run: runs-on: ubuntu-latest + permissions: + contents: read strategy: fail-fast: false matrix: @@ -47,6 +49,48 @@ jobs: doctrine-migrations: '^3.6' doctrine-fixtures-bundle: '^3.5' coverage: xdebug + - description: 'PostgreSQL 18' + php: '8.4' + symfony-versions: '7.4.*' + doctrine-orm-versions: '^3.0' + doctrine-migrations: '3.9.*' + doctrine-fixtures-bundle: '4.0.*' + coverage: 'none' + postgres: '18' + + services: + postgres: + image: postgres:${{ matrix.postgres || '14' }}-alpine + env: + POSTGRES_USER: app + POSTGRES_PASSWORD: app + POSTGRES_DB: app + POSTGRES_HOST_AUTH_METHOD: scram-sha-256 + POSTGRES_INITDB_ARGS: --auth=scram-sha-256 + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U app" + --health-interval 2s + --health-timeout 5s + --health-retries 20 + pgbouncer: + image: edoburu/pgbouncer:v1.24.1-p1 + env: + DB_HOST: postgres + DB_USER: app + DB_PASSWORD: app + DB_NAME: app + AUTH_TYPE: scram-sha-256 + POOL_MODE: transaction + DEFAULT_POOL_SIZE: 1 + MAX_PREPARED_STATEMENTS: 200 + ports: + - 6432:5432 + + env: + DIRECT_DATABASE_URL: pgsql://app:app@127.0.0.1:5432/app + PGBOUNCER_DATABASE_URL: pgsql://app:app@127.0.0.1:6432/app name: PHP ${{ matrix.php }} Symfony ${{ matrix.symfony-versions }} Doctrine ${{ matrix.doctrine-orm-versions }} Doctrine migrations ${{ matrix.doctrine-migrations }} Doctrine fixtures bundle ${{ matrix.doctrine-fixtures-bundle }} ${{ matrix.description }} steps: @@ -62,6 +106,7 @@ jobs: uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php }} + extensions: pdo_pgsql coverage: xdebug - name: Add PHPUnit matcher @@ -90,16 +135,82 @@ jobs: - name: Install dependencies run: composer install + - name: Wait for PgBouncer + run: timeout 60 sh -c 'until pg_isready -h 127.0.0.1 -p 6432 -U app; do sleep 1; done' + - name: Run PHPUnit tests - run: vendor/bin/phpunit + run: vendor/bin/phpunit --fail-on-skipped if: matrix.coverage == 'none' - name: PHPUnit tests and Log Code coverage - run: vendor/bin/phpunit --coverage-clover=coverage.xml + run: vendor/bin/phpunit --fail-on-skipped --coverage-clover=coverage.xml if: matrix.coverage == 'xdebug' + - name: Generate code coverage status + id: code-coverage-analyze + if: matrix.coverage == 'xdebug' + run: | + exit_code=0 + coverage_result="$(vendor/bin/coverage-check coverage.xml "${MIN_CODE_COVERAGE}")" || exit_code=$? + echo "${coverage_result}" + echo "analyze-text=${coverage_result}" >> "${GITHUB_OUTPUT}" + exit "${exit_code}" + env: + MIN_CODE_COVERAGE: 80 + + - name: Prepare code coverage comment + if: always() && matrix.coverage == 'xdebug' && steps.code-coverage-analyze.outputs.analyze-text + run: | + { + echo "[postgres-schema-bundle] (PHP ${MATRIX_PHP}, Symfony ${MATRIX_SYMFONY}) ${STATUS_ICON} ${ANALYZE_TEXT}" + echo "${STATUS_TEXT}" + } > coverage-comment.md + env: + MATRIX_PHP: ${{ matrix.php }} + MATRIX_SYMFONY: ${{ matrix.symfony-versions }} + ANALYZE_TEXT: ${{ steps.code-coverage-analyze.outputs.analyze-text }} + STATUS_ICON: ${{ steps.code-coverage-analyze.outcome == 'success' && '✅' || '❌' }} + STATUS_TEXT: >- + ${{ steps.code-coverage-analyze.outcome == 'success' + && 'Well done, sailor! 🚣' + || 'Please add more tests to your functionality :man_shrugging:' }} + + - name: Upload code coverage comment + if: always() && matrix.coverage == 'xdebug' && steps.code-coverage-analyze.outputs.analyze-text + uses: actions/upload-artifact@v4 + with: + name: coverage-comment + path: coverage-comment.md + retention-days: 1 + - name: Upload coverage reports to Codecov if: matrix.coverage == 'xdebug' uses: codecov/codecov-action@v4.0.1 with: token: ${{ secrets.CODECOV_TOKEN }} + + coverage-comment: + name: Code coverage comment + needs: run + if: >- + always() + && github.event_name == 'pull_request' + && github.event.pull_request.head.repo.full_name == github.repository + runs-on: ubuntu-latest + permissions: + pull-requests: write + steps: + - name: Download code coverage comment + id: download + continue-on-error: true + uses: actions/download-artifact@v4 + with: + name: coverage-comment + + - name: Create code coverage comment + if: steps.download.outcome == 'success' + uses: mshick/add-pr-comment@ec328af66588ab8f77cdeb2c264f14aba45bbf59 # v3.12.0 + with: + message-id: code-coverage-comment + refresh-message-position: true + message-path: coverage-comment.md diff --git a/.github/workflows/copilot-code-review.yml b/.github/workflows/copilot-code-review.yml new file mode 100644 index 0000000..bad1983 --- /dev/null +++ b/.github/workflows/copilot-code-review.yml @@ -0,0 +1,25 @@ +name: Copilot code review runner + +on: + workflow_dispatch: + +permissions: + contents: read + +jobs: + copilot-setup-steps: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + php-version: '8.3' + extensions: pdo_pgsql + coverage: none + + - name: Install dependencies + run: composer install --no-interaction --no-progress diff --git a/.github/workflows/security.yaml b/.github/workflows/security.yaml index 248d5d0..4cad68c 100644 --- a/.github/workflows/security.yaml +++ b/.github/workflows/security.yaml @@ -9,7 +9,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Setup PHP uses: shivammathur/setup-php@v2 diff --git a/.github/workflows/static-analysis.yaml b/.github/workflows/static-analysis.yaml index 8796ae1..209e35a 100644 --- a/.github/workflows/static-analysis.yaml +++ b/.github/workflows/static-analysis.yaml @@ -11,7 +11,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Setup PHP uses: shivammathur/setup-php@v2 @@ -27,7 +27,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Setup PHP uses: shivammathur/setup-php@v2 @@ -43,7 +43,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Setup PHP uses: shivammathur/setup-php@v2 diff --git a/README.md b/README.md index 6b15856..cc12b99 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,81 @@ class Kernel extends BaseKernel } ``` +### Search path mode (PgBouncer transaction pooling) + +By default the bundle sets `search_path` once per database session (`SET search_path TO ...`). +This is safe only with a direct connection or PgBouncer in **session** pooling mode. +Behind PgBouncer in **transaction** pooling mode a session setting stays on the pooled server connection +and leaks to other clients: their queries run in another schema. + +Switch to the transaction mode at kernel boot, next to `setSchemaResolver()`: + +```php +use Macpaw\PostgresSchemaBundle\Doctrine\SchemaConnection; +use Macpaw\PostgresSchemaBundle\Doctrine\SearchPathMode; + +SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); +``` + +`SchemaConnection::setSearchPathMode(SearchPathMode $mode, ?string $serverDefaultSchema = null)`: + +| Argument | Value | Behaviour | +|---|---|---| +| `$mode` | `SearchPathMode::Session` (default) | `SET search_path` once per connection; current behaviour | +| | `SearchPathMode::Transaction` | `SET LOCAL search_path` inside every transaction; the setting never outlives the transaction | +| `$serverDefaultSchema` | `null` (strict, default) | `SET LOCAL` for every schema, including the default one | +| | `'public'` or another schema | queries for this schema run without `SET LOCAL` and rely on the server default `search_path` | + +In transaction mode: + +- A statement outside a transaction runs as `BEGIN; SET LOCAL search_path TO ""`, the statement, then `COMMIT` + (`ROLLBACK` on error): two extra round trips. +- Inside a transaction `SET LOCAL` runs once after `BEGIN`, and again only when the schema changes or after + `ROLLBACK TO SAVEPOINT`. +- Plain queries, executed statements and prepared statements (Doctrine ORM) are all covered. +- The schema commands below only set the schema in the resolver; they do not issue a session `SET`. +- Manage transactions through the DBAL API: `beginTransaction()`, `commit()`, `rollBack()`, `transactional()`. + Raw `BEGIN`, `START TRANSACTION`, `COMMIT`, `ROLLBACK`, `END`, `ABORT` or `PREPARE TRANSACTION` SQL is rejected + with `TransactionSearchPathModeException`, as is any of it in a prepared statement or batched with other + statements. A single savepoint statement (`SAVEPOINT`, `RELEASE`, `ROLLBACK TO`), which DBAL uses for nested + transactions, runs as is. +- The SQL check fails closed. SQL with `;` or a leading comment that cannot be split reliably (unbalanced quotes, + comments or dollar quotes, or a PCRE limit error) is rejected; send such statements one by one. After every + statement the server transaction status (`PDO::inTransaction()`) is compared with the expected one: if transaction + control slipped past the check, the connection rolls back a transaction it left open and throws, and refuses + further statements in a DBAL transaction that the server no longer has. +- The mode is applied when a physical connection opens. An idle connection opened in the other mode is reopened; + changing the mode while a transaction is open throws `TransactionSearchPathModeException`. Set the mode in + `Kernel::boot()` before running queries: statements prepared before the switch keep the old mode. +- In the default `Session` mode the driver connection is not wrapped, and the bundle behaves exactly as before. + +Choosing `$serverDefaultSchema`: + +- Use `null` while any client of the same database still sets a session `search_path`: pods on an older bundle + version during rollout, session-mode services sharing the pool, manual `psql` sessions. + Otherwise default-schema queries inherit the leaked setting. +- Pass the schema only when every client of the database uses transaction mode and the server default + `search_path` (role or database setting) resolves to that schema. + For an environment whose default schema is not `public`, set it first with + `ALTER ROLE ... SET search_path` or `ALTER DATABASE ... SET search_path`. + +Limitations: + +- Only the `pdo_pgsql` driver is supported. It prepares a statement when it is executed, inside `SET LOCAL`. + The native `pgsql` driver prepares immediately, against the server default `search_path`, so it is rejected + with `UnsupportedDriverException`. +- `pdo_pgsql` uses server-side prepared statements, and a statement reused outside a transaction can run on + another server connection. PgBouncer must track prepared statements: version 1.21+ with + `max_prepared_statements` greater than 0 (the default since 1.22). +- A statement outside a transaction is committed before its result is fetched. This relies on `pdo_pgsql` + buffering the whole result (the default); do not enable lazy (unbuffered) result fetching. +- Statements that cannot run inside a transaction block (`CREATE INDEX CONCURRENTLY`, `VACUUM`, migrations with + `isTransactional() = false`) fail for a schema that needs `SET LOCAL`. +- `lastInsertId()` outside a transaction runs after the implicit `COMMIT`, possibly on another server connection + and without the schema `search_path`: `LASTVAL()` and `CURRVAL('')` can fail or return another value. + Fetch ids with `nextval()` (Doctrine ORM `SEQUENCE` strategy), `INSERT ... RETURNING`, or call `lastInsertId()` + inside the same explicit transaction as the insert. + Make sure you configure the context bundle properly: See https://github.com/MacPaw/schema-context-bundle/blob/develop/README.md @@ -83,7 +158,8 @@ schema_context: ## How it Works * A request comes in with a header like X-Tenant-Id: tenant123. * The SchemaRequestListener sets this schema in the context. -* When Doctrine connects to PostgreSQL, it sets the search_path to the specified schema. +* When Doctrine connects to PostgreSQL, it sets the search_path to the specified schema + (per session, or per transaction in `SearchPathMode::Transaction`). * If the schema does not exist or DB is not PostgreSQL, an exception is thrown. ## Optional Commands @@ -148,10 +224,22 @@ php bin/console doctrine:schema:fixtures:load [options] **Note:** These commands are optional and should only be registered if you're using the corresponding Doctrine features (migrations and/or fixtures) in your project. ## Testing -To run tests: +Unit tests: +```bash +vendor/bin/phpunit --testsuite unit +``` + +Integration tests run against PostgreSQL (14 by default, `POSTGRES_VERSION=18` for 18) behind PgBouncer in transaction pooling mode +with a single server connection, so every client shares one server session: ```bash -vendor/bin/phpunit +docker compose up -d --build +docker compose run --rm php composer install +docker compose run --rm php vendor/bin/phpunit +docker compose down ``` + +Without `DIRECT_DATABASE_URL` and `PGBOUNCER_DATABASE_URL` the integration tests are skipped. +CI runs them in every matrix job against PostgreSQL 14, and in one job against PostgreSQL 18. ## Contributing Feel free to open issues and submit pull requests. diff --git a/composer.json b/composer.json index 6b4e291..2cc421f 100644 --- a/composer.json +++ b/composer.json @@ -25,8 +25,9 @@ "doctrine/doctrine-fixtures-bundle": "^3.5 || ^4.0", "phpstan/phpstan": "^1.10", "phpunit/phpunit": "^10.0", - "squizlabs/php_codesniffer": "3.7.*", - "dg/bypass-finals": "^1.9" + "squizlabs/php_codesniffer": "^3.13", + "dg/bypass-finals": "^1.9", + "rregeer/phpunit-coverage-check": "^0.3" }, "config": { "allow-plugins": { diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0cf90d3 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,38 @@ +services: + postgres: + image: postgres:${POSTGRES_VERSION:-14}-alpine + environment: + POSTGRES_USER: app + POSTGRES_PASSWORD: app + POSTGRES_DB: app + POSTGRES_HOST_AUTH_METHOD: scram-sha-256 + POSTGRES_INITDB_ARGS: --auth=scram-sha-256 + healthcheck: + test: ["CMD", "pg_isready", "-U", "app"] + interval: 2s + retries: 20 + + pgbouncer: + image: edoburu/pgbouncer:v1.24.1-p1 + environment: + DB_HOST: postgres + DB_USER: app + DB_PASSWORD: app + DB_NAME: app + AUTH_TYPE: scram-sha-256 + POOL_MODE: transaction + DEFAULT_POOL_SIZE: 1 + MAX_PREPARED_STATEMENTS: 200 + depends_on: + postgres: + condition: service_healthy + + php: + build: docker/php + volumes: + - .:/app + environment: + DIRECT_DATABASE_URL: pgsql://app:app@postgres:5432/app + PGBOUNCER_DATABASE_URL: pgsql://app:app@pgbouncer:5432/app + depends_on: + - pgbouncer diff --git a/docker/php/Dockerfile b/docker/php/Dockerfile new file mode 100644 index 0000000..dfa7cbc --- /dev/null +++ b/docker/php/Dockerfile @@ -0,0 +1,8 @@ +FROM php:8.3-cli-alpine + +RUN apk add --no-cache postgresql-dev \ + && docker-php-ext-install pdo_pgsql pgsql + +COPY --from=composer:2 /usr/bin/composer /usr/bin/composer + +WORKDIR /app diff --git a/phpunit.xml.dist b/phpunit.xml.dist index 2ad4f0b..3adc3da 100644 --- a/phpunit.xml.dist +++ b/phpunit.xml.dist @@ -6,8 +6,12 @@ xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/10.5/phpunit.xsd" > - + ./tests + ./tests/Integration + + + ./tests/Integration diff --git a/src/Command/Doctrine/AbstractDoctrineSchemaCommand.php b/src/Command/Doctrine/AbstractDoctrineSchemaCommand.php index ff142a8..4cb4c27 100644 --- a/src/Command/Doctrine/AbstractDoctrineSchemaCommand.php +++ b/src/Command/Doctrine/AbstractDoctrineSchemaCommand.php @@ -6,6 +6,8 @@ use Doctrine\DBAL\Connection; use Error; +use Macpaw\PostgresSchemaBundle\Doctrine\SchemaConnection; +use Macpaw\PostgresSchemaBundle\Doctrine\SearchPathMode; use Macpaw\SchemaContextBundle\Service\BaggageSchemaResolver; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputArgument; @@ -56,6 +58,11 @@ protected function isSchemaExist(string $schema): bool protected function switchToSchema(string $schema): void { $this->schemaResolver->setSchema($schema); + + if (SchemaConnection::getSearchPathMode() === SearchPathMode::Transaction) { + return; + } + $quotedSchema = $this->connection->quoteIdentifier($schema); $this->connection->executeStatement("SET search_path TO {$quotedSchema}"); diff --git a/src/Doctrine/Middleware/TransactionSearchPathConnection.php b/src/Doctrine/Middleware/TransactionSearchPathConnection.php new file mode 100644 index 0000000..39bcc5a --- /dev/null +++ b/src/Doctrine/Middleware/TransactionSearchPathConnection.php @@ -0,0 +1,338 @@ +\w*)\$.*?\$\k\$' + . '|(?\/\*(?:[^\/*]|\/(?!\*)|\*(?!\/)|(?&comment))*\*\/)' + . '|--[^\n]*/s'; + /** Quotes, comment or dollar-quote delimiters left after removing non-code mean the SQL was not split reliably. */ + private const UNBALANCED_PATTERN = '/[\'"]|\/\*|\*\/|\$\w*\$/'; + private const PREPARED_2PC_PATTERN = '/^(?:COMMIT|ROLLBACK)\s+PREPARED\b/i'; + private const ROLLBACK_TO_PATTERN = '/^ROLLBACK\s+(?:(?:WORK|TRANSACTION)\s+)?TO\b/i'; + private const SAVEPOINT_PATTERN = '/^(?:SAVEPOINT|RELEASE)\b/i'; + private const TRANSACTION_CONTROL_PATTERN + = '/^(?:BEGIN|START\s+TRANSACTION|COMMIT|END|ABORT|ROLLBACK|PREPARE\s+TRANSACTION)\b/i'; + + private const KIND_SAVEPOINT = 'savepoint'; + private const KIND_ROLLBACK_TO = 'rollback_to'; + private const KIND_CONTROL = 'control'; + private const KIND_PASSTHROUGH = 'passthrough'; + + private bool $inTransaction = false; + private ?string $appliedSchema = null; + private bool $isAppliedSchemaKnown = true; + + /** + * @param Closure(): ?string $schemaProvider returns the quoted schema to apply, + * or null to keep the server default + * @param Closure(string, bool): void $onApply called with the quoted schema and whether + * the transaction was opened by the caller + * @param Closure(): bool $isServerInTransaction whether the server session has an open transaction + */ + public function __construct( + Connection $wrappedConnection, + private readonly Closure $schemaProvider, + private readonly Closure $onApply, + private readonly Closure $isServerInTransaction, + ) { + parent::__construct($wrappedConnection); + } + + public function prepare(string $sql): Statement + { + if ($this->classify($sql) !== null) { + throw TransactionSearchPathModeException::rawTransactionControl(); + } + + return new TransactionSearchPathStatement(parent::prepare($sql), $this); + } + + public function query(string $sql): Result + { + return $this->run($sql, fn (): Result => parent::query($sql)); + } + + public function exec(string $sql): int + { + return $this->run($sql, fn (): int => parent::exec($sql)); + } + + public function beginTransaction(): bool + { + $result = parent::beginTransaction(); + $this->startTransaction(); + + try { + $this->applyInTransaction(); + } catch (Throwable $exception) { + try { + parent::rollBack(); + } catch (Throwable) { + } + + $this->endTransaction(); + + throw $exception; + } + + return $result; + } + + public function commit(): bool + { + try { + return parent::commit(); + } finally { + $this->endTransaction(); + } + } + + public function rollBack(): bool + { + try { + return parent::rollBack(); + } finally { + $this->endTransaction(); + } + } + + /** + * @template T + * + * @param Closure(): T $operation + * + * @return T + */ + public function runWithSearchPath(Closure $operation): mixed + { + if ($this->inTransaction) { + // Outside a server transaction SET LOCAL is a no-op, so refuse to run after the transaction was lost. + if (!($this->isServerInTransaction)()) { + throw TransactionSearchPathModeException::rawTransactionControl(); + } + + $this->applyInTransaction(); + + return $this->runExpectingServerTransaction(true, $operation); + } + + $schema = ($this->schemaProvider)(); + + if ($schema === null) { + return $this->runExpectingServerTransaction(false, $operation); + } + + return $this->runInImplicitTransaction($schema, $operation); + } + + private function applyInTransaction(): void + { + $schema = ($this->schemaProvider)(); + + if ($this->isAppliedSchemaKnown && $schema === $this->appliedSchema) { + return; + } + + parent::exec('SET LOCAL search_path TO ' . ($schema ?? 'DEFAULT')); + $this->appliedSchema = $schema; + $this->isAppliedSchemaKnown = true; + ($this->onApply)($schema ?? 'DEFAULT', true); + } + + /** + * @template T + * + * @param Closure(): T $operation + * + * @return T + */ + private function runInImplicitTransaction(string $schema, Closure $operation): mixed + { + try { + parent::exec('BEGIN; SET LOCAL search_path TO ' . $schema); + ($this->onApply)($schema, false); + $result = $this->runExpectingServerTransaction(true, $operation); + } catch (Throwable $exception) { + $this->rollBackQuietly(); + + throw $exception; + } + + try { + parent::exec('COMMIT'); + } catch (Throwable $exception) { + $this->rollBackQuietly(); + + throw $exception; + } + + return $result; + } + + /** + * @template T + * + * @param Closure(): T $operation + * + * @return T + */ + private function run(string $sql, Closure $operation): mixed + { + return match ($this->classify($sql)) { + null => $this->runWithSearchPath($operation), + self::KIND_CONTROL => throw TransactionSearchPathModeException::rawTransactionControl(), + // No SET LOCAL before savepoint SQL: in an aborted transaction it would fail and block the recovery. + self::KIND_SAVEPOINT, self::KIND_PASSTHROUGH => $this->runExpectingServerTransaction( + $this->inTransaction, + $operation, + ), + self::KIND_ROLLBACK_TO => $this->runRollbackToSavepoint($operation), + }; + } + + /** + * @template T + * + * @param Closure(): T $operation + * + * @return T + */ + private function runRollbackToSavepoint(Closure $operation): mixed + { + try { + return $this->runExpectingServerTransaction($this->inTransaction, $operation); + } finally { + // A SET LOCAL issued after the savepoint is undone with it, so re-apply before the next query. + $this->isAppliedSchemaKnown = false; + } + } + + /** + * Runs a statement and checks that it left the server transaction status as expected. A mismatch means + * transaction control that the classification did not detect: the statement already ran, but nothing else + * runs on a connection whose transaction state no longer matches DBAL. + * + * @template T + * + * @param Closure(): T $operation + * + * @return T + */ + private function runExpectingServerTransaction(bool $expected, Closure $operation): mixed + { + $result = $operation(); + + if (($this->isServerInTransaction)() === $expected) { + return $result; + } + + if ($expected) { + $this->isAppliedSchemaKnown = false; + } else { + $this->rollBackQuietly(); + } + + throw TransactionSearchPathModeException::rawTransactionControl(); + } + + /** + * Classifies transaction control in a statement or a batch. Null means ordinary SQL. + * + * @return self::KIND_*|null + */ + private function classify(string $sql): ?string + { + // Fast path for the common case: one statement without a leading comment. + if (!str_contains($sql, ';') && preg_match('/^\s*(?:--|\/\*)/', $sql) !== 1) { + return $this->classifyStatement(ltrim($sql)); + } + + $code = preg_replace(self::NON_CODE_PATTERN, ' ', $sql); + + // Fail closed: a PCRE limit error or leftover delimiters mean the statements cannot be split reliably. + if ($code === null || preg_match(self::UNBALANCED_PATTERN, $code) !== 0) { + throw TransactionSearchPathModeException::unclassifiableSql(); + } + $statements = array_values(array_filter( + array_map(trim(...), explode(';', $code)), + static fn (string $statement): bool => $statement !== '', + )); + $kinds = array_map($this->classifyStatement(...), $statements); + $controlKinds = array_filter($kinds, static fn (?string $kind): bool => $kind !== null); + + if ($controlKinds === []) { + return null; + } + + // Statements after transaction control in a batch would run without SET LOCAL or end the transaction. + return count($statements) === 1 ? $controlKinds[0] : self::KIND_CONTROL; + } + + /** + * @return self::KIND_*|null + */ + private function classifyStatement(string $statement): ?string + { + return match (true) { + preg_match(self::PREPARED_2PC_PATTERN, $statement) === 1 => self::KIND_PASSTHROUGH, + preg_match(self::ROLLBACK_TO_PATTERN, $statement) === 1 => self::KIND_ROLLBACK_TO, + preg_match(self::SAVEPOINT_PATTERN, $statement) === 1 => self::KIND_SAVEPOINT, + preg_match(self::TRANSACTION_CONTROL_PATTERN, $statement) === 1 => self::KIND_CONTROL, + default => null, + }; + } + + private function startTransaction(): void + { + $this->inTransaction = true; + $this->appliedSchema = null; + $this->isAppliedSchemaKnown = true; + } + + private function endTransaction(): void + { + $this->inTransaction = false; + $this->appliedSchema = null; + $this->isAppliedSchemaKnown = true; + } + + private function rollBackQuietly(): void + { + try { + parent::exec('ROLLBACK'); + } catch (Throwable) { + } + } +} diff --git a/src/Doctrine/Middleware/TransactionSearchPathDriver.php b/src/Doctrine/Middleware/TransactionSearchPathDriver.php new file mode 100644 index 0000000..e273a15 --- /dev/null +++ b/src/Doctrine/Middleware/TransactionSearchPathDriver.php @@ -0,0 +1,55 @@ +isTransactionMode)()) { + return $connection; + } + + // Only pdo_pgsql defers statement preparation to execution, inside SET LOCAL. The check looks at the + // native connection, so it also works when other driver middlewares wrap the driver. + $native = method_exists($connection, 'getNativeConnection') ? $connection->getNativeConnection() : null; + + if (!$native instanceof PDO) { + throw new UnsupportedDriverException(get_debug_type($native)); + } + + return new TransactionSearchPathConnection( + $connection, + $this->schemaProvider, + $this->onApply, + $native->inTransaction(...), + ); + } +} diff --git a/src/Doctrine/Middleware/TransactionSearchPathStatement.php b/src/Doctrine/Middleware/TransactionSearchPathStatement.php new file mode 100644 index 0000000..c097b20 --- /dev/null +++ b/src/Doctrine/Middleware/TransactionSearchPathStatement.php @@ -0,0 +1,24 @@ +connection->runWithSearchPath(fn (): Result => parent::execute($params)); + } +} diff --git a/src/Doctrine/SchemaConnection.php b/src/Doctrine/SchemaConnection.php index 3eac8da..b75f11a 100644 --- a/src/Doctrine/SchemaConnection.php +++ b/src/Doctrine/SchemaConnection.php @@ -4,19 +4,74 @@ namespace Macpaw\PostgresSchemaBundle\Doctrine; +use Doctrine\Common\EventManager; +use Doctrine\DBAL\Configuration; use Doctrine\DBAL\Connection as DBALConnection; +use Doctrine\DBAL\Driver; +use Doctrine\DBAL\DriverManager; use Doctrine\DBAL\Platforms\PostgreSQLPlatform; +use Macpaw\PostgresSchemaBundle\Doctrine\Middleware\TransactionSearchPathDriver; +use Macpaw\PostgresSchemaBundle\Exception\TransactionSearchPathModeException; use Macpaw\PostgresSchemaBundle\Exception\UnsupportedPlatformException; use Macpaw\SchemaContextBundle\Logger\DebugLogger; use Macpaw\SchemaContextBundle\Service\BaggageSchemaResolver; use Throwable; +/** + * @phpstan-import-type Params from DriverManager + */ class SchemaConnection extends DBALConnection { private static ?BaggageSchemaResolver $schemaResolver = null; private static ?DebugLogger $logger = null; + private static SearchPathMode $searchPathMode = SearchPathMode::Session; + private static ?string $serverDefaultSchema = null; + /** Whether the open physical connection was created in transaction mode; null when not connected. */ + private ?bool $isConnectedInTransactionMode = null; private ?string $currentSchema = null; + /** + * @param array $params + * @phpstan-param Params $params + */ + public function __construct( + array $params, + Driver $driver, + ?Configuration $config = null, + ?EventManager $eventManager = null, + ) { + // The mode is decided per physical connection on connect: in Session mode the driver connection is + // returned unwrapped, so statements go to the database exactly as before. + $driver = new TransactionSearchPathDriver( + $driver, + self::isTransactionModeActive(...), + self::resolveTransactionSearchPath(...), + self::logTransactionSearchPath(...), + ); + + parent::__construct($params, $driver, $config, $eventManager); + } + + /** + * The mode applies to physical connections opened after the call; an idle open connection is reopened + * in the new mode, and switching while a transaction is open throws. + * + * @param string|null $serverDefaultSchema null (strict, default) applies SET LOCAL to every query, ignoring any + * session search_path another client left on the pooled server + * connection. A schema name skips SET LOCAL for that schema and relies + * on the server default search_path (role/database setting). + */ + public static function setSearchPathMode(SearchPathMode $mode, ?string $serverDefaultSchema = null): void + { + self::$searchPathMode = $mode; + self::$serverDefaultSchema = $serverDefaultSchema; + } + + public static function getSearchPathMode(): SearchPathMode + { + return self::$searchPathMode; + } + public static function setSchemaResolver(BaggageSchemaResolver $resolver): void { self::$schemaResolver = $resolver; @@ -29,16 +84,21 @@ public static function setLogger(DebugLogger $logger): void public function close(): void { - self::$logger?->logSchemaResetByConnectionClose($this->currentSchema ?? ''); + if ($this->currentSchema !== null) { + self::$logger?->logSchemaResetByConnectionClose($this->currentSchema); + } parent::close(); $this->currentSchema = null; + $this->isConnectedInTransactionMode = null; } public function rollBack(): bool { - self::$logger?->logSchemaResetByTransactionRollback($this->currentSchema ?? ''); + if ($this->currentSchema !== null) { + self::$logger?->logSchemaResetByTransactionRollback($this->currentSchema); + } $result = parent::rollBack(); @@ -49,7 +109,29 @@ public function rollBack(): bool public function connect(): bool { + $isTransactionMode = self::isTransactionModeActive(); + + $isModeChanged = $this->isConnectedInTransactionMode !== null + && $this->isConnectedInTransactionMode !== $isTransactionMode; + + if ($isModeChanged) { + if ($this->isTransactionActive()) { + throw TransactionSearchPathModeException::modeChangedInTransaction(); + } + + $this->close(); + } + $isNewConnection = parent::connect(); + $this->isConnectedInTransactionMode = $isTransactionMode; + + if ($isTransactionMode) { + if ($isNewConnection) { + $this->ensurePostgreSql(); + } + + return $isNewConnection; + } if (self::$schemaResolver === null) { return $isNewConnection; @@ -77,6 +159,35 @@ public function connect(): bool return $isNewConnection; } + private static function isTransactionModeActive(): bool + { + return self::$searchPathMode === SearchPathMode::Transaction && self::$schemaResolver !== null; + } + + private static function resolveTransactionSearchPath(): ?string + { + if (self::$schemaResolver === null) { + return null; + } + + $schema = self::$schemaResolver->getSchema(); + + if (self::$serverDefaultSchema !== null && $schema === self::$serverDefaultSchema) { + return null; + } + + return '"' . str_replace('"', '""', $schema) . '"'; + } + + /** + * @param bool $isCallerTransaction true when SET LOCAL runs in a transaction opened by the caller, + * false when the bundle wraps a single statement in its own transaction + */ + private static function logTransactionSearchPath(string $quotedSchema, bool $isCallerTransaction): void + { + self::$logger?->logApplySearchPath($quotedSchema, false, $isCallerTransaction); + } + private function getActualSearchPath(): ?string { if ($this->_conn === null) { diff --git a/src/Doctrine/SearchPathMode.php b/src/Doctrine/SearchPathMode.php new file mode 100644 index 0000000..376566d --- /dev/null +++ b/src/Doctrine/SearchPathMode.php @@ -0,0 +1,14 @@ + */ + private array $sent = []; + + private bool $isServerInTransaction = false; + + /** @var array server transaction status after each SQL */ + private array $serverTransactionAfter = []; + + public function testImplicitTransactionClosedByUndetectedControlIsRejectedWithoutCommit(): void + { + $connection = $this->connection(self::PREVIEW_SCHEMA); + $this->serverTransactionAfter[self::INSERT] = false; + + $this->expectRawTransactionControl(fn () => $connection->exec(self::INSERT)); + + self::assertSame([self::APPLY_IMPLICIT, self::INSERT, 'ROLLBACK'], $this->sent); + } + + public function testTransactionLeftOpenByUndetectedControlIsRolledBack(): void + { + $connection = $this->connection(null); + $this->serverTransactionAfter[self::INSERT] = true; + + $this->expectRawTransactionControl(fn () => $connection->exec(self::INSERT)); + + self::assertSame([self::INSERT, 'ROLLBACK'], $this->sent); + self::assertFalse($this->isServerInTransaction); + } + + public function testStatementsAfterCallerTransactionWasLostAreNotSent(): void + { + $connection = $this->connection(self::PREVIEW_SCHEMA); + $connection->beginTransaction(); + $this->serverTransactionAfter[self::INSERT] = false; + + $this->expectRawTransactionControl(fn () => $connection->exec(self::INSERT)); + $this->expectRawTransactionControl(fn () => $connection->exec('SELECT 1')); + + self::assertSame(['SET LOCAL search_path TO "preview_a"', self::INSERT], $this->sent); + } + + /** + * @return iterable + */ + public static function unclassifiableSqlProvider(): iterable + { + yield 'unterminated string' => ["SELECT 1; SELECT 'open"]; + yield 'unterminated identifier' => ['SELECT 1; SELECT "open']; + yield 'unterminated block comment' => ['SELECT 1; /* open COMMIT']; + yield 'stray comment end' => ['SELECT 1 */; COMMIT']; + yield 'unterminated dollar quote' => ['SELECT $body$ x; COMMIT']; + } + + #[DataProvider('unclassifiableSqlProvider')] + public function testUnclassifiableSqlIsRejectedBeforeSending(string $sql): void + { + $connection = $this->connection(self::PREVIEW_SCHEMA); + + foreach ([fn () => $connection->exec($sql), fn () => $connection->prepare($sql)] as $call) { + try { + $call(); + self::fail('Unclassifiable SQL must be rejected.'); + } catch (TransactionSearchPathModeException $exception) { + self::assertStringContainsString('cannot split this SQL', $exception->getMessage()); + } + } + + self::assertSame([], $this->sent); + } + + public function testPcreFailureIsRejectedBeforeSending(): void + { + $connection = $this->connection(self::PREVIEW_SCHEMA); + $backtrackLimit = ini_get('pcre.backtrack_limit'); + $jit = ini_get('pcre.jit'); + ini_set('pcre.jit', '0'); + ini_set('pcre.backtrack_limit', '1'); + + try { + $this->expectExceptionObject(TransactionSearchPathModeException::unclassifiableSql()); + $connection->exec("SELECT 1; SELECT \$x\$ long body; with semicolons \$x\$"); + } finally { + ini_set('pcre.backtrack_limit', (string) $backtrackLimit); + ini_set('pcre.jit', (string) $jit); + self::assertSame([], $this->sent); + } + } + + public function testMatchingServerTransactionStatusRunsAsBefore(): void + { + $connection = $this->connection(self::PREVIEW_SCHEMA); + + $connection->exec(self::INSERT); + $connection->beginTransaction(); + $connection->exec(self::INSERT); + $connection->commit(); + + self::assertSame( + [ + self::APPLY_IMPLICIT, + self::INSERT, + 'COMMIT', + 'SET LOCAL search_path TO "preview_a"', + self::INSERT, + ], + $this->sent, + ); + } + + private function expectRawTransactionControl(callable $call): void + { + try { + $call(); + self::fail('Undetected transaction control must be rejected.'); + } catch (TransactionSearchPathModeException $exception) { + self::assertSame( + TransactionSearchPathModeException::rawTransactionControl()->getMessage(), + $exception->getMessage(), + ); + } + } + + private function connection(?string $schema): TransactionSearchPathConnection + { + $wrapped = $this->createMock(DriverConnection::class); + $wrapped->method('exec')->willReturnCallback(function (string $sql): int { + $this->sent[] = $sql; + $this->isServerInTransaction = match (true) { + isset($this->serverTransactionAfter[$sql]) => $this->serverTransactionAfter[$sql], + str_starts_with($sql, 'BEGIN') => true, + $sql === 'COMMIT', $sql === 'ROLLBACK' => false, + default => $this->isServerInTransaction, + }; + + return 0; + }); + $wrapped->method('beginTransaction')->willReturnCallback(function (): bool { + $this->isServerInTransaction = true; + + return true; + }); + $wrapped->method('commit')->willReturnCallback(function (): bool { + $this->isServerInTransaction = false; + + return true; + }); + + return new TransactionSearchPathConnection( + $wrapped, + static fn (): ?string => $schema, + static function (): void { + }, + fn (): bool => $this->isServerInTransaction, + ); + } +} diff --git a/tests/Doctrine/SchemaConnectionTest.php b/tests/Doctrine/SchemaConnectionTest.php index 3228416..ff4e679 100644 --- a/tests/Doctrine/SchemaConnectionTest.php +++ b/tests/Doctrine/SchemaConnectionTest.php @@ -8,13 +8,18 @@ use Doctrine\DBAL\Configuration; use Doctrine\DBAL\Driver; use Doctrine\DBAL\Driver\Connection as DriverConnection; +use Doctrine\DBAL\Driver\Middleware\AbstractConnectionMiddleware; use Doctrine\DBAL\Platforms\PostgreSQLPlatform; use Doctrine\DBAL\Platforms\MySQLPlatform; use Macpaw\PostgresSchemaBundle\Doctrine\SchemaConnection; +use Macpaw\PostgresSchemaBundle\Doctrine\SearchPathMode; +use Macpaw\PostgresSchemaBundle\Exception\UnsupportedDriverException; use Macpaw\PostgresSchemaBundle\Exception\UnsupportedPlatformException; use Macpaw\SchemaContextBundle\Logger\DebugLogger; use Macpaw\SchemaContextBundle\Service\BaggageSchemaResolver; +use PDO; use PHPUnit\Framework\TestCase; +use stdClass; class SchemaConnectionTest extends TestCase { @@ -159,4 +164,65 @@ public function testThrowsForUnsupportedPlatform(): void $connection->connect(); } + + public function testTransactionModeThrowsForUnsupportedPlatform(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $driverConnection = $this->createMock(AbstractConnectionMiddleware::class); + $driverConnection->method('getNativeConnection')->willReturn($this->createMock(PDO::class)); + $driverConnection->expects($this->never())->method('exec'); + $driver = $this->createMock(Driver::class); + $driver->method('connect')->willReturn($driverConnection); + + $connection = $this->getMockBuilder(SchemaConnection::class) + ->setConstructorArgs([[], $driver, new Configuration(), new EventManager()]) + ->onlyMethods(['getDatabasePlatform']) + ->getMock(); + $connection->method('getDatabasePlatform')->willReturn(new MySQLPlatform()); + + SchemaConnection::setSchemaResolver( + new BaggageSchemaResolver('public', 'development', ['development'], new DebugLogger()), + ); + + try { + $this->expectException(UnsupportedPlatformException::class); + $connection->connect(); + } finally { + SchemaConnection::setSearchPathMode(SearchPathMode::Session); + } + } + + public function testTransactionModeRejectsNonPdoConnection(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $driverConnection = $this->createMock(AbstractConnectionMiddleware::class); + $driverConnection->method('getNativeConnection')->willReturn(new stdClass()); + $driverConnection->expects($this->never())->method('exec'); + $driver = $this->createMock(Driver::class); + $driver->method('connect')->willReturn($driverConnection); + $connection = new SchemaConnection([], $driver, new Configuration(), new EventManager()); + + SchemaConnection::setSchemaResolver( + new BaggageSchemaResolver('public', 'development', ['development'], new DebugLogger()), + ); + + try { + $this->expectException(UnsupportedDriverException::class); + $connection->connect(); + } finally { + SchemaConnection::setSearchPathMode(SearchPathMode::Session); + } + } + + public function testSessionModeDoesNotWrapDriverConnection(): void + { + $driverConnection = $this->createMock(DriverConnection::class); + $driver = $this->createMock(Driver::class); + $driver->method('connect')->willReturn($driverConnection); + $driver->method('getDatabasePlatform')->willReturn(new PostgreSQLPlatform()); + + $connection = new SchemaConnection([], $driver, new Configuration(), new EventManager()); + + self::assertSame($driverConnection, $connection->getWrappedConnection()); + } } diff --git a/tests/Integration/SearchPathPgbouncerTest.php b/tests/Integration/SearchPathPgbouncerTest.php new file mode 100644 index 0000000..77080e9 --- /dev/null +++ b/tests/Integration/SearchPathPgbouncerTest.php @@ -0,0 +1,479 @@ + */ + private array $connections = []; + + public static function setUpBeforeClass(): void + { + if (self::url('DIRECT_DATABASE_URL') === null || self::url('PGBOUNCER_DATABASE_URL') === null) { + return; + } + + $direct = DriverManager::getConnection(self::params('DIRECT_DATABASE_URL')); + $previewSchema = $direct->quoteIdentifier(self::PREVIEW_SCHEMA); + + $direct->executeStatement('DROP SCHEMA IF EXISTS ' . $previewSchema . ' CASCADE'); + $direct->executeStatement('DROP TABLE IF EXISTS public.marker'); + $direct->executeStatement('CREATE SCHEMA ' . $previewSchema); + $direct->executeStatement('CREATE TABLE public.marker (id serial PRIMARY KEY, v text NOT NULL)'); + $direct->executeStatement( + 'CREATE TABLE ' . $previewSchema . '.marker (id serial PRIMARY KEY, v text NOT NULL)', + ); + $direct->executeStatement( + 'CREATE TABLE ' . $previewSchema . '.deferred_unique' + . ' (v text, CONSTRAINT deferred_unique_v UNIQUE (v) DEFERRABLE INITIALLY DEFERRED)', + ); + $direct->insert('public.marker', ['v' => self::PUBLIC_MARKER]); + $direct->insert(self::PREVIEW_SCHEMA . '.marker', ['v' => self::PREVIEW_MARKER]); + $direct->close(); + } + + protected function setUp(): void + { + if (self::url('PGBOUNCER_DATABASE_URL') === null || self::url('DIRECT_DATABASE_URL') === null) { + self::markTestSkipped( + 'PGBOUNCER_DATABASE_URL and DIRECT_DATABASE_URL must be set; run via docker compose.', + ); + } + + $this->resetServerSessions(); + $this->resolver = new BaggageSchemaResolver('public', 'development', ['development'], new DebugLogger()); + SchemaConnection::setSchemaResolver($this->resolver); + } + + protected function tearDown(): void + { + foreach ($this->connections as $connection) { + $connection->close(); + } + + $this->connections = []; + SchemaConnection::setSearchPathMode(SearchPathMode::Session); + SchemaConnection::setLogger(new DebugLogger()); + } + + public function testSessionModeLeaksPreviewSchemaToDefaultClient(): void + { + $default = $this->connect(); + $preview = $this->connect(); + + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + self::assertSame(self::PREVIEW_MARKER, $this->readMarker($preview, self::PREVIEW_SCHEMA)); + + self::assertSame( + self::PREVIEW_MARKER, + $this->readMarker($default, null), + 'Session SET search_path is expected to leak behind transaction pooling.', + ); + } + + public function testSessionModeKeepsSessionSearchPathAfterSavepointRollback(): void + { + $preview = $this->connect(); + + self::assertSame(self::PREVIEW_MARKER, $this->readMarker($preview, self::PREVIEW_SCHEMA)); + $preview->beginTransaction(); + $preview->executeStatement('SAVEPOINT raw_sp'); + $preview->executeStatement('ROLLBACK TO SAVEPOINT raw_sp'); + + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + $preview->commit(); + } + + public function testLenientTransactionModeKeepsDefaultClientOnServerDefaultSchema(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction, 'public'); + $default = $this->connect(); + $preview = $this->connect(); + + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + self::assertSame(self::PREVIEW_MARKER, $this->readMarker($preview, self::PREVIEW_SCHEMA)); + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + self::assertSame('"$user", public', $default->fetchOne('SHOW search_path')); + } + + public function testStrictTransactionModeIgnoresSearchPathLeakedByASessionClient(): void + { + $this->leakSessionSearchPath(self::PREVIEW_SCHEMA); + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $default = $this->connect(); + + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + } + + public function testLenientTransactionModeInheritsSearchPathLeakedByASessionClient(): void + { + $this->leakSessionSearchPath(self::PREVIEW_SCHEMA); + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction, 'public'); + $default = $this->connect(); + + self::assertSame( + self::PREVIEW_MARKER, + $this->readMarker($default, null), + 'Skipping SET LOCAL for the server default relies on no client setting a session search_path.', + ); + } + + public function testTransactionModeAppliesSchemaToPreparedStatements(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $default = $this->connect(); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + $marker = $preview->prepare(self::MARKER_QUERY . ' WHERE id = ?')->executeQuery([1])->fetchOne(); + + self::assertSame(self::PREVIEW_MARKER, $marker); + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + } + + public function testTransactionModeWritesExplicitTransactionToResolvedSchema(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + $preview->transactional(static function (Connection $connection): void { + $connection->insert('marker', ['v' => 'tx-write']); + $connection->executeStatement('UPDATE marker SET v = v WHERE v = ?', ['tx-write']); + }); + + self::assertSame(1, $this->countDirect(self::PREVIEW_SCHEMA, 'tx-write')); + self::assertSame(0, $this->countDirect('public', 'tx-write')); + } + + public function testTransactionModeReappliesSchemaAfterSavepointRollback(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $preview->setNestTransactionsWithSavepoints(true); + + $this->resolver->setSchema(null); + $preview->beginTransaction(); + $preview->beginTransaction(); + self::assertSame(self::PREVIEW_MARKER, $this->readMarker($preview, self::PREVIEW_SCHEMA)); + $preview->rollBack(); + + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + $preview->commit(); + } + + public function testTransactionModeRollsBackSavepointOfAbortedTransactionAfterSchemaChange(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $preview->setNestTransactionsWithSavepoints(true); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + $preview->beginTransaction(); + $preview->beginTransaction(); + + try { + $preview->fetchOne('SELECT missing_column FROM marker'); + self::fail('Query with a missing column must fail.'); + } catch (DBALException) { + } + + $this->resolver->setSchema(null); + $preview->rollBack(); + + self::assertSame(self::PUBLIC_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + $preview->commit(); + } + + /** + * @return iterable + */ + public static function rawTransactionControlProvider(): iterable + { + yield 'begin' => ['BEGIN']; + yield 'start transaction with comment' => ['/* app */ START TRANSACTION']; + yield 'commit' => ['COMMIT']; + yield 'rollback and chain' => ["-- app\nROLLBACK AND CHAIN"]; + yield 'batch' => ["BEGIN; INSERT INTO marker (v) VALUES ('batch'); COMMIT"]; + yield 'nested comment before commit' => [ + "/* outer /* inner */ tail */ COMMIT; INSERT INTO marker (v) VALUES ('batch')", + ]; + } + + #[DataProvider('rawTransactionControlProvider')] + public function testTransactionModeRejectsRawTransactionControl(string $sql): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + foreach (['executeStatement', 'executeQuery'] as $method) { + try { + $preview->{$method}($sql); + self::fail(sprintf('%s() must reject raw transaction control.', $method)); + } catch (TransactionSearchPathModeException) { + } + } + + // The safety net throws the same exception after an undetected COMMIT already ran the rest of the batch, + // so check both schemas: the batch must be rejected before anything is sent. + self::assertSame(0, $this->countDirect(self::PREVIEW_SCHEMA, 'batch')); + self::assertSame(0, $this->countDirect('public', 'batch')); + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + } + + public function testIdleConnectionSwitchesToModeSetAfterItWasCreated(): void + { + $preview = $this->connect(); + self::assertSame(self::PREVIEW_MARKER, $this->readMarker($preview, self::PREVIEW_SCHEMA)); + + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $default = $this->connect(); + + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + self::assertInstanceOf(TransactionSearchPathConnection::class, $preview->getWrappedConnection()); + } + + public function testModeChangeInsideTransactionIsRejected(): void + { + $preview = $this->connect(); + $preview->beginTransaction(); + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + + try { + $this->expectException(TransactionSearchPathModeException::class); + $preview->fetchOne('SELECT 1'); + } finally { + SchemaConnection::setSearchPathMode(SearchPathMode::Session); + $preview->rollBack(); + } + } + + /** + * @return iterable + */ + public static function savepointBatchProvider(): iterable + { + yield 'savepoint then update' => ["SAVEPOINT s1; UPDATE marker SET v = v WHERE v = 'none'"]; + yield 'statement then commit' => ["UPDATE marker SET v = v WHERE v = 'none'; COMMIT"]; + yield 'prepare transaction' => ["PREPARE TRANSACTION 'tx1'"]; + } + + #[DataProvider('savepointBatchProvider')] + public function testTransactionModeRejectsTransactionControlInBatchInsideTransaction(string $sql): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + $preview->beginTransaction(); + + try { + $this->expectException(TransactionSearchPathModeException::class); + $preview->executeStatement($sql); + } finally { + $preview->rollBack(); + } + } + + public function testTransactionModeAllowsStatementsWithSemicolonsInLiterals(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + $preview->executeStatement("INSERT INTO marker (v) VALUES ('a; COMMIT')"); + $preview->executeStatement("INSERT INTO marker (v) VALUES (E'it\\'s; COMMIT')"); + + self::assertSame(1, $this->countDirect(self::PREVIEW_SCHEMA, 'a; COMMIT')); + self::assertSame(1, $this->countDirect(self::PREVIEW_SCHEMA, "it's; COMMIT")); + } + + public function testTransactionModeClosesImplicitTransactionWhenApplyFails(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + SchemaConnection::setLogger(new DebugLogger(self::failingApplyLogger())); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + try { + $preview->fetchOne(self::MARKER_QUERY); + self::fail('The failing logger must abort the query.'); + } catch (RuntimeException) { + } + + self::assertSame(0, $this->countOpenServerTransactions()); + } + + public function testTransactionModeClosesExplicitTransactionWhenApplyFails(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $preview->fetchOne('SELECT 1'); + SchemaConnection::setLogger(new DebugLogger(self::failingApplyLogger())); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + try { + $preview->beginTransaction(); + self::fail('The failing logger must abort beginTransaction().'); + } catch (RuntimeException) { + } + + self::assertSame(0, $this->countOpenServerTransactions()); + } + + public function testTransactionModeRecoversWhenImplicitCommitFails(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + try { + $preview->executeStatement("INSERT INTO deferred_unique (v) VALUES ('dup'), ('dup')"); + self::fail('The deferred unique constraint must fail on COMMIT.'); + } catch (DBALException) { + } + + self::assertSame(0, $this->countOpenServerTransactions()); + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + } + + public function testTransactionModeRecoversAfterFailedStatement(): void + { + SchemaConnection::setSearchPathMode(SearchPathMode::Transaction); + $default = $this->connect(); + $preview = $this->connect(); + $this->resolver->setSchema(self::PREVIEW_SCHEMA); + + try { + $preview->fetchOne('SELECT missing_column FROM marker'); + self::fail('Query with a missing column must fail.'); + } catch (DBALException) { + } + + self::assertSame(self::PREVIEW_MARKER, $preview->fetchOne(self::MARKER_QUERY)); + self::assertSame(self::PUBLIC_MARKER, $this->readMarker($default, null)); + } + + private static function failingApplyLogger(): AbstractLogger + { + return new class extends AbstractLogger { + /** @param array $context */ + public function log($level, string|Stringable $message, array $context = []): void + { + if ((string) $message === 'schema_context_apply_search_path') { + throw new RuntimeException('Logger failure'); + } + } + }; + } + + private function countOpenServerTransactions(): int + { + $direct = DriverManager::getConnection(self::params('DIRECT_DATABASE_URL')); + $this->connections[] = $direct; + + return (int) $direct->fetchOne( + "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND state LIKE 'idle in%'", + ); + } + + private function leakSessionSearchPath(string $schema): void + { + $plain = DriverManager::getConnection(self::params('PGBOUNCER_DATABASE_URL')); + $this->connections[] = $plain; + $plain->executeStatement('SET search_path TO ' . $plain->quoteIdentifier($schema)); + } + + private function resetServerSessions(): void + { + $direct = DriverManager::getConnection(self::params('DIRECT_DATABASE_URL')); + $direct->executeQuery( + 'SELECT pg_terminate_backend(pid) FROM pg_stat_activity' + . ' WHERE datname = current_database() AND pid <> pg_backend_pid() AND backend_type = ?', + ['client backend'], + ); + $direct->close(); + } + + private function countDirect(string $schema, string $value): int + { + $direct = DriverManager::getConnection(self::params('DIRECT_DATABASE_URL')); + $this->connections[] = $direct; + + return (int) $direct->fetchOne( + 'SELECT count(*) FROM ' . $direct->quoteIdentifier($schema) . '.marker WHERE v = ?', + [$value], + ); + } + + private function readMarker(Connection $connection, ?string $schema): mixed + { + $this->resolver->setSchema($schema); + + return $connection->fetchOne(self::MARKER_QUERY); + } + + private function connect(): Connection + { + $params = self::params('PGBOUNCER_DATABASE_URL'); + $params['wrapperClass'] = SchemaConnection::class; + + $connection = DriverManager::getConnection($params); + $this->connections[] = $connection; + + return $connection; + } + + private static function url(string $name): ?string + { + $url = getenv($name); + + return is_string($url) && $url !== '' ? $url : null; + } + + /** + * @return array{driver: 'pdo_pgsql', host: string, port: int, user: string, password: string, dbname: string} + */ + private static function params(string $name): array + { + $parts = parse_url((string) self::url($name)); + + return [ + 'driver' => 'pdo_pgsql', + 'host' => (string) ($parts['host'] ?? ''), + 'port' => (int) ($parts['port'] ?? 5432), + 'user' => (string) ($parts['user'] ?? ''), + 'password' => (string) ($parts['pass'] ?? ''), + 'dbname' => ltrim((string) ($parts['path'] ?? ''), '/'), + ]; + } +}