fix: connection-scoped reorder serialization; recursive SQL-leak scan with span-exact self-exclusion

- reorder lock is now per-(connection, loop) via module-level registry
  shared by all ChildrenDao instances over one wrapper, matching the
  migration runner's approach; race test now uses two DAO instances on
  one connection
- SQL-leak guard scans recursively (rglob); self-scan excludes only the
  two guard functions' exact source spans via ast line numbers, so code
  anywhere else in the file is still scanned
This commit is contained in:
Joshua King
2026-09-14 07:16:47 -04:00
parent f00ec03c78
commit 98e342df17
2 changed files with 78 additions and 34 deletions
+27 -10
View File
@@ -61,6 +61,26 @@ _CHILD_COLUMNS = (
) )
#: One asyncio.Lock per (connection wrapper, running loop), shared by
#: every ChildrenDao instance over the same database: the wrapper
#: rejects a second concurrent transaction with RuntimeError, so ALL
#: transactional DAO operations on one connection must queue here, not
#: per-instance. Keyed like the migration runner's lock; entries are
#: never evicted (a config entry holds one wrapper per process).
_CONNECTION_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
def _connection_lock(database) -> asyncio.Lock:
"""Return the shared transaction-serializing lock for ``database``."""
loop = asyncio.get_running_loop()
key = (id(database), id(loop))
lock = _CONNECTION_LOCKS.get(key)
if lock is None:
lock = asyncio.Lock()
_CONNECTION_LOCKS[key] = lock
return lock
class ChildrenDao: class ChildrenDao:
"""Typed async data access for the ``children`` table. """Typed async data access for the ``children`` table.
@@ -71,10 +91,6 @@ class ChildrenDao:
def __init__(self, database) -> None: def __init__(self, database) -> None:
self._database = database self._database = database
#: Serializes whole reorder() runs on this DAO instance: the
#: wrapper rejects concurrent transactions, so racing reorders
#: must queue here instead of colliding.
self._reorder_lock = asyncio.Lock()
async def create( async def create(
self, self,
@@ -184,11 +200,12 @@ class ChildrenDao:
caller bug. caller bug.
Concurrent reorder() calls on one connection are serialized by Concurrent reorder() calls on one connection are serialized by
a per-instance asyncio lock: the wrapper's transaction() a connection-scoped asyncio lock (shared across every ChildrenDao
rejects a second concurrent transaction with RuntimeError, so over the same wrapper, however many instances): the wrapper's
racing callers would crash instead of queueing. The lock makes transaction() rejects a second concurrent transaction with
the second call wait, then run against the first call's final RuntimeError, so racing callers would crash instead of
state. queueing. The lock makes the second call wait, then run
against the first call's final state.
""" """
if not ordered_ids: if not ordered_ids:
return return
@@ -197,7 +214,7 @@ class ChildrenDao:
"reorder() received duplicate child ids; every id must " "reorder() received duplicate child ids; every id must "
"appear at most once" "appear at most once"
) )
async with self._reorder_lock: async with _connection_lock(self._database):
async with self._database.transaction(): async with self._database.transaction():
offset = len(ordered_ids) offset = len(ordered_ids)
await self._database.execute_many( await self._database.execute_many(
+51 -24
View File
@@ -300,24 +300,28 @@ def test_reorder_rejects_duplicate_ids(tmp_path) -> None:
def test_reorder_concurrent_calls_serialize(tmp_path) -> None: def test_reorder_concurrent_calls_serialize(tmp_path) -> None:
"""Two concurrent reorders queue on the transaction, both succeed. """Concurrent reorders from DIFFERENT DAO instances queue safely.
The second call sees the first call's final positions as its Two ChildrenDao objects over one connection race two reorders: the
starting state; both end with a consistent 0..n-1 assignment. connection-scoped lock makes the second wait, then run against the
first call's final state. Without the lock the wrapper's
transaction() would reject the second concurrent transaction with
RuntimeError.
""" """
async def _main() -> None: async def _main() -> None:
database = NestQuestDatabase(_make_hass_mock()) database = NestQuestDatabase(_make_hass_mock())
await database.open(tmp_path / "reorder-concurrent.db") await database.open(tmp_path / "reorder-concurrent.db")
try: try:
await apply_migrations(database) await apply_migrations(database)
dao, _ = _daos(database) dao_a, _ = _daos(database)
a = await dao.create("Ada", NOW) a = await dao_a.create("Ada", NOW)
b = await dao.create("Bo", NOW) b = await dao_a.create("Bo", NOW)
task_1 = asyncio.ensure_future(dao.reorder([b.id, a.id])) dao_b, _ = _daos(database) # second instance, same connection
task_1 = asyncio.ensure_future(dao_a.reorder([b.id, a.id]))
await asyncio.sleep(0) await asyncio.sleep(0)
task_2 = asyncio.ensure_future(dao.reorder([a.id, b.id])) task_2 = asyncio.ensure_future(dao_b.reorder([a.id, b.id]))
await asyncio.gather(task_1, task_2) await asyncio.gather(task_1, task_2)
children = await dao.list_all() children = await dao_a.list_all()
orders = {c.display_name: c.sort_order for c in children} orders = {c.display_name: c.sort_order for c in children}
assert sorted(orders.values()) == [0, 1] assert sorted(orders.values()) == [0, 1]
finally: finally:
@@ -451,9 +455,9 @@ def test_admin_records_are_typed_not_raw_rows(tmp_path) -> None:
def test_children_and_admin_sql_lives_only_in_dao_module() -> None: def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
"""Guardrail: children/admin_users SQL may appear only in the DAO """Guardrail: children/admin_users SQL may appear only in the DAO
(and the schema DDL declarations). Scans the integration package (and the schema DDL declarations). Scans the integration package
AND the repo's test files, matching FROM/INTO/UPDATE/DELETE FROM/ AND the repo's test files RECURSIVELY, matching FROM/INTO/UPDATE/
JOIN against either table — including DELETE FROM children, the DELETE FROM/JOIN against either table — including DELETE FROM
hard-delete path the guardrail forbids. children, the hard-delete path the guardrail forbids.
Exclusions are narrowly justified: schema.py declares the tables' Exclusions are narrowly justified: schema.py declares the tables'
DDL; migrations.py applies that DDL; test_schema.py and DDL; migrations.py applies that DDL; test_schema.py and
@@ -486,11 +490,11 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
) )
offenders: list[str] = [] offenders: list[str] = []
for root in scan_roots: for root in scan_roots:
for py in sorted(root.glob("*.py")): for py in sorted(root.rglob("*.py")):
if py.name in allowed: if py.name in allowed:
continue continue
if sql_pattern.search(py.read_text()): if sql_pattern.search(py.read_text()):
offenders.append(py.name) offenders.append(str(py.relative_to(repo_root)))
assert offenders == [], ( assert offenders == [], (
f"SQL touching children/admin_users leaked into: {offenders}" f"SQL touching children/admin_users leaked into: {offenders}"
) )
@@ -498,23 +502,46 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: def test_dao_test_file_uses_dao_not_raw_table_sql() -> None:
"""This test module must exercise the DAO, not raw SQL, for these """This test module must exercise the DAO, not raw SQL, for these
tables — except inside this guard itself. Any SQL statement naming tables — except inside the two guard functions themselves, whose
children or admin_users in a string literal elsewhere here is a source spans are excluded precisely (by line range), so any code
violation. added elsewhere in this file, before or after the guards, is still
scanned.
""" """
import ast
import re import re
from pathlib import Path from pathlib import Path
text = Path(__file__).read_text() path = Path(__file__)
# Cut everything from the guard function to the end of file: the text = path.read_text()
# guard's own patterns mention the keywords in regex strings, not lines = text.splitlines(keepends=True)
# executable SQL. tree = ast.parse(text)
head = text[: text.index("def test_children_and_admin_sql_lives")] guard_names = {
"test_children_and_admin_sql_lives_only_in_dao_module",
"test_dao_test_file_uses_dao_not_raw_table_sql",
}
excluded: list[tuple[int, int]] = []
for node in ast.walk(tree):
if (
isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name in guard_names
):
excluded.append((node.lineno, node.end_lineno or node.lineno))
sql_pattern = re.compile( sql_pattern = re.compile(
r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM)\s+" r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM)\s+"
r'["\`]*(children|admin_users)\b', r'["\`]*(children|admin_users)\b',
re.IGNORECASE, re.IGNORECASE,
) )
assert sql_pattern.search(head) is None, ( offenders: list[str] = []
"tests must go through the DAO, not raw SQL, for these tables" excluded_ranges = set()
for start, end in excluded:
excluded_ranges.update(range(start, end + 1))
for number, line in enumerate(lines, start=1):
if number in excluded_ranges:
continue
if sql_pattern.search(line):
offenders.append(f"{path.name}:{number}")
assert offenders == [], (
"tests must go through the DAO, not raw SQL, for these tables; "
f"found: {offenders}"
) )