From b51eae50e8d40b92570c0210fab3d60e9bc0002b Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sat, 31 Oct 2026 14:29:46 -0700 Subject: [PATCH 1/2] Add DB.advisory_lock and LockTimeoutError for Postgres Transaction-scoped pg_advisory_xact_lock with str keys hashed by blake2b to a signed bigint, a transaction-local lock_timeout set through a bound set_config value, and SQLSTATE 55P03 mapped to LockTimeoutError. Other drivers raise NotImplementedError; calls outside a transaction raise RuntimeError. Co-Authored-By: Claude Opus 5.5 --- .../058-transaction-scoped-advisory-lock.md | 79 ++++++++ docs/adr/AGENTS.md | 1 + .../fastapi_startkit/masoniteorm/__init__.py | 3 +- .../masoniteorm/connections/advisory_lock.py | 25 +++ .../masoniteorm/connections/connection.py | 6 + .../connections/postgres_connection.py | 29 +++ .../masoniteorm/exceptions.py | 7 + .../masoniteorm/facades/DB.py | 8 + .../connections/test_advisory_lock.py | 178 ++++++++++++++++++ .../postgres/test_advisory_lock.py | 166 ++++++++++++++++ 10 files changed, 501 insertions(+), 1 deletion(-) create mode 100644 docs/adr/058-transaction-scoped-advisory-lock.md create mode 100644 fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/advisory_lock.py create mode 100644 fastapi_startkit/tests/masoniteorm/connections/test_advisory_lock.py create mode 100644 fastapi_startkit/tests/masoniteorm/postgres/test_advisory_lock.py diff --git a/docs/adr/058-transaction-scoped-advisory-lock.md b/docs/adr/058-transaction-scoped-advisory-lock.md new file mode 100644 index 00000000..58d8a330 --- /dev/null +++ b/docs/adr/058-transaction-scoped-advisory-lock.md @@ -0,0 +1,79 @@ +# 058 — Transaction-scoped advisory locks (`DB.advisory_lock`) and `LockTimeoutError` + +Date: 2026-10-31 + +## Problem + +Applications serialise critical sections across processes, such as OAuth refresh-token rotation, with +PostgreSQL advisory locks. Today they write raw SQL inside `transaction()`: + +```python +await connection.statement("SET LOCAL lock_timeout = '3000ms'") +await connection.statement("SELECT pg_advisory_xact_lock(42)") +``` + +and they catch the asyncpg `LockNotAvailableError` (SQLSTATE `55P03`) wrapped in SQLAlchemy's +`DBAPIError` to turn a lock-wait timeout into a client error. Every app repeats this. It leaks driver +exceptions into application code and invites string interpolation into `SET`, which does not accept +bind parameters. + +## Alternatives + +1. **Session-scoped locks** (`pg_advisory_lock` / `pg_advisory_unlock`). Rejected: the lock outlives + a rolled-back transaction if the unlock is missed, and pooled connections can return to the pool + still holding it. The transaction-scoped variant is released by PostgreSQL at commit or rollback. +2. **String keys hashed in SQL with `hashtext(:key)`.** Rejected: `hashtext` is an internal, + undocumented function. It returns a 32-bit `int4`, so it collides far more often than a 64-bit key, + and its output changed in the past (PostgreSQL 8.4). A key derived from it cannot be computed or + tested outside the database either. +3. **String keys hashed in Python with `hash()`.** Rejected: randomised per process (`PYTHONHASHSEED`). +4. **String keys hashed in Python with `blake2b`, 8-byte digest, read as a signed big-endian integer.** + Chosen. It is stable across processes, machines, Python versions and PostgreSQL versions. It uses + the full `bigint` key space, and it can be unit-tested without a database. +5. **Interpolating the timeout into `SET LOCAL lock_timeout = ...`.** Rejected: user input in SQL. + `set_config('lock_timeout', :value, true)` takes a bound value and has the same transaction-local + scope (`is_local = true`). + +## Decision + +- `Connection.advisory_lock(key, timeout_ms=None)` and `Connection.lock_timeout(timeout_ms)` exist on + every connection. The base `Connection` raises `NotImplementedError` naming the driver. + `PostgresConnection` implements them. +- `DB.advisory_lock(key, timeout_ms=None, connection=None)` and `DB.lock_timeout(timeout_ms, connection=None)` + forward to the named (or default) connection. +- Both must run inside an open transaction on that connection. Otherwise they raise `RuntimeError`, + because outside a transaction the lock would be released at once (each statement autocommits) and + `set_config(..., true)` would have no effect. +- Key: an `int` is used as-is and must fit a signed 64-bit `bigint`. A `str` is hashed as in + alternative 4. `bool` and other types raise `TypeError`, and an out-of-range `int` raises `ValueError`. +- Timeout: an `int` in `0..2147483647` (PostgreSQL's `lock_timeout` maximum, in milliseconds). `0` + disables the timeout, as in PostgreSQL. It is set with `SELECT set_config('lock_timeout', :ms, true)` + and lasts until the transaction ends. Later lock waits in the same transaction also use it, as with + the `SET LOCAL` this replaces. +- The lock is taken with `SELECT pg_advisory_xact_lock(:key)` and released by PostgreSQL at commit or + rollback. Taking it inside a nested transaction (savepoint) still holds it until the outer + transaction ends. +- A `DBAPIError` with SQLSTATE `55P03` raised while waiting is re-raised as + `fastapi_startkit.masoniteorm.LockTimeoutError(key, timeout_ms)` chained `from` the driver error. + Other errors propagate unchanged. PostgreSQL aborts the transaction on this error, so the caller's + `transaction()` block rolls back when the exception leaves it. + +## Implementation + +- `masoniteorm/connections/advisory_lock.py`: `advisory_lock_key(key)` and `lock_timeout_value(ms)` + (validation and hashing). +- `masoniteorm/exceptions.py`: `LockTimeoutError`, exported from `fastapi_startkit.masoniteorm`. +- `Connection`: `NotImplementedError` defaults. `PostgresConnection`: transaction check, `set_config`, + `pg_advisory_xact_lock`, `55P03` mapping. +- `DB` facade: `advisory_lock` and `lock_timeout`. +- The transaction modules (`manages_transactions.py`, `transactions.py`) are not changed. + +## Validation + +- Unit tests: key hashing (stable known values, signed range, `int` pass-through, bounds, type errors), + timeout validation, the no-transaction error, `NotImplementedError` on SQLite, `55P03` mapping and + pass-through of other driver errors, and facade forwarding. +- Live PostgreSQL tests (`tests/masoniteorm/postgres`): the lock is visible in `pg_locks` while held, + a second transaction blocks, a timeout raises `LockTimeoutError`, the lock is released on commit + and on rollback, and `lock_timeout` returns to its previous value after the transaction. +- `uv run pytest --ignore=tests/masoniteorm/postgres --cov` stays at or above 80%. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 4952cfc2..e9c62ed1 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -47,3 +47,4 @@ Read this index first, then open the records relevant to the change. Before impl | [052](052-sqlite-savepoint-isolation.md) | 2026-10-31 | SQLite savepoints start a real transaction | pysqlite/aiosqlite defer `BEGIN` until the first DML, so a savepoint opened first in an outer transaction became a top-level transaction and `RELEASE` committed it, leaking rows past `DatabaseTransaction`/`RefreshDatabase` rollbacks. `SQliteConnection` listens to the engine `savepoint` event and runs `BEGIN` when the driver connection is not in a transaction. On the shared `StaticPool` connection, a lost `BEGIN` race (`OperationalError` while a transaction is already open) is ignored, so concurrent nested transactions still succeed as on `main`. SQLAlchemy's `isolation_level=None` + `BEGIN` on every `begin` recipe was rejected because concurrent transactions on `SQLiteConfig`'s shared `StaticPool` connection then fail with `cannot start a transaction within a transaction`. | | [053](053-validated-reads-query-and-body.md) | 2026-10-31 | `Validated` reads the query string and the body | `Validated[Model]` declared a required `Body()`, so a `GET` with filters got `422 body: Field required` and apps bound models with `Query()`, skipping `authorize`, rules and `after`. The body parameter now defaults to `None`. Without a body, `GET`/`HEAD`/`OPTIONS`/`DELETE` validate `query_input`, while `POST`/`PUT`/`PATCH` still report `"": Field required` (per embed key when embedded), so a body-less cross-site form `POST` cannot pass on the query string (CSRF). `DELETE` always needs a CORS preflight. With a body, `request_input` merges `{**query, **body}` for an object; any other body, a JSON `null` included, passes through and fails with `model_type`. `query_input` keeps the keys pydantic accepts (`validation_alias`, `AliasChoices`, single-key `AliasPath`, the name with `validate_by_name`/`populate_by_name`). `collections.abc` `Sequence`/`Set` fields (not `str`/`bytes`) get every repeated value. Declared fields can be set from the URL on write methods with a JSON body, as with Laravel's `$request->all()`: keep sensitive fields off such models. `document_query_input` (installed by `FastAPIProvider`) replaces `requestBody` with `in: query` parameters on query-method routes. Chosen over a separate `ValidatedQuery` so one `FormRequest` serves `GET` and `POST`. | | [055](055-morph-name-equals-class-name.md) | 2026-10-31 | Morph relations accept the class name as the morph name | `MorphMany`/`MorphOne.get_record_key_lookup` raised `ValueError` whenever `Registry.get_morph_name` returned the class name, treating it as "not registered". `Model.__init_subclass__` registers every model with its class name as the default morph name, so every model without a `morph_map` alias (or mapped to its own class name) failed on `with_` and `load`. The lookup now returns `Registry.get_morph_name(cls)` as-is (`morph_map` aliases still win). `MorphMany`/`MorphOne.get_related` also forward `eagers` via `.with_(eagers or [])`, so nested loads such as `with_("media.lexeme")` work. `MorphTo`/`MorphToMany` were unaffected. | +| [058](058-transaction-scoped-advisory-lock.md) | 2026-10-31 | Transaction-scoped advisory locks and `LockTimeoutError` | `DB.advisory_lock(key, timeout_ms=None, connection=None)` / `Connection.advisory_lock` run `pg_advisory_xact_lock` (released at commit or rollback; no session-scoped variant), and `DB.lock_timeout(ms)` sets a transaction-local timeout with `set_config('lock_timeout', :ms, true)` (bound, no interpolation; validated `0..2147483647`). Both need an open transaction on the connection (`RuntimeError` otherwise), and non-Postgres connections raise `NotImplementedError`. An `int` key must fit `bigint`. A `str` key is hashed with `blake2b` (8 bytes, signed big-endian), which is stable across processes and versions, unlike `hashtext()` (internal, 32-bit, has changed) or `hash()`. SQLSTATE `55P03` while waiting becomes `LockTimeoutError` (exported from `fastapi_startkit.masoniteorm`) chained from the driver error. | diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py index c5723a1a..83ffe6ed 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py @@ -1,6 +1,6 @@ from .config.config import DBConfig, MySQLConfig, PostgresConfig, SQLiteConfig from .config.database import DatabaseConfig -from .exceptions import ModelNotFoundException +from .exceptions import LockTimeoutError, ModelNotFoundException from .facades import DB from .factory.factory import Factory from .migrations.Migration import Migration @@ -33,6 +33,7 @@ "SQLiteConfig", "Model", "ModelNotFoundException", + "LockTimeoutError", "DB", "Factory", "Migration", diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/advisory_lock.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/advisory_lock.py new file mode 100644 index 00000000..4d69f898 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/advisory_lock.py @@ -0,0 +1,25 @@ +import hashlib + +BIGINT_MIN = -(2**63) +BIGINT_MAX = 2**63 - 1 +LOCK_TIMEOUT_MAX_MS = 2**31 - 1 +LOCK_NOT_AVAILABLE = "55P03" + + +def advisory_lock_key(key: str | int) -> int: + if isinstance(key, str): + digest = hashlib.blake2b(key.encode("utf-8"), digest_size=8).digest() + return int.from_bytes(digest, "big", signed=True) + if isinstance(key, bool) or not isinstance(key, int): + raise TypeError(f"Advisory lock key must be a str or int, got {type(key).__name__}") + if not BIGINT_MIN <= key <= BIGINT_MAX: + raise ValueError(f"Advisory lock key {key} is outside the bigint range") + return key + + +def lock_timeout_value(timeout_ms: int) -> str: + if isinstance(timeout_ms, bool) or not isinstance(timeout_ms, int): + raise TypeError(f"lock_timeout must be an int number of milliseconds, got {type(timeout_ms).__name__}") + if not 0 <= timeout_ms <= LOCK_TIMEOUT_MAX_MS: + raise ValueError(f"lock_timeout must be between 0 and {LOCK_TIMEOUT_MAX_MS} milliseconds, got {timeout_ms}") + return str(timeout_ms) diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/connection.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/connection.py index be718159..b7533c4f 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/connection.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/connection.py @@ -116,3 +116,9 @@ async def statement(self, query: str, bindings: list[Any] | None = None) -> bool query, params = self.sql_alchemy_bindings(query, bindings) await self._execute(query, params) return True + + async def advisory_lock(self, key: str | int, timeout_ms: int | None = None) -> None: + raise NotImplementedError(f"Advisory locks are only supported on Postgres, not the {self.config.driver} driver") + + async def lock_timeout(self, timeout_ms: int) -> None: + raise NotImplementedError(f"lock_timeout is only supported on Postgres, not the {self.config.driver} driver") diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/postgres_connection.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/postgres_connection.py index ef3c297e..564d8876 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/postgres_connection.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/connections/postgres_connection.py @@ -2,12 +2,15 @@ from typing import Any from sqlalchemy import event +from sqlalchemy.exc import DBAPIError from sqlalchemy.ext.asyncio import AsyncEngine from fastapi_startkit.masoniteorm.config.config import DBConfig +from fastapi_startkit.masoniteorm.exceptions import LockTimeoutError from fastapi_startkit.masoniteorm.query.grammars import PostgresGrammar from fastapi_startkit.masoniteorm.query.processors import PostgresPostProcessor from fastapi_startkit.masoniteorm.schema.platforms import PostgresPlatform +from .advisory_lock import LOCK_NOT_AVAILABLE, advisory_lock_key, lock_timeout_value from .connection import Connection POSTGRES_EPOCH = datetime.datetime(2000, 1, 1) @@ -62,6 +65,32 @@ async def insert_get_id(self, query: str, bindings: list | None = None) -> int | row = result.fetchone() return row[0] if row is not None else None + async def advisory_lock(self, key: str | int, timeout_ms: int | None = None) -> None: + lock_key = advisory_lock_key(key) + timeout = None if timeout_ms is None else lock_timeout_value(timeout_ms) + self._require_transaction("advisory_lock") + if timeout is not None: + await self._set_lock_timeout(timeout) + try: + await self.statement("SELECT pg_advisory_xact_lock(?)", [lock_key]) + except DBAPIError as error: + if getattr(error.orig, "sqlstate", None) == LOCK_NOT_AVAILABLE: + raise LockTimeoutError(key, timeout_ms) from error + raise + + async def lock_timeout(self, timeout_ms: int) -> None: + value = lock_timeout_value(timeout_ms) + self._require_transaction("lock_timeout") + await self._set_lock_timeout(value) + + async def _set_lock_timeout(self, value: str) -> None: + await self.statement("SELECT set_config('lock_timeout', ?, true)", [value]) + + def _require_transaction(self, operation: str) -> None: + connection = self.connection + if connection is None or not connection.in_transaction(): + raise RuntimeError(f"{operation} requires an open transaction on the [{self.config.driver}] connection") + @classmethod def get_query_grammar(cls) -> type[PostgresGrammar]: return PostgresGrammar diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/exceptions.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/exceptions.py index 1ec5288e..4ba6277c 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/exceptions.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/exceptions.py @@ -21,3 +21,10 @@ def __init__(self, model: object, relation: str | None): f"Attempted to lazy load [{relation}] on model [{self.model}]. " f'Eager load it with {self.model}.with_("{relation}") or await the model\'s load("{relation}") first.' ) + + +class LockTimeoutError(Exception): + def __init__(self, key: str | int, timeout_ms: int | None): + self.key = key + self.timeout_ms = timeout_ms + super().__init__(f"Timed out waiting for advisory lock [{key}] after {timeout_ms} ms") diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/facades/DB.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/facades/DB.py index 51a3e0bd..f1cde50e 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/facades/DB.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/facades/DB.py @@ -53,6 +53,14 @@ async def delete(cls, query: str, bindings: list | None = None) -> int: async def statement(cls, query: str, bindings: list | None = None) -> bool: return await cls.instance().connection(None).statement(query, bindings) + @classmethod + async def advisory_lock(cls, key: str | int, timeout_ms: int | None = None, connection: str | None = None) -> None: + await cls.instance().connection(connection).advisory_lock(key, timeout_ms) + + @classmethod + async def lock_timeout(cls, timeout_ms: int, connection: str | None = None) -> None: + await cls.instance().connection(connection).lock_timeout(timeout_ms) + @classmethod async def begin_transaction(cls, name: str | None = None) -> None: await cls.instance().connection(name).begin_transaction() diff --git a/fastapi_startkit/tests/masoniteorm/connections/test_advisory_lock.py b/fastapi_startkit/tests/masoniteorm/connections/test_advisory_lock.py new file mode 100644 index 00000000..ee5ca5e3 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/connections/test_advisory_lock.py @@ -0,0 +1,178 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import create_async_engine + +from fastapi_startkit.masoniteorm import LockTimeoutError, PostgresConfig, SQLiteConfig +from fastapi_startkit.masoniteorm.connections.advisory_lock import ( + BIGINT_MAX, + BIGINT_MIN, + LOCK_TIMEOUT_MAX_MS, + advisory_lock_key, + lock_timeout_value, +) +from fastapi_startkit.masoniteorm.connections.postgres_connection import PostgresConnection +from fastapi_startkit.masoniteorm.connections.sqlite_connection import SQliteConnection +from fastapi_startkit.masoniteorm.facades.DB import DB + + +class DriverError(Exception): + def __init__(self, sqlstate: str): + super().__init__(sqlstate) + self.sqlstate = sqlstate + + +@pytest.fixture +async def postgres(): + engine = create_async_engine("postgresql+asyncpg://app:secret@localhost:1/unused") + try: + yield PostgresConnection(engine, PostgresConfig(driver="postgres")) + finally: + await engine.dispose() + + +@pytest.fixture +def in_transaction(postgres): + active = MagicMock(closed=False) + active.in_transaction.return_value = True + token = postgres._connection_context.set(active) + try: + yield postgres + finally: + postgres._connection_context.reset(token) + + +@pytest.mark.parametrize( + ("key", "expected"), + [ + ("oauth:refresh", -4881159482311254756), + ("", -1970711489451281740), + ("é", -3811523526868657873), + ], +) +def test_str_keys_hash_to_stable_signed_bigints(key, expected): + assert advisory_lock_key(key) == expected + + +def test_distinct_str_keys_hash_differently(): + assert advisory_lock_key("user:1") != advisory_lock_key("user:2") + + +@pytest.mark.parametrize("key", [0, 42, -7, BIGINT_MIN, BIGINT_MAX]) +def test_int_keys_pass_through(key): + assert advisory_lock_key(key) == key + + +@pytest.mark.parametrize("key", [BIGINT_MIN - 1, BIGINT_MAX + 1]) +def test_int_keys_outside_bigint_are_rejected(key): + with pytest.raises(ValueError, match="bigint range"): + advisory_lock_key(key) + + +@pytest.mark.parametrize("key", [True, 1.5, None, b"key"]) +def test_other_key_types_are_rejected(key): + with pytest.raises(TypeError, match="str or int"): + advisory_lock_key(key) + + +@pytest.mark.parametrize("timeout_ms", [0, 1, 3000, LOCK_TIMEOUT_MAX_MS]) +def test_valid_timeouts_become_strings(timeout_ms): + assert lock_timeout_value(timeout_ms) == str(timeout_ms) + + +@pytest.mark.parametrize("timeout_ms", [-1, LOCK_TIMEOUT_MAX_MS + 1]) +def test_out_of_range_timeouts_are_rejected(timeout_ms): + with pytest.raises(ValueError, match="between 0 and"): + lock_timeout_value(timeout_ms) + + +@pytest.mark.parametrize("timeout_ms", [False, 1.5, "100", "1; DROP TABLE users"]) +def test_non_int_timeouts_are_rejected(timeout_ms): + with pytest.raises(TypeError, match="int number of milliseconds"): + lock_timeout_value(timeout_ms) + + +async def test_advisory_lock_requires_a_transaction(postgres): + with pytest.raises(RuntimeError, match="advisory_lock requires an open transaction"): + await postgres.advisory_lock("key", timeout_ms=100) + + +async def test_lock_timeout_requires_a_transaction(postgres): + with pytest.raises(RuntimeError, match="lock_timeout requires an open transaction"): + await postgres.lock_timeout(100) + + +async def test_invalid_arguments_fail_before_any_query(in_transaction): + with patch.object(in_transaction, "statement", AsyncMock()) as statement: + with pytest.raises(TypeError): + await in_transaction.advisory_lock("key", timeout_ms="100") + with pytest.raises(ValueError): + await in_transaction.advisory_lock(BIGINT_MAX + 1) + statement.assert_not_awaited() + + +async def test_advisory_lock_binds_timeout_and_key(in_transaction): + with patch.object(in_transaction, "statement", AsyncMock()) as statement: + await in_transaction.advisory_lock("oauth:refresh", timeout_ms=250) + assert [call.args for call in statement.await_args_list] == [ + ("SELECT set_config('lock_timeout', ?, true)", ["250"]), + ("SELECT pg_advisory_xact_lock(?)", [-4881159482311254756]), + ] + + +async def test_advisory_lock_without_timeout_leaves_lock_timeout_alone(in_transaction): + with patch.object(in_transaction, "statement", AsyncMock()) as statement: + await in_transaction.advisory_lock(7) + statement.assert_awaited_once_with("SELECT pg_advisory_xact_lock(?)", [7]) + + +async def test_lock_timeout_binds_value(in_transaction): + with patch.object(in_transaction, "statement", AsyncMock()) as statement: + await in_transaction.lock_timeout(0) + statement.assert_awaited_once_with("SELECT set_config('lock_timeout', ?, true)", ["0"]) + + +async def test_lock_not_available_becomes_lock_timeout_error(in_transaction): + driver_error = DBAPIError("SELECT pg_advisory_xact_lock($1)", {}, DriverError("55P03")) + with patch.object(in_transaction, "statement", AsyncMock(side_effect=[True, driver_error])): + with pytest.raises(LockTimeoutError) as raised: + await in_transaction.advisory_lock("oauth:refresh", timeout_ms=100) + assert raised.value.key == "oauth:refresh" + assert raised.value.timeout_ms == 100 + assert raised.value.__cause__ is driver_error + assert "oauth:refresh" in str(raised.value) + + +async def test_other_driver_errors_propagate(in_transaction): + driver_error = DBAPIError("SELECT pg_advisory_xact_lock($1)", {}, DriverError("40P01")) + with patch.object(in_transaction, "statement", AsyncMock(side_effect=driver_error)): + with pytest.raises(DBAPIError) as raised: + await in_transaction.advisory_lock(1) + assert raised.value is driver_error + + +async def test_sqlite_raises_not_implemented(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'locks.sqlite3'}") + connection = SQliteConnection(engine, SQLiteConfig(driver="sqlite")) + try: + async with connection.transaction(): + with pytest.raises(NotImplementedError, match="only supported on Postgres, not the sqlite driver"): + await connection.advisory_lock("key") + with pytest.raises(NotImplementedError, match="only supported on Postgres, not the sqlite driver"): + await connection.lock_timeout(100) + finally: + await connection.close() + await engine.dispose() + + +async def test_facade_forwards_to_the_named_connection(): + connection = MagicMock(advisory_lock=AsyncMock(), lock_timeout=AsyncMock()) + manager = MagicMock() + manager.connection.return_value = connection + with patch.object(DB, "_instance", manager): + await DB.advisory_lock("key", 100, connection="dev") + await DB.lock_timeout(50) + assert [call.args for call in manager.connection.call_args_list] == [("dev",), (None,)] + connection.advisory_lock.assert_awaited_once_with("key", 100) + connection.lock_timeout.assert_awaited_once_with(50) diff --git a/fastapi_startkit/tests/masoniteorm/postgres/test_advisory_lock.py b/fastapi_startkit/tests/masoniteorm/postgres/test_advisory_lock.py new file mode 100644 index 00000000..6a3e561b --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/postgres/test_advisory_lock.py @@ -0,0 +1,166 @@ +import asyncio +from contextvars import Context + +import pytest +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import create_async_engine + +from fastapi_startkit.masoniteorm import LockTimeoutError, PostgresConfig +from fastapi_startkit.masoniteorm.connections.advisory_lock import advisory_lock_key +from fastapi_startkit.masoniteorm.connections.postgres_connection import PostgresConnection + +from .fixtures.db import URL + +KEY = "tests:advisory-lock" + + +def make_connection(**engine_options) -> PostgresConnection: + engine = create_async_engine(URL, **engine_options) + return PostgresConnection(engine, PostgresConfig(driver="postgres", url=URL, database="database_app_test")) + + +@pytest.fixture +async def connection(): + connection = make_connection() + try: + yield connection + finally: + await connection.close() + await connection.engine.dispose() + + +async def is_locked(connection: PostgresConnection, key: str | int) -> bool: + async def probe() -> bool: + async with connection.transaction(): + row = await connection.select_one( + "SELECT pg_try_advisory_xact_lock(?) AS acquired", [advisory_lock_key(key)] + ) + return not row["acquired"] + + return await asyncio.create_task(probe(), context=Context()) + + +async def hold(connection: PostgresConnection, key: str | int, acquired: asyncio.Event, release: asyncio.Event): + async with connection.transaction(): + await connection.advisory_lock(key) + acquired.set() + await release.wait() + + +async def test_lock_is_held_inside_the_transaction(connection): + async with connection.transaction(): + await connection.advisory_lock(KEY) + assert await is_locked(connection, KEY) + assert not await is_locked(connection, "tests:other-key") + + +async def test_int_key_is_locked_as_is(connection): + async with connection.transaction(): + await connection.advisory_lock(424242) + row = await connection.select_one( + "SELECT count(*) AS held FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid() " + "AND classid = 0 AND objid = 424242 AND objsubid = 1" + ) + assert row["held"] == 1 + + +async def test_second_transaction_blocks_until_the_holder_commits(connection): + acquired, release = asyncio.Event(), asyncio.Event() + holder = asyncio.create_task(hold(connection, KEY, acquired, release)) + await acquired.wait() + + waiter_acquired, waiter_release = asyncio.Event(), asyncio.Event() + waiter = asyncio.create_task(hold(connection, KEY, waiter_acquired, waiter_release)) + await asyncio.sleep(0.3) + assert not waiter_acquired.is_set() + + release.set() + await holder + await asyncio.wait_for(waiter_acquired.wait(), timeout=2) + waiter_release.set() + await waiter + + +async def test_timeout_raises_lock_timeout_error(connection): + acquired, release = asyncio.Event(), asyncio.Event() + holder = asyncio.create_task(hold(connection, KEY, acquired, release)) + await acquired.wait() + + async def contend(): + async with connection.transaction(): + await connection.advisory_lock(KEY, timeout_ms=100) + + try: + with pytest.raises(LockTimeoutError) as raised: + await asyncio.wait_for(asyncio.create_task(contend()), timeout=5) + finally: + release.set() + await holder + + assert raised.value.key == KEY + assert raised.value.timeout_ms == 100 + assert isinstance(raised.value.__cause__, DBAPIError) + assert connection.connection is None + + +async def test_lock_is_released_on_commit(connection): + async with connection.transaction(): + await connection.advisory_lock(KEY) + assert not await is_locked(connection, KEY) + + +async def test_lock_is_released_on_rollback(connection): + with pytest.raises(ZeroDivisionError): + async with connection.transaction(): + await connection.advisory_lock(KEY) + 1 / 0 + assert not await is_locked(connection, KEY) + + +async def test_lock_taken_in_a_savepoint_is_held_until_the_outer_transaction_ends(connection): + async with connection.transaction(): + async with connection.transaction(): + await connection.advisory_lock(KEY) + assert await is_locked(connection, KEY) + assert not await is_locked(connection, KEY) + + +async def test_lock_timeout_ends_with_the_transaction(): + connection = make_connection(pool_size=1, max_overflow=0) + try: + async with connection.transaction(): + before = await connection.select_one( + "SELECT current_setting('lock_timeout') AS value, pg_backend_pid() AS pid" + ) + async with connection.transaction(): + await connection.lock_timeout(1234) + during = await connection.select_one("SELECT current_setting('lock_timeout') AS value") + await connection.advisory_lock(KEY, timeout_ms=4321) + overridden = await connection.select_one("SELECT current_setting('lock_timeout') AS value") + async with connection.transaction(): + after = await connection.select_one( + "SELECT current_setting('lock_timeout') AS value, pg_backend_pid() AS pid" + ) + finally: + await connection.close() + await connection.engine.dispose() + + assert during["value"] == "1234ms" + assert overridden["value"] == "4321ms" + assert after["value"] == before["value"] + assert after["pid"] == before["pid"] + + +async def test_facade_takes_the_lock_on_the_named_connection(monkeypatch): + from fastapi_startkit.masoniteorm.facades.DB import DB as Facade + + from .fixtures.db import DB + + monkeypatch.setattr(Facade, "_instance", DB) + connection = DB.connection("dev") + try: + async with connection.transaction(): + await Facade.advisory_lock(KEY, timeout_ms=500, connection="dev") + assert await is_locked(connection, KEY) + finally: + await DB.clear() From e70f8c3d8e3647178310d4bc6238076a92ee6b07 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sat, 31 Oct 2026 14:30:09 -0700 Subject: [PATCH 2/2] Renumber advisory lock ADR to 060 Co-Authored-By: Claude Opus 5.5 --- ...advisory-lock.md => 060-transaction-scoped-advisory-lock.md} | 2 +- docs/adr/AGENTS.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename docs/adr/{058-transaction-scoped-advisory-lock.md => 060-transaction-scoped-advisory-lock.md} (98%) diff --git a/docs/adr/058-transaction-scoped-advisory-lock.md b/docs/adr/060-transaction-scoped-advisory-lock.md similarity index 98% rename from docs/adr/058-transaction-scoped-advisory-lock.md rename to docs/adr/060-transaction-scoped-advisory-lock.md index 58d8a330..e5960609 100644 --- a/docs/adr/058-transaction-scoped-advisory-lock.md +++ b/docs/adr/060-transaction-scoped-advisory-lock.md @@ -1,4 +1,4 @@ -# 058 — Transaction-scoped advisory locks (`DB.advisory_lock`) and `LockTimeoutError` +# 060 — Transaction-scoped advisory locks (`DB.advisory_lock`) and `LockTimeoutError` Date: 2026-10-31 diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index e9c62ed1..e82c51a4 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -47,4 +47,4 @@ Read this index first, then open the records relevant to the change. Before impl | [052](052-sqlite-savepoint-isolation.md) | 2026-10-31 | SQLite savepoints start a real transaction | pysqlite/aiosqlite defer `BEGIN` until the first DML, so a savepoint opened first in an outer transaction became a top-level transaction and `RELEASE` committed it, leaking rows past `DatabaseTransaction`/`RefreshDatabase` rollbacks. `SQliteConnection` listens to the engine `savepoint` event and runs `BEGIN` when the driver connection is not in a transaction. On the shared `StaticPool` connection, a lost `BEGIN` race (`OperationalError` while a transaction is already open) is ignored, so concurrent nested transactions still succeed as on `main`. SQLAlchemy's `isolation_level=None` + `BEGIN` on every `begin` recipe was rejected because concurrent transactions on `SQLiteConfig`'s shared `StaticPool` connection then fail with `cannot start a transaction within a transaction`. | | [053](053-validated-reads-query-and-body.md) | 2026-10-31 | `Validated` reads the query string and the body | `Validated[Model]` declared a required `Body()`, so a `GET` with filters got `422 body: Field required` and apps bound models with `Query()`, skipping `authorize`, rules and `after`. The body parameter now defaults to `None`. Without a body, `GET`/`HEAD`/`OPTIONS`/`DELETE` validate `query_input`, while `POST`/`PUT`/`PATCH` still report `"": Field required` (per embed key when embedded), so a body-less cross-site form `POST` cannot pass on the query string (CSRF). `DELETE` always needs a CORS preflight. With a body, `request_input` merges `{**query, **body}` for an object; any other body, a JSON `null` included, passes through and fails with `model_type`. `query_input` keeps the keys pydantic accepts (`validation_alias`, `AliasChoices`, single-key `AliasPath`, the name with `validate_by_name`/`populate_by_name`). `collections.abc` `Sequence`/`Set` fields (not `str`/`bytes`) get every repeated value. Declared fields can be set from the URL on write methods with a JSON body, as with Laravel's `$request->all()`: keep sensitive fields off such models. `document_query_input` (installed by `FastAPIProvider`) replaces `requestBody` with `in: query` parameters on query-method routes. Chosen over a separate `ValidatedQuery` so one `FormRequest` serves `GET` and `POST`. | | [055](055-morph-name-equals-class-name.md) | 2026-10-31 | Morph relations accept the class name as the morph name | `MorphMany`/`MorphOne.get_record_key_lookup` raised `ValueError` whenever `Registry.get_morph_name` returned the class name, treating it as "not registered". `Model.__init_subclass__` registers every model with its class name as the default morph name, so every model without a `morph_map` alias (or mapped to its own class name) failed on `with_` and `load`. The lookup now returns `Registry.get_morph_name(cls)` as-is (`morph_map` aliases still win). `MorphMany`/`MorphOne.get_related` also forward `eagers` via `.with_(eagers or [])`, so nested loads such as `with_("media.lexeme")` work. `MorphTo`/`MorphToMany` were unaffected. | -| [058](058-transaction-scoped-advisory-lock.md) | 2026-10-31 | Transaction-scoped advisory locks and `LockTimeoutError` | `DB.advisory_lock(key, timeout_ms=None, connection=None)` / `Connection.advisory_lock` run `pg_advisory_xact_lock` (released at commit or rollback; no session-scoped variant), and `DB.lock_timeout(ms)` sets a transaction-local timeout with `set_config('lock_timeout', :ms, true)` (bound, no interpolation; validated `0..2147483647`). Both need an open transaction on the connection (`RuntimeError` otherwise), and non-Postgres connections raise `NotImplementedError`. An `int` key must fit `bigint`. A `str` key is hashed with `blake2b` (8 bytes, signed big-endian), which is stable across processes and versions, unlike `hashtext()` (internal, 32-bit, has changed) or `hash()`. SQLSTATE `55P03` while waiting becomes `LockTimeoutError` (exported from `fastapi_startkit.masoniteorm`) chained from the driver error. | +| [060](060-transaction-scoped-advisory-lock.md) | 2026-10-31 | Transaction-scoped advisory locks and `LockTimeoutError` | `DB.advisory_lock(key, timeout_ms=None, connection=None)` / `Connection.advisory_lock` run `pg_advisory_xact_lock` (released at commit or rollback; no session-scoped variant), and `DB.lock_timeout(ms)` sets a transaction-local timeout with `set_config('lock_timeout', :ms, true)` (bound, no interpolation; validated `0..2147483647`). Both need an open transaction on the connection (`RuntimeError` otherwise), and non-Postgres connections raise `NotImplementedError`. An `int` key must fit `bigint`. A `str` key is hashed with `blake2b` (8 bytes, signed big-endian), which is stable across processes and versions, unlike `hashtext()` (internal, 32-bit, has changed) or `hash()`. SQLSTATE `55P03` while waiting becomes `LockTimeoutError` (exported from `fastapi_startkit.masoniteorm`) chained from the driver error. |