fix: settle all executor jobs before releasing the wrapper lock
This commit is contained in:
@@ -41,6 +41,13 @@ class NestQuestDatabase:
|
|||||||
not supported: a second ``transaction()`` entered while one is active
|
not supported: a second ``transaction()`` entered while one is active
|
||||||
raises :class:`RuntimeError`. ``close()`` serializes on the same lock
|
raises :class:`RuntimeError`. ``close()`` serializes on the same lock
|
||||||
and is idempotent.
|
and is idempotent.
|
||||||
|
|
||||||
|
Every connection-affecting executor job (connect, pragmas, statements,
|
||||||
|
transaction lifecycle and close) is awaited through the settlement
|
||||||
|
helper :meth:`_await_settled`: if the awaiting task is cancelled while
|
||||||
|
a job is in flight, the job still settles on the worker thread before
|
||||||
|
the cancellation propagates, so the wrapper lock is never released
|
||||||
|
while the executor thread is mid-job.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, hass: HomeAssistant) -> None:
|
def __init__(self, hass: HomeAssistant) -> None:
|
||||||
@@ -122,7 +129,9 @@ class NestQuestDatabase:
|
|||||||
raise
|
raise
|
||||||
return conn
|
return conn
|
||||||
|
|
||||||
self._conn = await self._hass.async_add_executor_job(_open)
|
self._conn = await self._await_settled(
|
||||||
|
self._hass.async_add_executor_job(_open)
|
||||||
|
)
|
||||||
return self
|
return self
|
||||||
|
|
||||||
async def execute(
|
async def execute(
|
||||||
@@ -136,7 +145,9 @@ class NestQuestDatabase:
|
|||||||
cursor = conn.execute(sql, parameters)
|
cursor = conn.execute(sql, parameters)
|
||||||
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
|
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
|
||||||
|
|
||||||
result = await self._hass.async_add_executor_job(_execute)
|
result = await self._await_settled(
|
||||||
|
self._hass.async_add_executor_job(_execute)
|
||||||
|
)
|
||||||
self._rowcount = result.rowcount
|
self._rowcount = result.rowcount
|
||||||
self._lastrowid = result.lastrowid
|
self._lastrowid = result.lastrowid
|
||||||
return result
|
return result
|
||||||
@@ -152,7 +163,9 @@ class NestQuestDatabase:
|
|||||||
cursor = conn.executemany(sql, parameters)
|
cursor = conn.executemany(sql, parameters)
|
||||||
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
|
return ExecutionResult(cursor.rowcount, cursor.lastrowid)
|
||||||
|
|
||||||
result = await self._hass.async_add_executor_job(_execute_many)
|
result = await self._await_settled(
|
||||||
|
self._hass.async_add_executor_job(_execute_many)
|
||||||
|
)
|
||||||
self._rowcount = result.rowcount
|
self._rowcount = result.rowcount
|
||||||
self._lastrowid = result.lastrowid
|
self._lastrowid = result.lastrowid
|
||||||
return result
|
return result
|
||||||
@@ -168,8 +181,8 @@ class NestQuestDatabase:
|
|||||||
cursor = conn.execute(sql, parameters)
|
cursor = conn.execute(sql, parameters)
|
||||||
return cursor.fetchone(), cursor.rowcount, cursor.lastrowid
|
return cursor.fetchone(), cursor.rowcount, cursor.lastrowid
|
||||||
|
|
||||||
row, rowcount, lastrowid = await self._hass.async_add_executor_job(
|
row, rowcount, lastrowid = await self._await_settled(
|
||||||
_fetch_one
|
self._hass.async_add_executor_job(_fetch_one)
|
||||||
)
|
)
|
||||||
self._rowcount = rowcount
|
self._rowcount = rowcount
|
||||||
self._lastrowid = lastrowid
|
self._lastrowid = lastrowid
|
||||||
@@ -186,8 +199,8 @@ class NestQuestDatabase:
|
|||||||
cursor = conn.execute(sql, parameters)
|
cursor = conn.execute(sql, parameters)
|
||||||
return cursor.fetchall(), cursor.rowcount, cursor.lastrowid
|
return cursor.fetchall(), cursor.rowcount, cursor.lastrowid
|
||||||
|
|
||||||
rows, rowcount, lastrowid = await self._hass.async_add_executor_job(
|
rows, rowcount, lastrowid = await self._await_settled(
|
||||||
_fetch_all
|
self._hass.async_add_executor_job(_fetch_all)
|
||||||
)
|
)
|
||||||
self._rowcount = rowcount
|
self._rowcount = rowcount
|
||||||
self._lastrowid = lastrowid
|
self._lastrowid = lastrowid
|
||||||
@@ -367,7 +380,7 @@ class NestQuestDatabase:
|
|||||||
def _close() -> None:
|
def _close() -> None:
|
||||||
conn.close()
|
conn.close()
|
||||||
|
|
||||||
await self._hass.async_add_executor_job(_close)
|
await self._await_settled(self._hass.async_add_executor_job(_close))
|
||||||
|
|
||||||
def _require_conn(self) -> sqlite3.Connection:
|
def _require_conn(self) -> sqlite3.Connection:
|
||||||
if self._conn is None:
|
if self._conn is None:
|
||||||
|
|||||||
@@ -525,6 +525,73 @@ async def test_cancel_during_commit_settles_then_propagates_cancellation(
|
|||||||
await database.close()
|
await database.close()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_cancel_during_in_transaction_execute_settles_then_rolls_back(
|
||||||
|
tmp_path,
|
||||||
|
) -> None:
|
||||||
|
"""NQ-PR12-009: cancel during an in-transaction statement must not race ROLLBACK.
|
||||||
|
|
||||||
|
The in-transaction INSERT executor job is gated: the transaction task
|
||||||
|
is cancelled while the statement is still in flight, then the worker
|
||||||
|
is allowed to complete it anyway. The ROLLBACK must only run after
|
||||||
|
the statement has settled — while the statement is in flight the
|
||||||
|
transaction is still open — and the cancelled statement's write must
|
||||||
|
not persist.
|
||||||
|
"""
|
||||||
|
hass, _registry = make_hass()
|
||||||
|
database = NestQuestDatabase(hass)
|
||||||
|
await database.open(tmp_path / "cancel-execute.db")
|
||||||
|
await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
|
||||||
|
await database.execute("INSERT INTO t (name) VALUES (?)", ("before",))
|
||||||
|
|
||||||
|
insert_started = asyncio.Event()
|
||||||
|
release_insert = asyncio.Event()
|
||||||
|
real_add = hass.async_add_executor_job
|
||||||
|
|
||||||
|
async def _gated_add_executor_job(fn, *args, **kwargs):
|
||||||
|
if getattr(fn, "__name__", "") == "_execute":
|
||||||
|
insert_started.set()
|
||||||
|
await release_insert.wait()
|
||||||
|
return await real_add(fn, *args, **kwargs)
|
||||||
|
|
||||||
|
hass.async_add_executor_job = _gated_add_executor_job
|
||||||
|
|
||||||
|
async def _txn() -> None:
|
||||||
|
async with database.transaction():
|
||||||
|
await database.execute("INSERT INTO t (name) VALUES (?)", ("lost",))
|
||||||
|
|
||||||
|
txn_task = asyncio.create_task(_txn())
|
||||||
|
await insert_started.wait()
|
||||||
|
txn_task.cancel()
|
||||||
|
await asyncio.sleep(0.05)
|
||||||
|
assert database._in_transaction is True, (
|
||||||
|
"ownership was released while the cancelled statement was in flight"
|
||||||
|
)
|
||||||
|
assert database._conn is not None
|
||||||
|
assert database._conn.in_transaction is True, (
|
||||||
|
"ROLLBACK ran before the cancelled statement settled"
|
||||||
|
)
|
||||||
|
release_insert.set()
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await txn_task
|
||||||
|
|
||||||
|
assert database._in_transaction is False
|
||||||
|
assert database._tx_task is None
|
||||||
|
assert database._conn is not None
|
||||||
|
assert database._conn.in_transaction is False
|
||||||
|
|
||||||
|
hass.async_add_executor_job = real_add
|
||||||
|
assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
|
||||||
|
("before",)
|
||||||
|
]
|
||||||
|
async with database.transaction():
|
||||||
|
await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
|
||||||
|
assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
|
||||||
|
("before",),
|
||||||
|
("kept",),
|
||||||
|
]
|
||||||
|
await database.close()
|
||||||
|
|
||||||
|
|
||||||
async def test_commit_and_rollback_failures_are_both_preserved(tmp_path) -> None:
|
async def test_commit_and_rollback_failures_are_both_preserved(tmp_path) -> None:
|
||||||
"""NQ-PR12-008: both errors survive when COMMIT and ROLLBACK both fail.
|
"""NQ-PR12-008: both errors survive when COMMIT and ROLLBACK both fail.
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user