DB.advisory_lock and LockTimeoutError (Postgres, transaction-scoped)

docs/adr/060-transaction-scoped-advisory-lock.md

@@ -0,0 +1,79 @@
+# 060 — 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%.

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. |
+| [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. |

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",

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)

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")

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

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")

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()