From ea4e19140087d6133ed8e02f938c08825b76cec3 Mon Sep 17 00:00:00 2001 From: Shane Maynard Date: Sun, 4 Oct 2026 08:32:33 -0400 Subject: [PATCH] fix: serialize SQLite close with cancelled worker transactions --- control/accounts.py | 6 ++- tests/unit/test_chat_packaging.py | 4 +- tests/unit/test_chat_security_review.py | 71 +++++++++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/control/accounts.py b/control/accounts.py index 92a1f88..4531857 100644 --- a/control/accounts.py +++ b/control/accounts.py @@ -103,7 +103,11 @@ class Accounts: os.chmod(companion, 0o600) def close(self): - self.db.close() + # Cancelling an asyncio.to_thread caller does not stop its SQLite worker. + # ChatStore/FeatureStore share this connection and lock; wait for any + # in-flight transaction before closing the native connection. + with self.lock: + self.db.close() def users(self): with self.lock: diff --git a/tests/unit/test_chat_packaging.py b/tests/unit/test_chat_packaging.py index 6475542..9858725 100644 --- a/tests/unit/test_chat_packaging.py +++ b/tests/unit/test_chat_packaging.py @@ -74,4 +74,6 @@ def test_ci_qualification_gates_images_and_deployment_without_live_conftest(): assert 'tests/unit/'+name in workflow assert 'pytest-asyncio -r control/requirements.txt' in workflow assert 'scripts/generate_hope_themes.py --check' in workflow - assert '-v "$PWD:/src:ro"' in workflow + assert 'git archive HEAD | docker run --rm -i' in workflow + assert 'tar -xf - -C /tmp/src' in workflow + assert '-v "$PWD:/src' not in workflow diff --git a/tests/unit/test_chat_security_review.py b/tests/unit/test_chat_security_review.py index 65780f8..390274d 100644 --- a/tests/unit/test_chat_security_review.py +++ b/tests/unit/test_chat_security_review.py @@ -1,6 +1,8 @@ """Offline cancellation/recovery regressions; no HTTP, NAS, secrets or model calls.""" import asyncio import functools +import sqlite3 +import threading import pytest @@ -66,6 +68,75 @@ def execution(tmp_path): accounts.close() +@offline_async +async def test_cancelled_sqlite_worker_is_joined_by_store_close(execution, monkeypatch, tmp_path): + """Task cancellation is not thread completion: close must serialize with SQL.""" + accounts, store, owner, _, _, rid = execution + entered = threading.Event() + release = threading.Event() + close_attempted = threading.Event() + close_entered = threading.Event() + original_lock, original_db = accounts.lock, accounts.db + original_row = store._row + closing_thread = threading.local() + + class ObservedLock: + def __enter__(self): + if getattr(closing_thread, 'closing', False): + close_attempted.set() + original_lock.acquire() + return self + + def __exit__(self, *args): + original_lock.release() + + class ObservedConnection: + def close(self): + close_entered.set() + close_attempted.set() + # Fail safely on the old implementation, without racing native SQL + # close in the test process and potentially segfaulting pytest. + assert original_lock._is_owned(), 'SQLite close must hold the shared store lock' + original_db.close() + + def blocked_row(*args): + entered.set() + assert release.wait(5), 'Offline worker release timed out' + return original_row(*args) + + def close(): + closing_thread.closing = True + accounts.close() + + accounts.lock = store.lock = ObservedLock() + monkeypatch.setattr(accounts, 'db', ObservedConnection()) + monkeypatch.setattr(store, '_row', blocked_row) + worker = asyncio.create_task(asyncio.to_thread(store.append_event, owner, rid, 'done', {'offline': True})) + closer = None + try: + assert await asyncio.to_thread(entered.wait, 2) + worker.cancel() + with pytest.raises(asyncio.CancelledError): + await worker + closer = asyncio.create_task(asyncio.to_thread(close)) + assert await asyncio.to_thread(close_attempted.wait, 2) + assert not close_entered.is_set(), 'Close must wait for the cancelled task\'s SQLite worker' + finally: + release.set() + try: + if closer is not None: + await closer + finally: + monkeypatch.setattr(accounts, 'db', original_db) + accounts.lock = store.lock = original_lock + assert close_entered.is_set() + # The worker's terminal replay event committed before close; no test bypass + # or change to the durable run/event contract is needed. + with sqlite3.connect(str(tmp_path / 'offline.sqlite')) as reopened: + assert reopened.execute('SELECT event, data FROM chat_events WHERE run_id=?', (rid,)).fetchall() == [('done', '{"offline": true}')] + accounts.close() # Teardown close remains idempotent. + + def coordinator(execution, adapter, *, upstream=None): accounts, store, owner, token, cid, rid = execution runs = ChatRuns(accounts, store, adapter=adapter)