Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions docs/adr/063-lock-with-a-value.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
# 063. `lock(value)` on the query builder, for `SKIP LOCKED` claims

Date: 2026-11-01

## Problem

A worker that claims pending rows from a shared table, such as a job queue or a per-user work list, needs concurrent workers to take different rows without waiting on each other. ADR 061 added `lock_for_update()` and `shared_lock()`, but a second worker that asks for the same rows with `FOR UPDATE` waits for the first to commit. Postgres 9.5+, MySQL 8.0.1+ and MariaDB 10.6+ solve this with `FOR UPDATE SKIP LOCKED`, and SQL Server with the `READPAST` hint. The builder could not render either, so applications wrote the claim as raw SQL.

## Alternatives

1. **A `skip_locked` flag: `lock_for_update(skip_locked=True)`.** It adds a new grammar key per option and grows with `NOWAIT`, `FOR NO KEY UPDATE` and the rest. Laravel has no such flag.
2. **`UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED) RETURNING ...` in one statement.** It needs `RETURNING` support in every grammar and hides the claim inside a single statement.
3. **Laravel's `lock($value = true)`.** `true` takes the update lock, `false` the shared lock, and a string is the dialect's lock clause, rendered verbatim. Laravel's database queue claims a job this way: `lock('FOR UPDATE SKIP LOCKED')->orderBy('id')->first()` in a transaction, then an update of that job by id.

## Decision

Option 3.

- `QueryBuilder.lock(value: bool | str = True)` sets `_lock` to `"update"` for `True`, `"share"` for `False`, or the string itself. `lock_for_update()` is `lock(True)` and `shared_lock()` is `lock(False)`. The last call wins. `Model.lock(value)` proxies to a new query.
- `BaseGrammar.process_locks()` renders the dialect clause for `"update"` and `"share"` and any other string verbatim, in the same slot: after `LIMIT`/`OFFSET` on Postgres and MySQL, and after the table on MSSQL (`WITH(ROWLOCK, UPDLOCK, READPAST)`). `SQLiteGrammar.process_locks()` renders nothing, as it already did for both locks. SQLite locks the whole database for a write transaction, so the same code runs in tests.
- A locked builder used as a `where_in` subquery keeps its lock, because the subquery compiles through the sub-builder's own grammar.
- A claim reads the matching rows with a lock inside `connection.transaction()`, iterates them and updates each by key, so the caller knows which rows it claimed without `RETURNING`.

ADR 061 rejected a stringly `lock(mode)` because an unknown mode silently rendered no lock. A string now renders verbatim, so a typo fails in the database instead of running unlocked.

A lock string is SQL, not a binding. It must be a literal written in the application's code, such as a module constant, and never built from request input, configuration a user can change, or any other untrusted value. `lock()` does not escape or validate it, just as `select_raw` does not, because the set of valid clauses differs by dialect and version.

The caller picks the clause for its dialect, as in Laravel's `DatabaseQueue::getLockForPopping()`. A string meant for one dialect is invalid SQL on another, except on SQLite, where every lock is a no-op.

## Validation

- `tests/masoniteorm/query/grammars/test_lock_value_grammar.py` snapshots `to_qmark()` on SQLite, MySQL, Postgres and MSSQL: `lock()` and `lock(True)` take the update lock, `lock(False)` the shared lock, a string renders verbatim after `ORDER BY` and `LIMIT` (after the table and with `TOP` on MSSQL, nothing on SQLite), a locked `where_in` subquery keeps its lock, and the last call wins.
- `tests/masoniteorm/postgres/test_postgres_skip_locked.py` runs concurrent transactions on Postgres. Three workers each read up to two pending rows with `lock("FOR UPDATE SKIP LOCKED")`, update each by id and hold their transaction. They claim 2, 2 and 1 of five rows, no row twice, and all five end up processing. A worker using a plain `lock()` waits for the holder and never gets the held row.
- `test_lock_grammar.py`, `test_sqlite_lock.py` and `test_postgres_lock.py` from ADR 061 pass unchanged.
- The full framework suite passes with coverage at or above the threshold.
1 change: 1 addition & 0 deletions docs/adr/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,4 @@ Read this index first, then open the records relevant to the change. Before impl
| [059](059-subquery-where-in-and-select-exists.md) | 2026-10-31 | Subqueries in `where_in`/`where_not_in`, `where_not_exists` and `add_select_exists` | Membership, anti-join and per-row exists flags needed `DB.select`: `where_in(col, builder)` raised `'QueryBuilder' object is not iterable` and no method emitted `NOT EXISTS` or an `EXISTS` column. `where_in`/`where_not_in` now take a builder or a callable (like `select_sub`) and render `col IN (subquery)` via `SubSelectExpression`; iterables are unchanged. `where_exists`/`or_where_exists` also take a callable; new `where_not_exists`/`or_where_not_exists`. `add_select_exists(alias, query)` appends `EXISTS(subquery) AS alias` (selecting `{table}.*` first when nothing is selected) through a `SubGroupExpression(exists=True)` and the grammar's `select_exists_string()`; MSSQL renders `CASE WHEN EXISTS(...) THEN 1 ELSE 0 END`. A shared `_resolve_subquery` raises one `TypeError`. Subquery bindings stay bound and land in SQL order. `Model` proxies all. Limits: Postgres returns `bool`, others `1`/`0`; `NOT IN` over a subquery with `NULL` matches nothing. Chosen over raw SQL or separate `*_sub` method names. |
| [061](061-builder-row-locks.md) | 2026-10-31 | Row locks on the query builder: `lock_for_update` and `shared_lock` | The grammars already rendered a per-dialect lock clause, but no builder method set it and `get_grammar()` never passed `lock`, so apps wrote `SELECT ... FOR UPDATE` by hand. `QueryBuilder.lock_for_update()`/`shared_lock()` set `_lock` to `"update"`/`"share"` (last call wins) and `get_grammar()` passes it: Postgres `FOR UPDATE`/`FOR SHARE`, MySQL `FOR UPDATE`/`LOCK IN SHARE MODE`, SQLite a no-op, and MSSQL now `WITH(ROWLOCK, UPDLOCK, HOLDLOCK)`/`WITH(ROWLOCK, HOLDLOCK)` instead of a bare `ROWLOCK` hint. `Model` proxies both. Only selects lock; `new()` groups and the `paginate()` count start unlocked. Chosen over update-first, whose `FOR NO KEY UPDATE` does not wait for an in-flight foreign-key insert, and over a stringly `lock(mode)`. Known limit: Postgres refuses a lock with aggregates, `DISTINCT` or `GROUP BY`. |
| [062](062-update-uses-the-updated-table.md) | 2026-11-01 | `update()` compiles against the updated table | `update()` built its grammar with `self.grammar()`, keeping the default table `users`, so MySQL and MSSQL prefixed every `SET`/`WHERE` column with `users` whatever table was updated (`UPDATE [queues] SET [users].[status] = ?`). It now passes `table=self._table`. Postgres and SQLite do not prefix update columns. `update()` still returns the row count; a queue claim reads locked rows, iterates and updates each by key, as Laravel's database queue does, so no `UPDATE ... RETURNING` is added. |
| [063](063-lock-with-a-value.md) | 2026-11-01 | `lock(value)` on the query builder, for `SKIP LOCKED` claims | Concurrent workers claiming pending rows waited on each other under `FOR UPDATE`, and the builder could not render `FOR UPDATE SKIP LOCKED`. Laravel's `QueryBuilder.lock(value=True)`: `True` takes the update lock, `False` the shared lock, a string is a trusted, code-literal dialect clause rendered verbatim (never user input) (e.g. `FOR UPDATE SKIP LOCKED`, MSSQL `WITH(ROWLOCK, UPDLOCK, READPAST)`); SQLite renders nothing. `lock_for_update()`/`shared_lock()` delegate to it and `Model.lock` proxies it. A locked `where_in` subquery keeps its lock. A claim reads locked rows in a transaction, iterates them and updates each by key, as Laravel's database queue does. Chosen over a `skip_locked` flag and over `UPDATE ... RETURNING`. |
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ def __init__(self, connection: "Connection", grammar: Any, processor: Any):
self._group_by = ()
self._having = ()
self._distinct = False
self._lock: Literal["update", "share", False] = False
self._lock: str | Literal[False] = False

self._sql = ""
self._bindings = ()
Expand Down Expand Up @@ -393,13 +393,18 @@ def distinct(self) -> "Self":
self._distinct = True
return self

def lock_for_update(self) -> "Self":
self._lock = "update"
def lock(self, value: bool | str = True) -> "Self":
if isinstance(value, str):
self._lock = value
else:
self._lock = "update" if value else "share"
return self

def lock_for_update(self) -> "Self":
return self.lock(True)

def shared_lock(self) -> "Self":
self._lock = "share"
return self
return self.lock(False)

async def aggregate(self, function: str, column: str):
self._aggregates += (AggregateExpression(function, column),)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,10 @@ def not_between(cls, column: str, low: Any, high: Any) -> "QueryBuilder[Self]":
def distinct(cls) -> "QueryBuilder[Self]":
return cls.query().distinct()

@classmethod
def lock(cls, value: bool | str = True) -> "QueryBuilder[Self]":
return cls.query().lock(value)

@classmethod
def lock_for_update(cls) -> "QueryBuilder[Self]":
return cls.query().lock_for_update()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -157,18 +157,6 @@ def _compile_select(self, qmark=False):
return self

def _compile_update(self, query=None, values=None, qmark=False):
"""Compiles an update query statement.

Arguments:
query -- The QueryBuilder instance (provides table/wheres context).
values {list} -- Runtime UpdateQueryExpression list.

Keyword Arguments:
qmark {bool} -- Use positional ? bindings. (default: {False})

Returns:
self
"""
self._action = "update"

if values is not None:
Expand Down Expand Up @@ -567,7 +555,7 @@ def process_offset(self):
return self.offset_string().format(offset=self._offset, limit=self._limit or 1)

def process_locks(self):
return self.locks.get(self.lock, "")
return self.locks.get(self.lock, self.lock or "")

def process_having(self, qmark=False):
"""Compiles having expression.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@ class SQLiteGrammar(BaseGrammar):

locks = {"share": "", "update": ""}

def process_locks(self):
return ""

def select_format(self):
return "SELECT {keyword} {columns} FROM {table} {joins} {wheres} {group_by} {having} {order_by} {limit} {offset} {lock}"

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
import asyncio
import contextlib

from fastapi_startkit.masoniteorm import Model

from .fixtures.db import DB, schema
from .test_case import TestCase

BLOCKED_FOR = 0.3
SETTLES_WITHIN = 5


class PskQueueItem(Model):
__table__ = "psk_queue_items"
__timestamps__ = None
id: int
status: str


def pending(batch):
return PskQueueItem.where("status", "pending").order_by("id").limit(batch)


class TestPostgresSkipLocked(TestCase):
async def asyncSetUp(self):
await super().asyncSetUp()
self.release = asyncio.Event()
self.holders = []
await schema.drop_table_if_exists("psk_queue_items")
async with await schema.create("psk_queue_items") as table:
table.increments("id")
table.string("status")
await PskQueueItem.query().insert([{"status": "pending"} for _ in range(5)])

async def asyncTearDown(self):
self.release.set()
for task in self.holders:
with contextlib.suppress(Exception, asyncio.CancelledError):
await asyncio.wait_for(task, SETTLES_WITHIN)
await schema.drop_table_if_exists("psk_queue_items")
await super().asyncTearDown()

async def claim_and_hold(self, batch, lock):
claimed = asyncio.get_running_loop().create_future()

async def worker():
async with DB.connection().transaction():
ids = []
for item in await lock(pending(batch)).get():
await PskQueueItem.where("id", item.id).update({"status": "processing"})
ids.append(item.id)
claimed.set_result(ids)
await self.release.wait()

self.holders.append(asyncio.ensure_future(worker()))
return claimed

async def test_concurrent_claims_skip_each_others_rows(self):
def skip_locked(query):
return query.lock("FOR UPDATE SKIP LOCKED")

first = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN)
second = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN)
third = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN)

self.assertEqual(len(first), 2)
self.assertEqual(len(second), 2)
self.assertEqual(len(third), 1)
self.assertEqual(len(set(first) | set(second) | set(third)), 5)

self.release.set()
await asyncio.gather(*self.holders)
self.assertEqual(await PskQueueItem.where("status", "processing").count(), 5)

async def test_a_plain_update_lock_waits_for_the_holder(self):
def skip_locked(query):
return query.lock("FOR UPDATE SKIP LOCKED")

held = await asyncio.wait_for(await self.claim_and_hold(1, skip_locked), SETTLES_WITHIN)
blocked = await self.claim_and_hold(1, lambda query: query.lock())

done, _ = await asyncio.wait({blocked}, timeout=BLOCKED_FOR)
self.assertEqual(done, set())
self.release.set()
self.assertNotIn(held[0], await asyncio.wait_for(blocked, SETTLES_WITHIN))

async def test_one_update_through_a_skip_locked_subquery_claims_a_free_row(self):
held = await asyncio.wait_for(
await self.claim_and_hold(1, lambda query: query.lock("FOR UPDATE SKIP LOCKED")), SETTLES_WITHIN
)
next_free = pending(1).select("id").lock("FOR UPDATE SKIP LOCKED")

updated = await asyncio.wait_for(
PskQueueItem.where("id", next_free).update({"status": "processing"}), SETTLES_WITHIN
)

processing = [item.id for item in await PskQueueItem.where("status", "processing").get()]
self.assertEqual(updated, 1)
self.assertEqual(len(processing), 1)
self.assertNotIn(held[0], processing)
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
import unittest

from fastapi_startkit.masoniteorm.models.builder import QueryBuilder
from fastapi_startkit.masoniteorm.query.grammars.MSSQLGrammar import MSSQLGrammar
from fastapi_startkit.masoniteorm.query.grammars.MySQLGrammar import MySQLGrammar
from fastapi_startkit.masoniteorm.query.grammars.PostgresGrammar import PostgresGrammar
from fastapi_startkit.masoniteorm.query.grammars.SQLiteGrammar import SQLiteGrammar

GRAMMARS = {
"sqlite": SQLiteGrammar,
"mysql": MySQLGrammar,
"postgres": PostgresGrammar,
"mssql": MSSQLGrammar,
}

SKIP_LOCKED = {
"sqlite": "FOR UPDATE SKIP LOCKED",
"mysql": "FOR UPDATE SKIP LOCKED",
"postgres": "FOR UPDATE SKIP LOCKED",
"mssql": "WITH(ROWLOCK, UPDLOCK, READPAST)",
}

EXPECTED = {
"sqlite": {
"update": 'SELECT * FROM "queues" WHERE "queues"."id" = ?',
"share": 'SELECT * FROM "queues" WHERE "queues"."id" = ?',
"string": 'SELECT * FROM "queues" WHERE "queues"."status" = ? ORDER BY "id" ASC LIMIT 1',
"subquery": 'SELECT * FROM "queues" WHERE "queues"."id" IN (SELECT "queues"."id" FROM "queues" WHERE "queues"."status" = ? LIMIT 1)',
},
"mysql": {
"update": "SELECT * FROM `queues` WHERE `queues`.`id` = ? FOR UPDATE",
"share": "SELECT * FROM `queues` WHERE `queues`.`id` = ? LOCK IN SHARE MODE",
"string": "SELECT * FROM `queues` WHERE `queues`.`status` = ? ORDER BY `id` ASC LIMIT 1 FOR UPDATE SKIP LOCKED",
"subquery": "SELECT * FROM `queues` WHERE `queues`.`id` IN (SELECT `queues`.`id` FROM `queues` WHERE `queues`.`status` = ? LIMIT 1 FOR UPDATE SKIP LOCKED)",
},
"postgres": {
"update": 'SELECT * FROM "queues" WHERE "queues"."id" = ? FOR UPDATE',
"share": 'SELECT * FROM "queues" WHERE "queues"."id" = ? FOR SHARE',
"string": 'SELECT * FROM "queues" WHERE "queues"."status" = ? ORDER BY "id" ASC LIMIT 1 FOR UPDATE SKIP LOCKED',
"subquery": 'SELECT * FROM "queues" WHERE "queues"."id" IN (SELECT "queues"."id" FROM "queues" WHERE "queues"."status" = ? LIMIT 1 FOR UPDATE SKIP LOCKED)',
},
"mssql": {
"update": "SELECT * FROM [queues] WITH(ROWLOCK, UPDLOCK, HOLDLOCK) WHERE [queues].[id] = ?",
"share": "SELECT * FROM [queues] WITH(ROWLOCK, HOLDLOCK) WHERE [queues].[id] = ?",
"string": "SELECT TOP 1 * FROM [queues] WITH(ROWLOCK, UPDLOCK, READPAST) WHERE [queues].[status] = ? ORDER BY [id] ASC",
"subquery": "SELECT * FROM [queues] WHERE [queues].[id] IN (SELECT TOP 1 [queues].[id] FROM [queues] WITH(ROWLOCK, UPDLOCK, READPAST) WHERE [queues].[status] = ?)",
},
}


def qb(grammar):
return QueryBuilder(connection=None, grammar=grammar, processor=None).table("queues")


class TestLockValueGrammar(unittest.TestCase):
def test_lock_true_takes_the_update_lock(self):
for name, grammar in GRAMMARS.items():
with self.subTest(grammar=name):
self.assertEqual(qb(grammar).where("id", 7).lock().to_qmark(), EXPECTED[name]["update"])
self.assertEqual(qb(grammar).where("id", 7).lock(True).to_qmark(), EXPECTED[name]["update"])

def test_lock_false_takes_the_shared_lock(self):
for name, grammar in GRAMMARS.items():
with self.subTest(grammar=name):
self.assertEqual(qb(grammar).where("id", 7).lock(False).to_qmark(), EXPECTED[name]["share"])

def test_a_lock_string_renders_verbatim_except_on_sqlite(self):
for name, grammar in GRAMMARS.items():
with self.subTest(grammar=name):
query = qb(grammar).where("status", "pending").order_by("id").limit(1).lock(SKIP_LOCKED[name])
self.assertEqual(query.to_qmark(), EXPECTED[name]["string"])
self.assertEqual(list(query._bindings), ["pending"])

def test_a_locked_subquery_keeps_its_lock(self):
for name, grammar in GRAMMARS.items():
with self.subTest(grammar=name):
claimable = qb(grammar).select("id").where("status", "pending").limit(1).lock(SKIP_LOCKED[name])
query = qb(grammar).where_in("id", claimable)
self.assertEqual(query.to_qmark(), EXPECTED[name]["subquery"])

def test_the_last_lock_call_wins(self):
for name, grammar in GRAMMARS.items():
with self.subTest(grammar=name):
query = qb(grammar).where("id", 7).lock("FOR UPDATE NOWAIT").lock_for_update()
self.assertEqual(query.to_qmark(), EXPECTED[name]["update"])
Loading