Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: make db wrapper executor-safe, serialized, and leak-proof

Browse Source
This commit is contained in:
Joshua King
2026-09-13 21:07:42 -04:00
parent ae7a064d63
commit a56a7e3d37
4 changed files with 525 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/nestquest/__init__.py
+22 -8
View File
@@ -50,11 +50,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await listener_hass.config_entries.async_reload(listener_entry.entry_id)
database = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
try:
remove_update_listener = entry.add_update_listener(_async_update_listener)
except BaseException:
await database.close()
raise
runtime_data = NestQuestRuntimeData(
entry_id=entry.entry_id,
options=dict(entry.options),
database=database,
remove_update_listener=entry.add_update_listener(_async_update_listener),
remove_update_listener=remove_update_listener,
)
entry.runtime_data = runtime_data
hass.data[DOMAIN][entry.entry_id] = runtime_data
@@ -71,13 +76,22 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
runtime_data = getattr(entry, "runtime_data", None)
if runtime_data is not None:
remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
if remove_update_listener is not None:
result = remove_update_listener()
if inspect.isawaitable(result):
await result
database = getattr(runtime_data, "database", None)
if database is not None:
await database.close()
unload_error: BaseException | None = None
try:
if remove_update_listener is not None:
result = remove_update_listener()
if inspect.isawaitable(result):
await result
except BaseException as err:
unload_error = err
finally:
database = getattr(runtime_data, "database", None)
if database is not None:
await database.close()
if unload_error is not None:
if getattr(entry, "runtime_data", None) is not None:
entry.runtime_data = None
raise unload_error
if getattr(entry, "runtime_data", None) is not None:
entry.runtime_data = None
return True
custom_components/nestquest/db.py
+181 -56
View File
@@ -1,27 +1,72 @@
"""Async SQLite connection wrapper for the NestQuest integration."""
from __future__ import annotations
import asyncio
import sqlite3
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from homeassistant.core import HomeAssistant
@dataclass(frozen=True)
class ExecutionResult:
"""Plain-data outcome of one statement.
``rowcount`` and ``lastrowid`` mirror the corresponding
:class:`sqlite3.Cursor` attributes; both are captured on the executor
thread so no sqlite3 object ever crosses back to the event loop.
"""
rowcount: int
lastrowid: int | None
class NestQuestDatabase:
"""Async wrapper around a single SQLite connection.
Every sqlite3 call is delegated to the Home Assistant executor through
``hass.async_add_executor_job`` so the event loop is never blocked.
``hass.async_add_executor_job`` so the event loop is never blocked, and
only plain Python data (row tuples and :class:`ExecutionResult`) is
returned to the event loop.
Every operation is serialized through one asyncio.Lock per wrapper. A
``transaction()`` holds that lock for its whole BEGIN..COMMIT span, so
interleaved coroutines cannot corrupt the transaction and concurrent
BEGINs are impossible; statements issued while a transaction is open
queue until it completes. Concurrent transactions on one wrapper are
not supported: a second ``transaction()`` entered while one is active
raises :class:`RuntimeError`. ``close()`` serializes on the same lock
and is idempotent.
"""
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the wrapper around a Home Assistant instance."""
self._hass = hass
self._conn: sqlite3.Connection | None = None
self._cursor: sqlite3.Cursor | None = None
self._rowcount: int = -1
self._lastrowid: int | None = None
self._lock: asyncio.Lock | None = None
self._lock_loop: asyncio.AbstractEventLoop | None = None
self._in_transaction = False
self._tx_task: asyncio.Task | None = None
def _get_lock(self) -> asyncio.Lock:
"""Return the per-wrapper lock, (re)bound to the running loop.
HA drives every wrapper call from one event loop, so the lock is
created once and reused; re-binding only when the running loop
differs keeps the wrapper usable from plain synchronous drivers
that run each call on a fresh loop.
"""
loop = asyncio.get_running_loop()
if self._lock is None or self._lock_loop is not loop:
self._lock = asyncio.Lock()
self._lock_loop = loop
return self._lock
@property
def connected(self) -> bool:
@@ -31,109 +76,189 @@ class NestQuestDatabase:
@property
def rowcount(self) -> int:
"""Row count of the most recent statement, or -1 when unset."""
return self._cursor.rowcount if self._cursor is not None else -1
return self._rowcount
@property
def lastrowid(self) -> int | None:
"""Row id of the most recent insert, or None when unset."""
return self._cursor.lastrowid if self._cursor is not None else None
return self._lastrowid
@asynccontextmanager
async def _serialized(self) -> AsyncIterator[None]:
"""Hold the wrapper lock unless the caller owns the active transaction.
A transaction holds the lock for its whole BEGIN..COMMIT span;
statements issued by the transaction's own task are already
serialized by it and must not re-acquire the non-reentrant lock.
Any other task queues on the lock until the transaction ends.
"""
if (
self._in_transaction
and self._tx_task is not None
and asyncio.current_task() is self._tx_task
):
yield
return
async with self._get_lock():
yield
async def open(self, path: Path | str) -> NestQuestDatabase:
"""Open the SQLite file with WAL journal mode and foreign keys on."""
if self._conn is not None:
return self
async with self._get_lock():
if self._conn is not None:
return self
def _open() -> sqlite3.Connection:
conn = sqlite3.connect(
path, check_same_thread=False, isolation_level=None
)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def _open() -> sqlite3.Connection:
conn = sqlite3.connect(
path, check_same_thread=False, isolation_level=None
)
try:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
except BaseException:
conn.close()
raise
return conn
self._conn = await self._hass.async_add_executor_job(_open)
self._conn = await self._hass.async_add_executor_job(_open)
return self
async def execute(
self, sql: str, parameters: Sequence[Any] = ()
) -> sqlite3.Cursor:
"""Run one statement in the executor and return its cursor."""
conn = self._require_conn()
) -> ExecutionResult:
"""Run one statement in the executor and return its plain result."""
async with self._serialized():
conn = self._require_conn()
def _execute() -> sqlite3.Cursor:
return conn.execute(sql, parameters)
def _execute() -> ExecutionResult:
cursor = conn.execute(sql, parameters)
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
self._cursor = await self._hass.async_add_executor_job(_execute)
return self._cursor
result = await self._hass.async_add_executor_job(_execute)
self._rowcount = result.rowcount
self._lastrowid = result.lastrowid
return result
async def execute_many(
self, sql: str, parameters: Sequence[Sequence[Any]]
) -> sqlite3.Cursor:
"""Run executemany in the executor and return its cursor."""
conn = self._require_conn()
) -> ExecutionResult:
"""Run executemany in the executor and return its plain result."""
async with self._serialized():
conn = self._require_conn()
def _execute_many() -> sqlite3.Cursor:
return conn.executemany(sql, parameters)
def _execute_many() -> ExecutionResult:
cursor = conn.executemany(sql, parameters)
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
self._cursor = await self._hass.async_add_executor_job(_execute_many)
return self._cursor
result = await self._hass.async_add_executor_job(_execute_many)
self._rowcount = result.rowcount
self._lastrowid = result.lastrowid
return result
async def fetch_one(
self, sql: str, parameters: Sequence[Any] = ()
) -> tuple | None:
"""Run a query in the executor and return its first row, or None."""
cursor = await self.execute(sql, parameters)
async with self._serialized():
conn = self._require_conn()
def _fetch_one() -> tuple | None:
return cursor.fetchone()
def _fetch_one() -> tuple[tuple | None, int, int | None]:
cursor = conn.execute(sql, parameters)
return cursor.fetchone(), cursor.rowcount, cursor.lastrowid
return await self._hass.async_add_executor_job(_fetch_one)
row, rowcount, lastrowid = await self._hass.async_add_executor_job(
_fetch_one
)
self._rowcount = rowcount
self._lastrowid = lastrowid
return row
async def fetch_all(
self, sql: str, parameters: Sequence[Any] = ()
) -> list[tuple]:
"""Run a query in the executor and return all of its rows."""
cursor = await self.execute(sql, parameters)
async with self._serialized():
conn = self._require_conn()
def _fetch_all() -> list[tuple]:
return cursor.fetchall()
def _fetch_all() -> tuple[list[tuple], int, int | None]:
cursor = conn.execute(sql, parameters)
return cursor.fetchall(), cursor.rowcount, cursor.lastrowid
return await self._hass.async_add_executor_job(_fetch_all)
rows, rowcount, lastrowid = await self._hass.async_add_executor_job(
_fetch_all
)
self._rowcount = rowcount
self._lastrowid = lastrowid
return rows
@asynccontextmanager
async def transaction(self) -> AsyncIterator[None]:
"""Run a block inside BEGIN/COMMIT, rolling back on any exception."""
conn = self._require_conn()
"""Run a block inside BEGIN/COMMIT, rolling back on any exception.
def _begin() -> None:
conn.execute("BEGIN")
The wrapper lock is held for the entire BEGIN..COMMIT span, so no
other statement can interleave inside the transaction. Entering a
second transaction while one is active raises RuntimeError.
"""
if self._in_transaction:
raise RuntimeError(
"A NestQuest transaction is already active on this "
"connection; concurrent transactions are not supported"
)
async with self._get_lock():
conn = self._require_conn()
self._in_transaction = True
self._tx_task = asyncio.current_task()
try:
await self._hass.async_add_executor_job(_begin)
try:
yield
except BaseException:
def _begin() -> None:
conn.execute("BEGIN")
def _rollback() -> None:
conn.execute("ROLLBACK")
await self._hass.async_add_executor_job(_begin)
try:
yield
except BaseException:
await self._hass.async_add_executor_job(_rollback)
raise
def _rollback() -> None:
conn.execute("ROLLBACK")
def _commit() -> None:
conn.execute("COMMIT")
await self._hass.async_add_executor_job(_rollback)
raise
await self._hass.async_add_executor_job(_commit)
def _commit() -> None:
conn.execute("COMMIT")
await self._hass.async_add_executor_job(_commit)
finally:
self._in_transaction = False
self._tx_task = None
async def close(self) -> None:
"""Close the connection in the executor; safe to call repeatedly."""
conn, self._conn, self._cursor = self._conn, None, None
if conn is None:
return
"""Close the connection in the executor; safe to call repeatedly.
def _close() -> None:
conn.close()
Serialized on the wrapper lock, so an in-flight operation or open
transaction always completes before the connection is closed.
"""
if (
self._in_transaction
and self._tx_task is not None
and asyncio.current_task() is self._tx_task
):
raise RuntimeError(
"Cannot close the NestQuest database from inside its own "
"transaction; exit the transaction first"
)
async with self._get_lock():
conn, self._conn = self._conn, None
self._rowcount, self._lastrowid = -1, None
if conn is None:
return
await self._hass.async_add_executor_job(_close)
def _close() -> None:
conn.close()
await self._hass.async_add_executor_job(_close)
def _require_conn(self) -> sqlite3.Connection:
if self._conn is None:
tests/test_db.py
+281 -2
View File
@@ -3,7 +3,10 @@ from __future__ import annotations
import ast
import asyncio
import sqlite3
import threading
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -101,7 +104,7 @@ def test_every_method_routes_sqlite3_through_executor(tmp_path) -> None:
assert row == ("a",)
assert rows == [("a",), ("b",), ("c",)]
assert hass.async_add_executor_job.await_count == calls_before + 7
assert hass.async_add_executor_job.await_count == calls_before + 5
_run(database.close())
@@ -118,11 +121,38 @@ async def test_every_wrapper_call_awaits_executor_job(tmp_path) -> None:
async with database.transaction():
pass
assert hass.async_add_executor_job.await_count == 9
assert hass.async_add_executor_job.await_count == 7
assert all(callable(c.args[0]) for c in hass.async_add_executor_job.call_args_list)
await database.close()
async def test_public_methods_return_only_plain_data(tmp_path) -> None:
"""execute/execute_many return plain dataclasses, never sqlite3 objects."""
hass = _make_hass_mock()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "plain.db")
result = await database.execute(
"CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"
)
assert not isinstance(result, (sqlite3.Connection, sqlite3.Cursor))
insert = await database.execute("INSERT INTO t (name) VALUES (?)", ("a",))
assert not isinstance(insert, (sqlite3.Connection, sqlite3.Cursor))
assert insert.rowcount == 1
assert insert.lastrowid == 1
many = await database.execute_many(
"INSERT INTO t (name) VALUES (?)", [("b",), ("c",)]
)
assert not isinstance(many, (sqlite3.Connection, sqlite3.Cursor))
assert many.rowcount == 2
row = await database.fetch_one("SELECT name FROM t WHERE id = 1")
assert row == ("a",)
rows = await database.fetch_all("SELECT name FROM t ORDER BY id")
assert rows == [("a",), ("b",), ("c",)]
for value in (*rows, row):
assert not isinstance(value, sqlite3.Cursor)
await database.close()
async def test_no_sqlite3_call_on_event_loop_with_real_executor(tmp_path) -> None:
"""With the conftest hass (real run_in_executor), queries still succeed."""
hass, _registry = make_hass()
@@ -133,6 +163,68 @@ async def test_no_sqlite3_call_on_event_loop_with_real_executor(tmp_path) -> Non
await database.close()
async def test_real_executor_never_touches_loop_thread(tmp_path, monkeypatch) -> None:
"""Every real sqlite3 call must run on a non-event-loop thread.
sqlite3.connect is instrumented so a proxy wraps the connection and
every cursor it hands out; each proxied call records the thread it ran
on. A recorded thread id equal to the event-loop thread would mean a
sqlite3 call crossed the executor boundary.
"""
loop_thread = threading.get_ident()
recorded: list[int] = []
guard = threading.Lock()
def _note() -> None:
with guard:
recorded.append(threading.get_ident())
class _Proxy:
"""Generic proxy recording the thread of every attribute call."""
def __init__(self, target: Any) -> None:
object.__setattr__(self, "_target", target)
def __getattr__(self, name: str):
attr = getattr(object.__getattribute__(self, "_target"), name)
if not callable(attr):
return attr
def _wrapped(*args, **kwargs):
_note()
result = attr(*args, **kwargs)
if isinstance(result, sqlite3.Cursor):
return _Proxy(result)
return result
return _wrapped
real_connect = sqlite3.connect
def _traced_connect(*args, **kwargs):
return _Proxy(real_connect(*args, **kwargs))
monkeypatch.setattr(db_module.sqlite3, "connect", _traced_connect)
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "threads.db")
await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)")
await database.execute("INSERT INTO t (id) VALUES (?)", (1,))
await database.execute_many("INSERT INTO t (id) VALUES (?)", [(2,), (3,)])
assert await database.fetch_one("SELECT id FROM t WHERE id = 1") == (1,)
assert await database.fetch_all("SELECT id FROM t") == [(1,), (2,), (3,)]
async with database.transaction():
await database.execute("INSERT INTO t (id) VALUES (?)", (4,))
await database.close()
assert recorded, "expected at least one recorded sqlite3 call"
assert loop_thread not in recorded, (
f"sqlite3 call ran on the event-loop thread {loop_thread}; "
f"recorded threads: {sorted(set(recorded))}"
)
# ---------------------------------------------------------------------------
# Source scan: sqlite3.* only inside executor-delegated functions
# ---------------------------------------------------------------------------
@@ -205,6 +297,8 @@ def test_close_is_idempotent(tmp_path) -> None:
_run(database.close())
assert database.connected is False
assert hass.async_add_executor_job.await_count == 2
assert database.rowcount == -1
assert database.lastrowid is None
def test_operations_after_close_raise_clear_error(tmp_path) -> None:
@@ -279,6 +373,191 @@ async def test_transaction_persists_across_connections(tmp_path) -> None:
await other.close()
# ---------------------------------------------------------------------------
# Serialization: one lock per wrapper, no cross-commit interleaving
# ---------------------------------------------------------------------------
async def test_concurrent_writers_do_not_interleave_commits(tmp_path) -> None:
"""Real-executor proof: nothing commits inside another transaction's span.
A transaction runner repeatedly counts rows, inserts, and counts again
inside BEGIN..COMMIT; the count must grow by exactly one, proving no
other writer committed in between even though several writer tasks fire
statements concurrently on real executor threads. Statements issued
while a transaction is open queue behind it on the wrapper lock.
"""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "serial.db")
await database.execute("CREATE TABLE t (kind TEXT, seq INTEGER)")
tx_rounds, writers, writer_rounds = 6, 5, 8
total = tx_rounds + writers * writer_rounds
async def _tx_runner() -> None:
for rnd in range(tx_rounds):
async with database.transaction():
rows = await database.fetch_all("SELECT COUNT(*) FROM t")
seen = rows[0][0] if rows else -1
await database.execute(
"INSERT INTO t (kind, seq) VALUES (?, ?)", ("tx", rnd)
)
rows = await database.fetch_all("SELECT COUNT(*) FROM t")
grown = rows[0][0] if rows else -1
if grown != seen + 1:
raise AssertionError(
f"interleaved commit: count {seen} -> {grown} "
f"in tx round {rnd}"
)
async def _writer(worker: int) -> None:
for rnd in range(writer_rounds):
await database.execute(
"INSERT INTO t (kind, seq) VALUES (?, ?)", ("w", worker)
)
await asyncio.gather(
_tx_runner(), *(_writer(w) for w in range(writers))
)
rows = await database.fetch_all("SELECT COUNT(*) FROM t")
assert rows[0][0] == total, f"expected {total} rows, got {rows[0][0]}"
await database.close()
async def test_transaction_span_excludes_concurrent_transaction(tmp_path) -> None:
"""A second transaction caller is rejected with a clear error."""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "reject.db")
await database.execute("CREATE TABLE t (id INTEGER)")
started = asyncio.Event()
async def _first() -> None:
async with database.transaction():
started.set()
await asyncio.sleep(0.05)
first = asyncio.create_task(_first())
await started.wait()
with pytest.raises(RuntimeError, match="transaction"):
async with database.transaction():
pass
await first
await database.execute("INSERT INTO t (id) VALUES (?)", (1,))
assert await database.fetch_one("SELECT id FROM t") == (1,)
await database.close()
async def test_transaction_blocks_other_statements_until_commit(tmp_path) -> None:
"""A statement issued during an open transaction queues behind it."""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "block.db")
await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
order: list[str] = []
started = asyncio.Event()
release = asyncio.Event()
async def _txn() -> None:
async with database.transaction():
await database.execute("INSERT INTO t (name) VALUES (?)", ("in_tx",))
order.append("tx-start")
started.set()
await release.wait()
order.append("tx-end")
async def _outsider() -> None:
await started.wait()
await database.execute("INSERT INTO t (name) VALUES (?)", ("after",))
order.append("outside")
txn_task = asyncio.create_task(_txn())
await started.wait()
outsider_task = asyncio.create_task(_outsider())
await asyncio.sleep(0.05)
assert order == ["tx-start"], f"outsider ran inside transaction: {order}"
release.set()
await asyncio.gather(txn_task, outsider_task)
assert order == ["tx-start", "tx-end", "outside"]
rows = await database.fetch_all("SELECT name FROM t ORDER BY id")
assert rows == [("in_tx",), ("after",)]
await database.close()
async def test_concurrent_transaction_is_rejected_with_clear_error(
tmp_path,
) -> None:
"""A second concurrent transaction() gets a clear RuntimeError."""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "concurrent-tx.db")
await database.execute("CREATE TABLE t (id INTEGER)")
started = asyncio.Event()
async def _first() -> None:
async with database.transaction():
started.set()
await asyncio.sleep(0.05)
first = asyncio.create_task(_first())
await started.wait()
with pytest.raises(RuntimeError, match="transaction"):
async with database.transaction():
pass
await first
await database.close()
async def test_close_waits_for_in_flight_transaction(tmp_path) -> None:
"""close() serializes: an open transaction finishes before shutdown."""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "closerace.db")
await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
started = asyncio.Event()
release = asyncio.Event()
async def _txn() -> None:
async with database.transaction():
await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
started.set()
await release.wait()
txn_task = asyncio.create_task(_txn())
await started.wait()
close_task = asyncio.create_task(database.close())
await asyncio.sleep(0.05)
assert database.connected is True, "close() raced an open transaction"
release.set()
await asyncio.gather(txn_task, close_task)
assert database.connected is False
witness = NestQuestDatabase(hass)
await witness.open(tmp_path / "closerace.db")
assert await witness.fetch_all("SELECT name FROM t") == [("kept",)]
await witness.close()
async def test_close_during_transaction_from_same_task_is_rejected(
tmp_path,
) -> None:
"""close() inside its own transaction body raises instead of deadlocking."""
hass, _registry = make_hass()
database = NestQuestDatabase(hass)
await database.open(tmp_path / "selfclose.db")
await database.execute("CREATE TABLE t (id INTEGER)")
with pytest.raises(RuntimeError, match="transaction"):
async with database.transaction():
await database.close()
assert database.connected is True
await database.close()
# ---------------------------------------------------------------------------
# rowcount / lastrowid surface
# ---------------------------------------------------------------------------
tests/test_lifecycle.py
+41 -1
View File
@@ -3,6 +3,8 @@ from __future__ import annotations
import asyncio
import pytest
from conftest import make_config_entry, make_hass, wire_entry_to_registry
from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
@@ -196,4 +198,42 @@ async def test_setup_awaitable_listener_removal_supported() -> None:
assert result is True
assert removed == [True]
assert hass.data == {}
assert entry.runtime_data is None
assert entry.runtime_data is None
async def test_setup_closes_database_when_listener_registration_fails() -> None:
"""A listener-registration failure must not leak the opened database."""
from custom_components.nestquest.db import NestQuestDatabase
class _Boom(Exception):
pass
entry = make_config_entry(entry_id="listener-boom")
entry.add_update_listener = lambda listener: (_ for _ in ()).throw(_Boom())
hass, _registry = make_hass()
with pytest.raises(_Boom):
await async_setup_entry(hass, entry)
assert hass.data.get(DOMAIN, {}) == {}
assert not hasattr(entry, "runtime_data")
async def test_unload_closes_database_even_when_listener_removal_raises() -> None:
"""Unload removes the listener, closes the DB in a finally, then raises."""
from custom_components.nestquest.db import NestQuestDatabase
class _RemoveBoom(Exception):
pass
def _remover():
raise _RemoveBoom()
entry = make_config_entry(entry_id="remove-boom")
entry.add_update_listener = lambda listener: _remover
hass, _registry = make_hass()
await async_setup_entry(hass, entry)
database = entry.runtime_data.database
with pytest.raises(_RemoveBoom):
await async_unload_entry(hass, entry)
assert database.connected is False, "database leaked despite remover error"
assert entry.runtime_data is None
assert DOMAIN not in hass.data